From 0f19dab5e048e49eebbdbab2c3043990ef952cd8 Mon Sep 17 00:00:00 2001
From: Robertbaelde <robert_baelde@hotmail.com>
Date: Wed, 31 May 2023 14:33:37 +0200
Subject: [PATCH 1/5] Projection Handler initial work

---
 src/Projections/CantLockProjection.php        | 11 ++++
 src/Projections/DefaultProjectionHandler.php  | 22 +++++++
 .../InMemoryProjectionStatusRepository.php    | 38 +++++++++++
 src/Projections/ProjectionHandler.php         | 34 ++++++++++
 src/Projections/ProjectionHandlerTest.php     | 65 +++++++++++++++++++
 src/Projections/ProjectionId.php              | 20 ++++++
 .../ProjectionStatusRepository.php            | 17 +++++
 7 files changed, 207 insertions(+)
 create mode 100644 src/Projections/CantLockProjection.php
 create mode 100644 src/Projections/DefaultProjectionHandler.php
 create mode 100644 src/Projections/InMemoryProjectionStatusRepository.php
 create mode 100644 src/Projections/ProjectionHandler.php
 create mode 100644 src/Projections/ProjectionHandlerTest.php
 create mode 100644 src/Projections/ProjectionId.php
 create mode 100644 src/Projections/ProjectionStatusRepository.php

diff --git a/src/Projections/CantLockProjection.php b/src/Projections/CantLockProjection.php
new file mode 100644
index 00000000..d15b6a29
--- /dev/null
+++ b/src/Projections/CantLockProjection.php
@@ -0,0 +1,11 @@
+<?php
+
+namespace EventSauce\EventSourcing\Projections;
+
+class CantLockProjection extends \Exception
+{
+    public static function becauseItIsAlreadyLocked(string $projectionId): self
+    {
+        return new self("Projection $projectionId is already locked");
+    }
+}
diff --git a/src/Projections/DefaultProjectionHandler.php b/src/Projections/DefaultProjectionHandler.php
new file mode 100644
index 00000000..4f95e914
--- /dev/null
+++ b/src/Projections/DefaultProjectionHandler.php
@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\PaginationCursor;
+
+class DefaultProjectionHandler
+{
+    public function __construct(
+        private ProjectionHandler $projectionHandler,
+        private ProjectionId $projectionId,
+        private PaginationCursor $initialCursor,
+    ) {
+    }
+
+    public function handle(): void
+    {
+        $this->projectionHandler->handle($this->projectionId, $this->initialCursor);
+    }
+}
diff --git a/src/Projections/InMemoryProjectionStatusRepository.php b/src/Projections/InMemoryProjectionStatusRepository.php
new file mode 100644
index 00000000..72b6fcca
--- /dev/null
+++ b/src/Projections/InMemoryProjectionStatusRepository.php
@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\PaginationCursor;
+
+class InMemoryProjectionStatusRepository implements ProjectionStatusRepository
+{
+    private array $state = [];
+    private array $locks = [];
+
+    public function getCursor(ProjectionId $projectionId): ?PaginationCursor
+    {
+        return $this->state[$projectionId->toString()] ?? null;
+    }
+
+    /**
+     * @throws CantLockProjection
+     */
+    public function getCursorAndLock(ProjectionId $projectionId): ?PaginationCursor
+    {
+        if (array_key_exists($projectionId->toString(), $this->locks)) {
+            throw CantLockProjection::becauseItIsAlreadyLocked($projectionId->toString());
+        }
+
+        $this->locks[$projectionId->toString()] = true;
+
+        return $this->state[$projectionId->toString()] ?? null;
+    }
+
+    public function persistCursorAndRelease(ProjectionId $projectionId, PaginationCursor $cursor): void
+    {
+        unset($this->locks[$projectionId->toString()]);
+        $this->state[$projectionId->toString()] = $cursor;
+    }
+}
diff --git a/src/Projections/ProjectionHandler.php b/src/Projections/ProjectionHandler.php
new file mode 100644
index 00000000..888cb733
--- /dev/null
+++ b/src/Projections/ProjectionHandler.php
@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\Message;
+use EventSauce\EventSourcing\MessageConsumer;
+use EventSauce\EventSourcing\MessageRepository;
+use EventSauce\EventSourcing\PaginationCursor;
+
+class ProjectionHandler
+{
+    public function __construct(
+        private MessageRepository $repository,
+        private MessageConsumer $consumer,
+        private ProjectionStatusRepository $projectionStatusRepository,
+    ) {
+    }
+
+    public function handle(ProjectionId $projectionId, PaginationCursor $initialCursor): void
+    {
+        $cursor = $this->projectionStatusRepository->getCursorAndLock($projectionId) ?? $initialCursor;
+
+        $messages = $this->repository->paginate($cursor);
+
+        /** @var Message $message */
+        foreach ($messages as $message) {
+            $this->consumer->handle($message);
+        }
+
+        $this->projectionStatusRepository->persistCursorAndRelease($projectionId, $messages->getReturn());
+    }
+}
diff --git a/src/Projections/ProjectionHandlerTest.php b/src/Projections/ProjectionHandlerTest.php
new file mode 100644
index 00000000..9c34aa96
--- /dev/null
+++ b/src/Projections/ProjectionHandlerTest.php
@@ -0,0 +1,65 @@
+<?php
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\CollectingMessageConsumer;
+use EventSauce\EventSourcing\EventStub;
+use EventSauce\EventSourcing\InMemoryMessageRepository;
+use EventSauce\EventSourcing\Message;
+use EventSauce\EventSourcing\OffsetCursor;
+use PHPUnit\Framework\TestCase;
+
+class ProjectionHandlerTest extends TestCase
+{
+    /** @test */
+    public function it_uses_initial_cursor_when_no_projection_state_exists()
+    {
+        $repository = new InMemoryMessageRepository();
+        $repository->persist(
+            new Message(new EventStub('1')),
+        );
+
+        $consumer = new CollectingMessageConsumer();
+
+        $projectionStatusRepository = new InMemoryProjectionStatusRepository();
+
+        $projectionHandler = new ProjectionHandler(
+            repository: $repository,
+            consumer: $consumer,
+            projectionStatusRepository: $projectionStatusRepository,
+        );
+
+        $projectionHandler->handle($this->getProjectionId(), OffsetCursor::fromStart());
+
+        $this->assertCount(1, $consumer->collectedMessages());
+        $this->assertEquals(OffsetCursor::fromStart()->plusOffset(1), $projectionStatusRepository->getCursor($this->getProjectionId()));
+    }
+
+    /** @test */
+    public function it_cant_start_a_new_play_while_projection_is_playing()
+    {
+        $repository = new InMemoryMessageRepository();
+        $repository->persist(
+            new Message(new EventStub('1')),
+        );
+
+        $consumer = new CollectingMessageConsumer();
+
+        $projectionStatusRepository = new InMemoryProjectionStatusRepository();
+        $projectionStatusRepository->getCursorAndLock($this->getProjectionId());
+
+        $projectionHandler = new ProjectionHandler(
+            repository: $repository,
+            consumer: $consumer,
+            projectionStatusRepository: $projectionStatusRepository,
+        );
+
+        $this->expectExceptionObject(CantLockProjection::becauseItIsAlreadyLocked('test'));
+        $projectionHandler->handle($this->getProjectionId(), OffsetCursor::fromStart());
+    }
+
+    private function getProjectionId(): ProjectionId
+    {
+        return ProjectionId::fromString('test');
+    }
+}
diff --git a/src/Projections/ProjectionId.php b/src/Projections/ProjectionId.php
new file mode 100644
index 00000000..70a181ed
--- /dev/null
+++ b/src/Projections/ProjectionId.php
@@ -0,0 +1,20 @@
+<?php
+
+namespace EventSauce\EventSourcing\Projections;
+
+class ProjectionId
+{
+    private function __construct(private string $id)
+    {
+
+    }
+    public function toString(): string
+    {
+        return $this->id;
+    }
+
+    public static function fromString(string $string): self
+    {
+        return new self($string);
+    }
+}
diff --git a/src/Projections/ProjectionStatusRepository.php b/src/Projections/ProjectionStatusRepository.php
new file mode 100644
index 00000000..2cd1b259
--- /dev/null
+++ b/src/Projections/ProjectionStatusRepository.php
@@ -0,0 +1,17 @@
+<?php
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\PaginationCursor;
+
+interface ProjectionStatusRepository
+{
+    public function getCursor(ProjectionId $projectionId): ?PaginationCursor;
+
+    /**
+     * @throws CantLockProjection
+     */
+    public function getCursorAndLock(ProjectionId $projectionId): ?PaginationCursor;
+
+    public function persistCursorAndRelease(ProjectionId $projectionId, PaginationCursor $cursor): void;
+}

