Feat: Database migration process

packages/Dbal/src/Configuration/DbalConfiguration.php

@@ -51,6 +51,8 @@ class DbalConfiguration
     private int $minimumTimeToRemoveMessageInMilliseconds = DeduplicationModule::REMOVE_MESSAGE_AFTER_7_DAYS;
     private int $deduplicationRemovalBatchSize = 1000;
 
+    private bool $initializeDatabaseTables = true;
+
     private function __construct()
     {
     }
@@ -357,4 +359,21 @@ public function getConsumerPositionTrackingConnectionReference(): string
     {
         return $this->consumerPositionTrackingConnectionReference;
     }
+
+    /**
+     * Controls whether database tables are automatically initialized on first use.
+     * When set to false, tables must be created manually using `ecotone:migration:database:setup --initialize`.
+     */
+    public function withAutomaticTableInitialization(bool $enabled): self
+    {
+        $self = clone $this;
+        $self->initializeDatabaseTables = $enabled;
+
+        return $self;
+    }
+
+    public function isAutomaticTableInitializationEnabled(): bool
+    {
+        return $this->initializeDatabaseTables;
+    }
 }

packages/Dbal/src/Configuration/DbalPublisherModule.php

@@ -3,6 +3,9 @@
 namespace Ecotone\Dbal\Configuration;
 
 use Ecotone\AnnotationFinder\AnnotationFinder;
+use Ecotone\Dbal\Database\DbalTableManagerReference;
+use Ecotone\Dbal\Database\EnqueueTableManager;
+use Ecotone\Dbal\DbalBackedMessageChannelBuilder;
 use Ecotone\Dbal\DbalOutboundChannelAdapterBuilder;
 use Ecotone\Messaging\Attribute\ModuleAnnotation;
 use Ecotone\Messaging\Config\Annotation\AnnotationModule;
