//! Swarm-wide view of what each hive last said about itself.
//!
//! Hives **offer** a snapshot upward; this daemon never reaches down to
//! collect one. That direction is the design, not an implementation
//! detail: the hive gateway has gone down in a way where every recovery
//! channel ran through the one broken thing, so a status path that
//! depended on the controller would have gone dark exactly when it was
//! needed to diagnose the controller's own network. A hive computes its
//! status locally regardless of whether the swarm can be reached.
//!
//! **The queue is the store.** A hive publishes into a `JetStream` KV
//! bucket, which retains the last value per key; this daemon reads that
//! bucket per request and keeps no copy. A cache here would be a second
//! answer to the same question, free to disagree with the first — and
//! the disagreement would surface as a hive reading healthy on a
//! dashboard while the bucket says otherwise.
//!
//! **Absence is the case this is built around** — the freshness states
//! and the reasoning behind each are in `docs/swarm/README.md`. The two
//! properties that constrain the code rather than describe it:
//! freshness is **derived at read time**, never stored (a stored
//! `healthy: bool` goes stale silently the moment nothing arrives), and
//! rows come from the **roster**, not the bucket, so an empty bucket
//! cannot render as a healthy swarm.
//!
//! One consequence worth stating because it is the opposite of what a
//! cache would give: losing the bucket degrades **to honesty**. Every
//! hive reads `never_reported` until its next publish, which is the true
//! answer — not a remembered "healthy" from before the loss.

use std::collections::BTreeMap;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use anyhow::{Context, Result};
use futures_util::TryStreamExt as _;
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;

use crate::HiveEntry;

/// The KV bucket hives publish their snapshots into.
///
/// A constant and not an option: reader and writer must name the same
/// bucket, and an option is a way for two deployments to disagree about
/// which one that is. Nothing about a bucket name is site-specific.
pub const BUCKET: &str = "hive-status";

/// Default age past which a snapshot is reported stale.
///
/// A threshold is a statement about how often hives offer, and that
/// cadence is decided by the publisher (a later slice), so this is a
/// default to be overridden rather than a constant to be relied on.
pub const DEFAULT_STALE_AFTER: Duration = Duration::from_mins(2);

/// Env var the NixOS module sets from
/// `services.hyperhive.swarm.controller.staleAfterSeconds`.
pub const STALE_AFTER_ENV: &str = "SWARM_CONTROLLER_STALE_AFTER_SECS";

/// How a hive's last report reads *now* — a function of the clock, not a
/// property of the report.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum Freshness {
    /// Reported within the staleness threshold.
    Fresh,
    /// Reported, but longer ago than the threshold. The payload is still
    /// rendered: "old" and "absent" are different answers and a consumer
    /// may want the last thing a hive managed to say.
    Stale,
    /// In the roster, has never offered a snapshot. Distinct from
    /// `Stale` because it separates "went quiet" from "never spoke" —
    /// the first is a fault, the second is usually a deployment that
    /// hasn't happened yet.
    NeverReported,
    /// Offered a snapshot but is not in the roster. Not an error this
    /// daemon can resolve, and not one it should hide.
    Unknown,
}

/// One row of the aggregate.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)]
pub struct HiveStatus {
    pub name: String,
    /// From the roster; `None` for a hive the roster doesn't list.
    pub domain: Option<String>,
    pub freshness: Freshness,
    /// When the snapshot was stored, unix seconds. `None` when nothing
    /// has been.
    ///
    /// This is the **bucket's** stamp, applied by the NATS server when
    /// the value landed — not a field inside the payload. A publisher
    /// therefore cannot make itself look fresher than it is, and a hive
    /// with a wrong clock skews its own payload rather than this.
    pub last_seen_unix: Option<i64>,
    /// Age at render time. Carried alongside `last_seen_unix` so a
    /// consumer with a different threshold need not re-derive it from a
    /// clock that may not match this host's.
    pub age_seconds: Option<u64>,
    /// Whatever the hive published, unopened — stored opaquely so the
    /// snapshot's contents stay settleable later without reworking the
    /// aggregate.
    ///
    /// `None` in two cases the consumer can tell apart by `freshness`:
    /// nothing has ever been published (`never_reported`), or something
    /// was published that is not JSON (any other freshness — the row
    /// still reports *when* the hive last spoke, and the read logs a
    /// warning naming it).
    pub snapshot: Option<serde_json::Value>,
}

/// A snapshot as retained by the bucket.
#[derive(Clone, Debug)]
struct Offered {
    received_at: SystemTime,
    /// `None` when the stored bytes are not JSON — see [`HiveStatus::snapshot`].
    payload: Option<serde_json::Value>,
}

