//! Tests for [`super`].

use super::super::apply::apply_remote_changes;
use super::super::db::configure_connection;
use super::super::schema::{SyncSchema, SyncTable};
use super::*;
use rusqlite::Connection;

fn node(n: u128) -> DeviceId {
    DeviceId::new(Uuid::from_u128(n))
}

fn schema() -> SyncSchema {
    SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])])
}

fn server_order_schema() -> SyncSchema {
    schema().conflict_strategy(ConflictStrategy::ServerOrder)
}

/// A device: in-memory DB with the note table + migration, and a node id.
fn device(n: u128) -> (Connection, DeviceId) {
    let conn = Connection::open_in_memory().unwrap();
    configure_connection(&conn).unwrap();
    conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
        .unwrap();
    conn.execute_batch(&schema().migration_sql()).unwrap();
    (conn, node(n))
}

/// Make a local edit (domain write → trigger captures it), stamp it at
/// `now_ms`, and return it as a pulled change from `node` (as a peer would
/// receive it via the server).
fn local_edit_as_pulled(
    conn: &Connection,
    node: DeviceId,
    id: &str,
    name: &str,
    now_ms: i64,
    seq: i64,
) -> PulledChange {
    conn.execute(
        "INSERT INTO note (id, name) VALUES (?1, ?2) \
             ON CONFLICT(id) DO UPDATE SET name = excluded.name",
        (id, name),
    )
    .unwrap();
    stamp_pending(conn, node, now_ms).unwrap();
    let entry = load_local_pending(conn, node)
        .unwrap()
        .into_iter()
        .find(|e| e.row_id == id)
        .unwrap();
    PulledChange {
        storage_version: None,
        entry,
        device_id: node,
        seq,
    }
}

/// A bare entry for the collapse tests: only table, row_id, op and HLC
/// matter there, so the payload names the entry for the assertion message.
fn entry(row_id: &str, op: ChangeOp, wall_ms: i64, node_n: u128, label: &str) -> ChangeEntry {
    ChangeEntry {
        table: "note".into(),
        op,
        row_id: row_id.into(),
        timestamp: Utc::now(),
        hlc: Hlc {
            wall_ms,
            counter: 0,
            node: node(node_n),
        },
        data: Some(serde_json::json!({ "name": label })),
        extra: serde_json::Map::default(),
    }
}

fn labels(entries: &[ChangeEntry]) -> Vec<String> {
    entries
        .iter()
        .map(|e| e.data.as_ref().unwrap()["name"].as_str().unwrap().into())
        .collect()
}

/// The collapse keeps the highest HLC per row, and it has to do so whichever
/// order the entries arrive in. Both directions are asserted because the
/// obvious way to get this wrong, comparing in the wrong direction, is
/// invisible when only the already-sorted order is tested: it then keeps the
/// last entry, which is also the newest.
#[test]
fn collapse_keeps_the_highest_hlc_per_row_in_either_order() {
    let older = entry("r1", ChangeOp::Update, 100, 1, "older");
    let newer = entry("r1", ChangeOp::Update, 200, 1, "newer");

    let ascending = collapse_max_hlc(vec![older.clone(), newer.clone()]);
    assert_eq!(labels(&ascending), ["newer"], "newest lost, arriving last");

    let descending = collapse_max_hlc(vec![newer, older]);
    assert_eq!(
        labels(&descending),
        ["newer"],
        "newest lost, arriving first"
    );
}

/// Operation-agnostic: a newer delete beats an older edit and an older
/// delete loses to a newer edit. The HLC decides, never the operation.
#[test]
fn collapse_ignores_the_operation() {
    let newer_delete = collapse_max_hlc(vec![
        entry("r1", ChangeOp::Update, 100, 1, "edit"),
        entry("r1", ChangeOp::Delete, 200, 1, "delete"),
    ]);
    assert_eq!(newer_delete.len(), 1);
    assert_eq!(newer_delete[0].op, ChangeOp::Delete);

    let older_delete = collapse_max_hlc(vec![
        entry("r1", ChangeOp::Delete, 100, 1, "delete"),
        entry("r1", ChangeOp::Update, 200, 1, "edit"),
    ]);
    assert_eq!(older_delete.len(), 1);
    assert_eq!(older_delete[0].op, ChangeOp::Update);
}

/// The collapse is per row: distinct rows all survive, and first-seen order
/// is preserved, which is what the doc comment promises the apply engine.
#[test]
fn collapse_is_per_row_and_keeps_first_seen_order() {
    let out = collapse_max_hlc(vec![
        entry("r2", ChangeOp::Update, 100, 1, "r2-old"),
        entry("r1", ChangeOp::Update, 100, 1, "r1-only"),
        entry("r2", ChangeOp::Update, 200, 1, "r2-new"),
    ]);
    assert_eq!(labels(&out), ["r2-new", "r1-only"]);
}

/// The case the shared order exists for. Two changes for one row at an
/// exact HLC tie: the winner has to be the same on every device, and the
/// only thing every device agrees on is the payload bytes. Arrival order is
/// not that thing, so the two orders must agree here.
#[test]
fn collapse_breaks_an_exact_hlc_tie_on_payload_not_arrival_order() {
    let a = entry("r1", ChangeOp::Update, 100, 1, "aaa");
    let b = entry("r1", ChangeOp::Update, 100, 1, "bbb");
    assert_eq!(a.hlc, b.hlc, "the tie is the premise of this test");

    let forwards = collapse_max_hlc(vec![a.clone(), b.clone()]);
    let backwards = collapse_max_hlc(vec![b, a]);
    assert_eq!(
        labels(&forwards),
        labels(&backwards),
        "two devices disagreed at an exact tie because they saw the batch in different orders"
    );
    assert_eq!(labels(&forwards), ["bbb"], "higher payload bytes win");
}

fn note_name(conn: &Connection, id: &str) -> Option<String> {
    conn.query_row("SELECT name FROM note WHERE id = ?1", [id], |r| r.get(0))
        .optional()
        .unwrap()
}

fn pull_apply(conn: &mut Connection, s: &SyncSchema, node: DeviceId, pulled: Vec<PulledChange>) {
    let now = Utc::now();
    let resolved = resolve_pull(conn, s, node, pulled, now, "").unwrap();
    apply_remote_changes(conn, s, &resolved, "").unwrap();
    record_committed(conn, resolved.as_slice()).unwrap();
}

