//! Generic job-DAG queue + desired-state reconciliation — the host-side
//! wrapper over the domain-agnostic [`hive_jobq`] scheduler. Jobs are nodes in
//! per-request DAGs (see [`templates`]); the special cases (graceful-stop
//! watcher, deferred-start follow-up, meta-update cascade) collapse into DAG
//! *shapes* over the shared node primitives ([`model::NodeKind`]).
//!
//! [`hive_jobq`] owns the graph, the two-class resource pool, and the roll-up
//! settle loop; this module maps hive-c0re's concepts onto it:
//! - [`model::NodeKind`] **is** the crate payload `N` directly — each variant
//!   carries the agent it targets ([`NodeKind::agent`]); the two resource
//!   classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease
//!   subtree-held), declared per node at its construction site;
//! - **a job has no container node.** A template declares its nodes and names
//!   the roots it wants back; `insert_job` returns those ids. Grouping is the
//!   parent axis (a root's rolled-up state *is* its subtree's), so membership is
//!   a graph walk with no host-side side-tables. The lease is owned by a subtree
//!   root and borrowed by its descendants (continuity);
//! - terminal work is an ordinary **tail node**
//!   ([`NodeKind::ResolveApproval`] / [`NodeKind::EmitRebuilt`]) that the builder
//!   appends in [`templates`], edged onto the job's group roots by the outcome it
//!   reports. Templates emit one tail per outcome and the graph runs exactly one,
//!   so nothing branches at runtime.
//!
//! The queue is runtime-only (no persistence): an empty graph on boot; desired
//! state is re-derived by the reconcile sweep. A single scheduler task
//! ([`scheduler::run_worker`]) drives it; concurrency comes from the build-slot
//! capacity, not multiple workers. Design: `docs/coordinator.md::Job queue`.

pub mod exec;
pub mod model;
pub mod power;
pub mod resource;
pub mod scheduler;
pub mod templates;
#[cfg(test)]
mod tests;

use std::sync::{Arc, Mutex};

use chrono::{DateTime, Utc};
use hive_jobq::resources::ResourceTable;
use hive_jobq::scheduler::{Outcome, Scheduler};
use hive_jobq::{Graph, NodeId};
use hive_jobq_wire::{GraphNode, GraphWire};
use tokio::sync::Notify;

pub use hive_jobq::TerminalState;
pub use model::{NodeKind, PermPayload, State};
use resource::Resource;

/// A job under construction: `hive_jobq`'s builder over this queue's payload
/// ([`NodeKind`]) and resource ([`Resource`]) types. Templates declare into a
/// borrowed one; only `hive_jobq` can make or insert it.
pub type JobBuilder = hive_jobq::JobBuilder<NodeKind, Resource>;

/// A handle to one node a template declared — where its edges, grouping and
/// resources are declared. `Copy`; naming a node as a dependency does not
/// consume the ability to name it again.
pub type Handle<'a> = hive_jobq::NodeRef<'a, NodeKind, Resource>;

/// How many terminal DAGs (`Done` / `Failed` / `Cancelled`) the snapshot
/// retains, newest first. A flat cap over the whole sorted list: the
/// dashboard renders one recent-builds list, so one number bounds it.
const MAX_HISTORY_DAGS: usize = 50;

/// Cap on stored node error strings.
const MAX_ERROR_LEN: usize = 2_000;

/// One live transient pill, derived from a running node.
///
/// A named struct rather than a tuple because three of its four fields are
/// easy to confuse at a call site: two are strings and two answer questions
/// nobody should have to guess at ("is this the agent or the label?", "does
/// this bool mean deliberate or running?").
#[derive(Debug, Clone)]
pub struct RunningTransient {
    /// The agent whose lease the node declared.
    pub agent: String,
    /// The node's own wire tag, rendered as the pill.
    pub label: String,
    /// Whether this operation is expected to take the container down — the
    /// crash watcher's input. See [`NodeKind::takes_container_down`].
    pub takes_container_down: bool,
    /// When the node started running, so the dashboard can tick elapsed
    /// seconds. Taken from the node itself, which is the true start of the
    /// operation rather than the moment a watcher noticed it.
    pub since: DateTime<Utc>,
}

/// The crate scheduler, specialised to this host's node + resource types.
///
/// A job is **just its nodes** — no container, no grouping side-tables. A
/// root's rolled-up state is its subtree's, so membership is a graph walk and
/// "which job is this node in" is [`JobQueue::root_of`]. One shared crate
/// [`Graph`] holds every job's nodes.
///
/// There is deliberately **no wrapper struct and no per-node side map**. The
/// last map held the `build_logs` row id; that link now lives on the log row
/// itself (`build_logs.node_id`). With nothing else to guard, the mutex holds
/// the scheduler *directly* — which is what lets `hive_jobq` drive the run loop
/// (it takes `&Arc<Mutex<Scheduler<..>>>`, a type a host-side wrapper could not
/// satisfy).
type Sched = Scheduler<NodeKind, Resource>;