/// Reads the aggregate out of the KV bucket.
///
/// Holds a NATS client rather than a bucket handle: the bucket is
/// resolved on first use and cached, so a controller that starts before
/// the bucket exists picks it up without a restart. Resolution failures
/// are not cached — [`tokio::sync::OnceCell::get_or_try_init`] retries —
/// which is what makes the queue coming up *after* this daemon a
/// non-event rather than a permanent degradation.
pub struct StatusReader {
    client: async_nats::Client,
    store: tokio::sync::OnceCell<async_nats::jetstream::kv::Store>,
    stale_after: Duration,
}

impl StatusReader {
    #[must_use]
    pub fn new(client: async_nats::Client, stale_after: Duration) -> Self {
        Self {
            client,
            store: tokio::sync::OnceCell::new(),
            stale_after,
        }
    }

    /// Reads [`STALE_AFTER_ENV`], falling back to
    /// [`DEFAULT_STALE_AFTER`]. A zero or unparseable value takes the
    /// default rather than failing startup — same rule as `load_hives`:
    /// a controller whose own config is wrong must still serve.
    #[must_use]
    pub fn stale_after_from_env() -> Duration {
        std::env::var(STALE_AFTER_ENV)
            .ok()
            .and_then(|raw| raw.trim().parse::<u64>().ok())
            .filter(|secs| *secs > 0)
            .map_or(DEFAULT_STALE_AFTER, Duration::from_secs)
    }

    /// The bucket handle, created on first use if nothing has made it yet.
    ///
    /// Whichever side arrives first creates it, and both sides want the
    /// same shape, so this is a race with one outcome. `history: 1` is
    /// the shape: the aggregate reads *the last thing each hive said*,
    /// and retaining more would be storage bought for a query nobody
    /// makes.
    async fn store(&self) -> Result<&async_nats::jetstream::kv::Store> {
        self.store
            .get_or_try_init(|| async {
                let js = async_nats::jetstream::new(self.client.clone());
                match js.get_key_value(BUCKET).await {
                    Ok(store) => Ok(store),
                    Err(e) => {
                        tracing::info!(
                            bucket = BUCKET,
                            reason = %e,
                            "status bucket not available, creating it"
                        );
                        js.create_key_value(async_nats::jetstream::kv::Config {
                            bucket: BUCKET.to_owned(),
                            description: "Last status snapshot offered by each hive".to_owned(),
                            history: 1,
                            ..Default::default()
                        })
                        .await
                        .with_context(|| format!("creating the {BUCKET} bucket"))
                    }
                }
            })
            .await
    }

    /// The aggregate, rendered against `now`.
    ///
    /// Every roster hive produces a row whether or not it has ever
    /// reported; a reporting hive outside the roster produces one too.
    pub async fn view(&self, roster: &[HiveEntry], now: SystemTime) -> Result<Vec<HiveStatus>> {
        // Only a CONNECTED client can be asked anything. `retry_on_initial_connect`
        // means the client exists before it is usable, and a JetStream request
        // made in that window does not fail — it WAITS, on every call, for
        // longer than any dashboard poll should take (measured: still going at
        // 15s against a queue that simply refuses the credential).
        //
        // Testing for `!= Connected` rather than `== Disconnected` is the whole
        // point: a client that has never connected once sits in `Pending`, so
        // the `Disconnected` test passes it straight through to the hang it was
        // written to prevent. That is exactly the case here — a controller
        // whose credential is wrong from boot never reaches `Disconnected`,
        // because it was never connected to begin with.
        //
        // Naming the state is also the better error: "not connected" is
        // actionable, a timeout is not.
        let state = self.client.connection_state();
        if state != async_nats::connection::State::Connected {
            anyhow::bail!("not connected to the swarm queue (client state: {state:?})");
        }

        let store = self.store().await?;

        // Keys first, then a fetch per key. The roster is a handful of
        // hives, so the round-trip count is not worth trading for a
        // watcher whose "I have seen everything current" condition is
        // one more thing to get right on a read path.
        let mut keys = store.keys().await.context("listing status bucket keys")?;
        let mut entries: BTreeMap<String, Offered> = BTreeMap::new();
        while let Some(key) = keys
            .try_next()
            .await
            .context("reading the status bucket's key list")?
        {
            let Some(entry) = store
                .entry(&key)
                .await
                .with_context(|| format!("reading status entry {key}"))?
            else {
                // Deleted between listing and fetching. Not an error:
                // the next read simply won't list it.
                continue;
            };
            let payload = match serde_json::from_slice(&entry.value) {
                Ok(value) => Some(value),
                Err(e) => {
                    tracing::warn!(
                        hive = %key,
                        error = %e,
                        "status snapshot is not JSON; reporting the timestamp without it"
                    );
                    None
                }
            };
            entries.insert(
                key,
                Offered {
                    received_at: to_system_time(entry.created.unix_timestamp()),
                    payload,
                },
            );
        }

        Ok(render(roster, &entries, now, self.stale_after))
    }
}

