mecha10_core/

node.rs

// Mecha10 Node Trait and Lifecycle Management
//
// This module defines the core Node trait and lifecycle management for Mecha10 nodes.
// Every Mecha10 node implements the NodeImpl trait to define its behavior.

use crate::{Context, HealthStatus, NodeState, Result};
use async_trait::async_trait;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tracing::{debug, error, info, warn};

// ============================================================================
// Health Priority
// ============================================================================

/// Node health reporting priority
///
/// Determines heartbeat frequency and timeout thresholds based on
/// node criticality. Safety-critical nodes report more frequently.
///
/// # Priority Levels
///
/// | Priority | Interval | TTL | Timeout | Use Case |
/// |----------|----------|-----|---------|----------|
/// | Critical | 1s | 5s | 3s | motor-controller, estop |
/// | Normal | 5s | 60s | 15s | sensors, AI nodes |
/// | Background | 30s | 120s | 90s | logging, diagnostics |
///
/// # Example
///
/// ```rust
/// use mecha10::prelude::*;
///
/// // Create config with critical priority
/// let config = HealthReportingConfig::critical();
///
/// // Or specify priority explicitly
/// let config = HealthReportingConfig {
///     priority: HealthPriority::Critical,
///     ..Default::default()
/// };
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum HealthPriority {
    /// Safety-critical nodes that require immediate failure detection.
    ///
    /// Examples: motor-controller, estop, safety-monitor
    ///
    /// - Interval: 1 second
    /// - TTL: 5 seconds
    /// - Timeout: 3 seconds
    Critical,

    /// Standard operational nodes (default).
    ///
    /// Examples: camera, sensors, simulation-bridge
    ///
    /// - Interval: 5 seconds
    /// - TTL: 60 seconds
    /// - Timeout: 15 seconds
    #[default]
    Normal,

    /// Background/auxiliary nodes where delayed detection is acceptable.
    ///
    /// Examples: logging, telemetry, diagnostics
    ///
    /// - Interval: 30 seconds
    /// - TTL: 120 seconds
    /// - Timeout: 90 seconds
    Background,
}

impl HealthPriority {
    /// Get the heartbeat interval for this priority level
    pub const fn interval(&self) -> Duration {
        match self {
            Self::Critical => Duration::from_secs(1),
            Self::Normal => Duration::from_secs(5),
            Self::Background => Duration::from_secs(30),
        }
    }

    /// Get the Redis TTL (time-to-live) in seconds for this priority level
    ///
    /// The TTL should be longer than the interval to allow for some jitter,
    /// but short enough that stale data expires quickly.
    pub const fn ttl_seconds(&self) -> u64 {
        match self {
            Self::Critical => 5,
            Self::Normal => 60,
            Self::Background => 120,
        }
    }

    /// Get the timeout threshold for monitoring
    ///
    /// A node is considered unresponsive if no health report is received
    /// within this duration.
    pub const fn timeout(&self) -> Duration {
        match self {
            Self::Critical => Duration::from_secs(3),
            Self::Normal => Duration::from_secs(15),
            Self::Background => Duration::from_secs(90),
        }
    }

    /// Get the timeout in microseconds (for comparison with timestamps)
    pub const fn timeout_micros(&self) -> u64 {
        match self {
            Self::Critical => 3_000_000,
            Self::Normal => 15_000_000,
            Self::Background => 90_000_000,
        }
    }
}

impl std::fmt::Display for HealthPriority {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Critical => write!(f, "critical"),
            Self::Normal => write!(f, "normal"),
            Self::Background => write!(f, "background"),
        }
    }
}

// ============================================================================
// Configuration
// ============================================================================

/// Configuration for node health reporting
///
/// # Example
///
/// ```rust
/// use mecha10::prelude::*;
///
/// // Default config (Normal priority, 5s interval)
/// let config = HealthReportingConfig::default();
///
/// // Critical priority (1s interval)
/// let config = HealthReportingConfig::critical();
///
/// // Background priority (30s interval)
/// let config = HealthReportingConfig::background();
///
/// // Custom config
/// let config = HealthReportingConfig {
///     enabled: true,
///     priority: HealthPriority::Normal,
///     store_history: false, // Disable history for high-frequency nodes
/// };
/// ```
#[derive(Debug, Clone)]
pub struct HealthReportingConfig {
    /// Whether health reporting is enabled (default: true)
    pub enabled: bool,

