//! Auto-update: rebuild the agents whose swarm-declared closure the hive has
//! not applied. An agent's `agent-closures` record names the store path the
//! swarm wants it to run; the agent's `applied/.<name>.closure` marker holds
//! the one the hive last swapped to ([`crate::attic::closure::pending`]).
//!
//! A pass runs on `hive-c0re serve` boot, then every [`PASS_INTERVAL`]. A boot
//! pass is how a hive that missed the swarm's deploy request catches up; the
//! periodic one covers a request lost while the hive was up. Each pass
//! rebuilds each stale agent whose `wanted` power intent is `Up` as one `Boot`
//! DAG, and the boot pass also converges every other drifted agent to its
//! persisted `wanted` via `Reconcile` DAGs. Rules that keep nix work minimal:
//!
//! 1. **Stale but wanted-offline agents** get no rebuild — their
//!    rebuild happens the first time they're started (the start
//!    submit path upgrades a stale start to rebuild+start).
//! 2. **Agents running the declared closure** are skipped.
//! 3. **A declaration the hive cannot read or act on** (no connection to the
//!    swarm queue, no record, the closure not in the cache, no readable running
//!    closure to compare with) rebuilds nothing.
//! 4. **A declaration is attempted once per process** by the periodic pass, and
//!    never while a rebuild of the agent is queued or running, so a closure
//!    that fails to deploy does not restart its agent every interval.
//!
//! See `docs/scheduler/coordinator.md::Boot reconcile`.

use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;

use crate::coordinator::Coordinator;
use crate::lifecycle::{self, AGENT_PREFIX};

/// How often the swarm's declarations are compared with what the hive runs
/// after the boot pass.
const PASS_INTERVAL: Duration = Duration::from_mins(5);

/// Resolve the current rev of `hyperhive_flake`. For a path on disk we
/// canonicalize (following symlinks) so a /etc/hyperhive → /nix/store/...
/// update yields a different string. For anything else we return None.
#[must_use]
pub fn current_flake_rev(hyperhive_flake: &str) -> Option<String> {
    let path = Path::new(hyperhive_flake);
    if !path.exists() {
        return None;
    }
    std::fs::canonicalize(path)
        .ok()
        .map(|p| p.display().to_string())
}

/// What one pass decided.
struct Plan {
    /// Stale ∧ wanted up, running agents first.
    rebuilds: Vec<String>,
    /// Agents whose observed power state drifted from `wanted`; empty after the
    /// boot pass.
    drifted: Vec<String>,
    deferred: usize,
    skipped: usize,
}

/// Whether an agent needs no rebuild this pass: nothing is pending, a rebuild
/// is already queued, or this process already attempted the declaration
/// (`attempted`) and the periodic pass does not retry it.
fn is_fresh(pending: Option<&str>, attempted: Option<&str>, rebuild_live: bool) -> bool {
    pending.is_none() || rebuild_live || pending == attempted
}

