//! The gates that ask a running thing whether it is serving: the staged
//! artifact on the build host, the deployed nodes, and the public pages through
//! the CDN.

use super::log::{append_to_log, gate_chunk_cb, stream_into_log};
use super::{GateCtx, NodeProbe};
use crate::classify;
use crate::domain::{GateKind, GateRunId};
use crate::outcome::{GateBlocker, GateFailure, GateOutcome, PassNote};
use anyhow::Result;
use ops_core::live_log::LiveLog;
use ops_core::remote::LogSink;
use ops_exec::{DiscardSink, sh_quote};

pub(super) async fn boot_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
    let bin: Option<(String,)> =
        sqlx::query_as("SELECT artifact_path FROM versions WHERE app = ? AND version = ?")
            .bind(&ctx.cfg.id)
            .bind(&ctx.version)
            .fetch_optional(&ctx.pool)
            .await?;
    let Some((bin,)) = bin else {
        return Ok(GateOutcome::blocked(GateBlocker::ArtifactMissing {
            version: ctx.version.clone(),
        }));
    };

    // Readiness smoke: start the binary and confirm it serves `GET /health`
    // within the window, not merely that the process stays up. Panics in main,
    // missing config, and port-bind failures still surface as an early exit; a
    // process that comes up but never serves /health is now its own failure.
    //
    // The server requires DATABASE_URL or it panics on config load before
    // we can observe anything. We point it at the scratch DB (already
    // migrated by the build step and refreshed by migration_dry_run if
    // that gate ran first). SCAN_ENABLED=false skips loading YARA rules
    // from /opt/makenotwork/yara-rules which doesn't exist on the build
    // host. SANDO_BOOT_SMOKE_PORT tells the smoke server which loopback port
    // to bind so we know where to probe. Other config has sane optional defaults.
    let mut cmd = tokio::process::Command::new(&bin);
    cmd.env("SANDO_BOOT_SMOKE", "1")
        .env("SANDO_BOOT_SMOKE_PORT", ctx.cfg.boot_smoke_port.to_string())
        .env("SCAN_ENABLED", "false")
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .kill_on_drop(true);
    if let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() {
        cmd.env("DATABASE_URL", scratch_url);
    }
    let log_path = ctx.log_path(GateKind::BootSmoke);
    let log_ref = ctx.log_ref(GateKind::BootSmoke);
    let mut child = match cmd.spawn() {
        Ok(c) => c,
        Err(e) => {
            // Spawn failures get a one-off log line via LiveLog so the
            // on-disk file still exists for `GET /logs/...`.
            let mut log = LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await;
            log.write_chunk(format!("spawn: {e}\n").as_bytes()).await;
            log.close().await;
            return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
                message: e.to_string(),
            })
            .with_log_ref(log_ref));
        }
    };

    // The boot smoke window is 3s. Drain stdout/stderr concurrently through
    // a shared LiveLog sink so the operator sees panics/log lines stream in
    // real time before the kill, AND the on-disk log gets the full byte
    // stream for post-mortem reads. The drainers exit when their pipe
    // closes — which happens when the child exits naturally or after kill.
    let log = std::sync::Arc::new(tokio::sync::Mutex::new(
        LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await,
    ));
    let stdout_task = tokio::spawn(stream_into_log(child.stdout.take(), log.clone()));
    let stderr_task = tokio::spawn(stream_into_log(child.stderr.take(), log.clone()));

    // Poll readiness across the 3s window instead of a flat sleep: GET /health
    // must return 2xx. A crash mid-window short-circuits to the exit-code
    // failure path (try_wait below); a process that stays up but never serves
    // /health is a distinct readiness failure.
    let probe_timeout = std::time::Duration::from_millis(500);
    let started = std::time::Instant::now();
    let window = std::time::Duration::from_secs(3);
    let mut probe_ok_after: Option<u32> = None;
    let mut last_probe_err = "never responded".to_string();
    let mut early_exit = None;
    while started.elapsed() < window {
        if let Some(status) = child.try_wait()? {
            early_exit = Some(status);
            break;
        }
        match tokio::time::timeout(probe_timeout, probe_health(ctx.cfg.boot_smoke_port)).await {
            Ok(Ok(())) => {
                probe_ok_after = Some(started.elapsed().as_millis() as u32);
                break;
            }
            Ok(Err(e)) => last_probe_err = e,
            Err(_) => last_probe_err = "probe timed out".to_string(),
        }
        tokio::time::sleep(std::time::Duration::from_millis(150)).await;
    }

    // Stop the child unless it already exited, then drain the log tasks.
    let exit = match early_exit {
        Some(status) => Some(status),
        None => {
            let e = child.try_wait()?;
            if e.is_none() {
                let _ = child.kill().await;
            }
            e
        }
    };
    // The streamed bytes already landed in the live log and the on-disk file for
    // the post-mortem reader. Drain the join handles to avoid hangs.
    let _ = stdout_task.await;
    let _ = stderr_task.await;
    // Unique owner of the Arc at this point (both tasks dropped their clones).
    if let Ok(mutex) = std::sync::Arc::try_unwrap(log) {
        mutex.into_inner().close().await;
    }

    match (exit, probe_ok_after) {
        // Exited on its own within the window — a crash/panic/bind failure.
        (Some(status), _) => {
            let failure = classify::classify_boot_smoke(status.code());
            Ok(GateOutcome::failed(failure).with_log_ref(log_ref))
        }
        // Stayed up and served /health — readiness proven.
        (None, Some(after_ms)) => {
            Ok(GateOutcome::passed(PassNote::HealthyProbe { after_ms }).with_log_ref(log_ref))
        }
        // Stayed up but never served /health — started, not ready.
        (None, None) => Ok(GateOutcome::failed(GateFailure::BootHealthProbeFailed {
            last_error: last_probe_err,
        })
        .with_log_ref(log_ref)),
    }
}

