weavatrix_memory/projection/memory/
apply.rs

1use super::{MemoryProjection, state::NodeRevision};
2use crate::{
3    domain::MemoryEvent,
4    error::{MemoryError, Result},
5    event::StoredEvent,
6    projection::Projection,
7};
8
9impl Projection<MemoryEvent> for MemoryProjection {
10    fn prepare_replay(&mut self, events: &[StoredEvent<MemoryEvent>]) {
11        let mut nodes = 0;
12        let mut facts = 0;
13        for event in events {
14            match event.payload {
15                MemoryEvent::NodeUpserted { .. } => nodes += 1,
16                MemoryEvent::FactRecorded { .. } => facts += 1,
17                MemoryEvent::FactRetracted { .. } => {}
18            }
19        }
20        self.nodes.reserve(nodes);
21        self.facts.reserve(facts);
22        self.node_lookup.reserve(nodes);
23        self.fact_lookup.reserve(facts);
24        self.incident_offsets.reserve(nodes);
25    }
26
27    fn apply(&mut self, event: &StoredEvent<MemoryEvent>) -> Result<()> {
28        if event.metadata.event_type != event.payload.event_type() {
29            return Err(MemoryError::InvalidValue {
30                field: "event_type",
31                reason: "must match the memory event payload",
32            });
33        }
34        match &event.payload {
35            MemoryEvent::NodeUpserted { node } => self.insert_node(NodeRevision {
36                node: node.clone(),
37                recorded_at: event.metadata.recorded_at,
38                position: event.metadata.global_position,
39            })?,
40            MemoryEvent::FactRecorded { fact } => {
41                if fact.recorded_at != event.metadata.recorded_at
42                    || fact.agent_id != event.metadata.agent_id
43                    || fact.session_id != event.metadata.session_id
44                {
45                    return Err(MemoryError::InvalidValue {
46                        field: "fact.envelope",
47                        reason: "fact provenance must match its event envelope",
48                    });
49                }
50                self.insert_fact(fact.clone())?;
51            }
52            MemoryEvent::FactRetracted {
53                fact_id,
54                valid_until,
55                evidence,
56            } => {
57                self.apply_retraction(
58                    event.metadata.recorded_at,
59                    event.metadata.global_position,
60                    fact_id,
61                    *valid_until,
62                    evidence,
63                )?;
64            }
65        }
66        self.last_global_position = Some(event.metadata.global_position);
67        Ok(())
68    }
69
70    fn apply_owned(&mut self, event: StoredEvent<MemoryEvent>) -> Result<()> {
71        let StoredEvent { metadata, payload } = event;
72        if metadata.event_type != payload.event_type() {
73            return Err(MemoryError::InvalidValue {
74                field: "event_type",
75                reason: "must match the memory event payload",
76            });
77        }
78        match payload {
79            MemoryEvent::NodeUpserted { node } => self.insert_node(NodeRevision {
80                node,
81                recorded_at: metadata.recorded_at,
82                position: metadata.global_position,
83            })?,
84            MemoryEvent::FactRecorded { fact } => {
85                if fact.recorded_at != metadata.recorded_at
86                    || fact.agent_id != metadata.agent_id
87                    || fact.session_id != metadata.session_id
88                {
89                    return Err(MemoryError::InvalidValue {
90                        field: "fact.envelope",
91                        reason: "fact provenance must match its event envelope",
92                    });
93                }
94                self.insert_fact(fact)?;
95            }
96            MemoryEvent::FactRetracted {
97                fact_id,
98                valid_until,
99                evidence,
100            } => {
101                self.apply_retraction(
102                    metadata.recorded_at,
103                    metadata.global_position,
104                    &fact_id,
105                    valid_until,
106                    &evidence,
107                )?;
108            }
109        }
110        self.last_global_position = Some(metadata.global_position);
111        Ok(())
112    }
113}
weavatrix_memory/projection/memory/apply.rs

weavatrix_memory/projection/memory/
apply.rs
