weavatrix_memory/projection/memory/
state.rs

1use super::index::IdIndex;
2use crate::{
3    domain::{MemoryFact, MemoryNode},
4    error::{MemoryError, Result},
5    id::{EntityId, FactId},
6    time::Timestamp,
7};
8use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
9use std::collections::{BTreeMap, HashMap};
10
11#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
12pub(crate) struct NodeRevision {
13    pub(crate) node: MemoryNode,
14    pub(crate) recorded_at: Timestamp,
15    pub(crate) position: u64,
16}
17
18#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
19pub(crate) struct NodeHistory {
20    pub(crate) first: NodeRevision,
21    pub(crate) later: Vec<NodeRevision>,
22}
23
24impl NodeHistory {
25    pub(super) const fn new(first: NodeRevision) -> Self {
26        Self {
27            first,
28            later: Vec::new(),
29        }
30    }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
34pub(crate) struct Supersession {
35    pub(crate) replacement: FactId,
36    pub(crate) valid_from: Timestamp,
37    pub(crate) recorded_at: Timestamp,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41pub(crate) struct Retraction {
42    pub(crate) valid_until: Timestamp,
43    pub(crate) recorded_at: Timestamp,
44}
45
46#[derive(Debug, Clone)]
47pub struct MemoryProjection {
48    pub(crate) nodes: Vec<NodeHistory>,
49    pub(crate) facts: Vec<MemoryFact>,
50    pub(crate) node_lookup: IdIndex<EntityId>,
51    pub(crate) fact_lookup: IdIndex<FactId>,
52    pub(crate) incident_offsets: Vec<usize>,
53    pub(crate) incident_facts: Vec<usize>,
54    pub(crate) incident_delta: HashMap<usize, Vec<usize>>,
55    pub(crate) supersessions: BTreeMap<FactId, Supersession>,
56    pub(crate) retractions: BTreeMap<FactId, Retraction>,
57    pub(crate) last_global_position: Option<u64>,
58}
59
60impl MemoryProjection {
61    pub(super) fn with_capacity(nodes: usize, facts: usize) -> Self {
62        Self {
63            nodes: Vec::with_capacity(nodes),
64            facts: Vec::with_capacity(facts),
65            node_lookup: IdIndex::with_capacity(nodes),
66            fact_lookup: IdIndex::with_capacity(facts),
67            incident_offsets: vec![0],
68            incident_facts: Vec::with_capacity(facts.saturating_mul(2)),
69            incident_delta: HashMap::new(),
70            supersessions: BTreeMap::new(),
71            retractions: BTreeMap::new(),
72            last_global_position: None,
73        }
74    }
75
76    pub(crate) fn rebuild_indexes(&mut self) -> Result<()> {
77        self.node_lookup.reserve(self.nodes.len());
78        for (index, revisions) in self.nodes.iter().enumerate() {
79            for revision in core::iter::once(&revisions.first).chain(&revisions.later) {
80                revision.node.validate()?;
81                if revision.node.id != revisions.first.node.id {
82                    return Err(MemoryError::InvalidValue {
83                        field: "projection.nodes",
84                        reason: "revision identifiers must match",
85                    });
86                }
87            }
88            if self
89                .node_lookup
90                .insert(revisions.first.node.id.clone(), index)
91                .is_some()
92            {
93                return Err(MemoryError::InvalidValue {
94                    field: "projection.nodes",
95                    reason: "node identifiers must be unique",
96                });
97            }
98        }
99        self.fact_lookup.reserve(self.facts.len());
100        let mut endpoints = Vec::with_capacity(self.facts.len());
101        for (index, fact) in self.facts.iter().enumerate() {
102            fact.validate()?;
103            let source = self.entity_index(&fact.source)?;
104            let target = self.entity_index(&fact.target)?;
105            if self.fact_lookup.insert(fact.id.clone(), index).is_some() {
106                return Err(MemoryError::ConflictingFact {
107                    id: fact.id.to_string(),
108                });
109            }
110            endpoints.push((source, target));
111        }
112        self.set_incidents(&endpoints)?;
113        self.validate_changes()
114    }
115
116    pub(super) fn set_incidents(&mut self, endpoints: &[(usize, usize)]) -> Result<()> {
117        let mut offsets = vec![0_usize; self.nodes.len() + 1];
118        for &(source, target) in endpoints {
119            offsets[source + 1] = offsets[source + 1]
120                .checked_add(1)
121                .ok_or(MemoryError::CapacityOverflow)?;
122            if target != source {
123                offsets[target + 1] = offsets[target + 1]
124                    .checked_add(1)
125                    .ok_or(MemoryError::CapacityOverflow)?;
126            }
127        }
128        for index in 1..offsets.len() {
129            offsets[index] = offsets[index]
130                .checked_add(offsets[index - 1])
131                .ok_or(MemoryError::CapacityOverflow)?;
132        }
133        let mut cursor = offsets[..self.nodes.len()].to_vec();
134        let mut incidents = vec![0_usize; *offsets.last().unwrap_or(&0)];
135        for (fact, &(source, target)) in endpoints.iter().enumerate() {
136            incidents[cursor[source]] = fact;
137            cursor[source] += 1;
138            if target != source {
139                incidents[cursor[target]] = fact;
140                cursor[target] += 1;
141            }
142        }
143        self.incident_offsets = offsets;
144        self.incident_facts = incidents;
145        self.incident_delta.clear();
146        Ok(())
147    }
148
149    fn entity_index(&self, id: &EntityId) -> Result<usize> {
150        self.node_lookup
151            .get(id)
152            .copied()
153            .ok_or_else(|| MemoryError::MissingEntity { id: id.to_string() })
154    }
155
156    fn validate_changes(&self) -> Result<()> {
157        for (prior, change) in &self.supersessions {
158            let prior_fact = self
159                .fact_lookup
160                .get(prior)
161                .map(|index| &self.facts[*index])
162                .ok_or_else(|| MemoryError::MissingFact {
163                    id: prior.to_string(),
164                })?;
165            let replacement = self
166                .fact_lookup
167                .get(&change.replacement)
168                .map(|index| &self.facts[*index])
169                .ok_or_else(|| MemoryError::MissingFact {
170                    id: change.replacement.to_string(),
171                })?;
172            if replacement.supersedes.as_ref() != Some(&prior_fact.id)
173                || replacement.valid_from != change.valid_from
174                || replacement.recorded_at != change.recorded_at
175            {
176                return Err(MemoryError::InvalidValue {
177                    field: "projection.supersessions",
178                    reason: "supersession index disagrees with replacement fact",
179                });
180            }
181        }
182        for (fact_id, change) in &self.retractions {
183            let fact = self
184                .fact_lookup
185                .get(fact_id)
186                .map(|index| &self.facts[*index])
187                .ok_or_else(|| MemoryError::MissingFact {
188                    id: fact_id.to_string(),
189                })?;
190            if change.valid_until <= fact.valid_from {
191                return Err(MemoryError::InvalidValue {
192                    field: "projection.retractions",
193                    reason: "retraction must follow fact validity",
194                });
195            }
196        }
197        Ok(())
198    }
199}
200
201impl Default for MemoryProjection {
202    fn default() -> Self {
203        Self::with_capacity(0, 0)
204    }
205}
206
207impl PartialEq for MemoryProjection {
208    fn eq(&self, other: &Self) -> bool {
209        self.nodes == other.nodes
210            && self.facts == other.facts
211            && self.supersessions == other.supersessions
212            && self.retractions == other.retractions
213            && self.last_global_position == other.last_global_position
214    }
215}
216
217impl Eq for MemoryProjection {}
218
219#[derive(Serialize)]
220struct ProjectionRef<'a> {
221    nodes: &'a [NodeHistory],
222    facts: &'a [MemoryFact],
223    supersessions: &'a BTreeMap<FactId, Supersession>,
224    retractions: &'a BTreeMap<FactId, Retraction>,
225    last_global_position: Option<u64>,
226}
227
228#[derive(Deserialize)]
229struct ProjectionData {
230    nodes: Vec<NodeHistory>,
231    facts: Vec<MemoryFact>,
232    supersessions: BTreeMap<FactId, Supersession>,
233    retractions: BTreeMap<FactId, Retraction>,
234    last_global_position: Option<u64>,
235}
236
237impl Serialize for MemoryProjection {
238    fn serialize<S>(&self, serializer: S) -> core::result::Result<S::Ok, S::Error>
239    where
240        S: Serializer,
241    {
242        ProjectionRef {
243            nodes: &self.nodes,
244            facts: &self.facts,
245            supersessions: &self.supersessions,
246            retractions: &self.retractions,
247            last_global_position: self.last_global_position,
248        }
249        .serialize(serializer)
250    }
251}
252
253impl<'de> Deserialize<'de> for MemoryProjection {
254    fn deserialize<D>(deserializer: D) -> core::result::Result<Self, D::Error>
255    where
256        D: Deserializer<'de>,
257    {
258        let data = ProjectionData::deserialize(deserializer)?;
259        let node_count = data.nodes.len();
260        let fact_count = data.facts.len();
261        let mut projection = Self {
262            nodes: data.nodes,
263            facts: data.facts,
264            node_lookup: IdIndex::with_capacity(node_count),
265            fact_lookup: IdIndex::with_capacity(fact_count),
266            incident_offsets: Vec::new(),
267            incident_facts: Vec::new(),
268            incident_delta: HashMap::new(),
269            supersessions: data.supersessions,
270            retractions: data.retractions,
271            last_global_position: data.last_global_position,
272        };
273        projection.rebuild_indexes().map_err(D::Error::custom)?;
274        Ok(projection)
275    }
276}
weavatrix_memory/projection/memory/state.rs

weavatrix_memory/projection/memory/
state.rs
