mothership 0.0.100

//! Bay - A docked ship using the docking protocol
//!
//! Bays are processes that communicate via Unix socket using the docking protocol.

use std::collections::HashMap;
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::Arc;
use std::time::Duration;

use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, Command};
use tokio::sync::Mutex;
use tracing::{debug, error, info};

use crate::charter::RouteConfig;
use crate::docking::DockingConnector;

use super::ship::is_stdout_suppressed;

/// Bay status
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum BayStatus {
    #[default]
    Pending,
    Starting,
    Docking,
    Docked,
    Failed,
    Stopped,
}

/// A bay in the fleet (docked process)
pub struct Bay {
    /// Bay name (unique identifier)
    pub name: String,
    /// Bay type (e.g., "websocket", "grpc")
    pub bay_type: String,
    /// Command to run
    pub command: String,
    /// Command arguments
    pub args: Vec<String>,
    /// Environment variables
    pub env: HashMap<String, String>,
    /// Dependencies (other ship/bay names)
    pub depends_on: Vec<String>,
    /// HTTP routes
    pub routes: Vec<RouteConfig>,
    /// Critical bay - crash kills fleet
    pub critical: bool,
    /// Configuration passed to ship via Moored message
    pub config: HashMap<String, String>,
    /// Unix socket path for docking
    socket_path: PathBuf,
    /// Current status
    status: Arc<Mutex<BayStatus>>,
    /// Child process handle
    child: Arc<Mutex<Option<Child>>>,
    /// Docking connector (once connected)
    connector: Arc<Mutex<Option<Arc<DockingConnector>>>>,
}