#[test]
fn stamp_pending_assigns_monotonic_hlcs() {
    let (conn, n) = device(1);
    conn.execute("INSERT INTO note (id, name) VALUES ('a', '1')", [])
        .unwrap();
    conn.execute("INSERT INTO note (id, name) VALUES ('b', '2')", [])
        .unwrap();
    assert_eq!(stamp_pending(&conn, n, 1000).unwrap(), 2);
    // Re-stamping is a no-op (both already stamped).
    assert_eq!(stamp_pending(&conn, n, 2000).unwrap(), 0);
    let stamps: Vec<(i64, i64)> = {
        let mut s = conn
            .prepare("SELECT hlc_wall, hlc_counter FROM sync_changelog ORDER BY id")
            .unwrap();
        s.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
            .unwrap()
            .map(|r| r.unwrap())
            .collect()
    };
    assert!(stamps.iter().all(|(w, _)| *w == 1000));
    assert_eq!(
        stamps[0].1 + 1,
        stamps[1].1,
        "counter increments within a wall_ms"
    );
}

#[test]
fn committed_ledger_advances_only() {
    let (conn, _) = device(1);
    let older = Hlc {
        wall_ms: 100,
        counter: 0,
        node: node(9),
    };
    let newer = Hlc {
        wall_ms: 200,
        counter: 0,
        node: node(9),
    };
    set_committed(&conn, "note", "r", &newer).unwrap();
    set_committed(&conn, "note", "r", &older).unwrap(); // must not regress
    assert_eq!(
        committed_hlc(&conn, "note", "r").unwrap().unwrap().wall_ms,
        200
    );
}

#[test]
fn conflicting_edits_converge_to_higher_hlc_both_directions() {
    // A edits at t=100, B edits at t=200 → B wins everywhere.
    let (mut a, an) = device(1);
    let (mut b, bn) = device(2);
    let a_change = local_edit_as_pulled(&a, an, "r", "from-A", 100, 1);
    let b_change = local_edit_as_pulled(&b, bn, "r", "from-B", 200, 1);

    pull_apply(&mut a, &schema(), an, vec![b_change]); // A pulls B (newer) → adopts B
    pull_apply(&mut b, &schema(), bn, vec![a_change]); // B pulls A (older) → keeps B

    assert_eq!(note_name(&a, "r").as_deref(), Some("from-B"));
    assert_eq!(note_name(&b, "r").as_deref(), Some("from-B"));
}

#[test]
fn gate_drops_repulled_older_change() {
    let (mut a, an) = device(1);
    let (b, bn) = device(2);
    // B's change is applied on A.
    let b_change = local_edit_as_pulled(&b, bn, "r", "v-old", 100, 1);
    pull_apply(&mut a, &schema(), an, vec![b_change.clone()]);
    // A then makes a NEWER local edit and commits it.
    let _a_new = local_edit_as_pulled(&a, an, "r", "v-new", 300, 2);
    // Mark A's edit committed (as a push would) so the gate has a committed clock.
    let a_pending = load_local_pending(&a, an);
    record_committed(&a, &a_pending.unwrap()).unwrap();
    // Re-pulling B's OLD change must be gated out (older than committed).
    let resolved = resolve_pull(&a, &schema(), an, vec![b_change], Utc::now(), "").unwrap();
    assert!(
        resolved.iter().all(|e| e.row_id != "r"),
        "stale re-pull must be gated"
    );
}

#[test]
fn newer_delete_beats_older_edit() {
    let (mut a, an) = device(1);
    let (b, bn) = device(2);
    // A has an older edit locally.
    local_edit_as_pulled(&a, an, "r", "edit", 100, 1);
    // B deletes the same row, newer.
    b.execute("INSERT INTO note (id, name) VALUES ('r', 'x')", [])
        .unwrap();
    stamp_pending(&b, bn, 50).unwrap();
    b.execute("DELETE FROM note WHERE id = 'r'", []).unwrap();
    stamp_pending(&b, bn, 200).unwrap();
    let del = load_local_pending(&b, bn)
        .unwrap()
        .into_iter()
        .find(|e| e.op == ChangeOp::Delete)
        .unwrap();
    let pulled = PulledChange {
        storage_version: None,
        entry: del,
        device_id: bn,
        seq: 2,
    };
    pull_apply(&mut a, &schema(), an, vec![pulled]);
    assert_eq!(
        note_name(&a, "r"),
        None,
        "newer delete wins over older edit"
    );
}

#[test]
fn server_order_applies_last_delivered_no_hlc() {
    let (mut a, an) = device(1);
    let s = server_order_schema();
    // Two pulled changes for the same row; server order = last wins, HLC ignored.
    let (src, sn) = device(2);
    let first = local_edit_as_pulled(&src, sn, "r", "first", 999, 1); // higher wall
    let (src2, sn2) = device(3);
    let second = local_edit_as_pulled(&src2, sn2, "r", "second", 1, 2); // lower wall, later seq
    let resolved = resolve_pull(&a, &s, an, vec![first, second], Utc::now(), "").unwrap();
    apply_remote_changes(&mut a, &s, &resolved, "").unwrap();
    assert_eq!(
        note_name(&a, "r").as_deref(),
        Some("second"),
        "server order: last delivered wins"
    );
}

/// The two guarantees `ResolvedChanges` documents, asserted on the same
/// input so the difference between them is the only variable. Under the HLC
/// strategy a batch carrying two changes for one row resolves to one entry;
/// under `ServerOrder` it deliberately stays two, because last-delivered-wins
/// is what an app choosing that strategy asked for.
#[test]
fn resolve_pull_collapses_a_row_under_hlc_and_does_not_under_server_order() {
    let (src, sn) = device(2);
    let first = local_edit_as_pulled(&src, sn, "r", "first", 100, 1);
    let second = local_edit_as_pulled(&src, sn, "r", "second", 200, 2);

    let (hlc_device, hn) = device(1);
    let under_hlc = resolve_pull(
        &hlc_device,
        &schema(),
        hn,
        vec![first.clone(), second.clone()],
        Utc::now(),
        "",
    )
    .unwrap();
    assert_eq!(
        under_hlc.len(),
        1,
        "the HLC strategy promises one entry per row; the apply order would \
             otherwise decide the value"
    );

    let (server_device, svn) = device(3);
    let under_server_order = resolve_pull(
        &server_device,
        &server_order_schema(),
        svn,
        vec![first, second],
        Utc::now(),
        "",
    )
    .unwrap();
    assert_eq!(
        under_server_order.len(),
        2,
        "ServerOrder must not collapse: last delivered wins is the strategy"
    );
}

/// A model of the pull pipeline.
///
/// Aimed at [`resolve_pull`] with a real `Connection`, deliberately, and not
/// at the pure conflict layer one step down. The one-entry-per-row invariant
/// only exists after the collapse, so `resolve_pull` is the lowest layer
/// where a max-HLC-wins specification is an honest thing to assert.
/// `CleanChanges::gated_at`, one layer down, promises only committed-clock
/// filtering, so three of these four properties do not hold there even
/// against correct code. See wiki `testing-posture`, Phase 3.
///
/// The four properties are the ones a sync engine lives or dies on, and each
/// is a different way for two devices to end up holding different bytes.
mod model {
    use super::*;
    use proptest::prelude::*;