/// The queue. Lives on `Coordinator` (one per hive-c0re process); a single
/// scheduler task ([`scheduler::run_worker`]) drives it.
pub struct JobQueue {
    /// The scheduler, held directly rather than behind a host-side wrapper —
    /// `hive_jobq`'s run-loop seam takes `&Arc<Mutex<Scheduler<..>>>`, so this
    /// *is* the type the crate drives.
    sched: Arc<Mutex<Sched>>,
    /// Wakes the scheduler when something new arrives or state changed.
    pub(crate) notify: Notify,
}

impl std::fmt::Debug for JobQueue {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("JobQueue").finish_non_exhaustive()
    }
}

impl Default for JobQueue {
    fn default() -> Self {
        Self::new(1)
    }
}

/// A node runner's `Result` as the scheduler's [`Outcome`].
///
/// The failure reason + `finished_at` are stamped onto the graph `Node` by the
/// scheduler (the reason rides `Outcome::Failed`); there is no host-side copy,
/// so nothing needs clearing on success.
fn outcome_of(result: Result<(), String>) -> Outcome {
    match result {
        Ok(()) => Outcome::Done,
        Err(e) => Outcome::Failed(truncate_error(&e)),
    }
}

// `insert_group` lived here: a `group_parent`-taking insert whose only
// remaining caller was the DAG container, everything under it. Runtime growth
// never went through it — an executor declares into the builder `hive_jobq`
// hands it, which parents the new work under the emitting node by
// construction. With no container to be the other kind of parent, the
// distinction it existed to express is gone.

