[Experimental] introduce event id & fix split stream logic in stream store

Makefile

@@ -43,6 +43,10 @@ phpunit-integration: vendor
 phpunit-integration-postgres: vendor                                            ## run phpunit integration tests on postgres
 	DB_URL="pdo-pgsql://postgres:postgres@localhost:5432/eventstore?charset=utf8" vendor/bin/phpunit --testsuite=integration
 
+.PHONY: phpunit-integration-mysql
+phpunit-integration-mysql: vendor                                               ## run phpunit integration tests on mysql
+	DB_URL="pdo-mysql://root@localhost:3306/eventstore?charset=utf8" vendor/bin/phpunit --testsuite=integration
+
 .PHONY: phpunit-unit
 phpunit-unit: vendor                                             				## run phpunit unit tests
 	XDEBUG_MODE=coverage vendor/bin/phpunit --testsuite=unit

baseline.xml

@@ -110,7 +110,7 @@
   </file>
   <file src="src/Store/StreamDoctrineDbalStoreStream.php">
     <ArgumentTypeCoercion>
-      <code><![CDATA[$data['playhead'] === null ? null : (int)$data['playhead']]]></code>
+      <code><![CDATA[(int)$data['playhead']]]></code>
     </ArgumentTypeCoercion>
   </file>
   <file src="src/Subscription/Engine/DefaultSubscriptionEngine.php">
@@ -371,7 +371,6 @@
   </file>
   <file src="tests/Unit/Subscription/Engine/SubscriptionManagerTest.php">
     <InvalidArgument>
-      <code><![CDATA[$result]]></code>
       <code><![CDATA[$subscriptions]]></code>
       <code><![CDATA[$subscriptions]]></code>
     </InvalidArgument>

docker-compose.yaml

@@ -5,4 +5,12 @@ services:
       - POSTGRES_PASSWORD=postgres
       - POSTGRES_DB=eventstore
     ports:
-      - 5432:5432
+      - 5432:5432
+
+  mysql:
+    image: mysql:8
+    environment:
+      - MYSQL_ALLOW_EMPTY_PASSWORD="yes"
+      - MYSQL_DATABASE="eventstore"
+    ports:
+      - 3306:3306

phpstan-baseline.neon

@@ -56,7 +56,7 @@ parameters:
 			path: src/Store/InMemoryStore.php
 
 		-
-			message: "#^Parameter \\#2 \\$playhead of class Patchlevel\\\\EventSourcing\\\\Store\\\\StreamHeader constructor expects int\\<1, max\\>\\|null, int\\|null given\\.$#"
+			message: "#^Parameter \\#1 \\$playhead of class Patchlevel\\\\EventSourcing\\\\Store\\\\Header\\\\PlayheadHeader constructor expects int\\<1, max\\>, int given\\.$#"
 			count: 1
 			path: src/Store/StreamDoctrineDbalStoreStream.php
 

src/Console/OutputStyle.php

@@ -5,13 +5,14 @@
 namespace Patchlevel\EventSourcing\Console;
 
 use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
-use Patchlevel\EventSourcing\Message\HeaderNotFound;
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Message\Serializer\HeadersSerializer;
 use Patchlevel\EventSourcing\Serializer\Encoder\Encoder;
 use Patchlevel\EventSourcing\Serializer\EventSerializer;
 use Patchlevel\EventSourcing\Store\ArchivedHeader;
-use Patchlevel\EventSourcing\Store\StreamHeader;
+use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
+use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
+use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
 use Patchlevel\EventSourcing\Store\StreamStartHeader;
 use Symfony\Component\Console\Style\SymfonyStyle;
 use Throwable;
@@ -49,13 +50,38 @@
 
         $customHeaders = array_filter(
             $message->headers(),
-            static fn ($header) => !$header instanceof StreamHeader
+            static fn ($header) => !$header instanceof StreamNameHeader
+                && !$header instanceof PlayheadHeader
+                && !$header instanceof RecordedOnHeader
                 && !$header instanceof AggregateHeader
                 && !$header instanceof ArchivedHeader
                 && !$header instanceof StreamStartHeader,
         );
 
-        $metaHeader = $this->metaHeader($message);
+        $streamName = null;
+        $playhead = null;
+        $recordedOn = null;
+
+        if ($message->hasHeader(AggregateHeader::class)) {
+            $header = $message->header(AggregateHeader::class);
+
+            $streamName = $header->streamName();
+            $playhead = $header->playhead;
+            $recordedOn = $header->recordedOn;
+        }
+
+        if ($message->hasHeader(StreamNameHeader::class)) {
+            $streamName = $message->header(StreamNameHeader::class)->streamName;
+        }
+
+        if ($message->hasHeader(PlayheadHeader::class)) {
+            $playhead = $message->header(PlayheadHeader::class)->playhead;
+        }
+
+        if ($message->hasHeader(RecordedOnHeader::class)) {
+            $recordedOn = $message->header(RecordedOnHeader::class)->recordedOn;
+        }
+
         $streamStart = $message->hasHeader(StreamStartHeader::class);
         $achieved = $message->hasHeader(ArchivedHeader::class);
 