    /// A fixed instant, so the poisoning guard is deterministic. Generated
    /// walls sit near it and well inside `MAX_HLC_DRIFT_MS`; poisoning is
    /// covered by its own example test, and letting it fire here would mean
    /// the properties were quietly asserting over an empty batch.
    const BASE_MS: i64 = 1_700_000_000_000;

    fn now() -> DateTime<Utc> {
        DateTime::from_timestamp_millis(BASE_MS).unwrap()
    }

    /// Three rows, three devices, three wall readings, and every range here
    /// is narrow on purpose.
    ///
    /// The interesting case is two changes for one row at an *exact* HLC
    /// tie with differing payloads, because that is the only case the
    /// payload tiebreak in `change_order` serves. A wider generator makes it
    /// unreachable: a first attempt drew walls from a 100ms window and
    /// produced roughly one tie across a whole 256-case run, few enough that
    /// deleting the tiebreak outright left every property passing. Ties have
    /// to be common for these properties to observe anything, so the clock
    /// is generated with almost no entropy in it and the payload carries the
    /// variation instead.
    fn batch() -> impl Strategy<Value = Vec<PulledChange>> {
        let one = (0u8..3, 0u8..3, 0i64..3, 0u32..2, 0u8..4).prop_map(
            |(row, dev, wall_off, counter, payload)| {
                let node = node(u128::from(dev));
                PulledChange {
                    storage_version: None,
                    entry: ChangeEntry {
                        table: "note".into(),
                        op: ChangeOp::Update,
                        row_id: format!("r{row}"),
                        timestamp: now(),
                        hlc: Hlc {
                            wall_ms: BASE_MS + wall_off,
                            counter,
                            node,
                        },
                        data: Some(serde_json::json!({
                            "id": format!("r{row}"),
                            "name": format!("v{payload}"),
                        })),
                        extra: serde_json::Map::default(),
                    },
                    device_id: node,
                    seq: 0,
                }
            },
        );
        proptest::collection::vec(one, 0..6)
    }

    /// The whole observable state of a device: what each row holds, and what
    /// the committed ledger says about it. Both matter. A pipeline that
    /// wrote the right value but recorded the wrong committed clock would
    /// gate its own next pull incorrectly, and comparing only the rows would
    /// not see it.
    fn state(conn: &Connection) -> Vec<(String, Option<String>, Option<Hlc>)> {
        ["r0", "r1", "r2"]
            .iter()
            .map(|r| {
                (
                    (*r).to_string(),
                    note_name(conn, r),
                    committed_hlc(conn, "note", r).unwrap(),
                )
            })
            .collect()
    }

    /// Run a batch through the real pipeline on a fresh device.
    fn pull(conn: &mut Connection, node: DeviceId, pulled: Vec<PulledChange>) {
        let s = schema();
        let resolved = resolve_pull(conn, &s, node, pulled, now(), "").unwrap();
        apply_remote_changes(conn, &s, &resolved, "").unwrap();
        record_committed(conn, resolved.as_slice()).unwrap();
    }

    /// The specification the pipeline is supposed to implement: per row, the
    /// highest wall clock wins, then the highest counter, then the highest
    /// device, then the highest payload bytes.
    ///
    /// Spelled out rather than delegated to `change_order`, and that is the
    /// whole point of it. An oracle that called `change_order` would agree
    /// with a broken `change_order`, which is the imitation-oracle failure
    /// from wiki `testing-posture` wearing a different hat: it was the first
    /// version of this function, and deleting the payload tiebreak left all
    /// four properties passing. This version fails when the rule changes,
    /// because it is a second statement of the rule rather than a reference
    /// to the first.
    fn expected_winner(pulled: &[PulledChange], row: &str) -> Option<String> {
        fn rank(e: &ChangeEntry) -> (i64, u32, Uuid, Vec<u8>) {
            (
                e.hlc.wall_ms,
                e.hlc.counter,
                e.hlc.node.as_uuid(),
                serde_json::to_vec(e.data.as_ref().unwrap()).unwrap(),
            )
        }
        pulled
            .iter()
            .map(|p| &p.entry)
            .filter(|e| e.row_id == row)
            .max_by_key(|e| rank(e))
            .map(|e| e.data.as_ref().unwrap()["name"].as_str().unwrap().into())
    }

    proptest! {
        /// **The pipeline implements max-HLC-wins.** The value each row ends
        /// up holding is the one from the change that wins under
        /// `change_order`, whatever else the batch contained.
        #[test]
        fn final_value_is_the_winner_under_change_order(pulled in batch()) {
            let (mut conn, n) = device(1);
            pull(&mut conn, n, pulled.clone());
            for row in ["r0", "r1", "r2"] {
                prop_assert_eq!(
                    note_name(&conn, row),
                    expected_winner(&pulled, row),
                    "row {} does not hold the winner",
                    row
                );
            }
        }

        /// **Batch order does not change the final state.** The server may
        /// deliver a batch in any order; two devices that see the same
        /// changes in different orders must agree afterwards. This is the
        /// property the whole change-ordering unification was for.
        #[test]
        fn order_within_a_batch_does_not_matter(pulled in batch()) {
            let (mut forwards, fnode) = device(1);
            pull(&mut forwards, fnode, pulled.clone());

            let mut reversed_batch = pulled;
            reversed_batch.reverse();
            let (mut backwards, bnode) = device(1);
            pull(&mut backwards, bnode, reversed_batch);

            prop_assert_eq!(
                state(&forwards),
                state(&backwards),
                "two devices diverged on batch order alone"
            );
        }

        /// **Replaying a batch is a no-op.** A pull that is retried, or a
        /// cursor that rewinds, must not change anything the first pass
        /// already settled. This is what the committed ledger exists for.
        #[test]
        fn replaying_a_batch_changes_nothing(pulled in batch()) {
            let (mut conn, n) = device(1);
            pull(&mut conn, n, pulled.clone());
            let after_first = state(&conn);
            pull(&mut conn, n, pulled);
            prop_assert_eq!(after_first, state(&conn), "a replayed batch moved the state");
        }

        /// **Committed clocks never go backwards.** The ledger is what gates
        /// stale re-pulls, so a regression there un-gates a change the device
        /// already superseded, and an old value overwrites a newer one.
        ///
        /// Two independent mechanisms hold this up, and this property
        /// observes the pair rather than either one: the committed-HLC gate
        /// drops a stale change before `set_committed` sees it, and
        /// `set_committed` refuses to regress even if one reaches it.
        /// Breaking either alone leaves this passing, which is what
        /// defense in depth means and is worth knowing before trusting a
        /// green run here; breaking both fails it. The individual clause in
        /// `set_committed` has its own example test,
        /// `committed_ledger_advances_only`.
        #[test]
        fn committed_clocks_only_advance(first in batch(), second in batch()) {
            let (mut conn, n) = device(1);
            pull(&mut conn, n, first);
            let before: Vec<Option<Hlc>> =
                state(&conn).into_iter().map(|(_, _, h)| h).collect();
            pull(&mut conn, n, second);
            let after: Vec<Option<Hlc>> =
                state(&conn).into_iter().map(|(_, _, h)| h).collect();

            for (before, after) in before.into_iter().zip(after) {
                match (before, after) {
                    (Some(b), Some(a)) => prop_assert!(
                        a >= b,
                        "committed clock went backwards: {:?} -> {:?}",
                        b, a
                    ),
                    (Some(b), None) => {
                        prop_assert!(false, "committed clock for a row disappeared: {:?}", b);
                    }
                    _ => {}
                }
            }
        }
    }
}