@@ -44,6 +47,21 @@ public function prepare(Configuration $messagingConfiguration, array $extensionO
         $registeredReferences = [];
         $applicationConfiguration = ExtensionObjectResolver::resolveUnique(ServiceConfiguration::class, $extensionObjects, ServiceConfiguration::createWithDefaults());
 
+        $dbalConfiguration = ExtensionObjectResolver::resolveUnique(DbalConfiguration::class, $extensionObjects, DbalConfiguration::createWithDefaults());
+        $dbalMessageChannels = ExtensionObjectResolver::resolve(DbalBackedMessageChannelBuilder::class, $extensionObjects);
+        $dbalPublishers = ExtensionObjectResolver::resolve(DbalMessagePublisherConfiguration::class, $extensionObjects);
+        $hasMessageQueues = ! empty($dbalMessageChannels) || ! empty($dbalPublishers);
+        $shouldAutoInitialize = $dbalConfiguration->isAutomaticTableInitializationEnabled();
+
+        $messagingConfiguration->registerServiceDefinition(
+            EnqueueTableManager::class,
+            new \Ecotone\Messaging\Config\Container\Definition(EnqueueTableManager::class, [
+                EnqueueTableManager::DEFAULT_TABLE_NAME,
+                $hasMessageQueues,
+                $shouldAutoInitialize,
+            ])
+        );
+
         foreach (ExtensionObjectResolver::resolve(DbalMessagePublisherConfiguration::class, $extensionObjects) as $dbalPublisher) {
             if (in_array($dbalPublisher->getReferenceName(), $registeredReferences)) {
                 throw ConfigurationException::create("Registering two publishers under same reference name {$dbalPublisher->getReferenceName()}. You need to create publisher with specific reference using `createWithReferenceName`.");
@@ -109,12 +127,16 @@ public function canHandle($extensionObject): bool
     {
         return
             $extensionObject instanceof DbalMessagePublisherConfiguration
-            || $extensionObject instanceof ServiceConfiguration;
+            || $extensionObject instanceof ServiceConfiguration
+            || $extensionObject instanceof DbalBackedMessageChannelBuilder
+            || $extensionObject instanceof DbalConfiguration;
     }
 
     public function getModuleExtensions(ServiceConfiguration $serviceConfiguration, array $serviceExtensions): array
     {
-        return [];
+        return [
+            new DbalTableManagerReference(EnqueueTableManager::class),
+        ];
     }
 
     public function getModulePackageName(): string

packages/Dbal/src/Database/DatabaseDropCommand.php

@@ -0,0 +1,51 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Dbal\Database;
+
+use Ecotone\Messaging\Attribute\ConsoleCommand;
+use Ecotone\Messaging\Attribute\ConsoleParameterOption;
+use Ecotone\Messaging\Config\ConsoleCommandResultSet;
+
+/**
+ * Console command handler for database drop operations.
+ *
+ * licence Apache-2.0
+ */
+class DatabaseDropCommand
+{
+    public function __construct(
+        private DatabaseSetupManager $databaseSetupManager,
+    ) {
+    }
+
+    #[ConsoleCommand('ecotone:migration:database:drop')]
+    public function drop(
+        #[ConsoleParameterOption] bool $force = false,
+        #[ConsoleParameterOption] bool $all = false,
+    ): ?ConsoleCommandResultSet {
+        $featureNames = $this->databaseSetupManager->getFeatureNames($all);
+
+        if (count($featureNames) === 0) {
+            return ConsoleCommandResultSet::create(
+                ['Status'],
+                [['No database tables registered for drop.']]
+            );
+        }
+
+        if ($force) {
+            $this->databaseSetupManager->dropAll($all);
+            return ConsoleCommandResultSet::create(
+                ['Feature', 'Status'],
+                array_map(fn (string $feature) => [$feature, 'Dropped'], $featureNames)
+            );
+        }
+
+        return ConsoleCommandResultSet::create(
+            ['Feature', 'Warning'],
+            array_map(fn (string $feature) => [$feature, 'Would be dropped (use --force to confirm)'], $featureNames)
+        );
+    }
+}
+

packages/Dbal/src/Database/DatabaseSetupCommand.php

@@ -0,0 +1,67 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Dbal\Database;
+
+use Ecotone\Messaging\Attribute\ConsoleCommand;
+use Ecotone\Messaging\Attribute\ConsoleParameterOption;
+use Ecotone\Messaging\Config\ConsoleCommandResultSet;
+
+/**
+ * Console command handler for database setup operations.
+ *
+ * licence Apache-2.0
+ */
+class DatabaseSetupCommand
+{
+    public function __construct(
+        private DatabaseSetupManager $databaseSetupManager,
+    ) {
+    }
+
+    #[ConsoleCommand('ecotone:migration:database:setup')]
+    public function setup(
+        #[ConsoleParameterOption] bool $initialize = false,
+        #[ConsoleParameterOption] bool $sql = false,
+        #[ConsoleParameterOption] bool $all = false,
+    ): ?ConsoleCommandResultSet {
+        $featureNames = $this->databaseSetupManager->getFeatureNames($all);
+
+        if (count($featureNames) === 0) {
+            return ConsoleCommandResultSet::create(
+                ['Status'],
+                [['No database tables registered for setup.']]
+            );
+        }
+
+        if ($sql) {
+            $statements = $this->databaseSetupManager->getCreateSqlStatements($all);
+            return ConsoleCommandResultSet::create(
+                ['SQL Statement'],
+                array_map(fn (string $statement) => [$statement], $statements)
+            );
+        }
+
+        if ($initialize) {
+            $this->databaseSetupManager->initializeAll($all);
+            return ConsoleCommandResultSet::create(
+                ['Feature', 'Status'],
+                array_map(fn (string $feature) => [$feature, 'Created'], $featureNames)
+            );
+        }
+
+        $initializationStatus = $this->databaseSetupManager->getInitializationStatus($all);
+        $rows = [];
+        foreach ($featureNames as $featureName) {
+            $isInitialized = $initializationStatus[$featureName] ?? false;
+            $rows[] = [$featureName, $isInitialized ? 'Yes' : 'No'];
+        }
+
+        return ConsoleCommandResultSet::create(
+            ['Feature', 'Initialized'],
+            $rows
+        );
+    }
+}
+

packages/Dbal/src/Database/DatabaseSetupManager.php

@@ -0,0 +1,162 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Ecotone\Dbal\Database;
+
+use Doctrine\DBAL\Connection;
+use Ecotone\Dbal\DbalReconnectableConnectionFactory;
+use Ecotone\Messaging\Config\Container\DefinedObject;
+use Ecotone\Messaging\Config\Container\Definition;
+use Enqueue\Dbal\DbalContext;
+use Interop\Queue\ConnectionFactory;
+
+/**
+ * Manages database setup and teardown for all registered table managers.
+ *
+ * licence Apache-2.0
+ */
+class DatabaseSetupManager implements DefinedObject
+{
+    /**
+     * @param DbalTableManager[] $tableManagers
+     */
+    public function __construct(
+        private ConnectionFactory $connectionFactory,
+        private array $tableManagers = [],
+    ) {
+    }
+
+    /**
+     * @return string[] List of feature names that require database tables
+     */
+    public function getFeatureNames(bool $includeInactive = false): array
+    {
+        return array_map(
+            fn (DbalTableManager $manager) => $manager->getFeatureName(),
+            $this->getManagers($includeInactive)
+        );
+    }
+
+    /**
+     * @return string[] SQL statements to create all tables
+     */
+    public function getCreateSqlStatements(bool $includeInactive = false): array
+    {
+        $connection = $this->getConnection();
+        $statements = [];
+
+        foreach ($this->getManagers($includeInactive) as $manager) {
+            $sql = $manager->getCreateTableSql($connection);
+            if (is_array($sql)) {
+                $statements = array_merge($statements, $sql);
+            } else {
+                $statements[] = $sql;
+            }
+        }
+
+        return $statements;
+    }
+
+    /**
+     * @return string[] SQL statements to drop all tables
+     */
+    public function getDropSqlStatements(bool $includeInactive = false): array
+    {
+        $connection = $this->getConnection();
+        $statements = [];
+
+        foreach ($this->getManagers($includeInactive) as $manager) {
+            $statements[] = $manager->getDropTableSql($connection);
+        }
+
+        return $statements;
+    }
+
+    /**
+     * Creates all tables.
+     */
+    public function initializeAll(bool $includeInactive = false): void
+    {
+        $connection = $this->getConnection();
+
+        foreach ($this->getManagers($includeInactive) as $manager) {
+            if ($manager->isInitialized($connection)) {
+                continue;
+            }
+
+            $manager->createTable($connection);
+        }
+    }
+
+    /**
+     * Drops all tables.
+     */
+    public function dropAll(bool $includeInactive = false): void
+    {
+        $connection = $this->getConnection();
+
+        foreach ($this->getManagers($includeInactive) as $manager) {
+            $manager->dropTable($connection);
+        }
+    }
+
+    /**
+     * Returns initialization status for each table manager.
+     *
+     * @return array<string, bool> Map of feature name to initialization status
+     */
+    public function getInitializationStatus(bool $includeInactive = false): array
+    {
+        $connection = $this->getConnection();
+        $status = [];
+
+        foreach ($this->getManagers($includeInactive) as $manager) {
+            $status[$manager->getFeatureName()] = $manager->isInitialized($connection);
+        }
+
+        return $status;
+    }
+
+    /**
+     * @return DbalTableManager[]
+     */
+    private function getManagers(bool $includeInactive): array
+    {
+        if ($includeInactive) {
+            return $this->tableManagers;
+        }
+
+        return array_filter(
+            $this->tableManagers,
+            fn (DbalTableManager $manager) => $manager->isActive()
+        );
+    }
+
+    private function getConnection(): Connection
+    {
+        /** @var DbalContext $context */
+        $context = $this->connectionFactory->createContext();
+
+        return $context->getDbalConnection();
+    }
+
+    public function getDefinition(): Definition
+    {
+        $tableManagerDefinitions = array_map(
+            fn (DbalTableManager $manager) => $manager->getDefinition(),
+            $this->tableManagers
+        );
+
+        return new Definition(
+            self::class,
+            [
+                new Definition(DbalReconnectableConnectionFactory::class, [
+                    $this->connectionFactory,
+                ]),
+                $tableManagerDefinitions,
+            ]
+        );
+    }
+}
+