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/InMemoryProjectionStatusRepository.php b/src/Projections/InMemoryProjectionStatusRepository.php
new file mode 100644
index 00000000..f5bd6485
--- /dev/null
+++ b/src/Projections/InMemoryProjectionStatusRepository.php
@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Projections;
+
+
+use EventSauce\EventSourcing\Subscriptions\Checkpoint;
+
+class InMemoryProjectionStatusRepository implements ProjectionStatusRepository
+{
+    private array $state = [];
+    private array $locks = [];
+
+    public function getCursor(ProjectionId $projectionId): ?Checkpoint
+    {
+        return $this->state[$projectionId->toString()] ?? null;
+    }
+
+    /**
+     * @throws CantLockProjection
+     */
+    public function getCheckpointAndLock(ProjectionId $projectionId): ?Checkpoint
+    {
+        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 persistCheckpointAndRelease(ProjectionId $projectionId, Checkpoint $checkpoint): void
+    {
+        unset($this->locks[$projectionId->toString()]);
+        $this->state[$projectionId->toString()] = $checkpoint;
+    }
+}
diff --git a/src/Projections/ProjectionHandler.php b/src/Projections/ProjectionHandler.php
new file mode 100644
index 00000000..a0b1ef3f
--- /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\Subscriptions\Checkpoint;
+use EventSauce\EventSourcing\Subscriptions\SubscriptionProvider;
+
+class ProjectionHandler
+{
+    public function __construct(
+        private SubscriptionProvider $subscription,
+        private MessageConsumer $consumer,
+        private ProjectionStatusRepository $projectionStatusRepository,
+    ) {
+    }
+
+    public function handle(ProjectionId $projectionId, Checkpoint $initialCheckpoint): void
+    {
+        $cursor = $this->projectionStatusRepository->getCheckpointAndLock($projectionId) ?? $initialCheckpoint;
+
+        $messages = $this->subscription->getEventsSinceCheckpoint($cursor);
+
+        /** @var Message $message */
+        foreach ($messages as $message) {
+            $this->consumer->handle($message);
+        }
+
+        $this->projectionStatusRepository->persistCheckpointAndRelease($projectionId, $messages->getReturn());
+    }
+}
diff --git a/src/Projections/ProjectionHandlerTest.php b/src/Projections/ProjectionHandlerTest.php
new file mode 100644
index 00000000..e213d2de
--- /dev/null
+++ b/src/Projections/ProjectionHandlerTest.php
@@ -0,0 +1,67 @@
+<?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 EventSauce\EventSourcing\Subscriptions\OffsetCheckpoint;
+use EventSauce\EventSourcing\Subscriptions\OffsetStreamSubscriptionProvider;
+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(
+            subscription: new OffsetStreamSubscriptionProvider($repository),
+            consumer: $consumer,
+            projectionStatusRepository: $projectionStatusRepository,
+        );
+
+        $projectionHandler->handle($this->getProjectionId(), OffsetCheckpoint::fromStart());
+
+        $this->assertCount(1, $consumer->collectedMessages());
+        $this->assertEquals(OffsetCheckpoint::forOffset(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->getCheckpointAndLock($this->getProjectionId());
+
+        $projectionHandler = new ProjectionHandler(
+            subscription: new OffsetStreamSubscriptionProvider($repository),
+            consumer: $consumer,
+            projectionStatusRepository: $projectionStatusRepository,
+        );
+
+        $this->expectExceptionObject(CantLockProjection::becauseItIsAlreadyLocked('test'));
+        $projectionHandler->handle($this->getProjectionId(), OffsetCheckpoint::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..4de79bcb
--- /dev/null
+++ b/src/Projections/ProjectionStatusRepository.php
@@ -0,0 +1,17 @@
+<?php
+
+namespace EventSauce\EventSourcing\Projections;
+
+use EventSauce\EventSourcing\Subscriptions\Checkpoint;
+
+interface ProjectionStatusRepository
+{
+    public function getCursor(ProjectionId $projectionId): ?Checkpoint;
+
+    /**
+     * @throws CantLockProjection
+     */
+    public function getCheckpointAndLock(ProjectionId $projectionId): ?Checkpoint;
+
+    public function persistCheckpointAndRelease(ProjectionId $projectionId, Checkpoint $checkpoint): void;
+}
diff --git a/src/Subscriptions/AggregateCheckpoint.php b/src/Subscriptions/AggregateCheckpoint.php
new file mode 100644
index 00000000..3ecc1c3a
--- /dev/null
+++ b/src/Subscriptions/AggregateCheckpoint.php
@@ -0,0 +1,30 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\AggregateRootId;
+
+final class AggregateCheckpoint implements Checkpoint
+{
+    private function __construct(
+        private AggregateRootId $aggregateRootId,
+        private int $aggregateRootVersion,
+    ) {
+    }
+
+    public static function forAggregateRootId(AggregateRootId $aggregateRootId, int $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..c6837e5a
--- /dev/null
+++ b/src/Subscriptions/AggregateRootIdVersionSubscriptionProvider.php
@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+use EventSauce\EventSourcing\MessageRepository;
+
+class AggregateRootIdVersionSubscriptionProvider implements SubscriptionProvider
+{
+    public function __construct(
+        private MessageRepository $messageRepository,
+    ) {
+    }
+
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint): \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;
+
+        $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/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/HeaderPartitioner.php b/src/Subscriptions/HeaderPartitioner.php
new file mode 100644
index 00000000..8069e013
--- /dev/null
+++ b/src/Subscriptions/HeaderPartitioner.php
@@ -0,0 +1,22 @@
+<?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
+    {
+        $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
new file mode 100644
index 00000000..e9c84ed6
--- /dev/null
+++ b/src/Subscriptions/OffsetCheckpoint.php
@@ -0,0 +1,36 @@
+<?php
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+final class OffsetCheckpoint implements Checkpoint
+{
+    private function __construct(
+        private int $offset,
+    ) {
+    }
+
+    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;
+    }
+
+    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..d24f6641
--- /dev/null
+++ b/src/Subscriptions/OffsetStreamSubscriptionProvider.php
@@ -0,0 +1,37 @@
+<?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,
+        private int $pageLimit = 100,
+    ) {
+    }
+
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint): \Generator
+    {
+        if ( ! $checkpoint instanceof OffsetCheckpoint) {
+            throw new \Exception('Invalid checkpoint type');
+        }
+
+        $cursor = OffsetCursor::fromOffset($checkpoint->getOffset(), $this->pageLimit);
+        $messages = $this->messageRepository->paginate($cursor);
+
+        yield from $messages;
+
+        $returnedCursor = $messages->getReturn();
+
+        if ( ! $returnedCursor instanceof OffsetCursor) {
+            throw new \Exception('Invalid returned cursor type');
+        }
+
+        return OffsetCheckpoint::forOffset($returnedCursor->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/PartitionedCheckpoint.php b/src/Subscriptions/PartitionedCheckpoint.php
new file mode 100644
index 00000000..78c45dba
--- /dev/null
+++ b/src/Subscriptions/PartitionedCheckpoint.php
@@ -0,0 +1,34 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+final 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..d8770b3b
--- /dev/null
+++ b/src/Subscriptions/PartitionedSubscriptionProvider.php
@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EventSauce\EventSourcing\Subscriptions;
+
+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,
+        private int $pageLimit = 100,
+    ) {
+    }
+
+    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(), $this->pageLimit);
+
+        $messages = $this->messageRepository->paginate($cursor);
+
+        /** @var Message $message */
+        foreach ($messages as $message) {
+            $partitionKey = $this->partitioner->getPartitionKey($message);
+            if ($partitionKey !== $checkpoint->getPartitionKey()) {
+                continue;
+            }
+            yield $message;
+        }
+
+        $returnedCursor = $messages->getReturn();
+
+        if ( ! $returnedCursor instanceof OffsetCursor) {
+            throw new \Exception('Invalid returned cursor type');
+        };
+
+        return $checkpoint->withOffset($returnedCursor->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;
+}
diff --git a/src/Subscriptions/SubscriptionProvider.php b/src/Subscriptions/SubscriptionProvider.php
new file mode 100644
index 00000000..95fea314
--- /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<mixed, Message, void, Checkpoint>
+     */
+    public function getEventsSinceCheckpoint(Checkpoint $checkpoint): Generator;
+}