grit-core 0.2.2

//! The oplog merge laws (Design Invariant 5): applying any interleaving of
//! two devices' op sequences converges to the same graph — application is
//! idempotent and commutative for concurrent ops. This must hold before any
//! sync mechanism exists; proptest hammers it with adversarial op mixes
//! (colliding ids, out-of-order invalidations, merges racing adds, purges
//! racing everything, merge cycles across devices).

use std::sync::Arc;

use grit_core::{GraphOp, Grit, Hlc, ManualClock, OplogEntry, Options};
use proptest::prelude::*;
use serde_json::json;
use uuid::Uuid;

const POOL: u128 = 6;

fn node_id(i: u8) -> Uuid {
    Uuid::from_u128(0x0100_0000 + (i as u128 % POOL))
}

fn edge_id(i: u8) -> Uuid {
    Uuid::from_u128(0x0200_0000 + (i as u128 % POOL))
}

fn episode_id(i: u8) -> Uuid {
    Uuid::from_u128(0x0300_0000 + (i as u128 % POOL))
}

/// Generator-level op description; deterministic ids from small pools so ops
/// collide on purpose.
#[derive(Debug, Clone)]
enum Spec {
    AddNode {
        node: u8,
        name_seed: u8,
    },
    AddEdge {
        edge: u8,
        src: u8,
        dst: u8,
        valid_at: Option<i64>,
    },
    AddEpisode {
        episode: u8,
        mentions: Vec<u8>,
    },
    Invalidate {
        edge: u8,
        at: i64,
    },
    /// `from` parity is forced per device (A merges even nodes, B odd) so two
    /// devices never issue conflicting merges of the SAME node — that is a
    /// Layer 2 protocol violation, outside the convergence contract. Cross
    /// merges (A: 2→3, B: 3→2) still happen and exercise cycle resolution.
    Merge {
        from: u8,
        into_offset: u8,
    },
    PurgeNode {
        node: u8,
    },
    PurgeEdge {
        edge: u8,
    },
    /// Per-field LWW: field chosen by seed parity, value from the seed —
    /// two devices routinely update the SAME field of the SAME node.
    UpdateNode {
        node: u8,
        field_seed: u8,
        value_seed: u8,
    },
}

fn spec_strategy() -> impl Strategy<Value = Spec> {
    prop_oneof![
        3 => (any::<u8>(), any::<u8>()).prop_map(|(node, name_seed)| Spec::AddNode { node, name_seed }),
        3 => (any::<u8>(), any::<u8>(), any::<u8>(), proptest::option::of(1_000i64..2_000))
            .prop_map(|(edge, src, dst, valid_at)| Spec::AddEdge { edge, src, dst, valid_at }),
        2 => (any::<u8>(), proptest::collection::vec(any::<u8>(), 0..3))
            .prop_map(|(episode, mentions)| Spec::AddEpisode { episode, mentions }),
        // Deliberately tiny invalid_at pool: two devices invalidating the same
        // edge at the same event time must collide on the
        // (edge_id, invalid_at) PK routinely, exercising the MIN-merge there.
        2 => (any::<u8>(), 1_000i64..1_004).prop_map(|(edge, at)| Spec::Invalidate { edge, at }),
        1 => (any::<u8>(), any::<u8>()).prop_map(|(from, into_offset)| Spec::Merge { from, into_offset }),
        1 => any::<u8>().prop_map(|node| Spec::PurgeNode { node }),
        1 => any::<u8>().prop_map(|edge| Spec::PurgeEdge { edge }),
        2 => (any::<u8>(), any::<u8>(), any::<u8>())
            .prop_map(|(node, field_seed, value_seed)| Spec::UpdateNode { node, field_seed, value_seed }),
    ]
}

