//! Boot reconcile: on `hive-c0re serve` boot, (a) run the config path
//! for agents whose per-agent rev marker is stale — one `Boot` DAG
//! (meta hyperhive lock bump) that grows a `Rebuild` subgraph for each
//! stale agent whose `wanted` power intent is `Up` — and (b)
//! converge every other drifted agent to its persisted `wanted` via
//! `Reconcile` DAGs. Two rules keep boot-time 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). The sweep
//!    parent still runs whenever *any* marker is stale so the meta
//!    hyperhive lock is bumped for those later start-upgrades.
//! 2. **Agents whose rev marker matches** the current hyperhive flake
//!    path are skipped — nothing changed, no nix work to do.
//!
//! Booting with no config change performs no meta commit — only
//! reconciles. See `docs/coordinator.md::Boot reconcile`.

use std::path::Path;
use std::sync::Arc;

use anyhow::Result;

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

/// 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())
}

/// Returns true when the applied repo has commits that have not yet been
/// deployed (i.e. the applied HEAD differs from the sha currently locked in
/// meta's flake.lock). This is the semantic the dashboard `needs_update` chip
/// conveys: "there is a config change ready to apply via rebuild."
///
/// Async on purpose: this runs per agent inside `container_view::build_all`,
/// which fires on the ~10s dashboard sweep, every `AgentStatus` request, and
/// every `rescan_containers_and_emit` after a lifecycle step. A synchronous
/// `git` fork here blocks a tokio worker for the whole exec — under
/// nix-build disk saturation that's long enough that concurrent sweeps
/// starved the runtime and stalled the per-agent sockets.
pub async fn agent_config_pending(name: &str, deployed_sha: Option<&str>) -> bool {
    let applied = crate::paths::applied_dir(name);
    let applied_head = tokio::process::Command::new("git")
        .args(["-C", &applied.to_string_lossy(), "rev-parse", "HEAD"])
        .output()
        .await
        .ok()
        .filter(|o| o.status.success())
        .and_then(|o| String::from_utf8(o.stdout).ok())
        .map(|s| s.trim().to_owned());

    match (applied_head.as_deref(), deployed_sha) {
        (Some(head), Some(sha)) => !head.starts_with(sha) && !sha.starts_with(head),
        _ => false,
    }
}

/// Whether this hive is "ruthless" — running with no root/manager agent at
/// all (no ruth). When true, hive-c0re skips the root-agent create/start
/// sweep entirely. Controlled by the host option
/// `services.hyperhive.ruthless`, threaded in via the `HYPERHIVE_RUTHLESS`
/// env var. Defaults to `false` when the var is unset (back-compat: the
/// root agent was always auto-managed before this opt-out existed); only
/// an explicit `true` / `1` / `yes` enables ruthless mode.
fn ruthless() -> bool {
    match std::env::var("HYPERHIVE_RUTHLESS") {
        Ok(v) => matches!(v.trim().to_ascii_lowercase().as_str(), "true" | "1" | "yes"),
        Err(_) => false,
    }
}

/// Auto-create the manager container on startup if it isn't already there.
/// hive-c0re manages the manager end-to-end: operators no longer declare
/// `containers.h-ruth` in their host NixOS config. Bypasses the approval
/// queue — the root/manager is auto-managed by default. Operators who
/// don't want a root agent at all set `services.hyperhive.ruthless = true`,
/// which short-circuits this whole function. Idempotent.
pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
    if ruthless() {
        tracing::info!(
            "ruthless mode (services.hyperhive.ruthless = true) - skipping root agent create/start"
        );
        return Ok(());
    }
    let existing = lifecycle::list().await.unwrap_or_default();
    let current_rev = current_flake_rev(&coord.hyperhive_flake);
    if existing
        .iter()
        .any(|c| c.strip_prefix(AGENT_PREFIX) == Some(MANAGER_NAME))
    {
        // Container exists already. If it predates the unified lifecycle
        // (no applied flake on disk) we must rebuild — otherwise it's
        // running whatever the host-declarative config was at create
        // time, with a wrong systemd unit and port.
        let applied_flake = crate::paths::applied_dir(MANAGER_NAME).join("flake.nix");
        if !applied_flake.exists() && current_rev.is_some() {
            tracing::warn!(
                "manager container exists but no applied flake — forcing rebuild to migrate"
            );
            if let Err(e) = coord.job_queue.submit(crate::job_queue::templates::rebuild(
                MANAGER_NAME,
                crate::job_queue::Source::AutoUpdate,
                "manager migration: no applied flake".to_owned(),
                true,
            )) {
                tracing::warn!(error = ?e, "manager migration rebuild submit failed");
            }
        } else {
            tracing::debug!("manager container already present");
        }
        // hive-c0re auto-manages the root/manager by default, so a
        // present-but-stopped root (e.g. a first-start failure on a fresh
        // install) is brought back up here: the startup sweep's rebuild only
        // restarts a container that was already running, so without this it
        // stays down until a manual `nixos-container start`. The sub-agent
        // `was_running` guard is intentionally left untouched. (Operators
        // opt out of this whole auto-management with
        // `services.hyperhive.ruthless = true`, gated at the top of
        // this function.)
        if !lifecycle::is_running(MANAGER_NAME).await {
            tracing::info!("manager container present but not running — starting");
            if let Err(e) = coord.power.set(MANAGER_NAME, crate::power::Wanted::Up) {
                tracing::warn!(error = ?e, "agent_power: set manager wanted=up failed");
            }
            if let Err(e) = lifecycle::start(MANAGER_NAME).await {
                tracing::warn!(error = ?e, "manager start failed");
            }
        }
        return Ok(());
    }
    tracing::info!("manager container missing — spawning");
    // lifecycle::spawn creates the runtime dir internally; no manual
    // ensure_agent_runtime_dir needed here.
    let runtime = crate::paths::agent_runtime_dir(MANAGER_NAME);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(MANAGER_NAME, runtime);
    lifecycle::spawn(MANAGER_NAME, &hive, &paths).await?;
    if let Err(e) = coord.power.set(MANAGER_NAME, crate::power::Wanted::Up) {
        tracing::warn!(error = ?e, "agent_power: set manager wanted=up failed");
    }
    if let Some(rev) = current_rev {
        let _ = std::fs::write(crate::paths::applied_rev_marker(MANAGER_NAME), &rev);
    }
    Ok(())
}

