gmysql

-module(gmysql@pool).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).

-export([disconnect/1, restart_connection/2, restart_all_connections/1, new_connection/1, checkout_connection/2, checkin_connection/2, with_connection/3, connect/3]).
-export_type([pool/0, message/0, state/0, slot/0, with_connection_error/0]).

-opaque pool() :: {pool, gleam@erlang@process:subject(message())}.

-opaque message() :: {update_state, fun((state()) -> state())} |
    {initialize, gleam@erlang@process:subject(message())} |
    shutdown |
    {restart, gmysql:connection()} |
    restart_all |
    start_new |
    {checkout,
        gleam@erlang@process:subject({ok, gmysql:connection()} | {error, nil})} |
    {checkin, gmysql:connection()} |
    {tick, gleam@erlang@process:subject(message())}.

-type state() :: {state,
        list(slot()),
        gmysql:config(),
        gleam@otp@intensity_tracker:intensity_tracker()}.

-type slot() :: {slot, gleam@option:option(gmysql:connection()), boolean()}.

-type with_connection_error() :: could_not_checkout.

-spec disconnect(pool()) -> nil.
disconnect(Pool) ->
    gleam@otp@actor:send(erlang:element(2, Pool), shutdown).

-spec restart_connection(pool(), gmysql:connection()) -> nil.
restart_connection(Pool, Connection) ->
    gleam@otp@actor:send(erlang:element(2, Pool), {restart, Connection}).

-spec restart_all_connections(pool()) -> nil.
restart_all_connections(Pool) ->
    gleam@otp@actor:send(erlang:element(2, Pool), restart_all).

-spec new_connection(pool()) -> nil.
new_connection(Pool) ->
    gleam@otp@actor:send(erlang:element(2, Pool), start_new).

-spec checkout_connection(pool(), integer()) -> {ok, gmysql:connection()} |
    {error, nil}.
checkout_connection(Pool, Timeout) ->
    gleam@otp@actor:call(
        erlang:element(2, Pool),
        fun(Field@0) -> {checkout, Field@0} end,
        Timeout
    ).

-spec checkin_connection(pool(), gmysql:connection()) -> nil.
checkin_connection(Pool, Connection) ->
    gleam@otp@actor:send(erlang:element(2, Pool), {checkin, Connection}).

-spec with_connection_loop(
    pool(),
    integer(),
    integer(),
    fun((gmysql:connection()) -> HRO)
) -> {ok, HRO} | {error, with_connection_error()}.
with_connection_loop(Pool, Until, Timeout, Function) ->
    gleam@bool:guard(
        os:system_time(millisecond) > Until,
        {error, could_not_checkout},
        fun() -> case checkout_connection(Pool, Timeout div 3) of
                {ok, Connection} ->
                    Result = Function(Connection),
                    checkin_connection(Pool, Connection),
                    {ok, Result};

                {error, _} ->
                    gleam_erlang_ffi:sleep(Timeout div 6),
                    with_connection_loop(Pool, Until, Timeout, Function)
            end end
    ).

-spec with_connection(pool(), integer(), fun((gmysql:connection()) -> HRL)) -> {ok,
        HRL} |
    {error, with_connection_error()}.
with_connection(Pool, Checkout_timeout, Function) ->
    Until = os:system_time(millisecond) + Checkout_timeout,
    with_connection_loop(Pool, Until, Checkout_timeout, Function).

-spec handle_update_state(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_update_state(Message, State) ->
    {update_state, Transform} = case Message of
        {update_state, _} -> Message;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Assertion pattern match failed"/utf8>>,
                        value => _assert_fail,
                        module => <<"gmysql/pool"/utf8>>,
                        function => <<"handle_update_state"/utf8>>,
                        line => 141})
    end,
    New_state = Transform(State),
    {continue, New_state, none}.