// ── Conflict stash ──
//
// LWW always discards one side; these pin that the discarded bytes are kept
// rather than dropped. The one that matters most is the superseded case,
// which never becomes a ConflictPair and so is invisible to resolve_lww.

#[derive(Debug, PartialEq)]
struct StashRow {
    table_name: String,
    row_id: String,
    scope: String,
    losing_side: String,
    losing_payload: Option<String>,
}

fn stash_rows(conn: &Connection) -> Vec<StashRow> {
    let mut stmt = conn
        .prepare(
            "SELECT table_name, row_id, scope, losing_side, losing_payload
                 FROM sync_conflict_stash ORDER BY id",
        )
        .unwrap();
    stmt.query_map([], |r| {
        Ok(StashRow {
            table_name: r.get(0)?,
            row_id: r.get(1)?,
            scope: r.get(2)?,
            losing_side: r.get(3)?,
            losing_payload: r.get(4)?,
        })
    })
    .unwrap()
    .map(|r| r.unwrap())
    .collect()
}

/// A remote change with an explicit HLC and payload, as a peer would send it.
fn remote_change(from: DeviceId, id: &str, name: &str, hlc: Hlc, seq: i64) -> PulledChange {
    PulledChange {
        storage_version: None,
        entry: ChangeEntry {
            table: "note".to_string(),
            op: ChangeOp::Update,
            row_id: id.to_string(),
            timestamp: Utc::now(),
            hlc,
            data: Some(serde_json::json!({ "id": id, "name": name })),
            extra: serde_json::Map::default(),
        },
        device_id: from,
        seq,
    }
}

/// Site 1: a newer remote change beats our pending edit, so our edit is the
/// one that vanishes from the row. It must be recoverable.
#[test]
fn stash_keeps_our_own_edit_when_the_remote_wins() {
    let (mut conn, n) = device(1);
    let peer = node(2);

    // Our pending edit, stamped early so it loses.
    conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
        .unwrap();
    stamp_pending(&conn, n, 1_000).unwrap();

    let newer = Hlc {
        wall_ms: 9_000,
        counter: 0,
        node: peer,
    };
    let resolved = resolve_pull(
        &conn,
        &schema(),
        n,
        vec![remote_change(peer, "n1", "theirs", newer, 1)],
        Utc::now(),
        "",
    )
    .unwrap();
    apply_remote_changes(&mut conn, &schema(), &resolved, "").unwrap();

    assert_eq!(note_name(&conn, "n1").as_deref(), Some("theirs"));
    let rows = stash_rows(&conn);
    assert_eq!(
        rows.len(),
        1,
        "our discarded edit must be stashed: {rows:?}"
    );
    assert_eq!(rows[0].losing_side, "local");
    assert_eq!(rows[0].row_id, "n1");
    assert!(
        rows[0].losing_payload.as_deref().unwrap().contains("mine"),
        "the stash must hold the discarded value, not a placeholder: {rows:?}"
    );
}

/// Site 2: our pending edit wins, so the other writer's change is discarded.
/// Their bytes are the ones that need keeping.
#[test]
fn stash_keeps_the_remote_edit_when_ours_wins() {
    let (conn, n) = device(1);
    let peer = node(2);

    conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
        .unwrap();
    stamp_pending(&conn, n, 9_000).unwrap();

    let older = Hlc {
        wall_ms: 1_000,
        counter: 0,
        node: peer,
    };
    let resolved = resolve_pull(
        &conn,
        &schema(),
        n,
        vec![remote_change(peer, "n1", "theirs", older, 1)],
        Utc::now(),
        "",
    )
    .unwrap();
    assert!(
        resolved.is_empty(),
        "the older remote change must not apply"
    );

    let rows = stash_rows(&conn);
    assert_eq!(
        rows.len(),
        1,
        "their discarded edit must be stashed: {rows:?}"
    );
    assert_eq!(rows[0].losing_side, "remote");
    assert!(
        rows[0]
            .losing_payload
            .as_deref()
            .unwrap()
            .contains("theirs")
    );
}

/// Site 3, the quiet one: no local pending edit contests the row, so no
/// ConflictPair is ever built. The change is dropped by the committed-HLC
/// gate alone, which is why a stash wired only into the match arms misses it.
#[test]
fn stash_keeps_a_change_superseded_by_the_committed_clock() {
    let (mut conn, n) = device(1);
    let peer = node(2);

    // Apply and commit a newer value, with nothing left pending afterwards.
    let newer = Hlc {
        wall_ms: 9_000,
        counter: 0,
        node: peer,
    };
    let resolved = resolve_pull(
        &conn,
        &schema(),
        n,
        vec![remote_change(peer, "n1", "current", newer, 1)],
        Utc::now(),
        "",
    )
    .unwrap();
    apply_remote_changes(&mut conn, &schema(), &resolved, "").unwrap();
    record_committed(&conn, resolved.as_slice()).unwrap();
    assert!(stash_rows(&conn).is_empty(), "nothing lost yet");

    // Now pull an older change for the same row. No pending edit contests it.
    let older = Hlc {
        wall_ms: 1_000,
        counter: 0,
        node: peer,
    };
    let resolved = resolve_pull(
        &conn,
        &schema(),
        n,
        vec![remote_change(peer, "n1", "stale", older, 2)],
        Utc::now(),
        "",
    )
    .unwrap();
    assert!(resolved.is_empty(), "the superseded change must not apply");
    assert_eq!(note_name(&conn, "n1").as_deref(), Some("current"));

    let rows = stash_rows(&conn);
    assert_eq!(
        rows.len(),
        1,
        "the superseded change must be stashed: {rows:?}"
    );
    assert_eq!(rows[0].losing_side, "remote");
    assert!(rows[0].losing_payload.as_deref().unwrap().contains("stale"));
}

