//! Publishing each subagent's terminal rows onto the swarm queue, as this
//! agent.
//!
//! Every line a subagent writes passes through `session`'s sink, which hands
//! it to [`Publisher::offer`]. The sink runs synchronously on the subagent's
//! output reader, so `offer` only queues. One task drains the queue: it
//! classifies each `stream-json` event into the rows the agent's own terminal
//! carries (`hive_sh4re::stream_enrich`) and publishes them on
//! `$SWARM.term.<agent>.sub.<subagent>` (`swarm_queue_client::subagent_term`).
//!
//! **Output only.** Nothing here subscribes: a subagent takes no input from
//! the swarm.
//!
//! **As the agent, never as the hive.** The connection presents the agent's
//! own queue credential, read from the swarm secret store under the agent's
//! store identity: the same credential and identity the harness uses, since
//! both daemons run as the agent's user. The hive's shared client is never
//! tried, because the queue grants it nothing under these subjects.
//!
//! **Publish only.** The rows land in `term-sub-<agent>`, the stream
//! `swarm-controller` creates for this agent; this daemon never opens it. A
//! row published while that stream does not exist yet still reaches a live
//! subscriber, but the swarm does not list the subagent from it.
//!
//! **Best-effort.** A subagent's run never waits on the queue: no store, a
//! refused credential, a full queue and a failed publish are each a log
//! line, and the run goes on.

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use anyhow::{Context as _, anyhow};
use hive_sh4re::term_msg::{ClassifyCtx, Level, TermMsg, fit, iso8601_utc};
use swarm_queue_client::subagent_term;
use swarm_secret_client::{
    SecretStore,
    client::{DEFAULT_CERT_MOUNT, ENV_ADDR, ENV_CACERT, Settings},
    policy, queue,
};
use tokio::sync::mpsc;

/// Names the agent. The store role, the store path, the queue token and the
/// subjects are all built from it.
const ENV_AGENT_NAME: &str = "HIVE_AGENT_NAME";

/// Where the swarm queue listens, as this container reaches it.
const ENV_NATS_URL: &str = "HIVE_AGENT_NATS_URL";

/// Lines waiting for the publish task. Past this, `offer` drops the line
/// rather than block a subagent's output reader.
const QUEUE_DEPTH: usize = 4096;

/// How long one read of the store may take; it runs inside a queue
/// connection attempt.
const STORE_READ_TIMEOUT: Duration = Duration::from_secs(10);

/// How long after a failed connect the task tries again. Rows that arrive
/// before a first connect succeeds are dropped.
const RETRY_AFTER: Duration = Duration::from_mins(1);

/// Room left under the server's payload limit for subject, headers and
/// framing, as in `hive-agent`'s `swarm_term`.
const HEADROOM: usize = 1024;

/// One line of a subagent's output, as the runtime's sink hands it over.
pub enum Output {
    /// A `stream-json` event.
    Event(serde_json::Value),
    /// A stdout line that was not JSON.
    Stdout(String),
    /// A stderr line.
    Stderr(String),
}

struct Line {
    subagent: String,
    output: Output,
    /// Unix seconds when the sink saw the line.
    ts: i64,
}

/// The sink's handle on the publish task.
pub struct Publisher {
    tx: mpsc::Sender<Line>,
    dropped: Arc<AtomicU64>,
}

impl Publisher {
    /// Start the publish task, or return `None` when this daemon has no
    /// queue address or no store to read the agent's credential from. Logs
    /// which, once.
    #[must_use]
    pub fn spawn() -> Option<Self> {
        let target = match Target::from_lookup(|k| std::env::var(k).ok()) {
            Ok(Some(target)) => target,
            Ok(None) => {
                tracing::info!(
                    "no swarm queue address or no secret store for this agent; subagent \
                     terminals are not published"
                );
                return None;
            }
            Err(e) => {
                tracing::warn!(
                    error = format!("{e:#}"),
                    "this agent's secret store coordinates are incomplete; subagent \
                     terminals are not published"
                );
                return None;
            }
        };
        swarm_queue_client::install_crypto_provider();
        tracing::info!(
            url = %target.url,
            stream = %subagent_term::stream_name(&target.agent),
            "publishing subagent terminals to the swarm queue as this agent"
        );
        let (publisher, rx) = Self::channel(QUEUE_DEPTH);
        tokio::spawn(run(rx, target, Arc::clone(&publisher.dropped)));
        Some(publisher)
    }

