refactor: simplify Repository and Serde traits, introduce AnyRepository, remove ProtobufSerde (#270)

Author: ar3s3ru

eventually-postgres/src/aggregate.rs

@@ -2,7 +2,7 @@ use std::marker::PhantomData;
 
 use async_trait::async_trait;
 use eventually::aggregate::Aggregate;
-use eventually::serde::{Deserializer, Serde, Serializer};
+use eventually::serde::Serde;
 use eventually::version::Version;
 use eventually::{aggregate, version};
 use sqlx::{PgPool, Postgres, Row};
@@ -15,7 +15,7 @@ where
     OutT: From<T>,
     OutEvt: From<T::Event>,
     TSerde: Serde<OutT>,
-    EvtSerde: Serializer<OutEvt>,
+    EvtSerde: Serde<OutEvt>,
 {
     pool: PgPool,
     aggregate_serde: TSerde,
@@ -32,7 +32,7 @@ where
     OutT: From<T>,
     OutEvt: From<T::Event>,
     TSerde: Serde<OutT>,
-    EvtSerde: Serializer<OutEvt>,
+    EvtSerde: Serde<OutEvt>,
 {
     pub async fn new(
         pool: PgPool,
@@ -100,7 +100,7 @@ where
     OutT: From<T> + Send + Sync,
     OutEvt: From<T::Event>,
     TSerde: Serde<OutT> + Send + Sync,
-    EvtSerde: Serializer<OutEvt>,
+    EvtSerde: Serde<OutEvt>,
 {
     async fn save_aggregate_state(
         &self,
@@ -138,7 +138,7 @@ where
 }
 
 #[async_trait]
-impl<T, OutT, OutEvt, TSerde, EvtSerde> aggregate::repository::Getter<T>
+impl<T, OutT, OutEvt, TSerde, EvtSerde> aggregate::Repository<T>
     for Repository<T, OutT, OutEvt, TSerde, EvtSerde>
 where
     T: Aggregate + TryFrom<OutT> + Send + Sync,
@@ -147,15 +147,16 @@ where
     OutT: From<T> + Send + Sync,
     OutEvt: From<T::Event> + Send + Sync,
     TSerde: Serde<OutT> + Send + Sync,
-    <TSerde as Deserializer<OutT>>::Error: std::error::Error + Send + Sync + 'static,
-    EvtSerde: Serializer<OutEvt> + Send + Sync,
+    <TSerde as Serde<OutT>>::Error: std::error::Error + Send + Sync + 'static,
+    EvtSerde: Serde<OutEvt> + Send + Sync,
 {
-    type Error = GetError;
+    type GetError = GetError;
+    type SaveError = SaveError;
 
     async fn get(
         &self,
         id: &T::Id,
-    ) -> Result<aggregate::Root<T>, aggregate::repository::GetError<Self::Error>> {
+    ) -> Result<aggregate::Root<T>, aggregate::repository::GetError<Self::GetError>> {
         let aggregate_id = id.to_string();
 
         let row = sqlx::query(
@@ -188,24 +189,8 @@ where
             aggregate,
         ))
     }
-}
-
-#[async_trait]
-impl<T, OutT, OutEvt, TSerde, EvtSerde> aggregate::repository::Saver<T>
-    for Repository<T, OutT, OutEvt, TSerde, EvtSerde>
-where
-    T: Aggregate + TryFrom<OutT> + Send + Sync,
-    <T as Aggregate>::Id: ToString,
-    <T as TryFrom<OutT>>::Error: std::error::Error + Send + Sync + 'static,
-    OutT: From<T> + Send + Sync,
-    OutEvt: From<T::Event> + Send + Sync,
-    TSerde: Serde<OutT> + Send + Sync,
-    <TSerde as Deserializer<OutT>>::Error: std::error::Error + Send + Sync + 'static,
-    EvtSerde: Serializer<OutEvt> + Send + Sync,
-{
-    type Error = SaveError;
 
-    async fn save(&self, root: &mut aggregate::Root<T>) -> Result<(), Self::Error> {
+    async fn save(&self, root: &mut aggregate::Root<T>) -> Result<(), Self::SaveError> {
         let events_to_commit = root.take_uncommitted_events();
 
         if events_to_commit.is_empty() {

eventually-postgres/src/event.rs

@@ -4,12 +4,11 @@ use std::string::ToString;
 use async_trait::async_trait;
 use chrono::Utc;
 use eventually::message::{Message, Metadata};
-use eventually::serde::{Deserializer, Serde, Serializer};
+use eventually::serde::Serde;
 use eventually::version::Version;
 use eventually::{event, version};
 use futures::future::ready;
 use futures::{StreamExt, TryStreamExt};
-use lazy_static::lazy_static;
 use regex::Regex;
 use sqlx::postgres::{PgDatabaseError, PgRow};
 use sqlx::{PgPool, Postgres, Row, Transaction};
@@ -60,7 +59,7 @@ impl From<AppendError> for Option<version::ConflictError> {
 
 pub(crate) async fn append_domain_event<Evt, OutEvt>(
     tx: &mut Transaction<'_, Postgres>,
-    serde: &impl Serializer<OutEvt>,
+    serde: &impl Serde<OutEvt>,
     event_stream_id: &str,
     event_version: i32,
     new_event_stream_version: i32,
@@ -97,7 +96,7 @@ where
 
 pub(crate) async fn append_domain_events<Evt, OutEvt>(
     tx: &mut Transaction<'_, Postgres>,
-    serde: &impl Serializer<OutEvt>,
+    serde: &impl Serde<OutEvt>,
     event_stream_id: &str,
     new_version: i32,
     events: Vec<event::Envelope<Evt>>,
@@ -176,7 +175,7 @@ where
     <Evt as TryFrom<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
     OutEvt: From<Evt> + Send + Sync,
     S: Serde<OutEvt> + Send + Sync,
-    <S as Deserializer<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
+    <S as Serde<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
 {
     fn event_row_to_persisted_event(
         &self,
@@ -213,7 +212,7 @@ where
     <Evt as TryFrom<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
     OutEvt: From<Evt> + Send + Sync,
     S: Serde<OutEvt> + Send + Sync,
-    <S as Deserializer<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
+    <S as Serde<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
 {
     type Error = StreamError;
 
@@ -250,7 +249,7 @@ where
     <Evt as TryFrom<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
     OutEvt: From<Evt> + Send + Sync,
     S: Serde<OutEvt> + Send + Sync,
-    <S as Deserializer<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
+    <S as Serde<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
 {
     type Error = AppendError;
 

eventually-postgres/tests/aggregate_repository.rs

@@ -1,5 +1,6 @@
-use eventually::aggregate::repository::{GetError, Getter, Saver};
-use eventually::serde::json::Json;
+use eventually::aggregate::repository::GetError;
+use eventually::aggregate::Repository;
+use eventually::serde::json::JsonSerde;
 use eventually::version;
 use eventually_postgres::aggregate;
 use futures::TryFutureExt;
@@ -15,8 +16,8 @@ async fn it_works() {
 
     let aggregate_repository = aggregate::Repository::new(
         pool,
-        Json::<setup::TestAggregate>::default(),
-        Json::<setup::TestDomainEvent>::default(),
+        JsonSerde::<setup::TestAggregate>::default(),
+        JsonSerde::<setup::TestDomainEvent>::default(),
     )
     .await
     .unwrap();
@@ -64,8 +65,8 @@ async fn it_detects_data_races_and_returns_conflict_error() {
 
     let aggregate_repository = aggregate::Repository::new(
         pool,
-        Json::<setup::TestAggregate>::default(),
-        Json::<setup::TestDomainEvent>::default(),
+        JsonSerde::<setup::TestAggregate>::default(),
+        JsonSerde::<setup::TestDomainEvent>::default(),
     )
     .await
     .unwrap();

eventually-postgres/tests/event_store.rs

@@ -1,11 +1,11 @@
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use eventually::event::{Appender, Persisted, StreamVersionExpected, Streamer, VersionSelect};
-use eventually::serde::json::Json;
+use eventually::serde::json::JsonSerde;
 use eventually::version;
 use eventually::version::Version;
 use eventually_postgres::event;
-use futures::{TryFutureExt, TryStreamExt};
+use futures::TryStreamExt;
 use rand::Rng;
 
 mod setup;
@@ -16,7 +16,7 @@ async fn append_with_no_version_check_works() {
         .await
         .expect("connection to the database should work");
 
-    let event_store = event::Store::new(pool, Json::<setup::TestDomainEvent>::default())
+    let event_store = event::Store::new(pool, JsonSerde::<setup::TestDomainEvent>::default())
         .await
         .unwrap();
 
@@ -72,7 +72,7 @@ async fn it_works_with_version_check_for_conflict() {
         .await
         .expect("connection to the database should work");
 
-    let event_store = event::Store::new(pool, Json::<setup::TestDomainEvent>::default())
+    let event_store = event::Store::new(pool, JsonSerde::<setup::TestDomainEvent>::default())
         .await
         .unwrap();
 
@@ -148,7 +148,7 @@ async fn it_handles_concurrent_writes_to_the_same_stream() {
         .await
         .expect("connection to the database should work");
 
-    let event_store = event::Store::new(pool, Json::<setup::TestDomainEvent>::default())
+    let event_store = event::Store::new(pool, JsonSerde::<setup::TestDomainEvent>::default())
         .await
         .unwrap();
 

eventually/Cargo.toml

@@ -21,8 +21,7 @@ default = []
 tracing = ["dep:tracing"]
 serde-prost = ["dep:prost"]
 serde-json = ["dep:serde_json"]
-serde-protobuf = ["dep:protobuf", "dep:protobuf-json-mapping"]
-full = ["serde-prost", "serde-json", "serde-protobuf", "tracing"]
+full = ["serde-prost", "serde-json", "tracing"]
 
 [dependencies]
 async-trait = "0.1.74"
@@ -31,9 +30,8 @@ thiserror = "1.0.50"
 prost = { version = "0.12.1", optional = true }
 serde_json = { version = "1.0.108", optional = true }
 serde = { version = "1.0.192", features = ["derive"] }
-protobuf = { version = "3.3.0", optional = true }
-protobuf-json-mapping = { version = "3.3.0", optional = true }
 tracing = { version = "0.1.40", features = ["async-await"], optional = true }
+anyhow = "1.0.75"
 
 [dev-dependencies]
 # NOTE: this is only used for test components and assertions.

eventually/src/aggregate.rs eventually/src/aggregate/mod.rseventually/src/aggregate.rs renamed to eventually/src/aggregate/mod.rs

@@ -387,8 +387,8 @@ pub(crate) mod test_user_domain {
 mod test {
     use std::error::Error;
 
-    use crate::aggregate::repository::{Getter, Saver};
     use crate::aggregate::test_user_domain::{User, UserEvent};
+    use crate::aggregate::Repository;
     use crate::event::store::EventStoreExt;
     use crate::{aggregate, event, version};