mecha10_cli/services/

topology.rs

//! Topology service for analyzing project structure
//!
//! This service provides static analysis of project topology including:
//! - Nodes and their enabled status
//! - Pub/sub topics (publishers and subscribers)
//! - Service ports (Redis, HTTP, database, etc.)

use anyhow::{Context, Result};
use regex::Regex;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};

use crate::services::ConfigService;
use crate::types::ProjectConfig;

/// Topology analysis result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Topology {
    pub project_name: String,
    pub redis: RedisInfo,
    pub services: Vec<ServiceInfo>,
    pub nodes: Vec<NodeTopology>,
    pub topics: Vec<TopicTopology>,
}

/// Redis connection information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RedisInfo {
    pub url: String,
    pub host: String,
    pub port: u16,
}

/// Service port information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServiceInfo {
    pub name: String,
    pub host: String,
    pub port: u16,
}

/// Node topology information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeTopology {
    pub name: String,
    pub package: String,
    pub enabled: bool,
    pub description: Option<String>,
    pub publishes: Vec<TopicRef>,
    pub subscribes: Vec<TopicRef>,
}

/// Topic reference with message type
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub struct TopicRef {
    pub path: String,
    pub message_type: Option<String>,
}

/// Topic topology information (grouped by topic)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TopicTopology {
    pub path: String,
    pub message_type: Option<String>,
    pub publishers: Vec<String>,
    pub subscribers: Vec<String>,
}

/// Topology service for static analysis
pub struct TopologyService {
    project_root: PathBuf,
}

impl TopologyService {
    /// Create a new topology service
    pub fn new(project_root: PathBuf) -> Self {
        Self { project_root }
    }

    /// Analyze project topology
    pub async fn analyze(&self) -> Result<Topology> {
        // Load project configuration
        let config_path = self.project_root.join("mecha10.json");
        let config = ConfigService::load_from(&config_path).await?;

        // Parse Redis URL
        let redis = self.parse_redis_url(&config.redis.url)?;

        // Extract service ports
        let services = self.extract_services(&config);

        // Analyze all enabled nodes
        let nodes = self.analyze_nodes(&config).await?;

        // Build topic-centric view
        let topics = self.build_topic_view(&nodes);

        Ok(Topology {
            project_name: config.name,
            redis,
            services,
            nodes,
            topics,
        })
    }

    /// Parse Redis URL into components
    ///
    /// Note: This method is public primarily for testing purposes.
    pub fn parse_redis_url(&self, url: &str) -> Result<RedisInfo> {
        // Handle redis://host:port format
        let url_pattern = Regex::new(r"redis://([^:]+):(\d+)")?;

        if let Some(caps) = url_pattern.captures(url) {
            let host = caps.get(1).map(|m| m.as_str()).unwrap_or("localhost");
            let port = caps.get(2).and_then(|m| m.as_str().parse::<u16>().ok()).unwrap_or(6379);

            Ok(RedisInfo {
                url: url.to_string(),
                host: host.to_string(),
                port,
            })
        } else {
            // Default fallback
            Ok(RedisInfo {
                url: url.to_string(),
                host: "localhost".to_string(),
                port: 6379,
            })
        }
    }

    /// Extract service port information from config
    fn extract_services(&self, config: &ProjectConfig) -> Vec<ServiceInfo> {
        let mut services = Vec::new();

        // HTTP API service
        if let Some(http_api) = &config.services.http_api {
            services.push(ServiceInfo {
                name: "HTTP API".to_string(),
                host: http_api.host.clone(),
                port: http_api.port,
            });
        }

        // Database service
        if let Some(db) = &config.services.database {
            // Try to parse postgres://host:port or similar
            if let Some(port) = self.extract_port_from_url(&db.url) {
                services.push(ServiceInfo {
                    name: "Database".to_string(),
                    host: self
                        .extract_host_from_url(&db.url)
                        .unwrap_or_else(|| "localhost".to_string()),
                    port,
                });
            }
        }

        services
    }

    /// Extract host from a URL string
    fn extract_host_from_url(&self, url: &str) -> Option<String> {
        let re = Regex::new(r"://([^:/@]+)").ok()?;
        re.captures(url)
            .and_then(|caps| caps.get(1).map(|m| m.as_str().to_string()))
    }

    /// Extract port from a URL string
    fn extract_port_from_url(&self, url: &str) -> Option<u16> {
        let re = Regex::new(r":(\d+)").ok()?;
        re.captures(url)
            .and_then(|caps| caps.get(1))
            .and_then(|m| m.as_str().parse().ok())
    }

