distribute

-module(distribute@connection_pool).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/connection_pool.gleam").
-export([destroy/1, new/2, with_connection/2, send_batch/3, send_batch_parallel/3, stats/1, stress_test/3]).
-export_type([pool/0, connection/0, pool_stats/0, pool_error/0, dynamic_/0, batch_result/1]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

-type pool() :: any().

-type connection() :: any().

-type pool_stats() :: {pool_stats, binary(), integer(), integer(), integer()}.

-type pool_error() :: pool_exhausted |
    {connection_failed, binary()} |
    {invalid_config, binary()} |
    invalid_pool.

-type dynamic_() :: any().

-type batch_result(HSD) :: {batch_result,
        integer(),
        integer(),
        integer(),
        list(HSD)}.

-file("src/distribute/connection_pool.gleam", 132).
?DOC(" Destroy the pool and release all resources\n").
-spec destroy(pool()) -> {ok, nil} | {error, pool_error()}.
destroy(Pool) ->
    Result = connection_pool_ffi:destroy_pool(Pool),
    case connection_pool_ffi:is_ok(Result) of
        true ->
            {ok, nil};

        false ->
            {error, invalid_pool}
    end.

-file("src/distribute/connection_pool.gleam", 228).
-spec unwrap_pool(dynamic_()) -> pool().
unwrap_pool(Result) ->
    erlang:element(2, Result).

-file("src/distribute/connection_pool.gleam", 77).
?DOC(" Create a new connection pool backed by ETS for atomic operations\n").
-spec new(binary(), integer()) -> {ok, pool()} | {error, pool_error()}.
new(Target_node, Max_connections) ->
    case Max_connections > 0 of
        false ->
            {error, {invalid_config, <<"max_connections must be > 0"/utf8>>}};

        true ->
            Result = connection_pool_ffi:new_pool(Target_node, Max_connections),
            case connection_pool_ffi:is_ok(Result) of
                true ->
                    {ok, unwrap_pool(Result)};

                false ->
                    {error,
                        {connection_failed,
                            connection_pool_ffi:get_error(Result)}}
            end
    end.

-file("src/distribute/connection_pool.gleam", 232).
-spec unwrap_connection(dynamic_()) -> connection().
unwrap_connection(Result) ->
    erlang:element(2, Result).

-file("src/distribute/connection_pool.gleam", 91).
?DOC(" Execute a function with a pooled connection (automatically released)\n").
-spec with_connection(pool(), fun((connection()) -> HSH)) -> {ok, HSH} |
    {error, pool_error()}.
with_connection(Pool, F) ->
    Conn_result = connection_pool_ffi:get_connection(Pool),
    case connection_pool_ffi:is_ok(Conn_result) of
        true ->
            Connection = unwrap_connection(Conn_result),
            Result = F(Connection),
            _ = connection_pool_ffi:release_connection(Connection),
            {ok, Result};

        false ->
            Error_msg = connection_pool_ffi:get_error(Conn_result),
            case Error_msg of
                <<"pool_exhausted"/utf8>> ->
                    {error, pool_exhausted};

                _ ->
                    {error, {connection_failed, Error_msg}}
            end
    end.

-file("src/distribute/connection_pool.gleam", 145).
?DOC(" Send multiple messages in a batch using a single connection\n").
-spec send_batch(
    pool(),
    list({binary(), HSO}),
    fun((binary(), HSO) -> {ok, nil} | {error, HSQ})
) -> {ok, list({ok, nil} | {error, HSQ})} | {error, pool_error()}.
send_batch(Pool, Messages, Send_fn) ->
    with_connection(
        Pool,
        fun(_) ->
            gleam@list:map(
                Messages,
                fun(Message) ->
                    {Name, Msg} = Message,
                    Send_fn(Name, Msg)
                end
            )
        end
    ).

-file("src/distribute/connection_pool.gleam", 160).
?DOC(" Send multiple messages in parallel (up to pool size) with error aggregation\n").
-spec send_batch_parallel(
    pool(),
    list({binary(), HSY}),
    fun((binary(), HSY) -> {ok, nil} | {error, HTA})
) -> {ok, batch_result(HTA)} | {error, pool_error()}.
send_batch_parallel(Pool, Messages, Send_fn) ->
    case send_batch(Pool, Messages, Send_fn) of
        {ok, Results} ->
            Successes = begin
                _pipe = gleam@list:filter(Results, fun(R) -> case R of
                            {ok, _} ->
                                true;

                            {error, _} ->
                                false
                        end end),
                erlang:length(_pipe)
            end,
            Failures = gleam@list:fold(Results, [], fun(Acc, R@1) -> case R@1 of
                        {error, E} ->
                            [E | Acc];

                        {ok, _} ->
                            Acc
                    end end),
            {ok,
                {batch_result,
                    erlang:length(Messages),
                    Successes,
                    erlang:length(Failures),
                    Failures}};

        {error, E@1} ->
            {error, E@1}
    end.

-file("src/distribute/connection_pool.gleam", 236).
-spec unwrap_stats_map(dynamic_()) -> dynamic_().
unwrap_stats_map(Result) ->
    erlang:element(2, Result).

-file("src/distribute/connection_pool.gleam", 243).
-spec get_map_string(dynamic_(), binary()) -> binary().
get_map_string(Map, Key) ->
    maps:get(erlang:binary_to_atom(Key), Map).

-file("src/distribute/connection_pool.gleam", 247).
-spec get_map_int(dynamic_(), binary()) -> integer().
get_map_int(Map, Key) ->
    maps:get(erlang:binary_to_atom(Key), Map).

-file("src/distribute/connection_pool.gleam", 115).
?DOC(" Get pool statistics (atomic read from ETS)\n").
-spec stats(pool()) -> {ok, pool_stats()} | {error, pool_error()}.
stats(Pool) ->
    Result = connection_pool_ffi:pool_stats(Pool),
    case connection_pool_ffi:is_ok(Result) of
        true ->
            Map = unwrap_stats_map(Result),
            {ok,
                {pool_stats,
                    get_map_string(Map, <<"target_node"/utf8>>),
                    get_map_int(Map, <<"max_connections"/utf8>>),
                    get_map_int(Map, <<"active_connections"/utf8>>),
                    get_map_int(Map, <<"available_connections"/utf8>>)}};

        false ->
            {error, invalid_pool}
    end.

-file("src/distribute/connection_pool.gleam", 199).
?DOC(
    " Run a concurrency stress-test against a pool. This spawns multiple worker processes\n"
    " each performing `ops` get/release cycles using the pool; returns final pool stats.\n"
).
-spec stress_test(pool(), integer(), integer()) -> {ok, pool_stats()} |
    {error, pool_error()}.
stress_test(Pool, Ops_per_worker, Workers) ->
    Res = connection_pool_ffi:stress_test_pool(Pool, Ops_per_worker, Workers),
    case connection_pool_ffi:is_ok(Res) of
        true ->
            Map = unwrap_stats_map(Res),
            {ok,
                {pool_stats,
                    get_map_string(Map, <<"target_node"/utf8>>),
                    get_map_int(Map, <<"max_connections"/utf8>>),
                    get_map_int(Map, <<"active_connections"/utf8>>),
                    get_map_int(Map, <<"available_connections"/utf8>>)}};

        false ->
            {error, invalid_pool}
    end.