/// Sort `names` in-place so parents precede their children in the topology.
/// Uses BFS from root agents (depth 0). Agents absent from `topo` sort last,
/// alphabetically within their tier. Stable within each depth tier.
pub fn topology_sort(
    names: &mut [String],
    topo: &std::collections::BTreeMap<String, Option<String>>,
) {
    use std::collections::{HashMap, VecDeque};
    // Build depth map using owned clones so the borrow on `names` is released
    // before the sort_by mutable borrow.
    let name_set: Vec<String> = names.to_vec();
    let mut depth: HashMap<String, usize> = HashMap::new();
    let mut queue: VecDeque<String> = VecDeque::new();
    // Seed roots: entries with no parent, or names not present in topo at all.
    for name in &name_set {
        if topo.get(name).is_none_or(Option::is_none) {
            depth.insert(name.clone(), 0);
            queue.push_back(name.clone());
        }
    }
    // BFS to assign depths to children.
    while let Some(parent) = queue.pop_front() {
        let d = depth[&parent] + 1;
        for name in &name_set {
            let is_child = topo.get(name).and_then(|p| p.as_deref()) == Some(parent.as_str());
            if is_child && !depth.contains_key(name) {
                depth.insert(name.clone(), d);
                queue.push_back(name.clone());
            }
        }
    }
    names.sort_by(|a, b| {
        let da = depth.get(a).copied().unwrap_or(usize::MAX);
        let db = depth.get(b).copied().unwrap_or(usize::MAX);
        da.cmp(&db).then(a.cmp(b))
    });
}

