First some basic suggestion: please enable Discussion for this project and/or make some Matrix channel etc. for users to hang out in, ask for help and socialize. Had it been already there, I would have just asked there. :)

I wanted to start using "event processing" (write stuff down in a log, and then have other stuff follow that log and react to it) in my Rust projects. I did not specifically thought about ES/CQRS but I think what I wanted to achieve might fit in this model.

Your crate looks like a solid, more informed base and I went through the book, read through the demo code and some of the source code and I think I can find myself here. I like that it's lightweight, to the point and supports Postgres. Great job.

I basically want to write a CI/CD bot to power my fake Rust/Nix software shop. I'm thinking - I consume github webhooks as "Commands", I track aggregates like "PR" (PR, commits in it, all the pushes, all the builds).

Q1: Are Querieseagerly and reactive? I mean - on every new event, I can expect the "simple query" to be called eagerly and in it's code I can emit any side effects(?) like starting a new CI build? Is it guaranteed? The ("it's useless in production"](https://github.com/serverlesstechnology/cqrs-demo/blob/50029766363aa0ff324ccd264bdeed532e5a236e/src/queries.rs#L12) is a bit confusing. I guess the framework itself does not keep track of which events each query processed already, so I would have to persist some "pointer" to skip emitting side effects for events I already reacted to before (some idempotency-like considerations), but otherwise that's it? BTW. You might want to consider building a concept of a "Reactor" or something that would wrap a Query and keep track of position in the event stream automatically - seems rather generic.

Q2: How do I approach multiple event streams including splitting between multiple "microservices"? The demos are kind of simplistic and include only one aggregate. I might just need to read more literature, since it's more of an architecture question, but if you could give me a pointer or two, I would appreciate it.

I guess I can have multiple programs using cqrs library pointing at the same DB in Postgres and one (or even more) of them writing events, and all following new events. I do realize that I can have things like CDC with Kafka/Kinesis etc. but it's kind of heavy for my tiny internal tools use, at least for now.

Anyway, thanks for your work on this project, and sooner or later I'll figure it out and maybe post some links with results.

Hi,

Given a single view composed by two aggregates roots, Vendor and Product:

pub struct VendorView {
    pub id: String,
    pub name: String,
    pub products: Vec<VendorViewProduct>,
}

pub struct VendorViewProduct {
    pub id: String,
    pub name: String,
    pub price: u32,
}

where id is the aggregate ID of each aggregate root, and the actual view_id is VendorView.id copied from the Vendor. This view stores all products of a single vendor. The Product aggregate has a vendor_id field to reference the Vendor.

I also wondered whether Product should not be a aggregate root at all, but in many scenarios/commands the Product itself is the root to be referenced, with its own stock, prices and so on. So it seems for this case the model is easier if Vendor and Products are separate aggregates.

To dispatch events to this view from separate CQRS instances of each aggregate we implement both View<Vendor> and View<Product> for VendorView. However we have no way to pass this vendor_id as view_id down to the Query because:

pub trait Query<A: Aggregate>: Send + Sync {
    async fn dispatch(&self, aggregate_id: &str, events: &[EventEnvelope<A>]);
}

// cqrs.rs
impl<A, ES> CqrsFramework<A, ES> {
    pub async fn execute_with_metadata(...) -> Result<(), AggregateError<A::Error>> {
let aggregate_context = self.store.load_aggregate(aggregate_id).await?;
        // here the aggregate_id is the Product's. Ok
        let aggregate_context = self.store.load_aggregate(aggregate_id).await?;
        // .....
        for processor in &self.queries {
            let dispatch_events = committed_events.as_slice();
            // for most queries, specially the ones exclusive to the Product, this is OK
            // but for VendorView this is not ok since Project.id != VendorView.id
            processor.dispatch(aggregate_id, dispatch_events).await;
        }
    }
}

One workaround I ended up doing was:

pub trait Query<A: Aggregate>: Send + Sync {
    async fn dispatch(&self, aggregate_id: &str, events: &[EventEnvelope<A>], secondary_id: Option<&str>);
}

pub trait Aggregate: Default + Serialize + DeserializeOwned + Sync + Send {
    fn secondary_id(&self) -> Option<String> {
        None
    }
}

impl Aggregate for Product {
    fn secondary_id(&self) -> Option<String> {
        Some(self.vendor_id.clone())
    }
}

