glats

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

-export([publish/4, publish_message/2, request/5, respond/4, subscribe/4, unsubscribe/2, server_info/1, active_subscriptions/1, new_inbox/0, connect/3]).
-export_type([message/0, publish_option/0, subscribe_option/0, server_info/0, connection_option/0, error/0, connection_message/0, subscription_message/0, state/0]).

-type message() :: {message,
        binary(),
        gleam@dict:dict(binary(), binary()),
        gleam@option:option(binary()),
        binary()}.

-type publish_option() :: {reply_to, binary()} |
    {headers, list({binary(), binary()})}.

-type subscribe_option() :: {queue_group, binary()}.

-type server_info() :: {server_info,
        binary(),
        binary(),
        binary(),
        binary(),
        binary(),
        integer(),
        boolean(),
        integer(),
        integer(),
        boolean(),
        gleam@option:option(boolean())}.

-type connection_option() :: {user_pass, binary(), binary()} |
    {token, binary()} |
    {n_key_seed, binary()} |
    {jwt, binary()} |
    {ca_cert, binary()} |
    {client_cert, binary(), binary()} |
    {inbox_prefix, binary()} |
    {connection_timeout, integer()} |
    enable_no_responders.

-type error() :: no_reply_topic | timeout | no_responders | unexpected.

-opaque connection_message() :: {subscribe,
        gleam@erlang@process:subject({ok, integer()} | {error, error()}),
        gleam@erlang@process:pid_(),
        binary(),
        list(subscribe_option())} |
    {unsubscribe,
        gleam@erlang@process:subject({ok, nil} | {error, error()}),
        integer()} |
    {publish,
        gleam@erlang@process:subject({ok, nil} | {error, error()}),
        binary(),
        binary(),
        list(publish_option())} |
    {request,
        gleam@erlang@process:subject(fun(() -> {ok, message()} |
            {error, error()})),
        binary(),
        binary(),
        list(publish_option()),
        integer()} |
    {get_server_info,
        gleam@erlang@process:subject({ok, server_info()} | {error, error()})} |
    {get_active_subscriptions,
        gleam@erlang@process:subject({ok, integer()} | {error, error()})} |
    {exited, gleam@erlang@process:exit_message()}.

-type subscription_message() :: {received_message,
        gleam@erlang@process:subject(connection_message()),
        integer(),
        gleam@option:option(integer()),
        message()}.

-type state() :: {state,
        gleam@erlang@process:pid_(),
        gleam@dict:dict(gleam@erlang@process:pid_(), integer())}.

-file("/home/arnar/Code/glats/src/glats.gleam", 276).
-spec handle_server_info(
    gleam@erlang@process:subject({ok, server_info()} | {error, error()}),
    state()
) -> gleam@otp@actor:next(any(), state()).
handle_server_info(From, State) ->
    _pipe = 'Elixir.Gnat':server_info(erlang:element(2, State)),
    _pipe@1 = 'Elixir.Glats':decode_server_info(_pipe),
    _pipe@2 = gleam@result:map_error(_pipe@1, fun(_) -> unexpected end),
    gleam@erlang@process:send(From, _pipe@2),
    {continue, State, none}.

-file("/home/arnar/Code/glats/src/glats.gleam", 287).
-spec handle_active_subscriptions(
    gleam@erlang@process:subject({ok, integer()} | {error, error()}),
    state()
) -> gleam@otp@actor:next(any(), state()).
handle_active_subscriptions(From, State) ->
    _pipe = 'Elixir.Gnat':active_subscriptions(erlang:element(2, State)),
    _pipe@1 = gleam@result:map_error(_pipe, fun(_) -> unexpected end),
    gleam@erlang@process:send(From, _pipe@1),
    {continue, State, none}.

-file("/home/arnar/Code/glats/src/glats.gleam", 297).
-spec handle_publish(
    gleam@erlang@process:subject({ok, nil} | {error, error()}),
    binary(),
    binary(),
    list(publish_option()),
    state()
) -> gleam@otp@actor:next(any(), state()).
handle_publish(From, Topic, Body, Opts, State) ->
    case begin
        _pipe = 'Elixir.Gnat':pub(erlang:element(2, State), Topic, Body, Opts),
        erlang:atom_to_binary(_pipe)
    end of
        <<"ok"/utf8>> ->
            gleam@erlang@process:send(From, {ok, nil});

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

