Introduce a reducer type of observer

[einari]

Today we have 2 options for projecting from events:

• Imperative Observers
• Declarative Projections

Between these there sits the potential for at least one more option; a reducer.

Imagine having reducers look something like this:

[Reducer("<some guid>")]
public class MyReducer
{
     public Task<MyReadModel> SomethingHappend(MyFirstEvent @event, EventContext context, MyReadModel? current)
     {
          current ??= new MyReadModel();
          current.SomeProperty = @event.SomeProperty;
          return Task.FromResult(current);
     }


     public Task<MyReadModel> SomethingElseHappened(MySecondEvent @event, EventContext context, MyReadModel? current)
     {
          current ??= new MyReadModel();
          current.SomeOtherProperty = @event.SomeOtherProperty;
          return Task.FromResult(current);
     }
}

The benefit of this is that we would decouple ourselves from the underlying storage mechanism. We could in fact store the results anywhere. They could be forwarded to other event sequences.

Bulks

Another benefit would be that its much easier to do bulks. So for a replay we could do a cursored approach where we'd do something like 1000 events at a time. Processing would be in memory for these and you'd then go and update on every Nth (1000) as your checkpoint. The observers offset would also then be updated on the checkpoint, when successful.

We could also be doing a hot-cold scenario for replays. This means that we could be running reducers towards a specific collection while catching up and swap it with the one used in production when done.

Keys

We would also need a way to provide how it should resolve the for the read model. If not provided it would typically use the EventSourceId. You should be able to provide the key cross cuttingly for the entire reducer if there is a property or composite of properties that are the same for all, or you should provide it per reducer method.

This could be achieved through attributes:

[Reducer("<some guid>", Key=nameof(ICommonInterface.TheKey)]
public class MyReducer
{
     [ModelKey(nameof(MyFirstEvent.SomeKey)]
     public Task<MyReadModel> SomethingHappend(MyFirstEvent @event, EventContext context, MyReadModel? current)
     {
          current ??= new MyReadModel();
          current.SomeProperty = @event.SomeProperty;
          return Task.FromResult(current);
     }

     [ModelKey(nameof(MySecondEvent.FirstKeyProperty, MySecondEvent.SecondKeyProperty)]
     public Task<MyReadModel> SomethingElseHappened(MySecondEvent @event, EventContext context, MyReadModel? current)
     {
          current ??= new MyReadModel();
          current.SomeOtherProperty = @event.SomeOtherProperty;
          return Task.FromResult(current);
     }
}

Sources and targets

Providing configuration for source EventSequence and target details. For instance if you want to target the output to specific collection in MongoDB:

[Reducer("<some guid>", Source="<guid for event sequence>", TargetType=ReducerTargetType.MongoDB, TargetTypeData="Collection Name")]
public class MyReducer
{
    // ... reducer methods
}

Or if you want to target an event sequence:

[Reducer("<some guid>", Source="<guid for event sequence>", TargetType=ReducerTargetType.EventSequence, TargetTypeData="<guid of event sequence>")]
public class MyReducer
{
    // ... reducer methods
}

[einari]

API, design and documentation inspiration - React Redux: https://www.linkedin.com/feed/update/urn:li:activity:7086677241870352384?utm_source=share&utm_medium=member_desktop

[einari]

Whats left after the initial "end-to-end" getting things working:

• [ ] Specs
• [x] Rename and move common parts that sits today in Projection (Sink++)
• [ ] Fully spec the MongoDB Sink and its supporting systems
• [x] Change Observer APIs and implementations to work with collection of appended events
• [x] Change CatchUp and Replay tasks to work in bulks
• [ ] Client validation of reducers
• [ ] Add support for Keys to Observers
• [ ] Add observer indexes based on keys for observers (including sub-task for rebuilding index)
• [ ] Review APIs and namespace structure
• [x] Remove requirement of nullable for initial read model state for client reducer
• [x] Support bulk Append
• [ ] Support bulk for Projections
• [ ] Documentation