    fn channel(depth: usize) -> (Self, mpsc::Receiver<Line>) {
        let (tx, rx) = mpsc::channel(depth);
        let publisher = Self {
            tx,
            dropped: Arc::new(AtomicU64::new(0)),
        };
        (publisher, rx)
    }

    /// Queue `output` from `subagent` for publishing. Never blocks: a full
    /// queue drops the line and counts it.
    pub fn offer(&self, subagent: &str, output: Output) {
        let ts = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX));
        let line = Line {
            subagent: subagent.to_owned(),
            output,
            ts,
        };
        if self.tx.try_send(line).is_err() {
            self.dropped.fetch_add(1, Ordering::Relaxed);
        }
    }
}

/// Where to publish, and what to connect with.
pub(crate) struct Target {
    pub(crate) url: String,
    pub(crate) agent: String,
    store: StoreSource,
}

/// This agent's queue secret as the store holds it.
struct StoreSource {
    settings: Settings,
    role: String,
    path: String,
}

impl Target {
    /// `None` when the queue address or the store address is absent, which
    /// is a container without a swarm queue or without a store.
    ///
    /// # Errors
    /// A store address without the agent name or the identity beside it.
    pub(crate) fn from_lookup(
        get: impl Fn(&str) -> Option<String>,
    ) -> anyhow::Result<Option<Self>> {
        let Some(url) = get(ENV_NATS_URL).filter(|v| !v.is_empty()) else {
            return Ok(None);
        };
        if get(ENV_ADDR).is_none_or(|v| v.is_empty()) {
            return Ok(None);
        }
        let agent = get(ENV_AGENT_NAME)
            .filter(|v| !v.is_empty())
            .with_context(|| format!("{ENV_AGENT_NAME} is unset or empty"))?;
        let settings =
            Settings::from_lookup(|k| get(k).filter(|v| k != ENV_CACERT || ca_is_usable(v)))
                .context("reading the swarm secret store's coordinates from the environment")?;
        let store = StoreSource {
            settings,
            role: policy::agent_object_name(&agent)?,
            path: queue::agent_queue_path(&agent)?,
        };
        Ok(Some(Self { url, agent, store }))
    }

    fn subject(&self, subagent: &str) -> String {
        subagent_term::subject(&self.agent, subagent)
    }
}

/// Whether the CA bundle at `path` is a file with bytes in it. Absent means
/// the container's own trust store.
fn ca_is_usable(path: &str) -> bool {
    std::fs::metadata(path).is_ok_and(|m| m.len() > 0)
}

impl StoreSource {
    /// The stored secret, read with a fresh login. Errors name the path and
    /// never the value.
    async fn read(&self) -> anyhow::Result<String> {
        let fetch = async {
            let store = SecretStore::connect(&self.settings, &self.role, DEFAULT_CERT_MOUNT)
                .await
                .context("logging in to the swarm secret store as this agent")?;
            let credential: Option<queue::AgentCredential> = store
                .read_optional(&self.path)
                .await
                .with_context(|| format!("reading {} from the store", self.path))?;
            credential
                .map(|c| c.value.trim().to_owned())
                .filter(|v| !v.is_empty())
                .with_context(|| format!("nothing is stored at {}", self.path))
        };
        tokio::time::timeout(STORE_READ_TIMEOUT, fetch)
            .await
            .map_err(|_| anyhow!("the store did not answer within {STORE_READ_TIMEOUT:?}"))?
    }
}

