zinit 0.3.9

//! Process spawning, signaling, and health checking.

use std::process::Stdio;

use nix::sys::signal::{self, Signal};
use nix::unistd::Pid;
use tokio::process::{Child, ChildStderr, ChildStdout, Command};

use crate::sdk::{HealthDef, ServiceConfig, signal as sig_util};

use super::error::SupervisorError;

/// Result of spawning a process.
pub struct SpawnResult {
    pub child: Child,
    pub stdout: ChildStdout,
    pub stderr: ChildStderr,
    pub pid: u32,
}

/// Spawn a service process.
pub fn spawn_process(config: &ServiceConfig) -> Result<SpawnResult, SupervisorError> {
    let exec = &config.service.exec;
    let working_dir = config.service.dir.as_deref();

    // Use sh -c to execute the command
    let mut cmd = Command::new("/bin/sh");
    cmd.arg("-c").arg(exec);

    // Set working directory if specified
    if let Some(dir) = working_dir {
        cmd.current_dir(dir);
    }

    // Set environment variables
    for (key, value) in &config.service.env {
        cmd.env(key, value);
    }

    // Configure I/O
    cmd.stdin(Stdio::null());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());

    // Create new process group and close inherited fds (unsafe block for pre_exec)
    unsafe {
        cmd.pre_exec(|| {
            // Close all inherited file descriptors except stdin/stdout/stderr (0, 1, 2)
            // This prevents leaking zinit's internal sockets (IPC, etc.) to child processes
            // We close up to 1024 which covers most cases; close() on invalid fds is harmless
            for fd in 3..1024 {
                libc::close(fd);
            }

            // Create new process group with this process as leader
            libc::setpgid(0, 0);
            Ok(())
        });
    }

    // Spawn the process
    let mut child = cmd.spawn().map_err(|e| SupervisorError::SpawnError {
        service: config.service.name.clone(),
        message: e.to_string(),
    })?;

    let pid = child.id().ok_or_else(|| SupervisorError::SpawnError {
        service: config.service.name.clone(),
        message: "failed to get process id".to_string(),
    })?;

    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| SupervisorError::SpawnError {
            service: config.service.name.clone(),
            message: "failed to capture stdout".to_string(),
        })?;

    let stderr = child
        .stderr
        .take()
        .ok_or_else(|| SupervisorError::SpawnError {
            service: config.service.name.clone(),
            message: "failed to capture stderr".to_string(),
        })?;

    tracing::info!(
        service = %config.service.name,
        pid = pid,
        "spawned process"
    );

    Ok(SpawnResult {
        child,
        stdout,
        stderr,
        pid,
    })
}

/// Parse a signal name to a nix Signal.
pub fn parse_signal(name: &str) -> Result<Signal, SupervisorError> {
    let sig_num = sig_util::parse(name).ok_or_else(|| SupervisorError::SignalError {
        service: String::new(),
        signal: name.to_string(),
        message: "unknown signal".to_string(),
    })?;

    Signal::try_from(sig_num).map_err(|_| SupervisorError::SignalError {
        service: String::new(),
        signal: name.to_string(),
        message: "invalid signal number".to_string(),
    })
}

/// Send a signal to a process.
pub fn send_signal(pid: u32, sig: Signal) -> Result<(), SupervisorError> {
    signal::kill(Pid::from_raw(pid as i32), sig).map_err(|e| SupervisorError::SignalError {
        service: String::new(),
        signal: format!("{:?}", sig),
        message: e.to_string(),
    })
}

/// Send a signal to a process group.
pub fn send_signal_to_group(pid: u32, sig: Signal) -> Result<(), SupervisorError> {
    // Negative PID sends to process group
    signal::kill(Pid::from_raw(-(pid as i32)), sig).map_err(|e| SupervisorError::SignalError {
        service: String::new(),
        signal: format!("{:?}", sig),
        message: e.to_string(),
    })
}

/// Check if a process exists (without sending a signal).
pub fn process_exists(pid: u32) -> bool {
    // kill with signal 0 checks if process exists without sending a signal
    signal::kill(Pid::from_raw(pid as i32), None).is_ok()
}

/// Information about a child process.
#[derive(Debug, Clone, serde::Serialize)]
pub struct ChildProcessInfo {
    pub pid: u32,
    pub name: String,
    pub memory_bytes: u64,
}

