mecha10-core 0.8.1

//! Redis Bridge
//!
//! Forwards messages between local Redis (on-robot) and control plane Redis (cloud).
//! This enables K8s-like distributed architecture where nodes can be scheduled
//! on robots or in the cloud.
//!
//! # Architecture
//!
//! ```text
//! Robot:
//!   Local Redis (localhost:6379)
//!     ↓↑ (bridged topics)
//!   RedisBridge
//!     ↓↑ (network)
//! Cloud:
//!   Control Plane Redis (shared across fleet)
//!     ↓↑
//!   Remote Nodes (K8s deployments)
//! ```
//!
//! # Topic Routing
//!
//! - **Local topics**: Stay on local Redis (e.g., `/sensor/camera/rgb` - high frequency)
//! - **Bridged topics**: Forwarded to control plane (e.g., `/system/heartbeat`, `/telemetry/*`)
//! - **Wildcard subscriptions**: Remote nodes can subscribe to `*/camera/rgb` (all robots)
//!
//! # Configuration
//!
//! ```yaml
//! bridge:
//!   enabled: true
//!   local_topics:
//!     - "/sensor/camera/*"     # Keep local (high frequency)
//!     - "/sensor/lidar/*"      # Keep local
//!   bridged_topics:
//!     - "/system/*"            # Bridge to cloud
//!     - "/telemetry/*"         # Bridge to cloud
//!     - "/logs/*"              # Bridge to cloud
//! ```

use crate::error::{Mecha10Error, Result};
use glob::Pattern;
use mecha10_messaging::MessageBus;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::sync::RwLock;
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BridgeConfig {
    pub enabled: bool,

    pub local_topics: Vec<String>,

    pub bridged_topics: Vec<String>,

    #[serde(default)]
    pub robot_id: Option<String>,
}

impl Default for BridgeConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            local_topics: vec![
                "/sensor/camera/*".to_string(),
                "/sensor/lidar/*".to_string(),
                "/sensor/imu/*".to_string(),
            ],
            bridged_topics: vec![
                "/system/*".to_string(),
                "/telemetry/*".to_string(),
                "/logs/*".to_string(),
                "/fleet/*".to_string(),
            ],
            robot_id: None,
        }
    }
}

impl BridgeConfig {
    /// Check if a topic should be bridged to control plane
    pub fn should_bridge(&self, topic: &str) -> bool {
        if !self.enabled {
            return false;
        }

        let matches_bridged = self.bridged_topics.iter().any(|pattern| {
            if let Ok(glob) = Pattern::new(pattern) {
                glob.matches(topic)
            } else {
                warn!("Invalid glob pattern in bridged_topics: {}", pattern);
                false
            }
        });

        let matches_local = self.local_topics.iter().any(|pattern| {
            if let Ok(glob) = Pattern::new(pattern) {
                glob.matches(topic)
            } else {
                warn!("Invalid glob pattern in local_topics: {}", pattern);
                false
            }
        });

        matches_bridged && !matches_local
    }

    /// Get the prefixed topic for control plane
    /// Example: "/camera/rgb" → "robot-123/camera/rgb"
    pub fn get_control_plane_topic(&self, topic: &str) -> String {
        if let Some(robot_id) = &self.robot_id {
            format!("{}{}", robot_id, topic)
        } else {
            topic.to_string()
        }
    }
}

struct ForwarderTask {
    handle: JoinHandle<()>,
    topic: String,
}

impl ForwarderTask {
    fn new(handle: JoinHandle<()>, topic: String) -> Self {
        Self { handle, topic }
    }

    fn abort(&self) {
        self.handle.abort();
    }
}

pub struct RedisBridge {
    config: BridgeConfig,

    local_bus: Arc<RwLock<MessageBus>>,

    control_plane_bus: Arc<RwLock<MessageBus>>,

    tasks: Arc<RwLock<Vec<ForwarderTask>>>,

    shutdown: Arc<RwLock<bool>>,
}