/// Connect presenting the agent's own token, read from the store on every
/// connection attempt so a re-minted secret is picked up on reconnect.
pub(crate) async fn connect(target: &Arc<Target>) -> anyhow::Result<async_nats::Client> {
    let url = target.url.clone();
    let target = Arc::clone(target);
    let agent = target.agent.clone();
    swarm_queue_client::connect_with_token(&url, None, &agent, move || {
        let target = Arc::clone(&target);
        // Spawned so the future handed back is only a `JoinHandle`, which is
        // `Sync` as async-nats's auth callback requires; the store read is not.
        let attempt = tokio::spawn(async move {
            let value = target.store.read().await.map_err(|e| format!("{e:#}"))?;
            swarm_queue_client::agent_token::format_agent_token(&target.agent, &value)
                .map_err(|e| e.to_string())
        });
        async move { attempt.await.map_err(|e| e.to_string())? }
    })
    .await
    .map_err(|e| anyhow!(swarm_queue_client::chain(&e)))
}

/// The connection, retried at most once per [`RETRY_AFTER`].
#[derive(Default)]
struct Queue {
    client: Option<async_nats::Client>,
    connect_after: Option<Instant>,
}

impl Queue {
    /// A client to publish on, connecting first when a connect is due.
    async fn ready(&mut self, target: &Arc<Target>) -> Option<async_nats::Client> {
        let now = Instant::now();
        if self.client.is_none() && self.connect_after.is_none_or(|t| now >= t) {
            match connect(target).await {
                Ok(client) => {
                    tracing::info!(url = %target.url, "subagent terminal: connected as this agent");
                    self.client = Some(client);
                }
                Err(e) => {
                    tracing::warn!(
                        error = format!("{e:#}"),
                        retry_in = ?RETRY_AFTER,
                        "subagent terminal: connecting with this agent's own queue credential \
                         failed; rows are dropped until a connect succeeds"
                    );
                    self.connect_after = Some(now + RETRY_AFTER);
                }
            }
        }
        self.client.clone()
    }
}

async fn run(mut rx: mpsc::Receiver<Line>, target: Target, dropped: Arc<AtomicU64>) {
    let target = Arc::new(target);
    let mut ctxs: HashMap<String, ClassifyCtx> = HashMap::new();
    let mut queue = Queue::default();
    while let Some(line) = rx.recv().await {
        let missed = dropped.swap(0, Ordering::Relaxed);
        if missed > 0 {
            tracing::warn!(missed, "subagent terminal: queue full, lines dropped");
        }
        let ctx = ctxs.entry(line.subagent.clone()).or_default();
        let rows = classify(&line.output, line.ts, ctx);
        if rows.is_empty() {
            continue;
        }
        let Some(client) = queue.ready(&target).await else {
            continue;
        };
        let subject = target.subject(&line.subagent);
        for row in rows {
            publish(&client, &subject, row).await;
        }
    }
}

/// The rows one line of output renders as, each stamped with `ts`.
fn classify(output: &Output, ts: i64, ctx: &mut ClassifyCtx) -> Vec<TermMsg> {
    let rows = match output {
        Output::Event(v) => hive_sh4re::stream_enrich::classify_stream_value(v, ctx),
        Output::Stdout(line) => vec![TermMsg::new(Level::Debug, line.clone())],
        Output::Stderr(line) => vec![TermMsg::new(Level::Warn, format!("stderr: {line}"))],
    };
    let ts = iso8601_utc(ts);
    rows.into_iter().map(|m| m.at(ts.as_str())).collect()
}

