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>>,
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
));
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(", ")))
},
};
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> {
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())
}