feat: migration delete tracks streaming tables (#614)

Author: dgafka

packages/PdoEventSourcing/src/Database/EventStreamTableManager.php

@@ -84,9 +84,26 @@ public function createTable(Connection $connection): void
 
     public function dropTable(Connection $connection): void
     {
+        if ($this->isInitialized($connection)) {
+            $this->dropStreamTables($connection);
+        }
         $connection->executeStatement($this->getDropTableSql($connection));
     }
 
+    private function dropStreamTables(Connection $connection): void
+    {
+        $streamTableNames = $connection->fetchFirstColumn(
+            "SELECT stream_name FROM {$this->tableName}"
+        );
+
+        foreach ($streamTableNames as $streamTableName) {
+            $dropSql = $this->isPostgres($connection)
+                ? "DROP TABLE IF EXISTS {$streamTableName}"
+                : "DROP TABLE IF EXISTS `{$streamTableName}`";
+            $connection->executeStatement($dropSql);
+        }
+    }
+
     public function isInitialized(Connection $connection): bool
     {
         return SchemaManagerCompatibility::tableExists($connection, $this->tableName);

packages/PdoEventSourcing/tests/Integration/EventStreamTest.php

@@ -4,6 +4,7 @@
 
 namespace Integration;
 
+use Ecotone\EventSourcing\Database\EventStreamTableManager;
 use Ecotone\EventSourcing\EventStore;
 use Ecotone\EventSourcing\Prooph\LazyProophEventStore;
 use Ecotone\Lite\EcotoneLite;
@@ -14,8 +15,10 @@
 use Enqueue\Dbal\DbalConnectionFactory;
 use Ramsey\Uuid\Uuid;
 use Test\Ecotone\EventSourcing\EventSourcingMessagingTestCase;
+use Test\Ecotone\EventSourcing\Fixture\Ticket\Command\RegisterTicket;
 use Test\Ecotone\EventSourcing\Fixture\Ticket\Event\TicketWasClosed;
 use Test\Ecotone\EventSourcing\Fixture\Ticket\Event\TicketWasRegistered;
+use Test\Ecotone\EventSourcing\Fixture\Ticket\Ticket;
 use Test\Ecotone\EventSourcing\Fixture\Ticket\TicketEventConverter;
 use Test\Ecotone\EventSourcing\Fixture\TicketWithSynchronousEventDrivenProjection\InProgressTicketList;
 
@@ -303,4 +306,38 @@ public function test_fetching_with_pagination()
             $events[0]->getPayload()
         );
     }
+
+    public function test_deleting_event_stream_table_also_deletes_stream_tables(): void
+    {
+        $ecotone = EcotoneLite::bootstrapFlowTestingWithEventStore(
+            containerOrAvailableServices: [
+                new InProgressTicketList($this->getConnection()),
+                new TicketEventConverter(),
+                DbalConnectionFactory::class => $this->getConnectionFactory(),
+            ],
+            configuration: ServiceConfiguration::createWithDefaults()
+                ->withEnvironment('prod')
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::EVENT_SOURCING_PACKAGE]))
+                ->withNamespaces([
+                    'Test\Ecotone\EventSourcing\Fixture\Ticket',
+                ]),
+            pathToRootCatalog: __DIR__ . '/../../',
+            runForProductionEventStore: true
+        );
+
+        $ecotone->sendCommand(new RegisterTicket('1', 'johny', 'alert'));
+
+        $connection = $this->getConnection();
+        $eventStreamsTable = LazyProophEventStore::DEFAULT_STREAM_TABLE;
+        $streamTableName = '_' . sha1(Ticket::class);
+
+        $this->assertTrue(self::tableExists($connection, $eventStreamsTable));
+        $this->assertTrue(self::tableExists($connection, $streamTableName));
+
+        $tableManager = new EventStreamTableManager($eventStreamsTable, true, true);
+        $tableManager->dropTable($connection);
+
+        $this->assertFalse(self::tableExists($connection, $streamTableName));
+        $this->assertFalse(self::tableExists($connection, $eventStreamsTable));
+    }
 }