impl<A, ES> CqrsFramework<A, ES> {
    pub async fn execute_with_metadata(...) -> Result<(), AggregateError<A::Error>> {
        // ..... I had to load_aggregate again to get the aggregate with the newly applied event
        let aggregate_context = self.store.load_aggregate(aggregate_id).await?;
        let aggregate = aggregate_context.aggregate();
        let secondary_id = aggregate.secondary_id();
       
        for processor in &self.queries {
            let dispatch_events = committed_events.as_slice();
            processor
                .dispatch(aggregate_id, dispatch_events, secondary_id.as_deref())
                .await;
        }
    }
}

Thanks!

Hi,

I've just read the book about this repo and am really excited about it!

However I am a bit worried about the size of the event store. Is there some kind of garbage collection for old events?

For instance let's say we have a system where only a known set of agents may write commands. For some time it makes sense to store the set of commands since due to networking latency, some commands may arrive delayed. However when we received some message from all writing agents, we may safely assume that the state of our system does not change at the point where we are sure to have received all commands to that point in time. Therefore we might calculate the state at that point, take it as new "start" and delete the events before, saving a lot of memory. Kind of like a concept called snapshots I read about, but to save storage, not computation time.

Or do I overestimate the problem the indefinitely growing event/command logs are?

Hi, I'm working on some views where I want to send the traditional created_at and updated_at fields to the front end. All the data for this can be easily derived from the timestamp field, so I'd like to use it, but it isn't surfaced by the EventEnvelope. I realise that currently there is no dependency on any DateTime libraries, so this isn't a trivial change, but it seems like the alternative is to store the timestamp in each event, which is less than ideal. Do you have any other thoughts on how this can be approached?

Thanks!

Hi again! One use case I've run into recently is sometimes I want to build an aggregate without a command to send it. It might be in lieu of a query where the query would be overkill or as part of a query processor chain. In order to do this, I can call load_aggregate on the EventStore object and call the aggregate() method. Unfortunately, there is no public handle to the EventStore member in the CqrsFramework struct, so I have to create duplicate EventSource objects that I pass along to my axum routes. This works fine, but it would be a lot nicer if either the EventSource was public or I could access it somehow. What are your thoughts?

I noticed that there is a CQRS-ES2 crate that is forked from this project; that project has made big updates about ownership and makes large changes, so should this project be considered unmaintained and should we instead be using that other crate?

I guess what I'm really asking here is if the CQRS-ES2 crate should be considered the modern replacement of this one that we should actually use instead?

I'm seeking guidance on the right CQRS crate to use for a new project.

https://github.com/serverlesstechnology/cqrs/blob/86d83dc5bb773742513bd2a60e42d219dff92e58/src/cqrs.rs#L187

        let committed_events = self
            .store
            .commit(resultant_events, aggregate_context, metadata)
            .await?;
        for processor in &self.queries {
            let dispatch_events = committed_events.as_slice();
            processor.dispatch(aggregate_id, dispatch_events).await;
        }

if the program crashes between committing and dispatching to queries, the queries might never see events. Is this on purpose? I would expect that to guarantee at least once delivery the dispatch should be done first, and commit later, but I'm somewhat of a noob.

A view payload {"name": "John"} is compatible with the view

struct ProductView {
    name: String,
}

But it will be incompatible if we change this read model to

struct ProductView {
    name: String,
    price: u32
}

