//! The engine's owned SQLite connection and bookkeeping-state primitives.
//!
//! `SyncStore` owns a private `rusqlite` connection to the app's database file
//! rather than borrowing the app's async pool, so the app's own persistence layer
//! is untouched. Under WAL, which the consuming apps already use, the engine's
//! short read/apply transactions coexist with the app's connections on the same
//! file. This module provides the low-level pieces the rest of the engine builds
//! on: opening/configuring that connection, the `sync_state` key-value store, the
//! crash-safe `applying_remote` echo-suppression transaction, and device
//! registration.

use std::path::PathBuf;
use std::sync::Arc;

use rusqlite::{Connection, OptionalExtension, TransactionBehavior, functions::FunctionFlags};
use sha2::{Digest, Sha256};
use uuid::Uuid;

use crate::error::{Result, SyncKitError};
use crate::ids::DeviceId;

/// Where the engine gets its SQLite connection.
///
/// [`DbSource::path`] is the common case, the engine opens the file itself.
/// [`DbSource::factory`] lets an app hand back a connection it configured (custom
/// pragmas, shared cache, a VFS); the engine still applies its own required setup
/// (WAL, `foreign_keys`, the `hash_row_id` function) on top.
#[derive(Clone)]
pub enum DbSource {
    /// Open `Connection::open(path)` on demand.
    Path(PathBuf),
    /// Call an app-provided factory to obtain a raw connection.
    Factory(Arc<dyn Fn() -> rusqlite::Result<Connection> + Send + Sync>),
}

impl std::fmt::Debug for DbSource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Path(p) => f.debug_tuple("Path").field(p).finish(),
            Self::Factory(_) => f.write_str("Factory(..)"),
        }
    }
}

impl DbSource {
    /// A source that opens the SQLite file at `path`.
    pub fn path(path: impl Into<PathBuf>) -> Self {
        Self::Path(path.into())
    }

    /// A source that calls `factory` for each connection.
    pub fn factory(
        factory: impl Fn() -> rusqlite::Result<Connection> + Send + Sync + 'static,
    ) -> Self {
        Self::Factory(Arc::new(factory))
    }

    /// Open and fully configure a connection: WAL journaling, foreign-key
    /// enforcement, and the `hash_row_id` function the generated triggers call.
    ///
    /// This is synchronous SQLite work; callers driving it from async code should
    /// run it on a blocking thread (the engine's push/pull loops do).
    pub fn open(&self) -> Result<Connection> {
        let conn = match self {
            Self::Path(p) => Connection::open(p)?,
            Self::Factory(f) => f()?,
        };
        configure_connection(&conn)?;
        Ok(conn)
    }
}

/// Apply the engine's required per-connection setup. Idempotent.
pub(crate) fn configure_connection(conn: &Connection) -> Result<()> {
    // WAL lets the engine's connection coexist with the app's; foreign_keys ON is
    // the default the apply path relaxes only for `references_unsynced` tables.
    conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
    register_hash_row_id(conn)?;
    ensure_scope_schema(conn)?;
    Ok(())
}

/// Idempotently bring a pre-Groups bookkeeping schema up to the multi-scope
/// shape: add `sync_changelog.scope` if the table exists without it, and ensure
/// `sync_scope_cursor` exists (seeded from the old single `pull_cursor`).
///
/// `CREATE TABLE IF NOT EXISTS` in [`SyncSchema::migration_sql`] handles a fresh
/// DB, but it cannot *add a column* to an existing changelog and SQLite has no
/// `ADD COLUMN IF NOT EXISTS`, so the ALTER is guarded on `pragma_table_info`
/// here. On a fresh DB (no bookkeeping yet) this is a no-op, `migration_sql`
/// creates everything with the column. Runs on every connection open; the guards
/// make it a cheap no-op once applied.
pub(crate) fn ensure_scope_schema(conn: &Connection) -> Result<()> {
    let bookkeeping_exists = conn
        .query_row(
            "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'sync_changelog'",
            [],
            |_| Ok(()),
        )
        .optional()?
        .is_some();
    if !bookkeeping_exists {
        return Ok(());
    }

    let has_scope = conn
        .query_row(
            "SELECT 1 FROM pragma_table_info('sync_changelog') WHERE name = 'scope'",
            [],
            |_| Ok(()),
        )
        .optional()?
        .is_some();
    if !has_scope {
        conn.execute_batch(
            "ALTER TABLE sync_changelog ADD COLUMN scope TEXT NOT NULL DEFAULT '';",
        )?;
    }

    conn.execute_batch(
        "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)
             SELECT '', CAST(value AS INTEGER) FROM sync_state WHERE key = 'pull_cursor';",
    )?;
    Ok(())
}

