glats

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

-export([handle_subscription/3, handle_request/3]).
-export_type([request/0, response/0, subscription_state/0, request_state/0]).

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

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

-type subscription_state() :: {subscription_state,
        gleam@erlang@process:subject(glats@connection:command()),
        fun((glats@message:message(), gleam@erlang@process:subject(glats@connection:command())) -> {ok,
                nil} |
            {error, binary()})}.

-type request_state() :: {request_state,
        gleam@erlang@process:subject(glats@connection:command()),
        fun((request(), gleam@erlang@process:subject(glats@connection:command())) -> {ok,
                response()} |
            {error, binary()})}.

-spec request_handler_loop(glats@message:message(), request_state()) -> gleam@otp@actor:next(request_state()).
request_handler_loop(Msg, State) ->
    case erlang:element(4, Msg) of
        {some, Reply_to} ->
            case (erlang:element(3, State))(
                {request, erlang:element(3, Msg), erlang:element(5, Msg)},
                erlang:element(2, State)
            ) of
                {ok, Response} ->
                    case glats@connection:publish_message(
                        erlang:element(2, State),
                        {message,
                            Reply_to,
                            erlang:element(2, Response),
                            none,
                            erlang:element(3, Response)}
                    ) of
                        {ok, nil} ->
                            {continue, State};

                        {error, Err} ->
                            {stop, {abnormal, Err}}
                    end;

                {error, Err@1} ->
                    {stop, {abnormal, Err@1}}
            end;

        none ->
            {continue, State}
    end.

-spec map_message(gleam@dynamic:dynamic()) -> glats@message:message().
map_message(Msg) ->
    _pipe = Msg,
    _pipe@1 = 'Elixir.Glats':convert_msg(_pipe),
    gleam@result:unwrap(
        _pipe@1,
        {message, <<"error"/utf8>>, gleam@map:new(), none, <<"body"/utf8>>}
    ).

-spec handle_subscription(
    gleam@erlang@process:subject(glats@connection:command()),
    binary(),
    fun((glats@message:message(), gleam@erlang@process:subject(glats@connection:command())) -> {ok,
            nil} |
        {error, binary()})
) -> {ok, gleam@erlang@process:subject(glats@message:message())} |
    {error, gleam@otp@actor:start_error()}.
handle_subscription(Conn, Subject, Handler) ->
    gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                Receiver = gleam@erlang@process:new_subject(),
                Selector = begin
                    _pipe = gleam_erlang_ffi:new_selector(),
                    gleam@erlang@process:selecting_anything(
                        _pipe,
                        fun map_message/1
                    )
                end,
                case glats@connection:subscribe(Conn, Receiver, Subject) of
                    {ok, _} ->
                        {ready, {subscription_state, Conn, Handler}, Selector};

                    {error, Err} ->
                        {failed, Err}
                end
            end,
            10000,
            fun(Msg, State) ->
                case (erlang:element(3, State))(Msg, erlang:element(2, State)) of
                    {ok, _} ->
                        {continue, State};

                    {error, _} ->
                        {stop, {abnormal, <<"handler returned error!"/utf8>>}}
                end
            end}
    ).

-spec handle_request(
    gleam@erlang@process:subject(glats@connection:command()),
    binary(),
    fun((request(), gleam@erlang@process:subject(glats@connection:command())) -> {ok,
            response()} |
        {error, binary()})
) -> {ok, gleam@erlang@process:subject(glats@message:message())} |
    {error, gleam@otp@actor:start_error()}.
handle_request(Conn, Subject, Handler) ->
    gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                Receiver = gleam@erlang@process:new_subject(),
                Selector = begin
                    _pipe = gleam_erlang_ffi:new_selector(),
                    gleam@erlang@process:selecting_anything(
                        _pipe,
                        fun map_message/1
                    )
                end,
                case glats@connection:subscribe(Conn, Receiver, Subject) of
                    {ok, _} ->
                        {ready, {request_state, Conn, Handler}, Selector};

                    {error, Err} ->
                        {failed, Err}
                end
            end,
            10000,
            fun request_handler_loop/2}
    ).