Checkpoint initial position

src/Core/src/Eventuous.Subscriptions/CheckpointInitialPosition.cs

@@ -0,0 +1,6 @@
+namespace Eventuous.Subscriptions; 
+
+public enum CheckpointInitialPosition {
+    Beginning,
+    End
+}

src/Core/src/Eventuous.Subscriptions/EventSubscriptionWithCheckpoint.cs

@@ -38,7 +38,8 @@ static ConsumePipe ConfigurePipe(ConsumePipe pipe, int concurrencyLimit)
 
     EventPosition?           LastProcessed           { get; set; }
     CheckpointCommitHandler? CheckpointCommitHandler { get; set; }
-    ICheckpointStore         CheckpointStore         { get; } = Ensure.NotNull(checkpointStore);
+
+    protected ICheckpointStore CheckpointStore { get; } = Ensure.NotNull(checkpointStore);
 
     protected SubscriptionKind Kind { get; } = kind;
 

src/Core/src/Eventuous.Subscriptions/SubscriptionOptions.cs

@@ -19,4 +19,5 @@ public abstract record SubscriptionOptions {
 public abstract record SubscriptionWithCheckpointOptions : SubscriptionOptions {
     public int CheckpointCommitBatchSize { get; set; } = 100;
     public int CheckpointCommitDelayMs   { get; set; } = 5000;
+    public CheckpointInitialPosition InitialPosition { get; set; } = CheckpointInitialPosition.Beginning;
 }

src/Core/test/Eventuous.Tests.Subscriptions.Base/SubscribeToAll.cs

@@ -15,6 +15,22 @@ SubscriptionFixtureBase<TContainer, TSubscription, TSubscriptionOptions, TCheckp
     where TSubscription : EventSubscription<TSubscriptionOptions>
     where TSubscriptionOptions : SubscriptionOptions
     where TCheckpointStore : class, ICheckpointStore {
+
+    protected async Task ShouldStartConsumptionFromEnd(CancellationToken cancellationToken) {
+        const int count = 10;
+
+        var commands = await GenerateAndHandleCommands(count);
+        var testEvents = commands.Select(ToEvent).ToList();
+
+        await fixture.StartSubscription();
+        await fixture.Handler.AssertCollection(TimeSpan.FromSeconds(2), []).Validate(cancellationToken);
+
+        commands = await GenerateAndHandleCommands(count);
+        testEvents = commands.Select(ToEvent).ToList();
+
+        await fixture.Handler.AssertCollection(TimeSpan.FromSeconds(2), [.. testEvents]).Validate(cancellationToken);
+        await fixture.StopSubscription();
+
     protected async Task ShouldConsumeProducedEvents(CancellationToken cancellationToken) {
         const int count = 10;
 

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/AllStreamSubscription.cs

@@ -84,7 +84,13 @@ protected override async ValueTask Subscribe(CancellationToken cancellationToken
 
         var (_, position) = await GetCheckpoint(cancellationToken).NoContext();
 
-        var fromAll = position == null ? FromAll.Start : FromAll.After(new(position.Value, position.Value));
+        FromAll GetPosition() {
+            if (position == null && Options.InitialPosition == CheckpointInitialPosition.End) return FromAll.End;
+            if (position == null) return FromAll.Start;
+            return FromAll.After(new(position.Value, position.Value));
+        }
+
+        var fromAll = GetPosition();
 
         Subscription = await Client.SubscribeToAllAsync(
                 fromAll,

src/KurrentDB/src/Eventuous.KurrentDB/Subscriptions/StreamSubscription.cs

@@ -91,7 +91,13 @@ public StreamSubscription(
     protected override async ValueTask Subscribe(CancellationToken cancellationToken) {
         var (_, position) = await GetCheckpoint(cancellationToken).NoContext();
 
-        var fromStream = position == null ? FromStream.Start : FromStream.After(StreamPosition.FromInt64((long)position));
+        FromStream GetStreamPosition() {
+            if (position == null && Options.InitialPosition == CheckpointInitialPosition.End) return FromStream.End;
+            if (position == null) return FromStream.Start;
+            return FromStream.After(StreamPosition.FromInt64((long)position));
+        }
+
+        var fromStream = GetStreamPosition();
 
         Subscription = await Client.SubscribeToStreamAsync(
                 Options.StreamName,

src/KurrentDB/test/Eventuous.Tests.KurrentDB/Subscriptions/SubscribeTests.cs

@@ -7,7 +7,18 @@
 
 namespace Eventuous.Tests.KurrentDB.Subscriptions;
 
-public class SubscribeToAll()
+    public class SubscribeToAllFromEnd()
+    : SubscribeToAllBase<EventStoreDbContainer, AllStreamSubscription, AllStreamSubscriptionOptions, TestCheckpointStore>(
+        new CatchUpSubscriptionFixture<AllStreamSubscription, AllStreamSubscriptionOptions, TestEventHandler>(opt => opt.InitialPosition = CheckpointInitialPosition.End, new("$all"), false)
+    ) {
+    [Test]
+    [Retry(3)]
+    public async Task Esdb_ShouldStartConsumptionFromEnd(CancellationToken cancellationToken) {
+        await ShouldStartConsumptionFromEnd(cancellationToken);
+    }
+}
+
+    public class SubscribeToAll()
     : SubscribeToAllBase<EventStoreDbContainer, AllStreamSubscription, AllStreamSubscriptionOptions, TestCheckpointStore>(
         new CatchUpSubscriptionFixture<AllStreamSubscription, AllStreamSubscriptionOptions, TestEventHandler>(_ => { }, new("$all"), false)
     ) {

src/Postgres/test/Eventuous.Tests.Postgres/Subscriptions/SubscribeTests.cs

@@ -7,6 +7,17 @@
 
 namespace Eventuous.Tests.Postgres.Subscriptions;
 
+[NotInParallel]
+public class SubscribeToAllFromEnd()
+    : SubscribeToAllBase<PostgreSqlContainer, PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions, PostgresCheckpointStore>(
+        new SubscriptionFixture<PostgresStore, PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions, TestEventHandler>(opt => opt.InitialPosition = Eventuous.Subscriptions.CheckpointInitialPosition.End, false)
+    ) {
+    [Test]
+    public async Task Postgres_ShouldStartConsumptionFromEnd(CancellationToken cancellationToken) {
+        await ShouldStartConsumptionFromEnd(cancellationToken);
+    }
+}
+
 [NotInParallel]
 public class SubscribeToAll()
     : SubscribeToAllBase<PostgreSqlContainer, PostgresAllStreamSubscription, PostgresAllStreamSubscriptionOptions, PostgresCheckpointStore>(

src/Relational/src/Eventuous.Sql.Base/Subscriptions/SqlSubscriptionBase.cs

@@ -214,6 +214,18 @@ protected override async ValueTask Subscribe(CancellationToken cancellationToken
         await BeforeSubscribe(cancellationToken).NoContext();
         var (_, position) = await GetCheckpoint(cancellationToken).NoContext();
 
+        if (position == null && Options.InitialPosition == CheckpointInitialPosition.End) {
+            var endOfStream = await GetSubscriptionEndOfStream(cancellationToken).NoContext();
+            if (endOfStream == EndOfStream.Invalid) {
+                throw new InvalidOperationException($"Could not get the end of the stream for subscription {SubscriptionId}");
+            }
+            await CheckpointStore.StoreCheckpoint(new Checkpoint(SubscriptionId, endOfStream.Position), true, cancellationToken).NoContext();
+            position = endOfStream.Position;
+            }
+            await CheckpointStore.StoreCheckpoint(new Checkpoint(SubscriptionId, endOfStream.Position), true, cancellationToken);
+            position = endOfStream.Position;
+        }
+
         _runner = new TaskRunner(token => PollingQuery(position, token)).Start();
     }