//! Tests for [`super`].

use super::super::db::get_sync_state_or;
use super::super::schema::SyncTable;
use super::*;
use std::sync::{Arc, Mutex};

/// A shared in-memory "server": an append-only personal log of (origin
/// device, entry), plus a group log of (group, origin device, entry).
#[derive(Clone, Default)]
struct FakeServer {
    log: Arc<Mutex<Vec<(DeviceId, ChangeEntry)>>>,
    group_log: Arc<Mutex<Vec<(GroupId, DeviceId, ChangeEntry)>>>,
    /// The storage version a pulled page appears to have been sealed under.
    /// Stands in for the `__sksv` the real transport reads out of the
    /// envelope; the fake log holds decrypted entries, so there is no
    /// envelope here to carry it.
    peer_version: Arc<Mutex<Option<u32>>>,
}

impl SyncTransport for FakeServer {
    fn group_scope_push(
        &self,
        group_id: GroupId,
        _gck_version: i32,
        device_id: DeviceId,
        changes: Vec<ChangeEntry>,
    ) -> impl Future<Output = Result<i64>> + Send {
        let group_log = self.group_log.clone();
        async move {
            let mut l = group_log.lock().unwrap();
            for c in changes {
                l.push((group_id, device_id, c));
            }
            Ok(l.len() as i64)
        }
    }

    async fn register_device(&self, _name: &str, _platform: &str) -> Result<Device> {
        Ok(Device {
            id: DeviceId::new(uuid::Uuid::from_u128(0xDE)),
            app_id: crate::ids::AppId::nil(),
            user_id: crate::ids::UserId::nil(),
            device_name: "fake".into(),
            platform: "test".into(),
            last_seen_at: Utc::now(),
            created_at: Utc::now(),
        })
    }

    fn push(
        &self,
        device_id: DeviceId,
        changes: Vec<ChangeEntry>,
    ) -> impl Future<Output = Result<i64>> + Send {
        let log = self.log.clone();
        async move {
            let mut l = log.lock().unwrap();
            for c in changes {
                l.push((device_id, c));
            }
            Ok(l.len() as i64)
        }
    }

    fn pull_rich(
        &self,
        _device_id: DeviceId,
        cursor: i64,
    ) -> impl Future<Output = Result<(Vec<PulledChange>, i64, bool)>> + Send {
        let log = self.log.clone();
        let peer_version = *self.peer_version.lock().unwrap();
        async move {
            let l = log.lock().unwrap();
            let out: Vec<PulledChange> = l
                .iter()
                .enumerate()
                .filter(|(i, _)| (*i as i64 + 1) > cursor)
                .map(|(i, (dev, entry))| PulledChange {
                    storage_version: peer_version,
                    entry: entry.clone(),
                    device_id: *dev,
                    seq: i as i64 + 1,
                })
                .collect();
            Ok((out, l.len() as i64, false))
        }
    }
}

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

/// A schema with a group-scoped table: `task` carries a local `group_id`
/// provenance column (not a synced column), declared via `group_scoped`.
fn group_schema() -> SyncSchema {
    SyncSchema::new(vec![
        SyncTable::full("task", &["id", "name"]).group_scoped("group_id"),
    ])
}

fn group_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
    let db = DbSource::path(path);
    let conn = db.open().unwrap();
    conn.execute_batch("CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);")
        .unwrap();
    conn.execute_batch(&group_schema().migration_sql()).unwrap();
    (db, DeviceId::new(uuid::Uuid::from_u128(n)))
}

