weavatrix_memory/projection/memory/
parts.rs

1use super::{
2    MemoryProjection,
3    state::{NodeHistory, NodeRevision},
4};
5use crate::{MemoryError, MemoryFact, MemoryNode, Result, Timestamp};
6use std::thread;
7
8#[derive(Clone, Copy)]
9struct PreparedFact {
10    source: usize,
11    target: usize,
12    id_hash: u64,
13}
14
15impl MemoryProjection {
16    /// Builds a current projection from already extracted nodes and facts.
17    ///
18    /// This bypasses event-envelope replay but retains domain, endpoint,
19    /// uniqueness, evidence, and supersession validation. All nodes are treated
20    /// as revisions recorded at `known_at`. Large fact sets are prepared in
21    /// parallel using scoped standard-library threads.
22    ///
23    /// # Errors
24    ///
25    /// Rejects duplicate identifiers, invalid nodes or facts, missing
26    /// endpoints, facts recorded after `known_at`, and invalid supersession
27    /// chains.
28    pub fn try_from_parts(
29        nodes: Vec<MemoryNode>,
30        facts: Vec<MemoryFact>,
31        known_at: Timestamp,
32        source_position: Option<u64>,
33    ) -> Result<Self> {
34        let mut projection = Self::with_capacity(nodes.len(), facts.len());
35        for (position, node) in nodes.into_iter().enumerate() {
36            node.validate()?;
37            let index = projection.nodes.len();
38            if projection
39                .node_lookup
40                .insert(node.id.clone(), index)
41                .is_some()
42            {
43                return Err(MemoryError::InvalidValue {
44                    field: "nodes",
45                    reason: "node identifiers must be unique",
46                });
47            }
48            projection.nodes.push(NodeHistory::new(NodeRevision {
49                node,
50                recorded_at: known_at,
51                position: u64::try_from(position).map_err(|_| MemoryError::CapacityOverflow)?,
52            }));
53        }
54        let prepared = prepare_facts(&projection, &facts, known_at)?;
55        for (index, (fact, prepared)) in facts.iter().zip(&prepared).enumerate() {
56            if projection
57                .fact_lookup
58                .insert_hashed(fact.id.clone(), index, prepared.id_hash)
59                .is_some()
60            {
61                return Err(MemoryError::ConflictingFact {
62                    id: fact.id.to_string(),
63                });
64            }
65        }
66        for fact in &facts {
67            if let Some(prior) = &fact.supersedes {
68                projection.apply_supersession(prior, fact)?;
69            }
70        }
71        let endpoints = prepared
72            .into_iter()
73            .map(|fact| (fact.source, fact.target))
74            .collect::<Vec<_>>();
75        projection.facts = facts;
76        projection.set_incidents(&endpoints)?;
77        projection.last_global_position = source_position;
78        Ok(projection)
79    }
80}
81
82fn prepare_facts(
83    projection: &MemoryProjection,
84    facts: &[MemoryFact],
85    known_at: Timestamp,
86) -> Result<Vec<PreparedFact>> {
87    let parallelism = thread::available_parallelism().map_or(1, usize::from);
88    let workers = parallelism.min(facts.len().div_ceil(16_384)).max(1);
89    if workers == 1 {
90        return facts
91            .iter()
92            .map(|fact| prepare_fact(projection, fact, known_at))
93            .collect();
94    }
95    let chunk_size = facts.len().div_ceil(workers);
96    thread::scope(|scope| {
97        let handles = facts
98            .chunks(chunk_size)
99            .map(|chunk| {
100                scope.spawn(move || {
101                    chunk
102                        .iter()
103                        .map(|fact| prepare_fact(projection, fact, known_at))
104                        .collect::<Result<Vec<_>>>()
105                })
106            })
107            .collect::<Vec<_>>();
108        let mut prepared = Vec::with_capacity(facts.len());
109        for handle in handles {
110            let mut chunk = handle.join().map_err(|_| MemoryError::InvalidValue {
111                field: "facts",
112                reason: "parallel fact preparation panicked",
113            })??;
114            prepared.append(&mut chunk);
115        }
116        Ok(prepared)
117    })
118}
119
120fn prepare_fact(
121    projection: &MemoryProjection,
122    fact: &MemoryFact,
123    known_at: Timestamp,
124) -> Result<PreparedFact> {
125    if fact.recorded_at > known_at {
126        return Err(MemoryError::InvalidValue {
127            field: "facts",
128            reason: "fact recorded_at must not exceed known_at",
129        });
130    }
131    fact.validate()?;
132    Ok(PreparedFact {
133        source: projection.require_entity(&fact.source)?,
134        target: projection.require_entity(&fact.target)?,
135        id_hash: projection.fact_lookup.hash(&fact.id),
136    })
137}
weavatrix_memory/projection/memory/parts.rs

weavatrix_memory/projection/memory/
parts.rs
