mecha10-core 0.6.2

//! Topic utility functions
//!
//! Provides common utilities for working with topics including:
//! - Topic path parsing and categorization
//! - Time formatting for topic metadata
//! - Topic metadata extraction from Redis streams
//! - Per-role publish/subscribe topic configuration for node configs

use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Extract category from topic path (first segment)
///
/// # Examples
/// ```
/// use mecha10_core::topic_utils::extract_category;
///
/// assert_eq!(extract_category("/sensor/camera"), "/sensor");
/// assert_eq!(extract_category("/actuator/motor/left"), "/actuator");
/// assert_eq!(extract_category("unknown"), "uncategorized");
/// ```
pub fn extract_category(topic: &str) -> String {
    let parts: Vec<&str> = topic.trim_start_matches('/').split('/').collect();
    if parts.is_empty() || parts[0].is_empty() {
        "uncategorized".to_string()
    } else {
        format!("/{}", parts[0])
    }
}

/// Format elapsed time in human-readable format
///
/// # Examples
/// ```
/// use mecha10_core::topic_utils::format_elapsed_time;
///
/// assert_eq!(format_elapsed_time(500), "500ms ago");
/// assert_eq!(format_elapsed_time(5000), "5s ago");
/// assert_eq!(format_elapsed_time(65000), "1m ago");
/// ```
pub fn format_elapsed_time(elapsed_ms: u64) -> String {
    if elapsed_ms < 1000 {
        format!("{}ms ago", elapsed_ms)
    } else if elapsed_ms < 60_000 {
        format!("{}s ago", elapsed_ms / 1000)
    } else if elapsed_ms < 3_600_000 {
        format!("{}m ago", elapsed_ms / 60_000)
    } else if elapsed_ms < 86_400_000 {
        format!("{}h ago", elapsed_ms / 3_600_000)
    } else {
        format!("{}d ago", elapsed_ms / 86_400_000)
    }
}

/// Topic metadata extracted from Redis stream info
#[derive(Debug, Clone)]
pub struct TopicMetadata {
    /// Topic path
    pub topic: String,
    /// Number of messages in the stream
    pub message_count: Option<usize>,
    /// Timestamp of last message (milliseconds since epoch)
    pub last_update_ms: Option<u64>,
    /// Timestamp of first message (milliseconds since epoch)
    pub first_timestamp_ms: Option<u64>,
    /// Timestamp of last message (milliseconds since epoch)
    pub last_timestamp_ms: Option<u64>,
}

impl TopicMetadata {
    /// Create new topic metadata
    pub fn new(topic: String) -> Self {
        Self {
            topic,
            message_count: None,
            last_update_ms: None,
            first_timestamp_ms: None,
            last_timestamp_ms: None,
        }
    }

    /// Set message count
    pub fn with_message_count(mut self, count: usize) -> Self {
        self.message_count = Some(count);
        self
    }

    /// Set last update timestamp
    pub fn with_last_update(mut self, timestamp_ms: u64) -> Self {
        self.last_update_ms = Some(timestamp_ms);
        self
    }

    /// Set timestamp range
    pub fn with_timestamp_range(mut self, first_ms: u64, last_ms: u64) -> Self {
        self.first_timestamp_ms = Some(first_ms);
        self.last_timestamp_ms = Some(last_ms);
        self
    }

    /// Calculate message rate (messages per second)
    pub fn calculate_rate(&self) -> Option<f64> {
        if let (Some(first_ms), Some(last_ms), Some(count)) =
            (self.first_timestamp_ms, self.last_timestamp_ms, self.message_count)
        {
            let duration_ms = last_ms.saturating_sub(first_ms);
            if duration_ms > 1000 && count > 0 {
                return Some((count as f64 * 1000.0) / duration_ms as f64);
            }
        }
        None
    }

    /// Get topic category
    pub fn category(&self) -> String {
        extract_category(&self.topic)
    }
}

/// Per-role publish/subscribe topic configuration shared by node config structs.
///
/// Node configs declare their topology as lists of role -> path maps so operators can
/// override individual topics from config without touching code, e.g.:
///
/// ```toml
/// [topics]
/// subscribes = [{ input = "/camera/rgb" }, { control = "/vision/object/control" }]
/// publishes = [{ output = "/vision/object/detections" }]
/// ```
///
/// The default here is intentionally empty - each node's own config `Default` impl
/// supplies its own per-role default topic paths, and the `*_topic()` accessor methods
/// fall back to those hardcoded defaults when a role is absent from `subscribes`/`publishes`.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct Topics {
    /// Topics this node publishes to, keyed by role (e.g. "output", "status")
    #[serde(default)]
    pub publishes: Vec<HashMap<String, String>>,

    /// Topics this node subscribes to, keyed by role (e.g. "input", "control")
    #[serde(default)]
    pub subscribes: Vec<HashMap<String, String>>,
}

impl Topics {
    /// Look up a subscribed topic by role, falling back to `default` if not configured
    pub fn subscribed_topic(&self, role: &str, default: &str) -> String {
        Self::lookup(&self.subscribes, role, default)
    }

    /// Look up a published topic by role, falling back to `default` if not configured
    pub fn published_topic(&self, role: &str, default: &str) -> String {
        Self::lookup(&self.publishes, role, default)
    }

    fn lookup(topics: &[HashMap<String, String>], role: &str, default: &str) -> String {
        topics
            .iter()
            .find_map(|t| t.get(role))
            .cloned()
            .unwrap_or_else(|| default.to_string())
    }
}