add inline projection 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

@@ -0,0 +1,79 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Projection;
+
+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\Hydrator;
+use Patchlevel\Hydrator\HydratorWithContext;
+use Patchlevel\Hydrator\MetadataHydrator;
+
+use function json_decode;
+use function json_encode;
+
+final class DoctrineProjectionStateStore implements ProjectionStateStore, DoctrineSchemaConfigurator
+{
+    public function __construct(
+        private readonly Connection $connection,
+        private readonly Hydrator $hydrator = new MetadataHydrator(),
+        private readonly SubscriberMetadataFactory $subscriberMetadataFactory = new AttributeSubscriberMetadataFactory(),
+        private readonly string $tableName = 'projection_state',
+    ) {
+    }
+
+    public function store(Projection $projection): void
+    {
+        $subscriberId = $this->subscriberId($projection);
+        $data = $this->hydrator->extract($projection);
+
+        $this->connection->insert($this->tableName, [
+            'id' => $subscriberId,
+            'state' => json_encode($data),
+        ]);
+    }
+
+    public function load(Projection $projection): void
+    {
+        $subscriberId = $this->subscriberId($projection);
+
+        $data = $this->connection->fetchAssociative(
+            'SELECT * FROM ' . $this->tableName . ' WHERE id = ?',
+            [$subscriberId],
+        );
+
+        if (!$data) {
+            return;
+        }
+
+        $this->hydrator->hydrate($projection::class, json_decode($data['state'], true), [HydratorWithContext::OBJECT_TO_POPULATE => $projection]);
+    }
+
+    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(Projection $projection): string
+    {
+        return $this->subscriberMetadataFactory->metadata($projection::class)->id;
+    }
+}

src/Subscription/Projection/DoctrineProjectionStateStore.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Subscription\Projection;
+
+use Patchlevel\EventSourcing\Subscription\Subscriber\BatchableSubscriber;
+use Patchlevel\Hydrator\Attribute\Ignore;
+
+abstract class Projection implements BatchableSubscriber
+{
+    public function __construct(
+        #[Ignore]
+        private readonly ProjectionStateStore $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/Projection/Projection.php

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

src/Subscription/Projection/ProjectionStateStore.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\Projection\Projection;
+use Patchlevel\EventSourcing\Tests\Integration\Subscription\Events\ProfileCreated;
+use Patchlevel\EventSourcing\Tests\Integration\Subscription\ProfileId;
+
+#[Projector('profile_inline')]
+final class ProfileInlineProjection extends Projection
+{
+    /** @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;
+    }
+}