//! Which fleet node should run the next microVM phase, and whether any can.
//!
//! # What this replaces
//!
//! Placement was `capable.first()` over a list ordered `last_seen DESC`
//! (`mission_orchestrator`, `nodes::online_for_backend`) — the most recently
//! heartbeated node. Among healthy nodes all heartbeating every 5s that is
//! arbitrary, and it consults nothing about load: two missions launched together
//! land on the same machine. It did not matter while one node held the only
//! rootfs image; all three do now.
//!
//! # Observed memory is not capacity
//!
//! The correctness core, and the reason this is not a one-line sort change. A VM
//! that booted 30 seconds ago holds a fraction of its 8 GiB claim — the guest has
//! not touched the rest — so `mem_pct` reports a sold-out node as nearly idle.
//! Ranking on utilisation alone would happily book five more VMs onto a node with
//! room for one. `capacity_of` therefore takes the WORSE of observed usage and
//! committed usage, and `a_sold_out_node_is_not_mistaken_for_an_idle_one` is the
//! negative control that pins it.
//!
//! Commitments come from two places that must be unioned by IDENTITY, never
//! added: `microvm_client::list` (booted VMs, including orphans nothing has
//! reaped) and `nodes::pinned_microvm_phases` (chosen but not yet booted). The
//! deterministic `vm_id_for` is what lets the same phase be recognised in both.
//!
//! # Fail-closed
//!
//! A node whose health is stale, whose daemon will not answer, or which is
//! draining is INELIGIBLE, not low-scoring. Unknown is not permission — the same
//! rule `nodes::online_for_backend` already applies to capabilities. The one
//! exception is Beszel metrics: they feed `headroom` as a tiebreak only, so stale
//! metrics demote a node instead of excluding it.

use cm_db::repo::node_metrics::EvalRow;
use cm_domain::NodeId;

/// Memory a phase VM claims. Re-exported from the executor so there is ONE number
/// — a scheduler and a launcher that disagree about VM size is a fleet that
/// overcommits by exactly their difference.
pub(crate) use crate::microvm_executor::MEM_MIB as MEM_PER_VM_MIB;

/// Held back for the host: the daemon, the OS, page cache, and the margin that
/// keeps a node out of swap. A node in swap makes every VM on it slow, so this is
/// cheaper than the alternative.
const HOST_RESERVE_MIB: i64 = 4096;

/// The floor we refuse to believe a host's own footprint is below. Without it, a
/// node reporting less used memory than its VMs have claimed would compute a
/// negative baseline and inflate its free memory.
const HOST_BASELINE_FLOOR_MIB: i64 = 2048;

/// Disk a VM may consume: an 8 GiB sparse rootfs plus room for the collected tar.
const DISK_PER_VM_GIB: i64 = 12;

/// Never let VM disk drive a node below this. `_outputs` and images live on the
/// same filesystem on some nodes.
const DISK_RESERVE_GIB: i64 = 20;

/// Health older than this and the node is ineligible. Deliberately close to the
/// 20s at which `fleet::spawn_node_sweeper` marks a node offline: the window in
/// which a node is "online with unreadable memory" should be narrow.
pub const MAX_HEALTH_AGE_SECS: f64 = 30.0;

/// Beszel metrics older than this rank as zero headroom. Only a tiebreak.
const MAX_METRICS_AGE_SECS: f64 = 60.0;

/// A node that can take at least one more phase VM.
#[derive(Debug, Clone, PartialEq)]
pub struct NodeCapacity {
    pub node_id: NodeId,
    pub name: String,
    /// How many MORE 8 GiB VMs fit.
    pub slots: i64,
    pub headroom: f64,
    pub committed_vms: i64,
    pub mem_total_mib: i64,
    pub used_eff_mib: i64,
    pub disk_free_gib: i64,
}

/// Why a node cannot take this phase. Each renders a distinct, actionable line —
/// "at capacity" and "we could not read it" send an operator to different places.
#[derive(Debug, Clone, PartialEq)]
pub enum Unfit {
    Draining,
    NotConnected,
    NoRecentHealth { age_secs: Option<f64> },
    CapacityUnknown { err: String },
    AtCapacity { committed: i64, used_eff_mib: i64, mem_total_mib: i64 },
    NoDisk { free_gib: i64 },
}

