glats

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

-export([decode_raw_msg/1, start_subscriber/3]).
-export_type([raw_message/0, state/2, message/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.

?MODULEDOC(false).

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

-type state(GFE, GFF) :: {state,
        GFE,
        gleam@erlang@process:subject(GFF),
        fun((GFE, raw_message()) -> GFF)}.

-opaque message() :: {received_message, raw_message()} |
    {decode_error, gleam@dynamic:dynamic_()} |
    subscriber_exited.

-file("src/glats/internal/subscription.gleam", 73).
?DOC(false).
-spec loop(message(), state(GFO, GFP)) -> gleam@otp@actor:next(any(), state(GFO, GFP)).
loop(Message, State) ->
    case Message of
        {received_message, Raw_msg} ->
            gleam@otp@actor:send(
                erlang:element(3, State),
                (erlang:element(4, State))(erlang:element(2, State), Raw_msg)
            ),
            {continue, State, none};

        {decode_error, Data} ->
            gleam_stdlib:println(
                <<"failed to decode: "/utf8,
                    (gleam@string:inspect(Data))/binary>>
            ),
            {continue, State, none};

        subscriber_exited ->
            gleam_stdlib:println(<<"subscriber exited"/utf8>>),
            {stop, normal}
    end.

-file("src/glats/internal/subscription.gleam", 109).
?DOC(false).
-spec decode_sid(fun((integer()) -> gleam@dynamic@decode:decoder(GHC))) -> gleam@dynamic@decode:decoder(GHC).
decode_sid(Next) ->
    gleam@dynamic@decode:optional_field(
        erlang:binary_to_atom(<<"sid"/utf8>>),
        -1,
        {decoder, fun gleam@dynamic@decode:decode_int/1},
        Next
    ).

-file("src/glats/internal/subscription.gleam", 114).
?DOC(false).
-spec decode_status(
    fun((gleam@option:option(integer())) -> gleam@dynamic@decode:decoder(GHF))
) -> gleam@dynamic@decode:decoder(GHF).
decode_status(Next) ->
    gleam@dynamic@decode:optional_field(
        erlang:binary_to_atom(<<"status"/utf8>>),
        none,
        gleam@dynamic@decode:optional(
            {decoder, fun gleam@dynamic@decode:decode_int/1}
        ),
        Next
    ).

-file("src/glats/internal/subscription.gleam", 126).
?DOC(false).
-spec decode_headers(
    fun((gleam@dict:dict(binary(), binary())) -> gleam@dynamic@decode:decoder(GHK))
) -> gleam@dynamic@decode:decoder(GHK).
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/internal/subscription.gleam", 137).
?DOC(false).
-spec decode_reply_to(
    fun((gleam@option:option(binary())) -> gleam@dynamic@decode:decoder(GHR))
) -> gleam@dynamic@decode:decoder(GHR).
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/internal/subscription.gleam", 93).
?DOC(false).
-spec decode_raw_msg(gleam@dynamic:dynamic_()) -> {ok, raw_message()} |
    {error, list(gleam@dynamic@decode:decode_error())}.
decode_raw_msg(Data) ->
    Decoder = begin
        decode_sid(
            fun(Sid) ->
                decode_status(
                    fun(Status) ->
                        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(
                                                            {raw_message,
                                                                Sid,
                                                                Status,
                                                                Topic,
                                                                Headers,
                                                                Reply_to,
                                                                Body}
                                                        )
                                                    end
                                                )
                                            end
                                        )
                                    end
                                )
                            end
                        )
                    end
                )
            end
        )
    end,
    gleam@dynamic@decode:run(Data, Decoder).

-file("src/glats/internal/subscription.gleam", 66).
?DOC(false).
-spec map_gnat_message(gleam@dynamic:dynamic_()) -> message().
map_gnat_message(Data) ->
    _pipe = Data,
    _pipe@1 = decode_raw_msg(_pipe),
    _pipe@2 = gleam@result:map(
        _pipe@1,
        fun(Field@0) -> {received_message, Field@0} end
    ),
    gleam@result:unwrap(_pipe@2, {decode_error, Data}).

-file("src/glats/internal/subscription.gleam", 37).
?DOC(false).
-spec start_subscriber(
    GFI,
    gleam@erlang@process:subject(GFJ),
    fun((GFI, raw_message()) -> GFJ)
) -> {ok, gleam@erlang@process:subject(message())} |
    {error, gleam@otp@actor:start_error()}.
start_subscriber(Conn, Receiver, Mapper) ->
    gleam@otp@actor:start_spec(
        {spec,
            fun() ->
                Monitor = gleam@erlang@process:monitor_process(
                    begin
                        _pipe = Receiver,
                        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,
                {ready, {state, Conn, Receiver, Mapper}, Selector}
            end,
            1000,
            fun loop/2}
    ).