About the implementation of CQRS, Projections, Snapshots, etc

[cgglyle]

Very nice project!

I wonder if there are any plans to add a CQRS extension? Because I don't see similar code in the project.

Although I saw Projections related code in v0.4.0 version, it has been removed for now. Projections is also an important part of the implementation and I'm also sure that Snapshots are also a very important part.

I'm interested in completing this part of the functionality if you can provide ideas about it. However I've tried writing parts of it before, but none of them were quite to my liking. Also I wonder if you would be interested in elaborating on the thought process.

Thanks

[ar3s3ru]

Hi @cgglyle, thank you for your interest and for opening the issue!

Indeed, v0.4.0 contained some abstraction/support for projections and has since been removed. The main reason for that is the complexity of building a cohesive API around, say, subscriptions.

I would like for eventually to be a very tiny library that enables certain DDD patterns like Aggregate Roots, Event Sourcing, CQRS, without the unnecessary complexity that many other libraries/frameworks in the market have. eventually could offer some opinionated integration for certain technologies, but it should not be part of the core library.

About projections

I go into some details in this comment (#338 (comment)), but essentially I'd like to offer:

1. A trait/interface for modeling event processors,
2. A wrapper for certain messaging technologies to accept/power/drive those event processors.

A simple event processor could look like this:

// eventually/src/event/processor.rs
#[async_trait]
pub trait Processor<T>: Send + Sync
where
    T: message::Message + Send + Sync
{
    type Error: Send + Sync;

    async fn process(&self, evt: event::Envelope<T>) -> Result<(), Self::Error>;
}

So then a projection would be a specific implementation of this event::Processor.

You can source this projection using a CDC tool (like Debezium, or Redpanda Connect) and stream it on a messaging platform like Kafka/RabbitMQ/etc..

We could build some integration for Kafka/RabbitMQ/etc. as a separate crate to create a consumer and call the event::Processor (with some prebuilt config for, e.g. at-least-once or exactly-once semantics, error handling, dead-lettering, etc.)

About snapshots

Snapshots are a different discussion. The way they are implemented at the moment for Postgres (the only backend supported currently) is through the aggregate::Repository, which saves the "snapshot" of the Aggregate Root state in the aggregates table and its events in the same transaction, so:

1. You always have a snapshot, and that's always the latest available version,
2. It is not event sourcing, since the snapshot is always used to fetch the Aggregate Root state during command evaluation.

More than happy to discuss how a proper event sourcing snapshotting functionality could look like here (I haven't one in mind at the moment).

[cgglyle]

Hi @ar3s3ru

It's nice to get a response from you.

I have realized that you may not want to add the Projections feature.

About Projections

But I have a little idea about Projections. I think that Projections is a special kind of Aggregate, and maybe you can just reuse the code of Aggregate.

Since we only need to replay the events we care about, we can either use Aggregate directly or extend out the Projections feature. Then write the relevant replay code in it.

Then use something like an event bus or CDC tool as you say to get the events and process them and store them in a database or whatever. (I personally feel that CDC would be more complicated)

This might be a viable option.

About Snapshots

Regarding snapshots, I think a snapshot manager could be designed to set up a snapshot schedule, say every 10 events to take a snapshot and then store it in the pg aggregate table with the last event version of the snapshot documented. When we need to read it we can get the last event version directly from the snapshot. Then read the last event version from the event store to the latest event, and then replay to get the latest state.

This way we still use event traceability to manage events and also get the snapshot functionality.

The code in the current library pg seems to store the state and event streams directly, and reads the state directly, and does not use the event traceability feature. Perhaps this part of the functionality could be integrated directly in the EventStore.

Of course, these are my ideas and I haven't thought of a better way to handle this. If you have better ways and ideas If I would be happy to communicate with you here.

Other

And I have another question, should the aggregate replay event return an error?

Events that are already stored in the database should be perfectly correct and perfectly recovered when read, otherwise doesn't that mean there is an error in the event stream?

I think this is a point that could be improved.

And regarding the event version change or database read errors should be marginal errors and shouldn't be handled in the core domain.

[ar3s3ru]

Hi @cgglyle, sorry for getting back to this so late 🙇🏻

But I have a little idea about Projections. I think that Projections is a special kind of Aggregate, and maybe you can just reuse the code of Aggregate.

Uuh I like that.

We could indeed define Projection as an Aggregate alias (since the current trait can be indeed used also to define a Projection in this manner, through apply).

In this case, we could even go as far as providing a pre-built event::Processor implementation that:

• Loads the latest Projection state through a Repository
• Runs Projection::apply(incoming_event)
• Store the new Projection state

We could potentially even reuse an AggregateRepository, but it should be non-event-sourced (though not a strict requirement) and we should be able to point the Repository to a custom-defined table. However, since we're dealing with Projections, most likely people would implement their own Repository implementation (read-optimized, etc.)

One issue in the project is that we don't have a comprehensive-yet-simple-enough example to use to benchmark these design proposals. Opened a PR some time ago but never finished it: #252

Regarding snapshots, I think a snapshot manager could be designed to set up a snapshot schedule, say every 10 events to take a snapshot and then store it in the pg aggregate table with the last event version of the snapshot documented. When we need to read it we can get the last event version directly from the snapshot. Then read the last event version from the event store to the latest event, and then replay to get the latest state.

Happy to explore this further. Can you elaborate on how this would look like concretely?

Ideally I would like to find a solution to this that doesn't break the current API.

And I have another question, should the aggregate replay event return an error?

Events that are already stored in the database should be perfectly correct and perfectly recovered when read, otherwise doesn't that mean there is an error in the event stream?

Couple of points here.

If you're referring to the fact that Aggregate::apply indeed returns a Result, this is primarily a defensive abstraction to handle scenarios that should not be possible (e.g. state=Some(...) && event=WasCreated -> that's something that should NOT happen).

I tried to experiment a lot with the Aggregate trait definition, but couldn't find a nice way to express this. Happy to hear options if you have some 😄

Also, I wouldn't assume that the event stream is perfectly correct. Human error is possible (and I've witnessed it) when dealing with event-sourced systems, especially if not everyone in the engineering team is well-versed in DDD/CQRS/ES; for many teams and companies, the natural instinct to fix production incidents is to go straight to the DB and waffle around.

