use crate::error::Result;
use crate::state::AppState;
use axum::extract::{Path, State, WebSocketUpgrade};
use axum::response::IntoResponse;
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use sqlx::Row;

pub fn router(state: AppState) -> Router {
    let prom = state.prom.clone();
    Router::new()
        .route("/state", get(get_state))
        .route("/promote/{tier}", post(promote))
        .route("/rollback/{tier}", post(rollback))
        .route("/rebuild", post(rebuild))
        .route("/backup/fetch", post(backup_fetch))
        .route("/events", get(events_ws))
        .with_state(state)
        .route("/metrics", get(crate::metrics::render).with_state(prom))
}

#[derive(Serialize)]
struct StateView {
    tiers: Vec<TierView>,
}

#[derive(Serialize)]
struct TierView {
    name: String,
    ord: i64,
    provisioned: bool,
    canary: String,
    current_version: Option<String>,
    previous_version: Option<String>,
    burn_in_started_at: Option<String>,
    nodes: Vec<String>,
    gates: Vec<GateView>,
}

#[derive(Serialize)]
struct GateView {
    kind: String,
    passed: Option<bool>,
    finished_at: Option<String>,
    detail: Option<String>,
}

async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> {
    let rows = sqlx::query(
        "SELECT t.name, t.ord, t.provisioned, t.canary,
                ts.current_version, ts.previous_version, ts.burn_in_started_at
         FROM tiers t
         LEFT JOIN tier_state ts ON ts.tier = t.name
         ORDER BY t.ord",
    )
    .fetch_all(&s.pool)
    .await?;

    let mut tiers = Vec::with_capacity(rows.len());
    for r in rows {
        let name: String = r.get("name");
        let current_version: Option<String> = r.get("current_version");

        let nodes: Vec<String> = sqlx::query_scalar("SELECT name FROM nodes WHERE tier = ? ORDER BY name")
            .bind(&name)
            .fetch_all(&s.pool)
            .await?;

        let gates: Vec<GateView> = if let Some(ver) = current_version.as_ref() {
            // Most recent gate_runs row per gate_kind for (tier, current_version).
            sqlx::query(
                "SELECT gate_kind, passed, finished_at, detail
                 FROM gate_runs g
                 WHERE tier = ?1 AND version = ?2
                   AND id = (SELECT MAX(id) FROM gate_runs
                             WHERE tier = ?1 AND version = ?2 AND gate_kind = g.gate_kind)
                 ORDER BY gate_kind",
            )
            .bind(&name)
            .bind(ver)
            .fetch_all(&s.pool)
            .await?
            .into_iter()
            .map(|gr| GateView {
                kind: gr.get("gate_kind"),
                passed: gr.get::<Option<i64>, _>("passed").map(|v| v != 0),
                finished_at: gr.get("finished_at"),
                detail: gr.get("detail"),
            })
            .collect()
        } else {
            Vec::new()
        };

        tiers.push(TierView {
            name,
            ord: r.get("ord"),
            provisioned: r.get::<i64, _>("provisioned") != 0,
            canary: r.get("canary"),
            current_version,
            previous_version: r.get("previous_version"),
            burn_in_started_at: r.get("burn_in_started_at"),
            nodes,
            gates,
        });
    }

    Ok(Json(StateView { tiers }))
}

#[derive(Deserialize)]
struct PromoteBody {
    version: String,
    #[serde(default)]
    hotfix: bool,
    #[serde(default)]
    reset_burn_in: bool,
}

