hayson_engine/storage/
record.rs

1use serde_json::{Map, Value};
2
3use crate::errors::HaysonCorruptLogError;
4use crate::schema::HaysonRecord;
5
6#[derive(Debug, Clone, Copy, PartialEq, Eq)]
7pub enum LogOp {
8    Put,
9    Del,
10}
11
12#[derive(Debug, Clone, PartialEq, Eq)]
13pub struct LogMeta {
14    pub key: String,
15    pub op: LogOp,
16    pub seq: u64,
17    pub ts: i64,
18}
19
20#[derive(Debug, Clone, PartialEq)]
21pub struct LogEntry {
22    pub meta: LogMeta,
23    pub data: Option<HaysonRecord>,
24}
25
26/// Serializes one log line: meta under `_k`/`_op`/`_seq`/`_ts`, fields alphabetically next to it.
27pub fn serialize_entry(meta: &LogMeta, data: Option<&HaysonRecord>) -> String {
28    let mut line = Map::new();
29    line.insert("_k".into(), Value::String(meta.key.clone()));
30    line.insert(
31        "_op".into(),
32        Value::String(match meta.op {
33            LogOp::Put => "put".into(),
34            LogOp::Del => "del".into(),
35        }),
36    );
37    line.insert("_seq".into(), Value::from(meta.seq));
38    line.insert("_ts".into(), Value::from(meta.ts));
39    if let Some(data) = data {
40        for (name, value) in data {
41            line.insert(name.clone(), value.clone());
42        }
43    }
44    Value::Object(line).to_string()
45}
46
47pub fn parse_entry(line: &str, path: &str, offset: u64) -> Result<LogEntry, HaysonCorruptLogError> {
48    let corrupt = |reason: String| HaysonCorruptLogError {
49        path: path.to_string(),
50        offset,
51        reason,
52    };
53
54    let parsed: Value =
55        serde_json::from_str(line).map_err(|cause| corrupt(format!("line is not valid JSON ({cause})")))?;
56    let source = parsed
57        .as_object()
58        .ok_or_else(|| corrupt("line is not a JSON object".into()))?;
59
60    let key = source
61        .get("_k")
62        .and_then(Value::as_str)
63        .filter(|s| !s.is_empty())
64        .ok_or_else(|| corrupt("line has no \"_k\"".into()))?
65        .to_string();
66
67    let op = match source.get("_op").and_then(Value::as_str) {
68        Some("put") => LogOp::Put,
69        Some("del") => LogOp::Del,
70        _ => return Err(corrupt(format!("line has an unknown \"_op\" {:?}", source.get("_op")))),
71    };
72
73    let seq = source.get("_seq").and_then(Value::as_u64);
74    let ts = source.get("_ts").and_then(Value::as_i64);
75    let (Some(seq), Some(ts)) = (seq, ts) else {
76        return Err(corrupt("line has a malformed \"_seq\" or \"_ts\"".into()));
77    };
78
79    let meta = LogMeta { key, op, seq, ts };
80    if op == LogOp::Del {
81        return Ok(LogEntry { meta, data: None });
82    }
83
84    let mut data = HaysonRecord::new();
85    for (name, value) in source {
86        if name.starts_with('_') {
87            continue;
88        }
89        data.insert(name.clone(), value.clone());
90    }
91    Ok(LogEntry { meta, data: Some(data) })
92}
93
94#[cfg(test)]
95mod tests {
96    use std::fs;
97    use std::path::PathBuf;
98
99    use super::*;
100
101    fn meta(key: &str, op: LogOp, seq: u64, ts: i64) -> LogMeta {
102        LogMeta {
103            key: key.to_string(),
104            op,
105            seq,
106            ts,
107        }
108    }
109
110    #[test]
111    fn puts_meta_under_the_reserved_prefix_next_to_the_payload() {
112        let mut data = HaysonRecord::new();
113        data.insert("name".into(), Value::String("Ania".into()));
114        let line = serialize_entry(&meta("user_1", LogOp::Put, 7, 42), Some(&data));
115        let parsed: Value = serde_json::from_str(&line).unwrap();
116        assert_eq!(
117            parsed,
118            serde_json::json!({ "_k": "user_1", "_op": "put", "_seq": 7, "_ts": 42, "name": "Ania" })
119        );
120    }
121
122    #[test]
123    fn writes_payload_fields_alphabetically_whatever_order_they_arrived_in() {
124        let m = meta("user_1", LogOp::Put, 7, 42);
125        let mut declared = HaysonRecord::new();
126        declared.insert("id".into(), Value::String("user_1".into()));
127        declared.insert("name".into(), Value::String("Ania".into()));
128        declared.insert("age".into(), Value::from(31));
129
130        let mut alphabetical = HaysonRecord::new();
131        alphabetical.insert("age".into(), Value::from(31));
132        alphabetical.insert("id".into(), Value::String("user_1".into()));
133        alphabetical.insert("name".into(), Value::String("Ania".into()));
134
135        let a = serialize_entry(&m, Some(&declared));
136        let b = serialize_entry(&m, Some(&alphabetical));
137        assert_eq!(a, b);
138        assert_eq!(
139            a,
140            r#"{"_k":"user_1","_op":"put","_seq":7,"_ts":42,"age":31,"id":"user_1","name":"Ania"}"#
141        );
142    }
143
144    #[test]
145    fn reads_a_line_written_before_the_field_order_was_canonical() {
146        let legacy = r#"{"_k":"user_1","_op":"put","_seq":7,"_ts":42,"id":"user_1","name":"Ania","age":31}"#;
147        let entry = parse_entry(legacy, "users.jsonl", 0).unwrap();
148        assert_eq!(entry.meta, meta("user_1", LogOp::Put, 7, 42));
149        let data = entry.data.unwrap();
150        assert_eq!(data.get("id"), Some(&Value::String("user_1".into())));
151        assert_eq!(data.get("name"), Some(&Value::String("Ania".into())));
152        assert_eq!(data.get("age"), Some(&Value::from(31)));
153    }
154
155    #[test]
156    fn parses_a_tombstone_without_a_payload() {
157        let line = serialize_entry(&meta("user_1", LogOp::Del, 8, 42), None);
158        let entry = parse_entry(&line, "users.jsonl", 0).unwrap();
159        assert_eq!(entry.meta, meta("user_1", LogOp::Del, 8, 42));
160        assert_eq!(entry.data, None);
161    }
162
163    #[test]
164    fn refuses_lines_that_are_not_usable() {
165        let cases = [
166            "{ nope",
167            "[]",
168            r#"{"_op":"put","_seq":1,"_ts":1}"#,
169            r#"{"_k":"a","_op":"merge","_seq":1,"_ts":1}"#,
170        ];
171        for line in cases {
172            assert!(parse_entry(line, "users.jsonl", 12).is_err());
173        }
174    }
175
176    /// Byte-for-byte parity with `packages/hayson/test/fixtures/v1/unique-field/users.jsonl`,
177    /// written by the TS engine — proof that the Rust serializer's output is bit-compatible.
178    #[test]
179    fn matches_the_ts_fixture_byte_for_byte() {
180        let mut root = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
181        root.push("../../packages/hayson/test/fixtures/v1/unique-field/users.jsonl");
182        let fixture = fs::read_to_string(&root).expect("TS fixture must exist");
183
184        let mut first = HaysonRecord::new();
185        first.insert("createdAt".into(), Value::from(1787587082007i64));
186        first.insert("email".into(), Value::String("ania@example.test".into()));
187        first.insert("id".into(), Value::String("user_1".into()));
188        first.insert("name".into(), Value::String("Ania Kowalska".into()));
189        let line1 = serialize_entry(&meta("user_1", LogOp::Put, 1, 1787587082007), Some(&first));
190
191        let mut second = HaysonRecord::new();
192        second.insert("createdAt".into(), Value::from(1787587082008i64));
193        second.insert("email".into(), Value::String("piotr@example.test".into()));
194        second.insert("id".into(), Value::String("user_2".into()));
195        second.insert("name".into(), Value::String("Piotr Nowak".into()));
196        let line2 = serialize_entry(&meta("user_2", LogOp::Put, 2, 1787587082008), Some(&second));
197
198        let produced = format!("{line1}\n{line2}\n");
199        assert_eq!(produced, fixture);
200    }
201}
hayson_engine/storage/record.rs

hayson_engine/storage/
record.rs
