defmodule Fief do
  @moduledoc """
  Consistent vnode ownership for Elixir clusters.

  Fief is instance-based: nothing runs under the `:fief` application itself.
  Start instances in your own supervision tree, Ecto/Oban-style:

      children = [
        {Fief,
         name: MyApp.Fief,
         authority: {Fief.Authority.Local, []},
         partitions: 64}
      ]

  The instance `:name` is the **namespace** — the identity of the coordination
  domain, which must match across every node participating in this instance.
  Multiple instances coexist in one BEAM and on one database.

  See `docs/design.md` for guarantees and protocols, `docs/implementation.md`
  for the code shape this module tree follows.
  """

  @doc """
  Optional instance-module sugar (implementation.md §9, deferred from M4 to
  M6): `defmodule MyApp.Fief do use Fief, partitions: 64, ... end` generates a
  module-named instance — `child_spec/1` (runtime opts override use-time
  opts; `:name` is always the module) plus `call/3`, `cast/2`, `owner_of/1`,
  `table/0`, `epoch/0`, `leader/0` wrapping the instance-first API. The core
  stays plain opts; this is convenience only.
  """
  defmacro __using__(use_opts) do
    quote do
      @doc false
      def child_spec(opts) do
        unquote(use_opts)
        |> Keyword.merge(opts)
        |> Keyword.put(:name, __MODULE__)
        |> Fief.child_spec()
      end

      @doc "Start this instance. Prefer `{#{inspect(__MODULE__)}, opts}` in a supervision tree."
      def start_link(opts \\ []) do
        Fief.start_link(
          unquote(use_opts)
          |> Keyword.merge(opts)
          |> Keyword.put(:name, __MODULE__)
        )
      end

      @doc "See `Fief.call/4`."
      def call(vnode, msg, opts \\ []), do: Fief.call(__MODULE__, vnode, msg, opts)

      @doc "See `Fief.cast/3`."
      def cast(vnode, msg), do: Fief.cast(__MODULE__, vnode, msg)

      @doc "See `Fief.owner_of/2`."
      def owner_of(vnode), do: Fief.owner_of(__MODULE__, vnode)

      @doc "See `Fief.table/1`."
      def table, do: Fief.table(__MODULE__)

      @doc "See `Fief.epoch/1`."
      def epoch, do: Fief.epoch(__MODULE__)

      @doc "See `Fief.leader/1`."
      def leader, do: Fief.leader(__MODULE__)
    end
  end

  @doc false
  def child_spec(opts) do
    name = Keyword.fetch!(opts, :name)

    %{
      id: {Fief, name},
      start: {Fief.Supervisor, :start_link, [opts]},
      type: :supervisor
    }
  end

  @doc "Start a Fief instance. Prefer `{Fief, opts}` in a supervision tree."
  @spec start_link(keyword()) :: Supervisor.on_start()
  def start_link(opts), do: Fief.Supervisor.start_link(opts)

  # -- the instance-first API surface (implementation.md §1) --------------------
  #
  # Vnode-level, key-free: keys (hashing, `Fief.call(key, msg)`) arrive with
  # `Fief.Key` at M6, layered on exactly these functions.

  @doc """
  Call `vnode` of `instance` with `msg` and await the impl's reply:
  `{:ok, reply}` or `{:error, reason}`. Runs in the caller (no router
  process); `opts` takes `:timeout` (ms), covering the entire `:moved`-retry
  loop. See `Fief.Router.call/4`.
  """
  @spec call(atom(), non_neg_integer(), term(), keyword()) ::
          {:ok, term()} | {:error, Fief.Router.call_error()}
  def call(instance, vnode, msg, opts \\ []), do: Fief.Router.call(instance, vnode, msg, opts)

  @doc "Fire-and-forget send to `vnode`'s owner. See `Fief.Router.cast/3`."
  @spec cast(atom(), non_neg_integer(), term()) :: :ok | {:error, Fief.Router.call_error()}
  def cast(instance, vnode, msg), do: Fief.Router.cast(instance, vnode, msg)

  @doc "The presumed owner of `vnode` — hint-grade. See `Fief.Router.owner_of/2`."
  @spec owner_of(atom(), non_neg_integer()) ::
          {:ok, term()} | {:error, :no_owner | :unreachable | :not_running}
  def owner_of(instance, vnode), do: Fief.Router.owner_of(instance, vnode)

  @doc """
  The cached vnode table — `{:ok, %{vnode => {owner, prev_owner, row_epoch}},
  epoch}` — hint-grade, from this node's routing cache (an ETS read; the
  Authority is the truth).
  """
  @spec table(atom()) :: {:ok, map(), non_neg_integer()} | {:error, :not_ready | :not_running}
  def table(instance) do
    tab = Fief.Router.Cache.table(instance)

    case :ets.lookup(tab, :table_epoch) do
      [{:table_epoch, epoch}] ->
        rows =
          :ets.select(tab, [
            {{{:route, :"$1"}, :"$2", :"$3", :"$4"}, [], [{{:"$1", {{:"$2", :"$3", :"$4"}}}}]}
          ])

        {:ok, Map.new(rows), epoch}

      [] ->
        {:error, :not_ready}
    end
  rescue
    ArgumentError -> {:error, :not_running}
  end

  @doc "The epoch the cached table was read at. Hint-grade, an ETS read."
  @spec epoch(atom()) :: {:ok, non_neg_integer()} | {:error, :not_ready | :not_running}
  def epoch(instance) do
    case :ets.lookup(Fief.Router.Cache.table(instance), :table_epoch) do
      [{:table_epoch, epoch}] -> {:ok, epoch}
      [] -> {:error, :not_ready}
    end
  rescue
    ArgumentError -> {:error, :not_running}
  end

  @doc """
  The current planner leader, `{:ok, {node_id, term}}`, answered from the
  Authority through the instance's Leadership adapter. `{:error, :disabled}`
  when the instance runs no leadership (`leadership: false` or no vnode impl).
  """
  @spec leader(atom()) ::
          {:ok, {term(), pos_integer()}}
          | {:error, :none | :unreachable | :disabled | :not_running}
  def leader(instance) do
    with {:ok, cfg} <- Fief.Router.config(instance) do
      case cfg.leadership do
        nil -> {:error, :disabled}
        {mod, name} -> mod.leader(name)
      end
    end
  end
end