mecha10-behavior-runtime 0.8.1

//! Action node that relays live target requests to the `instance-track` node's control topic.
//!
//! Mirrors `set_rl_target.rs`, but relays a `bbox` instead of a `target_class`.

use crate::{BehaviorNode, Context, NodeStatus};
use async_trait::async_trait;
use mecha10_core::messages::Message;
use mecha10_core::topics::Topic;
use mecha10_core::Receiver;
use serde::{Deserialize, Serialize};
use serde_json::Value;

#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
pub(crate) struct BoundingBox {
    x: f32,
    y: f32,
    width: f32,
    height: f32,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct InstanceTrackControlCommand {
    action: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    params: Option<Value>,
}

impl Message for InstanceTrackControlCommand {}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct InstanceTrackTargetRequest {
    #[serde(default)]
    pub(crate) bbox: Option<BoundingBox>,
}

impl Message for InstanceTrackTargetRequest {}

fn default_topic() -> String {
    "/instance-track/control".to_string()
}

/// The topic live target requests (clicks/drag-boxes on the dashboard's camera feed) arrive on.
/// `pub(crate)` so `InstanceTrackTaskNode` can subscribe to the same topic to track whether a
/// target is currently live, without hardcoding the string a second time.
pub(crate) fn default_request_topic() -> String {
    "/instance-track/target/request".to_string()
}

pub struct SetInstanceTrackTargetNode {
    topic: String,
    request_topic: String,
    target: Option<BoundingBox>,
    receiver: Option<Receiver<Value>>,
    published_initial: bool,
}

impl std::fmt::Debug for SetInstanceTrackTargetNode {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SetInstanceTrackTargetNode")
            .field("topic", &self.topic)
            .field("request_topic", &self.request_topic)
            .field("target", &self.target)
            .field("subscribed", &self.receiver.is_some())
            .field("published_initial", &self.published_initial)
            .finish()
    }
}

#[derive(Debug, Serialize, Deserialize)]
struct SetInstanceTrackTargetConfig {
    #[serde(default = "default_topic")]
    topic: String,
    #[serde(default = "default_request_topic")]
    request_topic: String,
    #[serde(default)]
    bbox: Option<BoundingBox>,
}

impl SetInstanceTrackTargetNode {
    /// Create a new SetInstanceTrackTargetNode from JSON configuration. See `SetRlTargetNode::from_json`'s
    /// doc comment for why `null` is normalized to `{}` here.
    pub fn from_json(config: Value) -> anyhow::Result<Self> {
        let config = if config.is_null() {
            serde_json::json!({})
        } else {
            config
        };
        let cfg: SetInstanceTrackTargetConfig = serde_json::from_value(config)?;
        Ok(Self {
            topic: cfg.topic,
            request_topic: cfg.request_topic,
            target: cfg.bbox,
            receiver: None,
            published_initial: false,
        })
    }

    async fn publish_set_target(&self, ctx: &Context, bbox: BoundingBox) -> anyhow::Result<()> {
        let command = InstanceTrackControlCommand {
            action: "set_target".to_string(),
            params: Some(serde_json::json!({ "bbox": bbox })),
        };

        let topic_static: &'static str = Box::leak(self.topic.clone().into_boxed_str());
        let topic = Topic::<InstanceTrackControlCommand>::new(topic_static);

        ctx.publish_to(topic, &command).await?;
        tracing::debug!("SetInstanceTrackTargetNode: published set_target({:?})", bbox);
        Ok(())
    }

    async fn publish_clear(&self, ctx: &Context) -> anyhow::Result<()> {
        let command = InstanceTrackControlCommand {
            action: "clear".to_string(),
            params: None,
        };

        let topic_static: &'static str = Box::leak(self.topic.clone().into_boxed_str());
        let topic = Topic::<InstanceTrackControlCommand>::new(topic_static);

        ctx.publish_to(topic, &command).await?;
        tracing::debug!("SetInstanceTrackTargetNode: published clear");
        Ok(())
    }