    /// Priority level determines interval and timeout (default: Normal)
    pub priority: HealthPriority,

    /// Whether to store health history in Redis (default: true)
    ///
    /// Disable for high-frequency critical nodes to reduce Redis load.
    /// History is stored as a sorted set with 24-hour retention.
    pub store_history: bool,
}

impl Default for HealthReportingConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            priority: HealthPriority::Normal,
            store_history: true,
        }
    }
}

impl HealthReportingConfig {
    /// Create config for critical priority nodes (1s interval)
    pub fn critical() -> Self {
        Self {
            enabled: true,
            priority: HealthPriority::Critical,
            store_history: false, // Disable history for critical nodes to reduce load
        }
    }

    /// Create config for normal priority nodes (5s interval)
    pub fn normal() -> Self {
        Self::default()
    }

    /// Create config for background priority nodes (30s interval)
    pub fn background() -> Self {
        Self {
            enabled: true,
            priority: HealthPriority::Background,
            store_history: true,
        }
    }

    /// Get the effective interval based on priority
    pub fn interval(&self) -> Duration {
        self.priority.interval()
    }

    /// Get the effective TTL based on priority
    pub fn ttl_seconds(&self) -> u64 {
        self.priority.ttl_seconds()
    }
}

/// Trait for node configurations that specify an update rate
///
/// This trait provides a standard interface for nodes that publish at a fixed rate.
/// Implement this trait on your node's config struct to enable convenient rate-based
/// interval creation.
///
/// # Example
///
/// ```rust
/// use mecha10::prelude::*;
///
/// #[derive(Debug, Clone, Serialize, Deserialize)]
/// pub struct CameraConfig {
///     pub fps: f32,
///     pub width: u32,
///     pub height: u32,
/// }
///
/// impl RateConfig for CameraConfig {
///     fn rate_hz(&self) -> f32 {
///         self.fps
///     }
/// }
///
/// // Now you can use the config to create intervals easily:
/// # async fn example(config: CameraConfig) {
/// let mut interval = config.create_interval();
/// loop {
///     interval.tick().await;
///     // Publish at configured rate
/// }
/// # }
/// ```
pub trait RateConfig {
    /// Get the update rate in Hertz (cycles per second)
    fn rate_hz(&self) -> f32;

    /// Create a tokio interval from the configured rate
    ///
    /// This is a convenience method that converts the rate to a Duration
    /// and creates a tokio::time::Interval.
    fn create_interval(&self) -> tokio::time::Interval {
        let duration = Duration::from_secs_f32(1.0 / self.rate_hz());
        tokio::time::interval(duration)
    }

    /// Get the period (time between updates) as a Duration
    fn period(&self) -> Duration {
        Duration::from_secs_f32(1.0 / self.rate_hz())
    }
}

// ============================================================================
// Health Status Helper
// ============================================================================

/// Helper methods for creating HealthStatus instances
pub trait HealthStatusExt {
    /// Create a healthy status
    fn healthy(message: impl Into<String>) -> Self;
    /// Create an unhealthy status
    fn unhealthy(message: impl Into<String>) -> Self;
}

impl HealthStatusExt for HealthStatus {
    fn healthy(message: impl Into<String>) -> Self {
        Self {
            healthy: true,
            last_check: crate::prelude::now_micros(),
            failure_count: 0,
            message: Some(message.into()),
        }
    }

    fn unhealthy(message: impl Into<String>) -> Self {
        Self {
            healthy: false,
            last_check: crate::prelude::now_micros(),
            failure_count: 1,
            message: Some(message.into()),
        }
    }
}

// ============================================================================
// NodeImpl Trait
// ============================================================================