    /// Analyze all enabled nodes
    async fn analyze_nodes(&self, config: &ProjectConfig) -> Result<Vec<NodeTopology>> {
        let mut nodes = Vec::new();

        // Analyze driver nodes
        for node in &config.nodes.drivers {
            if node.enabled {
                let topology = self
                    .analyze_node(&node.name, &node.path, node.description.as_deref())
                    .await?;
                nodes.push(topology);
            }
        }

        // Analyze custom nodes
        for node in &config.nodes.custom {
            if node.enabled {
                let topology = self
                    .analyze_node(&node.name, &node.path, node.description.as_deref())
                    .await?;
                nodes.push(topology);
            }
        }

        Ok(nodes)
    }

    /// Analyze a single node's configuration for topics
    async fn analyze_node(&self, name: &str, package_path: &str, description: Option<&str>) -> Result<NodeTopology> {
        // Try to load node config first
        let (publishes, subscribes) = self.load_topics_from_config(name).await.unwrap_or_else(|_| {
            // Fallback: try source parsing if config loading fails
            let source_path = match self.resolve_node_source(package_path) {
                Ok(path) if path.exists() => path,
                _ => return (Vec::new(), Vec::new()),
            };

            // Parse source file for topics (blocking operation in async context)
            match tokio::task::block_in_place(|| {
                let rt = tokio::runtime::Handle::current();
                rt.block_on(self.parse_node_source(&source_path))
            }) {
                Ok((pubs, subs)) => (pubs, subs),
                Err(_) => (Vec::new(), Vec::new()),
            }
        });

        Ok(NodeTopology {
            name: name.to_string(),
            package: package_path.to_string(),
            enabled: true,
            description: description.map(String::from),
            publishes,
            subscribes,
        })
    }

    /// Load topic information from node config file
    ///
    /// Supports multiple config formats:
    ///
    /// **Format 1: Array format (recommended)**
    /// ```json
    /// {
    ///   "topics": {
    ///     "publishes": [
    ///       { "output": "/vision/classification" },
    ///       { "status": "/motor/status" }
    ///     ],
    ///     "subscribes": [
    ///       { "input": "/camera/rgb" }
    ///     ]
    ///   }
    /// }
    /// ```
    ///
    /// **Format 2: Flat format (for nodes with many semantic topics)**
    /// ```json
    /// {
    ///   "topics": {
    ///     "command_in": "/ai/command",
    ///     "response_out": "/ai/response",
    ///     "camera_in": "/camera/rgb"
    ///   }
    /// }
    /// ```
    /// Field names ending in `_in` or `_sub` are subscribes, `_out` or `_pub` are publishes.
    ///
    /// **Format 3: Root-level fields (legacy)**
    /// ```json
    /// {
    ///   "input_topic": "/camera/rgb",
    ///   "output_topic": "/inference/detections",
    ///   "control_topic": "/inference/cmd"
    /// }
    /// ```
    async fn load_topics_from_config(&self, node_name: &str) -> Result<(Vec<TopicRef>, Vec<TopicRef>)> {
        // Try multiple config locations:
        // 1. User project: configs/common/nodes/{node_name}/config.json
        // 2. Framework monorepo: packages/nodes/{node_name}/configs/common/config.json
        let user_config_path = self
            .project_root
            .join("configs/common/nodes")
            .join(node_name)
            .join("config.json");

        let framework_config_path = self
            .project_root
            .join("packages/nodes")
            .join(node_name)
            .join("configs/common/config.json");

        let config_path = if user_config_path.exists() {
            user_config_path
        } else if framework_config_path.exists() {
            framework_config_path
        } else {
            anyhow::bail!(
                "Config file not found at {} or {}",
                user_config_path.display(),
                framework_config_path.display()
            );
        };

        // Load config as JSON
        let content = tokio::fs::read_to_string(&config_path).await?;
        let config: serde_json::Value = serde_json::from_str(&content)?;

        // Use the helper method to parse topics from JSON
        Ok(self.parse_topics_from_json(&config))
    }

