New feature: Sagas

composer.json

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

resources/migrations/doctrine/Version20251002195234.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\Migrations\AbstractMigration;
+
+final class Version20251002195234 extends AbstractMigration
+{
+    public function up(Schema $schema): void
+    {
+        $this->addSql(
+            <<<'SQL'
+            CREATE TABLE `saga_store` (
+              `saga_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`)
+            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+            SQL
+        );
+    }
+}

resources/migrations/phinx/20251002194212.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+use Phinx\Migration\AbstractMigration;
+
+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])
+            ->addColumn('saga_name', 'string', ['null' => false])
+            ->addColumn('payload', 'json', ['null' => false])
+            ->addColumn('created_at', 'timestamp', ['limit' => 6, 'null' => false])
+            ->addColumn('updated_at', 'timestamp', ['limit' => 6, 'null' => true])
+            ->create();
+    }
+}

resources/schema.sql

@@ -14,3 +14,12 @@ CREATE TABLE `event_store_relation` (
   PRIMARY KEY (`event_id`,`domain_tag`),
   CONSTRAINT `event_store_relation_ibfk_1` FOREIGN KEY (`event_id`) REFERENCES `event_store` (`id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+CREATE TABLE `saga_store` (
+  `saga_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`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

src/Saga/DoctrineDbalRdbmsSagaFactory.php

@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Gember\RdbmsEventStoreDoctrineDbal\Saga;
+
+use Gember\DependencyContracts\EventStore\Saga\RdbmsSaga;
+use DateTimeImmutable;
+use DateMalformedStringException;
+
+/**
+ * @phpstan-import-type SagaRow from DoctrineRdbmsSagaStoreRepository
+ */
+final readonly class DoctrineDbalRdbmsSagaFactory
+{
+    /**
+     * @param SagaRow $row
+     *
+     * @throws DateMalformedStringException
+     */
+    public function createFromRow(array $row): RdbmsSaga
+    {
+        return new RdbmsSaga(
+            $row['sagaName'],
+            $row['sagaId'],
+            $row['payload'],
+            new DateTimeImmutable($row['createdAt']),
+            $row['updatedAt'] !== null ? new DateTimeImmutable($row['updatedAt']) : null,
+        );
+    }
+}

src/Saga/DoctrineRdbmsSagaStoreRepository.php

@@ -0,0 +1,121 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Gember\RdbmsEventStoreDoctrineDbal\Saga;
+
+use Doctrine\DBAL\Connection;
+use Gember\DependencyContracts\EventStore\Saga\RdbmsSaga;
+use Gember\DependencyContracts\EventStore\Saga\RdbmsSagaStoreRepository;
+use Gember\DependencyContracts\EventStore\Saga\RdbmsSagaNotFoundException;
+use Gember\RdbmsEventStoreDoctrineDbal\Saga\TableSchema\SagaStoreTableSchema;
+use Override;
+use Stringable;
+use DateTimeImmutable;
+
+/**
+ * @phpstan-type SagaRow array{
+ *     sagaId: string,
+ *     sagaName: string,
+ *     payload: string,
+ *     createdAt: string,
+ *     updatedAt: null|string
+ * }
+ */
+final readonly class DoctrineRdbmsSagaStoreRepository implements RdbmsSagaStoreRepository
+{
+    public function __construct(
+        private Connection $connection,
+        private SagaStoreTableSchema $sagaStoreTableSchema,
+        private DoctrineDbalRdbmsSagaFactory $sagaFactory,
+    ) {}
+
+    #[Override]
+    public function get(string $sagaName, Stringable|string $sagaId): RdbmsSaga
+    {
+        $sagaStoreSchema = $this->sagaStoreTableSchema;
+
+        /** @var false|SagaRow $row */
+        $row = $this->connection->createQueryBuilder()
+            ->select(
+                <<<DQL
+                ss.{$sagaStoreSchema->sagaIdFieldName} as sagaId,
+                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))
+            ->andWhere(sprintf('ss.%s = :sagaName', $sagaStoreSchema->sagaNameFieldName))
+            ->setParameter('sagaId', (string) $sagaId)
+            ->setParameter('sagaName', $sagaName)
+            ->executeQuery()
+            ->fetchAssociative();
+
+        if (!$row) {
+            throw RdbmsSagaNotFoundException::withSagaId($sagaName, $sagaId);
+        }
+
+        return $this->sagaFactory->createFromRow($row);
+    }
+
+    #[Override]
+    public function save(
+        string $sagaName,
+        Stringable|string $sagaId,
+        string $payload,
+        DateTimeImmutable $now,
+    ): RdbmsSaga {
+        $sagaStoreSchema = $this->sagaStoreTableSchema;
+
+        try {
+            $previous = $this->get($sagaName, $sagaId);
+        } catch (RdbmsSagaNotFoundException) {
+            $this->connection->createQueryBuilder()
+                ->insert($sagaStoreSchema->tableName)
+                ->setValue($sagaStoreSchema->sagaIdFieldName, ':sagaId')
+                ->setValue($sagaStoreSchema->sagaNameFieldName, ':sagaName')
+                ->setValue($sagaStoreSchema->payloadFieldName, ':payload')
+                ->setValue($sagaStoreSchema->createdAtFieldName, ':createdAt')
+                ->setParameters([
+                    'sagaId' => $sagaId,
+                    'sagaName' => $sagaName,
+                    'payload' => $payload,
+                    'createdAt' => $now->format($sagaStoreSchema->createdAtFieldFormat),
+                ])
+                ->executeStatement();
+
+            return new RdbmsSaga(
+                $sagaName,
+                $sagaId,
+                $payload,
+                $now,
+                null,
+            );
+        }
+
+        $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(
+            $sagaName,
+            $sagaId,
+            $payload,
+            $previous->createdAt,
+            $now,
+        );
+    }
+}

src/Saga/TableSchema/SagaStoreTableSchema.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Gember\RdbmsEventStoreDoctrineDbal\Saga\TableSchema;
+
+final readonly class SagaStoreTableSchema
+{
+    public function __construct(
+        public string $tableName,
+        public string $sagaIdFieldName,
+        public string $sagaNameFieldName,
+        public string $payloadFieldName,
+        public string $createdAtFieldName,
+        public string $createdAtFieldFormat,
+        public string $updatedAtFieldName,
+        public string $updatedAtFieldFormat,
+    ) {}
+}

src/Saga/TableSchema/SagaTableSchemaFactory.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Gember\RdbmsEventStoreDoctrineDbal\Saga\TableSchema;
+
+final readonly class SagaTableSchemaFactory
+{
+    public static function createDefaultSagaStore(
+        string $tableName = 'saga_store',
+        string $sagaIdFieldName = 'saga_id',
+        string $sagaNameFieldName = 'saga_name',
+        string $payloadFieldName = 'payload',
+        string $createdAtFieldName = 'created_at',
+        string $createdAtFieldFormat = 'Y-m-d H:i:s.u',
+        string $updatedAtFieldName = 'updated_at',
+        string $updatedAtFieldFormat = 'Y-m-d H:i:s.u',
+    ): SagaStoreTableSchema {
+        return new SagaStoreTableSchema(
+            $tableName,
+            $sagaIdFieldName,
+            $sagaNameFieldName,
+            $payloadFieldName,
+            $createdAtFieldName,
+            $createdAtFieldFormat,
+            $updatedAtFieldName,
+            $updatedAtFieldFormat,
+        );
+    }
+}

tests/Saga/DoctrineDbalRdbmsSagaFactoryTest.php

@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Gember\RdbmsEventStoreDoctrineDbal\Test\Saga;
+
+use Gember\RdbmsEventStoreDoctrineDbal\Saga\DoctrineDbalRdbmsSagaFactory;
+use PHPUnit\Framework\TestCase;
+use PHPUnit\Framework\Attributes\Test;
+use DateTimeImmutable;
+
+/**
+ * @internal
+ */
+final class DoctrineDbalRdbmsSagaFactoryTest extends TestCase
+{
+    #[Test]
+    public function itShouldCreateRdbmsSaga(): void
+    {
+        $saga = (new DoctrineDbalRdbmsSagaFactory())->createFromRow([
+            'sagaName' => 'some.saga',
+            'sagaId' => '01K76G1PGKPZ047KDN25PFPEEV',
+            'payload' => '{"some":"data"}',
+            'createdAt' => '2018-12-01 12:05:08.234543',
+            'updatedAt' => null,
+        ]);
+
+        self::assertSame('some.saga', $saga->sagaName);
+        self::assertSame('01K76G1PGKPZ047KDN25PFPEEV', $saga->sagaId);
+        self::assertSame('{"some":"data"}', $saga->payload);
+        self::assertEquals(new DateTimeImmutable('2018-12-01 12:05:08.234543'), $saga->createdAt);
+        self::assertNull($saga->updatedAt);
+    }
+}