/// Core trait that all Mecha10 nodes must implement.
///
/// This trait defines the lifecycle of a node:
/// 1. `init()` - Initialize the node with configuration
/// 2. `run()` - Main execution loop
/// 3. `shutdown()` - Graceful shutdown
/// 4. `health_check()` - Report health status
///
/// # Example
///
/// ```rust
/// use mecha10::prelude::*;
///
/// #[derive(Debug)]
/// struct MyNode {
///     config: MyConfig,
/// }
///
/// #[derive(Debug, Deserialize)]
/// struct MyConfig {
///     rate: f32,
/// }
///
/// #[async_trait]
/// impl NodeImpl for MyNode {
///     type Config = MyConfig;
///
///     async fn init(config: Self::Config) -> Result<Self> {
///         info!("Initializing node with rate: {}", config.rate);
///         Ok(Self { config })
///     }
///
///     async fn run(&mut self, ctx: &Context) -> Result<()> {
///         let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?;
///
///         while let Some(image) = images.recv().await {
///             // Process image
///             info!("Received image: {}x{}", image.width, image.height);
///         }
///
///         Ok(())
///     }
/// }
/// ```
#[async_trait]
pub trait NodeImpl: Send + Sync + Debug + 'static {
    /// Configuration type for this node
    type Config: DeserializeOwned + Send + Sync + Debug;

    /// Initialize the node with the given configuration.
    ///
    /// This is called once when the node starts. Use this to:
    /// - Load configuration
    /// - Initialize resources (models, connections, etc.)
    /// - Validate prerequisites
    ///
    /// # Errors
    ///
    /// Return an error if initialization fails. The node will not start.
    async fn init(config: Self::Config) -> Result<Self>
    where
        Self: Sized;

    /// Main execution loop for the node.
    ///
    /// This is where your node's business logic lives. Typical patterns:
    /// - Subscribe to topics and process messages
    /// - Run periodic tasks
    /// - Publish results
    ///
    /// This method should run indefinitely until:
    /// - The context is cancelled (shutdown requested)
    /// - An unrecoverable error occurs
    ///
    /// # Errors
    ///
    /// Return an error if the node encounters an unrecoverable error.
    /// The node will transition to Error state and shutdown.
    async fn run(&mut self, ctx: &Context) -> Result<()>;

    /// Graceful shutdown hook.
    ///
    /// Called when the node is shutting down. Use this to:
    /// - Close connections
    /// - Flush buffers
    /// - Save state
    /// - Release resources
    ///
    /// The default implementation does nothing.
    ///
    /// # Errors
    ///
    /// Errors during shutdown are logged but don't prevent shutdown.
    async fn shutdown(&mut self) -> Result<()> {
        Ok(())
    }

    /// Health check for monitoring.
    ///
    /// Called periodically to check node health. Use this to:
    /// - Report current state
    /// - Check resource usage
    /// - Validate connections
    ///
    /// The default implementation returns healthy status.
    async fn health_check(&self) -> HealthStatus {
        HealthStatus {
            healthy: true,
            last_check: crate::prelude::now_micros(),
            failure_count: 0,
            message: Some("Node is running".to_string()),
        }
    }
}

// ============================================================================
// Node Runner
// ============================================================================

/// Wrapper that manages the lifecycle of a NodeImpl.
///
/// This provides:
/// - State management
/// - Lifecycle orchestration
/// - Error handling
/// - Health monitoring
#[derive(Clone)]
pub struct Node<T: NodeImpl> {
    node: Arc<Mutex<T>>,
    state: Arc<Mutex<NodeState>>,
}

impl<T: NodeImpl> Node<T> {
    /// Create a new node from a configuration.
    ///
    /// This calls `NodeImpl::init()` and wraps the result in a Node.
    ///
    /// # Errors
    ///
    /// Returns an error if initialization fails.
    pub async fn from_config(config: T::Config) -> Result<Self> {
        info!("Initializing node");

        let node = T::init(config).await?;

        Ok(Self {
            node: Arc::new(Mutex::new(node)),
            state: Arc::new(Mutex::new(NodeState::Starting)),
        })
    }

    /// Start the node with the given context.
    ///
    /// This orchestrates the full lifecycle:
    /// 1. Transition to Running state
    /// 2. Call `run()`
    /// 3. Call `shutdown()` on completion or error
    /// 4. Transition to final state
    ///
    /// # Errors
    ///
    /// Returns an error if the node fails during execution.
    pub async fn start(&self, ctx: Context) -> Result<()> {
        // Transition to running
        {
            let mut state = self.state.lock().await;
            *state = NodeState::Running;
            info!("Node transitioned to Running state");
        }

        // Run the node
        let result = {
            let mut node = self.node.lock().await;
            debug!("Starting node execution");
            node.run(&ctx).await
        };

        // Handle result
        match result {
            Ok(()) => {
                info!("Node completed successfully");
                self.shutdown().await?;
            }
            Err(e) => {
                error!("Node failed with error: {}", e);
                {
                    let mut state = self.state.lock().await;
                    *state = NodeState::Crashed;
                }
                self.shutdown().await?;
                return Err(e);
            }
        }

        Ok(())
    }

