//! Migration/trigger generation from a [`SyncSchema`].
//!
//! [`SyncSchema::migration_sql`] emits the full sync DDL, the `sync_state` and
//! `sync_changelog` bookkeeping tables, the HLC ledger, and one
//! INSERT/UPDATE/DELETE trigger set per table, derived entirely from the
//! manifest. Because the trigger's `json_object(...)` projection, the row-id
//! expression, and (later) the apply-side binding all read the same `columns`
//! and `pk`, the historical trigger/snapshot/whitelist drift is designed out.
//!
//! The generated form is deliberately *uniform* rather than a byte-for-byte copy
//! of any one app's hand-written triggers (which differ in naming, integer-key
//! casts, and whether a DELETE stores `NULL` or the key): every DELETE stores the
//! primary key in `data` so the pull side reconstructs the WHERE clause the same
//! way regardless of `RowIdScheme`. The golden tests assert *behaviour*, run the
//! generated triggers, mutate, inspect `sync_changelog`, not string identity.

use super::schema::{DeleteMode, RowIdScheme, SyncMode, SyncSchema, SyncTable};

/// Bookkeeping tables shared by every schema.
const BASE_TABLES: &str = "\
CREATE TABLE IF NOT EXISTS sync_state (
    key   TEXT PRIMARY KEY NOT NULL,
    value TEXT NOT NULL
);
INSERT OR IGNORE INTO sync_state (key, value) VALUES
    ('device_id', ''),
    ('pull_cursor', '0'),
    ('applying_remote', '0'),
    ('auto_sync_enabled', '1'),
    ('sync_interval_minutes', '15'),
    ('last_sync_at', ''),
    ('initial_snapshot_done', '0');