/// Realize a spec as a concrete op for `device` ("a" or "b").
fn to_op(spec: &Spec, device: &str) -> GraphOp {
    match spec {
        Spec::AddNode { node, name_seed } => GraphOp::AddNode {
            id: node_id(*node),
            kind: "k".into(),
            name: format!("node-{name_seed}"),
            summary: String::new(),
            attrs: json!({ "seed": name_seed }),
            group_id: String::new(),
        },
        Spec::AddEdge {
            edge,
            src,
            dst,
            valid_at,
        } => GraphOp::AddEdge {
            id: edge_id(*edge),
            src: node_id(*src),
            dst: node_id(*dst),
            rel: "R".into(),
            fact: format!("fact-{edge}"),
            attrs: json!({}),
            group_id: String::new(),
            valid_at: *valid_at,
            invalid_at: None,
        },
        Spec::AddEpisode { episode, mentions } => GraphOp::AddEpisode {
            id: episode_id(*episode),
            source: "test".into(),
            kind: String::new(),
            content: format!("episode-{episode}"),
            occurred_at: 1_500,
            group_id: String::new(),
            mentions: mentions.iter().map(|m| node_id(*m)).collect(),
        },
        Spec::Invalidate { edge, at } => GraphOp::InvalidateEdge {
            edge_id: edge_id(*edge),
            invalid_at: *at,
        },
        Spec::Merge { from, into_offset } => {
            // Force parity by device; ensure from != into.
            let from = (*from as u128 % POOL) as u8;
            let from = if device == "a" { from & !1 } else { from | 1 } % POOL as u8;
            let mut into = (from + 1 + (into_offset % (POOL as u8 - 1))) % POOL as u8;
            if into == from {
                into = (into + 1) % POOL as u8;
            }
            GraphOp::MergeNodes {
                from: node_id(from),
                into: node_id(into),
            }
        }
        Spec::PurgeNode { node } => GraphOp::Purge {
            ids: vec![node_id(*node)],
        },
        Spec::PurgeEdge { edge } => GraphOp::Purge {
            ids: vec![edge_id(*edge)],
        },
        Spec::UpdateNode {
            node,
            field_seed,
            value_seed,
        } => {
            let (name, summary, kind, attrs) = match field_seed % 4 {
                0 => (Some(format!("renamed-{value_seed}")), None, None, None),
                1 => (None, Some(format!("summary-{value_seed}")), None, None),
                2 => (None, None, Some(format!("kind-{value_seed}")), None),
                _ => (
                    Some(format!("renamed-{value_seed}")),
                    None,
                    None,
                    Some(json!({ "v": value_seed })),
                ),
            };
            GraphOp::UpdateNode {
                id: node_id(*node),
                name,
                summary,
                kind,
                attrs,
            }
        }
    }
}

/// Build the oplog entries one device would have produced. Wall times of the
/// two devices overlap (that's the point — the ops are concurrent).
fn entries(device: &str, specs: &[Spec]) -> Vec<OplogEntry> {
    let device_tag: u128 = if device == "a" {
        0xA0_0000_0000
    } else {
        0xB0_0000_0000
    };
    specs
        .iter()
        .enumerate()
        .map(|(i, spec)| OplogEntry {
            id: Uuid::from_u128(device_tag + i as u128),
            hlc: Hlc::new(1_000 + i as i64, 0, device),
            device_id: device.into(),
            op: to_op(spec, device),
        })
        .collect()
}

fn fresh_db(dir: &tempfile::TempDir, name: &str) -> Grit {
    let clock = Arc::new(ManualClock::new(50_000));
    Grit::open(dir.path().join(name), Options::new("local").clock(clock)).unwrap()
}

/// The graph-state fingerprint: everything except the oplog itself (whose
/// local seq/applied_at legitimately differ) — nodes, edges, episodes,
/// mentions, invalidation records, tombstones.
fn graph_dump(g: &Grit) -> String {
    let mut buf = Vec::new();
    g.export_jsonl(&mut buf).unwrap();
    let mut lines: Vec<&str> = std::str::from_utf8(&buf)
        .unwrap()
        .lines()
        .filter(|l| !l.contains("\"grit_export\"") && !l.contains("\"t\":\"oplog\""))
        .collect();
    lines.sort_unstable();
    lines.join("\n")
}