    /// Gracefully shutdown the node.
    ///
    /// This calls `shutdown()` and transitions to Stopped state.
    pub async fn shutdown(&self) -> Result<()> {
        info!("Shutting down node");

        // Transition to shutting down
        {
            let mut state = self.state.lock().await;
            *state = NodeState::Stopping;
        }

        // Call shutdown hook
        let result = {
            let mut node = self.node.lock().await;
            node.shutdown().await
        };

        // Transition to stopped
        {
            let mut state = self.state.lock().await;
            *state = NodeState::Stopped;
            info!("Node stopped");
        }

        if let Err(e) = result {
            warn!("Error during shutdown: {}", e);
            return Err(e);
        }

        Ok(())
    }

    /// Get the current state of the node.
    pub async fn state(&self) -> NodeState {
        self.state.lock().await.clone()
    }

    /// Perform a health check on the node.
    pub async fn health_check(&self) -> HealthStatus {
        let node = self.node.lock().await;
        node.health_check().await
    }

    /// Check if the node is healthy.
    pub async fn is_healthy(&self) -> bool {
        matches!(self.state().await, NodeState::Running | NodeState::Starting)
    }
}

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

/// Run a node with the given configuration and context.
///
/// This is a convenience function that:
/// 1. Initializes the node from config
/// 2. Starts the node with the context
/// 3. Handles errors
///
/// # Arguments
///
/// * `config` - Node-specific configuration
/// * `ctx` - Execution context
/// * `health_config` - Health reporting configuration
///
/// # Example
///
/// ```rust
/// use mecha10::prelude::*;
///
/// #[tokio::main]
/// async fn main() -> Result<()> {
///     let config = MyConfig { rate: 10.0 };
///     let ctx = Context::new("my_node");
///
///     // Use default health reporting (Normal priority, 5s interval)
///     run_node::<MyNode>(config, ctx, HealthReportingConfig::default()).await?;
///
///     // Critical priority for safety-critical nodes (1s interval)
///     run_node::<MyNode>(config, ctx, HealthReportingConfig::critical()).await?;
///
///     // Background priority for auxiliary nodes (30s interval)
///     run_node::<MyNode>(config, ctx, HealthReportingConfig::background()).await?;
///
///     Ok(())
/// }
/// ```
pub async fn run_node<T: NodeImpl>(
    config: T::Config,
    ctx: Context,
    health_config: HealthReportingConfig,
) -> Result<()> {
    // Initialize logging (safe to call multiple times)
    crate::prelude::init_logging();

    let node = Node::<T>::from_config(config).await?;

    // Start automatic health reporting
    if health_config.enabled {
        // Create clones for the health reporting task
        let node_for_health = node.node.clone();
        let ctx_for_health = ctx.clone();
        let priority = health_config.priority;
        let report_interval = health_config.interval();
        let ttl_seconds = health_config.ttl_seconds();
        let store_history = health_config.store_history;

        tokio::spawn(async move {
            use crate::health::HealthReporter;

            // Try to create health reporter with priority config
            let reporter =
                match HealthReporter::with_config(&ctx_for_health, priority, ttl_seconds, store_history).await {
                    Ok(r) => r,
                    Err(e) => {
                        warn!("Health reporting disabled: {}", e);
                        return;
                    }
                };

            info!(
                "Health reporting enabled: priority={}, interval={:?}, ttl={}s, history={}",
                priority, report_interval, ttl_seconds, store_history
            );

            let mut interval = tokio::time::interval(report_interval);

            loop {
                interval.tick().await;

                // Get health status from node
                let status = {
                    let node = node_for_health.lock().await;
                    node.health_check().await
                };

                // Convert types::HealthStatus to messages::system::HealthStatus
                let msg_status = crate::messages::system::HealthStatus {
                    timestamp: status.last_check,
                    level: if status.healthy {
                        crate::messages::system::HealthLevel::Ok
                    } else {
                        crate::messages::system::HealthLevel::Error
                    },
                    message: status.message.unwrap_or_else(|| "No message".to_string()),
                    diagnostics: None,
                };

                // Report to centralized monitoring
                if let Err(e) = reporter.report(msg_status).await {
                    debug!("Failed to report health: {}", e);
                }
            }
        });
    } else {
        info!("Health reporting disabled");
    }

    node.start(ctx).await
}