-file("/home/arnar/Code/glats/src/glats.gleam", 351).
-spec handle_unsubscribe(
    gleam@erlang@process:subject({ok, nil} | {error, error()}),
    integer(),
    state()
) -> gleam@otp@actor:next(any(), 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, unexpected})
    end,
    {continue, State, none}.

-file("/home/arnar/Code/glats/src/glats.gleam", 364).
-spec handle_subscribe(
    gleam@erlang@process:subject({ok, integer()} | {error, error()}),
    gleam@erlang@process:pid_(),
    binary(),
    list(subscribe_option()),
    state()
) -> gleam@otp@actor:next(any(), state()).
handle_subscribe(From, Subscriber, Topic, Opts, State) ->
    case 'Elixir.Gnat':sub(erlang:element(2, State), Subscriber, Topic, Opts) of
        {ok, Sid} ->
            case gleam_erlang_ffi:link(Subscriber) of
                true ->
                    gleam@erlang@process:send(From, {ok, Sid}),
                    {continue,
                        erlang:setelement(
                            3,
                            State,
                            gleam@dict:insert(
                                erlang:element(3, State),
                                Subscriber,
                                Sid
                            )
                        ),
                        none};

                false ->
                    gleam@erlang@process:send(From, {error, unexpected}),
                    {continue, State, none}
            end;

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

-file("/home/arnar/Code/glats/src/glats.gleam", 404).
-spec take_reply_topic(list(publish_option()), message()) -> list(publish_option()).
take_reply_topic(Prev, Message) ->
    case erlang:element(4, Message) of
        {some, Rt} ->
            gleam@list:prepend(Prev, {reply_to, Rt});

        none ->
            Prev
    end.

-file("/home/arnar/Code/glats/src/glats.gleam", 413).
-spec publish(
    gleam@erlang@process:subject(connection_message()),
    binary(),
    binary(),
    list(publish_option())
) -> {ok, nil} | {error, error()}.
publish(Conn, Topic, Body, Opts) ->
    gleam@erlang@process:call(
        Conn,
        fun(_capture) -> {publish, _capture, Topic, Body, Opts} end,
        5000
    ).

-file("/home/arnar/Code/glats/src/glats.gleam", 425).
-spec publish_message(
    gleam@erlang@process:subject(connection_message()),
    message()
) -> {ok, nil} | {error, error()}.
publish_message(Conn, Message) ->
    _pipe = [{headers, maps:to_list(erlang:element(3, Message))}],
    _pipe@1 = take_reply_topic(_pipe, Message),
    publish(
        Conn,
        erlang:element(2, Message),
        erlang:element(5, Message),
        _pipe@1
    ).

-file("/home/arnar/Code/glats/src/glats.gleam", 440).
-spec request(
    gleam@erlang@process:subject(connection_message()),
    binary(),
    binary(),
    list(publish_option()),
    integer()
) -> {ok, message()} | {error, error()}.
request(Conn, Topic, Body, Opts, Timeout) ->
    Make_request = gleam@erlang@process:call(
        Conn,
        fun(_capture) -> {request, _capture, Topic, Body, Opts, Timeout} end,
        1000
    ),
    Make_request().

-file("/home/arnar/Code/glats/src/glats.gleam", 460).
-spec respond(
    gleam@erlang@process:subject(connection_message()),
    message(),
    binary(),
    list(publish_option())
) -> {ok, nil} | {error, error()}.
respond(Conn, Message, Body, Opts) ->
    case erlang:element(4, Message) of
        {some, Rt} ->
            publish(Conn, Rt, Body, Opts);

        none ->
            {error, no_reply_topic}
    end.

-file("/home/arnar/Code/glats/src/glats.gleam", 476).
-spec subscription_mapper(
    gleam@erlang@process:subject(connection_message()),
    glats@internal@subscription:raw_message()
) -> subscription_message().
subscription_mapper(Conn, Raw_msg) ->
    {received_message,
        Conn,
        erlang:element(2, Raw_msg),
        erlang:element(3, Raw_msg),
        {message,
            erlang:element(4, Raw_msg),
            erlang:element(5, Raw_msg),
            erlang:element(6, Raw_msg),
            erlang:element(7, Raw_msg)}}.

