hayson-engine 0.27.0

use serde_json::Value;

use crate::table::{Table, TableError};

use super::entry::{BufferEntry, BufferOp};

#[derive(Debug, Clone, PartialEq)]
pub struct Conflict {
    pub table: String,
    pub key: String,
    pub versions: Vec<(String, BufferEntry)>,
}

#[derive(Debug, Clone, PartialEq, Default)]
pub struct MergeResult {
    pub clean: Vec<BufferEntry>,
    pub conflicts: Vec<Conflict>,
}

/// K-way merges every buffer entry by `buffered_at` — this is the relative order WITHIN one merge
/// operation only. It never looks at what is already committed in the database: whatever comes out
/// as `clean` still has to go through `apply_clean_entries`, which stages each entry through the
/// table's normal `put`/`patch`/`delete` path and so always lands after the current end of the log,
/// no matter how early `buffered_at` claims to be — this is what satisfies the "never retroactive"
/// rule, not anything in this function.
pub fn merge(mut entries: Vec<BufferEntry>) -> MergeResult {
    let order: Vec<usize> = {
        let mut idx: Vec<usize> = (0..entries.len()).collect();
        idx.sort_by(|&a, &b| entries[a].buffered_at.cmp(&entries[b].buffered_at));
        idx
    };
    entries = order.into_iter().map(|i| entries[i].clone()).collect();

    let mut conflicted_keys: Vec<(String, String)> = Vec::new();
    for entry in &entries {
        let group_key = (entry.table.clone(), entry.key.clone());
        if conflicted_keys.contains(&group_key) {
            continue;
        }
        let versions: Vec<&BufferEntry> = entries
            .iter()
            .filter(|other| other.table == entry.table && other.key == entry.key)
            .collect();
        let distinct_sources: std::collections::HashSet<&str> = versions.iter().map(|e| e.source.as_str()).collect();
        let first = versions.first().map(|e| (e.op, &e.data));
        let all_same_content = versions.iter().all(|e| Some((e.op, &e.data)) == first);
        if distinct_sources.len() > 1 && !all_same_content {
            conflicted_keys.push(group_key);
        }
    }

    let mut clean = Vec::new();
    let mut conflicts: Vec<Conflict> = Vec::new();
    for entry in entries {
        let group_key = (entry.table.clone(), entry.key.clone());
        if conflicted_keys.contains(&group_key) {
            match conflicts
                .iter_mut()
                .find(|c| c.table == entry.table && c.key == entry.key)
            {
                Some(conflict) => conflict.versions.push((entry.source.clone(), entry)),
                None => conflicts.push(Conflict {
                    table: entry.table.clone(),
                    key: entry.key.clone(),
                    versions: vec![(entry.source.clone(), entry)],
                }),
            }
        } else {
            clean.push(entry);
        }
    }

    MergeResult { clean, conflicts }
}

/// The integration point with the live database: stages every clean entry through `Table`'s
/// `put_from_buffer`/`patch_from_buffer`/`delete_from_buffer`, in the order `merge` produced —
/// never through a lower-level path that could smuggle in a historical `_seq`, and never through
/// the trigger-firing `put`/`patch`/`delete` (the entry's originating live write already fired
/// them once). A `TableError::Unique` here means the entry was clean against other buffers but
/// violates a unique constraint against the CURRENT registry state (which may have changed since
/// the entry was buffered) — that also becomes a `Conflict`, not a silent error. Any other
/// `TableError` (validation, not-found on a patch, ...) is surfaced to the caller instead of being
/// folded into a conflict, since it is not something conflict resolution can fix.
pub fn apply_clean_entries(table: &mut Table, clean: Vec<BufferEntry>) -> Result<Vec<Conflict>, TableError> {
    let mut conflicts = Vec::new();
    for entry in clean {
        let outcome = match entry.op {
            BufferOp::Put => {
                let document = Value::Object(entry.data.clone().unwrap_or_default());
                table.put_from_buffer(&document, false).map(|_| ())
            }
            BufferOp::Patch => {
                let changes = Value::Object(entry.data.clone().unwrap_or_default());
                table.patch_from_buffer(&entry.key, &changes, false).map(|_| ())
            }
            BufferOp::Delete => table.delete_from_buffer(&entry.key, false).map(|_| ()),
        };
        match outcome {
            Ok(()) => {}
            Err(TableError::Unique(_)) => conflicts.push(Conflict {
                table: entry.table.clone(),
                key: entry.key.clone(),
                versions: vec![(entry.source.clone(), entry)],
            }),
            Err(other) => return Err(other),
        }
    }
    Ok(conflicts)
}

#[cfg(test)]
mod tests {
    use serde_json::{json, Map};
    use tempfile::tempdir;

    use super::*;
    use crate::schema::parse_schema;

