//! Operator input from the swarm UI, taken straight off the swarm queue.
//!
//! swarm-controller sends each message on
//! [`swarm_queue_client::agent_input_subject`] for this agent. This task
//! subscribes to it and pushes each message into the serve loop's local
//! message channel, where it becomes a turn from the operator it names.
//! hive-c0re is not in the path, so the message has no broker row: the hive
//! dashboard's history and `recv` never see it, and `send` has no recipient
//! by the sender's name.
//!
//! **`from` is taken as given.** The controller is the only client granted
//! publish on the subject (`swarm-nats-auth`'s reader grant), and it names
//! the sender from the swarm gateway's authelia answer.
//!
//! **Core NATS, no reply.** A message sent while this task is not subscribed
//! is gone; the controller sends it as a request so the operator gets a 503
//! for that instead. No agent grant covers a reply subject, so nothing here
//! answers the request.

use futures_util::{Stream, StreamExt as _};
use hive_sh4re::inbox::DeliveredMessage;
use tokio::sync::mpsc::UnboundedSender;

use crate::swarm_queue::Connection;

/// Start the subscriber, if this agent has a queue credential and a label to
/// name its subject with.
pub fn spawn(tx: UnboundedSender<DeliveredMessage>) {
    if !crate::swarm_queue::configured() {
        return;
    }
    let agent = crate::identity::label();
    if agent.is_empty() {
        tracing::warn!("this agent has no label; not taking input from the swarm queue");
        return;
    }
    tokio::spawn(run(tx, agent));
}

async fn run(tx: UnboundedSender<DeliveredMessage>, agent: String) {
    let Some(Connection { client, .. }) = crate::swarm_queue::client().await else {
        return;
    };
    let subject = swarm_queue_client::agent_input_subject(&agent);
    let subscriber = match client.subscribe(subject.clone()).await {
        Ok(subscriber) => subscriber,
        Err(e) => {
            tracing::warn!(subject, error = %e, "swarm input: subscribe failed");
            return;
        }
    };
    tracing::info!(subject, "taking operator input from the swarm queue");
    forward(subscriber.map(|msg| msg.payload), &tx).await;
}

/// Push each decodable payload into `tx` until either end closes.
async fn forward<P: AsRef<[u8]>>(
    payloads: impl Stream<Item = P>,
    tx: &UnboundedSender<DeliveredMessage>,
) {
    let mut payloads = std::pin::pin!(payloads);
    while let Some(payload) = payloads.next().await {
        let Some(dm) = delivered(payload.as_ref()) else {
            continue;
        };
        if tx.send(dm).is_err() {
            // The serve loop is gone: the harness is shutting down.
            return;
        }
    }
}

/// The message a payload becomes, or `None` (logged) for one that is not an
/// [`swarm_queue_client::AgentInput`].
fn delivered(payload: &[u8]) -> Option<DeliveredMessage> {
    match serde_json::from_slice::<swarm_queue_client::AgentInput>(payload) {
        Ok(input) => {
            tracing::info!(from = %input.from, "swarm input: message received");
            Some(DeliveredMessage {
                from: input.from,
                body: input.body,
                id: 0,
                redelivered: false,
                in_reply_to: None,
            })
        }
        Err(e) => {
            tracing::warn!(error = %e, "swarm input: undecodable message dropped");
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use super::forward;
    use swarm_queue_client::AgentInput;

    fn payload(from: &str, body: &str) -> Vec<u8> {
        serde_json::to_vec(&AgentInput {
            from: from.to_owned(),
            body: body.to_owned(),
        })
        .expect("encodes")
    }

    #[tokio::test]
    async fn each_message_reaches_the_serve_loop_as_from_its_sender() {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let payloads = vec![
            payload("mara", "first"),
            b"not json".to_vec(),
            payload("argus", "second"),
        ];
        forward(futures_util::stream::iter(payloads), &tx).await;
        drop(tx);

        let mut got = Vec::new();
        while let Some(dm) = rx.recv().await {
            got.push(dm);
        }
        // The undecodable one is dropped, the ones either side of it are not.
        assert_eq!(got.len(), 2, "{got:?}");
        assert_eq!(
            (got[0].from.as_str(), got[0].body.as_str()),
            ("mara", "first")
        );
        assert_eq!(
            (got[1].from.as_str(), got[1].body.as_str()),
            ("argus", "second")
        );
        // No broker row behind it, so no `[msg #id]` tag in the prompt.
        assert!(got.iter().all(|dm| dm.id == 0 && dm.in_reply_to.is_none()));
    }

    #[tokio::test]
    async fn a_closed_serve_loop_stops_the_forwarder() {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        drop(rx);
        // An endless stream: returning at all proves the closed channel ended it.
        let payloads = futures_util::stream::repeat(payload("mara", "hi"));
        forward(payloads, &tx).await;
    }
}