weavatrix_memory/projection/memory/
state.rs

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

weavatrix_memory/projection/memory/
state.rs