/// An echo carries the same bytes as the winner, so nothing was lost. Without
/// this the stash fills with no-ops and stops being worth reading.
#[test]
fn stash_ignores_a_conflict_whose_payloads_are_identical() {
    let (conn, n) = device(1);
    let peer = node(2);

    conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'same')", [])
        .unwrap();
    stamp_pending(&conn, n, 1_000).unwrap();
    let local = load_local_pending(&conn, n).unwrap().pop().unwrap();

    // Byte-identical payload, newer clock: the remote wins and our identical
    // value goes away, which costs nothing.
    let mut remote = remote_change(
        peer,
        "n1",
        "x",
        Hlc {
            wall_ms: 9_000,
            counter: 0,
            node: peer,
        },
        1,
    );
    remote.entry.data = local.data.clone();

    resolve_pull(&conn, &schema(), n, vec![remote], Utc::now(), "").unwrap();
    assert!(
        stash_rows(&conn).is_empty(),
        "an identical payload is not a lost edit: {:?}",
        stash_rows(&conn)
    );
}

/// A clock-poisoned entry is refused as hostile, not outvoted. Stashing it
/// would hand whoever sent it a way to fill the stash.
#[test]
fn stash_ignores_a_clock_poisoned_drop() {
    let (conn, n) = device(1);
    let peer = node(2);
    let now = Utc::now();
    let poisoned = Hlc {
        wall_ms: now.timestamp_millis() + crate::conflict::MAX_HLC_DRIFT_MS * 10,
        counter: 0,
        node: peer,
    };

    let resolved = resolve_pull(
        &conn,
        &schema(),
        n,
        vec![remote_change(peer, "n1", "from the future", poisoned, 1)],
        now,
        "",
    )
    .unwrap();
    assert!(resolved.is_empty(), "a poisoned entry must not apply");
    assert!(
        stash_rows(&conn).is_empty(),
        "a poisoned entry is refused, not outvoted: {:?}",
        stash_rows(&conn)
    );
}

/// The scope a loss happened in is recorded, so a consuming app can tell a
/// personal conflict from one inside a shared group.
#[test]
fn stash_records_the_scope() {
    let (conn, n) = device(1);
    let peer = node(2);

    conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
        .unwrap();
    stamp_pending(&conn, n, 9_000).unwrap();

    let older = Hlc {
        wall_ms: 1_000,
        counter: 0,
        node: peer,
    };
    resolve_pull(
        &conn,
        &schema(),
        n,
        vec![remote_change(peer, "n1", "theirs", older, 1)],
        Utc::now(),
        "group-7",
    )
    .unwrap();

    assert_eq!(stash_rows(&conn)[0].scope, "group-7");
}

/// ServerOrder does not compare versions, so it has no loser to name.
#[test]
fn server_order_stashes_nothing() {
    let (mut conn, n) = device(1);
    let peer = node(2);
    let s = server_order_schema();

    conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'mine')", [])
        .unwrap();
    stamp_pending(&conn, n, 9_000).unwrap();

    let older = Hlc {
        wall_ms: 1_000,
        counter: 0,
        node: peer,
    };
    let resolved = resolve_pull(
        &conn,
        &s,
        n,
        vec![remote_change(peer, "n1", "theirs", older, 1)],
        Utc::now(),
        "",
    )
    .unwrap();
    apply_remote_changes(&mut conn, &s, &resolved, "").unwrap();

    assert!(stash_rows(&conn).is_empty());
}

// ── Field merge ──
//
// The merge semantics themselves are pinned in `conflict.rs`; what is new
// here is the wiring, which is where every one of these can go wrong
// independently of a correct merge: the base has to be recorded at the right
// moments, read back for the right tables, refused for counters, and left
// entirely alone for a table that did not opt in.

/// A four-column row, which is the point: a merge is only interesting when a
/// row has more columns than two devices are likely to both touch. `minutes`
/// is the counter.
const CARD_COLS: &[&str] = &["id", "title", "due", "minutes"];

fn card_schema(merges: bool) -> SyncSchema {
    let table = SyncTable::full("card", CARD_COLS);
    SyncSchema::new(vec![if merges {
        table.field_merge(&["minutes"])
    } else {
        table
    }])
}

fn card_device(n: u128, merges: bool) -> (Connection, DeviceId) {
    let conn = Connection::open_in_memory().unwrap();
    configure_connection(&conn).unwrap();
    conn.execute_batch(
        "CREATE TABLE card (id TEXT PRIMARY KEY, title TEXT, due TEXT, minutes INTEGER);",
    )
    .unwrap();
    conn.execute_batch(&card_schema(merges).migration_sql())
        .unwrap();
    (conn, node(n))
}

/// A wall reading to hang a card test's clocks off.
///
/// Real time, not a toy constant, because `resolve_pull` advances the local
/// clock to `now` on every pull it observes: an HLC of `2_000` would sit
/// decades below the clock it is meant to be newer than, and every one of
/// these tests would silently assert over a conflict that never happened.
fn card_t0() -> i64 {
    Utc::now().timestamp_millis()
}

/// A remote change for the card table with an explicit payload and wall clock.
fn card_change(from: DeviceId, wall_ms: i64, data: serde_json::Value) -> PulledChange {
    PulledChange {
        storage_version: None,
        entry: ChangeEntry {
            table: "card".into(),
            op: ChangeOp::Update,
            row_id: "c1".into(),
            timestamp: Utc::now(),
            hlc: Hlc {
                wall_ms,
                counter: 0,
                node: from,
            },
            data: Some(data),
            extra: serde_json::Map::default(),
        },
        device_id: from,
        seq: 1,
    }
}

/// The state both devices start from, established the way it really is: a
/// remote change applied and committed, which is also what records the base.
fn seed_base(conn: &mut Connection, node: DeviceId, peer: DeviceId, merges: bool, t0: i64) {
    let s = card_schema(merges);
    let base = card_change(
        peer,
        t0,
        serde_json::json!({"id": "c1", "title": "base", "due": null, "minutes": 0}),
    );
    pull_apply_with(conn, &s, node, vec![base]);
}

fn pull_apply_with(
    conn: &mut Connection,
    s: &SyncSchema,
    node: DeviceId,
    pulled: Vec<PulledChange>,
) {
    let resolved = resolve_pull(conn, s, node, pulled, Utc::now(), "").unwrap();
    apply_remote_changes(conn, s, &resolved, "").unwrap();
    record_committed(conn, resolved.as_slice()).unwrap();
}

