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

-export([publish_message/2, publish/3, request/4, respond/3, subscribe/3, unsubscribe/2, queue_subscribe/4, server_info/1, active_subscriptions/1, new_inbox/0, connect/3]).
-export_type([message/0, server_info/0, connection_option/0, connection_error/0, connection_message/0, subscription_message/0, subscription_actor_message/0, connection_state/0, subscription_actor_state/0]).

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

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

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

-type connection_error() :: timeout | no_responders | unexpected.

-opaque connection_message() :: {subscribe,
        gleam@erlang@process:subject({ok, integer()} | {error, binary()}),
        gleam@erlang@process:subject(subscription_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()}),
        message()} |
    {request,
        gleam@erlang@process:subject(fun(() -> {ok, message()} |
            {error, connection_error()})),
        binary(),
        binary(),
        integer()} |
    {get_server_info,
        gleam@erlang@process:subject({ok, server_info()} |
            {error, connection_error()})} |
    {get_active_subscriptions,
        gleam@erlang@process:subject({ok, integer()} |
            {error, connection_error()})} |
    {exited, gleam@erlang@process:exit_message()}.

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

-type subscription_actor_message() :: {get_sid,
        gleam@erlang@process:subject(integer())} |
    {incoming_message, integer(), 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(subscription_message())}.