And regarding the event version change or database read errors should be marginal errors and shouldn't be handled in the core domain.

I don't see that required to be handled in the domain - can you explain a bit further? Pointing at code and some examples help!

[cgglyle]

Hi @ar3s3ru

It's nice to get a response from you.

The part about Projection

I was thinking the same thing, maybe just reuse Aggregate and Root to provide replay functionality, but the actual projection storage should be left to the user, who may handle it specifically for projection.

Regarding the qpply error in Aggregate using Result is the right choice.

Because if the wrong event stream would cause a crash, the system would be much less reliable.

But I still think it's necessary to check the event stream as it's read in EventStored or in Streamer, checking for completeness, or version order, etc., and throwing errors as early as possible, instead of waiting until the core of the domain to find exceptions.

Is there a good solution for corrupt event streams? Manual processing? Or append some kind of Undo behavior? But the error occurs before the execution gets to the Undo event, or save the error for now, expecting the later event to be Undo and resume the stream? (I don't have a good idea)

I don't see that required to be handled in the domain - can you explain a bit further? Pointing at code and some examples help!

I'm referring to the type Error in Aggregate, I think the error applied should be a system fixed error, like ReplayFailed for example, and not a field error, because by default there is no problem with playback, and the only thing that could cause a problem with playback could be a system problem (programming error, database error, etc.), and to return A Domain error is very confusing, and may make the user think it is normal, but it is not.

A system or programming error is likely to be unrecoverable by the user, so it might be better to just remove the type Error?

About snapshots

I haven't thought about this yet. This part could be very complicated.

Ideally I would like to find a solution to this that doesn't break the current API.

On that note, since we're not at 1.x.x yet, it might be a good idea to make destructive changes to the system, and try to leave behind as much historical baggage as possible, so that we don't end up with a system that has a heavy history like Windows or Java Spring Boot.

Other

The Aggregate commands and playback in the project are split into apply and Root, which is different from the common all-in-one aggregation, which I think is excellent, and the separation of concerns is a good choice, which is a benefit I can think of.