mecha10_core/

topic_registry.rs

//! Topic Registry for Runtime Introspection and Validation
//!
//! Provides a central registry for tracking all active topics, their publishers,
//! subscribers, and message types. Enables runtime discovery, monitoring, and
//! debugging of the pub/sub system.
//!
//! # Use Cases
//!
//! - Discover what topics exist at runtime
//! - Find which nodes publish/subscribe to a topic
//! - Validate topic usage (orphaned topics, missing publishers/subscribers)
//! - Monitor topic activity (message rates, last activity)
//! - Debug communication issues
//! - Generate system topology diagrams
//!
//! # Example
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::topic_registry::{TopicRegistry, TopicInfo};
//!
//! # async fn example() -> Result<()> {
//! let registry = TopicRegistry::new();
//!
//! // Register a publisher
//! registry.register_publisher("/sensor/camera/rgb", "camera_node", "Image").await;
//!
//! // Register a subscriber
//! registry.register_subscriber("/sensor/camera/rgb", "vision_node", "Image").await;
//!
//! // Query topic info
//! if let Some(info) = registry.get_topic("/sensor/camera/rgb").await {
//!     println!("Topic: {}", info.topic);
//!     println!("Publishers: {:?}", info.publishers);
//!     println!("Subscribers: {:?}", info.subscribers);
//!     println!("Message type: {}", info.message_type);
//! }
//!
//! // List all topics
//! let topics = registry.list_topics().await;
//! for topic in topics {
//!     println!("- {}", topic);
//! }
//!
//! // Validate topology
//! let issues = registry.validate().await;
//! for issue in issues {
//!     println!("Warning: {}", issue);
//! }
//! # Ok(())
//! # }
//! ```

use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, info, warn};

// ============================================================================
// Topic Information
// ============================================================================

/// Information about a topic
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TopicInfo {
    /// Topic path (e.g., "/sensor/camera/rgb")
    pub topic: String,

    /// Message type name (e.g., "Image", "LaserScan")
    pub message_type: String,

    /// Nodes that publish to this topic
    pub publishers: Vec<String>,

    /// Nodes that subscribe to this topic
    pub subscribers: Vec<String>,

    /// When the topic was first registered
    pub created_at: u64,

    /// Last time a publisher was added
    pub last_publisher_added: Option<u64>,

    /// Last time a subscriber was added
    pub last_subscriber_added: Option<u64>,

    /// Optional description
    pub description: Option<String>,

    /// Custom metadata
    pub metadata: HashMap<String, String>,
}

impl TopicInfo {
    /// Create new topic info
    pub fn new(topic: &str, message_type: &str) -> Self {
        Self {
            topic: topic.to_string(),
            message_type: message_type.to_string(),
            publishers: Vec::new(),
            subscribers: Vec::new(),
            created_at: now_micros(),
            last_publisher_added: None,
            last_subscriber_added: None,
            description: None,
            metadata: HashMap::new(),
        }
    }

    /// Add a publisher
    pub fn add_publisher(&mut self, node_id: &str) {
        if !self.publishers.contains(&node_id.to_string()) {
            self.publishers.push(node_id.to_string());
            self.last_publisher_added = Some(now_micros());
        }
    }

    /// Add a subscriber
    pub fn add_subscriber(&mut self, node_id: &str) {
        if !self.subscribers.contains(&node_id.to_string()) {
            self.subscribers.push(node_id.to_string());
            self.last_subscriber_added = Some(now_micros());
        }
    }

    /// Remove a publisher
    pub fn remove_publisher(&mut self, node_id: &str) {
        self.publishers.retain(|p| p != node_id);
    }

    /// Remove a subscriber
    pub fn remove_subscriber(&mut self, node_id: &str) {
        self.subscribers.retain(|s| s != node_id);
    }

    /// Check if topic has any publishers
    pub fn has_publishers(&self) -> bool {
        !self.publishers.is_empty()
    }

    /// Check if topic has any subscribers
    pub fn has_subscribers(&self) -> bool {
        !self.subscribers.is_empty()
    }

    /// Check if topic is orphaned (no publishers or no subscribers)
    pub fn is_orphaned(&self) -> bool {
        self.publishers.is_empty() || self.subscribers.is_empty()
    }

    /// Get the number of publishers
    pub fn publisher_count(&self) -> usize {
        self.publishers.len()
    }

    /// Get the number of subscribers
    pub fn subscriber_count(&self) -> usize {
        self.subscribers.len()
    }
}

// ============================================================================
// Topic Registry
// ============================================================================

/// Central registry for tracking topics at runtime
///
/// Thread-safe registry that maintains information about all active topics,
/// their publishers, and subscribers.
#[derive(Clone)]
pub struct TopicRegistry {
    /// Topic information indexed by topic path
    topics: Arc<RwLock<HashMap<String, TopicInfo>>>,

    /// Node to published topics mapping
    node_publishers: Arc<RwLock<HashMap<String, HashSet<String>>>>,