impl JobQueue {
    #[must_use]
    pub fn new(build_slots: usize) -> Self {
        let mut table = ResourceTable::new();
        table.set_capacity(
            Resource::BuildSlot,
            u32::try_from(build_slots.max(1)).unwrap_or(u32::MAX),
        );
        Self {
            sched: Arc::new(Mutex::new(Scheduler::new(Graph::new(), table))),
            notify: Notify::new(),
        }
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, Sched> {
        self.sched.lock().expect("job_queue mutex poisoned")
    }

    /// Insert a job's nodes into the shared graph, then wake the run loop.
    ///
    /// Deliberately named for the [`hive_jobq`] primitive it wraps, because
    /// that is nearly all it is. **The wrapper earns its place on the wake**:
    /// the crate is sync and runtime-free — it holds no `Notify` at all — so
    /// the channel the run loop parks on belongs to the host, and something has
    /// to ping it. Left to call sites, an insert whose ping was forgotten would
    /// leave a correct DAG sitting unscheduled until an unrelated event
    /// happened along; nothing would fail, and no test in isolation would see
    /// it.
    ///
    /// Returns exactly what the primitive returns: the ids of the nodes the
    /// template named, in the order it named them.
    ///
    /// # Errors
    /// Propagates a graph-insert error (dependencies that aren't
    /// dependency-topological).
    pub fn insert_job(
        &self,
        declare: impl FnOnce(&JobBuilder) -> Vec<hive_jobq::NodeGuid>,
    ) -> anyhow::Result<Vec<NodeId>> {
        let mut inner = self.lock();
        let named = inner
            .insert_job(None, declare)
            .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
        drop(inner);
        self.notify.notify_one();
        Ok(named)
    }

    /// The scheduler itself, for `hive_jobq`'s run-loop seam
    /// (`Scheduler::claim_next`), which takes exactly this type.
    ///
    /// Handing out the `Arc` rather than wrapping each crate call keeps the
    /// host from growing a parallel API: the run loop uses `hive_jobq`'s
    /// functions directly, and this module stays the thin glue it is being
    /// reduced to.
    pub(crate) fn sched(&self) -> &Arc<Mutex<Sched>> {
        &self.sched
    }

    /// The id of the **group root** `node` belongs to, for log lines and the
    /// dashboard. Derived from the graph rather than carried alongside the node
    /// — the parent axis already knows it.
    ///
    /// Was `dag_of`, when a job's nodes hung under a container node that *was*
    /// the group. Without it the parent chain ends at whichever root the
    /// template declared, so this answers "which root owns this node", not
    /// "which DAG is this in" — there is no longer such a thing.
    #[must_use]
    pub fn root_of(&self, node: NodeId) -> Option<u64> {
        self.lock().graph().root_of(node).map(NodeId::get)
    }

    /// Cancel a DAG that hasn't started yet: every work node is still `Pending`,
    /// so each is cancelled. `false` once any work node is running or terminal —
    /// an in-flight nix build isn't interruptible.
    ///
    /// **Nodes that explicitly observe cancellation are spared** — a node whose
    /// edge names [`hive_jobq::TerminalState::Cancelled`] is asking to run when
    /// the work it follows was dropped, which is exactly what an approval tail
    /// needs: cancel the work, and the tail still fires to resolve the approval
    /// row rather than leaving it dangling forever.
    ///
    /// Nothing is special-cased by node kind. `AFTER_ANY` deliberately does *not*
    /// accept `Cancelled`, so an ordinary weak-edged step (rebuild's `Reconcile`,
    /// say) is cancelled along with everything else — there is nothing to converge
    /// when no node ever ran. Only a node that named `Cancelled` survives, and it
    /// survives because it asked to.
    ///
    /// `id` names **any node**, not specifically a DAG. Cancelling a group root
    /// drops that whole group (the cascade is the scheduler's), which is what
    /// the dashboard's whole-DAG cancel does; cancelling an interior node drops
    /// just that branch. Nothing here knows about DAGs.
    pub fn cancel(&self, id: u64) -> bool {
        let mut inner = self.lock();
        let Some(node) = inner.graph().resolve_id(id) else {
            return false;
        };
        if !inner.cancel_node(node) {
            return false;
        }
        drop(inner);
        self.notify.notify_one();
        true
    }

    /// The first failed node's error in `dag_id`, if any has failed yet.
    ///
    /// Unlike the roll-up summary this is readable *mid-flight*, which is the
    /// point: a compensation node runs `AfterAny` its subject, so when it asks,
    /// the DAG is still `Finishing` (the compensation node itself is running)
    /// while the node it is compensating for has already settled `Failed`. That
    /// lets the compensation annotate its bookkeeping with the reason the deploy
    /// failed, instead of having the error handed down from the node that hit
    /// it. `None` when nothing has failed — the ordinary success path.
    #[must_use]
    pub fn first_error(&self, dag_id: u64) -> Option<String> {
        let inner = self.lock();
        let node = find_node(&inner, dag_id)?;
        inner.graph().first_error(node).map(ToOwned::to_owned)
    }

    /// `(agent, label, takes_container_down)` for the live transient-pill set,
    /// recomputed from the nodes **actually running** — not from an intent a
    /// template declared at submit time. (A rebuild used to report `rebuilding`
    /// for its whole life: prebuild, stop, swap, tail and reconcile alike.)
    ///
    /// **Status is the only test**: every `Running` node that names an agent is
    /// in the set. Naming is targeting, not lease-holding — `Prebuild` /
    /// `MetaSync` are lease-exempt (the container keeps serving through them)
    /// but they *are* work on that agent, and the operator wants to see it.
    ///
    /// ⚠️ **So there can be more than one entry per agent**, which is the whole
    /// difference from the older lease-declaration test: lease-exemption is
    /// exactly what lets one DAG build for `a` while another holds `a`'s lease,
    /// so both are running and both name `a`. Anything keying this set by agent
    /// alone will silently drop one — see [`super::scheduler`].
    ///
    /// `label` is the node's own wire tag ([`NodeKind::as_str`]), the same
    /// vocabulary the graph wire ships as a node's label, so a pill and a
    /// graph node name an operation identically. `takes_container_down` is the
    /// crash watcher's
    /// input and does **not** ride the wire to the frontend — a `Start` pill and
    /// a `Stop` pill are both pills; only one means a vanished container is
    /// expected.
    ///
    /// Not the lease *owner* either: `resource_state()` answers "who holds the
    /// slot", a different question.
    #[must_use]
    pub fn running_transients(&self) -> Vec<RunningTransient> {
        let inner = self.lock();
        inner
            .graph()
            .nodes()
            .filter(|n| matches!(n.state, State::Running))
            .filter_map(|n| {
                // Status is the only test. The agent comes off the node's own
                // payload, not off a declared `Resource::Agent` edge: the
                // lease-exempt kinds (`Prebuild` / `MetaSync`) name an agent
                // without declaring its lease, and they are work on that agent
                // that the operator wants to see.
                //
                // Empty means an agentless container kind (`MetaLock`, `Dag`),
                // which targets no agent and lights nothing.
                let agent = n.payload.agent();
                if agent.is_empty() {
                    return None;
                }
                Some(RunningTransient {
                    agent: agent.to_owned(),
                    label: n.payload.as_str().to_owned(),
                    takes_container_down: n.payload.takes_container_down(),
                    // `started_at` is set when a node enters `Running`, and this
                    // only sees `Running` nodes — the fallback is unreachable in
                    // practice, and "just now" is the honest answer if it isn't.
                    since: n.started_at.unwrap_or_else(Utc::now),
                })
            })
            .collect()
    }

    /// Every node of every visible group, as generic graph nodes.
    ///
    /// **Nothing is hidden**, other than an explicit `states` ask. Group
    /// roots ride as ordinary nodes (so a consumer needs no special case for
    /// "the container" and reads the root's own `state` as the group's
    /// answer), and `Done` nodes stay by default (so a finished step is
    /// visible rather than vanishing from the payload, which is what makes a
    /// fast rebuild render as a single node).
    ///
    /// `states`, when given, keeps every **individual node** (root or
    /// descendant) whose own state is named — not just whole root groups.
    /// A still-live group (root not yet terminal) can otherwise hold any
    /// number of already-finished steps inside it; filtering only at the
    /// root leaves every one of those visible regardless of the ask, which
    /// is exactly the clutter a state filter exists to remove. `None` (or
    /// the full state set) is the unfiltered call, matching prior
    /// behaviour. A slice rather than a set: `State` derives `Eq` but not
    /// `Hash`, and the vocabulary is 7 variants — a linear check per node
    /// costs nothing at that size.
    ///
    /// A node whose *parent* got filtered out still rides with its original
    /// `parent` id — `<hive-jobq-graph>` (the one consumer) already treats
    /// an unresolvable parent as a new root (`buildTree`'s fallback), so a
    /// filtered-out ancestor surfaces a still-matching descendant one level
    /// higher rather than hiding or orphaning it.
    ///
    /// The projection itself is [`hive_jobq_wire`]'s; all this layer supplies
    /// is *which* nodes to show — see [`visible_roots`] for why the graph
    /// can't decide the root-visibility half of that for itself.
    #[must_use]
    pub fn graph_snapshot(&self, states: Option<&[State]>) -> Vec<GraphNode> {
        let inner = self.lock();
        let roots = visible_roots(&inner);
        let nodes = inner.graph().wire_snapshot(roots);
        filter_nodes_by_state(nodes, states)
    }

    /// Per-state counts over the **same** groups [`Queue::graph_snapshot`]
    /// serves.
    ///
    /// Supplies the same two things and nothing else: the lock, and
    /// [`visible_roots`]. The counting is [`hive_jobq_wire::state_rollup`]'s and
    /// is generic over the payload — this is a call site, not an implementation.
    #[must_use]
    pub fn state_rollup(&self) -> Vec<hive_jobq_wire::StateCount> {
        let inner = self.lock();
        hive_jobq_wire::state_rollup(inner.graph(), visible_roots(&inner))
    }

    /// One or more nodes plus their live subtrees, as generic wire nodes —
    /// the `QueueNodes` polling surface behind `hivectl`'s wait/progress
    /// loop. Goes through [`GraphWire::wire_snapshot`] — the same projection
    /// [`Self::graph_snapshot`] serves the dashboard with, differing only in
    /// *which* nodes it selects (caller-named ids and their subtrees, rather
    /// than every visible root). No `Done`-node filtering, no
    /// roll-up field (a node's own `state` answers that, see
    /// `hive_jobq_wire`'s doc comment). Looks each id up by identity
    /// alone — no assumption that it names a DAG container or a root;
    /// "just show whatever the backend sends" for whatever ids the caller
    /// asks about. Multiple ids in one call is the normal shape for a
    /// batch op (e.g. restarting every agent submits one root per agent) —
    /// callers should request the whole batch together rather than poll
    /// one id per round-trip.
    ///
    /// An id with no matching node in the graph is silently dropped from
    /// the result rather than erroring the whole batch — some ids in a
    /// batch may already be evicted while others are still live. Today
    /// that only happens for a genuinely unknown id: nothing prunes the
    /// graph yet (bounded-prune is a Stage-C follow-up; [`visible_roots`]
    /// bounds the *view*, not the graph), so a completed group's nodes keep riding here
    /// with a terminal `state` rather than disappearing — callers
    /// watching for "done" should read the root's `state`, not absence.
    #[must_use]
    pub fn node_subtrees(&self, ids: &[u64]) -> Vec<GraphNode> {
        let inner = self.lock();
        let roots: Vec<NodeId> = ids.iter().filter_map(|id| find_node(&inner, *id)).collect();
        inner.graph().wire_snapshot(roots)
    }
}

/// The graph node whose id equals `id`, whatever its kind or depth.
/// `NodeId` is un-fabricable from a raw `u64`, so this is a search.
fn find_node(sched: &Sched, id: u64) -> Option<NodeId> {
    sched
        .graph()
        .nodes()
        .find_map(|n| (n.id.get() == id).then_some(n.id))
}

/// [`Queue::graph_snapshot`]'s `states` ask, applied to the already-
/// projected node list: keeps every node — root or descendant — whose own
/// `state` is named.
///
/// Applied *after* [`GraphWire::wire_snapshot`] rather than as a root
/// pre-filter — narrowing which roots are visible at all is
/// [`visible_roots`]'s job (a different question: how much settled work is
/// retained, full stop); this is "of what's retained and live, which
/// individual nodes does the caller want shown right now." `None` (or an
/// unrecognised/absent query) is the identity filter.
fn filter_nodes_by_state(nodes: Vec<GraphNode>, states: Option<&[State]>) -> Vec<GraphNode> {
    let Some(states) = states else {
        return nodes;
    };
    nodes
        .into_iter()
        .filter(|n| states.contains(&n.state))
        .collect()
}

/// The visible **group** set for [`Queue::graph_snapshot`]: every live group
/// root, plus the newest [`MAX_HISTORY_DAGS`] settled ones.
///
/// Selected *structurally* — a root is a node with no parent. The typed
/// projection this replaced keyed on the since-removed container kind instead,
/// which made the visible set depend on one host node kind; nothing here knows
/// what a node means.
///
/// **This bound is load-bearing, not tidiness.** Nothing ever removes a node
/// from the graph (bounded pruning is a Stage-C follow-up), so serving
/// `graph.roots()` directly would grow the payload without limit for the whole
/// uptime of the daemon.
fn visible_roots(sched: &Sched) -> Vec<NodeId> {
    let roots: Vec<NodeId> = sched.graph().roots().map(|n| n.id).collect();
    let mut live: Vec<NodeId> = Vec::new();
    let mut terminal: Vec<(NodeId, i64, u64)> = Vec::new();
    for root in roots {
        if sched.graph().is_settled(root) == Some(true) {
            terminal.push((root, group_finished_at(sched, root), root.get()));
        } else {
            live.push(root);
        }
    }
    retain_history(live, terminal, MAX_HISTORY_DAGS)
}

/// When a whole group last finished: the newest `finished_at` across the root
/// **and** its descendants.
///
/// The root itself counts, because a group root can be an ordinary node with
/// no children at all — reading only descendants would date every such group
/// to the epoch and evict it first. (The typed path this replaced read
/// descendants only, and could get away with it: its roots were always DAG
/// containers, which always have children.)
fn group_finished_at(sched: &Sched, root: NodeId) -> i64 {
    sched
        .graph()
        .node(root)
        .and_then(|n| n.finished_at)
        .into_iter()
        .chain(
            sched
                .graph()
                .descendants(root)
                .filter_map(|n| n.finished_at),
        )
        .map(|t| t.timestamp())
        .max()
        .unwrap_or(0)
}

/// [`visible_roots`]'s policy, split from the graph it reads: keep every live
/// group, plus the newest `cap` terminal ones.
///
/// `terminal` rows are `(handle, finished_at, tiebreak)`. The tiebreak orders
/// DAGs that settled inside the same wall-clock second — which is *most* of
/// them under a burst, and all of them in a test, so it is load-bearing rather
/// than a formality.
///
/// Generic over the handle purely so this is reachable without a graph: a
/// `NodeId` cannot be fabricated, so a test that had to pass real ones could
/// only get them by submitting and running DAGs.
fn retain_history<T>(live: Vec<T>, mut terminal: Vec<(T, i64, u64)>, cap: usize) -> Vec<T> {
    // Newest first, so truncating to the cap keeps the most recent.
    terminal.sort_by(|a, b| b.1.cmp(&a.1).then(b.2.cmp(&a.2)));
    terminal.truncate(cap);
    let mut kept = live;
    kept.extend(terminal.into_iter().map(|(handle, _, _)| handle));
    kept
}

/// Truncate a node error to [`MAX_ERROR_LEN`] on a char boundary, appending `…`.
fn truncate_error(e: &str) -> String {
    if e.len() <= MAX_ERROR_LEN {
        return e.to_owned();
    }
    let cut = (0..=MAX_ERROR_LEN)
        .rev()
        .find(|i| e.is_char_boundary(*i))
        .unwrap_or(0);
    let mut msg = e[..cut].to_owned();
    msg.push('…');
    msg
}