Multi-Stage Tracked Sessions Closes GH-1809

src/Persistence/MartenTests/TestHelpers/AppendLetters.cs

@@ -0,0 +1,3 @@
+namespace MartenTests.TestHelpers;
+
+public record AppendLetters(Guid Id, string[] Events);

src/Persistence/MartenTests/TestHelpers/AppendLetters2.cs

@@ -0,0 +1,3 @@
+namespace MartenTests.TestHelpers;
+
+public record AppendLetters2(Guid Id, string[] Events);

src/Persistence/MartenTests/TestHelpers/AppendLetters2Handler.cs

@@ -0,0 +1,27 @@
+using Wolverine;
+using Wolverine.Marten;
+
+namespace MartenTests.TestHelpers;
+
+public static class AppendLetters2Handler
+{
+    [MartenStore(typeof(ILetterStore))]
+    public static (Events, OutgoingMessages) Handle(
+        AppendLetters2 command, 
+
+        [WriteAggregate(Required = false)]
+        LetterCounts aggregate)
+    {
+        switch (command.Events.Length)
+        {
+            case 0:
+                return ([], []);
+
+            case 1: 
+                return (new Events(command.Events[0].ToLetterEvents()), []);
+
+            default:
+                return (new Events(command.Events[0].ToLetterEvents()), [new AppendLetters2(command.Id, command.Events.Skip(1).ToArray())]);
+        }
+    }
+}

src/Persistence/MartenTests/TestHelpers/AppendLettersHandler.cs

@@ -0,0 +1,26 @@
+using Wolverine;
+using Wolverine.Marten;
+
+namespace MartenTests.TestHelpers;
+
+public static class AppendLettersHandler
+{
+    public static (Events, OutgoingMessages) Handle(
+        AppendLetters command, 
+
+        [WriteAggregate(Required = false)]
+        LetterCounts aggregate)
+    {
+        switch (command.Events.Length)
+        {
+            case 0:
+                return ([], []);
+
+            case 1: 
+                return (new Events(command.Events[0].ToLetterEvents()), []);
+
+            default:
+                return (new Events(command.Events[0].ToLetterEvents()), [new AppendLetters(command.Id, command.Events.Skip(1).ToArray())]);
+        }
+    }
+}

src/Persistence/MartenTests/TestHelpers/EEvent.cs

@@ -0,0 +1,3 @@
+namespace MartenTests.TestHelpers;
+
+public record EEvent;

src/Persistence/MartenTests/TestHelpers/ILetterStore.cs

@@ -0,0 +1,8 @@
+using Marten;
+
+namespace MartenTests.TestHelpers;
+
+public interface ILetterStore : IDocumentStore
+{
+
+}

src/Persistence/MartenTests/TestHelpers/LetterCounts.cs

@@ -0,0 +1,14 @@
+using JasperFx;
+
+namespace MartenTests.TestHelpers;
+
+public class LetterCounts: IRevisioned
+{
+    public Guid Id { get; set; }
+    public int ACount { get; set; }
+    public int BCount { get; set; }
+    public int CCount { get; set; }
+    public int DCount { get; set; }
+    public int ECount { get; set; }
+    public int Version { get; set; }
+}

src/Persistence/MartenTests/TestHelpers/LetterCountsProjection.cs

@@ -0,0 +1,42 @@
+using JasperFx.Events;
+using Marten.Events.Aggregation;
+using MartenTests.AggregateHandlerWorkflow;
+
+namespace MartenTests.TestHelpers;
+
+public class LetterCountsProjection: SingleStreamProjection<LetterCounts, Guid>
+{
+    public override LetterCounts Evolve(LetterCounts snapshot, Guid id, IEvent e)
+    {
+
+        switch (e.Data)
+        {
+            case AEvent _:
+                snapshot ??= new() { Id = id };
+                snapshot.ACount++;
+                break;
+
+            case BEvent _:
+                snapshot ??= new() { Id = id };
+                snapshot.BCount++;
+                break;
+
+            case CEvent _:
+                snapshot ??= new() { Id = id };
+                snapshot.CCount++;
+                break;
+
+            case DEvent _:
+                snapshot ??= new() { Id = id };
+                snapshot.DCount++;
+                break;
+
+            case EEvent _:
+                snapshot ??= new() { Id = id };
+                snapshot.ECount++;
+                break;
+        }
+
+        return snapshot;
+    }
+}

src/Persistence/MartenTests/TestHelpers/LetterEvents.cs

@@ -0,0 +1,38 @@
+using MartenTests.AggregateHandlerWorkflow;
+
+namespace MartenTests.TestHelpers;
+
+/// <summary>
+/// Basically an ObjectMother for the A/B/C/D/Event types
+/// </summary>
+public static class LetterEvents
+{
+    public static IEnumerable<object> ToLetterEvents(this string text)
+    {
+        foreach (var character in text.ToLowerInvariant())
+        {
+            switch (character)
+            {
+                case 'a':
+                    yield return new AEvent();
+                    break;
+
+                case 'b':
+                    yield return new BEvent();
+                    break;
+
+                case 'c':
+                    yield return new CEvent();
+                    break;
+
+                case 'd':
+                    yield return new DEvent();
+                    break;
+
+                case 'e':
+                    yield return new EEvent();
+                    break;
+            }
+        }
+    }
+}

