test: transport × persistence matrix for the distributed read-model flow

Cargo.toml

@@ -27,11 +27,10 @@ categories = ["data-structures", "database"]
 path = "src/lib.rs"
 
 [features]
-default = ["emitter", "bus"]
+default = ["emitter"]
 emitter = ["dep:event-emitter-rs"]
-bus = []
-http = ["bus", "dep:axum", "dep:reqwest", "dep:tokio"]
-grpc = ["bus", "dep:tonic", "dep:prost", "dep:tokio"]
+http = ["dep:axum", "dep:reqwest", "dep:tokio"]
+grpc = ["dep:tonic", "dep:prost", "dep:tokio"]
 postgres = ["dep:sqlx", "dep:tokio", "sqlx/postgres", "sqlx/runtime-tokio"]
 sqlite = ["dep:sqlx", "dep:tokio", "sqlx/runtime-tokio", "sqlx/sqlite"]
 nats = ["dep:async-nats", "dep:futures", "dep:tokio"]

src/aggregate/aggregate.rs

@@ -1,12 +1,7 @@
 use std::fmt;
-use std::marker::PhantomData;
 
 use crate::entity::{upcast_events, Entity, EventRecord, EventUpcaster};
-use crate::queued_repo::{GetAllWithOpts, GetWithOpts, ReadOpts, UnlockableRepository};
-use crate::repository::{
-    Commit, CommitBatch, Get, Repository, RepositoryError, TransactionalCommit,
-};
-use crate::snapshot::{SnapshotAggregateRepository, SnapshotStore, Snapshottable};
+use crate::repository::RepositoryError;
 
 /// Trait for domain aggregates that can be event-sourced.
 pub trait Aggregate: Sized + Default {
@@ -123,211 +118,3 @@ pub fn hydrate<A: Aggregate>(entity: Entity) -> Result<A, RepositoryError> {
 
     Ok(agg)
 }
-
-/// Extension trait adding aggregate-aware get method.
-pub trait GetAggregate: Get {
-    fn get_aggregate<A: Aggregate>(&self, id: &str) -> Result<Option<A>, RepositoryError>
-    where
-        Self: Sized,
-    {
-        let entity = self.get(id)?;
-        let Some(entity) = entity else {
-            return Ok(None);
-        };
-        Ok(Some(hydrate::<A>(entity)?))
-    }
-}
-
-impl<R: Get> GetAggregate for R {}
-
-/// Extension trait adding aggregate-aware get_all method.
-pub trait GetAllAggregates: Get {
-    fn get_all_aggregates<A: Aggregate>(&self, ids: &[&str]) -> Result<Vec<A>, RepositoryError>
-    where
-        Self: Sized,
-    {
-        let entities = self.get(ids)?;
-        let mut aggregates = Vec::with_capacity(entities.len());
-        for entity in entities {
-            aggregates.push(hydrate::<A>(entity)?);
-        }
-        Ok(aggregates)
-    }
-}
-
-impl<R: Get> GetAllAggregates for R {}
-
-/// Extension trait adding aggregate-aware commit methods.
-pub trait CommitAggregate: Commit {
-    fn commit_aggregate<A: Aggregate>(&self, aggregate: &mut A) -> Result<(), RepositoryError> {
-        self.commit(aggregate.entity_mut())
-    }
-
-    fn commit_all_aggregates<A: Aggregate>(
-        &self,
-        aggregates: &mut [&mut A],
-    ) -> Result<(), RepositoryError>
-    where
-        Self: TransactionalCommit,
-    {
-        let entities: Vec<&mut Entity> = aggregates
-            .iter_mut()
-            .map(|agg| (*agg).entity_mut())
-            .collect();
-        self.commit_batch(CommitBatch::new(entities))
-    }
-}
-
-impl<R: Commit> CommitAggregate for R {}
-
-/// Combined extension trait for full repository aggregate support.
-pub trait RepositoryExt: GetAggregate + GetAllAggregates + CommitAggregate {}
-
-impl<R: Repository> RepositoryExt for R {}
-
-/// Builder trait for creating typed aggregate repositories.
-pub trait AggregateBuilder: Sized {
-    fn aggregate<A: Aggregate>(self) -> AggregateRepository<Self, A> {
-        AggregateRepository::new(self)
-    }
-}
-
-impl<T> AggregateBuilder for T {}
-
-/// A repository wrapper that provides typed access to a specific aggregate type.
-pub struct AggregateRepository<R, A> {
-    repo: R,
-    _marker: PhantomData<A>,
-}
-
-impl<R, A> AggregateRepository<R, A> {
-    pub fn new(repo: R) -> Self {
-        AggregateRepository {
-            repo,
-            _marker: PhantomData,
-        }
-    }
-
-    pub fn repo(&self) -> &R {
-        &self.repo
-    }
-
-    pub fn repo_mut(&mut self) -> &mut R {
-        &mut self.repo
-    }
-}
-
-impl<R, A> AggregateRepository<R, A>
-where
-    R: Get,
-    A: Aggregate,
-{
-    pub fn get(&self, id: &str) -> Result<Option<A>, RepositoryError> {
-        let entity = self.repo.get(id)?;
-        let Some(entity) = entity else {
-            return Ok(None);
-        };
-        Ok(Some(hydrate::<A>(entity)?))
-    }
-}
-
-impl<R, A> AggregateRepository<R, A>
-where
-    R: Get,
-    A: Aggregate,
-{
-    pub fn get_all(&self, ids: &[&str]) -> Result<Vec<A>, RepositoryError> {
-        let entities = self.repo.get(ids)?;
-        let mut aggregates = Vec::with_capacity(entities.len());
-        for entity in entities {
-            aggregates.push(hydrate::<A>(entity)?);
-        }
-        Ok(aggregates)
-    }
-}
-
-impl<R, A> AggregateRepository<R, A>
-where
-    R: Commit,
-    A: Aggregate,
-{
-    pub fn commit(&self, aggregate: &mut A) -> Result<(), RepositoryError> {
-        self.repo.commit(aggregate.entity_mut())
-    }
-}
-
-impl<R, A> AggregateRepository<R, A>
-where
-    R: TransactionalCommit,
-    A: Aggregate,
-{
-    pub fn commit_all(&self, aggregates: &mut [&mut A]) -> Result<(), RepositoryError> {
-        let entities: Vec<&mut Entity> = aggregates
-            .iter_mut()
-            .map(|agg| (*agg).entity_mut())
-            .collect();
-        self.repo.commit_batch(CommitBatch::new(entities))
-    }
-}
-
-impl<R, A> AggregateRepository<R, A>
-where
-    R: UnlockableRepository,
-    A: Aggregate,
-{
-    pub fn abort(&self, aggregate: &A) -> Result<(), RepositoryError> {
-        self.repo.unlock(aggregate.entity().id())
-    }
-}
-
-impl<R, A> AggregateRepository<R, A>
-where
-    R: SnapshotStore,
-    A: Snapshottable,
-{
-    /// Wrap this repository with snapshot support at the given event frequency.
-    pub fn with_snapshots(self, frequency: u64) -> SnapshotAggregateRepository<R, A> {
-        SnapshotAggregateRepository::new(self, frequency)
-    }
-}
-
-impl<R, A> AggregateRepository<R, A>
-where
-    R: GetWithOpts,
-    A: Aggregate,
-{
-    /// Get an aggregate with options (e.g., to skip locking).
-    pub fn get_with(&self, id: &str, opts: ReadOpts) -> Result<Option<A>, RepositoryError> {
-        let entity = self.repo.get_with(id, opts)?;
-        let Some(entity) = entity else {
-            return Ok(None);
-        };
-        Ok(Some(hydrate::<A>(entity)?))
-    }
-
-    /// Non-locking read (alias for get_with no_lock).
-    pub fn peek(&self, id: &str) -> Result<Option<A>, RepositoryError> {
-        self.get_with(id, ReadOpts::no_lock())
-    }
-}
-
-impl<R, A> AggregateRepository<R, A>
-where
-    R: GetAllWithOpts,
-    A: Aggregate,
-{
-    /// Get all aggregates with options (e.g., to skip locking).
-    pub fn get_all_with(&self, ids: &[&str], opts: ReadOpts) -> Result<Vec<A>, RepositoryError> {
-        let entities = self.repo.get_all_with(ids, opts)?;
-        let mut aggregates = Vec::with_capacity(entities.len());
-        for entity in entities {
-            aggregates.push(hydrate::<A>(entity)?);
-        }
-        Ok(aggregates)
-    }
-
-    /// Non-locking read (alias for get_all_with no_lock).
-    pub fn peek_all(&self, ids: &[&str]) -> Result<Vec<A>, RepositoryError> {
-        self.get_all_with(ids, ReadOpts::no_lock())
-    }
-}

src/aggregate/async_aggregate.rs

@@ -1,6 +1,9 @@
 use std::marker::PhantomData;
 
 use crate::entity::Entity;
+use crate::queued_repo::{
+    AsyncGetAllWithOpts, AsyncGetWithOpts, AsyncUnlockableRepository, ReadOpts,
+};
 use crate::repository::{
     AsyncCommitBatch, AsyncGetStream, AsyncStreamWrite, AsyncTransactionalCommit, RepositoryError,
     StreamIdentity,
@@ -117,3 +120,71 @@ where
             .await
     }
 }
