//! Whether this hive could pull the latest closure of each of its agents from
//! the swarm's binary cache, and the answer as a log line and a gauge.
//!
//! Every [`PASS_INTERVAL`] each agent's record in `agent-closures` is looked up
//! and its narinfo fetched with the pull token and checked against the cache's
//! public key. Nothing is downloaded here; a rebuild downloads through
//! [`super::closure`].
//!
//! The verdict is the first of these that fails, in this order:
//! no credentials (`no-credentials`), no record (`no-record`), a record built
//! from another `internal/meta` commit than the head that pins the agent
//! differently (`rev-mismatch`), a record
//! the builder did not push (`not-cached`), the cache not holding the narinfo
//! (`not-cached`), a request that failed or was refused (`fetch-failed`), a
//! narinfo no signature of the key covers (`bad-signature`).

use std::sync::Mutex;
use std::time::Duration;

use anyhow::{Context as _, Result};
use swarm_queue_client::agent_closures::{self, Read};
use swarm_wire::agent_closures::AgentClosure;

use super::narinfo;
use super::reader::{self, Facts};
use crate::stats::sweep_health::{self, SweepHealth};

/// How often every agent is probed.
pub const PASS_INTERVAL: Duration = Duration::from_mins(5);

/// At [`PASS_INTERVAL`] this is ~15 minutes of failing passes.
const FAILURES_BEFORE_WARNING: u32 = 3;

/// Why a closure cannot be pulled; the gauge's `reason` label.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reason {
    NoRecord,
    RevMismatch,
    NotCached,
    NoCredentials,
    FetchFailed,
    BadSignature,
}

impl Reason {
    /// The label value.
    pub const fn label(self) -> &'static str {
        match self {
            Reason::NoRecord => "no-record",
            Reason::RevMismatch => "rev-mismatch",
            Reason::NotCached => "not-cached",
            Reason::NoCredentials => "no-credentials",
            Reason::FetchFailed => "fetch-failed",
            Reason::BadSignature => "bad-signature",
        }
    }
}

/// A closure that cannot be pulled, and what the log line says about it.
#[derive(Debug)]
pub struct Unpullable {
    pub reason: Reason,
    pub detail: String,
}

fn no(reason: Reason, detail: impl Into<String>) -> Unpullable {
    Unpullable {
        reason,
        detail: detail.into(),
    }
}

/// The cache's narinfo for a store path, so a test can stand in for it.
pub(crate) trait Cache {
    /// `None` when the cache does not hold the path.
    async fn narinfo(&self, facts: &Facts, store_path: &str) -> Result<Option<String>>;
}

struct Http(reqwest::Client);

impl Cache for Http {
    async fn narinfo(&self, facts: &Facts, store_path: &str) -> Result<Option<String>> {
        narinfo::fetch(&self.0, facts, store_path).await
    }
}

/// `record` when it was built from the `internal/meta` commit `head`, or from
/// one that pins the agent as `head` does (`same_pins`), and the builder pushed
/// it to the cache.
pub(crate) fn current_record<'a>(
    record: Option<&'a AgentClosure>,
    head: Option<&str>,
    same_pins: bool,
) -> Result<&'a AgentClosure, Unpullable> {
    let record =
        record.ok_or_else(|| no(Reason::NoRecord, "no build is recorded for the agent"))?;
    if head != Some(record.meta_rev.as_str()) && !same_pins {
        return Err(no(
            Reason::RevMismatch,
            format!(
                "built from {}, internal/meta is at {}",
                record.meta_rev,
                head.unwrap_or("nothing")
            ),
        ));
    }
    if !record.cached {
        return Err(no(
            Reason::NotCached,
            "the builder did not push the closure",
        ));
    }
    Ok(record)
}