-file("/home/arnar/Code/glats/src/glats.gleam", 506).
-spec subscribe(
    gleam@erlang@process:subject(connection_message()),
    gleam@erlang@process:subject(subscription_message()),
    binary(),
    list(subscribe_option())
) -> {ok, integer()} | {error, error()}.
subscribe(Conn, Subscriber, Topic, Opts) ->
    case glats@internal@subscription:start_subscriber(
        Conn,
        Subscriber,
        fun subscription_mapper/2
    ) of
        {ok, Sub} ->
            case gleam@erlang@process:call(
                Conn,
                fun(_capture) ->
                    {subscribe,
                        _capture,
                        begin
                            _pipe = Sub,
                            gleam@erlang@process:subject_owner(_pipe)
                        end,
                        Topic,
                        Opts}
                end,
                5000
            ) of
                {ok, Sid} ->
                    gleam@erlang@process:unlink(
                        begin
                            _pipe@1 = Sub,
                            gleam@erlang@process:subject_owner(_pipe@1)
                        end
                    ),
                    {ok, Sid};

                {error, Err} ->
                    gleam@erlang@process:kill(
                        begin
                            _pipe@2 = Sub,
                            gleam@erlang@process:subject_owner(_pipe@2)
                        end
                    ),
                    {error, Err}
            end;

        {error, _} ->
            {error, unexpected}
    end.

-file("/home/arnar/Code/glats/src/glats.gleam", 553).
-spec unsubscribe(gleam@erlang@process:subject(connection_message()), integer()) -> {ok,
        nil} |
    {error, error()}.
unsubscribe(Conn, Sid) ->
    gleam@erlang@process:call(
        Conn,
        fun(_capture) -> {unsubscribe, _capture, Sid} end,
        5000
    ).

-file("/home/arnar/Code/glats/src/glats.gleam", 563).
-spec server_info(gleam@erlang@process:subject(connection_message())) -> {ok,
        server_info()} |
    {error, error()}.
server_info(Conn) ->
    gleam@erlang@process:call(
        Conn,
        fun(Field@0) -> {get_server_info, Field@0} end,
        5000
    ).

-file("/home/arnar/Code/glats/src/glats.gleam", 573).
-spec active_subscriptions(gleam@erlang@process:subject(connection_message())) -> {ok,
        integer()} |
    {error, error()}.
active_subscriptions(Conn) ->
    gleam@erlang@process:call(
        Conn,
        fun(Field@0) -> {get_active_subscriptions, Field@0} end,
        5000
    ).

-file("/home/arnar/Code/glats/src/glats.gleam", 583).
-spec new_inbox() -> binary().
new_inbox() ->
    glats@internal@util:random_inbox(<<"_INBOX."/utf8>>).

-file("/home/arnar/Code/glats/src/glats.gleam", 608).
-spec apply_conn_option(
    gleam@dict:dict(binary(), gleam@dynamic:dynamic_()),
    connection_option()
) -> gleam@dict:dict(binary(), gleam@dynamic:dynamic_()).
apply_conn_option(Prev, Opt) ->
    case Opt of
        {user_pass, User, Pass} ->
            _pipe = Prev,
            _pipe@1 = gleam@dict:insert(
                _pipe,
                <<"username"/utf8>>,
                gleam_stdlib:identity(User)
            ),
            gleam@dict:insert(
                _pipe@1,
                <<"password"/utf8>>,
                gleam_stdlib:identity(Pass)
            );

        {token, Token} ->
            _pipe@2 = Prev,
            gleam@dict:insert(
                _pipe@2,
                <<"token"/utf8>>,
                gleam_stdlib:identity(Token)
            );

        {n_key_seed, Seed} ->
            _pipe@3 = Prev,
            gleam@dict:insert(
                _pipe@3,
                <<"nkey_seed"/utf8>>,
                gleam_stdlib:identity(Seed)
            );

        {jwt, Jwt} ->
            _pipe@4 = Prev,
            gleam@dict:insert(
                _pipe@4,
                <<"jwt"/utf8>>,
                gleam_stdlib:identity(Jwt)
            );

        {ca_cert, Path} ->
            _pipe@5 = Prev,
            _pipe@6 = gleam@dict:insert(
                _pipe@5,
                <<"tls"/utf8>>,
                gleam_stdlib:identity(true)
            ),
            gleam@dict:insert(
                _pipe@6,
                <<"cacertfile"/utf8>>,
                gleam_stdlib:identity(Path)
            );

        {client_cert, Cert, Key} ->
            _pipe@7 = Prev,
            _pipe@8 = gleam@dict:insert(
                _pipe@7,
                <<"tls"/utf8>>,
                gleam_stdlib:identity(true)
            ),
            _pipe@9 = gleam@dict:insert(
                _pipe@8,
                <<"certfile"/utf8>>,
                gleam_stdlib:identity(Cert)
            ),
            gleam@dict:insert(
                _pipe@9,
                <<"keyfile"/utf8>>,
                gleam_stdlib:identity(Key)
            );

        {inbox_prefix, Prefix} ->
            gleam@dict:insert(
                Prev,
                <<"inbox_prefix"/utf8>>,
                gleam_stdlib:identity(Prefix)
            );

        {connection_timeout, Timeout} ->
            gleam@dict:insert(
                Prev,
                <<"connection_timeout"/utf8>>,
                gleam_stdlib:identity(Timeout)
            );

        enable_no_responders ->
            gleam@dict:insert(
                Prev,
                <<"no_responders"/utf8>>,
                gleam_stdlib:identity(true)
            )
    end.

