mecha10_core/

health.rs

//! Centralized Health Monitoring
//!
//! Provides centralized health monitoring and aggregation for all nodes in a Mecha10 system.
//!
//! # Features
//!
//! - Automatic health reporting from nodes
//! - Centralized health aggregation in Redis
//! - System-wide health status
//! - Health history storage (24 hours, using Redis sorted sets)
//! - Health history queries by time range
//! - Alerting on health degradation
//! - Dashboard/CLI integration
//!
//! # Architecture
//!
//! ```text
//! ┌─────────┐     Health      ┌───────────────┐
//! │  Node A │────Status──────▶│               │
//! └─────────┘                 │               │
//!                             │  Health       │     ┌──────────┐
//! ┌─────────┐     Health      │  Monitor      │────▶│Dashboard │
//! │  Node B │────Status──────▶│  (Redis)      │     └──────────┘
//! └─────────┘                 │               │
//!                             │               │     ┌──────────┐
//! ┌─────────┐     Health      │               │────▶│Alerting  │
//! │  Node C │────Status──────▶│               │     └──────────┘
//! └─────────┘                 └───────────────┘
//! ```
//!
//! # Example
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::health::{HealthMonitor, HealthReporter};
//!
//! # async fn example() -> Result<()> {
//! // In your node
//! let ctx = Context::new("my-node").await?;
//! let reporter = HealthReporter::new(&ctx).await?;
//!
//! // Report health periodically
//! let mut interval = interval_at_hz(1.0); // 1 Hz
//! loop {
//!     interval.tick().await;
//!
//!     let status = HealthStatus::healthy()
//!         .with_diagnostic("cpu_usage", "25%")
//!         .with_diagnostic("memory_mb", "128");
//!
//!     reporter.report(status).await?;
//! }
//!
//! // Monitor system health
//! let monitor = HealthMonitor::connect("redis://localhost:6379").await?;
//!
//! // Get overall system health
//! let system_health = monitor.system_health().await?;
//! println!("System: {:?}", system_health.level);
//!
//! // Get health by node
//! let node_health = monitor.node_health("my-node").await?;
//!
//! // List unhealthy nodes
//! let unhealthy = monitor.unhealthy_nodes().await?;
//! # Ok(())
//! # }
//! ```

use crate::context::Context;
use crate::error::{Mecha10Error, Result};
use crate::messages::{HealthLevel, HealthStatus, Message};
use crate::node::HealthPriority;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;

// ============================================================================
// Health Reporter (Node Side)
// ============================================================================

/// Health reporter for nodes to publish their health status
///
/// Each node should create a HealthReporter and use it to periodically
/// report health status to the centralized monitoring system.
///
/// # Example
///
/// ```rust
/// use mecha10::prelude::*;
/// use mecha10::health::HealthReporter;
///
/// # async fn example(ctx: &Context) -> Result<()> {
/// let reporter = HealthReporter::new(ctx).await?;
///
/// // Report healthy status
/// reporter.report(HealthStatus::healthy()).await?;
///
/// // Report degraded status
/// reporter.report(
///     HealthStatus::degraded("High CPU usage")
///         .with_diagnostic("cpu", "95%")
/// ).await?;
/// # Ok(())
/// # }
/// ```
pub struct HealthReporter {
    node_id: String,
    redis_client: redis::Client,
    ctx: Arc<Context>,
    /// Health priority level
    priority: HealthPriority,
    /// TTL for Redis health key in seconds
    ttl_seconds: u64,
    /// Whether to store health history
    store_history: bool,
}

impl HealthReporter {
    /// Create a new health reporter with default settings (Normal priority)
    pub async fn new(ctx: &Context) -> Result<Self> {
        Self::with_config(ctx, HealthPriority::Normal, 60, true).await
    }