impl Bay {
    /// Create a new bay
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        name: String,
        bay_type: String,
        command: String,
        args: Vec<String>,
        env: HashMap<String, String>,
        depends_on: Vec<String>,
        routes: Vec<RouteConfig>,
        critical: bool,
        config: HashMap<String, String>,
    ) -> Self {
        // Socket path: $MS_SOCKET_DIR/<name>.sock or /tmp/mothership/<name>.sock
        let socket_dir = std::env::var("MS_SOCKET_DIR").unwrap_or_else(|_| {
            std::env::var("XDG_RUNTIME_DIR")
                .map(|d| format!("{}/mothership", d))
                .unwrap_or_else(|_| "/tmp/mothership".to_string())
        });
        let socket_path = PathBuf::from(socket_dir).join(format!("{}.sock", name));

        Self {
            name,
            bay_type,
            command,
            args,
            env,
            depends_on,
            routes,
            critical,
            config,
            socket_path,
            status: Arc::new(Mutex::new(BayStatus::Pending)),
            child: Arc::new(Mutex::new(None)),
            connector: Arc::new(Mutex::new(None)),
        }
    }

    /// Get current status
    pub async fn status(&self) -> BayStatus {
        *self.status.lock().await
    }

    /// Get the docking connector (if connected)
    pub async fn connector(&self) -> Option<Arc<DockingConnector>> {
        self.connector.lock().await.clone()
    }

    /// Launch the bay and establish docking connection
    pub async fn launch(&self) -> anyhow::Result<()> {
        info!(bay = %self.name, bay_type = %self.bay_type, "Launching bay");

        *self.status.lock().await = BayStatus::Starting;

        // Ensure socket directory exists
        if let Some(parent) = self.socket_path.parent() {
            std::fs::create_dir_all(parent)?;
        }

        // Remove stale socket
        let _ = std::fs::remove_file(&self.socket_path);

        // Spawn the bay process
        let mut cmd = Command::new(&self.command);

        let mothership_pid = std::process::id();
        let socket_dir = self
            .socket_path
            .parent()
            .unwrap()
            .to_string_lossy()
            .to_string();

        cmd.args(&self.args)
            .envs(&self.env)
            .env("NO_COLOR", "1")
            .env("MS_PID", mothership_pid.to_string())
            .env("MS_SHIP", &self.name)
            .env("MS_SOCKET_DIR", &socket_dir)
            .env(
                "MS_SOCKET_PATH",
                self.socket_path.to_string_lossy().to_string(),
            )
            .env("MS_BAY_TYPE", &self.bay_type)
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);

        // Set up process group and parent death signal
        let parent_pid = std::process::id() as i32;
        unsafe {
            cmd.pre_exec(move || {
                // Create new process group for clean termination
                let _ = libc::setpgid(0, 0);

                // Set up parent death notification (cross-platform)
                // On Linux: uses prctl(PR_SET_PDEATHSIG)
                // On macOS/FreeBSD: spawns kqueue watcher process
                let pgid = libc::getpid(); // After setpgid(0,0), our PID is our PGID
                super::parent_death::setup_parent_death_signal_preexec(
                    parent_pid,
                    pgid,
                    libc::SIGTERM,
                );

                Ok(())
            });
        }

        let mut child = cmd.spawn()?;
        let pid = child.id();

        // Log forwarding (pass through for JSON, prefix for plain text)
        let name = self.name.clone();
        if let Some(stdout) = child.stdout.take() {
            let name = name.clone();
            tokio::spawn(async move {
                let reader = BufReader::new(stdout);
                let mut lines = reader.lines();
                while let Ok(Some(line)) = lines.next_line().await {
                    if !is_stdout_suppressed() {
                        if line.starts_with('{') {
                            println!("{}", line);
                        } else {
                            println!("[{}] {}", name, line);
                        }
                    }
                }
            });
        }

        if let Some(stderr) = child.stderr.take() {
            let name = name.clone();
            tokio::spawn(async move {
                let reader = BufReader::new(stderr);
                let mut lines = reader.lines();
                while let Ok(Some(line)) = lines.next_line().await {
                    if !is_stdout_suppressed() {
                        if line.starts_with('{') {
                            eprintln!("{}", line);
                        } else {
                            eprintln!("[{}] {}", name, line);
                        }
                    }
                }
            });
        }

        *self.child.lock().await = Some(child);
        info!(bay = %self.name, pid = ?pid, "Bay process launched");

        // Wait for socket and establish docking connection
        *self.status.lock().await = BayStatus::Docking;
        self.establish_docking().await?;

        Ok(())
    }

    /// Wait for socket and establish docking connection
    async fn establish_docking(&self) -> anyhow::Result<()> {
        info!(bay = %self.name, socket = %self.socket_path.display(), "Waiting for docking socket");

        // Wait for socket to appear (max 30 seconds)
        let timeout = Duration::from_secs(30);
        let start = std::time::Instant::now();

        while start.elapsed() < timeout {
            if self.socket_path.exists() {
                break;
            }
            tokio::time::sleep(Duration::from_millis(100)).await;

            // Check if process died
            if !self.is_running().await {
                *self.status.lock().await = BayStatus::Failed;
                anyhow::bail!("Bay process died before creating socket");
            }
        }

        if !self.socket_path.exists() {
            *self.status.lock().await = BayStatus::Failed;
            anyhow::bail!("Timeout waiting for bay socket");
        }

        // Small delay for socket to be ready
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Connect via docking protocol
        match DockingConnector::connect(&self.name, self.socket_path.clone(), self.config.clone())
            .await
        {
            Ok(connector) => {
                info!(bay = %self.name, "Docking established");
                *self.connector.lock().await = Some(Arc::new(connector));
                *self.status.lock().await = BayStatus::Docked;
                Ok(())
            }
            Err(e) => {
                error!(bay = %self.name, error = %e, "Failed to establish docking");
                *self.status.lock().await = BayStatus::Failed;
                Err(e)
            }
        }
    }

    /// Check if process is running
    pub async fn is_running(&self) -> bool {
        let mut child_guard = self.child.lock().await;
        if let Some(child) = child_guard.as_mut() {
            match child.try_wait() {
                Ok(Some(_)) => false,
                Ok(None) => true,
                Err(_) => false,
            }
        } else {
            false
        }
    }

    /// Check if docking is established
    pub async fn is_docked(&self) -> bool {
        *self.status.lock().await == BayStatus::Docked
    }

    /// Terminate the bay
    pub async fn terminate(&self) -> anyhow::Result<()> {
        info!(bay = %self.name, "Terminating bay");

        // Drop connector first
        *self.connector.lock().await = None;

        // Terminate process
        let mut child_guard = self.child.lock().await;
        if let Some(child) = child_guard.as_mut() {
            use nix::sys::signal::{self, Signal};
            use nix::unistd::Pid;

            if let Some(pid) = child.id() {
                let pgid = pid as i32;
                if signal::killpg(Pid::from_raw(pgid), Signal::SIGTERM).is_err() {
                    let _ = signal::kill(Pid::from_raw(pgid), Signal::SIGTERM);
                }
            }

            let timeout = Duration::from_secs(5);
            tokio::select! {
                _ = tokio::time::sleep(timeout) => {
                    if let Some(pid) = child.id() {
                        let pgid = pid as i32;
                        if signal::killpg(Pid::from_raw(pgid), Signal::SIGKILL).is_err() {
                            let _ = signal::kill(Pid::from_raw(pgid), Signal::SIGKILL);
                        }
                    }
                    let _ = child.kill().await;
                }
                _ = child.wait() => {
                    debug!(bay = %self.name, "Bay exited gracefully");
                }
            }
        }

        // Cleanup socket
        let _ = std::fs::remove_file(&self.socket_path);

        *self.status.lock().await = BayStatus::Stopped;
        Ok(())
    }

    /// Get PID if running
    pub async fn pid(&self) -> Option<u32> {
        let child_guard = self.child.lock().await;
        child_guard.as_ref().and_then(|c| c.id())
    }

    /// Get socket path
    pub fn socket_path(&self) -> &PathBuf {
        &self.socket_path
    }
}

