mecha10-messaging 0.1.18

//! Mecha10 Messaging Layer
//!
//! This crate provides a Redis Streams-based pub/sub messaging system for
//! inter-node communication in the Mecha10 framework.
//!
//! # Features
//!
//! - Type-safe message serialization/deserialization
//! - Redis Streams backend for reliable message delivery
//! - Topic-based routing
//! - Consumer groups for load balancing
//! - Message acknowledgment and retry
//! - Automatic reconnection
//!
//! # Example
//!
//! ```rust
//! use mecha10_messaging::{MessageBus, Message};
//! use serde::{Deserialize, Serialize};
//!
//! #[derive(Debug, Serialize, Deserialize)]
//! struct LaserScan {
//!     ranges: Vec<f32>,
//!     timestamp: u64,
//! }
//!
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! let mut bus = MessageBus::connect("redis://localhost:6379", "robot-1").await?;
//!
//! // Subscribe to topic
//! let mut rx = bus.subscribe::<LaserScan>("/scan", "processing").await?;
//!
//! // Publish message
//! let scan = LaserScan { ranges: vec![1.0, 2.0, 3.0], timestamp: 12345 };
//! bus.publish("/scan", &scan).await?;
//!
//! // Receive message
//! if let Some(msg) = rx.recv().await {
//!     println!("Received scan: {:?}", msg.payload);
//!     msg.ack().await?;
//! }
//! # Ok(())
//! # }
//! ```

use redis::aio::MultiplexedConnection;
use redis::{AsyncCommands, Client, RedisError};
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use thiserror::Error;
use tokio::sync::{mpsc, RwLock};

/// Messaging errors
#[derive(Debug, Error)]
pub enum MessagingError {
    #[error("Redis error: {0}")]
    Redis(#[from] RedisError),

    #[error("Serialization error: {0}")]
    Serialization(#[from] serde_json::Error),

    #[error("Connection error: {0}")]
    Connection(String),

    #[error("Subscription error: {0}")]
    Subscription(String),

    #[error("Channel closed")]
    ChannelClosed,
}

/// Result type for messaging operations
pub type Result<T> = std::result::Result<T, MessagingError>;

/// Configuration for connection retry logic
#[derive(Debug, Clone)]
pub struct RetryConfig {
    /// Maximum number of connection attempts
    pub max_attempts: usize,
    /// Initial delay between retries
    pub initial_delay: Duration,
    /// Maximum delay between retries
    pub max_delay: Duration,
    /// Multiplier for exponential backoff
    pub multiplier: f64,
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_attempts: 5,
            initial_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(10),
            multiplier: 2.0,
        }
    }
}

/// A message envelope containing metadata and payload
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Message<T> {
    /// Message ID (Redis Stream ID)
    pub id: String,

    /// Topic the message was published to
    pub topic: String,

    /// Publisher node ID
    pub publisher: String,

    /// Timestamp (milliseconds since epoch)
    pub timestamp: u64,

    /// Message payload
    pub payload: T,

    /// Internal: Shared Redis connection for acknowledgment (avoids creating new connections)
    #[serde(skip)]
    ack_connection: Option<Arc<RwLock<MultiplexedConnection>>>,

    /// Internal: consumer group for acknowledgment
    #[serde(skip)]
    consumer_group: Option<String>,
}

impl<T> Message<T> {
    /// Acknowledge the message (mark as processed)
    ///
    /// Uses a shared connection to avoid creating new connections for each ack,
    /// which significantly reduces overhead at high message rates.
    pub async fn ack(&self) -> Result<()> {
        if let (Some(conn), Some(group)) = (&self.ack_connection, &self.consumer_group) {
            // Reuse shared connection for acknowledgment
            let mut c = conn.write().await;
            let _: () = c
                .xack(&self.topic, group, &[&self.id])
                .await
                .map_err(MessagingError::Redis)?;
        }
        Ok(())
    }

    /// Negative acknowledgment (mark for retry)
    pub async fn nack(&self) -> Result<()> {
        // Redis Streams will automatically retry pending messages
        // We just don't ACK it, and it will be redelivered
        Ok(())
    }
}