@@ -68,11 +94,11 @@
                 'streamStart',
                 'archived',
             ],
             [
                 [
-                    $metaHeader instanceof AggregateHeader ? $metaHeader->streamName() : $metaHeader->streamName,
-                    $metaHeader->playhead,
-                    $metaHeader->recordedOn?->format('Y-m-d H:i:s'),
+                    $streamName,
+                    $playhead,
+                    $recordedOn?->format('Y-m-d H:i:s'),
                     $streamStart ? 'yes' : 'no',
                     $achieved ? 'yes' : 'no',
                 ],
@@ -98,13 +124,4 @@
             $error = $error->getPrevious();
         } while ($error !== null);
     }
-
-    private function metaHeader(Message $message): AggregateHeader|StreamHeader
-    {
-        try {
-            return $message->header(AggregateHeader::class);
-        } catch (HeaderNotFound) {
-            return $message->header(StreamHeader::class);
-        }
-    }
 }

src/Message/Translator/AggregateToStreamHeaderTranslator.php

@@ -6,7 +6,9 @@
 
 use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
 use Patchlevel\EventSourcing\Message\Message;
-use Patchlevel\EventSourcing\Store\StreamHeader;
+use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
+use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
+use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
 
 /** @experimental */
 final class AggregateToStreamHeaderTranslator implements Translator
@@ -23,11 +25,9 @@ public function __invoke(Message $message): array
         return [
             $message
                 ->removeHeader(AggregateHeader::class)
-                ->withHeader(new StreamHeader(
-                    $aggregateHeader->streamName(),
-                    $aggregateHeader->playhead,
-                    $aggregateHeader->recordedOn,
-                )),
+                ->withHeader(new StreamNameHeader($aggregateHeader->streamName()))
+                ->withHeader(new PlayheadHeader($aggregateHeader->playhead))
+                ->withHeader(new RecordedOnHeader($aggregateHeader->recordedOn)),
         ];
     }
 }

src/Message/Translator/RecalculatePlayheadTranslator.php

@@ -5,9 +5,9 @@
 namespace Patchlevel\EventSourcing\Message\Translator;
 
 use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
-use Patchlevel\EventSourcing\Message\HeaderNotFound;
 use Patchlevel\EventSourcing\Message\Message;
-use Patchlevel\EventSourcing\Store\StreamHeader;
+use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
+use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
 
 use function array_key_exists;
 