    /// Resolve node source file path
    fn resolve_node_source(&self, package_path: &str) -> Result<PathBuf> {
        // Handle different path formats:
        // 1. Relative path: "./nodes/camera" or "./drivers/motor"
        // 2. Package name: "mecha10-nodes-camera"
        // 3. Framework package: "@mecha10/camera-node"

        if package_path.starts_with("./") || package_path.starts_with("../") {
            // Relative path
            let path = self.project_root.join(package_path).join("src/lib.rs");
            Ok(path)
        } else if package_path.starts_with("@mecha10/") || package_path.starts_with("mecha10-") {
            // Framework package - look in workspace packages
            let package_name = package_path
                .strip_prefix("@mecha10/")
                .or_else(|| package_path.strip_prefix("mecha10-"))
                .unwrap_or(package_path);

            // Strip "nodes-" or "drivers-" prefix if present (e.g., "mecha10-nodes-speaker" -> "speaker")
            let package_name = package_name
                .strip_prefix("nodes-")
                .or_else(|| package_name.strip_prefix("drivers-"))
                .unwrap_or(package_name);

            // Try to find framework root (MECHA10_FRAMEWORK_PATH or walk up to find workspace)
            let framework_root = self.find_framework_root()?;

            // Try packages/nodes/{name}/src/lib.rs
            let nodes_path = framework_root
                .join("packages/nodes")
                .join(package_name)
                .join("src/lib.rs");
            if nodes_path.exists() {
                return Ok(nodes_path);
            }

            // Try packages/drivers/{name}/src/lib.rs
            let drivers_path = framework_root
                .join("packages/drivers")
                .join(package_name)
                .join("src/lib.rs");
            if drivers_path.exists() {
                return Ok(drivers_path);
            }

            // Try packages/services/{name}/src/lib.rs
            let services_path = framework_root
                .join("packages/services")
                .join(package_name)
                .join("src/lib.rs");
            if services_path.exists() {
                return Ok(services_path);
            }

            // Not found, return the first attempt
            Ok(nodes_path)
        } else {
            // Assume it's a package name
            Ok(self.project_root.join("nodes").join(package_path).join("src/lib.rs"))
        }
    }

    /// Find the mecha10 framework root directory
    fn find_framework_root(&self) -> Result<PathBuf> {
        // First try MECHA10_FRAMEWORK_PATH environment variable
        if let Ok(framework_path) = std::env::var("MECHA10_FRAMEWORK_PATH") {
            let path = PathBuf::from(framework_path);
            if path.exists() {
                return Ok(path);
            }
        }

        // Check if we're already in the framework root (has packages/nodes directory)
        if self.project_root.join("packages/nodes").exists() {
            return Ok(self.project_root.clone());
        }

        // Walk up from project root to find workspace with packages/nodes
        let mut current = self.project_root.clone();
        loop {
            if current.join("packages/nodes").exists() {
                return Ok(current);
            }

            match current.parent() {
                Some(parent) => current = parent.to_path_buf(),
                None => break,
            }
        }

        // Fallback to project root
        Ok(self.project_root.clone())
    }

    /// Parse node source file for topic definitions and usage
    async fn parse_node_source(&self, source_path: &Path) -> Result<(Vec<TopicRef>, Vec<TopicRef>)> {
        let content = tokio::fs::read_to_string(source_path)
            .await
            .context(format!("Failed to read source file: {}", source_path.display()))?;

        // Extract topic constants and their types
        let topic_defs = self.extract_topic_definitions(&content);

        // Find publish calls
        let publishes = self.extract_publish_calls(&content, &topic_defs);

        // Find subscribe calls
        let subscribes = self.extract_subscribe_calls(&content, &topic_defs);

        Ok((publishes, subscribes))
    }