From 7197d60c855e065fea554b39ee40e9cd7278436a Mon Sep 17 00:00:00 2001
From: Robertbaelde <robert_baelde@hotmail.com>
Date: Wed, 31 May 2023 16:06:44 +0200
Subject: [PATCH 2/5] Introduce subscriptions

---
 src/Subscriptions/AggregateCheckpoint.php     | 30 ++++++++++++++++
 ...egateRootIdVersionSubscriptionProvider.php | 33 +++++++++++++++++
 ...eRootIdVersionSubscriptionProviderTest.php | 35 +++++++++++++++++++
 src/Subscriptions/Checkpoint.php              |  7 ++++
 src/Subscriptions/OffsetCheckpoint.php        | 31 ++++++++++++++++
 .../OffsetStreamSubscriptionProvider.php      | 30 ++++++++++++++++
 .../OffsetStreamSubscriptionProviderTest.php  | 31 ++++++++++++++++
 src/Subscriptions/SubscriptionProvider.php    | 14 ++++++++
 8 files changed, 211 insertions(+)
 create mode 100644 src/Subscriptions/AggregateCheckpoint.php
 create mode 100644 src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
 create mode 100644 src/Subscriptions/AggregateRootIdVersionSubscriptionProviderTest.php
 create mode 100644 src/Subscriptions/Checkpoint.php
 create mode 100644 src/Subscriptions/OffsetCheckpoint.php
 create mode 100644 src/Subscriptions/OffsetStreamSubscriptionProvider.php
 create mode 100644 src/Subscriptions/OffsetStreamSubscriptionProviderTest.php
 create mode 100644 src/Subscriptions/SubscriptionProvider.php

diff --git a/src/Subscriptions/AggregateCheckpoint.php b/src/Subscriptions/AggregateCheckpoint.php
new file mode 100644
index 00000000..0b29d680
--- /dev/null
+++ b/src/Subscriptions/AggregateCheckpoint.php
@@ -0,0 +1,30 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\AggregateRootId;
+
+class AggregateCheckpoint implements Checkpoint
+{
+    private function __construct(
+        private AggregateRootId $aggregateRootId,
+        private int $aggregateRootVersion,
+    ) {
+    }
+
+    public static function forAggregateRootId(AggregateRootId $aggregateRootId, $version = 0): static
+    {
+        return new static($aggregateRootId, $version);
+    }
+
+    public function getAggregateRootId(): AggregateRootId
+    {
+        return $this->aggregateRootId;
+    }
+
+    public function getVersion(): int
+    {
+        return $this->aggregateRootVersion;
+    }
+
+}
diff --git a/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php b/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
new file mode 100644
index 00000000..731efab5
--- /dev/null
+++ b/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
@@ -0,0 +1,33 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\DummyAggregateRootId;
+use EventSauce\EventSourcing\EventStub;
+use EventSauce\EventSourcing\Header;
+use EventSauce\EventSourcing\InMemoryMessageRepository;
+use EventSauce\EventSourcing\Message;
+use EventSauce\EventSourcing\MessageRepository;
+use Generator;
+
+class AggregateRootIdVersionSubscriptionProvider implements SubscriptionProvider
+{
+
+    public function __construct(
+        private MessageRepository $messageRepository)
+    {
+    }
+
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint, int $maxEvents = 100): Generator
+    {
+        if(!$checkpoint instanceof AggregateCheckpoint){
+            throw new \InvalidArgumentException('Checkpoint must be an instance of AggregateCheckpoint');
+        }
+
+        $messages = $this->messageRepository->retrieveAllAfterVersion($checkpoint->getAggregateRootId(), $checkpoint->getVersion());
+
+        yield from $messages;
+
+        return AggregateCheckpoint::forAggregateRootId($checkpoint->getAggregateRootId(), $messages->getReturn() ?? $checkpoint->getVersion());
+    }
+}
diff --git a/src/Subscriptions/AggregateRootIdVersionSubscriptionProviderTest.php b/src/Subscriptions/AggregateRootIdVersionSubscriptionProviderTest.php
new file mode 100644
index 00000000..d976e5fc
--- /dev/null
+++ b/src/Subscriptions/AggregateRootIdVersionSubscriptionProviderTest.php
@@ -0,0 +1,35 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\DummyAggregateRootId;
+use EventSauce\EventSourcing\EventStub;
+use EventSauce\EventSourcing\Header;
+use EventSauce\EventSourcing\InMemoryMessageRepository;
+use EventSauce\EventSourcing\Message;
+use PHPUnit\Framework\TestCase;
+
+class AggregateRootIdVersionSubscriptionProviderTest extends TestCase
+{
+    /** @test */
+    public function it_gets_aggregate_root_specific_events()
+    {
+        $idA = DummyAggregateRootId::generate();
+        $idB = DummyAggregateRootId::generate();
+
+        $messageRepository = new InMemoryMessageRepository();
+        $messageRepository->persist(
+            (new Message(new EventStub('1')))->withHeader(Header::AGGREGATE_ROOT_ID, $idA)->withHeader(Header::AGGREGATE_ROOT_VERSION, 1),
+            (new Message(new EventStub('2')))->withHeader(Header::AGGREGATE_ROOT_ID, $idB)->withHeader(Header::AGGREGATE_ROOT_VERSION, 1),
+        );
+
+        $provider = new AggregateRootIdVersionSubscriptionProvider($messageRepository);
+        $messages = $provider->getEventsSinceCheckpoint(AggregateCheckpoint::forAggregateRootId($idA));
+
+        $actualMessages = iterator_to_array($messages, false);
+        $this->assertCount(1, $actualMessages);
+        $checkpoint = $messages->getReturn();
+        $this->assertInstanceOf(AggregateCheckpoint::class, $checkpoint);
+        $this->assertEquals(1, $checkpoint->getVersion());
+    }
+}
diff --git a/src/Subscriptions/Checkpoint.php b/src/Subscriptions/Checkpoint.php
new file mode 100644
index 00000000..9a72f5b0
--- /dev/null
+++ b/src/Subscriptions/Checkpoint.php
@@ -0,0 +1,7 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+interface Checkpoint
+{
+}
diff --git a/src/Subscriptions/OffsetCheckpoint.php b/src/Subscriptions/OffsetCheckpoint.php
new file mode 100644
index 00000000..ebdb9521
--- /dev/null
+++ b/src/Subscriptions/OffsetCheckpoint.php
@@ -0,0 +1,31 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+class OffsetCheckpoint implements Checkpoint
+{
+    private function __construct(
+        private int $offset,
+    ) {
+    }
+
+    public static function forOffset(int $offset): static
+    {
+        return new static($offset);
+    }
+
+    public function getOffset(): int
+    {
+        return $this->offset;
+    }
+
+    public function toString(): string
+    {
+        return (string) $this->offset;
+    }
+
+    public static function fromString(string $string): static
+    {
+        return new static((int) $string);
+    }
+}
diff --git a/src/Subscriptions/OffsetStreamSubscriptionProvider.php b/src/Subscriptions/OffsetStreamSubscriptionProvider.php
new file mode 100644
index 00000000..784f695d
--- /dev/null
+++ b/src/Subscriptions/OffsetStreamSubscriptionProvider.php
@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\MessageRepository;
+use EventSauce\EventSourcing\OffsetCursor;
+
+class OffsetStreamSubscriptionProvider implements SubscriptionProvider
+{
+    public function __construct(
+        private MessageRepository $messageRepository,
+    ) {
+    }
+
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint, int $maxEvents = 100): \Generator
+    {
+        if ( ! $checkpoint instanceof OffsetCheckpoint) {
+            throw new \Exception('Invalid checkpoint type');
+        }
+
+        $cursor = OffsetCursor::fromOffset($checkpoint->getOffset(), $maxEvents);
+        $messages = $this->messageRepository->paginate($cursor);
+
+        yield from $messages;
+
+        return OffsetCheckpoint::forOffset($cursor->offset());
+    }
+}
diff --git a/src/Subscriptions/OffsetStreamSubscriptionProviderTest.php b/src/Subscriptions/OffsetStreamSubscriptionProviderTest.php
new file mode 100644
index 00000000..e774a702
--- /dev/null
+++ b/src/Subscriptions/OffsetStreamSubscriptionProviderTest.php
@@ -0,0 +1,31 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\EventStub;
+use EventSauce\EventSourcing\InMemoryMessageRepository;
+use EventSauce\EventSourcing\Message;
+use PHPUnit\Framework\TestCase;
+
+class OffsetStreamSubscriptionProviderTest extends TestCase
+{
+    /** @test */
+    public function it_gets_the_events_from_the_repository_since_origin()
+    {
+        $messageRepository = new InMemoryMessageRepository();
+        $messageRepository->persist(
+            new Message(new EventStub('1')),
+            new Message(new EventStub('2')),
+        );
+
+        $provider = new OffsetStreamSubscriptionProvider($messageRepository);
+        $messages = $provider->getEventsSinceCheckpoint(OffsetCheckpoint::forOffset(0), 100);
+
+        $actualMessages = iterator_to_array($messages, false);
+        $this->assertCount(2, $actualMessages);
+        $checkpoint = $messages->getReturn();
+        $this->assertInstanceOf(OffsetCheckpoint::class, $checkpoint);
+    }
+
+
+}
diff --git a/src/Subscriptions/SubscriptionProvider.php b/src/Subscriptions/SubscriptionProvider.php
new file mode 100644
index 00000000..40f508e3
--- /dev/null
+++ b/src/Subscriptions/SubscriptionProvider.php
@@ -0,0 +1,14 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\Message;
+use Generator;
+
+interface SubscriptionProvider
+{
+    /**
+     * @return Generator<int, Message, void, Checkpoint>
+     */
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint, int $maxEvents = 100): Generator;
+}