/// One readiness probe of the boot-smoke server: connect to `127.0.0.1:port`
/// and `GET /health`, returning `Ok(())` only on a `200`. A hand-rolled HTTP/1.0
/// request over a raw `TcpStream` keeps the outbound probe dependency-free
/// (reqwest is dev-only); the smoke server serves the one route over axum, which
/// speaks 1.0. `Err` carries a short reason for the operator's failure note. The
/// caller wraps each call in a timeout.
pub(super) async fn probe_health(port: u16) -> std::result::Result<(), String> {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let mut stream = tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port))
        .await
        .map_err(|e| format!("connect: {e}"))?;
    stream
        .write_all(b"GET /health HTTP/1.0\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        .await
        .map_err(|e| format!("write: {e}"))?;
    let mut buf = Vec::new();
    stream
        .read_to_end(&mut buf)
        .await
        .map_err(|e| format!("read: {e}"))?;
    let text = String::from_utf8_lossy(&buf);
    let status_line = text.lines().next().unwrap_or("");
    if status_line.contains(" 200 ") {
        Ok(())
    } else {
        Err(format!("unexpected status line: {status_line:?}"))
    }
}

/// `node_health` — the post-deploy gate that proves the *deployed nodes* are
/// serving, recording one outcome per (tier, version) that the next promote
/// checks. Distinct from `boot_smoke`, which boots the staged artifact on the
/// build host: this probes each node over the same executor the deploy used, so
/// a node that took a corrupt artifact, wrong-arch binary, or failed restart is
/// caught here rather than waved through. Fails closed: any
/// unhealthy node fails the gate, and an empty node set is `Blocked`.
/// Load the tier's public pages in a real browser and fail if the JavaScript
/// did not run.
///
/// The only gate here that crosses the CDN. `boot_smoke` runs on the build host
/// and `node_health` reaches a node over its executor, so between them nothing
/// requests the site the way a visitor does. A CDN holding a module from an
/// earlier deploy can fail to link against the fresh one beside it, and since
/// the bundle's entry point side-effect-imports every island, one bad link takes
/// all of them down together while every artifact is individually correct: right
/// markup, right stylesheets, every module answering 200 with current bytes.
/// Composition is observable in a browser and nowhere else.
///
/// Runs on the daemon host rather than on a node, because it is a *client*: it
/// should reach the site through whatever the public reaches it through, and a
/// probe that ran on the origin would inherit the blind spot this exists to
/// close.
pub(super) async fn page_smoke(ctx: &GateCtx) -> Result<GateOutcome> {
    let Some(cmd) = ctx.cfg.page_smoke_cmd.as_deref() else {
        // A service with no pages says so by configuring no command. Blocked
        // rather than passed: a gate that proves nothing must not read green.
        return Ok(GateOutcome::blocked(GateBlocker::NotConfigured {
            what: "page_smoke_cmd".into(),
        }));
    };
    let Some(base) = ctx.public_url.as_deref() else {
        return Ok(GateOutcome::blocked(GateBlocker::NotConfigured {
            what: "public_url".into(),
        }));
    };

    let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
    let child = tokio::process::Command::new("sh")
        .arg("-c")
        .arg(cmd)
        .env("BASE", base)
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .kill_on_drop(true)
        .spawn()?;

    let out = match tokio::time::timeout(ceiling, child.wait_with_output()).await {
        Ok(res) => res?,
        Err(_elapsed) => {
            return Ok(GateOutcome::failed(GateFailure::Timeout {
                gate: GateKind::PageSmoke,
                after_s: ctx.cfg.gate_timeout_secs as u32,
            })
            .with_log_ref(ctx.log_ref(GateKind::PageSmoke)));
        }
    };

    let log = format!(
        "{}{}",
        String::from_utf8_lossy(&out.stdout),
        String::from_utf8_lossy(&out.stderr)
    );
    append_to_log(&ctx.log_path(GateKind::PageSmoke), log.as_bytes()).await;

    if out.status.success() {
        return Ok(
            GateOutcome::passed(PassNote::PagesClean { base: base.into() })
                .with_log_ref(ctx.log_ref(GateKind::PageSmoke)),
        );
    }

    // The script prints one `FAIL <url>` line per bad page and indents the
    // reasons under it. Lift the first reason into the summary so a red gate
    // says what broke without anyone opening the log.
    let first = log
        .lines()
        .skip_while(|l| !l.starts_with("FAIL"))
        .nth(1)
        .map(str::trim)
        .filter(|l| !l.is_empty())
        .unwrap_or("see log");
    Ok(GateOutcome::failed(GateFailure::PagesBroken {
        base: base.into(),
        detail: first.to_string(),
    })
    .with_log_ref(ctx.log_ref(GateKind::PageSmoke)))
}

