IdentityCausationData.php

<?php

declare(strict_types=1);

namespace Funct\Ion\Common\Library\Cqrs\ServiceBus;

use JsonSerializable;
use Prooph\Common\Messaging\Message;

final class IdentityCausationData implements JsonSerializable
{
    public const PROPERTY_IDENTITY_ID = 'causation_identity';

    public const PROPERTY_IDENTITY_NAME = 'causation_identity_name';

    public const PROPERTY_IP = 'causation_ip';

    public static function withData(string $identityId, string $identityName, string $ipAddress): self
    {
        return new self($identityId, $identityName, $ipAddress);
    }

    public static function fromMessage(Message $message): ?self
    {
        return self::fromArray($message->metadata());
    }

    public static function fromArray(array $data): ?self
    {
        if (!array_key_exists(self::PROPERTY_IDENTITY_ID, $data)) {
            return null;
        }

        $name = null;
        if (array_key_exists(self::PROPERTY_IDENTITY_NAME, $data) && !empty($data[self::PROPERTY_IDENTITY_NAME])) {
            $name = $data[self::PROPERTY_IDENTITY_NAME];
        }

        $ip = null;
        if(array_key_exists(self::PROPERTY_IP, $data) && !empty($data[self::PROPERTY_IP])) {
            $ip = $data[self::PROPERTY_IP];
        }

        return new self($data[self::PROPERTY_IDENTITY_ID], $name, $ip);
    }

    private $identityId;

    private $identityName;

    private $ip;

    private function __construct(string $identity, ?string $identityName, ?string $ip)
    {
        $this->identityId = $identity;
        $this->identityName = $identityName;
        $this->ip = $ip;
    }

    public function identityId(): string
    {
        return $this->identityId;
    }

    public function identityName(): ?string
    {
        return $this->identityName;
    }

    public function ip(): ?string
    {
        return $this->ip;
    }

    public function toArray(): array
    {
        return [
            self::PROPERTY_IDENTITY_ID => $this->identityId,
            self::PROPERTY_IDENTITY_NAME => $this->identityName,
            self::PROPERTY_IP => $this->ip,
        ];
    }

    public function jsonSerialize(): array
    {
        return $this->toArray();
    }

    public function injectInto(Message $message): Message
    {
        $message = $message->withAddedMetadata(
            self::PROPERTY_IDENTITY_ID,
            $this->identityId
        );

        if (null !== $this->identityName) {
            $message = $message->withAddedMetadata(
                self::PROPERTY_IDENTITY_NAME,
                $this->identityName
            );
        }

        if (null !== $this->ip) {
            $message = $message->withAddedMetadata(
                self::PROPERTY_IP,
                $this->ip
            );
        }

        return $message;
    }
}

InjectSessionContextPlugin.php

<?php

declare(strict_types=1);

namespace Funct\Ion\Common\Library\Cqrs\ServiceBus;

use Funct\Ion\Common\Service\Authentication\SessionContext\CurrentSessionContext;
use geertw\IpAnonymizer\IpAnonymizer;
use Prooph\Common\Event\ActionEvent;
use Prooph\Common\Messaging\Message;
use Prooph\ServiceBus\CommandBus;
use Prooph\ServiceBus\MessageBus;
use Prooph\ServiceBus\Plugin\AbstractPlugin;
use Zend\Http\PhpEnvironment\RemoteAddress;

final class InjectSessionContextPlugin extends AbstractPlugin
{
    public const PRIORITY_INJECT_CONTEXT = 5000;

    private $currentSessionContext;
    public function __construct(CurrentSessionContext $context)
    {
        $this->currentSessionContext = $context;
    }

    public function attachToMessageBus(MessageBus $messageBus): void
    {
        $this->listenerHandlers[] = $messageBus->attach(
            MessageBus::EVENT_DISPATCH,
            function (ActionEvent $actionEvent) {
                return $this->injectSessionContext($actionEvent);
            },
            self::PRIORITY_INJECT_CONTEXT
        );
    }

    private function injectSessionContext(ActionEvent $event)
    {
        $message = $event->getParam(CommandBus::EVENT_PARAM_MESSAGE);
        if (!$message instanceof Message) {
            return;
        }

        if (!$this->currentSessionContext->hasActiveSession()) {
            return;
        }

        $causationData = IdentityCausationData::fromMessage($message);
        if (null !== $causationData) {
            return;
        }

        $identityCausationData = IdentityCausationData::withData(
            $this->currentSessionContext->getIdentityId(),
            $this->currentSessionContext->getIdentityName(),
            $this->getIpAddress()
        );

        $message = $identityCausationData->injectInto($message);
        $event->setParam(CommandBus::EVENT_PARAM_MESSAGE, $message);
    }

    private function getIpAddress()
    {
        $remoteAddressHelper = new RemoteAddress();
        $remoteAddress = $remoteAddressHelper->getIpAddress();

        if (empty($remoteAddress)) {
            return '127.0.0.1';
        }

        return IpAnonymizer::anonymizeIp($remoteAddress);
    }
}

RestoreSessionContextPlugin.php

<?php

declare(strict_types=1);

namespace Funct\Ion\Common\Library\Cqrs\ServiceBus;

use Funct\Ion\Common\Service\Authentication\SessionContext\CurrentSessionContext;
use Funct\Ion\Common\Service\Authentication\SessionContext\SessionContextRepositoryInterface;
use Prooph\Common\Event\ActionEvent;
use Prooph\Common\Messaging\Message;
use Prooph\ServiceBus\CommandBus;
use Prooph\ServiceBus\MessageBus;
use Prooph\ServiceBus\Plugin\AbstractPlugin;

final class RestoreSessionContextPlugin extends AbstractPlugin
{
    private $currentSessionContext;

    private $sessionContextRepository;

    private $inNestedContext;

    public function __construct(CurrentSessionContext $context, SessionContextRepositoryInterface $contextRepository)
    {
        $this->currentSessionContext = $context;
        $this->sessionContextRepository = $contextRepository;
        $this->inNestedContext = false;
    }

    public function attachToMessageBus(MessageBus $messageBus): void
    {
        $this->listenerHandlers[] = $messageBus->attach(
            MessageBus::EVENT_DISPATCH,
            function (ActionEvent $actionEvent) {
                $this->setSessionContext($actionEvent);
            },
            InjectSessionContextPlugin::PRIORITY_INJECT_CONTEXT - 1000
        );
        $this->listenerHandlers[] = $messageBus->attach(
            MessageBus::EVENT_FINALIZE,
            function () {
                $this->restoreSessionContext();
            },
            1000
        );
    }

    private function setSessionContext(ActionEvent $event): void
    {
        $message = $event->getParam(CommandBus::EVENT_PARAM_MESSAGE);
        if (!$message instanceof Message) {
            return;
        }

        $causationData = IdentityCausationData::fromMessage($message);
        $identity = $causationData->identityId();
        if ($this->currentSessionContext->hasActiveSession()) {
            $currentIdentity = $this->currentSessionContext->getIdentityId();
            if ($identity === $currentIdentity) {
                return;
            }
        }

        $context = $this
            ->sessionContextRepository
            ->getSessionContextForIdentity($identity);

        $this->currentSessionContext->startSession($context);
        $this->inNestedContext = true;
    }

    private function restoreSessionContext(): void
    {
        if (false === $this->inNestedContext) {
            return;
        }

        $this->currentSessionContext->endSession();
        $this->inNestedContext = false;
    }
}