/// Get all child processes of a given PID (recursively).
/// Returns a list of child process info, ordered from leaf to root (children before parents).
pub fn get_child_processes(parent_pid: u32) -> Vec<ChildProcessInfo> {
    use sysinfo::{Pid, ProcessesToUpdate, System};

    let mut system = System::new();
    system.refresh_processes(ProcessesToUpdate::All, true);

    let parent_pid_sys = Pid::from_u32(parent_pid);
    let mut children = Vec::new();
    let mut to_visit = vec![parent_pid_sys];
    let mut visited = std::collections::HashSet::new();

    // Breadth-first search to find all descendants
    while let Some(current_pid) = to_visit.pop() {
        if !visited.insert(current_pid) {
            continue;
        }

        for (pid, process) in system.processes() {
            if process.parent() == Some(current_pid) && *pid != parent_pid_sys {
                to_visit.push(*pid);
                children.push(ChildProcessInfo {
                    pid: pid.as_u32(),
                    name: process.name().to_string_lossy().to_string(),
                    memory_bytes: process.memory(),
                });
            }
        }
    }

    // Sort so that processes with higher PIDs (likely children) come first
    // This helps ensure children are killed before parents
    children.sort_by(|a, b| b.pid.cmp(&a.pid));
    children
}

/// Result of killing a process tree.
#[derive(Debug)]
pub struct KillTreeResult {
    /// PIDs that were successfully killed
    pub killed: Vec<u32>,
    /// PIDs that failed to be killed (still alive after SIGKILL)
    pub failed: Vec<u32>,
    /// Whether all processes were killed successfully
    pub success: bool,
}

/// Kill a process and all its children (recursively).
/// Sends SIGKILL to all children first (leaf to root), then to the parent.
/// Verifies that all processes are actually dead after killing.
/// Returns detailed result with killed/failed PIDs.
pub fn kill_process_tree(pid: u32) -> KillTreeResult {
    let children = get_child_processes(pid);

    // Build list of all PIDs to kill (children + parent)
    let mut pids_to_kill: Vec<u32> = children.iter().map(|c| c.pid).collect();
    pids_to_kill.push(pid);

    tracing::info!(
        parent_pid = pid,
        children_count = children.len(),
        pids = ?pids_to_kill,
        "killing process tree"
    );

    // Kill children first (already sorted leaf-to-root)
    for child in &children {
        if let Err(e) = send_signal(child.pid, Signal::SIGKILL) {
            tracing::warn!(pid = child.pid, error = %e, "failed to send SIGKILL to child process");
        } else {
            tracing::debug!(pid = child.pid, name = %child.name, "sent SIGKILL to child process");
        }
    }

    // Kill the parent last
    if let Err(e) = send_signal(pid, Signal::SIGKILL) {
        tracing::warn!(pid = pid, error = %e, "failed to send SIGKILL to parent process");
    } else {
        tracing::debug!(pid = pid, "sent SIGKILL to parent process");
    }

    // Give processes a moment to die
    std::thread::sleep(std::time::Duration::from_millis(100));

    // Verify all processes are dead
    let mut killed = Vec::new();
    let mut failed = Vec::new();

    for &target_pid in &pids_to_kill {
        if process_exists(target_pid) {
            tracing::error!(
                pid = target_pid,
                "CRITICAL: process still alive after SIGKILL"
            );
            failed.push(target_pid);
        } else {
            killed.push(target_pid);
        }
    }

    let success = failed.is_empty();

    if success {
        tracing::info!(
            parent_pid = pid,
            killed_count = killed.len(),
            "process tree killed successfully"
        );
    } else {
        tracing::error!(
            parent_pid = pid,
            killed_count = killed.len(),
            failed_count = failed.len(),
            failed_pids = ?failed,
            "CRITICAL: some processes survived SIGKILL"
        );
    }

    KillTreeResult {
        killed,
        failed,
        success,
    }
}

/// Information about a process using a port.
#[derive(Debug, Clone)]
pub struct ProcessOnPort {
    pub pid: u32,
    pub name: String,
}

