//! Applying resolved remote changes to the local database.
//!
//! [`apply_remote_changes`] takes changes the conflict layer has already resolved
//! and writes them, in foreign-key-safe order, inside the crash-safe
//! `applying_remote` transaction from [`super::db`]. All per-table policy declared
//! in the [`SyncSchema`](super::schema::SyncSchema) resolves here: `Full` vs
//! `PartialUpdate` upserts, `Hard`/`Ignore`/`Tombstone` deletes, `preserve_local`
//! secrets, `insert_defaults`, NOT-NULL null-tolerance, the `exclude_where` import
//! guard, hashed-row reconstruction from the payload, and `references_unsynced`
//! FK relaxation.
//!
//! A single row failing a constraint is skipped and logged, never fatal, so a
//! poison row cannot wedge cursor advancement; a genuine (non-constraint) SQLite
//! error rolls the whole batch back.

use std::collections::{HashMap, HashSet};

use rusqlite::types::ToSql;
use rusqlite::{Connection, ErrorCode, Transaction};
use serde_json::Value;
use tracing::warn;

use super::db::with_applying_remote;
use super::schema::{DeleteMode, SyncMode, SyncSchema, SyncTable};
use crate::error::Result;
use crate::types::{ChangeEntry, ChangeOp};

/// A JSON null to borrow when a payload omits a whitelisted column.
static JSON_NULL: Value = Value::Null;

/// Summary of an apply pass.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct ApplyOutcome {
    /// Rows written (insert/update/delete/tombstone).
    pub applied: usize,
    /// Rows intentionally dropped or unappliable (excluded, unknown table,
    /// missing key, or a skipped constraint violation).
    pub skipped: usize,
    /// Distinct tables an applied change touched, for selective UI invalidation.
    pub changed_tables: HashSet<String>,
}

/// Apply resolved remote `changes` to `conn` in FK-safe order.
///
/// Upserts run parents-first (schema declaration order); deletes run
/// children-first (reverse). If any change targets a `references_unsynced` table,
/// foreign-key enforcement is disabled for the whole apply, SQLite only honours
/// a `foreign_keys` pragma change *outside* a transaction, so it is toggled around
/// the `applying_remote` transaction, not inside it.
pub fn apply_remote_changes(
    conn: &mut Connection,
    schema: &SyncSchema,
    changes: &[ChangeEntry],
    scope: &str,
) -> Result<ApplyOutcome> {
    let by_name: HashMap<&str, &SyncTable> = schema.tables.iter().map(|t| (t.name, t)).collect();

    let fk_off = changes.iter().any(|c| {
        by_name
            .get(c.table.as_str())
            .is_some_and(|t| t.references_unsynced)
    });
    if fk_off {
        conn.execute_batch("PRAGMA foreign_keys=OFF")?;
    }

    let outcome =
        with_applying_remote(conn, |tx| apply_inner(tx, schema, &by_name, changes, scope));

    if fk_off {
        // Restore enforcement regardless of the apply result. A failed restore is
        // self-healing: the engine re-asserts foreign_keys=ON when it next opens
        // the connection (see db::configure_connection).
        if let Err(e) = conn.execute_batch("PRAGMA foreign_keys=ON") {
            warn!("failed to restore foreign_keys=ON after apply: {e}");
        }
    }

    outcome
}

fn apply_inner(
    tx: &Transaction<'_>,
    schema: &SyncSchema,
    by_name: &HashMap<&str, &SyncTable>,
    changes: &[ChangeEntry],
    scope: &str,
) -> Result<ApplyOutcome> {
    let mut out = ApplyOutcome::default();

    // Changes targeting a table the schema doesn't know are dropped (defensive,
    // a hostile or newer server could send one), not applied.
    let unknown = changes
        .iter()
        .filter(|c| !by_name.contains_key(c.table.as_str()))
        .count();
    if unknown > 0 {
        warn!("dropping {unknown} change(s) for unknown table(s)");
        out.skipped += unknown;
    }

    // Precompute NOT NULL columns for each Full-mode table once, not per row.
    let mut not_null: HashMap<&str, HashSet<String>> = HashMap::new();
    for t in &schema.tables {
        if matches!(t.mode, SyncMode::Full) {
            not_null.insert(t.name, not_null_columns(tx, t.name)?);
        }
    }

    // Upserts: parents first.
    for table in &schema.tables {
        for change in changes.iter().filter(|c| {
            c.table == table.name && matches!(c.op, ChangeOp::Insert | ChangeOp::Update)
        }) {
            step(
                &mut out,
                table,
                apply_upsert(tx, table, change, &not_null, scope),
            )?;
        }
    }

    // Deletes: children first.
    for table in schema.tables.iter().rev() {
        for change in changes
            .iter()
            .filter(|c| c.table == table.name && matches!(c.op, ChangeOp::Delete))
        {
            step(&mut out, table, apply_delete(tx, table, change))?;
        }
    }

    Ok(out)
}