From 3c5e37fb134b851df088ddb76a893a6b59806758 Mon Sep 17 00:00:00 2001
From: Robertbaelde <robert_baelde@hotmail.com>
Date: Wed, 31 May 2023 16:26:32 +0200
Subject: [PATCH 3/5] Add partitioned subscriber

---
 .../AggregateRootIdPartitioner.php            | 30 ++++++++++++
 .../AggregateRootPartitionedOffsetCursor.php  | 48 +++++++++++++++++++
 ...gregateRootPartitionedOffsetCursorTest.php | 19 ++++++++
 .../PartitionedProjectionHandler.php          | 24 ++++++++++
 src/Projections/Partitioner.php               | 10 ++++
 src/Subscriptions/HeaderPartitioner.php       | 18 +++++++
 src/Subscriptions/PartitionedCheckpoint.php   | 34 +++++++++++++
 .../PartitionedSubscriptionProvider.php       | 41 ++++++++++++++++
 .../PartitionedSubscriptionProviderTest.php   | 36 ++++++++++++++
 src/Subscriptions/Partitioner.php             | 10 ++++
 10 files changed, 270 insertions(+)
 create mode 100644 src/Projections/AggregateRootIdPartitioner.php
 create mode 100644 src/Projections/AggregateRootPartitionedOffsetCursor.php
 create mode 100644 src/Projections/AggregateRootPartitionedOffsetCursorTest.php
 create mode 100644 src/Projections/PartitionedProjectionHandler.php
 create mode 100644 src/Projections/Partitioner.php
 create mode 100644 src/Subscriptions/HeaderPartitioner.php
 create mode 100644 src/Subscriptions/PartitionedCheckpoint.php
 create mode 100644 src/Subscriptions/PartitionedSubscriptionProvider.php
 create mode 100644 src/Subscriptions/PartitionedSubscriptionProviderTest.php
 create mode 100644 src/Subscriptions/Partitioner.php