    /// Create a new health reporter with custom configuration
    ///
    /// # Arguments
    ///
    /// * `ctx` - Node context
    /// * `priority` - Health priority level (affects timeout calculations)
    /// * `ttl_seconds` - TTL for the Redis health key
    /// * `store_history` - Whether to store health history in sorted sets
    pub async fn with_config(
        ctx: &Context,
        priority: HealthPriority,
        ttl_seconds: u64,
        store_history: bool,
    ) -> Result<Self> {
        let redis_url = Self::get_redis_url()?;
        let redis_client = redis::Client::open(redis_url.as_str())
            .map_err(|e| Mecha10Error::Other(format!("Failed to connect to Redis for health reporting: {}", e)))?;

        Ok(Self {
            node_id: ctx.node_id().to_string(),
            redis_client,
            ctx: Arc::new(ctx.clone()),
            priority,
            ttl_seconds,
            store_history,
        })
    }

    /// Report health status
    ///
    /// Stores the health status in Redis with a TTL. If the node stops reporting,
    /// the health status will expire and the node will be marked as unavailable.
    pub async fn report(&self, status: HealthStatus) -> Result<()> {
        use redis::AsyncCommands;

        let mut conn = self
            .redis_client
            .get_multiplexed_async_connection()
            .await
            .map_err(|e| Mecha10Error::Other(format!("Failed to get Redis connection: {}", e)))?;

        // Create health report with priority
        let report = HealthReport {
            node_id: self.node_id.clone(),
            status,
            priority: self.priority,
            reported_at: crate::prelude::now_micros(),
        };

        // Store in Redis with configured TTL
        let key = format!("mecha10:health:{}", self.node_id);
        let value = serde_json::to_string(&report)
            .map_err(|e| Mecha10Error::Other(format!("Failed to serialize health report: {}", e)))?;

        conn.set_ex::<_, _, ()>(&key, value.clone(), self.ttl_seconds)
            .await
            .map_err(|e| Mecha10Error::Other(format!("Failed to store health status: {}", e)))?;

        // Store in health history if enabled
        if self.store_history {
            let history_key = format!("mecha10:health:history:{}", self.node_id);
            let score = report.reported_at as f64; // Use timestamp as score for sorting
            conn.zadd::<_, _, _, ()>(&history_key, &value, score)
                .await
                .map_err(|e| Mecha10Error::Other(format!("Failed to store health history: {}", e)))?;

            // Keep only last 24 hours of history (86400 seconds)
            let cutoff_time = (report.reported_at - 86_400_000_000) as f64; // 24 hours ago in microseconds
            conn.zrembyscore::<_, _, _, ()>(&history_key, f64::MIN, cutoff_time)
                .await
                .map_err(|e| Mecha10Error::Other(format!("Failed to clean old health history: {}", e)))?;
        }

        // Publish to health topic for real-time monitoring (includes priority)
        self.ctx.publish_raw("/system/health", &report).await?;

        Ok(())
    }

    /// Report healthy status
    pub async fn report_healthy(&self) -> Result<()> {
        self.report(HealthStatus::healthy()).await
    }

    /// Report degraded status
    pub async fn report_degraded(&self, message: impl Into<String>) -> Result<()> {
        self.report(HealthStatus::degraded(message)).await
    }

    /// Report unhealthy status
    pub async fn report_unhealthy(&self, message: impl Into<String>) -> Result<()> {
        self.report(HealthStatus::unhealthy(message)).await
    }

    fn get_redis_url() -> Result<String> {
        // Use Context's get_redis_url() for consistency - it checks env vars and config properly
        let url = Context::get_redis_url()?;
        tracing::debug!("HealthReporter using Redis URL: {}", url);
        Ok(url)
    }
}

// ============================================================================
// Health Monitor (Centralized)
// ============================================================================

/// Health report stored in Redis
///
/// Contains the node's health status along with priority information
/// for dashboard display and monitoring timeout calculations.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthReport {
    /// Node identifier
    pub node_id: String,
    /// Health status at the time of report
    pub status: HealthStatus,
    /// Node priority level (affects monitoring timeout)
    #[serde(default)]
    pub priority: HealthPriority,
    /// Timestamp when the report was created (microseconds since epoch)
    pub reported_at: u64,
}

// Implement Message trait so HealthReport can be published via pub/sub
impl Message for HealthReport {}

