From a16e56c61dfe4702a5a82c8b1d8941d26f462700 Mon Sep 17 00:00:00 2001 From: David Badura Date: Sat, 3 Oct 2026 21:31:42 +0200 Subject: [PATCH] Respect split stream events when building decision models An event marked with #[SplitStream] holds the whole state that is relevant from that point on. Aggregates already use it to skip older events, but decision models always read the complete history of their projections. Before reading, the builders now look up the last split event of every projection that handles one, using the projection's own tags, and read from there. Projections only get events from their own split event onwards, so a projection without split events in the same decision model still gets its whole history. Nothing is archived, the append condition stays the same. A projection can opt out by overriding splitEvents(). --- docs/split-stream.md | 66 +++++++ .../StoreDecisionModelBuilder.php | 7 +- src/Projection/BasicProjection.php | 14 +- src/Projection/CompositeProjection.php | 14 +- src/Projection/SplitStreamPositions.php | 68 +++++++ src/Projection/SplitStreamProvider.php | 17 ++ src/Projection/StoreProjectionBuilder.php | 7 +- .../StoreDecisionModelBuilderTest.php | 167 ++++++++++++++++++ tests/Unit/Fixture/VisitsProjection.php | 40 +++++ tests/Unit/Projection/BasicProjectionTest.php | 17 ++ .../Projection/CompositeProjectionTest.php | 24 +++ .../Projection/StoreProjectionBuilderTest.php | 19 ++ 12 files changed, 454 insertions(+), 6 deletions(-) create mode 100644 src/Projection/SplitStreamPositions.php create mode 100644 src/Projection/SplitStreamProvider.php create mode 100644 tests/Unit/Fixture/VisitsProjection.php diff --git a/docs/split-stream.md b/docs/split-stream.md index ad10e6122..13430283a 100644 --- a/docs/split-stream.md +++ b/docs/split-stream.md @@ -82,6 +82,72 @@ This impacts only the aggregate loaded by the repository. Subscriptions will sti You can combine this feature with the snapshot feature to increase the performance even more. ::: +## Dynamic Consistency Boundary + +Projections of the [dynamic consistency boundary](dynamic-consistency-boundary.md) also respect `#[SplitStream]`. +If a projection has an apply method for such an event, +the decision model is only built from the last of these events matching the projection's tags onwards. +No configuration is needed and nothing is archived, other projections and subscriptions still see all events. + +```php +use Patchlevel\EventSourcing\Attribute\Apply; +use Patchlevel\EventSourcing\Projection\BasicProjection; + +final class Balance extends BasicProjection +{ + public function __construct( + private readonly BankAccountId $bankAccountId, + ) { + } + + public function initialState(): int + { + return 0; + } + + /** @return list */ + protected function tagFilter(): array + { + return ["bank_account:{$this->bankAccountId->toString()}"]; + } + + #[Apply] + public function applyBalanceReported(int $state, BalanceReported $event): int + { + return $event->balanceInCents; + } + + #[Apply] + public function applyMoneyDeposited(int $state, MoneyDeposited $event): int + { + return $state + $event->amountInCents; + } +} +``` +:::note +Only the split events of the projection itself are considered. +A `BalanceReported` of another bank account does not cut off the history of this one, +and other projections in the same decision model still get all events they need. +::: + +If a projection handles a split event but needs the whole history anyway, for example to count them, +you can opt out by overriding `splitEvents`. + +```php +use Patchlevel\EventSourcing\Projection\BasicProjection; + +final class NumberOfBalanceReports extends BasicProjection +{ + // ... + + /** @return list */ + public function splitEvents(): array + { + return []; + } +} +``` + ## Learn more * [How to use message decorator](message-decorator.md) diff --git a/src/DecisionModel/StoreDecisionModelBuilder.php b/src/DecisionModel/StoreDecisionModelBuilder.php index 7eaef7b6b..dad0e51d9 100644 --- a/src/DecisionModel/StoreDecisionModelBuilder.php +++ b/src/DecisionModel/StoreDecisionModelBuilder.php @@ -6,10 +6,12 @@ use Patchlevel\EventSourcing\Projection\CompositeProjection; use Patchlevel\EventSourcing\Projection\Projection; +use Patchlevel\EventSourcing\Projection\SplitStreamPositions; 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 __construct( public function build( array $projections, ): DecisionModel { - $projection = new CompositeProjection($projections); + $from = SplitStreamPositions::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(); diff --git a/src/Projection/BasicProjection.php b/src/Projection/BasicProjection.php index 8a9d0b5c8..a022e9644 100644 --- a/src/Projection/BasicProjection.php +++ b/src/Projection/BasicProjection.php @@ -5,6 +5,7 @@ namespace Patchlevel\EventSourcing\Projection; use Patchlevel\EventSourcing\Attribute\Apply; +use Patchlevel\EventSourcing\Attribute\SplitStream; use Patchlevel\EventSourcing\Message\Message; use Patchlevel\EventSourcing\Store\SubQuery; use ReflectionClass; @@ -13,10 +14,12 @@ use ReflectionNamedType; use ReflectionUnionType; +use function array_filter; use function array_key_exists; use function array_keys; use function array_map; use function array_merge; +use function array_values; use function class_exists; /** @@ -24,7 +27,7 @@ * @template S = mixed * @implements Projection */ -abstract class BasicProjection implements Projection, SubQueryProvider +abstract class BasicProjection implements Projection, SubQueryProvider, SplitStreamProvider { /** @var array|null $applyMethods */ private array|null $applyMethods = null; @@ -67,6 +70,15 @@ public function subQuery(): SubQuery return $this->subQuery; } + /** @return list */ + public function splitEvents(): array + { + return array_values(array_filter( + $this->eventTypeFilter(), + static fn (string $eventClass): bool => (new ReflectionClass($eventClass))->getAttributes(SplitStream::class) !== [], + )); + } + /** @return list */ protected function eventTypeFilter(): array { diff --git a/src/Projection/CompositeProjection.php b/src/Projection/CompositeProjection.php index 4271aea8c..409880ceb 100644 --- a/src/Projection/CompositeProjection.php +++ b/src/Projection/CompositeProjection.php @@ -5,6 +5,7 @@ namespace Patchlevel\EventSourcing\Projection; use Patchlevel\EventSourcing\Message\Message; +use Patchlevel\EventSourcing\Store\Header\IndexHeader; use Patchlevel\EventSourcing\Store\Query; use function array_map; @@ -12,9 +13,13 @@ /** @experimental */ final class CompositeProjection { - /** @param array $projections */ + /** + * @param array $projections + * @param array $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 initialState(): array 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); } diff --git a/src/Projection/SplitStreamPositions.php b/src/Projection/SplitStreamPositions.php new file mode 100644 index 000000000..4adfcdfba --- /dev/null +++ b/src/Projection/SplitStreamPositions.php @@ -0,0 +1,68 @@ + $projections + * + * @return array + */ + 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 SplitStreamProvider || !$projection instanceof SubQueryProvider) { + continue; + } + + $splitEvents = $projection->splitEvents(); + + if ($splitEvents === []) { + continue; + } + + $subQuery = $projection->subQuery(); + + $lookups[$name] = new SubQuery( + $subQuery->tags, + $splitEvents, + $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; + } +} diff --git a/src/Projection/SplitStreamProvider.php b/src/Projection/SplitStreamProvider.php new file mode 100644 index 000000000..6034b4533 --- /dev/null +++ b/src/Projection/SplitStreamProvider.php @@ -0,0 +1,17 @@ + + */ + public function splitEvents(): array; +} diff --git a/src/Projection/StoreProjectionBuilder.php b/src/Projection/StoreProjectionBuilder.php index 852ee0d08..da7f01da3 100644 --- a/src/Projection/StoreProjectionBuilder.php +++ b/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 __construct( public function build( array $projections, ): array { - $projection = new CompositeProjection($projections); + $from = SplitStreamPositions::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(); diff --git a/tests/Unit/DecisionModel/StoreDecisionModelBuilderTest.php b/tests/Unit/DecisionModel/StoreDecisionModelBuilderTest.php index 5de35fa13..e03567e1e 100644 --- a/tests/Unit/DecisionModel/StoreDecisionModelBuilderTest.php +++ b/tests/Unit/DecisionModel/StoreDecisionModelBuilderTest.php @@ -4,11 +4,14 @@ namespace Patchlevel\EventSourcing\Tests\Unit\DecisionModel; +use Patchlevel\EventSourcing\Attribute\Apply; use Patchlevel\EventSourcing\DecisionModel\StoreDecisionModelBuilder; use Patchlevel\EventSourcing\Message\Message; use Patchlevel\EventSourcing\Message\Stream; +use Patchlevel\EventSourcing\Projection\BasicProjection; use Patchlevel\EventSourcing\Store\AppendCondition; use Patchlevel\EventSourcing\Store\AppendStore; +use Patchlevel\EventSourcing\Store\Header\IndexHeader; use Patchlevel\EventSourcing\Store\Header\TagsHeader; use Patchlevel\EventSourcing\Store\InMemoryStore; use Patchlevel\EventSourcing\Store\Query; @@ -18,6 +21,9 @@ use Patchlevel\EventSourcing\Tests\Unit\Fixture\LastEmailProjection; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\SplittingEvent; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\VisitsProjection; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\TestCase; @@ -115,4 +121,165 @@ public function testEmptyStream(): void self::assertEquals(new AppendCondition($expectedQuery, 0), $state->appendCondition); } + + public function testSplitStream(): void + { + $store = new InMemoryStore(); + $store->save( + $this->visited('1', ['profile-1']), + $this->visited('1', ['profile-1']), + $this->split(10, ['profile-1']), + $this->visited('1', ['profile-1']), + $this->split(99, ['profile-2']), + $this->visited('1', ['profile-1']), + ); + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'visits' => new VisitsProjection(['profile-1']), + ]); + + self::assertSame(['visits' => 12], $state->state); + self::assertSame(6, $state->appendCondition->after); + } + + public function testSplitStreamReadsFromLastSplitEvent(): void + { + $splitMessage = $this->split(10, ['profile-1'])->withHeader(new IndexHeader(3)); + + $expectedQuery = new Query( + new SubQuery(['profile-1'], [ProfileVisited::class, SplittingEvent::class]), + ); + + $store = new class ($splitMessage) implements AppendStore { + /** @var list */ + public array $calls = []; + + public function __construct( + private readonly Message $splitMessage, + ) { + } + + /** @param iterable $messages */ + public function append(iterable $messages, AppendCondition|null $appendCondition = null): void + { + } + + public function query(Query $query, int $from = 0): Stream + { + $this->calls[] = [$query, $from]; + + return new Stream([3 => $this->splitMessage]); + } + }; + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'visits' => new VisitsProjection(['profile-1']), + ]); + + self::assertSame(['visits' => 10], $state->state); + self::assertEquals(new AppendCondition($expectedQuery, 3), $state->appendCondition); + self::assertEquals( + [ + [new Query(new SubQuery(['profile-1'], [SplittingEvent::class], null, true)), 0], + [$expectedQuery, 3], + ], + $store->calls, + ); + } + + public function testSplitStreamWithoutSplitEvent(): void + { + $store = new InMemoryStore(); + $store->save( + $this->visited('1', ['profile-1']), + $this->visited('1', ['profile-1']), + ); + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'visits' => new VisitsProjection(['profile-1']), + ]); + + self::assertSame(['visits' => 2], $state->state); + } + + public function testSplitStreamWithProjectionWithoutSplitEvents(): void + { + $store = new InMemoryStore(); + $store->save( + Message::create(new ProfileCreated(ProfileId::fromString('1'), Email::fromString('info@patchlevel.de'))) + ->withHeader(new TagsHeader(['profile-1'])), + $this->visited('1', ['profile-1']), + $this->split(10, ['profile-1']), + $this->visited('1', ['profile-1']), + ); + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'created' => new IncrementProjection(0, ['profile-1']), + 'visits' => new VisitsProjection(['profile-1']), + ]); + + self::assertSame(['created' => 1, 'visits' => 11], $state->state); + } + + public function testSplitStreamOptOut(): void + { + $store = new InMemoryStore(); + $store->save( + $this->split(10, ['profile-1']), + $this->split(20, ['profile-1']), + ); + + $builder = new StoreDecisionModelBuilder($store); + + $state = $builder->build([ + 'splits' => new class extends BasicProjection { + public function initialState(): int + { + return 0; + } + + #[Apply] + public function applySplittingEvent(int $state, SplittingEvent $event): int + { + return $state + 1; + } + + /** @return list */ + public function splitEvents(): array + { + return []; + } + + /** @return list */ + protected function tagFilter(): array + { + return ['profile-1']; + } + }, + ]); + + self::assertSame(['splits' => 2], $state->state); + } + + /** @param list $tags */ + private function visited(string $id, array $tags): Message + { + return Message::create(new ProfileVisited(ProfileId::fromString($id))) + ->withHeader(new TagsHeader($tags)); + } + + /** @param list $tags */ + private function split(int $visits, array $tags): Message + { + return Message::create(new SplittingEvent(Email::fromString('info@patchlevel.de'), $visits)) + ->withHeader(new TagsHeader($tags)); + } } diff --git a/tests/Unit/Fixture/VisitsProjection.php b/tests/Unit/Fixture/VisitsProjection.php new file mode 100644 index 000000000..eefc02003 --- /dev/null +++ b/tests/Unit/Fixture/VisitsProjection.php @@ -0,0 +1,40 @@ + $tags */ + public function __construct( + private readonly array $tags, + ) { + } + + public function initialState(): int + { + return 0; + } + + #[Apply] + public function applyProfileVisited(int $state, ProfileVisited $event): int + { + return $state + 1; + } + + #[Apply] + public function applySplittingEvent(int $state, SplittingEvent $event): int + { + return $event->visits; + } + + /** @return list */ + protected function tagFilter(): array + { + return $this->tags; + } +} diff --git a/tests/Unit/Projection/BasicProjectionTest.php b/tests/Unit/Projection/BasicProjectionTest.php index 12e41ca46..c2951494f 100644 --- a/tests/Unit/Projection/BasicProjectionTest.php +++ b/tests/Unit/Projection/BasicProjectionTest.php @@ -12,9 +12,12 @@ use Patchlevel\EventSourcing\Store\Header\TagsHeader; use Patchlevel\EventSourcing\Store\SubQuery; use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\IncrementProjection; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\SplittingEvent; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\VisitsProjection; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\TestCase; use Stringable; @@ -53,6 +56,20 @@ public function applyProfileCreated(int $state, ProfileCreated $event): int self::assertSame($projection->subQuery(), $projection->subQuery()); } + public function testSplitEvents(): void + { + $projection = new VisitsProjection([]); + + self::assertSame([SplittingEvent::class], $projection->splitEvents()); + } + + public function testNoSplitEvents(): void + { + $projection = new IncrementProjection(0); + + self::assertSame([], $projection->splitEvents()); + } + public function testApplyWithNonMatchingMessage(): void { $projection = new class extends BasicProjection { diff --git a/tests/Unit/Projection/CompositeProjectionTest.php b/tests/Unit/Projection/CompositeProjectionTest.php index 4d965d3f0..c5d83fd05 100644 --- a/tests/Unit/Projection/CompositeProjectionTest.php +++ b/tests/Unit/Projection/CompositeProjectionTest.php @@ -7,6 +7,7 @@ use Patchlevel\EventSourcing\Message\Message; use Patchlevel\EventSourcing\Projection\CompositeProjection; use Patchlevel\EventSourcing\Projection\Projection; +use Patchlevel\EventSourcing\Store\Header\IndexHeader; use Patchlevel\EventSourcing\Store\Header\StreamNameHeader; use Patchlevel\EventSourcing\Store\Header\TagsHeader; use Patchlevel\EventSourcing\Store\Query; @@ -77,6 +78,29 @@ public function testApplyDelegatesToEachProjection(): void self::assertSame(11, $newState['y']); } + public function testApplySkipsMessagesBeforeFrom(): void + { + $composite = new CompositeProjection( + [ + 'x' => new IncrementProjection(0, ['match']), + 'y' => new IncrementProjection(0, ['match']), + ], + ['y' => 2], + ); + + $message = Message::create(new ProfileCreated( + ProfileId::fromString('test'), + Email::fromString('foo@example.com'), + )) + ->withHeader(new TagsHeader(['match'])); + + $state = $composite->initialState(); + $state = $composite->apply($state, $message->withHeader(new IndexHeader(1))); + $state = $composite->apply($state, $message->withHeader(new IndexHeader(2))); + + self::assertSame(['x' => 2, 'y' => 1], $state); + } + public function testApplySkipsNonMatchingProjection(): void { $composite = new CompositeProjection([ diff --git a/tests/Unit/Projection/StoreProjectionBuilderTest.php b/tests/Unit/Projection/StoreProjectionBuilderTest.php index 509e41465..22f90b8c5 100644 --- a/tests/Unit/Projection/StoreProjectionBuilderTest.php +++ b/tests/Unit/Projection/StoreProjectionBuilderTest.php @@ -10,10 +10,14 @@ use Patchlevel\EventSourcing\Projection\StoreProjectionBuilder; use Patchlevel\EventSourcing\Store\AppendStore; use Patchlevel\EventSourcing\Store\Header\TagsHeader; +use Patchlevel\EventSourcing\Store\InMemoryStore; use Patchlevel\EventSourcing\Tests\Unit\Fixture\Email; use Patchlevel\EventSourcing\Tests\Unit\Fixture\IncrementProjection; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileCreated; use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileId; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\ProfileVisited; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\SplittingEvent; +use Patchlevel\EventSourcing\Tests\Unit\Fixture\VisitsProjection; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\TestCase; @@ -42,4 +46,19 @@ public function testBuild(): void self::assertSame(['count' => 2], $builder->build($projections)); } + + public function testBuildWithSplitStream(): void + { + $store = new InMemoryStore(); + $store->save( + Message::create(new ProfileVisited(ProfileId::fromString('1')))->withHeader(new TagsHeader(['match'])), + Message::create(new SplittingEvent(Email::fromString('foo@bar.com'), 5)) + ->withHeader(new TagsHeader(['match'])), + Message::create(new ProfileVisited(ProfileId::fromString('1')))->withHeader(new TagsHeader(['match'])), + ); + + $builder = new StoreProjectionBuilder($store); + + self::assertSame(['visits' => 6], $builder->build(['visits' => new VisitsProjection(['match'])])); + } }