diff --git a/src/Projections/AggregateRootIdPartitioner.php b/src/Projections/AggregateRootIdPartitioner.php
new file mode 100644
index 00000000..a69f3dbb
--- /dev/null
+++ b/src/Projections/AggregateRootIdPartitioner.php
@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\ClassNameInflector;
+use EventSauce\EventSourcing\DotSeparatedSnakeCaseInflector;
+use EventSauce\EventSourcing\Message;
+
+class AggregateRootIdPartitioner implements Partitioner
+{
+    private ClassNameInflector $classNameInflector;
+
+    public function __construct(
+        ?ClassNameInflector $classNameInflector = null,
+    ) {
+        $this->classNameInflector = $classNameInflector ?: new DotSeparatedSnakeCaseInflector();
+    }
+
+    public function getPartitionKey(Message $message): string
+    {
+        $aggregateRootId = $message->aggregateRootId();
+        if ($aggregateRootId === null) {
+            return 'no_id';
+        }
+
+        return $this->classNameInflector->instanceToType($aggregateRootId) . '_' . $aggregateRootId->toString();
+    }
+}
diff --git a/src/Projections/AggregateRootPartitionedOffsetCursor.php b/src/Projections/AggregateRootPartitionedOffsetCursor.php
new file mode 100644
index 00000000..67102e48
--- /dev/null
+++ b/src/Projections/AggregateRootPartitionedOffsetCursor.php
@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\AggregateRootId;
+use EventSauce\EventSourcing\PaginationCursor;
+
+class AggregateRootPartitionedOffsetCursor implements PaginationCursor
+{
+    private function __construct(
+        private AggregateRootId $aggregateRootId,
+        private int $offset
+    ) {
+    }
+
+    public static function fromStart(AggregateRootId $aggregateRootId): static
+    {
+        return new static(
+            aggregateRootId: $aggregateRootId,
+            offset: 0
+        );
+    }
+
+    public function toString(): string
+    {
+        return implode('###', [
+            get_class($this->aggregateRootId),
+            $this->aggregateRootId->toString(),
+            (string) $this->offset,
+        ]);
+    }
+
+    public static function fromString(string $cursor): static
+    {
+        $cursor = explode('###', $cursor);
+        return new static(
+            aggregateRootId: new $cursor[0]($cursor[1]),
+            offset: (int) $cursor[2]
+        );
+    }
+
+    public function isAtStart(): bool
+    {
+        return $this->offset === 0;
+    }
+}
diff --git a/src/Projections/AggregateRootPartitionedOffsetCursorTest.php b/src/Projections/AggregateRootPartitionedOffsetCursorTest.php
new file mode 100644
index 00000000..55940363
--- /dev/null
+++ b/src/Projections/AggregateRootPartitionedOffsetCursorTest.php
@@ -0,0 +1,19 @@
+<?php
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\DummyAggregateRootId;
+use PHPUnit\Framework\TestCase;
+
+class AggregateRootPartitionedOffsetCursorTest extends TestCase
+{
+    /** @test */
+    public function it_can_construct_a_aggregate_root_partitioned_offset_cursor_from_start()
+    {
+        $id = DummyAggregateRootId::generate();
+        $cursor = AggregateRootPartitionedOffsetCursor::fromStart($id);
+
+        $string = $cursor->toString();
+        $this->assertEquals($cursor, AggregateRootPartitionedOffsetCursor::fromString($string));
+    }
+}
diff --git a/src/Projections/PartitionedProjectionHandler.php b/src/Projections/PartitionedProjectionHandler.php
new file mode 100644
index 00000000..04b87fac
--- /dev/null
+++ b/src/Projections/PartitionedProjectionHandler.php
@@ -0,0 +1,24 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\Message;
+use EventSauce\EventSourcing\PaginationCursor;
+
+class PartitionedProjectionHandler
+{
+    public function __construct(
+        private ProjectionHandler $projectionHandler,
+        private ProjectionId $projectionId,
+        private Partitioner $partitioner,
+        private PaginationCursor $initialCursor,
+    ) {
+    }
+
+    public function handle(Message $message): void
+    {
+        $this->projectionHandler->handle($this->projectionId, $this->initialCursor);
+    }
+}
diff --git a/src/Projections/Partitioner.php b/src/Projections/Partitioner.php
new file mode 100644
index 00000000..515f59d6
--- /dev/null
+++ b/src/Projections/Partitioner.php
@@ -0,0 +1,10 @@
+<?php
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\Message;
+
+interface Partitioner
+{
+    public function getPartitionKey(Message $message): string;
+}
diff --git a/src/Subscriptions/HeaderPartitioner.php b/src/Subscriptions/HeaderPartitioner.php
new file mode 100644
index 00000000..c5d2c494
--- /dev/null
+++ b/src/Subscriptions/HeaderPartitioner.php
@@ -0,0 +1,18 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\Message;
+
+class HeaderPartitioner implements Partitioner
+{
+    public function __construct(
+        private string $headerName,
+    ) {
+    }
+
+    public function getPartitionKey(Message $message): ?string
+    {
+        return $message->header($this->headerName);
+    }
+}
diff --git a/src/Subscriptions/PartitionedCheckpoint.php b/src/Subscriptions/PartitionedCheckpoint.php
new file mode 100644
index 00000000..c7e4ced9
--- /dev/null
+++ b/src/Subscriptions/PartitionedCheckpoint.php
@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+class PartitionedCheckpoint implements Checkpoint
+{
+    private function __construct(
+        private string $partitionKey,
+        private int $offset,
+    ) {
+    }
+
+    public static function fromOrigin(string $partitionKey): static
+    {
+        return new PartitionedCheckpoint($partitionKey, 0);
+    }
+
+    public function withOffset(int $offset): static
+    {
+        return new static($this->partitionKey, $offset);
+    }
+
+    public function getOffset(): int
+    {
+        return $this->offset;
+    }
+
+    public function getPartitionKey(): string
+    {
+        return $this->partitionKey;
+    }
+}
diff --git a/src/Subscriptions/PartitionedSubscriptionProvider.php b/src/Subscriptions/PartitionedSubscriptionProvider.php
new file mode 100644
index 00000000..a1f3fa6a
--- /dev/null
+++ b/src/Subscriptions/PartitionedSubscriptionProvider.php
@@ -0,0 +1,41 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\AntiCorruptionLayer\MessageFilter;
+use EventSauce\EventSourcing\Message;
+use EventSauce\EventSourcing\MessageRepository;
+use EventSauce\EventSourcing\OffsetCursor;
+
+class PartitionedSubscriptionProvider implements SubscriptionProvider
+{
+    public function __construct(
+        private MessageRepository $messageRepository,
+        private Partitioner $partitioner,
+    ) {
+    }
+
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint, int $maxEvents = 100): \Generator
+    {
+        if ( ! $checkpoint instanceof PartitionedCheckpoint) {
+            throw new \InvalidArgumentException('Checkpoint must be an instance of PartitionedCheckpoint');
+        }
+
+        $cursor = OffsetCursor::fromOffset($checkpoint->getOffset(), $maxEvents);
+
+        $messages = $this->messageRepository->paginate($cursor);
+
+        /** @var Message $message */
+        foreach ($messages as $message) {
+            $partitionKey = $this->partitioner->getPartitionKey($message);
+            if ($partitionKey !== $checkpoint->getPartitionKey()) {
+                continue;
+            }
+            yield $message;
+        }
+
+        return $checkpoint->withOffset($messages->getReturn()->offset());
+    }
+}
diff --git a/src/Subscriptions/PartitionedSubscriptionProviderTest.php b/src/Subscriptions/PartitionedSubscriptionProviderTest.php
new file mode 100644
index 00000000..b6141aa8
--- /dev/null
+++ b/src/Subscriptions/PartitionedSubscriptionProviderTest.php
@@ -0,0 +1,36 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\DummyAggregateRootId;
+use EventSauce\EventSourcing\EventStub;
+use EventSauce\EventSourcing\Header;
+use EventSauce\EventSourcing\InMemoryMessageRepository;
+use EventSauce\EventSourcing\Message;
+use PHPUnit\Framework\TestCase;
+
+class PartitionedSubscriptionProviderTest extends TestCase
+{
+    /** @test */
+    public function it_can_get_a_partition_of_a_stream()
+    {
+
+        $messageRepository = new InMemoryMessageRepository();
+        $messageRepository->persist(
+            (new Message(new EventStub('1')))->withHeader('test-partition-key', 'partition-1'),
+            (new Message(new EventStub('1')))->withHeader('test-partition-key', 'partition-2'),
+        );
+
+        $provider = new PartitionedSubscriptionProvider($messageRepository, new HeaderPartitioner('test-partition-key'));
+        $messages = $provider->getEventsSinceCheckpoint(PartitionedCheckpoint::fromOrigin('partition-1'));
+
+        $actualMessages = iterator_to_array($messages, false);
+        $this->assertCount(1, $actualMessages);
+        $checkpoint = $messages->getReturn();
+        $this->assertInstanceOf(PartitionedCheckpoint::class, $checkpoint);
+
+        // Cursor has offset of 2, since filtered out messages need to be counted in the stream offset
+        $this->assertEquals(2, $checkpoint->getOffset());
+
+    }
+}
diff --git a/src/Subscriptions/Partitioner.php b/src/Subscriptions/Partitioner.php
new file mode 100644
index 00000000..8e6ed5eb
--- /dev/null
+++ b/src/Subscriptions/Partitioner.php
@@ -0,0 +1,10 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\Message;
+
+interface Partitioner
+{
+    public function getPartitionKey(Message $message): ?string;
+}

