hayson-engine 0.17.0

use serde_json::{Map, Value};

use crate::errors::HaysonCorruptLogError;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BufferOp {
    Put,
    Patch,
    Delete,
}

#[derive(Debug, Clone, PartialEq)]
pub struct BufferEntry {
    pub table: String,
    pub key: String,
    pub op: BufferOp,
    pub data: Option<Map<String, Value>>,
    pub buffered_at: i64,
    pub source: String,
}

pub fn serialize_buffer_entry(entry: &BufferEntry) -> String {
    let mut line = Map::new();
    line.insert("_table".into(), Value::String(entry.table.clone()));
    line.insert("_key".into(), Value::String(entry.key.clone()));
    line.insert(
        "_op".into(),
        Value::String(
            match entry.op {
                BufferOp::Put => "put",
                BufferOp::Patch => "patch",
                BufferOp::Delete => "delete",
            }
            .into(),
        ),
    );
    line.insert("_at".into(), Value::from(entry.buffered_at));
    line.insert("_source".into(), Value::String(entry.source.clone()));
    if let Some(data) = &entry.data {
        for (name, value) in data {
            line.insert(name.clone(), value.clone());
        }
    }
    Value::Object(line).to_string()
}

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

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

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

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

    let op = match source_obj.get("_op").and_then(Value::as_str) {
        Some("put") => BufferOp::Put,
        Some("patch") => BufferOp::Patch,
        Some("delete") => BufferOp::Delete,
        _ => {
            return Err(corrupt(format!(
                "line has an unknown \"_op\" {:?}",
                source_obj.get("_op")
            )))
        }
    };

    let buffered_at = source_obj
        .get("_at")
        .and_then(Value::as_i64)
        .ok_or_else(|| corrupt("line has a malformed \"_at\"".into()))?;

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

    if op == BufferOp::Delete {
        return Ok(BufferEntry {
            table,
            key,
            op,
            data: None,
            buffered_at,
            source,
        });
    }

    let mut data = Map::new();
    for (name, value) in source_obj {
        if name.starts_with('_') {
            continue;
        }
        data.insert(name.clone(), value.clone());
    }
    Ok(BufferEntry {
        table,
        key,
        op,
        data: Some(data),
        buffered_at,
        source,
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    fn entry(op: BufferOp, data: Option<Map<String, Value>>) -> BufferEntry {
        BufferEntry {
            table: "users".into(),
            key: "u1".into(),
            op,
            data,
            buffered_at: 1_700_000_000_000,
            source: "branch-a".into(),
        }
    }

    #[test]
    fn round_trips_a_put_with_its_payload() {
        let mut data = Map::new();
        data.insert("name".into(), Value::String("Ania".into()));
        let original = entry(BufferOp::Put, Some(data));

        let line = serialize_buffer_entry(&original);
        let parsed = parse_buffer_entry(&line, "buf.jsonl", 0).unwrap();
        assert_eq!(parsed, original);
    }

    #[test]
    fn round_trips_a_delete_without_a_payload() {
        let original = entry(BufferOp::Delete, None);
        let line = serialize_buffer_entry(&original);
        let parsed = parse_buffer_entry(&line, "buf.jsonl", 0).unwrap();
        assert_eq!(parsed, original);
    }

    #[test]
    fn carries_no_final_seq_field() {
        let mut data = Map::new();
        data.insert("name".into(), Value::String("Ania".into()));
        let line = serialize_buffer_entry(&entry(BufferOp::Put, Some(data)));
        let parsed: Value = serde_json::from_str(&line).unwrap();
        assert!(parsed.get("_seq").is_none());
    }

    #[test]
    fn refuses_lines_that_are_not_usable() {
        let cases = [
            "{ nope",
            "[]",
            r#"{"_key":"u1","_op":"put","_at":1,"_source":"a"}"#,
            r#"{"_table":"users","_op":"put","_at":1,"_source":"a"}"#,
            r#"{"_table":"users","_key":"u1","_op":"merge","_at":1,"_source":"a"}"#,
            r#"{"_table":"users","_key":"u1","_op":"put","_source":"a"}"#,
        ];
        for line in cases {
            assert!(parse_buffer_entry(line, "buf.jsonl", 0).is_err());
        }
    }
}