    /// Drains every pending request off `receiver`, returning the last one seen (if any) - only
    /// the most recent request matters, same as `SetRlTargetNode::poll_latest_request`. A
    /// request with `bbox: null` (or no `bbox` key at all) means "clear".
    fn poll_latest_request(receiver: &mut Receiver<Value>) -> Option<Option<BoundingBox>> {
        let mut latest = None;
        while let Ok(value) = receiver.try_recv() {
            match serde_json::from_value::<InstanceTrackTargetRequest>(value) {
                Ok(request) => latest = Some(request.bbox),
                Err(e) => tracing::warn!("SetInstanceTrackTargetNode: dropping malformed target request: {}", e),
            }
        }
        latest
    }
}

#[async_trait]
impl BehaviorNode for SetInstanceTrackTargetNode {
    async fn tick(&mut self, ctx: &Context) -> anyhow::Result<NodeStatus> {
        if self.receiver.is_none() {
            self.receiver = Some(ctx.subscribe_raw::<Value>(&self.request_topic).await?);
        }

        if !self.published_initial {
            self.published_initial = true;
            if let Some(bbox) = self.target {
                self.publish_set_target(ctx, bbox).await?;
            }
        }

        let latest = Self::poll_latest_request(self.receiver.as_mut().expect("just subscribed above"));
        if let Some(bbox) = latest {
            self.target = bbox;
            match bbox {
                Some(bbox) => self.publish_set_target(ctx, bbox).await?,
                None => self.publish_clear(ctx).await?,
            }
        }

        Ok(if self.target.is_some() {
            NodeStatus::Running
        } else {
            NodeStatus::Failure
        })
    }