/// Subscriber handle for receiving messages
pub struct Subscriber<T> {
    rx: mpsc::UnboundedReceiver<Message<T>>,
    topic: String,
}

impl<T> Subscriber<T> {
    /// Receive the next message
    pub async fn recv(&mut self) -> Option<Message<T>> {
        self.rx.recv().await
    }

    /// Get the topic this subscriber is listening to
    pub fn topic(&self) -> &str {
        &self.topic
    }
}

/// Message bus for pub/sub communication
pub struct MessageBus {
    connection: Arc<RwLock<MultiplexedConnection>>,
    node_id: String,
    namespace: String,
    redis_url: String,
    subscriptions: Arc<RwLock<HashMap<String, tokio::task::JoinHandle<()>>>>,
}

impl MessageBus {
    /// Connect to Redis and create a new message bus
    ///
    /// Uses exponential backoff retry logic for resilient connection establishment.
    ///
    /// # Arguments
    ///
    /// * `redis_url` - Redis connection URL (e.g., "redis://localhost:6379")
    /// * `node_id` - Unique identifier for this node
    ///
    /// # Example
    ///
    /// ```rust
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let bus = MessageBus::connect("redis://localhost:6379", "camera-node").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect(redis_url: &str, node_id: &str) -> Result<Self> {
        Self::connect_with_retry(redis_url, node_id, RetryConfig::default()).await
    }