async fn promote(
    State(s): State<AppState>,
    Path(tier): Path<String>,
    Json(body): Json<PromoteBody>,
) -> Result<Json<serde_json::Value>> {
    let idx = s.topo.tiers.iter().position(|t| t.name == tier)
        .ok_or(crate::error::Error::NotFound)?;
    if idx == 0 {
        return Err(crate::error::Error::GateBlocked(
            "cannot /promote to the first tier; use /rebuild".into(),
        ));
    }
    let target = &s.topo.tiers[idx];
    let source = &s.topo.tiers[idx - 1];

    // 1. Predecessor must have all of its gates green for this version (with
    //    optional hotfix override that skips burn_in).
    let pending = unsatisfied_gates(&s.pool, &source.name, &body.version, body.hotfix).await?;
    if !pending.is_empty() {
        return Err(crate::error::Error::GateBlocked(format!(
            "{} gate(s) not satisfied on tier {}: {}",
            pending.len(),
            source.name,
            pending.join(", "),
        )));
    }

    // 2. Look up the artifact for this version.
    let bin: Option<(String,)> = sqlx::query_as(
        "SELECT artifact_path FROM versions WHERE version = ?",
    )
    .bind(&body.version)
    .fetch_optional(&s.pool)
    .await
    .map_err(crate::error::Error::Db)?;
    let Some((bin,)) = bin else {
        return Err(crate::error::Error::NotFound);
    };
    let bin_path = std::path::PathBuf::from(bin);

    // 3. Deploy to each node. Sequential canary is the only policy
    //    implemented in v0; parallel is a one-line change once we trust the
    //    sequential path.
    for node in &target.nodes {
        crate::deploy::deploy_node(node, &body.version, &bin_path)
            .await
            .map_err(crate::error::Error::Other)?;
        let now = chrono::Utc::now().to_rfc3339();
        sqlx::query(
            "INSERT INTO deploys (version, tier, node, started_at, finished_at, outcome, hotfix, reset_burn_in)
             VALUES (?, ?, ?, ?, ?, 'ok', ?, ?)",
        )
        .bind(&body.version).bind(&target.name).bind(&node.name)
        .bind(&now).bind(&now)
        .bind(body.hotfix as i64).bind(body.reset_burn_in as i64)
        .execute(&s.pool).await.map_err(crate::error::Error::Db)?;
    }

    // 4. Advance tier_state. burn_in_started_at is set to now so the target
    //    tier's burn_in gate starts ticking. reset_burn_in on the *source*
    //    tier nulls its clock only when the operator explicitly asked for it.
    let prev: Option<String> = sqlx::query_scalar(
        "SELECT current_version FROM tier_state WHERE tier = ?",
    )
    .bind(&target.name)
    .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?.flatten();
    sqlx::query(
        "UPDATE tier_state SET previous_version = ?, current_version = ?, burn_in_started_at = ?
         WHERE tier = ?",
    )
    .bind(prev)
    .bind(&body.version)
    .bind(chrono::Utc::now().to_rfc3339())
    .bind(&target.name)
    .execute(&s.pool).await.map_err(crate::error::Error::Db)?;

    if body.reset_burn_in {
        sqlx::query("UPDATE tier_state SET burn_in_started_at = NULL WHERE tier = ?")
            .bind(&source.name)
            .execute(&s.pool).await.map_err(crate::error::Error::Db)?;
    }

    tracing::info!(
        version = %body.version, tier = %target.name,
        hotfix = body.hotfix, reset_burn_in = body.reset_burn_in,
        "promote complete",
    );

    Ok(Json(serde_json::json!({
        "tier": target.name,
        "version": body.version,
        "nodes_deployed": target.nodes.iter().map(|n| n.name.clone()).collect::<Vec<_>>(),
    })))
}

/// Returns the kinds of gates on `tier` that have not (yet) passed for
/// `version`. `hotfix` suppresses the burn_in requirement only.
async fn unsatisfied_gates(
    pool: &sqlx::SqlitePool,
    tier: &str,
    version: &str,
    hotfix: bool,
) -> std::result::Result<Vec<String>, crate::error::Error> {
    // We need the configured gate list for the tier to know what *should*
    // pass. The route handler has Topology in hand and could pass it in, but
    // the DB also captures it implicitly via gate_runs rows. Simplest correct
    // answer: re-read from topology via tier name; the caller has it.
    // For now we inspect the latest gate_runs.
    let rows: Vec<(String, Option<i64>)> = sqlx::query_as(
        "SELECT gate_kind, passed FROM gate_runs g
         WHERE tier = ?1 AND version = ?2
           AND id = (SELECT MAX(id) FROM gate_runs
                     WHERE tier = ?1 AND version = ?2 AND gate_kind = g.gate_kind)",
    )
    .bind(tier).bind(version)
    .fetch_all(pool).await.map_err(crate::error::Error::Db)?;
    let mut bad = Vec::new();
    for (kind, passed) in rows {
        if hotfix && kind == "burn_in" {
            continue;
        }
        if passed.unwrap_or(0) == 0 {
            bad.push(kind);
        }
    }
    Ok(bad)
}