+
+impl<R, A> AsyncAggregateRepository<R, A>
+where
+    R: AsyncGetWithOpts,
+    A: Aggregate + Send,
+{
+    /// Load an aggregate with options (e.g. `ReadOpts::no_lock()` to skip the
+    /// queue lock when the repository is a `queued_async()` wrapper).
+    pub async fn get_with(&self, id: &str, opts: ReadOpts) -> Result<Option<A>, RepositoryError> {
+        let identity = stream_identity_for::<A>(id)?;
+        let Some(entity) = self.repo.get_stream_with(&identity, opts).await? else {
+            return Ok(None);
+        };
+        Ok(Some(hydrate::<A>(entity)?))
+    }
+
+    /// Non-locking read (alias for `get_with(ReadOpts::no_lock())`).
+    pub async fn peek(&self, id: &str) -> Result<Option<A>, RepositoryError> {
+        self.get_with(id, ReadOpts::no_lock()).await
+    }
+}
+
+impl<R, A> AsyncAggregateRepository<R, A>
+where
+    R: AsyncGetAllWithOpts,
+    A: Aggregate + Send,
+{
+    /// Load aggregates for the provided ids with options.
+    pub async fn get_all_with(
+        &self,
+        ids: &[&str],
+        opts: ReadOpts,
+    ) -> Result<Vec<A>, RepositoryError> {
+        let identities = ids
+            .iter()
+            .map(|id| stream_identity_for::<A>(id))
+            .collect::<Result<Vec<_>, _>>()?;
+        let entities = self.repo.get_streams_with(&identities, opts).await?;
+        let mut aggregates = Vec::with_capacity(entities.len());
+        for entity in entities {
+            aggregates.push(hydrate::<A>(entity)?);
+        }
+        Ok(aggregates)
+    }
+
+    /// Non-locking multi-read (alias for `get_all_with(ReadOpts::no_lock())`).
+    pub async fn peek_all(&self, ids: &[&str]) -> Result<Vec<A>, RepositoryError> {
+        self.get_all_with(ids, ReadOpts::no_lock()).await
+    }
+}
+
+impl<R, A> AsyncAggregateRepository<R, A>
+where
+    R: AsyncUnlockableRepository,
+    A: Aggregate,
+{
+    /// Release the lock held for an aggregate after an aborted load.
+    pub fn abort(&self, aggregate: &A) -> Result<(), RepositoryError> {
+        let identity = stream_identity_for::<A>(aggregate.entity().id())?;
+        self.repo.unlock(&identity)
+    }
+
+    /// Release the lock held for an aggregate id.
+    pub fn unlock(&self, id: &str) -> Result<(), RepositoryError> {
+        let identity = stream_identity_for::<A>(id)?;
+        self.repo.unlock(&identity)
+    }
+}

src/aggregate/mod.rs

@@ -1,8 +1,5 @@
 mod aggregate;
 mod async_aggregate;
 
-pub use aggregate::{
-    hydrate, Aggregate, AggregateBuilder, AggregateRepository, CommitAggregate, GetAggregate,
-    GetAllAggregates, RepositoryExt,
-};
+pub use aggregate::{hydrate, Aggregate};
 pub use async_aggregate::{AsyncAggregateBuilder, AsyncAggregateRepository};