-module(kura_ets_pool).
-moduledoc """
ETS pool implementation. There are no connections to pool: the "pool"
is a table registry, one gen_server per pool that owns all the ETS
data tables so they survive the (arbitrary, short-lived) processes
that read and write through the repo API.

## Layout

- One gen_server registered under the pool name. It owns every table.
- One named, protected meta table (named after the pool atom) mapping
  `TableName :: binary()` to the data table tid. Callers resolve
  tables with a plain `ets:lookup/2`; only creation goes through the
  server, which serializes concurrent create races.
- One `set, public` data table per source table, created on demand at
  first use (mirroring etso's TableServer-per-schema design).

`checkout/2` hands back the pool name itself: reads and writes go
straight to the public data tables, so there is nothing to lease.

## Example

```erlang
{ok, _Pid} = kura_ets_pool:start_pool(my_pool, #{}),
{ok, Tid} = kura_ets_pool:ensure_table(my_pool, ~"users"),
ok = kura_ets_pool:reset(my_pool),
ok = kura_ets_pool:stop_pool(my_pool).
```
""".

-behaviour(kura_pool).
-behaviour(kura_capabilities).
-behaviour(gen_server).

-export([
    start_pool/2,
    stop_pool/1,
    checkout/2,
    checkin/2,
    give_away/3,
    capabilities/0
]).

-export([ensure_table/2, reset/1]).

-export([init/1, handle_call/3, handle_cast/2]).

%%----------------------------------------------------------------------
%% kura_pool callbacks
%%----------------------------------------------------------------------

-spec start_pool(kura_pool:name(), kura_pool:opts()) -> {ok, pid()} | {error, term()}.
start_pool(Name, _Opts) ->
    %% Unlinked start: the caller is kura's application master, which
    %% must not be linked to (and killed with) a pool process.
    gen_server:start({local, Name}, ?MODULE, Name, []).

-spec stop_pool(kura_pool:name()) -> ok.
stop_pool(Name) ->
    case erlang:whereis(Name) of
        undefined -> ok;
        Pid -> gen_server:stop(Pid)
    end.

-spec checkout(kura_pool:name(), kura_pool:checkout_opts()) ->
    {ok, kura_pool:conn(), kura_pool:token()} | {error, term()}.
checkout(Name, _Opts) ->
    case erlang:whereis(Name) of
        undefined -> {error, no_pool};
        _Pid -> {ok, Name, Name}
    end.

-spec checkin(kura_pool:name(), kura_pool:token()) -> ok.
checkin(_Name, _Token) ->
    ok.

-spec give_away(kura_pool:token(), pid(), term()) -> ok | {error, term()}.
give_away(_Token, NewOwner, _GiftData) when is_pid(NewOwner) ->
    %% Nothing per-conn to transfer; the checkin-after-transfer
    %% guarantee holds trivially because checkin is a no-op.
    ok;
give_away(_Token, _NotPid, _GiftData) ->
    {error, badarg}.

%%----------------------------------------------------------------------
%% kura_capabilities
%%----------------------------------------------------------------------

-doc """
The ETS backend returns the affected row from insert/update/delete
(the moral equivalent of `RETURNING *`), and nothing else from the
standard SQL capability set. Notably absent: `transactions` — writes
are applied immediately with no rollback.
""".
-spec capabilities() -> kura_capabilities:capability_set().
capabilities() ->
    [returning].

%%----------------------------------------------------------------------
%% Table registry API (used by kura_ets_query)
%%----------------------------------------------------------------------

-doc "Return the data table for `TableName`, creating it on first use.".
-spec ensure_table(kura_pool:name(), binary()) -> {ok, ets:tid()} | {error, term()}.
ensure_table(Pool, TableName) when is_binary(TableName) ->
    case ets:whereis(Pool) of
        undefined ->
            {error, no_pool};
        _Meta ->
            case ets:lookup(Pool, TableName) of
                [{_, Tid}] -> {ok, Tid};
                [] -> gen_server:call(Pool, {ensure_table, TableName})
            end
    end.

-doc "Delete all rows from every table in the pool. Handy between tests.".
-spec reset(kura_pool:name()) -> ok | {error, term()}.
reset(Pool) ->
    case erlang:whereis(Pool) of
        undefined -> {error, no_pool};
        _Pid -> gen_server:call(Pool, reset)
    end.

%%----------------------------------------------------------------------
%% gen_server callbacks
%%----------------------------------------------------------------------

-spec init(atom()) -> {ok, atom()}.
init(Name) ->
    _Meta = ets:new(Name, [named_table, protected, set]),
    {ok, Name}.

-spec handle_call(term(), gen_server:from(), atom()) -> {reply, term(), atom()}.
handle_call({ensure_table, TableName}, _From, Name) ->
    case ets:lookup(Name, TableName) of
        [{_, Tid}] ->
            {reply, {ok, Tid}, Name};
        [] ->
            Tid = ets:new(kura_ets_table, [set, public]),
            true = ets:insert(Name, {TableName, Tid}),
            {reply, {ok, Tid}, Name}
    end;
handle_call(reset, _From, Name) ->
    lists:foreach(
        fun({_TableName, Tid}) -> true = ets:delete_all_objects(Tid) end,
        ets:tab2list(Name)
    ),
    {reply, ok, Name};
handle_call(Request, _From, Name) ->
    {reply, {error, {unknown_call, Request}}, Name}.

-spec handle_cast(term(), atom()) -> {noreply, atom()}.
handle_cast(_Msg, Name) ->
    {noreply, Name}.