pub(super) async fn node_health(ctx: &GateCtx) -> Result<GateOutcome> {
    if ctx.nodes.is_empty() {
        return Ok(GateOutcome::blocked(GateBlocker::NoNodesToProbe));
    }
    for probe in &ctx.nodes {
        if let Err(detail) = probe_node(probe).await {
            return Ok(GateOutcome::failed(GateFailure::NodeUnhealthy {
                node: probe.node.to_string(),
                detail,
            }));
        }
    }
    Ok(GateOutcome::passed(PassNote::NodesHealthy {
        nodes: ctx.nodes.len() as u32,
    }))
}

/// Probe one node over its executor: confirm the unit is active post-restart
/// and, when a `health_url` is configured, that it serves a 2xx. Retries across
/// ~10s because the service may still be restarting / warming. Runs under the
/// read-only `Observe(Health)` capability (every Sando node grants it), so the
/// probe needs no deploy authority. `Ok(())` = healthy; `Err(detail)` carries a
/// short reason for the gate's failure note.
async fn probe_node(probe: &NodeProbe) -> std::result::Result<(), String> {
    use ops_exec::{Action, ObserveKind, Step};
    let svc = sh_quote(&probe.service);
    let url = probe
        .health_url
        .as_deref()
        .map_or_else(|| "''".to_string(), sh_quote);
    // One executor round-trip with the retry loop on the node: is-active, then
    // (if a url is set) curl it for a 2xx. Exit 0 only when both hold.
    let script = format!(
        "svc={svc}; url={url}; \
         for _ in $(seq 1 10); do \
           if systemctl is-active --quiet \"$svc\"; then \
             if [ -z \"$url\" ] || curl -fsS --max-time 5 \"$url\" >/dev/null 2>&1; then exit 0; fi; \
           fi; \
           sleep 1; \
         done; \
         echo 'service not active or health url not 2xx after retries' >&2; exit 1"
    );
    let step = Step::shell(Action::Observe(ObserveKind::Health), script);
    let mut sink = DiscardSink;
    let out = probe
        .executor
        .run_streaming(&step, &mut sink)
        .await
        .map_err(|e| format!("probe spawn: {e}"))?;
    if out.status.success() {
        Ok(())
    } else {
        let code = out
            .status
            .code()
            .map_or_else(|| "signal".to_string(), |c| c.to_string());
        let stderr: String = String::from_utf8_lossy(&out.stderr)
            .chars()
            .take(200)
            .collect();
        Err(format!("exit {code}: {stderr}"))
    }
}