/// Classify every agent by what the swarm declares + persisted `wanted`
/// intent. `attempted` holds the declaration this process last rebuilt each
/// agent for; the agents it adds to `Plan::rebuilds` are recorded in it.
async fn classify(
    coord: &Arc<Coordinator>,
    attempted: &mut HashMap<String, String>,
) -> Option<Plan> {
    let containers = match lifecycle::list().await {
        Ok(c) => c,
        Err(e) => {
            tracing::warn!(error = ?e, "auto-update: nixos-container list failed");
            return None;
        }
    };

    // Resolve container names to logical agent names, then sort. The
    // parent field this used to depth-sort by is gone; with no
    // hierarchy left to respect, alphabetical is the whole order — and it
    // is exactly what the depth sort already produced once every agent
    // was a root.
    let mut logical_names: Vec<String> = containers
        .iter()
        .filter_map(|c| c.strip_prefix(AGENT_PREFIX).map(str::to_owned))
        .collect();
    logical_names.sort();

    // `get_or_seed` doubles as the one-time migration: an
    // agent without an `agent_power` row is seeded from its observed
    // state (running ⇒ Up), after which the DB is authoritative.
    // stale ∧ wanted=Up → rebuild. The bool is the agent's observed
    // running state, used to order running agents first before submit.
    let mut rebuilds: Vec<(String, bool)> = Vec::new();
    let mut drifted: Vec<String> = Vec::new(); // fresh ∧ wanted≠observed → reconcile
    let mut deferred = 0usize;
    let mut skipped = 0usize;
    for name in &logical_names {
        let running = lifecycle::is_running(name).await;
        let wanted = match coord.power.get_or_seed(name, running) {
            Ok(w) => Some(w),
            Err(e) => {
                tracing::warn!(%name, error = ?e, "agent_power read failed — reconciling so it surfaces");
                None
            }
        };
        let pending = crate::attic::closure::pending_path(name).await;
        let fresh = is_fresh(
            pending.as_deref(),
            attempted.get(name).map(String::as_str),
            coord.job_queue.rebuild_live(name),
        );
        match boot_action(wanted, fresh, running) {
            BootAction::Rebuild => {
                if let Some(path) = pending {
                    attempted.insert(name.clone(), path);
                }
                rebuilds.push((name.clone(), running));
            }
            BootAction::Defer { reconcile } => {
                deferred += 1;
                tracing::debug!(%name, "auto-update: stale but offline — deferring rebuild to on-start");
                if reconcile {
                    drifted.push(name.clone());
                }
            }
            BootAction::Skip { reconcile } => {
                skipped += 1;
                if reconcile {
                    drifted.push(name.clone());
                }
            }
            BootAction::Unreadable => drifted.push(name.clone()),
        }
    }

    // Rebuild running agents first. All entries are wanted=Up;
    // among them, warm the live/serving agents onto the declared closure before the
    // stopped-but-wanted-up ones so the scarce build slots hit uptime-critical
    // agents first. Stable sort keeps the alphabetical order within each
    // running/stopped group. The `drifted` reconciles aren't sorted
    // — they hold no build slot and run concurrently, so their order is moot.
    rebuilds.sort_by_key(|(_, running)| !running);
    Some(Plan {
        rebuilds: rebuilds.into_iter().map(|(name, _)| name).collect(),
        drifted,
        deferred,
        skipped,
    })
}

/// Auto-update (see the module doc): the boot pass, then a rebuild-only pass
/// every [`PASS_INTERVAL`] until shutdown. Returns Ok even if some submissions
/// failed.
pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
    let mut shutdown = coord.shutdown_rx();
    let mut attempted = HashMap::new();
    if let Some(plan) = classify(&coord, &mut attempted).await {
        tracing::info!(
            total = plan.rebuilds.len() + plan.deferred + plan.skipped,
            rebuilds = plan.rebuilds.len(),
            reconciles = plan.drifted.len(),
            deferred = plan.deferred,
            up_to_date = plan.skipped,
            "boot reconcile"
        );
        submit_boot_tree(
            &coord,
            &plan.rebuilds,
            plan.drifted,
            plan.deferred,
            plan.skipped,
        );
    }
    submit_startup_sweep_nodes(&coord);
    loop {
        tokio::select! {
            () = tokio::time::sleep(PASS_INTERVAL) => {}
            _ = shutdown.changed() => return Ok(()),
        }
        if let Some(plan) = classify(&coord, &mut attempted).await
            && !plan.rebuilds.is_empty()
        {
            tracing::info!(
                rebuilds = plan.rebuilds.len(),
                "auto-update: the swarm declared new closures"
            );
            submit_boot_tree(
                &coord,
                &plan.rebuilds,
                Vec::new(),
                plan.deferred,
                plan.skipped,
            );
        }
    }
}

/// What the boot sweep does with one agent.
#[derive(Debug, PartialEq, Eq)]
enum BootAction {
    /// Stale ∧ wanted up: rebuild to the swarm-declared closure. The DAG's
    /// tail `Reconcile` brings the agent (back) up, covering both the
    /// running-stale and stopped-but-wanted-up cases.
    Rebuild,
    /// Stale ∧ wanted offline: no boot-time nix work — the start submit
    /// path upgrades a stale start to a rebuild.
    Defer { reconcile: bool },
    /// Already running the declared closure, or nothing to deploy.
    Skip { reconcile: bool },
    /// The intent could not be read.
    Unreadable,
}

