rewrite outbox store

src/WatchServer/MessageSerializer.php → src/EventBus/Serializer/MessageSerializer.php

@@ -2,7 +2,7 @@
 
 declare(strict_types=1);
 
-namespace Patchlevel\EventSourcing\WatchServer;
+namespace Patchlevel\EventSourcing\EventBus\Serializer;
 
 use Patchlevel\EventSourcing\EventBus\Message;
 

src/EventBus/Serializer/PhpNativeMessageSerializer.php

@@ -0,0 +1,28 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\EventBus\Serializer;
+
+use Patchlevel\EventSourcing\EventBus\Message;
+
+use function base64_decode;
+use function base64_encode;
+use function serialize;
+use function unserialize;
+
+final class PhpNativeMessageSerializer implements MessageSerializer
+{
+    public function serialize(Message $message): string
+    {
+        return base64_encode(serialize($message));
+    }
+
+    public function deserialize(string $content): Message
+    {
+        return unserialize(
+            base64_decode($content),
+            ['allowed_classes' => true],
+        );
+    }
+}

src/Outbox/DoctrineOutboxStore.php

@@ -8,10 +8,8 @@
 use Doctrine\DBAL\Schema\Schema;
 use Doctrine\DBAL\Types\Types;
 use Patchlevel\EventSourcing\EventBus\Message;
+use Patchlevel\EventSourcing\EventBus\Serializer\MessageSerializer;
 use Patchlevel\EventSourcing\Schema\SchemaConfigurator;
-use Patchlevel\EventSourcing\Serializer\EventSerializer;
-use Patchlevel\EventSourcing\Serializer\SerializedEvent;
-use Patchlevel\EventSourcing\Store\DoctrineHelper;
 use Patchlevel\EventSourcing\Store\WrongQueryResult;
 
 use function array_map;