/// Riffle two sequences into one, preserving each device's internal order.
fn interleave(a: &[OplogEntry], b: &[OplogEntry], pattern: &[bool]) -> Vec<OplogEntry> {
    let (mut ai, mut bi) = (0, 0);
    let mut out = Vec::with_capacity(a.len() + b.len());
    for take_a in pattern {
        if *take_a && ai < a.len() {
            out.push(a[ai].clone());
            ai += 1;
        } else if bi < b.len() {
            out.push(b[bi].clone());
            bi += 1;
        }
    }
    out.extend_from_slice(&a[ai..]);
    out.extend_from_slice(&b[bi..]);
    out
}

/// Regression (found by audit, 2026-07-09): two devices invalidating the SAME
/// edge at the SAME event time collide on the (edge_id, invalid_at) PK; the
/// surviving recorded_at/hlc must not depend on apply order — recorded_at
/// drives as_at time travel, so first-writer-wins here meant two synced
/// devices answered historical queries differently.
#[test]
fn colliding_invalidations_keep_min_recording() {
    let dir = tempfile::tempdir().unwrap();
    let edge = edge_id(0);
    let mk = |id: u128, wall: i64, dev: &str, op: GraphOp| OplogEntry {
        id: Uuid::from_u128(id),
        hlc: Hlc::new(wall, 0, dev),
        device_id: dev.into(),
        op,
    };
    let setup = [
        mk(
            0xA1,
            1_000,
            "a",
            to_op(
                &Spec::AddNode {
                    node: 0,
                    name_seed: 0,
                },
                "a",
            ),
        ),
        mk(
            0xA2,
            1_001,
            "a",
            to_op(
                &Spec::AddNode {
                    node: 1,
                    name_seed: 1,
                },
                "a",
            ),
        ),
        mk(
            0xA3,
            1_002,
            "a",
            to_op(
                &Spec::AddEdge {
                    edge: 0,
                    src: 0,
                    dst: 1,
                    valid_at: None,
                },
                "a",
            ),
        ),
    ];
    // Same (edge, invalid_at), different recording walls.
    let inv_late = mk(
        0xA4,
        1_005,
        "a",
        GraphOp::InvalidateEdge {
            edge_id: edge,
            invalid_at: 2_000,
        },
    );
    let inv_early = mk(
        0xB1,
        1_003,
        "b",
        GraphOp::InvalidateEdge {
            edge_id: edge,
            invalid_at: 2_000,
        },
    );

    let g1 = fresh_db(&dir, "g1.db");
    for e in setup.iter().chain([&inv_late, &inv_early]) {
        g1.apply_remote(e.clone()).unwrap();
    }
    let g2 = fresh_db(&dir, "g2.db");
    for e in [&inv_early]
        .into_iter()
        .chain(setup.iter())
        .chain([&inv_late])
    {
        g2.apply_remote(e.clone()).unwrap();
    }
    let dump = graph_dump(&g1);
    assert_eq!(
        dump,
        graph_dump(&g2),
        "collision resolution diverged by apply order"
    );
    assert!(
        dump.contains("\"recorded_at\":1003"),
        "must keep the earliest recording:\n{dump}"
    );
}

