//! Tests for [`super`].

use super::*;
use crate::types::{ChangeEntry, ChangeOp, hlc_legacy_floor};
use rusqlite::Connection;
use serde_json::json;

use super::super::db::configure_connection;
use super::super::schema::{SyncSchema, SyncTable};

fn upsert(table: &str, row_id: &str, data: Value) -> ChangeEntry {
    ChangeEntry {
        table: table.into(),
        op: ChangeOp::Insert,
        row_id: row_id.into(),
        timestamp: chrono::Utc::now(),
        hlc: hlc_legacy_floor(),
        data: Some(data),
        extra: serde_json::Map::default(),
    }
}

fn delete(table: &str, row_id: &str, data: Value) -> ChangeEntry {
    ChangeEntry {
        op: ChangeOp::Delete,
        ..upsert(table, row_id, data)
    }
}

fn schema() -> SyncSchema {
    SyncSchema::new(vec![
        SyncTable::full("parent", &["id", "name"]),
        SyncTable::full("child", &["id", "parent_id", "note"]),
        SyncTable::full("acct", &["id", "name"])
            .preserve_local(&["secret"])
            .insert_defaults(&[("secret", "")]),
        SyncTable::full("tagpair", &["a", "b"]).pk(&["a", "b"]),
        SyncTable::full("items", &["id", "is_read", "is_starred"])
            .partial_update(&["is_read", "is_starred"])
            .ignore_deletes(),
        SyncTable::full("samp", &["hash", "name", "deleted_at"])
            .pk(&["hash"])
            .hashed()
            .tombstone("deleted_at"),
        SyncTable::full("cfg", &["key", "value"])
            .pk(&["key"])
            .exclude_where("{row}.key NOT LIKE 'sync\\_%' ESCAPE '\\'"),
        SyncTable::full("reffer", &["id", "ext_id"]).references_unsynced(),
        // `kind` is NOT NULL *with a default*, the only shape in which
        // omitting a column and binding an explicit NULL differ observably.
        SyncTable::full("note", &["id", "body", "kind"]),
        // A preserved column that is also a whitelist column, so a payload
        // can carry it and the ON CONFLICT SET has to refuse it.
        SyncTable::full("vault", &["id", "label", "token"]).preserve_local(&["token"]),
        // Partial update on a composite key: two WHERE bindings, not one.
        SyncTable::full("pairflag", &["a", "b", "flag"])
            .pk(&["a", "b"])
            .partial_update(&["flag"]),
        // INTEGER PRIMARY KEY, so a text id is a datatype mismatch: a SQLite
        // failure that is not a constraint violation.
        SyncTable::full("tally", &["id", "label"]),
    ])
}

fn db() -> Connection {
    let conn = Connection::open_in_memory().unwrap();
    configure_connection(&conn).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) ON DELETE CASCADE, note TEXT);
            CREATE TABLE acct (id TEXT PRIMARY KEY, name TEXT, secret TEXT NOT NULL);
            CREATE TABLE tagpair (a TEXT, b TEXT, PRIMARY KEY (a, b));
            CREATE TABLE items (id TEXT PRIMARY KEY, is_read INTEGER, is_starred INTEGER, title TEXT);
            CREATE TABLE ghost (id INTEGER PRIMARY KEY);
            CREATE TABLE samp (hash TEXT PRIMARY KEY, name TEXT, deleted_at INTEGER);
            CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);
            CREATE TABLE reffer (id TEXT PRIMARY KEY, ext_id INTEGER NOT NULL REFERENCES ghost(id));
            CREATE TABLE note (id TEXT PRIMARY KEY, body TEXT, kind TEXT NOT NULL DEFAULT 'plain');
            CREATE TABLE vault (id TEXT PRIMARY KEY, label TEXT, token TEXT);
            CREATE TABLE pairflag (a TEXT, b TEXT, flag INTEGER, PRIMARY KEY (a, b));
            CREATE TABLE tally (id INTEGER PRIMARY KEY, label TEXT);
            ",
    )
    .unwrap();
    let s = schema();
    conn.execute_batch(&s.migration_sql()).unwrap();
    conn
}

