mecha10-core 0.8.0

//! Dependency Resolution for Node Startup/Shutdown Ordering
//!
//! Provides topological sorting and dependency graph analysis to ensure nodes
//! start and stop in the correct order based on their dependencies.
//!
//! # Use Cases
//!
//! - Start nodes in dependency order (dependencies before dependents)
//! - Stop nodes in reverse order (dependents before dependencies)
//! - Detect circular dependencies
//! - Validate dependency references
//! - Group nodes by dependency level for parallel startup
//!
//! # Example
//!
//! ```rust
//! use mecha10_core::dependency::{DependencyGraph, StartupPlan};
//! use mecha10_core::types::NodeConfig;
//!
//! # fn node(id: &str, depends_on: Vec<String>) -> NodeConfig {
//! #     NodeConfig {
//! #         id: id.to_string(),
//! #         name: None,
//! #         executable: String::new(),
//! #         args: vec![],
//! #         env: Default::default(),
//! #         mode: Default::default(),
//! #         auto_restart: true,
//! #         health_check_interval_s: 5,
//! #         publishes: vec![],
//! #         subscribes: vec![],
//! #         depends_on,
//! #         placement: Default::default(),
//! #         resources: None,
//! #         scaling: None,
//! #     }
//! # }
//! # async fn example() -> mecha10_core::Result<()> {
//! let nodes = vec![
//!     node("camera", vec![]),
//!     node("vision", vec!["camera".into()]),
//!     node("planner", vec!["vision".into()]),
//! ];
//!
//! // Build dependency graph
//! let graph = DependencyGraph::from_configs(&nodes)?;
//!
//! // Get startup order
//! let startup_order = graph.topological_sort()?;
//! // Result: ["camera", "vision", "planner"]
//!
//! // Get shutdown order (reverse)
//! let shutdown_order = graph.reverse_topological_sort()?;
//! // Result: ["planner", "vision", "camera"]
//!
//! // Get parallel startup plan (nodes grouped by level)
//! let plan = graph.startup_plan()?;
//! // Level 0: ["camera"] - no dependencies
//! // Level 1: ["vision"] - depends on camera
//! // Level 2: ["planner"] - depends on vision
//! # Ok(())
//! # }
//! ```

use crate::error::{Mecha10Error, Result};
use crate::types::NodeConfig;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet, VecDeque};
use tracing::{debug, info};

#[derive(Debug, Clone)]
pub struct DependencyGraph {
    dependencies: HashMap<String, Vec<String>>,

    dependents: HashMap<String, Vec<String>>,

    nodes: HashSet<String>,
}

impl DependencyGraph {
    /// Create a new empty dependency graph
    pub fn new() -> Self {
        Self {
            dependencies: HashMap::new(),
            dependents: HashMap::new(),
            nodes: HashSet::new(),
        }
    }

    /// Build dependency graph from node configurations
    ///
    /// # Arguments
    ///
    /// * `configs` - Slice of node configurations
    ///
    /// # Returns
    ///
    /// Dependency graph or error if dependencies are invalid
    ///
    /// # Errors
    ///
    /// - Circular dependencies detected
    /// - Undefined dependency references
    pub fn from_configs(configs: &[NodeConfig]) -> Result<Self> {
        let mut graph = Self::new();

        for config in configs {
            graph.add_node(&config.id);
        }

        for config in configs {
            for dep in &config.depends_on {
                graph.add_dependency(&config.id, dep)?;
            }
        }

        graph.validate()?;

        Ok(graph)
    }

    /// Add a node to the graph
    pub fn add_node(&mut self, node_id: &str) {
        self.nodes.insert(node_id.to_string());
        self.dependencies.entry(node_id.to_string()).or_default();
        self.dependents.entry(node_id.to_string()).or_default();
    }

