From 147a1bd0711ca622078e068035b7462c77868cbe Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 26 Jul 2026 11:05:43 -0500 Subject: [PATCH] feat(#370): FetchStreamStatePlan / FetchStreamPlan + batched event fetches + StreamEventState / StreamEvents (marten#5053 parity) Parity port of marten#5053 (supersedes marten#5043, original plans by @uniquelau). Polecat already had the matching IQueryPlan / IBatchQueryPlan / QueryListPlan abstractions and the Stream* result family, so this is a port rather than a design exercise. Both halves land here rather than plans-only. Through Wolverine's fetch-specification feature a plan implementing only IBatchQueryPlan produces uncompilable generated code, so a partial implementation would have been actively worse than none. Part 1 -- the batched event surface, which did not exist. IBatchedQuery had QueryByPlan, Load/LoadMany/CheckExists, EventsExist and FetchForWritingByTags but no Events surface at all. New IBatchEvents, reached as batch.Events, with FetchStreamState and FetchStream in Guid and string overloads, FetchStream carrying the same optional version/timestamp/fromVersion filters as the standalone fetch. The two new batch items compose their SQL from the SAME canonical projections (PcStreamsRowReader.SelectColumns, PcEventsRowReader.ComposeSelectColumns) and hydrate through the SAME readers as the standalone path, with the same per-batch hoists, so batched and standalone cannot drift apart across a schema migration. The batch ensures the event store schema when it holds an event item (#219 create-on-first-use), flagged rather than eager so a document-only batch pays nothing. Part 2 -- FetchStreamStatePlan and FetchStreamPlan, each implementing both IQueryPlan and IBatchQueryPlan, each with Guid/string constructor overloads. Part 3 -- StreamEventState and StreamEvents in Polecat.AspNetCore, the event-side siblings of StreamAggregate, plus WriteStreamState / WriteEvents extensions on IQuerySession. Neither writes the framework's own types to the wire because neither can: StreamState.AggregateType and IEvent.EventType are System.Type and STJ refuses to serialize those, so StreamStateResponse / EventResponse project them down -- aggregate type to its simple name, IEvent's assembly-qualified DotNetTypeName deliberately off the wire with EventTypeName as the discriminator. Property names match Marten's DTOs so clients can move between stores. FetchStream yields an empty list both for a missing stream and for a filter excluding every event, so StreamEvents.OnEmptyStatus defaults to 404 with 200 as the opt-out for fromVersion paging. Bodies buffer through an ArrayBufferWriter and set Content-Length. Also fixed, because the port surfaced it and it would have shipped a dead wire field: StreamState.AggregateType was NEVER populated. pc_streams.type has always been projected and never read back, and the JasperFx path that would register the alias (StreamAction.PrepareEvents -> AggregateAliasFor) is one Polecat's QuickAppend closed-shape writer never goes through, so EventGraph.AggregateTypeFor could never resolve it. New EventGraph.TryResolveAggregateType resolves against the registered projections first (the strategy the event store explorer already used), caches the hit, and returns null rather than throwing for an alias this deployment does not know -- a stream tagged by a deployment that knew a type this one does not must still report its version and timestamps. This fixes FetchStreamStateAsync for every caller, not just the new endpoint. Tests: batch_event_fetching (the surface directly, including a dedicated timestamp case and an interleaved batch where a mis-ordered result set surfaces as a cross-wired answer), fetching_stream_query_plans (each plan standalone and batched, missing stream, version cap, fromVersion, shared round trip, plus a string-identity store), and 12 Alba tests over real Minimal API endpoints. Docs in batched-queries.md, events/querying.md and documents/aspnetcore.md. Full suite green: 1576 passed on net10, AspNetCore 65, EF Core 37. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01ETF4fye5MJk3kGX5kunrRk --- docs/documents/aspnetcore.md | 52 ++++ docs/documents/querying/batched-queries.md | 53 ++++ docs/events/querying.md | 27 ++ src/Polecat.AspNetCore.Testing/Program.cs | 23 ++ .../stream_event_result_types_tests.cs | 248 ++++++++++++++++++ .../EventStreamExtensions.cs | 123 +++++++++ .../EventStreamResponses.cs | 151 +++++++++++ src/Polecat.AspNetCore/StreamEventState.cs | 98 +++++++ src/Polecat.AspNetCore/StreamEvents.cs | 117 +++++++++ .../Batching/batch_event_fetching.cs | 223 ++++++++++++++++ .../Querying/fetching_stream_query_plans.cs | 232 ++++++++++++++++ src/Polecat/Batching/IBatchEvents.cs | 51 ++++ src/Polecat/Batching/IBatchedQuery.cs | 7 + src/Polecat/Events/EventGraph.cs | 41 +++ .../Events/Internal/PcStreamsRowReader.cs | 17 +- src/Polecat/Events/QueryEventStore.cs | 2 +- src/Polecat/IQueryPlan.cs | 121 +++++++++ src/Polecat/Internal/Batching/BatchEvents.cs | 56 ++++ src/Polecat/Internal/Batching/BatchedQuery.cs | 16 ++ .../Internal/Batching/FetchStreamBatchItem.cs | 107 ++++++++ .../Batching/FetchStreamStateBatchItem.cs | 52 ++++ 21 files changed, 1815 insertions(+), 2 deletions(-) create mode 100644 src/Polecat.AspNetCore.Testing/stream_event_result_types_tests.cs create mode 100644 src/Polecat.AspNetCore/EventStreamExtensions.cs create mode 100644 src/Polecat.AspNetCore/EventStreamResponses.cs create mode 100644 src/Polecat.AspNetCore/StreamEventState.cs create mode 100644 src/Polecat.AspNetCore/StreamEvents.cs create mode 100644 src/Polecat.Tests/Batching/batch_event_fetching.cs create mode 100644 src/Polecat.Tests/Querying/fetching_stream_query_plans.cs create mode 100644 src/Polecat/Batching/IBatchEvents.cs create mode 100644 src/Polecat/Internal/Batching/BatchEvents.cs create mode 100644 src/Polecat/Internal/Batching/FetchStreamBatchItem.cs create mode 100644 src/Polecat/Internal/Batching/FetchStreamStateBatchItem.cs diff --git a/docs/documents/aspnetcore.md b/docs/documents/aspnetcore.md index 82ad010e..83cb0587 100644 --- a/docs/documents/aspnetcore.md +++ b/docs/documents/aspnetcore.md @@ -125,6 +125,58 @@ Returns `200 application/json` with the latest projected aggregate state, or `40 stream exists. A constructor overload accepts `string` ids for stores configured with string-keyed streams. +## Streaming Event Stream Metadata and Events + +`StreamEventState` and `StreamEvents` are the event-side siblings of `StreamAggregate`, backed by +the `FetchStreamStatePlan` / `FetchStreamPlan` query plans — so the same plan can be batched through +`IBatchedQuery.QueryByPlan()` and returned from an endpoint. + +```csharp +app.MapGet("/orders/{id:guid}/state", (Guid id, IQuerySession s) => new StreamEventState(s, id)); +app.MapGet("/orders/{id:guid}/events", (Guid id, IQuerySession s) => new StreamEvents(s, id)); +``` + +Both take `Guid`, `string` and pre-built-plan constructors, and both implement `IResult` and +`IEndpointMetadataProvider` so OpenAPI advertises the right `200` and `404` shapes. + +### StreamEventState vs StreamAggregate + +- **`StreamEventState`** writes the stream's *metadata* — version, created/last timestamps, archived flag. +- **`StreamAggregate`** writes the projected aggregate *state* built from the stream's events. + +### The response DTOs + +Neither result writes the framework's own types to the wire, because neither can. `StreamState.AggregateType` +and `IEvent.EventType` are `System.Type`, and System.Text.Json refuses to serialize those: + +``` +NotSupportedException: Serialization and deserialization of 'System.Type' instances +is not supported. Path: $.AggregateType. +``` + +So the bodies are `StreamStateResponse` and `EventResponse`: the aggregate type reduces to its simple +name, and `IEvent`'s assembly-qualified `DotNetTypeName` is deliberately kept off the wire — use +`EventTypeName`, Polecat's stable event type alias, as the client-side discriminator. The property names +match Marten's equivalents, so a client can move between the two stores unchanged. + +### Empty streams are ambiguous + +`FetchStream` yields an empty list both for a stream that does not exist and for a filter that excludes +every event, and the two cannot be told apart. `StreamEvents` exposes `OnEmptyStatus`, defaulting to `404` +to match the other single-resource results. Set it to `200` to return an empty array instead — which is +what you want when paging forward with `fromVersion` and running off the end is expected: + +```csharp +app.MapGet("/orders/{id:guid}/events", (Guid id, long fromVersion, IQuerySession s) => + new StreamEvents(s, id, fromVersion: fromVersion) + { + OnEmptyStatus = StatusCodes.Status200OK + }); +``` + +Both responses set `Content-Length`, and serialization buffers through an `ArrayBufferWriter` so the +JSON never round-trips through a .NET string. + ### StreamOne vs StreamAggregate - **`StreamOne`** is for regular documents — objects stored via `session.Store()` and diff --git a/docs/documents/querying/batched-queries.md b/docs/documents/querying/batched-queries.md index 3b28fddd..ae79e9d6 100644 --- a/docs/documents/querying/batched-queries.md +++ b/docs/documents/querying/batched-queries.md @@ -67,3 +67,56 @@ Query plans can also be used independently: ```cs var orders = await session.QueryByPlanAsync(new ActiveOrdersPlan()); ``` + +## Batched Event Store Fetches + +`batch.Events` exposes the batched counterparts of `FetchStreamStateAsync` and `FetchStreamAsync`, so a +raw event-stream read can share the batch's single round trip with document loads and LINQ queries: + +```cs +var batch = session.CreateBatchQuery(); + +var stateTask = batch.Events.FetchStreamState(streamId); +var eventsTask = batch.Events.FetchStream(streamId); +var orderTask = batch.Load(orderId); + +await batch.Execute(); + +var state = await stateTask; // StreamState?, null when the stream does not exist +var events = await eventsTask; // IReadOnlyList, empty when the stream does not exist +var order = await orderTask; +``` + +Both come in `Guid` and `string` overloads for the two stream identity modes, and `FetchStream` carries the +same optional `version` / `timestamp` / `fromVersion` filters as `FetchStreamAsync`: + +```cs +// Events up to and including version 5 +var capped = batch.Events.FetchStream(streamId, version: 5); + +// Everything appended from version 10 onward +var tail = batch.Events.FetchStream(streamId, fromVersion: 10); +``` + +## Event Stream Query Plans + +`FetchStreamStatePlan` and `FetchStreamPlan` wrap those fetches as query plans. Both implement **both** +`IQueryPlan` and `IBatchQueryPlan`, so the same plan instance works standalone or in a batch: + +```cs +// Standalone +var state = await session.QueryByPlanAsync(new FetchStreamStatePlan(streamId)); +var events = await session.QueryByPlanAsync(new FetchStreamPlan(streamId, version: 5)); + +// Batched — one round trip +var batch = session.CreateBatchQuery(); +var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId)); +var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamId)); +await batch.Execute(); +``` + +::: tip +Implementing both interfaces matters beyond convenience. Through Wolverine's fetch-specification feature, +a plan that implements only `IBatchQueryPlan` produces uncompilable generated code — so a custom plan +you intend to route through a handler's `Load` should implement the pair as well. +::: diff --git a/docs/events/querying.md b/docs/events/querying.md index 8ef7cb3e..1e8ba017 100644 --- a/docs/events/querying.md +++ b/docs/events/querying.md @@ -17,6 +17,33 @@ foreach (var @event in events) Events are returned in version order. Archived streams are automatically excluded. +## Stream Fetches as Query Plans + +`FetchStreamStatePlan` and `FetchStreamPlan` wrap the two raw stream fetches as reusable query plans. +Both implement `IQueryPlan` **and** `IBatchQueryPlan`, so the same plan works standalone or inside +a batched query, and both offer `Guid streamId` / `string streamKey` constructor overloads: + +```cs +// Standalone +var state = await session.QueryByPlanAsync(new FetchStreamStatePlan(streamId)); +var events = await session.QueryByPlanAsync(new FetchStreamPlan(streamId, version: 5)); + +// Batched — one round trip +var batch = session.CreateBatchQuery(); +var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId)); +var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamId)); +await batch.Execute(); +``` + +`FetchStreamStatePlan` yields `null` when the stream does not exist; `FetchStreamPlan` yields an empty +list and carries `FetchStream`'s optional `version` / `timestamp` / `fromVersion` arguments. + +The underlying batched fetchers are also available directly as `batch.Events.FetchStreamState(...)` and +`batch.Events.FetchStream(...)` — see [Batched Queries](/documents/querying/batched-queries#batched-event-store-fetches). + +To return either straight from an ASP.NET Core endpoint, see the `StreamEventState` and `StreamEvents` +result types in [ASP.NET Core Integration](/documents/aspnetcore#streaming-event-stream-metadata-and-events). + ## AggregateStreamAsync Replay events to build the current aggregate state: diff --git a/src/Polecat.AspNetCore.Testing/Program.cs b/src/Polecat.AspNetCore.Testing/Program.cs index a2ddd53b..f484ae17 100644 --- a/src/Polecat.AspNetCore.Testing/Program.cs +++ b/src/Polecat.AspNetCore.Testing/Program.cs @@ -59,6 +59,29 @@ new StreamPagedByCursor( session.Query().OrderBy(x => x.Number).ThenBy(x => x.Id), cursor, pageSize)); +// #370 StreamEventState endpoint — stream metadata (version/timestamps/archived) or 404 +app.MapGet("/api/streams/{id:guid}/state", (Guid id, IQuerySession session) => + new StreamEventState(session, id)); + +// #370 StreamEvents endpoint — the stream's raw events as a JSON array, or 404 when empty +app.MapGet("/api/streams/{id:guid}/events", (Guid id, IQuerySession session) => + new StreamEvents(session, id)); + +// OnEmptyStatus opt-out — an empty stream answers 200 with an empty array rather than 404, which is +// what a caller paging forward with fromVersion wants when it runs off the end. +app.MapGet("/api/streams/{id:guid}/events-empty200", (Guid id, long? fromVersion, IQuerySession session) => + new StreamEvents(session, id, fromVersion: fromVersion ?? 0) + { + OnEmptyStatus = StatusCodes.Status200OK + }); + +// Pre-built plan constructor — a handler can build the plan once and either batch it or return it +app.MapGet("/api/streams/{id:guid}/events-by-plan", (Guid id, IQuerySession session) => + new StreamEvents(session, new FetchStreamPlan(id, version: 1))); + +app.MapGet("/api/streams/{id:guid}/state-by-plan", (Guid id, IQuerySession session) => + new StreamEventState(session, new FetchStreamStatePlan(id))); + app.Run(); namespace Polecat.AspNetCore.Testing diff --git a/src/Polecat.AspNetCore.Testing/stream_event_result_types_tests.cs b/src/Polecat.AspNetCore.Testing/stream_event_result_types_tests.cs new file mode 100644 index 00000000..fb4f4d36 --- /dev/null +++ b/src/Polecat.AspNetCore.Testing/stream_event_result_types_tests.cs @@ -0,0 +1,248 @@ +using System.Text.Json; +using Alba; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; +using Xunit; + +namespace Polecat.AspNetCore.Testing; + +/// +/// #370 (parity with marten#5053): the StreamEventState / StreamEvents endpoint result +/// types over real Minimal API endpoints. +/// +public class stream_event_result_types_tests : IAsyncLifetime +{ + private IAlbaHost _host = null!; + + public async Task InitializeAsync() + { + _host = await AlbaHost.For(); + + var store = (DocumentStore)_host.Services.GetRequiredService(); + await store.Database.ApplyAllConfiguredChangesToDatabaseAsync(); + } + + public async Task DisposeAsync() => await _host.DisposeAsync(); + + [Fact] + public async Task stream_event_state_returns_200_with_the_metadata() + { + var streamId = await StartPartyAsync(); + + var result = await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{streamId}/state"); + s.StatusCodeShouldBeOk(); + s.ContentTypeShouldBe("application/json"); + }); + + var state = Read(result); + state.Id.ShouldBe(streamId); + state.Version.ShouldBe(2); + state.IsArchived.ShouldBeFalse(); + state.Created.ShouldNotBe(default); + state.LastTimestamp.ShouldNotBe(default); + } + + [Fact] + public async Task stream_event_state_returns_404_for_a_missing_stream() + { + await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{Guid.NewGuid()}/state"); + s.StatusCodeShouldBe(404); + }); + } + + [Fact] + public async Task stream_event_state_accepts_a_prebuilt_plan() + { + var streamId = await StartPartyAsync(); + + var result = await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{streamId}/state-by-plan"); + s.StatusCodeShouldBeOk(); + }); + + Read(result).Version.ShouldBe(2); + } + + [Fact] + public async Task stream_events_returns_200_with_the_serialized_events() + { + var streamId = await StartPartyAsync(); + + var result = await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{streamId}/events"); + s.StatusCodeShouldBeOk(); + s.ContentTypeShouldBe("application/json"); + }); + + var events = Read(result); + events.Length.ShouldBe(2); + events[0].Version.ShouldBe(1); + events[1].Version.ShouldBe(2); + events.ShouldAllBe(x => x.StreamId == streamId); + events.ShouldAllBe(x => x.Id != Guid.Empty); + events.ShouldAllBe(x => x.Sequence > 0); + } + + /// + /// The event body itself has to survive the DTO projection — an endpoint that returned only + /// metadata would pass every other assertion here. + /// + [Fact] + public async Task stream_events_writes_the_event_body() + { + var streamId = await StartPartyAsync(); + + var result = await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{streamId}/events"); + s.StatusCodeShouldBeOk(); + }); + + var body = result.ReadAsText(); + body.ShouldContain("Fellowship"); + body.ShouldContain("Frodo"); + } + + /// + /// The reason the DTOs exist at all: IEvent.EventType is a , and STJ + /// throws NotSupportedException outright on those. EventTypeName is the alias a client + /// discriminates on; the assembly-qualified .NET type name is deliberately kept off the wire. + /// + [Fact] + public async Task stream_events_writes_the_alias_and_not_the_dotnet_type() + { + var streamId = await StartPartyAsync(); + + var result = await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{streamId}/events"); + s.StatusCodeShouldBeOk(); + }); + + var events = Read(result); + events.ShouldAllBe(x => !string.IsNullOrEmpty(x.EventTypeName)); + + var body = result.ReadAsText(); + body.ShouldNotContain("Polecat.AspNetCore.Testing, Version="); + body.ShouldNotContain("\"EventType\""); + } + + [Fact] + public async Task stream_events_returns_404_for_a_missing_stream_by_default() + { + await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{Guid.NewGuid()}/events"); + s.StatusCodeShouldBe(404); + }); + } + + [Fact] + public async Task on_empty_status_opts_out_of_the_404() + { + var result = await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{Guid.NewGuid()}/events-empty200"); + s.StatusCodeShouldBeOk(); + }); + + Read(result).ShouldBeEmpty(); + } + + /// + /// The case OnEmptyStatus exists for: paging forward with fromVersion and running off the end of + /// a stream that really does exist is expected, not a 404. + /// + [Fact] + public async Task on_empty_status_covers_paging_off_the_end_of_a_real_stream() + { + var streamId = await StartPartyAsync(); + + var result = await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{streamId}/events-empty200?fromVersion=99"); + s.StatusCodeShouldBeOk(); + }); + + Read(result).ShouldBeEmpty(); + } + + [Fact] + public async Task stream_events_accepts_a_prebuilt_plan_and_honors_its_filters() + { + var streamId = await StartPartyAsync(); + + var result = await _host.Scenario(s => + { + s.Get.Url($"/api/streams/{streamId}/events-by-plan"); + s.StatusCodeShouldBeOk(); + }); + + // The endpoint's plan caps at version 1 + var events = Read(result); + events.Length.ShouldBe(1); + events[0].Version.ShouldBe(1); + } + + [Fact] + public async Task both_results_set_content_length() + { + var streamId = await StartPartyAsync(); + + foreach (var url in new[] { $"/api/streams/{streamId}/state", $"/api/streams/{streamId}/events" }) + { + var result = await _host.Scenario(s => + { + s.Get.Url(url); + s.StatusCodeShouldBeOk(); + }); + + result.Context.Response.ContentLength.ShouldNotBeNull($"{url} must set Content-Length"); + result.Context.Response.ContentLength!.Value.ShouldBeGreaterThan(0); + } + } + + [Fact] + public async Task both_results_advertise_their_openapi_metadata() + { + var sources = _host.Services.GetServices(); + var endpoints = sources.SelectMany(x => x.Endpoints).OfType(); + + foreach (var pattern in new[] { "/api/streams/{id:guid}/state", "/api/streams/{id:guid}/events" }) + { + var endpoint = endpoints.Single(x => x.RoutePattern.RawText == pattern); + var produces = endpoint.Metadata + .OfType() + .ToList(); + + produces.ShouldContain(x => x.StatusCode == 200, $"{pattern} should advertise a 200"); + produces.ShouldContain(x => x.StatusCode == 404, $"{pattern} should advertise a 404"); + } + } + + private async Task StartPartyAsync() + { + var store = _host.Services.GetRequiredService(); + var streamId = Guid.NewGuid(); + + await using var session = store.LightweightSession(); + session.Events.StartStream(streamId, + new StreamingQuestStarted("Fellowship"), + new StreamingMembersJoined(["Frodo", "Sam"])); + await session.SaveChangesAsync(); + + return streamId; + } + + private static T Read(IScenarioResult result) + { + return JsonSerializer.Deserialize(result.ReadAsText(), + new JsonSerializerOptions(JsonSerializerDefaults.Web))!; + } +} diff --git a/src/Polecat.AspNetCore/EventStreamExtensions.cs b/src/Polecat.AspNetCore/EventStreamExtensions.cs new file mode 100644 index 00000000..856bdf46 --- /dev/null +++ b/src/Polecat.AspNetCore/EventStreamExtensions.cs @@ -0,0 +1,123 @@ +using System.Buffers; +using System.Diagnostics.CodeAnalysis; +using System.Text.Json; +using Microsoft.AspNetCore.Http; + +namespace Polecat.AspNetCore; + +/// +/// #370: the body writers behind and , +/// exposed as extensions so a handler that does not want the +/// IResult wrapper can write the same responses itself. +/// +public static class EventStreamExtensions +{ + /// + /// Resolve a and write the resulting stream metadata to the + /// response as JSON, or 404 when the stream does not exist. + /// + /// The response body is a rather than Polecat's + /// StreamState — see that type for why. + /// + /// + /// + /// + /// + /// + /// Defaults to 200 + [RequiresDynamicCode("Serializes StreamStateResponse with System.Text.Json, which uses runtime codegen.")] + [RequiresUnreferencedCode("Reflects over StreamStateResponse via System.Text.Json.")] + public static async Task WriteStreamState( + this IQuerySession session, + FetchStreamStatePlan plan, + HttpContext context, + string contentType = "application/json", + int onFoundStatus = StatusCodes.Status200OK) + { + ArgumentNullException.ThrowIfNull(session); + ArgumentNullException.ThrowIfNull(plan); + ArgumentNullException.ThrowIfNull(context); + + var state = await plan.Fetch(session, context.RequestAborted).ConfigureAwait(false); + + if (state == null) + { + context.Response.StatusCode = StatusCodes.Status404NotFound; + context.Response.ContentLength = 0; + return; + } + + await WriteJson(StreamStateResponse.From(state), context, contentType, onFoundStatus) + .ConfigureAwait(false); + } + + /// + /// Resolve a and write the resulting raw events to the + /// response as a JSON array. + /// + /// FetchStream yields an empty list both for a stream that does not exist and for a filter + /// that excludes every event, so the two cases cannot be told apart here. + /// decides which answer the endpoint gives; it defaults to + /// 404 to match the other single-resource results. Pass 200 to return an empty JSON + /// array instead. + /// + /// + /// Each element is an rather than Polecat's IEvent — see that + /// type for why. + /// + /// + /// + /// + /// + /// + /// Defaults to 200 + /// Defaults to 404 + [RequiresDynamicCode("Serializes EventResponse[] with System.Text.Json, which uses runtime codegen for each event's Data payload.")] + [RequiresUnreferencedCode("Reflects over EventResponse and each event's Data payload via System.Text.Json.")] + public static async Task WriteEvents( + this IQuerySession session, + FetchStreamPlan plan, + HttpContext context, + string contentType = "application/json", + int onFoundStatus = StatusCodes.Status200OK, + int onEmptyStatus = StatusCodes.Status404NotFound) + { + ArgumentNullException.ThrowIfNull(session); + ArgumentNullException.ThrowIfNull(plan); + ArgumentNullException.ThrowIfNull(context); + + var events = await plan.Fetch(session, context.RequestAborted).ConfigureAwait(false); + + if (events.Count == 0 && onEmptyStatus == StatusCodes.Status404NotFound) + { + context.Response.StatusCode = StatusCodes.Status404NotFound; + context.Response.ContentLength = 0; + return; + } + + await WriteJson(EventResponse.From(events), context, contentType, + events.Count == 0 ? onEmptyStatus : onFoundStatus).ConfigureAwait(false); + } + + /// + /// Serialize and write it to the response with Content-Length set. + /// Buffers through an so the JSON never round-trips through a + /// .NET string on its way to the socket. + /// + [RequiresDynamicCode("Serializes the response DTO with System.Text.Json, which uses runtime codegen for the event Data payload.")] + [RequiresUnreferencedCode("Reflects over the response DTO and the event Data payload via System.Text.Json.")] + private static async Task WriteJson(T value, HttpContext context, string contentType, int statusCode) + { + var buffer = new ArrayBufferWriter(); + await using (var writer = new Utf8JsonWriter(buffer)) + { + JsonSerializer.Serialize(writer, value); + } + + context.Response.StatusCode = statusCode; + context.Response.ContentType = contentType; + context.Response.ContentLength = buffer.WrittenCount; + await context.Response.Body.WriteAsync(buffer.WrittenMemory, context.RequestAborted) + .ConfigureAwait(false); + } +} diff --git a/src/Polecat.AspNetCore/EventStreamResponses.cs b/src/Polecat.AspNetCore/EventStreamResponses.cs new file mode 100644 index 00000000..1c9c10c0 --- /dev/null +++ b/src/Polecat.AspNetCore/EventStreamResponses.cs @@ -0,0 +1,151 @@ +using JasperFx.Events; + +namespace Polecat.AspNetCore; + +/// +/// The HTTP wire shape written by for a single event stream's +/// metadata. +/// +/// is not written directly because StreamState.AggregateType is a +/// , and System.Text.Json refuses to serialize instances +/// ("Serialization and deserialization of 'System.Type' instances is not supported"). This record +/// projects the aggregate type down to its simple name and is a stable contract for HTTP clients. +/// +/// +/// Property names match Marten's StreamStateResponse so a client can move between the two +/// stores unchanged. +/// +/// +public sealed record StreamStateResponse +{ + /// Identity of the stream when using Guid identity; for string-keyed streams. + public Guid Id { get; init; } + + /// Identity of the stream when using string identity; null for Guid-keyed streams. + public string? Key { get; init; } + + /// Current version of the stream, i.e. the count of events. + public long Version { get; init; } + + /// Simple name of the aggregate type the stream was tagged with, when it was tagged at all. + public string? AggregateTypeName { get; init; } + + /// The last time this stream was appended to. + public DateTimeOffset LastTimestamp { get; init; } + + /// The time at which this stream was created. + public DateTimeOffset Created { get; init; } + + /// Whether the stream has been archived. + public bool IsArchived { get; init; } + + /// + /// Project a onto the wire shape. + /// + public static StreamStateResponse From(StreamState state) + { + ArgumentNullException.ThrowIfNull(state); + + return new StreamStateResponse + { + Id = state.Id, + Key = state.Key, + Version = state.Version, + AggregateTypeName = state.AggregateType?.Name, + LastTimestamp = state.LastTimestamp, + Created = state.Created, + IsArchived = state.IsArchived + }; + } +} + +/// +/// The HTTP wire shape written by for one raw event in a stream. +/// +/// is not written directly because IEvent.EventType is a +/// , which System.Text.Json refuses to serialize. DotNetTypeName — the +/// assembly qualified .NET type name — is deliberately left off the wire as well; use +/// , Polecat's stable event type alias, to discriminate event types on +/// the client. +/// +/// +/// Property names match Marten's EventResponse so a client can move between the two stores +/// unchanged. +/// +/// +public sealed record EventResponse +{ + /// Unique identifier of the event. + public Guid Id { get; init; } + + /// The event's position within its stream. + public long Version { get; init; } + + /// The event's sequential position across the entire event store. + public long Sequence { get; init; } + + /// Owning stream's id when using Guid identity; otherwise. + public Guid StreamId { get; init; } + + /// Owning stream's key when using string identity; null otherwise. + public string? StreamKey { get; init; } + + /// Polecat's event type alias — the stable discriminator for clients. + public string? EventTypeName { get; init; } + + /// The time at which the event was captured. + public DateTimeOffset Timestamp { get; init; } + + /// The owning tenant id. + public string? TenantId { get; init; } + + /// Whether the event has been archived. + public bool IsArchived { get; init; } + + /// Optional causation id metadata. + public string? CausationId { get; init; } + + /// Optional correlation id metadata. + public string? CorrelationId { get; init; } + + /// Optional user defined metadata. Null when no headers were set. + public Dictionary? Headers { get; init; } + + /// The event body itself. + public object Data { get; init; } = default!; + + /// + /// Project an onto the wire shape. + /// + public static EventResponse From(IEvent @event) + { + ArgumentNullException.ThrowIfNull(@event); + + return new EventResponse + { + Id = @event.Id, + Version = @event.Version, + Sequence = @event.Sequence, + StreamId = @event.StreamId, + StreamKey = @event.StreamKey, + EventTypeName = @event.EventTypeName, + Timestamp = @event.Timestamp, + TenantId = @event.TenantId, + IsArchived = @event.IsArchived, + CausationId = @event.CausationId, + CorrelationId = @event.CorrelationId, + Headers = @event.Headers, + Data = @event.Data + }; + } + + /// + /// Project a list of onto the wire shape. + /// + public static EventResponse[] From(IReadOnlyList events) + { + ArgumentNullException.ThrowIfNull(events); + + return events.Select(From).ToArray(); + } +} diff --git a/src/Polecat.AspNetCore/StreamEventState.cs b/src/Polecat.AspNetCore/StreamEventState.cs new file mode 100644 index 00000000..968327a1 --- /dev/null +++ b/src/Polecat.AspNetCore/StreamEventState.cs @@ -0,0 +1,98 @@ +using System.Diagnostics.CodeAnalysis; +using System.Reflection; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Metadata; + +namespace Polecat.AspNetCore; + +/// +/// Minimal-API endpoint return value that writes the high level metadata of a single event stream — +/// Polecat's StreamState — to the response as JSON. Backed by +/// , the same query plan that can be batched through +/// IBatchedQuery.QueryByPlan(). +/// +/// Returns HTTP 404 when the stream does not exist, (default 200) +/// when it does. +/// +/// +/// The response body is a , not Polecat's StreamState +/// directly: StreamState.AggregateType is a and System.Text.Json refuses +/// to serialize those, so the aggregate type is projected down to its simple name. +/// +/// +/// StreamEventState vs StreamAggregate. Use when you want the +/// stream's metadata — version, timestamps, archived flag. Use +/// when you want the projected aggregate state built from the stream's events. +/// +/// +public sealed class StreamEventState : IResult, IEndpointMetadataProvider +{ + private readonly IQuerySession _session; + private readonly FetchStreamStatePlan _plan; + + /// + /// Write the stream metadata for the Guid-identified stream . + /// + public StreamEventState(IQuerySession session, Guid streamId) + : this(session, new FetchStreamStatePlan(streamId)) + { + } + + /// + /// Write the stream metadata for the string-keyed stream . + /// + public StreamEventState(IQuerySession session, string streamKey) + : this(session, new FetchStreamStatePlan( + streamKey ?? throw new ArgumentNullException(nameof(streamKey)))) + { + } + + /// + /// Write the stream metadata resolved by an existing . Lets a + /// handler build the plan once and either batch it or return it straight from an endpoint. + /// + public StreamEventState(IQuerySession session, FetchStreamStatePlan plan) + { + _session = session ?? throw new ArgumentNullException(nameof(session)); + _plan = plan ?? throw new ArgumentNullException(nameof(plan)); + } + + /// + /// Status code written when the stream is found. Defaults to 200. + /// + public int OnFoundStatus { get; init; } = StatusCodes.Status200OK; + + /// + /// Response content type. Defaults to application/json. + /// + public string ContentType { get; init; } = "application/json"; + + /// + [UnconditionalSuppressMessage("Trimming", "IL2046", + Justification = "IResult.ExecuteAsync is not RUC-annotated; the contract lives on this override.")] + [UnconditionalSuppressMessage("AOT", "IL3051", + Justification = "IResult.ExecuteAsync is not RDC-annotated; the contract lives on this override.")] + [RequiresDynamicCode("Serializes StreamStateResponse with System.Text.Json, which uses runtime codegen.")] + [RequiresUnreferencedCode("Reflects over StreamStateResponse via System.Text.Json.")] + public Task ExecuteAsync(HttpContext httpContext) + { + ArgumentNullException.ThrowIfNull(httpContext); + + return _session.WriteStreamState(_plan, httpContext, ContentType, OnFoundStatus); + } + + /// + /// Populates endpoint metadata so OpenAPI correctly advertises a + /// 200: StreamStateResponse and 404 response for this endpoint. + /// + public static void PopulateMetadata(MethodInfo method, EndpointBuilder builder) + { + ArgumentNullException.ThrowIfNull(builder); + + builder.Metadata.Add(new ProducesResponseTypeMetadata( + StatusCodes.Status200OK, typeof(StreamStateResponse), ["application/json"])); + builder.Metadata.Add(new ProducesResponseTypeMetadata( + StatusCodes.Status404NotFound, typeof(void), [])); + } +} diff --git a/src/Polecat.AspNetCore/StreamEvents.cs b/src/Polecat.AspNetCore/StreamEvents.cs new file mode 100644 index 00000000..0dcc7f5b --- /dev/null +++ b/src/Polecat.AspNetCore/StreamEvents.cs @@ -0,0 +1,117 @@ +using System.Diagnostics.CodeAnalysis; +using System.Reflection; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Metadata; + +namespace Polecat.AspNetCore; + +/// +/// Minimal-API endpoint return value that writes the raw events of a single event stream to the +/// response as a JSON array. Backed by , the +/// same query plan that can be batched through IBatchedQuery.QueryByPlan(), and carrying the +/// same optional version, timestamp and fromVersion filters as +/// FetchStreamAsync(). +/// +/// FetchStream yields an empty list both for a stream that does not exist and for a filter +/// that excludes every event, so the two cannot be told apart here. +/// decides which answer the endpoint gives; it defaults to 404 to match the other +/// single-resource results. Set it to 200 to return an empty JSON array instead — the right choice +/// for an endpoint that pages through a stream with fromVersion, where running off the end is +/// expected rather than exceptional. +/// +/// +/// Elements are , not Polecat's IEvent directly: +/// IEvent.EventType is a and System.Text.Json refuses to serialize those. +/// Use EventTypeName, Polecat's stable event type alias, to discriminate event types client +/// side; the assembly qualified .NET type name is deliberately not written to the wire. +/// +/// +public sealed class StreamEvents : IResult, IEndpointMetadataProvider +{ + private readonly IQuerySession _session; + private readonly FetchStreamPlan _plan; + + /// + /// Write the events of the Guid-identified stream . + /// + /// + /// + /// If set, writes events up to and including this version + /// If set, writes events captured on or before this timestamp + /// If set, writes events on or from this version + public StreamEvents(IQuerySession session, Guid streamId, long version = 0, + DateTimeOffset? timestamp = null, long fromVersion = 0) + : this(session, new FetchStreamPlan(streamId, version, timestamp, fromVersion)) + { + } + + /// + /// Write the events of the string-keyed stream . + /// + /// + /// + /// If set, writes events up to and including this version + /// If set, writes events captured on or before this timestamp + /// If set, writes events on or from this version + public StreamEvents(IQuerySession session, string streamKey, long version = 0, + DateTimeOffset? timestamp = null, long fromVersion = 0) + : this(session, new FetchStreamPlan( + streamKey ?? throw new ArgumentNullException(nameof(streamKey)), version, timestamp, fromVersion)) + { + } + + /// + /// Write the events resolved by an existing . Lets a handler build + /// the plan once and either batch it or return it straight from an endpoint. + /// + public StreamEvents(IQuerySession session, FetchStreamPlan plan) + { + _session = session ?? throw new ArgumentNullException(nameof(session)); + _plan = plan ?? throw new ArgumentNullException(nameof(plan)); + } + + /// + /// Status code written when the stream yields at least one event. Defaults to 200. + /// + public int OnFoundStatus { get; init; } = StatusCodes.Status200OK; + + /// + /// Status code written when the stream yields no events at all. Defaults to 404. + /// Set to 200 to write an empty JSON array instead. + /// + public int OnEmptyStatus { get; init; } = StatusCodes.Status404NotFound; + + /// + /// Response content type. Defaults to application/json. + /// + public string ContentType { get; init; } = "application/json"; + + /// + [UnconditionalSuppressMessage("Trimming", "IL2046", + Justification = "IResult.ExecuteAsync is not RUC-annotated; the contract lives on this override.")] + [UnconditionalSuppressMessage("AOT", "IL3051", + Justification = "IResult.ExecuteAsync is not RDC-annotated; the contract lives on this override.")] + [RequiresDynamicCode("Serializes EventResponse[] with System.Text.Json, which uses runtime codegen for each event's Data payload.")] + [RequiresUnreferencedCode("Reflects over EventResponse and each event's Data payload via System.Text.Json.")] + public Task ExecuteAsync(HttpContext httpContext) + { + ArgumentNullException.ThrowIfNull(httpContext); + + return _session.WriteEvents(_plan, httpContext, ContentType, OnFoundStatus, OnEmptyStatus); + } + + /// + /// Populates endpoint metadata so OpenAPI correctly advertises a + /// 200: EventResponse[] and 404 response for this endpoint. + /// + public static void PopulateMetadata(MethodInfo method, EndpointBuilder builder) + { + ArgumentNullException.ThrowIfNull(builder); + + builder.Metadata.Add(new ProducesResponseTypeMetadata( + StatusCodes.Status200OK, typeof(EventResponse[]), ["application/json"])); + builder.Metadata.Add(new ProducesResponseTypeMetadata( + StatusCodes.Status404NotFound, typeof(void), [])); + } +} diff --git a/src/Polecat.Tests/Batching/batch_event_fetching.cs b/src/Polecat.Tests/Batching/batch_event_fetching.cs new file mode 100644 index 00000000..388eaceb --- /dev/null +++ b/src/Polecat.Tests/Batching/batch_event_fetching.cs @@ -0,0 +1,223 @@ +using JasperFx.Events.Projections; +using Polecat.Projections; +using Polecat.Tests.Harness; +using Shouldly; + +namespace Polecat.Tests.Batching; + +/// +/// #370: IBatchedQuery.Events — the batched counterparts of FetchStreamStateAsync and +/// FetchStreamAsync. These pin the surface directly rather than through the query plans that +/// sit on top of it, and pin that the batched read agrees with the standalone one row for row: the +/// batch item composes its own SQL, so drift between the two paths is the risk worth guarding. +/// +[Collection("integration")] +public class batch_event_fetching : IntegrationContext +{ + public batch_event_fetching(DefaultStoreFixture fixture) : base(fixture) + { + } + + [Fact] + public async Task fetch_stream_state_in_a_batch() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + var fetcher = batch.Events.FetchStreamState(streamId); + await batch.Execute(); + + var state = await fetcher; + state.ShouldNotBeNull(); + state.Id.ShouldBe(streamId); + state.Version.ShouldBe(3); + } + + /// + /// #370: pc_streams.type was always projected and never read, so + /// StreamState.AggregateType came back null on every stream — which would have made the + /// StreamStateResponse.AggregateTypeName wire field structurally dead. Both the standalone + /// and batched reads resolve it now. + /// + [Fact] + public async Task stream_state_reports_the_aggregate_type_it_was_tagged_with() + { + await StoreOptions(opts => + opts.Projections.Add>( + ProjectionLifecycle.Live)); + + var streamId = Guid.NewGuid(); + await using (var session = theStore.LightweightSession()) + { + session.Events.StartStream(streamId, new QuestStarted("Tagged")); + await session.SaveChangesAsync(); + } + + await using var query = theStore.QuerySession(); + + (await query.Events.FetchStreamStateAsync(streamId))!.AggregateType.ShouldBe(typeof(BatchTaggedAggregate)); + + var batch = query.CreateBatchQuery(); + var fetcher = batch.Events.FetchStreamState(streamId); + await batch.Execute(); + (await fetcher)!.AggregateType.ShouldBe(typeof(BatchTaggedAggregate)); + } + + /// + /// An untagged stream simply has no aggregate type — not an error, and not a reason to fail the + /// metadata read. + /// + [Fact] + public async Task stream_state_reports_no_aggregate_type_for_an_untagged_stream() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + var fetcher = batch.Events.FetchStreamState(streamId); + await batch.Execute(); + + (await fetcher)!.AggregateType.ShouldBeNull(); + } + + [Fact] + public async Task fetch_stream_state_is_null_for_a_missing_stream() + { + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + var fetcher = batch.Events.FetchStreamState(Guid.NewGuid()); + await batch.Execute(); + + (await fetcher).ShouldBeNull(); + } + + [Fact] + public async Task fetch_stream_in_a_batch_matches_the_standalone_fetch() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var expected = await query.Events.FetchStreamAsync(streamId); + + var batch = query.CreateBatchQuery(); + var fetcher = batch.Events.FetchStream(streamId); + await batch.Execute(); + + var actual = await fetcher; + actual.Count.ShouldBe(expected.Count); + actual.Select(x => x.Id).ShouldBe(expected.Select(x => x.Id)); + actual.Select(x => x.Version).ShouldBe(expected.Select(x => x.Version)); + actual.Select(x => x.Sequence).ShouldBe(expected.Select(x => x.Sequence)); + actual.ShouldAllBe(x => x.StreamId == streamId); + actual[0].Data.ShouldBeOfType().Name.ShouldBe("Quest 1"); + } + + [Fact] + public async Task fetch_stream_applies_the_optional_filters() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + var capped = batch.Events.FetchStream(streamId, version: 2); + var from = batch.Events.FetchStream(streamId, fromVersion: 2); + var window = batch.Events.FetchStream(streamId, version: 2, fromVersion: 2); + await batch.Execute(); + + (await capped).Select(x => x.Version).ShouldBe([1, 2]); + (await from).Select(x => x.Version).ShouldBe([2, 3]); + (await window).Select(x => x.Version).ShouldBe([2]); + } + + /// + /// The timestamp filter gets its own test because it is the one parameter whose binding differs in + /// kind from the rest — a DateTimeOffset going through the batch's ICommandBuilder rather than the + /// standalone fetch's AddWithValue. + /// + [Fact] + public async Task fetch_stream_applies_the_timestamp_filter() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var all = await query.Events.FetchStreamAsync(streamId); + var cutoff = all[^1].Timestamp; + + var batch = query.CreateBatchQuery(); + var upToCutoff = batch.Events.FetchStream(streamId, timestamp: cutoff); + var beforeEverything = batch.Events.FetchStream(streamId, timestamp: all[0].Timestamp.AddMinutes(-5)); + await batch.Execute(); + + (await upToCutoff).Count.ShouldBe(3); + (await beforeEverything).ShouldBeEmpty(); + } + + [Fact] + public async Task fetch_stream_is_empty_for_a_missing_stream() + { + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + var fetcher = batch.Events.FetchStream(Guid.NewGuid()); + await batch.Execute(); + + (await fetcher).ShouldBeEmpty(); + } + + [Fact] + public async Task several_event_fetches_and_a_document_load_resolve_in_one_batch() + { + var first = await StartQuestStreamAsync(); + var second = await StartQuestStreamAsync(); + + var target = new Target { Id = Guid.NewGuid(), Color = "Green", Number = 42 }; + await using (var session = theStore.LightweightSession()) + { + session.Store(target); + await session.SaveChangesAsync(); + } + + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + + // Interleaved deliberately: every item reads the result set at its own ordinal, so an item that + // consumed the wrong one would show up as a cross-wired answer here rather than as a clean error. + var firstState = batch.Events.FetchStreamState(first); + var doc = batch.Load(target.Id); + var secondEvents = batch.Events.FetchStream(second); + var secondState = batch.Events.FetchStreamState(second); + var firstEvents = batch.Events.FetchStream(first); + await batch.Execute(); + + (await firstState)!.Id.ShouldBe(first); + (await secondState)!.Id.ShouldBe(second); + (await firstEvents).ShouldAllBe(x => x.StreamId == first); + (await secondEvents).ShouldAllBe(x => x.StreamId == second); + (await doc)!.Number.ShouldBe(42); + } + + private async Task StartQuestStreamAsync() + { + var streamId = Guid.NewGuid(); + await using var session = theStore.LightweightSession(); + session.Events.StartStream(streamId, + new QuestStarted("Quest 1"), new QuestStarted("Quest 2"), new QuestStarted("Quest 3")); + await session.SaveChangesAsync(); + return streamId; + } +} + +/// +/// Tagged onto a stream by StartStream<T> and registered as a projection, which is what +/// makes the persisted alias resolvable — Polecat's QuickAppend writer stores +/// AggregateType.Name directly and never goes through StreamAction.PrepareEvents, so the +/// registered projections are the source of truth for resolving it back. +/// +public class BatchTaggedAggregate +{ + public Guid Id { get; set; } + + public void Apply(QuestStarted e) + { + } +} diff --git a/src/Polecat.Tests/Querying/fetching_stream_query_plans.cs b/src/Polecat.Tests/Querying/fetching_stream_query_plans.cs new file mode 100644 index 00000000..fb2f2208 --- /dev/null +++ b/src/Polecat.Tests/Querying/fetching_stream_query_plans.cs @@ -0,0 +1,232 @@ +using JasperFx; +using JasperFx.Events; +using Polecat.Tests.Harness; +using Polecat.TestUtils; +using Shouldly; + +namespace Polecat.Tests.Querying; + +/// +/// #370 (parity with marten#5053): and +/// wrap the raw event-stream fetches as query plans. Both implement +/// both and , so each is +/// exercised standalone and through a batch here. +/// +[Collection("integration")] +public class fetching_stream_query_plans : IntegrationContext +{ + public fetching_stream_query_plans(DefaultStoreFixture fixture) : base(fixture) + { + } + + [Fact] + public async Task fetch_stream_state_plan_standalone() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var state = await query.QueryByPlanAsync(new FetchStreamStatePlan(streamId)); + + state.ShouldNotBeNull(); + state.Id.ShouldBe(streamId); + state.Version.ShouldBe(3); + state.IsArchived.ShouldBeFalse(); + } + + [Fact] + public async Task fetch_stream_state_plan_batched() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + var fetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId)); + await batch.Execute(); + + var state = await fetcher; + state.ShouldNotBeNull(); + state.Id.ShouldBe(streamId); + state.Version.ShouldBe(3); + } + + [Fact] + public async Task fetch_stream_state_plan_yields_null_for_a_missing_stream() + { + await using var query = theStore.QuerySession(); + + (await query.QueryByPlanAsync(new FetchStreamStatePlan(Guid.NewGuid()))).ShouldBeNull(); + + var batch = query.CreateBatchQuery(); + var fetcher = batch.QueryByPlan(new FetchStreamStatePlan(Guid.NewGuid())); + await batch.Execute(); + (await fetcher).ShouldBeNull(); + } + + [Fact] + public async Task fetch_stream_plan_standalone() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var events = await query.QueryByPlanAsync(new FetchStreamPlan(streamId)); + + events.Count.ShouldBe(3); + events.Select(x => x.Version).ShouldBe([1, 2, 3]); + events[0].Data.ShouldBeOfType().Name.ShouldBe("Quest 1"); + } + + [Fact] + public async Task fetch_stream_plan_batched() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + var fetcher = batch.QueryByPlan(new FetchStreamPlan(streamId)); + await batch.Execute(); + + var events = await fetcher; + events.Count.ShouldBe(3); + events.Select(x => x.Version).ShouldBe([1, 2, 3]); + events.ShouldAllBe(x => x.StreamId == streamId); + } + + [Fact] + public async Task fetch_stream_plan_honors_the_version_cap() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + + // Standalone and batched must apply the filter identically — the batched item composes its own + // SQL, so the cap is the thing most likely to drift between the two paths. + var standalone = await query.QueryByPlanAsync(new FetchStreamPlan(streamId, version: 2)); + standalone.Count.ShouldBe(2); + + var batch = query.CreateBatchQuery(); + var fetcher = batch.QueryByPlan(new FetchStreamPlan(streamId, version: 2)); + await batch.Execute(); + (await fetcher).Count.ShouldBe(2); + } + + [Fact] + public async Task fetch_stream_plan_honors_from_version() + { + var streamId = await StartQuestStreamAsync(); + + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + var fetcher = batch.QueryByPlan(new FetchStreamPlan(streamId, fromVersion: 3)); + await batch.Execute(); + + var events = await fetcher; + events.Count.ShouldBe(1); + events[0].Version.ShouldBe(3); + } + + [Fact] + public async Task fetch_stream_plan_yields_an_empty_list_for_a_missing_stream() + { + await using var query = theStore.QuerySession(); + + (await query.QueryByPlanAsync(new FetchStreamPlan(Guid.NewGuid()))).ShouldBeEmpty(); + + var batch = query.CreateBatchQuery(); + var fetcher = batch.QueryByPlan(new FetchStreamPlan(Guid.NewGuid())); + await batch.Execute(); + (await fetcher).ShouldBeEmpty(); + } + + [Fact] + public async Task both_plans_share_one_round_trip_with_document_loads() + { + var streamId = await StartQuestStreamAsync(); + + var target = new Target { Id = Guid.NewGuid(), Color = "Blue", Number = 7 }; + await using (var session = theStore.LightweightSession()) + { + session.Store(target); + await session.SaveChangesAsync(); + } + + await using var query = theStore.QuerySession(); + var batch = query.CreateBatchQuery(); + var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId)); + var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamId)); + var docFetcher = batch.Load(target.Id); + await batch.Execute(); + + (await stateFetcher)!.Version.ShouldBe(3); + (await eventsFetcher).Count.ShouldBe(3); + (await docFetcher)!.Number.ShouldBe(7); + } + + private async Task StartQuestStreamAsync() + { + var streamId = Guid.NewGuid(); + await using var session = theStore.LightweightSession(); + session.Events.StartStream(streamId, + new QuestStarted("Quest 1"), new QuestStarted("Quest 2"), new QuestStarted("Quest 3")); + await session.SaveChangesAsync(); + return streamId; + } +} + +/// +/// #370: the string-identity half. Stream identity is fixed at store construction, so the +/// streamKey constructor overloads need their own store rather than the shared Guid fixture. +/// +public class fetching_stream_query_plans_by_string_key : IAsyncLifetime +{ + private const string Schema = "fetch_stream_plans_str"; + private DocumentStore _store = null!; + + public Task InitializeAsync() + { + _store = DocumentStore.For(opts => + { + opts.ConnectionString = ConnectionSource.ConnectionString; + opts.DatabaseSchemaName = Schema; + opts.AutoCreateSchemaObjects = AutoCreate.All; + opts.UseNativeJsonType = ConnectionSource.SupportsNativeJson; + opts.Events.StreamIdentity = StreamIdentity.AsString; + }); + + return Task.CompletedTask; + } + + public Task DisposeAsync() + { + _store.Dispose(); + return Task.CompletedTask; + } + + [Fact] + public async Task both_plans_resolve_a_string_keyed_stream() + { + var streamKey = "quest/" + Guid.NewGuid().ToString("N"); + + await using (var session = _store.LightweightSession()) + { + session.Events.StartStream(streamKey, new QuestStarted("A"), new QuestStarted("B")); + await session.SaveChangesAsync(); + } + + await using var query = _store.QuerySession(); + + var state = await query.QueryByPlanAsync(new FetchStreamStatePlan(streamKey)); + state.ShouldNotBeNull(); + state.Key.ShouldBe(streamKey); + state.Version.ShouldBe(2); + + var batch = query.CreateBatchQuery(); + var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamKey)); + var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamKey)); + await batch.Execute(); + + (await stateFetcher)!.Key.ShouldBe(streamKey); + var events = await eventsFetcher; + events.Count.ShouldBe(2); + events.ShouldAllBe(x => x.StreamKey == streamKey); + } +} diff --git a/src/Polecat/Batching/IBatchEvents.cs b/src/Polecat/Batching/IBatchEvents.cs new file mode 100644 index 00000000..58db4acd --- /dev/null +++ b/src/Polecat/Batching/IBatchEvents.cs @@ -0,0 +1,51 @@ +using JasperFx.Events; + +namespace Polecat.Batching; + +/// +/// The event-store surface of a batched query — the batched counterparts of +/// and +/// . +/// +/// +/// #370 (parity with marten#5053). Reached through . Every method +/// returns immediately with an unresolved ; the task completes when +/// runs the whole batch as a single round trip and walks its +/// result sets in order. +/// +public interface IBatchEvents +{ + /// + /// Fetch the high level metadata about the stream identified by . + /// Yields null if the stream does not exist. + /// + Task FetchStreamState(Guid streamId); + + /// + /// Fetch the high level metadata about the stream identified by . + /// Yields null if the stream does not exist. + /// + Task FetchStreamState(string streamKey); + + /// + /// Fetch the raw events of the stream identified by . Yields an empty + /// list if the stream does not exist. + /// + /// + /// If set, fetches events up to and including this version + /// If set, fetches events captured on or before this timestamp + /// If set, fetches events on or from this version + Task> FetchStream(Guid streamId, long version = 0, + DateTimeOffset? timestamp = null, long fromVersion = 0); + + /// + /// Fetch the raw events of the stream identified by . Yields an empty + /// list if the stream does not exist. + /// + /// + /// If set, fetches events up to and including this version + /// If set, fetches events captured on or before this timestamp + /// If set, fetches events on or from this version + Task> FetchStream(string streamKey, long version = 0, + DateTimeOffset? timestamp = null, long fromVersion = 0); +} diff --git a/src/Polecat/Batching/IBatchedQuery.cs b/src/Polecat/Batching/IBatchedQuery.cs index c2c1333a..95e538e8 100644 --- a/src/Polecat/Batching/IBatchedQuery.cs +++ b/src/Polecat/Batching/IBatchedQuery.cs @@ -14,6 +14,13 @@ public interface IBatchedQuery /// IQuerySession Parent { get; } + /// + /// The batched event store fetches — FetchStreamState and FetchStream — so a raw + /// stream read can share the batch's single round trip with document loads and LINQ queries. + /// See (#370). + /// + IBatchEvents Events { get; } + /// /// Check if a document of type T with the given Guid id exists in the database /// without loading or deserializing the document. diff --git a/src/Polecat/Events/EventGraph.cs b/src/Polecat/Events/EventGraph.cs index b675408d..4c0bc2c5 100644 --- a/src/Polecat/Events/EventGraph.cs +++ b/src/Polecat/Events/EventGraph.cs @@ -341,6 +341,47 @@ public override Type AggregateTypeFor(string aggregateTypeName) $"Unknown aggregate type name '{aggregateTypeName}'."); } + /// + /// #370: resolve the alias persisted in pc_streams.type back to its aggregate type, or null + /// when this deployment has no registration for it. + /// + /// + /// + /// Unlike this never throws: a stream tagged by a deployment that + /// knew a type this one does not must still report its version and timestamps. + /// + /// + /// _aggregateTypes alone is not enough. It is only populated by + /// , which JasperFx.Events calls from + /// StreamAction.PrepareEvents — a path Polecat's QuickAppend closed-shape writer does not go + /// through, since the SQL Server dialect writes stream.AggregateType?.Name straight into the + /// column. So the registered projections are the primary source and the map is the fallback, + /// matching what the event store explorer's ResolveAggregateType already does. + /// + /// + internal Type? TryResolveAggregateType(string? aggregateTypeName) + { + if (string.IsNullOrEmpty(aggregateTypeName)) return null; + + if (_aggregateTypes.TryGetValue(aggregateTypeName, out var known)) return known; + + foreach (var source in _options.Projections.All) + { + foreach (var published in source.PublishedTypes()) + { + if (string.Equals(published.Name, aggregateTypeName, StringComparison.Ordinal) + || string.Equals(published.FullName, aggregateTypeName, StringComparison.Ordinal)) + { + // Cache it so the next row on this reader is a dictionary hit. + _aggregateTypes.TryAdd(aggregateTypeName, published); + return published; + } + } + } + + return null; + } + public override string AggregateAliasFor(Type aggregateType) { _aggregateTypes.TryAdd(aggregateType.Name, aggregateType); diff --git a/src/Polecat/Events/Internal/PcStreamsRowReader.cs b/src/Polecat/Events/Internal/PcStreamsRowReader.cs index 573f0f9a..ef68fbb0 100644 --- a/src/Polecat/Events/Internal/PcStreamsRowReader.cs +++ b/src/Polecat/Events/Internal/PcStreamsRowReader.cs @@ -44,7 +44,17 @@ internal static string SelectColumnsWithAlias(string alias) => /// . The caller is responsible for /// await reader.ReadAsync(...) beforehand. /// - internal static StreamState ReadStreamState(DbDataReader reader, StreamIdentity streamIdentity) + /// + /// + /// + /// Supply to resolve the type column's alias into + /// . #370: the column has always been projected but never + /// read, so AggregateType came back null on every stream that was in fact tagged with an + /// aggregate type. An unregistered alias resolves to null rather than throwing — a stream tagged + /// by a deployment that knew a type this one does not is not an error for a metadata read. + /// + internal static StreamState ReadStreamState(DbDataReader reader, StreamIdentity streamIdentity, + EventGraph? events = null) { var state = new StreamState { @@ -54,6 +64,11 @@ internal static StreamState ReadStreamState(DbDataReader reader, StreamIdentity IsArchived = reader.GetBoolean(6) }; + if (events != null && !reader.IsDBNull(1)) + { + state.AggregateType = events.TryResolveAggregateType(reader.GetString(1)); + } + if (streamIdentity == StreamIdentity.AsGuid) { state.Id = reader.GetGuid(0); diff --git a/src/Polecat/Events/QueryEventStore.cs b/src/Polecat/Events/QueryEventStore.cs index 20d23824..9c069588 100644 --- a/src/Polecat/Events/QueryEventStore.cs +++ b/src/Polecat/Events/QueryEventStore.cs @@ -288,7 +288,7 @@ private async Task> FetchStreamInternalAsync(object stream await using var reader = await _session.ExecuteReaderAsync(cmd, token); if (await reader.ReadAsync(token)) { - return PcStreamsRowReader.ReadStreamState(reader, _events.StreamIdentity); + return PcStreamsRowReader.ReadStreamState(reader, _events.StreamIdentity, _events); } return null; diff --git a/src/Polecat/IQueryPlan.cs b/src/Polecat/IQueryPlan.cs index 7b0e2bda..ae2add8f 100644 --- a/src/Polecat/IQueryPlan.cs +++ b/src/Polecat/IQueryPlan.cs @@ -1,3 +1,4 @@ +using JasperFx.Events; using Polecat.Batching; using Polecat.Internal.Batching; using Polecat.Linq; @@ -55,3 +56,123 @@ Task> IBatchQueryPlan>.Fetch(IBatchedQuery que return Query(query.Parent).ToListAsync(); } } + +/// +/// Query plan for the high level metadata of a single event stream, identified by either a Guid +/// stream id or a string stream key. Yields null if the stream does not exist. +/// +/// +/// #370 (parity with marten#5053). Implements both and +/// , so the same plan instance works with +/// session.QueryByPlanAsync() and batch.QueryByPlan(). Implementing only the batched +/// half matters beyond convenience: through Wolverine's fetch-specification feature, a plan that is +/// only an produces uncompilable generated code. +/// +public class FetchStreamStatePlan : IQueryPlan, IBatchQueryPlan +{ + private readonly Guid _streamId; + private readonly string? _streamKey; + + /// + /// Fetch the stream state for the stream identified by . + /// + public FetchStreamStatePlan(Guid streamId) + { + _streamId = streamId; + } + + /// + /// Fetch the stream state for the stream identified by . + /// + public FetchStreamStatePlan(string streamKey) + { + _streamKey = streamKey ?? throw new ArgumentNullException(nameof(streamKey)); + } + + public Task Fetch(IQuerySession session, CancellationToken token) + { + ArgumentNullException.ThrowIfNull(session); + + return _streamKey is not null + ? session.Events.FetchStreamStateAsync(_streamKey, token) + : session.Events.FetchStreamStateAsync(_streamId, token); + } + + public Task Fetch(IBatchedQuery query) + { + ArgumentNullException.ThrowIfNull(query); + + return _streamKey is not null + ? query.Events.FetchStreamState(_streamKey) + : query.Events.FetchStreamState(_streamId); + } +} + +/// +/// Query plan for the raw events of a single event stream, identified by either a Guid stream id or +/// a string stream key, carrying FetchStream's optional version / timestamp / +/// fromVersion filters. Yields an empty list if the stream does not exist. +/// +/// +/// #370 (parity with marten#5053). Implements both and +/// — see for why the pair +/// matters. +/// +public class FetchStreamPlan : IQueryPlan>, IBatchQueryPlan> +{ + private readonly Guid _streamId; + private readonly string? _streamKey; + private readonly long _version; + private readonly DateTimeOffset? _timestamp; + private readonly long _fromVersion; + + /// + /// Fetch the events for the stream identified by . + /// + /// + /// If set, queries for events up to and including this version + /// If set, queries for events captured on or before this timestamp + /// If set, queries for events on or from this version + public FetchStreamPlan(Guid streamId, long version = 0, DateTimeOffset? timestamp = null, + long fromVersion = 0) + { + _streamId = streamId; + _version = version; + _timestamp = timestamp; + _fromVersion = fromVersion; + } + + /// + /// Fetch the events for the stream identified by . + /// + /// + /// If set, queries for events up to and including this version + /// If set, queries for events captured on or before this timestamp + /// If set, queries for events on or from this version + public FetchStreamPlan(string streamKey, long version = 0, DateTimeOffset? timestamp = null, + long fromVersion = 0) + { + _streamKey = streamKey ?? throw new ArgumentNullException(nameof(streamKey)); + _version = version; + _timestamp = timestamp; + _fromVersion = fromVersion; + } + + public Task> Fetch(IQuerySession session, CancellationToken token) + { + ArgumentNullException.ThrowIfNull(session); + + return _streamKey is not null + ? session.Events.FetchStreamAsync(_streamKey, _version, _timestamp, _fromVersion, token) + : session.Events.FetchStreamAsync(_streamId, _version, _timestamp, _fromVersion, token); + } + + public Task> Fetch(IBatchedQuery query) + { + ArgumentNullException.ThrowIfNull(query); + + return _streamKey is not null + ? query.Events.FetchStream(_streamKey, _version, _timestamp, _fromVersion) + : query.Events.FetchStream(_streamId, _version, _timestamp, _fromVersion); + } +} diff --git a/src/Polecat/Internal/Batching/BatchEvents.cs b/src/Polecat/Internal/Batching/BatchEvents.cs new file mode 100644 index 00000000..5dbabc2b --- /dev/null +++ b/src/Polecat/Internal/Batching/BatchEvents.cs @@ -0,0 +1,56 @@ +using JasperFx.Events; +using Polecat.Batching; +using Polecat.Events; + +namespace Polecat.Internal.Batching; + +/// +/// #370: over . Each call appends one item to +/// the batch and hands back its unresolved task; the tasks complete when +/// walks the result sets. +/// +internal class BatchEvents : IBatchEvents +{ + private readonly BatchedQuery _parent; + private readonly QuerySession _session; + + public BatchEvents(BatchedQuery parent, QuerySession session) + { + _parent = parent; + _session = session; + } + + public Task FetchStreamState(Guid streamId) => AddStateItem(streamId); + + public Task FetchStreamState(string streamKey) + => AddStateItem(streamKey ?? throw new ArgumentNullException(nameof(streamKey))); + + public Task> FetchStream(Guid streamId, long version = 0, + DateTimeOffset? timestamp = null, long fromVersion = 0) + => AddStreamItem(streamId, version, timestamp, fromVersion); + + public Task> FetchStream(string streamKey, long version = 0, + DateTimeOffset? timestamp = null, long fromVersion = 0) + => AddStreamItem(streamKey ?? throw new ArgumentNullException(nameof(streamKey)), + version, timestamp, fromVersion); + + private Task AddStateItem(object streamId) + { + var item = new FetchStreamStateBatchItem(EventGraph(), streamId, _session.TenantId); + _parent.RequireEventStore(); + _parent.AddItem(item); + return item.Result; + } + + private Task> AddStreamItem(object streamId, long version, + DateTimeOffset? timestamp, long fromVersion) + { + var item = new FetchStreamBatchItem(EventGraph(), _session.Serializer, streamId, _session.TenantId, + version, timestamp, fromVersion); + _parent.RequireEventStore(); + _parent.AddItem(item); + return item.Result; + } + + private EventGraph EventGraph() => _session.Options.EventGraph; +} diff --git a/src/Polecat/Internal/Batching/BatchedQuery.cs b/src/Polecat/Internal/Batching/BatchedQuery.cs index 5e7405f4..97dbcdef 100644 --- a/src/Polecat/Internal/Batching/BatchedQuery.cs +++ b/src/Polecat/Internal/Batching/BatchedQuery.cs @@ -26,8 +26,19 @@ public BatchedQuery(QuerySession session, DocumentProviderRegistry providers, public IQuerySession Parent => _session; + // #370: lazily built so a batch that never touches the event store pays nothing for it. + public IBatchEvents Events => _events ??= new BatchEvents(this, _session); + private IBatchEvents? _events; + internal void AddItem(IBatchQueryItem item) => _items.Add(item); + // #219 create-on-first-use: the standalone FetchStream/FetchStreamState calls ensure the event store + // schema before reading. A batched fetch has to do the same, or the very first event read against a + // fresh store fails on a missing table. Flagged rather than ensured eagerly so a document-only batch + // never pays for it. + internal void RequireEventStore() => _needsEventStore = true; + private bool _needsEventStore; + internal void TrackProvider(DocumentProvider provider) => _involvedProviders.Add(provider); public Task CheckExists(Guid id) where T : class => AddCheckExists(id); @@ -104,6 +115,11 @@ public async Task Execute(CancellationToken token = default) // Ensure tables exist for all involved document types await _tableEnsurer.EnsureTablesAsync(_involvedProviders, token); + if (_needsEventStore) + { + await _tableEnsurer.EnsureEventStoreSchemaAsync(token); + } + // Build the combined batch (connection-less; lifetime sets it at execution) await using var batch = new SqlBatch(); var builder = new BatchBuilder(batch); diff --git a/src/Polecat/Internal/Batching/FetchStreamBatchItem.cs b/src/Polecat/Internal/Batching/FetchStreamBatchItem.cs new file mode 100644 index 00000000..5cc55c6c --- /dev/null +++ b/src/Polecat/Internal/Batching/FetchStreamBatchItem.cs @@ -0,0 +1,107 @@ +using System.Data; +using System.Data.Common; +using System.Diagnostics.CodeAnalysis; +using JasperFx.Events; +using Polecat.Events; +using Polecat.Events.Internal; +using Polecat.Serialization; +using Weasel.SqlServer; + +namespace Polecat.Internal.Batching; + +/// +/// #370: the batched half of QueryEventStore.FetchStreamAsync, carrying the same optional +/// version / timestamp / fromVersion filters. Composes its projection with +/// and hydrates through the same readers as the standalone fetch, +/// so the two can never drift apart across a schema migration. +/// +[UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", + Justification = "Class-level: hydrates IEvent batches via PcEventsRowReader (routed through ISerializer.FromJson). Event types are preserved by EventGraph registration on the caller side per the AOT publishing guide.")] +[UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", + Justification = "Class-level: ISerializer.FromJson and Event.MakeGenericType are annotated RDC. AOT consumers register concrete event types ahead of time.")] +internal class FetchStreamBatchItem : IBatchQueryItem +{ + private readonly TaskCompletionSource> _tcs = new(); + private readonly EventGraph _events; + private readonly ISerializer _serializer; + private readonly object _streamId; + private readonly string _tenantId; + private readonly long _version; + private readonly DateTimeOffset? _timestamp; + private readonly long _fromVersion; + + public FetchStreamBatchItem(EventGraph events, ISerializer serializer, object streamId, string tenantId, + long version, DateTimeOffset? timestamp, long fromVersion) + { + _events = events; + _serializer = serializer; + _streamId = streamId; + _tenantId = tenantId; + _version = version; + _timestamp = timestamp; + _fromVersion = fromVersion; + } + + public Task> Result => _tcs.Task; + + public void WriteSql(ICommandBuilder builder) + { + builder.Append( + $"SELECT {PcEventsRowReader.ComposeSelectColumns(_events.EventOptions)} FROM {_events.EventsTableName} WHERE stream_id = "); + builder.AppendParameter(_streamId, _streamId is string ? SqlDbType.VarChar : null); + builder.Append(" AND tenant_id = "); + builder.AppendParameter(_tenantId, SqlDbType.VarChar); + builder.Append(" AND is_archived = 0"); + + if (_version > 0) + { + builder.Append(" AND version <= "); + builder.AppendParameter(_version); + } + + if (_timestamp.HasValue) + { + builder.Append(" AND timestamp <= "); + builder.AppendParameter(_timestamp.Value); + } + + if (_fromVersion > 0) + { + builder.Append(" AND version >= "); + builder.AppendParameter(_fromVersion); + } + + builder.Append(" ORDER BY version;\n"); + } + + public async Task ReadResultSetAsync(DbDataReader reader, CancellationToken token) + { + var ctx = new EventHydrationContext(_events, _serializer, _streamId, defaultTenantId: _tenantId); + + // Same per-batch hoists as the standalone fetch: metadata ordinals computed once, a single-slot + // type→mapping cache, and the StreamIdentity specialization picked once rather than per row. + var slots = MetadataSlots.Compute(_events.EventOptions); + var cache = new EventTypeCache(); + + var results = new List(); + + if (_events.StreamIdentity == StreamIdentity.AsGuid) + { + while (await reader.ReadAsync(token).ConfigureAwait(false)) + { + var @event = PcEventsRowReader.ReadEventAsGuid(reader, ctx, slots, ref cache); + if (@event != null) results.Add(@event); + } + } + else + { + while (await reader.ReadAsync(token).ConfigureAwait(false)) + { + var @event = PcEventsRowReader.ReadEventAsString(reader, ctx, slots, ref cache); + if (@event != null) results.Add(@event); + } + } + + _tcs.SetResult(results); + } +} diff --git a/src/Polecat/Internal/Batching/FetchStreamStateBatchItem.cs b/src/Polecat/Internal/Batching/FetchStreamStateBatchItem.cs new file mode 100644 index 00000000..33f17644 --- /dev/null +++ b/src/Polecat/Internal/Batching/FetchStreamStateBatchItem.cs @@ -0,0 +1,52 @@ +using System.Data; +using System.Data.Common; +using JasperFx.Events; +using Polecat.Events; +using Polecat.Events.Internal; +using Weasel.SqlServer; + +namespace Polecat.Internal.Batching; + +/// +/// #370: the batched half of QueryEventStore.FetchStreamStateAsync. Reads the same +/// pc_streams column projection through the same , so a +/// batched fetch and a standalone one can never drift apart across a schema migration. +/// +internal class FetchStreamStateBatchItem : IBatchQueryItem +{ + private readonly TaskCompletionSource _tcs = new(); + private readonly EventGraph _events; + private readonly object _streamId; + private readonly string _tenantId; + + public FetchStreamStateBatchItem(EventGraph events, object streamId, string tenantId) + { + _events = events; + _streamId = streamId; + _tenantId = tenantId; + } + + public Task Result => _tcs.Task; + + public void WriteSql(ICommandBuilder builder) + { + builder.Append($"SELECT {PcStreamsRowReader.SelectColumns} FROM {_events.StreamsTableName} WHERE id = "); + builder.AppendParameter(_streamId, _streamId is string ? SqlDbType.VarChar : null); + builder.Append(" AND tenant_id = "); + builder.AppendParameter(_tenantId, SqlDbType.VarChar); + builder.Append(";\n"); + } + + public async Task ReadResultSetAsync(DbDataReader reader, CancellationToken token) + { + if (await reader.ReadAsync(token).ConfigureAwait(false)) + { + _tcs.SetResult(PcStreamsRowReader.ReadStreamState(reader, _events.StreamIdentity, _events)); + } + else + { + // A stream that does not exist is null, not an error — same answer as the standalone fetch. + _tcs.SetResult(null); + } + } +}