    /// Node to subscribed topics mapping
    node_subscribers: Arc<RwLock<HashMap<String, HashSet<String>>>>,
}

impl TopicRegistry {
    /// Create a new topic registry
    pub fn new() -> Self {
        info!("Creating new topic registry");
        Self {
            topics: Arc::new(RwLock::new(HashMap::new())),
            node_publishers: Arc::new(RwLock::new(HashMap::new())),
            node_subscribers: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// Register a publisher for a topic
    ///
    /// # Arguments
    ///
    /// * `topic` - Topic path
    /// * `node_id` - Publishing node identifier
    /// * `message_type` - Message type name
    pub async fn register_publisher(&self, topic: &str, node_id: &str, message_type: &str) {
        let mut topics = self.topics.write().await;

        let topic_info = topics
            .entry(topic.to_string())
            .or_insert_with(|| TopicInfo::new(topic, message_type));

        topic_info.add_publisher(node_id);

        // Update node publishers mapping
        let mut node_pubs = self.node_publishers.write().await;
        node_pubs
            .entry(node_id.to_string())
            .or_default()
            .insert(topic.to_string());

        debug!(
            "Registered publisher: node='{}', topic='{}', type='{}'",
            node_id, topic, message_type
        );
    }

    /// Register a subscriber for a topic
    ///
    /// # Arguments
    ///
    /// * `topic` - Topic path
    /// * `node_id` - Subscribing node identifier
    /// * `message_type` - Message type name
    pub async fn register_subscriber(&self, topic: &str, node_id: &str, message_type: &str) {
        let mut topics = self.topics.write().await;

        let topic_info = topics
            .entry(topic.to_string())
            .or_insert_with(|| TopicInfo::new(topic, message_type));

        topic_info.add_subscriber(node_id);

        // Update node subscribers mapping
        let mut node_subs = self.node_subscribers.write().await;
        node_subs
            .entry(node_id.to_string())
            .or_default()
            .insert(topic.to_string());

        debug!(
            "Registered subscriber: node='{}', topic='{}', type='{}'",
            node_id, topic, message_type
        );
    }

    /// Unregister a publisher
    pub async fn unregister_publisher(&self, topic: &str, node_id: &str) {
        let mut topics = self.topics.write().await;

        if let Some(info) = topics.get_mut(topic) {
            info.remove_publisher(node_id);

            // Clean up topic if no publishers or subscribers remain
            if info.publishers.is_empty() && info.subscribers.is_empty() {
                topics.remove(topic);
                debug!("Removed topic '{}' (no publishers or subscribers)", topic);
            }
        }

        // Update node publishers mapping
        let mut node_pubs = self.node_publishers.write().await;
        if let Some(topics_set) = node_pubs.get_mut(node_id) {
            topics_set.remove(topic);
            if topics_set.is_empty() {
                node_pubs.remove(node_id);
            }
        }

        debug!("Unregistered publisher: node='{}', topic='{}'", node_id, topic);
    }

    /// Unregister a subscriber
    pub async fn unregister_subscriber(&self, topic: &str, node_id: &str) {
        let mut topics = self.topics.write().await;

        if let Some(info) = topics.get_mut(topic) {
            info.remove_subscriber(node_id);

            // Clean up topic if no publishers or subscribers remain
            if info.publishers.is_empty() && info.subscribers.is_empty() {
                topics.remove(topic);
                debug!("Removed topic '{}' (no publishers or subscribers)", topic);
            }
        }

        // Update node subscribers mapping
        let mut node_subs = self.node_subscribers.write().await;
        if let Some(topics_set) = node_subs.get_mut(node_id) {
            topics_set.remove(topic);
            if topics_set.is_empty() {
                node_subs.remove(node_id);
            }
        }

        debug!("Unregistered subscriber: node='{}', topic='{}'", node_id, topic);
    }

    /// Unregister all topics for a node (when node stops)
    pub async fn unregister_node(&self, node_id: &str) {
        // Unregister as publisher
        let topics_published = {
            let node_pubs = self.node_publishers.read().await;
            node_pubs.get(node_id).cloned().unwrap_or_default()
        };

        for topic in topics_published {
            self.unregister_publisher(&topic, node_id).await;
        }

        // Unregister as subscriber
        let topics_subscribed = {
            let node_subs = self.node_subscribers.read().await;
            node_subs.get(node_id).cloned().unwrap_or_default()
        };

        for topic in topics_subscribed {
            self.unregister_subscriber(&topic, node_id).await;
        }

        info!("Unregistered all topics for node '{}'", node_id);
    }

    /// Get information about a specific topic
    pub async fn get_topic(&self, topic: &str) -> Option<TopicInfo> {
        let topics = self.topics.read().await;
        topics.get(topic).cloned()
    }

    /// List all registered topics
    pub async fn list_topics(&self) -> Vec<String> {
        let topics = self.topics.read().await;
        topics.keys().cloned().collect()
    }

    /// Get all topic information
    pub async fn get_all_topics(&self) -> Vec<TopicInfo> {
        let topics = self.topics.read().await;
        topics.values().cloned().collect()
    }

    /// Find topics by message type
    pub async fn find_by_message_type(&self, message_type: &str) -> Vec<TopicInfo> {
        let topics = self.topics.read().await;
        topics
            .values()
            .filter(|info| info.message_type == message_type)
            .cloned()
            .collect()
    }

    /// Find topics published by a node
    pub async fn topics_published_by(&self, node_id: &str) -> Vec<String> {
        let node_pubs = self.node_publishers.read().await;
        node_pubs
            .get(node_id)
            .cloned()
            .unwrap_or_default()
            .into_iter()
            .collect()
    }

    /// Find topics subscribed to by a node
    pub async fn topics_subscribed_by(&self, node_id: &str) -> Vec<String> {
        let node_subs = self.node_subscribers.read().await;
        node_subs
            .get(node_id)
            .cloned()
            .unwrap_or_default()
            .into_iter()
            .collect()
    }

    /// Get total number of topics
    pub async fn topic_count(&self) -> usize {
        let topics = self.topics.read().await;
        topics.len()
    }

    /// Validate the topic topology
    ///
    /// Returns a list of validation issues:
    /// - Topics with no publishers
    /// - Topics with no subscribers
    /// - Type mismatches
    pub async fn validate(&self) -> Vec<String> {
        let mut issues = Vec::new();
        let topics = self.topics.read().await;

        for (topic, info) in topics.iter() {
            if info.publishers.is_empty() {
                issues.push(format!(
                    "Topic '{}' has no publishers (subscribed by: {:?})",
                    topic, info.subscribers
                ));
            }

            if info.subscribers.is_empty() {
                issues.push(format!(
                    "Topic '{}' has no subscribers (published by: {:?})",
                    topic, info.publishers
                ));
            }
        }

        if !issues.is_empty() {
            warn!("Found {} topology issues", issues.len());
        }

        issues
    }

    /// Get topology statistics
    pub async fn stats(&self) -> TopologyStats {
        let topics = self.topics.read().await;

        let total_topics = topics.len();
        let total_publishers: usize = topics.values().map(|t| t.publishers.len()).sum();
        let total_subscribers: usize = topics.values().map(|t| t.subscribers.len()).sum();

        let orphaned_topics = topics.values().filter(|t| t.is_orphaned()).count();

        let topics_without_publishers = topics.values().filter(|t| t.publishers.is_empty()).count();

        let topics_without_subscribers = topics.values().filter(|t| t.subscribers.is_empty()).count();

        TopologyStats {
            total_topics,
            total_publishers,
            total_subscribers,
            orphaned_topics,
            topics_without_publishers,
            topics_without_subscribers,
        }
    }

    /// Export topology as DOT graph (for Graphviz visualization)
    pub async fn export_dot(&self) -> String {
        let mut dot = String::from("digraph TopicTopology {\n");
        dot.push_str("  rankdir=LR;\n");
        dot.push_str("  node [shape=box];\n\n");

        let topics = self.topics.read().await;

        // Add topic nodes
        for (topic, info) in topics.iter() {
            let label = format!("{}\\n({})", topic, info.message_type);
            dot.push_str(&format!(
                "  \"{}\" [label=\"{}\", shape=ellipse, style=filled, fillcolor=lightblue];\n",
                topic, label
            ));
        }

        dot.push('\n');

        // Add publisher edges
        for (topic, info) in topics.iter() {
            for pub_node in &info.publishers {
                dot.push_str(&format!(
                    "  \"{}\" -> \"{}\" [label=\"pub\", color=green];\n",
                    pub_node, topic
                ));
            }

            for sub_node in &info.subscribers {
                dot.push_str(&format!(
                    "  \"{}\" -> \"{}\" [label=\"sub\", color=blue];\n",
                    topic, sub_node
                ));
            }
        }

        dot.push_str("}\n");
        dot
    }

    /// Clear all registered topics (for testing)
    pub async fn clear(&self) {
        let mut topics = self.topics.write().await;
        topics.clear();

        let mut node_pubs = self.node_publishers.write().await;
        node_pubs.clear();

        let mut node_subs = self.node_subscribers.write().await;
        node_subs.clear();

        info!("Cleared topic registry");
    }
}

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

// ============================================================================
// Topology Statistics
// ============================================================================

/// Statistics about the topic topology
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TopologyStats {
    /// Total number of topics
    pub total_topics: usize,

    /// Total number of publisher connections
    pub total_publishers: usize,

    /// Total number of subscriber connections
    pub total_subscribers: usize,

    /// Number of orphaned topics (no publishers OR no subscribers)
    pub orphaned_topics: usize,

    /// Topics with no publishers
    pub topics_without_publishers: usize,

    /// Topics with no subscribers
    pub topics_without_subscribers: usize,
}

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

fn now_micros() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_micros() as u64
}