#[tokio::test]
async fn push_scope_drains_only_its_own_scope() {
    let dir = tempdir();
    let (db, node) = group_device(&dir.join("g.db"), 7);
    let server = FakeServer::default();
    let gid = GroupId::new(uuid::Uuid::from_u128(0x6971));

    // A personal task (group_id NULL) and a group task (group_id = gid).
    {
        let c = db.open().unwrap();
        c.execute(
            "INSERT INTO task (id, name, group_id) VALUES ('p', 'personal', NULL)",
            [],
        )
        .unwrap();
        c.execute(
            "INSERT INTO task (id, name, group_id) VALUES ('g', 'grouped', ?1)",
            [gid.to_string()],
        )
        .unwrap();
    }

    // Personal push drains only the personal row.
    let pushed = push_scope(&db, &server, &group_schema(), node, SyncScope::Personal)
        .await
        .unwrap();
    assert_eq!(pushed, 1);
    assert_eq!(server.log.lock().unwrap().len(), 1);
    assert_eq!(server.log.lock().unwrap()[0].1.row_id, "p");
    assert!(server.group_log.lock().unwrap().is_empty());

    // Group push drains only the group row, to that group.
    let pushed = push_scope(
        &db,
        &server,
        &group_schema(),
        node,
        SyncScope::Group {
            id: gid,
            gck_version: 1,
        },
    )
    .await
    .unwrap();
    assert_eq!(pushed, 1);
    let gl = server.group_log.lock().unwrap();
    assert_eq!(gl.len(), 1);
    assert_eq!(gl[0].0, gid);
    assert_eq!(gl[0].2.row_id, "g");
    // The personal log did not grow.
    assert_eq!(server.log.lock().unwrap().len(), 1);
}

fn device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
    let db = DbSource::path(path);
    let conn = db.open().unwrap();
    conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
        .unwrap();
    conn.execute_batch(&schema().migration_sql()).unwrap();
    (db, DeviceId::new(uuid::Uuid::from_u128(n)))
}

/// The gate's end-to-end case: a peer on a different manifest is refused
/// before anything is applied, and the cursor does not move, so the page is
/// still there once both sides agree.
#[tokio::test]
async fn a_peer_on_another_storage_version_is_refused_and_nothing_lands() {
    let dir = tempdir();
    let (writer, writer_node) = device(&dir.join("a.db"), 1);
    let (reader, reader_node) = device(&dir.join("b.db"), 2);
    let server = FakeServer::default();
    let gated = schema().storage_version(4);

    edit(&writer, "n1", "from the newer device");
    push_scope(&writer, &server, &gated, writer_node, SyncScope::Personal)
        .await
        .unwrap();
    // The peer is a manifest ahead.
    *server.peer_version.lock().unwrap() = Some(5);

    let err = pull_scope(&reader, &server, &gated, reader_node, SyncScope::Personal)
        .await
        .unwrap_err();
    let r = match err {
        SyncKitError::StorageVersion(r) => r,
        other => panic!("expected a storage-version refusal, got {other:?}"),
    };
    assert_eq!((r.mine, r.theirs), (4, 5));
    assert_eq!(r.message(), "Update this device.");

    // No partial write, no dropped records, and the cursor is where it was.
    let conn = reader.open().unwrap();
    let rows: i64 = conn
        .query_row("SELECT COUNT(*) FROM note", [], |r| r.get(0))
        .unwrap();
    assert_eq!(rows, 0, "nothing was applied");
    assert_eq!(
        get_scope_cursor(&conn, "").unwrap(),
        0,
        "the cursor did not advance past a page that was never applied"
    );

    // Once the reader catches up, the same page applies.
    let matched = schema().storage_version(5);
    pull_scope(&reader, &server, &matched, reader_node, SyncScope::Personal)
        .await
        .unwrap();
    let name: String = conn
        .query_row("SELECT name FROM note WHERE id = 'n1'", [], |r| r.get(0))
        .unwrap();
    assert_eq!(name, "from the newer device");
}

/// An app that has not adopted the gate must be entirely unaffected.
#[tokio::test]
async fn an_undeclared_manifest_pulls_a_stamped_page_as_before() {
    let dir = tempdir();
    let (writer, writer_node) = device(&dir.join("a.db"), 1);
    let (reader, reader_node) = device(&dir.join("b.db"), 2);
    let server = FakeServer::default();

    edit(&writer, "n1", "hello");
    push_scope(
        &writer,
        &server,
        &schema(),
        writer_node,
        SyncScope::Personal,
    )
    .await
    .unwrap();
    *server.peer_version.lock().unwrap() = Some(9);

    pull_scope(
        &reader,
        &server,
        &schema(),
        reader_node,
        SyncScope::Personal,
    )
    .await
    .unwrap();
    let conn = reader.open().unwrap();
    let name: String = conn
        .query_row("SELECT name FROM note WHERE id = 'n1'", [], |r| r.get(0))
        .unwrap();
    assert_eq!(name, "hello");
}