    /// Extract topic constant definitions from source
    ///
    /// Note: This method is public primarily for testing purposes.
    pub fn extract_topic_definitions(&self, content: &str) -> HashMap<String, TopicRef> {
        let mut topics = HashMap::new();

        // Match: pub const TOPIC_NAME: Topic<MessageType> = Topic::new("/path");
        let topic_pattern =
            Regex::new(r#"pub\s+const\s+([A-Z_]+):\s*Topic<([^>]+)>\s*=\s*Topic::new\("([^"]+)"\)"#).unwrap();

        for caps in topic_pattern.captures_iter(content) {
            let const_name = caps.get(1).map(|m| m.as_str()).unwrap_or("");
            let message_type = caps.get(2).map(|m| m.as_str().trim()).unwrap_or("");
            let topic_path = caps.get(3).map(|m| m.as_str()).unwrap_or("");

            topics.insert(
                const_name.to_string(),
                TopicRef {
                    path: topic_path.to_string(),
                    message_type: Some(message_type.to_string()),
                },
            );
        }

        topics
    }

    /// Extract publish_to calls
    ///
    /// Note: This method is public primarily for testing purposes.
    pub fn extract_publish_calls(&self, content: &str, topic_defs: &HashMap<String, TopicRef>) -> Vec<TopicRef> {
        let mut publishes = Vec::new();
        let mut seen = HashSet::new();

        // Match: ctx.publish_to(TOPIC_NAME, ...)
        // or: publish_to(TOPIC_NAME, ...)
        // or: ctx.publish_to(topics::TOPIC_NAME, ...)
        // Use a more flexible pattern that handles whitespace and optional module prefix
        let publish_pattern = Regex::new(r"publish_to\s*\(\s*(?:[a-z_]+::)?([A-Z_][A-Z0-9_]*)\s*,").unwrap();

        for caps in publish_pattern.captures_iter(content) {
            if let Some(const_name) = caps.get(1) {
                let const_name = const_name.as_str();
                if let Some(topic) = topic_defs.get(const_name) {
                    if seen.insert(topic.path.clone()) {
                        publishes.push(topic.clone());
                    }
                }
            }
        }

        publishes
    }

    /// Extract subscribe calls
    ///
    /// Note: This method is public primarily for testing purposes.
    pub fn extract_subscribe_calls(&self, content: &str, topic_defs: &HashMap<String, TopicRef>) -> Vec<TopicRef> {
        let mut subscribes = Vec::new();
        let mut seen = HashSet::new();

        // Match: ctx.subscribe::<MessageType>(TOPIC_NAME)
        // or: subscribe(TOPIC_NAME)
        // or: ctx.subscribe::<MessageType>(topics::TOPIC_NAME)
        // Use a more flexible pattern that handles whitespace, generics, and optional module prefix
        let subscribe_pattern =
            Regex::new(r"subscribe\s*(?:::\s*<[^>]+>\s*)?\(\s*(?:[a-z_]+::)?([A-Z_][A-Z0-9_]*)\s*\)").unwrap();

        for caps in subscribe_pattern.captures_iter(content) {
            if let Some(const_name) = caps.get(1) {
                let const_name = const_name.as_str();
                if let Some(topic) = topic_defs.get(const_name) {
                    if seen.insert(topic.path.clone()) {
                        subscribes.push(topic.clone());
                    }
                }
            }
        }

        subscribes
    }

    /// Build topic-centric view from node-centric data
    fn build_topic_view(&self, nodes: &[NodeTopology]) -> Vec<TopicTopology> {
        let mut topic_map: HashMap<String, TopicTopology> = HashMap::new();

        for node in nodes {
            // Add publishers
            for topic_ref in &node.publishes {
                let topic = topic_map
                    .entry(topic_ref.path.clone())
                    .or_insert_with(|| TopicTopology {
                        path: topic_ref.path.clone(),
                        message_type: topic_ref.message_type.clone(),
                        publishers: Vec::new(),
                        subscribers: Vec::new(),
                    });
                if !topic.publishers.contains(&node.name) {
                    topic.publishers.push(node.name.clone());
                }
            }

            // Add subscribers
            for topic_ref in &node.subscribes {
                let topic = topic_map
                    .entry(topic_ref.path.clone())
                    .or_insert_with(|| TopicTopology {
                        path: topic_ref.path.clone(),
                        message_type: topic_ref.message_type.clone(),
                        publishers: Vec::new(),
                        subscribers: Vec::new(),
                    });
                if !topic.subscribers.contains(&node.name) {
                    topic.subscribers.push(node.name.clone());
                }
            }
        }

        let mut topics: Vec<TopicTopology> = topic_map.into_values().collect();
        topics.sort_by(|a, b| a.path.cmp(&b.path));
        topics
    }

    /// Parse topics from JSON config value
    ///
    /// This is a helper method extracted for testing. It contains the core parsing logic
    /// from `load_topics_from_config` but operates on a JSON value instead of a file path.
    fn parse_topics_from_json(&self, config: &serde_json::Value) -> (Vec<TopicRef>, Vec<TopicRef>) {
        let mut publishes = Vec::new();
        let mut subscribes = Vec::new();

        // Try Format 1: Array format (topics.publishes/topics.subscribes)
        if let Some(topics) = config.get("topics").and_then(|v| v.as_object()) {
            // Check if using array format
            if topics.contains_key("publishes") || topics.contains_key("subscribes") {
                // Format 1: Array format
                if let Some(pubs) = topics.get("publishes").and_then(|v| v.as_array()) {
                    for topic in pubs {
                        if let Some(topic_obj) = topic.as_object() {
                            for (_semantic_name, topic_path) in topic_obj {
                                if let Some(path_str) = topic_path.as_str() {
                                    publishes.push(TopicRef {
                                        path: path_str.to_string(),
                                        message_type: None,
                                    });
                                }
                            }
                        }
                    }
                }

                if let Some(subs) = topics.get("subscribes").and_then(|v| v.as_array()) {
                    for topic in subs {
                        if let Some(topic_obj) = topic.as_object() {
                            for (_semantic_name, topic_path) in topic_obj {
                                if let Some(path_str) = topic_path.as_str() {
                                    subscribes.push(TopicRef {
                                        path: path_str.to_string(),
                                        message_type: None,
                                    });
                                }
                            }
                        }
                    }
                }
            } else {
                // Format 2: Flat format - infer pub/sub from field names
                for (field_name, topic_path) in topics {
                    if let Some(path_str) = topic_path.as_str() {
                        // Classify based on field name suffix
                        if field_name.ends_with("_in") || field_name.ends_with("_sub") || field_name.contains("input") {
                            subscribes.push(TopicRef {
                                path: path_str.to_string(),
                                message_type: None,
                            });
                        } else if field_name.ends_with("_out")
                            || field_name.ends_with("_pub")
                            || field_name.contains("output")
                        {
                            publishes.push(TopicRef {
                                path: path_str.to_string(),
                                message_type: None,
                            });
                        }
                        // If field name doesn't match patterns, skip it (might be control/status)
                    }
                }
            }
        }

        // Try Format 3: Root-level topic fields (fallback)
        if publishes.is_empty() && subscribes.is_empty() {
            // Check for input_topic, output_topic, control_topic pattern
            if let Some(input) = config.get("input_topic").and_then(|v| v.as_str()) {
                subscribes.push(TopicRef {
                    path: input.to_string(),
                    message_type: None,
                });
            }
            if let Some(output) = config.get("output_topic").and_then(|v| v.as_str()) {
                publishes.push(TopicRef {
                    path: output.to_string(),
                    message_type: None,
                });
            }
            if let Some(control) = config.get("control_topic").and_then(|v| v.as_str()) {
                subscribes.push(TopicRef {
                    path: control.to_string(),
                    message_type: None,
                });
            }
        }

        (publishes, subscribes)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::PathBuf;

    fn create_test_service() -> TopologyService {
        TopologyService::new(PathBuf::from("/tmp/test"))
    }

    #[test]
    fn test_parse_array_format() {
        let service = create_test_service();
        let config: serde_json::Value = serde_json::json!({
            "topics": {
                "publishes": [
                    { "output": "/vision/classification" },
                    { "status": "/motor/status" }
                ],
                "subscribes": [
                    { "input": "/camera/rgb" }
                ]
            }
        });

        let (publishes, subscribes) = service.parse_topics_from_json(&config);

        assert_eq!(publishes.len(), 2);
        assert_eq!(subscribes.len(), 1);
        assert_eq!(publishes[0].path, "/vision/classification");
        assert_eq!(publishes[1].path, "/motor/status");
        assert_eq!(subscribes[0].path, "/camera/rgb");
    }

    #[test]
    fn test_parse_flat_format() {
        let service = create_test_service();
        let config: serde_json::Value = serde_json::json!({
            "topics": {
                "command_in": "/ai/command",
                "response_out": "/ai/response",
                "camera_in": "/camera/rgb",
                "nav_goal_out": "/nav/goal",
                "motor_cmd_out": "/motor/cmd_vel",
                "behavior_out": "/behavior/execute"
            }
        });

        let (publishes, subscribes) = service.parse_topics_from_json(&config);

        // Should have 4 publishes (_out suffix)
        assert_eq!(publishes.len(), 4);
        assert!(publishes.iter().any(|t| t.path == "/ai/response"));
        assert!(publishes.iter().any(|t| t.path == "/nav/goal"));
        assert!(publishes.iter().any(|t| t.path == "/motor/cmd_vel"));
        assert!(publishes.iter().any(|t| t.path == "/behavior/execute"));

        // Should have 2 subscribes (_in suffix)
        assert_eq!(subscribes.len(), 2);
        assert!(subscribes.iter().any(|t| t.path == "/ai/command"));
        assert!(subscribes.iter().any(|t| t.path == "/camera/rgb"));
    }

    #[test]
    fn test_parse_root_level_format() {
        let service = create_test_service();
        let config: serde_json::Value = serde_json::json!({
            "input_topic": "/robot/sensors/camera/rgb",
            "output_topic": "/inference/detections",
            "control_topic": "/inference/cmd"
        });

        let (publishes, subscribes) = service.parse_topics_from_json(&config);

        assert_eq!(publishes.len(), 1);
        assert_eq!(subscribes.len(), 2);
        assert_eq!(publishes[0].path, "/inference/detections");
        assert!(subscribes.iter().any(|t| t.path == "/robot/sensors/camera/rgb"));
        assert!(subscribes.iter().any(|t| t.path == "/inference/cmd"));
    }
}