use crate::config::global_xbp_paths;
use crate::sdk::command::CommandRunner;
use crate::sdk::{command_debug_log, command_failure_message};
use serde::Deserialize;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::time::timeout;
use tracing::debug;
const PM2_WINDOWS_COMMAND_TIMEOUT: Duration = Duration::from_secs(20);
const PM2_WINDOWS_INSTALL_CHECK_TIMEOUT: Duration = Duration::from_secs(8);
fn pm2_invocation(args: &[&str]) -> (&'static str, Vec<String>) {
if cfg!(target_os = "windows") {
let mut invocation_args = vec!["/C".to_string(), "pm2".to_string()];
invocation_args.extend(args.iter().map(|arg| (*arg).to_string()));
("cmd", invocation_args)
} else {
(
"pm2",
args.iter()
.map(|arg| (*arg).to_string())
.collect::<Vec<_>>(),
)
}
}
fn pm2_timeout_for(args: &[&str], inherit_io: bool) -> Option<Duration> {
if !cfg!(target_os = "windows") {
return None;
}
if inherit_io && matches!(args.first().copied(), Some("logs" | "monitor")) {
return None;
}
Some(PM2_WINDOWS_COMMAND_TIMEOUT)
}
async fn check_pm2_installed() -> Result<(), String> {
let (program, invocation_args) = pm2_invocation(&["--version"]);
let invocation_refs: Vec<&str> = invocation_args.iter().map(|arg| arg.as_str()).collect();
let runner = CommandRunner::new(false);
let check_result = if cfg!(target_os = "windows") {
match timeout(
PM2_WINDOWS_INSTALL_CHECK_TIMEOUT,
runner.run(program, &invocation_refs),
)
.await
{
Ok(result) => result,
Err(_) => {
return Err(format!(
"PM2 version check timed out after {}s.\nhelp: run `pm2 ping` and restart the PM2 daemon before retrying.",
PM2_WINDOWS_INSTALL_CHECK_TIMEOUT.as_secs()
));
}
}
} else {
runner.run(program, &invocation_refs).await
};
match check_result {
Ok(outcome) if outcome.output.status.success() => Ok(()),
Ok(outcome) => Err(format!(
"PM2 is not available.\nstdout: {}\nstderr: {}\nhelp: install PM2 with `npm install -g pm2` and retry.",
outcome.stdout,
outcome.stderr
)),
Err(err) => Err(format!(
"Failed to check PM2 installation: {}\nhelp: install PM2 with `npm install -g pm2` and retry.",
err
)),
}
}
pub async fn available() -> Result<(), String> {
check_pm2_installed().await
}
pub async fn list(debug: bool) -> Result<(), String> {
check_pm2_installed().await?;
run_pm2_command(&["list"], debug, true).await
}
pub async fn logs(project: Option<String>, debug: bool) -> Result<(), String> {
let mut args = vec!["logs".to_string()];
if let Some(name) = project {
args.push(name);
}
let arg_refs = args.iter().map(|arg| arg.as_str()).collect::<Vec<_>>();
run_pm2_command(&arg_refs, debug, true).await
}
pub async fn stop(name: &str, debug: bool) -> Result<(), String> {
run_pm2_command(&["stop", name], debug, false).await?;
Ok(())
}
pub async fn delete(name: &str, debug: bool) -> Result<(), String> {
run_pm2_command(&["delete", name], debug, false).await?;
Ok(())
}
pub async fn start(
name: &str,
command: &str,
log_dir: Option<&PathBuf>,
envs: Option<&HashMap<String, String>>,
debug: bool,
) -> Result<(), String> {
start_in_dir(name, command, None, log_dir, envs, debug).await
}
pub async fn start_in_dir(
name: &str,
command: &str,
working_dir: Option<&Path>,
log_dir: Option<&PathBuf>,
envs: Option<&HashMap<String, String>>,
debug: bool,
) -> Result<(), String> {
check_pm2_installed().await?;
let cwd = working_dir
.map(Path::to_path_buf)
.or_else(|| std::env::current_dir().ok());
let mut command_tokens = split_start_command(command);
if command_tokens.is_empty() {
return Err("Start command is empty.".to_string());
}
if let Some(resolved) = resolve_start_program_path(&command_tokens[0], cwd.as_deref()) {
command_tokens[0] = resolved;
}
validate_start_program(&command_tokens[0], cwd.as_deref())?;
let mut args: Vec<String> = vec!["start".into()];
if let Some(log_path) = log_dir {
let stdout_log = log_path.join(format!("{}-stdout.log", name));
let stderr_log = log_path.join(format!("{}-stderr.log", name));
fs::create_dir_all(log_path)
.map_err(|e| format!("Failed to create log directory {}: {}", log_path.display(), e))?;
args.push("--name".into());
args.push(name.to_string());
args.push("--log".into());
args.push(stdout_log.to_string_lossy().to_string());
args.push("--error".into());
args.push(stderr_log.to_string_lossy().to_string());
} else {
args.push("--name".into());
args.push(name.to_string());
}
args.push("--interpreter".into());
args.push("none".into());
args.push("--".into());
args.extend(command_tokens);
let arg_refs: Vec<&str> = args.iter().map(|s| s.as_str()).collect();
let (program, invocation_args) = pm2_invocation(&arg_refs);
let invocation_refs: Vec<&str> = invocation_args.iter().map(|arg| arg.as_str()).collect();
let runner = CommandRunner::new(debug);
let outcome = if let Some(timeout_window) = pm2_timeout_for(&arg_refs, false) {
match timeout(
timeout_window,
runner.run_builder(program, &invocation_refs, |cmd| {
if let Some(dir) = cwd.as_ref() {
cmd.current_dir(dir);
}
if let Some(envs) = envs {
cmd.envs(envs);
}
}),
)
.await
{
Ok(result) => result.map_err(|e| e.to_string())?,
Err(_) => {
return Err(format!(
"pm2 start timed out after {}s.\nhelp: run `pm2 ping` and ensure the PM2 daemon is responsive.",
timeout_window.as_secs()
));
}
}
} else {
runner
.run_builder(program, &invocation_refs, |cmd| {
if let Some(dir) = cwd.as_ref() {
cmd.current_dir(dir);
}
if let Some(envs) = envs {
cmd.envs(envs);
}
})
.await
.map_err(|e| e.to_string())?
};
if !outcome.output.status.success() {
return Err(command_failure_message(
"pm2",
&arg_refs,
&outcome.output,
Some(
"verify the start binary exists (Windows: target/release/<name>.exe), \
run `pm2 ping`, and try `xbp service build <name>` first.",
),
));
}
if !outcome.stdout.trim().is_empty() {
print!("{}", outcome.stdout);
}
Ok(())
}
fn split_start_command(command: &str) -> Vec<String> {
let mut tokens = Vec::new();
let mut current = String::new();
let mut chars = command.chars().peekable();
let mut in_single = false;
let mut in_double = false;
while let Some(ch) = chars.next() {
match ch {
'\'' if !in_double => {
in_single = !in_single;
}
'"' if !in_single => {
in_double = !in_double;
}
c if c.is_whitespace() && !in_single && !in_double => {
if !current.is_empty() {
tokens.push(std::mem::take(&mut current));
}
}
c => current.push(c),
}
}
if !current.is_empty() {
tokens.push(current);
}
tokens
}
fn resolve_start_program_path(program: &str, working_dir: Option<&Path>) -> Option<String> {
let path = PathBuf::from(program);
if path.is_absolute() {
return resolve_windows_exe_if_missing(&path).map(|p| p.to_string_lossy().to_string());
}
let base = working_dir?;
let candidate = base.join(program.trim_start_matches("./").trim_start_matches(".\\"));
if candidate.exists() {
return Some(path_for_pm2_arg(&candidate, base));
}
if let Some(with_exe) = resolve_windows_exe_if_missing(&candidate) {
return Some(path_for_pm2_arg(&with_exe, base));
}
None
}
fn path_for_pm2_arg(absolute: &Path, working_dir: &Path) -> String {
absolute
.strip_prefix(working_dir)
.map(|relative| {
let rendered = relative.to_string_lossy().replace('\\', "/");
if rendered.is_empty() {
absolute.to_string_lossy().replace('\\', "/")
} else {
format!("./{rendered}")
}
})
.unwrap_or_else(|_| absolute.to_string_lossy().replace('\\', "/"))
}
fn resolve_windows_exe_if_missing(path: &Path) -> Option<PathBuf> {
if path.exists() {
return Some(path.to_path_buf());
}
if cfg!(windows) {
let with_exe = path.with_extension("exe");
if with_exe != path && with_exe.exists() {
return Some(with_exe);
}
let mut raw = path.as_os_str().to_os_string();
raw.push(".exe");
let candidate = PathBuf::from(raw);
if candidate.exists() {
return Some(candidate);
}
}
None
}
fn validate_start_program(program: &str, working_dir: Option<&Path>) -> Result<(), String> {
let path = PathBuf::from(program);
let resolved = if path.is_absolute() {
path.clone()
} else if let Some(dir) = working_dir {
dir.join(program.trim_start_matches("./").trim_start_matches(".\\"))
} else {
path.clone()
};
let exists = resolved.exists()
|| resolve_windows_exe_if_missing(&resolved).is_some()
|| !program.contains('/') && !program.contains('\\') && !program.starts_with('.');
if exists {
return Ok(());
}
let mut hint = format!(
"Start program `{}` was not found{}.",
program,
working_dir
.map(|dir| format!(" under {}", dir.display()))
.unwrap_or_default()
);
if cfg!(windows) && !program.ends_with(".exe") {
hint.push_str(" On Windows, release binaries usually end with `.exe`.");
}
hint.push_str(" Run the service build first (`xbp service build`), then retry start.");
Err(hint)
}
pub async fn cleanup(debug: bool) -> Result<(), String> {
let output = run_pm2_capture(&["list", "--no-color"], debug).await?;
if !output.status.success() {
return Err(command_failure_message(
"pm2",
&["list", "--no-color"],
&output,
Some("run `xbp list` to inspect PM2 process state."),
));
}
let stdout = String::from_utf8_lossy(&output.stdout);
let mut processes_to_delete = Vec::new();
for line in stdout.lines() {
if line.contains("stopped") || line.contains("errored") {
let parts: Vec<&str> = line.split('│').collect();
if parts.len() >= 3 {
let name = parts[2].trim();
if !name.is_empty() && name != "name" {
processes_to_delete.push(name.to_string());
}
}
}
}
for process_name in processes_to_delete {
if debug {
debug!("Deleting stopped/errored process: {}", process_name);
}
delete(&process_name, debug).await?;
}
let save_output = run_pm2_capture(&["save"], debug).await?;
if !save_output.status.success() {
return Err(command_failure_message(
"pm2",
&["save"],
&save_output,
Some("run `xbp snapshot` to verify PM2 snapshot persistence."),
));
}
Ok(())
}
pub async fn save(debug: bool) -> Result<(), String> {
let output = run_pm2_capture(&["save"], debug).await?;
command_debug_log(debug, "pm2", &["save"], &output, |msg| debug!("{}", msg));
if !output.status.success() {
return Err(command_failure_message(
"pm2",
&["save"],
&output,
Some("run `xbp snapshot` to persist and verify PM2 state."),
));
}
Ok(())
}
pub async fn snapshot(debug: bool) -> Result<PathBuf, String> {
save(debug).await?;
let prettylist_output = run_pm2_capture(&["jlist"], debug).await?;
let processes: Value = serde_json::from_slice(&prettylist_output.stdout)
.map_err(|e| format!("Failed to parse pm2 jlist output: {}", e))?;
let snapshot_root = pm2_snapshot_root()?;
fs::create_dir_all(&snapshot_root)
.map_err(|e| format!("Failed to create snapshot directory: {}", e))?;
let timestamp = chrono::Local::now().format("%Y%m%d-%H%M%S").to_string();
let snapshot_dir = snapshot_root.join(×tamp);
fs::create_dir_all(&snapshot_dir)
.map_err(|e| format!("Failed to create snapshot directory: {}", e))?;
let snapshot_json_path = snapshot_dir.join("pm2-jlist.json");
fs::write(
&snapshot_json_path,
serde_json::to_string_pretty(&processes)
.map_err(|e| format!("Failed to serialize PM2 snapshot: {}", e))?,
)
.map_err(|e| format!("Failed to write PM2 snapshot: {}", e))?;
if let Ok(pretty_output) = run_pm2_capture(&["prettylist"], debug).await {
let prettylist_path = snapshot_dir.join("pm2-prettylist.txt");
let _ = fs::write(prettylist_path, pretty_output.stdout);
}
let dump_path = pm2_dump_path();
if dump_path.exists() {
let snapshot_dump = snapshot_dir.join("dump.pm2");
fs::copy(&dump_path, &snapshot_dump)
.map_err(|e| format!("Failed to copy PM2 dump file: {}", e))?;
}
let latest_dir = snapshot_root.join("latest");
if latest_dir.exists() {
let _ = fs::remove_dir_all(&latest_dir);
}
copy_dir_contents(&snapshot_dir, &latest_dir)?;
let metadata_path = snapshot_dir.join("snapshot.json");
fs::write(
&metadata_path,
serde_json::to_string_pretty(&json!({
"created_at": chrono::Local::now().to_rfc3339(),
"pm2_home": pm2_home_dir().display().to_string(),
"process_count": processes.as_array().map(|items| items.len()).unwrap_or(0),
"files": {
"jlist": snapshot_json_path.file_name().and_then(|n| n.to_str()).unwrap_or("pm2-jlist.json"),
"dump": if snapshot_dir.join("dump.pm2").exists() { Some("dump.pm2") } else { None }
}
}))
.map_err(|e| format!("Failed to serialize PM2 snapshot metadata: {}", e))?,
)
.map_err(|e| format!("Failed to write PM2 snapshot metadata: {}", e))?;
Ok(snapshot_dir)
}
pub async fn resurrect(debug: bool) -> Result<(), String> {
match run_pm2_command(&["resurrect"], debug, true).await {
Ok(()) => Ok(()),
Err(primary_err) => {
if let Err(restore_err) = restore_latest_snapshot_dump(debug).await {
return Err(format!(
"pm2 resurrect failed: {}. Snapshot restore also failed: {}",
primary_err, restore_err
));
}
run_pm2_command(&["resurrect"], debug, true)
.await
.map_err(|retry_err| {
format!(
"pm2 resurrect failed: {}. Restored latest snapshot dump but retry failed: {}",
primary_err, retry_err
)
})
}
}
}
pub async fn flush(target: Option<&str>, debug: bool) -> Result<(), String> {
let mut args = vec!["flush".to_string()];
if let Some(target) = target {
args.push(target.to_string());
}
let arg_refs = args.iter().map(|arg| arg.as_str()).collect::<Vec<_>>();
run_pm2_command(&arg_refs, debug, true).await
}
pub async fn monitor(debug: bool) -> Result<(), String> {
run_pm2_command(&["monitor"], debug, true).await
}
pub async fn env(target: &str, debug: bool) -> Result<(), String> {
let resolved_id = resolve_process_id(target, debug).await?;
run_pm2_command(&["env", &resolved_id], debug, true).await
}
async fn resolve_process_id(target: &str, debug: bool) -> Result<String, String> {
if target.chars().all(|c| c.is_ascii_digit()) {
return Ok(target.to_string());
}
let output = run_pm2_capture(&["jlist"], debug).await?;
let processes: Vec<Pm2Process> = serde_json::from_slice(&output.stdout)
.map_err(|e| format!("Failed to parse pm2 jlist output: {}", e))?;
resolve_process_id_from_list(target, &processes)
}
async fn run_pm2_capture(args: &[&str], debug: bool) -> Result<std::process::Output, String> {
check_pm2_installed().await?;
let runner = CommandRunner::new(debug);
let (program, invocation_args) = pm2_invocation(args);
let invocation_refs: Vec<&str> = invocation_args.iter().map(|arg| arg.as_str()).collect();
let outcome = if let Some(timeout_window) = pm2_timeout_for(args, false) {
match timeout(timeout_window, runner.run(program, &invocation_refs)).await {
Ok(result) => result.map_err(|e| e.to_string())?,
Err(_) => {
return Err(format!(
"pm2 {} timed out after {}s.\nhelp: run `pm2 ping` and ensure the PM2 daemon is responsive.",
args.join(" "),
timeout_window.as_secs()
));
}
}
} else {
runner
.run(program, &invocation_refs)
.await
.map_err(|e| e.to_string())?
};
if !outcome.output.status.success() {
return Err(command_failure_message(
"pm2",
args,
&outcome.output,
Some("run `xbp --help` for available PM2 proxy commands."),
));
}
Ok(outcome.output)
}
async fn run_pm2_command(args: &[&str], debug: bool, inherit_io: bool) -> Result<(), String> {
check_pm2_installed().await?;
let runner = CommandRunner::new(debug);
let (program, invocation_args) = pm2_invocation(args);
let invocation_refs: Vec<&str> = invocation_args.iter().map(|arg| arg.as_str()).collect();
if inherit_io {
let status = if let Some(timeout_window) = pm2_timeout_for(args, true) {
match timeout(
timeout_window,
runner.run_with_stdio(program, &invocation_refs),
)
.await
{
Ok(result) => result.map_err(|e| e.to_string())?,
Err(_) => {
return Err(format!(
"pm2 {} timed out after {}s.\nhelp: run `pm2 ping` and ensure the PM2 daemon is responsive.",
args.join(" "),
timeout_window.as_secs()
));
}
}
} else {
runner
.run_with_stdio(program, &invocation_refs)
.await
.map_err(|e| e.to_string())?
};
if !status.success() {
return Err(format!(
"pm2 {} exited with status {}.\nhelp: run `xbp --help` and verify PM2 is healthy with `pm2 status`.",
args.join(" "),
status
));
}
return Ok(());
}
let output = run_pm2_capture(args, debug).await?;
let stdout = String::from_utf8_lossy(&output.stdout);
if !stdout.trim().is_empty() {
print!("{}", stdout);
}
Ok(())
}
#[derive(Debug, Deserialize)]
struct Pm2Process {
name: String,
pm_id: i64,
}
fn resolve_process_id_from_list(target: &str, processes: &[Pm2Process]) -> Result<String, String> {
let process = processes
.iter()
.find(|process| process.name == target)
.ok_or_else(|| {
format!(
"PM2 process '{}' not found.\nhelp: run `xbp list` to see names or `xbp env <pm2-id>` with a numeric id.",
target
)
})?;
Ok(process.pm_id.to_string())
}
fn pm2_snapshot_root() -> Result<PathBuf, String> {
let paths = global_xbp_paths()?;
Ok(paths.root_dir.join("snapshots").join("pm2"))
}
fn pm2_home_dir() -> PathBuf {
pm2_home_dir_from(
std::env::var_os("PM2_HOME").map(PathBuf::from),
dirs::home_dir(),
)
}
fn pm2_dump_path() -> PathBuf {
pm2_home_dir().join("dump.pm2")
}
fn pm2_home_dir_from(pm2_home: Option<PathBuf>, home_dir: Option<PathBuf>) -> PathBuf {
pm2_home
.or_else(|| home_dir.map(|home| home.join(".pm2")))
.unwrap_or_else(|| PathBuf::from(".pm2"))
}
async fn restore_latest_snapshot_dump(debug: bool) -> Result<(), String> {
let latest_dir = pm2_snapshot_root()?.join("latest");
let latest_dump = latest_dir.join("dump.pm2");
if !latest_dump.exists() {
return Err("No PM2 snapshot dump found at the latest snapshot location".to_string());
}
let target_dump = pm2_dump_path();
if let Some(parent) = target_dump.parent() {
fs::create_dir_all(parent)
.map_err(|e| format!("Failed to create PM2 home directory: {}", e))?;
}
fs::copy(&latest_dump, &target_dump)
.map_err(|e| format!("Failed to restore PM2 dump from snapshot: {}", e))?;
if debug {
debug!(
"restored PM2 dump from {} to {}",
latest_dump.display(),
target_dump.display()
);
}
Ok(())
}
fn copy_dir_contents(source: &Path, destination: &Path) -> Result<(), String> {
fs::create_dir_all(destination)
.map_err(|e| format!("Failed to create destination snapshot directory: {}", e))?;
for entry in fs::read_dir(source).map_err(|e| format!("Failed to read snapshot dir: {}", e))? {
let entry = entry.map_err(|e| format!("Failed to read snapshot entry: {}", e))?;
let source_path = entry.path();
let destination_path = destination.join(entry.file_name());
if source_path.is_dir() {
copy_dir_contents(&source_path, &destination_path)?;
} else {
fs::copy(&source_path, &destination_path)
.map_err(|e| format!("Failed to copy snapshot file: {}", e))?;
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::{
copy_dir_contents, pm2_home_dir_from, resolve_process_id_from_list, split_start_command,
Pm2Process,
};
use std::fs;
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};
fn temp_dir(label: &str) -> PathBuf {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("time")
.as_nanos();
let dir = std::env::temp_dir().join(format!("xbp-pm2-{}-{}", label, nanos));
fs::create_dir_all(&dir).expect("create temp dir");
dir
}
#[test]
fn splits_start_command_into_argv_tokens() {
assert_eq!(
split_start_command("./target/release/mollie-rs --port 8080"),
vec![
"./target/release/mollie-rs".to_string(),
"--port".to_string(),
"8080".to_string()
]
);
assert_eq!(
split_start_command(r#""C:\Program Files\app\bin" --flag "a b""#),
vec![
r"C:\Program Files\app\bin".to_string(),
"--flag".to_string(),
"a b".to_string()
]
);
}
#[test]
fn resolves_pm2_name_to_id() {
let processes = vec![
Pm2Process {
name: "api".to_string(),
pm_id: 4,
},
Pm2Process {
name: "worker".to_string(),
pm_id: 7,
},
];
let resolved =
resolve_process_id_from_list("worker", &processes).expect("worker should resolve");
assert_eq!(resolved, "7");
}
#[test]
fn errors_when_pm2_name_is_missing() {
let processes = vec![Pm2Process {
name: "api".to_string(),
pm_id: 4,
}];
let error =
resolve_process_id_from_list("missing", &processes).expect_err("missing should fail");
assert!(error.contains("PM2 process 'missing' not found"));
assert!(error.contains("xbp list"));
}
#[test]
fn pm2_home_dir_prefers_pm2_home_override() {
let expected = if cfg!(windows) {
PathBuf::from(r"C:\tmp\pm2-home")
} else {
PathBuf::from("/tmp/pm2-home")
};
assert_eq!(
pm2_home_dir_from(Some(expected.clone()), Some(PathBuf::from("/unused-home"))),
expected
);
}
#[test]
fn pm2_home_dir_falls_back_to_home_dot_pm2() {
let home = if cfg!(windows) {
PathBuf::from(r"C:\Users\floris")
} else {
PathBuf::from("/home/floris")
};
assert_eq!(
pm2_home_dir_from(None, Some(home.clone())),
home.join(".pm2")
);
}
#[test]
fn copy_dir_contents_copies_nested_files() {
let source = temp_dir("source");
let destination = temp_dir("destination");
let nested = source.join("nested");
fs::create_dir_all(&nested).expect("create nested");
fs::write(source.join("dump.pm2"), "pm2").expect("write top-level file");
fs::write(nested.join("processes.json"), "{}").expect("write nested file");
copy_dir_contents(&source, &destination).expect("copy");
assert_eq!(
fs::read_to_string(destination.join("dump.pm2")).expect("read copied top-level file"),
"pm2"
);
assert_eq!(
fs::read_to_string(destination.join("nested").join("processes.json"))
.expect("read copied nested file"),
"{}"
);
let _ = fs::remove_dir_all(source);
let _ = fs::remove_dir_all(destination);
}
}