    fn users_schema() -> crate::schema::TableSchema {
        parse_schema(&json!({
            "schemaVersion": 1,
            "table": "users",
            "primaryKey": "id",
            "fields": {
                "id": { "type": "string" },
                "name": { "type": "string", "required": true },
                "email": { "type": "string", "unique": true },
            },
        }))
        .unwrap()
    }

    fn put(key: &str, source: &str, buffered_at: i64, name: &str) -> BufferEntry {
        let mut data = Map::new();
        data.insert("id".into(), Value::String(key.into()));
        data.insert("name".into(), Value::String(name.into()));
        BufferEntry {
            table: "users".into(),
            key: key.into(),
            op: BufferOp::Put,
            data: Some(data),
            buffered_at,
            source: source.into(),
        }
    }

    fn self_audit_trigger() -> crate::catalog::TriggerEntry {
        crate::catalog::TriggerEntry {
            name: "users_insert_self_audit".into(),
            table: "users".into(),
            event: crate::catalog::TriggerEvent::Insert,
            effect: crate::catalog::TriggerEffect {
                table: "users".into(),
                fields: vec![
                    crate::catalog::TriggerFieldMapping {
                        target: "id".into(),
                        source: crate::catalog::TriggerValueSource::Constant(json!("audit-u1")),
                    },
                    crate::catalog::TriggerFieldMapping {
                        target: "name".into(),
                        source: crate::catalog::TriggerValueSource::Constant(json!("audit")),
                    },
                ],
            },
        }
    }

    #[test]
    fn merging_a_buffer_never_fires_a_trigger_the_live_write_already_fired() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("users.jsonl");
        let mut table = Table::open(users_schema(), &path).unwrap();
        table.set_triggers(vec![self_audit_trigger()]);

        let entry = put("u1", "branch-a", 1, "Ania");
        let result = merge(vec![entry]);
        assert!(result.conflicts.is_empty());

        apply_clean_entries(&mut table, result.clean).unwrap();

        assert!(table.get("u1").unwrap().is_some());
        assert!(table.get("audit-u1").unwrap().is_none());
    }

    #[test]
    fn two_folders_without_conflicts_merge_in_buffered_at_order() {
        let a = put("u2", "branch-a", 200, "Piotr");
        let b = put("u1", "branch-b", 100, "Ania");
        let result = merge(vec![a.clone(), b.clone()]);

        assert!(result.conflicts.is_empty());
        assert_eq!(result.clean, vec![b, a]);
    }

    #[test]
    fn the_same_key_changed_differently_in_two_buffers_is_a_conflict_not_clean() {
        let a = put("u1", "branch-a", 100, "Ania");
        let b = put("u1", "branch-b", 200, "Zosia");
        let result = merge(vec![a.clone(), b.clone()]);

        assert!(result.clean.is_empty());
        assert_eq!(result.conflicts.len(), 1);
        let conflict = &result.conflicts[0];
        assert_eq!(conflict.key, "u1");
        assert_eq!(conflict.versions.len(), 2);
    }

    #[test]
    fn a_buffer_entry_earlier_than_an_already_committed_write_never_jumps_ahead_of_it() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("users.jsonl");
        let mut table = Table::open(users_schema(), &path).unwrap();

        table.put(&json!({ "id": "base", "name": "Committed" }), false).unwrap();

        let earlier = put("late-arriving", "branch-a", 1, "FromTheOldBranch");
        let result = merge(vec![earlier]);
        assert!(result.conflicts.is_empty());

        apply_clean_entries(&mut table, result.clean).unwrap();

        let raw = std::fs::read_to_string(&path).unwrap();
        let lines: Vec<&str> = raw.lines().collect();
        assert_eq!(lines.len(), 2);
        assert!(lines[0].contains("\"base\""));
        assert!(lines[1].contains("\"late-arriving\""));
    }

    #[test]
    fn a_write_clean_across_buffers_still_conflicts_on_unique_against_current_db_state() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("users.jsonl");
        let mut table = Table::open(users_schema(), &path).unwrap();

        table
            .put(&json!({ "id": "u1", "name": "Ania", "email": "a@x.io" }), false)
            .unwrap();

        let mut data = Map::new();
        data.insert("id".into(), Value::String("u2".into()));
        data.insert("name".into(), Value::String("Zosia".into()));
        data.insert("email".into(), Value::String("a@x.io".into()));
        let entry = BufferEntry {
            table: "users".into(),
            key: "u2".into(),
            op: BufferOp::Put,
            data: Some(data),
            buffered_at: 1,
            source: "branch-a".into(),
        };

        let result = merge(vec![entry]);
        assert!(result.conflicts.is_empty());

        let conflicts = apply_clean_entries(&mut table, result.clean).unwrap();
        assert_eq!(conflicts.len(), 1);
        assert_eq!(conflicts[0].key, "u2");
        assert_eq!(table.get("u2").unwrap(), None);
    }
}