/// Whether the closure `record` names can be pulled with `facts`. `head` is the
/// `internal/meta` commit a deploy would build from; `same_pins` is whether the
/// record's commit pins the agent as `head` does.
pub(crate) async fn judge(
    facts: Option<&Facts>,
    record: Option<&AgentClosure>,
    head: Option<&str>,
    same_pins: bool,
    cache: &impl Cache,
) -> Result<(), Unpullable> {
    let facts = facts.ok_or_else(|| {
        no(
            Reason::NoCredentials,
            "the swarm has given this hive no cache facts",
        )
    })?;
    let record = current_record(record, head, same_pins)?;
    let text = match cache.narinfo(facts, &record.store_path).await {
        Ok(Some(text)) => text,
        Ok(None) => return Err(no(Reason::NotCached, "the cache does not hold the narinfo")),
        Err(e) => return Err(no(Reason::FetchFailed, format!("{e:#}"))),
    };
    narinfo::verify(&text, &record.store_path, facts.public_key())
        .map_err(|e| no(Reason::BadSignature, format!("{e:#}")))
}

/// The last pass's verdict per agent: `None` when it can be pulled.
static VERDICTS: Mutex<Vec<(String, Option<Reason>)>> = Mutex::new(Vec::new());

/// The verdicts of the last pass, for the gauge.
pub fn verdicts() -> Vec<(String, Option<Reason>)> {
    VERDICTS.lock().map(|g| g.clone()).unwrap_or_default()
}

/// Probe every agent now and every [`PASS_INTERVAL`] until shutdown.
pub async fn follow(client: async_nats::Client, mut shutdown: tokio::sync::watch::Receiver<bool>) {
    let mut health = SweepHealth::new("attic_pull_probe", "warn", FAILURES_BEFORE_WARNING);
    loop {
        match pass(&client).await {
            Ok(()) => health.record_ok(),
            Err(e) => {
                tracing::warn!(error = %format!("{e:#}"), "attic: the pull probe failed; the last verdicts stand");
                let err = format!("{e:#}");
                health.record_err(|ctx| {
                    let age = ctx.since_last_ok.map_or_else(
                        || "no success this session".to_owned(),
                        |d| format!("last ok {} ago", sweep_health::fmt_age(d)),
                    );
                    format!(
                        "checking whether agents' closures can be pulled is failing ({} consecutive, {age}) \
                         — deploys are unaffected: {err}",
                        ctx.consecutive
                    )
                });
            }
        }
        tokio::select! {
            () = tokio::time::sleep(PASS_INTERVAL) => {}
            _ = shutdown.changed() => {
                tracing::info!("attic: shutdown signal received");
                return;
            }
        }
    }
}

/// One pass: publish the facts, then judge every agent.
///
/// # Errors
/// A read that failed (the store, the queue, the forge), which leaves the last
/// verdicts as they are: the failure says nothing about any agent.
async fn pass(client: &async_nats::Client) -> Result<()> {
    swarm_queue_client::ensure_connected(client)?;
    let facts = reader::sync().await?;
    let agents = crate::lifecycle::agents_for_meta_listing()
        .await
        .context("enumerating this hive's agents")?;
    let head = match &facts {
        Some(_) => meta_head().await?,
        None => None,
    };
    let http = Http(narinfo::client()?);
    let mut now = Vec::with_capacity(agents.len());
    for spec in agents {
        let record = match &facts {
            Some(_) => read_record(client, &spec.name).await?,
            None => None,
        };
        let same_pins = match (&record, head.as_deref()) {
            (Some(r), Some(head)) => pins_unchanged(&spec.name, &r.meta_rev, head).await,
            _ => false,
        };
        let verdict = judge(
            facts.as_ref(),
            record.as_ref(),
            head.as_deref(),
            same_pins,
            &http,
        )
        .await;
        now.push((spec.name, verdict.as_ref().err().map(|u| u.reason)));
        report(&now[now.len() - 1], verdict.err().map(|u| u.detail));
    }
    if let Ok(mut g) = VERDICTS.lock() {
        *g = now;
    }
    Ok(())
}