It will then result in DeserializationError(Error("missing fieldevents", line: 0, column: 0). Is it safe to assume that we should add a deserialization fallback in case it fails or even an explicit "view upcast" (I don't even know if this is technically correct) in the same fashion as the event upcast?

Incompatible payloads will exist from the moment the new read model is deployed until the events are replayed to rebuild the views, which can take some time to be fully replayed and updated.

About event replay, I see that there is only PersistedEventRepository::get_events (by single aggregate_id) and ViewRepository::update_view (for single view). I forked the repos to add (it's still WIP):

// PersistedEventRepository
   async fn get_multiple_aggregate_events<A: Aggregate>(
        &self,
        aggregate_ids: Vec<&str>,
    ) -> Result<HashMap<String, Vec<SerializedEvent>>, PersistenceError>;
    
// ViewRepository
    async fn update_views(&self, views: Vec<(V, ViewContext)>) -> Result<(), PersistenceError>;

So we can batch select and update aggregate and views. What are your thoughts?

Hi, thanks for all your work. I'm learning a lot about how crqs/es works using your module and applying it against a real world modelling scenario. I saw in your book that you say:

A single table can be used for all aggregate events and this is usually the ideal setup for development and testing. For production systems it is recommended that each aggregate have a table solely dedicated to its' events.

I looked through the code, but it appears that this isn't supported at this point unless I'm missing something. I didn't see anything else in the upcoming 0.3 commit log either, but I may have missed something. Is this in development or on the roadmap? Cheers!

Sending events over channels is possible because DomainEvents can be serialized and deserialized before sending and after receiving.

However, in cases where a command is to be sent, a channel sending and receiving aggregate commands cannot generally be constructed and has to be handled on a case by case basis for each aggregate command type.

One way to fix this is to add a constraint to the Aggregate trait such as:

#[async_trait]
pub trait Aggregate: Default + Serialize + DeserializeOwned + Sync + Send {
    /// Specifies the inbound command used to make changes in the state of the Aggregate.
    type Command: Serialize + DeserializeOwned + fmt::Debug + Sync + Send;
...
}

See https://github.com/serverlesstechnology/cqrs/pull/32

Hello guys

We were PoCing around your crate when found this :

https://github.com/serverlesstechnology/cqrs/blob/09fc47b886739d9fb34d7cc05015b05162a13f48/src/cqrs.rs#L154

So the events are consumed to the projections from the command handler ? What if the machine stops ? What if the view database have issues ? What will happen when the service restarts ?

this is an experiment to test the approach described in #46

removing the upcast API reduces the amount of code to maintain by around 500 lines. Would require some documentation updates to demonstrate to users how to implement upcasting using serde.

Benefits:

• simplify the API (smaller surface, less maintenance)
• replaces a stringly-typed callback interface with a strongly-typed declarative approach
• (optionally) separates internal and serialised representations of an event

Todo:

• [ ] update documentation to show upcast pattern using serde
• [ ] investigate helpers/macros for reducing boilerplate in pattern

I wanted to present an alternative pattern for event upcasting. This relies on serde's ability to deserialize untagged enum representations.

I've used this pattern before for backwards compatibility of configuration files-

The gist is to separate the internal representation of an event from its serialised representation. It's serialised representation is an untagged union of all historical versions of the Event. You then add an infallible conversion from the union to the current version, and let serde do the rest.

use serde::{Deserialize, Serialize};

mod legacy {
    //! Previous versions of the `Event` enum, for backwards compatibility
    use serde::{Deserialize, Serialize};

    #[derive(Serialize, Deserialize)]
    pub enum V1 {}

    #[derive(Serialize, Deserialize)]
    pub enum V2 {}
}

// This is version 3 of the 'event'
#[derive(Serialize, Deserialize)]
#[serde(from = "EventRep")]
pub enum Event {}

#[derive(Serialize, Deserialize)]
#[serde(untagged)]
enum EventRep {
    V1(legacy::V1),
    V2(legacy::V2),
    V3(Event)
}

impl From<EventRep> for Event {
    fn from(value: EventRep) -> Self {
        match value {
            EventRep::V1(_) => todo!(),
            EventRep::V2(_) => todo!(),
            EventRep::V3(event) => event,
        }
    }
}

For fallible conversions, you could also use #[serde(try_from = "EventRep"].

Implementing upcasting this way simplifies the implementation of the framework, and removes the 'stringly' typed upcasting API in favour of a strongly-typed pattern. The downside is possibly more cognitive load on downstream users to implement this themselves and to get it right.

Obviously this is a breaking change, but i'm interested to get your thoughts.

I'd say it's likely to be possible to simplify some of the boilerplate with a derive macro, if such a thing doesn't already exist in the wild

the demo application should probably be moved into this repo, and this repo modified to be a cargo workspace

this ensures that this library and the demo application are updated in lockstep.

I had a quick go at this, but ran into some subtle version issues. I suspect postgres-es would also need to be added to the workspace to prevent multiple different versions of cqrs being pulled into the dependency tree

adds github continuous integration workflows to check formatting and tests

note that this PR is a prerequisite for merging #36 (since this PR prevents dependabot from inadvertently bumping the minimum compiler version for downstream users). #36 is very high-value for this crate since it is security conscious. Dependabot will automatically open PRs to address security vulnerabilities (as well as plain old outdated dependencies)

you can see the results of these checks here - https://github.com/danieleades/cqrs/pull/1