add stateful subscriber feature

Author: DavidBadura

composer.json

@@ -34,7 +34,7 @@
     "php": "~8.2.0 || ~8.3.0 || ~8.4.0 || ~8.5.0",
     "doctrine/dbal": "^4.4.0",
     "doctrine/migrations": "^3.3.2",
-    "patchlevel/hydrator": "^1.8.0",
+    "patchlevel/hydrator": "^1.19.0",
     "patchlevel/worker": "^1.4.0",
     "psr/cache": "^2.0.0 || ^3.0.0",
     "psr/clock": "^1.0",

composer.lock

@@ -162,6 +162,18 @@ parameters:
 			count: 1
 			path: src/Subscription/RetryStrategy/ClockBasedRetryStrategy.php
 
+		-
+			message: '#^Parameter \#1 \$json of function json_decode expects string, mixed given\.$#'
+			identifier: argument.type
+			count: 1
+			path: src/Subscription/StatefulSubscriber/DoctrineStatefulSubscriberStore.php
+
+		-
+			message: '#^Parameter \#2 \$data of method Patchlevel\\Hydrator\\HydratorWithContext\:\:hydrate\(\) expects array\<string, mixed\>, mixed given\.$#'
+			identifier: argument.type
+			count: 1
+			path: src/Subscription/StatefulSubscriber/DoctrineStatefulSubscriberStore.php
+
 		-
 			message: '#^Parameter \#3 \$errorContext of class Patchlevel\\EventSourcing\\Subscription\\SubscriptionError constructor expects list\<array\{namespace\: string, short_name\: string, class\: class\-string, message\: string, code\: int\|string, file\: string, line\: int, trace\: list\<array\{file\?\: string, line\?\: int, function\?\: string, class\?\: string, type\?\: string, args\?\: array\<mixed\>\}\>\}\>\|null, mixed given\.$#'
 			identifier: argument.type

phpstan-baseline.neon

@@ -0,0 +1,85 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\StatefulSubscriber;
+
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\DBAL\Types\Types;
+use Patchlevel\EventSourcing\Metadata\Subscriber\AttributeSubscriberMetadataFactory;
+use Patchlevel\EventSourcing\Metadata\Subscriber\SubscriberMetadataFactory;
+use Patchlevel\EventSourcing\Schema\DoctrineHelper;
+use Patchlevel\EventSourcing\Schema\DoctrineSchemaConfigurator;
+use Patchlevel\Hydrator\HydratorWithContext;
+use Patchlevel\Hydrator\MetadataHydrator;
+
+use function json_decode;
+use function json_encode;
+
+final readonly class DoctrineStatefulSubscriberStore implements StatefulSubscriberStore, DoctrineSchemaConfigurator
+{
+    public function __construct(
+        private Connection $connection,
+        private HydratorWithContext $hydrator = new MetadataHydrator(),
+        private SubscriberMetadataFactory $subscriberMetadataFactory = new AttributeSubscriberMetadataFactory(),
+        private string $tableName = 'stateful_subscriber_state',
+    ) {
+    }
+
+    public function store(StatefulSubscriber $subscriber): void
+    {
+        $subscriberId = $this->subscriberId($subscriber);
+        $data = $this->hydrator->extract($subscriber);
+
+        $this->connection->insert(
+            $this->tableName,
+            [
+                'id' => $subscriberId,
+                'state' => json_encode($data),
+            ],
+        );
+    }
+
+    public function load(StatefulSubscriber $subscriber): void
+    {
+        $subscriberId = $this->subscriberId($subscriber);
+
+        $data = $this->connection->fetchAssociative(
+            'SELECT * FROM ' . $this->tableName . ' WHERE id = ?',
+            [$subscriberId],
+        );
+
+        if (!$data) {
+            return;
+        }
+
+        $this->hydrator->hydrate(
+            $subscriber::class,
+            json_decode($data['state'], true),
+            [HydratorWithContext::OBJECT_TO_POPULATE => $subscriber],
+        );
+    }
+
+    public function configureSchema(Schema $schema, Connection $connection): void
+    {
+        if (!DoctrineHelper::sameDatabase($this->connection, $connection)) {
+            return;
+        }
+
+        $table = $schema->createTable($this->tableName);
+
+        $table->addColumn('id', Types::STRING)
+            ->setLength(255)
+            ->setNotnull(true);
+        $table->addColumn('state', Types::JSON)
+            ->setNotnull(true);
+
+        $table->setPrimaryKey(['id']);
+    }
+
+    public function subscriberId(StatefulSubscriber $projection): string
+    {
+        return $this->subscriberMetadataFactory->metadata($projection::class)->id;
+    }
+}

src/Subscription/StatefulSubscriber/DoctrineStatefulSubscriberStore.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\StatefulSubscriber;
+
+use Patchlevel\EventSourcing\Subscription\Subscriber\BatchableSubscriber;
+use Patchlevel\Hydrator\Attribute\Ignore;
+
+abstract class StatefulSubscriber implements BatchableSubscriber
+{
+    public function __construct(
+        #[Ignore]
+        private readonly StatefulSubscriberStore $store,
+    ) {
+        $this->store->load($this);
+    }
+
+    public function beginBatch(): void
+    {
+        // do nothing
+    }
+
+    public function commitBatch(): void
+    {
+        $this->store->store($this);
+    }
+
+    public function rollbackBatch(): void
+    {
+        $this->store->load($this);
+    }
+
+    public function forceCommit(): bool
+    {
+        return false;
+    }
+}

src/Subscription/StatefulSubscriber/StatefulSubscriber.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\StatefulSubscriber;
+
+interface StatefulSubscriberStore
+{
+    public function store(StatefulSubscriber $subscriber): void;
+
+    public function load(StatefulSubscriber $subscriber): void;
+}

src/Subscription/StatefulSubscriber/StatefulSubscriberStore.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Integration\Subscription\Subscriber;
+
+use Patchlevel\EventSourcing\Attribute\Projector;
+use Patchlevel\EventSourcing\Attribute\Subscribe;
+use Patchlevel\EventSourcing\Subscription\StatefulSubscriber\StatefulSubscriber;
+use Patchlevel\EventSourcing\Tests\Integration\Subscription\Events\ProfileCreated;
+use Patchlevel\EventSourcing\Tests\Integration\Subscription\ProfileId;
+
+#[Projector('profile_inline')]
+final class ProfileInlineStatefulSubscriber extends StatefulSubscriber
+{
+    /** @var array<string, string> */
+    public array $profiles = [];
+
+    #[Subscribe(ProfileCreated::class)]
+    public function handleProfileCreated(ProfileCreated $profileCreated): void
+    {
+        $this->profiles[$profileCreated->profileId->toString()] = $profileCreated->name;
+    }
+
+    public function findById(ProfileId $id): string|null
+    {
+        return $this->profiles[$id->toString()] ?? null;
+    }
+}