/// Whether `agent` has the same pins at `built` as at `head`. A commit that
/// cannot be read counts as different, which leaves the verdict to the commit
/// ids.
async fn pins_unchanged(agent: &str, built: &str, head: &str) -> bool {
    if built == head {
        return true;
    }
    let read = async {
        let archetype = crate::archetypes::of(agent)?.is_some();
        let source = crate::forge::internal_meta_source()
            .await
            .context("no forge token")?;
        crate::meta_rev::pins_unchanged(&source, agent, archetype, built, head).await
    };
    read.await.unwrap_or_else(|e| {
        tracing::debug!(%agent, error = %format!("{e:#}"), "attic: cannot compare the agent's pins");
        false
    })
}

pub(crate) async fn read_record(
    client: &async_nats::Client,
    agent: &str,
) -> Result<Option<AgentClosure>> {
    match agent_closures::read(client, agent)
        .await
        .map_err(|e| anyhow::anyhow!(e))
        .with_context(|| format!("reading the closure record of {agent}"))?
    {
        Read::Current(record) => Ok(record),
        Read::NoBucket => Ok(None),
    }
}

/// The `internal/meta` head, or `None` when the swarm has not published it or
/// this hive has no forge token.
async fn meta_head() -> Result<Option<String>> {
    use crate::meta_rev::MetaSource as _;
    match crate::forge::internal_meta_source().await {
        Some(source) => source
            .meta_head()
            .await
            .context("reading internal/meta's head"),
        None => Ok(None),
    }
}

/// The log line: at info when the verdict differs from the last pass, else
/// debug.
fn report((agent, verdict): &(String, Option<Reason>), detail: Option<String>) {
    let changed = verdicts()
        .iter()
        .find(|(a, _)| a == agent)
        .is_none_or(|(_, v)| v != verdict);
    let reason = verdict.map_or("", Reason::label);
    let pullable = verdict.is_none();
    let detail = detail.unwrap_or_default();
    if changed {
        tracing::info!(%agent, pullable, reason, %detail, "attic: can this hive pull the agent's latest closure");
    } else {
        tracing::debug!(%agent, pullable, reason, %detail, "attic: can this hive pull the agent's latest closure");
    }
}

#[cfg(test)]
mod tests {
    use base64::Engine as _;
    use base64::engine::general_purpose::STANDARD;
    use ed25519_dalek::{Signer as _, SigningKey};

    use super::*;

    const PATH: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas";
    const HASH: &str = "sha256:0mdqa9w1p6cmli6976v4wi0sw9r4p5prkj7lzfd1877wk11c9c73";
    const REV: &str = "a1b2c3";

    fn key() -> SigningKey {
        SigningKey::from_bytes(&[7; 32])
    }

    fn facts() -> Facts {
        let public = format!(
            "swarm:{}",
            STANDARD.encode(key().verifying_key().to_bytes())
        );
        Facts::parse("tok", "https://cache.example", "swarm", &public).expect("facts")
    }

    fn record() -> AgentClosure {
        AgentClosure {
            meta_rev: REV.to_owned(),
            store_path: PATH.to_owned(),
            system: "x86_64-linux".to_owned(),
            built_unix: 1,
            cached: true,
        }
    }

    fn narinfo_signed_by(key: &SigningKey) -> String {
        let sig = key.sign(format!("1;{PATH};{HASH};120;").as_bytes());
        format!(
            "StorePath: {PATH}\nNarHash: {HASH}\nNarSize: 120\nReferences: \nSig: swarm:{}\n",
            STANDARD.encode(sig.to_bytes())
        )
    }