/// These tests exercise the applier, not the pipeline that feeds it, so they
/// build the resolved batch directly rather than routing every case through
/// `resolve_pull`.
fn apply(conn: &mut Connection, changes: &[ChangeEntry]) -> ApplyOutcome {
    let changes = ResolvedChanges::for_test(changes.to_vec());
    apply_remote_changes(conn, &schema(), &changes, "").unwrap()
}

#[test]
fn full_insert_then_update_via_on_conflict() {
    let mut conn = db();
    let o = apply(
        &mut conn,
        &[upsert("parent", "p1", json!({"id":"p1","name":"a"}))],
    );
    assert_eq!(o.applied, 1);
    assert!(o.changed_tables.contains("parent"));
    apply(
        &mut conn,
        &[upsert("parent", "p1", json!({"id":"p1","name":"b"}))],
    );
    let name: String = conn
        .query_row("SELECT name FROM parent WHERE id='p1'", [], |r| r.get(0))
        .unwrap();
    assert_eq!(name, "b");
}

#[test]
fn on_conflict_update_does_not_cascade_to_children() {
    let mut conn = db();
    apply(
        &mut conn,
        &[
            upsert("parent", "p1", json!({"id":"p1","name":"a"})),
            upsert(
                "child",
                "c1",
                json!({"id":"c1","parent_id":"p1","note":"n"}),
            ),
        ],
    );
    // Re-upsert the parent; the child must survive (ON CONFLICT DO UPDATE, not REPLACE).
    apply(
        &mut conn,
        &[upsert("parent", "p1", json!({"id":"p1","name":"a2"}))],
    );
    let kids: i64 = conn
        .query_row("SELECT COUNT(*) FROM child", [], |r| r.get(0))
        .unwrap();
    assert_eq!(kids, 1);
}

#[test]
fn preserve_local_and_insert_defaults() {
    let mut conn = db();
    // First insert: secret defaults to '' (satisfies NOT NULL); payload never carries it.
    apply(
        &mut conn,
        &[upsert("acct", "a1", json!({"id":"a1","name":"n1"}))],
    );
    // Locally the user sets a real secret.
    conn.execute("UPDATE acct SET secret='hunter2' WHERE id='a1'", [])
        .unwrap();
    // A remote update to config columns must NOT clobber the local secret.
    apply(
        &mut conn,
        &[upsert("acct", "a1", json!({"id":"a1","name":"n2"}))],
    );
    let (name, secret): (String, String) = conn
        .query_row("SELECT name, secret FROM acct WHERE id='a1'", [], |r| {
            Ok((r.get(0)?, r.get(1)?))
        })
        .unwrap();
    assert_eq!(name, "n2");
    assert_eq!(
        secret, "hunter2",
        "preserved secret must survive a remote update"
    );
}

#[test]
fn null_tolerance_omits_not_null_but_keeps_nullable_null() {
    let mut conn = db();
    apply(
        &mut conn,
        &[upsert("parent", "p1", json!({"id":"p1","name":"start"}))],
    );
    // name is nullable → an explicit null clears it.
    apply(
        &mut conn,
        &[upsert("parent", "p1", json!({"id":"p1","name":null}))],
    );
    let name: Option<String> = conn
        .query_row("SELECT name FROM parent WHERE id='p1'", [], |r| r.get(0))
        .unwrap();
    assert_eq!(name, None);
    // A null for a NOT NULL column (child.parent_id) is omitted, so an insert
    // takes no value for it → constraint violation → deferred, not fatal.
    let o = apply(
        &mut conn,
        &[upsert(
            "child",
            "c1",
            json!({"id":"c1","parent_id":null,"note":"x"}),
        )],
    );
    assert_eq!(o.applied, 0);
    assert_eq!(o.deferred.len(), 1);
    assert_eq!(o.deferred[0].row_id, "c1");
}