#[cfg(test)]
mod tests {
    use super::super::run;
    use super::*;
    use crate::domain::TierId;
    use crate::events;
    use crate::topology::Gate;
    use sqlx::sqlite::SqlitePoolOptions;
    use std::collections::HashMap;

    /// Spawn a one-shot loopback server that answers the first connection with
    /// `status_line` + a tiny body, then closes. Returns the bound port.
    async fn oneshot_http(status_line: &'static str) -> u16 {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
            .await
            .unwrap();
        let port = listener.local_addr().unwrap().port();
        tokio::spawn(async move {
            if let Ok((mut sock, _)) = listener.accept().await {
                let mut scratch = [0u8; 1024];
                let _ = sock.read(&mut scratch).await; // drain the request line
                let resp =
                    format!("{status_line}\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok");
                let _ = sock.write_all(resp.as_bytes()).await;
            }
        });
        port
    }

    #[tokio::test]
    async fn probe_health_ok_on_200() {
        let port = oneshot_http("HTTP/1.1 200 OK").await;
        assert!(probe_health(port).await.is_ok());
    }

    #[tokio::test]
    async fn probe_health_err_on_non_200() {
        let port = oneshot_http("HTTP/1.1 503 Service Unavailable").await;
        let err = probe_health(port).await.unwrap_err();
        assert!(err.contains("status line"), "{err}");
    }

    #[tokio::test]
    async fn probe_health_err_on_connection_refused() {
        // Bind then drop to get an almost-certainly-free port nothing listens on.
        let port = {
            let l = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0))
                .await
                .unwrap();
            l.local_addr().unwrap().port()
        };
        let err = probe_health(port).await.unwrap_err();
        // Under load the kernel does not always refuse the connect: a socket
        // left in TIME_WAIT on that port completes the handshake and then
        // resets, so the failure surfaces on the write or the read instead.
        // Refused-on-connect and reset-on-write/read are the same fact, that
        // nothing is serving the port, and no other outcome counts as a pass.
        let refused = err.starts_with("connect: ") && err.contains("refused");
        let reset =
            (err.starts_with("write: ") || err.starts_with("read: ")) && err.contains("reset");
        assert!(refused || reset, "{err}");
    }

    /// node_health fails closed when there are no nodes to probe: a serving tier
    /// should always carry nodes, so an empty set is a misconfiguration that must
    /// block promotion, not pass it.
    #[tokio::test]
    async fn node_health_blocks_with_no_nodes() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        crate::db::migrate(&pool).await.unwrap();
        sqlx::query(
            "INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('b', 2, 1, 'sequential')",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query("INSERT INTO tier_state (tier) VALUES ('b')")
            .execute(&pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')")
            .execute(&pool).await.unwrap();

        let cfg = std::sync::Arc::new(crate::config::AppConfig::for_tests());
        let ctx = GateCtx {
            public_url: None,
            pool: pool.clone(),
            cfg,
            tier: TierId::new("b"),
            version: "0.1.0".parse().unwrap(),
            worktree: None,
            bundle: None,
            events: events::channel(),
            nodes: Vec::new(), // no nodes -> fail closed
            build_id: None,
            aux_dirs: HashMap::new(),
        };
        let out = run(&ctx, &Gate::NodeHealth).await.unwrap();
        assert_eq!(out.status_str(), "blocked");
        assert!(!out.is_passed());
        let row: (Option<String>, Option<String>) =
            sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1")
                .fetch_one(&pool)
                .await
                .unwrap();
        let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap();
        assert_eq!(json["status"]["blocker"]["kind"], "no_nodes_to_probe");
    }
}