-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]).

-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.

-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("src/glats.gleam", 277).
-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("src/glats.gleam", 288).
-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("src/glats.gleam", 298).
-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("src/glats.gleam", 352).
-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("src/glats.gleam", 365).
-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,
                        begin
                            _record = State,
                            {state,
                                erlang:element(2, _record),
                                gleam@dict:insert(
                                    erlang:element(3, State),
                                    Subscriber,
                                    Sid
                                )}
                        end,
                        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("src/glats.gleam", 405).
-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("src/glats.gleam", 414).
?DOC(" Publishes a single message to NATS on a provided topic.\n").
-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("src/glats.gleam", 426).
?DOC(
    " Publishes a single message to NATS using the data from a provided `Message`\n"
    " record.\n"
).
-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("src/glats.gleam", 441).
?DOC(
    " Sends a request and listens for a response synchronously.\n"
    " When connection is established with option `EnableNoResponders`,\n"
    " `Error(NoResponders)` will be returned immediately if no subscriber\n"
    " exists for the topic.\n"
    "\n"
    " See [request-reply pattern docs.](https://docs.nats.io/nats-concepts/core-nats/reqreply)\n"
    "\n"
    " To handle a request from NATS see `handler.handle_request`.\n"
).
-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("src/glats.gleam", 461).
?DOC(" Sends a respond to a Message's reply_to topic.\n").
-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("src/glats.gleam", 477).
-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("src/glats.gleam", 507).
?DOC(
    " Subscribes to a NATS topic that can be received on the\n"
    " provided OTP subject.\n"
    "\n"
    " To subscribe as a member of a queue group, add option\n"
    " `QueueGroup(String)` to list of options as the last\n"
    " parameter.\n"
    "\n"
    " ```gleam\n"
    " subscribe(conn, subject, \"my.topic\", [QueueGroup(\"my-group\")])\n"
    " ```\n"
    "\n"
    " See [Queue Groups docs.](https://docs.nats.io/nats-concepts/core-nats/queue)\n"
).
-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("src/glats.gleam", 554).
?DOC(" Unsubscribe from a subscription by providing the subscription ID.\n").
-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("src/glats.gleam", 564).
?DOC(" Returns server info provided by the connected NATS server.\n").
-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("src/glats.gleam", 574).
?DOC(" Returns the number of active subscriptions for the connection.\n").
-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("src/glats.gleam", 584).
?DOC(" Returns a new random inbox.\n").
-spec new_inbox() -> binary().
new_inbox() ->
    glats@internal@util:random_inbox(<<"_INBOX."/utf8>>).

-file("src/glats.gleam", 609).
-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("src/glats.gleam", 644).
-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) ->
            {gleam_stdlib:identity(erlang:binary_to_atom(erlang:element(1, O))),
                erlang:element(2, O)}
        end
    ),
    _pipe@3 = gleam@dynamic:properties(_pipe@2),
    gleam@dict:insert(Prev, <<"ssl_opts"/utf8>>, _pipe@3).

-file("src/glats.gleam", 591).
-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("src/glats.gleam", 671).
-spec decode_headers(
    fun((gleam@dict:dict(binary(), binary())) -> gleam@dynamic@decode:decoder(GUO))
) -> gleam@dynamic@decode:decoder(GUO).
decode_headers(Next) ->
    gleam@dynamic@decode:optional_field(
        erlang:binary_to_atom(<<"headers"/utf8>>),
        maps:new(),
        gleam@dynamic@decode:dict(
            {decoder, fun gleam@dynamic@decode:decode_string/1},
            {decoder, fun gleam@dynamic@decode:decode_string/1}
        ),
        Next
    ).

-file("src/glats.gleam", 682).
-spec decode_reply_to(
    fun((gleam@option:option(binary())) -> gleam@dynamic@decode:decoder(GUV))
) -> gleam@dynamic@decode:decoder(GUV).
decode_reply_to(Next) ->
    gleam@dynamic@decode:optional_field(
        erlang:binary_to_atom(<<"reply_to"/utf8>>),
        none,
        gleam@dynamic@decode:optional(
            {decoder, fun gleam@dynamic@decode:decode_string/1}
        ),
        Next
    ).

-file("src/glats.gleam", 655).
-spec decode_msg(gleam@dynamic:dynamic_()) -> {ok, message()} |
    {error, list(gleam@dynamic@decode:decode_error())}.
decode_msg(Data) ->
    Decoder = begin
        gleam@dynamic@decode:field(
            erlang:binary_to_atom(<<"topic"/utf8>>),
            {decoder, fun gleam@dynamic@decode:decode_string/1},
            fun(Topic) ->
                decode_headers(
                    fun(Headers) ->
                        decode_reply_to(
                            fun(Reply_to) ->
                                gleam@dynamic@decode:field(
                                    erlang:binary_to_atom(<<"body"/utf8>>),
                                    {decoder,
                                        fun gleam@dynamic@decode:decode_string/1},
                                    fun(Body) ->
                                        gleam@dynamic@decode:success(
                                            {message,
                                                Topic,
                                                Headers,
                                                Reply_to,
                                                Body}
                                        )
                                    end
                                )
                            end
                        )
                    end
                )
            end
        )
    end,
    gleam@dynamic@decode:run(Data, Decoder).

-file("src/glats.gleam", 317).
-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("src/glats.gleam", 237).
-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,
                                begin
                                    _record = State,
                                    {state,
                                        erlang:element(2, _record),
                                        gleam@dict:delete(
                                            erlang:element(3, State),
                                            erlang:element(2, Em)
                                        )}
                                end,
                                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("src/glats.gleam", 211).
?DOC(
    " Starts an actor that handles a connection to NATS using the provided\n"
    " settings.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " connect(\n"
    "   \"localhost\",\n"
    "   4222,\n"
    "   [\n"
    "     CACert(\"/tmp/nats/ca.crt\"),\n"
    "     InboxPrefix(\"_INBOX.custom.prefix.\"),\n"
    "   ],\n"
    " )\n"
    " ```\n"
).
-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}
    ).