impl Unfit {
    pub fn reason(&self) -> String {
        match self {
            Unfit::Draining => "draining".into(),
            Unfit::NotConnected => "daemon not connected".into(),
            Unfit::NoRecentHealth { age_secs } => match age_secs {
                Some(a) => format!("health {a:.0}s stale (max {MAX_HEALTH_AGE_SECS:.0}s)"),
                None => "never reported health".into(),
            },
            Unfit::CapacityUnknown { err } => format!("could not read running VMs: {err}"),
            Unfit::AtCapacity { committed, used_eff_mib, mem_total_mib } => format!(
                "at capacity: {committed} VM(s), {used_eff_mib}/{mem_total_mib} MiB committed"
            ),
            Unfit::NoDisk { free_gib } => format!("only {free_gib} GiB free"),
        }
    }
}

/// Why placement produced no node. Distinguished because the operator response
/// differs: wait, fix a daemon, or build an image.
#[derive(Debug, Clone)]
pub enum PlacementError {
    /// No node has the image / KVM at all. Not a capacity problem.
    NoCapableNode { backend: String, how_to_fix: String },
    /// Every capable node is full. Transient — the caller should queue.
    FleetAtCapacity { report: String },
    /// We could not READ capacity. Must never be reported as "full".
    FleetUnreadable { report: String },
    /// An explicit target exists but cannot take the work.
    TargetUnfit { node: NodeId, why: Unfit },
}

impl PlacementError {
    /// Whether the caller should wait and retry rather than fail the work.
    pub fn is_transient(&self) -> bool {
        matches!(
            self,
            PlacementError::FleetAtCapacity { .. } | PlacementError::FleetUnreadable { .. }
        )
    }

    pub fn message(&self) -> String {
        match self {
            PlacementError::NoCapableNode { backend, how_to_fix } => {
                format!("no online node can run backend {backend:?} — {how_to_fix}")
            }
            PlacementError::FleetAtCapacity { report } => format!(
                "fleet at capacity — a phase VM runs up to 60 min; this phase waits for a slot.\n{report}"
            ),
            PlacementError::FleetUnreadable { report } => format!(
                "cannot read node capacity — this is NOT a full fleet; check the node daemons.\n{report}"
            ),
            PlacementError::TargetUnfit { node, why } => format!(
                "node {} cannot take this phase: {}",
                node.as_uuid(),
                why.reason()
            ),
        }
    }
}

/// The whole capacity decision for one node, as pure arithmetic.
///
/// Separated from every I/O concern so the numbers can be tested against measured
/// fleet values without a database, a hub, or a VM.
pub fn capacity_of(
    node_id: NodeId,
    name: &str,
    mem_total_mib: i64,
    mem_used_mib: i64,
    disk_free_gib: i64,
    committed_vms: i64,
    headroom: f64,
) -> Result<NodeCapacity, Unfit> {
    // What the host itself costs, inferred by subtracting what the VMs claimed.
    // Floored because a node reporting less used than its VMs claim would
    // otherwise produce a negative baseline and invent free memory.
    let host_baseline =
        (mem_used_mib - committed_vms * MEM_PER_VM_MIB as i64).max(HOST_BASELINE_FLOOR_MIB);
    let committed_use = committed_vms * MEM_PER_VM_MIB as i64 + host_baseline;

    // The worse of the two views. Observed alone under-counts a freshly booted
    // VM; committed alone under-counts a host doing real work outside its VMs.
    let used_eff = mem_used_mib.max(committed_use);
    let free = mem_total_mib - used_eff - HOST_RESERVE_MIB;
    let slots = if free <= 0 { 0 } else { free / MEM_PER_VM_MIB as i64 };

    if disk_free_gib - DISK_PER_VM_GIB < DISK_RESERVE_GIB {
        return Err(Unfit::NoDisk { free_gib: disk_free_gib });
    }
    if slots < 1 {
        return Err(Unfit::AtCapacity {
            committed: committed_vms,
            used_eff_mib: used_eff,
            mem_total_mib,
        });
    }
    Ok(NodeCapacity {
        node_id,
        name: name.to_string(),
        slots,
        headroom,
        committed_vms,
        mem_total_mib,
        used_eff_mib: used_eff,
        disk_free_gib,
    })
}