#[test]
fn all_pk_table_uses_insert_or_ignore() {
    let mut conn = db();
    let o = apply(
        &mut conn,
        &[upsert("tagpair", "x", json!({"a":"x","b":"y"}))],
    );
    assert_eq!(o.applied, 1);
    // Re-applying the same all-PK row is a no-op, not an error.
    let o2 = apply(
        &mut conn,
        &[upsert("tagpair", "x", json!({"a":"x","b":"y"}))],
    );
    assert_eq!(o2.applied, 1); // executed, 0 rows changed, still Ok
    let n: i64 = conn
        .query_row("SELECT COUNT(*) FROM tagpair", [], |r| r.get(0))
        .unwrap();
    assert_eq!(n, 1);
}

#[test]
fn partial_update_touches_only_set_columns() {
    let mut conn = db();
    conn.execute(
        "INSERT INTO items (id, is_read, is_starred, title) VALUES ('i1', 0, 0, 'keep')",
        [],
    )
    .unwrap();
    apply(
        &mut conn,
        &[ChangeEntry {
            op: ChangeOp::Update,
            ..upsert("items", "i1", json!({"id":"i1","is_read":1,"is_starred":0}))
        }],
    );
    let (read, title): (i64, String) = conn
        .query_row("SELECT is_read, title FROM items WHERE id='i1'", [], |r| {
            Ok((r.get(0)?, r.get(1)?))
        })
        .unwrap();
    assert_eq!(read, 1);
    assert_eq!(
        title, "keep",
        "partial update must not touch non-set columns"
    );
}

#[test]
fn hard_delete_and_ignore_delete() {
    let mut conn = db();
    apply(
        &mut conn,
        &[upsert("parent", "p1", json!({"id":"p1","name":"a"}))],
    );
    let o = apply(&mut conn, &[delete("parent", "p1", json!({"id":"p1"}))]);
    assert_eq!(o.applied, 1);
    assert_eq!(
        conn.query_row("SELECT COUNT(*) FROM parent", [], |r| r.get::<_, i64>(0))
            .unwrap(),
        0
    );

    // items ignore deletes.
    conn.execute(
        "INSERT INTO items (id, is_read, is_starred) VALUES ('i1', 1, 0)",
        [],
    )
    .unwrap();
    let o2 = apply(&mut conn, &[delete("items", "i1", json!({"id":"i1"}))]);
    assert_eq!(o2.applied, 0);
    assert_eq!(
        conn.query_row("SELECT COUNT(*) FROM items", [], |r| r.get::<_, i64>(0))
            .unwrap(),
        1
    );
}

#[test]
fn tombstone_delete_sets_column_and_keeps_earliest() {
    let mut conn = db();
    conn.execute("INSERT INTO samp (hash, name) VALUES ('h1', 's')", [])
        .unwrap();
    // A hashed table's delete carries the PK in data; the opaque row_id is ignored.
    apply(
        &mut conn,
        &[delete("samp", "OPAQUE_HASH", json!({"hash":"h1"}))],
    );
    let (present, del): (i64, Option<i64>) = conn
        .query_row(
            "SELECT COUNT(*), MAX(deleted_at) FROM samp WHERE hash='h1'",
            [],
            |r| Ok((r.get(0)?, r.get(1)?)),
        )
        .unwrap();
    assert_eq!(present, 1, "tombstone keeps the row");
    let first = del.unwrap();
    // Re-deleting keeps the earliest instant (COALESCE).
    apply(
        &mut conn,
        &[delete("samp", "OPAQUE_HASH", json!({"hash":"h1"}))],
    );
    let second: i64 = conn
        .query_row("SELECT deleted_at FROM samp WHERE hash='h1'", [], |r| {
            r.get(0)
        })
        .unwrap();
    assert_eq!(first, second);
}

