POC: command bus

src/Attribute/Handle.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Attribute;
+
+use Attribute;
+
+#[Attribute(Attribute::TARGET_METHOD | Attribute::IS_REPEATABLE)]
+final class Handle
+{
+    /** @param class-string|null $commandClass */
+    public function __construct(
+        public readonly string|null $commandClass = null,
+    ) {
+    }
+}

src/Attribute/HandledBy.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Attribute;
+
+use Attribute;
+use Patchlevel\EventSourcing\Aggregate\AggregateRoot;
+
+#[Attribute(Attribute::TARGET_CLASS)]
+final class HandledBy
+{
+    /** @param class-string<AggregateRoot> $aggregateClass */
+    public function __construct(
+        public readonly string $aggregateClass,
+    ) {
+    }
+}

src/CommandBus/AggregateHandlerProvider.php

@@ -0,0 +1,88 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\CommandBus;
+
+use Patchlevel\EventSourcing\Aggregate\AggregateRoot;
+use Patchlevel\EventSourcing\Attribute\Handle;
+use Patchlevel\EventSourcing\Attribute\HandledBy;
+use Patchlevel\EventSourcing\CommandBus\Handler\DefaultHandlerFactory;
+use ReflectionClass;
+use ReflectionMethod;
+
+final class AggregateHandlerProvider implements HandlerProvider
+{
+    public function __construct(
+        private readonly DefaultHandlerFactory $handlerFactory,
+    ) {
+    }
+
+    /**
+     * @param class-string $commandClass
+     *
+     * @throws HandlerNotFound
+     */
+    public function handlerForCommand(string $commandClass): HandlerDescriptor
+    {
+        $aggregateClass = $this->aggregateClass($commandClass);
+
+        $reflectionClass = new ReflectionClass($aggregateClass);
+
+        foreach ($reflectionClass->getMethods() as $method) {
+            $attributes = $method->getAttributes(Handle::class);
+
+            if ($attributes === []) {
+                continue;
+            }
+
+            $handleClass = $this->handleClass($attributes[0]->newInstance(), $method);
+
+            if ($handleClass !== $commandClass) {
+                continue;
+            }
+
+            if ($method->isStatic()) {
+                return new HandlerDescriptor($this->handlerFactory->createHandler($aggregateClass, $method->getName()));
+            }
+
+            return new HandlerDescriptor($this->handlerFactory->updateHandler($aggregateClass, $method->getName()));
+        }
+
+        throw new HandlerNotFound($commandClass);
+    }
+
+    /**
+     * @param class-string $commandClass
+     *
+     * @return class-string<AggregateRoot>
+     */
+    private function aggregateClass(string $commandClass): string
+    {
+        $reflectionClass = new ReflectionClass($commandClass);
+        $attributes = $reflectionClass->getAttributes(HandledBy::class);
+
+        if ($attributes === []) {
+            throw new HandlerNotFound($commandClass);
+        }
+
+        $handledBy = $attributes[0]->newInstance();
+
+        return $handledBy->aggregateClass;
+    }
+
+    private function handleClass(Handle $handle, ReflectionMethod $reflectionMethod): string|null
+    {
+        if ($handle->commandClass !== null) {
+            return $handle->commandClass;
+        }
+
+        $parameters = $reflectionMethod->getParameters();
+
+        if ($parameters === []) {
+            return null;
+        }
+
+        return $parameters[0]->getType()->getName();
+    }
+}

src/CommandBus/CommandBus.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\CommandBus;
+
+interface CommandBus
+{
+    public function dispatch(object $command): void;
+}

src/CommandBus/DefaultCommandBus.php

@@ -0,0 +1,71 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\CommandBus;
+
+use Patchlevel\EventSourcing\CommandBus\Handler\DefaultHandlerFactory;
+use Patchlevel\EventSourcing\Repository\RepositoryManager;
+use Psr\Log\LoggerInterface;
+
+use function array_shift;
+use function sprintf;
+
+final class DefaultCommandBus implements CommandBus
+{
+    /** @var array<object> */
+    private array $queue;
+    private bool $processing;
+
+    public function __construct(
+        private readonly HandlerProvider $handlerProvider,
+        private readonly LoggerInterface|null $logger = null,
+    ) {
+        $this->queue = [];
+        $this->processing = false;
+    }
+
+    public function dispatch(object $command): void
+    {
+        $this->logger?->debug(sprintf(
+            'CommandBus: Add message "%s" to queue.',
+            $command::class,
+        ));
+
+        $this->queue[] = $command;
+
+        if ($this->processing) {
+            $this->logger?->debug('CommandBus: Is already processing, dont start new processing.');
+
+            return;
+        }
+
+        try {
+            $this->processing = true;
+
+            $this->logger?->debug('CommandBus: Start processing queue.');
+
+            while ($command = array_shift($this->queue)) {
+                $handler = $this->handlerProvider->handlerForCommand($command::class);
+
+                ($handler->callable())($command);
+            }
+        } finally {
+            $this->processing = false;
+
+            $this->logger?->debug('CommandBus: Finished processing queue.');
+        }
+    }
+
+    public static function createDefault(
+        RepositoryManager $repositoryManager,
+        LoggerInterface|null $logger = null,
+    ): self {
+        return new self(
+            new AggregateHandlerProvider(
+                new DefaultHandlerFactory($repositoryManager),
+            ),
+            $logger,
+        );
+    }
+}