    /// Add a dependency relationship
    ///
    /// # Arguments
    ///
    /// * `node_id` - The dependent node
    /// * `dependency` - The node that `node_id` depends on
    ///
    /// # Errors
    ///
    /// Returns error if dependency node doesn't exist
    pub fn add_dependency(&mut self, node_id: &str, dependency: &str) -> Result<()> {
        if !self.nodes.contains(node_id) {
            return Err(Mecha10Error::Configuration(format!(
                "Node '{}' not found in graph",
                node_id
            )));
        }

        if !self.nodes.contains(dependency) {
            return Err(Mecha10Error::Configuration(format!(
                "Dependency '{}' not found (required by '{}')",
                dependency, node_id
            )));
        }

        self.dependencies
            .entry(node_id.to_string())
            .or_default()
            .push(dependency.to_string());

        self.dependents
            .entry(dependency.to_string())
            .or_default()
            .push(node_id.to_string());

        Ok(())
    }

    /// Validate the dependency graph
    ///
    /// Checks for:
    /// - Circular dependencies
    /// - Undefined dependency references
    pub fn validate(&self) -> Result<()> {
        self.check_circular_dependencies()?;

        for (node_id, deps) in &self.dependencies {
            for dep in deps {
                if !self.nodes.contains(dep) {
                    return Err(Mecha10Error::Configuration(format!(
                        "Node '{}' depends on undefined node '{}'",
                        node_id, dep
                    )));
                }
            }
        }

        Ok(())
    }

    /// Check for circular dependencies using DFS
    fn check_circular_dependencies(&self) -> Result<()> {
        let mut visited = HashSet::new();
        let mut rec_stack = HashSet::new();

        for node in &self.nodes {
            if !visited.contains(node) && self.has_cycle_dfs(node, &mut visited, &mut rec_stack)? {
                return Err(Mecha10Error::Configuration(format!(
                    "Circular dependency detected involving node '{}'",
                    node
                )));
            }
        }

        Ok(())
    }

    /// DFS helper for cycle detection
    fn has_cycle_dfs(
        &self,
        node: &str,
        visited: &mut HashSet<String>,
        rec_stack: &mut HashSet<String>,
    ) -> Result<bool> {
        visited.insert(node.to_string());
        rec_stack.insert(node.to_string());

        if let Some(deps) = self.dependencies.get(node) {
            for dep in deps {
                if !visited.contains(dep) {
                    if self.has_cycle_dfs(dep, visited, rec_stack)? {
                        return Ok(true);
                    }
                } else if rec_stack.contains(dep) {
                    return Ok(true);
                }
            }
        }

        rec_stack.remove(node);
        Ok(false)
    }

    /// Perform topological sort using Kahn's algorithm
    ///
    /// Returns nodes in dependency order (dependencies before dependents).
    /// This is the order for starting nodes.
    ///
    /// # Returns
    ///
    /// Vector of node IDs in topological order
    ///
    /// # Errors
    ///
    /// Returns error if graph has cycles
    pub fn topological_sort(&self) -> Result<Vec<String>> {
        let mut in_degree: HashMap<String, usize> = HashMap::new();
        let mut result = Vec::new();
        let mut queue = VecDeque::new();

        for node in &self.nodes {
            let degree = self.dependencies.get(node).map(|d| d.len()).unwrap_or(0);
            in_degree.insert(node.clone(), degree);

            if degree == 0 {
                queue.push_back(node.clone());
            }
        }

        while let Some(node) = queue.pop_front() {
            result.push(node.clone());

            if let Some(dependents) = self.dependents.get(&node) {
                for dependent in dependents {
                    if let Some(degree) = in_degree.get_mut(dependent) {
                        *degree -= 1;
                        if *degree == 0 {
                            queue.push_back(dependent.clone());
                        }
                    }
                }
            }
        }

        if result.len() != self.nodes.len() {
            return Err(Mecha10Error::Configuration(
                "Circular dependency detected during topological sort".to_string(),
            ));
        }

        debug!("Topological sort result: {:?}", result);
        Ok(result)
    }

    /// Get reverse topological sort (for shutdown order)
    ///
    /// Returns nodes in reverse dependency order (dependents before dependencies).
    /// This is the order for stopping nodes.
    pub fn reverse_topological_sort(&self) -> Result<Vec<String>> {
        let mut order = self.topological_sort()?;
        order.reverse();
        debug!("Reverse topological sort result: {:?}", order);
        Ok(order)
    }

