weavatrix_memory/projection/
replay.rs

1use crate::{EventId, MemoryError, Result, StoredEvent, StreamId};
2use serde::{Deserialize, Serialize};
3use std::collections::{BTreeMap, BTreeSet};
4
5pub trait Projection<E>: Default {
6    /// Applies one validated stored event.
7    ///
8    /// # Errors
9    ///
10    /// Returns domain-specific invariant violations.
11    fn apply(&mut self, event: &StoredEvent<E>) -> Result<()>;
12}
13
14#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
15pub struct ReplayCursor {
16    pub global_position: Option<u64>,
17    pub stream_versions: BTreeMap<StreamId, u64>,
18}
19
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub struct ProjectionSnapshot<P> {
22    pub cursor: ReplayCursor,
23    pub projection: P,
24}
25
26/// Replays a complete, zero-based event sequence after validating its cursors.
27///
28/// # Errors
29///
30/// Rejects global-position or per-stream-version gaps and projection errors.
31pub fn replay<E, P>(events: &[StoredEvent<E>]) -> Result<P>
32where
33    P: Projection<E>,
34{
35    replay_tracked(events).map(|(projection, _)| projection)
36}
37
38/// Replays a complete sequence and returns the cursor needed for a snapshot.
39///
40/// # Errors
41///
42/// Rejects invalid event ordering, duplicate IDs, and projection errors.
43pub fn replay_tracked<E, P>(events: &[StoredEvent<E>]) -> Result<(P, ReplayCursor)>
44where
45    P: Projection<E>,
46{
47    apply_sequence(P::default(), ReplayCursor::default(), events, true)
48}
49
50/// Resumes a materialized projection from its exact event cursor.
51///
52/// # Errors
53///
54/// Rejects gaps between the snapshot cursor and the supplied tail, invalid
55/// stream versions, duplicate IDs within the tail, and projection errors.
56pub fn resume<E, P>(
57    snapshot: ProjectionSnapshot<P>,
58    tail: &[StoredEvent<E>],
59) -> Result<(P, ReplayCursor)>
60where
61    P: Projection<E>,
62{
63    apply_sequence(snapshot.projection, snapshot.cursor, tail, false)
64}
65
66fn apply_sequence<E, P>(
67    mut projection: P,
68    mut cursor: ReplayCursor,
69    events: &[StoredEvent<E>],
70    require_zero_start: bool,
71) -> Result<(P, ReplayCursor)>
72where
73    P: Projection<E>,
74{
75    let mut event_ids = BTreeSet::<EventId>::new();
76    for event in events {
77        if !event_ids.insert(event.metadata.id.clone()) {
78            return Err(MemoryError::DuplicateEvent {
79                id: event.metadata.id.to_string(),
80            });
81        }
82        let expected_position = cursor.global_position.map_or(Ok(0), |position| {
83            position.checked_add(1).ok_or(MemoryError::CapacityOverflow)
84        })?;
85        if require_zero_start && cursor.global_position.is_none() && expected_position != 0 {
86            return Err(MemoryError::InvalidReplay {
87                reason: "complete replay must start at zero".to_owned(),
88            });
89        }
90        if event.metadata.global_position != expected_position {
91            return Err(MemoryError::InvalidReplay {
92                reason: format!(
93                    "global position {}, expected {expected_position}",
94                    event.metadata.global_position
95                ),
96            });
97        }
98        let expected_version = cursor
99            .stream_versions
100            .get(&event.metadata.stream_id)
101            .map_or(Ok(0), |version| {
102                version.checked_add(1).ok_or(MemoryError::CapacityOverflow)
103            })?;
104        if event.metadata.stream_version != expected_version {
105            return Err(MemoryError::InvalidReplay {
106                reason: format!(
107                    "stream {} version {}, expected {expected_version}",
108                    event.metadata.stream_id, event.metadata.stream_version
109                ),
110            });
111        }
112        projection.apply(event)?;
113        cursor
114            .stream_versions
115            .insert(event.metadata.stream_id.clone(), expected_version);
116        cursor.global_position = Some(expected_position);
117    }
118    Ok((projection, cursor))
119}
weavatrix_memory/projection/replay.rs

weavatrix_memory/projection/
replay.rs