#[test]
fn exclude_where_guards_import_both_ways() {
    let mut conn = db();
    let o = apply(
        &mut conn,
        &[
            upsert(
                "cfg",
                "sync_cursor",
                json!({"key":"sync_cursor","value":"9"}),
            ), // excluded
            upsert("cfg", "theme", json!({"key":"theme","value":"dark"})), // included
        ],
    );
    assert_eq!(o.applied, 1);
    let keys: Vec<String> = {
        let mut s = conn.prepare("SELECT key FROM cfg ORDER BY key").unwrap();
        s.query_map([], |r| r.get(0))
            .unwrap()
            .map(|r| r.unwrap())
            .collect()
    };
    assert_eq!(keys, vec!["theme".to_string()]);
    // A hostile delete of an excluded key is also dropped.
    conn.execute(
        "INSERT INTO cfg (key, value) VALUES ('sync_secret', 'x')",
        [],
    )
    .unwrap();
    let o2 = apply(
        &mut conn,
        &[delete("cfg", "sync_secret", json!({"key":"sync_secret"}))],
    );
    assert_eq!(o2.applied, 0);
    assert_eq!(
        conn.query_row(
            "SELECT COUNT(*) FROM cfg WHERE key='sync_secret'",
            [],
            |r| r.get::<_, i64>(0)
        )
        .unwrap(),
        1
    );
}

#[test]
fn fk_ordering_parents_before_children_children_before_parents() {
    let mut conn = db();
    // Child listed before parent in the batch, but FK enforced, must still apply
    // because the engine orders upserts parents-first.
    let o = apply(
        &mut conn,
        &[
            upsert(
                "child",
                "c1",
                json!({"id":"c1","parent_id":"p1","note":"n"}),
            ),
            upsert("parent", "p1", json!({"id":"p1","name":"a"})),
        ],
    );
    assert_eq!(o.applied, 2);
    // Delete both; children-first ordering means the child goes before the parent.
    let o2 = apply(
        &mut conn,
        &[
            delete("parent", "p1", json!({"id":"p1"})),
            delete("child", "c1", json!({"id":"c1"})),
        ],
    );
    assert_eq!(o2.applied, 2);
}

#[test]
fn references_unsynced_relaxes_fk() {
    let mut conn = db();
    // reffer.ext_id points at a ghost row that does not exist and is not synced.
    // Without FK relaxation this would be a constraint violation.
    let o = apply(
        &mut conn,
        &[upsert("reffer", "r1", json!({"id":"r1","ext_id":999}))],
    );
    assert_eq!(
        o.applied, 1,
        "references_unsynced disables FK for the apply"
    );
    // FK enforcement is restored afterward.
    let fk: i64 = conn
        .query_row("PRAGMA foreign_keys", [], |r| r.get(0))
        .unwrap();
    assert_eq!(fk, 1);
}

#[test]
fn constraint_violation_is_skipped_not_fatal() {
    let mut conn = db();
    // First row violates FK (no parent p9); second is valid. Batch must not abort.
    let o = apply(
        &mut conn,
        &[
            upsert(
                "child",
                "bad",
                json!({"id":"bad","parent_id":"p9","note":"x"}),
            ),
            upsert("parent", "p1", json!({"id":"p1","name":"ok"})),
        ],
    );
    assert_eq!(o.applied, 1);
    assert_eq!(o.deferred.len(), 1, "the poison row is held, not lost");
    assert_eq!(
        conn.query_row("SELECT COUNT(*) FROM parent", [], |r| r.get::<_, i64>(0))
            .unwrap(),
        1
    );
}

#[test]
fn unknown_table_change_is_deferred_not_dropped() {
    let mut conn = db();
    let o = apply(&mut conn, &[upsert("nonexistent", "x", json!({"id":"x"}))]);
    assert_eq!(o.applied, 0);
    // Deferred rather than rejected: the table may exist after a client
    // upgrade, and then the held entry applies.
    assert_eq!(o.deferred.len(), 1);
    assert_eq!(o.deferred[0].table, "nonexistent");
    assert!(o.rejected.is_empty());
}

