distribute

-module(distribute@receiver).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/receiver.gleam").
-export([receive_typed/3, selecting_typed/4, start_receiver/3, start_distributed_worker/5]).
-export_type([receive_error/0, next/1]).

-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 receive_error() :: {decode_error, distribute@codec:decode_error()} |
    timeout.

-type next(GDE) :: {continue, GDE} | stop | {stop_abnormal, binary()}.

-file("src/distribute/receiver.gleam", 28).
?DOC(" Receive and decode one message.\n").
-spec receive_typed(
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, GDG} | {error, distribute@codec:decode_error()}),
    integer()
) -> {ok, GDG} | {error, receive_error()}.
receive_typed(Subject, Decoder, Timeout_ms) ->
    Selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        gleam@erlang@process:select_map(_pipe, Subject, fun(Msg) -> Msg end)
    end,
    case gleam_erlang_ffi:select(Selector, Timeout_ms) of
        {ok, Binary} ->
            case Decoder(Binary) of
                {ok, Value} ->
                    {ok, Value};

                {error, Err} ->
                    {error, {decode_error, Err}}
            end;

        {error, nil} ->
            {error, timeout}
    end.

-file("src/distribute/receiver.gleam", 52).
?DOC(" Add a typed handler to a `Selector`.\n").
-spec selecting_typed(
    gleam@erlang@process:selector(GDK),
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, GDN} | {error, distribute@codec:decode_error()}),
    fun(({ok, GDN} | {error, receive_error()}) -> GDK)
) -> gleam@erlang@process:selector(GDK).
selecting_typed(Selector, Subject, Decoder, Mapper) ->
    gleam@erlang@process:select_map(
        Selector,
        Subject,
        fun(Binary) -> case Decoder(Binary) of
                {ok, Value} ->
                    Mapper({ok, Value});

                {error, Err} ->
                    Mapper({error, {decode_error, Err}})
            end end
    ).

-file("src/distribute/receiver.gleam", 130).
-spec translate_next(next(GEI)) -> gleam@otp@actor:next(GEI, bitstring()).
translate_next(Next) ->
    case Next of
        {continue, S} ->
            gleam@otp@actor:continue(S);

        stop ->
            gleam@otp@actor:stop();

        {stop_abnormal, Reason} ->
            gleam@otp@actor:stop_abnormal(Reason)
    end.

-file("src/distribute/receiver.gleam", 72).
?DOC(
    " Start an OTP actor that decodes binary messages and forwards\n"
    " them to `handler`. Returns the raw `Subject(BitArray)`.\n"
).
-spec start_receiver(
    GDS,
    fun((bitstring()) -> {ok, GDT} | {error, distribute@codec:decode_error()}),
    fun((GDT, GDS) -> next(GDS))
) -> {ok, gleam@erlang@process:subject(bitstring())} |
    {error, gleam@otp@actor:start_error()}.
start_receiver(Initial_state, Decoder, Handler) ->
    _pipe = gleam@otp@actor:new(Initial_state),
    _pipe@1 = gleam@otp@actor:on_message(
        _pipe,
        fun(State, Binary) -> case Decoder(Binary) of
                {ok, Message} ->
                    translate_next(Handler(Message, State));

                {error, _} ->
                    gleam@otp@actor:continue(State)
            end end
    ),
    _pipe@2 = gleam@otp@actor:start(_pipe@1),
    gleam@result:map(_pipe@2, fun(Started) -> erlang:element(3, Started) end).

-file("src/distribute/receiver.gleam", 94).
?DOC(
    " Start a gen_statem actor with a deterministic name-based tag.\n"
    " Remote nodes can reconstruct the Subject via `registry.lookup`.\n"
).
-spec start_distributed_worker(
    binary(),
    GDZ,
    fun((GEA) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, GEA} | {error, distribute@codec:decode_error()}),
    fun((GEA, GDZ) -> next(GDZ))
) -> {ok, gleam@otp@actor:started(distribute@global:global_subject(GEA))} |
    {error, gleam@otp@actor:start_error()}.
start_distributed_worker(Name, Initial_state, Encoder, Decoder, Handler) ->
    _pipe@3 = gleam@otp@actor:new_with_initialiser(
        5000,
        fun(_) ->
            Gs = distribute@global:from_name(
                Name,
                erlang:self(),
                Encoder,
                Decoder
            ),
            Selector = begin
                _pipe = gleam_erlang_ffi:new_selector(),
                gleam@erlang@process:select(
                    _pipe,
                    distribute@global:subject(Gs)
                )
            end,
            {ok,
                begin
                    _pipe@1 = gleam@otp@actor:initialised(Initial_state),
                    _pipe@2 = gleam@otp@actor:selecting(_pipe@1, Selector),
                    gleam@otp@actor:returning(_pipe@2, Gs)
                end}
        end
    ),
    _pipe@4 = gleam@otp@actor:on_message(
        _pipe@3,
        fun(State, Binary) -> case Decoder(Binary) of
                {ok, Message} ->
                    translate_next(Handler(Message, State));

                {error, _} ->
                    gleam@otp@actor:continue(State)
            end end
    ),
    gleam@otp@actor:start(_pipe@4).