xbp_deploy/

verifier.rs

use std::sync::Arc;

use async_trait::async_trait;
use futures::future::join_all;
use serde::{Deserialize, Serialize};
use xbp_k8s::{K8sAdapter, KubeTarget};

use crate::context::DeployContext;
use crate::error::{DeployError, Result};
use crate::lock::load_lock;
use crate::providers::is_kubernetes_provider;
use crate::types::DeployPlan;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VerifyResult {
    pub ok: bool,
    pub lines: Vec<String>,
    pub drifts: Vec<String>,
}

#[async_trait]
pub trait DeployVerifier: Send + Sync {
    async fn verify(&self, ctx: &DeployContext, plan: &DeployPlan) -> Result<VerifyResult>;
}

pub struct DefaultDeployVerifier {
    pub k8s: Arc<dyn K8sAdapter>,
}

#[async_trait]
impl DeployVerifier for DefaultDeployVerifier {
    async fn verify(&self, ctx: &DeployContext, plan: &DeployPlan) -> Result<VerifyResult> {
        let mut lines: Vec<String> = Vec::new();
        let mut drifts: Vec<String> = Vec::new();

        // Only invoke kubectl verify when the plan actually includes k8s services.
        let has_k8s = plan
            .services
            .iter()
            .any(|s| is_kubernetes_provider(&s.provider))
            || !plan.k8s_plan.services.is_empty();

        if has_k8s {
            let target: KubeTarget = KubeTarget {
                context: plan.k8s_plan.context.clone(),
                namespace: plan.k8s_plan.default_namespace.clone(),
                kubeconfig: ctx
                    .config
                    .kubernetes
                    .as_ref()
                    .and_then(|k| k.kubeconfig.clone()),
            };

            match self.k8s.verify(&target, &plan.to_k8s_plan()).await {
                Ok(vlines) => {
                    for line in vlines {
                        // kubectl verify historically logged rollout/pod failures as soft
                        // notes only — that made `Result: ok` while the namespace/workload
                        // was missing. Treat error-shaped lines as drifts.
                        if k8s_verify_line_is_failure(&line) {
                            drifts.push(line.clone());
                        }
                        lines.push(line);
                    }
                }
                Err(e) => {
                    drifts.push(format!("k8s verify: {e}"));
                    lines.push(format!("k8s verify error: {e}"));
                }
            }
        } else {
            lines.push("kubernetes verify skipped (no k8s services)".into());
        }

        // Digest lock comparison
        if let Some(lock) = load_lock(&ctx.config.lock_file)? {
            if lock.hash != plan.hash && !plan.hash.is_empty() && !lock.hash.is_empty() {
                drifts.push(format!(
                    "plan hash drift: lock={} plan={}",
                    lock.hash, plan.hash
                ));
            }
            for svc in &plan.services {
                if let Some(expected) = lock.images.get(&svc.name) {
                    match &svc.digest {
                        Some(d) if d != &expected.digest => {
                            drifts.push(format!(
                                "[{}] digest mismatch lock={} plan={}",
                                svc.name, expected.digest, d
                            ));
                        }
                        None => {
                            drifts.push(format!(
                                "[{}] missing plan digest (lock has {})",
                                svc.name, expected.digest
                            ));
                        }
                        _ => lines.push(format!("[{}] digest matches lock", svc.name)),
                    }
                }
            }
            for (name, img) in &lock.images {
                if !plan.services.iter().any(|s| s.name == *name) {
                    drifts.push(format!("[{name}] present in lock but not in plan"));
                }
                let _ = img;
            }
        } else {
            lines.push("no deploy-lock.json (skip lock compare)".into());
        }

        // Parallel health
        if !ctx.flags.skip_health {
            let futs = plan.services.iter().flat_map(|svc| {
                svc.deploy.health.iter().map(move |url| {
                    let name = svc.name.clone();
                    let url = url.clone();
                    async move {
                        let status = probe(&url).await;
                        (name, url, status)
                    }
                })
            });
            for (name, url, status) in join_all(futs).await {
                match status {
                    Ok(code) if (200..400).contains(&code) => {
                        lines.push(format!("[{name}] health {url} → {code}"));
                    }
                    Ok(code) => drifts.push(format!("[{name}] health {url} → {code}")),
                    Err(e) => drifts.push(format!("[{name}] health {url}: {e}")),
                }
            }
        }

        let ok: bool = drifts.is_empty();
        if !ok {
            lines.push(format!("drift count: {}", drifts.len()));
        }
        Ok(VerifyResult { ok, lines, drifts })
    }
}