/// UpdateNode's convergence corners, directed: an update arriving before its
/// node exists must fold when the node lands; a lower-HLC AddNode replay
/// rewrites row content and must not clobber an applied update; two devices
/// updating the same field converge to the highest HLC in every order.
#[test]
fn update_node_races_converge() {
    let dir = tempfile::tempdir().unwrap();
    let node = node_id(0);
    let mk = |id: u128, wall: i64, dev: &str, op: GraphOp| OplogEntry {
        id: Uuid::from_u128(id),
        hlc: Hlc::new(wall, 0, dev),
        device_id: dev.into(),
        op,
    };
    let add = mk(
        0xA1,
        1_000,
        "a",
        GraphOp::AddNode {
            id: node,
            kind: "k".into(),
            name: "base".into(),
            summary: "base summary".into(),
            attrs: json!({}),
            group_id: String::new(),
        },
    );
    // Same id minted on device b with a LOWER hlc — the replay case where
    // AddNode's lowest-wins upsert rewrites the whole row.
    let add_replay = mk(
        0xB1,
        900,
        "b",
        GraphOp::AddNode {
            id: node,
            kind: "k".into(),
            name: "replay-base".into(),
            summary: "replay summary".into(),
            attrs: json!({}),
            group_id: String::new(),
        },
    );
    let update_early = mk(
        0xA2,
        1_100,
        "a",
        GraphOp::UpdateNode {
            id: node,
            name: None,
            summary: Some("early revision".into()),
            kind: None,
            attrs: None,
        },
    );
    let update_late = mk(
        0xB2,
        1_200,
        "b",
        GraphOp::UpdateNode {
            id: node,
            name: Some("promoted".into()),
            summary: Some("late revision".into()),
            kind: None,
            attrs: None,
        },
    );

    let all = [&add, &add_replay, &update_early, &update_late];
    let orders: [[usize; 4]; 4] = [
        [0, 1, 2, 3], // adds, then updates
        [2, 3, 0, 1], // updates arrive before any node exists
        [0, 2, 3, 1], // add-replay lands last, after updates applied
        [3, 2, 1, 0], // fully reversed
    ];
    let mut dumps = Vec::new();
    for (i, order) in orders.iter().enumerate() {
        let g = fresh_db(&dir, &format!("update-race-{i}.db"));
        for &j in order {
            g.apply_remote(all[j].clone()).unwrap();
        }
        // Same-field race: highest HLC wins; untouched field keeps base
        // (which is the LOWEST-hlc AddNode by the creation-collision rule).
        let n = g.node(node).unwrap().unwrap();
        assert_eq!(n.summary, "late revision", "order {order:?}");
        assert_eq!(n.name, "promoted", "order {order:?}");
        assert_eq!(n.kind, "k", "order {order:?}");
        dumps.push(graph_dump(&g));
    }
    assert!(
        dumps.windows(2).all(|w| w[0] == w[1]),
        "update races diverged by apply order"
    );
}

proptest! {
    #![proptest_config(ProptestConfig { cases: 48, ..ProptestConfig::default() })]

    /// Any interleaving of two devices' sequences converges, and re-applying
    /// everything a second time (idempotency) changes nothing.
    #[test]
    fn interleavings_converge(
        specs_a in proptest::collection::vec(spec_strategy(), 1..10),
        specs_b in proptest::collection::vec(spec_strategy(), 1..10),
        pattern in proptest::collection::vec(any::<bool>(), 20),
    ) {
        let dir = tempfile::tempdir().unwrap();
        let a = entries("a", &specs_a);
        let b = entries("b", &specs_b);

        // Device 1 hears all of A, then all of B.
        let g1 = fresh_db(&dir, "g1.db");
        for entry in a.iter().chain(b.iter()) {
            g1.apply_remote(entry.clone()).unwrap();
        }

        // Device 2 hears an arbitrary riffle of the same ops.
        let g2 = fresh_db(&dir, "g2.db");
        for entry in interleave(&a, &b, &pattern) {
            g2.apply_remote(entry).unwrap();
        }

        let dump1 = graph_dump(&g1);
        prop_assert_eq!(&dump1, &graph_dump(&g2), "orders diverged");

        // Idempotency: replaying every op is a no-op, and reports as such.
        for entry in a.iter().chain(b.iter()) {
            prop_assert!(!g1.apply_remote(entry.clone()).unwrap());
        }
        prop_assert_eq!(&dump1, &graph_dump(&g1), "replay was not idempotent");
    }
}