Add store agnostic AppendEvents/StartStream side effects

docs/guide/handlers/persistence.md

@@ -19,6 +19,7 @@ These all speak one vocabulary, and none of it names your database:
 | The whole method to be an event sourced command handler | `[DeciderFunction]` |
 | An event sourced model spanning several streams, matched by tag | `[DcbModel]` |
 | To write a document back | `Storage.Store` / `Insert` / `Update` / `Delete` / `Nothing<T>` |
+| To append events to a stream | [`Storage.AppendEvents` / `Storage.StartStream`](/guide/handlers/side-effects#event-side-effects) |
 
 ## Automatically Loading Entities to Method Parameters <Badge type="tip" text="3.6" />
 

docs/guide/handlers/side-effects.md

@@ -260,3 +260,85 @@ public static class StoreManyHandler
 
 The `UnitOfWork<T>` is really just a `List<IStorageAction<T>>` that can relay zero to many storage
 actions to your underlying persistence tooling. 
+
+## Event Side Effects <Badge type="tip" text="6.28" />
+
+`Storage.Store()` and friends write *documents*. Their counterparts for an *event stream* are
+`Storage.StartStream()` and `Storage.AppendEvents()`, and they work the same way — return one from a
+handler or HTTP endpoint and Wolverine relays it to your event store:
+
+```cs
+public static class InvoiceHandler
+{
+    // Notice there is no IDocumentSession anywhere in this class
+    public static StartStream Handle(CreateInvoice command)
+        => Storage.StartStream(command.Id, new InvoiceCreated(command.Amount));
+
+    public static AppendEvents Handle(ApproveInvoice command)
+        => Storage.AppendEvents(command.Id, new InvoiceApproved(command.ApprovedBy));
+}
+```
+
+That handler is a pure function of its input, trivially unit testable without a database, and — the
+actual point — **valid against Marten, Polecat, or Fisher without changing a line**. The work is
+expressed entirely in terms of `JasperFx.Events.IEventOperations`, the shared write-side event API all
+three implement, and each store's `IntegrateWithWolverine()` registers the variable source that hands
+Wolverine the right `IEventOperations` for the active session.
+
+Both accept streams identified by `Guid` or by string key, and several events at once:
+
+```cs
+Storage.AppendEvents(streamId, new ItemReady("shoes"), new OrderReady());
+Storage.AppendEvents("order-1234", new OrderShipped());
+
+// Start a stream for a known aggregate type
+Storage.StartStream<Order>(streamId, new OrderCreated(items));
+
+// Optimistic concurrency -- abort if the stream has moved on from this version
+Storage.AppendEvents(streamId, expectedVersion: 3, new OrderShipped());
+```
+
+An `AppendEvents` carrying no events is a deliberate no-op rather than an empty write, so a decision
+function is free to conclude that nothing happened:
+
+```cs
+public static AppendEvents Handle(MaybeApproveInvoice command)
+    => command.Approve
+        ? Storage.AppendEvents(command.Id, new InvoiceApproved("approver"))
+        : Storage.AppendEvents(command.Id);
+```
+
+Like storage actions, these compose with tuple returns, so a handler can append events *and* cascade a
+message:
+
+```cs
+public static (AppendEvents, InvoiceApprovalNoticed) Handle(ApproveInvoiceAndNotify command)
+    => (Storage.AppendEvents(command.Id, new InvoiceApproved(command.ApprovedBy)),
+        new InvoiceApprovalNoticed(command.Id));
+```
+
+### Ancillary Stores
+
+Returning one of these from a handler marked with `[Storage(typeof(IMyStore))]` appends to that
+*ancillary* store rather than the application's primary one — no extra configuration. The attribute
+swaps the session the chain resolves, and these side effects follow it.
+
+```cs
+[Storage(typeof(ICritterWatchStore))]
+public static AppendEvents Handle(ApproveInvoice command)
+    => Storage.AppendEvents(command.Id, new InvoiceApproved(command.ApprovedBy));
+```
+
+::: tip
+Appending through the event store only *queues* the work into that store's unit of work — something
+still has to commit it. Wolverine enrolls the chain in the event store's transaction for you when it
+sees one of these return values, so the events commit together with any outgoing messages through the
+outbox. You do not need `[Transactional]` or `AutoApplyTransactions()` for this to work.
+:::
+
+::: warning
+These require an event store. An application with no Marten, Polecat, or Fisher integration fails at
+bootstrapping time with an error naming the offending handler, rather than a codegen failure that
+says nothing about the real mistake. If more than one event store is registered, mark the handler with
+`[Storage(typeof(IYourStore))]` to say which one you mean.
+:::

src/Persistence/FisherTests/event_side_effects.cs

@@ -0,0 +1,98 @@
+using Fisher;
+using JasperFx;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine;
+using Wolverine.Fisher;
+using Wolverine.Attributes;
+using Wolverine.Persistence;
+using Wolverine.Tracking;
+
+namespace FisherTests;
+
+// The Fisher half of the store agnostic event side effects -- the handler below is character for character
+// what the Marten and Polecat suites run.
+public class event_side_effects : IAsyncLifetime
+{
+    private FisherTestDatabase theDatabase = null!;
+    private IHost _host = null!;
+
+    public async ValueTask InitializeAsync()
+    {
+        theDatabase = Servers.CreateDatabase("event_side_effects");
+
+        _host = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(FiInvoiceHandler));
+                opts.Durability.Mode = DurabilityMode.Solo;
+                opts.Policies.AutoApplyTransactions();
+
+                opts.Services.AddFisher(m =>
+                    {
+                        m.Connection(theDatabase.ConnectionString);
+                        m.AutoCreateSchemaObjects = AutoCreate.All;
+                    })
+                    .ApplyAllDatabaseChangesOnStartup()
+                    .IntegrateWithWolverine();
+            }).StartAsync();
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _host.StopAsync();
+        _host.Dispose();
+        theDatabase.Dispose();
+    }
+
+    [Fact]
+    public async Task start_stream_creates_the_stream_and_its_events()
+    {
+        var id = Guid.NewGuid();
+
+        await _host.InvokeMessageAndWaitAsync(new CreateFiInvoice(id, 100));
+
+        await using var session = _host.Services.GetRequiredService<IDocumentStore>().LightweightSession();
+        var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken);
+
+        events.Count.ShouldBe(1);
+        events[0].Data.ShouldBeOfType<FiInvoiceCreated>().Amount.ShouldBe(100);
+    }
+
+    [Fact]
+    public async Task append_events_adds_to_an_existing_stream()
+    {
+        var id = Guid.NewGuid();
+        await _host.InvokeMessageAndWaitAsync(new CreateFiInvoice(id, 100));
+
+        await _host.InvokeMessageAndWaitAsync(new ApproveFiInvoice(id, "kareem"));
+
+        await using var session = _host.Services.GetRequiredService<IDocumentStore>().LightweightSession();
+        var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken);
+
+        events.Count.ShouldBe(2);
+        events[1].Data.ShouldBeOfType<FiInvoiceApproved>().ApprovedBy.ShouldBe("kareem");
+    }
+}
+
+public record FiInvoiceCreated(decimal Amount);
+
+public record FiInvoiceApproved(string ApprovedBy);
+
+public record CreateFiInvoice(Guid Id, decimal Amount);
+
+public record ApproveFiInvoice(Guid Id, string ApprovedBy);
+
+// [WolverineIgnore] because these throw at BOOTSTRAP when no event store is registered, and this is a
+// shared test assembly: conventional discovery in any other host here -- an in-memory saga host, say --
+// would find them and fail that host's startup. The tests above include them explicitly instead.
+[WolverineIgnore]
+public static class FiInvoiceHandler
+{
+    public static StartStream Handle(CreateFiInvoice command)
+        => Storage.StartStream(command.Id, new FiInvoiceCreated(command.Amount));
+
+    public static AppendEvents Handle(ApproveFiInvoice command)
+        => Storage.AppendEvents(command.Id, new FiInvoiceApproved(command.ApprovedBy));
+}