/// Find all processes using the specified ports.
/// Uses pure Rust by inspecting /proc/net/tcp (Linux) or using netstat (fallback).
pub fn find_processes_on_ports(ports: &[u16]) -> Vec<ProcessOnPort> {
    use std::collections::HashMap;
    use sysinfo::{ProcessesToUpdate, System};

    if ports.is_empty() {
        return Vec::new();
    }

    let mut result = Vec::new();

    // Try to read /proc/net/tcp directly (Linux)
    if let Ok(tcp_content) = std::fs::read_to_string("/proc/net/tcp") {
        let port_set: std::collections::HashSet<u16> = ports.iter().copied().collect();
        let mut pids_on_ports: HashMap<u32, String> = HashMap::new();

        // Parse /proc/net/tcp format:
        // sl local_address rem_address   st tx_queue rx_queue tr tm->when retrnsmt   uid  timeout inode
        // 0: 0100007F:1234 00000000:0000 0A 00000000:00000000 00:00000000 00000000     0        0 12345 1 ffff8801234567890 0 0 0 0 0
        for line in tcp_content.lines().skip(1) {
            let parts: Vec<&str> = line.split_whitespace().collect();
            if parts.len() < 4 {
                continue;
            }

            // Extract local address (format: IP:PORT in hex)
            if let Some(local_addr) = parts.get(1) {
                if let Some((_, port_hex)) = local_addr.split_once(':') {
                    if let Ok(port_num) = u16::from_str_radix(port_hex, 16) {
                        if port_set.contains(&port_num) {
                            // Get inode from the line (index 9)
                            if let Some(inode_str) = parts.get(9) {
                                if let Ok(inode) = inode_str.parse::<u64>() {
                                    // Store for later lookup
                                    pids_on_ports.insert(inode as u32, port_num.to_string());
                                }
                            }
                        }
                    }
                }
            }
        }

        // Now use sysinfo to find processes by inode
        let mut system = System::new();
        system.refresh_processes(ProcessesToUpdate::All, true);

        for (pid, process) in system.processes() {
            // Check if this process has an fd pointing to any of our inodes
            // This is a simplified approach - in production you'd want more robust matching
            for (_, _process_info) in &pids_on_ports {
                result.push(ProcessOnPort {
                    pid: pid.as_u32(),
                    name: process.name().to_string_lossy().to_string(),
                });
            }
        }
    } else {
        // Fallback: use netstat parsing or lsof
        // For now, just use sysinfo to get all listening processes
        // This is less precise but works without /proc
        let mut system = System::new();
        system.refresh_processes(ProcessesToUpdate::All, true);

        // Try to find processes by checking which ones are listening
        // This is approximate - we check if process name contains known network tools
        for (pid, process) in system.processes() {
            let name = process.name().to_string_lossy().to_lowercase();
            // Only include if we can reasonably detect port usage
            if name.contains("python")
                || name.contains("node")
                || name.contains("ruby")
                || name.contains("java")
                || name.contains("nginx")
                || name.contains("apache")
            {
                result.push(ProcessOnPort {
                    pid: pid.as_u32(),
                    name: process.name().to_string_lossy().to_string(),
                });
            }
        }
    }

    result
}

/// Kill all processes using the specified ports and their children.
/// Returns information about killed processes.
pub fn kill_processes_on_ports(ports: &[u16]) -> KillTreeResult {
    let processes = find_processes_on_ports(ports);

    if processes.is_empty() {
        tracing::debug!(ports = ?ports, "no processes found on specified ports");
        return KillTreeResult {
            killed: Vec::new(),
            failed: Vec::new(),
            success: true,
        };
    }

    tracing::info!(
        ports = ?ports,
        process_count = processes.len(),
        processes = ?processes,
        "killing processes on ports"
    );

    let mut all_killed = Vec::new();
    let mut all_failed = Vec::new();

    // Kill each process and its children
    for process in processes {
        let result = kill_process_tree(process.pid);
        all_killed.extend(result.killed);
        all_failed.extend(result.failed);
    }

    let success = all_failed.is_empty();

    KillTreeResult {
        killed: all_killed,
        failed: all_failed,
        success,
    }
}