/// Fold one row's result into the outcome. A constraint violation is skipped and
/// logged; any other SQLite error is returned so `?` rolls the whole batch back
/// (converting to `SyncKitError` at the caller).
fn step(
    out: &mut ApplyOutcome,
    table: &SyncTable,
    result: rusqlite::Result<bool>,
) -> rusqlite::Result<()> {
    match result {
        Ok(true) => {
            out.applied += 1;
            out.changed_tables.insert(table.name.to_string());
            Ok(())
        }
        Ok(false) => {
            out.skipped += 1;
            Ok(())
        }
        Err(e) if is_constraint_violation(&e) => {
            warn!(
                table = table.name,
                "skipping row on constraint violation: {e}"
            );
            out.skipped += 1;
            Ok(())
        }
        Err(e) => Err(e),
    }
}

/// Names of the NOT NULL columns of `table`, read from the live schema so the
/// list can never drift from the migrations.
fn not_null_columns(tx: &Transaction<'_>, table: &str) -> rusqlite::Result<HashSet<String>> {
    let mut stmt = tx.prepare(&format!("PRAGMA table_info({table})"))?;
    let rows = stmt.query_map([], |r| {
        let name: String = r.get("name")?;
        let notnull: i64 = r.get("notnull")?;
        Ok((name, notnull != 0))
    })?;
    let mut set = HashSet::new();
    for row in rows {
        let (name, nn) = row?;
        if nn {
            set.insert(name);
        }
    }
    Ok(set)
}

fn field<'a>(data: &'a Value, col: &str) -> &'a Value {
    data.get(col).unwrap_or(&JSON_NULL)
}

/// Bind a JSON value as a SQLite parameter.
fn json_to_sql(v: &Value) -> Box<dyn ToSql> {
    match v {
        Value::String(s) => Box::new(s.clone()),
        Value::Number(n) => {
            if let Some(i) = n.as_i64() {
                Box::new(i)
            } else if let Some(f) = n.as_f64() {
                Box::new(f)
            } else {
                Box::new(n.to_string())
            }
        }
        Value::Bool(b) => Box::new(i64::from(*b)),
        Value::Null => Box::new(rusqlite::types::Null),
        other => Box::new(other.to_string()),
    }
}

fn is_constraint_violation(e: &rusqlite::Error) -> bool {
    matches!(e, rusqlite::Error::SqliteFailure(err, _) if err.code == ErrorCode::ConstraintViolation)
}

// upsert

fn apply_upsert(
    tx: &Transaction<'_>,
    table: &SyncTable,
    change: &ChangeEntry,
    not_null: &HashMap<&str, HashSet<String>>,
    scope: &str,
) -> rusqlite::Result<bool> {
    let Some(data) = change.data.as_ref().filter(|v| v.is_object()) else {
        return Ok(false); // an upsert with no object payload cannot be applied
    };
    if is_excluded(tx, table, Some(data))? {
        return Ok(false);
    }
    match &table.mode {
        SyncMode::PartialUpdate { set } => apply_partial_update(tx, table, data, set),
        SyncMode::Full => {
            let nn = not_null
                .get(table.name)
                .expect("Full table not_null precomputed");
            apply_full_upsert(tx, table, data, nn, scope)
        }
    }
}

