//! The sync half of portable config: a [`ConfigSpec`] as a [`SyncTable`].
//!
//! [`synckit-config`] holds the local store an app reads and writes and the
//! posture declaration; this turns that declaration into the sync engine's
//! terms. An app calls [`config_sync_table`] to fold config into its
//! [`SyncSchema`] and [`install_policy`] once at open to seed the filter the
//! generated trigger reads. After that the app's `ConfigStore` writes replicate
//! transparently — the consumer never learns whether sync is wired.
//!
//! ## The posture filter, and why it is an allowlist
//!
//! A config key/value table syncs every key unless told otherwise. The
//! [`exclude_where`](SyncTable::exclude_where) predicate here admits a key only
//! if a `config_key_policy` row marks it replicated:
//!
//! ```sql
//! EXISTS (SELECT 1 FROM config_key_policy p WHERE p.key = {row}.key AND p.replicated = 1)
//! ```
//!
//! An **allowlist**, not a denylist, and that is the security property. A key
//! with no policy row does not match, so it never enters the changelog — an
//! unclassified key stays on the device. audiofiles reached the same predicate
//! the hard way: its earlier `key != 'loose_files'` denylist never covered
//! `mirror_path`/`mirror_enabled`, letting a hostile server steer a local write
//! root across the boundary (fuzz-2026-07-21 #3). The policy table is seeded from
//! the spec, whose default posture is `Local`, so the allowlist is
//! [`ConfigSpec`]'s fail-closed rule carried into SQL.

use rusqlite::Connection;
use synckit_config::ConfigSpec;

use super::schema::SyncTable;
use crate::error::Result;

/// The policy table the trigger predicate joins. One row per declared key.
///
/// Keyed by `key` alone: one config table per app, so keys do not collide across
/// tables. A second config table would need a `table` column here; no app has
/// one, and adding it is cheap when one does.
const POLICY_TABLE: &str = "config_key_policy";

/// The `exclude_where` predicate: admit a key only if its policy row replicates.
///
/// Fixed SQL, so a `&'static str`. `{row}` is substituted with the trigger's
/// `NEW`/`OLD` alias by the migration layer, symmetrically on export and import.
const POLICY_FILTER: &str = "EXISTS (SELECT 1 FROM config_key_policy p \
     WHERE p.key = {row}.key AND p.replicated = 1)";

/// The [`SyncTable`] for a config: a key/value table filtered by posture.
///
/// Fold the result into the app's [`SyncSchema`](super::schema::SyncSchema). The
/// generated triggers capture a change only for a key the policy admits, so a
/// `Local` key is written to the store and never enqueued. Pair with
/// [`install_policy`], which seeds the table the filter reads.
pub fn config_sync_table(spec: &ConfigSpec) -> SyncTable {
    SyncTable::full(spec.table(), &["key", "value"])
        .pk(&["key"])
        .exclude_where(POLICY_FILTER)
}