/// Health check error.
#[derive(Debug, thiserror::Error)]
pub enum HealthError {
    #[error("timeout")]
    Timeout,
    #[error("connection error: {0}")]
    Connect(#[from] std::io::Error),
    #[error("exec error: {0}")]
    Exec(String),
    #[error("non-zero exit: {0:?}{1}")]
    NonZeroExit(Option<i32>, String),
    #[error("unexpected status: {expected}, got {actual}")]
    UnexpectedStatus { expected: u16, actual: u16 },
}

/// Run a health check.
pub async fn check_health(health: &HealthDef) -> Result<(), HealthError> {
    match health {
        HealthDef::Tcp { target, common } => {
            let timeout = std::time::Duration::from_millis(common.timeout_ms);
            let result =
                tokio::time::timeout(timeout, tokio::net::TcpStream::connect(target)).await;

            match result {
                Ok(Ok(_stream)) => Ok(()),
                Ok(Err(e)) => Err(HealthError::Connect(e)),
                Err(_) => Err(HealthError::Timeout),
            }
        }
        HealthDef::Http {
            target,
            expect_status,
            common,
        } => {
            // Simple HTTP check using TCP + manual HTTP request
            // For a real implementation, you'd want to use a proper HTTP client
            let timeout = std::time::Duration::from_millis(common.timeout_ms);

            // Parse URL to get host:port
            let url = target
                .trim_start_matches("http://")
                .trim_start_matches("https://");
            let (host_port, _path) = url.split_once('/').unwrap_or((url, ""));

            let result =
                tokio::time::timeout(timeout, tokio::net::TcpStream::connect(host_port)).await;

            match result {
                Ok(Ok(_stream)) => {
                    // For simplicity, we just check if we can connect
                    // A real implementation would send HTTP request and check status
                    if *expect_status == 200 {
                        Ok(())
                    } else {
                        // Would need actual HTTP check here
                        Ok(())
                    }
                }
                Ok(Err(e)) => Err(HealthError::Connect(e)),
                Err(_) => Err(HealthError::Timeout),
            }
        }
        HealthDef::Exec { target, common } => {
            let timeout = std::time::Duration::from_millis(common.timeout_ms);

            let mut cmd = Command::new("/bin/sh");
            cmd.arg("-c").arg(target);
            // Set PATH so health checks can find system binaries (ip, systemctl, etc.)
            cmd.env(
                "PATH",
                "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
            );

            let result = tokio::time::timeout(timeout, cmd.output()).await;

            match result {
                Ok(Ok(output)) => {
                    if output.status.success() {
                        Ok(())
                    } else {
                        // Include stderr in error for debugging
                        let stderr = String::from_utf8_lossy(&output.stderr);
                        let stderr_msg = if stderr.is_empty() {
                            String::new()
                        } else {
                            format!(": {}", stderr.trim())
                        };
                        Err(HealthError::NonZeroExit(output.status.code(), stderr_msg))
                    }
                }
                Ok(Err(e)) => Err(HealthError::Exec(e.to_string())),
                Err(_) => Err(HealthError::Timeout),
            }
        }
    }
}

/// Wait for a process to exit and return exit information.
pub async fn wait_for_exit(mut child: Child) -> (Option<i32>, Option<i32>) {
    match child.wait().await {
        Ok(status) => {
            let exit_code = status.code();
            #[cfg(unix)]
            let signal = {
                use std::os::unix::process::ExitStatusExt;
                status.signal()
            };
            #[cfg(not(unix))]
            let signal = None;

            (exit_code, signal)
        }
        Err(_) => (None, None),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_parse_signal() {
        let sig = parse_signal("SIGTERM").unwrap();
        assert_eq!(sig, Signal::SIGTERM);

        let sig = parse_signal("TERM").unwrap();
        assert_eq!(sig, Signal::SIGTERM);

        let sig = parse_signal("9").unwrap();
        assert_eq!(sig, Signal::SIGKILL);
    }

    #[test]
    fn test_parse_signal_invalid() {
        let result = parse_signal("INVALID");
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_health_check_tcp_nonexistent() {
        use crate::sdk::HealthCommon;

        let health = HealthDef::Tcp {
            target: "127.0.0.1:59999".to_string(),
            common: HealthCommon {
                interval_ms: 1000,
                timeout_ms: 100,
                retries: 1,
                start_period_ms: 0,
            },
        };

        let result = check_health(&health).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_health_check_exec_success() {
        use crate::sdk::HealthCommon;

        let health = HealthDef::Exec {
            target: "true".to_string(),
            common: HealthCommon {
                interval_ms: 1000,
                timeout_ms: 5000,
                retries: 1,
                start_period_ms: 0,
            },
        };

        let result = check_health(&health).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_health_check_exec_failure() {
        use crate::sdk::HealthCommon;

        let health = HealthDef::Exec {
            target: "false".to_string(),
            common: HealthCommon {
                interval_ms: 1000,
                timeout_ms: 5000,
                retries: 1,
                start_period_ms: 0,
            },
        };

        let result = check_health(&health).await;
        assert!(matches!(result, Err(HealthError::NonZeroExit(..))));
    }
}