//! Agent roster storage — the set of agent names the hive knows about.
//! Persisted as a JSON array at `/var/lib/hyperhive/meta/topology.json`,
//! alongside the meta `flake.nix`, so roster changes thread through the
//! same git commit log as deploys.
//!
//! **There is no hierarchy here any more.** The file used to be a map of
//! `name → parent | null` and this module owned the parent/child tree that
//! fed `<parent>` / `<children>` routing, the reparenting API and the
//! dashboard's tree view. All of that is gone; what the file is
//! *for* now is the one thing that survived the removal — naming every
//! agent, which is the set a
//! [`hive_sh4re::permissions::Capability::ManageRootAgent`] holder gets
//! bind-mounted ([`all_agents`]).
//!
//! Broader-than-your-own bind-mount grants are **not** stored here: they
//! hang off the `ManageRootAgent` capability in `capabilities.json`. See
//! `lifecycle::set_nspawn_flags`.

use std::collections::BTreeSet;
use std::path::{Path, PathBuf};

use serde::Deserialize;

const TOPOLOGY_FILE: &str = "topology.json";

#[must_use]
pub fn topology_path() -> PathBuf {
    crate::paths::meta_root().join(TOPOLOGY_FILE)
}

/// On-disk shapes [`read_from`] accepts. The array is what [`reconcile`] writes; the
/// map is the legacy `name → parent | null` format, kept readable so a
/// hive that upgrades across this change keeps its roster instead of
/// blanking it until the next `reconcile` pass — and a blank roster is not
/// a cosmetic gap, it is every `ManageRootAgent` holder losing its mounts
/// for the length of that window.
#[derive(Deserialize)]
#[serde(untagged)]
enum OnDisk {
    Roster(BTreeSet<String>),
    /// The value was the parent name; only the keys carry over.
    WithParents(std::collections::BTreeMap<String, Option<String>>),
}

/// A missing file is the empty roster (a fresh install that hasn't run
/// through `meta::sync_agents` yet).
impl Default for OnDisk {
    fn default() -> Self {
        Self::Roster(BTreeSet::new())
    }
}

/// Snapshot of the agent roster in the file at `path` (normally
/// [`topology_path`]). The file is small (one line per agent), so it is
/// re-read rather than cached — keeps the source of truth on disk.
///
/// An absent file is the empty set. A file that exists but can't be read
/// or parsed is an error, so [`reconcile`] never writes over it.
fn read_from(path: &Path) -> std::io::Result<BTreeSet<String>> {
    Ok(match super::read_map::<OnDisk>(path)? {
        OnDisk::Roster(names) => names,
        OnDisk::WithParents(map) => map.into_keys().collect(),
    })
}

/// Every agent the roster knows about, in name order. This is the set
/// a [`hive_sh4re::permissions::Capability::ManageRootAgent`] holder gets
/// bind-mounted, and it is deliberately unfiltered: that capability means
/// "may manage any agent", so the set is all of them.
///
/// An unreadable roster is logged and yields no agents, so the holder
/// starts without cross-agent mounts rather than failing to start.
#[must_use]
pub fn all_agents() -> Vec<String> {
    all_agents_at(&topology_path())
}

fn all_agents_at(path: &Path) -> Vec<String> {
    match read_from(path) {
        Ok(topo) => all_agents_in(&topo),
        Err(e) => {
            tracing::error!(
                path = %path.display(),
                error = ?e,
                "topology unreadable — no cross-agent mounts"
            );
            Vec::new()
        }
    }
}

/// Pure form of [`all_agents`] for unit tests.
#[must_use]
pub fn all_agents_in(topo: &BTreeSet<String>) -> Vec<String> {
    topo.iter().cloned().collect()
}

/// Reconcile `topology.json` against the current agent set. Adds any agent
/// missing from the file; removes entries for agents no longer present.
/// Returns true when the file changed and should be re-committed by the
/// caller. The write is atomic, as a sorted array (`BTreeSet` iterates in
/// key order) to keep git diffs minimal.
///
/// `pending` lists agents that have a provisioned proposed config repo
/// but no container yet (provisioned, not yet spawned). They are KEPT
/// (not dropped) so an agent that exists on disk but has never booted is
/// still a name the hive knows about.
///
/// Fails without writing when the existing file can't be read or parsed:
/// rebuilding it from the live set alone would drop every pending name.
pub fn reconcile(agent_names: &[String], pending: &[String]) -> std::io::Result<bool> {
    reconcile_at(&topology_path(), agent_names, pending)
}

fn reconcile_at(path: &Path, agent_names: &[String], pending: &[String]) -> std::io::Result<bool> {
    let (next, changed) = apply_reconcile(&read_from(path)?, agent_names, pending);
    if changed {
        super::write_map(path, &next)?;
    }
    Ok(changed)
}

/// Pure form of [`reconcile`] for unit tests. Adds missing live agents,
/// drops entries for agents that are neither live nor pending-init, and
/// reports whether anything changed.
#[must_use]
pub fn apply_reconcile(
    current: &BTreeSet<String>,
    agent_names: &[String],
    pending: &[String],
) -> (BTreeSet<String>, bool) {
    let mut next = current.clone();
    let mut changed = false;
    for name in agent_names {
        if next.insert(name.clone()) {
            changed = true;
        }
    }
    let known: std::collections::HashSet<&String> =
        agent_names.iter().chain(pending.iter()).collect();
    next.retain(|name| {
        let keep = known.contains(name);
        if !keep {
            changed = true;
        }
        keep
    });
    (next, changed)
}