-spec publish_message(
    gleam@erlang@process:subject(connection_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(),
    integer()
) -> {ok, message()} | {error, connection_error()}.
request(Conn, Subject, Message, Timeout) ->
    Make_request = gleam@erlang@process:call(
        Conn,
        fun(_capture) -> {request, _capture, Subject, Message, Timeout} end,
        1000
    ),
    Make_request().

-spec respond(
    gleam@erlang@process:subject(connection_message()),
    message(),
    binary()
) -> {ok, nil} | {error, binary()}.
respond(Conn, Message, Body) ->
    case erlang:element(4, Message) of
        {some, Rt} ->
            publish(Conn, Rt, Body);

        none ->
            {error, <<"no reply to subject"/utf8>>}
    end.

-spec subscribe(
    gleam@erlang@process:subject(connection_message()),
    gleam@erlang@process:subject(subscription_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};

        {incoming_message, Sid, Msg} ->
            gleam@otp@actor:send(
                erlang:element(4, State),
                {received_message, erlang:element(2, State), Sid, 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(subscription_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 server_info(gleam@erlang@process:subject(connection_message())) -> {ok,
        server_info()} |
    {error, connection_error()}.
server_info(Conn) ->
    gleam@erlang@process:call(
        Conn,
        fun(Field@0) -> {get_server_info, Field@0} end,
        5000
    ).

-spec active_subscriptions(gleam@erlang@process:subject(connection_message())) -> {ok,
        integer()} |
    {error, connection_error()}.
active_subscriptions(Conn) ->
    gleam@erlang@process:call(
        Conn,
        fun(Field@0) -> {get_active_subscriptions, Field@0} end,
        5000
    ).

-spec new_inbox() -> binary().
new_inbox() ->
    glats@internal@util:random_inbox(<<"_INBOX."/utf8>>).

-spec apply_conn_option(
    gleam@map:map_(binary(), gleam@dynamic:dynamic()),
    connection_option()
) -> gleam@map:map_(binary(), gleam@dynamic:dynamic()).
apply_conn_option(Prev, Opt) ->
    case Opt of
        {ca_cert, Path} ->
            _pipe = Prev,
            _pipe@1 = gleam@map:insert(
                _pipe,
                <<"tls"/utf8>>,
                gleam@dynamic:from(true)
            ),
            gleam@map:insert(
                _pipe@1,
                <<"cacertfile"/utf8>>,
                gleam@dynamic:from(Path)
            );

        {client_cert, Cert, Key} ->
            _pipe@2 = Prev,
            _pipe@3 = gleam@map:insert(
                _pipe@2,
                <<"tls"/utf8>>,
                gleam@dynamic:from(true)
            ),
            _pipe@4 = gleam@map:insert(
                _pipe@3,
                <<"certfile"/utf8>>,
                gleam@dynamic:from(Cert)
            ),
            gleam@map:insert(
                _pipe@4,
                <<"keyfile"/utf8>>,
                gleam@dynamic:from(Key)
            );

        {inbox_prefix, Prefix} ->
            gleam@map:insert(
                Prev,
                <<"inbox_prefix"/utf8>>,
                gleam@dynamic:from(Prefix)
            );

        {connection_timeout, Timeout} ->
            gleam@map:insert(
                Prev,
                <<"connection_timeout"/utf8>>,
                gleam@dynamic:from(Timeout)
            );

        enable_no_responders ->
            gleam@map:insert(
                Prev,
                <<"no_responders"/utf8>>,
                gleam@dynamic:from(true)
            )
    end.

-spec add_ssl_opts(gleam@map:map_(binary(), gleam@dynamic:dynamic())) -> gleam@map:map_(binary(), gleam@dynamic:dynamic()).
add_ssl_opts(Prev) ->
    _pipe = gleam@map:take(
        Prev,
        [<<"cacertfile"/utf8>>, <<"certfile"/utf8>>, <<"keyfile"/utf8>>]
    ),
    _pipe@1 = gleam@map: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@dynamic:from(_pipe@2),
    gleam@map:insert(Prev, <<"ssl_opts"/utf8>>, _pipe@3).

-spec build_settings(binary(), integer(), list(connection_option())) -> gleam@map:map_(gleam@erlang@atom:atom_(), gleam@dynamic:dynamic()).
build_settings(Host, Port, Opts) ->
    _pipe = [{<<"host"/utf8>>, gleam@dynamic:from(Host)},
        {<<"port"/utf8>>, gleam@dynamic:from(Port)}],
    _pipe@1 = gleam@map: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@map:take(
        _pipe@3,
        [<<"host"/utf8>>,
            <<"port"/utf8>>,
            <<"tls"/utf8>>,
            <<"ssl_opts"/utf8>>,
            <<"inbox_prefix"/utf8>>,
            <<"connection_timeout"/utf8>>,
            <<"no_responders"/utf8>>]
    ),
    _pipe@5 = gleam@map: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
    ),
    gleam@map:from_list(_pipe@6).

-spec atom_field(
    binary(),
    fun((gleam@dynamic:dynamic()) -> {ok, GOQ} |
        {error, list(gleam@dynamic:decode_error())})
) -> fun((gleam@dynamic:dynamic()) -> {ok, GOQ} |
    {error, list(gleam@dynamic:decode_error())}).
atom_field(Key, Value) ->
    gleam@dynamic:field(erlang:binary_to_atom(Key), Value).

-spec headers(gleam@dynamic:dynamic()) -> {ok,
        gleam@map:map_(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 gleam@map:from_list/1),
    gleam@result:'or'(_pipe@2, {ok, gleam@map:new()}).

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

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

-spec map_gnat_message(gleam@dynamic:dynamic()) -> subscription_actor_message().
map_gnat_message(Data) ->
    Sid_ = begin
        _pipe = Data,
        (gleam@dynamic:field(
            erlang:binary_to_atom(<<"sid"/utf8>>),
            fun gleam@dynamic:int/1
        ))(_pipe)
    end,
    case Sid_ of
        {ok, Sid} ->
            _pipe@1 = Data,
            _pipe@2 = decode_msg(_pipe@1),
            _pipe@3 = gleam@result:map(
                _pipe@2,
                fun(_capture) -> {incoming_message, Sid, _capture} end
            ),
            gleam@result:unwrap(_pipe@3, {decode_error, Data});

        {error, _} ->
            {decode_error, Data}
    end.

-spec handle_publish(
    gleam@erlang@process:subject({ok, nil} | {error, binary()}),
    message(),
    connection_state()
) -> gleam@otp@actor:next(connection_state()).
handle_publish(From, Message, State) ->
    Opts = case erlang:element(4, Message) of
        {some, Rt} ->
            [{erlang:binary_to_atom(<<"reply_to"/utf8>>), Rt}];

        none ->
            []
    end,
    case begin
        _pipe = 'Elixir.Gnat':pub(
            erlang:element(2, State),
            erlang:element(2, Message),
            erlang:element(5, Message),
            Opts
        ),
        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(fun(() -> {ok, message()} |
        {error, connection_error()})),
    binary(),
    binary(),
    any(),
    connection_state()
) -> gleam@otp@actor:next(connection_state()).
handle_request(From, Subject, Message, Timeout, State) ->
    Opts = [{erlang:binary_to_atom(<<"receive_timeout"/utf8>>),
            gleam@dynamic:from(Timeout)}],
    Req_func = fun() ->
        case 'Elixir.Gnat':request(
            erlang:element(2, State),
            Subject,
            Message,
            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}.

-spec start_subscription_actor(
    gleam@erlang@process:subject(subscription_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 map_gnat_message/1
                    )
                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(subscription_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_active_subscriptions(
    gleam@erlang@process:subject({ok, integer()} | {error, connection_error()}),
    connection_state()
) -> gleam@otp@actor:next(connection_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}.

-spec handle_server_info(
    gleam@erlang@process:subject({ok, server_info()} |
        {error, connection_error()}),
    connection_state()
) -> gleam@otp@actor:next(connection_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}.

-spec handle_command(connection_message(), connection_state()) -> gleam@otp@actor:next(connection_state()).
handle_command(Message, State) ->
    case Message of
        {exited, Em} ->
            {stop, erlang:element(3, Em)};

        {publish, From, Msg} ->
            handle_publish(From, Msg, State);

        {request, From@1, Subject, Msg@1, Timeout} ->
            handle_request(From@1, Subject, Msg@1, Timeout, 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);

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

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

        _ ->
            {continue, State}
    end.

-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),
                Subject = gleam@erlang@process:new_subject(),
                Selector = begin
                    _pipe = gleam_erlang_ffi:new_selector(),
                    _pipe@1 = gleam@erlang@process:selecting_trapped_exits(
                        _pipe,
                        fun(Field@0) -> {exited, Field@0} end
                    ),
                    gleam@erlang@process:selecting(
                        _pipe@1,
                        Subject,
                        fun(Msg) -> Msg end
                    )
                end,
                case 'Elixir.Gnat':start_link(build_settings(Host, Port, Opts)) of
                    {ok, Pid} ->
                        {ready, {connection_state, Pid, Subject}, Selector};

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