mecha10_core/context/

mod.rs

//! Node execution context
//!
//! Provides a high-level API for nodes to interact with the framework,
//! including type-safe pub/sub, configuration access, and lifecycle management.
//!
//! This module has been refactored into smaller sub-modules for better organization:
//! - `infra_config`: Infrastructure configuration types
//! - `receiver`: Message receiver wrapper

mod builder;
mod infra_config;
mod receiver;

pub use builder::ContextBuilder;
pub use infra_config::{BridgeConfig, InfrastructureConfig, MinioConfig, MlflowConfig, PostgresConfig, RedisConfig};
pub use receiver::Receiver;

use crate::error::{Mecha10Error, Result};
use crate::messages::Message;
use crate::state::FilesystemStateManager;
use crate::topics::Topic;
use anyhow::Context as _;
use serde::{de::DeserializeOwned, Serialize};
use std::sync::Arc;
use tokio::sync::RwLock;

// Import the complete Context struct and all implementations from the backup file
// This maintains backward compatibility while improving code organization

pub struct Context {
    /// Node identifier (includes instance if set: "node_name" or "node_name/instance")
    node_id: String,

    /// Optional instance name
    instance: Option<String>,

    /// Local message bus for on-robot communication (wrapped in Arc for sharing across tasks)
    #[cfg(feature = "messaging")]
    message_bus: Arc<RwLock<mecha10_messaging::MessageBus>>,

    /// Control plane message bus for robot-to-cloud communication (optional)
    #[cfg(feature = "messaging")]
    control_plane_bus: Arc<RwLock<Option<mecha10_messaging::MessageBus>>>,

    /// Redis bridge for forwarding messages between local and control plane
    #[cfg(feature = "messaging")]
    redis_bridge: Arc<RwLock<Option<crate::redis_bridge::RedisBridge>>>,

    /// Shutdown signal
    shutdown: Arc<RwLock<bool>>,

