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 std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tracing::{debug, error, info, warn};

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

/// Configuration for node health reporting
#[derive(Debug, Clone)]
pub struct HealthReportingConfig {
    /// Interval between health reports (default: 5 seconds)
    pub interval: Duration,
    /// Whether health reporting is enabled (default: true)
    pub enabled: bool,
}

impl Default for HealthReportingConfig {
    fn default() -> Self {
        Self {
            interval: Duration::from_secs(5),
            enabled: true,
        }
    }
}

/// 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 (5 second interval)
///     run_node::<MyNode>(config, ctx, HealthReportingConfig::default()).await?;
///
///     // Or customize
///     run_node::<MyNode>(
///         config,
///         ctx,
///         HealthReportingConfig {
///             interval: Duration::from_secs(10),
///             enabled: true,
///         }
///     ).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 a clone for the health reporting task
        let node_for_health = node.node.clone();
        let ctx_for_health = ctx.clone();
        let report_interval = health_config.interval;

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

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

            info!("Health reporting enabled with interval: {:?}", report_interval);

            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
}