/// System-wide health summary
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SystemHealth {
    /// Overall system health level
    pub level: HealthLevel,

    /// Total number of nodes
    pub total_nodes: usize,

    /// Number of healthy nodes
    pub healthy_nodes: usize,

    /// Number of degraded nodes
    pub degraded_nodes: usize,

    /// Number of unhealthy nodes
    pub unhealthy_nodes: usize,

    /// Number of unresponsive nodes (haven't reported recently)
    pub unresponsive_nodes: usize,

    /// Timestamp when this summary was generated
    pub timestamp: u64,

    /// Details by node
    pub nodes: HashMap<String, NodeHealthInfo>,
}

/// Health information for a single node
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeHealthInfo {
    /// Node ID
    pub node_id: String,

    /// Current health status
    pub status: HealthStatus,

    /// Node priority level
    #[serde(default)]
    pub priority: HealthPriority,

    /// When the status was last reported
    pub reported_at: u64,

    /// How long ago the status was reported (microseconds)
    pub age_micros: u64,

    /// Whether the node is responsive (reported recently)
    pub responsive: bool,
}

/// Centralized health monitor
///
/// Aggregates health status from all nodes and provides system-wide health information.
///
/// # Example
///
/// ```rust
/// use mecha10::health::HealthMonitor;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// let monitor = HealthMonitor::connect("redis://localhost:6379").await?;
///
/// // Get system-wide health
/// let health = monitor.system_health().await?;
/// println!("System health: {:?}", health.level);
/// println!("Healthy: {}/{}", health.healthy_nodes, health.total_nodes);
///
/// // List unhealthy nodes
/// for node in monitor.unhealthy_nodes().await? {
///     println!("Unhealthy: {} - {}", node.node_id, node.status.message);
/// }
/// # Ok(())
/// # }
/// ```
pub struct HealthMonitor {
    redis_client: redis::Client,
    cache: Arc<RwLock<Option<(SystemHealth, std::time::Instant)>>>,
    cache_ttl: Duration,
}

impl HealthMonitor {
    /// Connect to Redis for health monitoring
    pub async fn connect(redis_url: &str) -> Result<Self> {
        let redis_client = redis::Client::open(redis_url)
            .map_err(|e| Mecha10Error::Other(format!("Failed to connect to Redis: {}", e)))?;

        Ok(Self {
            redis_client,
            cache: Arc::new(RwLock::new(None)),
            cache_ttl: Duration::from_secs(5), // Cache for 5 seconds
        })
    }

    /// Get system-wide health summary
    pub async fn system_health(&self) -> Result<SystemHealth> {
        // Check cache first
        {
            let cache = self.cache.read().await;
            if let Some((health, cached_at)) = cache.as_ref() {
                if cached_at.elapsed() < self.cache_ttl {
                    return Ok(health.clone());
                }
            }
        }

        // Fetch fresh data
        let health = self.fetch_system_health().await?;

        // Update cache
        {
            let mut cache = self.cache.write().await;
            *cache = Some((health.clone(), std::time::Instant::now()));
        }

        Ok(health)
    }

