//! The settle loop — drives a [`Graph`] to completion over the resource pool.
//!
//! [`Scheduler::settle`] claims every currently-runnable pending node (its
//! [`Dep::Node`] edges satisfied *and* all its [`Dep::Resource`] units acquired
//! atomically), marks it `Running`, holds its resource guards, and returns the
//! newly-started ids for the caller's runner to execute. The runner reports each
//! node's result back with [`Scheduler::complete`]; a running node may grow its
//! own sub-group first via [`Scheduler::append`]. Concurrency is emergent from
//! resource capacity — there is no separate active-node cap.
//!
//! A resource is held for the acquiring node's *entire subtree* lifetime: the
//! owned guard is released only when that node and every descendant is terminal
//! (`group_terminal`), not when the node's own work finishes. Single-owner and
//! synchronous — the caller drives `settle` / `complete`; no async or locking
//! lives here (that's the runner's job, one layer up).
//!
//! Recursive-lock re-entrancy (a sub-node reusing an ancestor group's lock) and
//! the eager `AfterOk` failure cascade are layered on top of this owned core.

use std::collections::HashMap;
use std::hash::Hash;

use crate::guard::{ResourceGuard, SharedResources};
use crate::resources::ResourceTable;
use crate::{Dep, DepWhen, Graph, GraphError, NodeId, State};

/// The result of a node's own execution, reported to [`Scheduler::complete`].
///
/// `Cancelled` is not an outcome a runner reports — it is scheduler-driven (an
/// `AfterOk` dependency failed), so a runner only ever says `Done` or `Failed`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Outcome {
    /// The node's work succeeded.
    Done,
    /// The node's work failed.
    Failed,
}

/// Drives a [`Graph`] over a shared resource pool: claim runnable nodes, hold
/// their resources for the subtree's lifetime, release on subtree-terminal.
pub struct Scheduler<N, R: Clone + Eq + Hash> {
    graph: Graph<N, R>,
    resources: SharedResources<R>,
    /// Owned resource guards, keyed by the node that acquired them. Dropped
    /// (releasing the units) when that node's whole subtree is terminal.
    owned: HashMap<NodeId, Vec<ResourceGuard<R>>>,
    /// The single re-entrancy slot per `(ancestor-holder, resource)`: the id of
    /// the descendant currently *borrowing* that ancestor's lock. Present ⇒ the
    /// slot is taken, so no other descendant may re-enter the same lock until
    /// the borrower's subtree is terminal — "only one node at a time within".
    borrow_slots: HashMap<(NodeId, R), NodeId>,
}

impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
    /// A scheduler over `graph` with `resources` as the capacity pool.
    #[must_use]
    pub fn new(graph: Graph<N, R>, resources: ResourceTable<R>) -> Self {
        Self {
            graph,
            resources: SharedResources::new(resources),
            owned: HashMap::new(),
            borrow_slots: HashMap::new(),
        }
    }

    /// The graph, for inspection (state, hierarchy, UI rendering).
    #[must_use]
    pub fn graph(&self) -> &Graph<N, R> {
        &self.graph
    }

    /// Append a node — e.g. a running node growing its own sub-group. Delegates
    /// to [`Graph::insert`]; call [`Scheduler::settle`] afterwards to start it
    /// once it is runnable.
    ///
    /// # Errors
    /// Propagates [`GraphError`] for a dangling parent or dependency id.
    pub fn append(
        &mut self,
        payload: N,
        deps: Vec<Dep<R>>,
        parent: Option<NodeId>,
    ) -> Result<NodeId, GraphError> {
        self.graph.insert(payload, deps, parent)
    }

    /// Claim every currently-runnable pending node and start it: node-deps
    /// satisfied and all resource-deps acquired atomically (all-or-nothing).
    /// Each claimed node is marked `Running`, its owned guards held, and its id
    /// returned for the runner to execute. A single pass suffices — a node
    /// started here is `Running`, not terminal, so it cannot satisfy another
    /// node's dependency in the same pass; it only consumes resources.
    pub fn settle(&mut self) -> Vec<NodeId> {
        let pending: Vec<NodeId> = self
            .graph
            .nodes()
            .filter(|n| n.state == State::Pending)
            .map(|n| n.id)
            .collect();
        let mut started = Vec::new();
        for id in pending {
            if self.node_deps_satisfied(id) && self.try_start(id) {
                started.push(id);
            }
        }
        started
    }

    /// Try to start node `id`: classify each resource dep as *owned* (no
    /// ancestor holds it → acquire real units) or *borrowed* (an ancestor group
    /// already holds it → re-enter, gated by the one re-entrancy slot), then
    /// take everything atomically or nothing. Returns whether it started.
    fn try_start(&mut self, id: NodeId) -> bool {
        let mut owned_reqs = Vec::new();
        let mut borrows = Vec::new();
        for (name, count) in self.resource_reqs(id) {
            if let Some(ancestor) = self.ancestor_owning(id, &name) {
                // Re-entrant reuse: allowed only if the slot is free.
                if self.borrow_slots.contains_key(&(ancestor, name.clone())) {
                    return false;
                }
                borrows.push((ancestor, name));
            } else {
                owned_reqs.push((name, count));
            }
        }
        // Owned units are all-or-nothing; borrow slots were all confirmed free
        // above, so this is the only fallible step. Nothing mutated until here.
        let Some(guard) = self.resources.acquire(owned_reqs) else {
            return false;
        };
        self.owned.entry(id).or_default().push(guard);
        for slot in borrows {
            self.borrow_slots.insert(slot, id);
        }
        self.graph.set_state(id, State::Running);
        true
    }

    /// The nearest ancestor of `id` that *owns* (holds real units of) `name`,
    /// or `None` if no ancestor holds it (⇒ `id` must own-acquire it itself).
    fn ancestor_owning(&self, id: NodeId, name: &R) -> Option<NodeId> {
        let mut cursor = self.graph.node(id)?.parent;
        while let Some(ancestor) = cursor {
            if self.node_owns(ancestor, name) {
                return Some(ancestor);
            }
            cursor = self.graph.node(ancestor)?.parent;
        }
        None
    }

    /// Whether node `holder` holds an owned guard covering resource `name`.
    fn node_owns(&self, holder: NodeId, name: &R) -> bool {
        self.owned.get(&holder).is_some_and(|guards| {
            guards
                .iter()
                .any(|g| g.held().iter().any(|(n, _)| n == name))
        })
    }

    /// Report a running node's own execution result. Sets its state, then
    /// releases the owned guards of every node whose whole subtree has become
    /// terminal — a parent keeps its lock until its last descendant finishes.
    /// Call [`Scheduler::settle`] again afterwards to start newly-unblocked work.
    pub fn complete(&mut self, id: NodeId, outcome: Outcome) {
        let state = match outcome {
            Outcome::Done => State::Done,
            Outcome::Failed => State::Failed,
        };
        self.graph.set_state(id, state);
        if outcome == Outcome::Failed {
            self.cascade_cancel(id);
        }
        self.release_settled_subtrees();
    }

    /// Eagerly cancel the transitive `AfterOk` dependents of a just-failed node:
    /// they can never run (a strong dependency failed), so mark them `Cancelled`
    /// now — before they could claim resources. A dependent is always still
    /// `Pending` here (a `Running` node's `AfterOk` deps were `Done` when it
    /// started, and `Done` is terminal), so no resources need releasing.
    fn cascade_cancel(&mut self, failed: NodeId) {
        let mut stack = vec![failed];
        while let Some(dep) = stack.pop() {
            let dependents: Vec<NodeId> = self
                .graph
                .nodes()
                .filter(|n| {
                    n.state == State::Pending
                        && n.deps.iter().any(
                            |d| matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == dep),
                        )
                })
                .map(|n| n.id)
                .collect();
            for d in dependents {
                self.graph.set_state(d, State::Cancelled);
                stack.push(d);
            }
        }
    }

    /// Release everything whose subtree has become terminal: drop the owned
    /// guards of any holder (→ frees its units) and free any re-entrancy slot
    /// held by a borrower — both are held for the whole subtree lifetime.
    fn release_settled_subtrees(&mut self) {
        let holders: Vec<NodeId> = self.owned.keys().copied().collect();
        for holder in holders {
            if self.graph.group_terminal(holder) {
                self.owned.remove(&holder); // drops guards → releases the units
            }
        }
        self.borrow_slots
            .retain(|_, borrower| !self.graph.group_terminal(*borrower));
    }

    /// Whether every [`Dep::Node`] edge of `id` is satisfied. `Dep::Resource`
    /// edges are handled by the atomic acquire in [`Scheduler::settle`], not here.
    fn node_deps_satisfied(&self, id: NodeId) -> bool {
        let Some(node) = self.graph.node(id) else {
            return false;
        };
        node.deps.iter().all(|dep| match dep {
            Dep::Resource { .. } => true,
            Dep::Node { id, when } => self.dep_node_satisfied(*id, *when),
        })
    }

    /// Whether a node/group dependency `id` satisfies edge kind `when`. A group
    /// is depended on as a whole: `AfterAny` needs its subtree terminal (any
    /// outcome), `AfterOk` needs its whole subtree to have succeeded.
    fn dep_node_satisfied(&self, id: NodeId, when: DepWhen) -> bool {
        match when {
            DepWhen::AfterAny => self.graph.group_terminal(id),
            DepWhen::AfterOk => self.subtree_all_done(id),
        }
    }

    /// Whether `id` and every descendant reached [`State::Done`] — the success
    /// condition for an `AfterOk` edge onto a (possibly group) node.
    fn subtree_all_done(&self, id: NodeId) -> bool {
        let Some(node) = self.graph.node(id) else {
            return false;
        };
        node.state == State::Done && self.graph.children(id).all(|c| self.subtree_all_done(c.id))
    }

    /// The `(name, count)` resource units `id` must hold to run.
    fn resource_reqs(&self, id: NodeId) -> Vec<(R, u32)> {
        let Some(node) = self.graph.node(id) else {
            return Vec::new();
        };
        node.deps
            .iter()
            .filter_map(|dep| match dep {
                Dep::Resource { name, count } => Some((name.clone(), *count)),
                Dep::Node { .. } => None,
            })
            .collect()
    }
}

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

    fn res(name: &str) -> String {
        name.to_owned()
    }

    /// A graph + a resource table with `build-slot` set to `slots`.
    fn scheduler_with_slots(slots: u32) -> Scheduler<&'static str, String> {
        let mut table = ResourceTable::new();
        table.set_capacity(res("build-slot"), slots);
        Scheduler::new(Graph::new(), table)
    }

    fn slot_dep() -> Vec<Dep<String>> {
        vec![Dep::Resource {
            name: res("build-slot"),
            count: 1,
        }]
    }

    fn resource_dep(name: &str) -> Vec<Dep<String>> {
        vec![Dep::Resource {
            name: res(name),
            count: 1,
        }]
    }

    #[test]
    fn resource_node_starts_then_releases_on_complete() {
        let mut s = scheduler_with_slots(1);
        let n = s.append("build", slot_dep(), None).expect("insert");
        // settle claims it (a slot is free) and marks it Running.
        assert_eq!(s.settle(), vec![n]);
        assert_eq!(s.graph().node(n).unwrap().state, State::Running);
        // Slot is held.
        assert!(s.resources.with(|t| t.available(&res("build-slot")) == 0));
        // Completing it releases the slot (subtree is just this node).
        s.complete(n, Outcome::Done);
        assert_eq!(s.graph().node(n).unwrap().state, State::Done);
        assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1));
    }

    #[test]
    fn build_slot_cap_limits_concurrency_and_release_unblocks() {
        let mut s = scheduler_with_slots(2);
        let a = s.append("a", slot_dep(), None).expect("a");
        let b = s.append("b", slot_dep(), None).expect("b");
        let c = s.append("c", slot_dep(), None).expect("c");
        // cap 2 → a + b start, c blocks on the exhausted slot.
        let started = s.settle();
        assert_eq!(started, vec![a, b]);
        assert_eq!(s.graph().node(c).unwrap().state, State::Pending);
        // a finishes → its slot frees → c can now start.
        s.complete(a, Outcome::Done);
        assert_eq!(s.settle(), vec![c]);
        assert_eq!(s.graph().node(c).unwrap().state, State::Running);
    }

    #[test]
    fn parent_holds_resource_until_child_subtree_done() {
        let mut s = scheduler_with_slots(1);
        // Parent grabs the single build-slot and runs.
        let parent = s.append("parent", slot_dep(), None).expect("parent");
        assert_eq!(s.settle(), vec![parent]);
        // Parent grows a child (no resource dep of its own) and finishes its
        // OWN work — but its subtree is not terminal, so it keeps the slot.
        let child = s.append("child", vec![], Some(parent)).expect("child");
        s.complete(parent, Outcome::Done);
        assert!(
            s.resources.with(|t| t.available(&res("build-slot")) == 0),
            "parent must keep its lock while a child is still pending/running"
        );
        // The child starts and completes → now the whole subtree is terminal →
        // the parent's slot is released exactly once.
        assert_eq!(s.settle(), vec![child]);
        s.complete(child, Outcome::Done);
        assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1));
    }

    #[test]
    fn recursive_lock_serializes_re_entrant_descendants() {
        // `agent/foo` is unconfigured → default capacity 1.
        let mut s = Scheduler::new(Graph::new(), ResourceTable::new());
        let agent = res("agent/foo");
        // A group node owns agent/foo and runs.
        let group = s
            .append("group", resource_dep("agent/foo"), None)
            .expect("group");
        assert_eq!(s.settle(), vec![group]);
        assert!(s.resources.with(|t| t.available(&agent) == 0));
        // Two sub-nodes each need agent/foo → they re-enter the group's lock,
        // but only ONE at a time (the single re-entrancy slot).
        let c1 = s
            .append("c1", resource_dep("agent/foo"), Some(group))
            .expect("c1");
        let c2 = s
            .append("c2", resource_dep("agent/foo"), Some(group))
            .expect("c2");
        let started = s.settle();
        assert_eq!(
            started,
            vec![c1],
            "only one descendant may borrow at a time"
        );
        assert_eq!(s.graph().node(c2).unwrap().state, State::Pending);
        // The lock was NOT re-acquired — still just the group's one unit held.
        assert!(s.resources.with(|t| t.available(&agent) == 0));
        // c1 finishes → its borrow slot frees → c2 can now re-enter.
        s.complete(c1, Outcome::Done);
        assert_eq!(s.settle(), vec![c2]);
        assert_eq!(s.graph().node(c2).unwrap().state, State::Running);
        // Still no double-acquire; the group's single unit is the only hold.
        assert!(s.resources.with(|t| t.available(&agent) == 0));
    }

    #[test]
    fn failed_after_ok_dep_cancels_dependents_but_after_any_still_runs() {
        let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
        let root = s.append("root", vec![], None).expect("root");
        let strong1 = s
            .append(
                "strong1",
                vec![Dep::Node {
                    id: root,
                    when: DepWhen::AfterOk,
                }],
                None,
            )
            .expect("strong1");
        let strong2 = s
            .append(
                "strong2",
                vec![Dep::Node {
                    id: strong1,
                    when: DepWhen::AfterOk,
                }],
                None,
            )
            .expect("strong2");
        let weak = s
            .append(
                "weak",
                vec![Dep::Node {
                    id: root,
                    when: DepWhen::AfterAny,
                }],
                None,
            )
            .expect("weak");
        assert_eq!(s.settle(), vec![root]);
        s.complete(root, Outcome::Failed);
        // The AfterOk chain strong1→strong2 is eagerly cancelled (a strong dep
        // failed)…
        assert_eq!(s.graph().node(strong1).unwrap().state, State::Cancelled);
        assert_eq!(s.graph().node(strong2).unwrap().state, State::Cancelled);
        // …but the AfterAny dependent still runs — it converges regardless.
        assert_eq!(s.settle(), vec![weak]);
    }
}