async fn rollback(
    State(s): State<AppState>,
    Path(tier): Path<String>,
) -> Result<Json<serde_json::Value>> {
    let target = s.topo.tiers.iter().find(|t| t.name == tier)
        .ok_or(crate::error::Error::NotFound)?;

    let row: Option<(Option<String>, Option<String>)> = sqlx::query_as(
        "SELECT current_version, previous_version FROM tier_state WHERE tier = ?",
    )
    .bind(&tier)
    .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?;
    let (Some(current), Some(previous)) = row.unwrap_or((None, None)) else {
        return Err(crate::error::Error::GateBlocked(
            "no previous_version to roll back to".into(),
        ));
    };

    let bin: Option<(String,)> = sqlx::query_as(
        "SELECT artifact_path FROM versions WHERE version = ?",
    )
    .bind(&previous)
    .fetch_optional(&s.pool).await.map_err(crate::error::Error::Db)?;
    let Some((bin,)) = bin else {
        return Err(crate::error::Error::GateBlocked(
            format!("previous version {previous} has no artifact_path; rollback impossible"),
        ));
    };
    let bin_path = std::path::PathBuf::from(bin);

    for node in &target.nodes {
        crate::deploy::deploy_node(node, &previous, &bin_path)
            .await
            .map_err(crate::error::Error::Other)?;
    }

    sqlx::query(
        "UPDATE tier_state SET current_version = ?, previous_version = ?, burn_in_started_at = NULL
         WHERE tier = ?",
    )
    .bind(&previous)
    .bind(&current)
    .bind(&tier)
    .execute(&s.pool).await.map_err(crate::error::Error::Db)?;

    tracing::warn!(tier = %tier, from = %current, to = %previous, "rollback complete");

    Ok(Json(serde_json::json!({
        "tier": tier,
        "rolled_back_from": current,
        "now_running": previous,
    })))
}

#[derive(Deserialize, Default)]
struct RebuildBody {
    /// Specific sha to build. If absent, resolve `topo.repo.branch` from the bare repo.
    #[serde(default)]
    sha: Option<String>,
}

async fn rebuild(
    State(s): State<AppState>,
    body: Option<Json<RebuildBody>>,
) -> Result<Json<serde_json::Value>> {
    let body = body.map(|Json(b)| b).unwrap_or_default();
    let sha = match body.sha {
        Some(s) => s,
        None => crate::git::resolve_ref(
            std::path::Path::new(&s.topo.repo.bare_path),
            &s.topo.repo.branch,
        )
        .await
        .map_err(crate::error::Error::Other)?,
    };

    tracing::info!(sha = %sha, "rebuild requested");

    let pool = s.pool.clone();
    let cfg = s.cfg.clone();
    let topo = s.topo.clone();
    let sha_for_task = sha.clone();
    tokio::spawn(async move {
        if let Err(e) = crate::build::build_and_run_mm(pool, cfg, topo, sha_for_task.clone()).await {
            tracing::error!(sha = %sha_for_task, error = %e, "rebuild pipeline failed");
        }
    });

    Ok(Json(serde_json::json!({ "accepted": true, "sha": sha })))
}

async fn backup_fetch(State(s): State<AppState>) -> Result<Json<serde_json::Value>> {
    let fb = crate::backup::fetch(&s.pool, &s.cfg, &s.topo)
        .await
        .map_err(crate::error::Error::Other)?;
    Ok(Json(serde_json::json!({
        "source": fb.source,
        "local_path": fb.local_path,
        "byte_size": fb.byte_size,
    })))
}

async fn events_ws(ws: WebSocketUpgrade, State(_s): State<AppState>) -> impl IntoResponse {
    ws.on_upgrade(|_socket| async move {
        // tail of deploy/gate events for the TUI
    })
}