@@ -20,9 +18,11 @@
 
 final class DoctrineOutboxStore implements OutboxStore, SchemaConfigurator
 {
+    public const HEADER_OUTBOX_IDENTIFIER = 'outboxIdentifier';
+
     public function __construct(
         private readonly Connection $connection,
-        private readonly EventSerializer $serializer,
+        private readonly MessageSerializer $messageSerializer,
         private readonly string $outboxTable = 'outbox',
     ) {
     }
@@ -32,24 +32,10 @@ public function saveOutboxMessage(Message ...$messages): void
         $this->connection->transactional(
             function (Connection $connection) use ($messages): void {
                 foreach ($messages as $message) {
-                    $event = $message->event();
-
-                    $data = $this->serializer->serialize($event);
-
                     $connection->insert(
                         $this->outboxTable,
                         [
-                            'aggregate' => $message->aggregateName(),
-                            'aggregate_id' => $message->aggregateId(),
-                            'playhead' => $message->playhead(),
-                            'event' => $data->name,
-                            'payload' => $data->payload,
-                            'recorded_on' => $message->recordedOn(),
-                            'custom_headers' => $message->customHeaders(),
-                        ],
-                        [
-                            'recorded_on' => Types::DATETIMETZ_IMMUTABLE,
-                            'custom_headers' => Types::JSON,
+                            'message' => $this->messageSerializer->serialize($message),
                         ],
                     );
                 }
@@ -66,20 +52,15 @@ public function retrieveOutboxMessages(int|null $limit = null): array
             ->setMaxResults($limit)
             ->getSQL();
 
-        /** @var list<array{aggregate: string, aggregate_id: string, playhead: string|int, event: string, payload: string, recorded_on: string, custom_headers: string}> $result */
+        /** @var list<array{id: int, message: string}> $result */
         $result = $this->connection->fetchAllAssociative($sql);
-        $platform = $this->connection->getDatabasePlatform();
 
         return array_map(
-            function (array $data) use ($platform) {
-                $event = $this->serializer->deserialize(new SerializedEvent($data['event'], $data['payload']));
-
-                return Message::create($event)
-                    ->withAggregateName($data['aggregate'])
-                    ->withAggregateId($data['aggregate_id'])
-                    ->withPlayhead(DoctrineHelper::normalizePlayhead($data['playhead'], $platform))
-                    ->withRecordedOn(DoctrineHelper::normalizeRecordedOn($data['recorded_on'], $platform))
-                    ->withCustomHeaders(DoctrineHelper::normalizeCustomHeaders($data['custom_headers'], $platform));
+            function (array $data) {
+                $message = $this->messageSerializer->deserialize($data['message']);
+
+                return $message
+                    ->withCustomHeader(self::HEADER_OUTBOX_IDENTIFIER, $data['id']);
             },
             $result,
         );
@@ -90,14 +71,9 @@ public function markOutboxMessageConsumed(Message ...$messages): void
         $this->connection->transactional(
             function (Connection $connection) use ($messages): void {
                 foreach ($messages as $message) {
-                    $connection->delete(
-                        $this->outboxTable,
-                        [
-                            'aggregate' => $message->aggregateName(),
-                            'aggregate_id' => $message->aggregateId(),
-                            'playhead' => $message->playhead(),
-                        ],
-                    );
+                    $id = $message->customHeader(self::HEADER_OUTBOX_IDENTIFIER);
+
+                    $connection->delete($this->outboxTable, ['id' => $id]);
                 }
             },
         );
@@ -123,24 +99,13 @@ public function configureSchema(Schema $schema, Connection $connection): void
     {
         $table = $schema->createTable($this->outboxTable);
 
-        $table->addColumn('aggregate', Types::STRING)
-            ->setLength(255)
-            ->setNotnull(true);
-        $table->addColumn('aggregate_id', Types::STRING)
-            ->setLength(32)
-            ->setNotnull(true);
-        $table->addColumn('playhead', Types::INTEGER)
-            ->setNotnull(true);
-        $table->addColumn('event', Types::STRING)
-            ->setLength(255)
-            ->setNotnull(true);
-        $table->addColumn('payload', Types::JSON)
-            ->setNotnull(true);
-        $table->addColumn('recorded_on', Types::DATETIMETZ_IMMUTABLE)
-            ->setNotnull(false);
-        $table->addColumn('custom_headers', Types::JSON)
-            ->setNotnull(true);
-
-        $table->setPrimaryKey(['aggregate', 'aggregate_id', 'playhead']);
+        $table->addColumn('id', Types::INTEGER)
+            ->setNotnull(true)
+            ->setAutoincrement(true);
+        $table->addColumn('message', Types::STRING)
+            ->setNotnull(true)
+            ->setLength(16_000);
+
+        $table->setPrimaryKey(['id']);
     }
 }

src/WatchServer/SocketWatchServer.php

@@ -5,6 +5,7 @@
 namespace Patchlevel\EventSourcing\WatchServer;
 
 use Closure;
+use Patchlevel\EventSourcing\EventBus\Serializer\MessageSerializer;
 use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use RuntimeException;

src/WatchServer/SocketWatchServerClient.php

@@ -5,6 +5,7 @@
 namespace Patchlevel\EventSourcing\WatchServer;
 
 use Patchlevel\EventSourcing\EventBus\Message;
+use Patchlevel\EventSourcing\EventBus\Serializer\MessageSerializer;
 
 use function fclose;
 use function restore_error_handler;

tests/Integration/Outbox/OutboxTest.php

@@ -7,6 +7,7 @@
 use Doctrine\DBAL\Connection;
 use Patchlevel\EventSourcing\EventBus\ChainEventBus;
 use Patchlevel\EventSourcing\EventBus\DefaultConsumer;
+use Patchlevel\EventSourcing\EventBus\Serializer\PhpNativeMessageSerializer;
 use Patchlevel\EventSourcing\Outbox\DoctrineOutboxStore;
 use Patchlevel\EventSourcing\Outbox\EventBusPublisher;
 use Patchlevel\EventSourcing\Outbox\OutboxEventBus;
@@ -58,7 +59,7 @@ public function testSuccessful(): void
 
         $outboxStore = new DoctrineOutboxStore(
             $this->connection,
-            $serializer,
+            new PhpNativeMessageSerializer(),
             'outbox',
         );
 

tests/Unit/EventBus/Serializer/PhpNativeMessageSerializerTest.php

@@ -0,0 +1,70 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Unit\EventBus\Serializer;
+
+use DateTimeImmutable;
+use Patchlevel\EventSourcing\EventBus\Message;
+use Patchlevel\EventSourcing\EventBus\Serializer\PhpNativeMessageSerializer;
+use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId;
+use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited;
+use PHPUnit\Framework\TestCase;
+use Prophecy\PhpUnit\ProphecyTrait;
+
+/** @covers \Patchlevel\EventSourcing\EventBus\Serializer\PhpNativeMessageSerializer */
+final class PhpNativeMessageSerializerTest extends TestCase
+{
+    use ProphecyTrait;
+
+    public function testSerialize(): void
+    {
+        $event = new ProfileVisited(
+            ProfileId::fromString('foo'),
+        );
+
+        $message = Message::create($event)
+            ->withRecordedOn(new DateTimeImmutable('2020-01-01T20:00:00.000000+0100'));
+
+        $nativeSerializer = new PhpNativeMessageSerializer();
+
+        $content = $nativeSerializer->serialize($message);
+
+        self::assertEquals('Tzo0MToiUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UiOjg6e3M6NTY6IgBQYXRjaGxldmVsXEV2ZW50U291cmNpbmdcRXZlbnRCdXNcTWVzc2FnZQBhZ2dyZWdhdGVOYW1lIjtOO3M6NTQ6IgBQYXRjaGxldmVsXEV2ZW50U291cmNpbmdcRXZlbnRCdXNcTWVzc2FnZQBhZ2dyZWdhdGVJZCI7TjtzOjUxOiIAUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UAcGxheWhlYWQiO047czo1MzoiAFBhdGNobGV2ZWxcRXZlbnRTb3VyY2luZ1xFdmVudEJ1c1xNZXNzYWdlAHJlY29yZGVkT24iO086MTc6IkRhdGVUaW1lSW1tdXRhYmxlIjozOntzOjQ6ImRhdGUiO3M6MjY6IjIwMjAtMDEtMDEgMjA6MDA6MDAuMDAwMDAwIjtzOjEzOiJ0aW1lem9uZV90eXBlIjtpOjE7czo4OiJ0aW1lem9uZSI7czo2OiIrMDE6MDAiO31zOjU3OiIAUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UAbmV3U3RyZWFtU3RhcnQiO2I6MDtzOjUxOiIAUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UAYXJjaGl2ZWQiO2I6MDtzOjU2OiIAUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UAY3VzdG9tSGVhZGVycyI7YTowOnt9czo0ODoiAFBhdGNobGV2ZWxcRXZlbnRTb3VyY2luZ1xFdmVudEJ1c1xNZXNzYWdlAGV2ZW50IjtPOjU4OiJQYXRjaGxldmVsXEV2ZW50U291cmNpbmdcVGVzdHNcVW5pdFxGaXh0dXJlXFByb2ZpbGVWaXNpdGVkIjoxOntzOjk6InZpc2l0b3JJZCI7Tzo1MzoiUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXFRlc3RzXFVuaXRcRml4dHVyZVxQcm9maWxlSWQiOjE6e3M6NTc6IgBQYXRjaGxldmVsXEV2ZW50U291cmNpbmdcVGVzdHNcVW5pdFxGaXh0dXJlXFByb2ZpbGVJZABpZCI7czozOiJmb28iO319fQ==', $content);
+    }
+
+    public function testDeserialize(): void
+    {
+        $event = new ProfileVisited(
+            ProfileId::fromString('foo'),
+        );
+
+        $nativeSerializer = new PhpNativeMessageSerializer();
+
+        $message = $nativeSerializer->deserialize('Tzo0MToiUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UiOjg6e3M6NTY6IgBQYXRjaGxldmVsXEV2ZW50U291cmNpbmdcRXZlbnRCdXNcTWVzc2FnZQBhZ2dyZWdhdGVOYW1lIjtOO3M6NTQ6IgBQYXRjaGxldmVsXEV2ZW50U291cmNpbmdcRXZlbnRCdXNcTWVzc2FnZQBhZ2dyZWdhdGVJZCI7TjtzOjUxOiIAUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UAcGxheWhlYWQiO047czo1MzoiAFBhdGNobGV2ZWxcRXZlbnRTb3VyY2luZ1xFdmVudEJ1c1xNZXNzYWdlAHJlY29yZGVkT24iO086MTc6IkRhdGVUaW1lSW1tdXRhYmxlIjozOntzOjQ6ImRhdGUiO3M6MjY6IjIwMjAtMDEtMDEgMjA6MDA6MDAuMDAwMDAwIjtzOjEzOiJ0aW1lem9uZV90eXBlIjtpOjE7czo4OiJ0aW1lem9uZSI7czo2OiIrMDE6MDAiO31zOjU3OiIAUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UAbmV3U3RyZWFtU3RhcnQiO2I6MDtzOjUxOiIAUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UAYXJjaGl2ZWQiO2I6MDtzOjU2OiIAUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXEV2ZW50QnVzXE1lc3NhZ2UAY3VzdG9tSGVhZGVycyI7YTowOnt9czo0ODoiAFBhdGNobGV2ZWxcRXZlbnRTb3VyY2luZ1xFdmVudEJ1c1xNZXNzYWdlAGV2ZW50IjtPOjU4OiJQYXRjaGxldmVsXEV2ZW50U291cmNpbmdcVGVzdHNcVW5pdFxGaXh0dXJlXFByb2ZpbGVWaXNpdGVkIjoxOntzOjk6InZpc2l0b3JJZCI7Tzo1MzoiUGF0Y2hsZXZlbFxFdmVudFNvdXJjaW5nXFRlc3RzXFVuaXRcRml4dHVyZVxQcm9maWxlSWQiOjE6e3M6NTc6IgBQYXRjaGxldmVsXEV2ZW50U291cmNpbmdcVGVzdHNcVW5pdFxGaXh0dXJlXFByb2ZpbGVJZABpZCI7czozOiJmb28iO319fQ==');
+
+        self::assertEquals($event, $message->event());
+        self::assertEquals([
+            'recordedOn' => new DateTimeImmutable('2020-01-01T20:00:00.000000+0100'),
+            'newStreamStart' => false,
+            'archived' => false,
+        ], $message->headers());
+    }
+
+    public function testEquals(): void
+    {
+        $event = new ProfileVisited(
+            ProfileId::fromString('foo'),
+        );
+
+        $message = Message::create($event)
+            ->withRecordedOn(new DateTimeImmutable('2020-01-01T20:00:00.000000+0100'));
+
+        $nativeSerializer = new PhpNativeMessageSerializer();
+
+        $content = $nativeSerializer->serialize($message);
+        $clonedMessage = $nativeSerializer->deserialize($content);
+
+        self::assertEquals($message, $clonedMessage);
+    }
+}