fn apply_full_upsert(
    tx: &Transaction<'_>,
    table: &SyncTable,
    data: &Value,
    not_null: &HashSet<String>,
    scope: &str,
) -> rusqlite::Result<bool> {
    // A remote null for a NOT NULL column is invalid input the changelog never
    // emits; omit such columns so a new row takes the schema default and an
    // existing row keeps its value. Nullable columns keep null so legitimate
    // clears still propagate.
    let present: Vec<&str> = table
        .columns
        .iter()
        .copied()
        .filter(|c| !(field(data, c).is_null() && not_null.contains(*c)))
        .collect();

    // INSERT columns = present whitelist columns + any insert_defaults not already
    // present (e.g. a NOT NULL preserved secret gets its default on first insert).
    let mut insert_cols: Vec<&str> = present.clone();
    let mut default_vals: Vec<&str> = Vec::new();
    for &(col, val) in table.insert_defaults {
        if !insert_cols.contains(&col) {
            insert_cols.push(col);
            default_vals.push(val);
        }
    }
    if insert_cols.is_empty() {
        return Ok(false);
    }

    // Group provenance: stamp a group-scoped table's `group_id` column from the
    // pull scope (never from the payload) so the pulled row is tagged with the
    // group it came from. Personal scope ("") leaves group_id at its default
    // (NULL). Injected into the INSERT columns only, never into the ON CONFLICT
    // SET, so an existing row's group never changes.
    let group_stamp: Option<&str> = table.group_scope.filter(|_| !scope.is_empty());
    if let Some(gcol) = group_stamp {
        insert_cols.push(gcol);
    }

    let pk: HashSet<&str> = table.pk.iter().copied().collect();
    let preserve: HashSet<&str> = table.preserve_local.iter().copied().collect();
    // ON CONFLICT update touches present columns that are neither PK nor a
    // preserved local secret, so an existing row's secrets are never overwritten.
    let update_cols: Vec<&str> = present
        .iter()
        .copied()
        .filter(|c| !pk.contains(c) && !preserve.contains(c))
        .collect();

    let placeholders = (1..=insert_cols.len())
        .map(|i| format!("?{i}"))
        .collect::<Vec<_>>()
        .join(", ");

    let sql = if update_cols.is_empty() {
        // Every column is part of the PK (or nothing to update), ignore conflicts
        // rather than DELETE+INSERT, which would cascade to children.
        format!(
            "INSERT OR IGNORE INTO {} ({}) VALUES ({})",
            table.name,
            insert_cols.join(", "),
            placeholders
        )
    } else {
        let set = update_cols
            .iter()
            .map(|c| format!("{c} = excluded.{c}"))
            .collect::<Vec<_>>()
            .join(", ");
        format!(
            "INSERT INTO {} ({}) VALUES ({}) ON CONFLICT({}) DO UPDATE SET {}",
            table.name,
            insert_cols.join(", "),
            placeholders,
            table.pk.join(", "),
            set
        )
    };

    let mut params: Vec<Box<dyn ToSql>> = present
        .iter()
        .map(|c| json_to_sql(field(data, c)))
        .collect();
    params.extend(
        default_vals
            .iter()
            .map(|v| Box::new((*v).to_string()) as Box<dyn ToSql>),
    );
    if group_stamp.is_some() {
        params.push(Box::new(scope.to_string()) as Box<dyn ToSql>);
    }

    exec(tx, &sql, &params)?;
    Ok(true)
}

fn apply_partial_update(
    tx: &Transaction<'_>,
    table: &SyncTable,
    data: &Value,
    set: &[&str],
) -> rusqlite::Result<bool> {
    let Some(pk) = pk_bindings(table, data, None) else {
        return Ok(false); // no primary key to target
    };
    let set_present: Vec<&str> = set
        .iter()
        .copied()
        .filter(|c| data.get(*c).is_some())
        .collect();
    if set_present.is_empty() {
        return Ok(false);
    }

    let mut params: Vec<Box<dyn ToSql>> = Vec::new();
    let mut idx = 1;
    let set_sql = set_present
        .iter()
        .map(|c| {
            let s = format!("{c} = ?{idx}");
            idx += 1;
            params.push(json_to_sql(field(data, c)));
            s
        })
        .collect::<Vec<_>>()
        .join(", ");
    let where_sql = pk
        .into_iter()
        .map(|(c, v)| {
            let s = format!("{c} = ?{idx}");
            idx += 1;
            params.push(v);
            s
        })
        .collect::<Vec<_>>()
        .join(" AND ");

    let sql = format!("UPDATE {} SET {set_sql} WHERE {where_sql}", table.name);
    exec(tx, &sql, &params)?;
    Ok(true)
}

// delete

fn apply_delete(
    tx: &Transaction<'_>,
    table: &SyncTable,
    change: &ChangeEntry,
) -> rusqlite::Result<bool> {
    if matches!(table.deletes, DeleteMode::Ignore) {
        return Ok(false);
    }
    if is_excluded(tx, table, change.data.as_ref())? {
        return Ok(false);
    }
    let Some(pk) = pk_bindings(
        table,
        change.data.as_ref().unwrap_or(&JSON_NULL),
        Some(&change.row_id),
    ) else {
        return Ok(false); // can't reconstruct the key, skip
    };

    let mut params: Vec<Box<dyn ToSql>> = Vec::new();
    let mut idx = 1;
    let where_sql = pk
        .into_iter()
        .map(|(c, v)| {
            let s = format!("{c} = ?{idx}");
            idx += 1;
            params.push(v);
            s
        })
        .collect::<Vec<_>>()
        .join(" AND ");

    let sql = match &table.deletes {
        // A remote delete on a tombstone table must not hard-DELETE: a local
        // ON DELETE CASCADE would wipe organizational state the remote never
        // meant to touch. COALESCE keeps the earliest tombstone instant.
        DeleteMode::Tombstone { column } => format!(
            "UPDATE {} SET {column} = COALESCE({column}, unixepoch()) WHERE {where_sql}",
            table.name
        ),
        DeleteMode::Hard => format!("DELETE FROM {} WHERE {where_sql}", table.name),
        DeleteMode::Ignore => unreachable!("returned above"),
    };
    exec(tx, &sql, &params)?;
    Ok(true)
}