#[cfg(test)]
mod tests {
    use super::{
        BTreeSet, OnDisk, TOPOLOGY_FILE, all_agents_at, all_agents_in, apply_reconcile, read_from,
        reconcile_at,
    };

    fn roster_three() -> BTreeSet<String> {
        ["alice", "bob", "carol"]
            .into_iter()
            .map(str::to_owned)
            .collect()
    }

    #[test]
    fn apply_reconcile_adds_missing_live_agent() {
        let live = vec![
            crate::lifecycle::MANAGER_NAME.to_owned(),
            "newbie".to_owned(),
        ];
        let (next, changed) = apply_reconcile(&BTreeSet::new(), &live, &[]);
        assert!(changed);
        assert!(next.contains("newbie"));
        assert!(next.contains(crate::lifecycle::MANAGER_NAME));
    }

    #[test]
    fn apply_reconcile_drops_vanished_agent() {
        let live = vec!["alice".to_owned()];
        let (next, changed) = apply_reconcile(&roster_three(), &live, &[]);
        assert!(changed);
        assert!(!next.contains("bob"));
        assert!(!next.contains("carol"));
        assert!(next.contains("alice"));
    }

    #[test]
    fn apply_reconcile_keeps_pending_init_agent() {
        // `dora` was provisioned but has no container yet, so it's absent
        // from the live set. It must NOT be dropped.
        let mut roster = roster_three();
        roster.insert("dora".to_owned());
        let live = vec!["alice".to_owned(), "bob".to_owned(), "carol".to_owned()];
        let pending = vec!["dora".to_owned()];
        let (next, changed) = apply_reconcile(&roster, &live, &pending);
        assert!(!changed, "no change expected: {next:?}");
        assert!(next.contains("dora"));
    }

    #[test]
    fn apply_reconcile_is_a_noop_when_already_in_sync() {
        let live = vec!["alice".to_owned(), "bob".to_owned(), "carol".to_owned()];
        let (next, changed) = apply_reconcile(&roster_three(), &live, &[]);
        assert!(!changed);
        assert_eq!(next, roster_three());
    }

    #[test]
    fn all_agents_in_returns_every_name_sorted() {
        assert_eq!(
            all_agents_in(&roster_three()),
            vec!["alice", "bob", "carol"]
        );
    }

    #[test]
    fn all_agents_in_empty_roster_returns_empty() {
        assert!(all_agents_in(&BTreeSet::new()).is_empty());
    }

    /// The upgrade path. A hive whose `topology.json` still carries the
    /// legacy `name → parent` map must read as the same roster, parent
    /// values discarded — otherwise the first read after the upgrade hands
    /// `ManageRootAgent` holders an empty mount set.
    #[test]
    fn a_pre_4472_parent_map_reads_as_its_key_set() {
        let raw = r#"{"alice": "bob", "bob": null, "carol": "bob"}"#;
        let parsed: OnDisk = serde_json::from_str(raw).expect("legacy map parses");
        let names = match parsed {
            OnDisk::Roster(n) => n,
            OnDisk::WithParents(m) => m.into_keys().collect(),
        };
        assert_eq!(names, roster_three());
    }

    #[test]
    fn a_roster_array_round_trips() {
        let raw = serde_json::to_string(&roster_three()).expect("serialises");
        let parsed: OnDisk = serde_json::from_str(&raw).expect("array parses");
        let names = match parsed {
            OnDisk::Roster(n) => n,
            OnDisk::WithParents(m) => m.into_keys().collect(),
        };
        assert_eq!(names, roster_three());
    }

    const TRUNCATED: &str = "[\n  \"alice\",\n  \"bob\",\n  \"do";

    fn corrupt_file() -> (tempfile::TempDir, std::path::PathBuf) {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join(TOPOLOGY_FILE);
        std::fs::write(&path, TRUNCATED).expect("seed");
        (dir, path)
    }

    #[test]
    fn reconcile_leaves_a_corrupt_file_untouched() {
        let (_dir, path) = corrupt_file();
        let live = vec!["alice".to_owned(), "bob".to_owned()];
        let err =
            reconcile_at(&path, &live, &[]).expect_err("a corrupt file must not be rewritten");
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
        assert_eq!(std::fs::read(&path).expect("read"), TRUNCATED.as_bytes());
    }

    #[test]
    fn all_agents_of_a_corrupt_file_is_empty() {
        let (_dir, path) = corrupt_file();
        assert!(all_agents_at(&path).is_empty());
    }

    #[test]
    fn reconcile_round_trips_through_a_missing_file_and_leaves_no_temp_file() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join(TOPOLOGY_FILE);
        assert!(read_from(&path).expect("missing file").is_empty());
        let live = vec!["alice".to_owned(), "bob".to_owned(), "carol".to_owned()];
        assert!(reconcile_at(&path, &live, &[]).expect("reconcile"));
        assert_eq!(read_from(&path).expect("read"), roster_three());
        let entries: Vec<_> = std::fs::read_dir(dir.path())
            .expect("read_dir")
            .map(|e| e.expect("entry").file_name())
            .collect();
        assert_eq!(entries, vec![std::ffi::OsString::from(TOPOLOGY_FILE)]);
    }
}