async fn probe(url: &str) -> std::result::Result<u16, String> {
    use tokio::process::Command;
    let output: std::process::Output = Command::new("curl")
        .args(["-sS", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "15", url])
        .output()
        .await
        .map_err(|e| e.to_string())?;
    if !output.status.success() {
        return Err(String::from_utf8_lossy(&output.stderr).trim().to_string());
    }
    String::from_utf8_lossy(&output.stdout)
        .trim()
        .parse()
        .map_err(|_| "invalid status".into())
}

/// Whether a kubectl verify log line indicates a real cluster failure.
///
/// Matches lines like:
/// - `[athena] rollout: kubectl failed: Error from server (NotFound): namespaces "athena" not found`
/// - `[athena] rollout: Error from server (NotFound): deployments.apps "athena" not found`
/// - `[athena] pods: error: ...`
pub fn k8s_verify_line_is_failure(line: &str) -> bool {
    let trimmed = line.trim();
    let body = trimmed
        .split_once(']')
        .map(|(_, rest)| rest.trim())
        .unwrap_or(trimmed);
    let lower = body.to_ascii_lowercase();

    // Successful rollout status messages are not failures.
    if lower.contains("successfully rolled out")
        || lower.contains("rollout complete")
        || lower.contains("deployment \"") && lower.contains("successfully")
    {
        return false;
    }

    // Deployment / namespace missing — always a hard failure (even outside rollout: prefix).
    if lower.contains("not found")
        && (lower.contains("deployment")
            || lower.contains("deployments.apps")
            || lower.contains("namespace")
            || lower.contains("namespaces \""))
    {
        return true;
    }

    if body.starts_with("rollout:")
        || body.starts_with("pods:")
        || lower.contains("rollout:")
        || lower.contains("kubectl failed")
    {
        if lower.contains("error")
            || lower.contains("failed")
            || lower.contains("not found")
            || lower.contains("timed out")
            || lower.contains("timeout")
            || lower.contains("forbidden")
            || lower.contains("unauthorized")
            || lower.contains("connection refused")
            || lower.contains("unable to connect")
        {
            return true;
        }
        // Empty pods list after a label query — soft signal only when selector was used
        // and kubectl returned nothing useful.
        if body.starts_with("pods:") {
            let detail = body.trim_start_matches("pods:").trim();
            if detail.is_empty() {
                return false;
            }
        }
    }
    false
}

#[cfg(test)]
mod tests {
    use super::k8s_verify_line_is_failure;

    #[test]
    fn verify_treats_missing_namespace_rollout_as_failure() {
        assert!(k8s_verify_line_is_failure(
            r#"[athena] rollout: kubectl failed: Error from server (NotFound): namespaces "athena" not found"#
        ));
        assert!(k8s_verify_line_is_failure(
            "[athena] rollout: Error from server (NotFound): deployments.apps \"athena\" not found"
        ));
        assert!(!k8s_verify_line_is_failure(
            "[athena] deployment \"athena\" successfully rolled out"
        ));
        assert!(!k8s_verify_line_is_failure("[athena] pods: pod/athena-abc"));
        assert!(!k8s_verify_line_is_failure("[athena] pods:"));
        assert!(!k8s_verify_line_is_failure(
            "[athena] health https://example.com/ → 200"
        ));
    }
}

/// Ensure verify surfaces as DeployError when ok=false if desired by caller.
pub fn verify_to_result(v: VerifyResult) -> Result<VerifyResult> {
    if v.ok {
        Ok(v)
    } else {
        Err(DeployError::Verify(v.drifts.join("; ")))
    }
}