#[test]
fn an_excluded_row_is_filtered_not_held() {
    let mut conn = db();
    // cfg's include predicate is "key NOT LIKE 'sync_%'", so a sync_ key is
    // excluded on import. That is policy, not failure, and must never reach
    // the dead-letter.
    let o = apply(
        &mut conn,
        &[upsert(
            "cfg",
            "sync_token",
            json!({"key":"sync_token","value":"x"}),
        )],
    );
    assert_eq!(o.applied, 0);
    assert_eq!(o.filtered, 1);
    assert!(o.deferred.is_empty());
    assert!(o.rejected.is_empty());
}

#[test]
fn a_payloadless_upsert_is_rejected_not_deferred() {
    let mut conn = db();
    let mut change = upsert("parent", "p1", json!({"id":"p1"}));
    change.data = None;
    let o = apply(&mut conn, &[change]);
    assert_eq!(o.applied, 0);
    assert_eq!(
        o.rejected.len(),
        1,
        "identical bytes would fail identically"
    );
    assert!(o.deferred.is_empty());
}

#[test]
fn fk_sweep_catches_what_the_batch_wide_relaxation_hides() {
    let mut conn = db();
    // `reffer` declares references_unsynced, so the whole apply runs with
    // foreign_keys=OFF. Without the sweep, the child row below lands with a
    // missing parent and nothing is reported.
    let o = apply(
        &mut conn,
        &[
            upsert("reffer", "r1", json!({"id":"r1","ext_id":404})),
            upsert(
                "child",
                "c1",
                json!({"id":"c1","parent_id":"missing","note":"x"}),
            ),
        ],
    );

    assert_eq!(
        conn.query_row("SELECT COUNT(*) FROM child", [], |r| r.get::<_, i64>(0))
            .unwrap(),
        0,
        "the orphan is removed, not left to resurface later"
    );
    assert_eq!(
        conn.query_row("SELECT COUNT(*) FROM reffer", [], |r| r.get::<_, i64>(0))
            .unwrap(),
        1,
        "the table the relaxation exists for keeps its row"
    );
    assert_eq!(o.applied, 1);
    assert_eq!(o.deferred.len(), 1);
    assert_eq!(o.deferred[0].table, "child");
    assert_eq!(o.deferred[0].row_id, "c1");
}

#[test]
fn fk_sweep_leaves_a_row_whose_parent_is_present() {
    let mut conn = db();
    let o = apply(
        &mut conn,
        &[
            upsert("reffer", "r1", json!({"id":"r1","ext_id":404})),
            upsert("parent", "p1", json!({"id":"p1","name":"ok"})),
            upsert(
                "child",
                "c1",
                json!({"id":"c1","parent_id":"p1","note":"x"}),
            ),
        ],
    );
    assert_eq!(o.applied, 3);
    assert!(o.deferred.is_empty(), "a satisfied FK is not a violation");
}

#[test]
fn unapplied_totals_the_three_not_applied_kinds() {
    let mut conn = db();
    let o = apply(
        &mut conn,
        &[
            // applied: an ordinary parent row.
            upsert("parent", "p1", json!({"id":"p1","name":"a"})),
            // filtered x3: two excluded cfg keys and one delete on a table
            // that ignores deletes.
            upsert("cfg", "sync_a", json!({"key":"sync_a","value":"1"})),
            upsert("cfg", "sync_b", json!({"key":"sync_b","value":"2"})),
            delete("items", "i1", json!({"id":"i1"})),
            // rejected x1: no payload, so the same bytes always fail.
            ChangeEntry {
                data: None,
                ..upsert("parent", "p2", json!({"id":"p2"}))
            },
            // deferred x2: an unknown table and a missing FK parent.
            upsert("nonexistent", "x", json!({"id":"x"})),
            upsert(
                "child",
                "c1",
                json!({"id":"c1","parent_id":"absent","note":"n"}),
            ),
        ],
    );

    assert_eq!(o.applied, 1);
    assert_eq!(o.filtered, 3);
    assert_eq!(o.rejected.len(), 1);
    assert_eq!(o.deferred.len(), 2);
    // The derived total is the sum of exactly those three, and excludes
    // `applied`. Each part is non-zero and distinct, so no constant and no
    // pair of the three adds up to it.
    assert_eq!(o.unapplied(), 6);
    assert_eq!(
        o.unapplied(),
        o.filtered + o.rejected.len() + o.deferred.len()
    );
}