/// A bucket timestamp as a [`SystemTime`].
///
/// A pre-epoch stamp is not representable here and is not a thing a NATS
/// server produces; treating it as the epoch renders the row as
/// extremely stale, which is the safe direction — a nonsense timestamp
/// must never read as fresh.
fn to_system_time(unix_seconds: i64) -> SystemTime {
    u64::try_from(unix_seconds).map_or(UNIX_EPOCH, |secs| UNIX_EPOCH + Duration::from_secs(secs))
}

/// Turn a roster plus whatever the bucket held into the rendered rows.
///
/// Split out of [`StatusReader::view`] deliberately: this is where every
/// rule the acceptance criterion cares about lives, and keeping it a
/// pure function means those rules are tested against a table rather
/// than against a running NATS server.
fn render(
    roster: &[HiveEntry],
    entries: &BTreeMap<String, Offered>,
    now: SystemTime,
    stale_after: Duration,
) -> Vec<HiveStatus> {
    let mut rows: Vec<HiveStatus> = roster
        .iter()
        .map(|hive| match entries.get(&hive.name) {
            Some(offered) => row(
                hive.name.clone(),
                Some(hive.domain.clone()),
                offered,
                now,
                stale_after,
            ),
            None => HiveStatus {
                name: hive.name.clone(),
                domain: Some(hive.domain.clone()),
                freshness: Freshness::NeverReported,
                last_seen_unix: None,
                age_seconds: None,
                snapshot: None,
            },
        })
        .collect();

    rows.extend(
        entries
            .iter()
            .filter(|(name, _)| !roster.iter().any(|hive| &&hive.name == name))
            .map(|(name, offered)| {
                let mut unknown = row(name.clone(), None, offered, now, stale_after);
                unknown.freshness = Freshness::Unknown;
                unknown
            }),
    );
    rows
}

fn row(
    name: String,
    domain: Option<String>,
    offered: &Offered,
    now: SystemTime,
    stale_after: Duration,
) -> HiveStatus {
    // A snapshot stamped in the future (clock skew between the NATS
    // server and this host) yields no age rather than a negative one,
    // and is treated as fresh — the honest reading of "this arrived, I
    // cannot tell how long ago".
    let age = now.duration_since(offered.received_at).ok();
    let freshness = match age {
        Some(age) if age > stale_after => Freshness::Stale,
        _ => Freshness::Fresh,
    };
    HiveStatus {
        name,
        domain,
        freshness,
        last_seen_unix: offered
            .received_at
            .duration_since(UNIX_EPOCH)
            .ok()
            .and_then(|d| i64::try_from(d.as_secs()).ok()),
        age_seconds: age.map(|age| age.as_secs()),
        snapshot: offered.payload.clone(),
    }
}

#[cfg(test)]
mod tests {
    use super::{DEFAULT_STALE_AFTER, Freshness, Offered, STALE_AFTER_ENV, StatusReader, render};
    use crate::HiveEntry;
    use std::collections::BTreeMap;
    use std::time::{Duration, SystemTime};

    fn roster() -> Vec<HiveEntry> {
        vec![
            HiveEntry {
                name: "pr1ma".to_owned(),
                domain: "pr1ma.example.com".to_owned(),
            },
            HiveEntry {
                name: "umbra".to_owned(),
                domain: "umbra.example.com".to_owned(),
            },
        ]
    }

    fn entries(rows: &[(&str, SystemTime)]) -> BTreeMap<String, Offered> {
        rows.iter()
            .map(|(name, at)| {
                (
                    (*name).to_owned(),
                    Offered {
                        received_at: *at,
                        payload: Some(serde_json::json!({ "ok": true })),
                    },
                )
            })
            .collect()
    }

