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();
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 {
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());
}
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());
}
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())
}
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();
if lower.contains("successfully rolled out")
|| lower.contains("rollout complete")
|| lower.contains("deployment \"") && lower.contains("successfully")
{
return false;
}
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;
}
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"
));
}
}
pub fn verify_to_result(v: VerifyResult) -> Result<VerifyResult> {
if v.ok {
Ok(v)
} else {
Err(DeployError::Verify(v.drifts.join("; ")))
}
}