//! Reconcile sando.toml into SQLite at startup.
//!
//! Tiers and nodes are config-driven; mutable per-tier state (current version,
//! burn-in clock) lives in tier_state and must survive across syncs. Stale
//! rows (tier or node removed from the TOML) are deleted, but tier_state for
//! a removed tier is preserved silently — the FK is cleared by deleting the
//! parent last. If you actually need to forget a retired tier, do it by hand.

use crate::domain::AppId;
use crate::topology::Topology;
use anyhow::Result;
use sqlx::SqlitePool;

/// Reconcile one product's topology into the tier and node tables.
///
/// Every statement is scoped to `app`. Without that, syncing one product would
/// treat every other product's tiers as removed from config and delete them:
/// the stale-row sweep asks "which rows are not in this TOML", and one TOML has
/// never described more than one product.
pub async fn sync(pool: &SqlitePool, app: &AppId, topo: &Topology) -> Result<()> {
    let mut tx = pool.begin().await?;

    let want_tiers: Vec<&str> = topo.tiers.iter().map(|t| t.name.as_str()).collect();
    let want_nodes: Vec<(&str, &str)> = topo
        .tiers
        .iter()
        .flat_map(|t| {
            t.nodes
                .iter()
                .map(move |n| (t.name.as_str(), n.name.as_str()))
        })
        .collect();

    // Drop stale nodes first (FK to tiers).
    let existing_nodes: Vec<(String, String)> =
        sqlx::query_as("SELECT name, tier FROM nodes WHERE app = ?")
            .bind(app)
            .fetch_all(&mut *tx)
            .await?;
    for (name, tier) in existing_nodes {
        if !want_nodes.iter().any(|(t, n)| *t == tier && *n == name) {
            sqlx::query("DELETE FROM nodes WHERE app = ? AND name = ?")
                .bind(app)
                .bind(&name)
                .execute(&mut *tx)
                .await?;
        }
    }

    // Drop stale tiers. tier_state rows referencing them are preserved by
    // clearing the FK target only after a manual cleanup — for now we just
    // refuse to delete a tier that still has tier_state with non-null version.
    let existing_tiers: Vec<String> = sqlx::query_scalar("SELECT name FROM tiers WHERE app = ?")
        .bind(app)
        .fetch_all(&mut *tx)
        .await?;
    for t in existing_tiers {
        if !want_tiers.contains(&t.as_str()) {
            let in_use: Option<String> = sqlx::query_scalar(
                "SELECT current_version FROM tier_state WHERE app = ? AND tier = ?",
            )
            .bind(app)
            .bind(&t)
            .fetch_optional(&mut *tx)
            .await?
            .flatten();
            anyhow::ensure!(
                in_use.is_none(),
                "refusing to remove tier {t} from app `{app}`'s config: tier_state still pins a \
                 version. clean it up by hand before editing the topology.",
            );
            sqlx::query("DELETE FROM tier_state WHERE app = ? AND tier = ?")
                .bind(app)
                .bind(&t)
                .execute(&mut *tx)
                .await?;
            sqlx::query("DELETE FROM tiers WHERE app = ? AND name = ?")
                .bind(app)
                .bind(&t)
                .execute(&mut *tx)
                .await?;
        }
    }

    // Upsert tiers in declaration order; `ord` mirrors that order so the
    // promotion sequence is queryable without re-reading the TOML.
    for (i, t) in topo.tiers.iter().enumerate() {
        sqlx::query(
            "INSERT INTO tiers (app, name, ord, provisioned, canary)
             VALUES (?, ?, ?, ?, ?)
             ON CONFLICT(app, name) DO UPDATE SET
                 ord = excluded.ord,
                 provisioned = excluded.provisioned,
                 canary = excluded.canary",
        )
        .bind(app)
        .bind(&t.name)
        .bind(i as i64)
        .bind(t.provisioned as i64)
        .bind(t.canary.as_str())
        .execute(&mut *tx)
        .await?;

        sqlx::query("INSERT OR IGNORE INTO tier_state (app, tier) VALUES (?, ?)")
            .bind(app)
            .bind(&t.name)
            .execute(&mut *tx)
            .await?;

        for n in &t.nodes {
            sqlx::query(
                "INSERT INTO nodes (app, name, tier, ssh_target, release_root)
                 VALUES (?, ?, ?, ?, ?)
                 ON CONFLICT(app, name) DO UPDATE SET
                     tier = excluded.tier,
                     ssh_target = excluded.ssh_target,
                     release_root = excluded.release_root",
            )
            .bind(app)
            .bind(&n.name)
            .bind(&t.name)
            .bind(&n.ssh_target)
            .bind(&n.release_root)
            .execute(&mut *tx)
            .await?;
        }
    }

    tx.commit().await?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::topology::{BackupConfig, CanaryPolicy, Gate, Node, RepoConfig, Tier, Topology};
    use sqlx::sqlite::SqlitePoolOptions;

    async fn fresh_pool() -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();
        pool
    }

    fn app() -> AppId {
        AppId::default()
    }

    fn topo(tiers: Vec<Tier>) -> Topology {
        Topology {
            repo: Some(RepoConfig {
                bare_path: "/tmp/x".into(),
                branch: "main".into(),
                upstream: None,
            }),
            backup: vec![BackupConfig {
                name: "server".into(),
                source: "file:///tmp/b".into(),
                local_path: "/tmp/b".into(),
            }],
            tiers,
            aux_repos: Vec::new(),
        }
    }

    fn tier(name: &str, provisioned: bool, nodes: Vec<Node>) -> Tier {
        Tier {
            name: name.into(),
            provisioned,
            gates: vec![Gate::BootSmoke],
            canary: CanaryPolicy::Sequential,
            nodes,
        }
    }

    fn node(name: &str) -> Node {
        Node {
            platform: None,
            name: name.into(),
            ssh_target: format!("deploy@{name}"),
            release_root: "/opt/mnw".into(),
            service_name: "makenotwork.service".into(),
            health_url: None,
            config_check_env_file: None,
            actuate: crate::topology::default_actuate(),
            observe: crate::topology::default_observe(),
            companions: Vec::new(),
        }
    }

    #[tokio::test]
    async fn syncs_tiers_nodes_and_inits_tier_state() {
        let pool = fresh_pool().await;
        let t = topo(vec![
            tier("host", true, vec![]),
            tier("a", true, vec![node("testnot-1")]),
            tier("c", false, vec![]),
        ]);

        sync(&pool, &app(), &t).await.unwrap();

        let tier_names: Vec<String> = sqlx::query_scalar("SELECT name FROM tiers ORDER BY ord")
            .fetch_all(&pool)
            .await
            .unwrap();
        assert_eq!(tier_names, vec!["host", "a", "c"]);

        let node_names: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes")
            .fetch_all(&pool)
            .await
            .unwrap();
        assert_eq!(node_names, vec!["testnot-1"]);

        let state_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(state_count, 3);
    }

    #[tokio::test]
    async fn second_sync_is_idempotent() {
        let pool = fresh_pool().await;
        let t = topo(vec![
            tier("host", true, vec![]),
            tier("a", true, vec![node("n1")]),
        ]);
        sync(&pool, &app(), &t).await.unwrap();
        sync(&pool, &app(), &t).await.unwrap();

        let nodes: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM nodes")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(nodes, 1);
        let states: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(states, 2);
    }

    #[tokio::test]
    async fn removing_node_from_config_drops_row() {
        let pool = fresh_pool().await;
        let t1 = topo(vec![tier("a", true, vec![node("n1"), node("n2")])]);
        sync(&pool, &app(), &t1).await.unwrap();
        let t2 = topo(vec![tier("a", true, vec![node("n1")])]);
        sync(&pool, &app(), &t2).await.unwrap();

        let nodes: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes")
            .fetch_all(&pool)
            .await
            .unwrap();
        assert_eq!(nodes, vec!["n1"]);
    }

    /// Syncing one product leaves every other product's tiers and nodes alone.
    ///
    /// The sweep at the top of `sync` asks "which rows are not in this TOML",
    /// and a TOML describes one product. Unscoped, syncing pom at startup would
    /// answer "all of MNW's" and delete them — and since startup syncs each app
    /// in turn, the last one to run would be the only one left standing.
    #[tokio::test]
    async fn syncing_one_app_does_not_touch_another() {
        let pool = fresh_pool().await;
        let mnw = AppId::new("mnw");
        let pom = AppId::new("pom");
        sync(
            &pool,
            &mnw,
            &topo(vec![tier("host", true, vec![node("n1")])]),
        )
        .await
        .unwrap();
        sync(
            &pool,
            &pom,
            &topo(vec![tier("host", true, vec![node("n2")])]),
        )
        .await
        .unwrap();

        // Both survive, and a tier name they share is two rows, not one.
        let tiers: Vec<(String, String)> =
            sqlx::query_as("SELECT app, name FROM tiers ORDER BY app")
                .fetch_all(&pool)
                .await
                .unwrap();
        assert_eq!(
            tiers,
            vec![
                ("mnw".to_string(), "host".to_string()),
                ("pom".to_string(), "host".to_string())
            ]
        );
        let nodes: Vec<(String, String)> =
            sqlx::query_as("SELECT app, name FROM nodes ORDER BY app")
                .fetch_all(&pool)
                .await
                .unwrap();
        assert_eq!(
            nodes,
            vec![
                ("mnw".to_string(), "n1".to_string()),
                ("pom".to_string(), "n2".to_string())
            ]
        );
        // And one tier_state row each, not one shared.
        let states: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tier_state")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(states, 2);
    }

    /// A tier pinned in one product does not block removing the same-named tier
    /// from another.
    #[tokio::test]
    async fn a_pin_in_one_app_does_not_block_another_apps_edit() {
        let pool = fresh_pool().await;
        let mnw = AppId::new("mnw");
        let pom = AppId::new("pom");
        let two = topo(vec![tier("host", true, vec![]), tier("a", true, vec![])]);
        sync(&pool, &mnw, &two).await.unwrap();
        sync(&pool, &pom, &two).await.unwrap();

        // MNW pins a version on tier a.
        sqlx::query("INSERT INTO versions (app, version, git_sha, built_at, artifact_path) VALUES ('mnw','0.1.0','deadbeef','2026-05-22T00:00:00Z','/r/0.1.0')")
            .execute(&pool).await.unwrap();
        sqlx::query(
            "UPDATE tier_state SET current_version = '0.1.0' WHERE app = 'mnw' AND tier = 'a'",
        )
        .execute(&pool)
        .await
        .unwrap();

        // pom dropping ITS tier a is fine; MNW's pin is not pom's business.
        sync(&pool, &pom, &topo(vec![tier("host", true, vec![])]))
            .await
            .unwrap();
        // MNW dropping the same tier is still refused.
        let err = sync(&pool, &mnw, &topo(vec![tier("host", true, vec![])]))
            .await
            .unwrap_err();
        assert!(err.to_string().contains("tier_state still pins"), "{err}");
    }

    #[tokio::test]
    async fn refuses_to_drop_tier_with_pinned_version() {
        let pool = fresh_pool().await;
        let t1 = topo(vec![tier("host", true, vec![]), tier("a", true, vec![])]);
        sync(&pool, &app(), &t1).await.unwrap();

        // Simulate a version being deployed on tier a.
        sqlx::query("INSERT INTO versions (app, version, git_sha, built_at, artifact_path) VALUES ('mnw', '0.1.0', 'deadbeef', '2026-05-22T00:00:00Z', '/r/0.1.0')")
            .execute(&pool).await.unwrap();
        sqlx::query(
            "UPDATE tier_state SET current_version = '0.1.0' WHERE app = 'mnw' AND tier = 'a'",
        )
        .execute(&pool)
        .await
        .unwrap();

        let t2 = topo(vec![tier("host", true, vec![])]);
        let err = sync(&pool, &app(), &t2).await.unwrap_err();
        assert!(
            err.to_string().contains("tier_state still pins"),
            "got: {err}"
        );
    }
}