weavatrix-memory 0.2.0

Event-sourced, bitemporal context compiler for coding agents
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

use crate::{EventId, MemoryError, Result, StoredEvent, StreamId};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};

pub trait Projection<E>: Default {
    /// Applies one validated stored event.
    ///
    /// # Errors
    ///
    /// Returns domain-specific invariant violations.
    fn apply(&mut self, event: &StoredEvent<E>) -> Result<()>;
}

#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReplayCursor {
    pub global_position: Option<u64>,
    pub stream_versions: BTreeMap<StreamId, u64>,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProjectionSnapshot<P> {
    pub cursor: ReplayCursor,
    pub projection: P,
}

/// Replays a complete, zero-based event sequence after validating its cursors.
///
/// # Errors
///
/// Rejects global-position or per-stream-version gaps and projection errors.
pub fn replay<E, P>(events: &[StoredEvent<E>]) -> Result<P>
where
    P: Projection<E>,
{
    replay_tracked(events).map(|(projection, _)| projection)
}

/// Replays a complete sequence and returns the cursor needed for a snapshot.
///
/// # Errors
///
/// Rejects invalid event ordering, duplicate IDs, and projection errors.
pub fn replay_tracked<E, P>(events: &[StoredEvent<E>]) -> Result<(P, ReplayCursor)>
where
    P: Projection<E>,
{
    apply_sequence(P::default(), ReplayCursor::default(), events, true)
}

/// Resumes a materialized projection from its exact event cursor.
///
/// # Errors
///
/// Rejects gaps between the snapshot cursor and the supplied tail, invalid
/// stream versions, duplicate IDs within the tail, and projection errors.
pub fn resume<E, P>(
    snapshot: ProjectionSnapshot<P>,
    tail: &[StoredEvent<E>],
) -> Result<(P, ReplayCursor)>
where
    P: Projection<E>,
{
    apply_sequence(snapshot.projection, snapshot.cursor, tail, false)
}

fn apply_sequence<E, P>(
    mut projection: P,
    mut cursor: ReplayCursor,
    events: &[StoredEvent<E>],
    require_zero_start: bool,
) -> Result<(P, ReplayCursor)>
where
    P: Projection<E>,
{
    let mut event_ids = BTreeSet::<EventId>::new();
    for event in events {
        if !event_ids.insert(event.metadata.id.clone()) {
            return Err(MemoryError::DuplicateEvent {
                id: event.metadata.id.to_string(),
            });
        }
        let expected_position = cursor.global_position.map_or(Ok(0), |position| {
            position.checked_add(1).ok_or(MemoryError::CapacityOverflow)
        })?;
        if require_zero_start && cursor.global_position.is_none() && expected_position != 0 {
            return Err(MemoryError::InvalidReplay {
                reason: "complete replay must start at zero".to_owned(),
            });
        }
        if event.metadata.global_position != expected_position {
            return Err(MemoryError::InvalidReplay {
                reason: format!(
                    "global position {}, expected {expected_position}",
                    event.metadata.global_position
                ),
            });
        }
        let expected_version = cursor
            .stream_versions
            .get(&event.metadata.stream_id)
            .map_or(Ok(0), |version| {
                version.checked_add(1).ok_or(MemoryError::CapacityOverflow)
            })?;
        if event.metadata.stream_version != expected_version {
            return Err(MemoryError::InvalidReplay {
                reason: format!(
                    "stream {} version {}, expected {expected_version}",
                    event.metadata.stream_id, event.metadata.stream_version
                ),
            });
        }
        projection.apply(event)?;
        cursor
            .stream_versions
            .insert(event.metadata.stream_id.clone(), expected_version);
        cursor.global_position = Some(expected_position);
    }
    Ok((projection, cursor))
}