CREATE TABLE IF NOT EXISTS sync_changelog (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    table_name  TEXT NOT NULL,
    op          TEXT NOT NULL,
    row_id      TEXT NOT NULL,
    timestamp   TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    data        TEXT,
    pushed      INTEGER NOT NULL DEFAULT 0,
    hlc_wall    INTEGER,
    hlc_counter INTEGER,
    -- SyncKit Groups scope: '' = personal, otherwise the group id this change
    -- belongs to. Set by the triggers from the row's group_scoped column. An
    -- existing (pre-Groups) changelog is migrated to add this column by
    -- `ensure_scope_schema` (db.rs), since CREATE IF NOT EXISTS cannot alter it.
    scope       TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_sync_changelog_unpushed
    ON sync_changelog(pushed) WHERE pushed = 0;

-- Per-scope pull cursor. '' holds the personal cursor, migrated once from the
-- pre-Groups single `pull_cursor` sync_state value; each group id gets its own.
CREATE TABLE IF NOT EXISTS sync_scope_cursor (
    scope  TEXT PRIMARY KEY NOT NULL,
    cursor INTEGER NOT NULL
);
INSERT OR IGNORE INTO sync_scope_cursor (scope, cursor)
    VALUES ('', CAST((SELECT value FROM sync_state WHERE key = 'pull_cursor') AS INTEGER));

-- Committed HLC per row (conflict gating).
CREATE TABLE IF NOT EXISTS sync_committed_hlc (
    table_name  TEXT NOT NULL,
    row_id      TEXT NOT NULL,
    hlc_wall    INTEGER NOT NULL,
    hlc_counter INTEGER NOT NULL,
    hlc_node    TEXT NOT NULL,
    PRIMARY KEY (table_name, row_id)
) WITHOUT ROWID;
";

/// Provisioned once, never synced. Only emitted when a table hashes its row-id.
const SALT_SEED: &str = "\nINSERT OR IGNORE INTO sync_state (key, value) VALUES
    ('row_id_salt', lower(hex(randomblob(32))));\n";

/// The changelog echo-suppression guard shared by every trigger.
const APPLYING_GUARD: &str = "(SELECT value FROM sync_state WHERE key = 'applying_remote') != '1'";

impl SyncSchema {
    /// Emit the complete sync DDL for this schema: bookkeeping tables, the HLC
    /// ledger, an optional `row_id_salt`, and every table's triggers.
    pub fn migration_sql(&self) -> String {
        let mut out = String::from(BASE_TABLES);
        if self.any_hashed() {
            out.push_str(SALT_SEED);
        }
        for table in &self.tables {
            out.push('\n');
            out.push_str(&triggers_for(table));
        }
        out
    }
}

/// The wire row-id expression for a table under the given row alias (`NEW`/`OLD`
/// in a trigger, or the table name in the initial-snapshot SELECT). Shared with
/// [`super::sync::create_initial_snapshot`] so the snapshot produces byte-identical
/// row-ids to the triggers.
pub(crate) fn row_id_expr(table: &SyncTable, alias: &str) -> String {
    let concat = table
        .pk
        .iter()
        .map(|c| format!("CAST({alias}.{c} AS TEXT)"))
        .collect::<Vec<_>>()
        .join(" || ':' || ");
    match table.row_id {
        RowIdScheme::PrimaryKey => concat,
        RowIdScheme::Hashed => format!(
            "hash_row_id((SELECT value FROM sync_state WHERE key = 'row_id_salt'), {concat})"
        ),
    }
}

/// A `json_object('c', ALIAS.c, ...)` projection over `cols`. Shared with the
/// initial-snapshot SELECT so snapshot payloads match trigger payloads exactly.
pub(crate) fn json_object(alias: &str, cols: &[&str]) -> String {
    let pairs = cols
        .iter()
        .map(|c| format!("'{c}', {alias}.{c}"))
        .collect::<Vec<_>>()
        .join(", ");
    format!("json_object({pairs})")
}

/// The `WHEN` clause: the applying-remote guard, plus this table's exclusion
/// predicate (with `{row}` bound to `alias`) if any.
fn when_clause(table: &SyncTable, alias: &str) -> String {
    match table.exclude_where {
        Some(pred) => format!("{APPLYING_GUARD} AND ({})", pred.replace("{row}", alias)),
        None => APPLYING_GUARD.to_string(),
    }
}

/// One `CREATE TRIGGER` statement (with a preceding idempotent DROP).
fn trigger(name: &str, timing: &str, table: &str, when: &str, values: &str) -> String {
    format!(
        "DROP TRIGGER IF EXISTS {name};\n\
         CREATE TRIGGER {name}\n\
         AFTER {timing} ON {table}\n\
         WHEN {when}\n\
         BEGIN\n\
         INSERT INTO sync_changelog (table_name, op, row_id, data, scope)\n\
         VALUES {values};\n\
         END;\n"
    )
}

/// The changelog `scope` expression for a table under the given row alias. A
/// `group_scoped` table reads its provenance column (`NULL` -> personal `''`);
/// every other table is personal-only, so its scope is the literal `''`.
fn scope_expr(table: &SyncTable, alias: &str) -> String {
    match table.group_scope {
        Some(col) => format!("COALESCE({alias}.{col}, '')"),
        None => "''".to_string(),
    }
}

/// The INSERT/UPDATE/DELETE trigger set for one table.
fn triggers_for(table: &SyncTable) -> String {
    let name = table.name;
    let emitted = table.emitted_columns();
    let mut out = String::new();

    // INSERT, only for Full-mode tables. A PartialUpdate table never inserts
    // (it only reflects state changes on rows that already exist locally).
    if matches!(table.mode, SyncMode::Full) {
        let values = format!(
            "('{name}', 'INSERT', {}, {}, {})",
            row_id_expr(table, "NEW"),
            json_object("NEW", &emitted),
            scope_expr(table, "NEW"),
        );
        out.push_str(&trigger(
            &format!("sync_{name}_insert"),
            "INSERT",
            name,
            &when_clause(table, "NEW"),
            &values,
        ));
    }

    // UPDATE, always. PartialUpdate additionally guards on one of its `set`
    // columns actually changing (null-safe `IS NOT`).
    let update_when = match &table.mode {
        SyncMode::Full => when_clause(table, "NEW"),
        SyncMode::PartialUpdate { set } => {
            let changed = set
                .iter()
                .map(|c| format!("OLD.{c} IS NOT NEW.{c}"))
                .collect::<Vec<_>>()
                .join(" OR ");
            format!("{} AND ({changed})", when_clause(table, "NEW"))
        }
    };
    let update_values = format!(
        "('{name}', 'UPDATE', {}, {}, {})",
        row_id_expr(table, "NEW"),
        json_object("NEW", &emitted),
        scope_expr(table, "NEW"),
    );
    out.push_str(&trigger(
        &format!("sync_{name}_update"),
        "UPDATE",
        name,
        &update_when,
        &update_values,
    ));

    // DELETE, generated unless deletes are ignored. Tombstone tables still emit
    // a normal export trigger (a *local* delete must propagate); the tombstone is
    // an apply-side concern. The payload carries the primary key so the pull side
    // reconstructs the WHERE clause without the (possibly hashed) row-id.
    if !matches!(table.deletes, DeleteMode::Ignore) {
        let delete_values = format!(
            "('{name}', 'DELETE', {}, {}, {})",
            row_id_expr(table, "OLD"),
            json_object("OLD", table.pk),
            scope_expr(table, "OLD"),
        );
        out.push_str(&trigger(
            &format!("sync_{name}_delete"),
            "DELETE",
            name,
            &when_clause(table, "OLD"),
            &delete_values,
        ));
    }

    out
}

#[cfg(test)]
mod tests {
    use super::super::schema::{SyncSchema, SyncTable};
    use rusqlite::Connection;
    use sha2::{Digest, Sha256};

    /// A row observed in `sync_changelog`.
    #[derive(Debug, PartialEq, Eq)]
    struct Entry {
        table: String,
        op: String,
        row_id: String,
        data: Option<String>,
    }

    /// Build an in-memory DB with the domain tables the tests mutate, register a
    /// deterministic `hash_row_id`, then run the generated migration.
    fn setup(schema: &SyncSchema) -> Connection {
        let conn = Connection::open_in_memory().unwrap();
        // Deterministic stand-in for the engine's real hash_row_id (B2). Any
        // stable non-identity function proves the row-id was hashed, not leaked.
        conn.create_scalar_function(
            "hash_row_id",
            2,
            rusqlite::functions::FunctionFlags::SQLITE_DETERMINISTIC,
            |ctx| {
                let salt: String = ctx.get(0)?;
                let key: String = ctx.get(1)?;
                let mut h = Sha256::new();
                h.update(salt.as_bytes());
                h.update(key.as_bytes());
                Ok(hex::encode(h.finalize()))
            },
        )
        .unwrap();

        conn.execute_batch(
            "
            CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);
            CREATE TABLE samp (hash TEXT PRIMARY KEY, name TEXT, deleted_at INTEGER);
            CREATE TABLE tags (sample_hash TEXT, tag TEXT, PRIMARY KEY (sample_hash, tag));
            CREATE TABLE items (id TEXT PRIMARY KEY, is_read INTEGER, is_starred INTEGER, title TEXT);
            CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);
            ",
        )
        .unwrap();

        conn.execute_batch(&schema.migration_sql()).unwrap();
        conn
    }

    /// The representative manifest: one table per trigger-affecting policy.
    fn schema() -> SyncSchema {
        SyncSchema::new(vec![
            SyncTable::full("note", &["id", "name"]),
            SyncTable::full("samp", &["hash", "name", "deleted_at"])
                .pk(&["hash"])
                .hashed()
                .tombstone("deleted_at"),
            SyncTable::full("tags", &["sample_hash", "tag"])
                .pk(&["sample_hash", "tag"])
                .hashed(),
            SyncTable::full("items", &["id", "is_read", "is_starred"])
                .partial_update(&["is_read", "is_starred"])
                .ignore_deletes(),
            SyncTable::full("cfg", &["key", "value"])
                .pk(&["key"])
                .exclude_where("{row}.key NOT LIKE 'sync\\_%' ESCAPE '\\'"),
        ])
    }

    fn changelog(conn: &Connection) -> Vec<Entry> {
        let mut stmt = conn
            .prepare("SELECT table_name, op, row_id, data FROM sync_changelog ORDER BY id")
            .unwrap();
        let rows = stmt
            .query_map([], |r| {
                Ok(Entry {
                    table: r.get(0)?,
                    op: r.get(1)?,
                    row_id: r.get(2)?,
                    data: r.get(3)?,
                })
            })
            .unwrap();
        rows.map(|r| r.unwrap()).collect()
    }

    #[test]
    fn full_cleartext_roundtrips_insert_update_delete() {
        let conn = setup(&schema());
        conn.execute("INSERT INTO note (id, name) VALUES ('n1', 'hello')", [])
            .unwrap();
        conn.execute("UPDATE note SET name = 'bye' WHERE id = 'n1'", [])
            .unwrap();
        conn.execute("DELETE FROM note WHERE id = 'n1'", [])
            .unwrap();

        let log = changelog(&conn);
        assert_eq!(log.len(), 3);
        assert_eq!(
            (log[0].op.as_str(), log[0].row_id.as_str()),
            ("INSERT", "n1")
        );
        assert!(
            log[0]
                .data
                .as_deref()
                .unwrap()
                .contains("\"name\":\"hello\"")
        );
        assert_eq!(log[1].op, "UPDATE");
        // Cleartext delete: row_id is the key, data carries the PK (not NULL).
        assert_eq!(
            (log[2].op.as_str(), log[2].row_id.as_str()),
            ("DELETE", "n1")
        );
        assert_eq!(log[2].data.as_deref(), Some("{\"id\":\"n1\"}"));
    }

    #[test]
    fn applying_remote_flag_suppresses_capture() {
        let conn = setup(&schema());
        conn.execute(
            "UPDATE sync_state SET value = '1' WHERE key = 'applying_remote'",
            [],
        )
        .unwrap();
        conn.execute("INSERT INTO note (id, name) VALUES ('n2', 'x')", [])
            .unwrap();
        assert!(changelog(&conn).is_empty(), "echo must not be captured");
    }

    #[test]
    fn hashed_rowid_hides_key_and_delete_carries_pk() {
        let conn = setup(&schema());
        conn.execute("INSERT INTO samp (hash, name) VALUES ('deadbeef', 's')", [])
            .unwrap();
        conn.execute("DELETE FROM samp WHERE hash = 'deadbeef'", [])
            .unwrap();

        let log = changelog(&conn);
        assert_eq!(log.len(), 2);
        // The wire row-id must not be the cleartext content hash.
        assert_ne!(log[0].row_id, "deadbeef");
        assert_eq!(log[0].row_id.len(), 64, "sha256 hex");
        assert!(
            log[0]
                .data
                .as_deref()
                .unwrap()
                .contains("\"hash\":\"deadbeef\"")
        );
        // Tombstone tables STILL emit a normal export DELETE; row-id hashed, PK in data.
        assert_eq!(log[1].op, "DELETE");
        assert_eq!(log[1].row_id, log[0].row_id);
        assert_eq!(log[1].data.as_deref(), Some("{\"hash\":\"deadbeef\"}"));
    }

    #[test]
    fn hashed_composite_key_concatenates_before_hashing() {
        let conn = setup(&schema());
        conn.execute(
            "INSERT INTO tags (sample_hash, tag) VALUES ('abc', 'kick')",
            [],
        )
        .unwrap();
        let log = changelog(&conn);
        assert_eq!(log.len(), 1);
        // Composite key "abc:kick" is concatenated then hashed; both parts
        // survive in data. Salt is random per run, so assert shape not value.
        assert_eq!(log[0].row_id.len(), 64);
        assert!(
            log[0]
                .data
                .as_deref()
                .unwrap()
                .contains("\"sample_hash\":\"abc\"")
        );
        assert!(log[0].data.as_deref().unwrap().contains("\"tag\":\"kick\""));
    }

    #[test]
    fn partial_update_is_update_only_and_change_gated() {
        let conn = setup(&schema());
        // Insert never captured (no INSERT trigger for PartialUpdate).
        conn.execute(
            "INSERT INTO items (id, is_read, is_starred, title) VALUES ('i1', 0, 0, 't')",
            [],
        )
        .unwrap();
        assert!(changelog(&conn).is_empty());

        // Updating a non-set column does not fire (change guard on is_read/is_starred).
        conn.execute("UPDATE items SET title = 't2' WHERE id = 'i1'", [])
            .unwrap();
        assert!(changelog(&conn).is_empty());

        // Updating a set column fires exactly one UPDATE carrying pk + set cols.
        conn.execute("UPDATE items SET is_read = 1 WHERE id = 'i1'", [])
            .unwrap();
        let log = changelog(&conn);
        assert_eq!(log.len(), 1);
        assert_eq!(
            (log[0].op.as_str(), log[0].row_id.as_str()),
            ("UPDATE", "i1")
        );
        let data = log[0].data.as_deref().unwrap();
        assert!(data.contains("\"is_read\":1") && data.contains("\"is_starred\":0"));
    }

    #[test]
    fn ignore_deletes_emits_no_delete_row() {
        let conn = setup(&schema());
        conn.execute(
            "INSERT INTO items (id, is_read, is_starred) VALUES ('i2', 1, 0)",
            [],
        )
        .unwrap();
        conn.execute("UPDATE items SET is_read = 0 WHERE id = 'i2'", [])
            .unwrap(); // baseline capture
        let before = changelog(&conn).len();
        conn.execute("DELETE FROM items WHERE id = 'i2'", [])
            .unwrap();
        assert_eq!(
            changelog(&conn).len(),
            before,
            "ignored delete must not capture"
        );
    }

    #[test]
    fn exclude_where_filters_symmetrically_on_export() {
        let conn = setup(&schema());
        conn.execute(
            "INSERT INTO cfg (key, value) VALUES ('sync_cursor', '5')",
            [],
        )
        .unwrap(); // excluded
        conn.execute("INSERT INTO cfg (key, value) VALUES ('theme', 'dark')", [])
            .unwrap(); // included
        let log = changelog(&conn);
        assert_eq!(log.len(), 1);
        assert_eq!(log[0].row_id, "theme");
    }

    /// A row observed in `sync_changelog` including its scope.
    #[derive(Debug, PartialEq, Eq)]
    struct ScopedEntry {
        op: String,
        row_id: String,
        scope: String,
    }

    fn scoped_changelog(conn: &Connection) -> Vec<ScopedEntry> {
        let mut stmt = conn
            .prepare("SELECT op, row_id, scope FROM sync_changelog ORDER BY id")
            .unwrap();
        let rows = stmt
            .query_map([], |r| {
                Ok(ScopedEntry {
                    op: r.get(0)?,
                    row_id: r.get(1)?,
                    scope: r.get(2)?,
                })
            })
            .unwrap();
        rows.map(|r| r.unwrap()).collect()
    }

    #[test]
    fn group_scoped_table_stamps_scope_from_group_id() {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(
            "CREATE TABLE task (id TEXT PRIMARY KEY, name TEXT, group_id TEXT);
             CREATE TABLE cfg (key TEXT PRIMARY KEY, value TEXT);",
        )
        .unwrap();
        let schema = SyncSchema::new(vec![
            SyncTable::full("task", &["id", "name", "group_id"]).group_scoped("group_id"),
            SyncTable::full("cfg", &["key", "value"]).pk(&["key"]),
        ]);
        conn.execute_batch(&schema.migration_sql()).unwrap();

        // A personal task row (group_id NULL) -> scope ''.
        conn.execute(
            "INSERT INTO task (id, name, group_id) VALUES ('t1', 'a', NULL)",
            [],
        )
        .unwrap();
        // A group task row -> scope = the group id.
        conn.execute(
            "INSERT INTO task (id, name, group_id) VALUES ('t2', 'b', 'grp-9')",
            [],
        )
        .unwrap();
        // Moving a task into a group is an UPDATE that re-scopes it.
        conn.execute("UPDATE task SET group_id = 'grp-9' WHERE id = 't1'", [])
            .unwrap();
        // A personal-only (non-group_scoped) table is always scope ''.
        conn.execute("INSERT INTO cfg (key, value) VALUES ('theme', 'dark')", [])
            .unwrap();
        // A delete carries the group scope of the deleted row (OLD).
        conn.execute("DELETE FROM task WHERE id = 't2'", [])
            .unwrap();

        let log = scoped_changelog(&conn);
        assert_eq!(
            log,
            vec![
                ScopedEntry {
                    op: "INSERT".into(),
                    row_id: "t1".into(),
                    scope: String::new()
                },
                ScopedEntry {
                    op: "INSERT".into(),
                    row_id: "t2".into(),
                    scope: "grp-9".into()
                },
                ScopedEntry {
                    op: "UPDATE".into(),
                    row_id: "t1".into(),
                    scope: "grp-9".into()
                },
                ScopedEntry {
                    op: "INSERT".into(),
                    row_id: "theme".into(),
                    scope: String::new()
                },
                ScopedEntry {
                    op: "DELETE".into(),
                    row_id: "t2".into(),
                    scope: "grp-9".into()
                },
            ]
        );
    }

    #[test]
    fn scope_cursor_seeded_from_legacy_pull_cursor() {
        let conn = Connection::open_in_memory().unwrap();
        let schema = SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]);
        conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
            .unwrap();
        conn.execute_batch(&schema.migration_sql()).unwrap();
        // Personal scope cursor exists, seeded from pull_cursor's default '0'.
        let personal: i64 = conn
            .query_row(
                "SELECT cursor FROM sync_scope_cursor WHERE scope = ''",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(personal, 0);
    }

    #[test]
    fn salt_seeded_only_when_a_table_is_hashed() {
        // schema() has hashed tables → salt present.
        let with = setup(&schema());
        let salt: i64 = with
            .query_row(
                "SELECT COUNT(*) FROM sync_state WHERE key = 'row_id_salt'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(salt, 1);

        // A schema with no hashed table must not seed a salt.
        let plain = SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]);
        assert!(!plain.migration_sql().contains("row_id_salt"));
    }
}