mothership 0.0.100

//! Mission Control
//!
//! Coordinates fleet launches, monitors ship health, and manages shutdown sequences.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use tokio::sync::{Mutex, Notify};
use tokio::time::sleep;
use tracing::{debug, error, info, warn};

use super::bay::Bay;
use super::ship::{Ship, ShipBuilder, ShipSnapshot};
use super::telemetry::HealthChecker;
use crate::charter::Manifest;
use crate::docking::DockingConnector;

#[cfg(target_os = "linux")]
fn validate_pid1_runtime() -> anyhow::Result<()> {
    if std::process::id() == 1 {
        anyhow::bail!(
            "mothership cannot run fleet supervision as PID 1 on Linux. \
             Use an init process (`docker run --init`, compose `init: true`, or `tini`)."
        );
    }
    Ok(())
}

#[cfg(not(target_os = "linux"))]
fn validate_pid1_runtime() -> anyhow::Result<()> {
    Ok(())
}

/// Fleet command center
pub struct Fleet {
    /// All ships (by name)
    pub ships: HashMap<String, Arc<Ship>>,
    /// All bays (by name)
    pub bays: HashMap<String, Arc<Bay>>,
    /// Group membership (group name -> ship names)
    pub groups: HashMap<String, Vec<String>>,
    /// Bay type membership (bay_type -> bay names)
    pub bay_types: HashMap<String, Vec<String>>,
    /// Health checker
    health_checker: HealthChecker,
    /// Shutdown flag
    shutdown: Arc<Mutex<bool>>,
}

impl Fleet {
    /// Create fleet from manifest
    pub fn from_manifest(manifest: &Manifest) -> Self {
        let mut ships = HashMap::new();
        let mut groups = HashMap::new();
        let mut bays = HashMap::new();
        let mut bay_types = HashMap::new();

        // Flatten all fleet groups into ships
        for (group_name, group_ships) in &manifest.fleet {
            let mut group_ship_names = Vec::new();

            for ship_config in group_ships {
                let ship = Arc::new(
                    ShipBuilder::new(&ship_config.name, &ship_config.command)
                        .group(group_name)
                        .args(ship_config.args.clone())
                        .env(ship_config.env.clone())
                        .bind(ship_config.bind.clone())
                        .healthcheck(ship_config.healthcheck.clone())
                        .depends_on(ship_config.depends_on.clone())
                        .routes(ship_config.routes.clone())
                        .critical(ship_config.critical)
                        .oneshot(ship_config.oneshot)
                        .build(),
                );
                group_ship_names.push(ship_config.name.clone());
                ships.insert(ship_config.name.clone(), ship);
            }

            groups.insert(group_name.clone(), group_ship_names);
        }

        // Flatten all bays
        for (bay_type_name, type_bays) in &manifest.bays {
            let mut type_bay_names = Vec::new();

            for bay_config in type_bays {
                let bay = Arc::new(Bay::new(
                    bay_config.name.clone(),
                    bay_type_name.clone(),
                    bay_config.command.clone(),
                    bay_config.args.clone(),
                    bay_config.env.clone(),
                    bay_config.depends_on.clone(),
                    bay_config.routes.clone(),
                    bay_config.critical,
                    bay_config.config.clone(),
                ));
                type_bay_names.push(bay_config.name.clone());
                bays.insert(bay_config.name.clone(), bay);
            }

            bay_types.insert(bay_type_name.clone(), type_bay_names);
        }

        Self {
            ships,
            bays,
            groups,
            bay_types,
            health_checker: HealthChecker::new(),
            shutdown: Arc::new(Mutex::new(false)),
        }
    }

    /// Launch the entire fleet (ships and bays)
    pub async fn launch(&self) -> anyhow::Result<()> {
        validate_pid1_runtime()?;

        let total = self.ships.len() + self.bays.len();
        info!(
            "Launching fleet ({} ships, {} bays)",
            self.ships.len(),
            self.bays.len()
        );

        if total == 0 {
            info!("No ships or bays to launch");
            return Ok(());
        }

        // Launch ships and bays in dependency order
        let launch_order = self.resolve_launch_order();
        for name in launch_order {
            // Try ships first
            if let Some(ship) = self.ships.get(&name) {
                // Wait for dependencies first
                for dep in &ship.depends_on {
                    self.wait_for_dependency(dep).await?;
                }

                // Launch this ship
                ship.launch().await?;

                // Handle oneshot vs long-running
                if ship.oneshot {
                    // Wait for oneshot to complete
                    let code = ship.wait().await?;
                    if code != 0 {
                        if ship.critical {
                            return Err(anyhow::anyhow!(
                                "Oneshot job '{}' failed with exit code {}",
                                name,
                                code
                            ));
                        }
                        warn!(
                            ship = %name,
                            exit_code = code,
                            "Non-critical oneshot failed"
                        );
                    } else {
                        info!(ship = %name, "Oneshot job completed");
                    }
                } else {
                    // Wait for health check or mark healthy
                    self.wait_for_ready(ship).await?;
                    info!(ship = %name, group = %ship.group, "Ship launched");
                }
            }
            // Try bays
            else if let Some(bay) = self.bays.get(&name) {
                // Wait for dependencies first
                for dep in &bay.depends_on {
                    self.wait_for_dependency(dep).await?;
                }

                // Launch and dock
                bay.launch().await?;
                info!(bay = %name, bay_type = %bay.bay_type, "Bay docked");
            }
        }

        info!("Fleet launched successfully");
        Ok(())
    }

