use std::path::Path;
use std::process::Stdio;
use colored::Colorize;
use tokio::process::Command;
use crate::strategies::XbpConfig;
use crate::utils::command_exists;
use super::types::{DeployMode, DeployStepPlan};
pub async fn execute_step(
project_root: &Path,
config: &XbpConfig,
step: &DeployStepPlan,
mode: DeployMode,
yes: bool,
debug: bool,
) -> Result<Vec<String>, String> {
match step.provider.as_str() {
"kubernetes" => execute_kubernetes(project_root, config, step, mode, yes, debug).await,
"kubernetes-operator" => {
execute_kubernetes_operator(project_root, config, step, mode, yes, debug).await
}
"worker" => execute_worker(project_root, step, mode, debug).await,
"pm2" | "systemd" => execute_local_runtime(step, mode),
other => Ok(vec![format!(
"[{}] provider `{other}` has no executor yet (plan recorded)",
step.service
)]),
}
}
async fn execute_kubernetes(
project_root: &Path,
config: &XbpConfig,
step: &DeployStepPlan,
mode: DeployMode,
yes: bool,
debug: bool,
) -> Result<Vec<String>, String> {
let mut lines = Vec::new();
lines.push(format!(
"[{}] provider=kubernetes mode={:?}",
step.service, mode
));
if let Some(image) = &step.image_ref {
lines.push(format!("[{}] image {}", step.service, image));
if matches!(mode, DeployMode::Verify | DeployMode::Run | DeployMode::Status) {
if let Some(msg) = try_verify_image(image).await {
lines.push(format!("[{}] {msg}", step.service));
}
}
}
if matches!(mode, DeployMode::Status | DeployMode::Verify) {
if let Some(workload) = &step.workload {
let ns = step.namespace.as_deref();
match kubectl(
&["rollout", "status", workload, "--timeout=30s"],
config,
ns,
debug,
)
.await
{
Ok(out) => lines.push(format!("[{}] rollout: {}", step.service, out.lines().next().unwrap_or("ok"))),
Err(e) => {
if matches!(mode, DeployMode::Verify) {
return Err(format!("[{}] rollout status failed: {e}", step.service));
}
lines.push(format!("[{}] rollout status: {e}", step.service));
}
}
}
probe_health(step, &mut lines).await?;
return Ok(lines);
}
maybe_confirm_context(config, yes, &mut lines)?;
for path in &step.manifest_paths {
let p = Path::new(path);
if !p.exists() {
lines.push(format!(
"[{}] skip missing manifest path {}",
step.service, path
));
continue;
}
let args = ["apply", "-f", path.as_str()];
let ns = step.namespace.as_deref();
let out = kubectl(&args, config, ns, debug).await?;
lines.push(format!(
"[{}] apply {}: {}",
step.service,
path,
summarize_kubectl(&out)
));
let _ = project_root; }
if let Some(workload) = &step.workload {
let ns = step.namespace.as_deref();
match kubectl(
&["rollout", "status", workload, "--timeout=180s"],
config,
ns,
debug,
)
.await
{
Ok(out) => lines.push(format!("[{}] {}", step.service, summarize_kubectl(&out))),
Err(e) => return Err(format!("[{}] rollout failed: {e}", step.service)),
}
}
probe_health(step, &mut lines).await?;
Ok(lines)
}
async fn execute_kubernetes_operator(
project_root: &Path,
config: &XbpConfig,
step: &DeployStepPlan,
mode: DeployMode,
yes: bool,
debug: bool,
) -> Result<Vec<String>, String> {
let mut lines = Vec::new();
lines.push(format!(
"[{}] provider=kubernetes-operator mode={:?}",
step.service, mode
));
if matches!(mode, DeployMode::Status | DeployMode::Verify) {
if let Some(selector) = &step.selector {
let ns = step.namespace.as_deref();
let out = kubectl(
&["get", "pods", "-l", selector, "-o", "wide"],
config,
ns,
debug,
)
.await
.unwrap_or_else(|e| e);
lines.push(format!("[{}] pods: {}", step.service, summarize_kubectl(&out)));
}
return Ok(lines);
}
maybe_confirm_context(config, yes, &mut lines)?;
for path in [&step.crds_path, &step.install_path].into_iter().flatten() {
if !Path::new(path).exists() {
lines.push(format!("[{}] missing path {}", step.service, path));
continue;
}
let ns = step.namespace.as_deref();
let out = kubectl(&["apply", "-f", path], config, ns, debug).await?;
lines.push(format!(
"[{}] apply {}: {}",
step.service,
path,
summarize_kubectl(&out)
));
}
if let Some(selector) = &step.selector {
let ns = step.namespace.as_deref();
let _ = kubectl(
&[
"wait",
"--for=condition=Ready",
"pod",
"-l",
selector,
"--timeout=180s",
],
config,
ns,
debug,
)
.await;
lines.push(format!("[{}] waited for selector {selector}", step.service));
}
let _ = project_root;
Ok(lines)
}
async fn execute_worker(
project_root: &Path,
step: &DeployStepPlan,
mode: DeployMode,
debug: bool,
) -> Result<Vec<String>, String> {
let mut lines = vec![format!(
"[{}] provider=worker mode={:?}",
step.service, mode
)];
if matches!(mode, DeployMode::Status | DeployMode::Verify) {
lines.push(format!(
"[{}] worker verify: ensure wrangler whoami / deployments manually or via `xbp workers`",
step.service
));
return Ok(lines);
}
let cwd = project_root.to_path_buf();
if command_exists("npx") {
let mut cmd = Command::new("npx");
cmd.arg("wrangler")
.arg("deploy")
.current_dir(&cwd)
.stdout(Stdio::piped())
.stderr(Stdio::piped());
if debug {
lines.push(format!("[{}] running: npx wrangler deploy", step.service));
}
match cmd.output().await {
Ok(out) if out.status.success() => {
lines.push(format!("[{}] wrangler deploy ok", step.service));
}
Ok(out) => {
let err = String::from_utf8_lossy(&out.stderr);
lines.push(format!(
"[{}] wrangler deploy skipped/failed: {}",
step.service,
err.lines().next().unwrap_or("non-zero exit")
));
}
Err(e) => lines.push(format!("[{}] wrangler not run: {e}", step.service)),
}
} else {
lines.push(format!(
"[{}] npx not available; configure worker deploy via `xbp workers deploy`",
step.service
));
}
Ok(lines)
}
fn execute_local_runtime(step: &DeployStepPlan, mode: DeployMode) -> Result<Vec<String>, String> {
Ok(vec![format!(
"[{}] provider={} mode={:?} — use `xbp service start {}` / redeploy for local runtime",
step.service, step.provider, mode, step.service
)])
}
async fn probe_health(step: &DeployStepPlan, lines: &mut Vec<String>) -> Result<(), String> {
for url in &step.health {
match reqwest_get_status(url).await {
Ok(code) if (200..400).contains(&code) => {
lines.push(format!("[{}] health {} → {code}", step.service, url));
}
Ok(code) => {
return Err(format!(
"[{}] health {} returned HTTP {code}",
step.service, url
));
}
Err(e) => {
return Err(format!("[{}] health {} failed: {e}", step.service, url));
}
}
}
Ok(())
}
async fn reqwest_get_status(url: &str) -> Result<u16, String> {
if command_exists("curl") {
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());
}
let code = String::from_utf8_lossy(&output.stdout).trim().to_string();
return code
.parse::<u16>()
.map_err(|_| format!("invalid status code `{code}`"));
}
Err("curl not available for health probes".to_string())
}
async fn try_verify_image(image: &str) -> Option<String> {
if command_exists("crane") {
let output = Command::new("crane")
.args(["digest", image])
.output()
.await
.ok()?;
if output.status.success() {
let digest = String::from_utf8_lossy(&output.stdout).trim().to_string();
return Some(format!("oci digest {digest}"));
}
}
if command_exists("docker") {
let output = Command::new("docker")
.args(["manifest", "inspect", image])
.output()
.await
.ok()?;
if output.status.success() {
return Some("oci manifest present (docker)".to_string());
}
}
Some("oci verify skipped (crane/docker not available)".to_string())
}
fn maybe_confirm_context(
config: &XbpConfig,
yes: bool,
lines: &mut Vec<String>,
) -> Result<(), String> {
let require = config
.kubernetes
.as_ref()
.and_then(|k| k.require_context_confirmation_for_apply)
.unwrap_or(true);
if !require || yes {
return Ok(());
}
if !std::io::IsTerminal::is_terminal(&std::io::stdin()) {
return Err(
"kubernetes.require_context_confirmation_for_apply is true; pass --yes for non-interactive apply"
.to_string(),
);
}
let ctx = config
.kubernetes
.as_ref()
.and_then(|k| k.default_context.clone())
.filter(|s| !s.is_empty())
.unwrap_or_else(|| "(current-context)".to_string());
let ok = dialoguer::Confirm::new()
.with_prompt(format!(
"Apply Kubernetes changes using context `{ctx}`?"
))
.default(false)
.interact()
.map_err(|e| e.to_string())?;
if !ok {
return Err("deploy cancelled (context not confirmed)".to_string());
}
lines.push(format!("context confirmed: {ctx}"));
Ok(())
}
async fn kubectl(
args: &[&str],
config: &XbpConfig,
namespace: Option<&str>,
debug: bool,
) -> Result<String, String> {
if !command_exists("kubectl") {
return Err("kubectl not found in PATH".to_string());
}
let mut cmd = Command::new("kubectl");
if let Some(ctx) = config
.kubernetes
.as_ref()
.and_then(|k| k.default_context.as_deref())
.map(str::trim)
.filter(|s| !s.is_empty())
{
cmd.args(["--context", ctx]);
}
if let Some(ns) = namespace.map(str::trim).filter(|s| !s.is_empty()) {
cmd.args(["-n", ns]);
}
if let Some(kubeconfig) = config
.kubernetes
.as_ref()
.and_then(|k| k.kubeconfig.as_deref())
.map(str::trim)
.filter(|s| !s.is_empty())
{
cmd.env("KUBECONFIG", kubeconfig);
}
cmd.args(args);
cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
if debug {
eprintln!(
"{} kubectl {}",
"debug".dimmed(),
args.join(" ")
);
}
let output = cmd.output().await.map_err(|e| e.to_string())?;
if !output.status.success() {
let err = String::from_utf8_lossy(&output.stderr);
let out = String::from_utf8_lossy(&output.stdout);
return Err(format!("{}{}", err.trim(), out.trim()));
}
Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
}
fn summarize_kubectl(out: &str) -> String {
out.lines()
.take(3)
.collect::<Vec<_>>()
.join(" | ")
.chars()
.take(200)
.collect()
}