"""Deterministic duty assignment — the mesh's leaderless brain.

Every node runs this same pure function over the same gossiped inputs (the
live node set + the LWW placement overrides) and lands on the same answer, so
the mesh needs no election and has no split-brain window: when a node dies or
runs out of tokens, every survivor recomputes and the duty has *already*
moved. Determinism comes from total ordering — every ranking ends in the node
id tie-break.

Eligibility for a duty:
- the node has the duty enabled (per-node toggle), and
- if the placement is token-aware, the node is not out of tokens
  (``tokens == "out"``); low-token nodes stay eligible but rank behind
  same-strategy peers with full tokens.

Strategies (ranking among eligible nodes):
- ``weakest-first``    highest tier number first (tier 1 = strongest machine)
- ``strongest-first``  lowest tier number first
- ``local-first``      the given local node first, the rest weakest-first
- ``surplus-first``    most spare quota first, where "spare" is *relative*: each
  node's budget left divided by the clock left until that budget resets (see
  ``protocol.NodeInfo.surplus``). Ties fall back to weakest-first. This is the
  default, and the load balancer — dispatchers rank targets by it (see
  ``config.dispatch_strategy``) and it also drives displayed duty ownership.

  Ranking on the raw remaining budget instead would be wrong in both directions:
  it would hoard an account whose quota resets in hours (leaving budget to expire
  unspent) while hammering one whose larger balance still has to stretch across
  a week. Only the ratio tells the two apart.
"""

from __future__ import annotations

from dataclasses import dataclass

from . import config, protocol
from .config import Placement, PlacementOverrides
from .protocol import NodeInfo

_TOKEN_RANK = {"ok": 0, "low": 1, "out": 2}


@dataclass(frozen=True)
class DutyAssignment:
    duty: str
    assigned: tuple[str, ...]  # node ids, in rank order
    # Unmet platform requirements: [(platform, missing_count)].
    shortfall: tuple[tuple[str, int], ...] = ()

    @property
    def satisfied(self) -> bool:
        return not self.shortfall

    def to_dict(self) -> dict:
        return {
            "duty": self.duty,
            "assigned": list(self.assigned),
            "shortfall": [{"platform": p, "missing": m} for p, m in self.shortfall],
        }


def _eligible(nodes: list[NodeInfo], duty_id: str, placement: Placement) -> list[NodeInfo]:
    out = []
    for n in nodes:
        if not n.duty_enabled(duty_id):
            continue
        if placement.token_aware and n.tokens == "out":
            continue
        out.append(n)
    return out


def _ranked(nodes: list[NodeInfo], strategy: str, local_id: str) -> list[NodeInfo]:
    def key(n: NodeInfo):
        tok = _TOKEN_RANK.get(n.tokens, 1)
        if strategy == "strongest-first":
            return (tok, n.tier, n.id)
        if strategy == "local-first":
            return (tok, n.id != local_id, -n.tier, n.id)
        if strategy == "surplus-first":
            # Most spare quota first, measured as pace — budget left over clock
            # left until reset — so the comparison is relative, not absolute. A
            # node sitting on a big balance that resets tonight outranks one whose
            # larger balance still has to cover a week. Surplus leads; token rank
            # then tier break ties, which also lands neutral-surplus nodes on the
            # weakest-first ordering. Bucketed so drift can't reshuffle peers that
            # are, for routing purposes, equally flush.
            return (-protocol.surplus_bucket(n.surplus()), tok, -n.tier, n.id)
        # weakest-first (and any unknown strategy from a newer peer)
        return (tok, -n.tier, n.id)

    return sorted(nodes, key=key)


def assign_duty(
    duty_id: str,
    nodes: list[NodeInfo],
    overrides: PlacementOverrides | None = None,
    local_id: str = "",
) -> DutyAssignment:
    """Assign one duty over the given live nodes.

    With a platform ``spread`` (e.g. the bundle E2E's one-linux-plus-one-macos)
    each requirement is filled from that platform's ranked candidates; a node
    fills at most one slot. Without a spread, the single best-ranked node owns
    the duty. Requirements that can't be met are reported as ``shortfall`` —
    the duty still gets whatever coverage exists.
    """
    placement = config.placement_for(duty_id, overrides)
    pool = _ranked(_eligible(nodes, duty_id, placement), placement.strategy, local_id)

    if not placement.spread:
        return DutyAssignment(duty_id, (pool[0].id,) if pool else (),
                              () if pool else (("any", 1),))

    assigned: list[str] = []
    shortfall: list[tuple[str, int]] = []
    taken: set[str] = set()
    for platform, count in placement.spread:
        got = 0
        for n in pool:
            if got == count:
                break
            if n.platform == platform and n.id not in taken:
                taken.add(n.id)
                assigned.append(n.id)
                got += 1
        if got < count:
            shortfall.append((platform, count - got))
    return DutyAssignment(duty_id, tuple(assigned), tuple(shortfall))


def assign_all(
    nodes: list[NodeInfo],
    overrides: PlacementOverrides | None = None,
    local_id: str = "",
) -> dict[str, DutyAssignment]:
    """Every duty in the shared catalog, assigned. The topology snapshot and
    the dispatch router both come through here, so what the panel shows is by
    construction what dispatch will do."""
    return {
        duty_id: assign_duty(duty_id, nodes, overrides, local_id)
        for duty_id in config.duty_ids()
    }


def slot_candidates(
    duty_id: str,
    nodes: list[NodeInfo],
    overrides: PlacementOverrides | None = None,
    local_id: str = "",
    strategy: str | None = None,
) -> list[tuple[str, list[str]]]:
    """Per-slot failover lists for executing a dispatch.

    A spread duty runs one job per slot (the bundle E2E = a linux slot AND a
    macos slot); each slot gets its own ranked candidate list so a failed
    target falls over to the next machine *of the required platform*. The
    executor is responsible for not landing two slots on one node. A no-spread
    duty is a single ``("any", ranked)`` slot.

    ``strategy`` overrides only the *ranking* (not eligibility or spread) — the
    dispatcher passes ``config.dispatch_strategy()`` here so target selection is
    the load-balancing decision (surplus-first by default), independent of the
    duty's displayed placement strategy.
    """
    placement = config.placement_for(duty_id, overrides)
    rank_strategy = strategy or placement.strategy
    pool = _ranked(_eligible(nodes, duty_id, placement), rank_strategy, local_id)
    if not placement.spread:
        return [("any", [n.id for n in pool])]
    slots: list[tuple[str, list[str]]] = []
    for platform, count in placement.spread:
        of_platform = [n.id for n in pool if n.platform == platform]
        # One job per slot, and the executor never lands two slots on one node, so
        # more slots than nodes-of-platform can never be filled — every extra slot is
        # a guaranteed "no eligible node" failure. Bounding here keeps a hostile/typo
        # `count` (a signed placement override's spread is attacker-influenceable on an
        # open mesh — see config._parse_spread) from materializing an arbitrarily large
        # list and OOM-ing the node the moment it dispatches this duty. Keep a floor of
        # one slot so a platform with zero live nodes still surfaces a single failure
        # (mirrors assign_duty's shortfall) instead of vanishing silently.
        n_slots = min(count, max(1, len(of_platform)))
        slots.extend((platform, of_platform) for _ in range(n_slots))
    return slots
