defmodule SpacetimeDB.Table do
  @moduledoc """
  An ETS-backed local mirror of a single SpacetimeDB table.

  `SpacetimeDB.Table` is a GenServer that owns a named ETS table and keeps it
  in sync with the server via `apply_update/2`.  All query functions
  (`all/1`, `get/2`, `filter/2`, `count/1`) read directly from ETS — they
  never go through the GenServer process, so they are non-blocking and can be
  called from any process at full ETS speed.

  ## Setup

      children = [
        {SpacetimeDB.Table,
         name: :players,
         schema: MyGame.Player,
         table_name: "Player",
         primary_key: :id},

        {SpacetimeDB,
         host: "localhost",
         database: "game",
         handler: MyGame.Handler}
      ]

  ## Wiring updates

  Call `apply_update/2` from your handler callbacks.  It accepts
  `TransactionUpdate`, `InitialSubscription`, `SubscribeApplied`, or a bare
  `TableUpdate` struct.  Only rows whose `table_name` matches the `:table_name`
  option are touched.

      defmodule MyGame.Handler do
        @behaviour SpacetimeDB.Handler

        @impl true
        def on_initial_subscription(sub, _),
          do: SpacetimeDB.Table.apply_update(:players, sub)

        @impl true
        def on_transaction_update(update, _),
          do: SpacetimeDB.Table.apply_update(:players, update)
      end

  ## Querying

      SpacetimeDB.Table.all(:players)
      #=> [%MyGame.Player{id: 1, name: "Alice"}, ...]

      SpacetimeDB.Table.get(:players, 1)
      #=> {:ok, %MyGame.Player{id: 1, name: "Alice"}}

      SpacetimeDB.Table.filter(:players, fn p -> p.health > 50 end)
      #=> [%MyGame.Player{...}, ...]

      SpacetimeDB.Table.count(:players)
      #=> 42

  ## BSATN vs JSON rows

  - **BSATN mode** (default): row inserts/deletes arrive as raw binaries and are
    decoded via `schema.decode/1` (generated by `SpacetimeDB.BSATN.Schema`).
  - **JSON mode**: rows arrive as string-keyed maps and are coerced to structs via
    `String.to_existing_atom/1` on the keys.

  ## Options

  | Key | Type | Required | Description |
  |-----|------|----------|-------------|
  | `:name` | `atom()` | yes | Registered name for the GenServer and ETS lookup |
  | `:schema` | `module()` | yes | Module that implements `decode/1` and `encode/1` (via `SpacetimeDB.BSATN.Schema`) |
  | `:table_name` | `String.t()` | yes | SpacetimeDB table name to track (must match `TableUpdate.table_name`) |
  | `:primary_key` | `atom()` | yes | Field name used as the ETS key |
  """

  use GenServer

  require Logger

  alias SpacetimeDB.Types

  # ---------------------------------------------------------------------------
  # Public API
  # ---------------------------------------------------------------------------

  @doc "Start and link the table mirror process."
  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts) do
    name = Keyword.fetch!(opts, :name)
    GenServer.start_link(__MODULE__, opts, name: name)
  end

  @doc false
  def child_spec(opts) do
    name = Keyword.fetch!(opts, :name)
    %{id: {__MODULE__, name}, start: {__MODULE__, :start_link, [opts]}, restart: :permanent}
  end

  # ---------------------------------------------------------------------------
  # Query API  (direct ETS reads — no GenServer call)
  # ---------------------------------------------------------------------------

  @doc "Return all rows as a list of structs."
  @spec all(atom()) :: [struct()]
  def all(name) do
    name
    |> ets_name()
    |> :ets.tab2list()
    |> Enum.map(fn {_k, v} -> v end)
  end

  @doc "Look up a single row by primary key.  Returns `{:ok, row}` or `:error`."
  @spec get(atom(), term()) :: {:ok, struct()} | :error
  def get(name, key) do
    tab = ets_name(name)

    case :ets.lookup(tab, key) do
      [{^key, row}] -> {:ok, row}
      [] -> :error
    end
  end

  @doc "Return all rows matching the predicate function."
  @spec filter(atom(), (struct() -> boolean())) :: [struct()]
  def filter(name, fun), do: Enum.filter(all(name), fun)

  @doc "Return the number of rows currently in the table."
  @spec count(atom()) :: non_neg_integer()
  def count(name), do: :ets.info(ets_name(name), :size)

  # ---------------------------------------------------------------------------
  # Update API  (direct ETS writes — no GenServer call)
  # ---------------------------------------------------------------------------

  @doc """
  Apply a server message to this table mirror.

  Accepts `TransactionUpdate`, `InitialSubscription`, `SubscribeApplied`, or a
  bare `TableUpdate`.  Only rows whose `table_name` matches the table configured
  at `start_link` time are affected.
  """
  @spec apply_update(atom(), term()) :: :ok
  def apply_update(name, %Types.TransactionUpdate{tables: tables}),
    do: apply_matching(name, tables)

  def apply_update(name, %Types.InitialSubscription{tables: tables}),
    do: apply_matching(name, tables)

  def apply_update(name, %Types.SubscribeApplied{tables: tables}),
    do: apply_matching(name, tables)

  def apply_update(name, %Types.TableUpdate{} = tu),
    do: do_apply(name, tu)

  def apply_update(_name, _other), do: :ok

  # ---------------------------------------------------------------------------
  # GenServer — owns the ETS table
  # ---------------------------------------------------------------------------

  @impl GenServer
  def init(opts) do
    name = Keyword.fetch!(opts, :name)
    schema = Keyword.fetch!(opts, :schema)
    table_name = Keyword.fetch!(opts, :table_name)
    primary_key = Keyword.fetch!(opts, :primary_key)

    tab = :ets.new(ets_name(name), [:named_table, :public, :set, {:read_concurrency, true}])

    # Store config in persistent_term so apply_update/query functions can read
    # it without a GenServer call.
    :persistent_term.put(config_key(name), %{
      schema: schema,
      table_name: table_name,
      primary_key: primary_key
    })

    {:ok, %{name: name, tab: tab}}
  end

  @impl GenServer
  def terminate(_reason, %{name: name, tab: tab}) do
    :persistent_term.erase(config_key(name))
    :ets.delete(tab)
    :ok
  end

  # ---------------------------------------------------------------------------
  # Private — update helpers
  # ---------------------------------------------------------------------------

  defp apply_matching(name, table_updates) do
    %{table_name: watched} = config(name)

    Enum.each(table_updates, fn tu ->
      if tu.table_name == watched, do: do_apply(name, tu)
    end)
  end

  defp do_apply(name, %Types.TableUpdate{inserts: inserts, deletes: deletes}) do
    %{schema: schema, primary_key: pk} = config(name)
    tab = ets_name(name)

    for row <- inserts do
      case decode_row(schema, row) do
        {:ok, struct} -> :ets.insert(tab, {Map.fetch!(struct, pk), struct})
        {:error, reason} -> Logger.warning("[SpacetimeDB.Table] insert decode failed: #{inspect(reason)}")
      end
    end

    for row <- deletes do
      case decode_row(schema, row) do
        {:ok, struct} -> :ets.delete(tab, Map.fetch!(struct, pk))
        {:error, reason} -> Logger.warning("[SpacetimeDB.Table] delete decode failed: #{inspect(reason)}")
      end
    end

    :ok
  end

  # BSATN rows arrive as binaries
  defp decode_row(schema, row) when is_binary(row) do
    case schema.decode(row) do
      {:ok, struct, _rest} -> {:ok, struct}
      {:error, _} = err -> err
    end
  end

  # JSON rows arrive as string-keyed maps
  defp decode_row(schema, row) when is_map(row) do
    attrs =
      Map.new(row, fn {k, v} ->
        {String.to_existing_atom(k), v}
      end)

    {:ok, struct!(schema, attrs)}
  rescue
    e -> {:error, Exception.message(e)}
  end

  defp decode_row(_schema, other), do: {:error, {:unexpected_row_type, other}}

  # ---------------------------------------------------------------------------
  # Private — naming helpers
  # ---------------------------------------------------------------------------

  defp ets_name(name), do: :"__stdb_table_#{name}"
  defp config_key(name), do: {__MODULE__, name}
  defp config(name), do: :persistent_term.get(config_key(name))
end