//! Squid's deny-list file: the destination names the swarm refuses this hive's
//! agents, whatever allows them.
//!
//! swarm-controller keeps the deny list beside the allow-list
//! ([`swarm_wire::egress_allow::DenyList`], its own key of the same bucket); each
//! entry is for every agent or for one. Every [`INTERVAL`] this reads it and
//! renders [`ENV_DENY_FILE`], squid configuration that
//! `nix/host-modules/hive-squid.nix` `include`s after the destination denies
//! and before `deny !authed` and every allow, so a denied name is refused for
//! every user in every mode. A swarm-wide entry is one `dstdomain` ACL and a
//! `deny` that names no user, so even a client without a credential gets a 403;
//! an agent's entry also names that agent's squid user. Each rule ends in the
//! [`MARK`] annotation, which the proxy logs as its own `egress_denied` line.
//!
//! No entry renders an empty file, which squid reads as no rules, so a hive
//! whose swarm has no deny list configures squid as it did before this existed.
//!
//! The revision of the list last rendered is what [`applied`] answers and the
//! hive's status snapshot carries, so the swarm can see a hive that is behind.
//! A hive that cannot read the list keeps the file it has, and reports nothing
//! it did not render.
//!
//! Every entry is checked again with
//! [`swarm_wire::egress_allow::normalize_domain`], since the names end up in
//! squid's configuration, and skipped when it fails. An entry another entry of
//! its list covers is dropped, because squid ignores the narrower of two
//! overlapping names in one ACL.

use std::fmt::Write as _;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use anyhow::{Context, Result};
use swarm_queue_client::egress_allow::{self, DenyRead};
use swarm_wire::egress_allow::{DenyList, covered_by, normalize_domain};

use super::squid_allow::reduced;
use super::squid_users::{read_or_empty, write_if_changed};
use crate::stats::sweep_health::{self, SweepHealth};

/// How often the list is read.
pub const INTERVAL: Duration = Duration::from_secs(30);

/// At [`INTERVAL`] this is ~90 seconds of failing reads.
const FAILURES_BEFORE_WARNING: u32 = 3;

/// The squid configuration file this renders. Set by `nix/host-modules/hive-c0re`
/// from `services.hyperhive.deploy.hive-controller.squid.denyFile`.
pub const ENV_DENY_FILE: &str = "HIVE_SQUID_DENY_FILE";

/// The annotation a deny rule puts on a request, as `key=value`. The `note` ACL
/// in `nix/host-modules/hive-squid.nix` reads the key.
const MARK: &str = "egress_deny=1";

/// The ACL that applies [`MARK`].
const MARK_ACL: &str = "deny_hit_mark";

/// The revision last rendered plus one; `0` before the first pass has rendered
/// or confirmed anything.
static APPLIED: AtomicU64 = AtomicU64::new(0);

/// The revision of the swarm's deny list this hive's file holds, `0` when the
/// swarm has none, or `None` until a pass has said so.
#[must_use]
pub fn applied() -> Option<u64> {
    APPLIED.load(Ordering::Relaxed).checked_sub(1)
}

fn set_applied(revision: u64) {
    APPLIED.store(revision.saturating_add(1), Ordering::Relaxed);
}

/// Read the list now and every [`INTERVAL`] until shutdown, when this host has
/// squid's deny file.
pub async fn follow(client: async_nats::Client, mut shutdown: tokio::sync::watch::Receiver<bool>) {
    let Some(file) = std::env::var_os(ENV_DENY_FILE)
        .filter(|v| !v.is_empty())
        .map(PathBuf::from)
    else {
        return;
    };
    let mut health = SweepHealth::new("squid_deny_render", "warn", FAILURES_BEFORE_WARNING);
    loop {
        match pass(&client, &file).await {
            Ok(()) => health.record_ok(),
            Err(e) => {
                tracing::warn!(error = ?e, "squid deny list: the pass failed");
                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!(
                        "reading the swarm's egress deny list is failing ({} consecutive, \
                         {age}) — squid keeps the denies last written: {err}",
                        ctx.consecutive
                    )
                });
            }
        }
        tokio::select! {
            () = tokio::time::sleep(INTERVAL) => {}
            _ = shutdown.changed() => return,
        }
    }
}

