weavatrix_memory/projection/memory/
parts.rs

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

weavatrix_memory/projection/memory/
parts.rs