/// Admission inputs drawn from an `EvalRow`, or why the node is ineligible.
///
/// Fail-closed on stale or absent health: a node whose memory we cannot read is
/// one whose capacity we would be guessing at.
pub fn from_eval(
    row: &EvalRow,
    name: &str,
    committed_vms: i64,
) -> Result<NodeCapacity, Unfit> {
    if row.status == "draining" {
        return Err(Unfit::Draining);
    }
    let fresh = row
        .health_age_secs
        .is_some_and(|a| a <= MAX_HEALTH_AGE_SECS);
    let (Some(total), Some(used)) = (row.mem_total_bytes, row.mem_used_bytes) else {
        return Err(Unfit::NoRecentHealth { age_secs: row.health_age_secs });
    };
    if !fresh || total <= 0 {
        return Err(Unfit::NoRecentHealth { age_secs: row.health_age_secs });
    }
    const MIB: i64 = 1024 * 1024;
    const GIB: i64 = 1024 * 1024 * 1024;
    capacity_of(
        row.node_id,
        name,
        total / MIB,
        used / MIB,
        row.disk_free_bytes.unwrap_or(0) / GIB,
        committed_vms,
        row.headroom_fresh(MAX_METRICS_AGE_SECS),
    )
}

/// Rank admissible nodes: most free slots first, then live headroom, then id.
///
/// Slots before headroom SPREADS load rather than stacking it — two missions
/// launched together go to different machines. Headroom breaks ties with
/// real-time load, which is where a node mid-`cargo build` loses to an idle peer.
/// Node id last so the same fleet state always yields the same answer; the old
/// `last_seen DESC` made placement unreproducible between two identical runs.
pub fn rank(mut fit: Vec<NodeCapacity>) -> Vec<NodeCapacity> {
    fit.sort_by(|a, b| {
        b.slots
            .cmp(&a.slots)
            .then(
                b.headroom
                    .partial_cmp(&a.headroom)
                    .unwrap_or(std::cmp::Ordering::Equal),
            )
            .then(a.node_id.as_uuid().cmp(&b.node_id.as_uuid()))
    });
    fit
}

/// One line per node, for logs and for the message an operator reads.
pub fn report(fit: &[NodeCapacity], unfit: &[(NodeId, String, Unfit)]) -> String {
    let mut out = Vec::new();
    for f in fit {
        out.push(format!(
            "  {}: {} slot(s) free, {} VM(s) committed, {}/{} MiB, headroom {:.0}",
            f.name, f.slots, f.committed_vms, f.used_eff_mib, f.mem_total_mib, f.headroom
        ));
    }
    for (_, name, why) in unfit {
        out.push(format!("  {name}: UNFIT — {}", why.reason()));
    }
    if out.is_empty() {
        out.push("  (no capable nodes)".into());
    }
    out.join("\n")
}

/// Count a node's commitments, unioning booted VMs with pinned-not-yet-booted
/// phases BY IDENTITY.
///
/// A composed graph's step VMs (`...-s0`, `-s1`) each count: each is a real
/// Firecracker process holding 8 GiB. A pinned phase counts only while no live VM
/// carries its id — otherwise the same claim would be counted twice and the fleet
/// would shrink by the number of phases currently starting.
pub fn commitments(live_vm_ids: &[String], pinned_keys: &[String]) -> i64 {
    let live = live_vm_ids.len() as i64;
    let unbooted = pinned_keys
        .iter()
        .filter(|k| !live_vm_ids.iter().any(|v| v.starts_with(k.as_str())))
        .count() as i64;
    live + unbooted
}