-spec handle_crashed_connection(gleam@erlang@process:exit_message()) -> message().
handle_crashed_connection(Exit_message) ->
    {exit_message, Pid, Reason} = Exit_message,
    Connection = gmysql_ffi:from_pid(Pid),
    case Reason of
        normal ->
            {update_state,
                fun(State) ->
                    erlang:setelement(
                        2,
                        State,
                        gleam@list:filter(
                            erlang:element(2, State),
                            fun(Slot) ->
                                erlang:element(2, Slot) =:= {some, Connection}
                            end
                        )
                    )
                end};

        _ ->
            {update_state,
                fun(State@1) ->
                    erlang:setelement(
                        2,
                        State@1,
                        gleam@list:map(
                            erlang:element(2, State@1),
                            fun(Slot@1) ->
                                case erlang:element(2, Slot@1) =:= {some,
                                    Connection} of
                                    true ->
                                        erlang:setelement(2, Slot@1, none);

                                    false ->
                                        Slot@1
                                end
                            end
                        )
                    )
                end}
    end.

-spec handle_init(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_init(Message, State) ->
    {initialize, Self} = case Message of
        {initialize, _} -> Message;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Assertion pattern match failed"/utf8>>,
                        value => _assert_fail,
                        module => <<"gmysql/pool"/utf8>>,
                        function => <<"handle_init"/utf8>>,
                        line => 177})
    end,
    gleam_erlang_ffi:trap_exits(true),
    Selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        _pipe@1 = gleam@erlang@process:selecting(
            _pipe,
            Self,
            fun gleam@function:identity/1
        ),
        gleam@erlang@process:selecting_trapped_exits(
            _pipe@1,
            fun handle_crashed_connection/1
        )
    end,
    gleam@erlang@process:send(Self, {tick, Self}),
    {continue, State, {some, Selector}}.

-spec handle_shutdown(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_shutdown(_, State) ->
    gleam@list:each(
        erlang:element(2, State),
        fun(Slot) -> case erlang:element(2, Slot) of
                none ->
                    nil;

                {some, Conn} ->
                    gmysql_ffi:close(Conn)
            end end
    ),
    {stop, normal}.

-spec handle_restart(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_restart(Message, State) ->
    {restart, Conn} = case Message of
        {restart, _} -> Message;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Assertion pattern match failed"/utf8>>,
                        value => _assert_fail,
                        module => <<"gmysql/pool"/utf8>>,
                        function => <<"handle_restart"/utf8>>,
                        line => 202})
    end,
    Pid = gmysql_ffi:to_pid(Conn),
    gleam@erlang@process:kill(Pid),
    gleam@otp@actor:continue(State).

-spec handle_restart_all(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_restart_all(_, State) ->
    gleam@list:each(
        erlang:element(2, State),
        fun(Slot) -> case erlang:element(2, Slot) of
                none ->
                    nil;

                {some, Conn} ->
                    _pipe = gmysql_ffi:to_pid(Conn),
                    gleam@erlang@process:kill(_pipe)
            end end
    ),
    gleam@otp@actor:continue(State).

-spec handle_start_new(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_start_new(_, State) ->
    gleam@otp@actor:continue(
        erlang:setelement(
            2,
            State,
            [{slot, none, false} | erlang:element(2, State)]
        )
    ).

-spec handle_checkout(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_checkout(Message, State) ->
    {checkout, Client} = case Message of
        {checkout, _} -> Message;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Assertion pattern match failed"/utf8>>,
                        value => _assert_fail,
                        module => <<"gmysql/pool"/utf8>>,
                        function => <<"handle_checkout"/utf8>>,
                        line => 230})
    end,
    {Connection@1, Slots} = gleam@list:map_fold(
        erlang:element(2, State),
        none,
        fun(Connection, Slot) -> case {Connection, Slot} of
                {{some, _}, _} ->
                    {Connection, Slot};

                {none, {slot, {some, Conn}, false}} ->
                    {{some, Conn}, erlang:setelement(3, Slot, true)};

                {_, _} ->
                    {Connection, Slot}
            end end
    ),
    _pipe = Connection@1,
    _pipe@1 = gleam@option:to_result(_pipe, nil),
    gleam@erlang@process:send(Client, _pipe@1),
    gleam@otp@actor:continue(erlang:setelement(2, State, Slots)).

-spec handle_checkin(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_checkin(Message, State) ->
    {checkin, Connection} = case Message of
        {checkin, _} -> Message;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Assertion pattern match failed"/utf8>>,
                        value => _assert_fail,
                        module => <<"gmysql/pool"/utf8>>,
                        function => <<"handle_checkin"/utf8>>,
                        line => 250})
    end,
    gleam@otp@actor:continue(
        erlang:setelement(
            2,
            State,
            gleam@list:map(
                erlang:element(2, State),
                fun(Slot) ->
                    case erlang:element(2, Slot) =:= {some, Connection} of
                        true ->
                            erlang:setelement(3, Slot, false);

                        false ->
                            Slot
                    end
                end
            )
        )
    ).

