Add new generic on PgStore and converter trait to decouple persistence from Aggregate::Event

.clippy.toml

@@ -1 +1 @@
-msrv = "1.58.0"
+msrv = "1.74.0"

.github/workflows/ci.yml

@@ -12,6 +12,11 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
+      - run: |
+          rustup override set 1.74.0
+          rustup component add rustfmt
+          rustup component add clippy
+          rustup component add rust-docs
       - uses: Swatinem/rust-cache@v2
       - uses: taiki-e/install-action@cargo-make
       - run: cargo make fmt-check
@@ -25,6 +30,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
+      - run: rustup override set 1.74.0
       - uses: Swatinem/rust-cache@v2
       - uses: taiki-e/install-action@cargo-make
       - name: Add hosts entries

Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 name = "esrs"
 readme = "README.md"
 repository = "https://github.com/primait/event_sourcing.rs"
-rust-version = "1.58.0"
+rust-version = "1.74.0"
 version = "0.14.0"
 
 [package.metadata.docs.rs]

Makefile.toml

@@ -78,6 +78,7 @@ script = [
     "cargo run --example readme --features=postgres",
     "cargo run --example rebuilder --features=rebuilder,postgres",
     "cargo run --example saga --features=postgres",
+    "cargo run --example schema --features=postgres",
     "cargo run --example shared_view --features=postgres",
     "cargo run --example store_crud --features=postgres",
     "cargo run --example transactional_view --features=postgres",
@@ -95,6 +96,7 @@ script = [
     "cargo clippy --example readme --features=postgres -- -D warnings",
     "cargo clippy --example rebuilder --features=rebuilder,postgres -- -D warnings",
     "cargo clippy --example saga --features=postgres -- -D warnings",
+    "cargo clippy --example schema --features=postgres -- -D warnings",
     "cargo clippy --example shared_view --features=postgres -- -D warnings",
     "cargo clippy --example store_crud --features=postgres -- -D warnings",
     "cargo clippy --example transactional_view --features=postgres -- -D warnings",

README.md

@@ -212,6 +212,14 @@ store.persist(&mut state, events).await?;
 To alleviate the burden of writing all of this code, you can leverage the `AggregateManager` helper. An `AggregateManager`
 could be considered as a synchronous `CommandBus`.
 
+##### Decoupling `Aggregate::Event` from the database using `Schema`
+
+To avoid strong coupling between the domain events represented by `Aggregate::Event` and the persistence layer. It is possible to introduce a `Schema` type on the `PgStore`.
+
+This type must implement `Schema` and `Persistable`. The mechanism enables the domain events to evolve more freely. For example, it is possible to deprecate an event variant making use of the schema (see [deprecating events example](examples/schema/deprecating_events.rs)). Additionally, this mechanism can be used as an alternative for upcasting (see [upcasting example](examples/schema/upcasting.rs)).
+
+For an example of how to introduce a schema to an existing application see [introducing schema example](examples/schema/adding_schema.rs).
+
 ```rust
 let manager: AggregateManager<_> = AggregateManager::new(store);
 manager.handle_command(Default::default(), BookCommand::Buy { num_of_copies: 1 }).await

examples/common/a.rs

@@ -38,4 +38,4 @@ pub struct EventA {
 }
 
 #[cfg(feature = "upcasting")]
-impl esrs::event::Upcaster for EventA {}
+impl esrs::sql::event::Upcaster for EventA {}

examples/common/b.rs

@@ -38,4 +38,4 @@ pub struct EventB {
 }
 
 #[cfg(feature = "upcasting")]
-impl esrs::event::Upcaster for EventB {}
+impl esrs::sql::event::Upcaster for EventB {}

examples/common/basic/mod.rs

@@ -38,7 +38,7 @@ pub struct BasicEvent {
 }
 
 #[cfg(feature = "upcasting")]
-impl esrs::event::Upcaster for BasicEvent {}
+impl esrs::sql::event::Upcaster for BasicEvent {}
 
 #[allow(dead_code)]
 #[derive(Debug, thiserror::Error)]

examples/readme/main.rs

@@ -99,7 +99,7 @@ pub enum BookEvent {
 }
 
 #[cfg(feature = "upcasting")]
-impl esrs::event::Upcaster for BookEvent {}
+impl esrs::sql::event::Upcaster for BookEvent {}
 
 #[derive(Debug, Error)]
 pub enum BookError {

examples/saga/aggregate.rs

@@ -36,4 +36,4 @@ pub enum SagaEvent {
 }
 
 #[cfg(feature = "upcasting")]
-impl esrs::event::Upcaster for SagaEvent {}
+impl esrs::sql::event::Upcaster for SagaEvent {}

examples/schema/adding_schema.rs

@@ -0,0 +1,184 @@
+//! This example demonstrates how it is possible to add a schema to a store.
+//!
+//! The module `before_schema` represents the code in the system before the introduction of the
+//! schema. The first part of the example shows the state of the system before the schema is
+//! introduced.
+//!
+//! The module `after_schema` represents the code in the system after the introduction of the
+//! schema. Similarly, the second part of the example shows how the to setup the PgStore in this
+//! state and how the previously persisted events are visible via the schema.
+//!
+//! Note that the introduction of the schema removes the need for `Aggregate::Event` to implement
+//! `Persistable` instead relies on the implementation of `Schema` and the fact that `Schema`
+//! implements `Persistable`.
+
+use esrs::manager::AggregateManager;
+use serde::{Deserialize, Serialize};
+
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use esrs::store::postgres::{PgStore, PgStoreBuilder};
+use esrs::store::EventStore;
+use esrs::AggregateState;
+
+use crate::common::CommonError;
+
+pub(crate) enum Command {}
+
+mod before_schema {
+    //! This module represents the code of the initial iteration of the system before the
+    //! introduction of the schema.
+    use super::*;
+    pub(crate) struct Aggregate;
+
+    impl esrs::Aggregate for Aggregate {
+        const NAME: &'static str = "introducing_schema";
+        type State = SchemaState;
+        type Command = Command;
+        type Event = Event;
+        type Error = CommonError;
+
+        fn handle_command(_state: &Self::State, command: Self::Command) -> Result<Vec<Self::Event>, Self::Error> {
+            match command {}
+        }
+
+        fn apply_event(state: Self::State, payload: Self::Event) -> Self::State {
+            let mut events = state.events;
+            events.push(payload);
+
+            Self::State { events }
+        }
+    }
+
+    #[derive(Default)]
+    pub(crate) struct SchemaState {
+        pub(crate) events: Vec<Event>,
+    }
+
+    // Required only before the schema is introduced, after the schema is introduced
+    // we do not need to make the type serializable. See other schema examples.
+    #[derive(Deserialize, Serialize)]
+    pub enum Event {
+        A,
+        B { contents: String },
+        C { count: u64 },
+    }
+
+    #[cfg(feature = "upcasting")]
+    impl esrs::sql::event::Upcaster for Event {}
+}
+
+mod after_schema {
+    //! This module represents the code after the introduction of the schema.
+    use super::*;
+    pub(crate) struct Aggregate;
+
+    impl esrs::Aggregate for Aggregate {
+        const NAME: &'static str = "introducing_schema";
+        type State = SchemaState;
+        type Command = Command;
+        type Event = Event;
+        type Error = CommonError;
+
+        fn handle_command(_state: &Self::State, command: Self::Command) -> Result<Vec<Self::Event>, Self::Error> {
+            match command {}
+        }
+
+        fn apply_event(state: Self::State, payload: Self::Event) -> Self::State {
+            let mut events = state.events;
+            events.push(payload);
+
+            Self::State { events }
+        }
+    }
+
+    #[derive(Default)]
+    pub(crate) struct SchemaState {
+        pub(crate) events: Vec<Event>,
+    }
+
+    pub enum Event {
+        A,
+        B { contents: String },
+        C { count: u64 },
+    }
+    #[derive(Deserialize, Serialize)]
+    pub enum Schema {
+        A,
+        B { contents: String },
+        C { count: u64 },
+    }
+
+    impl esrs::store::postgres::Schema<Event> for Schema {
+        fn from_event(value: Event) -> Self {
+            match value {
+                Event::A => Schema::A,
+                Event::B { contents } => Schema::B { contents },
+                Event::C { count } => Schema::C { count },
+            }
+        }
+
+        fn to_event(self) -> Option<Event> {
+            match self {
+                Self::A => Some(Event::A),
+                Self::B { contents } => Some(Event::B { contents }),
+                Self::C { count } => Some(Event::C { count }),
+            }
+        }
+    }
+
+    #[cfg(feature = "upcasting")]
+    impl esrs::sql::event::Upcaster for Schema {}
+}
+
+pub(crate) async fn example(pool: PgPool) {
+    let aggregate_id: Uuid = Uuid::new_v4();
+
+    // Initial state of system before introduction of schema
+    {
+        use before_schema::{Aggregate, Event};
+
+        let store: PgStore<Aggregate> = PgStoreBuilder::new(pool.clone()).try_build().await.unwrap();
+
+        let events = vec![
+            Event::A,
+            Event::B {
+                contents: "hello world".to_owned(),
+            },
+            Event::C { count: 42 },
+        ];
+
+        let mut state = AggregateState::with_id(aggregate_id);
+        let events = store.persist(&mut state, events).await.unwrap();
+
+        assert_eq!(events.len(), 3);
+    }
+
+    // After introducing Schema
+    {
+        use after_schema::{Aggregate, Event, Schema};
+
+        let store: PgStore<Aggregate, Schema> = PgStoreBuilder::new(pool.clone())
+            .with_schema::<Schema>()
+            .try_build()
+            .await
+            .unwrap();
+
+        let events = vec![
+            Event::A,
+            Event::B {
+                contents: "goodbye world".to_owned(),
+            },
+            Event::C { count: 42 },
+        ];
+
+        let manager = AggregateManager::new(store.clone());
+        let mut state = manager.load(aggregate_id).await.unwrap().unwrap();
+        let _ = store.persist(&mut state, events).await.unwrap();
+
+        let events = manager.load(aggregate_id).await.unwrap().unwrap().into_inner().events;
+
+        assert_eq!(events.len(), 6);
+    }
+}