    /// Get startup plan with nodes grouped by dependency level
    ///
    /// Nodes at the same level have no dependencies on each other and can
    /// be started in parallel.
    ///
    /// # Returns
    ///
    /// StartupPlan with nodes grouped by level
    pub fn startup_plan(&self) -> Result<StartupPlan> {
        let mut levels: Vec<Vec<String>> = Vec::new();
        let mut processed = HashSet::new();
        let mut current_level = Vec::new();

        for node in &self.nodes {
            if self.dependencies.get(node).map(|d| d.is_empty()).unwrap_or(true) {
                current_level.push(node.clone());
                processed.insert(node.clone());
            }
        }

        if !current_level.is_empty() {
            levels.push(current_level.clone());
        }

        while processed.len() < self.nodes.len() {
            let mut next_level = Vec::new();

            for node in &self.nodes {
                if processed.contains(node) {
                    continue;
                }

                if let Some(deps) = self.dependencies.get(node) {
                    if deps.iter().all(|dep| processed.contains(dep)) {
                        next_level.push(node.clone());
                        processed.insert(node.clone());
                    }
                }
            }

            if next_level.is_empty() && processed.len() < self.nodes.len() {
                return Err(Mecha10Error::Configuration(
                    "Circular dependency detected in startup plan".to_string(),
                ));
            }

            if !next_level.is_empty() {
                levels.push(next_level);
            }
        }

        info!("Startup plan created with {} levels", levels.len());
        for (i, level) in levels.iter().enumerate() {
            debug!("  Level {}: {:?}", i, level);
        }

        Ok(StartupPlan { levels })
    }

    /// Get shutdown plan (reverse of startup plan)
    pub fn shutdown_plan(&self) -> Result<ShutdownPlan> {
        let startup = self.startup_plan()?;
        let mut levels = startup.levels;
        levels.reverse();

        info!("Shutdown plan created with {} levels", levels.len());
        for (i, level) in levels.iter().enumerate() {
            debug!("  Level {}: {:?}", i, level);
        }

        Ok(ShutdownPlan { levels })
    }

    /// Get all dependencies of a node (transitive closure)
    ///
    /// Returns all nodes that must be started before this node.
    pub fn get_all_dependencies(&self, node_id: &str) -> HashSet<String> {
        Self::transitive_closure(&self.dependencies, node_id)
    }

    /// Get all dependents of a node (transitive closure)
    ///
    /// Returns all nodes that must be stopped before this node can be stopped.
    pub fn get_all_dependents(&self, node_id: &str) -> HashSet<String> {
        Self::transitive_closure(&self.dependents, node_id)
    }

    /// Walk `edges` (either the dependency or dependent adjacency map) from `node_id`,
    /// collecting the full transitive closure. Shared by [`Self::get_all_dependencies`] and
    /// [`Self::get_all_dependents`], which differ only in which map they traverse.
    fn transitive_closure(edges: &HashMap<String, Vec<String>>, node_id: &str) -> HashSet<String> {
        let mut result = HashSet::new();
        let mut to_visit = vec![node_id.to_string()];

        while let Some(current) = to_visit.pop() {
            if let Some(deps) = edges.get(&current) {
                for dep in deps {
                    if result.insert(dep.clone()) {
                        to_visit.push(dep.clone());
                    }
                }
            }
        }

        result
    }

    /// Check if a node can be started
    ///
    /// Returns true if all dependencies are in the provided running set.
    pub fn can_start(&self, node_id: &str, running: &HashSet<String>) -> bool {
        if let Some(deps) = self.dependencies.get(node_id) {
            deps.iter().all(|dep| running.contains(dep))
        } else {
            true
        }
    }

    /// Check if a node can be stopped
    ///
    /// Returns true if no dependents are in the provided running set.
    pub fn can_stop(&self, node_id: &str, running: &HashSet<String>) -> bool {
        if let Some(deps) = self.dependents.get(node_id) {
            !deps.iter().any(|dep| running.contains(dep))
        } else {
            true
        }
    }

