Handle RdbmsSaga with multiple SagaIds

jerowork · jerowork · commit d3b1d553585c · 2025-10-16T19:33:24.000+02:00
diff --git a/src/Saga/DoctrineDbalRdbmsSagaFactory.php b/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,
diff --git a/src/Saga/DoctrineRdbmsSagaStoreRepository.php b/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,
diff --git a/tests/Saga/DoctrineDbalRdbmsSagaFactoryTest.php b/tests/Saga/DoctrineDbalRdbmsSagaFactoryTest.php
@@ -18,15 +18,17 @@ final class DoctrineDbalRdbmsSagaFactoryTest extends TestCase
     public function itShouldCreateRdbmsSaga(): void
     {
         $saga = (new DoctrineDbalRdbmsSagaFactory())->createFromRow([
+            'id' => '01K7Q14MMW2FQP2JS1RHQS7QXP',
             'sagaName' => 'some.saga',
-            'sagaId' => '01K76G1PGKPZ047KDN25PFPEEV',
+            'sagaIds' => ['01K76G1PGKPZ047KDN25PFPEEV', '01K7Q13CG3A3PQCC98XYSE67K1'],
             'payload' => '{"some":"data"}',
             'createdAt' => '2018-12-01 12:05:08.234543',
             'updatedAt' => null,
         ]);
 
+        self::assertSame('01K7Q14MMW2FQP2JS1RHQS7QXP', $saga->id);
         self::assertSame('some.saga', $saga->sagaName);
-        self::assertSame('01K76G1PGKPZ047KDN25PFPEEV', $saga->sagaId);
+        self::assertSame(['01K76G1PGKPZ047KDN25PFPEEV', '01K7Q13CG3A3PQCC98XYSE67K1'], $saga->sagaIds);
         self::assertSame('{"some":"data"}', $saga->payload);
         self::assertEquals(new DateTimeImmutable('2018-12-01 12:05:08.234543'), $saga->createdAt);
         self::assertNull($saga->updatedAt);
diff --git a/tests/Saga/DoctrineRdbmsSagaStoreRepositoryTest.php b/tests/Saga/DoctrineRdbmsSagaStoreRepositoryTest.php
@@ -10,6 +10,7 @@
 use Gember\RdbmsEventStoreDoctrineDbal\Saga\DoctrineDbalRdbmsSagaFactory;
 use Gember\RdbmsEventStoreDoctrineDbal\Saga\DoctrineRdbmsSagaStoreRepository;
 use Gember\RdbmsEventStoreDoctrineDbal\Saga\TableSchema\SagaTableSchemaFactory;
+use Gember\RdbmsEventStoreDoctrineDbal\Test\TestDoubles\TestIdentityGenerator;
 use PHPUnit\Framework\Attributes\Test;
 use PHPUnit\Framework\TestCase;
 use Override;
@@ -21,6 +22,7 @@
 final class DoctrineRdbmsSagaStoreRepositoryTest extends TestCase
 {
     private DoctrineRdbmsSagaStoreRepository $repository;
+    private TestIdentityGenerator $identityGenerator;
 
     #[Override]
     protected function setUp(): void
@@ -33,7 +35,9 @@ protected function setUp(): void
         $this->repository = new DoctrineRdbmsSagaStoreRepository(
             $connection,
             SagaTableSchemaFactory::createDefaultSagaStore(),
+            SagaTableSchemaFactory::createDefaultSagaStoreRelation(),
             new DoctrineDbalRdbmsSagaFactory(),
+            $this->identityGenerator = new TestIdentityGenerator(),
         );
     }
 
@@ -48,17 +52,21 @@ public function itShouldThrowExceptionWhenSagaNotFound(): void
     #[Test]
     public function itShouldSaveAndGetSaga(): void
     {
+        $this->identityGenerator->ids[] = '01K7Q083CX4T7Z0NT5CKEX8NEJ';
+
         $this->repository->save(
             'some.saga',
-            '01K76GDQ5RT71G7HQVNR264KD4',
             '{"some":"data"}',
             new DateTimeImmutable('2025-10-10 12:00:34'),
+            '01K76GDQ5RT71G7HQVNR264KD4',
+            '01K7Q033P5174AXA054FFAHW2F',
         );
 
         $saga = $this->repository->get('some.saga', '01K76GDQ5RT71G7HQVNR264KD4');
 
+        self::assertSame('01K7Q083CX4T7Z0NT5CKEX8NEJ', $saga->id);
         self::assertSame('some.saga', $saga->sagaName);
-        self::assertSame('01K76GDQ5RT71G7HQVNR264KD4', $saga->sagaId);
+        self::assertSame(['01K76GDQ5RT71G7HQVNR264KD4', '01K7Q033P5174AXA054FFAHW2F'], $saga->sagaIds);
         self::assertSame('{"some":"data"}', $saga->payload);
         self::assertEquals(new DateTimeImmutable('2025-10-10 12:00:34'), $saga->createdAt);
         self::assertNull($saga->updatedAt);
@@ -67,24 +75,29 @@ public function itShouldSaveAndGetSaga(): void
     #[Test]
     public function itShouldSaveExistingSaga(): void
     {
+        $this->identityGenerator->ids[] = '01K7Q083CX4T7Z0NT5CKEX8NEJ';
+
         $this->repository->save(
             'some.saga',
-            '01K76GDQ5RT71G7HQVNR264KD4',
             '{"some":"data"}',
             new DateTimeImmutable('2025-10-10 12:00:34'),
+            '01K76GDQ5RT71G7HQVNR264KD4',
+            '01K7Q0GR8ABHBZG8QCGTXJXJ7T',
         );
 
         $this->repository->save(
             'some.saga',
-            '01K76GDQ5RT71G7HQVNR264KD4',
             '{"some":"updated"}',
             new DateTimeImmutable('2025-10-10 13:30:12'),
+            '01K76GDQ5RT71G7HQVNR264KD4',
+            '01K7Q0JGY9ZMX11K75AAY5J78R',
         );
 
         $saga = $this->repository->get('some.saga', '01K76GDQ5RT71G7HQVNR264KD4');
 
+        self::assertSame('01K7Q083CX4T7Z0NT5CKEX8NEJ', $saga->id);
         self::assertSame('some.saga', $saga->sagaName);
-        self::assertSame('01K76GDQ5RT71G7HQVNR264KD4', $saga->sagaId);
+        self::assertSame(['01K76GDQ5RT71G7HQVNR264KD4', '01K7Q0JGY9ZMX11K75AAY5J78R'], $saga->sagaIds);
         self::assertSame('{"some":"updated"}', $saga->payload);
         self::assertEquals(new DateTimeImmutable('2025-10-10 12:00:34'), $saga->createdAt);
         self::assertEquals(new DateTimeImmutable('2025-10-10 13:30:12'), $saga->updatedAt);
diff --git a/tests/TestDoubles/TestIdentityGenerator.php b/tests/TestDoubles/TestIdentityGenerator.php
@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Gember\RdbmsEventStoreDoctrineDbal\Test\TestDoubles;
+
+use Gember\DependencyContracts\Util\Generator\Identity\IdentityGenerator;
+use Override;
+
+final class TestIdentityGenerator implements IdentityGenerator
+{
+    /**
+     * @var list<string>
+     */
+    public array $ids = [];
+
+    #[Override]
+    public function generate(): string
+    {
+        return (string) array_shift($this->ids);
+    }
+}
diff --git a/tests/schema.sql b/tests/schema.sql
