Introduce specialized stream query handler for CQRS

[LarsSkovslund]

Abstract

There are situations when using a projection to build a query side read model is not desirable. One example is producing an audit log from events on a single stream. You can achieve this today by implementing a command handler that consumes the desired events to build up an internal read model and return it through the context.ResponseObject property.

Motivation

To make these kind of single stream quires more explicit in naming and behaviour a new type of handler could be introduced.

Suggestion

Initial suggestion from @technight

[technight]

I imagine the introduction of the following interfaces:

public interface IQueryHandler<in TQuery, TResult>
    where TQuery : IQuery
{
    ValueTask<TResult> ExecuteAsync(
        TQuery query,
        CancellationToken cancellationToken);
}


public interface IQuery
{
    EventStreamId GetEventStreamId();
}

This is essentially a strongly typed, read-only version of the ICommandHandler interface.

This would probably also require query specific versions of:

• CommandProcessor
• CommandProcessorFactory
• StateProjector
• CommandHandlerMetadata

---

Implementing the IQueryHandler could look something like this:

public class MyQueryHandler : 
    IQueryHandler<MyQuery, MyResult>,
    IConsumeEvent<MyEvent>
{
    private MyResult result = new();
    
    public ValueTask<MyResult> ExecuteAsync(MyQuery command, CancellationToken cancellationToken)
    {
        return ValueTask.FromResult(result);
    }

    public void Consume(MyEvent evt, EventMetadata metadata)
    {
        // Modify result based on events from stream
    }
}

---

From a consumers point of view I imagine using it like this:

var processor = serviceProvider.GetRequiredService<IQueryProcessor<MyQuery, MyResult>>();
var query = new MyQuery(streamId, otherStuffPotentiallyUsedByQuery);
var result = await processor.ExecuteAsync(query);

result will now be of type MyResult and will not need a Cast or Pattern Match as is required when setting context.ResponseObject.

I am still uncertain if we may still want the processor to actually return a result object like CommandResult which could include similar information like stream id, current stream version, and whether the stream exists or not. Something like:

public record QueryResult<TResponse>(
    EventStreamId Id,
    StreamVersion Version,
    ResultType Result,
    TResponse Response);

I am leaning towards just returning the result without wrapping it, but I don't know if there are use cases where it would be beneficial to have the above information included.

[LarsSkovslund]

Thanks for a good and the detailed description 👍

Below constrains is not documented any ware, so the is my opportunity to get it down on writing

The ICommandHandler<T> and IProjection are build with a set of design goals;

• Must be testable without any infrastructure dependencies
• Do not expose any primitives from Atc.Cosmos.EventStore on it's public interfaces
• Must use a push model for consuming event, ideally through the IConsumeEvent family of interfaces
• Must not control the process of fetching events e.g. fetching and storing behaviour is controlled by the Command and not the CommandHandler

To avoid any confusion when using the word Query in the context of CQRS we should name the interfaces IStreamQueryXXX to indicate the behaviour is querying over a single stream of data, As this library is not providing any build-in support for storing and query read model generated through projections.

Lets look at the query handler interface and its behaviour.

public interface IStreamQueryHandler<in TQuery, TResult>
    where TQuery : IStreamQuery
{
    ValueTask<TResult> ExecuteAsync(TQuery query, CancellationToken cancellationToken);
}

public interface IStreamQuery
{
    EventStreamId GetEventStreamId();
}

Examining IStreamQuery interface, it's purpose is controlling which stream to query on but are there other needs/use cases requiring infrastructure configurations? If we look at a common read model query need, support for paging/continuation token or only changes since a specific event. Extending IStreamQuery with an options to specify the version (position) to project data from in the stream, we can optimize on the number of events read from the stream.

public interface IStreamQuery
{
    // Instruct the infrastructure to read from an event stream with returned id.
    EventStreamId GetEventStreamId();

    // Specifies the position in the stream to project events from.
    EventStreamVersion? FromPosition { get; set; }

    // Limit the number of events to project.
    long? MaxEvents { get; set; }
}

With the above interface specification we can implement an abstract StreamQueryBase record to be reused.

public abstract record StreamQueryBase<TStreamId>(
    TStreamId StreamId,
    EventStreamVersion? FromPosition = default,
    long? MaxEvents = default) : IStreamQuery
    where TStreamId : EventStreamId
{
    EventStreamId IStreamQuery.GetEventStreamId() => StreamId;
}

---

Looking at IStreamQueryHandler<TQuery, TResult> we have one input (TQuery) to control how the query behaves in order to produce the desired result (TResult). The suggested solution would expect the filtering process and read model projection to happen at the same time. This will only be possible if all events are stored locally in a member list when they are consumed so the ExecuteAsync can perform the projection and filtering. If we introduce a Prepare step to be invoked before any consume events we have a hook for the missing behaviour.

// Implement interface to build a read model from a given stream.
// To consume events used in the query you must implement one
// or more of the following interface IConsumeEvent(Async)<T>
public interface IStreamQueryHandler<in TQuery, TResult>
{
  // Prepare is called before query handler is starting to consume events,
  // this will give the implementor a hook point to control how events are
  // transformed into the query result (TResult).
  ValuerTask PrepareAsync(TQuery parameters, CancellationToken cancellationToken)):

  // Gets the projected read model.
  ValueTask<TResult> ExecuteAsync(CancellationToken cancellationToken);
}

[technight]

Great refinements, I like them a lot 👍

And it's a very good point with PrepareAsync in IStreamQueryHandler.
I suppose the PrepareAsync can simply be invoked by the potential StreamQueryProcessor? So that the consumer will still only have to do processor.ExecuteAsync(query) and the processor will then be responsible for:

1. Call Handler.PrepareAsync
2. Do state projection
3. Call Handler.ExecuteAsync

What's your thought on the result of IStreamQueryProcessor.ExecuteAsync? Do we wrap the result with additional stream information, or do we simply return the plain version?