    /// Get nodes that depend on this node
    pub fn get_direct_dependents(&self, node_id: &str) -> Vec<String> {
        Self::direct_edges(&self.dependents, node_id)
    }

    /// Get nodes this node depends on
    pub fn get_direct_dependencies(&self, node_id: &str) -> Vec<String> {
        Self::direct_edges(&self.dependencies, node_id)
    }

    fn direct_edges(edges: &HashMap<String, Vec<String>>, node_id: &str) -> Vec<String> {
        edges.get(node_id).cloned().unwrap_or_default()
    }

    /// Get the number of nodes in the graph
    pub fn node_count(&self) -> usize {
        self.nodes.len()
    }

    /// Get all node IDs
    pub fn all_nodes(&self) -> Vec<String> {
        self.nodes.iter().cloned().collect()
    }
}

impl Default for DependencyGraph {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StartupPlan {
    pub levels: Vec<Vec<String>>,
}

impl StartupPlan {
    /// Get total number of levels
    pub fn level_count(&self) -> usize {
        level_count(&self.levels)
    }

    /// Get nodes at a specific level
    pub fn get_level(&self, level: usize) -> Option<&Vec<String>> {
        self.levels.get(level)
    }

    /// Get total number of nodes
    pub fn node_count(&self) -> usize {
        plan_node_count(&self.levels)
    }

    /// Get maximum parallelism (largest level size)
    pub fn max_parallelism(&self) -> usize {
        max_parallelism(&self.levels)
    }

    /// Flatten to sequential order
    pub fn flatten(&self) -> Vec<String> {
        flatten_levels(&self.levels)
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ShutdownPlan {
    pub levels: Vec<Vec<String>>,
}

impl ShutdownPlan {
    /// Get total number of levels
    pub fn level_count(&self) -> usize {
        level_count(&self.levels)
    }

    /// Get nodes at a specific level
    pub fn get_level(&self, level: usize) -> Option<&Vec<String>> {
        self.levels.get(level)
    }

    /// Get total number of nodes
    pub fn node_count(&self) -> usize {
        plan_node_count(&self.levels)
    }

    /// Get maximum parallelism (largest level size)
    pub fn max_parallelism(&self) -> usize {
        max_parallelism(&self.levels)
    }

    /// Flatten to sequential order
    pub fn flatten(&self) -> Vec<String> {
        flatten_levels(&self.levels)
    }
}

/// Shared level-based queries used by both [`StartupPlan`] and [`ShutdownPlan`], which both
/// wrap `Vec<Vec<String>>` "nodes grouped by level" but can't share an impl block since they're
/// distinct public types.
fn level_count(levels: &[Vec<String>]) -> usize {
    levels.len()
}

fn plan_node_count(levels: &[Vec<String>]) -> usize {
    levels.iter().map(|l| l.len()).sum()
}

fn max_parallelism(levels: &[Vec<String>]) -> usize {
    levels.iter().map(|l| l.len()).max().unwrap_or(0)
}

fn flatten_levels(levels: &[Vec<String>]) -> Vec<String> {
    levels.iter().flatten().cloned().collect()
}

/// Create a simple dependency graph from node ID mappings
///
/// # Example
///
/// ```
/// use mecha10_core::dependency::create_graph;
///
/// # fn example() -> mecha10_core::Result<()> {
/// let graph = create_graph(&[
///     ("camera", vec![]),
///     ("vision", vec!["camera"]),
///     ("planner", vec!["vision"]),
/// ])?;
/// # Ok(())
/// # }
/// ```
pub fn create_graph(nodes: &[(&str, Vec<&str>)]) -> Result<DependencyGraph> {
    let mut graph = DependencyGraph::new();

    for (node_id, _) in nodes {
        graph.add_node(node_id);
    }

    for (node_id, deps) in nodes {
        for dep in deps {
            graph.add_dependency(node_id, dep)?;
        }
    }

    graph.validate()?;
    Ok(graph)
}