    /// Connect to Redis with custom retry configuration
    ///
    /// # Arguments
    ///
    /// * `redis_url` - Redis connection URL
    /// * `node_id` - Unique identifier for this node
    /// * `retry_config` - Retry configuration for exponential backoff
    ///
    /// # Example
    ///
    /// ```rust
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let retry_config = RetryConfig {
    ///     max_attempts: 10,
    ///     initial_delay: Duration::from_millis(200),
    ///     max_delay: Duration::from_secs(30),
    ///     multiplier: 2.0,
    /// };
    /// let bus = MessageBus::connect_with_retry(
    ///     "redis://localhost:6379",
    ///     "camera-node",
    ///     retry_config
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect_with_retry(redis_url: &str, node_id: &str, retry_config: RetryConfig) -> Result<Self> {
        let client = Client::open(redis_url).map_err(MessagingError::Redis)?;

        let mut attempt = 0;
        let mut delay = retry_config.initial_delay;

        loop {
            attempt += 1;

            match client.get_multiplexed_async_connection().await {
                Ok(connection) => {
                    if attempt > 1 {
                        tracing::info!(
                            node_id = node_id,
                            attempts = attempt,
                            "Successfully connected to Redis after retries"
                        );
                    }

                    return Ok(Self {
                        connection: Arc::new(RwLock::new(connection)),
                        node_id: node_id.to_string(),
                        namespace: "mecha10".to_string(),
                        redis_url: redis_url.to_string(),
                        subscriptions: Arc::new(RwLock::new(HashMap::new())),
                    });
                }
                Err(e) => {
                    if attempt >= retry_config.max_attempts {
                        tracing::error!(
                            node_id = node_id,
                            attempts = attempt,
                            error = %e,
                            "Failed to connect to Redis after maximum retry attempts"
                        );
                        return Err(MessagingError::Connection(format!(
                            "Failed to connect to Redis after {} attempts: {}",
                            attempt, e
                        )));
                    }

                    tracing::warn!(
                        node_id = node_id,
                        attempt = attempt,
                        max_attempts = retry_config.max_attempts,
                        delay_ms = delay.as_millis(),
                        error = %e,
                        "Redis connection failed, retrying"
                    );

                    tokio::time::sleep(delay).await;

                    // Calculate next delay with exponential backoff
                    delay = std::cmp::min(
                        Duration::from_secs_f64(delay.as_secs_f64() * retry_config.multiplier),
                        retry_config.max_delay,
                    );
                }
            }
        }
    }

    /// Set the namespace for topic isolation
    ///
    /// Namespaces allow multiple robots/fleets to use the same Redis instance
    /// without topic collisions.
    pub fn set_namespace(&mut self, namespace: &str) {
        self.namespace = namespace.to_string();
    }

    /// Publish a message to a topic
    ///
    /// # Arguments
    ///
    /// * `topic` - Topic name (e.g., "/scan", "/odom")
    /// * `payload` - Message payload (must be serializable)
    ///
    /// # Example
    ///
    /// ```rust
    /// # use serde::{Serialize, Deserialize};
    /// # #[derive(Serialize, Deserialize)]
    /// # struct Point { x: f32, y: f32 }
    /// # async fn example(bus: &mut MessageBus) -> Result<(), Box<dyn std::error::Error>> {
    /// bus.publish("/position", &Point { x: 1.0, y: 2.0 }).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn publish<T: Serialize>(&mut self, topic: &str, payload: &T) -> Result<()> {
        let full_topic = format!("{}:{}", self.namespace, topic);

        let envelope = serde_json::json!({
            "publisher": self.node_id,
            "timestamp": std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_millis() as u64,
            "payload": payload,
        });

        let data = serde_json::to_string(&envelope)?;

        let mut conn = self.connection.write().await;

        // Use XADD with MAXLEN to keep approximately the last N messages in the stream
        // The ~ makes it approximate/efficient (Redis can trim at any point)
        // This prevents Redis from evicting empty streams and ensures stream persistence
        //
        // For camera/sensor streams that need low latency, calculate buffer based on target latency
        // Target: 200ms buffer to balance latency vs frame loss
        //
        // Buffer calculation based on camera FPS (configurable via MECHA10_CAMERA_FPS env var):
        //   15 FPS → 3 frames (200ms)
        //   20 FPS → 4 frames (200ms)
        //   30 FPS → 6 frames (200ms)
        //   60 FPS → 12 frames (200ms)
        let maxlen = if topic.contains("/camera/") || topic.contains("/video/") || topic.contains("/image/") {
            // Read camera FPS from environment (default: 30)
            let fps = std::env::var("MECHA10_CAMERA_FPS")
                .ok()
                .and_then(|s| s.parse::<f32>().ok())
                .unwrap_or(30.0);

            // Calculate buffer size for ~200ms latency
            let target_latency_ms = 200.0;
            let buffer_frames = ((fps * target_latency_ms) / 1000.0).ceil() as usize;
            buffer_frames.max(2) // Minimum 2 frames to prevent underflow
        } else {
            100 // Default: keep last 100 messages for other topics
        };

        let _: () = conn
            .xadd_maxlen(
                &full_topic,
                redis::streams::StreamMaxlen::Approx(maxlen),
                "*",
                &[("data", data)],
            )
            .await
            .map_err(MessagingError::Redis)?;

        Ok(())
    }

