defmodule TradeMachine.GMGradeV2.BenchmarkRotation do
  @moduledoc """
  Deterministic 240-minute rotation for a real NBA team's completed season.

  A user's scenario rotation is an explicit input — they choose the starters and
  distribute the minutes. The 30 benchmark teams have no such author, but AVC's
  Competitive Strength scores are *percentiles against those 30 teams*, so the
  benchmark rotations have to be manufactured. This module manufactures them,
  and AVC's interpretation note 3 sets the bar:

  > Every real NBA team needs a reproducible starter designation and 240-minute
  > allocation calculated under the same role and fatigue rules as a user
  > scenario.

  "Same rules as a user scenario" is taken literally here (DECISIONS D-39): the
  240-minute total, the five-starter count and the 48-minute-per-player ceiling
  are **read from `TradeMachine.GMGradeV2.RotationValidation`**, never restated
  as local constants, and every rotation this module produces is run back
  through `RotationValidation.validate/2` before it is returned. A benchmark
  rotation that a user would not be allowed to submit is not returned at all —
  it comes back as `{:error, …}`. See "Why the output is re-validated" below.

  ## The algorithm (version: `ModelVersion.versions()[:benchmarks]`)

  1. **Eligibility.** A row needs a `player_id` and positive Season Minutes.
     Every rejected row is recorded in `:excluded` with a reason — see "Nothing
     is dropped silently".
  2. **Rank** by Season Minutes descending, ties broken by `player_id`
     ascending. This single ordering is the module's canonical order and is
     reused for the pool cutoff, the starter tie-break and the rounding
     remainder, so there is one thing to reason about rather than three.
  3. **Pool.** Walk the ranking accumulating minutes-per-game until 240 is
     covered, including the player who crosses it — a game *is* 240 player
     minutes, so this selects the players a real rotation actually consumes
     (9–11 per team in 2025-26). Always at least five players, so the
     five-starter rule is satisfiable.
  4. **Allocate** each pool member minutes in proportion to their minutes per
     game, scaled so the total is exactly 240, with the 48-minute ceiling
     applied by water-filling: cap whoever exceeds it, redistribute their excess
     across the rest, repeat until nobody does.
  5. **Round** to whole minutes and correct the residue (see "Rounding").
  6. **Starters** — the top five in the pool by games started, ties broken by
     Season Minutes then `player_id`, with the rule that actually ran reported
     in `:starter_basis` (see "Which starter rule ran").

  ## Season Minutes is *not* the `total_minutes` column (DECISIONS D-38)

  The ranking quantity is `games_played × minutes_per_game` via
  `PlayerImpact.season_minutes/1` — AVC's "Season Minutes" (D-11), the same
  quantity that drives the 1,200-minute Reliability Factor and the 800-minute
  Experience Factor. `player_stats` *also* carries a literal `total_minutes`
  column, and the two disagree: `minutes_per_game` is stored rounded to one
  decimal, so the product accumulates error across a season (221 of 422 rows in
  2025-26 differ by more than a minute, worst case 4.3). Ranking by the stored
  column would be marginally more accurate in isolation and is the wrong choice
  anyway — a player must not be able to rank above a teammate here while
  ranking below him in the reliability adjustment applied to the very same
  rotation. Measured, not assumed: across the 30 teams the two keys select an
  identical pool, and differ in ordering on two teams.

  ## Rounding

  Proportional targets are rounded to the nearest whole minute (whole minutes
  because `RotationValidation` rejects fractional assignments), which leaves a
  residue of a few minutes either way. The residue is handed out ±1 at a time
  down the canonical order — highest Season Minutes first, wrapping if it runs
  past the end — skipping anyone who would breach the 48-minute ceiling or go
  negative.

  This is the task's prescription (D-40) rather than the largest-remainder
  method that would minimise deviation from exact proportionality. Priced
  before adopting: summed across the 30 real 2025-26 teams, total departure
  from the exact proportional split is 89.8 minutes here against 77.8 under
  largest-remainder, and no single player is off by more than 1.5 minutes. It
  costs little because rounding to nearest is already unbiased — only the small
  leftover is directional, not a whole minute per player as a floor-and-top-up
  scheme would be. It is worth knowing that it *is* a choice: a top-of-rotation
  tiebreak, not a neutral one.

  ## Which starter rule ran (DECISIONS D-47)

  `player_stats.games_started` is NULL for every row in every database audited,
  so the `games_started` criterion has never actually executed and "top five by
  games started" silently degrades into "top five by Season Minutes". The
  degraded rule still produces a perfectly legal 240-minute, five-starter
  rotation, which is exactly why it went unnoticed — so the rule that ran is
  reported in `:starter_basis` rather than left to be inferred, and consumers
  freeze it into the benchmark document.

  ## Nothing is dropped silently

  The phase constraint is that missing data is recorded rather than imputed. A
  team whose roster rows are half empty must not quietly produce a rotation
  built from the surviving half and look identical to a complete one, so
  `:excluded` names every row that did not become an entry and why
  (`:missing_player_id`, `:missing_minutes`, `:no_minutes`, `:outside_rotation`,
  `:not_a_map`). `:outside_rotation` is the ordinary case — a deep bench player
  who lost the cutoff — and is listed alongside the others so a caller
  reconciling roster size against rotation size gets one complete account.

  Structurally impossible values (negative games or minutes) raise, via
  `PlayerImpact.season_minutes/1`. Duplicate `player_id`s are an error rather
  than being merged or de-duplicated: two rows for one player in one season is a
  caller bug, and silently picking one of them would produce a rotation nobody
  asked for. The check runs over every row handed in, **including the ones
  eligibility would drop** — a duplicate whose second row has no minutes is
  invisible to this module's own output and still poisons a caller that keys its
  own maps off the same rows. See `check_duplicates/1`.

  ## Determinism (DECISIONS D-50)

  The same rows in any order produce an identical result, and that has to hold
  for the *whole* result, not just for `:entries`. Ineligible rows are therefore
  sorted by `{player_id, reason}` before they are returned — reading them off in
  arrival order would make `:excluded` a function of however Postgres happened
  to return the team's rows, on a value that gets frozen into a benchmark
  document. Everything else reduces over the canonically ordered pool.

  ## Why the output is re-validated

  Running the finished rotation back through `RotationValidation.validate/2`
  makes "if this returns `{:ok, _}`, the rotation is legal" true by construction
  instead of true by argument. Every invariant — 240 exactly, five starters,
  nothing over 48, nothing negative or fractional, no duplicates — is then
  enforced by the same code path that polices user submissions, so an arithmetic
  regression in the allocator surfaces as a loud `{:error, …}` rather than as a
  benchmark that is quietly 239 minutes long and skews every percentile computed
  against it.

  ## Purity

  No `Repo` access and no ambient state, matching the rest of `GMGradeV2`. The
  caller loads one team's rows for one season — `NBA.Stats.list_player_stats_for_season_full/1`
  grouped by `:team_id` is the intended source — and owns the season filter;
  this module never checks which season it was handed.
  """

  alias TradeMachine.GMGradeV2.ModelVersion
  alias TradeMachine.GMGradeV2.PlayerImpact
  alias TradeMachine.GMGradeV2.RotationValidation

  @typedoc "A rotation slot, in the shape `RotationValidation` normalizes to."
  @type entry :: %{player_id: binary(), minutes: integer(), starter: boolean()}

  @typedoc "A row that did not become an entry, and why."
  @type exclusion :: %{player_id: binary() | nil, reason: atom()}

  @typedoc """
  Which rule actually designated the starting five, and over how much of the
  pool. `:basis` is `:games_started` when at least one candidate carried a
  measured value, and `:season_minutes_fallback` when none did — the latter is
  the current state of every real team, since the column is unpopulated (D-47).
  """
  @type starter_basis :: %{
          basis: :games_started | :season_minutes_fallback,
          covered: non_neg_integer(),
          pool_size: non_neg_integer()
        }

  @typedoc "A built benchmark rotation plus the diagnostics behind it."
  @type t :: %{
          entries: [entry()],
          excluded: [exclusion()],
          pool_size: non_neg_integer(),
          scale: float(),
          starter_basis: starter_basis(),
          version: String.t()
        }

  @type error :: %{code: atom(), message: String.t()}

  @doc """
  The benchmark algorithm version, frozen onto every benchmark this rotation
  feeds. Bump `:benchmarks` in `ModelVersion.versions/0` when the rules above
  change, or old benchmarks stop being reproducible.
  """
  @spec version() :: String.t()
  def version, do: Map.fetch!(ModelVersion.versions(), :benchmarks)

  @doc """
  Builds one team's benchmark rotation from its `player_stats` rows for a single
  season.

  Accepts `%TradeMachine.NBA.PlayerStats{}` structs or plain maps carrying the
  same atom keys (`:player_id`, `:games_played`, `:games_started`,
  `:minutes_per_game`) — the projection from
  `NBA.Stats.list_player_stats_for_season_full/1` works as-is.

  Returns `{:ok, t()}` with entries in canonical order (most Season Minutes
  first), or `{:error, t:error/0}` when no legal rotation exists for the rows
  given.
  """
  @spec build(Enumerable.t()) :: {:ok, t()} | {:error, error()}
  def build(rows) do
    rows = Enum.to_list(rows)
    {candidates, ineligible} = classify(rows)
    ranked = Enum.sort_by(candidates, &rank_key/1)

    with :ok <- check_duplicates(rows),
         {:ok, pool} <- select_pool(ranked, ineligible),
         {:ok, targets} <- allocate(pool) do
      finish(pool, targets, ineligible ++ outside_rotation(ranked, pool))
    end
  end

  # ── Eligibility ─────────────────────────────────────────────────────

  defp classify(rows) do
    {candidates, ineligible} =
      Enum.reduce(rows, {[], []}, fn row, {ok, bad} ->
        case candidate(row) do
          {:ok, candidate} -> {[candidate | ok], bad}
          {:excluded, exclusion} -> {ok, [exclusion | bad]}
        end
      end)

    # Candidates are about to be sorted into canonical order; the exclusions are
    # not, so they are sorted here instead of being handed back in whatever order
    # the caller's query produced (D-50). `nil` player ids sort first — Erlang's
    # term order puts atoms below binaries — which is stable, not meaningful.
    {Enum.reverse(candidates), Enum.sort_by(ineligible, &{&1.player_id, &1.reason})}
  end

  defp candidate(row) when is_map(row) do
    player_id = Map.get(row, :player_id)
    # Raises on negative games or minutes, and returns nil — never 0 — when
    # either input is missing (D-31).
    minutes = PlayerImpact.season_minutes(row)

    cond do
      is_nil(player_id) -> {:excluded, exclusion(nil, :missing_player_id)}
      is_nil(minutes) -> {:excluded, exclusion(player_id, :missing_minutes)}
      minutes <= 0 -> {:excluded, exclusion(player_id, :no_minutes)}
      true -> {:ok, build_candidate(row, player_id, minutes)}
    end
  end

  defp candidate(_row), do: {:excluded, exclusion(nil, :not_a_map)}

  defp build_candidate(row, player_id, minutes) do
    %{
      player_id: player_id,
      # Positive Season Minutes guarantees positive minutes per game, so the
      # allocator never divides by a zero weight.
      minutes_per_game: Map.fetch!(row, :minutes_per_game) / 1,
      season_minutes: minutes / 1,
      games_started: Map.get(row, :games_started)
    }
  end

  defp exclusion(player_id, reason), do: %{player_id: player_id, reason: reason}

  # Over the *raw* rows, not over the eligible candidates. A second row for a
  # player that happens to be ineligible — no minutes, say — never reaches
  # `classify/1`'s candidate list, so checking the candidates would wave the pair
  # through: the rotation builds cleanly from the eligible row and looks
  # entirely normal. It is `BenchmarkBuilder` that then pays, because it keys its
  # own per-player maps off the same raw rows (`Map.new(rows, …)`) and Postgres
  # guarantees no row order — so which of the two rows supplies the player's
  # frozen impact becomes a coin flip, and a benchmark that must be identical
  # under any row order (D-50) is not. Measured on a five-player fixture: the
  # team's `star_power` came out −1.5 or 4.825 depending purely on which
  # duplicate arrived first.
  defp check_duplicates(rows) do
    duplicates =
      rows
      |> Enum.filter(&is_map/1)
      |> Enum.map(&Map.get(&1, :player_id))
      |> Enum.reject(&is_nil/1)
      |> Enum.frequencies()
      |> Enum.filter(fn {_player_id, count} -> count > 1 end)
      |> Enum.map(fn {player_id, _count} -> player_id end)
      |> Enum.sort()

    if duplicates == [] do
      :ok
    else
      {:error,
       error(
         :duplicate_player,
         "Each player may appear once per season; got repeated rows for: " <>
           Enum.join(duplicates, ", ")
       )}
    end
  end

  # ── Ranking and pool selection ──────────────────────────────────────

  # The canonical order: most Season Minutes first, `player_id` breaking ties so
  # two players with identical minutes never swap between runs.
  defp rank_key(candidate), do: {-candidate.season_minutes, candidate.player_id}

  defp select_pool(ranked, ineligible) do
    starters = RotationValidation.starter_count()

    pool =
      case take_until_covered(ranked) do
        covered when length(covered) < starters -> Enum.take(ranked, starters)
        covered -> covered
      end

    cond do
      pool == [] ->
        {:error,
         error(
           :no_eligible_players,
           "No player has a positive minutes total this season (#{summarize(ineligible)})."
         )}

      length(pool) < starters ->
        {:error,
         error(
           :insufficient_players,
           "A rotation needs at least #{starters} players with recorded minutes; got " <>
             "#{length(pool)} (#{summarize(ineligible)})."
         )}

      true ->
        {:ok, pool}
    end
  end

  # A build that fails has no `:excluded` list to return, so the reason rows
  # were dropped goes into the message instead — otherwise "no eligible
  # players" gives an operator nothing to act on.
  defp summarize([]), do: "no rows excluded"

  defp summarize(ineligible) do
    counts =
      ineligible
      |> Enum.frequencies_by(& &1.reason)
      |> Enum.sort()
      |> Enum.map_join(", ", fn {reason, count} -> "#{reason}: #{count}" end)

    "excluded — " <> counts
  end

  # Take players down the ranking until their minutes per game cover a full
  # game, including whoever crosses the line.
  defp take_until_covered(ranked) do
    budget = RotationValidation.total_minutes()

    {taken, _covered} =
      Enum.reduce_while(ranked, {[], 0.0}, fn candidate, {taken, covered} ->
        if covered >= budget do
          {:halt, {taken, covered}}
        else
          {:cont, {[candidate | taken], covered + candidate.minutes_per_game}}
        end
      end)

    Enum.reverse(taken)
  end

  defp outside_rotation(ranked, pool) do
    selected = MapSet.new(pool, & &1.player_id)

    for candidate <- ranked, not MapSet.member?(selected, candidate.player_id) do
      exclusion(candidate.player_id, :outside_rotation)
    end
  end

  # ── Minute allocation ───────────────────────────────────────────────

  defp allocate(pool) do
    cap = RotationValidation.max_minutes()
    budget = RotationValidation.total_minutes()

    # Stated rather than assumed: five starters at 48 happens to be exactly 240
    # today, so the starter-count floor already guarantees this. Retuning either
    # constant would break that coincidence silently.
    if length(pool) * cap < budget do
      {:error,
       error(
         :rotation_infeasible,
         "#{length(pool)} players capped at #{cap} minutes cannot cover #{budget}."
       )}
    else
      {:ok, water_fill(pool, budget, cap, %{})}
    end
  end

  # Proportional shares under a ceiling: scale the uncapped players to fill the
  # remaining budget, and if that pushes anyone past the cap, pin them there and
  # re-scale whoever is left. Terminates because each pass either finishes or
  # removes at least one player from the uncapped set, and `uncapped` can never
  # empty while the feasibility check above holds (`remaining <= n × cap` is
  # preserved by each pass, and the whole set cannot exceed the cap at once).
  defp water_fill([], _remaining, _cap, allocated), do: allocated

  defp water_fill(uncapped, remaining, cap, allocated) do
    weight = Enum.reduce(uncapped, 0.0, &(&1.minutes_per_game + &2))
    scale = remaining / weight

    case Enum.split_with(uncapped, &(&1.minutes_per_game * scale > cap)) do
      {[], _} ->
        Enum.reduce(uncapped, allocated, fn candidate, acc ->
          Map.put(acc, candidate.player_id, candidate.minutes_per_game * scale)
        end)

      {over, rest} ->
        allocated = Enum.reduce(over, allocated, &Map.put(&2, &1.player_id, cap / 1))
        water_fill(rest, remaining - length(over) * cap, cap, allocated)
    end
  end

  # ── Rounding to whole minutes ───────────────────────────────────────

  defp round_to_budget(pool, targets) do
    cap = RotationValidation.max_minutes()

    minutes =
      Map.new(pool, fn candidate ->
        rounded = targets |> Map.fetch!(candidate.player_id) |> round() |> min(cap) |> max(0)
        {candidate.player_id, rounded}
      end)

    # Summed over `pool`, not over the map: these are integers so the total is
    # order-independent either way, but reducing over a map's iteration order is
    # the habit D-50 exists to break.
    assigned = Enum.reduce(pool, 0, &(Map.fetch!(minutes, &1.player_id) + &2))
    distribute(minutes, pool, RotationValidation.total_minutes() - assigned, cap)
  end

  # Hand the residue out ±1 at a time down the canonical order, wrapping if it
  # runs past the end, skipping anyone at a boundary. `pool` is already in
  # canonical order, so "highest-minute players first" is just its own order.
  defp distribute(minutes, _pool, 0, _cap), do: minutes

  defp distribute(minutes, pool, residue, cap) do
    step = if residue > 0, do: 1, else: -1
    eligible = Enum.filter(pool, &adjustable?(Map.fetch!(minutes, &1.player_id), step, cap))

    case Enum.take(eligible, abs(residue)) do
      # Unreachable while the rotation is feasible. Returning rather than
      # looping leaves the shortfall for `RotationValidation` to reject, which
      # is the point of re-validating: a bug here cannot escape as a silently
      # short rotation.
      [] ->
        minutes

      adjusted ->
        minutes =
          Enum.reduce(adjusted, minutes, fn candidate, acc ->
            Map.update!(acc, candidate.player_id, &(&1 + step))
          end)

        distribute(minutes, pool, residue - step * length(adjusted), cap)
    end
  end

  # D-40 skips "anyone at the 48-minute ceiling **or at zero**", and that reads on
  # the handout, not on the direction. A player whose proportional target rounded
  # to zero did not make the rotation on the merits; handing him the residue's
  # spare minute promotes a non-participant, and because D-46 populates Star Power
  # from entries with `minutes > 0` it also silently enlarges that population in a
  # frozen benchmark. The `-1` clause is unchanged: it already refuses to go
  # negative, which is the same boundary read from the other side.
  defp adjustable?(minutes, 1, cap), do: minutes > 0 and minutes < cap
  defp adjustable?(minutes, -1, _cap), do: minutes > 0

  # ── Starters ────────────────────────────────────────────────────────

  defp designate_starters(pool) do
    pool
    |> Enum.sort_by(&starter_key/1)
    |> Enum.take(RotationValidation.starter_count())
    |> MapSet.new(& &1.player_id)
  end

  # A missing `games_started` is unknown, not zero, so it sorts *below* a
  # measured 0 rather than tying with it (D-40). The leading flag carries that
  # instead of a sentinel number, so nothing arithmetic depends on the encoding.
  defp starter_key(%{games_started: started} = candidate) when is_number(started) do
    {0, -started, -candidate.season_minutes, candidate.player_id}
  end

  defp starter_key(candidate) do
    {1, 0, -candidate.season_minutes, candidate.player_id}
  end

  # ── Assembly ────────────────────────────────────────────────────────

  defp finish(pool, targets, excluded) do
    minutes = round_to_budget(pool, targets)
    starters = designate_starters(pool)

    entries =
      Enum.map(pool, fn candidate ->
        %{
          player_id: candidate.player_id,
          minutes: Map.fetch!(minutes, candidate.player_id),
          starter: MapSet.member?(starters, candidate.player_id)
        }
      end)

    case RotationValidation.validate(entries, Enum.map(pool, & &1.player_id)) do
      {:ok, validated} ->
        {:ok,
         %{
           entries: validated,
           excluded: excluded,
           pool_size: length(pool),
           scale: scale(pool),
           starter_basis: starter_basis(pool),
           version: version()
         }}

      {:error, errors} ->
        {:error,
         error(
           :invalid_rotation,
           "Built a rotation the scenario validator rejects: " <>
             Enum.map_join(errors, "; ", & &1.message)
         )}
    end
  end

  # Which rule actually decided the starting five, as opposed to which rule is
  # written above. `starter_key/1`'s nil clause is a deliberate ordering choice
  # (D-40: unknown sorts below a measured 0), but it is silent — and when *no*
  # candidate carries a `games_started` value, every key collapses to the
  # `{1, 0, -season_minutes, player_id}` form and "top five by games started"
  # degrades into "top five by Season Minutes" with nothing to show for it.
  #
  # That is not hypothetical (D-47): `games_started` is NULL for 100% of
  # `player_stats` rows in dev, in the worktree DBs, and in the prod snapshot,
  # so the criterion has never once executed. Reporting the basis is this
  # module's application of the phase-wide rule that missing data must never
  # silently become a number — here the absence degrades an *ordering* rather
  # than a value, which is exactly why it went unnoticed. Consumers freeze this
  # into the benchmark document so a reader years from now can tell which rule
  # produced the five.
  defp starter_basis(pool) do
    covered = Enum.count(pool, &is_number(&1.games_started))

    %{
      basis: if(covered == 0, do: :season_minutes_fallback, else: :games_started),
      covered: covered,
      pool_size: length(pool)
    }
  end

  # Diagnostics only: the *nominal* proportional scale, before the 48-minute
  # ceiling is applied — when the ceiling binds, no player is actually scaled by
  # this. Above 1.0 means the pool played less than a full game's worth of
  # minutes and was stretched to fill 240; real teams sit at 0.89–1.00.
  defp scale(pool) do
    RotationValidation.total_minutes() / Enum.reduce(pool, 0.0, &(&1.minutes_per_game + &2))
  end

  defp error(code, message), do: %{code: code, message: message}
end