glats

-module(glats@handler).
-compile([no_auto_import, nowarn_unused_vars]).

-export([handle_request/5]).
-export_type([request/0, response/0, request_outcome/1, request_handler_state/1]).

-type request() :: {request, gleam@map:map_(binary(), binary()), binary()}.

-type response() :: {response,
        gleam@map:map_(binary(), binary()),
        gleam@option:option(binary()),
        binary()}.

-type request_outcome(HCS) :: {reply, response(), HCS} |
    {stop, gleam@erlang@process:exit_reason()}.

-type request_handler_state(HCT) :: {request_handler_state,
        gleam@erlang@process:subject(glats:connection_message()),
        integer(),
        fun((request(), HCT) -> request_outcome(HCT)),
        HCT}.

-spec request_handler_msg(
    gleam@erlang@process:subject(glats:connection_message()),
    glats:message(),
    request_handler_state(HDD)
) -> gleam@otp@actor:next(request_handler_state(HDD)).
request_handler_msg(Conn, Msg, State) ->
    case erlang:element(4, Msg) of
        {some, Reply_to} ->
            Req = {request, erlang:element(3, Msg), erlang:element(5, Msg)},
            case (erlang:element(4, State))(Req, erlang:element(5, State)) of
                {reply, Res, New_inner} ->
                    Pub_res = glats:publish_message(
                        Conn,
                        {message,
                            Reply_to,
                            erlang:element(2, Res),
                            erlang:element(3, Res),
                            erlang:element(4, Res)}
                    ),
                    case Pub_res of
                        {ok, nil} ->
                            {continue, erlang:setelement(5, State, New_inner)};

                        {error, Err} ->
                            {stop, {abnormal, gleam@string:inspect(Err)}}
                    end;

                {stop, Reason} ->
                    {stop, Reason}
            end;

        none ->
            {continue, State}
    end.

-spec request_handler_loop(
    glats:subscription_message(),
    request_handler_state(HDA)
) -> gleam@otp@actor:next(request_handler_state(HDA)).
request_handler_loop(Message, State) ->
    case Message of
        {received_message, Conn, _, _, Msg} ->
            request_handler_msg(Conn, Msg, State);

        _ ->
            {continue, State}
    end.

-spec handle_request(
    gleam@erlang@process:subject(glats:connection_message()),
    HCW,
    binary(),
    list(glats:subscribe_option()),
    fun((request(), HCW) -> request_outcome(HCW))
) -> {ok, gleam@erlang@process:subject(glats:subscription_message())} |
    {error, gleam@otp@actor:start_error()}.
handle_request(Conn, State, Topic, Opts, Handler) ->
    gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                Subscriber = gleam@erlang@process:new_subject(),
                Selector = begin
                    _pipe = gleam_erlang_ffi:new_selector(),
                    gleam@erlang@process:selecting(
                        _pipe,
                        Subscriber,
                        fun(Msg) -> Msg end
                    )
                end,
                case glats:subscribe(Conn, Subscriber, Topic, Opts) of
                    {ok, Sid} ->
                        {ready,
                            {request_handler_state, Conn, Sid, Handler, State},
                            Selector};

                    {error, Err} ->
                        {failed, gleam@string:inspect(Err)}
                end
            end,
            5000,
            fun request_handler_loop/2}
    ).