/// Read a scope's pull cursor (`''` = personal), defaulting to `0` when absent.
pub fn get_scope_cursor(conn: &Connection, scope: &str) -> Result<i64> {
    conn.query_row(
        "SELECT cursor FROM sync_scope_cursor WHERE scope = ?1",
        [scope],
        |r| r.get(0),
    )
    .optional()
    .map(|o| o.unwrap_or(0))
    .map_err(Into::into)
}

/// Persist a scope's pull cursor.
pub fn set_scope_cursor(conn: &Connection, scope: &str, cursor: i64) -> Result<()> {
    conn.execute(
        "INSERT INTO sync_scope_cursor (scope, cursor) VALUES (?1, ?2) \
         ON CONFLICT(scope) DO UPDATE SET cursor = excluded.cursor",
        (scope, cursor),
    )?;
    Ok(())
}

/// Register `hash_row_id(salt, key) -> TEXT`: lowercase hex of
/// `SHA-256(salt || ":" || key)`. Deterministic, keyed by the per-vault
/// `row_id_salt`, so the wire row-id of a content-bearing table never carries
/// cleartext. Matches the audiofiles definition byte-for-byte so its existing
/// hashed row-ids stay stable across the port.
fn register_hash_row_id(conn: &Connection) -> Result<()> {
    conn.create_scalar_function(
        "hash_row_id",
        2,
        FunctionFlags::SQLITE_DETERMINISTIC | FunctionFlags::SQLITE_UTF8,
        |ctx| {
            let salt: String = ctx.get(0)?;
            let key: String = ctx.get(1)?;
            Ok(hash_row_id(&salt, &key))
        },
    )?;
    Ok(())
}

/// The `hash_row_id` computation, exposed for the apply path (which must derive
/// the same wire row-id in Rust when it needs to address a hashed row).
pub(crate) fn hash_row_id(salt: &str, key: &str) -> String {
    let mut hasher = Sha256::new();
    hasher.update(salt.as_bytes());
    hasher.update(b":");
    hasher.update(key.as_bytes());
    let digest = hasher.finalize();
    let mut hex = String::with_capacity(64);
    for byte in digest {
        use std::fmt::Write;
        let _ = write!(hex, "{byte:02x}");
    }
    hex
}

// sync_state key-value store

/// Read a `sync_state` value, or `None` if the key is absent.
pub fn get_sync_state(conn: &Connection, key: &str) -> Result<Option<String>> {
    conn.query_row("SELECT value FROM sync_state WHERE key = ?1", [key], |r| {
        r.get::<_, String>(0)
    })
    .optional()
    .map_err(Into::into)
}

/// Read a `sync_state` value, falling back to `default` when absent.
pub fn get_sync_state_or(conn: &Connection, key: &str, default: &str) -> Result<String> {
    Ok(get_sync_state(conn, key)?.unwrap_or_else(|| default.to_string()))
}

/// Write a `sync_state` value, inserting or updating.
pub fn set_sync_state(conn: &Connection, key: &str, value: &str) -> Result<()> {
    conn.execute(
        "INSERT INTO sync_state (key, value) VALUES (?1, ?2) \
         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
        (key, value),
    )?;
    Ok(())
}

/// Count unpushed changelog entries.
pub fn count_pending_changes(conn: &Connection) -> Result<i64> {
    conn.query_row(
        "SELECT COUNT(*) FROM sync_changelog WHERE pushed = 0",
        [],
        |r| r.get(0),
    )
    .map_err(Into::into)
}

// applying_remote echo-suppression

/// Run `apply` with the `applying_remote` flag raised, inside a single
/// `IMMEDIATE` transaction.
///
/// The flag is set to `'1'` and back to `'0'` inside the same transaction as the
/// writes `apply` performs. The changelog triggers, evaluating on this same
/// connection, see the uncommitted `'1'` and suppress the echo; on commit the
/// persisted value is `'0'` again. If `apply` returns `Err` (or the process dies
/// mid-transaction) the whole transaction, including the flag flip, rolls back,
/// so the flag can never be stranded at `'1'` and silently swallow later local
/// edits. That stranding is exactly the failure the apps defended against with a
/// start-of-sync reset; here it is structurally impossible.
pub fn with_applying_remote<T>(
    conn: &mut Connection,
    apply: impl FnOnce(&rusqlite::Transaction<'_>) -> Result<T>,
) -> Result<T> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    set_sync_state(&tx, "applying_remote", "1")?;
    let out = apply(&tx)?;
    set_sync_state(&tx, "applying_remote", "0")?;
    tx.commit()?;
    Ok(out)
}

