mecha10-core 0.8.0

//! Type-safe and raw publish/subscribe methods.

#![cfg(feature = "messaging")]

use super::{Context, Receiver};
use crate::error::{Mecha10Error, Result};
use crate::messages::Message;
use crate::topics::Topic;
use anyhow::Context as _;
use serde::{de::DeserializeOwned, Serialize};

impl Context {
    /// 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_core::prelude::*;
    /// use mecha10_core::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(())
    /// # }
    /// ```
    pub async fn subscribe<T>(&self, topic: Topic<T>) -> Result<Receiver<T>>
    where
        T: Message + DeserializeOwned + Send + 'static,
    {
        let path = topic.path();
        self.subscribe_raw(path).await
    }

    /// 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_core::prelude::*;
    /// use mecha10_core::topics::sensor;
    /// use mecha10_core::messages::Image;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let image = Image {
    ///     timestamp: now_micros(),
    ///     width: 640,
    ///     height: 480,
    ///     encoding: "rgb8".to_string(),
    ///     data: vec![],
    ///     frame_id: None,
    /// };
    ///
    /// // Type checked: image must be of type Image
    /// ctx.publish_to(sensor::CAMERA_RGB, &image).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn publish_to<T>(&self, topic: Topic<T>, message: &T) -> Result<()>
    where
        T: Message + Serialize,
    {
        self.publish_raw(topic.path(), message).await
    }

    /// 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,
    {
        self.publish_raw(topic_path, message).await
    }

    /// 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_core::prelude::*;
    /// use mecha10_core::topics::sensor;
    /// use mecha10_core::messages::Image;
    ///
    /// # async fn example() -> std::result::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![],
    ///     frame_id: None,
    /// };
    ///
    /// // Publishes to "/sensor/camera/rgb/left"
    /// ctx.publish_to_scoped(sensor::CAMERA_RGB, &image).await?;
    /// # Ok(())
    /// # }
    /// ```
    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_core::prelude::*;
    /// use mecha10_core::topics::actuator;
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// // Velocity + torque PWM values for this motor
    /// let cmd = vec![1.0, 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(())
    /// # }
    /// ```
    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_core::prelude::*;
    /// use mecha10_core::topics::sensor;
    ///
    /// # async fn example() -> std::result::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(())
    /// # }
    /// ```
    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.
    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;

        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),
            })?;

        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();

        tokio::spawn(async move {
            let mut sub = subscriber;
            while let Some(msg) = sub.recv().await {
                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)"
                    );
                }
                if tx.send(msg.payload).is_err() {
                    break;
                }
            }
        });

        Ok(Receiver::from(rx))
    }

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