Add checkpoints for decision models

docs/dynamic-consistency-boundary.md

@@ -299,6 +299,67 @@ final class GuestAlreadyCheckedIn extends BasicProjection
     }
 }
 ```
+### Checkpoints
+
+Over time, a projection may have to load a lot of events.
+If an event contains the whole state the projection needs, you can mark its apply method with `#[Checkpoint]`.
+The decision model then only loads events from the last of these events onwards,
+looked up with the tags of the projection.
+
+```php
+use Patchlevel\EventSourcing\Attribute\Apply;
+use Patchlevel\EventSourcing\Attribute\Checkpoint;
+use Patchlevel\EventSourcing\Identifier\Uuid;
+use Patchlevel\EventSourcing\Projection\BasicProjection;
+
+final class NumberOfGuestsInHotel extends BasicProjection
+{
+    public function __construct(
+        private readonly Uuid $hotelId,
+    ) {
+    }
+
+    public function initialState(): int
+    {
+        return 0;
+    }
+
+    /** @return list<string> */
+    protected function tagFilter(): array
+    {
+        return ["hotel:{$this->hotelId->toString()}"];
+    }
+
+    #[Apply]
+    #[Checkpoint]
+    public function applyGuestsCounted(int $state, GuestsCounted $event): int
+    {
+        return $event->numberOfGuests;
+    }
+
+    #[Apply]
+    public function applyGuestIsCheckedIn(int $state, GuestIsCheckedIn $event): int
+    {
+        return $state + 1;
+    }
+
+    #[Apply]
+    public function applyGuestIsCheckedOut(int $state, GuestIsCheckedOut $event): int
+    {
+        return $state - 1;
+    }
+}
+```
+:::note
+A checkpoint only affects the projection that declares it.
+Other projections in the same decision model and subscriptions still get all events they need.
+Nothing is archived and the append condition is the same as without a checkpoint.
+:::
+
+:::tip
+If a projection only ever needs its last event, you can override `lastEventIsEnough()` to return `true` instead.
+:::
+
 ## Define handlers
 
 We’ll implement three command handlers corresponding to our commands.

src/Attribute/Checkpoint.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Attribute;
+
+use Attribute;
+
+/**
+ * Marks an apply method of a projection whose event contains the whole state the projection needs.
+ * Events before the last of these events are not loaded anymore.
+ *
+ * @experimental
+ */
+#[Attribute(Attribute::TARGET_METHOD)]
+final class Checkpoint
+{
+}

src/DecisionModel/StoreDecisionModelBuilder.php

@@ -4,12 +4,14 @@
 
 namespace Patchlevel\EventSourcing\DecisionModel;
 
+use Patchlevel\EventSourcing\Projection\CheckpointPositions;
 use Patchlevel\EventSourcing\Projection\CompositeProjection;
 use Patchlevel\EventSourcing\Projection\Projection;
 use Patchlevel\EventSourcing\Store\AppendCondition;
 use Patchlevel\EventSourcing\Store\AppendStore;
 
 use function max;
+use function min;
 
 /** @experimental */
 final class StoreDecisionModelBuilder implements DecisionModelBuilder
@@ -23,10 +25,11 @@
     public function build(
         array $projections,
     ): DecisionModel {
-        $projection = new CompositeProjection($projections);
+        $from = CheckpointPositions::resolve($this->store, $projections);
+        $projection = new CompositeProjection($projections, $from);
 
         $query = $projection->query();
-        $stream = $this->store->query($query);
+        $stream = $this->store->query($query, $from === [] ? 0 : min($from));
 
         $state = $projection->initialState();
 

src/Projection/ApplyMethodDetectionError.php

@@ -80,4 +80,14 @@ public static function argumentTypeIsMissing(string $methodName): self
             ),
         );
     }
+
+    public static function checkpointWithoutApply(string $methodName): self
+    {
+        return new self(
+            sprintf(
+                'The method "%s" has a #[Checkpoint] attribute, but no #[Apply] attribute.',
+                $methodName,
+            ),
+        );
+    }
 }

src/Projection/BasicProjection.php

@@ -5,6 +5,7 @@
 namespace Patchlevel\EventSourcing\Projection;
 
 use Patchlevel\EventSourcing\Attribute\Apply;
+use Patchlevel\EventSourcing\Attribute\Checkpoint;
 use Patchlevel\EventSourcing\Message\Message;
 use Patchlevel\EventSourcing\Store\SubQuery;
 use ReflectionClass;
@@ -24,7 +25,7 @@
  * @template S = mixed
  * @implements Projection<S>
  */
-abstract class BasicProjection implements Projection, SubQueryProvider
+abstract class BasicProjection implements Projection, SubQueryProvider, CheckpointProvider
 {
     /** @var array<class-string, string>|null $applyMethods */
     private array|null $applyMethods = null;
@@ -67,6 +68,22 @@
         return $this->subQuery;
     }
 