    async fn fetch_system_health(&self) -> Result<SystemHealth> {
        use redis::AsyncCommands;

        let mut conn = self
            .redis_client
            .get_multiplexed_async_connection()
            .await
            .map_err(|e| Mecha10Error::Other(format!("Failed to get Redis connection: {}", e)))?;

        // Get all health keys (filter out history keys)
        let pattern = "mecha10:health:*";
        let keys: Vec<String> = conn
            .keys(pattern)
            .await
            .map_err(|e| Mecha10Error::Other(format!("Failed to list health keys: {}", e)))?;

        // Filter out history keys
        let keys: Vec<_> = keys.into_iter().filter(|k| !k.contains(":history:")).collect();

        let mut nodes = HashMap::new();
        let mut healthy_count = 0;
        let mut degraded_count = 0;
        let mut unhealthy_count = 0;
        let mut unresponsive_count = 0;

        let now = crate::prelude::now_micros();

        for key in keys {
            let value: Option<String> = conn
                .get(&key)
                .await
                .map_err(|e| Mecha10Error::Other(format!("Failed to get health value: {}", e)))?;

            if let Some(value) = value {
                if let Ok(report) = serde_json::from_str::<HealthReport>(&value) {
                    let age_micros = now.saturating_sub(report.reported_at);
                    // Use priority-based timeout threshold
                    let timeout_micros = report.priority.timeout_micros();
                    let responsive = age_micros < timeout_micros;

                    if !responsive {
                        unresponsive_count += 1;
                    } else {
                        match report.status.level {
                            HealthLevel::Ok => healthy_count += 1,
                            HealthLevel::Degraded => degraded_count += 1,
                            HealthLevel::Error => unhealthy_count += 1,
                            HealthLevel::Unknown => unresponsive_count += 1,
                        }
                    }

                    nodes.insert(
                        report.node_id.clone(),
                        NodeHealthInfo {
                            node_id: report.node_id,
                            status: report.status,
                            priority: report.priority,
                            reported_at: report.reported_at,
                            age_micros,
                            responsive,
                        },
                    );
                }
            }
        }

        // Determine overall system health
        let level = if unhealthy_count > 0 || unresponsive_count > nodes.len() / 2 {
            HealthLevel::Error
        } else if degraded_count > 0 || unresponsive_count > 0 {
            HealthLevel::Degraded
        } else if healthy_count > 0 {
            HealthLevel::Ok
        } else {
            HealthLevel::Unknown
        };

        Ok(SystemHealth {
            level,
            total_nodes: nodes.len(),
            healthy_nodes: healthy_count,
            degraded_nodes: degraded_count,
            unhealthy_nodes: unhealthy_count,
            unresponsive_nodes: unresponsive_count,
            timestamp: now,
            nodes,
        })
    }

    /// Get health status for a specific node
    pub async fn node_health(&self, node_id: &str) -> Result<Option<NodeHealthInfo>> {
        let system_health = self.system_health().await?;
        Ok(system_health.nodes.get(node_id).cloned())
    }

    /// Get list of all healthy nodes
    pub async fn healthy_nodes(&self) -> Result<Vec<NodeHealthInfo>> {
        let system_health = self.system_health().await?;
        Ok(system_health
            .nodes
            .values()
            .filter(|n| n.responsive && n.status.level == HealthLevel::Ok)
            .cloned()
            .collect())
    }

    /// Get list of degraded nodes
    pub async fn degraded_nodes(&self) -> Result<Vec<NodeHealthInfo>> {
        let system_health = self.system_health().await?;
        Ok(system_health
            .nodes
            .values()
            .filter(|n| n.responsive && n.status.level == HealthLevel::Degraded)
            .cloned()
            .collect())
    }

    /// Get list of unhealthy nodes
    pub async fn unhealthy_nodes(&self) -> Result<Vec<NodeHealthInfo>> {
        let system_health = self.system_health().await?;
        Ok(system_health
            .nodes
            .values()
            .filter(|n| n.responsive && n.status.level == HealthLevel::Error)
            .cloned()
            .collect())
    }

    /// Get list of unresponsive nodes (haven't reported recently)
    pub async fn unresponsive_nodes(&self) -> Result<Vec<NodeHealthInfo>> {
        let system_health = self.system_health().await?;
        Ok(system_health
            .nodes
            .values()
            .filter(|n| !n.responsive)
            .cloned()
            .collect())
    }

    /// Clear the cache (force fresh data on next query)
    pub async fn clear_cache(&self) {
        let mut cache = self.cache.write().await;
        *cache = None;
    }