    /// Wait for a dependency (ship or bay) to be ready
    async fn wait_for_dependency(&self, dep: &str) -> anyhow::Result<()> {
        if let Some(dep_ship) = self.ships.get(dep) {
            self.wait_for_ready(dep_ship).await?;
        } else if let Some(dep_bay) = self.bays.get(dep) {
            // Wait for bay to be docked
            let timeout = Duration::from_secs(60);
            let start = std::time::Instant::now();
            while start.elapsed() < timeout {
                if dep_bay.is_docked().await {
                    return Ok(());
                }
                sleep(Duration::from_millis(500)).await;
            }
            anyhow::bail!("Timeout waiting for bay '{}' to dock", dep);
        } else {
            warn!(dependency = %dep, "Dependency not found, skipping");
        }
        Ok(())
    }

    /// Resolve launch order based on dependencies (topological sort)
    fn resolve_launch_order(&self) -> Vec<String> {
        let mut order = Vec::new();

        // Collect all vessel names (ships + bays)
        let mut remaining: Vec<_> = self
            .ships
            .keys()
            .cloned()
            .chain(self.bays.keys().cloned())
            .collect();

        // Helper to check if a name is a known vessel
        let is_vessel =
            |name: &str| -> bool { self.ships.contains_key(name) || self.bays.contains_key(name) };

        // Helper to get dependencies
        let get_deps = |name: &str| -> &[String] {
            if let Some(ship) = self.ships.get(name) {
                &ship.depends_on
            } else if let Some(bay) = self.bays.get(name) {
                &bay.depends_on
            } else {
                &[]
            }
        };

        while !remaining.is_empty() {
            let mut progress = false;
            remaining.retain(|name| {
                let deps = get_deps(name);
                let deps_satisfied = deps
                    .iter()
                    .all(|dep| order.contains(dep) || !is_vessel(dep));
                if deps_satisfied {
                    order.push(name.clone());
                    progress = true;
                    return false;
                }
                true
            });

            if !progress && !remaining.is_empty() {
                warn!("Possible circular dependency in fleet: {:?}", remaining);
                order.append(&mut remaining);
            }
        }

        order
    }

    /// Wait for a ship to become ready (healthy or completed)
    async fn wait_for_ready(&self, ship: &Ship) -> anyhow::Result<()> {
        let name = ship.display_name();

        // Oneshot jobs - wait for completion
        if ship.oneshot {
            // Already handled in launch()
            return Ok(());
        }

        // Check for healthcheck URL
        let url = match ship.healthcheck_url() {
            Some(url) => url,
            None => {
                // No healthcheck - mark healthy after short delay
                sleep(Duration::from_millis(500)).await;
                ship.mark_healthy().await;
                ship.reset_breaker().await;
                return Ok(());
            }
        };

        info!(ship = name, url = %url, "Waiting for health check");

        // Initial delay to allow app to boot (connect to DB, etc.)
        sleep(self.health_checker.initial_delay()).await;

        let max_attempts = 30; // 30 * 2s = 60s max wait
        for attempt in 1..=max_attempts {
            if !ship.is_running().await {
                return Err(anyhow::anyhow!("Ship {} died during startup", name));
            }

            if self.health_checker.check_http(name, &url).await {
                ship.mark_healthy().await;
                ship.reset_breaker().await;
                return Ok(());
            }

            if attempt < max_attempts {
                sleep(Duration::from_secs(2)).await;
            }
        }

        Err(anyhow::anyhow!(
            "Ship {} failed health check after {} attempts",
            name,
            max_attempts
        ))
    }

