xbp_deploy/

runner.rs

use std::sync::Arc;

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

use crate::context::{DeployContext, DeployMode};
use crate::error::{DeployError, Result};
use crate::history::{record_from_plan, DeployHistoryStore};
use crate::lock::{lock_from_plan, write_lock};
use crate::types::DeployPlan;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeployResult {
    pub ok: bool,
    pub lines: Vec<String>,
    pub failed_services: Vec<String>,
    pub error: Option<String>,
}

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

pub struct DefaultDeployRunner {
    pub k8s: Arc<dyn K8sAdapter>,
    pub promoter: Option<Arc<dyn OciPromoter>>,
    /// Optional confirmation callback: returns true to proceed.
    pub confirm: Option<Arc<dyn Fn(String) -> bool + Send + Sync>>,
}

#[async_trait]
impl DeployRunner for DefaultDeployRunner {
    async fn run(&self, ctx: &DeployContext, plan: &DeployPlan) -> Result<DeployResult> {
        let mut lines = Vec::new();
        let mut failed = Vec::new();

        lines.push(format!(
            "run target={} env={} dry_run={} services={}",
            plan.target.label(),
            plan.env,
            ctx.flags.dry_run,
            plan.order.len()
        ));
        lines.push(format!(
            "context={:?} namespace={:?}",
            plan.k8s_plan.context, plan.k8s_plan.default_namespace
        ));

        // Promote mode: OCI only.
        if matches!(ctx.mode, DeployMode::Promote) {
            return self.promote(ctx, plan).await;
        }

        if ctx.require_confirmation() {
            let msg = format!(
                "Apply deploy `{}` to env `{}` (context={:?})?",
                plan.target.label(),
                plan.env,
                plan.k8s_plan.context
            );
            let ok = self
                .confirm
                .as_ref()
                .map(|f| f(msg))
                .unwrap_or(false);
            if !ok {
                return Err(DeployError::Cancelled);
            }
            lines.push("context confirmed".into());
        }

        let target = 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()),
        };

        if !ctx.flags.skip_apply {
            let opts = K8sApplyOptions {
                dry_run: ctx.flags.dry_run,
                server_side: false,
                prune: false,
            };
            match self.k8s.apply(&target, &plan.to_k8s_plan(), &opts).await {
                Ok(apply_lines) => lines.extend(apply_lines),
                Err(e) => {
                    failed.push("kubernetes".into());
                    lines.push(format!("ERROR: {e}"));
                }
            }
        } else {
            lines.push("skip-apply".into());
        }

        if !ctx.flags.skip_rollout && !ctx.flags.dry_run && failed.is_empty() {
            for svc in &plan.services {
                if let Some(workload) = &svc.deploy.workload {
                    let mut t = target.clone();
                    if t.namespace.is_none() {
                        t.namespace = svc.deploy.namespace.clone();
                    }
                    match self.k8s.rollout(&t, workload, "180s").await {
                        Ok(out) => lines.push(format!(
                            "[{}] {}",
                            svc.name,
                            out.lines().next().unwrap_or("ok")
                        )),
                        Err(e) => {
                            failed.push(svc.name.clone());
                            lines.push(format!("[{}] rollout failed: {e}", svc.name));
                        }
                    }
                }
            }
        }

        if !ctx.flags.skip_health && !ctx.flags.dry_run && failed.is_empty() {
            let health_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 { (name, url.clone(), http_status(&url).await) }
                })
            });
            for (name, url, result) in join_all(health_futs).await {
                match result {
                    Ok(code) if (200..400).contains(&code) => {
                        lines.push(format!("[{name}] health {url} → {code}"));
                    }
                    Ok(code) => {
                        failed.push(name.clone());
                        lines.push(format!("[{name}] health {url} → {code}"));
                    }
                    Err(e) => {
                        failed.push(name.clone());
                        lines.push(format!("[{name}] health {url} failed: {e}"));
                    }
                }
            }
        }

        let ok = failed.is_empty();
        let result = DeployResult {
            ok,
            lines: lines.clone(),
            failed_services: failed.clone(),
            error: if ok {
                None
            } else {
                Some(format!("failed: {}", failed.join(", ")))
            },
        };

        // History + lock
        if matches!(ctx.mode, DeployMode::Run) && !ctx.flags.dry_run {
            let store = DeployHistoryStore::new(&ctx.config.history_dir);
            let record = record_from_plan(
                plan,
                ok,
                if ok {
                    "deploy succeeded".into()
                } else {
                    result.error.clone().unwrap_or_default()
                },
                result.error.clone(),
                plan.k8s_plan.context.clone(),
            );
            let _ = store.write_record(&record);
            if ok {
                let lock = lock_from_plan(plan);
                let _ = write_lock(&ctx.config.lock_file, &lock);
            }
        }

        Ok(result)
    }
}

impl DefaultDeployRunner {
    async fn promote(&self, ctx: &DeployContext, plan: &DeployPlan) -> Result<DeployResult> {
        let tag = ctx
            .flags
            .promote_tag
            .as_deref()
            .ok_or_else(|| DeployError::Validation("promote requires a target tag".into()))?;
        let Some(promoter) = &self.promoter else {
            return Err(DeployError::Oci("no OCI promoter configured".into()));
        };
        let mut lines = Vec::new();
        let mut failed = Vec::new();
        for svc in &plan.services {
            let Some(image) = &svc.image else {
                continue;
            };
            let target = image.with_tag(tag);
            if ctx.flags.dry_run {
                lines.push(format!(
                    "[{}] dry-run promote {} -> {}",
                    svc.name,
                    image.reference(),
                    target.reference()
                ));
                continue;
            }
            match promoter.promote(image, &target, true).await {
                Ok(()) => lines.push(format!(
                    "[{}] promoted {} -> {}",
                    svc.name,
                    image.reference(),
                    target.reference()
                )),
                Err(e) => {
                    failed.push(svc.name.clone());
                    lines.push(format!("[{}] promote failed: {e}", svc.name));
                }
            }
        }
        Ok(DeployResult {
            ok: failed.is_empty(),
            lines,
            failed_services: failed.clone(),
            error: if failed.is_empty() {
                None
            } else {
                Some(format!("promote failed: {}", failed.join(", ")))
            },
        })
    }
}

async fn http_status(url: &str) -> std::result::Result<u16, String> {
    // Minimal dependency: use std + tokio process curl fallback, or reqwest if available.
    // Keep deploy crate free of reqwest — use TCP-less approach via `std::process` curl optional.
    use tokio::process::Command;
    let 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())
}