/// Survey every capable node: which can take a phase VM, and why the rest cannot.
///
/// `vm_list` is asked of each candidate in parallel with a short deadline. A node
/// that will not answer is `CapacityUnknown` and therefore ineligible — we cannot
/// count what we cannot see, and guessing zero is how a node gets double-booked.
pub async fn survey(
    pool: &sqlx::PgPool,
    hub: &crate::fleet::NodeHub,
    workspace_id: uuid::Uuid,
    backend: Option<&str>,
) -> Result<(Vec<NodeCapacity>, Vec<(NodeId, String, Unfit)>), String> {
    let candidates = cm_db::repo::nodes::online_for_backend(pool, workspace_id, backend)
        .await
        .map_err(|e| format!("looking up nodes for backend {backend:?}: {e}"))?;
    if candidates.is_empty() {
        return Ok((Vec::new(), Vec::new()));
    }

    let evals = cm_db::repo::node_metrics::eval_all(pool)
        .await
        .map_err(|e| format!("reading node metrics: {e}"))?;
    let pinned = cm_db::repo::nodes::pinned_microvm_phases(pool, workspace_id)
        .await
        .map_err(|e| format!("reading pinned phases: {e}"))?;

    let names = node_names(pool, workspace_id).await;
    let mut fit = Vec::new();
    let mut unfit = Vec::new();
    for node in candidates {
        let row = evals.iter().find(|e| e.node_id == node);
        let name = names
            .get(&node.as_uuid())
            .cloned()
            .unwrap_or_else(|| node.as_uuid().to_string()[..8].to_string());

        // Not connected: nothing can be asked of it, and nothing can run on it.
        if !hub.is_connected(node) {
            unfit.push((node, name, Unfit::NotConnected));
            continue;
        }
        let Some(row) = row else {
            unfit.push((node, name, Unfit::NoRecentHealth { age_secs: None }));
            continue;
        };

        // Commitments: booted VMs unioned with phases pinned here but not yet
        // booted, by the deterministic id both sides agree on.
        let live = match crate::microvm_client::list(hub, node).await {
            Ok(v) => v,
            Err(e) => {
                unfit.push((node, name, Unfit::CapacityUnknown { err: e }));
                continue;
            }
        };
        let keys: Vec<String> = pinned
            .iter()
            .filter(|(n, _, _)| *n == node)
            .map(|(_, phase, iter)| crate::microvm_executor::vm_id_for(*phase, *iter, None))
            .collect();
        let committed = commitments(&live, &keys);

        match from_eval(row, &name, committed) {
            Ok(c) => fit.push(c),
            Err(why) => unfit.push((node, name, why)),
        }
    }
    Ok((rank(fit), unfit))
}

/// Node names for readable reports. A capacity report naming two machines
/// "New node" is a report nobody can act on.
async fn node_names(
    pool: &sqlx::PgPool,
    workspace_id: uuid::Uuid,
) -> std::collections::HashMap<uuid::Uuid, String> {
    sqlx::query_as::<_, (uuid::Uuid, String)>(
        "SELECT id, name FROM nodes WHERE workspace_id = $1",
    )
    .bind(workspace_id)
    .fetch_all(pool)
    .await
    .unwrap_or_default()
    .into_iter()
    .collect()
}