src/CommandBus/Handler/AggregateIdNotFound.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\CommandBus\Handler;
+
+use RuntimeException;
+
+use function sprintf;
+
+final class AggregateIdNotFound extends RuntimeException
+{
+    /** @param class-string $commandClass */
+    public function __construct(string $commandClass)
+    {
+        parent::__construct(sprintf('Missing `Id` Attribute in command %s', $commandClass));
+    }
+}

src/CommandBus/Handler/CreateAggregateHandler.php

@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\CommandBus\Handler;
+
+use Patchlevel\EventSourcing\Aggregate\AggregateRoot;
+use Patchlevel\EventSourcing\Repository\RepositoryManager;
+
+final class CreateAggregateHandler
+{
+    /** @param class-string<AggregateRoot> $aggregateClass */
+    public function __construct(
+        private readonly RepositoryManager $repositoryManager,
+        private readonly string $aggregateClass,
+        private readonly string $methodName,
+    ) {
+    }
+
+    public function __invoke(object $command): void
+    {
+        $repository = $this->repositoryManager->get($this->aggregateClass);
+
+        $aggregate = $this->aggregateClass::{$this->methodName}($command);
+
+        $repository->save($aggregate);
+    }
+}

src/CommandBus/Handler/DefaultHandlerFactory.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\CommandBus\Handler;
+
+use Patchlevel\EventSourcing\Repository\RepositoryManager;
+
+final class DefaultHandlerFactory implements HandlerFactory
+{
+    public function __construct(
+        private readonly RepositoryManager $repositoryManager,
+    ) {
+    }
+
+    public function createHandler(string $aggregateClass, string $method): callable
+    {
+        return new CreateAggregateHandler(
+            $this->repositoryManager,
+            $aggregateClass,
+            $method,
+        );
+    }
+
+    public function updateHandler(string $aggregateClass, string $method): callable
+    {
+        return new UpdateAggregateHandler(
+            $this->repositoryManager,
+            $aggregateClass,
+            $method,
+        );
+    }
+}

src/CommandBus/Handler/HandlerFactory.php

@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\CommandBus\Handler;
+
+use Patchlevel\EventSourcing\Aggregate\AggregateRoot;
+
+interface HandlerFactory
+{
+    /**
+     * @param class-string<AggregateRoot> $aggregateClass
+     *
+     * @return callable(object): void
+     */
+    public function createHandler(string $aggregateClass, string $method): callable;
+
+    /**
+     * @param class-string<AggregateRoot> $aggregateClass
+     *
+     * @return callable(object): void
+     */
+    public function updateHandler(string $aggregateClass, string $method): callable;
+}

src/CommandBus/Handler/UpdateAggregateHandler.php

@@ -0,0 +1,51 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\CommandBus\Handler;
+
+use Patchlevel\EventSourcing\Aggregate\AggregateRoot;
+use Patchlevel\EventSourcing\Aggregate\AggregateRootId;
+use Patchlevel\EventSourcing\Attribute\Id;
+use Patchlevel\EventSourcing\Repository\RepositoryManager;
+use ReflectionClass;
+
+final class UpdateAggregateHandler
+{
+    /** @param class-string<AggregateRoot> $aggregateClass */
+    public function __construct(
+        private readonly RepositoryManager $repositoryManager,
+        private readonly string $aggregateClass,
+        private readonly string $methodName,
+    ) {
+    }
+
+    public function __invoke(object $command): void
+    {
+        $aggregateRootId = $this->aggregateRootId($command);
+        $repository = $this->repositoryManager->get($this->aggregateClass);
+
+        $aggregate = $repository->load($aggregateRootId);
+
+        $aggregate->{$this->methodName}($command);
+
+        $repository->save($aggregate);
+    }
+
+    private function aggregateRootId(object $command): AggregateRootId
+    {
+        $reflectionClass = new ReflectionClass($command);
+
+        foreach ($reflectionClass->getProperties() as $property) {
+            $attributes = $property->getAttributes(Id::class);
+
+            if ($attributes === []) {
+                continue;
+            }
+
+            return $property->getValue($command);
+        }
+
+        throw new AggregateIdNotFound($command::class);
+    }
+}