From e042d8c2fc35e0bb5b2698e79001b7073599299e Mon Sep 17 00:00:00 2001
From: Robertbaelde <robert_baelde@hotmail.com>
Date: Wed, 31 May 2023 16:42:52 +0200
Subject: [PATCH 4/5] Integrate subscriptions into projections

---
 .../AggregateRootIdPartitioner.php            | 30 ------------
 .../AggregateRootPartitionedOffsetCursor.php  | 48 -------------------
 ...gregateRootPartitionedOffsetCursorTest.php | 19 --------
 src/Projections/DefaultProjectionHandler.php  | 22 ---------
 .../InMemoryProjectionStatusRepository.php    | 11 +++--
 .../PartitionedProjectionHandler.php          | 24 ----------
 src/Projections/Partitioner.php               | 10 ----
 src/Projections/ProjectionHandler.php         | 14 +++---
 src/Projections/ProjectionHandlerTest.php     | 14 +++---
 .../ProjectionStatusRepository.php            |  8 ++--
 ...egateRootIdVersionSubscriptionProvider.php |  5 +-
 src/Subscriptions/OffsetCheckpoint.php        |  5 ++
 .../OffsetStreamSubscriptionProvider.php      |  7 +--
 .../PartitionedSubscriptionProvider.php       |  5 +-
 src/Subscriptions/SubscriptionProvider.php    |  2 +-
 15 files changed, 41 insertions(+), 183 deletions(-)
 delete mode 100644 src/Projections/AggregateRootIdPartitioner.php
 delete mode 100644 src/Projections/AggregateRootPartitionedOffsetCursor.php
 delete mode 100644 src/Projections/AggregateRootPartitionedOffsetCursorTest.php
 delete mode 100644 src/Projections/DefaultProjectionHandler.php
 delete mode 100644 src/Projections/PartitionedProjectionHandler.php
 delete mode 100644 src/Projections/Partitioner.php

diff --git a/src/Projections/AggregateRootIdPartitioner.php b/src/Projections/AggregateRootIdPartitioner.php
deleted file mode 100644
index a69f3dbb..00000000
--- a/src/Projections/AggregateRootIdPartitioner.php
+++ /dev/null
@@ -1,30 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace EventSauce\EventSourcing\Projections;
-
-use EventSauce\EventSourcing\ClassNameInflector;
-use EventSauce\EventSourcing\DotSeparatedSnakeCaseInflector;
-use EventSauce\EventSourcing\Message;
-
-class AggregateRootIdPartitioner implements Partitioner
-{
-    private ClassNameInflector $classNameInflector;
-
-    public function __construct(
-        ?ClassNameInflector $classNameInflector = null,
-    ) {
-        $this->classNameInflector = $classNameInflector ?: new DotSeparatedSnakeCaseInflector();
-    }
-
-    public function getPartitionKey(Message $message): string
-    {
-        $aggregateRootId = $message->aggregateRootId();
-        if ($aggregateRootId === null) {
-            return 'no_id';
-        }
-
-        return $this->classNameInflector->instanceToType($aggregateRootId) . '_' . $aggregateRootId->toString();
-    }
-}
diff --git a/src/Projections/AggregateRootPartitionedOffsetCursor.php b/src/Projections/AggregateRootPartitionedOffsetCursor.php
deleted file mode 100644
index 67102e48..00000000
--- a/src/Projections/AggregateRootPartitionedOffsetCursor.php
+++ /dev/null
@@ -1,48 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace EventSauce\EventSourcing\Projections;
-
-use EventSauce\EventSourcing\AggregateRootId;
-use EventSauce\EventSourcing\PaginationCursor;
-
-class AggregateRootPartitionedOffsetCursor implements PaginationCursor
-{
-    private function __construct(
-        private AggregateRootId $aggregateRootId,
-        private int $offset
-    ) {
-    }
-
-    public static function fromStart(AggregateRootId $aggregateRootId): static
-    {
-        return new static(
-            aggregateRootId: $aggregateRootId,
-            offset: 0
-        );
-    }
-
-    public function toString(): string
-    {
-        return implode('###', [
-            get_class($this->aggregateRootId),
-            $this->aggregateRootId->toString(),
-            (string) $this->offset,
-        ]);
-    }
-
-    public static function fromString(string $cursor): static
-    {
-        $cursor = explode('###', $cursor);
-        return new static(
-            aggregateRootId: new $cursor[0]($cursor[1]),
-            offset: (int) $cursor[2]
-        );
-    }
-
-    public function isAtStart(): bool
-    {
-        return $this->offset === 0;
-    }
-}
diff --git a/src/Projections/AggregateRootPartitionedOffsetCursorTest.php b/src/Projections/AggregateRootPartitionedOffsetCursorTest.php
deleted file mode 100644
index 55940363..00000000
--- a/src/Projections/AggregateRootPartitionedOffsetCursorTest.php
+++ /dev/null
@@ -1,19 +0,0 @@
-<?php
-
-namespace EventSauce\EventSourcing\Projections;
-
-use EventSauce\EventSourcing\DummyAggregateRootId;
-use PHPUnit\Framework\TestCase;
-
-class AggregateRootPartitionedOffsetCursorTest extends TestCase
-{
-    /** @test */
-    public function it_can_construct_a_aggregate_root_partitioned_offset_cursor_from_start()
-    {
-        $id = DummyAggregateRootId::generate();
-        $cursor = AggregateRootPartitionedOffsetCursor::fromStart($id);
-
-        $string = $cursor->toString();
-        $this->assertEquals($cursor, AggregateRootPartitionedOffsetCursor::fromString($string));
-    }
-}
diff --git a/src/Projections/DefaultProjectionHandler.php b/src/Projections/DefaultProjectionHandler.php
deleted file mode 100644
index 4f95e914..00000000
--- a/src/Projections/DefaultProjectionHandler.php
+++ /dev/null
@@ -1,22 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace EventSauce\EventSourcing\Projections;
-
-use EventSauce\EventSourcing\PaginationCursor;
-
-class DefaultProjectionHandler
-{
-    public function __construct(
-        private ProjectionHandler $projectionHandler,
-        private ProjectionId $projectionId,
-        private PaginationCursor $initialCursor,
-    ) {
-    }
-
-    public function handle(): void
-    {
-        $this->projectionHandler->handle($this->projectionId, $this->initialCursor);
-    }
-}
diff --git a/src/Projections/InMemoryProjectionStatusRepository.php b/src/Projections/InMemoryProjectionStatusRepository.php
index 72b6fcca..f5bd6485 100644
--- a/src/Projections/InMemoryProjectionStatusRepository.php
+++ b/src/Projections/InMemoryProjectionStatusRepository.php
@@ -4,14 +4,15 @@
 
 namespace EventSauce\EventSourcing\Projections;
 
