Add support for EventsCritereon and ToIndexCriteroen in InMemoryStore. Also add IndexHeader, EventIdHeader and RecordedOnHeader to the messages in the InMemoryStore and fixed FromIndexCriteroen and transactional behaviour.

Author: DanielBadura

baseline.xml

@@ -159,11 +159,6 @@
       <code><![CDATA[$dateTimeType->convertToPHPValue($data['recorded_on'], $platform)]]></code>
     </MixedArgument>
   </file>
-  <file src="src/Store/InMemoryStore.php">
-    <MixedPropertyTypeCoercion>
-      <code><![CDATA[$this->messages]]></code>
-    </MixedPropertyTypeCoercion>
-  </file>
   <file src="src/Store/StreamDoctrineDbalStore.php">
     <DeprecatedMethod>
       <code><![CDATA[setPrimaryKey]]></code>

deptrac-baseline.yaml

@@ -14,3 +14,12 @@ deptrac:
       - Patchlevel\EventSourcing\Aggregate\AggregateRoot
     Patchlevel\EventSourcing\Attribute\Subscriber:
       - Patchlevel\EventSourcing\Subscription\RunMode
+    Patchlevel\EventSourcing\Store\DoctrineDbalStore:
+      - Patchlevel\EventSourcing\Aggregate\AggregateHeader
+    Patchlevel\EventSourcing\Store\DoctrineDbalStoreStream:
+      - Patchlevel\EventSourcing\Aggregate\AggregateHeader
+    Patchlevel\EventSourcing\Store\InMemoryStore:
+      - Patchlevel\EventSourcing\Aggregate\AggregateHeader
+      - Patchlevel\EventSourcing\Metadata\Event\EventRegistry
+    Patchlevel\EventSourcing\Store\MissingEventRegistry:
+      - Patchlevel\EventSourcing\Metadata\Event\EventRegistry

deptrac.yaml

@@ -186,8 +186,6 @@ deptrac:
       - Cryptography
       - MetadataAggregate
     Store:
-      - Aggregate
-      - Attribute
       - Clock
       - Message
       - Metadata

src/Store/Header/IndexHeader.php

@@ -4,10 +4,7 @@
 
 namespace Patchlevel\EventSourcing\Store\Header;
 
-/**
- * @psalm-immutable
- * @experimental
- */
+/** @psalm-immutable */
 final class IndexHeader
 {
     /** @param positive-int $index */

src/Store/InMemoryStore.php

@@ -6,21 +6,30 @@
 
 use Closure;
 use Patchlevel\EventSourcing\Aggregate\AggregateHeader;
+use Patchlevel\EventSourcing\Clock\SystemClock;
 use Patchlevel\EventSourcing\Message\HeaderNotFound;
 use Patchlevel\EventSourcing\Message\Message;
+use Patchlevel\EventSourcing\Metadata\Event\EventRegistry;
 use Patchlevel\EventSourcing\Store\Criteria\AggregateIdCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\AggregateNameCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\ArchivedCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\Criteria;
+use Patchlevel\EventSourcing\Store\Criteria\EventsCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\FromIndexCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\FromPlayheadCriterion;
 use Patchlevel\EventSourcing\Store\Criteria\StreamCriterion;
+use Patchlevel\EventSourcing\Store\Criteria\ToIndexCriterion;
+use Patchlevel\EventSourcing\Store\Header\EventIdHeader;
+use Patchlevel\EventSourcing\Store\Header\IndexHeader;
 use Patchlevel\EventSourcing\Store\Header\PlayheadHeader;
+use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
 use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
+use Psr\Clock\ClockInterface;
+use Ramsey\Uuid\Uuid;
+use Throwable;
 
 use function array_filter;
 use function array_map;
-use function array_push;
 use function array_reverse;
 use function array_slice;
 use function array_unique;
@@ -35,10 +44,16 @@
 
 final class InMemoryStore implements StreamStore
 {
-    /** @param array<positive-int|0, Message> $messages */
+    /** @var array<0|positive-int, Message> */
+    private array $messages = [];
+
+    /** @param list<Message> $messages */
     public function __construct(
-        private array $messages = [],
+        array $messages = [],
+        private readonly EventRegistry|null $eventRegistry = null,
+        private readonly ClockInterface $clock = new SystemClock(),
     ) {
+        $this->save(...$messages);
     }
 
     public function load(
@@ -71,7 +86,27 @@ public function count(Criteria|null $criteria = null): int
 
     public function save(Message ...$messages): void
     {
-        array_push($this->messages, ...$messages);
+        $this->transactional(function () use ($messages): void {
+            $count = count($this->messages);
+
+            foreach ($messages as $message) {
+                $count++;
+
+                if (!$message->hasHeader(IndexHeader::class)) {
+                    $message = $message->withHeader(new IndexHeader($count));
+                }
+
+                if (!$message->hasHeader(EventIdHeader::class)) {
+                    $message = $message->withHeader(new EventIdHeader(Uuid::uuid7()->toString()));
+                }
+
+                if (!$message->hasHeader(RecordedOnHeader::class)) {
+                    $message = $message->withHeader(new RecordedOnHeader($this->clock->now()));
+                }
+
+                $this->messages[] = $message;
+            }
+        });
     }
 
     /**
@@ -81,7 +116,14 @@ public function save(Message ...$messages): void
      */
     public function transactional(Closure $function): void
     {
-        $function();
+        $messages = $this->messages;
+        try {
+            $function();
+        } catch (Throwable $e) {
+            $this->messages = $messages;
+
+            throw $e;
+        }
     }
 
     /** @return list<string> */
@@ -134,9 +176,11 @@ private function filter(Criteria|null $criteria): array
             return $this->messages;
         }
 
+        $eventRegistry = $this->eventRegistry;
+
         return array_filter(
             $this->messages,
-            static function (Message $message, int $index) use ($criteria): bool {
+            static function (Message $message) use ($criteria, $eventRegistry): bool {
                 foreach ($criteria->all() as $criterion) {
                     switch ($criterion::class) {
                         case AggregateIdCriterion::class:
@@ -222,7 +266,35 @@ static function (Message $message, int $index) use ($criteria): bool {
 
                             break;
                         case FromIndexCriterion::class:
-                            if ($index < $criterion->fromIndex) {
+                            try {
+                                $index = $message->header(IndexHeader::class)->index;
+                            } catch (HeaderNotFound) {
+                                return false;
+                            }
+
+                            if ($index <= $criterion->fromIndex) {
+                                return false;
+                            }
+
+                            break;
+                        case ToIndexCriterion::class:
+                            try {
+                                $index = $message->header(IndexHeader::class)->index;
+                            } catch (HeaderNotFound) {
+                                return false;
+                            }
+
+                            if ($index >= $criterion->toIndex) {
+                                return false;
+                            }
+
+                            break;
+                        case EventsCriterion::class:
+                            if ($eventRegistry === null) {
+                                throw new MissingEventRegistry($criterion::class);
+                            }
+
+                            if (!in_array($eventRegistry->eventName($message->event()::class), $criterion->events)) {
                                 return false;
                             }
 

src/Store/MissingEventRegistry.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Store;
+
+use Patchlevel\EventSourcing\Metadata\Event\EventRegistry;
+use RuntimeException;
+
+use function sprintf;
+
+final class MissingEventRegistry extends RuntimeException
+{
+    /** @param class-string $criterionClass */
+    public function __construct(string $criterionClass)
+    {
+        parent::__construct(sprintf('criterion %s not supported without an %s given', $criterionClass, EventRegistry::class));
+    }
+}