weavatrix_memory/projection/memory/
binary.rs

1mod domain;
2mod io;
3
4use super::{
5    index::IdIndex,
6    state::{MemoryProjection, NodeHistory, NodeRevision, Retraction, Supersession},
7};
8use crate::{Codec, FactId, ProjectionSnapshot, ReplayCursor, Result, StreamId, Timestamp};
9use domain::{read_fact, read_node, write_fact, write_node};
10use io::{Reader, Writer, codec};
11use std::collections::{BTreeMap, HashMap};
12
13const HEADER: &[u8; 8] = b"WMEMCB01";
14
15#[derive(Debug, Clone, Copy, Default)]
16pub struct CompactSnapshotCodec;
17
18impl Codec<ProjectionSnapshot<MemoryProjection>> for CompactSnapshotCodec {
19    fn encode(&self, value: &ProjectionSnapshot<MemoryProjection>) -> Result<Vec<u8>> {
20        let mut writer = Writer::new(HEADER);
21        write_cursor(&mut writer, &value.cursor)?;
22        write_projection(&mut writer, &value.projection)?;
23        Ok(writer.finish())
24    }
25
26    fn decode(&self, bytes: &[u8]) -> Result<ProjectionSnapshot<MemoryProjection>> {
27        let mut reader = Reader::new(bytes, HEADER)?;
28        let cursor = read_cursor(&mut reader)?;
29        let projection = read_projection(&mut reader)?;
30        reader.finish()?;
31        if cursor.global_position != projection.last_global_position {
32            return Err(codec("cursor and projection positions disagree"));
33        }
34        Ok(ProjectionSnapshot { cursor, projection })
35    }
36}
37
38fn write_cursor(writer: &mut Writer, cursor: &ReplayCursor) -> Result<()> {
39    writer.optional_u64(cursor.global_position);
40    writer.usize(cursor.stream_versions.len())?;
41    for (stream, version) in &cursor.stream_versions {
42        writer.string(stream.as_str())?;
43        writer.varint(*version);
44    }
45    Ok(())
46}
47
48fn read_cursor(reader: &mut Reader<'_>) -> Result<ReplayCursor> {
49    let global_position = reader.optional_u64()?;
50    let count = reader.count()?;
51    let mut stream_versions = BTreeMap::new();
52    for _ in 0..count {
53        let stream = StreamId::new(reader.string()?)?;
54        let version = reader.varint()?;
55        if stream_versions.insert(stream, version).is_some() {
56            return Err(codec("duplicate stream in replay cursor"));
57        }
58    }
59    Ok(ReplayCursor {
60        global_position,
61        stream_versions,
62    })
63}
64
65fn write_projection(writer: &mut Writer, projection: &MemoryProjection) -> Result<()> {
66    writer.usize(projection.nodes.len())?;
67    for history in &projection.nodes {
68        writer.usize(history.later.len() + 1)?;
69        write_revision(writer, &history.first)?;
70        for revision in &history.later {
71            write_revision(writer, revision)?;
72        }
73    }
74    writer.usize(projection.facts.len())?;
75    for fact in &projection.facts {
76        write_fact(writer, fact)?;
77    }
78    writer.usize(projection.supersessions.len())?;
79    for (prior, change) in &projection.supersessions {
80        writer.string(prior.as_str())?;
81        writer.string(change.replacement.as_str())?;
82        writer.signed(change.valid_from.as_unix_micros());
83        writer.signed(change.recorded_at.as_unix_micros());
84    }
85    writer.usize(projection.retractions.len())?;
86    for (fact, change) in &projection.retractions {
87        writer.string(fact.as_str())?;
88        writer.signed(change.valid_until.as_unix_micros());
89        writer.signed(change.recorded_at.as_unix_micros());
90    }
91    writer.optional_u64(projection.last_global_position);
92    Ok(())
93}
94
95fn read_projection(reader: &mut Reader<'_>) -> Result<MemoryProjection> {
96    let node_count = reader.count()?;
97    let mut nodes = Vec::with_capacity(node_count);
98    for _ in 0..node_count {
99        nodes.push(read_history(reader)?);
100    }
101    let fact_count = reader.count()?;
102    let mut facts = Vec::with_capacity(fact_count);
103    for _ in 0..fact_count {
104        facts.push(read_fact(reader)?);
105    }
106    let supersessions = read_supersessions(reader)?;
107    let retractions = read_retractions(reader)?;
108    let last_global_position = reader.optional_u64()?;
109    let mut projection = MemoryProjection {
110        nodes,
111        facts,
112        node_lookup: IdIndex::with_capacity(node_count),
113        fact_lookup: IdIndex::with_capacity(fact_count),
114        incident_offsets: Vec::new(),
115        incident_facts: Vec::new(),
116        incident_delta: HashMap::new(),
117        supersessions,
118        retractions,
119        last_global_position,
120    };
121    projection.rebuild_indexes()?;
122    Ok(projection)
123}
124
125fn write_revision(writer: &mut Writer, revision: &NodeRevision) -> Result<()> {
126    write_node(writer, &revision.node)?;
127    writer.signed(revision.recorded_at.as_unix_micros());
128    writer.varint(revision.position);
129    Ok(())
130}
131
132fn read_history(reader: &mut Reader<'_>) -> Result<NodeHistory> {
133    let count = reader.count()?;
134    if count == 0 {
135        return Err(codec("node revision history must not be empty"));
136    }
137    let first = read_revision(reader)?;
138    let mut later = Vec::with_capacity(count - 1);
139    for _ in 1..count {
140        later.push(read_revision(reader)?);
141    }
142    Ok(NodeHistory { first, later })
143}
144
145fn read_revision(reader: &mut Reader<'_>) -> Result<NodeRevision> {
146    Ok(NodeRevision {
147        node: read_node(reader)?,
148        recorded_at: Timestamp::from_unix_micros(reader.signed()?),
149        position: reader.varint()?,
150    })
151}
152
153fn read_supersessions(reader: &mut Reader<'_>) -> Result<BTreeMap<FactId, Supersession>> {
154    let count = reader.count()?;
155    let mut changes = BTreeMap::new();
156    for _ in 0..count {
157        let prior = FactId::new(reader.string()?)?;
158        let change = Supersession {
159            replacement: FactId::new(reader.string()?)?,
160            valid_from: Timestamp::from_unix_micros(reader.signed()?),
161            recorded_at: Timestamp::from_unix_micros(reader.signed()?),
162        };
163        if changes.insert(prior, change).is_some() {
164            return Err(codec("duplicate supersession"));
165        }
166    }
167    Ok(changes)
168}
169
170fn read_retractions(reader: &mut Reader<'_>) -> Result<BTreeMap<FactId, Retraction>> {
171    let count = reader.count()?;
172    let mut changes = BTreeMap::new();
173    for _ in 0..count {
174        let fact = FactId::new(reader.string()?)?;
175        let change = Retraction {
176            valid_until: Timestamp::from_unix_micros(reader.signed()?),
177            recorded_at: Timestamp::from_unix_micros(reader.signed()?),
178        };
179        if changes.insert(fact, change).is_some() {
180            return Err(codec("duplicate retraction"));
181        }
182    }
183    Ok(changes)
184}
weavatrix_memory/projection/memory/binary.rs

weavatrix_memory/projection/memory/
binary.rs
