Improving Saga handling: Allow multiple sagaIds per Saga

composer.json

@@ -23,7 +23,7 @@
     "require": {
         "php": "^8.3",
         "doctrine/dbal": "^4.0",
-        "gember/dependency-contracts": "^0.2.1"
+        "gember/dependency-contracts": "^0.3"
     },
     "require-dev": {
         "captainhook/captainhook": "^5.23",

resources/migrations/doctrine/Version20251002195234.php

@@ -12,12 +12,23 @@ public function up(Schema $schema): void
         $this->addSql(
             <<<'SQL'
             CREATE TABLE `saga_store` (
-              `saga_id` varchar(50) COLLATE utf8mb4_unicode_ci NOT NULL,
+              `id` varchar(50) COLLATE utf8mb4_unicode_ci NOT NULL,
               `saga_name` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL,
               `payload` json NOT NULL,
               `created_at` timestamp(6) NOT NULL,
               `updated_at` timestamp(6) NULL DEFAULT NULL,
-              PRIMARY KEY (`saga_id`, `saga_name`)
+              PRIMARY KEY (`id`)
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+            SQL
+        );
+
+        $this->addSql(
+            <<<'SQL'
+            CREATE TABLE `saga_store_relation` (
+              `id` varchar(50) COLLATE utf8mb4_unicode_ci NOT NULL,
+              `saga_id` varchar(50) COLLATE utf8mb4_unicode_ci NOT NULL,
+              PRIMARY KEY (`id`,`saga_id`),
+              CONSTRAINT `saga_store_relation_ibfk_1` FOREIGN KEY (`id`) REFERENCES `saga_store` (`id`)
             ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
             SQL
         );

resources/migrations/phinx/20251002194212.php

@@ -8,8 +8,8 @@ final class V20251002194212 extends AbstractMigration
 {
     public function change(): void
     {
-        $this->table('saga_store', ['id' => false, 'primary_key' => ['saga_id', 'saga_name']])
-            ->addColumn('saga_id', 'string', ['limit' => 50, 'null' => false])
+        $this->table('saga_store', ['id' => false, 'primary_key' => 'id'])
+            ->addColumn('id', 'string', ['limit' => 50, 'null' => false])
             ->addColumn('saga_name', 'string', ['null' => false])
             ->addColumn('payload', 'json', ['null' => false])
             ->addColumn('created_at', 'timestamp', ['limit' => 6, 'null' => false])

resources/migrations/phinx/20251015202332.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+use Phinx\Migration\AbstractMigration;
+
+final class V20251015202332 extends AbstractMigration
+{
+    public function change(): void
+    {
+        $this->table('saga_store_relation', ['id' => false, 'primary_key' => ['id', 'saga_id']])
+            ->addColumn('id', 'string', ['limit' => 50, 'null' => false])
+            ->addColumn('saga_id', 'string', ['limit' => 50, 'null' => false])
+            ->addForeignKey('id', 'saga_store', 'id')
+            ->create();
+    }
+}

resources/schema.sql

@@ -16,10 +16,17 @@ CREATE TABLE `event_store_relation` (
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
 
 CREATE TABLE `saga_store` (
-  `saga_id` varchar(50) COLLATE utf8mb4_unicode_ci NOT NULL,
+  `id` varchar(50) COLLATE utf8mb4_unicode_ci NOT NULL,
   `saga_name` varchar(255) COLLATE utf8mb4_unicode_ci NOT NULL,
   `payload` json NOT NULL,
   `created_at` timestamp(6) NOT NULL,
   `updated_at` timestamp(6) NULL DEFAULT NULL,
-  PRIMARY KEY (`saga_id`, `saga_name`)
+  PRIMARY KEY (`id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+CREATE TABLE `saga_store_relation` (
+  `id` varchar(50) COLLATE utf8mb4_unicode_ci NOT NULL,
+  `saga_id` varchar(50) COLLATE utf8mb4_unicode_ci NOT NULL,
+  PRIMARY KEY (`id`,`saga_id`),
+  CONSTRAINT `saga_store_relation_ibfk_1` FOREIGN KEY (`id`) REFERENCES `saga_store` (`id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

src/Saga/DoctrineDbalRdbmsSagaFactory.php

@@ -21,8 +21,9 @@
     public function createFromRow(array $row): RdbmsSaga
     {
         return new RdbmsSaga(
+            $row['id'],
             $row['sagaName'],
-            $row['sagaId'],
+            $row['sagaIds'],
             $row['payload'],
             new DateTimeImmutable($row['createdAt']),
             $row['updatedAt'] !== null ? new DateTimeImmutable($row['updatedAt']) : null,

src/Saga/DoctrineRdbmsSagaStoreRepository.php

@@ -4,18 +4,23 @@
 
 namespace Gember\RdbmsEventStoreDoctrineDbal\Saga;
 
+use Doctrine\DBAL\ArrayParameterType;
 use Doctrine\DBAL\Connection;
 use Gember\DependencyContracts\EventStore\Saga\RdbmsSaga;
 use Gember\DependencyContracts\EventStore\Saga\RdbmsSagaStoreRepository;
 use Gember\DependencyContracts\EventStore\Saga\RdbmsSagaNotFoundException;
+use Gember\DependencyContracts\Util\Generator\Identity\IdentityGenerator;
+use Gember\RdbmsEventStoreDoctrineDbal\Saga\TableSchema\SagaStoreRelationTableSchema;
 use Gember\RdbmsEventStoreDoctrineDbal\Saga\TableSchema\SagaStoreTableSchema;
 use Override;
 use Stringable;
 use DateTimeImmutable;
+use Throwable;
 
 /**
  * @phpstan-type SagaRow array{
- *     sagaId: string,
+ *     id: string,
+ *     sagaIds: list<string>,
  *     sagaName: string,
  *     payload: string,
  *     createdAt: string,
@@ -27,92 +32,199 @@
     public function __construct(
         private Connection $connection,
         private SagaStoreTableSchema $sagaStoreTableSchema,
+        private SagaStoreRelationTableSchema $sagaStoreRelationTableSchema,
         private DoctrineDbalRdbmsSagaFactory $sagaFactory,
+        private IdentityGenerator $identityGenerator,
     ) {}
 
     #[Override]
-    public function get(string $sagaName, Stringable|string $sagaId): RdbmsSaga
+    public function get(string $sagaName, Stringable|string ...$sagaIds): RdbmsSaga
     {
         $sagaStoreSchema = $this->sagaStoreTableSchema;
+        $sagaStoreRelationSchema = $this->sagaStoreRelationTableSchema;
 
-        /** @var false|SagaRow $row */
+        /** @var list<array{
+         *     id: string,
+         *     sagaId: string,
+         *     sagaName: string,
+         *     payload: string,
+         *     createdAt: string,
+         *     updatedAt: string|null
+         * }> $row */
         $row = $this->connection->createQueryBuilder()
             ->select(
                 <<<DQL
-                ss.{$sagaStoreSchema->sagaIdFieldName} as sagaId,
+                ss.{$sagaStoreSchema->idFieldName} as id,
                 ss.{$sagaStoreSchema->sagaNameFieldName} as sagaName,
                 ss.{$sagaStoreSchema->payloadFieldName} as payload,
                 ss.{$sagaStoreSchema->createdAtFieldName} as createdAt,
                 ss.{$sagaStoreSchema->updatedAtFieldName} as updatedAt
                 DQL
             )
             ->from($sagaStoreSchema->tableName, 'ss')
-            ->where(sprintf('ss.%s = :sagaId', $sagaStoreSchema->sagaIdFieldName))
+            ->join('ss', $sagaStoreRelationSchema->tableName, 'ssr', sprintf(
+                'ss.%s = ssr.%s',
+                $sagaStoreSchema->idFieldName,
+                $sagaStoreRelationSchema->idFieldName,
+            ))
+            ->where(sprintf('ssr.%s IN (:sagaIds)', $sagaStoreRelationSchema->sagaIdFieldName))
             ->andWhere(sprintf('ss.%s = :sagaName', $sagaStoreSchema->sagaNameFieldName))
-            ->setParameter('sagaId', (string) $sagaId)
+            ->setParameter(
+                'sagaIds',
+                array_map(fn($sagaId) => (string) $sagaId, $sagaIds),
+                ArrayParameterType::STRING,
+            )
             ->setParameter('sagaName', $sagaName)
+            ->setMaxResults(1)
             ->executeQuery()
             ->fetchAssociative();
 
         if (!$row) {
-            throw RdbmsSagaNotFoundException::withSagaId($sagaName, $sagaId);
+            throw RdbmsSagaNotFoundException::create($sagaName, ...$sagaIds);
         }
 
-        return $this->sagaFactory->createFromRow($row);
+        $sagaIdRows = $this->connection->createQueryBuilder()
+            ->select(
+                <<<DQL
+                ssr.{$sagaStoreRelationSchema->sagaIdFieldName} as sagaId
+                DQL
+            )
+            ->from($sagaStoreRelationSchema->tableName, 'ssr')
+            ->executeQuery()
+            ->fetchAllAssociative();
+
+        /** @var list<SagaRow> $payload */
+        $payload = $row;
+        $payload['sagaIds'] = array_map(fn($sagaIdRow) => $sagaIdRow['sagaId'], $sagaIdRows);
+
+        return $this->sagaFactory->createFromRow($payload);
     }
 
     #[Override]
     public function save(
         string $sagaName,
-        Stringable|string $sagaId,
         string $payload,
         DateTimeImmutable $now,
+        Stringable|string ...$sagaIds,
+    ): RdbmsSaga {
+        try {
+            $previous = $this->get($sagaName, ...$sagaIds);
+        } catch (RdbmsSagaNotFoundException) {
+            return $this->create($sagaName, $payload, $now, ...$sagaIds);
+        }
+
+        return $this->update($previous, $sagaName, $payload, $now, ...$sagaIds);
+    }
+
+    private function create(
+        string $sagaName,
+        string $payload,
+        DateTimeImmutable $now,
+        Stringable|string ...$sagaIds,
     ): RdbmsSaga {
+        $id = $this->identityGenerator->generate();
+
         $sagaStoreSchema = $this->sagaStoreTableSchema;
+        $sagaStoreRelationSchema = $this->sagaStoreRelationTableSchema;
+
+        $this->connection->beginTransaction();
 
         try {
-            $previous = $this->get($sagaName, $sagaId);
-        } catch (RdbmsSagaNotFoundException) {
             $this->connection->createQueryBuilder()
                 ->insert($sagaStoreSchema->tableName)
-                ->setValue($sagaStoreSchema->sagaIdFieldName, ':sagaId')
+                ->setValue($sagaStoreSchema->idFieldName, ':id')
                 ->setValue($sagaStoreSchema->sagaNameFieldName, ':sagaName')
                 ->setValue($sagaStoreSchema->payloadFieldName, ':payload')
                 ->setValue($sagaStoreSchema->createdAtFieldName, ':createdAt')
                 ->setParameters([
-                    'sagaId' => $sagaId,
+                    'id' => $id,
                     'sagaName' => $sagaName,
                     'payload' => $payload,
                     'createdAt' => $now->format($sagaStoreSchema->createdAtFieldFormat),
                 ])
                 ->executeStatement();
 
-            return new RdbmsSaga(
-                $sagaName,
-                $sagaId,
-                $payload,
-                $now,
-                null,
-            );
+            foreach ($sagaIds as $sagaId) {
+                $this->connection->createQueryBuilder()
+                    ->insert($sagaStoreRelationSchema->tableName)
+                    ->setValue($sagaStoreRelationSchema->idFieldName, ':id')
+                    ->setValue($sagaStoreRelationSchema->sagaIdFieldName, ':sagaId')
+                    ->setParameters([
+                        'id' => $id,
+                        'sagaId' => $sagaId,
+                    ])
+                    ->executeStatement();
+            }
+
+            $this->connection->commit();
+        } catch (Throwable $exception) {
+            $this->connection->rollBack();
+
+            throw $exception;
         }
 
-        $this->connection->createQueryBuilder()
-            ->update($sagaStoreSchema->tableName)
-            ->where(sprintf('%s = :sagaId', $sagaStoreSchema->sagaIdFieldName))
-            ->andWhere(sprintf('%s = :sagaName', $sagaStoreSchema->sagaNameFieldName))
-            ->set($sagaStoreSchema->payloadFieldName, ':payload')
-            ->set($sagaStoreSchema->updatedAtFieldName, ':updatedAt')
-            ->setParameters([
-                'sagaId' => $sagaId,
-                'sagaName' => $sagaName,
-                'payload' => $payload,
-                'updatedAt' => $now->format($sagaStoreSchema->updatedAtFieldFormat),
-            ])
-            ->executeStatement();
+        return new RdbmsSaga(
+            $id,
+            $sagaName,
+            array_values($sagaIds),
+            $payload,
+            $now,
+            null,
+        );
+    }
+
+    private function update(
+        RdbmsSaga $previous,
+        string $sagaName,
+        string $payload,
+        DateTimeImmutable $now,
+        Stringable|string ...$sagaIds,
+    ): RdbmsSaga {
+        $sagaStoreSchema = $this->sagaStoreTableSchema;
+        $sagaStoreRelationSchema = $this->sagaStoreRelationTableSchema;
+
+        $this->connection->beginTransaction();
+
+        try {
+            $this->connection->createQueryBuilder()
+                ->update($sagaStoreSchema->tableName)
+                ->where(sprintf('%s = :id', $sagaStoreSchema->idFieldName))
+                ->set($sagaStoreSchema->payloadFieldName, ':payload')
+                ->set($sagaStoreSchema->updatedAtFieldName, ':updatedAt')
+                ->setParameter('id', $previous->id)
+                ->setParameter('payload', $payload)
+                ->setParameter('updatedAt', $now->format($sagaStoreSchema->updatedAtFieldFormat))
+                ->executeStatement();
+
+            $this->connection->createQueryBuilder()
+                ->delete($sagaStoreRelationSchema->tableName)
+                ->where(sprintf('%s = :id', $sagaStoreRelationSchema->idFieldName))
+                ->setParameter('id', $previous->id)
+                ->executeStatement();
+
+            foreach ($sagaIds as $sagaId) {
+                $this->connection->createQueryBuilder()
+                    ->insert($sagaStoreRelationSchema->tableName)
+                    ->setValue($sagaStoreRelationSchema->idFieldName, ':id')
+                    ->setValue($sagaStoreRelationSchema->sagaIdFieldName, ':sagaId')
+                    ->setParameters([
+                        'id' => $previous->id,
+                        'sagaId' => $sagaId,
+                    ])
+                    ->executeStatement();
+            }
+
+            $this->connection->commit();
+        } catch (Throwable $exception) {
+            $this->connection->rollBack();
+
+            throw $exception;
+        }
 
         return new RdbmsSaga(
+            $previous->id,
             $sagaName,
-            $sagaId,
+            array_values($sagaIds),
             $payload,
             $previous->createdAt,
             $now,