Add FetchStreamStatePlan + FetchStreamPlan: raw event stream fetches as batchable query plans

docs/cSpell.json

@@ -3,6 +3,7 @@
   "language": "en",
   "words": [
     "jasperfx",
+    "batchable",
     "TimescaleDB",
     "timescaledb",
     "hypertable",

docs/documents/querying/compiled-queries.md

@@ -733,7 +733,7 @@ public interface IBatchQueryPlan<T>
     Task<T> Fetch(IBatchedQuery query);
 }
 ```
-<sup><a href='https://github.com/JasperFx/marten/blob/master/src/Marten/IQueryPlan.cs#L21-L33' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_ibatchqueryplan' title='Start of snippet'>anchor</a></sup>
+<sup><a href='https://github.com/JasperFx/marten/blob/master/src/Marten/IQueryPlan.cs#L22-L34' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_ibatchqueryplan' title='Start of snippet'>anchor</a></sup>
 <!-- endSnippet -->
 
 And because we expect this to be very common, there is convenience base class named `QueryListPlan<T>` for querying lists of `T` data that can be used for both querying directly against an `IQuerySession` and for batch querying. The usage within a batched query is shown below from the Marten tests:
@@ -778,3 +778,70 @@ public async Task use_as_batch()
 ```
 <sup><a href='https://github.com/JasperFx/marten/blob/master/src/DocumentDbTests/Reading/query_plans.cs#L34-L71' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_using_query_plan_in_batch_query' title='Start of snippet'>anchor</a></sup>
 <!-- endSnippet -->
+
+### Query Plans for Event Streams
+
+Marten ships two concrete query plans for the raw event stream fetches so that they can also be used as batchable
+specifications. `FetchStreamStatePlan` fetches the high level `StreamState` metadata about a single stream (yielding
+`null` when the stream does not exist), while `FetchStreamPlan` fetches the raw events of a single stream (yielding an
+empty list when the stream does not exist) with support for the optional `version`, `timestamp`, and `fromVersion`
+filters of `FetchStreamAsync()`. Both plans accept either a `Guid` stream id or a `string` stream key to cover both
+stream identity styles.
+
+Using `FetchStreamPlan` standalone against a session:
+
+<!-- snippet: sample_using_fetch_stream_plan -->
+<a id='snippet-sample_using_fetch_stream_plan'></a>
+```cs
+[Fact]
+public async Task fetch_stream_by_query_plan()
+{
+    var streamId = theSession.Events.StartStream<Quest>(new QuestStarted { Name = "Destroy the One Ring" },
+        new MembersJoined(1, "Hobbiton", "Frodo", "Sam"),
+        new MembersJoined(2, "Bree", "Aragorn")).Id;
+    await theSession.SaveChangesAsync();
+
+    var events = await theSession.QueryByPlanAsync(new FetchStreamPlan(streamId));
+
+    events.Count.ShouldBe(3);
+    events[0].Data.ShouldBeOfType<QuestStarted>();
+}
+```
+<sup><a href='https://github.com/JasperFx/marten/blob/master/src/EventSourcingTests/fetching_stream_query_plans.cs#L51-L67' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_using_fetch_stream_plan' title='Start of snippet'>anchor</a></sup>
+<!-- endSnippet -->
+
+And because both plans also implement `IBatchQueryPlan<T>`, they can be combined with any other registered queries
+within a [batched query](/documents/querying/batched-queries) to fetch a stream's state and its raw events in one
+database round trip:
+
+<!-- snippet: sample_fetch_stream_plans_in_batch -->
+<a id='snippet-sample_fetch_stream_plans_in_batch'></a>
+```cs
+[Fact]
+public async Task use_both_plans_in_one_batch()
+{
+    var streamId = theSession.Events.StartStream<Quest>(new QuestStarted { Name = "Destroy the One Ring" },
+        new MembersJoined(1, "Hobbiton", "Frodo", "Sam")).Id;
+    await theSession.SaveChangesAsync();
+
+    // Start a batch query
+    var batch = theSession.CreateBatchQuery();
+
+    // Fetching the stream state and the raw events of the same stream
+    // in one database round trip
+    var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId));
+    var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamId));
+
+    // Execute the batch query
+    await batch.Execute();
+
+    var state = await stateFetcher;
+    var events = await eventsFetcher;
+
+    state.ShouldNotBeNull();
+    state.Version.ShouldBe(2);
+    events.Count.ShouldBe(2);
+}
+```
+<sup><a href='https://github.com/JasperFx/marten/blob/master/src/EventSourcingTests/fetching_stream_query_plans.cs#L105-L133' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_fetch_stream_plans_in_batch' title='Start of snippet'>anchor</a></sup>
+<!-- endSnippet -->