-spec handle_tick(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_tick(Message, State) ->
    {tick, Self} = case Message of
        {tick, _} -> Message;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Assertion pattern match failed"/utf8>>,
                        value => _assert_fail,
                        module => <<"gmysql/pool"/utf8>>,
                        function => <<"handle_tick"/utf8>>,
                        line => 266})
    end,
    {Rate_limiter, Slots} = gleam@list:map_fold(
        erlang:element(2, State),
        erlang:element(4, State),
        fun(Rate_limit, Slot) ->
            case {gleam@otp@intensity_tracker:add_event(Rate_limit), Slot} of
                {{ok, Limiter}, {slot, none, _}} ->
                    case gmysql:connect(erlang:element(3, State)) of
                        {ok, Connection} ->
                            {Limiter, {slot, {some, Connection}, false}};

                        {error, _} ->
                            {Limiter, Slot}
                    end;

                {_, {slot, none, _}} ->
                    {Rate_limit, erlang:setelement(3, Slot, false)};

                {_, Slot@1} ->
                    {Rate_limit, Slot@1}
            end
        end
    ),
    gleam@erlang@process:send_after(Self, 250, {tick, Self}),
    gleam@otp@actor:continue(
        erlang:setelement(4, erlang:setelement(2, State, Slots), Rate_limiter)
    ).

-spec handle_message(message(), state()) -> gleam@otp@actor:next(message(), state()).
handle_message(Message, State) ->
    case Message of
        {update_state, _} ->
            handle_update_state(Message, State);

        {initialize, _} ->
            handle_init(Message, State);

        shutdown ->
            handle_shutdown(Message, State);

        {restart, _} ->
            handle_restart(Message, State);

        restart_all ->
            handle_restart_all(Message, State);

        start_new ->
            handle_start_new(Message, State);

        {checkout, _} ->
            handle_checkout(Message, State);

        {checkin, _} ->
            handle_checkin(Message, State);

        {tick, _} ->
            handle_tick(Message, State)
    end.

-spec connect(gmysql:config(), integer(), integer()) -> pool().
connect(Config, Count, Max_connections_per_second) ->
    Slots = begin
        _pipe = gleam@iterator:repeat({slot, none, false}),
        _pipe@1 = gleam@iterator:take(_pipe, Count),
        gleam@iterator:to_list(_pipe@1)
    end,
    _assert_subject = gleam@otp@actor:start(
        {state,
            Slots,
            Config,
            gleam@otp@intensity_tracker:new(Max_connections_per_second, 1000)},
        fun handle_message/2
    ),
    {ok, Actor} = case _assert_subject of
        {ok, _} -> _assert_subject;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Assertion pattern match failed"/utf8>>,
                        value => _assert_fail,
                        module => <<"gmysql/pool"/utf8>>,
                        function => <<"connect"/utf8>>,
                        line => 46})
    end,
    gleam@erlang@process:send(Actor, {initialize, Actor}),
    {pool, Actor}.