//! `GET /api/messages/stream` — an SSE feed of every message between agents,
//! and from non-agent senders to agents, read from the agents' terminal rows
//! on the swarm queue. No hive is involved.
//!
//! **Two kinds of row make an event.** An agent's `send` call leaves a
//! [`SEND_ICON`] row on its own terminal; the sender is that terminal's
//! agent, the recipient is in the summary. A turn started by a message leaves
//! a [`TURN_ICON`] row on the receiver's terminal; it makes an event only when
//! the sender is not in the roster (`reminder`, `compact`, the operator), since
//! an agent sender's `send` row already made one.
//!
//! **Not every message appears.** Messages an agent drains with an in-turn
//! `recv` start no turn and leave no `TURN ←` row, and a `send` row exists
//! only while the sender's terminal publisher was up. A sender added to the
//! roster after the connection opened counts as a non-agent until the next
//! connection, so its messages show twice.
//!
//! **History, then live, with no gap.** The live subscription on
//! `$SWARM.term.*` (one token, so subagent subjects are excluded) opens first.
//! Then each roster agent's stream gives its newest
//! [`agent_term::REPLAY_ROWS`] rows, which are filtered by `since_minutes`,
//! merged by `ts` and sent; a live event equal to a history event is dropped.
//! History is bounded by those 100 rows per agent, not only by `since_minutes`.

use std::collections::HashSet;
use std::convert::Infallible;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use axum::extract::rejection::QueryRejection;
use axum::extract::{Query, State};
use axum::http::StatusCode;
use axum::response::sse::{Event, KeepAlive, Sse};
use futures_util::{Stream, StreamExt as _, stream};
use hive_types::term_msg::{SEND_ICON, TURN_ICON, send_recipient, turn_sender};
use serde::Deserialize;
use swarm_controller_api::messages::SwarmMessage;
use swarm_wire::agent_term;
use utoipa::IntoParams;

use crate::routes::{AppState, error_problem};

/// `since_minutes` when the query names none.
const DEFAULT_SINCE_MINUTES: u64 = 60;

/// The largest `since_minutes`: the stream's own retention.
const MAX_SINCE_MINUTES: u64 = 24 * 60;

/// How long a stream that has delivered its rows stays quiet before the
/// replay counts as caught up.
const HISTORY_IDLE: Duration = Duration::from_millis(500);

/// How many agents' histories are read at once.
const HISTORY_CONCURRENCY: usize = 16;

/// `stream_messages`'s query.
#[derive(Deserialize, IntoParams)]
pub struct MessagesQuery {
    /// How many minutes of history to send before the live tail. Default
    /// 60, at most 1440; 0 sends live messages only.
    since_minutes: Option<u64>,
}