docs/events/querying.md

@@ -258,6 +258,47 @@ public class fetching_stream_state: IntegrationContext
 
 Furthermore, `StreamState` contains metadata for when the stream was created, `StreamState.Created`, and when the stream was last updated, `StreamState.LastTimestamp`.
 
+## Stream Query Plans
+
+The stream fetches above are also available as reusable [query plans](/documents/querying/compiled-queries#query-plans-):
+`FetchStreamStatePlan` wraps `FetchStreamState()`/`FetchStreamStateAsync()` and `FetchStreamPlan` wraps `FetchStream()`/`FetchStreamAsync()`.
+Both plans accept either a `Guid` stream id or a `string` stream key, and `FetchStreamPlan` carries the optional
+`version`, `timestamp`, and `fromVersion` filters. Because the plans implement both `IQueryPlan<T>` and
+`IBatchQueryPlan<T>`, the same object works standalone with `IQuerySession.QueryByPlanAsync()` or combined with any
+other registered queries into a single database round trip within a batched query:
+
+<!-- snippet: sample_fetch_stream_plans_in_batch -->
+<a id='snippet-sample_fetch_stream_plans_in_batch'></a>
+```cs
+[Fact]
+public async Task use_both_plans_in_one_batch()
+{
+    var streamId = theSession.Events.StartStream<Quest>(new QuestStarted { Name = "Destroy the One Ring" },
+        new MembersJoined(1, "Hobbiton", "Frodo", "Sam")).Id;
+    await theSession.SaveChangesAsync();
+
+    // Start a batch query
+    var batch = theSession.CreateBatchQuery();
+
+    // Fetching the stream state and the raw events of the same stream
+    // in one database round trip
+    var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId));
+    var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamId));
+
+    // Execute the batch query
+    await batch.Execute();
+
+    var state = await stateFetcher;
+    var events = await eventsFetcher;
+
+    state.ShouldNotBeNull();
+    state.Version.ShouldBe(2);
+    events.Count.ShouldBe(2);
+}
+```
+<sup><a href='https://github.com/JasperFx/marten/blob/master/src/EventSourcingTests/fetching_stream_query_plans.cs#L105-L133' title='Snippet source file'>snippet source</a> | <a href='#snippet-sample_fetch_stream_plans_in_batch' title='Start of snippet'>anchor</a></sup>
+<!-- endSnippet -->
+
 ## 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:

src/EventSourcingTests/fetching_stream_query_plans.cs

@@ -0,0 +1,134 @@
+using System;
+using System.Threading.Tasks;
+using JasperFx.Events;
+using Marten;
+using Marten.Testing.Harness;
+using Shouldly;
+using Xunit;
+
+namespace EventSourcingTests;
+
+public class fetching_stream_query_plans: OneOffConfigurationsContext
+{
+    [Fact]
+    public async Task fetch_stream_state_by_query_plan()
+    {
+        var streamId = theSession.Events.StartStream<Quest>(new QuestStarted { Name = "Destroy the One Ring" },
+            new MembersJoined(1, "Hobbiton", "Frodo", "Sam")).Id;
+        await theSession.SaveChangesAsync();
+
+        var state = await theSession.QueryByPlanAsync(new FetchStreamStatePlan(streamId));
+
+        state.ShouldNotBeNull();
+        state.Id.ShouldBe(streamId);
+        state.Version.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task fetch_stream_state_by_query_plan_with_string_identity()
+    {
+        StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsString);
+
+        theSession.Events.Append("one-ring", new QuestStarted { Name = "Destroy the One Ring" },
+            new MembersJoined(1, "Hobbiton", "Frodo", "Sam"));
+        await theSession.SaveChangesAsync();
+
+        var state = await theSession.QueryByPlanAsync(new FetchStreamStatePlan("one-ring"));
+
+        state.ShouldNotBeNull();
+        state.Key.ShouldBe("one-ring");
+        state.Version.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task fetch_stream_state_by_query_plan_for_missing_stream_is_null()
+    {
+        var state = await theSession.QueryByPlanAsync(new FetchStreamStatePlan(Guid.NewGuid()));
+
+        state.ShouldBeNull();
+    }
+
+    #region sample_using_fetch_stream_plan
+
+    [Fact]
+    public async Task fetch_stream_by_query_plan()
+    {
+        var streamId = theSession.Events.StartStream<Quest>(new QuestStarted { Name = "Destroy the One Ring" },
+            new MembersJoined(1, "Hobbiton", "Frodo", "Sam"),
+            new MembersJoined(2, "Bree", "Aragorn")).Id;
+        await theSession.SaveChangesAsync();
+
+        var events = await theSession.QueryByPlanAsync(new FetchStreamPlan(streamId));
+
+        events.Count.ShouldBe(3);
+        events[0].Data.ShouldBeOfType<QuestStarted>();
+    }
+
+    #endregion
+
+    [Fact]
+    public async Task fetch_stream_by_query_plan_with_string_identity()
+    {
+        StoreOptions(opts => opts.Events.StreamIdentity = StreamIdentity.AsString);
+
+        theSession.Events.Append("one-ring", new QuestStarted { Name = "Destroy the One Ring" },
+            new MembersJoined(1, "Hobbiton", "Frodo", "Sam"));
+        await theSession.SaveChangesAsync();
+
+        var events = await theSession.QueryByPlanAsync(new FetchStreamPlan("one-ring"));
+
+        events.Count.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task fetch_stream_by_query_plan_with_version_cap()
+    {
+        var streamId = theSession.Events.StartStream<Quest>(new QuestStarted { Name = "Destroy the One Ring" },
+            new MembersJoined(1, "Hobbiton", "Frodo", "Sam"),
+            new MembersJoined(2, "Bree", "Aragorn")).Id;
+        await theSession.SaveChangesAsync();
+
+        var events = await theSession.QueryByPlanAsync(new FetchStreamPlan(streamId, version: 2));
+
+        events.Count.ShouldBe(2);
+        events[^1].Version.ShouldBe(2);
+    }
+
+    [Fact]
+    public async Task fetch_stream_by_query_plan_for_missing_stream_is_empty()
+    {
+        var events = await theSession.QueryByPlanAsync(new FetchStreamPlan(Guid.NewGuid()));
+
+        events.ShouldBeEmpty();
+    }
+
+    #region sample_fetch_stream_plans_in_batch
+
+    [Fact]
+    public async Task use_both_plans_in_one_batch()
+    {
+        var streamId = theSession.Events.StartStream<Quest>(new QuestStarted { Name = "Destroy the One Ring" },
+            new MembersJoined(1, "Hobbiton", "Frodo", "Sam")).Id;
+        await theSession.SaveChangesAsync();
+
+        // Start a batch query
+        var batch = theSession.CreateBatchQuery();
+
+        // Fetching the stream state and the raw events of the same stream
+        // in one database round trip
+        var stateFetcher = batch.QueryByPlan(new FetchStreamStatePlan(streamId));
+        var eventsFetcher = batch.QueryByPlan(new FetchStreamPlan(streamId));
+
+        // Execute the batch query
+        await batch.Execute();
+
+        var state = await stateFetcher;
+        var events = await eventsFetcher;
+
+        state.ShouldNotBeNull();
+        state.Version.ShouldBe(2);
+        events.Count.ShouldBe(2);
+    }
+
+    #endregion
+}

src/Marten/IQueryPlan.cs

@@ -3,6 +3,7 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using JasperFx.Events;
 using Marten.Linq;
 using Marten.Services.BatchQuerying;
 
@@ -65,3 +66,106 @@ Task<IReadOnlyList<T>> IBatchQueryPlan<IReadOnlyList<T>>.Fetch(IBatchedQuery que
         return query.AddItem(handler);
     }
 }
+
+/// <summary>
+/// Query plan to fetch the high level metadata about a single event stream identified by
+/// either a Guid stream id or a string stream key. Can be used both individually with
+/// IQuerySession.QueryByPlanAsync() and with IBatchedQuery.QueryByPlan(). Yields null
+/// if the stream does not exist
+/// </summary>
+public class FetchStreamStatePlan : IQueryPlan<StreamState?>, IBatchQueryPlan<StreamState?>
+{
+    private readonly Guid _streamId;
+    private readonly string? _streamKey;
+
+    /// <summary>
+    /// Fetch the stream state for the stream identified by <paramref name="streamId"/>
+    /// </summary>
+    /// <param name="streamId"></param>
+    public FetchStreamStatePlan(Guid streamId)
+    {
+        _streamId = streamId;
+    }
+
+    /// <summary>
+    /// Fetch the stream state for the stream identified by <paramref name="streamKey"/>
+    /// </summary>
+    /// <param name="streamKey"></param>
+    public FetchStreamStatePlan(string streamKey)
+    {
+        _streamKey = streamKey;
+    }
+
+    public Task<StreamState?> Fetch(IQuerySession session, CancellationToken token)
+    {
+        return _streamKey is not null
+            ? session.Events.FetchStreamStateAsync(_streamKey, token)
+            : session.Events.FetchStreamStateAsync(_streamId, token);
+    }
+
+    public async Task<StreamState?> Fetch(IBatchedQuery query)
+    {
+        return _streamKey is not null
+            ? await query.Events.FetchStreamState(_streamKey).ConfigureAwait(false)
+            : await query.Events.FetchStreamState(_streamId).ConfigureAwait(false);
+    }
+}
+
+/// <summary>
+/// Query plan to fetch the raw events for a single event stream identified by either a
+/// Guid stream id or a string stream key. Can be used both individually with
+/// IQuerySession.QueryByPlanAsync() and with IBatchedQuery.QueryByPlan(). Yields an
+/// empty list if the stream does not exist
+/// </summary>
+public class FetchStreamPlan : IQueryPlan<IReadOnlyList<IEvent>>, IBatchQueryPlan<IReadOnlyList<IEvent>>
+{
+    private readonly Guid _streamId;
+    private readonly string? _streamKey;
+    private readonly long _version;
+    private readonly DateTimeOffset? _timestamp;
+    private readonly long _fromVersion;
+
+    /// <summary>
+    /// Fetch the events for the stream identified by <paramref name="streamId"/>
+    /// </summary>
+    /// <param name="streamId"></param>
+    /// <param name="version">If set, queries for events up to and including this version</param>
+    /// <param name="timestamp">If set, queries for events captured on or before this timestamp</param>
+    /// <param name="fromVersion">If set, queries for events on or from this version</param>
+    public FetchStreamPlan(Guid streamId, long version = 0, DateTimeOffset? timestamp = null, long fromVersion = 0)
+    {
+        _streamId = streamId;
+        _version = version;
+        _timestamp = timestamp;
+        _fromVersion = fromVersion;
+    }
+
+    /// <summary>
+    /// Fetch the events for the stream identified by <paramref name="streamKey"/>
+    /// </summary>
+    /// <param name="streamKey"></param>
+    /// <param name="version">If set, queries for events up to and including this version</param>
+    /// <param name="timestamp">If set, queries for events captured on or before this timestamp</param>
+    /// <param name="fromVersion">If set, queries for events on or from this version</param>
+    public FetchStreamPlan(string streamKey, long version = 0, DateTimeOffset? timestamp = null, long fromVersion = 0)
+    {
+        _streamKey = streamKey;
+        _version = version;
+        _timestamp = timestamp;
+        _fromVersion = fromVersion;
+    }
+
+    public Task<IReadOnlyList<IEvent>> Fetch(IQuerySession session, CancellationToken token)
+    {
+        return _streamKey is not null
+            ? session.Events.FetchStreamAsync(_streamKey, _version, _timestamp, _fromVersion, token)
+            : session.Events.FetchStreamAsync(_streamId, _version, _timestamp, _fromVersion, token);
+    }
+
+    public Task<IReadOnlyList<IEvent>> Fetch(IBatchedQuery query)
+    {
+        return _streamKey is not null
+            ? query.Events.FetchStream(_streamKey, _version, _timestamp, _fromVersion)
+            : query.Events.FetchStream(_streamId, _version, _timestamp, _fromVersion);
+    }
+}