fn card(conn: &Connection) -> (Option<String>, Option<String>, Option<i64>) {
    conn.query_row(
        "SELECT title, due, minutes FROM card WHERE id = 'c1'",
        [],
        |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
    )
    .unwrap()
}

/// The case the whole feature exists for. Two devices edit different columns
/// of one row; both edits survive, where LWW would have discarded one whole
/// edit for touching a row it never contested.
#[test]
fn opted_in_table_merges_edits_to_different_fields() {
    let (mut conn, n) = card_device(1, true);
    let peer = node(2);
    let t0 = card_t0();
    seed_base(&mut conn, n, peer, true, t0);

    // Local: set the due date. Remote, newer: retitle.
    conn.execute("UPDATE card SET due = '2026-09-01' WHERE id = 'c1'", [])
        .unwrap();
    stamp_pending(&conn, n, t0 + 1_000).unwrap();
    let remote = card_change(
        peer,
        t0 + 2_000,
        serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
    );

    pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);

    let (title, due, _) = card(&conn);
    assert_eq!(title.as_deref(), Some("theirs"), "remote's field was lost");
    assert_eq!(
        due.as_deref(),
        Some("2026-09-01"),
        "the local edit was discarded for contesting a field it never touched"
    );
    assert!(
        stash_rows(&conn).is_empty(),
        "a merge that contested nothing lost nothing, so it must not stash: {:?}",
        stash_rows(&conn)
    );
}

/// A field both sides moved is a real contest. The merge hands it to the HLC
/// winner (the same rule LWW would apply), and the loser's version is stashed,
/// because a value did go away.
#[test]
fn a_contested_field_goes_to_the_hlc_winner_and_stashes_the_loser() {
    let (mut conn, n) = card_device(1, true);
    let peer = node(2);
    let t0 = card_t0();
    seed_base(&mut conn, n, peer, true, t0);

    conn.execute(
        "UPDATE card SET title = 'mine', due = '2026-09-01' WHERE id = 'c1'",
        [],
    )
    .unwrap();
    stamp_pending(&conn, n, t0 + 1_000).unwrap();
    let remote = card_change(
        peer,
        t0 + 2_000,
        serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
    );

    pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);

    let (title, due, _) = card(&conn);
    assert_eq!(
        title.as_deref(),
        Some("theirs"),
        "the contested field must go to the newer HLC"
    );
    assert_eq!(
        due.as_deref(),
        Some("2026-09-01"),
        "an uncontested field must survive even when the same row lost a contest"
    );

    let rows = stash_rows(&conn);
    assert_eq!(
        rows.len(),
        1,
        "the losing title must be recoverable: {rows:?}"
    );
    assert_eq!(rows[0].losing_side, "local");
    assert!(rows[0].losing_payload.as_deref().unwrap().contains("mine"));
}

/// The failure a value merge cannot see. Two devices each log thirty minutes;
/// merging takes one side's absolute total and the other half-hour is gone
/// with nothing recording that it existed. Refusing to merge is worse for the
/// uncontested fields and better for the truth: LWW discards one whole edit
/// and stashes it, so the loss stays visible and recoverable.
#[test]
fn a_contested_counter_refuses_to_merge_and_falls_back_to_lww() {
    let (mut conn, n) = card_device(1, true);
    let peer = node(2);
    let t0 = card_t0();
    seed_base(&mut conn, n, peer, true, t0);

    // Both sides increment `minutes`, and each also moves a field the other
    // did not, so a merge would visibly have kept both.
    conn.execute(
        "UPDATE card SET minutes = minutes + 30, due = '2026-09-01' WHERE id = 'c1'",
        [],
    )
    .unwrap();
    stamp_pending(&conn, n, t0 + 1_000).unwrap();
    let remote = card_change(
        peer,
        t0 + 2_000,
        serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 30}),
    );

    pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);

    let (title, due, minutes) = card(&conn);
    assert_eq!(minutes, Some(30), "a merged counter would still read 30");
    assert_eq!(title.as_deref(), Some("theirs"));
    assert_eq!(
        due, None,
        "the row must hold one side whole, not a merge of both"
    );

    let rows = stash_rows(&conn);
    assert_eq!(
        rows.len(),
        1,
        "the discarded half-hour must stay visible in the stash: {rows:?}"
    );
    assert!(
        rows[0]
            .losing_payload
            .as_deref()
            .unwrap()
            .contains("2026-09-01"),
        "the stash must hold the whole discarded edit: {rows:?}"
    );
}

/// A counter only one side moved is not the counter problem: nothing has to
/// be reconstructed, so the merge runs and that side's value carries across.
/// Without this the counter declaration would cost every row on the table its
/// merge, which is most of the feature.
#[test]
fn an_uncontested_counter_does_not_block_the_merge() {
    let (mut conn, n) = card_device(1, true);
    let peer = node(2);
    let t0 = card_t0();
    seed_base(&mut conn, n, peer, true, t0);

    conn.execute("UPDATE card SET minutes = minutes + 30 WHERE id = 'c1'", [])
        .unwrap();
    stamp_pending(&conn, n, t0 + 1_000).unwrap();
    let remote = card_change(
        peer,
        t0 + 2_000,
        serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
    );

    pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);

    let (title, _, minutes) = card(&conn);
    assert_eq!(
        minutes,
        Some(30),
        "an uncontested counter must carry across"
    );
    assert_eq!(title.as_deref(), Some("theirs"));
}

/// The default must be no change in behaviour. The identical conflict on a
/// table that did not opt in resolves the way it always has: one whole edit
/// wins, the other is stashed.
#[test]
fn a_table_that_did_not_opt_in_still_resolves_by_lww() {
    let (mut conn, n) = card_device(1, false);
    let peer = node(2);
    let t0 = card_t0();
    seed_base(&mut conn, n, peer, false, t0);

    conn.execute("UPDATE card SET due = '2026-09-01' WHERE id = 'c1'", [])
        .unwrap();
    stamp_pending(&conn, n, t0 + 1_000).unwrap();
    let remote = card_change(
        peer,
        t0 + 2_000,
        serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
    );

    pull_apply_with(&mut conn, &card_schema(false), n, vec![remote]);

    let (title, due, _) = card(&conn);
    assert_eq!(title.as_deref(), Some("theirs"));
    assert_eq!(
        due, None,
        "without the opt-in the whole remote edit wins, exactly as before"
    );
    assert_eq!(stash_rows(&conn).len(), 1, "and the loser is still stashed");

    let bases: i64 = conn
        .query_row("SELECT COUNT(*) FROM sync_row_snapshot", [], |r| r.get(0))
        .unwrap();
    assert_eq!(bases, 0, "a table that did not opt in must store no base");
}