/// Offer one row. Every failure is terminal for that row and for nothing else.
async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) {
    // Unconnected, the client buffers rather than fails, and reports the
    // library's default payload limit rather than the server's.
    if let Err(e) = swarm_queue_client::ensure_connected(client) {
        tracing::warn!(error = %swarm_queue_client::chain(&e), "subagent terminal: publish skipped");
        return;
    }
    let limit = swarm_queue_client::max_payload(client).saturating_sub(HEADROOM);
    let summary = msg.summary.clone();
    let Some(msg) = fit(msg, limit) else {
        tracing::warn!(
            summary,
            limit,
            "subagent terminal: row does not fit even without its body, dropped"
        );
        return;
    };
    let payload = match serde_json::to_vec(&msg) {
        Ok(payload) => payload,
        Err(e) => {
            tracing::warn!(error = %e, "subagent terminal: serialising failed, row dropped");
            return;
        }
    };
    if let Err(e) = client.publish(subject.to_owned(), payload.into()).await {
        tracing::warn!(error = %e, "subagent terminal: publish failed, row dropped");
    }
}

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

    fn env(pairs: &[(&str, &str)]) -> impl Fn(&str) -> Option<String> {
        let pairs: Vec<(String, String)> = pairs
            .iter()
            .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
            .collect();
        move |k| pairs.iter().find(|(n, _)| n == k).map(|(_, v)| v.clone())
    }

    const STORE: [(&str, &str); 3] = [
        ("BAO_ADDR", "https://bao.t.local:8200"),
        ("BAO_CLIENT_CERT", "/run/credentials/c"),
        ("BAO_CLIENT_KEY", "/run/credentials/k"),
    ];

    fn full() -> Vec<(&'static str, &'static str)> {
        let mut pairs = STORE.to_vec();
        pairs.push((ENV_NATS_URL, "tls://nats.t.local:4222"));
        pairs.push((ENV_AGENT_NAME, "iris"));
        pairs
    }

    /// Each subagent publishes on its own subject under this agent's family,
    /// the one the queue grants the agent.
    #[test]
    fn a_subagent_publishes_under_the_agents_own_family() {
        let target = Target::from_lookup(env(&full()))
            .expect("complete")
            .expect("configured");
        assert_eq!(target.subject("scout"), "$SWARM.term.iris.sub.scout");
        assert_eq!(target.subject("w-2"), "$SWARM.term.iris.sub.w-2");
        assert_eq!(target.store.path, "swarm/agents/iris/queue");
    }

    #[test]
    fn without_a_queue_address_or_a_store_nothing_is_published() {
        let no_url: Vec<_> = full()
            .into_iter()
            .filter(|(k, _)| *k != ENV_NATS_URL)
            .collect();
        assert!(Target::from_lookup(env(&no_url)).expect("legal").is_none());
        let no_store: Vec<_> = full()
            .into_iter()
            .filter(|(k, _)| *k != "BAO_ADDR")
            .collect();
        assert!(
            Target::from_lookup(env(&no_store))
                .expect("legal")
                .is_none()
        );
    }

    #[test]
    fn a_store_without_the_agent_name_is_an_error() {
        let no_name: Vec<_> = full()
            .into_iter()
            .filter(|(k, _)| *k != ENV_AGENT_NAME)
            .collect();
        assert!(Target::from_lookup(env(&no_name)).is_err());
    }

    /// The sink must never wait on the publisher: a full queue drops the line
    /// and counts it.
    #[test]
    fn a_full_queue_drops_and_counts_rather_than_blocks() {
        let (publisher, _rx) = Publisher::channel(1);
        publisher.offer("scout", Output::Stdout("one".to_owned()));
        publisher.offer("scout", Output::Stdout("two".to_owned()));
        assert_eq!(publisher.dropped.load(Ordering::Relaxed), 1);
    }

    #[test]
    fn every_kind_of_line_renders_as_stamped_rows() {
        let mut ctx = ClassifyCtx::default();
        let ts = 1_789_302_903;
        let event = serde_json::json!({
            "type": "assistant",
            "message": { "content": [{ "type": "text", "text": "looking" }] }
        });
        let rows = classify(&Output::Event(event), ts, &mut ctx);
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].ts, "2026-09-13T12:35:03Z");

        let rows = classify(&Output::Stderr("boom".to_owned()), ts, &mut ctx);
        assert_eq!(rows[0].level, Level::Warn);
        assert_eq!(rows[0].summary, "stderr: boom");

        let rows = classify(&Output::Stdout("chatter".to_owned()), ts, &mut ctx);
        assert_eq!(rows[0].level, Level::Debug);
        assert_eq!(rows[0].ts, "2026-09-13T12:35:03Z");
    }
}