zinit 0.3.9

//! Public API methods for IPC.

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

use crate::sdk::{LogLine, ReloadResult, ServiceConfig, ServiceState, ServiceStatus, WhyBlocked};

use crate::server::error::{SupervisorError, SupervisorResult};
use crate::server::graph::{Service, ServiceId};
use crate::server::log;
use crate::server::process::{parse_signal, send_signal_to_group};
use nix::sys::signal::Signal;

use super::SYSTEM_CONFIG_DIR;
use super::Supervisor;
use super::events::TimeoutKind;

impl Supervisor {
    /// List all services - returns just names.
    pub async fn list_services(&self) -> Vec<String> {
        let graph = self.graph.read().await;
        graph
            .all_services()
            .filter_map(|id| graph.get(id).map(|s| s.name.clone()))
            .collect()
    }

    /// Get service status (simplified).
    pub async fn get_status(&self, name: &str) -> SupervisorResult<ServiceStatus> {
        let graph = self.graph.read().await;
        let id = graph
            .get_by_name(name)
            .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))?;

        let service = graph.get(id).unwrap();
        Ok(ServiceStatus::from_state(
            service.name.clone(),
            &service.state,
        ))
    }

    /// Start a service.
    pub async fn start_service(&mut self, name: &str) -> SupervisorResult<()> {
        let id = {
            let graph = self.graph.read().await;
            graph
                .get_by_name(name)
                .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))?
        };

        self.try_start_service(id).await;
        Ok(())
    }

    /// Stop a service.
    /// This cascades to dependents: all services that depend on this one
    /// are stopped first (in reverse topological order), then this service is stopped.
    /// Dependencies (services this one depends on) are NOT stopped automatically.
    pub async fn stop_service(&mut self, name: &str) -> SupervisorResult<()> {
        // First, get all running dependents that need to be stopped
        let dependents_to_stop: Vec<String> = {
            let graph = self.graph.read().await;
            let id = graph
                .get_by_name(name)
                .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))?;

            // Get all transitive dependents in order (most dependent first)
            graph
                .all_dependents_ordered(id)
                .into_iter()
                .filter_map(|dep_id| {
                    let dep = graph.get(dep_id)?;
                    // Only include running services
                    if dep.state.is_active() || dep.state.pid().is_some() {
                        Some(dep.name.clone())
                    } else {
                        None
                    }
                })
                .collect()
        };

        // Stop all running dependents first (cascade) - MUST succeed or we abort
        if !dependents_to_stop.is_empty() {
            tracing::info!(
                service = %name,
                dependents = ?dependents_to_stop,
                "stopping dependents first"
            );

            for dep_name in &dependents_to_stop {
                // Recursively stop each dependent (this handles their dependents too)
                if let Err(e) = self.stop_single_service(dep_name).await {
                    return Err(SupervisorError::CascadeStopFailed {
                        operation: "stop".to_string(),
                        service: name.to_string(),
                        dependent: dep_name.clone(),
                        reason: e.to_string(),
                    });
                }
            }
        }

        // Now stop the requested service
        self.stop_single_service(name).await
    }

    /// Internal helper: stop a single service without cascade logic.
    pub(crate) async fn stop_single_service(&mut self, name: &str) -> SupervisorResult<()> {
        let (id, pid, stop_signal, stop_timeout) = {
            let graph = self.graph.read().await;
            let id = graph
                .get_by_name(name)
                .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))?;

            let service = graph.get(id).unwrap();

            // Check if service is already stopped or stopping
            if matches!(
                service.state,
                ServiceState::Inactive
                    | ServiceState::Exited { .. }
                    | ServiceState::Stopping { .. }
            ) {
                tracing::debug!(service = %name, state = %service.state.name(), "service already stopped or stopping");
                return Ok(());
            }

            // Targets (pid=0) don't need signal, just mark inactive
            let pid = service.state.pid();
            if pid == Some(0) || pid.is_none() {
                drop(graph);
                let mut graph = self.graph.write().await;
                graph.set_state(id, ServiceState::Inactive);
                tracing::info!(service = %name, "target/service marked inactive");
                return Ok(());
            }

            let pid = pid.unwrap();

            let stop_signal = service
                .service_config()
                .map(|c| c.lifecycle.stop_signal.clone())
                .unwrap_or_else(|| "SIGTERM".to_string());

            let stop_timeout = service
                .service_config()
                .map(|c| c.lifecycle.stop_timeout_ms)
                .unwrap_or(10000);

            (id, pid, stop_signal, stop_timeout)
        };

        let sig = parse_signal(&stop_signal)?;
        send_signal_to_group(pid, sig)?;

        {
            let mut graph = self.graph.write().await;
            graph.set_state(id, ServiceState::Stopping { pid });
        }

        self.schedule_timeout(id, TimeoutKind::Stop, stop_timeout);

        tracing::info!(service = %name, signal = %stop_signal, pgid = pid, "stopping service (process group)");
        Ok(())
    }

    /// Stop a service and wait for it to actually stop.
    /// Enforces: 5s graceful (SIGTERM), then kills entire process tree with verification.
    /// Used during cascade stop for removal operations.
    async fn stop_and_wait(
        &mut self,
        service_name: &str,
        operation: &str,
        parent_name: &str,
    ) -> SupervisorResult<()> {
        const GRACEFUL_TIMEOUT: Duration = Duration::from_secs(5);
        const POLL_INTERVAL: Duration = Duration::from_millis(100);

        // Get service info and send SIGTERM
        let pid = {
            let graph = self.graph.read().await;
            let id = match graph.get_by_name(service_name) {
                Some(id) => id,
                None => return Ok(()), // Service doesn't exist, consider it stopped
            };
            let service = match graph.get(id) {
                Some(s) => s,
                None => return Ok(()),
            };

            // Already stopped?
            if matches!(
                service.state,
                ServiceState::Inactive | ServiceState::Exited { .. } | ServiceState::Failed { .. }
            ) {
                return Ok(());
            }

            service.state.pid()
        };

        // If no PID (target service), just mark inactive
        let pid = match pid {
            Some(p) if p > 0 => p,
            _ => {
                // Mark as inactive and return
                let graph = self.graph.read().await;
                if let Some(id) = graph.get_by_name(service_name) {
                    drop(graph);
                    let mut graph = self.graph.write().await;
                    graph.set_state(id, ServiceState::Inactive);
                }
                return Ok(());
            }
        };

        // Send SIGTERM
        if let Err(e) = send_signal_to_group(pid, Signal::SIGTERM) {
            tracing::warn!(service = %service_name, pid = pid, error = %e, "failed to send SIGTERM");
        }

        // Update state to Stopping
        {
            let graph = self.graph.read().await;
            if let Some(id) = graph.get_by_name(service_name) {
                drop(graph);
                let mut graph = self.graph.write().await;
                graph.set_state(id, ServiceState::Stopping { pid });
            }
        }

        // Wait up to 5s for graceful stop
        let start = std::time::Instant::now();
        while start.elapsed() < GRACEFUL_TIMEOUT {
            tokio::time::sleep(POLL_INTERVAL).await;

            let is_stopped = {
                let graph = self.graph.read().await;
                graph
                    .get_by_name(service_name)
                    .and_then(|id| graph.get(id))
                    .is_none_or(|s| {
                        matches!(
                            s.state,
                            ServiceState::Inactive
                                | ServiceState::Exited { .. }
                                | ServiceState::Failed { .. }
                        )
                    })
            };

            if is_stopped {
                tracing::debug!(service = %service_name, "stopped gracefully");
                return Ok(());
            }
        }

        // Graceful timeout exceeded - use kill_process_tree to kill process and all children
        // This enumerates all children via sysinfo, kills them leaf-to-root, then verifies all are dead
        tracing::warn!(service = %service_name, pid = pid, "graceful stop timeout (5s), killing process tree");

        use crate::server::process::kill_process_tree;
        let kill_result = kill_process_tree(pid);

        if !kill_result.success {
            // Some processes survived SIGKILL - this is a serious problem
            tracing::error!(
                service = %service_name,
                pid = pid,
                failed_pids = ?kill_result.failed,
                "CRITICAL: some processes survived SIGKILL in process tree"
            );
            return Err(SupervisorError::CascadeStopFailed {
                operation: operation.to_string(),
                service: parent_name.to_string(),
                dependent: service_name.to_string(),
                reason: format!(
                    "processes {:?} did not die after SIGKILL",
                    kill_result.failed
                ),
            });
        }

        // All processes killed successfully, update state
        tracing::debug!(
            service = %service_name,
            pid = pid,
            killed_count = kill_result.killed.len(),
            "process tree killed successfully"
        );

        // Force state update since process is confirmed dead
        let graph = self.graph.read().await;
        if let Some(id) = graph.get_by_name(service_name) {
            drop(graph);
            let mut graph = self.graph.write().await;
            if let Some(service) = graph.get(id) {
                if matches!(
                    service.state,
                    ServiceState::Stopping { .. } | ServiceState::Running { .. }
                ) {
                    graph.set_state(
                        id,
                        ServiceState::Failed {
                            reason: crate::sdk::FailureReason::Signal { signal: 9 }, // SIGKILL
                        },
                    );
                }
            }
        }

        Ok(())
    }

    /// Restart a service.
    pub async fn restart_service(&mut self, name: &str) -> SupervisorResult<()> {
        self.stop_service(name).await?;
        Ok(())
    }

    /// Send a signal to a service.
    pub async fn kill_service(&mut self, name: &str, signal: Option<&str>) -> SupervisorResult<()> {
        let graph = self.graph.read().await;
        let id = graph
            .get_by_name(name)
            .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))?;

        let service = graph.get(id).unwrap();
        let pid = service
            .state
            .pid()
            .ok_or_else(|| SupervisorError::ServiceNotRunning(name.to_string()))?;

        let sig = parse_signal(signal.unwrap_or("SIGTERM"))?;
        send_signal_to_group(pid, sig)?;

        tracing::info!(service = %name, signal = ?sig, pgid = pid, "sent signal to process group");
        Ok(())
    }

    /// Get why a service is blocked.
    pub async fn why_blocked(&self, name: &str) -> SupervisorResult<WhyBlocked> {
        let graph = self.graph.read().await;
        let id = graph
            .get_by_name(name)
            .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))?;

        let service = graph.get(id).unwrap();
        let blocked = matches!(service.state, ServiceState::Blocked { .. });

        let (waiting_on, conflicts_with, port_conflict, process_conflict) =
            match graph.can_start(id) {
                Ok(()) => (vec![], vec![], None, None),
                Err(reason) => {
                    let port_msg = match &reason {
                        crate::server::error::BlockedReason::PortConflict { ports, services } => {
                            Some(format!(
                                "port {} in use by service {}",
                                ports
                                    .iter()
                                    .map(|p| p.to_string())
                                    .collect::<Vec<_>>()
                                    .join(", "),
                                services.join(", ")
                            ))
                        }
                        crate::server::error::BlockedReason::ExternalPortConflict {
                            port,
                            pid,
                            process_name,
                        } => {
                            let proc_info = match (pid, process_name) {
                                (Some(p), Some(name)) => format!("{} (PID {})", name, p),
                                (Some(p), None) => format!("PID {}", p),
                                (None, Some(name)) => name.clone(),
                                (None, None) => "unknown process".to_string(),
                            };
                            Some(format!(
                                "port {} in use by external process: {}",
                                port, proc_info
                            ))
                        }
                        _ => None,
                    };

                    let proc_conflict = match &reason {
                        crate::server::error::BlockedReason::ProcessNameConflict {
                            filter,
                            processes,
                        } => Some(crate::sdk::responses::ProcessConflictDetails {
                            filter: filter.clone(),
                            processes: processes.clone(),
                        }),
                        _ => None,
                    };

                    (
                        reason.waiting_on(),
                        reason.conflicts_with(),
                        port_msg,
                        proc_conflict,
                    )
                }
            };

        let ascii = graph
            .format_why_blocked(name)
            .unwrap_or_else(|| "Unknown service".to_string());

        Ok(WhyBlocked {
            name: name.to_string(),
            blocked,
            waiting_on,
            conflicts_with,
            port_conflict,
            process_conflict,
            ascii,
        })
    }

    /// Get the dependency tree.
    pub async fn get_tree(&self) -> String {
        let graph = self.graph.read().await;
        graph.format_tree()
    }

    /// Add a new service.
    pub async fn add_service(&mut self, config: ServiceConfig) -> SupervisorResult<()> {
        let errors = crate::sdk::validate::validate_service(&config);
        if !errors.is_empty() {
            return Err(SupervisorError::Validation(errors.join(", ")));
        }

        let (id, should_autostart) = {
            let mut graph = self.graph.write().await;
            let service = Service::from_service(config);
            let autostart = service.should_autostart();
            let id = graph.add_service(service)?;
            // Link this service's dependencies to create graph edges
            // This is critical for cascade stop to work correctly
            let _missing = graph.link_service_dependencies(id);
            (id, autostart)
        };

        // Only try to start if status is "start" (the default)
        // If status is "stop" or "ignore", don't auto-start
        if should_autostart {
            self.try_start_service(id).await;
        } else {
            tracing::debug!(service_id = ?id, "service added with status != start, not auto-starting");
        }

        Ok(())
    }

    /// Remove a service.
    /// This will:
    /// 1. Stop and REMOVE all dependent services (cascade removal)
    /// 2. Stop this service if running and wait for it to fully stop
    /// 3. Clean up all associated state (timers, tasks, log buffers)
    /// 4. Remove the service from the graph
    /// 5. Delete the config file from disk (if not ephemeral)
    pub async fn remove_service(&mut self, name: &str) -> SupervisorResult<()> {
        // Check if the service exists
        let (id, is_ephemeral) = {
            let graph = self.graph.read().await;
            let id = graph
                .get_by_name(name)
                .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))?;
            let is_ephemeral = graph
                .get(id)
                .map(|service| service.ephemeral)
                .unwrap_or(true);
            (id, is_ephemeral)
        };

        // Get all dependents (services that depend on this one) - they must be removed first
        // Order: most dependent first (C before B before A in chain A<-B<-C)
        let dependents_to_remove: Vec<String> = {
            let graph = self.graph.read().await;
            graph
                .all_dependents_ordered(id)
                .into_iter()
                .filter_map(|dep_id| graph.get(dep_id).map(|dep| dep.name.clone()))
                .collect()
        };

        // Recursively remove all dependents first (this will stop them)
        // Process in order: most dependent first
        for dep_name in &dependents_to_remove {
            tracing::info!(
                service = %name,
                dependent = %dep_name,
                "cascade removing dependent service"
            );
            // Recursively call remove_service - this handles stopping, cleanup, and removal
            // Box::pin to handle recursion in async
            Box::pin(self.remove_service(dep_name)).await?;
        }

        // Re-fetch the service ID since cascade removals may have invalidated it
        // (petgraph's NodeIndex can become stale after node removals)
        let id = {
            let graph = self.graph.read().await;
            graph
                .get_by_name(name)
                .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))?
        };

        // Now stop this service if it's running (same 5s graceful + SIGKILL logic)
        let is_active = {
            let graph = self.graph.read().await;
            graph
                .get(id)
                .is_some_and(|service| service.state.is_active())
        };

        if is_active {
            // Use the same stop_and_wait logic but we can't fail on error for the main service
            // since we've already stopped all dependents - log and continue
            if let Err(e) = self.stop_and_wait(name, "remove", name).await {
                tracing::warn!(service = %name, error = %e, "failed to stop service cleanly, proceeding with removal");
            }
        }

        // Clean up all associated state BEFORE removing from graph
        // Cancel any pending timers
        self.cancel_timeout(id, TimeoutKind::Start);
        self.cancel_timeout(id, TimeoutKind::Stop);
        self.cancel_timeout(id, TimeoutKind::HealthCheck);
        self.cancel_timeout(id, TimeoutKind::RestartDelay);

        // Remove from pending restarts
        self.pending_restarts.remove(&id);

        // Remove health check attempts
        self.health_attempts.remove(&id);

        // Abort and remove process monitoring task
        if let Some(handle) = self.process_tasks.remove(&id) {
            handle.abort();
        }

        // Clean up log buffer
        log::remove_buffer(&self.log_buffers, id).await;

        // Remove from graph
        let mut graph = self.graph.write().await;
        graph.remove_service(name)?;
        drop(graph);

        // Delete the config file from disk if it exists (non-ephemeral services)
        if !is_ephemeral {
            for ext in &["toml", "yaml", "yml"] {
                let config_file = self.config_dir.join(format!("{}.{}", name, ext));
                if config_file.exists() {
                    if let Err(e) = std::fs::remove_file(&config_file) {
                        return Err(SupervisorError::ConfigDeleteFailed {
                            service: name.to_string(),
                            reason: format!("{}: {}", config_file.display(), e),
                        });
                    }
                    tracing::info!(service = name, path = %config_file.display(), "deleted service config file");
                    break; // Only one config file per service
                }
            }
        }

        Ok(())
    }
    /// Reload configuration.
    pub async fn reload(&mut self) -> SupervisorResult<ReloadResult> {
        let mut graph = self.graph.write().await;

        // Build old_id -> name mapping before reload
        let old_id_to_name: HashMap<ServiceId, String> = graph
            .all_services()
            .filter_map(|id| graph.get(id).map(|s| (id, s.name.clone())))
            .collect();

        let result = if self.pid1_mode {
            let system_dir = std::path::Path::new(SYSTEM_CONFIG_DIR);
            let (result, new_system_names) =
                graph.reload_from_directories(Some(system_dir), &self.config_dir)?;
            self.system_service_names = new_system_names;
            result
        } else {
            graph.reload_from_directory(&self.config_dir)?
        };

        // Build name -> new_id mapping after reload
        let name_to_new_id: HashMap<String, ServiceId> = graph
            .all_services()
            .filter_map(|id| graph.get(id).map(|s| (s.name.clone(), id)))
            .collect();

        // Release lock before remapping
        drop(graph);

        // Remap all ServiceId-keyed maps
        self.remap_service_ids(&old_id_to_name, &name_to_new_id)
            .await;

        tracing::info!(
            added = result.added.len(),
            removed = result.removed.len(),
            changed = result.changed.len(),
            pid1_mode = self.pid1_mode,
            "configuration reloaded"
        );

        Ok(result)
    }

    /// Get logs for a service.
    pub async fn get_logs(
        &self,
        name: &str,
        lines: Option<usize>,
    ) -> SupervisorResult<Vec<LogLine>> {
        let graph = self.graph.read().await;
        let id = graph
            .get_by_name(name)
            .ok_or_else(|| SupervisorError::ServiceNotFound(name.to_string()))?;

        drop(graph);

        Ok(log::get_logs(&self.log_buffers, id, lines).await)
    }
}