/// Reconstruct the primary-key bindings for a row.
///
/// Prefers the payload (our generated triggers always carry the PK in `data`,
/// even for hashed rows whose wire row-id is opaque). Falls back to the wire
/// `row_id` for single-key tables written by an older client that stored the key
/// there; a composite key with no payload cannot be reconstructed and yields
/// `None`.
fn pk_bindings(
    table: &SyncTable,
    data: &Value,
    row_id: Option<&str>,
) -> Option<Vec<(&'static str, Box<dyn ToSql>)>> {
    if let Some(obj) = data.as_object() {
        let mut out = Vec::with_capacity(table.pk.len());
        let mut complete = true;
        for pk in table.pk {
            match obj.get(*pk) {
                Some(v) if !v.is_null() => out.push((*pk, json_to_sql(v))),
                _ => {
                    complete = false;
                    break;
                }
            }
        }
        if complete && !out.is_empty() {
            return Some(out);
        }
    }
    // Payload didn't carry the key, fall back to the wire row_id (single-PK only).
    match (row_id, table.pk) {
        (Some(id), [pk]) => Some(vec![(*pk, Box::new(id.to_string()) as Box<dyn ToSql>)]),
        _ => None,
    }
}

// exclusion

/// Whether `table`'s `exclude_where` predicate excludes this row. The predicate
/// is the *include* condition; a row is excluded when it does not hold. Evaluated
/// in SQLite against a one-row derived table so the same predicate text drives
/// both the export trigger and this import guard. No predicate, or no payload to
/// test, means not excluded.
fn is_excluded(
    tx: &Transaction<'_>,
    table: &SyncTable,
    data: Option<&Value>,
) -> rusqlite::Result<bool> {
    let Some(pred) = table.exclude_where else {
        return Ok(false);
    };
    let Some(obj) = data.and_then(|v| v.as_object()) else {
        return Ok(false);
    };

    let cols = table.columns;
    let derived = cols
        .iter()
        .enumerate()
        .map(|(i, c)| format!("?{} AS {c}", i + 1))
        .collect::<Vec<_>>()
        .join(", ");
    // COALESCE(..., 0): a NULL predicate (e.g. NULL key) is treated as included.
    let sql = format!(
        "SELECT COALESCE(NOT ({}), 0) FROM (SELECT {derived}) AS x",
        pred.replace("{row}", "x")
    );
    let params: Vec<Box<dyn ToSql>> = cols
        .iter()
        .map(|c| json_to_sql(field_opt(obj, c)))
        .collect();
    let excluded: i64 = tx.query_row(&sql, params_slice(&params).as_slice(), |r| r.get(0))?;
    Ok(excluded != 0)
}

fn field_opt<'a>(obj: &'a serde_json::Map<String, Value>, col: &str) -> &'a Value {
    obj.get(col).unwrap_or(&JSON_NULL)
}

// exec helpers

fn params_slice(params: &[Box<dyn ToSql>]) -> Vec<&dyn ToSql> {
    params.iter().map(std::convert::AsRef::as_ref).collect()
}

fn exec(tx: &Transaction<'_>, sql: &str, params: &[Box<dyn ToSql>]) -> rusqlite::Result<()> {
    tx.execute(sql, params_slice(params).as_slice())?;
    Ok(())
}

#[cfg(test)]
mod tests {
    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(),
        ])
    }

    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));
            ",
        )
        .unwrap();
        let s = schema();
        conn.execute_batch(&s.migration_sql()).unwrap();
        conn
    }

    fn apply(conn: &mut Connection, changes: &[ChangeEntry]) -> ApplyOutcome {
        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 → skipped, 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.skipped, 1);
    }

    #[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.skipped, 1);
        assert_eq!(
            conn.query_row("SELECT COUNT(*) FROM parent", [], |r| r.get::<_, i64>(0))
                .unwrap(),
            1
        );
    }

    #[test]
    fn unknown_table_change_is_skipped() {
        let mut conn = db();
        let o = apply(&mut conn, &[upsert("nonexistent", "x", json!({"id":"x"}))]);
        assert_eq!(o.applied, 0);
        assert_eq!(o.skipped, 1);
    }
}