src/Persistence/MartenTests/TestHelpers/catch_up_and_then_do_nothing.cs

@@ -0,0 +1,121 @@
+using IntegrationTests;
+using JasperFx.Core;
+using JasperFx.Events.Daemon;
+using JasperFx.Events.Projections;
+using Marten;
+using Marten.Events;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine;
+using Wolverine.Marten;
+using Wolverine.Tracking;
+
+namespace MartenTests.TestHelpers;
+
+public class catch_up_and_then_do_nothing : IAsyncLifetime
+{
+    private IHost _host;
+
+    public async Task InitializeAsync()
+    {
+
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Services.AddMarten(m =>
+                {
+                    m.DisableNpgsqlLogging = true;
+
+                    m.Connection(Servers.PostgresConnectionString);
+                    m.DatabaseSchemaName = "letters3";
+
+                    m.Projections.Add<LetterCountsProjection>(ProjectionLifecycle.Async);
+                }).AddAsyncDaemon(DaemonMode.Solo).IntegrateWithWolverine();
+
+                opts.Services.AddMartenStore<ILetterStore>(m =>
+                {
+                    m.DisableNpgsqlLogging = true;
+
+                    m.Connection(Servers.PostgresConnectionString);
+                    m.DatabaseSchemaName = "letters4";
+
+                    m.Projections.Add<LetterCountsProjection>(ProjectionLifecycle.Async);
+                }).IntegrateWithWolverine();
+
+                opts.Durability.Mode = DurabilityMode.Solo;
+            }).StartAsync();
+    }
+
+    public async Task DisposeAsync()
+    {
+        await _host.StopAsync();
+    }
+
+    [Fact]
+    public async Task with_main_store()
+    {
+        await _host.ResetAllMartenDataAsync();
+
+        var id = Guid.NewGuid();
+
+        // Setting up some other aggregates first
+        using var session = _host.DocumentStore().LightweightSession();
+        session.Events.StartStream<LetterCounts>("AABBCCDDEE".ToLetterEvents());
+        session.Events.StartStream<LetterCounts>("AABBCCDDEE".ToLetterEvents());
+        session.Events.StartStream<LetterCounts>("AABBCCDDEE".ToLetterEvents());
+        await session.SaveChangesAsync();
+        await _host.WaitForNonStaleProjectionDataAsync(5.Seconds());
+
+        (await session.Query<LetterCounts>().CountAsync()).ShouldBe(3);
+
+
+        var tracked = await _host.TrackActivity()
+            .ResetAllMartenDataFirst()
+            .PauseThenCatchUpOnMartenDaemonActivity(CatchUpMode.AndDoNothing)
+            .InvokeMessageAndWaitAsync(new AppendLetters(id, ["AAAACCCCBDEEE", "ABCDECCC", "BBBA", "DDDAE"]));
+
+        // Proving that previous data was wiped out
+
+        var all = await session.Query<LetterCounts>().ToListAsync();
+        var counts = all.Single();
+        counts.Id.ShouldBe(id);
+
+        counts.ACount.ShouldBe(7);
+        counts.BCount.ShouldBe(5);
+        counts.CCount.ShouldBe(8);
+    }
+
+    [Fact]
+    public async Task with_ancillary_store()
+    {
+        await _host.ResetAllMartenDataAsync<ILetterStore>();
+
+        var id = Guid.NewGuid();
+
+        // Setting up some other aggregates first
+        using var session = _host.DocumentStore<ILetterStore>().LightweightSession();
+        session.Events.StartStream<LetterCounts>("AABBCCDDEE".ToLetterEvents());
+        session.Events.StartStream<LetterCounts>("AABBCCDDEE".ToLetterEvents());
+        session.Events.StartStream<LetterCounts>("AABBCCDDEE".ToLetterEvents());
+        await session.SaveChangesAsync();
+        await _host.WaitForNonStaleProjectionDataAsync<ILetterStore>(5.Seconds());
+
+        (await session.Query<LetterCounts>().CountAsync()).ShouldBe(3);
+
+
+        var tracked = await _host.TrackActivity()
+            .ResetAllMartenDataFirst<ILetterStore>()
+            .PauseThenCatchUpOnMartenDaemonActivity<ILetterStore>(CatchUpMode.AndDoNothing)
+            .InvokeMessageAndWaitAsync(new AppendLetters2(id, ["AAAACCCCBDEEE", "ABCDECCC", "BBBA", "DDDAE"]));
+
+        // Proving that previous data was wiped out
+
+        var all = await session.Query<LetterCounts>().ToListAsync();
+        var counts = all.Single();
+        counts.Id.ShouldBe(id);
+
+        counts.ACount.ShouldBe(7);
+        counts.BCount.ShouldBe(5);
+        counts.CCount.ShouldBe(8);
+    }
+}