/// Choose a node for a phase, honouring an explicit target as a REQUEST.
///
/// `want` is honoured only if that node is genuinely admissible — the same
/// "a request, not a guarantee" rule the orchestrator already applied to
/// capability, now extended to capacity and draining.
pub async fn choose(
    pool: &sqlx::PgPool,
    hub: &crate::fleet::NodeHub,
    workspace_id: uuid::Uuid,
    backend: Option<&str>,
    want: Option<uuid::Uuid>,
) -> Result<NodeId, PlacementError> {
    let how_to_fix = format!(
        "needs /dev/kvm + firecracker (scripts/fc-node-setup.sh) AND the {} rootfs \
         built on that node (scripts/fc-build-rootfs.sh <host> <image> {})",
        backend.unwrap_or("default"),
        backend.unwrap_or("<name>")
    );
    let (fit, unfit) = survey(pool, hub, workspace_id, backend).await.map_err(|e| {
        PlacementError::FleetUnreadable { report: format!("  survey failed: {e}") }
    })?;

    if fit.is_empty() && unfit.is_empty() {
        return Err(PlacementError::NoCapableNode {
            backend: backend.unwrap_or("default").to_string(),
            how_to_fix,
        });
    }
    let report = report(&fit, &unfit);

    if let Some(want) = want {
        if let Some(c) = fit.iter().find(|c| c.node_id.as_uuid() == want) {
            return Ok(c.node_id);
        }
        if let Some((n, _, why)) = unfit.iter().find(|(n, _, _)| n.as_uuid() == want) {
            return Err(PlacementError::TargetUnfit { node: *n, why: why.clone() });
        }
        // Target is not even a candidate — fall through and place it somewhere
        // capable rather than failing, since the target was advisory.
    }

    if let Some(best) = fit.into_iter().next() {
        return Ok(best.node_id);
    }

    // Nothing fit. Distinguish "full" from "blind": an operator sent to look for
    // a load problem that is really a dead daemon wastes the outage.
    let blind = unfit.iter().all(|(_, _, w)| {
        matches!(w, Unfit::CapacityUnknown { .. } | Unfit::NotConnected | Unfit::NoRecentHealth { .. })
    });
    Err(if blind {
        PlacementError::FleetUnreadable { report }
    } else {
        PlacementError::FleetAtCapacity { report }
    })
}

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

    fn nid(n: u128) -> NodeId {
        NodeId::from(uuid::Uuid::from_u128(n))
    }

    /// THE test. Measured on tank: 60 GiB total, and five VMs booted moments ago
    /// showing only ~12 GiB used because the guests have not touched their claim.
    ///
    /// Observed-usage-only arithmetic says (61440-12000-4096)/8192 = 5 more VMs.
    /// The node has room for ONE. Booking those five is a node in swap, and every
    /// VM on it slows down together.
    #[test]
    fn a_sold_out_node_is_not_mistaken_for_an_idle_one() {
        let observed_only =
            capacity_of(nid(1), "tank", 61440, 12000, 800, 0, 50.0).expect("fits");
        assert_eq!(
            observed_only.slots, 5,
            "this is what utilisation alone claims — the bug being fixed"
        );

        let with_commitments =
            capacity_of(nid(1), "tank", 61440, 12000, 800, 5, 50.0).expect("fits");
        assert_eq!(
            with_commitments.slots, 1,
            "five 8 GiB claims are already spoken for, whatever the guests have touched"
        );
    }

    /// The measured idle fleet. Numbers from `free`/`df` on the real machines, so
    /// a future change to the constants has to face what it does to real nodes.
    #[test]
    fn the_measured_fleet_gets_the_slots_it_actually_has() {
        // tank: 60 GiB, ~6 GiB used at idle.
        let tank = capacity_of(nid(1), "tank", 61440, 6144, 869, 0, 90.0).unwrap();
        assert_eq!(tank.slots, 6);
        // architect: 60 GiB, ~7 GiB used.
        let arch = capacity_of(nid(2), "architect", 61440, 7168, 388, 0, 90.0).unwrap();
        assert_eq!(arch.slots, 6);
        // morpheus: 31 GiB — deliberately the conservative 2, not 3. Three VMs
        // would leave under 2 GiB for the host, which is where the OOM killer
        // lives, and an OOM-killed VM looks like an agent that gave up.
        let morph = capacity_of(nid(3), "morpheus", 31744, 5120, 312, 0, 90.0).unwrap();
        assert_eq!(morph.slots, 2);
    }

    /// Spread, don't stack; then real load; then determinism.
    #[test]
    fn ranking_prefers_free_slots_then_headroom_then_a_stable_order() {
        let a = capacity_of(nid(1), "a", 61440, 6144, 800, 0, 40.0).unwrap(); // 6 slots
        let b = capacity_of(nid(2), "b", 61440, 6144, 800, 3, 90.0).unwrap(); // 3 slots
        assert_eq!(rank(vec![b.clone(), a.clone()])[0].name, "a", "more slots wins");

        // Equal slots → the node under less real load.
        let busy = capacity_of(nid(3), "busy", 61440, 6144, 800, 0, 10.0).unwrap();
        let idle = capacity_of(nid(4), "idle", 61440, 6144, 800, 0, 95.0).unwrap();
        assert_eq!(rank(vec![busy.clone(), idle.clone()])[0].name, "idle");

        // Equal on both → same answer twice. `last_seen DESC` could not promise this.
        let x = capacity_of(nid(9), "x", 61440, 6144, 800, 0, 50.0).unwrap();
        let y = capacity_of(nid(8), "y", 61440, 6144, 800, 0, 50.0).unwrap();
        assert_eq!(rank(vec![x.clone(), y.clone()])[0].name, "y");
        assert_eq!(rank(vec![y, x])[0].name, "y");
    }

    /// A booted VM and its pinned phase row are ONE claim, not two.
    #[test]
    fn commitments_union_by_identity_rather_than_adding() {
        let live = vec!["m-abc123def456-0".to_string(), "m-abc123def456-0-s2".to_string()];
        // Same phase as the live VMs: already counted.
        assert_eq!(commitments(&live, &["m-abc123def456-0".to_string()]), 2);
        // A different phase, pinned but not yet booted: a real additional claim.
        assert_eq!(
            commitments(&live, &["m-999888777666-0".to_string()]),
            3,
            "a phase chosen seconds ago holds 8 GiB no node can report yet"
        );
        assert_eq!(commitments(&[], &["m-1-0".into(), "m-2-0".into()]), 2);
    }

    /// Disk is a hard gate, and it is checked BEFORE capacity so the message
    /// names the real problem.
    #[test]
    fn a_node_short_of_disk_is_refused_even_with_memory_to_spare() {
        let e = capacity_of(nid(1), "tank", 61440, 6144, 25, 0, 90.0).unwrap_err();
        assert!(matches!(e, Unfit::NoDisk { free_gib: 25 }), "{e:?}");
        assert!(e.reason().contains("25 GiB"));
    }

    /// Stale metrics may cost a tie; they may never win one, and they may never
    /// exclude a node — that is health's job.
    #[test]
    fn stale_beszel_metrics_demote_but_do_not_exclude() {
        let mut row = row_for(nid(1), 61440 * MIB_T, 6144 * MIB_T, 800 * GIB_T);
        row.metrics_age_secs = Some(3600.0);
        row.health_age_secs = Some(3.0);
        row.cpu_pct = Some(5.0);
        let fit = from_eval(&row, "tank", 0).expect("still eligible");
        assert_eq!(fit.headroom, 95.0, "fresh health carries the headroom");

        row.health_age_secs = Some(3600.0);
        assert!(
            matches!(from_eval(&row, "tank", 0), Err(Unfit::NoRecentHealth { .. })),
            "stale HEALTH is exclusion, because memory is then a guess"
        );
    }

    /// The two failures an operator must never confuse.
    #[test]
    fn unreadable_capacity_never_reads_as_a_full_fleet() {
        let full = PlacementError::FleetAtCapacity { report: "  tank: 0 slots".into() };
        let blind = PlacementError::FleetUnreadable { report: "  tank: UNFIT".into() };
        assert!(full.message().contains("at capacity"));
        assert!(blind.message().contains("cannot read"));
        assert!(
            !blind.message().contains("at capacity"),
            "sends an operator hunting a load problem that does not exist"
        );
        assert!(full.is_transient() && blind.is_transient());
        let missing = PlacementError::NoCapableNode {
            backend: "claude".into(),
            how_to_fix: "build the image".into(),
        };
        assert!(!missing.is_transient(), "a missing image will not fix itself by waiting");
    }

    const MIB_T: i64 = 1024 * 1024;
    const GIB_T: i64 = 1024 * 1024 * 1024;

    fn row_for(node_id: NodeId, total: i64, used: i64, disk_free: i64) -> EvalRow {
        EvalRow {
            node_id,
            workspace_id: cm_domain::WorkspaceId::from(uuid::Uuid::from_u128(1)),
            status: "online".into(),
            cpu_pct: None,
            mem_pct: None,
            disk_pct: None,
            gpu_pct: None,
            temp_max: None,
            load1: None,
            mem_total_bytes: Some(total),
            mem_used_bytes: Some(used),
            disk_free_bytes: Some(disk_free),
            health_age_secs: Some(3.0),
            metrics_age_secs: Some(3.0),
        }
    }

    /// A draining node is ineligible, not merely unattractive. The microVM path
    /// never checked this before: a mission pinned before a drain kept feeding
    /// VMs to a node an operator had cordoned.
    #[test]
    fn a_draining_node_is_ineligible() {
        let mut row = row_for(nid(1), 61440 * MIB_T, 6144 * MIB_T, 800 * GIB_T);
        row.status = "draining".into();
        assert_eq!(from_eval(&row, "tank", 0), Err(Unfit::Draining));
    }
}