glats

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

-export([publish_message/2, publish/3, request/3, subscribe/3, unsubscribe/2, queue_subscribe/4, connect/1]).
-export_type([server_error/0, connection_message/0, subscription_actor_message/0, connection_state/0, subscription_actor_state/0]).

-type server_error() :: timeout | no_responders | unknown.

-opaque connection_message() :: {subscribe,
        gleam@erlang@process:subject({ok, integer()} | {error, binary()}),
        gleam@erlang@process:subject(glats@message:message()),
        binary(),
        gleam@option:option(binary())} |
    {unsubscribe,
        gleam@erlang@process:subject({ok, nil} | {error, binary()}),
        integer()} |
    {publish,
        gleam@erlang@process:subject({ok, nil} | {error, binary()}),
        glats@message:message()} |
    {request,
        gleam@erlang@process:subject({ok, glats@message:message()} |
            {error, server_error()}),
        binary(),
        binary()}.

-type subscription_actor_message() :: {get_sid,
        gleam@erlang@process:subject(integer())} |
    {received_message, glats@message:message()} |
    {decode_error, gleam@dynamic:dynamic()} |
    subscriber_exited.

-type connection_state() :: {connection_state,
        gleam@erlang@process:pid_(),
        gleam@erlang@process:subject(connection_message())}.

-type subscription_actor_state() :: {subscription_actor_state,
        gleam@erlang@process:subject(connection_message()),
        integer(),
        gleam@erlang@process:subject(glats@message:message())}.

-spec publish_message(
    gleam@erlang@process:subject(connection_message()),
    glats@message:message()
) -> {ok, nil} | {error, binary()}.
publish_message(Conn, Message) ->
    gleam@erlang@process:call(
        Conn,
        fun(_capture) -> {publish, _capture, Message} end,
        5000
    ).

-spec publish(
    gleam@erlang@process:subject(connection_message()),
    binary(),
    binary()
) -> {ok, nil} | {error, binary()}.
publish(Conn, Subject, Message) ->
    publish_message(Conn, {message, Subject, gleam@map:new(), none, Message}).

-spec request(
    gleam@erlang@process:subject(connection_message()),
    binary(),
    binary()
) -> {ok, glats@message:message()} | {error, server_error()}.
request(Conn, Subject, Message) ->
    gleam@erlang@process:call(
        Conn,
        fun(_capture) -> {request, _capture, Subject, Message} end,
        5000
    ).

-spec subscribe(
    gleam@erlang@process:subject(connection_message()),
    gleam@erlang@process:subject(glats@message:message()),
    binary()
) -> {ok, integer()} | {error, binary()}.
subscribe(Conn, Subscriber, Subject) ->
    gleam@erlang@process:call(
        Conn,
        fun(_capture) -> {subscribe, _capture, Subscriber, Subject, none} end,
        5000
    ).

-spec unsubscribe(gleam@erlang@process:subject(connection_message()), integer()) -> {ok,
        nil} |
    {error, binary()}.
unsubscribe(Conn, Sid) ->
    gleam@erlang@process:call(
        Conn,
        fun(_capture) -> {unsubscribe, _capture, Sid} end,
        5000
    ).

-spec subscription_loop(
    subscription_actor_message(),
    subscription_actor_state()
) -> gleam@otp@actor:next(subscription_actor_state()).
subscription_loop(Message, State) ->
    case Message of
        {get_sid, From} ->
            gleam@otp@actor:send(From, erlang:element(3, State)),
            {continue, State};

        {received_message, Msg} ->
            gleam@otp@actor:send(erlang:element(4, State), Msg),
            {continue, State};

        {decode_error, Data} ->
            gleam@io:debug(Data),
            {continue, State};

        subscriber_exited ->
            gleam@io:println(<<"subscriber exited"/utf8>>),
            unsubscribe(erlang:element(2, State), erlang:element(3, State)),
            {stop, normal}
    end.

-spec queue_subscribe(
    gleam@erlang@process:subject(connection_message()),
    gleam@erlang@process:subject(glats@message:message()),
    binary(),
    binary()
) -> {ok, integer()} | {error, binary()}.
queue_subscribe(Conn, Subscriber, Subject, Group) ->
    gleam@erlang@process:call(
        Conn,
        fun(_capture) ->
            {subscribe, _capture, Subscriber, Subject, {some, Group}}
        end,
        5000
    ).

-spec add_opt_to_list(
    list({binary(), gleam@dynamic:dynamic()}),
    binary(),
    any()
) -> list({binary(), gleam@dynamic:dynamic()}).
add_opt_to_list(Old, Key, Value) ->
    _pipe = Old,
    gleam@list:append(_pipe, [{Key, gleam@dynamic:from(Value)}]).