-use EventSauce\EventSourcing\PaginationCursor;
+
+use EventSauce\EventSourcing\Subscriptions\Checkpoint;
 
 class InMemoryProjectionStatusRepository implements ProjectionStatusRepository
 {
     private array $state = [];
     private array $locks = [];
 
-    public function getCursor(ProjectionId $projectionId): ?PaginationCursor
+    public function getCursor(ProjectionId $projectionId): ?Checkpoint
     {
         return $this->state[$projectionId->toString()] ?? null;
     }
@@ -19,7 +20,7 @@ public function getCursor(ProjectionId $projectionId): ?PaginationCursor
     /**
      * @throws CantLockProjection
      */
-    public function getCursorAndLock(ProjectionId $projectionId): ?PaginationCursor
+    public function getCheckpointAndLock(ProjectionId $projectionId): ?Checkpoint
     {
         if (array_key_exists($projectionId->toString(), $this->locks)) {
             throw CantLockProjection::becauseItIsAlreadyLocked($projectionId->toString());
@@ -30,9 +31,9 @@ public function getCursorAndLock(ProjectionId $projectionId): ?PaginationCursor
         return $this->state[$projectionId->toString()] ?? null;
     }
 
-    public function persistCursorAndRelease(ProjectionId $projectionId, PaginationCursor $cursor): void
+    public function persistCheckpointAndRelease(ProjectionId $projectionId, Checkpoint $checkpoint): void
     {
         unset($this->locks[$projectionId->toString()]);
-        $this->state[$projectionId->toString()] = $cursor;
+        $this->state[$projectionId->toString()] = $checkpoint;
     }
 }
diff --git a/src/Projections/PartitionedProjectionHandler.php b/src/Projections/PartitionedProjectionHandler.php
deleted file mode 100644
index 04b87fac..00000000
--- a/src/Projections/PartitionedProjectionHandler.php
+++ /dev/null
@@ -1,24 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace EventSauce\EventSourcing\Projections;
-
-use EventSauce\EventSourcing\Message;
-use EventSauce\EventSourcing\PaginationCursor;
-
-class PartitionedProjectionHandler
-{
-    public function __construct(
-        private ProjectionHandler $projectionHandler,
-        private ProjectionId $projectionId,
-        private Partitioner $partitioner,
-        private PaginationCursor $initialCursor,
-    ) {
-    }
-
-    public function handle(Message $message): void
-    {
-        $this->projectionHandler->handle($this->projectionId, $this->initialCursor);
-    }
-}
diff --git a/src/Projections/Partitioner.php b/src/Projections/Partitioner.php
deleted file mode 100644
index 515f59d6..00000000
--- a/src/Projections/Partitioner.php
+++ /dev/null
@@ -1,10 +0,0 @@
-<?php
-
-namespace EventSauce\EventSourcing\Projections;
-
-use EventSauce\EventSourcing\Message;
-
-interface Partitioner
-{
-    public function getPartitionKey(Message $message): string;
-}
diff --git a/src/Projections/ProjectionHandler.php b/src/Projections/ProjectionHandler.php
index 888cb733..a0b1ef3f 100644
--- a/src/Projections/ProjectionHandler.php
+++ b/src/Projections/ProjectionHandler.php
@@ -6,29 +6,29 @@
 
 use EventSauce\EventSourcing\Message;
 use EventSauce\EventSourcing\MessageConsumer;
-use EventSauce\EventSourcing\MessageRepository;
-use EventSauce\EventSourcing\PaginationCursor;
+use EventSauce\EventSourcing\Subscriptions\Checkpoint;
+use EventSauce\EventSourcing\Subscriptions\SubscriptionProvider;
 
 class ProjectionHandler
 {
     public function __construct(
-        private MessageRepository $repository,
+        private SubscriptionProvider $subscription,
         private MessageConsumer $consumer,
         private ProjectionStatusRepository $projectionStatusRepository,
     ) {
     }
 
-    public function handle(ProjectionId $projectionId, PaginationCursor $initialCursor): void
+    public function handle(ProjectionId $projectionId, Checkpoint $initialCheckpoint): void
     {
-        $cursor = $this->projectionStatusRepository->getCursorAndLock($projectionId) ?? $initialCursor;
+        $cursor = $this->projectionStatusRepository->getCheckpointAndLock($projectionId) ?? $initialCheckpoint;
 
-        $messages = $this->repository->paginate($cursor);
+        $messages = $this->subscription->getEventsSinceCheckpoint($cursor);
 
         /** @var Message $message */
         foreach ($messages as $message) {
             $this->consumer->handle($message);
         }
 
-        $this->projectionStatusRepository->persistCursorAndRelease($projectionId, $messages->getReturn());
+        $this->projectionStatusRepository->persistCheckpointAndRelease($projectionId, $messages->getReturn());
     }
 }
diff --git a/src/Projections/ProjectionHandlerTest.php b/src/Projections/ProjectionHandlerTest.php
index 9c34aa96..e213d2de 100644
--- a/src/Projections/ProjectionHandlerTest.php
+++ b/src/Projections/ProjectionHandlerTest.php
@@ -7,6 +7,8 @@
 use EventSauce\EventSourcing\InMemoryMessageRepository;
 use EventSauce\EventSourcing\Message;
 use EventSauce\EventSourcing\OffsetCursor;
+use EventSauce\EventSourcing\Subscriptions\OffsetCheckpoint;
+use EventSauce\EventSourcing\Subscriptions\OffsetStreamSubscriptionProvider;
 use PHPUnit\Framework\TestCase;
 
 class ProjectionHandlerTest extends TestCase
@@ -24,15 +26,15 @@ public function it_uses_initial_cursor_when_no_projection_state_exists()
         $projectionStatusRepository = new InMemoryProjectionStatusRepository();
 
         $projectionHandler = new ProjectionHandler(
-            repository: $repository,
+            subscription: new OffsetStreamSubscriptionProvider($repository),
             consumer: $consumer,
             projectionStatusRepository: $projectionStatusRepository,
         );
 
-        $projectionHandler->handle($this->getProjectionId(), OffsetCursor::fromStart());
+        $projectionHandler->handle($this->getProjectionId(), OffsetCheckpoint::fromStart());
 
         $this->assertCount(1, $consumer->collectedMessages());
-        $this->assertEquals(OffsetCursor::fromStart()->plusOffset(1), $projectionStatusRepository->getCursor($this->getProjectionId()));
+        $this->assertEquals(OffsetCheckpoint::forOffset(1), $projectionStatusRepository->getCursor($this->getProjectionId()));
     }
 
     /** @test */
@@ -46,16 +48,16 @@ public function it_cant_start_a_new_play_while_projection_is_playing()
         $consumer = new CollectingMessageConsumer();
 
         $projectionStatusRepository = new InMemoryProjectionStatusRepository();
