myko/core/saga/
traits.rs

1//! Core saga traits and types for reactive event processing.
2//!
3//! Sagas are stream processors that react to events and emit typed commands.
4
5use std::sync::Arc;
6
7use futures::StreamExt;
8
9use crate::{
10    command::{CommandContext, CommandError, CommandHandler, CommandIdStatic},
11    event::{MEvent, MEventType},
12    item::Eventable,
13};
14
15/// Type alias for the event stream input to sagas.
16pub type EventStream = futures::stream::BoxStream<'static, MEvent>;
17
18/// Type alias for type-erased saga commands.
19pub type CommandStream = futures::stream::BoxStream<'static, Box<dyn SagaCommand>>;
20
21/// Type-erased executable command produced by a saga.
22pub trait SagaCommand: Send + Sync + 'static {
23    fn execute_boxed(self: Box<Self>, ctx: CommandContext) -> Result<(), CommandError>;
24    fn command_name(&self) -> &'static str;
25}
26
27impl<C> SagaCommand for C
28where
29    C: CommandHandler + CommandIdStatic + Send + Sync + 'static,
30{
31    fn execute_boxed(self: Box<Self>, ctx: CommandContext) -> Result<(), CommandError> {
32        let _ = (*self).execute(ctx)?;
33        Ok(())
34    }
35
36    fn command_name(&self) -> &'static str {
37        C::COMMAND_ID
38    }
39}
40
41/// Trait for saga handlers that process events and emit commands.
42pub trait Saga: Send + Sync + 'static {
43    /// Accumulated state type for this saga. Use `()` for stateless sagas.
44    type State: Default + Send + Sync + 'static;
45
46    /// Typed command emitted by this saga.
47    type Command: SagaCommand;
48
49    /// Human-readable name for logging and debugging.
50    fn name() -> &'static str
51    where
52        Self: Sized;
53
54    /// Build the event processing pipeline.
55    fn build(
56        events: EventStream,
57        ctx: Arc<super::context::SagaContext>,
58    ) -> futures::stream::BoxStream<'static, Self::Command>
59    where
60        Self: Sized;
61}
62
63/// Ergonomic handler trait for macro-declared sagas.
64///
65/// `#[myko_saga(...)]` can generate `Saga` + registration boilerplate while this trait
66/// contains the typed business logic.
67pub trait SagaHandler: Send + Sync + 'static {
68    /// Event item type consumed by this saga.
69    type EventItem: Eventable;
70    /// Command type emitted by this saga.
71    type Command: SagaCommand;
72    /// Event change type consumed by this saga.
73    const EVENT_TYPE: MEventType;
74    /// Handle one matching event and optionally emit a command.
75    fn handle(
76        item: Self::EventItem,
77        event: MEvent,
78        ctx: Arc<super::context::SagaContext>,
79    ) -> Option<Self::Command>;
80}
81
82/// Type-erased saga trait for dynamic dispatch.
83pub trait AnySaga: Send + Sync + 'static {
84    /// Get the saga's name.
85    fn name(&self) -> &'static str;
86
87    /// Build the event processing pipeline.
88    fn build_boxed(
89        &self,
90        events: EventStream,
91        ctx: Arc<super::context::SagaContext>,
92    ) -> CommandStream;
93
94    /// Create the initial state for this saga.
95    fn create_state(&self) -> Box<dyn std::any::Any + Send + Sync>;
96}
97
98impl<S: Saga> AnySaga for S {
99    fn name(&self) -> &'static str {
100        S::name()
101    }
102
103    fn build_boxed(
104        &self,
105        events: EventStream,
106        ctx: Arc<super::context::SagaContext>,
107    ) -> CommandStream {
108        Box::pin(S::build(events, ctx).map(|cmd| Box::new(cmd) as Box<dyn SagaCommand>))
109    }
110
111    fn create_state(&self) -> Box<dyn std::any::Any + Send + Sync> {
112        Box::new(S::State::default())
113    }
114}
115
116/// Registration entry for saga discovery via inventory.
117pub struct SagaRegistration {
118    /// Unique identifier for this saga.
119    pub saga_id: &'static str,
120    /// Factory function to create an instance of the saga.
121    pub create: fn() -> Arc<dyn AnySaga>,
122    /// Entity type this saga listens for (for dispatch filtering).
123    pub event_entity_type: &'static str,
124    /// Change type this saga listens for (for dispatch filtering).
125    pub event_change_type: MEventType,
126}
127
128inventory::collect!(SagaRegistration);
myko/core/saga/traits.rs

myko/core/saga/
traits.rs