#[utoipa::path(
    get,
    path = "/api/messages/stream",
    params(MessagesQuery),
    responses(
        (status = 200, description = "server-sent event stream; each event's `data` is \
            one `SwarmMessage` as JSON: first the recent history, oldest first, then \
            live messages. Built from agents' terminal rows, so a message an agent \
            drained with an in-turn `recv` is missing",
            body = SwarmMessage, content_type = "text/event-stream"),
        (status = 400, description = "`since_minutes` is not a number (problem+json)",
            body = String),
        (status = 503, description = "no swarm queue or identity bridge is configured on \
            this host, the roster could not be read, or the queue could not be reached \
            (problem+json)", body = String),
        (status = 500, description = "the subscribe failed (problem+json)", body = String),
    ),
    tag = "agents"
)]
pub(crate) async fn stream_messages(
    State(state): State<AppState>,
    query: Result<Query<MessagesQuery>, QueryRejection>,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, problem_details::ProblemDetails> {
    let Query(query) = query.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?;
    let minutes = query
        .since_minutes
        .unwrap_or(DEFAULT_SINCE_MINUTES)
        .min(MAX_SINCE_MINUTES);
    let Some(status) = state.status.as_ref() else {
        return Err(error_problem(
            StatusCode::SERVICE_UNAVAILABLE,
            "no swarm queue is configured on this host",
        ));
    };
    let roster = crate::agent_input::roster(&state).await?;
    let client = status.queue_client();
    swarm_queue_client::ensure_connected(&client).map_err(|e| {
        error_problem(
            StatusCode::SERVICE_UNAVAILABLE,
            &swarm_queue_client::chain(&e),
        )
    })?;

    let subscriber =
        crate::term_stream::subscribe(&client, swarm_wire::term_subject_wildcard(), "messages")
            .await?;
    let roster: Arc<HashSet<String>> = Arc::new(roster.into_iter().collect());
    let history = history(&client, &roster, minutes).await;
    let seen: HashSet<SwarmMessage> = history.iter().cloned().collect();
    let live = live(subscriber, roster).filter(move |m| std::future::ready(!seen.contains(m)));
    let events = stream::iter(history)
        .chain(live)
        .map(|m| Ok(message_event(&m)));
    Ok(Sse::new(events).keep_alive(KeepAlive::default()))
}

fn message_event(message: &SwarmMessage) -> Event {
    // A struct of strings always serializes.
    Event::default().data(serde_json::to_string(message).unwrap_or_default())
}

/// A unix-seconds stamp as ISO 8601 / RFC 3339 UTC, the spelling a row's `ts`
/// carries, so the two compare as strings. It must equal
/// `hive_sh4re::term_msg::iso8601_utc`, which stamps those rows.
///
/// A stamp outside the representable range renders as the epoch.
fn iso8601_utc(unix_seconds: i64) -> String {
    chrono::DateTime::from_timestamp(unix_seconds, 0)
        .unwrap_or_default()
        .to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
}

/// The messages in the newest rows of each of `roster`'s streams from the last
/// `minutes`, oldest first. An agent whose stream cannot be read adds none.
async fn history(
    client: &async_nats::Client,
    roster: &Arc<HashSet<String>>,
    minutes: u64,
) -> Vec<SwarmMessage> {
    if minutes == 0 {
        return Vec::new();
    }
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX));
    let cutoff = iso8601_utc(now.saturating_sub(i64::try_from(minutes * 60).unwrap_or(i64::MAX)));
    let mut messages: Vec<SwarmMessage> = stream::iter(roster.iter().cloned().collect::<Vec<_>>())
        .map(|agent| agent_history(client.clone(), agent, roster.clone()))
        .buffer_unordered(HISTORY_CONCURRENCY)
        .flat_map(stream::iter)
        .filter(|m| std::future::ready(m.ts >= cutoff))
        .collect()
        .await;
    messages.sort_by(|a, b| a.ts.cmp(&b.ts));
    messages
}

/// The messages in `agent`'s replayed rows, read until the stream goes quiet.
async fn agent_history(
    client: async_nats::Client,
    agent: String,
    roster: Arc<HashSet<String>>,
) -> Vec<SwarmMessage> {
    let replayed = crate::term_stream::replay(
        &client,
        &agent_term::stream_name(&agent),
        swarm_wire::term_subject(&agent),
        None,
    )
    .await;
    let mut rows = match replayed {
        Ok(rows) => rows,
        Err(error) => {
            tracing::info!(%agent, %error, "messages stream: no history for this agent");
            return Vec::new();
        }
    };
    let mut messages = Vec::new();
    while let Ok(Some(row)) = tokio::time::timeout(HISTORY_IDLE, rows.next()).await {
        messages.extend(message_of(&agent, &row.payload, |a| roster.contains(a)));
    }
    messages
}

/// The messages in the rows arriving on `$SWARM.term.<agent>` subjects.
fn live(
    subscriber: async_nats::Subscriber,
    roster: Arc<HashSet<String>>,
) -> impl Stream<Item = SwarmMessage> {
    subscriber.filter_map(move |msg| {
        let message = agent_of(msg.subject.as_str())
            .and_then(|agent| message_of(agent, &msg.payload, |a| roster.contains(a)));
        std::future::ready(message)
    })
}