/// A row whose base was never recorded, or was dropped by a delete. There is
/// nothing to merge against, so the conflict falls through to LWW rather than
/// merging against a base it invented.
#[test]
fn a_missing_base_falls_back_to_lww_without_failing() {
    let (mut conn, n) = card_device(1, true);
    let peer = node(2);
    let t0 = card_t0();
    seed_base(&mut conn, n, peer, true, t0);
    conn.execute("DELETE FROM sync_row_snapshot", []).unwrap();

    conn.execute("UPDATE card SET due = '2026-09-01' WHERE id = 'c1'", [])
        .unwrap();
    stamp_pending(&conn, n, t0 + 1_000).unwrap();
    let remote = card_change(
        peer,
        t0 + 2_000,
        serde_json::json!({"id": "c1", "title": "theirs", "due": null, "minutes": 0}),
    );

    pull_apply_with(&mut conn, &card_schema(true), n, vec![remote]);

    let (title, due, _) = card(&conn);
    assert_eq!(title.as_deref(), Some("theirs"));
    assert_eq!(due, None, "with no base the whole remote edit wins");
    assert_eq!(stash_rows(&conn).len(), 1);
}

/// Applying a remote change re-bases the row, so the *next* conflict measures
/// both sides against what this device last received rather than against the
/// version it first saw. Without this the base would go stale and every later
/// merge would report fields as changed that nobody touched.
#[test]
fn applying_a_remote_change_rebases_the_row() {
    let (mut conn, n) = card_device(1, true);
    let peer = node(2);
    let t0 = card_t0();
    seed_base(&mut conn, n, peer, true, t0);

    pull_apply_with(
        &mut conn,
        &card_schema(true),
        n,
        vec![card_change(
            peer,
            t0 + 2_000,
            serde_json::json!({"id": "c1", "title": "second", "due": null, "minutes": 0}),
        )],
    );

    let base = super::snapshot::load(&conn, "card", "c1");
    assert_eq!(
        base["title"], "second",
        "the base must track the latest applied version, not the first"
    );
}

/// A delete drops the base with the row. A stale base would otherwise be
/// handed to a merge for whatever next occupies the key, describing a version
/// of a different row.
#[test]
fn deleting_a_row_drops_its_base() {
    let (mut conn, n) = card_device(1, true);
    let peer = node(2);
    let t0 = card_t0();
    seed_base(&mut conn, n, peer, true, t0);
    assert!(super::snapshot::load(&conn, "card", "c1").is_object());

    let mut delete = card_change(peer, t0 + 4_000, serde_json::json!({"id": "c1"}));
    delete.entry.op = ChangeOp::Delete;
    pull_apply_with(&mut conn, &card_schema(true), n, vec![delete]);

    assert_eq!(
        super::snapshot::load(&conn, "card", "c1"),
        serde_json::Value::Null
    );
}

// ── Dependent columns ──

/// A row whose `state` carries a `state_at` derived from it, the shape
/// GoingsOn's `status`/`completed_at` has.
fn dep_schema() -> SyncSchema {
    SyncSchema::new(vec![
        SyncTable::full("job", &["id", "state", "state_at", "note"])
            .field_merge(&[])
            .dependent_columns(&[&["state", "state_at"]]),
    ])
}

fn dep_device(n: u128) -> (Connection, DeviceId) {
    let conn = Connection::open_in_memory().unwrap();
    configure_connection(&conn).unwrap();
    conn.execute_batch(
        "CREATE TABLE job (id TEXT PRIMARY KEY, state TEXT, state_at TEXT, note TEXT);",
    )
    .unwrap();
    conn.execute_batch(&dep_schema().migration_sql()).unwrap();
    (conn, node(n))
}

fn job_change(from: DeviceId, wall_ms: i64, data: serde_json::Value) -> PulledChange {
    PulledChange {
        storage_version: None,
        entry: ChangeEntry {
            table: "job".into(),
            op: ChangeOp::Update,
            row_id: "j1".into(),
            timestamp: Utc::now(),
            hlc: Hlc {
                wall_ms,
                counter: 0,
                node: from,
            },
            data: Some(data),
            extra: serde_json::Map::default(),
        },
        device_id: from,
        seq: 1,
    }
}

fn job(conn: &Connection) -> (Option<String>, Option<String>, Option<String>) {
    conn.query_row(
        "SELECT state, state_at, note FROM job WHERE id = 'j1'",
        [],
        |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
    )
    .unwrap()
}

/// The failure the declaration exists for, and the reason it cannot be fixed
/// in the app: one device moves the row to `started` (leaving `state_at`
/// alone), the other to `done` (stamping it). Only the second moved
/// `state_at`, so a column-by-column merge treats it as uncontested and keeps
/// it whichever way `state` falls, producing a started job with a completion
/// time. The group forces both from one side.
#[test]
fn a_contested_dependent_group_is_taken_whole_from_one_side() {
    for (local_newer, expect) in [(true, ("started", None)), (false, ("done", Some("T")))] {
        let (mut conn, n) = dep_device(1);
        let peer = node(2);
        let t0 = card_t0();
        let s = dep_schema();

        pull_apply_with(
            &mut conn,
            &s,
            n,
            vec![job_change(
                peer,
                t0,
                serde_json::json!({"id": "j1", "state": "pending", "state_at": null, "note": "n"}),
            )],
        );

        // Local moves state only; remote moves state and its derived stamp.
        conn.execute("UPDATE job SET state = 'started' WHERE id = 'j1'", [])
            .unwrap();
        let (local_ms, remote_ms) = if local_newer {
            (t0 + 3_000, t0 + 2_000)
        } else {
            (t0 + 1_000, t0 + 2_000)
        };
        stamp_pending(&conn, n, local_ms).unwrap();
        pull_apply_with(
            &mut conn,
            &s,
            n,
            vec![job_change(
                peer,
                remote_ms,
                serde_json::json!({"id": "j1", "state": "done", "state_at": "T", "note": "n"}),
            )],
        );

        let (state, state_at, _) = job(&conn);
        assert_eq!(
            (state.as_deref(), state_at.as_deref()),
            (Some(expect.0), expect.1),
            "the group must land as one device's version of it (local_newer = {local_newer})"
        );
    }
}