    /// What the cache answers for the narinfo.
    struct Answers(Result<Option<String>, &'static str>);

    impl Cache for Answers {
        async fn narinfo(&self, _: &Facts, _: &str) -> Result<Option<String>> {
            self.0.clone().map_err(|e| anyhow::anyhow!(e))
        }
    }

    async fn reason(
        facts: Option<&Facts>,
        record: Option<&AgentClosure>,
        head: Option<&str>,
        cache: Answers,
    ) -> Option<Reason> {
        judge(facts, record, head, false, &cache)
            .await
            .err()
            .map(|u| u.reason)
    }

    fn good() -> Answers {
        Answers(Ok(Some(narinfo_signed_by(&key()))))
    }

    #[tokio::test]
    async fn a_signed_narinfo_for_the_head_build_is_pullable() {
        assert_eq!(
            reason(Some(&facts()), Some(&record()), Some(REV), good()).await,
            None
        );
    }

    #[tokio::test]
    async fn no_facts_is_no_credentials_even_with_a_record() {
        let r = reason(None, Some(&record()), Some(REV), good()).await;
        assert_eq!(r, Some(Reason::NoCredentials));
    }

    #[tokio::test]
    async fn no_record_is_no_record() {
        let r = reason(Some(&facts()), None, Some(REV), good()).await;
        assert_eq!(r, Some(Reason::NoRecord));
    }

    #[tokio::test]
    async fn a_record_from_another_commit_is_a_rev_mismatch() {
        let r = reason(Some(&facts()), Some(&record()), Some("other"), good()).await;
        assert_eq!(r, Some(Reason::RevMismatch));
        let r = reason(Some(&facts()), Some(&record()), None, good()).await;
        assert_eq!(r, Some(Reason::RevMismatch), "no head to compare with");
    }

    #[tokio::test]
    async fn a_record_from_another_commit_with_the_same_pins_is_pullable() {
        let (facts, record) = (facts(), record());
        let judged = judge(Some(&facts), Some(&record), Some("other"), true, &good()).await;
        assert!(judged.is_ok());
        let unpushed = AgentClosure {
            cached: false,
            ..record
        };
        let judged = judge(Some(&facts), Some(&unpushed), Some("other"), true, &good()).await;
        assert_eq!(judged.err().map(|u| u.reason), Some(Reason::NotCached));
    }

    #[tokio::test]
    async fn an_unpushed_record_or_a_missing_narinfo_is_not_cached() {
        let unpushed = AgentClosure {
            cached: false,
            ..record()
        };
        let r = reason(Some(&facts()), Some(&unpushed), Some(REV), good()).await;
        assert_eq!(r, Some(Reason::NotCached));
        let r = reason(
            Some(&facts()),
            Some(&record()),
            Some(REV),
            Answers(Ok(None)),
        )
        .await;
        assert_eq!(r, Some(Reason::NotCached));
    }

    #[tokio::test]
    async fn a_failed_request_is_fetch_failed() {
        let r = reason(
            Some(&facts()),
            Some(&record()),
            Some(REV),
            Answers(Err("401")),
        )
        .await;
        assert_eq!(r, Some(Reason::FetchFailed));
    }

    #[tokio::test]
    async fn a_narinfo_the_key_did_not_sign_is_a_bad_signature() {
        let forged = Answers(Ok(Some(narinfo_signed_by(&SigningKey::from_bytes(
            &[9; 32],
        )))));
        let r = reason(Some(&facts()), Some(&record()), Some(REV), forged).await;
        assert_eq!(r, Some(Reason::BadSignature));
    }

    #[test]
    fn the_labels_are_the_documented_ones() {
        let labels = [
            Reason::NoRecord,
            Reason::RevMismatch,
            Reason::NotCached,
            Reason::NoCredentials,
            Reason::FetchFailed,
            Reason::BadSignature,
        ]
        .map(Reason::label);
        assert_eq!(
            labels,
            [
                "no-record",
                "rev-mismatch",
                "not-cached",
                "no-credentials",
                "fetch-failed",
                "bad-signature"
            ]
        );
    }
}