    /// Monitor fleet health and restart crashed ships
    pub async fn monitor(&self, shutdown_notify: Arc<Notify>) {
        let interval = self.health_checker.interval();

        loop {
            if *self.shutdown.lock().await {
                break;
            }

            for (name, ship) in &self.ships {
                // Skip oneshot jobs
                if ship.oneshot {
                    continue;
                }

                if ship.is_backing_off().await {
                    if !ship.backoff_expired().await {
                        continue;
                    }
                    ship.clear_backoff().await;
                    ship.increment_restart().await;
                    if let Err(e) = ship.launch().await {
                        error!(ship = %name, error = %e, "Failed to restart ship");
                        let breaker_open = ship.record_crash().await;
                        if breaker_open {
                            ship.enter_backoff().await;
                        }
                        if ship.critical {
                            error!("Critical ship failed to restart, shutting down fleet");
                            *self.shutdown.lock().await = true;
                            shutdown_notify.notify_one();
                            return;
                        }
                    } else if ship.healthcheck_url().is_none() {
                        ship.mark_healthy().await;
                        ship.reset_breaker().await;
                    }
                    continue;
                }

                if !ship.is_running().await {
                    ship.mark_failed().await;
                    let breaker_open = ship.record_crash().await;
                    if breaker_open {
                        warn!(
                            ship = %name,
                            critical = ship.critical,
                            "Restart circuit open, entering backoff"
                        );
                        ship.enter_backoff().await;
                        if ship.critical {
                            error!("Critical ship entered backoff, shutting down fleet");
                            *self.shutdown.lock().await = true;
                            shutdown_notify.notify_one();
                            return;
                        }
                        continue;
                    }

                    ship.increment_restart().await;
                    if let Err(e) = ship.launch().await {
                        error!(ship = %name, error = %e, "Failed to restart ship");
                        let breaker_open = ship.record_crash().await;
                        if breaker_open {
                            ship.enter_backoff().await;
                        }
                        if ship.critical {
                            error!("Critical ship failed to restart, shutting down fleet");
                            *self.shutdown.lock().await = true;
                            shutdown_notify.notify_one();
                            return;
                        }
                    } else if ship.healthcheck_url().is_none() {
                        ship.mark_healthy().await;
                        ship.reset_breaker().await;
                    }
                } else if let Some(url) = ship.healthcheck_url() {
                    // Skip health-triggered restarts during grace period
                    let in_grace = ship.in_grace_period().await;
                    let healthy = self.health_checker.check_http(name, &url).await;

                    if healthy {
                        ship.record_health_success().await;
                        ship.mark_healthy().await;
                    } else if in_grace {
                        // During grace period, mark unhealthy but don't trigger restart
                        debug!(
                            ship = %name,
                            "Health check failed during grace period, waiting"
                        );
                        ship.mark_unhealthy().await;
                    } else {
                        ship.mark_unhealthy().await;
                        let breaker_open = ship.record_health_failure().await;
                        if breaker_open {
                            warn!(
                                ship = %name,
                                critical = ship.critical,
                                "Health check failures opened restart circuit"
                            );
                            let _ = ship.terminate().await;
                            ship.enter_backoff().await;
                            if ship.critical {
                                error!("Critical ship entered backoff, shutting down fleet");
                                *self.shutdown.lock().await = true;
                                shutdown_notify.notify_one();
                                return;
                            }
                        }
                    }
                }
            }

            sleep(interval).await;
        }
    }

    /// Graceful shutdown of the fleet
    pub async fn shutdown(&self) {
        info!("Initiating fleet shutdown");
        *self.shutdown.lock().await = true;

        // Shutdown in reverse dependency order
        let shutdown_order: Vec<_> = self.resolve_launch_order().into_iter().rev().collect();

        for name in &shutdown_order {
            if let Some(ship) = self.ships.get(name) {
                // Skip oneshot (already completed)
                if ship.oneshot {
                    continue;
                }

                info!(ship = %name, "Shutting down ship");
                let _ = ship.terminate().await;
            } else if let Some(bay) = self.bays.get(name) {
                info!(bay = %name, "Undocking bay");
                let _ = bay.terminate().await;
            }
        }

        info!("Fleet shutdown complete");
    }

    /// Get all ships for iteration
    pub fn all_ships(&self) -> Vec<Arc<Ship>> {
        self.ships.values().cloned().collect()
    }

    /// Get all bays for iteration
    pub fn all_bays(&self) -> Vec<Arc<Bay>> {
        self.bays.values().cloned().collect()
    }

    /// Get a bay's docking connector by name
    pub async fn get_bay_connector(&self, name: &str) -> Option<Arc<DockingConnector>> {
        if let Some(bay) = self.bays.get(name) {
            bay.connector().await
        } else {
            None
        }
    }

    /// Get snapshots of all ships (for sync access like TUI)
    pub async fn ships_snapshot(&self) -> Vec<ShipSnapshot> {
        let mut snapshots = Vec::new();
        for ship in self.ships.values() {
            snapshots.push(ship.snapshot().await);
        }
        // Sort by group then name for consistent display
        snapshots.sort_by(|a, b| {
            a.group()
                .cmp(b.group())
                .then_with(|| a.name().cmp(b.name()))
        });
        snapshots
    }

    /// Get logs for a ship by index
    pub async fn ship_logs(&self, index: usize, limit: usize) -> Vec<super::ship::LogEntry> {
        let snapshots = self.ships_snapshot().await;
        if index < snapshots.len() {
            let name = snapshots[index].name();
            if let Some(ship) = self.ships.get(name) {
                return ship.logs(limit).await;
            }
        }
        vec![]
    }

    /// Get ship count
    pub fn ship_count(&self) -> usize {
        self.ships.len()
    }

    /// Get bay count
    pub fn bay_count(&self) -> usize {
        self.bays.len()
    }
}