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); + } + } +}