#[test]
fn unapplied_is_zero_on_a_clean_apply() {
    let mut conn = db();
    let o = apply(
        &mut conn,
        &[upsert("parent", "p1", json!({"id":"p1","name":"a"}))],
    );
    assert_eq!(o.applied, 1);
    assert_eq!(o.unapplied(), 0, "an applied row is not an unapplied one");
}

#[test]
fn a_composite_delete_keys_off_the_payload_not_the_row_id() {
    let mut conn = db();
    apply(
        &mut conn,
        &[
            upsert("tagpair", "ignored", json!({"a":"x","b":"y"})),
            upsert("tagpair", "ignored", json!({"a":"x","b":"z"})),
        ],
    );
    // The wire row id names nothing this table can be keyed by; the payload
    // carries the whole composite key, so exactly one row goes.
    let o = apply(
        &mut conn,
        &[delete("tagpair", "not-a-key", json!({"a":"x","b":"y"}))],
    );
    assert_eq!(o.applied, 1);
    assert!(o.rejected.is_empty());
    let left: Vec<String> = conn
        .prepare("SELECT b FROM tagpair WHERE a='x' ORDER BY b")
        .unwrap()
        .query_map([], |r| r.get(0))
        .unwrap()
        .map(std::result::Result::unwrap)
        .collect();
    assert_eq!(left, vec!["z".to_string()], "the wrong row was deleted");
}

#[test]
fn a_null_component_of_a_composite_key_is_rejected_not_bound() {
    let mut conn = db();
    apply(
        &mut conn,
        &[
            upsert("tagpair", "ignored", json!({"a":"x","b":"y"})),
            upsert("tagpair", "ignored", json!({"a":"x","b":"z"})),
        ],
    );
    // `b` is JSON null, so the key cannot be reconstructed. Binding the
    // partial key would delete every row sharing `a`.
    let o = apply(
        &mut conn,
        &[delete("tagpair", "x", json!({"a":"x","b":null}))],
    );
    assert_eq!(o.applied, 0);
    assert_eq!(o.rejected.len(), 1);
    assert_eq!(o.rejected[0].table, "tagpair");
    assert!(o.deferred.is_empty());
    let rows: i64 = conn
        .query_row("SELECT COUNT(*) FROM tagpair", [], |r| r.get(0))
        .unwrap();
    assert_eq!(rows, 2, "a null key component must touch nothing");
}

#[test]
fn a_partial_update_with_a_null_key_is_rejected() {
    let mut conn = db();
    conn.execute(
        "INSERT INTO items (id, is_read, is_starred, title) VALUES ('i1', 0, 0, 't')",
        [],
    )
    .unwrap();
    // The partial-update path never falls back to the wire row id, so a null
    // `id` in the payload has no key at all.
    let o = apply(
        &mut conn,
        &[upsert("items", "i1", json!({"id":null,"is_read":1}))],
    );
    assert_eq!(o.applied, 0);
    assert_eq!(o.rejected.len(), 1);
    assert!(o.deferred.is_empty());
    let is_read: i64 = conn
        .query_row("SELECT is_read FROM items WHERE id='i1'", [], |r| r.get(0))
        .unwrap();
    assert_eq!(is_read, 0, "the keyless update must not have landed");
}

#[test]
fn a_present_key_in_the_payload_beats_the_wire_row_id() {
    let mut conn = db();
    conn.execute(
        "INSERT INTO items (id, is_read, is_starred, title) VALUES ('i1', 0, 0, 'one')",
        [],
    )
    .unwrap();
    conn.execute(
        "INSERT INTO items (id, is_read, is_starred, title) VALUES ('i2', 0, 0, 'two')",
        [],
    )
    .unwrap();
    let o = apply(
        &mut conn,
        &[upsert("items", "i1", json!({"id":"i2","is_read":1}))],
    );
    assert_eq!(o.applied, 1);
    let read: Vec<i64> = conn
        .prepare("SELECT is_read FROM items ORDER BY id")
        .unwrap()
        .query_map([], |r| r.get(0))
        .unwrap()
        .map(std::result::Result::unwrap)
        .collect();
    assert_eq!(read, vec![0, 1], "the payload key names the row to update");
}