impl RedisBridge {
    /// Create a new Redis bridge
    ///
    /// # Arguments
    ///
    /// * `config` - Bridge configuration
    /// * `local_bus` - Local Redis message bus
    /// * `control_plane_bus` - Control plane Redis message bus
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use mecha10_core::redis_bridge::{BridgeConfig, RedisBridge};
    /// # use mecha10_messaging::MessageBus;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let config = BridgeConfig::default();
    /// let local_bus = MessageBus::connect("redis://localhost:6379", "robot-1").await?;
    /// let cp_bus = MessageBus::connect("redis://control-plane:6379", "robot-1").await?;
    ///
    /// let bridge = RedisBridge::new(config, local_bus, cp_bus);
    /// bridge.start().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn new(config: BridgeConfig, local_bus: MessageBus, control_plane_bus: MessageBus) -> Self {
        Self {
            config,
            local_bus: Arc::new(RwLock::new(local_bus)),
            control_plane_bus: Arc::new(RwLock::new(control_plane_bus)),
            tasks: Arc::new(RwLock::new(Vec::new())),
            shutdown: Arc::new(RwLock::new(false)),
        }
    }

    /// Start the bridge
    ///
    /// Spawns forwarder tasks for each bridged topic pattern.
    /// Messages matching bridged patterns will be forwarded from local to control plane.
    pub async fn start(&self) -> Result<()> {
        if !self.config.enabled {
            info!("Redis bridge is disabled in configuration");
            return Ok(());
        }

        info!(
            "Starting Redis bridge with {} bridged topic patterns",
            self.config.bridged_topics.len()
        );

        for pattern in &self.config.bridged_topics {
            self.start_forwarder(pattern).await?;
        }

        info!("Redis bridge started successfully");
        Ok(())
    }

    /// Start a forwarder task for a specific topic pattern
    async fn start_forwarder(&self, pattern: &str) -> Result<()> {
        debug!("Starting forwarder for pattern: {}", pattern);

        let topic = pattern.to_string();
        let config = self.config.clone();
        let local_bus = Arc::clone(&self.local_bus);
        let control_plane_bus = Arc::clone(&self.control_plane_bus);
        let shutdown = Arc::clone(&self.shutdown);

        let handle = tokio::spawn(async move {
            if let Err(e) = Self::forward_messages(topic.clone(), config, local_bus, control_plane_bus, shutdown).await
            {
                error!("Forwarder task error for topic {}: {}", topic, e);
            }
        });

        let mut tasks = self.tasks.write().await;
        tasks.push(ForwarderTask::new(handle, pattern.to_string()));

        Ok(())
    }

    /// Forward messages from local to control plane
    #[allow(clippy::too_many_arguments)]
    async fn forward_messages(
        topic: String,
        config: BridgeConfig,
        local_bus: Arc<RwLock<MessageBus>>,
        control_plane_bus: Arc<RwLock<MessageBus>>,
        shutdown: Arc<RwLock<bool>>,
    ) -> Result<()> {
        let mut local = local_bus.write().await;
        let mut subscriber = local
            .subscribe::<serde_json::Value>(&topic, "bridge")
            .await
            .map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to subscribe to topic {}: {}", topic, e),
                suggestion: "Check local Redis connection".to_string(),
            })?;
        drop(local);

        debug!("Subscribed to local topic: {}", topic);

        while !*shutdown.read().await {
            if let Some(msg) = subscriber.recv().await {
                debug!("Received message on topic {} from {}", msg.topic, msg.publisher);

                let cp_topic = config.get_control_plane_topic(&topic);

                let mut cp_bus = control_plane_bus.write().await;
                if let Err(e) = cp_bus.publish(&cp_topic, &msg.payload).await {
                    warn!(
                        "Failed to forward message from {} to control plane topic {}: {}",
                        topic, cp_topic, e
                    );
                } else {
                    debug!("Forwarded message to control plane topic: {}", cp_topic);
                }
                drop(cp_bus);

                if let Err(e) = msg.ack().await {
                    warn!("Failed to acknowledge message on topic {}: {}", topic, e);
                }
            }
        }

        debug!("Forwarder task stopped for topic: {}", topic);
        Ok(())
    }

    /// Stop the bridge gracefully
    ///
    /// Stops all forwarder tasks and waits for them to complete.
    pub async fn stop(&self) -> Result<()> {
        info!("Stopping Redis bridge...");

        *self.shutdown.write().await = true;

        let tasks = self.tasks.read().await;
        for task in tasks.iter() {
            debug!("Stopping forwarder task for topic: {}", task.topic);
            task.abort();
        }
        drop(tasks);

        self.tasks.write().await.clear();

        info!("Redis bridge stopped");
        Ok(())
    }

    /// Check if the bridge is running
    pub async fn is_running(&self) -> bool {
        !*self.shutdown.read().await
    }

    /// Get the number of active forwarder tasks
    pub async fn active_tasks(&self) -> usize {
        self.tasks.read().await.len()
    }
}