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.