hayson_engine/buffer/
entry.rs

1use serde_json::{Map, Value};
2
3use crate::errors::HaysonCorruptLogError;
4
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6pub enum BufferOp {
7    Put,
8    Patch,
9    Delete,
10}
11
12#[derive(Debug, Clone, PartialEq)]
13pub struct BufferEntry {
14    pub table: String,
15    pub key: String,
16    pub op: BufferOp,
17    pub data: Option<Map<String, Value>>,
18    pub buffered_at: i64,
19    pub source: String,
20}
21
22pub fn serialize_buffer_entry(entry: &BufferEntry) -> String {
23    let mut line = Map::new();
24    line.insert("_table".into(), Value::String(entry.table.clone()));
25    line.insert("_key".into(), Value::String(entry.key.clone()));
26    line.insert(
27        "_op".into(),
28        Value::String(
29            match entry.op {
30                BufferOp::Put => "put",
31                BufferOp::Patch => "patch",
32                BufferOp::Delete => "delete",
33            }
34            .into(),
35        ),
36    );
37    line.insert("_at".into(), Value::from(entry.buffered_at));
38    line.insert("_source".into(), Value::String(entry.source.clone()));
39    if let Some(data) = &entry.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_buffer_entry(line: &str, path: &str, line_no: usize) -> Result<BufferEntry, HaysonCorruptLogError> {
48    let corrupt = |reason: String| HaysonCorruptLogError {
49        path: path.to_string(),
50        offset: line_no as u64,
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_obj = parsed
57        .as_object()
58        .ok_or_else(|| corrupt("line is not a JSON object".into()))?;
59
60    let table = source_obj
61        .get("_table")
62        .and_then(Value::as_str)
63        .filter(|s| !s.is_empty())
64        .ok_or_else(|| corrupt("line has no \"_table\"".into()))?
65        .to_string();
66
67    let key = source_obj
68        .get("_key")
69        .and_then(Value::as_str)
70        .filter(|s| !s.is_empty())
71        .ok_or_else(|| corrupt("line has no \"_key\"".into()))?
72        .to_string();
73
74    let op = match source_obj.get("_op").and_then(Value::as_str) {
75        Some("put") => BufferOp::Put,
76        Some("patch") => BufferOp::Patch,
77        Some("delete") => BufferOp::Delete,
78        _ => {
79            return Err(corrupt(format!(
80                "line has an unknown \"_op\" {:?}",
81                source_obj.get("_op")
82            )))
83        }
84    };
85
86    let buffered_at = source_obj
87        .get("_at")
88        .and_then(Value::as_i64)
89        .ok_or_else(|| corrupt("line has a malformed \"_at\"".into()))?;
90
91    let source = source_obj
92        .get("_source")
93        .and_then(Value::as_str)
94        .filter(|s| !s.is_empty())
95        .ok_or_else(|| corrupt("line has no \"_source\"".into()))?
96        .to_string();
97
98    if op == BufferOp::Delete {
99        return Ok(BufferEntry {
100            table,
101            key,
102            op,
103            data: None,
104            buffered_at,
105            source,
106        });
107    }
108
109    let mut data = Map::new();
110    for (name, value) in source_obj {
111        if name.starts_with('_') {
112            continue;
113        }
114        data.insert(name.clone(), value.clone());
115    }
116    Ok(BufferEntry {
117        table,
118        key,
119        op,
120        data: Some(data),
121        buffered_at,
122        source,
123    })
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129
130    fn entry(op: BufferOp, data: Option<Map<String, Value>>) -> BufferEntry {
131        BufferEntry {
132            table: "users".into(),
133            key: "u1".into(),
134            op,
135            data,
136            buffered_at: 1_700_000_000_000,
137            source: "branch-a".into(),
138        }
139    }
140
141    #[test]
142    fn round_trips_a_put_with_its_payload() {
143        let mut data = Map::new();
144        data.insert("name".into(), Value::String("Ania".into()));
145        let original = entry(BufferOp::Put, Some(data));
146
147        let line = serialize_buffer_entry(&original);
148        let parsed = parse_buffer_entry(&line, "buf.jsonl", 0).unwrap();
149        assert_eq!(parsed, original);
150    }
151
152    #[test]
153    fn round_trips_a_delete_without_a_payload() {
154        let original = entry(BufferOp::Delete, None);
155        let line = serialize_buffer_entry(&original);
156        let parsed = parse_buffer_entry(&line, "buf.jsonl", 0).unwrap();
157        assert_eq!(parsed, original);
158    }
159
160    #[test]
161    fn carries_no_final_seq_field() {
162        let mut data = Map::new();
163        data.insert("name".into(), Value::String("Ania".into()));
164        let line = serialize_buffer_entry(&entry(BufferOp::Put, Some(data)));
165        let parsed: Value = serde_json::from_str(&line).unwrap();
166        assert!(parsed.get("_seq").is_none());
167    }
168
169    #[test]
170    fn refuses_lines_that_are_not_usable() {
171        let cases = [
172            "{ nope",
173            "[]",
174            r#"{"_key":"u1","_op":"put","_at":1,"_source":"a"}"#,
175            r#"{"_table":"users","_op":"put","_at":1,"_source":"a"}"#,
176            r#"{"_table":"users","_key":"u1","_op":"merge","_at":1,"_source":"a"}"#,
177            r#"{"_table":"users","_key":"u1","_op":"put","_source":"a"}"#,
178        ];
179        for line in cases {
180            assert!(parse_buffer_entry(line, "buf.jsonl", 0).is_err());
181        }
182    }
183}
hayson_engine/buffer/entry.rs

hayson_engine/buffer/
entry.rs