    /// State manager for persistent node state (lazy initialized)
    state_manager: Arc<RwLock<Option<Arc<crate::state::ConcreteStateManager>>>>,
}
impl Clone for Context {
    fn clone(&self) -> Self {
        Self {
            node_id: self.node_id.clone(),
            instance: self.instance.clone(),
            #[cfg(feature = "messaging")]
            message_bus: Arc::clone(&self.message_bus),
            #[cfg(feature = "messaging")]
            control_plane_bus: Arc::clone(&self.control_plane_bus),
            #[cfg(feature = "messaging")]
            redis_bridge: Arc::clone(&self.redis_bridge),
            shutdown: Arc::clone(&self.shutdown),
            state_manager: Arc::clone(&self.state_manager),
        }
    }
}
impl Context {
    /// Create a new context
    ///
    /// # Arguments
    ///
    /// * `node_id` - Unique identifier for this node
    ///
    /// # 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> {
        #[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: node_id.to_string(),
            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()
    }

    /// Load infrastructure configuration based on environment
    ///
    /// Priority order:
    /// 1. mecha10.json (project configuration) - redis.url field
    /// 2. Environment variables (MECHA10_REDIS_URL or REDIS_URL)
    /// 3. config/infrastructure.<environment>.yaml
    ///
    /// Set via MECHA10_ENVIRONMENT environment variable.
    fn load_infrastructure_config() -> Result<InfrastructureConfig> {
        use crate::config::load_config;
        use std::env;
        use std::path::PathBuf;

        // 1. Try loading from mecha10.json FIRST (project configuration)
        if let Ok(redis_url) = Self::try_load_redis_from_mecha10_json() {
            return Ok(InfrastructureConfig {
                redis: crate::context::infra_config::RedisConfig {
                    url: redis_url,
                    max_connections: 10,
                    connection_timeout_ms: 5000,
                },
                control_plane_redis: None,
                bridge: None,
                postgres: None,
                mlflow: None,
                minio: None,
            });
        }

        // 2. Check for environment variable overrides
        // If REDIS_URL or MECHA10_REDIS_URL is set, use that and create minimal config
        if let Ok(redis_url) = env::var("MECHA10_REDIS_URL").or_else(|_| env::var("REDIS_URL")) {
            return Ok(InfrastructureConfig {
                redis: crate::context::infra_config::RedisConfig {
                    url: redis_url,
                    max_connections: 10,
                    connection_timeout_ms: 5000,
                },
                control_plane_redis: None,
                bridge: None,
                postgres: None,
                mlflow: None,
                minio: None,
            });
        }

        // 3. Load from infrastructure.yaml
        let environment = env::var("MECHA10_ENVIRONMENT").unwrap_or_else(|_| "development".to_string());

        let config_filename = format!("infrastructure.{}.yaml", environment);

        // Try multiple paths in order:
        // 1. Current directory (for running from workspace root)
        // 2. Workspace root (for running tests from target/debug/deps)
        // 3. Two levels up (for running tests from packages/core/target/debug/deps)
        let paths_to_try = vec![
            PathBuf::from(format!("config/{}", config_filename)),
            PathBuf::from(format!("../../config/{}", config_filename)),
            PathBuf::from(format!("../../../../config/{}", config_filename)),
        ];

        for path in &paths_to_try {
            if path.exists() {
                return load_config(path)
                    .with_context(|| {
                        format!(
                            "Failed to load infrastructure config for environment '{}' from '{}'",
                            environment,
                            path.display()
                        )
                    })
                    .map_err(|e| Mecha10Error::Configuration(format!("{:#}", e)));
            }
        }

        // If no config file found, return error with helpful message
        Err(Mecha10Error::Configuration(format!(
            "Failed to load infrastructure config for environment '{}'. \
             Tried paths: {}. \
             Make sure the file exists, set MECHA10_REDIS_URL environment variable, \
             or configure redis.url in mecha10.json.",
            environment,
            paths_to_try
                .iter()
                .map(|p| p.display().to_string())
                .collect::<Vec<_>>()
                .join(", ")
        )))
    }

    /// Try to load Redis URL from mecha10.json
    ///
    /// Looks for mecha10.json in current directory and reads redis.url field.
    fn try_load_redis_from_mecha10_json() -> Result<String> {
        use std::path::PathBuf;

        // Try multiple paths to find mecha10.json
        let paths_to_try = vec![
            PathBuf::from("mecha10.json"),
            PathBuf::from("../mecha10.json"),
            PathBuf::from("../../mecha10.json"),
        ];

        for path in &paths_to_try {
            if !path.exists() {
                continue;
            }

            // Read and parse mecha10.json
            let content = std::fs::read_to_string(path)
                .map_err(|e| Mecha10Error::Configuration(format!("Failed to read {}: {}", path.display(), e)))?;

            let json: serde_json::Value = serde_json::from_str(&content)
                .map_err(|e| Mecha10Error::Configuration(format!("Failed to parse {}: {}", path.display(), e)))?;

            // Extract redis.url
            if let Some(redis_url) = json.get("redis").and_then(|r| r.get("url")).and_then(|u| u.as_str()) {
                return Ok(redis_url.to_string());
            }
        }

        Err(Mecha10Error::Configuration(
            "No mecha10.json found with redis.url configuration".to_string(),
        ))
    }

    /// Get Redis URL from infrastructure configuration (helper for internal use)
    fn get_redis_url_from_config(config: &InfrastructureConfig) -> Result<String> {
        // First try environment variables (for backward compatibility and testing)
        if let Ok(url) = std::env::var("MECHA10_REDIS_URL") {
            return Ok(url);
        }
        if let Ok(url) = std::env::var("REDIS_URL") {
            return Ok(url);
        }

        // Use config
        Ok(config.redis.url.clone())
    }

    /// Get Redis URL from infrastructure configuration
    pub fn get_redis_url() -> Result<String> {
        let config = Self::load_infrastructure_config()?;
        Self::get_redis_url_from_config(&config)
    }

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

    /// Load node configuration with environment-aware file-level fallback
    ///
    /// This method implements the V2 configuration system with:
    /// - Environment-specific configs (dev/staging/production)
    /// - File-level fallback (no key-by-key merging)
    /// - Automatic path resolution
    ///
    /// **Fallback Priority:**
    /// 1. `configs/{env}/nodes/{node-name}/config.json` (environment-specific)
    /// 2. `configs/common/nodes/{node-name}/config.json` (common fallback)
    /// 3. `T::default()` (code default)
    ///
    /// **Environment Detection:**
    /// - Uses `MECHA10_ENVIRONMENT` env var (default: "dev")
    /// - Supported values: "dev", "staging", "production", or custom
    ///
    /// # Type Parameters
    ///
    /// * `T` - Configuration type that implements `DeserializeOwned` + `Default`
    ///
    /// # Arguments
    ///
    /// * `node_name` - Name of the node (e.g., "simulation-bridge", "camera-streamer")
    ///
    /// # Returns
    ///
    /// The loaded configuration of type `T`, using the first available source
    /// in the fallback chain.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Debug, Clone, Deserialize, Serialize, Default)]
    /// struct SimBridgeConfig {
    ///     camera_width: u32,
    ///     camera_height: u32,
    ///     camera_fps: u32,
    /// }
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Loads from configs/{env}/nodes/simulation-bridge/config.json
    /// // Falls back to configs/common/nodes/simulation-bridge/config.json
    /// // Falls back to SimBridgeConfig::default()
    /// let config: SimBridgeConfig = ctx.load_node_config("simulation-bridge").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn load_node_config<T>(&self, node_name: &str) -> Result<T>
    where
        T: DeserializeOwned + Default,
    {
        use crate::config::load_config;
        use std::env;
        use std::path::PathBuf;

        // Get environment from env var (default: "dev")
        let environment = env::var("MECHA10_ENVIRONMENT").unwrap_or_else(|_| "dev".to_string());

        // Try multiple base paths (for running from different locations)
        let base_paths = vec![
            PathBuf::from("."),           // Current directory
            PathBuf::from("../.."),       // From target/debug/deps
            PathBuf::from("../../../.."), // From packages/*/target/debug/deps
        ];

        // Build fallback chain for each base path
        for base_path in &base_paths {
            // 1. Try environment-specific config
            let env_config_path = base_path
                .join("configs")
                .join(&environment)
                .join("nodes")
                .join(node_name)
                .join("config.json");

            if env_config_path.exists() {
                tracing::debug!(
                    "Loading node config for '{}' from environment-specific path: {}",
                    node_name,
                    env_config_path.display()
                );
                return load_config(&env_config_path)
                    .with_context(|| {
                        format!(
                            "Failed to load node config for '{}' from '{}'",
                            node_name,
                            env_config_path.display()
                        )
                    })
                    .map_err(|e| Mecha10Error::Configuration(format!("{:#}", e)));
            }

            // 2. Try common config
            let common_config_path = base_path
                .join("configs")
                .join("common")
                .join("nodes")
                .join(node_name)
                .join("config.json");

            if common_config_path.exists() {
                tracing::debug!(
                    "Loading node config for '{}' from common path: {}",
                    node_name,
                    common_config_path.display()
                );
                return load_config(&common_config_path)
                    .with_context(|| {
                        format!(
                            "Failed to load node config for '{}' from '{}'",
                            node_name,
                            common_config_path.display()
                        )
                    })
                    .map_err(|e| Mecha10Error::Configuration(format!("{:#}", e)));
            }
        }

        // 3. Fall back to default
        tracing::debug!(
            "No config file found for node '{}', using Default implementation",
            node_name
        );
        Ok(T::default())
    }

    /// 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()
    }

    /// Get control plane Redis URL from infrastructure configuration
    pub fn get_control_plane_redis_url() -> Result<Option<String>> {
        let config = Self::load_infrastructure_config()?;
        Ok(config.control_plane_redis.map(|c| c.url))
    }

    /// Subscribe to a topic with type-safe message handling
    ///
    /// # Arguments
    ///
    /// * `topic` - Type-safe topic to subscribe to
    ///
    /// # Returns
    ///
    /// A receiver that will yield messages of type `T`
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::topics::sensor;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Type is inferred as Receiver<Image>
    /// let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?;
    ///
    /// while let Some(image) = images.recv().await {
    ///     println!("Image: {}x{}", image.width, image.height);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn subscribe<T>(&self, topic: Topic<T>) -> Result<Receiver<T>>
    where
        T: Message + DeserializeOwned + Send + 'static,
    {
        let mut bus = self.message_bus.write().await;

        // Create a unique consumer group for this node
        let consumer_group = format!("{}-{}", self.node_id, topic.path().replace('/', "-"));

        let subscriber = bus
            .subscribe::<T>(topic.path(), &consumer_group)
            .await
            .with_context(|| {
                format!(
                    "Failed to subscribe to topic '{}' with consumer group '{}' from node '{}'",
                    topic.path(),
                    consumer_group,
                    self.node_id
                )
            })
            .map_err(|e| Mecha10Error::MessagingError {
                message: format!("{:#}", e),
                suggestion: format!("Check that Redis is running and topic '{}' is valid", topic.path()),
            })?;

        // Convert to our Receiver type
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();

        // Spawn task to forward messages
        tokio::spawn(async move {
            let mut sub = subscriber;
            while let Some(msg) = sub.recv().await {
                // Auto-acknowledge message before moving payload
                if let Err(e) = msg.ack().await {
                    tracing::warn!(
                        topic = %msg.topic,
                        message_id = %msg.id,
                        error = %e,
                        "Failed to acknowledge message (will be redelivered on reconnect)"
                    );
                }
                // Extract payload and forward
                if tx.send(msg.payload).is_err() {
                    break; // Receiver dropped
                }
            }
        });

        Ok(Receiver::from(rx))
    }

    /// Publish a message to a specific topic
    ///
    /// # Arguments
    ///
    /// * `topic` - Type-safe topic to publish to
    /// * `message` - Message to publish (type checked against topic)
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::topics::sensor;
    /// use mecha10::messages::Image;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let image = Image {
    ///     timestamp: now_micros(),
    ///     width: 640,
    ///     height: 480,
    ///     encoding: "rgb8".to_string(),
    ///     data: vec![],
    /// };
    ///
    /// // Type checked: image must be of type Image
    /// ctx.publish_to(sensor::CAMERA_RGB, &image).await?;
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn publish_to<T>(&self, topic: Topic<T>, message: &T) -> Result<()>
    where
        T: Message + Serialize,
    {
        let mut bus = self.message_bus.write().await;

        bus.publish(topic.path(), message)
            .await
            .with_context(|| {
                format!(
                    "Failed to publish to topic '{}' from node '{}'",
                    topic.path(),
                    self.node_id
                )
            })
            .map_err(|e| Mecha10Error::MessagingError {
                message: format!("{:#}", e),
                suggestion: "Check that Redis is running and accepting connections".to_string(),
            })?;

        Ok(())
    }

    /// Publish a message to a dynamic topic path (for runtime-generated topics like camera streams)
    ///
    /// This is useful when the topic path is determined at runtime and cannot be a `&'static str`.
    ///
    /// # Arguments
    /// * `topic_path` - The topic path as a String (e.g., "robot/sensors/camera/front/compressed")
    /// * `message` - The message to publish
    pub async fn publish_to_path<T>(&self, topic_path: &str, message: &T) -> Result<()>
    where
        T: Message + Serialize,
    {
        let mut bus = self.message_bus.write().await;

        bus.publish(topic_path, message)
            .await
            .with_context(|| {
                format!(
                    "Failed to publish to topic '{}' from node '{}'",
                    topic_path, self.node_id
                )
            })
            .map_err(|e| Mecha10Error::MessagingError {
                message: format!("{:#}", e),
                suggestion: "Check that Redis is running and accepting connections".to_string(),
            })?;

        Ok(())
    }

    /// Publish a message to an instance-scoped topic
    ///
    /// This automatically appends the context's instance name to the topic path.
    /// If no instance is set, this behaves like `publish_to()`.
    ///
    /// # Arguments
    ///
    /// * `topic` - Type-safe topic to publish to
    /// * `message` - Message to publish (type checked against topic)
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::topics::sensor;
    /// use mecha10::messages::Image;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let ctx = Context::new("camera_fake").await?.with_instance("left");
    ///
    /// let image = Image {
    ///     timestamp: now_micros(),
    ///     width: 640,
    ///     height: 480,
    ///     encoding: "rgb8".to_string(),
    ///     data: vec![],
    /// };
    ///
    /// // Publishes to "/sensor/camera/rgb/left"
    /// ctx.publish_to_scoped(sensor::CAMERA_RGB, &image).await?;
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn publish_to_scoped<T>(&self, topic: Topic<T>, message: &T) -> Result<()>
    where
        T: Message + Serialize,
    {
        let scoped_path = if let Some(instance) = &self.instance {
            format!("{}/{}", topic.path(), instance)
        } else {
            topic.path().to_string()
        };

        self.publish_raw(&scoped_path, message).await
    }

    /// Publish a message to a topic with an explicit instance suffix
    ///
    /// This allows publishing to a specific instance's topic regardless of
    /// the context's own instance name.
    ///
    /// # Arguments
    ///
    /// * `topic` - Type-safe topic to publish to
    /// * `instance` - Instance name to append to topic path
    /// * `message` - Message to publish (type checked against topic)
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::topics::actuator;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let cmd = MotorCommand { velocity: 1.0, torque: 0.0 };
    ///
    /// // Send to specific motor instances
    /// ctx.publish_to_instance(actuator::MOTOR_CMD, "front_left", &cmd).await?;
    /// ctx.publish_to_instance(actuator::MOTOR_CMD, "front_right", &cmd).await?;
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn publish_to_instance<T>(&self, topic: Topic<T>, instance: &str, message: &T) -> Result<()>
    where
        T: Message + Serialize,
    {
        let instance_path = format!("{}/{}", topic.path(), instance);
        self.publish_raw(&instance_path, message).await
    }

    /// Subscribe to an instance-scoped topic
    ///
    /// This automatically appends the context's instance name to the topic path.
    /// If no instance is set, this behaves like `subscribe()`.
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::topics::sensor;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let ctx = Context::new("vision_node").await?;
    ///
    /// // Subscribe to left camera (topic: "/sensor/camera/rgb/left")
    /// let mut left_images = ctx.subscribe_scoped(sensor::CAMERA_RGB, "left").await?;
    ///
    /// // Subscribe to right camera (topic: "/sensor/camera/rgb/right")
    /// let mut right_images = ctx.subscribe_scoped(sensor::CAMERA_RGB, "right").await?;
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn subscribe_scoped<T>(&self, topic: Topic<T>, instance: &str) -> Result<Receiver<T>>
    where
        T: Message + DeserializeOwned + Send + 'static,
    {
        let scoped_path = format!("{}/{}", topic.path(), instance);
        self.subscribe_raw(&scoped_path).await
    }

    /// Subscribe to a topic using a raw string path (for RPC and advanced use cases)
    ///
    /// # Arguments
    ///
    /// * `topic_path` - Raw topic path string
    ///
    /// # Returns
    ///
    /// A receiver that will yield messages of type `T`
    ///
    /// # Note
    ///
    /// This is a low-level method. Prefer using `subscribe()` with type-safe topics
    /// for normal use cases. This method is primarily for RPC and dynamic topic handling.
    #[cfg(feature = "messaging")]
    pub async fn subscribe_raw<T>(&self, topic_path: &str) -> Result<Receiver<T>>
    where
        T: Message + DeserializeOwned + Send + 'static,
    {
        let mut bus = self.message_bus.write().await;

        // Create a unique consumer group for this node
        let consumer_group = format!("{}-{}", self.node_id, topic_path.replace('/', "-"));

        let subscriber = bus
            .subscribe::<T>(topic_path, &consumer_group)
            .await
            .with_context(|| {
                format!(
                    "Failed to subscribe to raw topic '{}' with consumer group '{}' from node '{}'",
                    topic_path, consumer_group, self.node_id
                )
            })
            .map_err(|e| Mecha10Error::MessagingError {
                message: format!("{:#}", e),
                suggestion: format!("Check that Redis is running and topic '{}' is valid", topic_path),
            })?;

        // Convert to our Receiver type
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();

        // Spawn task to forward messages
        tokio::spawn(async move {
            let mut sub = subscriber;
            while let Some(msg) = sub.recv().await {
                // Auto-acknowledge message before moving payload
                if let Err(e) = msg.ack().await {
                    tracing::warn!(
                        topic = %msg.topic,
                        message_id = %msg.id,
                        error = %e,
                        "Failed to acknowledge message (will be redelivered on reconnect)"
                    );
                }
                // Extract payload and forward
                if tx.send(msg.payload).is_err() {
                    break; // Receiver dropped
                }
            }
        });

        Ok(Receiver::from(rx))
    }

    /// Discover topics matching a glob pattern
    ///
    /// Queries Redis for all topics matching the given pattern without subscribing.
    /// Useful for inspecting available topics before subscribing.
    ///
    /// # Pattern Syntax
    ///
    /// - `*` matches any sequence of characters within a path segment
    /// - `**` matches any sequence of path segments
    /// - `{a,b}` matches either a or b (future enhancement)
    ///
    /// # Arguments
    ///
    /// * `pattern` - Glob pattern to match topics against
    ///
    /// # Returns
    ///
    /// A vector of topic paths that match the pattern
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Discover all camera topics
    /// let topics = ctx.discover_topics("/sensor/camera/*").await?;
    /// println!("Found {} camera topics: {:?}", topics.len(), topics);
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn discover_topics(&self, pattern: &str) -> Result<Vec<String>> {
        use crate::aggregated::pattern::convert_glob_to_redis;

        let bus = self.message_bus.read().await;
        let redis_pattern = convert_glob_to_redis(pattern);

        bus.discover_topics(&redis_pattern)
            .await
            .with_context(|| {
                format!(
                    "Failed to discover topics matching pattern '{}' (redis pattern: '{}') from node '{}'",
                    pattern, redis_pattern, self.node_id
                )
            })
            .map_err(|e| Mecha10Error::MessagingError {
                message: format!("{:#}", e),
                suggestion: "Check that Redis is running and the pattern is valid".to_string(),
            })
    }

    /// Subscribe to multiple topics matching a glob pattern
    ///
    /// Returns an aggregated receiver that multiplexes all matching topics into a single stream.
    /// Each message is tagged with its source topic path.
    ///
    /// # Pattern Syntax
    ///
    /// - `*` matches any sequence of characters within a path segment
    /// - `**` matches any sequence of path segments
    /// - `{a,b}` matches either a or b (future enhancement)
    ///
    /// # Arguments
    ///
    /// * `pattern` - Glob pattern to match topics against
    ///
    /// # Returns
    ///
    /// An `AggregatedReceiver<T>` that yields `(topic_path, message)` tuples
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Subscribe to all cameras with one call
    /// let mut cameras = ctx.subscribe_aggregated::<ImageMessage>("/sensor/camera/*").await?;
    ///
    /// while let Some((topic, image)) = cameras.recv().await {
    ///     let camera_name = topic.rsplit('/').next().unwrap_or("unknown");
    ///     println!("Camera {}: {}x{}", camera_name, image.width, image.height);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn subscribe_aggregated<T>(&self, pattern: &str) -> Result<crate::aggregated::AggregatedReceiver<T>>
    where
        T: Message + DeserializeOwned + Send + 'static,
    {
        // 1. Discover all topics matching pattern
        let matching_topics = self.discover_topics(pattern).await?;

        if matching_topics.is_empty() {
            return Err(Mecha10Error::MessagingError {
                message: format!("No topics found matching pattern: {}", pattern),
                suggestion: "Check that the pattern is correct and topics exist".to_string(),
            });
        }

        // 2. Subscribe to each topic
        let mut receivers = Vec::new();
        for topic_path in matching_topics {
            let rx = self.subscribe_raw::<T>(&topic_path).await?;
            // Extract receiver and wrap in ReceiverType enum
            let receiver_type = match rx.inner {
                crate::context::receiver::ReceiverInner::Unbounded(unbounded) => {
                    crate::aggregated::ReceiverType::Unbounded(unbounded)
                }
                crate::context::receiver::ReceiverInner::Bounded(bounded) => {
                    crate::aggregated::ReceiverType::Bounded(bounded)
                }
            };
            receivers.push((topic_path.clone(), receiver_type));
        }

        // 3. Return aggregated receiver with support for both channel types
        Ok(crate::aggregated::AggregatedReceiver::new_mixed(
            pattern.to_string(),
            receivers,
        ))
    }

    /// Publish a message to a topic using a raw string path (for RPC and advanced use cases)
    ///
    /// # Arguments
    ///
    /// * `topic_path` - Raw topic path string
    /// * `message` - Message to publish
    ///
    /// # Note
    ///
    /// This is a low-level method. Prefer using `publish_to()` with type-safe topics
    /// for normal use cases. This method is primarily for RPC and dynamic topic handling.
    #[cfg(feature = "messaging")]
    pub async fn publish_raw<T>(&self, topic_path: &str, message: &T) -> Result<()>
    where
        T: Message + Serialize,
    {
        let mut bus = self.message_bus.write().await;

        bus.publish(topic_path, message)
            .await
            .with_context(|| {
                format!(
                    "Failed to publish to raw topic '{}' from node '{}'",
                    topic_path, self.node_id
                )
            })
            .map_err(|e| Mecha10Error::MessagingError {
                message: format!("{:#}", e),
                suggestion: "Check that Redis is running and accepting connections".to_string(),
            })?;

        Ok(())
    }

    /// 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)
    }

    /// Subscribe to all camera topics
    ///
    /// Convenience method for `subscribe_aggregated::<Image>("/sensor/camera/*")`
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let mut cameras = ctx.subscribe_all_cameras().await?;
    /// while let Some((topic, image)) = cameras.recv().await {
    ///     let camera_name = topic.rsplit('/').next().unwrap_or("unknown");
    ///     println!("Camera {}: {}x{}", camera_name, image.width, image.height);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn subscribe_all_cameras(&self) -> Result<crate::aggregated::AggregatedReceiver<crate::types::Image>> {
        self.subscribe_aggregated("/sensor/camera/*").await
    }

    /// Subscribe to all sensor topics
    ///
    /// Subscribes to all sensor data including cameras, LiDAR, IMU, odometry, etc.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Subscribe to all sensors (returns untyped messages)
    /// let topics = ctx.discover_topics("/sensor/**").await?;
    /// println!("Found {} sensor topics", topics.len());
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn discover_all_sensors(&self) -> Result<Vec<String>> {
        self.discover_topics("/sensor/**").await
    }

    /// Subscribe to all LiDAR topics
    ///
    /// Convenience method for `subscribe_aggregated::<LaserScan>("/sensor/lidar/*")`
    #[cfg(feature = "messaging")]
    pub async fn subscribe_all_lidars(&self) -> Result<crate::aggregated::AggregatedReceiver<crate::types::LaserScan>> {
        self.subscribe_aggregated("/sensor/lidar/*").await
    }

    /// Subscribe to all IMU topics
    ///
    /// Convenience method for `subscribe_aggregated::<IMU>("/sensor/imu/*")`
    #[cfg(feature = "messaging")]
    pub async fn subscribe_all_imus(&self) -> Result<crate::aggregated::AggregatedReceiver<crate::types::IMU>> {
        self.subscribe_aggregated("/sensor/imu/*").await
    }

    /// Subscribe to all motor topics
    ///
    /// Convenience method for `subscribe_aggregated::<MotorStatus>("/actuator/motor/*")`
    #[cfg(feature = "messaging")]
    pub async fn subscribe_all_motors(
        &self,
    ) -> Result<crate::aggregated::AggregatedReceiver<crate::actuator::MotorStatus>> {
        self.subscribe_aggregated("/actuator/motor/*").await
    }

    /// Subscribe to a topic pattern with wildcard
    ///
    /// Generic helper for subscribing to multiple topics matching a pattern.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    /// use mecha10::topics::sensor;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Subscribe to all instances of a specific topic type
    /// let mut cameras = ctx.subscribe_pattern(sensor::CAMERA_RGB, "*").await?;
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn subscribe_pattern<T>(
        &self,
        topic: crate::topics::Topic<T>,
        pattern: &str,
    ) -> Result<crate::aggregated::AggregatedReceiver<T>>
    where
        T: Message + DeserializeOwned + Send + 'static,
    {
        let full_pattern = format!("{}/{}", topic.path(), pattern);
        self.subscribe_aggregated(&full_pattern).await
    }

    /// Get or create a state manager for this node
    ///
    /// The state manager provides persistent storage for node state across restarts.
    /// By default, it uses filesystem storage in `./state/` directory.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Debug, Clone, Serialize, Deserialize)]
    /// struct MyState {
    ///     counter: u64,
    ///     position: (f64, f64),
    /// }
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let state_mgr = ctx.state_manager().await?;
    ///
    /// // Load previous state
    /// let mut state = state_mgr
    ///     .load::<MyState>("my_node")
    ///     .await?
    ///     .unwrap_or(MyState {
    ///         counter: 0,
    ///         position: (0.0, 0.0),
    ///     });
    ///
    /// // Update state
    /// state.counter += 1;
    ///
    /// // Save state
    /// state_mgr.save("my_node", &state).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn state_manager(&self) -> Result<Arc<crate::state::ConcreteStateManager>> {
        let mut mgr = self.state_manager.write().await;

        if let Some(existing) = mgr.as_ref() {
            return Ok(Arc::clone(existing));
        }

        // Create default filesystem-based state manager
        let state_dir = std::env::var("MECHA10_STATE_DIR").unwrap_or_else(|_| "./state".to_string());
        let fs_mgr = FilesystemStateManager::new(&state_dir).await?;
        let arc_mgr = Arc::new(crate::state::ConcreteStateManager::Filesystem(fs_mgr));

        *mgr = Some(Arc::clone(&arc_mgr));

        Ok(arc_mgr)
    }

    /// Set a custom state manager for this context
    ///
    /// This allows using alternative backends (Redis, PostgreSQL, etc.) or custom implementations.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    /// use std::sync::Arc;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Use in-memory state manager for testing
    /// let mem_mgr = Arc::new(MemoryStateManager::new());
    /// ctx.set_state_manager(mem_mgr).await;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn set_state_manager(&self, manager: Arc<crate::state::ConcreteStateManager>) {
        let mut mgr = self.state_manager.write().await;
        *mgr = Some(manager);
    }

    // ========================================
    // Topic Discovery Helpers
    // ========================================

    /// List all sensor topics currently active in the system
    ///
    /// Returns all topics under `/sensor/*` including cameras, LiDAR, IMU, GPS, etc.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let sensors = ctx.list_sensor_topics().await?;
    /// println!("Available sensors:");
    /// for topic in sensors {
    ///     println!("  - {}", topic);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn list_sensor_topics(&self) -> Result<Vec<String>> {
        self.discover_topics("/sensor/*").await
    }

    /// List all actuator topics currently active in the system
    ///
    /// Returns all topics under `/actuator/*` including motors, servos, grippers, etc.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let actuators = ctx.list_actuator_topics().await?;
    /// println!("Available actuators:");
    /// for topic in actuators {
    ///     println!("  - {}", topic);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn list_actuator_topics(&self) -> Result<Vec<String>> {
        self.discover_topics("/actuator/*").await
    }

    /// List all topics in the system
    ///
    /// Returns all active topics regardless of category.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let all_topics = ctx.list_all_topics().await?;
    /// println!("All active topics ({} total):", all_topics.len());
    /// for topic in all_topics {
    ///     println!("  - {}", topic);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn list_all_topics(&self) -> Result<Vec<String>> {
        self.discover_topics("*").await
    }

    /// List all camera topics
    ///
    /// Returns all topics under `/sensor/camera/*`.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let cameras = ctx.list_camera_topics().await?;
    /// println!("Available cameras:");
    /// for topic in cameras {
    ///     println!("  - {}", topic);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn list_camera_topics(&self) -> Result<Vec<String>> {
        self.discover_topics("/sensor/camera/*").await
    }

    /// List all LiDAR topics
    ///
    /// Returns all topics under `/sensor/lidar/*`.
    #[cfg(feature = "messaging")]
    pub async fn list_lidar_topics(&self) -> Result<Vec<String>> {
        self.discover_topics("/sensor/lidar/*").await
    }

    /// List all IMU topics
    ///
    /// Returns all topics under `/sensor/imu/*`.
    #[cfg(feature = "messaging")]
    pub async fn list_imu_topics(&self) -> Result<Vec<String>> {
        self.discover_topics("/sensor/imu/*").await
    }

    /// List all motor topics
    ///
    /// Returns all topics under `/actuator/motor/*`.
    #[cfg(feature = "messaging")]
    pub async fn list_motor_topics(&self) -> Result<Vec<String>> {
        self.discover_topics("/actuator/motor/*").await
    }

    /// Print a formatted summary of all active topics
    ///
    /// Useful for debugging and introspection. Groups topics by category
    /// and displays them in a human-readable format.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use mecha10::prelude::*;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// ctx.print_topic_summary().await?;
    /// // Output:
    /// // === Active Topics ===
    /// // Sensors (3):
    /// //   - /sensor/camera/front
    /// //   - /sensor/lidar/main
    /// //   - /sensor/imu/base
    /// // Actuators (2):
    /// //   - /actuator/motor/left
    /// //   - /actuator/motor/right
    /// // =====================
    /// # Ok(())
    /// # }
    /// ```
    #[cfg(feature = "messaging")]
    pub async fn print_topic_summary(&self) -> Result<()> {
        let sensors = self.list_sensor_topics().await.unwrap_or_default();
        let actuators = self.list_actuator_topics().await.unwrap_or_default();
        let all_topics = self.list_all_topics().await?;

        let other_topics: Vec<String> = all_topics
            .into_iter()
            .filter(|t| !t.starts_with("/sensor/") && !t.starts_with("/actuator/"))
            .collect();

        println!("=== Active Topics ===");

        if !sensors.is_empty() {
            println!("Sensors ({}):", sensors.len());
            for topic in sensors {
                println!("  - {}", topic);
            }
        }

        if !actuators.is_empty() {
            println!("Actuators ({}):", actuators.len());
            for topic in actuators {
                println!("  - {}", topic);
            }
        }

        if !other_topics.is_empty() {
            println!("Other ({}):", other_topics.len());
            for topic in other_topics {
                println!("  - {}", topic);
            }
        }

        println!("=====================");

        Ok(())
    }
}