src/Persistence/MartenTests/AncillaryStores/event_side_effects_against_an_ancillary_store.cs

@@ -0,0 +1,131 @@
+using IntegrationTests;
+using JasperFx.Resources;
+using Marten;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Shouldly;
+using Wolverine;
+using Wolverine.Marten;
+using Wolverine.Attributes;
+using Wolverine.Persistence;
+using Wolverine.Tracking;
+
+namespace MartenTests.AncillaryStores;
+
+/// <summary>
+///     Storage.AppendEvents() / Storage.StartStream() have to reach an <b>ancillary</b> event store, not just
+///     the application's primary one. This is the case that matters for CritterWatch, which registers its own
+///     store rather than the host application's.
+/// </summary>
+/// <remarks>
+///     The mechanism is worth stating because it is why this works with no extra plumbing: the shared
+///     IEventOperations variable source resolves whichever <c>IDocumentSession</c> the chain has, and
+///     <c>[Storage(typeof(IEventSideEffectStore))]</c> has already swapped that session for the ancillary
+///     store's outbox-enrolled one at the front of the chain's middleware. Asserting the negative -- that the
+///     events did NOT land in the main store -- is the point; a side effect that silently fell back to the
+///     primary store would still satisfy a positive-only assertion against the ancillary one.
+/// </remarks>
+public class event_side_effects_against_an_ancillary_store : IAsyncLifetime
+{
+    private IHost theHost = null!;
+
+    public async ValueTask InitializeAsync()
+    {
+        theHost = await Host.CreateDefaultBuilder()
+            .UseWolverine(opts =>
+            {
+                opts.Durability.MessageStorageSchemaName = "wolverine";
+                opts.Durability.Mode = DurabilityMode.Solo;
+                opts.Policies.AutoApplyTransactions();
+
+                opts.Services.AddMarten(m =>
+                {
+                    m.Connection(Servers.PostgresConnectionString);
+                    m.DatabaseSchemaName = "evt_side_effect_main";
+                    m.Events.DatabaseSchemaName = "evt_side_effect_main";
+                }).IntegrateWithWolverine();
+
+                opts.Services.AddMartenStore<IEventSideEffectStore>(m =>
+                {
+                    m.Connection(Servers.PostgresConnectionString);
+                    m.DatabaseSchemaName = "evt_side_effect_ancillary";
+                    m.Events.DatabaseSchemaName = "evt_side_effect_ancillary";
+                }).IntegrateWithWolverine();
+
+                opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(AncillaryInvoiceHandler));
+
+                opts.Services.AddResourceSetupOnStartup();
+            }).StartAsync();
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await theHost.StopAsync();
+        theHost.Dispose();
+    }
+
+    [Fact]
+    public async Task start_stream_lands_in_the_targeted_store_and_not_the_main_one()
+    {
+        var id = Guid.NewGuid();
+
+        await theHost.InvokeMessageAndWaitAsync(new CreateAncillaryInvoice(id, 250));
+
+        // ...landed in the ancillary store
+        var ancillary = theHost.Services.GetRequiredService<IEventSideEffectStore>();
+        await using (var session = ancillary.LightweightSession())
+        {
+            var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken);
+            events.Count.ShouldBe(1);
+            events[0].Data.ShouldBeOfType<AncillaryInvoiceCreated>().Amount.ShouldBe(250);
+        }
+
+        // ...and demonstrably NOT in the main store
+        await using (var main = theHost.DocumentStore().LightweightSession())
+        {
+            var events = await main.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken);
+            events.ShouldBeEmpty();
+        }
+    }
+
+    [Fact]
+    public async Task append_events_also_routes_to_the_targeted_store()
+    {
+        var id = Guid.NewGuid();
+        await theHost.InvokeMessageAndWaitAsync(new CreateAncillaryInvoice(id, 250));
+
+        await theHost.InvokeMessageAndWaitAsync(new ApproveAncillaryInvoice(id, "petrovic"));
+
+        var ancillary = theHost.Services.GetRequiredService<IEventSideEffectStore>();
+        await using var session = ancillary.LightweightSession();
+        var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken);
+
+        events.Count.ShouldBe(2);
+        events[1].Data.ShouldBeOfType<AncillaryInvoiceApproved>().ApprovedBy.ShouldBe("petrovic");
+    }
+}
+
+public interface IEventSideEffectStore : IDocumentStore;
+
+public record AncillaryInvoiceCreated(decimal Amount);
+
+public record AncillaryInvoiceApproved(string ApprovedBy);
+
+public record CreateAncillaryInvoice(Guid Id, decimal Amount);
+
+public record ApproveAncillaryInvoice(Guid Id, string ApprovedBy);
+
+// [WolverineIgnore] because these throw at BOOTSTRAP when no event store is registered, and this is a
+// shared test assembly: conventional discovery in any other host here -- an in-memory saga host, say --
+// would find them and fail that host's startup. The tests above include them explicitly instead.
+[WolverineIgnore]
+public static class AncillaryInvoiceHandler
+{
+    [Storage(typeof(IEventSideEffectStore))]
+    public static StartStream Handle(CreateAncillaryInvoice command)
+        => Storage.StartStream(command.Id, new AncillaryInvoiceCreated(command.Amount));
+
+    [Storage(typeof(IEventSideEffectStore))]
+    public static AppendEvents Handle(ApproveAncillaryInvoice command)
+        => Storage.AppendEvents(command.Id, new AncillaryInvoiceApproved(command.ApprovedBy));
+}