/// The same split by the other route, and the one that was briefly wrong.
///
/// Here no single column is contested: one device moves `state`, the other
/// moves only `state_at`. A field-by-field pass sees two disjoint changes,
/// merges both, and lands one column from each device, which is the same
/// broken pair as the contested case. So the trigger is both sides having
/// touched the group *anywhere*, not both having touched the same column.
#[test]
fn a_group_touched_by_both_sides_on_different_columns_is_still_taken_whole() {
    let (mut conn, n) = dep_device(1);
    let peer = node(2);
    let t0 = card_t0();
    let s = dep_schema();

    pull_apply_with(
        &mut conn,
        &s,
        n,
        vec![job_change(
            peer,
            t0,
            serde_json::json!({"id": "j1", "state": "pending", "state_at": null, "note": "n"}),
        )],
    );

    // Local moves only `state`; remote moves only `state_at`. Disjoint.
    conn.execute("UPDATE job SET state = 'started' WHERE id = 'j1'", [])
        .unwrap();
    stamp_pending(&conn, n, t0 + 3_000).unwrap();
    pull_apply_with(
        &mut conn,
        &s,
        n,
        vec![job_change(
            peer,
            t0 + 2_000,
            serde_json::json!({"id": "j1", "state": "pending", "state_at": "T", "note": "n"}),
        )],
    );

    let (state, state_at, _) = job(&conn);
    assert_eq!(
        (state.as_deref(), state_at.as_deref()),
        (Some("started"), None),
        "the group took one column from each device: a state neither one held"
    );
}

/// The group only fires when it is contested. A device that moves the group
/// while the other moves an unrelated column still gets a merge, which is the
/// whole reason the table opted in.
#[test]
fn an_uncontested_dependent_group_still_merges() {
    let (mut conn, n) = dep_device(1);
    let peer = node(2);
    let t0 = card_t0();
    let s = dep_schema();

    pull_apply_with(
        &mut conn,
        &s,
        n,
        vec![job_change(
            peer,
            t0,
            serde_json::json!({"id": "j1", "state": "pending", "state_at": null, "note": "n"}),
        )],
    );

    // Local edits the unrelated column; only remote touches the group.
    conn.execute("UPDATE job SET note = 'mine' WHERE id = 'j1'", [])
        .unwrap();
    stamp_pending(&conn, n, t0 + 1_000).unwrap();
    pull_apply_with(
        &mut conn,
        &s,
        n,
        vec![job_change(
            peer,
            t0 + 2_000,
            serde_json::json!({"id": "j1", "state": "done", "state_at": "T", "note": "n"}),
        )],
    );

    let (state, state_at, note) = job(&conn);
    assert_eq!(
        (state.as_deref(), state_at.as_deref()),
        (Some("done"), Some("T")),
        "an uncontested group must carry across intact"
    );
    assert_eq!(
        note.as_deref(),
        Some("mine"),
        "declaring a group must not cost the table its merge on other columns"
    );
}

/// The stash is bounded. Unbounded, a pathological sync loop grows it without
/// limit.
#[test]
fn stash_is_trimmed_to_its_ceiling() {
    let (conn, _) = device(1);
    for i in 0..(super::stash::MAX_STASH_ROWS + 50) {
        conn.execute(
            "INSERT INTO sync_conflict_stash
                   (table_name, row_id, losing_side, losing_payload, losing_hlc, losing_device, winning_hlc)
                 VALUES ('note', ?1, 'remote', '{}', '1:0:x', 'dev', '2:0:y')",
            [i.to_string()],
        )
        .unwrap();
    }
    super::stash::trim_stash(&conn).unwrap();

    let kept: i64 = conn
        .query_row("SELECT COUNT(*) FROM sync_conflict_stash", [], |r| r.get(0))
        .unwrap();
    assert_eq!(kept, super::stash::MAX_STASH_ROWS);

    // The newest survive: the oldest losses are the least likely to be acted on.
    let oldest_kept: String = conn
        .query_row(
            "SELECT row_id FROM sync_conflict_stash ORDER BY id ASC LIMIT 1",
            [],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(oldest_kept, "50");
}

/// `stamp_pending` has to leave the advanced clock on disk, not only on the
/// changelog rows. The wall component is high (5_000_000) and the row count
/// is three, so the persisted clock is a value no fresh `Hlc::zero` can
/// coincide with, and the second stamp runs at a LOWER now_ms (1000) so a
/// clock that failed to persist would visibly restart at 1000 instead of
/// continuing the counter at 5_000_000.
#[test]
fn stamp_pending_persists_the_advanced_clock() {
    let (conn, n) = device(1);
    for id in ["r1", "r2", "r3"] {
        conn.execute("INSERT INTO note (id, name) VALUES (?1, 'v')", [id])
            .unwrap();
    }
    assert_eq!(stamp_pending(&conn, n, 5_000_000).unwrap(), 3);

    // Three ticks at one wall component: adopt 5_000_000 with counter 0,
    // then bump twice.
    assert_eq!(
        load_clock(&conn, n).unwrap(),
        Hlc {
            wall_ms: 5_000_000,
            counter: 2,
            node: n
        },
        "the clock reached by stamping must survive a reload"
    );

    conn.execute("INSERT INTO note (id, name) VALUES ('r4', 'v')", [])
        .unwrap();
    assert_eq!(stamp_pending(&conn, n, 1000).unwrap(), 1);
    let r4 = load_local_pending(&conn, n)
        .unwrap()
        .into_iter()
        .find(|e| e.row_id == "r4")
        .unwrap();
    assert_eq!(
        r4.hlc,
        Hlc {
            wall_ms: 5_000_000,
            counter: 3,
            node: n
        },
        "a later stamp at an earlier now_ms must continue the reloaded clock"
    );
}

/// `observe` has to leave the merged clock on disk: the whole point is that
/// a subsequent local write outranks the remote it observed. The remote sits
/// far in the future (9_000_000_000) and the local stamp that follows runs
/// at now_ms 2000, so if the merge were not persisted the new stamp would
/// land at wall 2000 and lose to the remote by seven orders of magnitude.
#[test]
fn observe_persists_the_merged_clock() {
    let (conn, n) = device(1);
    let remote = Hlc {
        wall_ms: 9_000_000_000,
        counter: 5,
        node: node(2),
    };
    observe(&conn, n, [remote], 1000).unwrap();
    assert_eq!(
        load_clock(&conn, n).unwrap(),
        Hlc {
            wall_ms: 9_000_000_000,
            counter: 6,
            node: n
        },
        "observe must persist the remote wall and a strictly greater counter"
    );

    conn.execute("INSERT INTO note (id, name) VALUES ('r', 'v')", [])
        .unwrap();
    assert_eq!(stamp_pending(&conn, n, 2000).unwrap(), 1);
    let stamped = load_local_pending(&conn, n).unwrap().remove(0).hlc;
    assert_eq!(
        stamped,
        Hlc {
            wall_ms: 9_000_000_000,
            counter: 7,
            node: n
        }
    );
    assert!(
        stamped > remote,
        "a local write after observing must outrank the observed remote"
    );
}