/// Create and seed the policy table the [`config_sync_table`] filter reads.
///
/// Idempotent, and safe to call at every open: it recreates the table's rows
/// from the spec each time, so a key whose posture changed in code takes effect
/// on the next start, and a key removed from the spec loses its row and stops
/// replicating. The seed is the spec's [`policy_rows`](ConfigSpec::policy_rows),
/// so an undeclared key has no row and the filter cannot admit it.
///
/// Wrapped in a transaction so a crash mid-seed cannot leave the policy table
/// describing a set of keys that never all existed at once.
pub fn install_policy(conn: &mut Connection, spec: &ConfigSpec) -> Result<()> {
    let tx = conn.transaction()?;
    tx.execute_batch(&format!(
        "CREATE TABLE IF NOT EXISTS {POLICY_TABLE} (\
             key        TEXT PRIMARY KEY,\
             replicated INTEGER NOT NULL\
         ) WITHOUT ROWID;"
    ))?;
    // Rebuild the rows from the spec each open. Delete-then-insert rather than
    // upsert so a key dropped from the spec loses its row (and stops syncing)
    // rather than lingering with a stale posture.
    tx.execute(&format!("DELETE FROM {POLICY_TABLE}"), [])?;
    {
        let mut stmt = tx.prepare(&format!(
            "INSERT INTO {POLICY_TABLE} (key, replicated) VALUES (?1, ?2)"
        ))?;
        for row in spec.policy_rows() {
            stmt.execute(rusqlite::params![row.key, i64::from(row.replicated)])?;
        }
    }
    tx.commit()?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::store::schema::SyncSchema;
    use synckit_config::{ConfigStore, Posture};

    const SPEC: ConfigSpec = ConfigSpec::new(
        "app_config",
        &[
            ("theme", Posture::Synced),
            ("sidebar_visible", Posture::Synced),
            ("mirror_path", Posture::Local),
        ],
    );

    /// The changelog row-ids captured so far, in order.
    fn captured(conn: &Connection) -> Vec<String> {
        let mut stmt = conn
            .prepare("SELECT row_id FROM sync_changelog ORDER BY id")
            .unwrap();
        stmt.query_map([], |r| r.get::<_, String>(0))
            .unwrap()
            .map(std::result::Result::unwrap)
            .collect()
    }

    /// A DB with the config table, its store, its policy, and its sync triggers.
    ///
    /// The order matters and is the order an app opens in: the store creates the
    /// config table, `install_policy` seeds the filter, then the schema's
    /// migration installs the triggers that read it.
    fn wired() -> (Connection, ConfigStore) {
        let mut conn = Connection::open_in_memory().unwrap();
        let store = ConfigStore::open(&conn, &SPEC).unwrap();
        install_policy(&mut conn, &SPEC).unwrap();
        let sql = SyncSchema::new(vec![config_sync_table(&SPEC)]).migration_sql();
        conn.execute_batch(&sql).unwrap();
        (conn, store)
    }

    // The point of the whole layer: a Synced key is captured, a Local one is
    // not, and neither the store nor the caller said anything about sync.
    #[test]
    fn a_synced_key_is_captured_and_a_local_key_is_not() {
        let (conn, store) = wired();
        store.set(&conn, "theme", "akari-night").unwrap();
        store.set(&conn, "mirror_path", "/mnt/samples").unwrap();

        assert_eq!(
            captured(&conn),
            ["theme"],
            "the preference syncs; the local path never enters the changelog",
        );
    }

    // The allowlist's reason for being: a key nobody classified must not sync,
    // even though the table syncs by default. This is the fuzz-hardened rule in
    // SQL rather than in Rust.
    #[test]
    fn an_undeclared_key_is_never_captured() {
        let (conn, store) = wired();
        store
            .set(&conn, "some_new_local_path", "/etc/secret")
            .unwrap();
        assert!(
            captured(&conn).is_empty(),
            "an unclassified key has no policy row and cannot be admitted",
        );
    }

    #[test]
    fn every_synced_key_captures() {
        let (conn, store) = wired();
        store.set(&conn, "theme", "flatwhite").unwrap();
        store.set(&conn, "sidebar_visible", "true").unwrap();
        assert_eq!(captured(&conn), ["theme", "sidebar_visible"]);
    }

    // Re-seeding drops a key removed from the spec, so it stops syncing rather
    // than lingering with a stale posture.
    #[test]
    fn reseeding_a_narrower_spec_stops_the_dropped_key_syncing() {
        let (mut conn, store) = wired();
        store.set(&conn, "theme", "flatwhite").unwrap();
        assert_eq!(captured(&conn), ["theme"]);

        const NARROWER: ConfigSpec =
            ConfigSpec::new("app_config", &[("mirror_path", Posture::Local)]);
        install_policy(&mut conn, &NARROWER).unwrap();

        conn.execute("DELETE FROM sync_changelog", []).unwrap();
        store.set(&conn, "theme", "akari-night").unwrap();
        assert!(
            captured(&conn).is_empty(),
            "theme is no longer in the policy, so it no longer syncs",
        );
    }

    #[test]
    fn install_policy_is_idempotent() {
        let (mut conn, _store) = wired();
        // Every open re-seeds; a second call must not error or duplicate rows.
        install_policy(&mut conn, &SPEC).unwrap();
        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM config_key_policy", [], |r| r.get(0))
            .unwrap();
        assert_eq!(count, 3, "one row per declared key, not doubled");
    }
}