fn edit(db: &DbSource, id: &str, name: &str) {
    let conn = db.open().unwrap();
    conn.execute(
        "INSERT INTO note (id, name) VALUES (?1, ?2) ON CONFLICT(id) DO UPDATE SET name = excluded.name",
        (id, name),
    )
    .unwrap();
}

/// A schema whose `child` table has a real foreign key to `parent`, so a
/// child arriving first violates a constraint instead of quietly landing.
fn fk_schema() -> SyncSchema {
    SyncSchema::new(vec![
        SyncTable::full("parent", &["id", "name"]),
        SyncTable::full("child", &["id", "parent_id"]),
    ])
}

fn fk_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
    let db = DbSource::path(path);
    let conn = db.open().unwrap();
    conn.execute_batch(
        "CREATE TABLE parent (id TEXT PRIMARY KEY, name TEXT);
             CREATE TABLE child (id TEXT PRIMARY KEY, parent_id TEXT NOT NULL REFERENCES parent(id));",
    )
    .unwrap();
    conn.execute_batch(&fk_schema().migration_sql()).unwrap();
    (db, DeviceId::new(uuid::Uuid::from_u128(n)))
}

/// Put an entry on the fake server as if another device had pushed it.
fn serve(server: &FakeServer, table: &str, row_id: &str, data: serde_json::Value) {
    server.log.lock().unwrap().push((
        DeviceId::new(uuid::Uuid::from_u128(0xAA)),
        ChangeEntry {
            table: table.into(),
            op: ChangeOp::Insert,
            row_id: row_id.into(),
            timestamp: Utc::now(),
            hlc: crate::types::hlc_legacy_floor(),
            data: Some(data),
            extra: serde_json::Map::default(),
        },
    ));
}

fn row_count(db: &DbSource, table: &str) -> i64 {
    db.open()
        .unwrap()
        .query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0))
        .unwrap()
}

#[tokio::test]
async fn a_child_that_arrives_before_its_parent_is_held_and_lands_on_the_next_pull() {
    let dir = tempdir();
    let (db, node) = fk_device(&dir.join("fk.db"), 11);
    let server = FakeServer::default();

    // The child arrives alone. Its parent does not exist yet, so it cannot be
    // written; before the hold existed this row was gone for good, because the
    // cursor moved past it and the server never sends an entry twice.
    serve(
        &server,
        "child",
        "c1",
        serde_json::json!({"id":"c1","parent_id":"p1"}),
    );
    let out = pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
        .await
        .unwrap();

    assert_eq!(out.applied, 0);
    assert_eq!(out.deferred, 1);
    assert_eq!(row_count(&db, "child"), 0);
    {
        let conn = db.open().unwrap();
        assert_eq!(
            get_scope_cursor(&conn, "").unwrap(),
            1,
            "the cursor still advances; the hold is what makes that safe"
        );
        assert_eq!(deferred::counts(&conn, "").unwrap().deferred, 1);
    }

    // The parent lands on the next pull, and the held child rides in with it.
    serve(
        &server,
        "parent",
        "p1",
        serde_json::json!({"id":"p1","name":"p"}),
    );
    let out = pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
        .await
        .unwrap();

    assert_eq!(out.applied, 2, "the new parent plus the retried child");
    assert_eq!(out.deferred, 0);
    assert_eq!(row_count(&db, "child"), 1);
    assert_eq!(
        deferred::counts(&db.open().unwrap(), "").unwrap().total(),
        0,
        "a held entry that lands is cleared"
    );
}

#[tokio::test]
async fn a_parent_that_never_arrives_stops_being_retried_at_the_cap() {
    let dir = tempdir();
    let (db, node) = fk_device(&dir.join("fk_cap.db"), 12);
    let server = FakeServer::default();

    serve(
        &server,
        "child",
        "c1",
        serde_json::json!({"id":"c1","parent_id":"nope"}),
    );

    // Each pull spends one attempt. The first holds it; MAX_ATTEMPTS more
    // exhaust it. An empty page still runs the retry, which is the point.
    for _ in 0..=deferred::MAX_ATTEMPTS {
        pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
            .await
            .unwrap();
    }

    let conn = db.open().unwrap();
    let counts = deferred::counts(&conn, "").unwrap();
    assert_eq!(counts.deferred, 0, "no longer retried");
    assert_eq!(counts.rejected, 1, "but still visible, not discarded");
    let listed = deferred::list(&conn, "").unwrap();
    assert_eq!(listed[0].row_id, "c1");
    assert_eq!(listed[0].attempts, deferred::MAX_ATTEMPTS);
}

