glats

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

-export([publish_message/2, publish/3, request/3, subscribe/3, start/1]).
-export_type([settings/0, command/0, server_error/0, state/0]).

-type settings() :: {settings,
        gleam@option:option(binary()),
        gleam@option:option(integer()),
        gleam@option:option(boolean()),
        gleam@option:option(gleam@map:map_(binary(), binary()))}.

-opaque command() :: {subscribe,
        gleam@erlang@process:subject({ok, binary()} | {error, binary()}),
        gleam@erlang@process:subject(gleam@dynamic:dynamic()),
        binary()} |
    {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 server_error() :: timeout | no_responders | unknown.

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

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

-spec publish(gleam@erlang@process:subject(command()), binary(), binary()) -> {ok,
        nil} |
    {error, binary()}.
publish(Conn, Subject, Message) ->
    publish_message(Conn, glats@message:new_message(Subject, Message)).

-spec request(gleam@erlang@process:subject(command()), 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,
        10000
    ).

-spec subscribe(
    gleam@erlang@process:subject(command()),
    gleam@erlang@process:subject(gleam@dynamic:dynamic()),
    binary()
) -> {ok, binary()} | {error, binary()}.
subscribe(Conn, Receiver, Subject) ->
    gleam@erlang@process:call(
        Conn,
        fun(_capture) -> {subscribe, _capture, Receiver, Subject} end,
        10000
    ).

-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()}), 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()}), 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()}), 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()}), 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(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(),
    state()
) -> gleam@otp@actor:next(state()).
handle_publish(From, Message, State) ->
    case begin
        _pipe = 'Elixir.Gnat':pub(
            erlang:element(2, State),
            erlang:element(2, Message),
            erlang:element(5, Message),
            gleam@dynamic:from([])
        ),
        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_subscribe(
    gleam@erlang@process:subject({ok, binary()} | {error, binary()}),
    gleam@erlang@process:subject(any()),
    binary(),
    state()
) -> gleam@otp@actor:next(state()).
handle_subscribe(From, Receiver, Subject, State) ->
    _pipe@1 = 'Elixir.Gnat':sub(
        erlang:element(2, State),
        begin
            _pipe = Receiver,
            gleam@erlang@process:subject_owner(_pipe)
        end,
        Subject,
        gleam@dynamic:from([])
    ),
    gleam@erlang@process:send(From, _pipe@1),
    {continue, State}.

-spec handle_request(
    gleam@erlang@process:subject({ok, glats@message:message()} |
        {error, server_error()}),
    binary(),
    binary(),
    state()
) -> gleam@otp@actor:next(state()).
handle_request(From, Subject, Message, State) ->
    case 'Elixir.Gnat':request(
        erlang:element(2, State),
        Subject,
        Message,
        gleam@dynamic:from([])
    ) of
        {ok, Msg} ->
            _pipe = 'Elixir.Glats':convert_bare_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 handle_command(command(), state()) -> gleam@otp@actor:next(state()).
handle_command(Cmd, State) ->
    case Cmd of
        {publish, From, Message} ->
            handle_publish(From, Message, State);

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

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

        _ ->
            {continue, State}
    end.

-spec start(settings()) -> {ok, gleam@erlang@process:subject(command())} |
    {error, gleam@otp@actor:start_error()}.
start(Settings) ->
    gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                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}, Selector};

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