weavatrix_memory/projection/memory/
state.rs

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

weavatrix_memory/projection/memory/
state.rs