-spec take_host(
    list({binary(), gleam@dynamic:dynamic()}),
    glats@settings:settings()
) -> list({binary(), gleam@dynamic:dynamic()}).
take_host(Old, Settings) ->
    case erlang:element(2, Settings) of
        {some, Host} ->
            _pipe = Old,
            add_opt_to_list(_pipe, <<"host"/utf8>>, Host);

        none ->
            Old
    end.

-spec take_port(
    list({binary(), gleam@dynamic:dynamic()}),
    glats@settings:settings()
) -> list({binary(), gleam@dynamic:dynamic()}).
take_port(Old, Settings) ->
    case erlang:element(3, Settings) of
        {some, Port} ->
            _pipe = Old,
            add_opt_to_list(_pipe, <<"port"/utf8>>, Port);

        none ->
            Old
    end.

-spec take_tls(
    list({binary(), gleam@dynamic:dynamic()}),
    glats@settings:settings()
) -> list({binary(), gleam@dynamic:dynamic()}).
take_tls(Old, Settings) ->
    case erlang:element(4, Settings) of
        {some, Tls} ->
            _pipe = Old,
            add_opt_to_list(_pipe, <<"tls"/utf8>>, Tls);

        none ->
            Old
    end.

-spec take_ssl_opts(
    list({binary(), gleam@dynamic:dynamic()}),
    glats@settings:settings()
) -> list({binary(), gleam@dynamic:dynamic()}).
take_ssl_opts(Old, Settings) ->
    case erlang:element(5, Settings) of
        {some, Ssl_opts} ->
            _pipe = Old,
            add_opt_to_list(_pipe, <<"ssl_opts"/utf8>>, Ssl_opts);

        none ->
            Old
    end.

-spec build_settings(glats@settings:settings()) -> gleam@map:map_(binary(), gleam@dynamic:dynamic()).
build_settings(Settings) ->
    _pipe = [],
    _pipe@1 = take_host(_pipe, Settings),
    _pipe@2 = take_port(_pipe@1, Settings),
    _pipe@3 = take_tls(_pipe@2, Settings),
    _pipe@4 = take_ssl_opts(_pipe@3, Settings),
    gleam@map:from_list(_pipe@4).

-spec handle_publish(
    gleam@erlang@process:subject({ok, nil} | {error, binary()}),
    glats@message:message(),
    connection_state()
) -> gleam@otp@actor:next(connection_state()).
handle_publish(From, Message, State) ->
    case begin
        _pipe = 'Elixir.Gnat':pub(
            erlang:element(2, State),
            erlang:element(2, Message),
            erlang:element(5, Message),
            []
        ),
        erlang:atom_to_binary(_pipe)
    end of
        <<"ok"/utf8>> ->
            gleam@erlang@process:send(From, {ok, nil});

        _ ->
            gleam@erlang@process:send(
                From,
                {error, <<"unknown publish error"/utf8>>}
            )
    end,
    {continue, State}.

-spec handle_request(
    gleam@erlang@process:subject({ok, glats@message:message()} |
        {error, server_error()}),
    binary(),
    binary(),
    connection_state()
) -> gleam@otp@actor:next(connection_state()).
handle_request(From, Subject, Message, State) ->
    case 'Elixir.Gnat':request(erlang:element(2, State), Subject, Message, []) of
        {ok, Msg} ->
            _pipe = glats@internal@decoder:decode_msg(Msg),
            _pipe@1 = gleam@result:map_error(_pipe, fun(_) -> unknown end),
            gleam@erlang@process:send(From, _pipe@1);

        {error, Err} ->
            _pipe@2 = case erlang:atom_to_binary(Err) of
                <<"timeout"/utf8>> ->
                    {error, timeout};

                <<"no_responders"/utf8>> ->
                    {error, no_responders};

                _ ->
                    {error, unknown}
            end,
            gleam@erlang@process:send(From, _pipe@2)
    end,
    {continue, State}.

-spec start_subscription_actor(
    gleam@erlang@process:subject(glats@message:message()),
    binary(),
    gleam@option:option(binary()),
    connection_state()
) -> {ok, gleam@erlang@process:subject(subscription_actor_message())} |
    {error, gleam@otp@actor:start_error()}.