#[test]
fn a_non_constraint_sqlite_failure_rolls_the_whole_batch_back() {
    let mut conn = db();
    // tally.id is an INTEGER PRIMARY KEY, so a text id SQLite cannot coerce
    // raises SQLITE_MISMATCH, not SQLITE_CONSTRAINT. Only a constraint
    // violation is survivable; anything else means the batch cannot be
    // trusted, so it must surface as Err rather than as a deferred row.
    let changes = ResolvedChanges::for_test(vec![
        upsert("parent", "p1", json!({"id":"p1","name":"a"})),
        upsert("tally", "t1", json!({"id":"notanint","label":"x"})),
    ]);
    let e = apply_remote_changes(&mut conn, &schema(), &changes, "")
        .expect_err("a non-constraint SQLite error must not be swallowed as deferred");
    assert!(matches!(e, crate::error::SyncKitError::Database(_)));
    assert_eq!(
        conn.query_row("SELECT COUNT(*) FROM parent", [], |r| r.get::<_, i64>(0))
            .unwrap(),
        0,
        "the valid row earlier in the same batch rolls back with it"
    );
    assert_eq!(
        conn.query_row("SELECT COUNT(*) FROM tally", [], |r| r.get::<_, i64>(0))
            .unwrap(),
        0
    );
}

#[test]
fn a_null_for_a_defaulted_not_null_column_is_omitted_not_bound() {
    let mut conn = db();
    // note.kind is NOT NULL DEFAULT 'plain'. Omitting it (what the NOT NULL
    // set is read for) takes the default; binding an explicit NULL would
    // violate instead, so the two are distinguishable here.
    let o = apply(
        &mut conn,
        &[upsert(
            "note",
            "n1",
            json!({"id":"n1","body":"b1","kind":null}),
        )],
    );
    assert_eq!(o.applied, 1);
    assert!(o.deferred.is_empty(), "an omitted column takes its default");
    let kind: String = conn
        .query_row("SELECT kind FROM note WHERE id='n1'", [], |r| r.get(0))
        .unwrap();
    assert_eq!(kind, "plain");

    // Same on the update leg: the column is left out of the ON CONFLICT SET,
    // so a local value stands rather than being nulled.
    conn.execute("UPDATE note SET kind='code' WHERE id='n1'", [])
        .unwrap();
    let o2 = apply(
        &mut conn,
        &[upsert(
            "note",
            "n1",
            json!({"id":"n1","body":"b2","kind":null}),
        )],
    );
    assert_eq!(o2.applied, 1);
    assert!(o2.deferred.is_empty());
    let (body, kind): (String, String) = conn
        .query_row("SELECT body, kind FROM note WHERE id='n1'", [], |r| {
            Ok((r.get(0)?, r.get(1)?))
        })
        .unwrap();
    assert_eq!(body, "b2", "a nullable column still updates");
    assert_eq!(
        kind, "code",
        "a NOT NULL column the payload nulled keeps its local value"
    );
}

#[test]
fn a_preserved_column_carried_in_the_payload_is_still_not_overwritten() {
    let mut conn = db();
    // vault.token is both a whitelist column and preserve_local, so the
    // payload can carry it and the ON CONFLICT SET must still refuse it.
    // Only the preserve filter keeps it out; the PK filter would not.
    let o = apply(
        &mut conn,
        &[upsert(
            "vault",
            "v1",
            json!({"id":"v1","label":"l1","token":"seed"}),
        )],
    );
    assert_eq!(o.applied, 1);
    conn.execute("UPDATE vault SET token='local' WHERE id='v1'", [])
        .unwrap();
    apply(
        &mut conn,
        &[upsert(
            "vault",
            "v1",
            json!({"id":"v1","label":"l2","token":"remote"}),
        )],
    );
    let (label, token): (String, String) = conn
        .query_row("SELECT label, token FROM vault WHERE id='v1'", [], |r| {
            Ok((r.get(0)?, r.get(1)?))
        })
        .unwrap();
    assert_eq!(label, "l2", "a non-preserved column still updates");
    assert_eq!(
        token, "local",
        "preserve_local outranks a payload that carries the column"
    );
}