#[tokio::test]
async fn a_deferred_entry_is_not_recorded_as_committed() {
    let dir = tempdir();
    let (db, node) = fk_device(&dir.join("fk_gate.db"), 13);
    let server = FakeServer::default();

    serve(
        &server,
        "child",
        "c1",
        serde_json::json!({"id":"c1","parent_id":"p1"}),
    );
    pull_scope(&db, &server, &fk_schema(), node, SyncScope::Personal)
        .await
        .unwrap();

    // Recording an unapplied row's HLC would gate its own retry out on the
    // next pull, since the gate drops anything not newer than what is
    // committed, and the hold would be a queue that never drains.
    assert!(
        super::super::hlc::committed_hlc(&db.open().unwrap(), "child", "c1")
            .unwrap()
            .is_none()
    );
}

fn stamp_at(db: &DbSource, node: DeviceId, now_ms: i64) {
    stamp_pending(&db.open().unwrap(), node, now_ms).unwrap();
}

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

#[tokio::test]
async fn two_device_push_pull_converges_to_higher_hlc() {
    let dir = tempdir();
    let (da, na) = device(&dir.join("a.db"), 1);
    let (db_, nb) = device(&dir.join("b.db"), 2);
    let server = FakeServer::default();

    // A edits first (t=100), B edits the same row later (t=200) → B wins.
    edit(&da, "r", "from-A");
    stamp_at(&da, na, 100);
    edit(&db_, "r", "from-B");
    stamp_at(&db_, nb, 200);

    push_changes(&da, &server, &schema(), na).await.unwrap();
    push_changes(&db_, &server, &schema(), nb).await.unwrap();

    pull_changes(&da, &server, &schema(), na).await.unwrap();
    pull_changes(&db_, &server, &schema(), nb).await.unwrap();

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

#[tokio::test]
async fn push_marks_rows_and_advances_cursor() {
    let dir = tempdir();
    let (da, na) = device(&dir.join("a.db"), 1);
    let server = FakeServer::default();
    edit(&da, "r1", "x");
    edit(&da, "r2", "y");

    let pushed = push_changes(&da, &server, &schema(), na).await.unwrap();
    assert_eq!(pushed, 2);
    // All local rows are marked pushed.
    let unpushed: i64 = da
        .open()
        .unwrap()
        .query_row(
            "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
            [],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(unpushed, 0);

    // A fresh device pulls both and advances its cursor to 2.
    let (db_, nb) = device(&dir.join("b.db"), 2);
    let out = pull_changes(&db_, &server, &schema(), nb).await.unwrap();
    assert_eq!(out.applied, 2);
    assert_eq!(note_name(&db_, "r1").as_deref(), Some("x"));
    // Personal pull advances the personal ('') scope cursor.
    let cursor = get_scope_cursor(&db_.open().unwrap(), "").unwrap();
    assert_eq!(cursor, 2);
}

/// The push half of the field-merge base.
///
/// A pull is the obvious moment a row becomes common ground and it is only
/// half of them: once the server takes an edit, that edit is what a peer will
/// pull, so it is the version the two devices next diverge from. Re-basing
/// only on pull would leave the base stuck at whatever this device last
/// *received*, and every merge afterwards would report this device's own
/// already-shared edits as changes, handing itself fields it never contested.
#[tokio::test]
async fn an_acknowledged_push_rebases_the_row() {
    let dir = tempdir();
    let db = DbSource::path(dir.join("a.db"));
    {
        let conn = db.open().unwrap();
        conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
            .unwrap();
        conn.execute_batch(&merge_schema().migration_sql()).unwrap();
    }
    let node = DeviceId::new(uuid::Uuid::from_u128(1));
    let server = FakeServer::default();

    edit(&db, "r1", "mine");
    assert_eq!(
        base(&db, "r1"),
        serde_json::Value::Null,
        "nothing shared yet"
    );

    push_changes(&db, &server, &merge_schema(), node)
        .await
        .unwrap();
    assert_eq!(
        base(&db, "r1")["name"],
        "mine",
        "the server took this edit, so it is the version a peer will pull"
    );

    // And a pushed delete drops the base rather than leaving it describing a
    // row that no longer exists.
    db.open()
        .unwrap()
        .execute("DELETE FROM note WHERE id = 'r1'", [])
        .unwrap();
    push_changes(&db, &server, &merge_schema(), node)
        .await
        .unwrap();
    assert_eq!(base(&db, "r1"), serde_json::Value::Null);
}

/// A table that did not opt in must store no base, so the storage cost lands
/// only where someone asked for it.
#[tokio::test]
async fn a_push_stores_no_base_for_a_table_that_did_not_opt_in() {
    let dir = tempdir();
    let (db, node) = device(&dir.join("a.db"), 1);
    let server = FakeServer::default();

    edit(&db, "r1", "mine");
    push_changes(&db, &server, &schema(), node).await.unwrap();

    assert_eq!(base(&db, "r1"), serde_json::Value::Null);
}

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

fn base(db: &DbSource, row_id: &str) -> serde_json::Value {
    super::snapshot::load(&db.open().unwrap(), "note", row_id)
}

#[test]
fn initial_snapshot_captures_existing_rows_once() {
    let dir = tempdir();
    let (da, _) = device(&dir.join("a.db"), 1);
    let conn = da.open().unwrap();
    // Pre-existing rows inserted with triggers suppressed (as if before sync).
    conn.execute("INSERT INTO note (id, name) VALUES ('r1', 'a')", [])
        .unwrap();
    conn.execute("INSERT INTO note (id, name) VALUES ('r2', 'b')", [])
        .unwrap();
    // (the inserts above DID fire triggers; clear the changelog to simulate a
    // pre-sync backfill scenario cleanly)
    conn.execute("DELETE FROM sync_changelog", []).unwrap();

    let n = create_initial_snapshot(&conn, &schema()).unwrap();
    assert_eq!(n, 2);
    // Idempotent: a second snapshot adds nothing.
    assert_eq!(create_initial_snapshot(&conn, &schema()).unwrap(), 0);
    let done = get_sync_state_or(&conn, "initial_snapshot_done", "0").unwrap();
    assert_eq!(done, "1");
}

#[test]
fn retention_caps_pushed_but_keeps_unpushed() {
    let dir = tempdir();
    let (da, _) = device(&dir.join("a.db"), 1);
    let conn = da.open().unwrap();
    conn.execute("DELETE FROM sync_changelog", []).unwrap();
    // 5 pushed + 2 unpushed entries.
    for i in 0..5 {
        conn.execute(
            "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',1)",
            [format!("p{i}")],
        )
        .unwrap();
    }
    for i in 0..2 {
        conn.execute(
            "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',0)",
            [format!("u{i}")],
        )
        .unwrap();
    }
    let before: i64 = conn
        .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
        .unwrap();
    // Cap to 3: keep the 3 most recent rows, drop older PUSHED ones only.
    let dropped = enforce_changelog_retention(&conn, 3).unwrap();
    let after: i64 = conn
        .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
        .unwrap();
    let unpushed: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
            [],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(unpushed, 2, "unpushed entries are never dropped");
    // The 3 newest rows by id are the 2 unpushed plus the newest pushed one,
    // so exactly the 4 oldest pushed rows go. A count that is not 4 - a
    // hardcoded 0 or 1, or the row count of some other query - disagrees.
    assert_eq!(
        dropped, 4,
        "the four oldest pushed rows are the only ones dropped"
    );
    assert_eq!(
        dropped,
        (before - after) as u64,
        "the returned count is the number of rows actually deleted"
    );
    assert_eq!(after, 3, "the cap is the number of surviving rows here");
}

#[test]
fn retention_keeps_exactly_cap_rows_and_drops_the_next_one() {
    let dir = tempdir();
    let (da, _) = device(&dir.join("retention_edge.db"), 27);
    let conn = da.open().unwrap();
    conn.execute("DELETE FROM sync_changelog", []).unwrap();
    // 6 pushed rows, ids ascending with insertion order.
    for i in 0..6 {
        conn.execute(
            "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed) VALUES ('note','INSERT',?1,'{}',1)",
            [format!("p{i}")],
        )
        .unwrap();
    }
    let ids: Vec<i64> = {
        let mut st = conn
            .prepare("SELECT id FROM sync_changelog ORDER BY id ASC")
            .unwrap();
        let rows: Vec<i64> = st
            .query_map([], |r| r.get(0))
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        rows
    };
    assert_eq!(ids.len(), 6);

    // Cap 4 of 6 keeps the newest four and drops two. Both sides of the
    // boundary are named: ids[1] is the last row dropped, ids[2] the first
    // row kept, so an off-by-one in the LIMIT changes the answer.
    let dropped = enforce_changelog_retention(&conn, 4).unwrap();
    assert_eq!(dropped, 2, "6 rows capped at 4 drops 2");
    let survivors: Vec<i64> = {
        let mut st = conn
            .prepare("SELECT id FROM sync_changelog ORDER BY id ASC")
            .unwrap();
        st.query_map([], |r| r.get(0))
            .unwrap()
            .map(|r| r.unwrap())
            .collect()
    };
    assert_eq!(survivors, ids[2..].to_vec(), "the newest four rows survive");
    assert!(
        !survivors.contains(&ids[1]),
        "the row just past the cap is gone"
    );

    // Re-running at the same cap is a no-op: nothing is left to drop, so a
    // constant return value of 1 or 2 disagrees with 0 here.
    assert_eq!(
        enforce_changelog_retention(&conn, 4).unwrap(),
        0,
        "a second pass at the same cap drops nothing"
    );
}

#[tokio::test]
async fn push_drains_past_the_batch_limit() {
    let dir = tempdir();
    let (db, node) = device(&dir.join("big.db"), 21);
    let server = FakeServer::default();

    // One more than two full batches, so the drain has to come back for a
    // third: a loop that stops after the first batch loses 501 rows and
    // still reports a clean sync.
    let total = PUSH_BATCH_LIMIT * 2 + 1;
    {
        let conn = db.open().unwrap();
        conn.execute("BEGIN", []).unwrap();
        for i in 0..total {
            conn.execute(
                "INSERT INTO note (id, name) VALUES (?1, 'n')",
                [format!("r{i:04}")],
            )
            .unwrap();
        }
        conn.execute("COMMIT", []).unwrap();
        let pending: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(pending as usize, total, "the triggers logged every insert");
    }

    let pushed = push_scope(&db, &server, &schema(), node, SyncScope::Personal)
        .await
        .unwrap();
    assert_eq!(pushed as usize, total);

    {
        let log = server.log.lock().unwrap();
        assert_eq!(log.len(), total, "every row reached the server");
        let mut seen: Vec<String> = log.iter().map(|(_, e)| e.row_id.clone()).collect();
        seen.sort();
        seen.dedup();
        assert_eq!(
            seen.len(),
            total,
            "no row was sent twice in place of another"
        );
    }

    let left: i64 = db
        .open()
        .unwrap()
        .query_row(
            "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
            [],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(left, 0, "every row is marked pushed");

    // A second drain has nothing to do and sends nothing.
    let again = push_scope(&db, &server, &schema(), node, SyncScope::Personal)
        .await
        .unwrap();
    assert_eq!(again, 0);
    assert_eq!(server.log.lock().unwrap().len(), total);
}

/// A schema wide enough for one pull to produce all four outcomes at once.
fn mix_schema() -> SyncSchema {
    SyncSchema::new(vec![
        SyncTable::full("parent", &["id", "name"]),
        SyncTable::full("child", &["id", "parent_id"]),
        SyncTable::full("cfg", &["key", "value"])
            .pk(&["key"])
            .exclude_where("{row}.key NOT LIKE 'sync\\_%' ESCAPE '\\'"),
    ])
}

fn mix_device(path: &std::path::Path, n: u128) -> (DbSource, DeviceId) {
    let db = DbSource::path(path);
    let conn = db.open().unwrap();
    conn.execute_batch(
        "CREATE TABLE parent (id TEXT PRIMARY KEY, name TEXT);
             CREATE TABLE child (id TEXT PRIMARY KEY, parent_id TEXT NOT NULL REFERENCES parent(id));
             CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);",
    )
    .unwrap();
    conn.execute_batch(&mix_schema().migration_sql()).unwrap();
    (db, DeviceId::new(uuid::Uuid::from_u128(n)))
}

/// Put an arbitrary entry on the fake server, including one with no payload.
fn serve_entry(server: &FakeServer, entry: ChangeEntry) {
    server
        .log
        .lock()
        .unwrap()
        .push((DeviceId::new(uuid::Uuid::from_u128(0xAA)), entry));
}

#[tokio::test]
async fn pull_reports_all_four_outcome_counters() {
    let dir = tempdir();
    let (db, node) = mix_device(&dir.join("mix.db"), 22);
    let server = FakeServer::default();

    // applied x4
    for i in 1..=4 {
        serve(
            &server,
            "parent",
            &format!("p{i}"),
            serde_json::json!({"id": format!("p{i}"), "name": "p"}),
        );
    }
    // filtered x3: cfg's include predicate excludes the sync_ prefix.
    for k in ["sync_a", "sync_b", "sync_c"] {
        serve(
            &server,
            "cfg",
            k,
            serde_json::json!({"key": k, "value": "v"}),
        );
    }
    // rejected x1: no payload, so the same bytes always fail.
    serve_entry(
        &server,
        ChangeEntry {
            table: "parent".into(),
            op: ChangeOp::Insert,
            row_id: "p9".into(),
            timestamp: Utc::now(),
            hlc: crate::types::hlc_legacy_floor(),
            data: None,
            extra: serde_json::Map::default(),
        },
    );
    // deferred x2: a parent that is not in this page and never will be.
    for i in 1..=2 {
        serve(
            &server,
            "child",
            &format!("c{i}"),
            serde_json::json!({"id": format!("c{i}"), "parent_id": "absent"}),
        );
    }

    let out = pull_scope(&db, &server, &mix_schema(), node, SyncScope::Personal)
        .await
        .unwrap();

    assert_eq!(out.applied, 4);
    assert_eq!(out.filtered, 3);
    assert_eq!(out.rejected, 1);
    assert_eq!(out.deferred, 2);
    assert!(out.changed_tables.contains("parent"));
    assert!(
        !out.changed_tables.contains("child"),
        "nothing landed in child"
    );
    assert_eq!(row_count(&db, "parent"), 4);
    assert_eq!(row_count(&db, "child"), 0);
    assert_eq!(row_count(&db, "cfg"), 0);
}

#[test]
fn cleanup_changelog_returns_what_it_deleted() {
    let dir = tempdir();
    let (da, _) = device(&dir.join("cleanup.db"), 23);
    let conn = da.open().unwrap();
    conn.execute("DELETE FROM sync_changelog", []).unwrap();

    // 4 pushed and old (deletable), 2 pushed and recent, 3 unpushed and old.
    let insert = |row_id: &str, pushed: i64, ts: &str| {
        conn.execute(
            "INSERT INTO sync_changelog (table_name, op, row_id, data, pushed, timestamp) \
             VALUES ('note','INSERT',?1,'{}',?2,?3)",
            rusqlite::params![row_id, pushed, ts],
        )
        .unwrap();
    };
    let old = "2000-01-01T00:00:00.000Z";
    let now = "2999-01-01T00:00:00.000Z";
    for i in 0..4 {
        insert(&format!("old{i}"), 1, old);
    }
    for i in 0..2 {
        insert(&format!("new{i}"), 1, now);
    }
    for i in 0..3 {
        insert(&format!("unp{i}"), 0, old);
    }

    let before: i64 = conn
        .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
        .unwrap();
    let removed = cleanup_changelog(&conn).unwrap();
    let after: i64 = conn
        .query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
        .unwrap();

    assert_eq!(removed, 4, "only the old pushed rows go");
    assert_eq!(
        removed,
        (before - after) as u64,
        "the returned count is the number of rows actually deleted"
    );
    let unpushed: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
            [],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(unpushed, 3, "an unpushed row is never pruned by age");

    // Nothing left to prune, so a second pass reports zero.
    assert_eq!(cleanup_changelog(&conn).unwrap(), 0);
}

// minimal temp-dir helper (no external dep)
fn tempdir() -> std::path::PathBuf {
    let mut p = std::env::temp_dir();
    // Unique-ish per test via a monotonic counter; tests run in the same
    // process so a static AtomicU64 keeps paths distinct.
    use std::sync::atomic::{AtomicU64, Ordering};
    static N: AtomicU64 = AtomicU64::new(0);
    p.push(format!(
        "synckit_b5_{}_{}",
        std::process::id(),
        N.fetch_add(1, Ordering::Relaxed)
    ));
    std::fs::create_dir_all(&p).unwrap();
    p
}