    fn t0() -> SystemTime {
        SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000)
    }

    /// The whole point of the module: an empty bucket must not render as
    /// a healthy swarm. Rows come from the roster, so silence is visible.
    #[test]
    fn an_empty_bucket_renders_every_hive_as_never_reported() {
        let rows = render(&roster(), &BTreeMap::new(), t0(), DEFAULT_STALE_AFTER);
        assert_eq!(rows.len(), 2, "a row per roster hive, not per report");
        assert!(
            rows.iter().all(|r| r.freshness == Freshness::NeverReported),
            "nothing published means nothing known — not healthy"
        );
        assert!(rows.iter().all(|r| r.snapshot.is_none()));
    }

    /// Freshness is derived from the clock at read time, so the same
    /// stored value reads differently as it ages. The boundary is where
    /// an off-by-one would hide, so it is pinned in both directions.
    #[test]
    fn the_threshold_boundary_is_inclusive() {
        let stale_after = Duration::from_secs(90);
        let stored = entries(&[("pr1ma", t0())]);

        assert_eq!(
            render(
                &roster(),
                &stored,
                t0() + Duration::from_secs(90),
                stale_after
            )[0]
            .freshness,
            Freshness::Fresh,
            "exactly at the threshold is inside it"
        );
        assert_eq!(
            render(
                &roster(),
                &stored,
                t0() + Duration::from_secs(91),
                stale_after
            )[0]
            .freshness,
            Freshness::Stale,
            "one second past is outside it"
        );
    }

    /// One hive reporting must not make its silent neighbour look
    /// healthy — the failure mode of any aggregate that renders only
    /// what it has.
    #[test]
    fn a_reporting_hive_does_not_vouch_for_a_silent_one() {
        let rows = render(
            &roster(),
            &entries(&[("pr1ma", t0())]),
            t0(),
            DEFAULT_STALE_AFTER,
        );
        assert_eq!(rows[0].name, "pr1ma");
        assert_eq!(rows[0].freshness, Freshness::Fresh);
        assert_eq!(rows[1].name, "umbra");
        assert_eq!(rows[1].freshness, Freshness::NeverReported);
    }

    /// An observation the daemon cannot explain is surfaced, not dropped.
    #[test]
    fn a_hive_outside_the_roster_is_surfaced_as_unknown() {
        let rows = render(
            &roster(),
            &entries(&[("ghost", t0())]),
            t0(),
            DEFAULT_STALE_AFTER,
        );
        assert_eq!(rows.len(), 3, "two roster hives plus the stranger");
        let ghost = rows.last().expect("rows is non-empty");
        assert_eq!(ghost.name, "ghost");
        assert_eq!(ghost.freshness, Freshness::Unknown);
        assert!(
            ghost.domain.is_none(),
            "the roster is where a domain comes from, and this hive isn't in it"
        );
    }

    /// Clock skew must not produce a negative age or a panic. A snapshot
    /// stamped in the future reads fresh with no age — "it arrived, I
    /// cannot tell how long ago".
    #[test]
    fn a_future_timestamp_yields_no_age_rather_than_a_wrong_one() {
        let rows = render(
            &roster(),
            &entries(&[("pr1ma", t0() + Duration::from_secs(30))]),
            t0(),
            Duration::from_secs(90),
        );
        assert_eq!(rows[0].freshness, Freshness::Fresh);
        assert_eq!(rows[0].age_seconds, None);
    }

    /// A hive that published something unreadable still gets its
    /// timestamp reported: *when* it last spoke is exactly what this
    /// aggregate is for, and dropping the row would read as silence.
    #[test]
    fn an_unparseable_payload_still_reports_when_it_arrived() {
        let mut stored = BTreeMap::new();
        stored.insert(
            "pr1ma".to_owned(),
            Offered {
                received_at: t0(),
                payload: None,
            },
        );

        let row = &render(&roster(), &stored, t0(), DEFAULT_STALE_AFTER)[0];
        assert_eq!(
            row.freshness,
            Freshness::Fresh,
            "unreadable is not the same as absent — freshness is what \
             separates them on the wire"
        );
        assert!(row.snapshot.is_none());
        assert_eq!(row.last_seen_unix, Some(1_700_000_000));
    }

    /// SAFETY: single-threaded mutation of a process env var no other
    /// test in this crate reads; restored before returning. One test
    /// rather than four for the same reason `load_hives`'s is — the
    /// parallel runner would race them.
    #[test]
    fn stale_after_from_env_covers_missing_bogus_zero_and_valid() {
        unsafe {
            std::env::remove_var(STALE_AFTER_ENV);
        }
        assert_eq!(StatusReader::stale_after_from_env(), DEFAULT_STALE_AFTER);

        unsafe {
            std::env::set_var(STALE_AFTER_ENV, "not a number");
        }
        assert_eq!(
            StatusReader::stale_after_from_env(),
            DEFAULT_STALE_AFTER,
            "a controller whose own config is wrong must still serve"
        );

        unsafe {
            std::env::set_var(STALE_AFTER_ENV, "0");
        }
        assert_eq!(
            StatusReader::stale_after_from_env(),
            DEFAULT_STALE_AFTER,
            "zero would make every snapshot instantly stale — take the default"
        );

        unsafe {
            std::env::set_var(STALE_AFTER_ENV, "300");
        }
        assert_eq!(StatusReader::stale_after_from_env(), Duration::from_mins(5));

        unsafe {
            std::env::remove_var(STALE_AFTER_ENV);
        }
    }
}