@@ -19,42 +19,30 @@
     /** @return list<Message> */
     public function __invoke(Message $message): array
     {
-        try {
-            $header = $message->header(AggregateHeader::class);
-        } catch (HeaderNotFound) {
-            try {
-                $header = $message->header(StreamHeader::class);
-            } catch (HeaderNotFound) {
-                return [$message];
-            }
-        }
+        if ($message->hasHeader(StreamNameHeader::class) && $message->hasHeader(PlayheadHeader::class)) {
+            $streamName = $message->header(StreamNameHeader::class)->streamName;
 
-        $stream = $header instanceof StreamHeader ? $header->streamName : $header->streamName();
+            $playhead = $this->nextPlayhead($streamName);
 
-        $playhead = $this->nextPlayhead($stream);
-
-        if ($header->playhead === $playhead) {
-            return [$message];
+            return [
+                $message->withHeader(new PlayheadHeader($playhead)),
+            ];
         }
 
-        if ($header instanceof StreamHeader) {
+        if ($message->hasHeader(AggregateHeader::class)) {
+            $header = $message->header(AggregateHeader::class);
+
             return [
-                $message->withHeader(new StreamHeader(
-                    $header->streamName,
-                    $playhead,
+                $message->withHeader(new AggregateHeader(
+                    $header->aggregateName,
+                    $header->aggregateId,
+                    $this->nextPlayhead($header->streamName()),
                     $header->recordedOn,
                 )),
             ];
         }
 
-        return [
-            $message->withHeader(new AggregateHeader(
-                $header->aggregateName,
-                $header->aggregateId,
-                $playhead,
-                $header->recordedOn,
-            )),
-        ];
+        return [$message];
     }
 
     public function reset(): void

src/Message/Translator/UntilEventTranslator.php

@@ -6,9 +6,8 @@
 
 use DateTimeImmutable;
 use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
-use Patchlevel\EventSourcing\Message\HeaderNotFound;
 use Patchlevel\EventSourcing\Message\Message;
-use Patchlevel\EventSourcing\Store\StreamHeader;
+use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
 
 final class UntilEventTranslator implements Translator
 {
@@ -20,22 +19,26 @@
     /** @return list<Message> */
     public function __invoke(Message $message): array
     {
-        try {
+        if ($message->hasHeader(AggregateHeader::class)) {
             $header = $message->header(AggregateHeader::class);
-        } catch (HeaderNotFound) {
-            try {
-                $header = $message->header(StreamHeader::class);
-            } catch (HeaderNotFound) {
+
+            if ($header->recordedOn < $this->until) {
                 return [$message];
             }
+
+            return [];
         }
 
-        $recordedOn = $header->recordedOn;
+        if ($message->hasHeader(RecordedOnHeader::class)) {
+            $header = $message->header(RecordedOnHeader::class);
+
+            if ($header->recordedOn < $this->until) {
+                return [$message];
+            }
 
-        if ($recordedOn < $this->until) {
-            return [$message];
+            return [];
         }
 
-        return [];
+        return [$message];
     }
 }

src/Metadata/Message/MessageHeaderRegistry.php

@@ -6,7 +6,10 @@
 
 use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
 use Patchlevel\EventSourcing\Store\ArchivedHeader;
-use Patchlevel\EventSourcing\Store\StreamHeader;
+use Patchlevel\EventSourcing\Store\Header\EventIdHeader;
+use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
+use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
+use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
 use Patchlevel\EventSourcing\Store\StreamStartHeader;
 
 use function array_flip;
@@ -73,10 +76,13 @@ public function headerNames(): array
     public static function createWithInternalHeaders(array $headerNameToClassMap = []): self
     {
         $internalHeaders = [
-            'stream' => StreamHeader::class,
+            'streamName' => StreamNameHeader::class,
+            'playhead' => PlayheadHeader::class,
+            'recordedOn' => RecordedOnHeader::class,
             'aggregate' => AggregateHeader::class,
             'archived' => ArchivedHeader::class,
             'newStreamStart' => StreamStartHeader::class,
+            'eventId' => EventIdHeader::class,
         ];
 
         return new self($headerNameToClassMap + $internalHeaders);

src/Repository/DefaultRepository.php

@@ -17,9 +17,11 @@
 use Patchlevel\EventSourcing\Snapshot\SnapshotStore;
 use Patchlevel\EventSourcing\Snapshot\SnapshotVersionInvalid;
 use Patchlevel\EventSourcing\Store\Criteria\CriteriaBuilder;
+use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
+use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
+use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
 use Patchlevel\EventSourcing\Store\Store;
 use Patchlevel\EventSourcing\Store\Stream;
-use Patchlevel\EventSourcing\Store\StreamHeader;
 use Patchlevel\EventSourcing\Store\StreamStore;
 use Patchlevel\EventSourcing\Store\UniqueConstraintViolation;
 use Psr\Clock\ClockInterface;
@@ -142,11 +144,7 @@ public function load(AggregateRootId $id): AggregateRoot
             }
 
             if ($this->useStreamHeader) {
-                $playhead = $firstMessage->header(StreamHeader::class)->playhead;
-
-                if ($playhead === null) {
-                    throw new AggregateNotFound($this->metadata->className, $id);
-                }
+                $playhead = $firstMessage->header(PlayheadHeader::class)->playhead;
             } else {
                 $playhead = $firstMessage->header(AggregateHeader::class)->playhead;
             }
@@ -245,27 +243,35 @@ public function save(AggregateRoot $aggregate): void
 
             $aggregateName = $this->metadata->name;
 
-            $useStreamHeader = $this->useStreamHeader;
+            $streamName = $this->useStreamHeader ? StreamNameTranslator::streamName($aggregateName, $aggregateId) : null;
 
             $messages = array_map(
-                static function (object $event) use ($aggregateName, $aggregateId, &$playhead, $messageDecorator, $clock, $useStreamHeader) {
-                    if ($useStreamHeader) {
-                        $header = new StreamHeader(
-                            StreamNameTranslator::streamName($aggregateName, $aggregateId),
-                            ++$playhead,
-                            $clock->now(),
-                        );
+                static function (object $event) use (
+                    $aggregateName,
+                    $aggregateId,
+                    &$playhead,
+                    $messageDecorator,
+                    $clock,
+                    $streamName,
+                ) {
+                    $message = Message::create($event);
+
+                    if ($streamName !== null) {
+                        $message = $message
+                            ->withHeader(new StreamNameHeader($streamName))
+                            ->withHeader(new PlayheadHeader(++$playhead))
+                            ->withHeader(new RecordedOnHeader($clock->now()));
                     } else {
-                        $header = new AggregateHeader(
-                            $aggregateName,
-                            $aggregateId,
-                            ++$playhead,
-                            $clock->now(),
+                        $message = $message->withHeader(
+                            new AggregateHeader(
+                                $aggregateName,
+                                $aggregateId,
+                                ++$playhead,
+                                $clock->now(),
+                            ),
                         );
                     }
 
-                    $message = Message::create($event)->withHeader($header);
-
                     if ($messageDecorator) {
                         return $messageDecorator($message);
                     }