substrate_core/
event_store_port.rs

1//! EventStorePort — append-only per-aggregate event log with global ordering.
2//!
3//! Core defines the contract and replay/projection helpers; `store-sqlite`
4//! provides the durable SQLite implementation.
5
6use serde::{Deserialize, Serialize};
7use uuid::Uuid;
8
9use crate::domain::TaskState;
10
11/// A persisted domain event with monotonic per-aggregate and global sequence.
12#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
13pub struct EventEnvelope<E> {
14    /// Aggregate this event belongs to.
15    pub aggregate_id: Uuid,
16    /// Monotonic sequence within the aggregate (0-based).
17    pub aggregate_seq: u64,
18    /// Monotonic sequence across all aggregates.
19    pub global_seq: u64,
20    /// Domain event payload.
21    pub event: E,
22    /// Unix epoch seconds when the event was recorded.
23    pub occurred_at: i64,
24}
25
26/// Append-only event log port.
27///
28/// Implementations MUST guarantee:
29/// - events for an aggregate are returned in `aggregate_seq` order;
30/// - `global_seq` is strictly monotonic across all appends;
31/// - duplicate `(aggregate_id, aggregate_seq)` appends are rejected.
32pub trait EventStorePort: Send + Sync {
33    /// Error type returned by event-store operations.
34    type Error: std::error::Error + Send + Sync + 'static;
35    /// Serde-serializable domain event type stored by this port.
36    type Event: Serialize + for<'de> Deserialize<'de> + Clone + Send + Sync;
37
38    /// Append `event` to the log for `aggregate_id`.
39    ///
40    /// `expected_seq` is the sequence number this event will receive (equal to
41    /// the current event count for the aggregate). Mismatch rejects the append.
42    fn append(
43        &self,
44        aggregate_id: Uuid,
45        expected_seq: u64,
46        event: &Self::Event,
47    ) -> Result<EventEnvelope<Self::Event>, Self::Error>;
48
49    /// Load all events for `aggregate_id` in `aggregate_seq` order.
50    fn load(&self, aggregate_id: Uuid) -> Result<Vec<EventEnvelope<Self::Event>>, Self::Error>;
51}
52
53/// Fold an ordered event stream into aggregate state.
54pub trait Projection {
55    /// Rebuilt aggregate state.
56    type State;
57    /// Domain event type consumed by this projection.
58    type Event;
59
60    /// State before any events are applied.
61    fn initial() -> Self::State;
62
63    /// Apply a single event to `state`, returning the new state.
64    fn apply(state: Self::State, event: &Self::Event) -> Self::State;
65}
66
67/// Replay `events` through `P`, returning the folded state.
68pub fn replay<P: Projection>(events: &[P::Event]) -> P::State {
69    events
70        .iter()
71        .fold(P::initial(), |state, event| P::apply(state, event))
72}
73
74// ---------------------------------------------------------------------------
75// Task lifecycle demonstration projection
76// ---------------------------------------------------------------------------
77
78/// Task lifecycle events for event-sourced replay.
79#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
80#[serde(tag = "kind", rename_all = "snake_case")]
81pub enum TaskLifecycleEvent {
82    /// Task created with prompt and working directory.
83    Created {
84        /// Instruction handed to an engine.
85        prompt: String,
86        /// Working directory.
87        cwd: String,
88    },
89    /// Lifecycle transition enforced by the FSM at replay time.
90    Advanced {
91        /// Target lifecycle state.
92        to: TaskState,
93    },
94}
95
96/// State rebuilt from a [`TaskLifecycleEvent`] stream.
97#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
98pub struct TaskProjectionState {
99    /// Aggregate identity (set on first `Created`).
100    pub id: Option<Uuid>,
101    /// Task prompt.
102    pub prompt: String,
103    /// Working directory.
104    pub cwd: String,
105    /// Current lifecycle state.
106    pub state: TaskState,
107}
108
109/// Projection that folds [`TaskLifecycleEvent`] into [`TaskProjectionState`].
110pub struct TaskLifecycleProjection;
111
112impl Projection for TaskLifecycleProjection {
113    type State = TaskProjectionState;
114    type Event = TaskLifecycleEvent;
115
116    fn initial() -> Self::State {
117        TaskProjectionState {
118            id: None,
119            prompt: String::new(),
120            cwd: String::new(),
121            state: TaskState::Submitted,
122        }
123    }
124
125    fn apply(mut state: Self::State, event: &Self::Event) -> Self::State {
126        match event {
127            TaskLifecycleEvent::Created { prompt, cwd } => {
128                state.prompt = prompt.clone();
129                state.cwd = cwd.clone();
130                state.state = TaskState::Submitted;
131            }
132            TaskLifecycleEvent::Advanced { to } => {
133                if TaskState::can_transition(state.state, *to) {
134                    state.state = *to;
135                }
136            }
137        }
138        state
139    }
140}
141
142/// Replay task lifecycle events for `aggregate_id` from `store`.
143pub fn replay_task_state<S>(store: &S, aggregate_id: Uuid) -> Result<TaskProjectionState, S::Error>
144where
145    S: EventStorePort<Event = TaskLifecycleEvent>,
146{
147    let envelopes = store.load(aggregate_id)?;
148    let events: Vec<TaskLifecycleEvent> = envelopes.into_iter().map(|e| e.event).collect();
149    Ok(replay::<TaskLifecycleProjection>(&events))
150}
substrate_core/event_store_port.rs

substrate_core/
event_store_port.rs
