super_cache

defmodule SuperCache.Cluster.TxnRegistry do
  @moduledoc """
  In-memory transaction log for the three-phase commit protocol.

  Each in-flight transaction is stored as a map entry keyed by its
  `txn_id` binary.  The log is held in an ETS table owned by this
  GenServer so that:

  - Reads are lock-free (`:public` + `:read_concurrency`).
  - The table survives coordinator or participant crashes as long as the
    owning VM is alive (the ETS table is linked to this GenServer's
    lifetime, not to the caller's).
  - On node restart, `SuperCache.Cluster.ThreePhaseCommit.recover/0` can
    iterate the table to resolve in-doubt transactions.

  ## Transaction states

  | State           | Meaning                                                   |
  |-----------------|-----------------------------------------------------------|
  | `:prepared`     | PREPARE accepted; waiting for PRE_COMMIT or ABORT        |
  | `:pre_committed`| PRE_COMMIT acknowledged; commit is safe even after crash |
  | `:committed`    | COMMIT applied; entry removed immediately after          |
  | `:aborted`      | ABORT received; entry removed immediately after          |

  ## Memory management

  Committed and aborted entries are deleted synchronously by
  `ThreePhaseCommit` after each successful/failed transaction.
  Stale `:prepared` entries (e.g. from a coordinator crash before
  PRE_COMMIT) are cleaned up by `recover/0` on the next startup.
  """

  use GenServer, restart: :permanent, shutdown: 5_000

  require Logger
  require SuperCache.Log

  @table __MODULE__

  # ── Lifecycle ────────────────────────────────────────────────────────────────

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)

  @impl true
  def init(_opts) do
    @table = :ets.new(@table, [
      :set,
      :public,
      :named_table,
      {:read_concurrency, true},
      {:write_concurrency, true}
    ])

    Logger.info("super_cache, txn_registry, ETS table #{inspect(@table)} ready")
    {:ok, %{}}
  end

  # ── Public API ───────────────────────────────────────────────────────────────

  @doc """
  Register a new transaction in state `:prepared`.

  `replicas` is the list of participant nodes (may be empty on participants
  themselves — they only need `ops` and `partition_idx`).
  """
  @spec register(binary, non_neg_integer, list, [node]) :: :ok
  def register(txn_id, partition_idx, ops, replicas) do
    entry = %{
      txn_id:        txn_id,
      partition_idx: partition_idx,
      ops:           ops,
      replicas:      replicas,
      state:         :prepared,
      inserted_at:   System.monotonic_time(:millisecond)
    }

    :ets.insert(@table, {txn_id, entry})
    :ok
  end

  @doc "Transition `txn_id` to `:pre_committed`."
  @spec mark_pre_committed(binary) :: :ok
  def mark_pre_committed(txn_id) do
    update_state(txn_id, :pre_committed)
  end

  @doc "Look up a transaction by id.  Returns the entry map or `nil`."
  @spec get(binary) :: map | nil
  def get(txn_id) do
    case :ets.lookup(@table, txn_id) do
      [{^txn_id, entry}] -> entry
      []                 -> nil
    end
  end

  @doc "Remove a transaction entry (called after commit or abort)."
  @spec remove(binary) :: :ok
  def remove(txn_id) do
    :ets.delete(@table, txn_id)
    :ok
  end

  @doc "Return all transaction entries as `[{txn_id, entry}]`."
  @spec list_all() :: [{binary, map}]
  def list_all() do
    :ets.tab2list(@table)
  end

  @doc "Return the number of in-flight transactions (useful for tests)."
  @spec count() :: non_neg_integer
  def count() do
    :ets.info(@table, :size)
  end

  # ── Private ──────────────────────────────────────────────────────────────────

  defp update_state(txn_id, new_state) do
    case :ets.lookup(@table, txn_id) do
      [{^txn_id, entry}] ->
        :ets.insert(@table, {txn_id, %{entry | state: new_state}})
        :ok

      [] ->
        Logger.warning("super_cache, txn_registry, update_state: txn=#{txn_id} not found")
        :ok
    end
  end
end