mecha10_core/

dependency.rs

//! 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::dependency::{DependencyGraph, StartupPlan};
//! use mecha10::types::NodeConfig;
//!
//! # async fn example() -> mecha10::Result<()> {
//! let nodes = vec![
//!     NodeConfig { id: "camera".into(), depends_on: vec![], /* ... */ },
//!     NodeConfig { id: "vision".into(), depends_on: vec!["camera".into()], /* ... */ },
//!     NodeConfig { id: "planner".into(), depends_on: 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};

// ============================================================================
// Dependency Graph
// ============================================================================

/// Dependency graph for node relationships
///
/// Maintains a directed graph of node dependencies where an edge from A to B
/// means "A depends on B" (B must start before A).
#[derive(Debug, Clone)]
pub struct DependencyGraph {
    /// Node dependencies: node_id -> vec of dependencies
    dependencies: HashMap<String, Vec<String>>,

    /// Reverse dependencies: node_id -> vec of dependents
    dependents: HashMap<String, Vec<String>>,

    /// All node IDs
    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();

        // Add all nodes first
        for config in configs {
            graph.add_node(&config.id);
        }

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

        // Validate graph
        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<()> {
        // Ensure both nodes exist
        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
            )));
        }

        // Add to dependencies map
        self.dependencies
            .entry(node_id.to_string())
            .or_default()
            .push(dependency.to_string());

        // Add to reverse dependencies map
        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<()> {
        // Check for circular dependencies
        self.check_circular_dependencies()?;

        // Check all dependencies exist
        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); // Cycle detected
                }
            }
        }

        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();

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

            // Nodes with no dependencies can start immediately
            if degree == 0 {
                queue.push_back(node.clone());
            }
        }

        // Process nodes with no dependencies
        while let Some(node) = queue.pop_front() {
            result.push(node.clone());

            // Reduce in-degree for all dependents
            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());
                        }
                    }
                }
            }
        }

        // Check if all nodes were processed
        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();

        // Find nodes with no dependencies (level 0)
        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());
        }

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

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

                // Check if all dependencies are processed
                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() {
                // No progress - must be a cycle
                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> {
        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) = self.dependencies.get(&current) {
                for dep in deps {
                    if result.insert(dep.clone()) {
                        to_visit.push(dep.clone());
                    }
                }
            }
        }

        result
    }

    /// 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> {
        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) = self.dependents.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 // No dependencies
        }
    }

    /// 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 // No dependents
        }
    }

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

    /// Get nodes this node depends on
    pub fn get_direct_dependencies(&self, node_id: &str) -> Vec<String> {
        self.dependencies.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()
    }
}

// ============================================================================
// Startup Plan
// ============================================================================

/// Startup plan with nodes grouped by dependency level
///
/// Nodes at the same level can be started in parallel.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StartupPlan {
    /// Nodes grouped by dependency level
    /// Level 0: No dependencies
    /// Level N: All dependencies are in levels 0..N-1
    pub levels: Vec<Vec<String>>,
}

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

    /// 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 {
        self.levels.iter().map(|l| l.len()).sum()
    }

    /// Get maximum parallelism (largest level size)
    pub fn max_parallelism(&self) -> usize {
        self.levels.iter().map(|l| l.len()).max().unwrap_or(0)
    }

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

// ============================================================================
// Shutdown Plan
// ============================================================================

/// Shutdown plan with nodes grouped by dependency level (reversed)
///
/// Nodes at the same level can be stopped in parallel.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ShutdownPlan {
    /// Nodes grouped by dependency level (reversed)
    pub levels: Vec<Vec<String>>,
}

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

    /// 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 {
        self.levels.iter().map(|l| l.len()).sum()
    }

    /// Get maximum parallelism (largest level size)
    pub fn max_parallelism(&self) -> usize {
        self.levels.iter().map(|l| l.len()).max().unwrap_or(0)
    }

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

// ============================================================================
// Helper Functions
// ============================================================================

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

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

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

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