Merge pull request #820 from patchlevel/conection-registry-dbal-cleanup-task-handler

allow to pass connection registry in dbal cleanup task handler

Author: DavidBadura

docs/pages/aggregate.md

@@ -365,7 +365,7 @@ final class Profile extends BasicAggregateRoot
 
 ## Shared apply context
 
-When working with [micro-aggregates](./aggregate.md#micro-aggregates), 
+When working with [micro-aggregates](./aggregate.md#micro-aggregates),
 it’s common that events are applied by different aggregates.
 As a result, an aggregate may receive events it does not handle, which can lead to multiple “missing apply” warnings.
 

docs/pages/subscription.md

@@ -422,6 +422,11 @@ Default, we provide the following cleanup tasks for `doctrine/dbal`:
 | `DropIndexTask` | Drops an index from a table. |
 | `DropTableTask` | Drops a table.               |
 
+!!! note
+
+    If you are passing connection registry, you can use the connection name as parameter.
+    The `connectionName` parameter is optional and defaults to the default connection.
+    
 !!! tip
 
     You can create your own cleanup tasks and handler.
@@ -1054,18 +1059,45 @@ Lastly, we have to add the new handler to `DefaultCleaner`,
 which is responsible for cleaning up subscriptions.
 
 ```php
-use Patchlevel\EventSourcing\Subscription\Cleanup\Dbal\DbalCleanupTaskHandler;
 use Patchlevel\EventSourcing\Subscription\Cleanup\DefaultCleaner;
 
 $cleaner = new DefaultCleaner([
-    new DbalCleanupTaskHandler($projectionConnection),
     new MongodbCleanupTaskHandler($mongodbDatabase),
 ]);
 ```
 !!! warning
 
     You need to pass the Cleaner to the Subscription Engine.
 
+#### Dbal Cleanup Task Handler
+
+We provide a Dbal cleanup task handler by default.
+More information about the available tasks can be found in the [Dbal Cleanup Tasks](#dbal-cleanup-tasks) documentation.
+
+```php
+use Doctrine\Dbal\Connection;
+use Patchlevel\EventSourcing\Subscription\Cleanup\Dbal\DbalCleanupTaskHandler;
+use Patchlevel\EventSourcing\Subscription\Cleanup\DefaultCleaner;
+
+/** @var Connection $connection */
+$cleaner = new DefaultCleaner([
+    new DbalCleanupTaskHandler($connection),
+]);
+```
+If you have multiple database connections and want to use the `DbalCleanupTaskHandler` to clean up the respective databases,
+you can also pass a `ConnectionRegistry` (from `doctrine/persistence`) to the `DbalCleanupTaskHandler`.
+Then you can pass the connection name as parameter in the cleanup task and the handler will use the corresponding connection to execute the task.
+
+```php
+use Doctrine\Persistence\ConnectionRegistry;
+use Patchlevel\EventSourcing\Subscription\Cleanup\Dbal\DbalCleanupTaskHandler;
+use Patchlevel\EventSourcing\Subscription\Cleanup\DefaultCleaner;
+
+/** @var ConnectionRegistry $connectionRegistry */
+$cleaner = new DefaultCleaner([
+    new DbalCleanupTaskHandler($connectionRegistry),
+]);
+```
 ### Subscriber Accessor
 
 The subscriber accessor repository is responsible for providing the subscribers to the subscription engine.

src/Subscription/Cleanup/Dbal/ConnectionNameNotSupported.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Cleanup\Dbal;
+
+use RuntimeException;
+
+use function sprintf;
+
+final class ConnectionNameNotSupported extends RuntimeException
+{
+    public function __construct(string $connectionName)
+    {
+        parent::__construct(
+            sprintf(
+                'Connection name "%s" is not supported. Only a single connection is available. Please use the connection registry.',
+                $connectionName,
+            ),
+        );
+    }
+}

src/Subscription/Cleanup/Dbal/DbalCleanupTaskHandler.php

@@ -5,26 +5,27 @@
 namespace Patchlevel\EventSourcing\Subscription\Cleanup\Dbal;
 
 use Doctrine\DBAL\Connection;
+use Doctrine\Persistence\ConnectionRegistry;
 use Patchlevel\EventSourcing\Subscription\Cleanup\CleanupTaskHandler;
 use Patchlevel\EventSourcing\Subscription\Cleanup\CleanupTaskNotSupported;
 
 final class DbalCleanupTaskHandler implements CleanupTaskHandler
 {
     public function __construct(
-        private readonly Connection $connection,
+        private readonly Connection|ConnectionRegistry $connection,
     ) {
     }
 
     public function __invoke(object $task): void
     {
         if ($task instanceof DropTableTask) {
-            $this->connection->createSchemaManager()->dropTable($task->table);
+            $this->connection($task->connectionName)->createSchemaManager()->dropTable($task->table);
 
             return;
         }
 
         if ($task instanceof DropIndexTask) {
-            $this->connection->createSchemaManager()->dropIndex($task->index, $task->table);
+            $this->connection($task->connectionName)->createSchemaManager()->dropIndex($task->index, $task->table);
 
             return;
         }
@@ -36,4 +37,23 @@ public function supports(object $task): bool
     {
         return $task instanceof DropTableTask || $task instanceof DropIndexTask;
     }
+
+    private function connection(string|null $connectionName): Connection
+    {
+        if ($this->connection instanceof ConnectionRegistry) {
+            $connection = $this->connection->getConnection($connectionName);
+
+            if (!$connection instanceof Connection) {
+                throw new UnexpectedConnectionType($connectionName, Connection::class, $connection::class);
+            }
+
+            return $connection;
+        }
+
+        if ($connectionName === null) {
+            return $this->connection;
+        }
+
+        throw new ConnectionNameNotSupported($connectionName);
+    }
 }

src/Subscription/Cleanup/Dbal/DropIndexTask.php

@@ -9,6 +9,7 @@ final class DropIndexTask
     public function __construct(
         public readonly string $index,
         public readonly string $table,
+        public readonly string|null $connectionName = null,
     ) {
     }
 }

src/Subscription/Cleanup/Dbal/DropTableTask.php

@@ -8,6 +8,7 @@ final class DropTableTask
 {
     public function __construct(
         public readonly string $table,
+        public readonly string|null $connectionName = null,
     ) {
     }
 }

src/Subscription/Cleanup/Dbal/UnexpectedConnectionType.php

@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Cleanup\Dbal;
+
+use RuntimeException;
+
+use function sprintf;
+
+final class UnexpectedConnectionType extends RuntimeException
+{
+    /**
+     * @param class-string $expected
+     * @param class-string $actual
+     */
+    public function __construct(string|null $connectionName, string $expected, string $actual)
+    {
+        parent::__construct(
+            sprintf(
+                'Expected connection "%s" to be of type "%s", got "%s"',
+                $connectionName ?? 'default (null)',
+                $expected,
+                $actual,
+            ),
+        );
+    }
+}

tests/Unit/Subscription/Cleanup/Dbal/DbalCleanupTaskHandlerTest.php

@@ -6,10 +6,13 @@
 
 use Doctrine\DBAL\Connection;
 use Doctrine\DBAL\Schema\AbstractSchemaManager;
+use Doctrine\Persistence\ConnectionRegistry;
 use Patchlevel\EventSourcing\Subscription\Cleanup\CleanupTaskNotSupported;
+use Patchlevel\EventSourcing\Subscription\Cleanup\Dbal\ConnectionNameNotSupported;
 use Patchlevel\EventSourcing\Subscription\Cleanup\Dbal\DbalCleanupTaskHandler;
 use Patchlevel\EventSourcing\Subscription\Cleanup\Dbal\DropIndexTask;
 use Patchlevel\EventSourcing\Subscription\Cleanup\Dbal\DropTableTask;
+use Patchlevel\EventSourcing\Subscription\Cleanup\Dbal\UnexpectedConnectionType;
 use PHPUnit\Framework\Attributes\CoversClass;
 use PHPUnit\Framework\TestCase;
 use stdClass;
@@ -78,4 +81,51 @@ public function testHandleDropIndex(): void
 
         $handler(new DropIndexTask('foo', 'bar'));
     }
+
+    public function testHandleWithConnectionRegistry(): void
+    {
+        $schemaManager = $this->createMock(AbstractSchemaManager::class);
+        $schemaManager->expects($this->once())->method('dropTable')->with('test');
+
+        $connection = $this->createMock(Connection::class);
+        $connection
+            ->expects($this->once())
+            ->method('createSchemaManager')
+            ->willReturn($schemaManager);
+
+        $registry = $this->createMock(ConnectionRegistry::class);
+        $registry
+            ->expects($this->once())
+            ->method('getConnection')
+            ->with('foo')
+            ->willReturn($connection);
+
+        $handler = new DbalCleanupTaskHandler($registry);
+
+        $handler(new DropTableTask('test', 'foo'));
+    }
+
+    public function testHandleWithConnectionRegistryAndUnexpectedType(): void
+    {
+        $registry = $this->createMock(ConnectionRegistry::class);
+        $registry
+            ->expects($this->once())
+            ->method('getConnection')
+            ->with('foo')
+            ->willReturn(new stdClass());
+
+        $handler = new DbalCleanupTaskHandler($registry);
+
+        $this->expectException(UnexpectedConnectionType::class);
+        $handler(new DropTableTask('test', 'foo'));
+    }
+
+    public function testHandleWithConnectionAndConnectionNameNotSupported(): void
+    {
+        $connection = $this->createMock(Connection::class);
+        $handler = new DbalCleanupTaskHandler($connection);
+
+        $this->expectException(ConnectionNameNotSupported::class);
+        $handler(new DropTableTask('test', 'foo'));
+    }
 }