-file("/home/arnar/Code/glats/src/glats.gleam", 643).
-spec add_ssl_opts(gleam@dict:dict(binary(), gleam@dynamic:dynamic_())) -> gleam@dict:dict(binary(), gleam@dynamic:dynamic_()).
add_ssl_opts(Prev) ->
    _pipe = gleam@dict:take(
        Prev,
        [<<"cacertfile"/utf8>>, <<"certfile"/utf8>>, <<"keyfile"/utf8>>]
    ),
    _pipe@1 = maps:to_list(_pipe),
    _pipe@2 = gleam@list:map(
        _pipe@1,
        fun(O) ->
            {erlang:binary_to_atom(erlang:element(1, O)), erlang:element(2, O)}
        end
    ),
    _pipe@3 = gleam_stdlib:identity(_pipe@2),
    gleam@dict:insert(Prev, <<"ssl_opts"/utf8>>, _pipe@3).

-file("/home/arnar/Code/glats/src/glats.gleam", 590).
-spec build_settings(binary(), integer(), list(connection_option())) -> gleam@dict:dict(gleam@erlang@atom:atom_(), gleam@dynamic:dynamic_()).
build_settings(Host, Port, Opts) ->
    _pipe = [{<<"host"/utf8>>, gleam_stdlib:identity(Host)},
        {<<"port"/utf8>>, gleam_stdlib:identity(Port)}],
    _pipe@1 = maps:from_list(_pipe),
    _pipe@2 = gleam@list:fold(Opts, _pipe@1, fun apply_conn_option/2),
    _pipe@3 = add_ssl_opts(_pipe@2),
    _pipe@4 = gleam@dict:take(
        _pipe@3,
        [<<"host"/utf8>>,
            <<"port"/utf8>>,
            <<"tls"/utf8>>,
            <<"ssl_opts"/utf8>>,
            <<"inbox_prefix"/utf8>>,
            <<"connection_timeout"/utf8>>,
            <<"no_responders"/utf8>>,
            <<"username"/utf8>>,
            <<"password"/utf8>>,
            <<"token"/utf8>>,
            <<"nkey_seed"/utf8>>,
            <<"jwt"/utf8>>]
    ),
    _pipe@5 = maps:to_list(_pipe@4),
    _pipe@6 = gleam@list:map(
        _pipe@5,
        fun(I) ->
            {erlang:binary_to_atom(erlang:element(1, I)), erlang:element(2, I)}
        end
    ),
    maps:from_list(_pipe@6).

-file("/home/arnar/Code/glats/src/glats.gleam", 686).
-spec atom_field(
    binary(),
    fun((gleam@dynamic:dynamic_()) -> {ok, IDY} |
        {error, list(gleam@dynamic:decode_error())})
) -> fun((gleam@dynamic:dynamic_()) -> {ok, IDY} |
    {error, list(gleam@dynamic:decode_error())}).
atom_field(Key, Value) ->
    gleam@dynamic:field(erlang:binary_to_atom(Key), Value).

-file("/home/arnar/Code/glats/src/glats.gleam", 668).
-spec headers(gleam@dynamic:dynamic_()) -> {ok,
        gleam@dict:dict(binary(), binary())} |
    {error, list(gleam@dynamic:decode_error())}.
headers(Data) ->
    _pipe = Data,
    _pipe@1 = (atom_field(
        <<"headers"/utf8>>,
        gleam@dynamic:list(
            gleam@dynamic:tuple2(
                fun gleam@dynamic:string/1,
                fun gleam@dynamic:string/1
            )
        )
    ))(_pipe),
    _pipe@2 = gleam@result:map(_pipe@1, fun maps:from_list/1),
    gleam@result:'or'(_pipe@2, {ok, maps:new()}).

-file("/home/arnar/Code/glats/src/glats.gleam", 680).
-spec reply_to(gleam@dynamic:dynamic_()) -> {ok, gleam@option:option(binary())} |
    {error, list(gleam@dynamic:decode_error())}.