start_subscription_actor(Subscriber, Subject, Queue_group, State) ->
    gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                Monitor = gleam@erlang@process:monitor_process(
                    begin
                        _pipe = Subscriber,
                        gleam@erlang@process:subject_owner(_pipe)
                    end
                ),
                Selector = begin
                    _pipe@1 = gleam_erlang_ffi:new_selector(),
                    _pipe@2 = gleam@erlang@process:selecting_process_down(
                        _pipe@1,
                        Monitor,
                        fun(_) -> subscriber_exited end
                    ),
                    gleam@erlang@process:selecting_record2(
                        _pipe@2,
                        erlang:binary_to_atom(<<"msg"/utf8>>),
                        fun(Data) ->
                            _pipe@3 = Data,
                            _pipe@4 = glats@internal@decoder:decode_msg(_pipe@3),
                            _pipe@5 = gleam@result:map(
                                _pipe@4,
                                fun(Field@0) -> {received_message, Field@0} end
                            ),
                            gleam@result:unwrap(_pipe@5, {decode_error, Data})
                        end
                    )
                end,
                Opts = case Queue_group of
                    {some, Group} ->
                        [{erlang:binary_to_atom(<<"queue_group"/utf8>>), Group}];

                    none ->
                        []
                end,
                case 'Elixir.Gnat':sub(
                    erlang:element(2, State),
                    erlang:self(),
                    Subject,
                    Opts
                ) of
                    {ok, Sid} ->
                        {ready,
                            {subscription_actor_state,
                                erlang:element(3, State),
                                Sid,
                                Subscriber},
                            Selector};

                    {error, Err} ->
                        {failed, Err}
                end
            end,
            5000,
            fun subscription_loop/2}
    ).

-spec handle_subscribe(
    gleam@erlang@process:subject({ok, integer()} | {error, binary()}),
    gleam@erlang@process:subject(glats@message:message()),
    binary(),
    gleam@option:option(binary()),
    connection_state()
) -> gleam@otp@actor:next(connection_state()).
handle_subscribe(From, Subscriber, Subject, Queue_group, State) ->
    case start_subscription_actor(Subscriber, Subject, Queue_group, State) of
        {ok, Actor} ->
            case gleam@erlang@process:try_call(
                Actor,
                fun(Field@0) -> {get_sid, Field@0} end,
                1000
            ) of
                {ok, Sid} ->
                    gleam@erlang@process:send(From, {ok, Sid});

                {error, _} ->
                    gleam@erlang@process:send(
                        From,
                        {error, <<"subscribe failed"/utf8>>}
                    )
            end;

        {error, _} ->
            gleam@erlang@process:send(
                From,
                {error, <<"subscribe failed"/utf8>>}
            )
    end,
    {continue, State}.

-spec handle_unsubscribe(
    gleam@erlang@process:subject({ok, nil} | {error, binary()}),
    integer(),
    connection_state()
) -> gleam@otp@actor:next(connection_state()).
handle_unsubscribe(From, Sid, State) ->
    case begin
        _pipe = 'Elixir.Gnat':unsub(erlang:element(2, State), Sid, []),
        erlang:atom_to_binary(_pipe)
    end of
        <<"ok"/utf8>> ->
            gleam@erlang@process:send(From, {ok, nil});

        _ ->
            gleam@erlang@process:send(
                From,
                {error, <<"unknown unsubscribe error"/utf8>>}
            )
    end,
    {continue, State}.

-spec handle_command(connection_message(), connection_state()) -> gleam@otp@actor:next(connection_state()).
handle_command(Message, State) ->
    case Message of
        {publish, From, Msg} ->
            handle_publish(From, Msg, State);

        {request, From@1, Subject, Msg@1} ->
            handle_request(From@1, Subject, Msg@1, State);

        {subscribe, From@2, Subscriber, Subject@1, Queue_group} ->
            handle_subscribe(From@2, Subscriber, Subject@1, Queue_group, State);

        {unsubscribe, From@3, Sid} ->
            handle_unsubscribe(From@3, Sid, State);

        _ ->
            {continue, State}
    end.

-spec connect(glats@settings:settings()) -> {ok,
        gleam@erlang@process:subject(connection_message())} |
    {error, gleam@otp@actor:start_error()}.
connect(Settings) ->
    gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                Subject = gleam@erlang@process:new_subject(),
                Selector = gleam_erlang_ffi:new_selector(),
                case 'Elixir.Gnat':start_link(
                    begin
                        _pipe = Settings,
                        build_settings(_pipe)
                    end
                ) of
                    {ok, Pid} ->
                        {ready, {connection_state, Pid, Subject}, Selector};

                    {error, _} ->
                        {failed, <<"starting connection failed"/utf8>>}
                end
            end,
            5000,
            fun handle_command/2}
    ).