/// `wanted` is `None` when the store could not answer for this agent.
///
/// That case has no neutral stand-in, which is why it is a variant rather
/// than a fallback value: `Wanted::from_running(running)` is precisely the
/// value for which `reconcile_action` returns `Noop`, so guessing it here
/// puts the agent beyond every check below — a corrupt row then leaves one
/// `warn!` per boot and no other trace. The queued reconcile's own
/// `get_or_seed` fails as a per-agent node instead.
fn boot_action(wanted: Option<crate::power::Wanted>, fresh: bool, running: bool) -> BootAction {
    let Some(wanted) = wanted else {
        return BootAction::Unreadable;
    };
    let reconcile =
        crate::power::reconcile_action(wanted, running) != crate::power::ReconcileAction::Noop;
    if fresh {
        BootAction::Skip { reconcile }
    } else if wanted == crate::power::Wanted::Up {
        BootAction::Rebuild
    } else {
        BootAction::Defer { reconcile }
    }
}

/// Submit the boot-time knowledge and wanted-state sweeps as DAG
/// nodes — `KnowledgePull`, `WantedPull`. Unlike
/// [`submit_boot_tree`] this runs on **every** boot, quiet or not: these
/// aren't config-drift work, they're startup housekeeping that always needs
/// to happen, and the point of moving them here is exactly so they show up
/// as real work on the dashboard instead of an invisible `tokio::spawn` that
/// only surfaces on failure. Independent, build-slot- and lease-exempt
/// roots — no dependency edges between them, matching the existing
/// `Reconcile`-root pattern in [`boot_nodes`]. `KnowledgePull` holds its
/// sweep's own resource, so it queues behind any pass of the same sweep
/// already running.
fn submit_startup_sweep_nodes(coord: &Arc<Coordinator>) {
    use crate::job_queue::NodeKind;

    if let Err(e) = coord.job_queue.insert_job(|b| {
        let _ = crate::job_queue::templates::knowledge_pull(b);
        let _ = b.node(NodeKind::WantedPull);
        Vec::new()
    }) {
        tracing::warn!(error = ?e, "boot: startup sweep DAG insert failed");
    }
}

/// The boot DAG's node declarations, split out of [`submit_boot_tree`] so they
/// can be exercised without a live [`Coordinator`].
///
/// That split is not cosmetic: this path constructs nodes outside
/// `job_queue/`, so it is the one place a resource declaration can be forgotten
/// without any in-module test noticing. It has happened once already — the
/// boot `Reconcile`s silently declared nothing when kind-derived resources
/// were removed, which drops the agent lease a boot reconcile needs to not
/// race another DAG's container ops.
pub(crate) fn boot_nodes(
    builder: &crate::job_queue::JobBuilder,
    fanout: &[String],
    drifted: Vec<String>,
) {
    use crate::job_queue::NodeKind;
    use crate::job_queue::resource::Resource;

    // One graceful rebuild subgraph per stale agent, each an independent
    // group root: agents that were mid-turn when the host came up drain rather
    // than being cut off, and the rebuilds overlap on their own leases.
    for agent in fanout {
        let _ = crate::job_queue::templates::graceful_rebuild_nodes(builder, agent, None);
    }
    // One boot Reconcile per drifted agent — independent roots.
    for name in drifted {
        // Name the lease before the agent string moves into the kind.
        let lease = Resource::Agent(name.clone());
        let _ = builder
            .node(NodeKind::Reconcile { agent: name })
            .needs(lease);
    }
}

/// Submit this boot's work as **one DAG** (no anchor node, no per-agent
/// child DAGs). It holds one graceful rebuild subgraph per stale agent, and
/// every drifted agent gets a boot `Reconcile` as an independent root, so they
/// converge concurrently with the rebuilds. No-op when there's nothing to do.
fn submit_boot_tree(
    coord: &Arc<Coordinator>,
    fanout: &[String],
    drifted: Vec<String>,
    n_deferred: usize,
    n_skipped: usize,
) {
    // Fully-quiet boot (nothing to rebuild, nothing drifted) inserts nothing.
    if fanout.is_empty() && drifted.is_empty() {
        return;
    }
    // The summary the sweep used to hand the container as its `reason` is a log
    // line now: it was only ever stored on a node nobody read, and the counts
    // are worth having where they can actually be seen.
    tracing::info!(
        rebuilds = fanout.len(),
        reconciles = drifted.len(),
        deferred = n_deferred,
        up_to_date = n_skipped,
        "boot: sweep"
    );

    // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they
    // land; the boot DAG as a whole has no terminal side effect, so no tail.
    // The subgraphs also carry their own per-agent crash-watch suppression
    // during their `Swap` (applied at claim time); a reconcile-only boot needs
    // no transient.
    if let Err(e) = coord.job_queue.insert_job(|b| {
        boot_nodes(b, fanout, drifted);
        Vec::new()
    }) {
        tracing::warn!(error = ?e, "boot: sweep DAG insert failed");
    }
    coord.emit_rebuild_queue_snapshot();
}

