refactor: use Interface Segregation to split traits in individual ones (#238)

* refactor(eventually): split event::Store in event::Streamer and event::Appender

* refactor(eventually): split aggregate::Repository in aggregate::Saver and aggregate::Getter

* fix(examples/bank-accounting): use the new interfaces

* fix: tests broken from api changes are now passing

Author: ar3s3ru

eventually-postgres/src/aggregate.rs

@@ -5,7 +5,8 @@ use eventually::{
     aggregate,
     aggregate::Aggregate,
     serde::{Deserializer, Serde, Serializer},
-    version::{ConflictError, Version},
+    version,
+    version::Version,
 };
 use sqlx::{PgPool, Postgres, Row};
 
@@ -56,19 +57,25 @@ where
 }
 
 #[derive(Debug, thiserror::Error)]
-pub enum RepositoryError {
+pub enum GetError {
     #[error("failed to fetch the aggregate state row: {0}")]
     FetchAggregateRow(#[source] sqlx::Error),
     #[error("failed to deserialize the aggregate state from the database row: {0}")]
     DeserializeAggregate(#[source] Box<dyn std::error::Error + Send + Sync + 'static>),
     #[error("failed to convert the aggregate state into its domain type: {0}")]
     ConvertAggregate(#[source] Box<dyn std::error::Error + Send + Sync + 'static>),
+    #[error("database returned an error: {0}")]
+    Database(#[from] sqlx::Error),
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum SaveError {
     #[error("failed to begin a new transaction: {0}")]
     BeginTransaction(#[source] sqlx::Error),
     #[error("conflict error detected: {0})")]
-    Conflict(#[source] ConflictError),
+    Conflict(#[source] version::ConflictError),
     #[error("concurrent update detected, represented as a conflict error: {0})")]
-    Concurrency(#[source] ConflictError),
+    Concurrency(#[source] version::ConflictError),
     #[error("failed to save the new aggregate state: {0}")]
     SaveAggregateState(#[source] sqlx::Error),
     #[error("failed to append a new domain event: {0}")]
@@ -79,11 +86,11 @@ pub enum RepositoryError {
     Database(#[from] sqlx::Error),
 }
 
-impl From<RepositoryError> for Option<ConflictError> {
-    fn from(err: RepositoryError) -> Self {
+impl From<SaveError> for Option<version::ConflictError> {
+    fn from(err: SaveError) -> Self {
         match err {
-            RepositoryError::Conflict(v) => Some(v),
-            RepositoryError::Concurrency(v) => Some(v),
+            SaveError::Conflict(v) => Some(v),
+            SaveError::Concurrency(v) => Some(v),
             _ => None,
         }
     }
@@ -104,7 +111,7 @@ where
         aggregate_id: &str,
         expected_version: Version,
         root: &mut aggregate::Root<T>,
-    ) -> Result<(), RepositoryError> {
+    ) -> Result<(), SaveError> {
         let out_state = root.to_aggregate_type::<OutT>();
         let bytes_state = self.aggregate_serde.serialize(out_state);
 
@@ -117,13 +124,15 @@ where
             .execute(tx)
             .await
             .map_err(|err| match crate::check_for_conflict_error(&err) {
-                Some(err) => RepositoryError::Conflict(err),
+                Some(err) => SaveError::Conflict(err),
                 None => match err.as_database_error().and_then(|err| err.code()) {
-                    Some(code) if code == "40001" => RepositoryError::Concurrency(ConflictError {
-                        expected: expected_version,
-                        actual: root.version(),
-                    }),
-                    _ => RepositoryError::SaveAggregateState(err),
+                    Some(code) if code == "40001" => {
+                        SaveError::Concurrency(version::ConflictError {
+                            expected: expected_version,
+                            actual: root.version(),
+                        })
+                    }
+                    _ => SaveError::SaveAggregateState(err),
                 },
             })?;
 
@@ -132,7 +141,7 @@ where
 }
 
 #[async_trait]
-impl<T, OutT, OutEvt, TSerde, EvtSerde> aggregate::Repository<T>
+impl<T, OutT, OutEvt, TSerde, EvtSerde> aggregate::Getter<T>
     for Repository<T, OutT, OutEvt, TSerde, EvtSerde>
 where
     T: Aggregate + TryFrom<OutT> + Send + Sync,
@@ -144,7 +153,7 @@ where
     <TSerde as Deserializer<OutT>>::Error: std::error::Error + Send + Sync + 'static,
     EvtSerde: Serializer<OutEvt> + Send + Sync,
 {
-    type Error = RepositoryError;
+    type Error = GetError;
 
     async fn get(
         &self,
@@ -163,25 +172,41 @@ where
         .await
         .map_err(|err| match err {
             sqlx::Error::RowNotFound => aggregate::RepositoryGetError::AggregateRootNotFound,
-            _ => aggregate::RepositoryGetError::Inner(RepositoryError::FetchAggregateRow(err)),
+            _ => aggregate::RepositoryGetError::Inner(GetError::FetchAggregateRow(err)),
         })?;
 
-        let version: i32 = row.try_get("version").map_err(RepositoryError::Database)?;
-        let bytes_state: Vec<u8> = row.try_get("state").map_err(RepositoryError::Database)?;
+        let version: i32 = row.try_get("version").map_err(GetError::Database)?;
+        let bytes_state: Vec<u8> = row.try_get("state").map_err(GetError::Database)?;
 
         let aggregate: T = self
             .aggregate_serde
             .deserialize(bytes_state)
-            .map_err(|err| RepositoryError::DeserializeAggregate(Box::new(err)))
+            .map_err(|err| GetError::DeserializeAggregate(Box::new(err)))
             .and_then(|out_t| {
-                T::try_from(out_t).map_err(|err| RepositoryError::ConvertAggregate(Box::new(err)))
+                T::try_from(out_t).map_err(|err| GetError::ConvertAggregate(Box::new(err)))
             })?;
 
         Ok(aggregate::Root::rehydrate_from_state(
             version as Version,
             aggregate,
         ))
     }
+}
+
+#[async_trait]
+impl<T, OutT, OutEvt, TSerde, EvtSerde> aggregate::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 store(&self, root: &mut aggregate::Root<T>) -> Result<(), Self::Error> {
         let events_to_commit = root.take_uncommitted_events();
@@ -194,7 +219,7 @@ where
             .pool
             .begin()
             .await
-            .map_err(RepositoryError::BeginTransaction)?;
+            .map_err(SaveError::BeginTransaction)?;
 
         sqlx::query("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE DEFERRABLE")
             .execute(&mut tx)
@@ -214,11 +239,9 @@ where
             events_to_commit,
         )
         .await
-        .map_err(RepositoryError::AppendEvent)?;
+        .map_err(SaveError::AppendEvent)?;
 
-        tx.commit()
-            .await
-            .map_err(RepositoryError::CommitTransaction)?;
+        tx.commit().await.map_err(SaveError::CommitTransaction)?;
 
         Ok(())
     }

eventually-postgres/src/event.rs

@@ -209,8 +209,7 @@ where
     }
 }
 
-#[async_trait]
-impl<Id, Evt, OutEvt, S> event::Store for Store<Id, Evt, OutEvt, S>
+impl<Id, Evt, OutEvt, S> event::Streamer<Id, Evt> for Store<Id, Evt, OutEvt, S>
 where
     Id: ToString + Clone + Send + Sync,
     Evt: TryFrom<OutEvt> + Message + std::fmt::Debug + Send + Sync,
@@ -219,16 +218,9 @@ where
     S: Serde<OutEvt> + Send + Sync,
     <S as Deserializer<OutEvt>>::Error: std::error::Error + Send + Sync + 'static,
 {
-    type StreamId = Id;
-    type Event = Evt;
-    type StreamError = StreamError;
-    type AppendError = AppendError;
+    type Error = StreamError;
 
-    fn stream(
-        &self,
-        id: &Self::StreamId,
-        select: event::VersionSelect,
-    ) -> event::Stream<Self::StreamId, Self::Event, Self::StreamError> {
+    fn stream(&self, id: &Id, select: event::VersionSelect) -> event::Stream<Id, Evt, Self::Error> {
         let from_version: i32 = match select {
             event::VersionSelect::All => 0,
             event::VersionSelect::From(v) => v as i32,
@@ -251,13 +243,26 @@ where
             .and_then(move |row| ready(self.event_row_to_persisted_event(id.clone(), row)))
             .boxed()
     }
+}
+
+#[async_trait]
+impl<Id, Evt, OutEvt, S> event::Appender<Id, Evt> for Store<Id, Evt, OutEvt, S>
+where
+    Id: ToString + Clone + Send + Sync,
+    Evt: TryFrom<OutEvt> + Message + std::fmt::Debug + Send + Sync,
+    <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,
+{
+    type Error = AppendError;
 
     async fn append(
         &self,
-        id: Self::StreamId,
+        id: Id,
         version_check: event::StreamVersionExpected,
-        events: Vec<event::Envelope<Self::Event>>,
-    ) -> Result<Version, Self::AppendError> {
+        events: Vec<event::Envelope<Evt>>,
+    ) -> Result<Version, Self::Error> {
         let mut tx = self
             .pool
             .begin()

eventually-postgres/tests/aggregate_repository.rs

@@ -1,5 +1,5 @@
 use eventually::{
-    aggregate::{Repository, RepositoryGetError},
+    aggregate::{Getter, RepositoryGetError, Saver},
     serde::json::Json,
     version,
 };

eventually-postgres/tests/event_store.rs

@@ -1,7 +1,7 @@
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use eventually::{
-    event::{Persisted, Store, StreamVersionExpected, VersionSelect},
+    event::{Appender, Persisted, StreamVersionExpected, Streamer, VersionSelect},
     serde::json::Json,
     version,
     version::Version,

eventually/src/aggregate.rs

@@ -29,7 +29,8 @@ use crate::{event, message, version::Version};
 
 mod repository;
 pub use repository::{
-    EventSourced as EventSourcedRepository, GetError as RepositoryGetError, Repository,
+    EventSourced as EventSourcedRepository, GetError as RepositoryGetError, Getter, Repository,
+    Saver,
 };
 
 /// An Aggregate represents a Domain Model that, through an Aggregate [Root],
@@ -390,7 +391,7 @@ mod test {
     use crate::{
         aggregate,
         aggregate::test_user_domain::{User, UserEvent},
-        aggregate::Repository,
+        aggregate::{Getter, Saver},
         event,
         event::store::EventStoreExt,
         version,

eventually/src/aggregate/repository.rs

@@ -17,25 +17,44 @@ pub enum GetError<I> {
     Inner(#[from] I),
 }
 
-/// A Repository is an object that allows to load and save
-/// an [Aggregate Root][Root] from and to a persistent data store.
 #[async_trait]
-pub trait Repository<T>: Send + Sync
+pub trait Getter<T>: Send + Sync
 where
     T: Aggregate,
 {
-    /// The error type that can be returned by the Repository implementation
-    /// during loading or storing of an Aggregate Root.
-    type Error;
+    type Error: Send + Sync;
 
     /// Loads an Aggregate Root instance from the data store,
     /// referenced by its unique identifier.
     async fn get(&self, id: &T::Id) -> Result<aggregate::Root<T>, GetError<Self::Error>>;
+}
+
+#[async_trait]
+pub trait Saver<T>: Send + Sync
+where
+    T: Aggregate,
+{
+    type Error: Send + Sync;
 
     /// Stores a new version of an Aggregate Root instance to the data store.
     async fn store(&self, root: &mut aggregate::Root<T>) -> Result<(), Self::Error>;
 }
 
+/// A Repository is an object that allows to load and save
+/// an [Aggregate Root][Root] from and to a persistent data store.
+pub trait Repository<T>: Getter<T> + Saver<T> + Send + Sync
+where
+    T: Aggregate,
+{
+}
+
+impl<T, R> Repository<T> for R
+where
+    T: Aggregate,
+    R: Getter<T> + Saver<T> + Send + Sync,
+{
+}
+
 /// List of possible errors that can be returned by an [`EventSourced`] method.
 #[derive(Debug, thiserror::Error)]
 pub enum EventSourcedError<E, SE, AE> {
@@ -71,7 +90,7 @@ pub enum EventSourcedError<E, SE, AE> {
 pub struct EventSourced<T, S>
 where
     T: Aggregate,
-    S: event::Store<StreamId = T::Id, Event = T::Event>,
+    S: event::Store<T::Id, T::Event>,
 {
     store: S,
     aggregate: PhantomData<T>,
@@ -80,7 +99,7 @@ where
 impl<T, S> From<S> for EventSourced<T, S>
 where
     T: Aggregate,
-    S: event::Store<StreamId = T::Id, Event = T::Event>,
+    S: event::Store<T::Id, T::Event>,
 {
     fn from(store: S) -> Self {
         Self {
@@ -91,14 +110,18 @@ where
 }
 
 #[async_trait]
-impl<T, S> Repository<T> for EventSourced<T, S>
+impl<T, S> Getter<T> for EventSourced<T, S>
 where
     T: Aggregate,
     T::Id: Clone,
     T::Error: Debug,
-    S: event::Store<StreamId = T::Id, Event = T::Event>,
+    S: event::Store<T::Id, T::Event>,
 {
-    type Error = EventSourcedError<T::Error, S::StreamError, S::AppendError>;
+    type Error = EventSourcedError<
+        T::Error,
+        <S as event::Streamer<T::Id, T::Event>>::Error,
+        <S as event::Appender<T::Id, T::Event>>::Error,
+    >;
 
     async fn get(&self, id: &T::Id) -> Result<aggregate::Root<T>, GetError<Self::Error>> {
         let ctx = self
@@ -120,6 +143,21 @@ where
 
         ctx.ok_or(GetError::AggregateRootNotFound)
     }
+}
+
+#[async_trait]
+impl<T, S> Saver<T> for EventSourced<T, S>
+where
+    T: Aggregate,
+    T::Id: Clone,
+    T::Error: Debug,
+    S: event::Store<T::Id, T::Event>,
+{
+    type Error = EventSourcedError<
+        T::Error,
+        <S as event::Streamer<T::Id, T::Event>>::Error,
+        <S as event::Appender<T::Id, T::Event>>::Error,
+    >;
 
     async fn store(&self, root: &mut aggregate::Root<T>) -> Result<(), Self::Error> {
         let events_to_commit = root.take_uncommitted_events();