async fn pass(client: &async_nats::Client, file: &Path) -> Result<()> {
    // An unconnected client does not fail a JetStream request, it hangs on it.
    swarm_queue_client::ensure_connected(client)?;
    let agents: Vec<String> = crate::lifecycle::agents_for_meta_listing()
        .await
        .context("enumerating this hive's agents")?
        .into_iter()
        .map(|spec| spec.name)
        .collect();
    let read = egress_allow::read_deny(client)
        .await
        .map_err(|e| anyhow::anyhow!(e))
        .context("reading the swarm's egress deny list")?;
    if let Some(revision) = sync_with(read, &agents, file)? {
        set_applied(revision);
    }
    Ok(())
}

/// [`pass`] with the read and the agents given. The file is only written when
/// its rendering differs from what it holds. Answers the revision the file now
/// holds, or `None` when it cannot say.
fn sync_with(read: DenyRead, agents: &[String], file: &Path) -> Result<Option<u64>> {
    match read {
        DenyRead::Current { list, revision } => {
            let rendered = render(&list, agents);
            if write_if_changed(file, &rendered.text)? {
                tracing::info!(
                    file = %file.display(), revision,
                    "squid deny list: {} swarm, {} agent entries", rendered.swarm, rendered.agent
                );
            }
            if !rendered.skipped.is_empty() {
                tracing::warn!(
                    skipped = ?rendered.skipped,
                    "squid deny list: entries the swarm holds were skipped as invalid"
                );
            }
            Ok(Some(revision))
        }
        // Nothing to read says nothing about what the swarm wants, so a file
        // that holds denies stays. An empty file is the true "no list".
        DenyRead::NotSet => Ok(read_or_empty(file)?.is_empty().then_some(0)),
    }
}

struct Rendered {
    text: String,
    /// Swarm-wide names written.
    swarm: usize,
    /// Names written across this hive's agents.
    agent: usize,
    /// Entries left out because they are not valid names or agents.
    skipped: Vec<String>,
}