+    /** @return list<class-string> */
+    public function checkpointEvents(): array
+    {
+        $checkpointEvents = [];
+
+        foreach ($this->applyMethods() as $eventClass => $methodName) {
+            if ((new ReflectionMethod($this, $methodName))->getAttributes(Checkpoint::class) === []) {
+                continue;
+            }
+
+            $checkpointEvents[] = $eventClass;
+        }
+
+        return $checkpointEvents;
+    }
+
     /** @return list<class-string> */
     protected function eventTypeFilter(): array
     {
@@ -88,6 +105,10 @@
             $attributes = $method->getAttributes(Apply::class);
 
             if ($attributes === []) {
+                if ($method->getAttributes(Checkpoint::class) !== []) {
+                    throw ApplyMethodDetectionError::checkpointWithoutApply($method->getName());
+                }
+
                 continue;
             }
 

src/Projection/CheckpointPositions.php

@@ -0,0 +1,68 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Projection;
+
+use Patchlevel\EventSourcing\Store\AppendStore;
+use Patchlevel\EventSourcing\Store\Query;
+use Patchlevel\EventSourcing\Store\SubQuery;
+
+use function array_map;
+use function array_values;
+use function max;
+
+/** @internal */
+final class CheckpointPositions
+{
+    /**
+     * Returns for each projection the index from which on it needs events.
+     * That is the index of its last checkpoint event, or 0 if it has none.
+     *
+     * @param array<string, Projection> $projections
+     *
+     * @return array<string, positive-int|0>
+     */
+    public static function resolve(AppendStore $store, array $projections): array
+    {
+        $positions = array_map(static fn (): int => 0, $projections);
+        $lookups = [];
+
+        foreach ($projections as $name => $projection) {
+            if (!$projection instanceof CheckpointProvider || !$projection instanceof SubQueryProvider) {
+                continue;
+            }
+
+            $checkpointEvents = $projection->checkpointEvents();
+
+            if ($checkpointEvents === []) {
+                continue;
+            }
+
+            $subQuery = $projection->subQuery();
+
+            $lookups[$name] = new SubQuery(
+                $subQuery->tags,
+                $checkpointEvents,
+                $subQuery->streamName,
+                true,
+            );
+        }
+
+        if ($lookups === []) {
+            return $positions;
+        }
+
+        foreach ($store->query(new Query(...array_values($lookups))) as $index => $message) {
+            foreach ($lookups as $name => $lookup) {
+                if (!$lookup->match($message)) {
+                    continue;
+                }
+
+                $positions[$name] = max($positions[$name], $index);
+            }
+        }
+
+        return $positions;
+    }
+}

src/Projection/CheckpointProvider.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Projection;
+
+/** @experimental */
+interface CheckpointProvider
+{
+    /**
+     * Events which contain the whole state the projection needs.
+     * Everything before the last of these events is not loaded anymore.
+     *
+     * @return list<class-string>
+     */
+    public function checkpointEvents(): array;
+}

src/Projection/CompositeProjection.php

@@ -5,16 +5,21 @@
 namespace Patchlevel\EventSourcing\Projection;
 
 use Patchlevel\EventSourcing\Message\Message;
+use Patchlevel\EventSourcing\Store\Header\IndexHeader;
 use Patchlevel\EventSourcing\Store\Query;
 
 use function array_map;
 
 /** @experimental */
 final class CompositeProjection
 {
-    /** @param array<string, Projection> $projections */
+    /**
+     * @param array<string, Projection>     $projections
+     * @param array<string, positive-int|0> $from        messages before this index are not applied to the projection
+     */
     public function __construct(
         private readonly array $projections,
+        private readonly array $from = [],
     ) {
     }
 
@@ -51,6 +56,13 @@
     public function apply(mixed $state, Message $message): mixed
     {
         foreach ($this->projections as $name => $projection) {
+            if (
+                ($this->from[$name] ?? 0) > 0
+                && $message->header(IndexHeader::class)->index < $this->from[$name]
+            ) {
+                continue;
+            }
+
             $state[$name] = $projection->apply($state[$name], $message);
         }
 

src/Projection/StoreProjectionBuilder.php

@@ -6,6 +6,8 @@
 
 use Patchlevel\EventSourcing\Store\AppendStore;
 
+use function min;
+
 /** @experimental */
 final class StoreProjectionBuilder implements ProjectionBuilder
 {
@@ -22,10 +24,11 @@
     public function build(
         array $projections,
     ): array {
-        $projection = new CompositeProjection($projections);
+        $from = CheckpointPositions::resolve($this->store, $projections);
+        $projection = new CompositeProjection($projections, $from);
 
         $query = $projection->query();
-        $stream = $this->store->query($query);
+        $stream = $this->store->query($query, $from === [] ? 0 : min($from));
 
         $state = $projection->initialState();
 

tests/Benchmark/DynamicConsistencyBoundary/Events/CourseCapacityChanged.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Benchmark\DynamicConsistencyBoundary\Events;
+
+use Patchlevel\EventSourcing\Attribute\Event;
+use Patchlevel\EventSourcing\Attribute\EventTag;
+
+#[Event('course.capacity_changed')]
+final class CourseCapacityChanged
+{
+    public function __construct(
+        #[EventTag(prefix: 'course')]
+        public readonly string $courseId,
+        public readonly int $capacity,
+    ) {
+    }
+}

tests/Benchmark/DynamicConsistencyBoundary/Events/CourseDefined.php

@@ -0,0 +1,19 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Benchmark\DynamicConsistencyBoundary\Events;
+
+use Patchlevel\EventSourcing\Attribute\Event;
+use Patchlevel\EventSourcing\Attribute\EventTag;
+
+#[Event('course.defined')]
+final class CourseDefined
+{
+    public function __construct(
+        #[EventTag(prefix: 'course')]
+        public readonly string $courseId,
+        public readonly int $capacity,
+    ) {
+    }
+}

tests/Benchmark/DynamicConsistencyBoundary/Events/InvoiceCreated.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Patchlevel\EventSourcing\Tests\Benchmark\DynamicConsistencyBoundary\Events;
+
+use Patchlevel\EventSourcing\Attribute\Event;
+use Patchlevel\EventSourcing\Attribute\EventTag;
+
+#[Event('invoice.created')]
+final class InvoiceCreated
+{
+    public function __construct(
+        #[EventTag(prefix: 'invoice')]
+        public readonly int $invoiceNumber,
+    ) {
+    }
+}