#[test]
fn a_pre_existing_orphan_survives_a_sweep_of_its_own_table() {
    let mut conn = db();
    // An orphan left by an earlier relaxed apply. No entry in this batch
    // describes it, so there is nothing to hold and deleting it would lose
    // the row for good.
    conn.execute_batch(
        "PRAGMA foreign_keys=OFF;
             INSERT INTO child (id, parent_id, note) VALUES ('ghost1', 'gone', 'g');
             PRAGMA foreign_keys=ON;",
    )
    .unwrap();
    // `reffer` turns the batch-wide relaxation on, so the sweep runs over
    // `child`. The batch names the table but not this row.
    let o = apply(
        &mut conn,
        &[
            upsert("reffer", "r1", json!({"id":"r1","ext_id":404})),
            upsert("parent", "p1", json!({"id":"p1","name":"a"})),
            upsert(
                "child",
                "c1",
                json!({"id":"c1","parent_id":"p1","note":"n"}),
            ),
        ],
    );
    assert_eq!(o.applied, 3);
    assert!(
        o.deferred.is_empty(),
        "a row this batch never pulled cannot be deferred for retry"
    );
    let ids: Vec<String> = conn
        .prepare("SELECT id FROM child ORDER BY id")
        .unwrap()
        .query_map([], |r| r.get(0))
        .unwrap()
        .map(std::result::Result::unwrap)
        .collect();
    assert_eq!(
        ids,
        vec!["c1".to_string(), "ghost1".to_string()],
        "matching the table alone is not matching the row"
    );
}

#[test]
fn a_composite_partial_update_binds_every_key_component() {
    let mut conn = db();
    conn.execute_batch("INSERT INTO pairflag (a, b, flag) VALUES ('x','y',0), ('x','z',0);")
        .unwrap();
    let o = apply(
        &mut conn,
        &[ChangeEntry {
            op: ChangeOp::Update,
            ..upsert("pairflag", "x:z", json!({"a":"x","b":"z","flag":1}))
        }],
    );
    assert_eq!(o.applied, 1);
    let flags: Vec<(String, i64)> = conn
        .prepare("SELECT b, flag FROM pairflag ORDER BY b")
        .unwrap()
        .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
        .unwrap()
        .map(std::result::Result::unwrap)
        .collect();
    // Each key component needs its own placeholder: reusing the first would
    // compare `b` against the value of `a` and touch the wrong row, or none.
    assert_eq!(
        flags,
        vec![("y".to_string(), 0), ("z".to_string(), 1)],
        "only the row the whole composite key names is updated"
    );
}

#[test]
fn a_delete_with_no_payload_falls_back_to_the_wire_row_id() {
    let mut conn = db();
    apply(
        &mut conn,
        &[
            upsert("parent", "p1", json!({"id":"p1","name":"a"})),
            upsert("parent", "p2", json!({"id":"p2","name":"b"})),
        ],
    );
    // An older client wrote the key into the wire row id and sent no payload.
    // A single-PK table can still be addressed from it.
    let mut change = delete("parent", "p1", json!({}));
    change.data = None;
    let o = apply(&mut conn, &[change]);
    assert_eq!(o.applied, 1);
    assert!(
        o.rejected.is_empty(),
        "a single-PK delete is reconstructable from the row id alone"
    );
    let left: Vec<String> = conn
        .prepare("SELECT id FROM parent ORDER BY id")
        .unwrap()
        .query_map([], |r| r.get(0))
        .unwrap()
        .map(std::result::Result::unwrap)
        .collect();
    assert_eq!(
        left,
        vec!["p2".to_string()],
        "the row the wire id names, and only it, goes"
    );
}