/// The agent a main terminal subject belongs to. A subagent's subject has
/// more tokens and names none.
fn agent_of(subject: &str) -> Option<&str> {
    let agent = subject.strip_prefix(&swarm_wire::term_subject(""))?;
    (!agent.is_empty() && !agent.contains('.')).then_some(agent)
}

/// The fields of a terminal row this feed reads.
#[derive(Deserialize)]
struct RowView {
    #[serde(default)]
    ts: String,
    icon: Option<String>,
    #[serde(default)]
    summary: String,
    body: Option<String>,
    body_format: Option<String>,
}

/// The message in `agent`'s terminal row `payload`, if it holds one.
/// `is_agent` says whether a turn's sender is an agent.
fn message_of(
    agent: &str,
    payload: &[u8],
    is_agent: impl Fn(&str) -> bool,
) -> Option<SwarmMessage> {
    let row: RowView = serde_json::from_slice(payload).ok()?;
    let (from, to) = match row.icon.as_deref()? {
        SEND_ICON => (
            agent,
            send_recipient(&row.summary).filter(|to| !to.is_empty())?,
        ),
        TURN_ICON => {
            let sender = turn_sender(&row.summary).filter(|s| !s.is_empty() && !is_agent(s))?;
            (sender, agent)
        }
        _ => return None,
    };
    Some(SwarmMessage {
        ts: row.ts,
        from: from.to_owned(),
        to: to.to_owned(),
        body: row.body.unwrap_or_default(),
        body_format: row.body_format,
    })
}

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

    use serde_json::json;

    use crate::test_nats::TestNats;

    /// The spelling is the one the agents stamp rows with (RFC 3339, whole
    /// seconds, `Z`): the history cutoff is compared with `ts` as a string.
    #[test]
    fn iso8601_utc_spells_a_stamp_the_way_a_row_ts_does() {
        assert_eq!(iso8601_utc(0), "1970-01-01T00:00:00Z");
        assert_eq!(iso8601_utc(1_789_302_903), "2026-09-13T12:35:03Z");
    }

    #[test]
    fn iso8601_utc_renders_an_out_of_range_stamp_as_the_epoch() {
        assert_eq!(iso8601_utc(i64::MAX), "1970-01-01T00:00:00Z");
    }

    fn row(icon: &str, summary: &str, body: &str) -> bytes::Bytes {
        json!({
            "ts": "2026-10-18T12:00:00Z", "icon": icon, "level": "info",
            "summary": summary, "body": body, "body_format": "markdown",
        })
        .to_string()
        .into()
    }

    fn roster() -> Arc<HashSet<String>> {
        Arc::new(["atlas", "iris"].map(String::from).into())
    }

    async fn publish(client: &async_nats::Client, subject: String, payload: bytes::Bytes) {
        client.publish(subject, payload).await.expect("publish");
        client.flush().await.expect("flush");
    }

    async fn next(messages: &mut (impl Stream<Item = SwarmMessage> + Unpin)) -> SwarmMessage {
        tokio::time::timeout(Duration::from_secs(5), messages.next())
            .await
            .expect("a message arrived")
            .expect("the feed ended")
    }

    async fn nothing_more(messages: &mut (impl Stream<Item = SwarmMessage> + Unpin)) {
        if let Ok(Some(m)) = tokio::time::timeout(Duration::from_millis(500), messages.next()).await
        {
            panic!("an extra message: {m:?}");
        }
    }

    /// The last row published is a `send`, so reaching it proves the rows
    /// before it were read and dropped rather than not yet delivered.
    #[tokio::test]
    async fn live_rows_become_messages_and_the_rest_are_dropped() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let subscriber = client
            .subscribe(swarm_wire::term_subject_wildcard())
            .await
            .expect("subscribe");
        let mut messages = Box::pin(live(subscriber, roster()));
        let atlas = swarm_wire::term_subject("atlas");
        let iris = swarm_wire::term_subject("iris");

        publish(
            &client,
            atlas.clone(),
            row(SEND_ICON, "send* → iris · 3L", "hello"),
        )
        .await;
        let m = next(&mut messages).await;
        assert_eq!(
            (m.from.as_str(), m.to.as_str(), m.body.as_str()),
            ("atlas", "iris", "hello")
        );
        assert_eq!(m.ts, "2026-10-18T12:00:00Z");
        assert_eq!(m.body_format.as_deref(), Some("markdown"));

        // A reminder's turn is a message from `reminder` to the agent.
        publish(
            &client,
            iris.clone(),
            row(TURN_ICON, "TURN ← reminder", "stand up"),
        )
        .await;
        let m = next(&mut messages).await;
        assert_eq!(
            (m.from.as_str(), m.to.as_str(), m.body.as_str()),
            ("reminder", "iris", "stand up")
        );

        // A turn from an agent, a subagent subject, and other rows: no event.
        publish(
            &client,
            iris.clone(),
            row(TURN_ICON, "TURN ← atlas", "hello"),
        )
        .await;
        publish(
            &client,
            swarm_wire::subagent_term::subject("atlas", "scout"),
            row(SEND_ICON, "send* → iris", "from a subagent"),
        )
        .await;
        publish(&client, iris.clone(), row("🔧", "Bash", "ls")).await;
        publish(&client, iris, bytes::Bytes::from_static(b"not json")).await;
        publish(&client, atlas, row(SEND_ICON, "send* → iris", "last")).await;
        assert_eq!(next(&mut messages).await.body, "last");
        nothing_more(&mut messages).await;
    }

    #[tokio::test]
    async fn history_holds_only_rows_since_the_cutoff_oldest_first() {
        let nats = TestNats::start();
        let client = nats.client().await;
        for agent in ["atlas", "iris"] {
            swarm_queue_client::agent_term::open_or_create(&client, agent)
                .await
                .expect("the stream");
        }
        let stamped = |ts: &str, summary: &str, body: &str| -> bytes::Bytes {
            json!({"ts": ts, "icon": SEND_ICON, "level": "info", "summary": summary, "body": body})
                .to_string()
                .into()
        };
        let secs = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("after the epoch")
            .as_secs();
        let now = iso8601_utc(i64::try_from(secs).expect("fits"));
        let earlier = iso8601_utc(i64::try_from(secs - 600).expect("fits"));
        let atlas = swarm_wire::term_subject("atlas");
        let iris = swarm_wire::term_subject("iris");
        publish(&client, atlas, stamped(&now, "send* → iris", "newer")).await;
        publish(
            &client,
            iris.clone(),
            stamped(&earlier, "send* → atlas", "older"),
        )
        .await;
        publish(
            &client,
            iris,
            stamped("2000-01-01T00:00:00Z", "send* → atlas", "ancient"),
        )
        .await;
        let js = async_nats::jetstream::new(client.clone());
        for (agent, want) in [("atlas", 1), ("iris", 2)] {
            for _ in 0..100 {
                let mut s = js
                    .get_stream(agent_term::stream_name(agent))
                    .await
                    .expect("the stream");
                if s.info().await.expect("info").state.messages >= want {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(50)).await;
            }
        }
        let got = history(&client, &roster(), 60).await;
        assert_eq!(
            got.iter().map(|m| m.body.as_str()).collect::<Vec<_>>(),
            ["older", "newer"]
        );
        // No window reads no history.
        assert!(history(&client, &roster(), 0).await.is_empty());
    }

    #[test]
    fn a_row_without_a_party_makes_no_message() {
        let none = |_: &str| false;
        assert!(message_of("atlas", &row(SEND_ICON, "send* → ", "x"), none).is_none());
        assert!(message_of("atlas", &row(TURN_ICON, "TURN ← ", "x"), none).is_none());
        assert!(message_of("atlas", &row(SEND_ICON, "send* → iris", "x"), none).is_some());
    }

    #[test]
    fn only_main_terminal_subjects_name_an_agent() {
        assert_eq!(agent_of("$SWARM.term.atlas"), Some("atlas"));
        assert_eq!(agent_of("$SWARM.term.atlas.sub.scout"), None);
        assert_eq!(agent_of("$SWARM.term."), None);
    }
}