impl std::fmt::Debug for Bay {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Bay")
            .field("name", &self.name)
            .field("bay_type", &self.bay_type)
            .field("socket_path", &self.socket_path)
            .finish()
    }
}

#[cfg(all(test, unix))]
mod tests {
    use std::{collections::HashMap, fs, path::Path, process::Stdio, time::Duration};

    use tempfile::tempdir;
    use tokio::process::Command;
    use tokio::time::{Instant, sleep};

    use super::Bay;

    async fn wait_for_pid_file(path: &Path) -> i32 {
        let deadline = Instant::now() + Duration::from_secs(2);
        loop {
            if let Ok(contents) = fs::read_to_string(path)
                && let Ok(pid) = contents.trim().parse::<i32>()
                && pid > 0
            {
                return pid;
            }
            if Instant::now() >= deadline {
                panic!("timed out waiting for pid file: {}", path.display());
            }
            sleep(Duration::from_millis(20)).await;
        }
    }

    fn process_exists(pid: i32) -> bool {
        let result = unsafe { libc::kill(pid, 0) };
        if result == 0 {
            return true;
        }
        std::io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH)
    }

    async fn wait_for_exit(pid: i32) {
        let deadline = Instant::now() + Duration::from_secs(2);
        loop {
            if !process_exists(pid) {
                return;
            }
            if Instant::now() >= deadline {
                panic!("process {} still running after shutdown", pid);
            }
            sleep(Duration::from_millis(20)).await;
        }
    }

    #[tokio::test]
    async fn terminate_kills_process_group_and_cleans_socket() {
        let temp = tempdir().expect("temp dir");
        let pid_path = temp.path().join("bay-child.pid");
        let socket_path = temp.path().join("bay.sock");

        let mut bay = Bay::new(
            "test-bay".to_string(),
            "test".to_string(),
            "sh".to_string(),
            vec![],
            HashMap::new(),
            vec![],
            vec![],
            false,
            HashMap::new(),
        );
        bay.socket_path = socket_path.clone();

        fs::write(&socket_path, b"stub").expect("create socket stub");

        let mut cmd = Command::new("sh");
        cmd.args(["-c", "sleep 1000 & echo $! > \"$PID_FILE\"; wait"])
            .env("PID_FILE", pid_path.to_string_lossy().to_string())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true);

        unsafe {
            cmd.pre_exec(|| {
                if libc::setpgid(0, 0) != 0 {
                    return Err(std::io::Error::last_os_error());
                }
                Ok(())
            });
        }

        let child = cmd.spawn().expect("spawn bay child");
        *bay.child.lock().await = Some(child);

        let child_pid = wait_for_pid_file(&pid_path).await;
        bay.terminate().await.expect("terminate bay");

        assert!(!socket_path.exists(), "socket file not removed");
        wait_for_exit(child_pid).await;
    }
}