#[cfg(test)]
mod tests {
    use super::{BootAction, boot_action, is_fresh};
    use crate::power::Wanted;

    // The regression the `Unreadable` variant exists for: the arm used to
    // substitute `from_running(running)`, which is `Noop` against BOTH
    // observations — so no combination of inputs could reach a reconcile.
    #[test]
    fn an_unreadable_intent_is_never_settled_whatever_the_agent_is_doing() {
        for fresh in [true, false] {
            for running in [true, false] {
                assert_eq!(
                    boot_action(None, fresh, running),
                    BootAction::Unreadable,
                    "fresh={fresh} running={running}"
                );
            }
        }
    }

    #[test]
    fn a_fresh_agent_doing_what_it_should_is_skipped_without_a_reconcile() {
        assert_eq!(
            boot_action(Some(Wanted::Up), true, true),
            BootAction::Skip { reconcile: false }
        );
        assert_eq!(
            boot_action(Some(Wanted::Offline), true, false),
            BootAction::Skip { reconcile: false }
        );
    }

    #[test]
    fn a_fresh_agent_that_drifted_from_its_intent_is_reconciled() {
        assert_eq!(
            boot_action(Some(Wanted::Up), true, false),
            BootAction::Skip { reconcile: true }
        );
        assert_eq!(
            boot_action(Some(Wanted::Offline), true, true),
            BootAction::Skip { reconcile: true }
        );
    }

    #[test]
    fn a_stale_agent_wanted_up_is_rebuilt_running_or_not() {
        assert_eq!(
            boot_action(Some(Wanted::Up), false, true),
            BootAction::Rebuild
        );
        assert_eq!(
            boot_action(Some(Wanted::Up), false, false),
            BootAction::Rebuild
        );
    }

    #[test]
    fn a_stale_agent_wanted_offline_defers_and_still_reports_drift() {
        assert_eq!(
            boot_action(Some(Wanted::Offline), false, false),
            BootAction::Defer { reconcile: false }
        );
        assert_eq!(
            boot_action(Some(Wanted::Offline), false, true),
            BootAction::Defer { reconcile: true }
        );
    }

    /// A changed declaration is rebuilt with no operator step, and with no read
    /// of the forge: the boot pass reaches `Rebuild` from the record and marker
    /// alone.
    #[test]
    fn a_changed_declaration_for_a_wanted_up_agent_is_rebuilt_unprompted() {
        let fresh = is_fresh(Some("/nix/store/new"), None, false);
        assert_eq!(
            boot_action(Some(Wanted::Up), fresh, true),
            BootAction::Rebuild
        );
        let fresh = is_fresh(None, None, false);
        assert_eq!(
            boot_action(Some(Wanted::Up), fresh, true),
            BootAction::Skip { reconcile: false },
            "control: nothing declared, nothing rebuilt"
        );
    }

    #[test]
    fn a_declaration_is_attempted_once_and_never_beside_a_live_rebuild() {
        assert!(is_fresh(None, None, false), "nothing pending");
        assert!(
            is_fresh(None, Some("/nix/store/b"), false),
            "nothing pending now, whatever was attempted"
        );
        assert!(
            !is_fresh(Some("/nix/store/b"), None, false),
            "control: a new declaration"
        );
        assert!(
            is_fresh(Some("/nix/store/b"), None, true),
            "a rebuild is queued"
        );
        assert!(
            is_fresh(Some("/nix/store/b"), Some("/nix/store/b"), false),
            "already attempted"
        );
        assert!(
            !is_fresh(Some("/nix/store/c"), Some("/nix/store/b"), false),
            "a newer declaration is attempted again"
        );
    }
}