add only last event filter

Author: DavidBadura

src/CommandBus/InstantRetryCommandBus.php

@@ -6,6 +6,7 @@
 
 use Patchlevel\EventSourcing\Attribute\InstantRetry;
 use Patchlevel\EventSourcing\Repository\AggregateOutdated;
+use Patchlevel\EventSourcing\Store\AppendConditionNotMet;
 use ReflectionClass;
 use Throwable;
 
@@ -20,7 +21,7 @@ final class InstantRetryCommandBus implements CommandBus
     public function __construct(
         private readonly CommandBus $commandBus,
         private readonly int $defaultMaxRetries = 3,
-        private readonly array $defaultExceptions = [AggregateOutdated::class],
+        private readonly array $defaultExceptions = [AggregateOutdated::class, AppendConditionNotMet::class],
     ) {
     }
 

src/DCB/EventRouter.php

@@ -52,6 +52,7 @@ public function subQuery(): SubQuery
             $this->tagFilter(),
             $this->eventTypeFilter(),
             $this->streamName(),
+            $this->lastEventIsEnough(),
         );
     }
 
@@ -182,4 +183,9 @@ public function streamName(): string|null
     {
         return null;
     }
+
+    public function lastEventIsEnough(): bool
+    {
+        return false;
+    }
 }

src/Store/SubQuery.php

@@ -23,6 +23,7 @@ public function __construct(
         public readonly array $tags = [],
         public readonly array $events = [],
         public readonly string|null $streamName = null,
+        public readonly bool $onlyLastEvent = false,
     ) {
         sort($tags);
         sort($events);
@@ -53,7 +54,8 @@ public function equals(self $queryComponent): bool
     {
         return $this->streamName === $queryComponent->streamName
             && $this->tags === $queryComponent->tags
-            && $this->events === $queryComponent->events;
+            && $this->events === $queryComponent->events
+            && $this->onlyLastEvent === $queryComponent->onlyLastEvent;
     }
 
     /**
@@ -67,6 +69,8 @@ private function isSubset(array $needle, array $haystack): bool
 
     public function empty(): bool
     {
-        return $this->streamName === null && $this->tags === [] && $this->events === [];
+        return $this->streamName === null
+            && $this->tags === []
+            && $this->events === [];
     }
 }

src/Store/TaggableDoctrineDbalStore.php

@@ -803,11 +803,10 @@ private function unlock(): void
     private function uniqueParameterGenerator(): Closure
     {
         return static function () {
+            /** @var int $counter */
             static $counter = 0;
 
-            ++$counter;
-
-            return 'param' . $counter;
+            return 'param' . ++$counter;
         };
     }
 
@@ -868,6 +867,10 @@ private function queryCondition(QueryBuilder $builder, Query $query): void
                 );
             }
 
+            if ($subQuery->onlyLastEvent) {
+                $subQueryBuilder->select('MAX(id) AS id');
+            }
+
             $subqueries[] = $subQueryBuilder->getSQL();
         }
 

tests/Integration/DynamicConsistencyBoundary/Invoice/Projection/NextInvoiceNumberProjection.php

@@ -27,6 +27,11 @@ public function initialState(): int
     #[Apply]
     public function applyInvoiceCreated(int $state, InvoiceCreated $event): int
     {
-        return $state + 1;
+        return $event->invoiceNumber + 1;
+    }
+
+    public function lastEventIsEnough(): bool
+    {
+        return true;
     }
 }

tests/Integration/Store/TaggableDoctrineDbalStoreTest.php

@@ -687,6 +687,34 @@ public function testQueryTagAndEvent(): void
         self::assertStreamEquals([$message1], $stream);
     }
 
+    public function testOnlyLastEvent(): void
+    {
+        $profileId1 = ProfileId::generate();
+        $profileId2 = ProfileId::generate();
+
+        $message1 = Message::create(new ProfileCreated($profileId1, 'test'))
+            ->withHeader(new StreamNameHeader('foo'))
+            ->withHeader(new RecordedOnHeader(new DateTimeImmutable('2020-01-01 00:00:00')))
+            ->withHeader(new TagsHeader(['profile:' . $profileId1->toString()]));
+        $message2 = Message::create(new ProfileCreated($profileId2, 'test'))
+            ->withHeader(new StreamNameHeader('foo'))
+            ->withHeader(new RecordedOnHeader(new DateTimeImmutable('2020-01-01 00:00:00')))
+            ->withHeader(new TagsHeader(['profile:' . $profileId2->toString()]));
+
+        $this->store->append([$message1, $message2]);
+
+        $stream = $this->store->query(
+            new Query(
+                new SubQuery(
+                    events: [ProfileCreated::class],
+                    onlyLastEvent: true,
+                ),
+            ),
+        );
+
+        self::assertStreamEquals([$message2], $stream);
+    }
+
     public function testComplexQuery(): void
     {
         $profileId1 = ProfileId::generate();