bikoret/vav_pipeline/lib/vav_pipeline/manna_ledger.ex

defmodule VavPipeline.MannaLedger do
  @moduledoc """
  Manna consumption ledger — d'var yom b'yomo.

  Shemot 16:4 — "The people shall go out and gather the matter
  of a day in its day." Each cycle receives exactly what it needs.

  Shemot 16:19 — "Let no one leave any of it until morning."
  Root ותר (yoter) — do not hoard. At cycle boundary, surplus
  is released. Not rolled over. Not saved.

  Shemot 16:20 — "Some left over... and it bred worms and stank."
  Root באש (ba'ash) — hoarded resources breed corruption.

  This GenServer tracks:
  1. Today's allocated provision (from the manna file)
  2. Consumption against each resource category
  3. Remaining balance
  4. Cycle boundary enforcement (zeroing out surplus)

  Architecture §4, Step 26: Manna anti-hoarding enforcement
  and cycle-boundary surplus release.

  ## What is eaten stays eaten

  The day's consumption is written to `ledger.json` beside the provision
  file, after every cycle that eats. A ledger that lives only in memory
  forgets what was eaten whenever the process restarts: the same day's
  provision is gathered again, whole. While manna was an emptyDir that
  needed the kohen's hand; once the provision was put on a volume
  (2026-10-04) a crash loop alone would have refilled the jar, as often
  as it crashed. *Ish l'fi ochlo* — each according to what he eats
  (Shemot 16:18): the omer is measured once.

  - On the first load of a day — and at once when the process starts —
    if `ledger.json` carries TODAY's date, consumption resumes from it. Another day's file is not this day's
    record and is ignored.
  - At the day boundary the file is cleared with the surplus. A release
    within the same day clears the allocation but not the record.
  - The file is a convenience of the vessel, never a source of manna:
    it can only reduce what remains. If it cannot be read or written the
    ledger keeps counting in memory and says so; it does not stop the
    cycle.
  """

  use GenServer

  require Logger

  alias VavPipeline.{Manna, Chatzotzrot}

  @type resource_key :: :compute_units | :memory_mb | :api_calls_external |
                        :storage_delta_mb | :sanhedrin_panels_max

  @type ledger_state :: %{
          cycle_date: String.t() | nil,
          allocated: map(),
          consumed: map(),
          cycle_count: non_neg_integer(),
          last_enforcement: DateTime.t() | nil,
          hoarding_violations: non_neg_integer()
        }

  @zero_provision %{
    compute_units: 0,
    memory_mb: 0,
    api_calls_external: 0,
    storage_delta_mb: 0,
    sanhedrin_panels_max: 0
  }

  # ============================================================
  # Client API
  # ============================================================

  def start_link(opts \\ []) do
    name = Keyword.get(opts, :name, __MODULE__)
    GenServer.start_link(__MODULE__, opts, name: name)
  end

  @doc """
  Load today's provision from the manna file into the ledger.

  Called at the start of each cycle. Resets consumption to zero.
  If provision was already loaded for today, returns {:ok, :already_loaded}.
  """
  @spec load_provision() :: {:ok, :loaded | :already_loaded | :no_provision} | {:error, String.t()}
  def load_provision do
    GenServer.call(__MODULE__, :load_provision)
  end

  @doc """
  Consume resources from today's manna provision.

  D'var yom b'yomo — take only what the day requires.

  Parameters:
  - resource: which resource to consume (:compute_units, :api_calls_external, etc.)
  - amount: how much to consume

  Returns:
  - {:ok, remaining} — consumption recorded, remaining balance returned
  - {:error, :provision_exhausted} — no more of this resource available
  - {:error, :no_provision} — no provision loaded for today
  """
  @spec consume(resource_key(), pos_integer()) ::
          {:ok, non_neg_integer()} | {:error, atom()}
  def consume(resource, amount) when is_atom(resource) and is_integer(amount) and amount > 0 do
    GenServer.call(__MODULE__, {:consume, resource, amount})
  end

  @doc """
  Check remaining balance for a resource.
  """
  @spec remaining(resource_key()) :: non_neg_integer()
  def remaining(resource) when is_atom(resource) do
    GenServer.call(__MODULE__, {:remaining, resource})
  end

  @doc """
  Check if a resource has sufficient provision for the requested amount.
  """
  @spec available?(resource_key(), pos_integer()) :: boolean()
  def available?(resource, amount) when is_atom(resource) and is_integer(amount) do
    GenServer.call(__MODULE__, {:available?, resource, amount})
  end

  @doc """
  Get the full ledger state — allocated, consumed, remaining.
  """
  @spec ledger() :: ledger_state()
  def ledger do
    GenServer.call(__MODULE__, :ledger)
  end

  @doc """
  Enforce anti-hoarding — release all surplus at cycle boundary.

  Shemot 16:19 — al yoter mimenu ad boker.
  "Let no one leave any of it until morning."

  This zeroes out remaining allocation and records the enforcement.
  Called by BereishitCycle.erev_voker/2 at cycle boundary.

  Returns {:ok, %{released: map()}} with the amounts released (what was left over).
  """
  @spec enforce_cycle_boundary() :: {:ok, %{released: map()}}
  def enforce_cycle_boundary do
    GenServer.call(__MODULE__, :enforce_cycle_boundary)
  end

  @doc """
  Reset the ledger to initial state, and forget the persisted day.
  For testing only.
  """
  @spec reset() :: :ok
  def reset do
    GenServer.call(__MODULE__, :reset)
  end

  @doc """
  Where the day's consumption is kept: `ledger.json`, beside the
  provision file (on the manna volume in production).
  """
  @spec ledger_file() :: String.t()
  def ledger_file do
    Path.join(Path.dirname(Manna.provision_file()), "ledger.json")
  end

  # ============================================================
  # GenServer callbacks
  # ============================================================

  @impl true
  def init(_opts) do
    state = %{
      cycle_date: nil,
      allocated: @zero_provision,
      consumed: @zero_provision,
      cycle_count: 0,
      last_enforcement: nil,
      hoarding_violations: 0
    }

    Logger.info("MannaLedger: ledger ready — d'var yom b'yomo")
    {:ok, state, {:continue, :resume_day}}
  end

  # A process that starts in the middle of a day whose record is on disk
  # takes the day up at once, rather than at its first cycle — so that
  # /kodesh/status never shows an untouched jar that is in fact half
  # eaten. With no record for today there is nothing to resume, and the
  # first cycle gathers as it always did.
  @impl true
  def handle_continue(:resume_day, state) do
    today = Date.utc_today() |> Date.to_iso8601()

    if persisted_date() == today do
      {_reply, new_state} = load_for(today, state)
      {:noreply, new_state}
    else
      {:noreply, state}
    end
  end

  @impl true
  def handle_call(:reset, _from, _state) do
    forget_persisted()

    {:reply, :ok,
     %{
       cycle_date: nil,
       allocated: @zero_provision,
       consumed: @zero_provision,
       cycle_count: 0,
       last_enforcement: nil,
       hoarding_violations: 0
     }}
  end

  @impl true
  def handle_call(:load_provision, _from, state) do
    today = Date.utc_today() |> Date.to_iso8601()

    if state.cycle_date == today do
      {:reply, {:ok, :already_loaded}, state}
    else
      {reply, new_state} = load_for(today, state)
      {:reply, reply, new_state}
    end
  end

  @impl true
  def handle_call({:consume, resource, amount}, _from, state) do
    allocated = Map.get(state.allocated, resource, 0)
    consumed = Map.get(state.consumed, resource, 0)
    remaining = allocated - consumed

    cond do
      state.cycle_date == nil ->
        {:reply, {:error, :no_provision}, state}

      remaining <= 0 ->
        Logger.warning(
          "MannaLedger: #{resource} exhausted — " <>
            "allocated #{allocated}, consumed #{consumed}"
        )

        {:reply, {:error, :provision_exhausted}, state}

      amount > remaining ->
        # Partial consumption not allowed — don't take more than what remains
        Logger.warning(
          "MannaLedger: #{resource} insufficient — " <>
            "requested #{amount}, remaining #{remaining}"
        )

        {:reply, {:error, :provision_exhausted}, state}

      true ->
        new_consumed = Map.put(state.consumed, resource, consumed + amount)
        new_remaining = remaining - amount
        new_state = %{state | consumed: new_consumed}
        persist(new_state)

        # Signal when provision runs low (< 10% remaining)
        if new_remaining < allocated * 0.1 and new_remaining > 0 do
          Chatzotzrot.signal(:assembly_leaders,
            reason: "Manna low: #{resource} at #{new_remaining}/#{allocated}"
          )
        end

        {:reply, {:ok, new_remaining}, new_state}
    end
  end

  @impl true
  def handle_call({:remaining, resource}, _from, state) do
    allocated = Map.get(state.allocated, resource, 0)
    consumed = Map.get(state.consumed, resource, 0)
    {:reply, max(allocated - consumed, 0), state}
  end

  @impl true
  def handle_call({:available?, resource, amount}, _from, state) do
    allocated = Map.get(state.allocated, resource, 0)
    consumed = Map.get(state.consumed, resource, 0)
    {:reply, (allocated - consumed) >= amount, state}
  end

  @impl true
  def handle_call(:ledger, _from, state) do
    {:reply, state, state}
  end

  @impl true
  def handle_call(:enforce_cycle_boundary, _from, state) do
    now = DateTime.utc_now()

    # Calculate what was left over — the surplus that would breed worms
    released =
      Map.new(@zero_provision, fn {key, _} ->
        allocated = Map.get(state.allocated, key, 0)
        consumed = Map.get(state.consumed, key, 0)
        {key, max(allocated - consumed, 0)}
      end)

    total_released = released |> Map.values() |> Enum.sum()

    if total_released > 0 do
      Logger.info(
        "MannaLedger: cycle boundary — releasing surplus: #{inspect(released)}"
      )
    end

    new_state = %{
      state
      | allocated: @zero_provision,
        consumed: @zero_provision,
        cycle_date: nil,
        last_enforcement: now,
        hoarding_violations: state.hoarding_violations
    }

    # The day's record goes with the day's surplus — when the day is
    # over. A release called while it is still the ledger's day (the
    # late-provision re-gather in Manna.gather_for_cycle/0) lets go of
    # the allocation, not of the memory of what was eaten: the reload
    # that follows resumes today's count.
    if state.cycle_date != (Date.utc_today() |> Date.to_iso8601()) do
      forget_persisted()
    end

    {:reply, {:ok, %{released: released}}, new_state}
  end

  # ============================================================
  # Loading
  # ============================================================

  # Gather `today`'s provision into the ledger, resuming today's
  # consumption if this process is not the first to gather it.
  defp load_for(today, state) do
    case Manna.check() do
      {:ok, :no_provision} ->
        new_state = %{
          state
          | cycle_date: today,
            allocated: @zero_provision,
            consumed: @zero_provision,
            cycle_count: state.cycle_count + 1
        }

        Logger.warning("MannaLedger: no provision for #{today} — system will wait")
        {{:ok, :no_provision}, new_state}

      {:ok, manna} ->
        # What was already eaten today, if this process is not the
        # first to gather it.
        consumed = restore_consumed(today)

        new_state = %{
          state
          | cycle_date: today,
            allocated: manna.provision,
            consumed: consumed,
            cycle_count: state.cycle_count + 1
        }

        Logger.info(
          "MannaLedger: provision loaded for #{today} (#{manna.type}) — " <>
            "compute: #{manna.provision.compute_units}, " <>
            "api_calls: #{manna.provision.api_calls_external}, " <>
            "panels: #{manna.provision.sanhedrin_panels_max}" <>
            if(consumed == @zero_provision,
              do: "",
              else: " — resuming the day: already consumed #{inspect(consumed)}"
            )
        )

        {{:ok, :loaded}, new_state}

      {:error, reason} ->
        {{:error, reason}, state}
    end
  end

  # ============================================================
  # Persistence — ledger.json beside the provision
  # ============================================================

  # Written whole and renamed into place, so a reader (or a crash) never
  # meets half a file.
  defp persist(state) do
    path = ledger_file()
    tmp = path <> ".tmp"

    body =
      Jason.encode!(%{
        "cycle_date" => state.cycle_date,
        "consumed" => Map.new(state.consumed, fn {k, v} -> {Atom.to_string(k), v} end),
        "written_at" => DateTime.utc_now() |> DateTime.to_iso8601()
      })

    with :ok <- File.write(tmp, body),
         :ok <- File.rename(tmp, path) do
      :ok
    else
      {:error, reason} ->
        Logger.warning(
          "MannaLedger: could not persist consumption to #{path} (#{inspect(reason)}) — " <>
            "counting in memory only; a restart today would forget what was eaten"
        )

        :ok
    end
  end

  # Today's consumption from an earlier life of this process, or zeros.
  defp restore_consumed(today) do
    path = ledger_file()

    with {:ok, contents} <- File.read(path),
         {:ok, %{"cycle_date" => ^today, "consumed" => consumed}} when is_map(consumed) <-
           Jason.decode(contents),
         {:ok, restored} <- parse_consumed(consumed) do
      restored
    else
      {:error, :enoent} ->
        @zero_provision

      {:ok, %{"cycle_date" => other}} ->
        Logger.info("MannaLedger: #{path} is for #{inspect(other)}, not #{today} — not this day's record")
        @zero_provision

      other ->
        Logger.warning(
          "MannaLedger: #{path} is unreadable (#{inspect(other)}) — " <>
            "starting today's count from zero"
        )

        @zero_provision
    end
  end

  # Only the five known resources, only non-negative integers. Anything
  # else is not a ledger.
  defp parse_consumed(consumed) do
    Enum.reduce_while(@zero_provision, {:ok, %{}}, fn {key, _}, {:ok, acc} ->
      case Map.get(consumed, Atom.to_string(key), 0) do
        n when is_integer(n) and n >= 0 -> {:cont, {:ok, Map.put(acc, key, n)}}
        bad -> {:halt, {:error, {:bad_consumed, key, bad}}}
      end
    end)
  end

  # The date the record on disk is for, or nil.
  defp persisted_date do
    with {:ok, contents} <- File.read(ledger_file()),
         {:ok, %{"cycle_date" => date}} when is_binary(date) <- Jason.decode(contents) do
      date
    else
      _ -> nil
    end
  end

  defp forget_persisted do
    case File.rm(ledger_file()) do
      :ok -> :ok
      {:error, :enoent} -> :ok
      {:error, reason} ->
        Logger.warning("MannaLedger: could not clear #{ledger_file()} (#{inspect(reason)})")
        :ok
    end
  end
end