/// Defensively clear a stuck `applying_remote` flag at start of sync. With
/// [`with_applying_remote`] the flag cannot actually strand, but a database
/// written by an older, non-transactional client might carry a stale `'1'`.
pub fn clear_applying_remote(conn: &Connection) -> Result<()> {
    set_sync_state(conn, "applying_remote", "0")
}

// device identity

/// The registered device id, or `None` if this install has not registered yet.
pub fn device_id(conn: &Connection) -> Result<Option<DeviceId>> {
    match get_sync_state(conn, "device_id")? {
        Some(s) if !s.is_empty() => {
            let uuid = Uuid::parse_str(&s)
                .map_err(|e| SyncKitError::Database(format!("malformed device_id: {e}")))?;
            Ok(Some(DeviceId::new(uuid)))
        }
        _ => Ok(None),
    }
}

/// Persist the registered device id.
pub fn set_device_id(conn: &Connection, id: DeviceId) -> Result<()> {
    set_sync_state(conn, "device_id", &id.as_uuid().to_string())
}

#[cfg(test)]
mod tests {
    use super::super::schema::{SyncSchema, SyncTable};
    use super::*;

    /// An in-memory DB configured like the engine would, with a one-table
    /// migration applied so triggers exist to exercise the flag.
    fn db() -> Connection {
        let conn = Connection::open_in_memory().unwrap();
        configure_connection(&conn).unwrap();
        conn.execute_batch("CREATE TABLE note (id TEXT PRIMARY KEY, name TEXT);")
            .unwrap();
        let schema = SyncSchema::new(vec![SyncTable::full("note", &["id", "name"])]);
        conn.execute_batch(&schema.migration_sql()).unwrap();
        conn
    }

    fn changelog_count(conn: &Connection) -> i64 {
        conn.query_row("SELECT COUNT(*) FROM sync_changelog", [], |r| r.get(0))
            .unwrap()
    }

    #[test]
    fn hash_row_id_matches_audiofiles_formula() {
        // sha256("salt:key") lowercase hex, the exact audiofiles definition.
        let mut h = Sha256::new();
        h.update(b"salt:key");
        let expected = hex::encode(h.finalize());
        assert_eq!(hash_row_id("salt", "key"), expected);
        // Pinned against an independent SHA-256 as well, so the row id stays a
        // fixed wire value rather than one that merely agrees with whatever
        // hasher this crate happens to link.
        assert_eq!(
            hash_row_id("salt", "key"),
            "c82b2df7a36eec83a4b5a9d83093a209006ae9029b6cb13323500da263d9fd06"
        );
        assert_eq!(hash_row_id("salt", "key").len(), 64);
        // Salt actually keys the hash (not just concatenated blindly).
        assert_ne!(hash_row_id("a", "bc"), hash_row_id("ab", "c"));
    }

    #[test]
    fn hash_row_id_is_registered_and_callable_in_sql() {
        let conn = db();
        let via_sql: String = conn
            .query_row("SELECT hash_row_id('s', 'k')", [], |r| r.get(0))
            .unwrap();
        assert_eq!(via_sql, hash_row_id("s", "k"));
    }

    #[test]
    fn foreign_keys_enforced_by_default() {
        let conn = db();
        let fk: i64 = conn
            .query_row("PRAGMA foreign_keys", [], |r| r.get(0))
            .unwrap();
        assert_eq!(fk, 1);
    }

    #[test]
    fn sync_state_get_set_and_absent() {
        let conn = db();
        assert_eq!(get_sync_state(&conn, "nope").unwrap(), None);
        assert_eq!(
            get_sync_state_or(&conn, "nope", "fallback").unwrap(),
            "fallback"
        );
        set_sync_state(&conn, "k", "v1").unwrap();
        assert_eq!(get_sync_state(&conn, "k").unwrap().as_deref(), Some("v1"));
        set_sync_state(&conn, "k", "v2").unwrap(); // upsert
        assert_eq!(get_sync_state(&conn, "k").unwrap().as_deref(), Some("v2"));
    }

    #[test]
    fn applying_remote_suppresses_capture_and_commits_flag_low() {
        let mut conn = db();
        with_applying_remote(&mut conn, |tx| {
            tx.execute("INSERT INTO note (id, name) VALUES ('n1', 'x')", [])?;
            Ok(())
        })
        .unwrap();
        // The write landed, but its trigger was suppressed (echo not captured)...
        let n: i64 = conn
            .query_row("SELECT COUNT(*) FROM note", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n, 1);
        assert_eq!(changelog_count(&conn), 0);
        // ...and the persisted flag is back to '0'.
        assert_eq!(
            get_sync_state(&conn, "applying_remote").unwrap().as_deref(),
            Some("0")
        );
    }

