mecha10_core/context/

lifecycle.rs

//! Context construction, instance naming, shutdown signaling, and logging span.

use super::Context;
#[cfg(feature = "messaging")]
use crate::error::Mecha10Error;
use crate::error::Result;
#[cfg(feature = "messaging")]
use anyhow::Context as _;
use std::sync::Arc;
use tokio::sync::RwLock;

impl Context {
    /// Create a new context
    ///
    /// # Arguments
    ///
    /// * `node_id` - Unique identifier for this node (fallback if NODE_NAME env var not set)
    ///
    /// # Environment Variables
    ///
    /// * `NODE_NAME` - If set, overrides the `node_id` parameter. This allows node-runner
    ///   to set full identifiers like `@mecha10/speaker` for consistent naming across
    ///   the CLI, dashboard, and health reporting.
    ///
    /// # Example
    ///
    /// ```rust
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let ctx = Context::new("my-node").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn new(node_id: &str) -> Result<Self> {
        // Use NODE_NAME from environment if set (for consistent naming with node-runner)
        // This allows bundled nodes to report health with full identifiers like @mecha10/speaker
        let node_id = std::env::var("NODE_NAME").unwrap_or_else(|_| node_id.to_string());
        #[cfg(feature = "messaging")]
        let (message_bus, control_plane_bus, redis_bridge) = {
            // Load infrastructure config
            let config = Self::load_infrastructure_config()?;

            // Connect to local Redis
            let redis_url = Self::get_redis_url_from_config(&config)?;
            let local_bus = mecha10_messaging::MessageBus::connect(&redis_url, &node_id)
                .await
                .with_context(|| {
                    format!(
                        "Failed to connect local message bus for node '{}' to Redis at '{}'",
                        node_id, redis_url
                    )
                })
                .map_err(|e| Mecha10Error::MessagingError {
                    message: format!("{:#}", e),
                    suggestion: "Ensure local Redis is running at the configured URL".to_string(),
                })?;

            // Connect to control plane Redis if configured
            let cp_bus = if let Some(cp_config) = &config.control_plane_redis {
                let cp_url = &cp_config.url;
                match mecha10_messaging::MessageBus::connect(cp_url, &node_id).await {
                    Ok(bus) => {
                        tracing::info!(
                            "Connected to control plane Redis at '{}' for node '{}'",
                            cp_url,
                            node_id
                        );
                        Some(bus)
                    }
                    Err(e) => {
                        tracing::warn!(
                            "Failed to connect to control plane Redis at '{}': {}. Continuing with local Redis only.",
                            cp_url,
                            e
                        );
                        None
                    }
                }
            } else {
                None
            };

            // Initialize Redis bridge if both connections are available
            let bridge = if let Some(_cp_bus_ref) = &cp_bus {
                // Get bridge configuration (use default if not specified)
                let bridge_config = config.bridge.unwrap_or_default();

                if bridge_config.enabled {
                    // Create bridge with separate clones of the buses
                    let bridge_local_bus = mecha10_messaging::MessageBus::connect(&redis_url, &node_id)
                        .await
                        .map_err(|e| Mecha10Error::MessagingError {
                            message: format!("Failed to create bridge connection to local Redis: {}", e),
                            suggestion: "Check local Redis connection".to_string(),
                        })?;

                    let cp_url = config
                        .control_plane_redis
                        .as_ref()
                        .map(|c| c.url.clone())
                        .unwrap_or_default();
                    let bridge_cp_bus = mecha10_messaging::MessageBus::connect(&cp_url, &node_id)
                        .await
                        .map_err(|e| Mecha10Error::MessagingError {
                            message: format!("Failed to create bridge connection to control plane Redis: {}", e),
                            suggestion: "Check control plane Redis connection".to_string(),
                        })?;

                    // Convert from infra_config::BridgeConfig to redis_bridge::BridgeConfig
                    let redis_bridge_config = crate::redis_bridge::BridgeConfig {
                        enabled: bridge_config.enabled,
                        local_topics: bridge_config.local_topics,
                        bridged_topics: bridge_config.bridged_topics,
                        robot_id: bridge_config.robot_id,
                    };

                    let bridge =
                        crate::redis_bridge::RedisBridge::new(redis_bridge_config, bridge_local_bus, bridge_cp_bus);

                    // Start the bridge
                    if let Err(e) = bridge.start().await {
                        tracing::warn!("Failed to start Redis bridge: {}. Continuing without bridge.", e);
                        None
                    } else {
                        tracing::info!("Redis bridge started successfully");
                        Some(bridge)
                    }
                } else {
                    tracing::debug!("Redis bridge is disabled in configuration");
                    None
                }
            } else {
                None
            };

            (local_bus, cp_bus, bridge)
        };

        Ok(Self {
            node_id,
            instance: None,
            #[cfg(feature = "messaging")]
            message_bus: Arc::new(RwLock::new(message_bus)),
            #[cfg(feature = "messaging")]
            control_plane_bus: Arc::new(RwLock::new(control_plane_bus)),
            #[cfg(feature = "messaging")]
            redis_bridge: Arc::new(RwLock::new(redis_bridge)),
            shutdown: Arc::new(RwLock::new(false)),
            state_manager: Arc::new(RwLock::new(None)),
        })
    }