reply_to(Data) ->
    _pipe = Data,
    _pipe@1 = (gleam@dynamic:optional(
        atom_field(<<"reply_to"/utf8>>, fun gleam@dynamic:string/1)
    ))(_pipe),
    gleam@result:'or'(_pipe@1, {ok, none}).

-file("/home/arnar/Code/glats/src/glats.gleam", 654).
-spec decode_msg(gleam@dynamic:dynamic_()) -> {ok, message()} |
    {error, list(gleam@dynamic:decode_error())}.
decode_msg(Data) ->
    _pipe = Data,
    (gleam@dynamic:decode4(
        fun(Field@0, Field@1, Field@2, Field@3) -> {message, Field@0, Field@1, Field@2, Field@3} end,
        atom_field(<<"topic"/utf8>>, fun gleam@dynamic:string/1),
        fun headers/1,
        fun reply_to/1,
        atom_field(<<"body"/utf8>>, fun gleam@dynamic:string/1)
    ))(_pipe).

-file("/home/arnar/Code/glats/src/glats.gleam", 316).
-spec handle_request(
    gleam@erlang@process:subject(fun(() -> {ok, message()} | {error, error()})),
    binary(),
    binary(),
    list(publish_option()),
    integer(),
    state()
) -> gleam@otp@actor:next(any(), state()).
handle_request(From, Topic, Body, _, Timeout, State) ->
    Opts = [{erlang:binary_to_atom(<<"receive_timeout"/utf8>>),
            gleam_stdlib:identity(Timeout)}],
    Req_func = fun() ->
        case 'Elixir.Gnat':request(erlang:element(2, State), Topic, Body, Opts) of
            {ok, Msg} ->
                _pipe = decode_msg(Msg),
                gleam@result:map_error(_pipe, fun(_) -> unexpected end);

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

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

                    _ ->
                        {error, unexpected}
                end
        end
    end,
    gleam@erlang@process:send(From, Req_func),
    {continue, State, none}.

-file("/home/arnar/Code/glats/src/glats.gleam", 236).
-spec handle_command(connection_message(), state()) -> gleam@otp@actor:next(any(), state()).
handle_command(Message, State) ->
    case Message of
        {exited, Em} ->
            case erlang:element(2, Em) =:= erlang:element(2, State) of
                true ->
                    {stop, erlang:element(3, Em)};

                false ->
                    case gleam_stdlib:map_get(
                        erlang:element(3, State),
                        erlang:element(2, Em)
                    ) of
                        {ok, Sid} ->
                            'Elixir.Gnat':unsub(
                                erlang:element(2, State),
                                Sid,
                                []
                            ),
                            {continue,
                                erlang:setelement(
                                    3,
                                    State,
                                    gleam@dict:delete(
                                        erlang:element(3, State),
                                        erlang:element(2, Em)
                                    )
                                ),
                                none};

                        {error, nil} ->
                            gleam_stdlib:println(
                                <<"exited process not found in map of subscribers"/utf8>>
                            ),
                            {continue, State, none}
                    end
            end;

        {publish, From, Topic, Body, Opts} ->
            handle_publish(From, Topic, Body, Opts, State);

        {request, From@1, Topic@1, Body@1, Opts@1, Timeout} ->
            handle_request(From@1, Topic@1, Body@1, Opts@1, Timeout, State);

        {subscribe, From@2, Subscriber, Topic@2, Opts@2} ->
            handle_subscribe(From@2, Subscriber, Topic@2, Opts@2, State);

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

        {get_server_info, From@4} ->
            handle_server_info(From@4, State);

        {get_active_subscriptions, From@5} ->
            handle_active_subscriptions(From@5, State)
    end.

-file("/home/arnar/Code/glats/src/glats.gleam", 210).
-spec connect(binary(), integer(), list(connection_option())) -> {ok,
        gleam@erlang@process:subject(connection_message())} |
    {error, gleam@otp@actor:start_error()}.
connect(Host, Port, Opts) ->
    gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                gleam_erlang_ffi:trap_exits(true),
                Selector = begin
                    _pipe = gleam_erlang_ffi:new_selector(),
                    gleam@erlang@process:selecting_trapped_exits(
                        _pipe,
                        fun(Field@0) -> {exited, Field@0} end
                    )
                end,
                case 'Elixir.Gnat':start_link(build_settings(Host, Port, Opts)) of
                    {ok, Pid} ->
                        {ready, {state, Pid, maps:new()}, Selector};

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