    /// Subscribe to a topic with a consumer group
    ///
    /// Consumer groups allow multiple nodes to process messages in parallel,
    /// with each message delivered to only one consumer in the group.
    ///
    /// # Arguments
    ///
    /// * `topic` - Topic name (e.g., "/scan", "/odom")
    /// * `consumer_group` - Consumer group name (e.g., "processing", "logging")
    ///
    /// # Example
    ///
    /// ```rust
    /// # use serde::{Serialize, Deserialize};
    /// # #[derive(Serialize, Deserialize)]
    /// # struct LaserScan { ranges: Vec<f32> }
    /// # async fn example(bus: &mut MessageBus) -> Result<(), Box<dyn std::error::Error>> {
    /// let mut rx = bus.subscribe::<LaserScan>("/scan", "slam").await?;
    /// while let Some(msg) = rx.recv().await {
    ///     println!("Received: {:?}", msg.payload);
    ///     msg.ack().await?;
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn subscribe<T: DeserializeOwned + Send + 'static>(
        &mut self,
        topic: &str,
        consumer_group: &str,
    ) -> Result<Subscriber<T>> {
        let full_topic = format!("{}:{}", self.namespace, topic);
        let consumer_id = format!("{}:{}", self.node_id, uuid::Uuid::new_v4());

        // Create consumer group with retry logic (exponential backoff)
        let max_retries = 5;
        let mut retry_count = 0;
        let mut group_created = false;

        while retry_count < max_retries {
            let mut conn = self.connection.write().await;
            let result: std::result::Result<(), RedisError> =
                conn.xgroup_create_mkstream(&full_topic, consumer_group, "$").await;
            drop(conn);

            match result {
                Ok(_) => {
                    tracing::info!(
                        "✅ Created consumer group '{}' for topic '{}'",
                        consumer_group,
                        full_topic
                    );
                    group_created = true;
                    break;
                }
                Err(e) => {
                    let err_msg = e.to_string();
                    // Group already exists is OK
                    if err_msg.contains("BUSYGROUP") {
                        tracing::debug!(
                            "Consumer group '{}' already exists for topic '{}'",
                            consumer_group,
                            full_topic
                        );
                        group_created = true;
                        break;
                    }

                    retry_count += 1;
                    if retry_count < max_retries {
                        let delay = std::time::Duration::from_millis(100 * 2_u64.pow(retry_count - 1));
                        tracing::warn!(
                            "⚠️  Failed to create consumer group '{}' for topic '{}' (attempt {}/{}): {:?}. Retrying in {:?}...",
                            consumer_group, full_topic, retry_count, max_retries, e, delay
                        );
                        tokio::time::sleep(delay).await;
                    } else {
                        tracing::error!(
                            "❌ Failed to create consumer group '{}' for topic '{}' after {} attempts: {:?}",
                            consumer_group,
                            full_topic,
                            max_retries,
                            e
                        );
                        return Err(MessagingError::Subscription(format!(
                            "Failed to create consumer group after {} retries: {}",
                            max_retries, e
                        )));
                    }
                }
            }
        }

        if !group_created {
            return Err(MessagingError::Subscription(format!(
                "Failed to verify consumer group creation for topic '{}'",
                full_topic
            )));
        }

        // Create channel for messages
        let (tx, rx) = mpsc::unbounded_channel();

        // Spawn background task to read from stream
        let redis_url = self.redis_url.clone();
        let full_topic_clone = full_topic.clone();
        let consumer_group_clone = consumer_group.to_string();
        let consumer_id_clone = consumer_id.clone();

        let task = tokio::spawn(async move {
            tracing::info!(
                "📡 Background subscription task started for topic '{}' group '{}'",
                full_topic_clone,
                consumer_group_clone
            );

            // Track consecutive connection failures for client recreation
            let mut consecutive_failures: u32 = 0;
            const MAX_FAILURES_BEFORE_RECREATE: u32 = 5;

            // Current client - will be recreated if connection driver terminates
            let mut current_client = match Client::open(redis_url.as_str()) {
                Ok(c) => c,
                Err(e) => {
                    tracing::error!(
                        "❌ Failed to create Redis client for topic '{}': {:?}",
                        full_topic_clone,
                        e
                    );
                    return;
                }
            };

            // Create a dedicated shared connection for acknowledgments
            // This avoids creating new connections for each message ack
            let mut ack_connection: Arc<RwLock<MultiplexedConnection>> =
                match current_client.get_multiplexed_async_connection().await {
                    Ok(conn) => Arc::new(RwLock::new(conn)),
                    Err(e) => {
                        tracing::error!(
                            "❌ Failed to create ack connection for topic '{}': {:?}",
                            full_topic_clone,
                            e
                        );
                        return;
                    }
                };

            loop {
                match current_client.get_multiplexed_async_connection().await {
                    Ok(mut conn) => {
                        // Reset failure counter on successful connection
                        consecutive_failures = 0;
                        // Read messages from stream
                        let result: std::result::Result<redis::streams::StreamReadReply, RedisError> = conn
                            .xread_options(
                                &[&full_topic_clone],
                                &[">"],
                                &redis::streams::StreamReadOptions::default()
                                    .group(&consumer_group_clone, &consumer_id_clone)
                                    .count(10)
                                    .block(1000),
                            )
                            .await;

                        match result {
                            Ok(reply) => {
                                tracing::debug!(
                                    "📨 Received {} keys from stream '{}'",
                                    reply.keys.len(),
                                    full_topic_clone
                                );
                                for stream_key in reply.keys {
                                    for stream_id in stream_key.ids {
                                        // Parse message
                                        match stream_id.map.get("data") {
                                            Some(redis::Value::BulkString(bytes)) => {
                                                match std::str::from_utf8(bytes) {
                                                    Ok(json_str) => {
                                                        match serde_json::from_str::<serde_json::Value>(json_str) {
                                                            Ok(envelope) => {
                                                                match serde_json::from_value::<T>(
                                                                    envelope["payload"].clone(),
                                                                ) {
                                                                    Ok(payload) => {
                                                                        let msg = Message {
                                                                            id: stream_id.id.clone(),
                                                                            topic: full_topic_clone.clone(),
                                                                            publisher: envelope["publisher"]
                                                                                .as_str()
                                                                                .unwrap_or("")
                                                                                .to_string(),
                                                                            timestamp: envelope["timestamp"]
                                                                                .as_u64()
                                                                                .unwrap_or(0),
                                                                            payload,
                                                                            ack_connection: Some(
                                                                                ack_connection.clone(),
                                                                            ),
                                                                            consumer_group: Some(
                                                                                consumer_group_clone.clone(),
                                                                            ),
                                                                        };

                                                                        if tx.send(msg).is_err() {
                                                                            tracing::warn!("Channel closed, subscriber dropped for topic '{}'", full_topic_clone);
                                                                            return;
                                                                            // Subscriber dropped
                                                                        }
                                                                    }
                                                                    Err(e) => {
                                                                        tracing::warn!(
                                                                            "Failed to deserialize payload for topic '{}' msg '{}': {:?}",
                                                                            full_topic_clone, stream_id.id, e
                                                                        );
                                                                    }
                                                                }
                                                            }
                                                            Err(e) => {
                                                                tracing::warn!(
                                                                    "Failed to parse JSON envelope for topic '{}' msg '{}': {:?}",
                                                                    full_topic_clone, stream_id.id, e
                                                                );
                                                            }
                                                        }
                                                    }
                                                    Err(e) => {
                                                        tracing::warn!(
                                                            "Failed to decode UTF-8 for topic '{}' msg '{}': {:?}",
                                                            full_topic_clone,
                                                            stream_id.id,
                                                            e
                                                        );
                                                    }
                                                }
                                            }
                                            Some(other) => {
                                                tracing::warn!(
                                                    "Unexpected Redis value type for topic '{}' msg '{}': {:?}",
                                                    full_topic_clone,
                                                    stream_id.id,
                                                    other
                                                );
                                            }
                                            None => {
                                                tracing::warn!(
                                                    "Missing 'data' field for topic '{}' msg '{}'",
                                                    full_topic_clone,
                                                    stream_id.id
                                                );
                                            }
                                        }
                                    }
                                }
                            }
                            Err(e) => {
                                let err_msg = e.to_string();
                                tracing::warn!(
                                    "❌ XREAD failed for topic '{}' group '{}': {:?}",
                                    full_topic_clone,
                                    consumer_group_clone,
                                    e
                                );

                                // If consumer group doesn't exist, try to recreate it
                                if err_msg.contains("NOGROUP") {
                                    tracing::warn!(
                                        "🔄 Consumer group '{}' missing, attempting to recreate...",
                                        consumer_group_clone
                                    );
                                    let result: std::result::Result<(), RedisError> = conn
                                        .xgroup_create_mkstream(&full_topic_clone, &consumer_group_clone, "$")
                                        .await;
                                    match result {
                                        Ok(_) => {
                                            tracing::info!(
                                                "✅ Recreated consumer group '{}' for topic '{}'",
                                                consumer_group_clone,
                                                full_topic_clone
                                            );
                                        }
                                        Err(create_err) => {
                                            tracing::error!(
                                                "❌ Failed to recreate consumer group '{}': {:?}",
                                                consumer_group_clone,
                                                create_err
                                            );
                                        }
                                    }
                                }

                                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                            }
                        }
                    }
                    Err(e) => {
                        consecutive_failures += 1;
                        tracing::warn!(
                            "❌ Failed to get Redis connection for topic '{}' (attempt {}): {:?}",
                            full_topic_clone,
                            consecutive_failures,
                            e
                        );

                        // If we've failed too many times, recreate the client
                        if consecutive_failures >= MAX_FAILURES_BEFORE_RECREATE {
                            tracing::warn!(
                                "🔄 Recreating Redis client for topic '{}' after {} consecutive failures",
                                full_topic_clone,
                                consecutive_failures
                            );

                            match Client::open(redis_url.as_str()) {
                                Ok(new_client) => {
                                    current_client = new_client;
                                    consecutive_failures = 0;

                                    // Also recreate the ack connection
                                    match current_client.get_multiplexed_async_connection().await {
                                        Ok(conn) => {
                                            ack_connection = Arc::new(RwLock::new(conn));
                                            tracing::info!(
                                                "✅ Successfully recreated Redis client and ack connection for topic '{}'",
                                                full_topic_clone
                                            );
                                        }
                                        Err(e) => {
                                            tracing::error!(
                                                "❌ Failed to create new ack connection for topic '{}': {:?}",
                                                full_topic_clone,
                                                e
                                            );
                                        }
                                    }
                                }
                                Err(e) => {
                                    tracing::error!(
                                        "❌ Failed to recreate Redis client for topic '{}': {:?}",
                                        full_topic_clone,
                                        e
                                    );
                                }
                            }
                        }

                        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                    }
                }
            }
        });

        // Store task handle
        let mut subs = self.subscriptions.write().await;
        subs.insert(full_topic, task);

        Ok(Subscriber {
            rx,
            topic: topic.to_string(),
        })
    }

    /// Subscribe to multiple topics matching a wildcard pattern
    ///
    /// Enables aggregated subscriptions across multiple topics using wildcard patterns.
    /// Useful for collecting data from multiple robots (e.g., `*/camera/rgb` subscribes
    /// to camera feeds from all robots).
    ///
    /// # Arguments
    ///
    /// * `pattern` - Wildcard pattern with `*` as placeholder (e.g., "*/camera/rgb", "robot-*/scan")
    /// * `consumer_group` - Consumer group name for load balancing
    ///
    /// # Pattern Syntax
    ///
    /// - `*` matches any segment in the topic path
    /// - `*/camera/rgb` matches "robot-1/camera/rgb", "robot-2/camera/rgb", etc.
    /// - `robot-*/scan` matches "robot-1/scan", "robot-alpha/scan", etc.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use serde::{Serialize, Deserialize};
    /// # #[derive(Serialize, Deserialize)]
    /// # struct Frame { data: Vec<u8> }
    /// # async fn example(bus: &mut MessageBus) -> Result<(), Box<dyn std::error::Error>> {
    /// // Subscribe to camera feeds from ALL robots
    /// let mut rx = bus.subscribe_pattern::<Frame>("*/camera/rgb", "vision-processing").await?;
    ///
    /// while let Some(msg) = rx.recv().await {
    ///     println!("Received frame from robot: {}", msg.topic);
    ///     // Process frame from any robot
    ///     msg.ack().await?;
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Implementation
    ///
    /// This method:
    /// 1. Converts wildcard pattern to Redis KEYS pattern
    /// 2. Discovers all matching topics using `discover_topics()`
    /// 3. Subscribes to each discovered topic with the same consumer group
    /// 4. Multiplexes messages from all topics into a single receiver
    ///
    /// # Note
    ///
    /// This performs initial topic discovery using Redis KEYS command. For dynamic
    /// topic discovery (new topics appearing after subscription), consider polling
    /// `discover_topics()` periodically and creating new subscriptions.
    pub async fn subscribe_pattern<T: DeserializeOwned + Send + 'static>(
        &mut self,
        pattern: &str,
        consumer_group: &str,
    ) -> Result<Subscriber<T>> {
        // Convert wildcard pattern to Redis KEYS pattern
        // Example: "*/camera/rgb" -> "mecha10:*/camera/rgb"
        let full_pattern = format!("{}:{}", self.namespace, pattern);

        // Discover all topics matching the pattern
        let topics = self.discover_topics(&full_pattern).await?;

        if topics.is_empty() {
            tracing::warn!(
                pattern = pattern,
                "No topics found matching pattern, subscriber will not receive messages until topics are created"
            );
        }

        // Create a single channel for all matching topics
        let (tx, rx) = mpsc::unbounded_channel();

        // Subscribe to each discovered topic
        for topic in topics {
            let full_topic = format!("{}:{}", self.namespace, topic);
            let consumer_id = format!("{}:{}", self.node_id, uuid::Uuid::new_v4());

            // Create consumer group if it doesn't exist
            let mut conn = self.connection.write().await;
            let _: std::result::Result<(), RedisError> =
                conn.xgroup_create_mkstream(&full_topic, consumer_group, "$").await;
            drop(conn);

            // Spawn background task to read from this stream
            let redis_url = self.redis_url.clone();
            let full_topic_clone = full_topic.clone();
            let consumer_group_clone = consumer_group.to_string();
            let consumer_id_clone = consumer_id.clone();
            let tx_clone = tx.clone();
            let topic_clone = topic.clone();

            let task = tokio::spawn(async move {
                // Track consecutive connection failures for client recreation
                let mut consecutive_failures: u32 = 0;
                const MAX_FAILURES_BEFORE_RECREATE: u32 = 5;

                // Current client - will be recreated if connection driver terminates
                let mut current_client = match Client::open(redis_url.as_str()) {
                    Ok(c) => c,
                    Err(e) => {
                        tracing::error!(
                            "❌ Failed to create Redis client for topic '{}': {:?}",
                            full_topic_clone,
                            e
                        );
                        return;
                    }
                };

                // Create a dedicated shared connection for acknowledgments
                let mut ack_connection: Arc<RwLock<MultiplexedConnection>> =
                    match current_client.get_multiplexed_async_connection().await {
                        Ok(conn) => Arc::new(RwLock::new(conn)),
                        Err(e) => {
                            tracing::error!(
                                "❌ Failed to create ack connection for topic '{}': {:?}",
                                full_topic_clone,
                                e
                            );
                            return;
                        }
                    };

                loop {
                    match current_client.get_multiplexed_async_connection().await {
                        Ok(mut conn) => {
                            // Reset failure counter on successful connection
                            consecutive_failures = 0;
                            // Read messages from stream
                            let result: std::result::Result<redis::streams::StreamReadReply, RedisError> = conn
                                .xread_options(
                                    &[&full_topic_clone],
                                    &[">"],
                                    &redis::streams::StreamReadOptions::default()
                                        .group(&consumer_group_clone, &consumer_id_clone)
                                        .count(10)
                                        .block(1000),
                                )
                                .await;

                            match result {
                                Ok(reply) => {
                                    for stream_key in reply.keys {
                                        for stream_id in stream_key.ids {
                                            // Parse message
                                            if let Some(redis::Value::BulkString(bytes)) = stream_id.map.get("data") {
                                                if let Ok(json_str) = std::str::from_utf8(bytes) {
                                                    if let Ok(envelope) =
                                                        serde_json::from_str::<serde_json::Value>(json_str)
                                                    {
                                                        if let Ok(payload) =
                                                            serde_json::from_value::<T>(envelope["payload"].clone())
                                                        {
                                                            let msg = Message {
                                                                id: stream_id.id.clone(),
                                                                topic: topic_clone.clone(),
                                                                publisher: envelope["publisher"]
                                                                    .as_str()
                                                                    .unwrap_or("")
                                                                    .to_string(),
                                                                timestamp: envelope["timestamp"].as_u64().unwrap_or(0),
                                                                payload,
                                                                ack_connection: Some(ack_connection.clone()),
                                                                consumer_group: Some(consumer_group_clone.clone()),
                                                            };

                                                            if tx_clone.send(msg).is_err() {
                                                                return; // Subscriber dropped
                                                            }
                                                        }
                                                    }
                                                }
                                            }
                                        }
                                    }
                                }
                                Err(_) => {
                                    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                                }
                            }
                        }
                        Err(e) => {
                            consecutive_failures += 1;
                            tracing::warn!(
                                "❌ Failed to get Redis connection for topic '{}' (attempt {}): {:?}",
                                full_topic_clone,
                                consecutive_failures,
                                e
                            );

                            // If we've failed too many times, recreate the client
                            if consecutive_failures >= MAX_FAILURES_BEFORE_RECREATE {
                                tracing::warn!(
                                    "🔄 Recreating Redis client for topic '{}' after {} consecutive failures",
                                    full_topic_clone,
                                    consecutive_failures
                                );

                                match Client::open(redis_url.as_str()) {
                                    Ok(new_client) => {
                                        current_client = new_client;
                                        consecutive_failures = 0;

                                        // Also recreate the ack connection
                                        match current_client.get_multiplexed_async_connection().await {
                                            Ok(conn) => {
                                                ack_connection = Arc::new(RwLock::new(conn));
                                                tracing::info!(
                                                    "✅ Successfully recreated Redis client and ack connection for topic '{}'",
                                                    full_topic_clone
                                                );
                                            }
                                            Err(e) => {
                                                tracing::error!(
                                                    "❌ Failed to create new ack connection for topic '{}': {:?}",
                                                    full_topic_clone,
                                                    e
                                                );
                                            }
                                        }
                                    }
                                    Err(e) => {
                                        tracing::error!(
                                            "❌ Failed to recreate Redis client for topic '{}': {:?}",
                                            full_topic_clone,
                                            e
                                        );
                                    }
                                }
                            }

                            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                        }
                    }
                }
            });

            // Store task handle
            let mut subs = self.subscriptions.write().await;
            subs.insert(format!("pattern:{}:{}", pattern, full_topic), task);
        }

        Ok(Subscriber {
            rx,
            topic: pattern.to_string(),
        })
    }

    /// Discover topics matching a Redis KEYS pattern
    ///
    /// Uses Redis KEYS command to find all streams matching the given pattern.
    /// This is useful for aggregated topic subscriptions.
    ///
    /// # Arguments
    ///
    /// * `pattern` - Redis KEYS pattern (e.g., "mecha10:topic:/sensor/camera/*")
    ///
    /// # Returns
    ///
    /// A vector of topic paths (without the namespace prefix)
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # async fn example(bus: &mut MessageBus) -> Result<(), Box<dyn std::error::Error>> {
    /// // Discover all camera topics
    /// let topics = bus.discover_topics("mecha10:topic:/sensor/camera/*").await?;
    /// println!("Found {} camera topics", topics.len());
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Note
    ///
    /// This uses Redis KEYS command which can be slow on large datasets.
    /// For production use with many topics, consider using SCAN instead.
    pub async fn discover_topics(&self, pattern: &str) -> Result<Vec<String>> {
        use redis::AsyncCommands;

        let mut conn = self.connection.write().await;

        // Use KEYS command to find matching topics
        let keys: Vec<String> = conn.keys(pattern).await.map_err(MessagingError::Redis)?;

        // Strip namespace prefix from each key
        let prefix = format!("{}:", self.namespace);
        let topics: Vec<String> = keys
            .into_iter()
            .filter_map(|key| key.strip_prefix(&prefix).map(|s| s.to_string()))
            .collect();

        Ok(topics)
    }

    /// Unsubscribe from all topics and close connections
    pub async fn close(&mut self) -> Result<()> {
        let mut subs = self.subscriptions.write().await;
        for (_, task) in subs.drain() {
            task.abort();
        }
        Ok(())
    }
}