    /// Get health history for a node
    ///
    /// Returns health reports for the specified node over the given duration.
    /// History is stored for up to 24 hours.
    ///
    /// # Arguments
    ///
    /// * `node_id` - The node to query history for
    /// * `duration` - How far back to retrieve history (e.g., Duration::from_secs(3600) for 1 hour)
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use mecha10::prelude::*;
    /// # use mecha10::health::HealthMonitor;
    /// # use std::time::Duration;
    /// # async fn example() -> Result<()> {
    /// let monitor = HealthMonitor::connect("redis://localhost:6379").await?;
    ///
    /// // Get last hour of health history
    /// let history = monitor.node_health_history("camera_node", Duration::from_secs(3600)).await?;
    /// println!("Found {} health reports in the last hour", history.len());
    /// # Ok(())
    /// # }
    /// ```
    pub async fn node_health_history(&self, node_id: &str, duration: Duration) -> Result<Vec<HealthReport>> {
        use redis::AsyncCommands;

        let mut conn = self
            .redis_client
            .get_multiplexed_async_connection()
            .await
            .map_err(|e| Mecha10Error::Other(format!("Failed to connect to Redis: {}", e)))?;

        let history_key = format!("mecha10:health:history:{}", node_id);

        // Calculate time range (in microseconds)
        let now = crate::prelude::now_micros();
        let duration_micros = duration.as_micros() as u64;
        let start_time = now.saturating_sub(duration_micros);

        // Query sorted set by score (timestamp)
        let results: Vec<String> = conn
            .zrangebyscore(&history_key, start_time as f64, now as f64)
            .await
            .map_err(|e| Mecha10Error::Other(format!("Failed to retrieve health history: {}", e)))?;

        // Deserialize results
        let mut reports = Vec::new();
        for json_str in results {
            match serde_json::from_str::<HealthReport>(&json_str) {
                Ok(report) => reports.push(report),
                Err(e) => {
                    eprintln!("Failed to deserialize health report: {}", e);
                    continue;
                }
            }
        }

        Ok(reports)
    }
}

// ============================================================================
// Auto-Reporting Extension
// ============================================================================

/// Extension trait for Context to enable automatic health reporting
#[async_trait::async_trait]
pub trait HealthReportingExt {
    /// Start automatic health reporting with default settings (Normal priority, 5s interval)
    ///
    /// Spawns a background task that periodically reports health status.
    /// The health_check_fn is called to generate the current health status.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::health::HealthReportingExt;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Report healthy status every 5 seconds (default)
    /// ctx.start_health_reporting(|| async { HealthStatus::healthy() }).await?;
    /// # Ok(())
    /// # }
    /// ```
    async fn start_health_reporting<F, Fut>(&self, health_check_fn: F) -> Result<()>
    where
        F: Fn() -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = HealthStatus> + Send;

    /// Start automatic health reporting with custom configuration
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::health::HealthReportingExt;
    /// use mecha10::node::HealthPriority;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Report every second with Critical priority
    /// ctx.start_health_reporting_with_config(
    ///     HealthPriority::Critical,
    ///     Duration::from_secs(1),
    ///     || async { HealthStatus::healthy() }
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    async fn start_health_reporting_with_config<F, Fut>(
        &self,
        priority: HealthPriority,
        interval: Duration,
        health_check_fn: F,
    ) -> Result<()>
    where
        F: Fn() -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = HealthStatus> + Send;
}

#[async_trait::async_trait]
impl HealthReportingExt for Context {
    async fn start_health_reporting<F, Fut>(&self, health_check_fn: F) -> Result<()>
    where
        F: Fn() -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = HealthStatus> + Send,
    {
        self.start_health_reporting_with_config(HealthPriority::Normal, Duration::from_secs(5), health_check_fn)
            .await
    }

    async fn start_health_reporting_with_config<F, Fut>(
        &self,
        priority: HealthPriority,
        interval: Duration,
        health_check_fn: F,
    ) -> Result<()>
    where
        F: Fn() -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = HealthStatus> + Send,
    {
        let reporter = HealthReporter::with_config(self, priority, 60, true).await?;

        tracing::info!(
            "Health reporting enabled: priority={:?}, interval={:?}",
            priority,
            interval
        );

        tokio::spawn(async move {
            let mut ticker = tokio::time::interval(interval);

            loop {
                ticker.tick().await;

                let status = health_check_fn().await;

                if let Err(e) = reporter.report(status).await {
                    tracing::debug!("Failed to report health: {}", e);
                }
            }
        });

        Ok(())
    }
}