diff --git a/docs/guide/handlers/persistence.md b/docs/guide/handlers/persistence.md index 01d723d09..e52a8c835 100644 --- a/docs/guide/handlers/persistence.md +++ b/docs/guide/handlers/persistence.md @@ -18,6 +18,8 @@ These all speak one vocabulary, and none of it names your database: | An event sourced model's current state, read only | `[ReadModel]` | | The whole method to be an event sourced command handler | `[DeciderFunction]` | | An event sourced model spanning several streams, matched by tag | `[DcbModel]` | +| A stream's metadata — version, type, timestamps — **unfolded** | [`[StreamState]`](#raw-stream-reads) | +| A stream's raw events, **unfolded** | [`[StreamEvents]`](#raw-stream-reads) | | 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) | | Every document of a type | [`[All]`](#reading-every-document-of-a-type) | @@ -407,7 +409,7 @@ public static class ShipOrderHandler } } ``` -snippet source | anchor +snippet source | anchor Use `[DeciderFunction]` for the same workflow at the method or class level, where the model's @@ -431,7 +433,7 @@ public static class MarkItemReadyHandler } } ``` -snippet source | anchor +snippet source | anchor Use `[ReadModel]` when the handler only needs to *look at* the model. It resolves the current state @@ -450,7 +452,7 @@ public static class ReadOrderStatusHandler } } ``` -snippet source | anchor +snippet source | anchor By default `[WriteModel]` and `[DeciderFunction]` use an optimistic concurrency check at the point @@ -468,9 +470,133 @@ public static class ShipOrderExclusivelyHandler } } ``` -snippet source | anchor +snippet source | anchor +### Raw Stream Reads + +`[ReadModel]` gives you a model *folded* from a stream's events. Sometimes folding is exactly the wrong +thing: a timeline endpoint, an audit view, or a "what happened to this order?" screen needs the history +itself, and the fold has already collapsed it. `[StreamState]` and `[StreamEvents]` are the raw reads for +those handlers. + +`[StreamState]` resolves the stream's metadata — version, aggregate type, created and updated timestamps — +and `[StreamEvents]` resolves its events as `IReadOnlyList`. Neither one folds anything. Both are +store-agnostic in the same way the rest of this vocabulary is, and the store batches them with any other +batchable load on the same handler into a single round trip: + + + +```cs +public static class OrderTimelineHandler +{ + // [StreamState] gives you the stream's metadata -- version, aggregate type, created/updated + // timestamps -- and [StreamEvents] gives you the raw events, WITHOUT folding either into an + // aggregate. This is the read [ReadModel] cannot express, because folding has already thrown + // away the history this handler exists to serve. Both fetches batch into one round trip. + public static OrderTimeline Handle( + OrderTimelineQuery query, + [StreamState] StreamState state, + [StreamEvents] IReadOnlyList events) + { + return new OrderTimeline(state.Version, events.Select(x => x.EventTypeName).ToArray()); + } +} +``` +snippet source | anchor + + +Like `[ReadModel]`, `[StreamState]` takes its required-ness from the parameter's nullable annotation: +`StreamState state` stops the handler when the stream does not exist, and `StreamState? state` leaves +absence to you. + + + +```cs +public static class OptionalOrderTimelineHandler +{ + // Nullable annotation decides the default, exactly as it does for [ReadModel]: + // "StreamState state" is required and stops the handler when the stream does not exist, + // "StreamState? state" leaves absence to you + public static OrderTimeline Handle(OrderTimelineQuery query, [StreamState] StreamState? state) + { + return new OrderTimeline(state?.Version ?? 0, []); + } +} +``` +snippet source | anchor + + +::: warning The identity convention is not the one `[Entity]` uses +This is the easiest thing here to get wrong. `[Entity] Order order` can infer an identity member named +`OrderId`, because the parameter's own type names your entity. The parameter type here is `StreamState` — +the *store's* vocabulary, not your aggregate — so there is no aggregate name to infer from, and the +convention degrades to a member literally named `Id`. + +For anything else, name the member explicitly: + + + +```cs +public static class OrderAuditHandler +{ + // The identity convention here is NOT the one [Entity] and [ReadModel] use. Those infer + // "OrderId" from the parameter's own type; the parameter type here is StreamState, which + // names the store's vocabulary rather than your aggregate. So a bare [StreamState] resolves + // only a member literally named "Id" -- name the member explicitly for anything else. + public static OrderTimeline Handle( + OrderAuditQuery query, + [StreamState("OrderId")] StreamState state, + [StreamEvents("OrderId")] IReadOnlyList events) + { + return new OrderTimeline(state.Version, events.Select(x => x.EventTypeName).ToArray()); + } +} +``` +snippet source | anchor + + +A miss is not silent. Wolverine throws `InvalidEntityLoadUsageException` when the chain is compiled — at +startup, not at the first request. +::: + +#### There is deliberately no `Required` on `[StreamEvents]` + +`[StreamState]` has a `Required` property; `[StreamEvents]` does not, and that asymmetry is intentional +rather than an oversight. A missing stream yields an **empty list**, not null, so the null-guard model the +rest of the `[Entity]` family is built on has nothing to test — a guard would either never fire, or would +have to invent a count threshold. And "zero events" is a genuinely different question from "no such +stream." + +When a handler needs an existence guard, pair the two. `[StreamState]` reads the same stream in the same +batch and answers the question precisely: + +```csharp +public static OrderTimeline Handle( + OrderTimelineQuery query, + [StreamState] StreamState state, // non-nullable, so this is the not-found guard + [StreamEvents] IReadOnlyList events) +``` + +#### Choosing a store + +Because these parameter types are store vocabulary rather than your own types, Wolverine cannot ask "who +owns this?" the way it can for `[ReadModel] Order`. Resolution goes, in order: + +1. An explicit `AggregateType` — `[StreamState(AggregateType = typeof(Order))]` — which identifies the + owning store and nothing else. It does not change what is read. +2. The ancillary store the chain was routed to by `[Storage(typeof(IMyStore))]`. +3. The single registered event store integration, when there is only one. + +With more than one integration registered and no other signal, Wolverine throws at compile time with a +message naming both escape hatches rather than guessing. + +There are also Marten-specific spellings, `[MartenStreamState]` and `[MartenStreamEvents]`, which exist +for the same reason [`[ReadAggregate]`](/guide/durability/marten/event-sourcing) does: they **name** their +store instead of resolving one, so they still work in a host that called `AddMarten(...)` without +`IntegrateWithWolverine()`, where nothing ever registers a persistence strategy. Prefer the store-agnostic +`[StreamState]` and `[StreamEvents]` in new code. + ### Dynamic Consistency Boundaries `[WriteModel]` and `[DeciderFunction]` are both about *one* stream. When the decision spans several — @@ -507,7 +633,7 @@ public static class ReserveSeatHandler } } ``` -snippet source | anchor +snippet source | anchor Mark the parameter as `IEventBoundary` instead of `T` if you want the boundary handle itself — diff --git a/docs/guide/http/marten.md b/docs/guide/http/marten.md index 94e781ee1..a8e9ca1eb 100644 --- a/docs/guide/http/marten.md +++ b/docs/guide/http/marten.md @@ -152,7 +152,7 @@ public static OrderShipped Ship(ShipOrder2 command, [Aggregate] Order order) return new OrderShipped(); } ``` -snippet source | anchor +snippet source | anchor Using this version of the "aggregate workflow", you no longer have to supply a command in the request body, so you could @@ -171,7 +171,7 @@ public static OrderShipped Ship3([Aggregate] Order order) return new OrderShipped(); } ``` -snippet source | anchor +snippet source | anchor A couple other notes: @@ -264,7 +264,7 @@ public class Order public bool IsShipped() => Shipped.HasValue; } ``` -snippet source | anchor +snippet source | anchor To append a single event to an event stream from an HTTP endpoint, you can use a return value like so: @@ -283,7 +283,7 @@ public static OrderShipped Ship(ShipOrder command, Order order) return new OrderShipped(); } ``` -snippet source | anchor +snippet source | anchor Or potentially append multiple events using the `Events` type as a return value like this sample: @@ -319,7 +319,7 @@ public static (OrderStatus, Events) Post(MarkItemReady command, Order order) return (new OrderStatus(order.Id, order.IsReadyToShip()), events); } ``` -snippet source | anchor +snippet source | anchor ### Responding with the Updated Aggregate @@ -345,7 +345,7 @@ public static (UpdatedAggregate, Events) ConfirmDifferent(ConfirmOrder command, ); } ``` -snippet source | anchor +snippet source | anchor If you should happen to have a message handler or HTTP endpoint signature that uses multiple event streams, @@ -519,7 +519,7 @@ an HTTP endpoint method, use the `[ReadAggregate]` attribute like this: [WolverineGet("/orders/latest/{id}")] public static Order GetLatest(Guid id, [ReadAggregate] Order order) => order; ``` -snippet source | anchor +snippet source | anchor If the aggregate doesn't exist, the HTTP request will stop with a 404 status code. @@ -529,6 +529,45 @@ The aggregate/stream identity is found with the same rules as the `[Entity]` or 2. Look for a request body property or route argument named "EntityTypeId" 3. Look for a request body property or route argument named "Id" or "id" +## Reading a Stream Without Folding It + +`[ReadAggregate]` gives you the *folded* aggregate. When an endpoint's whole job is to serve the history — +a timeline view, an audit trail — folding has already thrown away what it needs. `[StreamState]` and +`[StreamEvents]` are the raw reads for those endpoints: + + + +```cs +// GH-3627. Timeline/audit shaped reads: the handler serves the event history itself, which +// [ReadAggregate] cannot express because folding has already collapsed what it needs. Both fetches +// batch into one round trip alongside any other batchable load on the same handler. +[WolverineGet("/orders/{id}/timeline")] +public static OrderTimeline GetTimeline( + Guid id, + [MartenStreamState] StreamState state, + [MartenStreamEvents] IReadOnlyList events) +{ + return new OrderTimeline(state.Version, events.Select(x => x.EventTypeName).ToArray()); +} +``` +snippet source | anchor + + +`[StreamState]` follows the same not-found rules as `[ReadAggregate]`: a non-nullable `StreamState` +parameter stops the request with a 404, and `StreamState?` leaves absence to the endpoint. + +The identity rules are **not** quite the same, though, and it is worth reading the +[full treatment in the persistence helpers guide](/guide/handlers/persistence.html#raw-stream-reads) +before you use them. `[ReadAggregate] Order order` can find a route argument named `OrderId` because the +parameter type names your aggregate; the parameter type here is `StreamState`, so only a route argument or +body property named `Id` resolves without help. Name it explicitly for anything else — +`[StreamState("orderId")]`. + +The `[MartenStreamState]` and `[MartenStreamEvents]` spellings used above name Marten directly, in the +same way `[ReadAggregate]` does. The store-agnostic `[StreamState]` and `[StreamEvents]` work identically +in any host that registered an event store integration with `IntegrateWithWolverine()`, and are the better +default in new code. + ### Compiled Query Resource Writer Policy Marten integration comes with an `IResourceWriterPolicy` policy that handles compiled queries as return types. @@ -539,7 +578,7 @@ Register it in `WolverineHttpOptions` like this: ```cs opts.UseMartenCompiledQueryResultPolicy(); ``` -snippet source | anchor +snippet source | anchor If you now return a compiled query from an Endpoint the result will get directly streamed to the client as JSON. Short circuiting JSON deserialization. @@ -566,7 +605,7 @@ public class ApprovedInvoicedCompiledQuery : ICompiledListQuery } } ``` -snippet source | anchor +snippet source | anchor ## Streaming JSON Responses diff --git a/src/Persistence/MartenTests/stream_state_and_events_handlers_3627.cs b/src/Persistence/MartenTests/stream_state_and_events_handlers_3627.cs new file mode 100644 index 000000000..2047e7248 --- /dev/null +++ b/src/Persistence/MartenTests/stream_state_and_events_handlers_3627.cs @@ -0,0 +1,112 @@ +using IntegrationTests; +using JasperFx.Events; +using JasperFx.Resources; +using Marten; +using MartenTests.AggregateHandlerWorkflow; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Shouldly; +using Wolverine; +using Wolverine.Marten; +using Wolverine.Persistence.EventSourcing; + +namespace MartenTests; + +public record FindTimeline(Guid Id); + +public record FindTimelineByAggregateId(Guid AggregateId); + +public record Timeline(long Version, string[] EventTypes); + +public static class FindTimelineHandler +{ + public static Timeline Handle( + FindTimeline query, + [StreamState] StreamState state, + [StreamEvents] IReadOnlyList events) + { + return new Timeline(state.Version, events.Select(x => x.EventTypeName).ToArray()); + } + + public static Timeline Handle( + FindTimelineByAggregateId query, + [StreamState("AggregateId")] StreamState state, + [StreamEvents("AggregateId")] IReadOnlyList events) + { + return new Timeline(state.Version, events.Select(x => x.EventTypeName).ToArray()); + } +} + +/// +/// GH-3627. The message handler side of [StreamState] / [StreamEvents]. The HTTP side is covered by +/// Wolverine.Http.Tests, but the handler path had no coverage, and its identity resolution differs in +/// a way that is easy to get wrong: the parameter type is StreamState, not your aggregate, so the +/// "<ParameterType>Id" convention looks for "StreamStateId". Only a member named "Id" or the +/// explicit named-argument form resolves. A miss is an InvalidEntityLoadUsageException at bootstrap. +/// +public class stream_state_and_events_handlers_3627 : PostgresqlContext, IAsyncLifetime +{ + private IHost theHost = null!; + private IDocumentStore theStore = null!; + + public async ValueTask InitializeAsync() + { + theHost = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.Durability.Mode = DurabilityMode.Solo; + opts.Discovery.DisableConventionalDiscovery().IncludeType(typeof(FindTimelineHandler)); + + opts.Services.AddMarten(m => + { + m.Connection(Servers.PostgresConnectionString); + m.DisableNpgsqlLogging = true; + }) + .UseLightweightSessions() + .IntegrateWithWolverine(); + + opts.Services.AddResourceSetupOnStartup(); + }).StartAsync(); + + theStore = theHost.Services.GetRequiredService(); + } + + public async ValueTask DisposeAsync() + { + await theHost.StopAsync(); + theHost.Dispose(); + } + + private async Task startStreamAsync() + { + var streamId = Guid.NewGuid(); + await using var session = theStore.LightweightSession(); + session.Events.StartStream(streamId, new AEvent(), new AEvent(), new CEvent()); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + return streamId; + } + + [Fact] + public async Task the_Id_convention_resolves_the_stream() + { + var streamId = await startStreamAsync(); + + var timeline = await theHost.MessageBus() + .InvokeAsync(new FindTimeline(streamId), TestContext.Current.CancellationToken); + + timeline.Version.ShouldBe(3); + timeline.EventTypes.Length.ShouldBe(3); + } + + [Fact] + public async Task the_named_argument_form_resolves_a_differently_named_property() + { + var streamId = await startStreamAsync(); + + var timeline = await theHost.MessageBus() + .InvokeAsync(new FindTimelineByAggregateId(streamId), TestContext.Current.CancellationToken); + + timeline.Version.ShouldBe(3); + timeline.EventTypes.Length.ShouldBe(3); + } +} diff --git a/src/Samples/DocumentationSamples/EventSourcedModelSamples.cs b/src/Samples/DocumentationSamples/EventSourcedModelSamples.cs index 5015946dc..eb5a04cc4 100644 --- a/src/Samples/DocumentationSamples/EventSourcedModelSamples.cs +++ b/src/Samples/DocumentationSamples/EventSourcedModelSamples.cs @@ -1,3 +1,4 @@ +using JasperFx.Events; using JasperFx.Events.Tags; using Wolverine.Persistence; using Wolverine.Persistence.EventSourcing; @@ -119,3 +120,62 @@ public static SeatReserved Handle(ReserveSeat command, [DcbModel] SeatAvailabili } #endregion + +public record OrderTimelineQuery(Guid Id); + +public record OrderAuditQuery(Guid OrderId); + +public record OrderTimeline(long Version, string[] EventTypes); + +#region sample_using_stream_state_and_events + +public static class OrderTimelineHandler +{ + // [StreamState] gives you the stream's metadata -- version, aggregate type, created/updated + // timestamps -- and [StreamEvents] gives you the raw events, WITHOUT folding either into an + // aggregate. This is the read [ReadModel] cannot express, because folding has already thrown + // away the history this handler exists to serve. Both fetches batch into one round trip. + public static OrderTimeline Handle( + OrderTimelineQuery query, + [StreamState] StreamState state, + [StreamEvents] IReadOnlyList events) + { + return new OrderTimeline(state.Version, events.Select(x => x.EventTypeName).ToArray()); + } +} + +#endregion + +#region sample_stream_state_with_named_identity + +public static class OrderAuditHandler +{ + // The identity convention here is NOT the one [Entity] and [ReadModel] use. Those infer + // "OrderId" from the parameter's own type; the parameter type here is StreamState, which + // names the store's vocabulary rather than your aggregate. So a bare [StreamState] resolves + // only a member literally named "Id" -- name the member explicitly for anything else. + public static OrderTimeline Handle( + OrderAuditQuery query, + [StreamState("OrderId")] StreamState state, + [StreamEvents("OrderId")] IReadOnlyList events) + { + return new OrderTimeline(state.Version, events.Select(x => x.EventTypeName).ToArray()); + } +} + +#endregion + +#region sample_stream_state_optional + +public static class OptionalOrderTimelineHandler +{ + // Nullable annotation decides the default, exactly as it does for [ReadModel]: + // "StreamState state" is required and stops the handler when the stream does not exist, + // "StreamState? state" leaves absence to you + public static OrderTimeline Handle(OrderTimelineQuery query, [StreamState] StreamState? state) + { + return new OrderTimeline(state?.Version ?? 0, []); + } +} + +#endregion