diff --git a/docs/guide/handlers/persistence.md b/docs/guide/handlers/persistence.md index 8cee9af0d..3e7d69ed1 100644 --- a/docs/guide/handlers/persistence.md +++ b/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` | +| To append events to a stream | [`Storage.AppendEvents` / `Storage.StartStream`](/guide/handlers/side-effects#event-side-effects) | ## Automatically Loading Entities to Method Parameters diff --git a/docs/guide/handlers/side-effects.md b/docs/guide/handlers/side-effects.md index d88532d3f..d6a13b975 100644 --- a/docs/guide/handlers/side-effects.md +++ b/docs/guide/handlers/side-effects.md @@ -260,3 +260,85 @@ public static class StoreManyHandler The `UnitOfWork` is really just a `List>` that can relay zero to many storage actions to your underlying persistence tooling. + +## Event Side Effects + +`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(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. +::: diff --git a/src/Persistence/FisherTests/event_side_effects.cs b/src/Persistence/FisherTests/event_side_effects.cs new file mode 100644 index 000000000..df00dc5f9 --- /dev/null +++ b/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().LightweightSession(); + var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken); + + events.Count.ShouldBe(1); + events[0].Data.ShouldBeOfType().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().LightweightSession(); + var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken); + + events.Count.ShouldBe(2); + events[1].Data.ShouldBeOfType().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)); +} diff --git a/src/Persistence/MartenTests/AncillaryStores/event_side_effects_against_an_ancillary_store.cs b/src/Persistence/MartenTests/AncillaryStores/event_side_effects_against_an_ancillary_store.cs new file mode 100644 index 000000000..7c7ab878e --- /dev/null +++ b/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; + +/// +/// Storage.AppendEvents() / Storage.StartStream() have to reach an ancillary 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. +/// +/// +/// The mechanism is worth stating because it is why this works with no extra plumbing: the shared +/// IEventOperations variable source resolves whichever IDocumentSession the chain has, and +/// [Storage(typeof(IEventSideEffectStore))] 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. +/// +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(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(); + 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().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(); + 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().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)); +} diff --git a/src/Persistence/MartenTests/event_side_effects.cs b/src/Persistence/MartenTests/event_side_effects.cs new file mode 100644 index 000000000..8b86f06ef --- /dev/null +++ b/src/Persistence/MartenTests/event_side_effects.cs @@ -0,0 +1,170 @@ +using IntegrationTests; +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; + +// Storage.AppendEvents() / Storage.StartStream() are store agnostic side effects expressed purely against +// JasperFx.Events' IEventOperations. This is the Marten proof; PolecatTests and FisherTests run the same +// handlers against their own stores. +public class event_side_effects : IAsyncLifetime +{ + private IHost _host = null!; + + public async ValueTask InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(InvoiceHandler)); + opts.Durability.Mode = DurabilityMode.Solo; + opts.Policies.AutoApplyTransactions(); + opts.Services.AddMarten(m => + { + m.DisableNpgsqlLogging = true; + m.Connection(Servers.PostgresConnectionString); + m.DatabaseSchemaName = "event_side_effects"; + }).IntegrateWithWolverine().UseLightweightSessions(); + }).StartAsync(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + [Fact] + public async Task start_stream_creates_the_stream_and_its_events() + { + var id = Guid.NewGuid(); + + await _host.InvokeMessageAndWaitAsync(new CreateInvoice(id, 100)); + + await using var session = _host.DocumentStore().LightweightSession(); + var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken); + + events.Count.ShouldBe(1); + events[0].Data.ShouldBeOfType().Amount.ShouldBe(100); + } + + [Fact] + public async Task append_events_adds_to_an_existing_stream() + { + var id = Guid.NewGuid(); + await _host.InvokeMessageAndWaitAsync(new CreateInvoice(id, 100)); + + await _host.InvokeMessageAndWaitAsync(new ApproveInvoice(id, "kareem")); + + await using var session = _host.DocumentStore().LightweightSession(); + var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken); + + events.Count.ShouldBe(2); + events[1].Data.ShouldBeOfType().ApprovedBy.ShouldBe("kareem"); + } + + [Fact] + public async Task append_several_events_at_once_keeps_their_order() + { + var id = Guid.NewGuid(); + await _host.InvokeMessageAndWaitAsync(new CreateInvoice(id, 100)); + + await _host.InvokeMessageAndWaitAsync(new CloseInvoice(id)); + + await using var session = _host.DocumentStore().LightweightSession(); + var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken); + + events.Count.ShouldBe(3); + events[1].Data.ShouldBeOfType(); + events[2].Data.ShouldBeOfType(); + } + + [Fact] + public async Task an_empty_append_is_a_no_op_rather_than_an_empty_stream_action() + { + var id = Guid.NewGuid(); + await _host.InvokeMessageAndWaitAsync(new CreateInvoice(id, 100)); + + // A decision function that concludes "nothing to do" is a legitimate outcome, and must not + // hand the store a stream action carrying no events + await _host.InvokeMessageAndWaitAsync(new MaybeApproveInvoice(id, false)); + + await using var session = _host.DocumentStore().LightweightSession(); + var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken); + + events.Count.ShouldBe(1); + } + + [Fact] + public async Task works_as_part_of_a_tuple_return() + { + var id = Guid.NewGuid(); + await _host.InvokeMessageAndWaitAsync(new CreateInvoice(id, 100)); + + var tracked = await _host.InvokeMessageAndWaitAsync(new ApproveInvoiceAndNotify(id, "sabonis")); + + // the cascaded message went out... + tracked.Sent.SingleMessage().InvoiceId.ShouldBe(id); + + // ...and the side effect in the same tuple still appended + await using var session = _host.DocumentStore().LightweightSession(); + var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken); + events.Count.ShouldBe(2); + events[1].Data.ShouldBeOfType().ApprovedBy.ShouldBe("sabonis"); + } +} + +public record InvoiceCreated(decimal Amount); + +public record InvoiceApproved(string ApprovedBy); + +public record InvoiceClosed; + +public record CreateInvoice(Guid Id, decimal Amount); + +public record ApproveInvoice(Guid Id, string ApprovedBy); + +public record CloseInvoice(Guid Id); + +public record MaybeApproveInvoice(Guid Id, bool Approve); + +public record ApproveInvoiceAndNotify(Guid Id, string ApprovedBy); + +public record InvoiceApprovalNoticed(Guid InvoiceId); + +// [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 InvoiceHandler +{ + // No IDocumentSession anywhere in this class -- that is the whole point + 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)); + + public static AppendEvents Handle(CloseInvoice command) + => Storage.AppendEvents(command.Id, new InvoiceApproved("auto"), new InvoiceClosed()); + + public static AppendEvents Handle(MaybeApproveInvoice command) + => command.Approve + ? Storage.AppendEvents(command.Id, new InvoiceApproved("maybe")) + : Storage.AppendEvents(command.Id); + + public static (AppendEvents, InvoiceApprovalNoticed) Handle(ApproveInvoiceAndNotify command) + => (Storage.AppendEvents(command.Id, new InvoiceApproved(command.ApprovedBy)), + new InvoiceApprovalNoticed(command.Id)); + + public static void Handle(InvoiceApprovalNoticed msg) + { + } +} diff --git a/src/Persistence/PolecatTests/event_side_effects.cs b/src/Persistence/PolecatTests/event_side_effects.cs new file mode 100644 index 000000000..6faaf65d1 --- /dev/null +++ b/src/Persistence/PolecatTests/event_side_effects.cs @@ -0,0 +1,94 @@ +using IntegrationTests; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Polecat; +using Shouldly; +using Wolverine; +using Wolverine.Attributes; +using Wolverine.Persistence; +using Wolverine.Polecat; +using Wolverine.Tracking; + +namespace PolecatTests; + +// The Polecat half of the store agnostic event side effects -- the handler below is character for character +// what the Marten and Fisher suites run. +public class event_side_effects : IAsyncLifetime +{ + private IHost _host = null!; + + public async ValueTask InitializeAsync() + { + _host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(PcInvoiceHandler)); + opts.Durability.Mode = DurabilityMode.Solo; + opts.Policies.AutoApplyTransactions(); + opts.Services.AddPolecat(m => + { + m.ConnectionString = Servers.SqlServerConnectionString; + m.DatabaseSchemaName = "pc_event_side_effects"; + }).IntegrateWithWolverine(); + }).StartAsync(); + + var store = (DocumentStore)_host.Services.GetRequiredService(); + await store.Database.ApplyAllConfiguredChangesToDatabaseAsync(); + } + + public async ValueTask DisposeAsync() + { + await _host.StopAsync(); + _host.Dispose(); + } + + [Fact] + public async Task start_stream_creates_the_stream_and_its_events() + { + var id = Guid.NewGuid(); + + await _host.InvokeMessageAndWaitAsync(new CreatePcInvoice(id, 100)); + + await using var session = _host.Services.GetRequiredService().LightweightSession(); + var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken); + + events.Count.ShouldBe(1); + events[0].Data.ShouldBeOfType().Amount.ShouldBe(100); + } + + [Fact] + public async Task append_events_adds_to_an_existing_stream() + { + var id = Guid.NewGuid(); + await _host.InvokeMessageAndWaitAsync(new CreatePcInvoice(id, 100)); + + await _host.InvokeMessageAndWaitAsync(new ApprovePcInvoice(id, "kareem")); + + await using var session = _host.Services.GetRequiredService().LightweightSession(); + var events = await session.Events.FetchStreamAsync(id, token: TestContext.Current.CancellationToken); + + events.Count.ShouldBe(2); + events[1].Data.ShouldBeOfType().ApprovedBy.ShouldBe("kareem"); + } +} + +public record PcInvoiceCreated(decimal Amount); + +public record PcInvoiceApproved(string ApprovedBy); + +public record CreatePcInvoice(Guid Id, decimal Amount); + +public record ApprovePcInvoice(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 PcInvoiceHandler +{ + public static StartStream Handle(CreatePcInvoice command) + => Storage.StartStream(command.Id, new PcInvoiceCreated(command.Amount)); + + public static AppendEvents Handle(ApprovePcInvoice command) + => Storage.AppendEvents(command.Id, new PcInvoiceApproved(command.ApprovedBy)); +} diff --git a/src/Persistence/Wolverine.Fisher/Codegen/SessionVariableSource.cs b/src/Persistence/Wolverine.Fisher/Codegen/SessionVariableSource.cs index b4d3cd369..cb1794f00 100644 --- a/src/Persistence/Wolverine.Fisher/Codegen/SessionVariableSource.cs +++ b/src/Persistence/Wolverine.Fisher/Codegen/SessionVariableSource.cs @@ -93,3 +93,50 @@ public override void GenerateCode(GeneratedMethod method, ISourceWriter writer) Next?.GenerateCode(method, writer); } } + +/// +/// Supplies the shared contract rather than +/// Fisher's own derived one, so that Wolverine's store agnostic Storage.AppendEvents() / +/// Storage.StartStream() side effects resolve against any registered event store. The sibling +/// EventOperationsSource above stays as it is because a handler asking for Fisher's own type +/// must keep getting a variable of exactly that type. +/// +internal class SharedEventOperationsSource : IVariableSource +{ + public bool Matches(Type type) + { + return type == typeof(JasperFx.Events.IEventOperations); + } + + public Variable Create(Type type) + { + return new SharedEventOperationsFrame().Variable; + } +} + +internal class SharedEventOperationsFrame : SyncFrame +{ + private Variable _session = null!; + + public SharedEventOperationsFrame() + { + Variable = new Variable(typeof(JasperFx.Events.IEventOperations), this); + } + + public Variable Variable { get; } + + public override IEnumerable FindVariables(IMethodVariables chain) + { + // Deliberately the session rather than a store: an ancillary store's [Storage] frame has already + // swapped which IDocumentSession this chain resolves, so this follows it for free. + _session = chain.FindVariable(typeof(IDocumentSession)); + yield return _session; + } + + public override void GenerateCode(GeneratedMethod method, ISourceWriter writer) + { + writer.Write( + $"{typeof(JasperFx.Events.IEventOperations)} {Variable.Usage} = {_session.Usage}.{nameof(IDocumentSession.Events)};"); + Next?.GenerateCode(method, writer); + } +} diff --git a/src/Persistence/Wolverine.Fisher/FisherIntegration.cs b/src/Persistence/Wolverine.Fisher/FisherIntegration.cs index 4c8ad08e1..fd8838bfe 100644 --- a/src/Persistence/Wolverine.Fisher/FisherIntegration.cs +++ b/src/Persistence/Wolverine.Fisher/FisherIntegration.cs @@ -50,6 +50,7 @@ public void Configure(WolverineOptions options) options.CodeGeneration.Sources.Add(new SessionVariableSource()); options.CodeGeneration.Sources.Add(new DocumentOperationsSource()); options.CodeGeneration.Sources.Add(new EventOperationsSource()); + options.CodeGeneration.Sources.Add(new SharedEventOperationsSource()); options.Policies.Add(); diff --git a/src/Persistence/Wolverine.Marten/Codegen/SessionVariableSource.cs b/src/Persistence/Wolverine.Marten/Codegen/SessionVariableSource.cs index cb6fee2ff..3930a8dd9 100644 --- a/src/Persistence/Wolverine.Marten/Codegen/SessionVariableSource.cs +++ b/src/Persistence/Wolverine.Marten/Codegen/SessionVariableSource.cs @@ -96,4 +96,51 @@ public override void GenerateCode(GeneratedMethod method, ISourceWriter writer) writer.Write($"{typeof(IEventStoreOperations)} {Variable.Usage} = {_session.Usage}.{nameof(IDocumentSession.Events)};"); Next?.GenerateCode(method, writer); } -} \ No newline at end of file +} + +/// +/// Supplies the shared contract rather than +/// Marten's own derived one, so that Wolverine's store agnostic Storage.AppendEvents() / +/// Storage.StartStream() side effects resolve against any registered event store. The sibling +/// EventOperationsSource above stays as it is because a handler asking for Marten's own type +/// must keep getting a variable of exactly that type. +/// +internal class SharedEventOperationsSource : IVariableSource +{ + public bool Matches(Type type) + { + return type == typeof(JasperFx.Events.IEventOperations); + } + + public Variable Create(Type type) + { + return new SharedEventOperationsFrame().Variable; + } +} + +internal class SharedEventOperationsFrame : SyncFrame +{ + private Variable _session = null!; + + public SharedEventOperationsFrame() + { + Variable = new Variable(typeof(JasperFx.Events.IEventOperations), this); + } + + public Variable Variable { get; } + + public override IEnumerable FindVariables(IMethodVariables chain) + { + // Deliberately the session rather than a store: an ancillary store's [Storage] frame has already + // swapped which IDocumentSession this chain resolves, so this follows it for free. + _session = chain.FindVariable(typeof(IDocumentSession)); + yield return _session; + } + + public override void GenerateCode(GeneratedMethod method, ISourceWriter writer) + { + writer.Write( + $"{typeof(JasperFx.Events.IEventOperations)} {Variable.Usage} = {_session.Usage}.{nameof(IDocumentSession.Events)};"); + Next?.GenerateCode(method, writer); + } +} diff --git a/src/Persistence/Wolverine.Marten/MartenIntegration.cs b/src/Persistence/Wolverine.Marten/MartenIntegration.cs index a00fdf3e1..67e128c84 100644 --- a/src/Persistence/Wolverine.Marten/MartenIntegration.cs +++ b/src/Persistence/Wolverine.Marten/MartenIntegration.cs @@ -73,6 +73,7 @@ public void Configure(WolverineOptions options) options.CodeGeneration.Sources.Add(new SessionVariableSource()); options.CodeGeneration.Sources.Add(new DocumentOperationsSource()); options.CodeGeneration.Sources.Add(new EventStoreOperationsSource()); + options.CodeGeneration.Sources.Add(new SharedEventOperationsSource()); options.Policies.Add(); diff --git a/src/Persistence/Wolverine.Polecat/Codegen/SessionVariableSource.cs b/src/Persistence/Wolverine.Polecat/Codegen/SessionVariableSource.cs index c649142dc..349f088ef 100644 --- a/src/Persistence/Wolverine.Polecat/Codegen/SessionVariableSource.cs +++ b/src/Persistence/Wolverine.Polecat/Codegen/SessionVariableSource.cs @@ -93,3 +93,50 @@ public override void GenerateCode(GeneratedMethod method, ISourceWriter writer) Next?.GenerateCode(method, writer); } } + +/// +/// Supplies the shared contract rather than +/// Polecat's own derived one, so that Wolverine's store agnostic Storage.AppendEvents() / +/// Storage.StartStream() side effects resolve against any registered event store. The sibling +/// EventOperationsSource above stays as it is because a handler asking for Polecat's own type +/// must keep getting a variable of exactly that type. +/// +internal class SharedEventOperationsSource : IVariableSource +{ + public bool Matches(Type type) + { + return type == typeof(JasperFx.Events.IEventOperations); + } + + public Variable Create(Type type) + { + return new SharedEventOperationsFrame().Variable; + } +} + +internal class SharedEventOperationsFrame : SyncFrame +{ + private Variable _session = null!; + + public SharedEventOperationsFrame() + { + Variable = new Variable(typeof(JasperFx.Events.IEventOperations), this); + } + + public Variable Variable { get; } + + public override IEnumerable FindVariables(IMethodVariables chain) + { + // Deliberately the session rather than a store: an ancillary store's [Storage] frame has already + // swapped which IDocumentSession this chain resolves, so this follows it for free. + _session = chain.FindVariable(typeof(IDocumentSession)); + yield return _session; + } + + public override void GenerateCode(GeneratedMethod method, ISourceWriter writer) + { + writer.Write( + $"{typeof(JasperFx.Events.IEventOperations)} {Variable.Usage} = {_session.Usage}.{nameof(IDocumentSession.Events)};"); + Next?.GenerateCode(method, writer); + } +} diff --git a/src/Persistence/Wolverine.Polecat/PolecatIntegration.cs b/src/Persistence/Wolverine.Polecat/PolecatIntegration.cs index bef5ef556..45dd5898f 100644 --- a/src/Persistence/Wolverine.Polecat/PolecatIntegration.cs +++ b/src/Persistence/Wolverine.Polecat/PolecatIntegration.cs @@ -55,6 +55,7 @@ public void Configure(WolverineOptions options) options.CodeGeneration.Sources.Add(new SessionVariableSource()); options.CodeGeneration.Sources.Add(new DocumentOperationsSource()); options.CodeGeneration.Sources.Add(new EventOperationsSource()); + options.CodeGeneration.Sources.Add(new SharedEventOperationsSource()); options.Policies.Add(); diff --git a/src/Testing/CoreTests/Persistence/event_side_effect_validation.cs b/src/Testing/CoreTests/Persistence/event_side_effect_validation.cs new file mode 100644 index 000000000..059db225b --- /dev/null +++ b/src/Testing/CoreTests/Persistence/event_side_effect_validation.cs @@ -0,0 +1,68 @@ +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine.Attributes; +using Wolverine.Persistence; +using Xunit; + +namespace CoreTests.Persistence; + +// Storage.AppendEvents() / Storage.StartStream() need an event store. Without one the failure used to be a +// raw codegen "cannot determine how to build variable of type IEventOperations" deep in startup, which says +// nothing about the actual mistake. These pin the helpful message instead. +public class event_side_effect_validation +{ + [Fact] + public async Task helpful_error_when_appending_with_no_event_store_registered() + { + var ex = await Should.ThrowAsync(async () => + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(NoStoreAppendHandler)); + }).StartAsync(); + }); + + ex.Message.ShouldContain("no registered event store"); + ex.Message.ShouldContain("IntegrateWithWolverine()"); + ex.Message.ShouldContain(nameof(AppendEvents)); + } + + [Fact] + public async Task helpful_error_when_starting_a_stream_with_no_event_store_registered() + { + var ex = await Should.ThrowAsync(async () => + { + using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(NoStoreStartHandler)); + }).StartAsync(); + }); + + ex.Message.ShouldContain("no registered event store"); + ex.Message.ShouldContain(nameof(StartStream)); + } +} + +// These deliberately break bootstrapping, so they must never be found by the conventional discovery +// that every other CoreTests host runs -- otherwise this file fails 500+ unrelated tests. +public record AppendSomething(Guid Id); + +public record StartSomething(Guid Id); + +public record SomethingHappened; + +[WolverineIgnore] +public static class NoStoreAppendHandler +{ + public static AppendEvents Handle(AppendSomething command) + => Storage.AppendEvents(command.Id, new SomethingHappened()); +} + +[WolverineIgnore] +public static class NoStoreStartHandler +{ + public static StartStream Handle(StartSomething command) + => Storage.StartStream(command.Id, new SomethingHappened()); +} diff --git a/src/Wolverine/Persistence/EventSideEffects.cs b/src/Wolverine/Persistence/EventSideEffects.cs new file mode 100644 index 000000000..316a88aee --- /dev/null +++ b/src/Wolverine/Persistence/EventSideEffects.cs @@ -0,0 +1,255 @@ +using System.Diagnostics.CodeAnalysis; +using System.Reflection; +using JasperFx; +using JasperFx.CodeGeneration; +using JasperFx.CodeGeneration.Frames; +using JasperFx.CodeGeneration.Model; +using JasperFx.Core.Reflection; +using JasperFx.Events; +using Wolverine.Configuration; +using Wolverine.Persistence.EventSourcing; +using Wolverine.Persistence.Sagas; + +namespace Wolverine.Persistence; + +/// +/// Shared codegen for and . +/// +internal static class EventSideEffectFrames +{ + /// + /// Build the frame that invokes the side effect's Execute(IEventOperations), having first enrolled + /// the chain in the event store's transaction. + /// + /// + /// The transaction enrollment is the reason these implement rather than + /// plain . Appending through IEventOperations only queues the work into + /// the store's unit of work; something still has to call SaveChangesAsync. AutoApplyTransactions + /// will not do it, because it only fires when a provider's CanApply recognizes a store type in the + /// chain -- and the entire point of these side effects is that a store agnostic handler names no store + /// type at all. Without this the events were silently queued and never committed. + /// + public static Frame BuildFrame( + [DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicMethods)] + Type effectType, IChain chain, Variable variable, GenerationRules rules, + IServiceContainer container) + { + var provider = findEventStoreProvider(effectType, chain, rules, container); + provider.ApplyTransactionSupport(chain, container); + + var method = effectType.GetMethod(nameof(AppendEvents.Execute), + BindingFlags.Public | BindingFlags.Instance)!; + + return new IfElseNullGuardFrame.IfNullGuardFrame(variable, + new MethodCall(effectType, method) + { + Target = variable, + CommentText = "Append events through the registered event store" + }); + } + + private static IPersistenceFrameProvider findEventStoreProvider(Type effectType, IChain chain, + GenerationRules rules, IServiceContainer container) + { + // An event side effect needs an *event* store, so narrow to the providers that also implement the + // event sourcing seam. Marten, Polecat and Fisher each implement both interfaces on one class. + var candidates = rules.PersistenceProviders() + .Where(x => x is IEventSourcingFrameProvider) + .ToArray(); + + if (candidates.Length == 1) + { + return candidates[0]; + } + + if (candidates.Length == 0) + { + throw new InvalidOperationException( + $"{chain.Description} returns {effectType.NameInCode()}, but this application has no registered " + + "event store. Storage.AppendEvents() and Storage.StartStream() require an event store " + + "integration -- call IntegrateWithWolverine() on your Marten, Polecat, or Fisher store " + + "registration."); + } + + // More than one event store is a real configuration, and there is nothing on the side effect itself + // that says which one it belongs to -- unlike IStorageAction, which resolves by entity type. + var narrowed = candidates.Where(x => x.CanApply(chain, container)).ToArray(); + if (narrowed.Length == 1) + { + return narrowed[0]; + } + + throw new InvalidOperationException( + $"{chain.Description} returns {effectType.NameInCode()}, but this application has " + + $"{candidates.Length} registered event stores and nothing on the side effect says which one to use. " + + "Mark the handler with [Storage(typeof(IYourStore))] to name the store explicitly."); + } +} + +/// +/// Append events to an existing event stream as a Wolverine , returned from a handler +/// or HTTP endpoint rather than written by reaching for a store's session. +/// +/// +/// +/// The work is expressed entirely against , the shared write-side event API +/// that Marten, Polecat and Fisher all implement, so a handler returning one of these is valid on any of +/// them. Each store's IntegrateWithWolverine() registers the variable source that hands Wolverine +/// the for the active session. +/// +/// +/// Build these through and friends rather than +/// constructing them directly. +/// +/// +public class AppendEvents : ISideEffectAware +{ + static Frame ISideEffectAware.BuildFrame(IChain chain, Variable variable, GenerationRules rules, + IServiceContainer container) + => EventSideEffectFrames.BuildFrame(typeof(AppendEvents), chain, variable, rules, container); + + internal AppendEvents(Guid streamId, object[] events) + { + StreamId = streamId; + Events = events; + } + + internal AppendEvents(string streamKey, object[] events) + { + StreamKey = streamKey; + Events = events; + } + + /// + /// The Guid identity of the target stream, for stores configured with Guid stream identity. + /// + public Guid? StreamId { get; } + + /// + /// The string key of the target stream, for stores configured with string stream identity. + /// + public string? StreamKey { get; } + + /// + /// The events to append, in order. + /// + public IReadOnlyList Events { get; } + + /// + /// The expected current version of the stream on the server. When set, the transaction is aborted if the + /// stream has moved on, giving optimistic concurrency. + /// + public long? ExpectedVersion { get; init; } + + // Named Execute rather than Apply because Wolverine's ISideEffect convention only looks for + // Execute/ExecuteAsync -- see SideEffectPolicy.findMethod. The IEventOperations parameter is resolved + // like any other, through the variable source each store registers. + public void Execute(IEventOperations operations) + { + // Returning an empty append is a legitimate "nothing happened" answer from a decision function, + // and every store would otherwise be handed a stream action carrying no events. + if (Events.Count == 0) + { + return; + } + + if (StreamId.HasValue) + { + if (ExpectedVersion.HasValue) + { + operations.Append(StreamId.Value, ExpectedVersion.Value, Events.ToArray()); + } + else + { + operations.Append(StreamId.Value, Events); + } + } + else + { + if (ExpectedVersion.HasValue) + { + operations.Append(StreamKey!, ExpectedVersion.Value, Events); + } + else + { + operations.Append(StreamKey!, Events); + } + } + } +} + +/// +/// Start a brand new event stream as a Wolverine . See +/// for the rationale; this is the same mechanism for the "this stream does not exist yet" case. +/// +public class StartStream : ISideEffectAware +{ + static Frame ISideEffectAware.BuildFrame(IChain chain, Variable variable, GenerationRules rules, + IServiceContainer container) + => EventSideEffectFrames.BuildFrame(typeof(StartStream), chain, variable, rules, container); + + internal StartStream(Guid streamId, Type? aggregateType, object[] events) + { + StreamId = streamId; + AggregateType = aggregateType; + Events = events; + } + + internal StartStream(string streamKey, Type? aggregateType, object[] events) + { + StreamKey = streamKey; + AggregateType = aggregateType; + Events = events; + } + + /// + /// The Guid identity for the new stream, for stores configured with Guid stream identity. + /// + public Guid? StreamId { get; } + + /// + /// The string key for the new stream, for stores configured with string stream identity. + /// + public string? StreamKey { get; } + + /// + /// The aggregate type this stream is for, when the store tracks one. Optional. + /// + public Type? AggregateType { get; } + + /// + /// The events to start the stream with, in order. + /// + public IReadOnlyList Events { get; } + + public void Execute(IEventOperations operations) + { + if (Events.Count == 0) + { + return; + } + + if (StreamId.HasValue) + { + if (AggregateType != null) + { + operations.StartStream(AggregateType, StreamId.Value, Events); + } + else + { + operations.StartStream(StreamId.Value, Events); + } + } + else + { + if (AggregateType != null) + { + operations.StartStream(AggregateType, StreamKey!, Events); + } + else + { + operations.StartStream(StreamKey!, Events); + } + } + } +} diff --git a/src/Wolverine/Persistence/Storage.cs b/src/Wolverine/Persistence/Storage.cs index 62a5d3b44..b9058afb6 100644 --- a/src/Wolverine/Persistence/Storage.cs +++ b/src/Wolverine/Persistence/Storage.cs @@ -56,6 +56,55 @@ public static class Storage /// public static Nothing Nothing() => new(); + /// + /// Append events to an existing event stream identified by a Guid. Works against any registered event + /// store -- Marten, Polecat or Fisher -- because it is expressed purely in terms of + /// . + /// + public static AppendEvents AppendEvents(Guid streamId, params object[] events) => new(streamId, events); + + /// + /// Append events to an existing event stream identified by a string key. Works against any registered + /// event store -- Marten, Polecat or Fisher. + /// + public static AppendEvents AppendEvents(string streamKey, params object[] events) => new(streamKey, events); + + /// + /// Append events to an existing event stream identified by a Guid, asserting the stream's current version + /// on the server for optimistic concurrency. + /// + public static AppendEvents AppendEvents(Guid streamId, long expectedVersion, params object[] events) + => new(streamId, events) { ExpectedVersion = expectedVersion }; + + /// + /// Append events to an existing event stream identified by a string key, asserting the stream's current + /// version on the server for optimistic concurrency. + /// + public static AppendEvents AppendEvents(string streamKey, long expectedVersion, params object[] events) + => new(streamKey, events) { ExpectedVersion = expectedVersion }; + + /// + /// Start a brand new event stream with a user supplied Guid identity. + /// + public static StartStream StartStream(Guid streamId, params object[] events) => new(streamId, null, events); + + /// + /// Start a brand new event stream with a user supplied string key. + /// + public static StartStream StartStream(string streamKey, params object[] events) => new(streamKey, null, events); + + /// + /// Start a brand new event stream for a known aggregate type with a user supplied Guid identity. + /// + public static StartStream StartStream(Guid streamId, params object[] events) where T : class + => new(streamId, typeof(T), events); + + /// + /// Start a brand new event stream for a known aggregate type with a user supplied string key. + /// + public static StartStream StartStream(string streamKey, params object[] events) where T : class + => new(streamKey, typeof(T), events); + [UnconditionalSuppressMessage("Trimming", "IL2072", Justification = "Variable.VariableType returns the effect's runtime Type without DAM annotation; TypeExtensions.Closes inspects the generic-interface graph for IStorageAction<>. The entity type is application-rooted (handler return type), preserved in any practical setup; AOT consumers register effect entity types via the persistence-frame provider registration.")] internal static bool TryApply(Variable effect, GenerationRules rules, IServiceContainer container, IChain chain)