/// The squid configuration for `list`, for the agents this hive runs. Empty for
/// no entry that applies.
///
/// The destination ACL comes first in each rule: squid checks ACLs left to
/// right, so a swarm-wide rule never reaches a `proxy_auth` ACL, which would
/// answer an anonymous client 407 instead of the 403.
///
/// A per-agent name gets two rules. For a request through the proxy port the
/// agent is the one whose credential came with it; `!icpt` comes first, since a
/// `proxy_auth` ACL evaluated on an intercepted request only logs a warning.
/// For an intercepted request the agent is the user the `hive_agent` helper
/// named (`ext_user`).
fn render(list: &DenyList, agents: &[String]) -> Rendered {
    let mut skipped = Vec::new();
    let mut swarm_wide = Vec::new();
    let mut per_agent: std::collections::BTreeMap<&str, Vec<String>> =
        std::collections::BTreeMap::new();
    for entry in &list.entries {
        let Ok(domain) = normalize_domain(&entry.domain) else {
            skipped.push(entry.domain.clone());
            continue;
        };
        match entry.agent.as_deref() {
            None => swarm_wide.push(domain),
            Some(agent) => {
                if hive_types::Ident::parse(agent).is_err() {
                    skipped.push(format!("{} for {agent}", entry.domain));
                } else if let Some(mine) = agents.iter().find(|a| *a == agent) {
                    per_agent.entry(mine.as_str()).or_default().push(domain);
                }
            }
        }
    }
    let swarm_wide = reduced(swarm_wide);

    let mut rules = String::new();
    if !swarm_wide.is_empty() {
        let _ = write!(
            rules,
            "acl deny_swarm_dst dstdomain {}\n\
             http_access deny deny_swarm_dst {MARK_ACL}\n",
            swarm_wide.join(" ")
        );
    }
    let mut agent_count = 0;
    for (agent, names) in per_agent {
        let names = reduced(
            names
                .into_iter()
                .filter(|d| !covered_by(d, &swarm_wide))
                .collect(),
        );
        if names.is_empty() {
            continue;
        }
        agent_count += names.len();
        let _ = write!(
            rules,
            "acl deny_agent_{agent}_dst dstdomain {}\n\
             acl deny_agent_{agent} proxy_auth {agent}\n\
             acl deny_agent_{agent}_ix ext_user {agent}\n\
             http_access deny deny_agent_{agent}_dst !icpt deny_agent_{agent} {MARK_ACL}\n\
             http_access deny deny_agent_{agent}_dst icpt deny_agent_{agent}_ix {MARK_ACL}\n",
            names.join(" ")
        );
    }
    let text = if rules.is_empty() {
        String::new()
    } else {
        format!("acl {MARK_ACL} annotate_transaction {MARK}\n{rules}")
    };
    Rendered {
        text,
        swarm: swarm_wide.len(),
        agent: agent_count,
        skipped,
    }
}

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

    fn list(entries: &[(&str, Option<&str>)]) -> DenyList {
        DenyList {
            entries: entries
                .iter()
                .map(|(d, a)| DenyEntry {
                    domain: (*d).to_owned(),
                    agent: a.map(str::to_owned),
                })
                .collect(),
        }
    }

    fn agents(names: &[&str]) -> Vec<String> {
        names.iter().map(|n| (*n).to_owned()).collect()
    }

    const MARK_LINE: &str = "acl deny_hit_mark annotate_transaction egress_deny=1\n";

    /// The one place a rule's shape is pinned: the destination ACL before any
    /// user ACL, and the annotation last.
    #[test]
    fn a_swarm_wide_name_is_one_destination_rule_that_names_no_user() {
        let r = render(
            &list(&[("evil.com", None), (".bad.io", None), ("EVIL.com", None)]),
            &agents(&["atlas"]),
        );
        assert_eq!(
            r.text,
            format!(
                "{MARK_LINE}acl deny_swarm_dst dstdomain .bad.io evil.com\n\
                 http_access deny deny_swarm_dst deny_hit_mark\n"
            )
        );
        assert!(!r.text.contains("proxy_auth"), "no user, so no 407");
        assert_eq!((r.swarm, r.agent), (2, 0));
    }

    #[test]
    fn an_agents_name_is_a_destination_then_user_rule_and_other_hives_agents_are_left_out() {
        let r = render(
            &list(&[
                ("pypi.org", Some("atlas")),
                ("crates.io", Some("atlas")),
                ("github.com", Some("iris")),
                ("github.com", Some("../argus")),
            ]),
            &agents(&["atlas", "argus"]),
        );
        assert_eq!(
            r.text,
            format!(
                "{MARK_LINE}acl deny_agent_atlas_dst dstdomain crates.io pypi.org\n\
                 acl deny_agent_atlas proxy_auth atlas\n\
                 acl deny_agent_atlas_ix ext_user atlas\n\
                 http_access deny deny_agent_atlas_dst !icpt deny_agent_atlas deny_hit_mark\n\
                 http_access deny deny_agent_atlas_dst icpt deny_agent_atlas_ix deny_hit_mark\n"
            )
        );
        assert_eq!((r.swarm, r.agent), (0, 2));
        assert_eq!(r.skipped, ["github.com for ../argus"]);
        assert!(!r.text.contains("iris"), "another hive's agent");
        assert!(!r.text.contains("argus"), "an invalid name");
    }

    /// An agent's entry adds to the swarm-wide rule and never replaces it: the
    /// swarm-wide rule renders whole whatever the agent lists, and a name it
    /// already covers is not repeated for the agent.
    #[test]
    fn an_agents_entry_adds_to_the_swarm_wide_rule_and_never_narrows_it() {
        let r = render(
            &list(&[
                (".evil.com", None),
                ("api.evil.com", Some("atlas")),
                ("bad.io", Some("atlas")),
            ]),
            &agents(&["atlas"]),
        );
        assert!(
            r.text.contains(
                "acl deny_swarm_dst dstdomain .evil.com\nhttp_access deny deny_swarm_dst"
            )
        );
        assert!(r.text.contains("deny_agent_atlas_dst dstdomain bad.io\n"));
        assert!(!r.text.contains("api.evil.com"), "{}", r.text);
        assert_eq!((r.swarm, r.agent), (1, 1));
    }

    /// No entry renders no byte, which is what keeps a hive with no deny list
    /// configured as it was.
    #[test]
    fn no_entry_that_applies_renders_an_empty_file() {
        assert_eq!(render(&DenyList::default(), &agents(&["atlas"])).text, "");
        // An agent this hive does not run, and an invalid name, render nothing.
        let r = render(
            &list(&[("evil.com", Some("iris")), ("1.2.3.4", None)]),
            &agents(&["atlas"]),
        );
        assert_eq!(r.text, "");
        assert_eq!(r.skipped, ["1.2.3.4"]);
        // Control: the same list with a name that applies is not empty.
        assert!(
            !render(&list(&[("evil.com", None)]), &agents(&[]))
                .text
                .is_empty()
        );
    }

    #[test]
    fn an_entry_that_is_not_a_name_never_reaches_squid_configuration() {
        let r = render(
            &list(&[
                ("evil.com", None),
                ("*.example.com", Some("atlas")),
                ("example.com\nhttp_access allow all", Some("atlas")),
                ("example.com\nhttp_access allow all", None),
            ]),
            &agents(&["atlas"]),
        );
        assert!(!r.text.contains("allow"), "{}", r.text);
        assert!(r.text.contains("dstdomain evil.com\n"));
        assert_eq!(r.skipped.len(), 3);
    }

    /// The value swarm-controller's `egress_deny` route publishes, byte for byte
    /// as its own test pins it, renders the rules for the agents named in it.
    #[test]
    fn the_controllers_published_value_renders_the_rules_it_names() {
        let wire = r#"{"entries":[{"domain":"bad.io","agent":"atlas"},{"domain":"evil.com"},{"domain":"evil.com","agent":"atlas"}]}"#;
        let decoded: DenyList = serde_json::from_str(wire).expect("decodes");
        let r = render(&decoded, &agents(&["atlas"]));
        assert!(r.text.contains("acl deny_swarm_dst dstdomain evil.com\n"));
        assert!(
            r.text
                .contains("acl deny_agent_atlas_dst dstdomain bad.io\n")
        );
        assert_eq!((r.swarm, r.agent), (1, 1));
    }

    struct Dir(tempfile::TempDir);

    impl Dir {
        fn file(&self) -> PathBuf {
            self.0.path().join("deny.conf")
        }
    }

    fn dir() -> Dir {
        Dir(tempfile::tempdir().expect("tempdir"))
    }

    fn text(d: &Dir) -> String {
        std::fs::read_to_string(d.file()).expect("the file is there")
    }

    fn current(entries: &[(&str, Option<&str>)], revision: u64) -> DenyRead {
        DenyRead::Current {
            list: list(entries),
            revision,
        }
    }

    #[test]
    fn the_file_follows_the_list_and_the_claimed_revision_follows_the_file() {
        let d = dir();
        let mine = agents(&["atlas"]);
        let sync = |read| sync_with(read, &mine, &d.file()).expect("renders");

        assert_eq!(sync(current(&[("evil.com", None)], 7)), Some(7));
        assert!(text(&d).contains("dstdomain evil.com"));

        // A second pass of the same list writes nothing.
        let ino = std::os::unix::fs::MetadataExt::ino(&std::fs::metadata(d.file()).unwrap());
        assert_eq!(sync(current(&[("evil.com", None)], 7)), Some(7));
        assert_eq!(
            std::os::unix::fs::MetadataExt::ino(&std::fs::metadata(d.file()).unwrap()),
            ino
        );

        // No answer keeps the denies in force and claims no revision.
        assert_eq!(sync(DenyRead::NotSet), None);
        assert!(text(&d).contains("dstdomain evil.com"), "denies are kept");

        // The list emptied: the file is empty and the revision moves.
        assert_eq!(sync(current(&[], 9)), Some(9));
        assert_eq!(text(&d), "");

        // An empty file with nothing to read is the true "no list": revision 0.
        assert_eq!(sync(DenyRead::NotSet), Some(0));
    }

    /// The only test that touches the process-wide value.
    #[test]
    fn nothing_is_applied_until_a_pass_says_so() {
        assert_eq!(applied(), None);
        set_applied(0);
        assert_eq!(applied(), Some(0));
        set_applied(12);
        assert_eq!(applied(), Some(12));
    }

    #[test]
    fn the_file_is_group_readable() {
        use std::os::unix::fs::PermissionsExt as _;
        let d = dir();
        sync_with(current(&[("evil.com", None)], 1), &agents(&[]), &d.file()).expect("renders");
        let mode = std::fs::metadata(d.file())
            .expect("stat")
            .permissions()
            .mode()
            & 0o777;
        assert_eq!(mode, 0o640);
    }
}