diff --git a/Cargo.lock b/Cargo.lock index 2ebba45..16db641 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10,7 +10,7 @@ checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" [[package]] name = "fmodel-decider-rust" -version = "0.2.0" +version = "0.3.0" dependencies = [ "pretty_assertions", ] diff --git a/Cargo.toml b/Cargo.toml index 9fc1f19..812a02c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "fmodel-decider-rust" -version = "0.2.0" +version = "0.3.0" edition = "2024" description = "Pure, type-safe, event-driven decision engine with progressive type refinement for event-sourced and state-stored systems" license = "Apache-2.0" diff --git a/src/application/event_repository.rs b/src/application/event_repository.rs index ec982f5..479cebe 100644 --- a/src/application/event_repository.rs +++ b/src/application/event_repository.rs @@ -1,4 +1,4 @@ -use crate::EventComputationTrait; +use crate::{CommandQueries, EventComputationTrait, EventMeta, IdempotencyKey}; // ================================================================================================ // EventRepository Trait @@ -12,9 +12,16 @@ use crate::EventComputationTrait; /// /// # Type Parameters /// -/// - `C`: Command type that triggers state changes -/// - `Ei`: Input event type (events loaded from storage) -/// - `Eo`: Output event type (events to be persisted) +/// - `C`: Command type that triggers state changes. Must implement [`IdempotencyKey`] so +/// repository implementations can deduplicate retried commands, and [`CommandQueries`] so +/// DCB-style repositories know which events to fetch. In multi-threaded mode, must also +/// be `Send + Sync` since it is moved into the `Send` future returned by `execute`. +/// - `Ei`: Input event type (events loaded from storage). Must implement [`EventMeta`] so +/// repository implementations can build secondary indexes. In multi-threaded mode, must +/// also be `Send + Sync`. +/// - `Eo`: Output event type (events to be persisted). Must implement [`EventMeta`] so +/// repository implementations can index newly persisted events. In multi-threaded mode, +/// must also be `Send + Sync`. /// /// # Associated Types /// @@ -43,12 +50,41 @@ use crate::EventComputationTrait; /// ```rust,no_run /// use std::collections::HashMap; /// use std::sync::{Arc, Mutex}; -/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider, EventMeta, IdempotencyKey, CommandQueries, QueryTuple, Tag}; /// /// # #[derive(Clone, Debug)] -/// # enum Command { OpenAccount { id: String } } +/// # enum Command { OpenAccount { id: String, idempotency_key: String } } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::OpenAccount { idempotency_key, .. } => idempotency_key, +/// # } +/// # } +/// # } +/// # impl CommandQueries for Command { +/// # fn queries(&self) -> Vec { +/// # match self { +/// # Command::OpenAccount { id, .. } => vec![QueryTuple { +/// # event_type: "AccountOpened".to_string(), +/// # tags: vec![Tag::new("id", id.clone())], +/// # }], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug)] /// # enum Event { AccountOpened { id: String } } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::AccountOpened { .. } => "AccountOpened", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # match self { +/// # Event::AccountOpened { id } => vec![Tag::new("id", id.clone())], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { opened: bool } /// @@ -58,7 +94,12 @@ use crate::EventComputationTrait; /// } /// /// # #[cfg(not(feature = "single-threaded"))] -/// # trait EventRepository: Send + Sync { +/// # trait EventRepository: Send + Sync +/// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, command: C, decider: &D) -> Result, Self::Error> /// # where D: EventComputationTrait + Send + Sync, D::Error: std::fmt::Debug; @@ -79,7 +120,7 @@ use crate::EventComputationTrait; /// { /// // 1. FETCH: Extract stream ID and load events /// let stream_id = match &command { -/// Command::OpenAccount { id } => id.clone(), +/// Command::OpenAccount { id, .. } => id.clone(), /// }; /// let mut events = self.events.lock().unwrap(); /// let current_events = events.get(&stream_id).cloned().unwrap_or_default(); @@ -105,7 +146,7 @@ use crate::EventComputationTrait; /// # let decider = AggregateDecider::new( /// # |c: &Command, _s: &State| -> Result, String> { /// # match c { -/// # Command::OpenAccount { id } => Ok(vec![Event::AccountOpened { id: id.clone() }]), +/// # Command::OpenAccount { id, .. } => Ok(vec![Event::AccountOpened { id: id.clone() }]), /// # } /// # }, /// # |s: &State, _e: &Event| { @@ -118,7 +159,7 @@ use crate::EventComputationTrait; /// # let repository = InMemoryEventRepository { /// # events: Arc::new(Mutex::new(HashMap::new())), /// # }; -/// let command = Command::OpenAccount { id: "acc-123".to_string() }; +/// let command = Command::OpenAccount { id: "acc-123".to_string(), idempotency_key: "req-1".to_string() }; /// /// // Execute command through repository /// let events = repository.execute(command, &decider).await?; @@ -148,7 +189,12 @@ use crate::EventComputationTrait; /// } /// ``` #[cfg(not(feature = "single-threaded"))] -pub trait EventRepository: Send + Sync { +pub trait EventRepository: Send + Sync +where + C: IdempotencyKey + CommandQueries + Send + Sync, + Ei: EventMeta + Send + Sync, + Eo: EventMeta + Send + Sync, +{ /// Error type for repository operations. /// /// This should capture both infrastructure errors (fetch/save failures) and domain @@ -184,19 +230,57 @@ pub trait EventRepository: Send + Sync { /// The exact mechanism depends on the underlying storage (transactions, optimistic /// locking, etc.). /// + /// # Idempotency + /// + /// Implementations should use `command.idempotency_key()` to detect retried commands: + /// when a command with a previously seen key is executed again, the original output + /// events should be returned without re-running the decider. + /// + /// # DCB Queries + /// + /// DCB-style implementations should use `command.queries()` to determine which events + /// to fetch for the FETCH stage, instead of assuming a single fixed stream. + /// /// # Example /// /// ```rust,no_run - /// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider}; + /// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider, EventMeta, IdempotencyKey, CommandQueries, QueryTuple, Tag}; /// # #[derive(Clone, Debug)] - /// # enum Command { Deposit(u32) } + /// # enum Command { Deposit { amount: u32, idempotency_key: String } } + /// # impl IdempotencyKey for Command { + /// # fn idempotency_key(&self) -> &str { + /// # match self { + /// # Command::Deposit { idempotency_key, .. } => idempotency_key, + /// # } + /// # } + /// # } + /// # impl CommandQueries for Command { + /// # fn queries(&self) -> Vec { + /// # Vec::new() + /// # } + /// # } /// # #[derive(Clone, Debug)] /// # enum Event { Deposited(u32) } + /// # impl EventMeta for Event { + /// # fn event_type(&self) -> &str { + /// # match self { + /// # Event::Deposited(_) => "Deposited", + /// # } + /// # } + /// # fn tags(&self) -> Vec { + /// # Vec::new() + /// # } + /// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { balance: u32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] - /// # trait EventRepository: Send + Sync { + /// # trait EventRepository: Send + Sync + /// # where + /// # C: IdempotencyKey + CommandQueries, + /// # Ei: EventMeta, + /// # Eo: EventMeta, + /// # { /// # type Error; /// # async fn execute(&self, command: C, decider: &D) -> Result, Self::Error> /// # where D: EventComputationTrait + Send + Sync; @@ -213,7 +297,7 @@ pub trait EventRepository: Send + Sync { /// # let decider = AggregateDecider::new( /// # |c: &Command, _s: &State| -> Result, String> { /// # match c { - /// # Command::Deposit(amount) => Ok(vec![Event::Deposited(*amount)]), + /// # Command::Deposit { amount, .. } => Ok(vec![Event::Deposited(*amount)]), /// # } /// # }, /// # |s: &State, e: &Event| { @@ -225,7 +309,7 @@ pub trait EventRepository: Send + Sync { /// # }, /// # || State::default(), /// # ); - /// let command = Command::Deposit(100); + /// let command = Command::Deposit { amount: 100, idempotency_key: "req-1".to_string() }; /// let events = repository.execute(command, &decider).await?; /// # Ok(()) /// # } @@ -256,6 +340,17 @@ pub trait EventRepository: Send + Sync { /// succeed and their events are persisted, or none are. The exact mechanism depends on the /// underlying storage (transactions, optimistic locking, etc.). /// + /// # Idempotency + /// + /// As with [`execute`](Self::execute), implementations should use each command's + /// `idempotency_key()` to detect and skip re-running previously handled commands within + /// the batch. + /// + /// # DCB Queries + /// + /// As with [`execute`](Self::execute), DCB-style implementations should use each + /// command's `queries()` to determine which events to fetch for it. + /// /// # Type Parameters /// /// - `D`: The decider implementing `EventComputationTrait` @@ -306,19 +401,46 @@ pub trait EventRepository: Send + Sync { /// use std::rc::Rc; /// use std::cell::RefCell; /// use std::collections::HashMap; -/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider, EventMeta, IdempotencyKey, CommandQueries, QueryTuple, Tag}; /// /// # #[derive(Clone, Debug)] -/// # enum Command { Deposit(u32) } +/// # enum Command { Deposit { amount: u32, idempotency_key: String } } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Deposit { idempotency_key, .. } => idempotency_key, +/// # } +/// # } +/// # } +/// # impl CommandQueries for Command { +/// # fn queries(&self) -> Vec { +/// # Vec::new() +/// # } +/// # } /// # #[derive(Clone, Debug)] /// # enum Event { Deposited(u32) } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::Deposited(_) => "Deposited", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # Vec::new() +/// # } +/// # } /// /// // Single-threaded repository using Rc instead of Arc /// struct SingleThreadedEventRepository { /// events: Rc>>>, /// } /// -/// # trait EventRepository { +/// # trait EventRepository +/// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, command: C, decider: &D) -> Result, Self::Error> /// # where D: EventComputationTrait; @@ -344,7 +466,12 @@ pub trait EventRepository: Send + Sync { /// # } /// ``` #[cfg(feature = "single-threaded")] -pub trait EventRepository { +pub trait EventRepository +where + C: IdempotencyKey + CommandQueries, + Ei: EventMeta, + Eo: EventMeta, +{ /// Error type for repository operations. type Error; diff --git a/src/application/event_sourced_command_handler.rs b/src/application/event_sourced_command_handler.rs index e102ddf..1398481 100644 --- a/src/application/event_sourced_command_handler.rs +++ b/src/application/event_sourced_command_handler.rs @@ -1,5 +1,5 @@ use super::event_repository::EventRepository; -use crate::EventComputationTrait; +use crate::{CommandQueries, EventComputationTrait, EventMeta, IdempotencyKey}; // ================================================================================================ // EventSourcedCommandHandler - Convenience Layer @@ -13,9 +13,10 @@ use crate::EventComputationTrait; /// /// # Type Parameters /// -/// - `C`: Command type that triggers state changes -/// - `Ei`: Input event type (events loaded from storage) -/// - `Eo`: Output event type (events to be persisted) +/// - `C`: Command type that triggers state changes. Must implement [`IdempotencyKey`] and +/// [`CommandQueries`]. +/// - `Ei`: Input event type (events loaded from storage). Must implement [`EventMeta`]. +/// - `Eo`: Output event type (events to be persisted). Must implement [`EventMeta`]. /// - `D`: The decider implementing `EventComputationTrait` /// - `R`: The repository implementing `EventRepository` /// @@ -30,16 +31,43 @@ use crate::EventComputationTrait; /// **Without handler (direct repository usage):** /// ```rust,no_run /// # use std::sync::Arc; -/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider, EventMeta, IdempotencyKey, CommandQueries, QueryTuple, Tag}; /// # #[derive(Clone, Debug)] -/// # enum Command { Deposit(u32) } +/// # enum Command { Deposit { amount: u32, idempotency_key: String } } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Deposit { idempotency_key, .. } => idempotency_key, +/// # } +/// # } +/// # } +/// # impl CommandQueries for Command { +/// # fn queries(&self) -> Vec { +/// # Vec::new() +/// # } +/// # } /// # #[derive(Clone, Debug)] /// # enum Event { Deposited(u32) } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::Deposited(_) => "Deposited", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # Vec::new() +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { balance: u32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] -/// # trait EventRepository: Send + Sync { +/// # trait EventRepository: Send + Sync +/// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, command: C, decider: &D) -> Result, Self::Error> /// # where D: EventComputationTrait + Send + Sync; @@ -59,9 +87,9 @@ use crate::EventComputationTrait; /// # ); /// # let repository = Arc::new(MyRepository); /// // Must pass decider on every call -/// let events1 = repository.execute(Command::Deposit(100), &decider).await?; -/// let events2 = repository.execute(Command::Deposit(200), &decider).await?; -/// let events3 = repository.execute(Command::Deposit(300), &decider).await?; +/// let events1 = repository.execute(Command::Deposit { amount: 100, idempotency_key: "req-1".to_string() }, &decider).await?; +/// let events2 = repository.execute(Command::Deposit { amount: 200, idempotency_key: "req-2".to_string() }, &decider).await?; +/// let events3 = repository.execute(Command::Deposit { amount: 300, idempotency_key: "req-3".to_string() }, &decider).await?; /// # Ok(()) /// # } /// ``` @@ -69,16 +97,43 @@ use crate::EventComputationTrait; /// **With handler:** /// ```rust,no_run /// # use std::sync::Arc; -/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider, EventMeta, IdempotencyKey, CommandQueries, QueryTuple, Tag}; /// # #[derive(Clone, Debug)] -/// # enum Command { Deposit(u32) } +/// # enum Command { Deposit { amount: u32, idempotency_key: String } } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Deposit { idempotency_key, .. } => idempotency_key, +/// # } +/// # } +/// # } +/// # impl CommandQueries for Command { +/// # fn queries(&self) -> Vec { +/// # Vec::new() +/// # } +/// # } /// # #[derive(Clone, Debug)] /// # enum Event { Deposited(u32) } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::Deposited(_) => "Deposited", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # Vec::new() +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { balance: u32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] -/// # trait EventRepository: Send + Sync { +/// # trait EventRepository: Send + Sync +/// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, command: C, decider: &D) -> Result, Self::Error> /// # where D: EventComputationTrait + Send + Sync; @@ -94,6 +149,9 @@ use crate::EventComputationTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # struct EventSourcedCommandHandler /// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, /// # D: EventComputationTrait + Send + Sync, /// # R: EventRepository + Send + Sync, /// # { @@ -104,6 +162,9 @@ use crate::EventComputationTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # impl EventSourcedCommandHandler /// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, /// # D: EventComputationTrait + Send + Sync, /// # R: EventRepository + Send + Sync, /// # { @@ -123,9 +184,9 @@ use crate::EventComputationTrait; /// # let repository = Arc::new(MyRepository); /// // Decider encapsulated in handler /// let handler = EventSourcedCommandHandler::new(decider, repository); -/// let events1 = handler.handle(Command::Deposit(100)).await?; -/// let events2 = handler.handle(Command::Deposit(200)).await?; -/// let events3 = handler.handle(Command::Deposit(300)).await?; +/// let events1 = handler.handle(Command::Deposit { amount: 100, idempotency_key: "req-1".to_string() }).await?; +/// let events2 = handler.handle(Command::Deposit { amount: 200, idempotency_key: "req-2".to_string() }).await?; +/// let events3 = handler.handle(Command::Deposit { amount: 300, idempotency_key: "req-3".to_string() }).await?; /// # Ok(()) /// # } /// ``` @@ -149,16 +210,60 @@ use crate::EventComputationTrait; /// /// ```rust,no_run /// use std::sync::Arc; -/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider, EventMeta, IdempotencyKey, CommandQueries, QueryTuple, Tag}; /// # #[derive(Clone, Debug)] -/// # enum Command { OpenAccount { id: String }, Deposit { id: String, amount: u32 } } +/// # enum Command { +/// # OpenAccount { id: String, idempotency_key: String }, +/// # Deposit { id: String, amount: u32, idempotency_key: String }, +/// # } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::OpenAccount { idempotency_key, .. } => idempotency_key, +/// # Command::Deposit { idempotency_key, .. } => idempotency_key, +/// # } +/// # } +/// # } +/// # impl CommandQueries for Command { +/// # fn queries(&self) -> Vec { +/// # match self { +/// # Command::OpenAccount { id, .. } => vec![QueryTuple { +/// # event_type: "AccountOpened".to_string(), +/// # tags: vec![Tag::new("id", id.clone())], +/// # }], +/// # Command::Deposit { id, .. } => vec![QueryTuple { +/// # event_type: "MoneyDeposited".to_string(), +/// # tags: vec![Tag::new("id", id.clone())], +/// # }], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug)] /// # enum Event { AccountOpened { id: String }, MoneyDeposited { id: String, amount: u32 } } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::AccountOpened { .. } => "AccountOpened", +/// # Event::MoneyDeposited { .. } => "MoneyDeposited", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # match self { +/// # Event::AccountOpened { id } => vec![Tag::new("id", id.clone())], +/// # Event::MoneyDeposited { id, .. } => vec![Tag::new("id", id.clone())], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { balance: u32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] -/// # trait EventRepository: Send + Sync { +/// # trait EventRepository: Send + Sync +/// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, command: C, decider: &D) -> Result, Self::Error> /// # where D: EventComputationTrait + Send + Sync; @@ -174,6 +279,9 @@ use crate::EventComputationTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # struct EventSourcedCommandHandler /// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, /// # D: EventComputationTrait + Send + Sync, /// # R: EventRepository + Send + Sync, /// # { @@ -184,6 +292,9 @@ use crate::EventComputationTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # impl EventSourcedCommandHandler /// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, /// # D: EventComputationTrait + Send + Sync, /// # R: EventRepository + Send + Sync, /// # { @@ -200,8 +311,8 @@ use crate::EventComputationTrait; /// let decider = Arc::new(AggregateDecider::new( /// |c: &Command, _s: &State| -> Result, String> { /// match c { -/// Command::OpenAccount { id } => Ok(vec![Event::AccountOpened { id: id.clone() }]), -/// Command::Deposit { id, amount } => Ok(vec![Event::MoneyDeposited { id: id.clone(), amount: *amount }]), +/// Command::OpenAccount { id, .. } => Ok(vec![Event::AccountOpened { id: id.clone() }]), +/// Command::Deposit { id, amount, .. } => Ok(vec![Event::MoneyDeposited { id: id.clone(), amount: *amount }]), /// } /// }, /// |s: &State, e: &Event| { @@ -219,15 +330,18 @@ use crate::EventComputationTrait; /// let handler = EventSourcedCommandHandler::new(decider, repository); /// /// // Handle commands without passing decider each time -/// let events = handler.handle(Command::OpenAccount { id: "123".to_string() }).await?; -/// let events = handler.handle(Command::Deposit { id: "123".to_string(), amount: 100 }).await?; -/// let events = handler.handle(Command::Deposit { id: "123".to_string(), amount: 50 }).await?; +/// let events = handler.handle(Command::OpenAccount { id: "123".to_string(), idempotency_key: "req-1".to_string() }).await?; +/// let events = handler.handle(Command::Deposit { id: "123".to_string(), amount: 100, idempotency_key: "req-2".to_string() }).await?; +/// let events = handler.handle(Command::Deposit { id: "123".to_string(), amount: 50, idempotency_key: "req-3".to_string() }).await?; /// # Ok(()) /// # } /// ``` #[cfg(not(feature = "single-threaded"))] pub struct EventSourcedCommandHandler where + C: IdempotencyKey + CommandQueries + Send + Sync, + Ei: EventMeta + Send + Sync, + Eo: EventMeta + Send + Sync, D: EventComputationTrait + Send + Sync, R: EventRepository + Send + Sync, { @@ -239,6 +353,9 @@ where #[cfg(not(feature = "single-threaded"))] impl EventSourcedCommandHandler where + C: IdempotencyKey + CommandQueries + Send + Sync, + Ei: EventMeta + Send + Sync, + Eo: EventMeta + Send + Sync, D: EventComputationTrait + Send + Sync, R: EventRepository + Send + Sync, { @@ -257,16 +374,43 @@ where /// /// ```rust,no_run /// use std::sync::Arc; - /// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider}; + /// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider, EventMeta, IdempotencyKey, CommandQueries, QueryTuple, Tag}; /// # #[derive(Clone, Debug)] - /// # enum Command { Deposit(u32) } + /// # enum Command { Deposit { amount: u32, idempotency_key: String } } + /// # impl IdempotencyKey for Command { + /// # fn idempotency_key(&self) -> &str { + /// # match self { + /// # Command::Deposit { idempotency_key, .. } => idempotency_key, + /// # } + /// # } + /// # } + /// # impl CommandQueries for Command { + /// # fn queries(&self) -> Vec { + /// # Vec::new() + /// # } + /// # } /// # #[derive(Clone, Debug)] /// # enum Event { Deposited(u32) } + /// # impl EventMeta for Event { + /// # fn event_type(&self) -> &str { + /// # match self { + /// # Event::Deposited(_) => "Deposited", + /// # } + /// # } + /// # fn tags(&self) -> Vec { + /// # Vec::new() + /// # } + /// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { balance: u32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] - /// # trait EventRepository: Send + Sync { + /// # trait EventRepository: Send + Sync + /// # where + /// # C: IdempotencyKey + CommandQueries, + /// # Ei: EventMeta, + /// # Eo: EventMeta, + /// # { /// # type Error; /// # async fn execute(&self, command: C, decider: &D) -> Result, Self::Error> /// # where D: EventComputationTrait + Send + Sync; @@ -282,6 +426,9 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # struct EventSourcedCommandHandler /// # where + /// # C: IdempotencyKey + CommandQueries, + /// # Ei: EventMeta, + /// # Eo: EventMeta, /// # D: EventComputationTrait + Send + Sync, /// # R: EventRepository + Send + Sync, /// # { @@ -292,6 +439,9 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # impl EventSourcedCommandHandler /// # where + /// # C: IdempotencyKey + CommandQueries, + /// # Ei: EventMeta, + /// # Eo: EventMeta, /// # D: EventComputationTrait + Send + Sync, /// # R: EventRepository + Send + Sync, /// # { @@ -335,16 +485,48 @@ where /// /// ```rust,no_run /// # use std::sync::Arc; - /// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider}; + /// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider, EventMeta, IdempotencyKey, CommandQueries, QueryTuple, Tag}; /// # #[derive(Clone, Debug)] - /// # enum Command { Deposit(u32), Withdraw(u32) } + /// # enum Command { + /// # Deposit { amount: u32, idempotency_key: String }, + /// # Withdraw { amount: u32, idempotency_key: String }, + /// # } + /// # impl IdempotencyKey for Command { + /// # fn idempotency_key(&self) -> &str { + /// # match self { + /// # Command::Deposit { idempotency_key, .. } => idempotency_key, + /// # Command::Withdraw { idempotency_key, .. } => idempotency_key, + /// # } + /// # } + /// # } + /// # impl CommandQueries for Command { + /// # fn queries(&self) -> Vec { + /// # Vec::new() + /// # } + /// # } /// # #[derive(Clone, Debug)] /// # enum Event { Deposited(u32), Withdrawn(u32) } + /// # impl EventMeta for Event { + /// # fn event_type(&self) -> &str { + /// # match self { + /// # Event::Deposited(_) => "Deposited", + /// # Event::Withdrawn(_) => "Withdrawn", + /// # } + /// # } + /// # fn tags(&self) -> Vec { + /// # Vec::new() + /// # } + /// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { balance: u32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] - /// # trait EventRepository: Send + Sync { + /// # trait EventRepository: Send + Sync + /// # where + /// # C: IdempotencyKey + CommandQueries, + /// # Ei: EventMeta, + /// # Eo: EventMeta, + /// # { /// # type Error; /// # async fn execute(&self, command: C, decider: &D) -> Result, Self::Error> /// # where D: EventComputationTrait + Send + Sync; @@ -360,6 +542,9 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # struct EventSourcedCommandHandler /// # where + /// # C: IdempotencyKey + CommandQueries, + /// # Ei: EventMeta, + /// # Eo: EventMeta, /// # D: EventComputationTrait + Send + Sync, /// # R: EventRepository + Send + Sync, /// # { @@ -370,6 +555,9 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # impl EventSourcedCommandHandler /// # where + /// # C: IdempotencyKey + CommandQueries, + /// # Ei: EventMeta, + /// # Eo: EventMeta, /// # D: EventComputationTrait + Send + Sync, /// # R: EventRepository + Send + Sync, /// # { @@ -389,9 +577,9 @@ where /// # let repository = Arc::new(MyRepository); /// # let handler = EventSourcedCommandHandler::new(decider, repository); /// // Handle multiple commands - /// let events1 = handler.handle(Command::Deposit(100)).await?; - /// let events2 = handler.handle(Command::Deposit(200)).await?; - /// let events3 = handler.handle(Command::Withdraw(50)).await?; + /// let events1 = handler.handle(Command::Deposit { amount: 100, idempotency_key: "req-1".to_string() }).await?; + /// let events2 = handler.handle(Command::Deposit { amount: 200, idempotency_key: "req-2".to_string() }).await?; + /// let events3 = handler.handle(Command::Withdraw { amount: 50, idempotency_key: "req-3".to_string() }).await?; /// # Ok(()) /// # } /// ``` @@ -446,15 +634,42 @@ where /// # #[cfg(feature = "single-threaded")] /// # { /// use std::rc::Rc; -/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{EventComputationTrait, AggregateDecider, EventMeta, IdempotencyKey, CommandQueries, QueryTuple, Tag}; /// # #[derive(Clone, Debug)] -/// # enum Command { Deposit(u32) } +/// # enum Command { Deposit { amount: u32, idempotency_key: String } } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Deposit { idempotency_key, .. } => idempotency_key, +/// # } +/// # } +/// # } +/// # impl CommandQueries for Command { +/// # fn queries(&self) -> Vec { +/// # Vec::new() +/// # } +/// # } /// # #[derive(Clone, Debug)] /// # enum Event { Deposited(u32) } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::Deposited(_) => "Deposited", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # Vec::new() +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { balance: u32 } /// # struct MyRepository; -/// # trait EventRepository { +/// # trait EventRepository +/// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, command: C, decider: &D) -> Result, Self::Error> /// # where D: EventComputationTrait; @@ -468,6 +683,9 @@ where /// # use std::marker::PhantomData; /// # struct EventSourcedCommandHandler /// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, /// # D: EventComputationTrait, /// # R: EventRepository, /// # { @@ -477,6 +695,9 @@ where /// # } /// # impl EventSourcedCommandHandler /// # where +/// # C: IdempotencyKey + CommandQueries, +/// # Ei: EventMeta, +/// # Eo: EventMeta, /// # D: EventComputationTrait, /// # R: EventRepository, /// # { @@ -502,6 +723,9 @@ where #[cfg(feature = "single-threaded")] pub struct EventSourcedCommandHandler where + C: IdempotencyKey + CommandQueries, + Ei: EventMeta, + Eo: EventMeta, D: EventComputationTrait, R: EventRepository, { @@ -513,6 +737,9 @@ where #[cfg(feature = "single-threaded")] impl EventSourcedCommandHandler where + C: IdempotencyKey + CommandQueries, + Ei: EventMeta, + Eo: EventMeta, D: EventComputationTrait, R: EventRepository, { diff --git a/src/application/event_sourced_query_handler.rs b/src/application/event_sourced_query_handler.rs index f2671a5..ddd0dde 100644 --- a/src/application/event_sourced_query_handler.rs +++ b/src/application/event_sourced_query_handler.rs @@ -1,26 +1,4 @@ -use crate::ViewTrait; - -// ================================================================================================ -// QueryTuple - Event Query Specification -// ================================================================================================ - -/// A query specification for fetching events from the event store. -/// -/// Consists of an event type and zero or more tags that identify the event stream to query. -/// This is the core crate's dependency-free equivalent of the TypeScript `QueryTuple` type -/// from fmodel-ts: `[...tags, eventType]`. -/// -/// # Fields -/// -/// - `event_type`: The event type identifier (e.g., `"RestaurantCreatedEvent"`) -/// - `tags`: Tags to filter by, in `"key:value"` format. Empty vec queries all events of the type. -#[derive(Debug, Clone)] -pub struct QueryTuple { - /// The event type identifier to query. - pub event_type: String, - /// Tags to filter by, in `"key:value"` format. - pub tags: Vec, -} +use crate::{EventMeta, QueryTuple, ViewTrait}; // ================================================================================================ // EventLoader Trait @@ -36,9 +14,14 @@ pub struct QueryTuple { /// /// # Type Parameters /// -/// - `E`: Event type to load +/// - `E`: Event type to load. Must implement [`EventMeta`], since `QueryTuple` queries by +/// the exact `event_type`/`tags` shape `EventMeta` produces. In multi-threaded mode, must +/// also be `Send + Sync` since it is part of the `Send` future returned by `load`. #[cfg(not(feature = "single-threaded"))] -pub trait EventLoader: Send + Sync { +pub trait EventLoader: Send + Sync +where + E: EventMeta + Send + Sync, +{ /// Error type for load operations. type Error; @@ -53,7 +36,10 @@ pub trait EventLoader: Send + Sync { /// /// See the multi-threaded `EventLoader` documentation for details. #[cfg(feature = "single-threaded")] -pub trait EventLoader { +pub trait EventLoader +where + E: EventMeta, +{ /// Error type for load operations. type Error; @@ -85,6 +71,7 @@ pub trait EventLoader { #[cfg(not(feature = "single-threaded"))] pub struct EventSourcedQueryHandler where + E: EventMeta + Send + Sync, V: ViewTrait + Send + Sync, L: EventLoader + Send + Sync, { @@ -96,6 +83,7 @@ where #[cfg(not(feature = "single-threaded"))] impl EventSourcedQueryHandler where + E: EventMeta + Send + Sync, V: ViewTrait + Send + Sync, L: EventLoader + Send + Sync, { @@ -148,6 +136,7 @@ where #[cfg(feature = "single-threaded")] pub struct EventSourcedQueryHandler where + E: EventMeta, V: ViewTrait, L: EventLoader, { @@ -159,6 +148,7 @@ where #[cfg(feature = "single-threaded")] impl EventSourcedQueryHandler where + E: EventMeta, V: ViewTrait, L: EventLoader, { diff --git a/src/application/materialized_view_handler.rs b/src/application/materialized_view_handler.rs index b85e625..eeae300 100644 --- a/src/application/materialized_view_handler.rs +++ b/src/application/materialized_view_handler.rs @@ -1,5 +1,5 @@ use super::view_repository::ViewRepository; -use crate::ViewTrait; +use crate::{EventMeta, ViewTrait}; // ================================================================================================ // MaterializedViewHandler - Convenience Layer @@ -13,7 +13,7 @@ use crate::ViewTrait; /// /// # Type Parameters /// -/// - `E`: Event type that triggers view updates +/// - `E`: Event type that triggers view updates. Must implement [`EventMeta`]. /// - `S`: State type (both current and evolved state) /// - `V`: The view implementing `ViewTrait` /// - `R`: The repository implementing `ViewRepository` @@ -29,14 +29,31 @@ use crate::ViewTrait; /// **Without handler (direct repository usage):** /// ```rust,no_run /// # use std::sync::Arc; -/// # use fmodel_decider_rust::{ViewTrait, Projection}; +/// # use fmodel_decider_rust::{ViewTrait, Projection, EventMeta, Tag}; /// # #[derive(Clone, Debug)] /// # enum Event { ItemAdded(String), ItemRemoved(String) } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::ItemAdded(_) => "ItemAdded", +/// # Event::ItemRemoved(_) => "ItemRemoved", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # match self { +/// # Event::ItemAdded(item) => vec![Tag::new("item", item.clone())], +/// # Event::ItemRemoved(item) => vec![Tag::new("item", item.clone())], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { items: Vec } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] -/// # trait ViewRepository: Send + Sync { +/// # trait ViewRepository: Send + Sync +/// # where +/// # E: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, event: E, view: &V) -> Result /// # where V: ViewTrait + Send + Sync; @@ -65,14 +82,31 @@ use crate::ViewTrait; /// **With handler:** /// ```rust,no_run /// # use std::sync::Arc; -/// # use fmodel_decider_rust::{ViewTrait, Projection}; +/// # use fmodel_decider_rust::{ViewTrait, Projection, EventMeta, Tag}; /// # #[derive(Clone, Debug)] /// # enum Event { ItemAdded(String), ItemRemoved(String) } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::ItemAdded(_) => "ItemAdded", +/// # Event::ItemRemoved(_) => "ItemRemoved", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # match self { +/// # Event::ItemAdded(item) => vec![Tag::new("item", item.clone())], +/// # Event::ItemRemoved(item) => vec![Tag::new("item", item.clone())], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { items: Vec } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] -/// # trait ViewRepository: Send + Sync { +/// # trait ViewRepository: Send + Sync +/// # where +/// # E: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, event: E, view: &V) -> Result /// # where V: ViewTrait + Send + Sync; @@ -88,6 +122,7 @@ use crate::ViewTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # struct MaterializedViewHandler /// # where +/// # E: EventMeta, /// # V: ViewTrait + Send + Sync, /// # R: ViewRepository + Send + Sync, /// # { @@ -98,6 +133,7 @@ use crate::ViewTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # impl MaterializedViewHandler /// # where +/// # E: EventMeta, /// # V: ViewTrait + Send + Sync, /// # R: ViewRepository + Send + Sync, /// # { @@ -142,14 +178,33 @@ use crate::ViewTrait; /// /// ```rust,no_run /// use std::sync::Arc; -/// # use fmodel_decider_rust::{ViewTrait, Projection}; +/// # use fmodel_decider_rust::{ViewTrait, Projection, EventMeta, Tag}; /// # #[derive(Clone, Debug)] /// # enum Event { AccountOpened { id: String }, MoneyDeposited { id: String, amount: u32 }, MoneyWithdrawn { id: String, amount: u32 } } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::AccountOpened { .. } => "AccountOpened", +/// # Event::MoneyDeposited { .. } => "MoneyDeposited", +/// # Event::MoneyWithdrawn { .. } => "MoneyWithdrawn", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # match self { +/// # Event::AccountOpened { id } => vec![Tag::new("id", id.clone())], +/// # Event::MoneyDeposited { id, .. } => vec![Tag::new("id", id.clone())], +/// # Event::MoneyWithdrawn { id, .. } => vec![Tag::new("id", id.clone())], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { balance: u32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] -/// # trait ViewRepository: Send + Sync { +/// # trait ViewRepository: Send + Sync +/// # where +/// # E: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, event: E, view: &V) -> Result /// # where V: ViewTrait + Send + Sync; @@ -165,6 +220,7 @@ use crate::ViewTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # struct MaterializedViewHandler /// # where +/// # E: EventMeta, /// # V: ViewTrait + Send + Sync, /// # R: ViewRepository + Send + Sync, /// # { @@ -175,6 +231,7 @@ use crate::ViewTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # impl MaterializedViewHandler /// # where +/// # E: EventMeta, /// # V: ViewTrait + Send + Sync, /// # R: ViewRepository + Send + Sync, /// # { @@ -221,6 +278,8 @@ use crate::ViewTrait; #[cfg(not(feature = "single-threaded"))] pub struct MaterializedViewHandler where + E: EventMeta + Send + Sync, + S: Send + Sync, V: ViewTrait + Send + Sync, R: ViewRepository + Send + Sync, { @@ -232,6 +291,8 @@ where #[cfg(not(feature = "single-threaded"))] impl MaterializedViewHandler where + E: EventMeta + Send + Sync, + S: Send + Sync, V: ViewTrait + Send + Sync, R: ViewRepository + Send + Sync, { @@ -250,14 +311,29 @@ where /// /// ```rust,no_run /// use std::sync::Arc; - /// # use fmodel_decider_rust::{ViewTrait, Projection}; + /// # use fmodel_decider_rust::{ViewTrait, Projection, EventMeta, Tag}; /// # #[derive(Clone, Debug)] /// # enum Event { ItemAdded(String) } + /// # impl EventMeta for Event { + /// # fn event_type(&self) -> &str { + /// # match self { + /// # Event::ItemAdded(_) => "ItemAdded", + /// # } + /// # } + /// # fn tags(&self) -> Vec { + /// # match self { + /// # Event::ItemAdded(item) => vec![Tag::new("item", item.clone())], + /// # } + /// # } + /// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { items: Vec } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] - /// # trait ViewRepository: Send + Sync { + /// # trait ViewRepository: Send + Sync + /// # where + /// # E: EventMeta, + /// # { /// # type Error; /// # async fn execute(&self, event: E, view: &V) -> Result /// # where V: ViewTrait + Send + Sync; @@ -273,6 +349,7 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # struct MaterializedViewHandler /// # where + /// # E: EventMeta, /// # V: ViewTrait + Send + Sync, /// # R: ViewRepository + Send + Sync, /// # { @@ -283,6 +360,7 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # impl MaterializedViewHandler /// # where + /// # E: EventMeta, /// # V: ViewTrait + Send + Sync, /// # R: ViewRepository + Send + Sync, /// # { @@ -331,14 +409,31 @@ where /// /// ```rust,no_run /// # use std::sync::Arc; - /// # use fmodel_decider_rust::{ViewTrait, Projection}; + /// # use fmodel_decider_rust::{ViewTrait, Projection, EventMeta, Tag}; /// # #[derive(Clone, Debug)] /// # enum Event { ItemAdded(String), ItemRemoved(String) } + /// # impl EventMeta for Event { + /// # fn event_type(&self) -> &str { + /// # match self { + /// # Event::ItemAdded(_) => "ItemAdded", + /// # Event::ItemRemoved(_) => "ItemRemoved", + /// # } + /// # } + /// # fn tags(&self) -> Vec { + /// # match self { + /// # Event::ItemAdded(item) => vec![Tag::new("item", item.clone())], + /// # Event::ItemRemoved(item) => vec![Tag::new("item", item.clone())], + /// # } + /// # } + /// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { items: Vec } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] - /// # trait ViewRepository: Send + Sync { + /// # trait ViewRepository: Send + Sync + /// # where + /// # E: EventMeta, + /// # { /// # type Error; /// # async fn execute(&self, event: E, view: &V) -> Result /// # where V: ViewTrait + Send + Sync; @@ -354,6 +449,7 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # struct MaterializedViewHandler /// # where + /// # E: EventMeta, /// # V: ViewTrait + Send + Sync, /// # R: ViewRepository + Send + Sync, /// # { @@ -364,6 +460,7 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # impl MaterializedViewHandler /// # where + /// # E: EventMeta, /// # V: ViewTrait + Send + Sync, /// # R: ViewRepository + Send + Sync, /// # { @@ -436,13 +533,28 @@ where /// # #[cfg(feature = "single-threaded")] /// # { /// use std::rc::Rc; -/// # use fmodel_decider_rust::{ViewTrait, Projection}; +/// # use fmodel_decider_rust::{ViewTrait, Projection, EventMeta, Tag}; /// # #[derive(Clone, Debug)] /// # enum Event { ItemAdded(String) } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::ItemAdded(_) => "ItemAdded", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # match self { +/// # Event::ItemAdded(item) => vec![Tag::new("item", item.clone())], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { items: Vec } /// # struct MyRepository; -/// # trait ViewRepository { +/// # trait ViewRepository +/// # where +/// # E: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, event: E, view: &V) -> Result /// # where V: ViewTrait; @@ -456,6 +568,7 @@ where /// # use std::marker::PhantomData; /// # struct MaterializedViewHandler /// # where +/// # E: EventMeta, /// # V: ViewTrait, /// # R: ViewRepository, /// # { @@ -465,6 +578,7 @@ where /// # } /// # impl MaterializedViewHandler /// # where +/// # E: EventMeta, /// # V: ViewTrait, /// # R: ViewRepository, /// # { @@ -495,6 +609,7 @@ where #[cfg(feature = "single-threaded")] pub struct MaterializedViewHandler where + E: EventMeta, V: ViewTrait, R: ViewRepository, { @@ -506,6 +621,7 @@ where #[cfg(feature = "single-threaded")] impl MaterializedViewHandler where + E: EventMeta, V: ViewTrait, R: ViewRepository, { diff --git a/src/application/state_repository.rs b/src/application/state_repository.rs index f81fbe3..ab09020 100644 --- a/src/application/state_repository.rs +++ b/src/application/state_repository.rs @@ -1,4 +1,4 @@ -use crate::StateComputationTrait; +use crate::{IdempotencyKey, StateComputationTrait}; // ================================================================================================ // StateRepository Trait @@ -12,8 +12,11 @@ use crate::StateComputationTrait; /// /// # Type Parameters /// -/// - `C`: Command type that triggers state changes -/// - `S`: State type (both current and new state) +/// - `C`: Command type that triggers state changes. Must implement [`IdempotencyKey`] so +/// repository implementations can deduplicate retried commands. In multi-threaded mode, +/// must also be `Send + Sync` since it is moved into the `Send` future returned by `execute`. +/// - `S`: State type (both current and new state). In multi-threaded mode, must also be +/// `Send + Sync`. /// /// # Associated Types /// @@ -50,10 +53,21 @@ use crate::StateComputationTrait; /// ```rust,no_run /// use std::collections::HashMap; /// use std::sync::{Arc, Mutex}; -/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider, IdempotencyKey}; /// /// # #[derive(Clone, Debug)] -/// # enum Command { Increment, Decrement } +/// # enum Command { +/// # Increment { idempotency_key: String }, +/// # Decrement { idempotency_key: String }, +/// # } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Increment { idempotency_key } => idempotency_key, +/// # Command::Decrement { idempotency_key } => idempotency_key, +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { count: i32 } /// @@ -63,7 +77,10 @@ use crate::StateComputationTrait; /// } /// /// # #[cfg(not(feature = "single-threaded"))] -/// # trait StateRepository: Send + Sync { +/// # trait StateRepository: Send + Sync +/// # where +/// # C: IdempotencyKey, +/// # { /// # type Error; /// # async fn execute(&self, command: C, component: &D) -> Result /// # where D: StateComputationTrait + Send + Sync, D::Error: std::fmt::Debug; @@ -104,8 +121,8 @@ use crate::StateComputationTrait; /// # let component = AggregateDecider::new( /// # |c: &Command, _s: &State| -> Result, String> { /// # match c { -/// # Command::Increment => Ok(vec![()]), -/// # Command::Decrement => Ok(vec![()]), +/// # Command::Increment { .. } => Ok(vec![()]), +/// # Command::Decrement { .. } => Ok(vec![()]), /// # } /// # }, /// # |s: &State, _e: &()| s.clone(), @@ -114,7 +131,7 @@ use crate::StateComputationTrait; /// # let repository = InMemoryStateRepository { /// # states: Arc::new(Mutex::new(HashMap::new())), /// # }; -/// let command = Command::Increment; +/// let command = Command::Increment { idempotency_key: "req-1".to_string() }; /// /// // Execute command through repository /// let state = repository.execute(command, &component).await?; @@ -144,7 +161,11 @@ use crate::StateComputationTrait; /// } /// ``` #[cfg(not(feature = "single-threaded"))] -pub trait StateRepository: Send + Sync { +pub trait StateRepository: Send + Sync +where + C: IdempotencyKey + Send + Sync, + S: Send + Sync, +{ /// Error type for repository operations. /// /// This should capture both infrastructure errors (fetch/save failures) and domain @@ -180,17 +201,33 @@ pub trait StateRepository: Send + Sync { /// persisted. The exact mechanism depends on the underlying storage (transactions, /// optimistic locking, etc.). /// + /// # Idempotency + /// + /// Implementations should use `command.idempotency_key()` to detect retried commands: + /// when a command with a previously seen key is executed again, the original persisted + /// state should be returned without re-running the component. + /// /// # Example /// /// ```rust,no_run - /// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider}; + /// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider, IdempotencyKey}; /// # #[derive(Clone, Debug)] - /// # enum Command { SetValue(i32) } + /// # enum Command { SetValue { value: i32, idempotency_key: String } } + /// # impl IdempotencyKey for Command { + /// # fn idempotency_key(&self) -> &str { + /// # match self { + /// # Command::SetValue { idempotency_key, .. } => idempotency_key, + /// # } + /// # } + /// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { value: i32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] - /// # trait StateRepository: Send + Sync { + /// # trait StateRepository: Send + Sync + /// # where + /// # C: IdempotencyKey, + /// # { /// # type Error; /// # async fn execute(&self, command: C, component: &D) -> Result /// # where D: StateComputationTrait + Send + Sync; @@ -209,7 +246,7 @@ pub trait StateRepository: Send + Sync { /// # |s: &State, _e: &()| s.clone(), /// # || State::default(), /// # ); - /// let command = Command::SetValue(42); + /// let command = Command::SetValue { value: 42, idempotency_key: "req-1".to_string() }; /// let state = repository.execute(command, &component).await?; /// # Ok(()) /// # } @@ -239,6 +276,12 @@ pub trait StateRepository: Send + Sync { /// succeed and the resulting state is persisted, or none are. The exact mechanism depends /// on the underlying storage (transactions, optimistic locking, etc.). /// + /// # Idempotency + /// + /// As with [`execute`](Self::execute), implementations should use each command's + /// `idempotency_key()` to detect and skip re-running previously handled commands within + /// the batch. + /// /// # Empty batch /// /// An empty `commands` vec has no computed result. Implementations should decide on a @@ -295,10 +338,17 @@ pub trait StateRepository: Send + Sync { /// use std::rc::Rc; /// use std::cell::RefCell; /// use std::collections::HashMap; -/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider, IdempotencyKey}; /// /// # #[derive(Clone, Debug)] -/// # enum Command { Increment } +/// # enum Command { Increment { idempotency_key: String } } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Increment { idempotency_key } => idempotency_key, +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { count: i32 } /// @@ -307,7 +357,10 @@ pub trait StateRepository: Send + Sync { /// states: Rc>>, /// } /// -/// # trait StateRepository { +/// # trait StateRepository +/// # where +/// # C: IdempotencyKey, +/// # { /// # type Error; /// # async fn execute(&self, command: C, component: &D) -> Result /// # where D: StateComputationTrait; @@ -331,7 +384,10 @@ pub trait StateRepository: Send + Sync { /// # } /// ``` #[cfg(feature = "single-threaded")] -pub trait StateRepository { +pub trait StateRepository +where + C: IdempotencyKey, +{ /// Error type for repository operations. type Error; diff --git a/src/application/state_stored_command_handler.rs b/src/application/state_stored_command_handler.rs index 863d4b3..2f8e02a 100644 --- a/src/application/state_stored_command_handler.rs +++ b/src/application/state_stored_command_handler.rs @@ -1,5 +1,5 @@ use super::state_repository::StateRepository; -use crate::StateComputationTrait; +use crate::{IdempotencyKey, StateComputationTrait}; // ================================================================================================ // StateStoredCommandHandler - Convenience Layer @@ -13,7 +13,7 @@ use crate::StateComputationTrait; /// /// # Type Parameters /// -/// - `C`: Command type that triggers state changes +/// - `C`: Command type that triggers state changes. Must implement [`IdempotencyKey`]. /// - `S`: State type (both current and new state) /// - `D`: The decider implementing `StateComputationTrait` /// - `R`: The repository implementing `StateRepository` @@ -29,14 +29,28 @@ use crate::StateComputationTrait; /// **Without handler (direct repository usage):** /// ```rust,no_run /// # use std::sync::Arc; -/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider, IdempotencyKey}; /// # #[derive(Clone, Debug)] -/// # enum Command { Increment, Decrement } +/// # enum Command { +/// # Increment { idempotency_key: String }, +/// # Decrement { idempotency_key: String }, +/// # } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Increment { idempotency_key } => idempotency_key, +/// # Command::Decrement { idempotency_key } => idempotency_key, +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { count: i32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] -/// # trait StateRepository: Send + Sync { +/// # trait StateRepository: Send + Sync +/// # where +/// # C: IdempotencyKey, +/// # { /// # type Error; /// # async fn execute(&self, command: C, component: &D) -> Result /// # where D: StateComputationTrait + Send + Sync; @@ -56,9 +70,9 @@ use crate::StateComputationTrait; /// # ); /// # let repository = Arc::new(MyRepository); /// // Must pass component on every call -/// let state1 = repository.execute(Command::Increment, &component).await?; -/// let state2 = repository.execute(Command::Increment, &component).await?; -/// let state3 = repository.execute(Command::Decrement, &component).await?; +/// let state1 = repository.execute(Command::Increment { idempotency_key: "req-1".to_string() }, &component).await?; +/// let state2 = repository.execute(Command::Increment { idempotency_key: "req-2".to_string() }, &component).await?; +/// let state3 = repository.execute(Command::Decrement { idempotency_key: "req-3".to_string() }, &component).await?; /// # Ok(()) /// # } /// ``` @@ -66,14 +80,28 @@ use crate::StateComputationTrait; /// **With handler:** /// ```rust,no_run /// # use std::sync::Arc; -/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider, IdempotencyKey}; /// # #[derive(Clone, Debug)] -/// # enum Command { Increment, Decrement } +/// # enum Command { +/// # Increment { idempotency_key: String }, +/// # Decrement { idempotency_key: String }, +/// # } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Increment { idempotency_key } => idempotency_key, +/// # Command::Decrement { idempotency_key } => idempotency_key, +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { count: i32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] -/// # trait StateRepository: Send + Sync { +/// # trait StateRepository: Send + Sync +/// # where +/// # C: IdempotencyKey, +/// # { /// # type Error; /// # async fn execute(&self, command: C, component: &D) -> Result /// # where D: StateComputationTrait + Send + Sync; @@ -89,6 +117,7 @@ use crate::StateComputationTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # struct StateStoredCommandHandler /// # where +/// # C: IdempotencyKey, /// # D: StateComputationTrait + Send + Sync, /// # R: StateRepository + Send + Sync, /// # { @@ -99,6 +128,7 @@ use crate::StateComputationTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # impl StateStoredCommandHandler /// # where +/// # C: IdempotencyKey, /// # D: StateComputationTrait + Send + Sync, /// # R: StateRepository + Send + Sync, /// # { @@ -118,9 +148,9 @@ use crate::StateComputationTrait; /// # let repository = Arc::new(MyRepository); /// // Component encapsulated in handler /// let handler = StateStoredCommandHandler::new(component, repository); -/// let state1 = handler.handle(Command::Increment).await?; -/// let state2 = handler.handle(Command::Increment).await?; -/// let state3 = handler.handle(Command::Decrement).await?; +/// let state1 = handler.handle(Command::Increment { idempotency_key: "req-1".to_string() }).await?; +/// let state2 = handler.handle(Command::Increment { idempotency_key: "req-2".to_string() }).await?; +/// let state3 = handler.handle(Command::Decrement { idempotency_key: "req-3".to_string() }).await?; /// # Ok(()) /// # } /// ``` @@ -144,14 +174,30 @@ use crate::StateComputationTrait; /// /// ```rust,no_run /// use std::sync::Arc; -/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider, IdempotencyKey}; /// # #[derive(Clone, Debug)] -/// # enum Command { Increment, Decrement, Reset } +/// # enum Command { +/// # Increment { idempotency_key: String }, +/// # Decrement { idempotency_key: String }, +/// # Reset { idempotency_key: String }, +/// # } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Increment { idempotency_key } => idempotency_key, +/// # Command::Decrement { idempotency_key } => idempotency_key, +/// # Command::Reset { idempotency_key } => idempotency_key, +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { count: i32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] -/// # trait StateRepository: Send + Sync { +/// # trait StateRepository: Send + Sync +/// # where +/// # C: IdempotencyKey, +/// # { /// # type Error; /// # async fn execute(&self, command: C, component: &D) -> Result /// # where D: StateComputationTrait + Send + Sync; @@ -167,6 +213,7 @@ use crate::StateComputationTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # struct StateStoredCommandHandler /// # where +/// # C: IdempotencyKey, /// # D: StateComputationTrait + Send + Sync, /// # R: StateRepository + Send + Sync, /// # { @@ -177,6 +224,7 @@ use crate::StateComputationTrait; /// # #[cfg(not(feature = "single-threaded"))] /// # impl StateStoredCommandHandler /// # where +/// # C: IdempotencyKey, /// # D: StateComputationTrait + Send + Sync, /// # R: StateRepository + Send + Sync, /// # { @@ -193,9 +241,9 @@ use crate::StateComputationTrait; /// let component = Arc::new(AggregateDecider::new( /// |c: &Command, _s: &State| -> Result, String> { /// match c { -/// Command::Increment => Ok(vec![()]), -/// Command::Decrement => Ok(vec![()]), -/// Command::Reset => Ok(vec![()]), +/// Command::Increment { .. } => Ok(vec![()]), +/// Command::Decrement { .. } => Ok(vec![()]), +/// Command::Reset { .. } => Ok(vec![()]), /// } /// }, /// |s: &State, _e: &()| { @@ -211,15 +259,17 @@ use crate::StateComputationTrait; /// let handler = StateStoredCommandHandler::new(component, repository); /// /// // Handle commands without passing component each time -/// let state = handler.handle(Command::Increment).await?; -/// let state = handler.handle(Command::Increment).await?; -/// let state = handler.handle(Command::Decrement).await?; +/// let state = handler.handle(Command::Increment { idempotency_key: "req-1".to_string() }).await?; +/// let state = handler.handle(Command::Increment { idempotency_key: "req-2".to_string() }).await?; +/// let state = handler.handle(Command::Decrement { idempotency_key: "req-3".to_string() }).await?; /// # Ok(()) /// # } /// ``` #[cfg(not(feature = "single-threaded"))] pub struct StateStoredCommandHandler where + C: IdempotencyKey + Send + Sync, + S: Send + Sync, D: StateComputationTrait + Send + Sync, R: StateRepository + Send + Sync, { @@ -231,6 +281,8 @@ where #[cfg(not(feature = "single-threaded"))] impl StateStoredCommandHandler where + C: IdempotencyKey + Send + Sync, + S: Send + Sync, D: StateComputationTrait + Send + Sync, R: StateRepository + Send + Sync, { @@ -249,14 +301,24 @@ where /// /// ```rust,no_run /// use std::sync::Arc; - /// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider}; + /// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider, IdempotencyKey}; /// # #[derive(Clone, Debug)] - /// # enum Command { Increment } + /// # enum Command { Increment { idempotency_key: String } } + /// # impl IdempotencyKey for Command { + /// # fn idempotency_key(&self) -> &str { + /// # match self { + /// # Command::Increment { idempotency_key } => idempotency_key, + /// # } + /// # } + /// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { count: i32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] - /// # trait StateRepository: Send + Sync { + /// # trait StateRepository: Send + Sync + /// # where + /// # C: IdempotencyKey, + /// # { /// # type Error; /// # async fn execute(&self, command: C, component: &D) -> Result /// # where D: StateComputationTrait + Send + Sync; @@ -272,6 +334,7 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # struct StateStoredCommandHandler /// # where + /// # C: IdempotencyKey, /// # D: StateComputationTrait + Send + Sync, /// # R: StateRepository + Send + Sync, /// # { @@ -282,6 +345,7 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # impl StateStoredCommandHandler /// # where + /// # C: IdempotencyKey, /// # D: StateComputationTrait + Send + Sync, /// # R: StateRepository + Send + Sync, /// # { @@ -325,14 +389,28 @@ where /// /// ```rust,no_run /// # use std::sync::Arc; - /// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider}; + /// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider, IdempotencyKey}; /// # #[derive(Clone, Debug)] - /// # enum Command { Increment, Decrement } + /// # enum Command { + /// # Increment { idempotency_key: String }, + /// # Decrement { idempotency_key: String }, + /// # } + /// # impl IdempotencyKey for Command { + /// # fn idempotency_key(&self) -> &str { + /// # match self { + /// # Command::Increment { idempotency_key } => idempotency_key, + /// # Command::Decrement { idempotency_key } => idempotency_key, + /// # } + /// # } + /// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { count: i32 } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] - /// # trait StateRepository: Send + Sync { + /// # trait StateRepository: Send + Sync + /// # where + /// # C: IdempotencyKey, + /// # { /// # type Error; /// # async fn execute(&self, command: C, component: &D) -> Result /// # where D: StateComputationTrait + Send + Sync; @@ -348,6 +426,7 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # struct StateStoredCommandHandler /// # where + /// # C: IdempotencyKey, /// # D: StateComputationTrait + Send + Sync, /// # R: StateRepository + Send + Sync, /// # { @@ -358,6 +437,7 @@ where /// # #[cfg(not(feature = "single-threaded"))] /// # impl StateStoredCommandHandler /// # where + /// # C: IdempotencyKey, /// # D: StateComputationTrait + Send + Sync, /// # R: StateRepository + Send + Sync, /// # { @@ -377,9 +457,9 @@ where /// # let repository = Arc::new(MyRepository); /// # let handler = StateStoredCommandHandler::new(component, repository); /// // Handle multiple commands - /// let state1 = handler.handle(Command::Increment).await?; - /// let state2 = handler.handle(Command::Increment).await?; - /// let state3 = handler.handle(Command::Decrement).await?; + /// let state1 = handler.handle(Command::Increment { idempotency_key: "req-1".to_string() }).await?; + /// let state2 = handler.handle(Command::Increment { idempotency_key: "req-2".to_string() }).await?; + /// let state3 = handler.handle(Command::Decrement { idempotency_key: "req-3".to_string() }).await?; /// # Ok(()) /// # } /// ``` @@ -433,13 +513,23 @@ where /// # #[cfg(feature = "single-threaded")] /// # { /// use std::rc::Rc; -/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider}; +/// # use fmodel_decider_rust::{StateComputationTrait, AggregateDecider, IdempotencyKey}; /// # #[derive(Clone, Debug)] -/// # enum Command { Increment } +/// # enum Command { Increment { idempotency_key: String } } +/// # impl IdempotencyKey for Command { +/// # fn idempotency_key(&self) -> &str { +/// # match self { +/// # Command::Increment { idempotency_key } => idempotency_key, +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { count: i32 } /// # struct MyRepository; -/// # trait StateRepository { +/// # trait StateRepository +/// # where +/// # C: IdempotencyKey, +/// # { /// # type Error; /// # async fn execute(&self, command: C, component: &D) -> Result /// # where D: StateComputationTrait; @@ -453,6 +543,7 @@ where /// # use std::marker::PhantomData; /// # struct StateStoredCommandHandler /// # where +/// # C: IdempotencyKey, /// # D: StateComputationTrait, /// # R: StateRepository, /// # { @@ -462,6 +553,7 @@ where /// # } /// # impl StateStoredCommandHandler /// # where +/// # C: IdempotencyKey, /// # D: StateComputationTrait, /// # R: StateRepository, /// # { @@ -487,6 +579,7 @@ where #[cfg(feature = "single-threaded")] pub struct StateStoredCommandHandler where + C: IdempotencyKey, D: StateComputationTrait, R: StateRepository, { @@ -498,6 +591,7 @@ where #[cfg(feature = "single-threaded")] impl StateStoredCommandHandler where + C: IdempotencyKey, D: StateComputationTrait, R: StateRepository, { diff --git a/src/application/view_repository.rs b/src/application/view_repository.rs index 4a776bb..af90c53 100644 --- a/src/application/view_repository.rs +++ b/src/application/view_repository.rs @@ -1,4 +1,4 @@ -use crate::ViewTrait; +use crate::{EventMeta, ViewTrait}; // ================================================================================================ // ViewRepository Trait @@ -12,8 +12,11 @@ use crate::ViewTrait; /// /// # Type Parameters /// -/// - `E`: Event type that triggers view updates -/// - `S`: State type (both current and evolved state) +/// - `E`: Event type that triggers view updates. Must implement [`EventMeta`] so repository +/// implementations can build secondary indexes. In multi-threaded mode, must also be +/// `Send + Sync` since it is moved into the `Send` future returned by `execute`. +/// - `S`: State type (both current and evolved state). In multi-threaded mode, must also be +/// `Send + Sync`. /// /// # Associated Types /// @@ -52,10 +55,24 @@ use crate::ViewTrait; /// ```rust,no_run /// use std::collections::HashMap; /// use std::sync::{Arc, Mutex}; -/// # use fmodel_decider_rust::{ViewTrait, Projection}; +/// # use fmodel_decider_rust::{ViewTrait, Projection, EventMeta, Tag}; /// /// # #[derive(Clone, Debug)] /// # enum Event { AccountOpened { id: String }, MoneyDeposited { id: String, amount: u32 } } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::AccountOpened { .. } => "AccountOpened", +/// # Event::MoneyDeposited { .. } => "MoneyDeposited", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # match self { +/// # Event::AccountOpened { id } => vec![Tag::new("id", id.clone())], +/// # Event::MoneyDeposited { id, .. } => vec![Tag::new("id", id.clone())], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { balance: u32 } /// @@ -65,7 +82,10 @@ use crate::ViewTrait; /// } /// /// # #[cfg(not(feature = "single-threaded"))] -/// # trait ViewRepository: Send + Sync { +/// # trait ViewRepository: Send + Sync +/// # where +/// # E: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, event: E, view: &V) -> Result /// # where V: ViewTrait + Send + Sync; @@ -152,7 +172,11 @@ use crate::ViewTrait; /// } /// ``` #[cfg(not(feature = "single-threaded"))] -pub trait ViewRepository: Send + Sync { +pub trait ViewRepository: Send + Sync +where + E: EventMeta + Send + Sync, + S: Send + Sync, +{ /// Error type for repository operations. /// /// This should capture both infrastructure errors (fetch/save failures) and view @@ -191,14 +215,29 @@ pub trait ViewRepository: Send + Sync { /// # Example /// /// ```rust,no_run - /// # use fmodel_decider_rust::{ViewTrait, Projection}; + /// # use fmodel_decider_rust::{ViewTrait, Projection, EventMeta, Tag}; /// # #[derive(Clone, Debug)] /// # enum Event { ItemAdded(String) } + /// # impl EventMeta for Event { + /// # fn event_type(&self) -> &str { + /// # match self { + /// # Event::ItemAdded(_) => "ItemAdded", + /// # } + /// # } + /// # fn tags(&self) -> Vec { + /// # match self { + /// # Event::ItemAdded(item) => vec![Tag::new("item", item.clone())], + /// # } + /// # } + /// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { items: Vec } /// # struct MyRepository; /// # #[cfg(not(feature = "single-threaded"))] - /// # trait ViewRepository: Send + Sync { + /// # trait ViewRepository: Send + Sync + /// # where + /// # E: EventMeta, + /// # { /// # type Error; /// # async fn execute(&self, event: E, view: &V) -> Result /// # where V: ViewTrait + Send + Sync; @@ -304,10 +343,22 @@ pub trait ViewRepository: Send + Sync { /// use std::rc::Rc; /// use std::cell::RefCell; /// use std::collections::HashMap; -/// # use fmodel_decider_rust::{ViewTrait, Projection}; +/// # use fmodel_decider_rust::{ViewTrait, Projection, EventMeta, Tag}; /// /// # #[derive(Clone, Debug)] /// # enum Event { ItemAdded(String) } +/// # impl EventMeta for Event { +/// # fn event_type(&self) -> &str { +/// # match self { +/// # Event::ItemAdded(_) => "ItemAdded", +/// # } +/// # } +/// # fn tags(&self) -> Vec { +/// # match self { +/// # Event::ItemAdded(item) => vec![Tag::new("item", item.clone())], +/// # } +/// # } +/// # } /// # #[derive(Clone, Debug, Default)] /// # struct State { items: Vec } /// @@ -316,7 +367,10 @@ pub trait ViewRepository: Send + Sync { /// views: Rc>>, /// } /// -/// # trait ViewRepository { +/// # trait ViewRepository +/// # where +/// # E: EventMeta, +/// # { /// # type Error; /// # async fn execute(&self, event: E, view: &V) -> Result /// # where V: ViewTrait; @@ -340,7 +394,10 @@ pub trait ViewRepository: Send + Sync { /// # } /// ``` #[cfg(feature = "single-threaded")] -pub trait ViewRepository { +pub trait ViewRepository +where + E: EventMeta, +{ /// Error type for repository operations. type Error; diff --git a/src/lib.rs b/src/lib.rs index d878b86..0aa68ba 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -183,8 +183,7 @@ mod workflow; pub use application::{ EventLoader, EventRepository, EventSourcedCommandHandler, EventSourcedQueryHandler, - MaterializedViewHandler, QueryTuple, StateRepository, StateStoredCommandHandler, - ViewRepository, + MaterializedViewHandler, StateRepository, StateStoredCommandHandler, ViewRepository, }; pub use decider::AggregateDecider; pub use dynamic_decider::DCBDecider; @@ -541,6 +540,48 @@ pub trait ProcessTrait: DeciderTrait fn pending(&self, state: &Si) -> Self::Actions; } +/// A single `key:value` tag attached to an event, used for secondary indexing. +/// +/// Encapsulates the `"key:value"` string convention structurally instead of leaving +/// callers to hand-format strings (e.g. `format!("id:{}", id)`), which removes the risk +/// of malformed tags: a missing separator, or a value that itself contains a colon and +/// breaks naive parsing downstream. +/// +/// `Tag` implements [`std::fmt::Display`], producing the canonical `"key:value"` wire +/// format that repository implementations index on. +/// +/// # Example +/// +/// ```rust +/// use fmodel_decider_rust::Tag; +/// +/// let tag = Tag::new("restaurantId", "123"); +/// assert_eq!(tag.to_string(), "restaurantId:123"); +/// ``` +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct Tag { + /// The tag's key (e.g. `"restaurantId"`). + pub key: String, + /// The tag's value (e.g. `"123"`). + pub value: String, +} + +impl Tag { + /// Creates a new tag from a key and value. + pub fn new(key: impl Into, value: impl Into) -> Self { + Self { + key: key.into(), + value: value.into(), + } + } +} + +impl std::fmt::Display for Tag { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}:{}", self.key, self.value) + } +} + /// Trait for events that carry metadata for indexing and querying. /// /// Used by repository implementations (e.g., FoundationDB) to build secondary @@ -556,7 +597,7 @@ pub trait ProcessTrait: DeciderTrait /// /// ## Tags /// -/// Tags are `"key:value"` strings extracted from the event's own fields. They serve +/// Tags are [`Tag`] values extracted from the event's own fields. They serve /// as the basis for secondary indexing and stream identification. Repository /// implementations use the power set of an event's tags to build indexes that /// support queries by any tag combination. @@ -564,7 +605,7 @@ pub trait ProcessTrait: DeciderTrait /// ## Example /// /// ```rust -/// use fmodel_decider_rust::EventMeta; +/// use fmodel_decider_rust::{EventMeta, Tag}; /// /// struct RestaurantCreatedEvent { /// restaurant_id: String, @@ -575,8 +616,8 @@ pub trait ProcessTrait: DeciderTrait /// "RestaurantCreatedEvent" /// } /// -/// fn tags(&self) -> Vec { -/// vec![format!("restaurantId:{}", self.restaurant_id)] +/// fn tags(&self) -> Vec { +/// vec![Tag::new("restaurantId", self.restaurant_id.clone())] /// } /// } /// ``` @@ -587,12 +628,31 @@ pub trait EventMeta { /// forming the first element of tag index and last event pointer keys. fn event_type(&self) -> &str; - /// Returns tags extracted from the event's fields in `"key:value"` format. + /// Returns tags extracted from the event's fields. /// /// Tags are used for secondary indexing and stream identification. Repository /// implementations build power set indexes over these tags to support queries /// by any combination. - fn tags(&self) -> Vec; + fn tags(&self) -> Vec; +} + +/// A query specification for fetching events from the event store. +/// +/// Consists of an event type and zero or more tags that identify the event stream to query. +/// This is the core crate's dependency-free equivalent of the TypeScript `QueryTuple` type +/// from fmodel-ts: `[...tags, eventType]`. +/// +/// # Fields +/// +/// - `event_type`: The event type identifier (e.g., `"RestaurantCreatedEvent"`) +/// - `tags`: Tags to filter by. Empty vec queries all events of the type. These are matched +/// against the tags an event exposes via [`EventMeta::tags`]. +#[derive(Debug, Clone)] +pub struct QueryTuple { + /// The event type identifier to query. + pub event_type: String, + /// Tags to filter by. + pub tags: Vec, } /// Implemented by commands that carry an idempotency key. @@ -604,3 +664,35 @@ pub trait IdempotencyKey { /// Returns the idempotency key for this command. fn idempotency_key(&self) -> &str; } + +/// Implemented by commands that declare which events are relevant to deciding them. +/// +/// Used by [`EventRepository`](crate::EventRepository) implementations backed by a +/// Dynamic Consistency Boundary (DCB) event store: instead of fetching a single fixed +/// stream, the repository loads exactly the events matching the query tuples this method +/// returns, then folds them through the decider before computing new events. This mirrors +/// how [`EventMeta::tags`] lets events declare their own indexing tags — here the command +/// declares which of those tags identify the events it needs to see. +/// +/// # Example +/// +/// ```rust +/// use fmodel_decider_rust::{CommandQueries, QueryTuple, Tag}; +/// +/// struct DepositCommand { +/// account_id: String, +/// } +/// +/// impl CommandQueries for DepositCommand { +/// fn queries(&self) -> Vec { +/// vec![QueryTuple { +/// event_type: "AccountEvent".to_string(), +/// tags: vec![Tag::new("accountId", self.account_id.clone())], +/// }] +/// } +/// } +/// ``` +pub trait CommandQueries { + /// Returns the query tuples identifying events relevant to deciding this command. + fn queries(&self) -> Vec; +}