defmodule TradeMachine.GMGradeV2.FrozenDocument do
  @moduledoc """
  Canonicalization and schema validation for the two write-once `data` documents
  — `TradeMachine.GMGradeV2.Benchmark` and
  `TradeMachine.GMGradeV2.ChallengeSnapshot`.

  Both columns are jsonb, both rows are frozen at insert and never updated, and
  neither can be corrected afterwards — only superseded. That makes the *default*
  answer for an unrecognised document the whole design question, and D-54 answers
  it: **canonicalize first, then validate a required schema, fail-closed.**

  ## Why this is one walker and not a pile of guards

  The previous implementation grew a new guard per review round — the 30-team
  invariant (rounds 1–3), then `dav`/`meta` type-checking (round 4), then
  `skills.player_percentiles` and the dual `:skills`/`"skills"` key spellings
  (round 5, which *defeated* the round-4 remediation). Five rounds, the same
  finding class at a new line each time, because enumerating bad shapes is
  strictly weaker than declaring the good one. Nothing here validates a shape it
  was told about; it validates against a declared schema and treats everything
  else as invalid.

  Two mechanisms, in order:

    1. `canonicalize/1` normalizes the document to string keys and **refuses**
       what it cannot normalize cleanly. A map may legally hold `:skills` and
       `"skills"` at once — Elixir treats them as two keys, **jsonb does not**:
       encoding emits the name twice and Postgres keeps the last. So a document
       whose atom-keyed section is well-formed can validate, get written, and
       read back as whatever the string-keyed duplicate held; the more carefully
       a validator checks the spelling it resolved, the more confidently it
       approves a row that will not contain it. There is no reading of "both
       spellings present" more likely to be intended than to be a builder
       assembling a document two different ways, so it is refused rather than
       resolved by a precedence rule. Callers store the canonical form, which is
       what makes the in-memory document and the read-back document the same
       object (D-42).

    2. `validate/2` walks a declared `t:spec/0` whose final clause is an
       **error**, never `[]`. A value that matches no clause is invalid by
       default. That single property is what stops the next unanticipated shape
       from being the sixth review round.

  The grammar is deliberately tiny — three node types, no escape hatch. There is
  no "anything goes" node, because that is precisely the fall-through this module
  exists to remove; an unconstrained leaf is expressed as the constraint that is
  actually meant (`{:map, []}` — an object, contents unconstrained).

  ## Spec grammar

      {:map, entries}              # a map; every named entry present and valid
      {:map, entries, opts}        # ... plus `min_size:`
      {:list, opts}                # a list; `of:` `min_length:` `length:`

  `entries` is a **list** of `{name, spec}` pairs rather than a map, so error
  order is authored rather than hash order (D-50), and extra keys are permitted:
  a later phase must be able to freeze additional sections or diagnostics without
  this module being the thing that blocks it (Phase 07 fills `dav`; Phase 06
  added `expected_salaries` to the snapshot and then declared it required).

  ## What "required" means here

  Required is about the keys a *consumer reads*, not about non-emptiness. The
  distinction is D-52's, and it is why the two look different in the schemas:
  `dav => %{}` is a documented state ("Phase 07 has not run"), so `dav` is
  `{:map, []}`; an empty `metrics` map is not a state — those arrays *are* the
  percentile tables every later grade is scored against, so a section carrying a
  full `teams` list and no arrays reads back as a well-formed document that can
  score nothing. Both are `%{}`; only one may pass.
  """

  # D-43: five raw arrays, not four — Two-Way Balance is frozen as its two
  # percentile inputs (`rotation_oepm`/`rotation_depm`) rather than as a raw
  # metric, and combined 70/30 at scoring time. Declared here rather than in the
  # builder so `Benchmark` can require exactly what `BenchmarkBuilder` writes
  # without either depending on the other, and so the two cannot drift.
  @competitive_strength_metrics [
    :star_power,
    :starting_five,
    :rotation_quality,
    :rotation_oepm,
    :rotation_depm
  ]

  @type name :: String.t()

  @type spec ::
          {:map, [{name(), spec()}]}
          | {:map, [{name(), spec()}], keyword()}
          | {:list, keyword()}

  @doc """
  The Competitive Strength raw metric names every benchmark must carry.

  One declaration, read by both the builder and the schema, so a metric cannot be
  added to the build and silently left optional at the persistence boundary.
  """
  @spec competitive_strength_metrics() :: [atom()]
  def competitive_strength_metrics, do: @competitive_strength_metrics

  # ── Canonicalization ───────────────────────────────────────────────────

  @doc """
  Normalizes a document to string keys, or refuses it.

  Returns `{:ok, canonical}` for a document whose every key is an atom or a
  string with no two keys normalizing to the same name, and `{:error, messages}`
  otherwise. Structs are values, not documents — a `DateTime` in `meta` is left
  alone, since its keys are fixed at compile time and cannot collide.

  Total: it never raises, including on exotic keys, because raising out of a
  validator is the failure mode a fail-closed boundary exists to remove (D-52).

      iex> alias TradeMachine.GMGradeV2.FrozenDocument
      iex> FrozenDocument.canonicalize(%{skills: %{teams: []}})
      {:ok, %{"skills" => %{"teams" => []}}}
      iex> FrozenDocument.canonicalize(%{:skills => %{}, "skills" => %{}})
      {:error, ["skills has both atom and string keys, which collide in jsonb"]}
  """
  @spec canonicalize(term()) :: {:ok, term()} | {:error, [String.t()]}
  def canonicalize(document) do
    case walk(document, []) do
      {canonical, []} -> {:ok, canonical}
      {_canonical, errors} -> {:error, errors |> Enum.uniq() |> Enum.sort()}
    end
  end

  defp walk(map, path) when is_map(map) and not is_struct(map) do
    {named, key_errors} =
      Enum.reduce(map, {[], []}, fn {key, value}, {named, errors} ->
        case key_name(key) do
          {:ok, name} ->
            {[{name, value} | named], errors}

          :error ->
            {named,
             [
               render(path) <> " has a key that is neither an atom nor a string: #{inspect(key)}"
               | errors
             ]}
        end
      end)

    collision_errors =
      named
      |> Enum.group_by(fn {name, _value} -> name end)
      |> Enum.reject(fn {_name, group} -> match?([_only], group) end)
      |> Enum.map(fn {name, _group} ->
        render(path ++ [name]) <> " has both atom and string keys, which collide in jsonb"
      end)

    {canonical, child_errors} =
      Enum.reduce(named, {%{}, []}, fn {name, value}, {acc, errors} ->
        {child, child_errors} = walk(value, path ++ [name])
        {Map.put(acc, name, child), errors ++ child_errors}
      end)

    {canonical, key_errors ++ collision_errors ++ child_errors}
  end

  # `is_list/1` is true of `[1 | 2]`, and every `Enum` traversal raises on one.
  # An improper list has no JSON representation at all, so it is refused here
  # rather than allowed to raise out of a validator whose whole contract is that
  # it does not (D-52).
  defp walk(list, path) when is_list(list) do
    if proper_list?(list) do
      {canonical, errors} =
        list
        |> Enum.with_index()
        |> Enum.reduce({[], []}, fn {value, index}, {acc, errors} ->
          {child, child_errors} = walk(value, path ++ ["[#{index}]"])
          {[child | acc], errors ++ child_errors}
        end)

      {Enum.reverse(canonical), errors}
    else
      {list, [render(path) <> " is an improper list, which has no JSON representation"]}
    end
  end

  defp walk(value, _path), do: {value, []}

  @doc """
  Whether a term is a proper list.

  Public because `is_list/1` admits an improper list while `length/1` raises on
  one, so any caller deriving a length from an untrusted document needs the same
  test this module uses.
  """
  @spec proper_list?(term()) :: boolean()
  def proper_list?([]), do: true
  def proper_list?([_head | tail]), do: proper_list?(tail)
  def proper_list?(_other), do: false

  defp key_name(key) when is_binary(key), do: {:ok, key}
  defp key_name(key) when is_atom(key), do: {:ok, Atom.to_string(key)}
  defp key_name(_key), do: :error

  # ── Schema validation ──────────────────────────────────────────────────

  @doc """
  Validates a canonicalized document against a `t:spec/0`.

  Returns `[]` for a conforming document and a list of rendered messages
  otherwise. Run this on the output of `canonicalize/1`, never on a raw
  document: every name in a spec is a string, so an atom-keyed document would
  report every required key as missing.
  """
  @spec validate(term(), spec()) :: [String.t()]
  def validate(document, spec), do: check(document, spec, [])

  defp check(value, {:map, entries}, path), do: check(value, {:map, entries, []}, path)

  defp check(value, {:map, entries, opts}, path)
       when is_map(value) and not is_struct(value) and is_list(opts) do
    min_size = Keyword.get(opts, :min_size, 0)

    size_errors =
      if map_size(value) < min_size do
        [render(path) <> " must not be empty"]
      else
        []
      end

    # For a node whose entries may each legitimately be empty, this is the only
    # place the "all of them empty" state can be refused. Counted over the
    # node's *declared* entries so an extra key cannot satisfy it.
    populated_errors =
      case Keyword.get(opts, :min_populated, 0) do
        min when min > 0 ->
          populated =
            Enum.count(entries, fn {name, _sub_spec} ->
              case Map.fetch(value, name) do
                {:ok, sub_value} -> populated?(sub_value)
                :error -> false
              end
            end)

          if populated < min do
            [
              render(path) <>
                " must hold at least #{min} non-empty entry, got #{populated} of #{length(entries)}"
            ]
          else
            []
          end

        _ ->
          []
      end

    entry_errors =
      Enum.flat_map(entries, fn {name, sub_spec} ->
        # `Map.fetch/2`, never `Map.get(a) || Map.get(b)`: both `nil || x` and
        # `false || x` fall through, which is how a stored `false` gets mapped
        # onto the benign "not supplied" branch (D-52).
        case Map.fetch(value, name) do
          {:ok, sub_value} -> check(sub_value, sub_spec, path ++ [name])
          :error -> [render(path ++ [name]) <> " is missing"]
        end
      end)

    size_errors ++ populated_errors ++ entry_errors
  end

  defp check(value, {:list, opts}, path) when is_list(value) and is_list(opts) do
    # `length/1` raises on an improper list, and `is_list/1` admits one. The
    # changeset path canonicalizes first, so this is only reachable through the
    # public `validate/2` — but a fail-closed walker that raises on one input
    # shape is not fail-closed.
    if proper_list?(value) do
      length_errors(value, opts, path) ++ element_errors(value, opts, path)
    else
      [render(path) <> " is an improper list, which has no JSON representation"]
    end
  end

  # The fail-closed default. Every value that reached no clause above is invalid,
  # including one whose spec this module does not recognise — the walker never
  # falls through to `[]`, which is the whole of D-49's intent made total.
  defp check(value, spec, path), do: [render(path) <> " " <> mismatch(spec, value)]

  # "Empty" is defined only for the containers a table can be; anything else is
  # a value, and a value is populated. Improper lists are handled by the list
  # clause, so this is never asked to measure one.
  defp populated?(value) when is_list(value), do: proper_list?(value) and value != []
  defp populated?(value) when is_map(value) and not is_struct(value), do: map_size(value) > 0
  defp populated?(_value), do: true

  defp length_errors(value, opts, path) do
    actual = length(value)
    min_length = Keyword.get(opts, :min_length, 0)

    cond do
      Keyword.has_key?(opts, :length) and actual != Keyword.fetch!(opts, :length) ->
        [
          render(path) <>
            " must hold exactly #{Keyword.fetch!(opts, :length)} #{element_noun(opts)}, got #{actual}"
        ]

      actual < min_length ->
        [render(path) <> " must hold at least #{min_length} #{element_noun(opts)}, got #{actual}"]

      true ->
        []
    end
  end

  defp element_errors(value, opts, path) do
    case Keyword.fetch(opts, :of) do
      :error ->
        []

      {:ok, of} ->
        value
        |> Enum.with_index()
        |> Enum.reject(fn {element, _index} -> element?(element, of) end)
        |> Enum.map(fn {_element, index} -> index end)
        |> case do
          [] -> []
          bad -> [render(path) <> " must hold only #{element_noun(opts)}#{render_indices(bad)}"]
        end
    end
  end

  defp element?(element, :number), do: is_number(element)
  defp element?(element, :map), do: is_map(element) and not is_struct(element)

  # The same fail-closed default `check/3` carries, one level down. Without it an
  # `of: :string` added by a later phase raises `FunctionClauseError` on the
  # frozen-write path instead of reporting — an unrecognised spec must invalidate,
  # never crash and never silently pass.
  defp element?(_element, _of), do: false

  defp element_noun(opts) do
    case Keyword.get(opts, :of) do
      :number -> "numeric values"
      :map -> "objects"
      nil -> "values"
      other -> "values of the unrecognised element type #{inspect(other)}"
    end
  end

  defp render_indices(indices) do
    shown = Enum.take(indices, 5)
    suffix = if length(indices) > 5, do: ", …", else: ""
    " (bad at index #{Enum.join(shown, ", ")}#{suffix})"
  end

  defp mismatch({:map, _entries}, value), do: mismatch({:map, [], []}, value)
  defp mismatch({:map, _entries, _opts}, value), do: "must be a map, got: #{inspect(value)}"

  defp mismatch({:list, opts}, value),
    do: "must be a list of #{element_noun(opts)}, got: #{inspect(value)}"

  defp mismatch(spec, value), do: "does not match #{inspect(spec)}, got: #{inspect(value)}"

  defp render([]), do: "the document"
  defp render(path), do: path |> Enum.join(".") |> String.replace(".[", "[")

  # ── Introspection (used by the D-54 property test) ─────────────────────

  @doc """
  Every path a spec requires, paired with the spec that governs it.

  Derived from the spec rather than restated, so the property test that asserts
  "dropping or corrupting any required key invalidates the document" widens
  automatically when a section is added — the point of D-54's *one* property test
  instead of N example tests.
  """
  @spec required_paths(spec()) :: [{[name()], spec()}]
  def required_paths({:map, entries}), do: required_paths({:map, entries, []})

  def required_paths({:map, entries, _opts}) do
    Enum.flat_map(entries, fn {name, sub_spec} ->
      [{[name], sub_spec}] ++
        Enum.map(required_paths(sub_spec), fn {path, spec} -> {[name | path], spec} end)
    end)
  end

  def required_paths(_spec), do: []

  @doc """
  Values that must never validate against `spec` — the corruption half of the
  D-54 property test.

  Derived from the node type so that every kind of malformation the grammar can
  express is exercised at every required path: the wrong primitive type, the
  documented-empty-vs-meaningfully-empty pair (D-52), and — for arrays — right
  cardinality with unusable contents, which is the shape that has the right
  length and still cannot be ranked against.
  """
  @spec invalid_samples(spec()) :: [term()]
  def invalid_samples({:map, entries}), do: invalid_samples({:map, entries, []})

  def invalid_samples({:map, entries, opts}) do
    # `%{}` is invalid for a node that requires *either* a minimum size or named
    # entries, and valid for one that requires neither — which is exactly the
    # `meta`-vs-`dav` pair. Deriving it from the node rather than listing it
    # keeps the property test asserting the distinction instead of assuming it.
    empty = if Keyword.get(opts, :min_size, 0) > 0 or entries != [], do: [%{}], else: []
    [nil, false, [], "not a map", 3] ++ empty
  end

  def invalid_samples({:list, opts}) do
    empty = if Keyword.get(opts, :min_length, 0) > 0, do: [[]], else: []

    wrong_contents =
      case {Keyword.get(opts, :of), Keyword.get(opts, :length)} do
        {nil, _} -> []
        {_of, nil} -> [[nil]]
        # Right cardinality, unusable contents — the case cardinality checks miss.
        {_of, length} -> [List.duplicate(nil, length), List.duplicate(0.0, length + 1)]
      end

    [nil, false, %{}, "not a list"] ++ empty ++ wrong_contents
  end

  def invalid_samples(_spec), do: [nil]

  # ── Changeset integration ──────────────────────────────────────────────

  @doc """
  Canonicalizes and validates a jsonb document field, storing the canonical form.

  One implementation for both frozen tables, so the two cannot drift into
  enforcing different things — the divergence that let `Benchmark` gain four
  rounds of hardening while `ChallengeSnapshot`, the row every entrant in a
  single challenge is scored against, kept the round-1 checks.

  `schema_fun` receives the *canonicalized* document and returns its
  `t:spec/0`, so a schema can depend on the document's own contents (a ranked
  array holding one value per team, say) without any caller re-deriving it.

  On success the field is replaced by its canonical string-keyed form: what is
  written is then exactly what jsonb reads back (D-42). On failure every problem
  is reported at once rather than one per insert attempt — a build that has to
  be rerun per error is a build nobody reruns.
  """
  @spec validate_document(Ecto.Changeset.t(), atom(), (term() -> spec())) :: Ecto.Changeset.t()
  def validate_document(changeset, field, schema_fun) do
    # `fetch_field/2`, NOT `fetch_change/2`. `cast/3` records no change when the
    # cast value equals the struct's current one (`Ecto.Type.equal?/3`), so
    # `fetch_change/2` answers `:error` for a document that was supplied, cast
    # cleanly, and is about to be written — `Repo.insert` merges the whole struct
    # into the changes regardless. Skipping there let a malformed document reach
    # jsonb with the schema never consulted, which is D-54's finding class at a
    # new line. `fetch_field/2` returns the *effective* value, which is exactly
    # the one Postgres will store.
    #
    # A `nil` is the one thing still worth skipping: the field was omitted or
    # explicitly nil, `validate_required/2` already says so, and a second error
    # for the same cause is noise.
    case Ecto.Changeset.fetch_field(changeset, field) do
      :error ->
        changeset

      {_source, nil} ->
        changeset

      {_source, document} ->
        case canonicalize(document) do
          {:ok, canonical} ->
            case validate(canonical, schema_fun.(canonical)) do
              [] ->
                changeset
                |> Ecto.Changeset.put_change(field, canonical)
                |> validate_encodable(field, canonical)

              errors ->
                add_errors(changeset, field, errors)
            end

          {:error, errors} ->
            add_errors(changeset, field, errors)
        end
    end
  end

  # A document can satisfy the schema and still fail to reach Postgres: a tuple
  # or a PID under `meta` raises inside the jsonb encode, turning an insert into
  # an exception rather than a changeset error. A fail-closed boundary that
  # crashes instead of reporting is still not reporting (D-52), and on a
  # write-once path the caller deserves an error it can act on. One encode of a
  # document built once per season is not a cost worth optimizing.
  # Encoding proves the document can be written; it does not prove that what
  # comes back is what went in. Canonicalization normalizes *keys*, so a leaf
  # value Jason TRANSFORMS rather than preserves — a `DateTime`, a `Date`, a
  # bare atom — passes the schema, encodes cleanly, and is stored as a string
  # while the struct returned to the caller still holds the original term. That
  # is exactly the in-memory/Postgres divergence D-42 exists to prevent, moved
  # from keys to values. Decoding what was just encoded and comparing closes it
  # by construction, and the builder's output already satisfies it.
  defp validate_encodable(changeset, field, canonical) do
    with {:ok, json} <- Jason.encode(canonical),
         {:ok, decoded} <- Jason.decode(json) do
      if decoded == canonical do
        changeset
      else
        Ecto.Changeset.add_error(
          changeset,
          field,
          "does not survive a JSON round trip unchanged: a value is rewritten by encoding " <>
            "(a struct, date, or bare atom stored as a string), so the frozen document would " <>
            "differ from the one in memory (D-42)"
        )
      end
    else
      {:error, error} ->
        Ecto.Changeset.add_error(changeset, field, "cannot be encoded as JSON: #{inspect(error)}")
    end
  end

  defp add_errors(changeset, field, errors) do
    Enum.reduce(errors, changeset, fn message, acc ->
      Ecto.Changeset.add_error(acc, field, message)
    end)
  end
end