    /// Set an instance name for this context (builder pattern)
    ///
    /// This allows the same node implementation to run multiple instances
    /// with different names. The node_id becomes "{node_name}/{instance}".
    ///
    /// # Example
    ///
    /// ```rust
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// // Single instance (no instance name)
    /// let ctx = Context::new("camera_fake").await?;
    /// // node_id: "camera_fake"
    ///
    /// // Multi-instance (with instance names)
    /// let left_ctx = Context::new("camera_fake").await?.with_instance("left");
    /// // node_id: "camera_fake/left"
    ///
    /// let right_ctx = Context::new("camera_fake").await?.with_instance("right");
    /// // node_id: "camera_fake/right"
    /// # Ok(())
    /// # }
    /// ```
    pub fn with_instance(mut self, instance: &str) -> Self {
        self.instance = Some(instance.to_string());
        self.node_id = format!("{}/{}", self.node_id, instance);
        self
    }

    /// Get the instance name if set
    ///
    /// Returns `None` for single-instance nodes, or `Some(instance)` for
    /// multi-instance nodes.
    pub fn instance(&self) -> Option<&str> {
        self.instance.as_deref()
    }

    /// Get the node ID
    pub fn node_id(&self) -> &str {
        &self.node_id
    }

    /// Check if control plane Redis connection is available
    #[cfg(feature = "messaging")]
    pub async fn has_control_plane(&self) -> bool {
        self.control_plane_bus.read().await.is_some()
    }

    /// Check if shutdown has been requested
    pub async fn is_shutdown(&self) -> bool {
        *self.shutdown.read().await
    }

    /// Request shutdown
    pub async fn shutdown(&self) {
        *self.shutdown.write().await = true;

        // Stop Redis bridge if running
        #[cfg(feature = "messaging")]
        {
            let bridge_opt = self.redis_bridge.read().await;
            if let Some(bridge) = bridge_opt.as_ref() {
                if let Err(e) = bridge.stop().await {
                    tracing::warn!("Failed to stop Redis bridge during shutdown: {}", e);
                }
            }
        }
    }

    /// Wait for shutdown signal
    pub async fn wait_for_shutdown(&self) {
        loop {
            if self.is_shutdown().await {
                break;
            }
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
        }
    }

    /// Enter a logging span with this node's context
    ///
    /// This returns a guard that, when entered, adds the node_id to all log messages.
    /// The span is automatically exited when the guard is dropped.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let _guard = ctx.logging_span().entered();
    /// info!("This message includes node_id automatically");
    /// debug!("So does this one");
    /// # Ok(())
    /// # }
    /// ```
    pub fn logging_span(&self) -> tracing::Span {
        tracing::info_span!("node", node_id = %self.node_id)
    }
}