    fn name(&self) -> &str {
        "SetInstanceTrackTargetNode"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_helpers::create_test_context;

    fn sample_bbox() -> BoundingBox {
        BoundingBox {
            x: 0.1,
            y: 0.2,
            width: 0.3,
            height: 0.4,
        }
    }

    #[test]
    fn from_json_reads_topic_request_topic_and_bbox() {
        let node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
            "topic": "/instance-track/control",
            "request_topic": "/instance-track/target/request",
            "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 }
        }))
        .unwrap();

        assert_eq!(node.topic, "/instance-track/control");
        assert_eq!(node.request_topic, "/instance-track/target/request");
        assert_eq!(node.target, Some(sample_bbox()));
    }

    #[test]
    fn from_json_defaults_topics_and_allows_no_target() {
        let node = SetInstanceTrackTargetNode::from_json(serde_json::json!({})).unwrap();

        assert_eq!(node.topic, "/instance-track/control");
        assert_eq!(node.request_topic, "/instance-track/target/request");
        assert_eq!(node.target, None);
    }

    #[test]
    fn from_json_accepts_null_the_same_as_an_empty_object() {
        let node = SetInstanceTrackTargetNode::from_json(Value::Null).unwrap();

        assert_eq!(node.topic, "/instance-track/control");
        assert_eq!(node.request_topic, "/instance-track/target/request");
        assert_eq!(node.target, None);
    }

    #[test]
    fn poll_latest_request_returns_none_when_empty() {
        let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        assert_eq!(SetInstanceTrackTargetNode::poll_latest_request(&mut receiver), None);
    }

    #[test]
    fn poll_latest_request_keeps_only_the_most_recent_value() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        tx.send(serde_json::json!({ "bbox": { "x": 0.0, "y": 0.0, "width": 0.1, "height": 0.1 } }))
            .unwrap();
        tx.send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
            .unwrap();

        assert_eq!(
            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
            Some(Some(sample_bbox()))
        );
    }

    #[test]
    fn poll_latest_request_accepts_a_null_bbox_as_a_clear_request() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        tx.send(serde_json::json!({ "bbox": null })).unwrap();

        assert_eq!(
            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
            Some(None)
        );
    }

    #[test]
    fn poll_latest_request_accepts_an_absent_bbox_key_as_a_clear_request() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        tx.send(serde_json::json!({})).unwrap();

        assert_eq!(
            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
            Some(None)
        );
    }

    #[test]
    fn poll_latest_request_drops_malformed_requests_and_keeps_polling() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let mut receiver = Receiver::from(rx);
        tx.send(serde_json::json!({ "bbox": "not-a-bbox" })).unwrap();
        tx.send(serde_json::json!({ "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 } }))
            .unwrap();

        assert_eq!(
            SetInstanceTrackTargetNode::poll_latest_request(&mut receiver),
            Some(Some(sample_bbox()))
        );
    }

    #[tokio::test]
    #[ignore]
    async fn tick_publishes_configured_target_once_then_stays_running() {
        let ctx = create_test_context("test_set_instance_track_target").await;

        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
            "topic": "/instance-track/control",
            "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 }
        }))
        .unwrap();

        let mut receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();

        let status = node.tick(&ctx).await.unwrap();
        assert_eq!(status, NodeStatus::Running);

        let received = tokio::time::timeout(std::time::Duration::from_millis(500), receiver.recv())
            .await
            .expect("timed out waiting for published command")
            .expect("channel closed without a message");

        assert_eq!(received["action"], "set_target");
        assert_eq!(received["params"]["bbox"]["x"], 0.1);

        let status = node.tick(&ctx).await.unwrap();
        assert_eq!(status, NodeStatus::Running);
        let timed_out = tokio::time::timeout(std::time::Duration::from_millis(200), receiver.recv()).await;
        assert!(timed_out.is_err(), "expected no second publish without a new request");
    }

    #[tokio::test]
    #[ignore]
    async fn tick_relays_a_live_request_published_on_request_topic() {
        let ctx = create_test_context("test_set_instance_track_target_request").await;

        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
            "topic": "/instance-track/control",
            "request_topic": "/instance-track/target/request"
        }))
        .unwrap();

        node.tick(&ctx).await.unwrap();

        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
        ctx.publish_raw(
            "/instance-track/target/request",
            &serde_json::json!({ "bbox": { "x": 0.5, "y": 0.5, "width": 0.1, "height": 0.1 } }),
        )
        .await
        .unwrap();

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        node.tick(&ctx).await.unwrap();

        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
            .await
            .expect("timed out waiting for relayed command")
            .expect("channel closed without a message");

        assert_eq!(received["action"], "set_target");
        assert_eq!(received["params"]["bbox"]["x"], 0.5);
        assert_eq!(
            node.tick(&ctx).await.unwrap(),
            NodeStatus::Running,
            "a live target must report Running"
        );
    }

    #[tokio::test]
    #[ignore = "Requires Redis infrastructure"]
    async fn tick_returns_failure_when_no_target_has_ever_been_set() {
        let ctx = create_test_context("test_set_instance_track_target_idle").await;

        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
            "request_topic": "/instance-track/target/request"
        }))
        .unwrap();

        let status = node.tick(&ctx).await.unwrap();
        assert_eq!(
            status,
            NodeStatus::Failure,
            "no target yet means cede control, not Success"
        );
    }

    #[tokio::test]
    #[ignore = "Requires Redis infrastructure"]
    async fn tick_returns_failure_after_a_clear_request_even_with_a_configured_initial_target() {
        let ctx = create_test_context("test_set_instance_track_target_cleared").await;

        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
            "request_topic": "/instance-track/target/request",
            "bbox": { "x": 0.1, "y": 0.2, "width": 0.3, "height": 0.4 }
        }))
        .unwrap();

        assert_eq!(node.tick(&ctx).await.unwrap(), NodeStatus::Running);

        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
            .await
            .unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        assert_eq!(node.tick(&ctx).await.unwrap(), NodeStatus::Failure);
    }

    #[tokio::test]
    #[ignore]
    async fn tick_relays_a_clear_request_published_on_request_topic() {
        let ctx = create_test_context("test_set_instance_track_target_clear").await;

        let mut node = SetInstanceTrackTargetNode::from_json(serde_json::json!({
            "topic": "/instance-track/control",
            "request_topic": "/instance-track/target/request"
        }))
        .unwrap();

        node.tick(&ctx).await.unwrap();

        let mut control_receiver = ctx.subscribe_raw::<Value>("/instance-track/control").await.unwrap();
        ctx.publish_raw("/instance-track/target/request", &serde_json::json!({ "bbox": null }))
            .await
            .unwrap();

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        node.tick(&ctx).await.unwrap();

        let received = tokio::time::timeout(std::time::Duration::from_millis(500), control_receiver.recv())
            .await
            .expect("timed out waiting for relayed command")
            .expect("channel closed without a message");

        assert_eq!(received["action"], "clear");
    }
}