    #[test]
    fn applying_remote_rolls_back_writes_and_flag_on_error() {
        let mut conn = db();
        let result: Result<()> = with_applying_remote(&mut conn, |tx| {
            tx.execute("INSERT INTO note (id, name) VALUES ('n2', 'y')", [])?;
            Err(SyncKitError::Internal("boom mid-apply".into())) // simulate a crash
        });
        assert!(result.is_err());
        // The write inside the failed transaction rolled back...
        let n: i64 = conn
            .query_row("SELECT COUNT(*) FROM note", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n, 0);
        // ...and the flag was never stranded at '1'.
        assert_eq!(
            get_sync_state(&conn, "applying_remote").unwrap().as_deref(),
            Some("0")
        );
    }

    #[test]
    fn normal_write_is_captured_outside_applying_remote() {
        let conn = db();
        conn.execute("INSERT INTO note (id, name) VALUES ('n3', 'z')", [])
            .unwrap();
        assert_eq!(changelog_count(&conn), 1);
    }

    #[test]
    fn device_id_roundtrip() {
        let conn = db();
        assert_eq!(device_id(&conn).unwrap(), None); // seeded as '' → None
        let id = DeviceId::new(Uuid::from_u128(0x42));
        set_device_id(&conn, id).unwrap();
        assert_eq!(device_id(&conn).unwrap(), Some(id));
    }

    #[test]
    fn malformed_device_id_is_an_error_not_a_panic() {
        let conn = db();
        set_sync_state(&conn, "device_id", "not-a-uuid").unwrap();
        assert!(device_id(&conn).is_err());
    }

    #[test]
    fn ensure_scope_schema_upgrades_a_legacy_changelog() {
        // Simulate a pre-Groups database: a sync_changelog WITHOUT the scope
        // column, and a pull_cursor mid-stream at 42.
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(
            "CREATE TABLE sync_state (key TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL);
             INSERT INTO sync_state (key, value) VALUES ('pull_cursor', '42');
             CREATE TABLE sync_changelog (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 table_name TEXT NOT NULL, op TEXT NOT NULL, row_id TEXT NOT NULL,
                 data TEXT, pushed INTEGER NOT NULL DEFAULT 0
             );",
        )
        .unwrap();
        // No scope column yet.
        let has_scope_before: bool = conn
            .query_row(
                "SELECT 1 FROM pragma_table_info('sync_changelog') WHERE name = 'scope'",
                [],
                |_| Ok(true),
            )
            .optional()
            .unwrap()
            .unwrap_or(false);
        assert!(!has_scope_before);

        // Idempotent: run it twice.
        ensure_scope_schema(&conn).unwrap();
        ensure_scope_schema(&conn).unwrap();

        // The scope column now exists and defaults to personal.
        conn.execute(
            "INSERT INTO sync_changelog (table_name, op, row_id) VALUES ('t','INSERT','r')",
            [],
        )
        .unwrap();
        let scope: String = conn
            .query_row(
                "SELECT scope FROM sync_changelog WHERE row_id = 'r'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(scope, "");

        // The personal cursor was migrated from the legacy pull_cursor value.
        assert_eq!(get_scope_cursor(&conn, "").unwrap(), 42);
        assert_eq!(get_scope_cursor(&conn, "grp-1").unwrap(), 0); // absent -> 0
        set_scope_cursor(&conn, "grp-1", 7).unwrap();
        assert_eq!(get_scope_cursor(&conn, "grp-1").unwrap(), 7);
    }

    #[test]
    fn salt_is_provisioned_once_and_stable_across_reruns() {
        let conn = Connection::open_in_memory().unwrap();
        configure_connection(&conn).unwrap();
        conn.execute_batch("CREATE TABLE samp (hash TEXT PRIMARY KEY, name TEXT);")
            .unwrap();
        let schema = SyncSchema::new(vec![
            SyncTable::full("samp", &["hash", "name"])
                .pk(&["hash"])
                .hashed(),
        ]);
        conn.execute_batch(&schema.migration_sql()).unwrap();
        let salt1 = get_sync_state(&conn, "row_id_salt").unwrap().unwrap();
        assert_eq!(salt1.len(), 64); // hex of 32 random bytes
        // Re-running the migration must not rotate the salt (INSERT OR IGNORE).
        conn.execute_batch(&schema.migration_sql()).unwrap();
        let salt2 = get_sync_state(&conn, "row_id_salt").unwrap().unwrap();
        assert_eq!(salt1, salt2);
    }
}