diff --git a/docs/documents/aspnetcore.md b/docs/documents/aspnetcore.md index a0f370b9dc..15af975682 100644 --- a/docs/documents/aspnetcore.md +++ b/docs/documents/aspnetcore.md @@ -233,7 +233,7 @@ bool found = await session.Events.StreamLatestJson(orderId, stream); ## Typed Streaming Result Types For Minimal API endpoints (and for frameworks like [Wolverine.Http](https://wolverinefx.net/guide/http/) -that dispatch any `IResult` return value), `Marten.AspNetCore` ships five typed +that dispatch any `IResult` return value), `Marten.AspNetCore` ships seven typed result wrappers that carry the streaming behavior above as endpoint return values while also contributing correct OpenAPI metadata: @@ -244,13 +244,15 @@ while also contributing correct OpenAPI metadata: | `StreamAggregate` | `IDocumentSession` + stream id — event-sourced | Single `T` | yes | | `StreamPaged` | `IQueryable` — regular Marten document query | Paged JSON envelope | no (empty page = 200) | | `StreamPagedByCursor` | `IQueryable` (with `OrderBy`/`ThenBy`) | { "items": T[], "nextCursor" } | no (empty array = 200) | +| `StreamEventState` | `IQuerySession` + stream id — event stream | Single `StreamStateResponse` | yes | +| `StreamEvents` | `IQuerySession` + stream id — event stream | JSON array `EventResponse[]` | yes (configurable) | Each type implements both `IResult` (so ASP.NET Minimal API dispatches it via `ExecuteAsync`) and `IEndpointMetadataProvider` (so Swashbuckle, NSwag, and the built-in OpenAPI generator see the right response shape), while delegating the -actual body write to `WriteSingle`/`WriteArray`/`WriteLatest`. Returning one -from an endpoint is a concise, typed alternative to writing the HTTP handshake -manually. +actual body write to `WriteSingle`/`WriteArray`/`WriteLatest`/`WriteStreamState`/`WriteEvents`. +Returning one from an endpoint is a concise, typed alternative to writing the HTTP +handshake manually. ### `StreamOne` — single document with 404 on miss @@ -315,7 +317,122 @@ Returns `200 application/json` with the JSON of the latest projected aggregate state, or `404` if no stream exists. A constructor overload accepts `string` ids for stores configured with string-keyed streams. -### StreamOne vs StreamAggregate +### `StreamEventState` — event stream metadata + +Writes the high level metadata of a single event stream — Marten's `StreamState` — as JSON, +or `404` when the stream does not exist: + + + +```cs +app.MapGet("/minimal/order/{id:guid}/state", + (Guid id, IQuerySession session) + => new StreamEventState(session, id)); +``` +snippet source | anchor + + +A constructor overload accepts a `string` stream key for stores configured with string-keyed +streams. + +The response body is a **`StreamStateResponse`**, not `StreamState` itself. `StreamState.AggregateType` +is a `System.Type`, and System.Text.Json refuses to serialize those outright +(`Serialization and deserialization of 'System.Type' instances is not supported`), so the +aggregate type is projected down to its simple name in `AggregateTypeName`: + +```json +{ + "id": "0198e1b4-5b1c-7a1e-9a3f-2f2f5b6c7d8e", + "key": null, + "version": 2, + "aggregateTypeName": "Order", + "lastTimestamp": "2026-07-26T09:41:02.113Z", + "created": "2026-07-26T09:41:02.098Z", + "isArchived": false +} +``` + +### `StreamEvents` — raw events of a stream + +Writes the raw events of a single event stream as a JSON array: + + + +```cs +app.MapGet("/minimal/order/{id:guid}/events", + (Guid id, IQuerySession session) + => new StreamEvents(session, id)); +``` +snippet source | anchor + + +`StreamEvents` carries the same optional `version`, `timestamp`, and `fromVersion` filters as +`FetchStreamAsync()`, and there is a `string` stream key overload as well. + +Elements are **`EventResponse`**, not `IEvent` itself — `IEvent.EventType` is a `System.Type` and +hits the same System.Text.Json wall as above. Use `eventTypeName`, Marten's stable event type +alias, to discriminate event types on the client. The assembly qualified .NET type name +(`DotNetTypeName`) is deliberately left off the wire: + +```json +[ + { + "id": "0198e1b4-5b1c-7a1e-9a3f-2f2f5b6c7d8e", + "version": 1, + "sequence": 41, + "streamId": "0198e1b4-5b1c-7a1e-9a3f-2f2f5b6c7d8e", + "streamKey": null, + "eventTypeName": "order_placed", + "timestamp": "2026-07-26T09:41:02.098Z", + "tenantId": "*DEFAULT*", + "isArchived": false, + "causationId": null, + "correlationId": null, + "headers": null, + "data": { "description": "Widget", "amount": 99.95 } + } +] +``` + +#### Empty streams: 404 or an empty array? + +`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` therefore exposes an +`OnEmptyStatus` that defaults to `404`, matching the other single-resource results. Set it to +`200` when running off the end of a stream is expected rather than exceptional — paging forward +with `fromVersion`, for example: + + + +```cs +// Paging forward through a stream: running off the end is expected, not a 404 +app.MapGet("/minimal/order/{id:guid}/events/from/{fromVersion:long}", + (Guid id, long fromVersion, IQuerySession session) + => new StreamEvents(session, id, fromVersion: fromVersion) + { + OnEmptyStatus = StatusCodes.Status200OK + }); +``` +snippet source | anchor + + +#### Sharing a query plan with a batched query + +Both results are backed by the [`FetchStreamStatePlan` / `FetchStreamPlan` query plans](/events/querying#stream-query-plans), +and both accept a pre-built plan. That lets a handler build the plan once and either batch it with +other queries into a single round trip or hand it straight back as an HTTP result: + +```csharp +var plan = new FetchStreamPlan(orderId, version: 5); + +// ...batch it alongside other queries +var fetcher = batch.QueryByPlan(plan); + +// ...or return it from an endpoint +return new StreamEvents(session, plan); +``` + +### Choosing between the result types - **`StreamOne`** is for regular Marten documents — plain objects persisted via `session.Store()` and queried with `session.Query()`. The query hits @@ -324,6 +441,12 @@ ids for stores configured with string-keyed streams. latest aggregate state by folding events from the event store (or reads a projected snapshot if one is configured). Use this when `T` is an event-sourced aggregate, not a stored document. +- **`StreamEventState`** returns a stream's _metadata_ — version, timestamps, + archived flag — rather than any projected state. Reach for it when a client + needs to know where a stream is up to, not what it currently looks like. +- **`StreamEvents`** returns the stream's _raw events_. Use it for audit trails, + event-log style UIs, and debugging endpoints, rather than as the read model for + ordinary consumers — those are better served by `StreamAggregate`. ### ETag / Conditional Requests diff --git a/docs/events/querying.md b/docs/events/querying.md index ba8dd14b2b..df476244f2 100644 --- a/docs/events/querying.md +++ b/docs/events/querying.md @@ -299,6 +299,13 @@ public async Task use_both_plans_in_one_batch() snippet source | anchor +::: tip +Both plans also back an ASP.NET Core endpoint return value — +[`StreamEventState` and `StreamEvents`](/documents/aspnetcore#typed-streaming-result-types-) write the +same data straight to an HTTP response, and accept a pre-built plan so a handler can share one plan +between a batched query and its HTTP result. +::: + ## Fetch a Single Event You can fetch the information for a single event by id, including its version number within the stream, by using `IEventStore.LoadAsync()` as shown below: diff --git a/src/IssueService/StreamingMinimalEndpoints.cs b/src/IssueService/StreamingMinimalEndpoints.cs index 0b0c31cb53..6a8f527018 100644 --- a/src/IssueService/StreamingMinimalEndpoints.cs +++ b/src/IssueService/StreamingMinimalEndpoints.cs @@ -123,6 +123,51 @@ public static IEndpointRouteBuilder MapStreamingMinimalEndpoints(this IEndpointR cursor, pageSize)); + // --- StreamEventState --- + + #region sample_minimal_api_stream_event_state + + app.MapGet("/minimal/order/{id:guid}/state", + (Guid id, IQuerySession session) + => new StreamEventState(session, id)); + + #endregion + + app.MapGet("/minimal/named-order/{id}/state", + (string id, IQuerySession session) + => new StreamEventState(session, id)); + + // --- StreamEvents --- + + #region sample_minimal_api_stream_events + + app.MapGet("/minimal/order/{id:guid}/events", + (Guid id, IQuerySession session) + => new StreamEvents(session, id)); + + #endregion + + app.MapGet("/minimal/named-order/{id}/events", + (string id, IQuerySession session) + => new StreamEvents(session, id)); + + #region sample_minimal_api_stream_events_from_version + + // Paging forward through a stream: running off the end is expected, not a 404 + app.MapGet("/minimal/order/{id:guid}/events/from/{fromVersion:long}", + (Guid id, long fromVersion, IQuerySession session) + => new StreamEvents(session, id, fromVersion: fromVersion) + { + OnEmptyStatus = StatusCodes.Status200OK + }); + + #endregion + + // Version cap, and the plan-accepting constructor + app.MapGet("/minimal/order/{id:guid}/events/upto/{version:long}", + (Guid id, long version, IQuerySession session) + => new StreamEvents(session, new FetchStreamPlan(id, version))); + return app; } } diff --git a/src/Marten.AspNetCore.Testing/stream_event_result_types_tests.cs b/src/Marten.AspNetCore.Testing/stream_event_result_types_tests.cs new file mode 100644 index 0000000000..282a8126ad --- /dev/null +++ b/src/Marten.AspNetCore.Testing/stream_event_result_types_tests.cs @@ -0,0 +1,248 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Alba; +using IssueService.Controllers; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Metadata; +using Microsoft.AspNetCore.Routing; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; +using Xunit; + +namespace Marten.AspNetCore.Testing; + +/// +/// Alba-based tests for and executing +/// against plain Minimal API endpoints (no Wolverine required). +/// +[Collection("integration")] +public class stream_event_result_types_tests: IntegrationContext +{ + private readonly IAlbaHost theHost; + + public stream_event_result_types_tests(AppFixture fixture): base(fixture) + { + theHost = fixture.Host; + } + + private async Task anOrderStream(params object[] events) + { + var orderId = Guid.NewGuid(); + var store = theHost.Services.GetRequiredService(); + await using var session = store.LightweightSession(); + session.Events.StartStream(orderId, events); + await session.SaveChangesAsync(); + + return orderId; + } + + // ───────────────────────── StreamEventState ───────────────────────── + + [Fact] + public async Task stream_event_state_returns_metadata_for_an_existing_stream() + { + var orderId = await anOrderStream(new OrderPlaced("Widget", 99.95m), new OrderShipped()); + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{orderId}/state"); + s.StatusCodeShouldBe(200); + s.ContentTypeShouldBe("application/json"); + }); + + var state = result.ReadAsJson(); + state.ShouldNotBeNull(); + state.Id.ShouldBe(orderId); + state.Version.ShouldBe(2); + state.IsArchived.ShouldBeFalse(); + state.Created.ShouldBeGreaterThan(DateTimeOffset.MinValue); + } + + [Fact] + public async Task stream_event_state_serializes_the_aggregate_type_as_a_name() + { + // StreamState.AggregateType is a System.Type, which System.Text.Json flatly refuses to + // serialize. StreamStateResponse projects it to a simple name instead. + var orderId = await anOrderStream(new OrderPlaced("Widget", 99.95m)); + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{orderId}/state"); + s.StatusCodeShouldBe(200); + }); + + var state = result.ReadAsJson(); + state.AggregateTypeName.ShouldBe(nameof(Order)); + } + + [Fact] + public async Task stream_event_state_sets_content_length() + { + var orderId = await anOrderStream(new OrderPlaced("Widget", 99.95m)); + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{orderId}/state"); + s.StatusCodeShouldBe(200); + }); + + result.Context.Response.ContentLength.HasValue.ShouldBeTrue(); + } + + [Fact] + public async Task stream_event_state_returns_404_for_a_missing_stream() + { + await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{Guid.NewGuid()}/state"); + s.StatusCodeShouldBe(404); + }); + } + + // ───────────────────────── StreamEvents ───────────────────────── + + [Fact] + public async Task stream_events_returns_the_raw_events_as_a_json_array() + { + var orderId = await anOrderStream(new OrderPlaced("Widget", 99.95m), new OrderShipped()); + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{orderId}/events"); + s.StatusCodeShouldBe(200); + s.ContentTypeShouldBe("application/json"); + }); + + var events = result.ReadAsJson>(); + events.Count.ShouldBe(2); + + events[0].Version.ShouldBe(1); + events[0].StreamId.ShouldBe(orderId); + events[0].EventTypeName.ShouldNotBeNullOrEmpty(); + events[0].Timestamp.ShouldBeGreaterThan(DateTimeOffset.MinValue); + + events[1].Version.ShouldBe(2); + events[1].EventTypeName.ShouldNotBe(events[0].EventTypeName); + } + + [Fact] + public async Task stream_events_returns_404_for_a_missing_stream_by_default() + { + await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{Guid.NewGuid()}/events"); + s.StatusCodeShouldBe(404); + }); + } + + [Fact] + public async Task stream_events_honors_a_version_cap_through_the_plan_constructor() + { + var orderId = await anOrderStream(new OrderPlaced("Widget", 99.95m), new OrderShipped()); + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{orderId}/events/upto/1"); + s.StatusCodeShouldBe(200); + }); + + var events = result.ReadAsJson>(); + events.Count.ShouldBe(1); + events[0].Version.ShouldBe(1); + } + + [Fact] + public async Task stream_events_can_opt_into_an_empty_array_instead_of_404() + { + var orderId = await anOrderStream(new OrderPlaced("Widget", 99.95m)); + + // fromVersion past the end of the stream — expected when paging forward, so this + // endpoint sets OnEmptyStatus to 200 rather than reporting the order as missing. + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{orderId}/events/from/99"); + s.StatusCodeShouldBe(200); + s.ContentTypeShouldBe("application/json"); + }); + + result.ReadAsJson>().ShouldBeEmpty(); + } + + [Fact] + public async Task stream_events_carries_the_event_body_on_the_data_property() + { + var orderId = await anOrderStream(new OrderPlaced("Widget", 99.95m)); + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{orderId}/events"); + s.StatusCodeShouldBe(200); + }); + + var events = result.ReadAsJson>(); + events[0].Data.Description.ShouldBe("Widget"); + events[0].Data.Amount.ShouldBe(99.95m); + } + + [Fact] + public async Task stream_events_sets_content_length() + { + var orderId = await anOrderStream(new OrderPlaced("Widget", 99.95m)); + + var result = await theHost.Scenario(s => + { + s.Get.Url($"/minimal/order/{orderId}/events"); + s.StatusCodeShouldBe(200); + }); + + result.Context.Response.ContentLength.HasValue.ShouldBeTrue(); + } + + // ───────────────────────── OpenAPI metadata ───────────────────────── + + [Fact] + public void stream_event_state_endpoint_advertises_produces_response_and_404() + { + var metadata = EndpointMetadataFor("GET", "/minimal/order/{id:guid}/state"); + + metadata.OfType() + .ShouldContain(m => m.StatusCode == 200 && m.Type == typeof(StreamStateResponse)); + metadata.OfType() + .ShouldContain(m => m.StatusCode == 404); + } + + [Fact] + public void stream_events_endpoint_advertises_produces_array_and_404() + { + var metadata = EndpointMetadataFor("GET", "/minimal/order/{id:guid}/events"); + + metadata.OfType() + .ShouldContain(m => m.StatusCode == 200 && m.Type == typeof(EventResponse[])); + metadata.OfType() + .ShouldContain(m => m.StatusCode == 404); + } + + private EndpointMetadataCollection EndpointMetadataFor(string method, string pattern) + { + var endpoint = theHost.Services.GetServices() + .SelectMany(x => x.Endpoints) + .OfType() + .FirstOrDefault(x => + x.RoutePattern.RawText == pattern && + x.Metadata.GetMetadata()!.HttpMethods.Contains(method)); + + endpoint.ShouldNotBeNull($"No endpoint found for {method} {pattern}"); + return endpoint.Metadata; + } +} + +/// +/// Strongly-typed view of the wire shape so the test can assert on the event body itself. +/// +public class OrderPlacedEventResponse +{ + public long Version { get; set; } + public OrderPlaced Data { get; set; } = default!; +} diff --git a/src/Marten.AspNetCore/EventStreamResponses.cs b/src/Marten.AspNetCore/EventStreamResponses.cs new file mode 100644 index 0000000000..a54ba08757 --- /dev/null +++ b/src/Marten.AspNetCore/EventStreamResponses.cs @@ -0,0 +1,150 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using JasperFx.Events; + +namespace Marten.AspNetCore; + +#region sample_event_stream_response_types + +/// +/// 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. +/// +/// +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 Marten onto the wire shape. + /// + public static StreamStateResponse From(StreamState state) + { + if (state == null) throw new ArgumentNullException(nameof(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 +/// , Marten's stable event type alias, to discriminate event types on +/// the client. +/// +/// +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; } + + /// Marten's event type alias — the stable discriminator for clients. + public string? EventTypeName { get; init; } + + /// The UTC 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 a Marten onto the wire shape. + /// + public static EventResponse From(IEvent @event) + { + if (@event == null) throw new ArgumentNullException(nameof(@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 Marten onto the wire shape. + /// + public static EventResponse[] From(IReadOnlyList events) + { + if (events == null) throw new ArgumentNullException(nameof(events)); + + return events.Select(From).ToArray(); + } +} + +#endregion diff --git a/src/Marten.AspNetCore/QueryableExtensions.cs b/src/Marten.AspNetCore/QueryableExtensions.cs index 323c425ac7..99175f3bfa 100644 --- a/src/Marten.AspNetCore/QueryableExtensions.cs +++ b/src/Marten.AspNetCore/QueryableExtensions.cs @@ -1,4 +1,5 @@ using System; +using System.Buffers; using System.IO; using System.Linq; using System.Threading.Tasks; @@ -478,4 +479,110 @@ public static async Task WriteLatest( } } + /// + /// Resolve a and write the resulting stream metadata to the + /// HttpContext response as JSON, or 404 when the stream does not exist. + /// + /// The response body is a rather than Marten's + /// StreamState — see that type for why. + /// + /// + /// + /// + /// + /// + /// Defaults to 200 + public static async Task WriteStreamState( + this IQuerySession session, + FetchStreamStatePlan plan, + HttpContext context, + string contentType = "application/json", + int onFoundStatus = 200 + ) + { + if (session == null) throw new ArgumentNullException(nameof(session)); + if (plan == null) throw new ArgumentNullException(nameof(plan)); + if (context == null) throw new ArgumentNullException(nameof(context)); + + var state = await plan.Fetch(session, context.RequestAborted).ConfigureAwait(false); + + if (state == null) + { + context.Response.StatusCode = 404; + context.Response.ContentLength = 0; + return; + } + + await writeJson(session, StreamStateResponse.From(state), context, contentType, onFoundStatus) + .ConfigureAwait(false); + } + + /// + /// Resolve a and write the resulting raw events to the HttpContext + /// response as a JSON array. + /// + /// Marten's 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 Marten's IEvent — see that + /// type for why. + /// + /// + /// + /// + /// + /// + /// Defaults to 200 + /// Defaults to 404 + public static async Task WriteEvents( + this IQuerySession session, + FetchStreamPlan plan, + HttpContext context, + string contentType = "application/json", + int onFoundStatus = 200, + int onEmptyStatus = 404 + ) + { + if (session == null) throw new ArgumentNullException(nameof(session)); + if (plan == null) throw new ArgumentNullException(nameof(plan)); + if (context == null) throw new ArgumentNullException(nameof(context)); + + var events = await plan.Fetch(session, context.RequestAborted).ConfigureAwait(false); + + if (events.Count == 0 && onEmptyStatus == 404) + { + context.Response.StatusCode = 404; + context.Response.ContentLength = 0; + return; + } + + await writeJson(session, EventResponse.From(events), context, contentType, + events.Count == 0 ? onEmptyStatus : onFoundStatus).ConfigureAwait(false); + } + + /// + /// Serialize with the store's configured serializer and write it to the + /// response, setting Content-Length. Buffers through a pooled + /// so the JSON never round-trips through a .NET string. + /// + private static async Task writeJson( + IQuerySession session, + object value, + HttpContext context, + string contentType, + int statusCode) + { + var buffer = new ArrayBufferWriter(); + session.DocumentStore.Options.Serializer().WriteTo(buffer, 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/Marten.AspNetCore/StreamEventState.cs b/src/Marten.AspNetCore/StreamEventState.cs new file mode 100644 index 0000000000..76c3ecd857 --- /dev/null +++ b/src/Marten.AspNetCore/StreamEventState.cs @@ -0,0 +1,93 @@ +using System; +using System.Reflection; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Metadata; + +namespace Marten.AspNetCore; + +/// +/// Minimal-API / Wolverine.Http endpoint return value that writes the high level metadata of a +/// single event stream — Marten's StreamState — to the +/// 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 Marten'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"; + + /// + public Task ExecuteAsync(HttpContext httpContext) + { + if (httpContext == null) throw new ArgumentNullException(nameof(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) + { + if (builder == null) throw new ArgumentNullException(nameof(builder)); + + builder.Metadata.Add(new ProducesResponseTypeMetadata( + StatusCodes.Status200OK, typeof(StreamStateResponse), new[] { "application/json" })); + builder.Metadata.Add(new ProducesResponseTypeMetadata( + StatusCodes.Status404NotFound, typeof(void), Array.Empty())); + } +} diff --git a/src/Marten.AspNetCore/StreamEvents.cs b/src/Marten.AspNetCore/StreamEvents.cs new file mode 100644 index 0000000000..c2c6b87e42 --- /dev/null +++ b/src/Marten.AspNetCore/StreamEvents.cs @@ -0,0 +1,112 @@ +using System; +using System.Reflection; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Metadata; + +namespace Marten.AspNetCore; + +/// +/// Minimal-API / Wolverine.Http endpoint return value that writes the raw events of a single event +/// stream to the 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(). +/// +/// Marten's 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 Marten's IEvent directly: +/// IEvent.EventType is a and System.Text.Json refuses to serialize those. +/// Use EventTypeName, Marten'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"; + + /// + public Task ExecuteAsync(HttpContext httpContext) + { + if (httpContext == null) throw new ArgumentNullException(nameof(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) + { + if (builder == null) throw new ArgumentNullException(nameof(builder)); + + builder.Metadata.Add(new ProducesResponseTypeMetadata( + StatusCodes.Status200OK, typeof(EventResponse[]), new[] { "application/json" })); + builder.Metadata.Add(new ProducesResponseTypeMetadata( + StatusCodes.Status404NotFound, typeof(void), Array.Empty())); + } +}