-        $projectionStatusRepository->getCursorAndLock($this->getProjectionId());
+        $projectionStatusRepository->getCheckpointAndLock($this->getProjectionId());
 
         $projectionHandler = new ProjectionHandler(
-            repository: $repository,
+            subscription: new OffsetStreamSubscriptionProvider($repository),
             consumer: $consumer,
             projectionStatusRepository: $projectionStatusRepository,
         );
 
         $this->expectExceptionObject(CantLockProjection::becauseItIsAlreadyLocked('test'));
-        $projectionHandler->handle($this->getProjectionId(), OffsetCursor::fromStart());
+        $projectionHandler->handle($this->getProjectionId(), OffsetCheckpoint::fromStart());
     }
 
     private function getProjectionId(): ProjectionId
diff --git a/src/Projections/ProjectionStatusRepository.php b/src/Projections/ProjectionStatusRepository.php
index 2cd1b259..4de79bcb 100644
--- a/src/Projections/ProjectionStatusRepository.php
+++ b/src/Projections/ProjectionStatusRepository.php
@@ -2,16 +2,16 @@
 
 namespace EventSauce\EventSourcing\Projections;
 
-use EventSauce\EventSourcing\PaginationCursor;
+use EventSauce\EventSourcing\Subscriptions\Checkpoint;
 
 interface ProjectionStatusRepository
 {
-    public function getCursor(ProjectionId $projectionId): ?PaginationCursor;
+    public function getCursor(ProjectionId $projectionId): ?Checkpoint;
 
     /**
      * @throws CantLockProjection
      */
-    public function getCursorAndLock(ProjectionId $projectionId): ?PaginationCursor;
+    public function getCheckpointAndLock(ProjectionId $projectionId): ?Checkpoint;
 
-    public function persistCursorAndRelease(ProjectionId $projectionId, PaginationCursor $cursor): void;
+    public function persistCheckpointAndRelease(ProjectionId $projectionId, Checkpoint $checkpoint): void;
 }
diff --git a/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php b/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
index 731efab5..b65b29d8 100644
--- a/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
+++ b/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
@@ -14,11 +14,12 @@ class AggregateRootIdVersionSubscriptionProvider implements SubscriptionProvider
 {
 
     public function __construct(
-        private MessageRepository $messageRepository)
+        private MessageRepository $messageRepository,
+    )
     {
     }
 
-    public function getEventsSinceCheckpoint(Checkpoint $checkpoint, int $maxEvents = 100): Generator
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint): Generator
     {
         if(!$checkpoint instanceof AggregateCheckpoint){
             throw new \InvalidArgumentException('Checkpoint must be an instance of AggregateCheckpoint');
diff --git a/src/Subscriptions/OffsetCheckpoint.php b/src/Subscriptions/OffsetCheckpoint.php
index ebdb9521..78361a1e 100644
--- a/src/Subscriptions/OffsetCheckpoint.php
+++ b/src/Subscriptions/OffsetCheckpoint.php
@@ -14,6 +14,11 @@ public static function forOffset(int $offset): static
         return new static($offset);
     }
 
+    public static function fromStart(): static
+    {
+        return new static(0);
+    }
+
     public function getOffset(): int
     {
         return $this->offset;
diff --git a/src/Subscriptions/OffsetStreamSubscriptionProvider.php b/src/Subscriptions/OffsetStreamSubscriptionProvider.php
index 784f695d..10c8a52e 100644
--- a/src/Subscriptions/OffsetStreamSubscriptionProvider.php
+++ b/src/Subscriptions/OffsetStreamSubscriptionProvider.php
@@ -11,20 +11,21 @@ class OffsetStreamSubscriptionProvider implements SubscriptionProvider
 {
     public function __construct(
         private MessageRepository $messageRepository,
+        private int $pageLimit = 100,
     ) {
     }
 
-    public function getEventsSinceCheckpoint(Checkpoint $checkpoint, int $maxEvents = 100): \Generator
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint): \Generator
     {
         if ( ! $checkpoint instanceof OffsetCheckpoint) {
             throw new \Exception('Invalid checkpoint type');
         }
 
-        $cursor = OffsetCursor::fromOffset($checkpoint->getOffset(), $maxEvents);
+        $cursor = OffsetCursor::fromOffset($checkpoint->getOffset(), $this->pageLimit);
         $messages = $this->messageRepository->paginate($cursor);
 
         yield from $messages;
 
-        return OffsetCheckpoint::forOffset($cursor->offset());
+        return OffsetCheckpoint::forOffset($messages->getReturn()->offset());
     }
 }
diff --git a/src/Subscriptions/PartitionedSubscriptionProvider.php b/src/Subscriptions/PartitionedSubscriptionProvider.php
index a1f3fa6a..515cfa9e 100644
--- a/src/Subscriptions/PartitionedSubscriptionProvider.php
+++ b/src/Subscriptions/PartitionedSubscriptionProvider.php
@@ -14,16 +14,17 @@ class PartitionedSubscriptionProvider implements SubscriptionProvider
     public function __construct(
         private MessageRepository $messageRepository,
         private Partitioner $partitioner,
+        private int $pageLimit = 100,
     ) {
     }
 
-    public function getEventsSinceCheckpoint(Checkpoint $checkpoint, int $maxEvents = 100): \Generator
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint): \Generator
     {
         if ( ! $checkpoint instanceof PartitionedCheckpoint) {
             throw new \InvalidArgumentException('Checkpoint must be an instance of PartitionedCheckpoint');
         }
 
-        $cursor = OffsetCursor::fromOffset($checkpoint->getOffset(), $maxEvents);
+        $cursor = OffsetCursor::fromOffset($checkpoint->getOffset(), $this->pageLimit);
 
         $messages = $this->messageRepository->paginate($cursor);
 
diff --git a/src/Subscriptions/SubscriptionProvider.php b/src/Subscriptions/SubscriptionProvider.php
index 40f508e3..d480acb9 100644
--- a/src/Subscriptions/SubscriptionProvider.php
+++ b/src/Subscriptions/SubscriptionProvider.php
@@ -10,5 +10,5 @@ interface SubscriptionProvider
     /**
      * @return Generator<int, Message, void, Checkpoint>
      */
-    public function getEventsSinceCheckpoint(Checkpoint $checkpoint, int $maxEvents = 100): Generator;
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint): Generator;
 }

From 3a55e778653e1d3e62c7f7b01696850bdaa9af47 Mon Sep 17 00:00:00 2001
From: Robertbaelde <robert_baelde@hotmail.com>
Date: Wed, 31 May 2023 16:53:25 +0200
Subject: [PATCH 5/5] Fix phpstan

---
 src/Subscriptions/AggregateCheckpoint.php     |  4 ++--
 ...egateRootIdVersionSubscriptionProvider.php | 23 +++++++++----------
 src/Subscriptions/HeaderPartitioner.php       |  6 ++++-
 src/Subscriptions/OffsetCheckpoint.php        |  2 +-
 .../OffsetStreamSubscriptionProvider.php      |  8 ++++++-
 src/Subscriptions/PartitionedCheckpoint.php   |  2 +-
 .../PartitionedSubscriptionProvider.php       |  9 ++++++--
 src/Subscriptions/SubscriptionProvider.php    |  2 +-
 8 files changed, 35 insertions(+), 21 deletions(-)

