hayson-engine 0.26.0

use serde_json::{Map, Value};

use crate::errors::HaysonCorruptLogError;
use crate::schema::HaysonRecord;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LogOp {
    Put,
    Del,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LogMeta {
    pub key: String,
    pub op: LogOp,
    pub seq: u64,
    pub ts: i64,
}

#[derive(Debug, Clone, PartialEq)]
pub struct LogEntry {
    pub meta: LogMeta,
    pub data: Option<HaysonRecord>,
}

/// Serializes one log line: meta under `_k`/`_op`/`_seq`/`_ts`, fields alphabetically next to it.
pub fn serialize_entry(meta: &LogMeta, data: Option<&HaysonRecord>) -> String {
    let mut line = Map::new();
    line.insert("_k".into(), Value::String(meta.key.clone()));
    line.insert(
        "_op".into(),
        Value::String(match meta.op {
            LogOp::Put => "put".into(),
            LogOp::Del => "del".into(),
        }),
    );
    line.insert("_seq".into(), Value::from(meta.seq));
    line.insert("_ts".into(), Value::from(meta.ts));
    if let Some(data) = data {
        for (name, value) in data {
            line.insert(name.clone(), value.clone());
        }
    }
    Value::Object(line).to_string()
}

pub fn parse_entry(line: &str, path: &str, offset: u64) -> Result<LogEntry, HaysonCorruptLogError> {
    let corrupt = |reason: String| HaysonCorruptLogError {
        path: path.to_string(),
        offset,
        reason,
    };

    let parsed: Value =
        serde_json::from_str(line).map_err(|cause| corrupt(format!("line is not valid JSON ({cause})")))?;
    let source = parsed
        .as_object()
        .ok_or_else(|| corrupt("line is not a JSON object".into()))?;

    let key = source
        .get("_k")
        .and_then(Value::as_str)
        .filter(|s| !s.is_empty())
        .ok_or_else(|| corrupt("line has no \"_k\"".into()))?
        .to_string();

    let op = match source.get("_op").and_then(Value::as_str) {
        Some("put") => LogOp::Put,
        Some("del") => LogOp::Del,
        _ => return Err(corrupt(format!("line has an unknown \"_op\" {:?}", source.get("_op")))),
    };

    let seq = source.get("_seq").and_then(Value::as_u64);
    let ts = source.get("_ts").and_then(Value::as_i64);
    let (Some(seq), Some(ts)) = (seq, ts) else {
        return Err(corrupt("line has a malformed \"_seq\" or \"_ts\"".into()));
    };

    let meta = LogMeta { key, op, seq, ts };
    if op == LogOp::Del {
        return Ok(LogEntry { meta, data: None });
    }

    let mut data = HaysonRecord::new();
    for (name, value) in source {
        if name.starts_with('_') {
            continue;
        }
        data.insert(name.clone(), value.clone());
    }
    Ok(LogEntry { meta, data: Some(data) })
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::path::PathBuf;

    use super::*;

    fn meta(key: &str, op: LogOp, seq: u64, ts: i64) -> LogMeta {
        LogMeta {
            key: key.to_string(),
            op,
            seq,
            ts,
        }
    }

    #[test]
    fn puts_meta_under_the_reserved_prefix_next_to_the_payload() {
        let mut data = HaysonRecord::new();
        data.insert("name".into(), Value::String("Ania".into()));
        let line = serialize_entry(&meta("user_1", LogOp::Put, 7, 42), Some(&data));
        let parsed: Value = serde_json::from_str(&line).unwrap();
        assert_eq!(
            parsed,
            serde_json::json!({ "_k": "user_1", "_op": "put", "_seq": 7, "_ts": 42, "name": "Ania" })
        );
    }

    #[test]
    fn writes_payload_fields_alphabetically_whatever_order_they_arrived_in() {
        let m = meta("user_1", LogOp::Put, 7, 42);
        let mut declared = HaysonRecord::new();
        declared.insert("id".into(), Value::String("user_1".into()));
        declared.insert("name".into(), Value::String("Ania".into()));
        declared.insert("age".into(), Value::from(31));

        let mut alphabetical = HaysonRecord::new();
        alphabetical.insert("age".into(), Value::from(31));
        alphabetical.insert("id".into(), Value::String("user_1".into()));
        alphabetical.insert("name".into(), Value::String("Ania".into()));

        let a = serialize_entry(&m, Some(&declared));
        let b = serialize_entry(&m, Some(&alphabetical));
        assert_eq!(a, b);
        assert_eq!(
            a,
            r#"{"_k":"user_1","_op":"put","_seq":7,"_ts":42,"age":31,"id":"user_1","name":"Ania"}"#
        );
    }

    #[test]
    fn reads_a_line_written_before_the_field_order_was_canonical() {
        let legacy = r#"{"_k":"user_1","_op":"put","_seq":7,"_ts":42,"id":"user_1","name":"Ania","age":31}"#;
        let entry = parse_entry(legacy, "users.jsonl", 0).unwrap();
        assert_eq!(entry.meta, meta("user_1", LogOp::Put, 7, 42));
        let data = entry.data.unwrap();
        assert_eq!(data.get("id"), Some(&Value::String("user_1".into())));
        assert_eq!(data.get("name"), Some(&Value::String("Ania".into())));
        assert_eq!(data.get("age"), Some(&Value::from(31)));
    }

    #[test]
    fn parses_a_tombstone_without_a_payload() {
        let line = serialize_entry(&meta("user_1", LogOp::Del, 8, 42), None);
        let entry = parse_entry(&line, "users.jsonl", 0).unwrap();
        assert_eq!(entry.meta, meta("user_1", LogOp::Del, 8, 42));
        assert_eq!(entry.data, None);
    }

    #[test]
    fn refuses_lines_that_are_not_usable() {
        let cases = [
            "{ nope",
            "[]",
            r#"{"_op":"put","_seq":1,"_ts":1}"#,
            r#"{"_k":"a","_op":"merge","_seq":1,"_ts":1}"#,
        ];
        for line in cases {
            assert!(parse_entry(line, "users.jsonl", 12).is_err());
        }
    }

    /// Byte-for-byte parity with `packages/hayson/test/fixtures/v1/unique-field/users.jsonl`,
    /// written by the TS engine — proof that the Rust serializer's output is bit-compatible.
    #[test]
    fn matches_the_ts_fixture_byte_for_byte() {
        let mut root = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        root.push("../../packages/hayson/test/fixtures/v1/unique-field/users.jsonl");
        let fixture = fs::read_to_string(&root).expect("TS fixture must exist");

        let mut first = HaysonRecord::new();
        first.insert("createdAt".into(), Value::from(1787587082007i64));
        first.insert("email".into(), Value::String("ania@example.test".into()));
        first.insert("id".into(), Value::String("user_1".into()));
        first.insert("name".into(), Value::String("Ania Kowalska".into()));
        let line1 = serialize_entry(&meta("user_1", LogOp::Put, 1, 1787587082007), Some(&first));

        let mut second = HaysonRecord::new();
        second.insert("createdAt".into(), Value::from(1787587082008i64));
        second.insert("email".into(), Value::String("piotr@example.test".into()));
        second.insert("id".into(), Value::String("user_2".into()));
        second.insert("name".into(), Value::String("Piotr Nowak".into()));
        let line2 = serialize_entry(&meta("user_2", LogOp::Put, 2, 1787587082008), Some(&second));

        let produced = format!("{line1}\n{line2}\n");
        assert_eq!(produced, fixture);
    }
}