/// Boot reconcile (see the module doc): classify every agent by rev
/// freshness + persisted `wanted` intent, submit one `Boot` DAG
/// (hyperhive lock bump growing an in-DAG rebuild subgraph per stale
/// wanted-up agent) when anything is stale, and `Reconcile` DAGs for
/// agents whose observed power state drifted from `wanted`. Returns Ok even
/// if some submissions failed.
pub async fn run(coord: Arc<Coordinator>) -> Result<()> {
    let containers = match lifecycle::list().await {
        Ok(c) => c,
        Err(e) => {
            tracing::warn!(error = ?e, "boot reconcile: nixos-container list failed");
            return Ok(());
        }
    };

    let current_rev = current_flake_rev(&coord.hyperhive_flake);

    // Resolve container names to logical agent names, then sort by
    // topology depth so parents are always rebuilt before their
    // children. Root agents (depth 0) go first; agents absent from
    // the topology file sort last (stable, alphabetical within tier).
    let mut logical_names: Vec<String> = containers
        .iter()
        .filter_map(|c| c.strip_prefix(AGENT_PREFIX).map(str::to_owned))
        .collect();
    let topo = crate::topology::read();
    topology_sort(&mut logical_names, &topo);

    // Classify. `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.
    let mut any_stale = false;
    // stale ∧ wanted=Up → sweep rebuild. The bool is the agent's observed
    // running state, used to order running agents first before submit.
    let mut fanout: Vec<(String, bool)> = Vec::new();
    let mut drifted: Vec<String> = Vec::new(); // fresh ∧ wanted≠observed → reconcile
    let mut n_deferred = 0usize;
    let mut n_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) => w,
            Err(e) => {
                tracing::warn!(%name, error = ?e, "agent_power read failed — assuming observed");
                crate::power::Wanted::from_running(running)
            }
        };
        let fresh = current_rev.as_ref().is_some_and(|rev| {
            std::fs::read_to_string(crate::paths::applied_rev_marker(name))
                .is_ok_and(|stored| stored == rev.as_str())
        });
        if fresh {
            n_skipped += 1;
        } else {
            any_stale = true;
            if wanted == crate::power::Wanted::Up {
                // Rebuild against the post-bump lock; the DAG's tail
                // Reconcile brings the agent (back) up — covering both
                // the running-stale and stopped-but-wanted-up cases.
                fanout.push((name.clone(), running));
                continue;
            }
            // Stale but wanted offline: no boot-time nix work — the
            // start submit path upgrades a stale start to a rebuild.
            n_deferred += 1;
            tracing::debug!(%name, "boot reconcile: stale but offline — deferring rebuild to on-start");
        }
        if crate::power::reconcile_action(wanted, running) != crate::power::ReconcileAction::Noop {
            drifted.push(name.clone());
        }
    }

    tracing::info!(
        total = containers.len(),
        rebuilds = fanout.len(),
        reconciles = drifted.len(),
        deferred = n_deferred,
        up_to_date = n_skipped,
        "boot reconcile"
    );

    // Rebuild running agents first. All fanout entries are wanted=Up;
    // among them, warm the live/serving agents onto the fresh config before the
    // stopped-but-wanted-up ones so the scarce build slots hit uptime-critical
    // agents first. Stable sort keeps topology order (parents before children)
    // within each running/stopped group. The `drifted` reconciles aren't sorted
    // — they hold no build slot and run concurrently, so their order is moot.
    fanout.sort_by_key(|(_, running)| !running);
    let fanout: Vec<String> = fanout.into_iter().map(|(name, _)| name).collect();

    submit_boot_tree(&coord, any_stale, fanout, drifted, n_deferred, n_skipped);
    Ok(())
}

/// Submit this boot's work as **one DAG** (no anchor node, no per-agent
/// child DAGs). Node 0 is the sweep `MetaLock` (only when
/// something is stale) — its executor bumps the hyperhive lock, then grows
/// one rebuild subgraph per stale agent into *this same* DAG (rooted on the
/// `MetaLock`, so they build against the post-bump lock; see
/// `exec::run_meta_lock`). Every drifted agent gets a boot `Reconcile` as an
/// independent root — a boot reconcile needs no lock bump, so it converges
/// concurrently with the sweep. No-op when there's nothing to do.
fn submit_boot_tree(
    coord: &Arc<Coordinator>,
    any_stale: bool,
    fanout: Vec<String>,
    drifted: Vec<String>,
    n_deferred: usize,
    n_skipped: usize,
) {
    use crate::job_queue::{DagSpec, NodeKind, NodeSpec, Source};

    // Fully-quiet boot (nothing stale, nothing drifted) submits nothing.
    if !any_stale && drifted.is_empty() {
        return;
    }
    let reason = format!(
        "boot: {} rebuild(s), {} reconcile(s), {} deferred (offline), {} up-to-date",
        fanout.len(),
        drifted.len(),
        n_deferred,
        n_skipped,
    );

    let mut nodes: Vec<NodeSpec> = Vec::new();
    // Sweep whenever ANY marker is stale — even when every stale agent is
    // wanted-offline: the hyperhive lock bump must land now so their later
    // start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock
    // ⇒ no meta commit on a no-change boot. The `fanout` list rides the
    // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs.
    if any_stale {
        nodes.push(NodeSpec {
            kind: NodeKind::MetaLock {
                sweep: true,
                fanout: Some(fanout),
                // A sweep bumps `hyperhive` alone (`lock_update_hyperhive`),
                // so it names no inputs.
                inputs: Vec::new(),
            },
            deps: Vec::new(),
            parent: None,
        });
    }
    // One boot Reconcile per drifted agent — independent roots.
    for name in drifted {
        nodes.push(NodeSpec {
            kind: NodeKind::Reconcile { agent: name },
            deps: Vec::new(),
            parent: None,
        });
    }

    let spec = DagSpec {
        // 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.
        source: Source::AutoUpdate,
        reason,
        // Rebuilding when the sweep will grow rebuild subgraphs (per-agent
        // crash-watch suppression during their Swap, applied at claim time);
        // a reconcile-only boot needs no transient.
        nodes,
    };
    if let Err(e) = coord.job_queue.submit(spec) {
        tracing::warn!(error = ?e, "boot: sweep DAG submit failed");
    }
    coord.emit_rebuild_queue_snapshot();
}