diff --git a/src/Subscriptions/AggregateCheckpoint.php b/src/Subscriptions/AggregateCheckpoint.php
index 0b29d680..3ecc1c3a 100644
--- a/src/Subscriptions/AggregateCheckpoint.php
+++ b/src/Subscriptions/AggregateCheckpoint.php
@@ -4,7 +4,7 @@
 
 use EventSauce\EventSourcing\AggregateRootId;
 
-class AggregateCheckpoint implements Checkpoint
+final class AggregateCheckpoint implements Checkpoint
 {
     private function __construct(
         private AggregateRootId $aggregateRootId,
@@ -12,7 +12,7 @@ private function __construct(
     ) {
     }
 
-    public static function forAggregateRootId(AggregateRootId $aggregateRootId, $version = 0): static
+    public static function forAggregateRootId(AggregateRootId $aggregateRootId, int $version = 0): static
     {
         return new static($aggregateRootId, $version);
     }
diff --git a/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php b/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
index b65b29d8..c6837e5a 100644
--- a/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
+++ b/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
@@ -1,27 +1,21 @@
 <?php
 
+declare(strict_types=1);
+
 namespace EventSauce\EventSourcing\Subscriptions;
 
-use EventSauce\EventSourcing\DummyAggregateRootId;
-use EventSauce\EventSourcing\EventStub;
-use EventSauce\EventSourcing\Header;
-use EventSauce\EventSourcing\InMemoryMessageRepository;
-use EventSauce\EventSourcing\Message;
 use EventSauce\EventSourcing\MessageRepository;
-use Generator;
 
 class AggregateRootIdVersionSubscriptionProvider implements SubscriptionProvider
 {
-
     public function __construct(
         private MessageRepository $messageRepository,
-    )
-    {
+    ) {
     }
 
-    public function getEventsSinceCheckpoint(Checkpoint $checkpoint): Generator
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint): \Generator
     {
-        if(!$checkpoint instanceof AggregateCheckpoint){
+        if ( ! $checkpoint instanceof AggregateCheckpoint) {
             throw new \InvalidArgumentException('Checkpoint must be an instance of AggregateCheckpoint');
         }
 
@@ -29,6 +23,11 @@ public function getEventsSinceCheckpoint(Checkpoint $checkpoint): Generator
 
         yield from $messages;
 
-        return AggregateCheckpoint::forAggregateRootId($checkpoint->getAggregateRootId(), $messages->getReturn() ?? $checkpoint->getVersion());
+        $version = $messages->getReturn() ?? $checkpoint->getVersion();
+        if ( ! is_int($version)) {
+            throw new \InvalidArgumentException('Version must be an integer');
+        }
+
+        return AggregateCheckpoint::forAggregateRootId($checkpoint->getAggregateRootId(), $version);
     }
 }
diff --git a/src/Subscriptions/HeaderPartitioner.php b/src/Subscriptions/HeaderPartitioner.php
index c5d2c494..8069e013 100644
--- a/src/Subscriptions/HeaderPartitioner.php
+++ b/src/Subscriptions/HeaderPartitioner.php
@@ -13,6 +13,10 @@ public function __construct(
 
     public function getPartitionKey(Message $message): ?string
     {
-        return $message->header($this->headerName);
+        $key = $message->header($this->headerName);
+        if(!is_string($key)) {
+            return null;
+        }
+        return $key;
     }
 }
diff --git a/src/Subscriptions/OffsetCheckpoint.php b/src/Subscriptions/OffsetCheckpoint.php
index 78361a1e..e9c84ed6 100644
--- a/src/Subscriptions/OffsetCheckpoint.php
+++ b/src/Subscriptions/OffsetCheckpoint.php
@@ -2,7 +2,7 @@
 
 namespace EventSauce\EventSourcing\Subscriptions;
 
-class OffsetCheckpoint implements Checkpoint
+final class OffsetCheckpoint implements Checkpoint
 {
     private function __construct(
         private int $offset,
diff --git a/src/Subscriptions/OffsetStreamSubscriptionProvider.php b/src/Subscriptions/OffsetStreamSubscriptionProvider.php
index 10c8a52e..d24f6641 100644
--- a/src/Subscriptions/OffsetStreamSubscriptionProvider.php
+++ b/src/Subscriptions/OffsetStreamSubscriptionProvider.php
@@ -26,6 +26,12 @@ public function getEventsSinceCheckpoint(Checkpoint $checkpoint): \Generator
 
         yield from $messages;
 
-        return OffsetCheckpoint::forOffset($messages->getReturn()->offset());
+        $returnedCursor = $messages->getReturn();
+
+        if ( ! $returnedCursor instanceof OffsetCursor) {
+            throw new \Exception('Invalid returned cursor type');
+        }
+
+        return OffsetCheckpoint::forOffset($returnedCursor->offset());
     }
 }
diff --git a/src/Subscriptions/PartitionedCheckpoint.php b/src/Subscriptions/PartitionedCheckpoint.php
index c7e4ced9..78c45dba 100644
--- a/src/Subscriptions/PartitionedCheckpoint.php
+++ b/src/Subscriptions/PartitionedCheckpoint.php
@@ -4,7 +4,7 @@
 
 namespace EventSauce\EventSourcing\Subscriptions;
 
-class PartitionedCheckpoint implements Checkpoint
+final class PartitionedCheckpoint implements Checkpoint
 {
     private function __construct(
         private string $partitionKey,
diff --git a/src/Subscriptions/PartitionedSubscriptionProvider.php b/src/Subscriptions/PartitionedSubscriptionProvider.php
index 515cfa9e..d8770b3b 100644
--- a/src/Subscriptions/PartitionedSubscriptionProvider.php
+++ b/src/Subscriptions/PartitionedSubscriptionProvider.php
@@ -4,7 +4,6 @@
 
 namespace EventSauce\EventSourcing\Subscriptions;
 
-use EventSauce\EventSourcing\AntiCorruptionLayer\MessageFilter;
 use EventSauce\EventSourcing\Message;
 use EventSauce\EventSourcing\MessageRepository;
 use EventSauce\EventSourcing\OffsetCursor;
@@ -37,6 +36,12 @@ public function getEventsSinceCheckpoint(Checkpoint $checkpoint): \Generator
             yield $message;
         }
 
-        return $checkpoint->withOffset($messages->getReturn()->offset());
+        $returnedCursor = $messages->getReturn();
+
+        if ( ! $returnedCursor instanceof OffsetCursor) {
+            throw new \Exception('Invalid returned cursor type');
+        };
+
+        return $checkpoint->withOffset($returnedCursor->offset());
     }
 }
diff --git a/src/Subscriptions/SubscriptionProvider.php b/src/Subscriptions/SubscriptionProvider.php
index d480acb9..95fea314 100644
--- a/src/Subscriptions/SubscriptionProvider.php
+++ b/src/Subscriptions/SubscriptionProvider.php
@@ -8,7 +8,7 @@
 interface SubscriptionProvider
 {
     /**
-     * @return Generator<int, Message, void, Checkpoint>
+     * @return Generator<mixed, Message, void, Checkpoint>
      */
     public function getEventsSinceCheckpoint(Checkpoint $checkpoint): Generator;
 }