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, receive_loop_typed/4, start_typed_receiver/3, start_global_receiver/3]).
-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(GXF) :: {continue, GXF} | stop | {stop_abnormal, binary()}.

-file("src/distribute/receiver.gleam", 23).
?DOC(
    " Receive a single typed message from a Subject with timeout.\n"
    "\n"
    " Waits for a message, decodes it using the provided decoder, and returns\n"
    " the decoded value. Returns an error on timeout or decode failure.\n"
).
-spec receive_typed(
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, GXH} | {error, distribute@codec:decode_error()}),
    integer()
) -> {ok, GXH} | {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", 63).
?DOC(
    " Add a typed message handler to a Selector.\n"
    "\n"
    " This function allows you to add a handler for messages from a specific\n"
    " Subject to an existing Selector. The handler will decode messages using\n"
    " the provided decoder and transform them with the mapper function.\n"
    "\n"
    " The mapper function receives a `Result` containing either the decoded value\n"
    " or a decode error. This allows the caller to handle malformed messages\n"
    " gracefully (e.g. by logging them or ignoring them).\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `selector`: The Selector to add the handler to.\n"
    " - `subject`: The Subject to receive messages from.\n"
    " - `decoder`: The Decoder to use for decoding binary messages.\n"
    " - `mapper`: A function to transform the decode result into the selector's\n"
    "   return type `b`.\n"
    "\n"
    " ## Returns\n"
    "\n"
    " The updated Selector with the new handler added.\n"
).
-spec selecting_typed(
    gleam@erlang@process:selector(GXL),
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, GXO} | {error, distribute@codec:decode_error()}),
    fun(({ok, GXO} | {error, receive_error()}) -> GXL)
) -> gleam@erlang@process:selector(GXL).
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", 106).
-spec do_receive_loop(
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, GXY} | {error, distribute@codec:decode_error()}),
    GYA,
    fun((GXY, GYA) -> GYA)
) -> nil.
do_receive_loop(Subject, Decoder, State, Handler) ->
    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) of
        Binary ->
            case Decoder(Binary) of
                {ok, Value} ->
                    New_state = Handler(Value, State),
                    do_receive_loop(Subject, Decoder, New_state, Handler);

                {error, _} ->
                    do_receive_loop(Subject, Decoder, State, Handler)
            end
    end.

-file("src/distribute/receiver.gleam", 97).
?DOC(
    " Receive messages in a loop with a typed handler function.\n"
    "\n"
    " This function continuously receives messages from a Subject, decodes them,\n"
    " and passes them to a handler function. The handler returns updated state,\n"
    " and the loop continues until externally stopped (process termination).\n"
    "\n"
    " Malformed messages that fail to decode are silently skipped.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `subject`: The Subject to receive messages from.\n"
    " - `decoder`: The Decoder for incoming binary messages.\n"
    " - `initial_state`: The initial state value.\n"
    " - `handler`: Function that processes decoded messages and returns updated state.\n"
    "\n"
    " ## Note\n"
    "\n"
    " This function runs forever. Use it in a spawned process that can be terminated\n"
    " externally, or structure your handler to perform side effects and ignore state\n"
    " if you want graceful shutdown.\n"
).
-spec receive_loop_typed(
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, GXU} | {error, distribute@codec:decode_error()}),
    GXW,
    fun((GXU, GXW) -> GXW)
) -> nil.
receive_loop_typed(Subject, Decoder, Initial_state, Handler) ->
    do_receive_loop(Subject, Decoder, Initial_state, Handler).

-file("src/distribute/receiver.gleam", 183).
?DOC(
    " Start an actor that receives and decodes typed messages.\n"
    "\n"
    " This is a convenience wrapper around `actor.start` that handles binary\n"
    " decoding automatically. The returned `Subject` accepts `BitArray` messages,\n"
    " making it suitable for use with `distribute`'s typed messaging API.\n"
    "\n"
    " Malformed messages are silently ignored.\n"
    "\n"
    " Note: This helper does not support custom selectors in the handler return\n"
    " value. If you need complex selector logic, use `selecting_typed` with a\n"
    " standard `actor.start_spec`.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `initial_state`: The initial state of the actor.\n"
    " - `decoder`: The decoder for incoming messages.\n"
    " - `handler`: The message handler function.\n"
    "\n"
    " ## Returns\n"
    "\n"
    " `Ok(Subject(BitArray))` on success, or `Error(StartError)`.\n"
).
-spec start_typed_receiver(
    GYB,
    fun((bitstring()) -> {ok, GYC} | {error, distribute@codec:decode_error()}),
    fun((GYC, GYB) -> next(GYB))
) -> {ok, gleam@erlang@process:subject(bitstring())} |
    {error, gleam@otp@actor:start_error()}.
start_typed_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} ->
                    case Handler(Message, State) of
                        {continue, New_state} ->
                            gleam@otp@actor:continue(New_state);

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

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

                {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", 225).
-spec global_loop(
    GYN,
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, GYP} | {error, distribute@codec:decode_error()}),
    fun((GYP, GYN) -> next(GYN))
) -> nil.
global_loop(State, Subject, Decoder, Handler) ->
    Selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        gleam@erlang@process:select_map(_pipe, Subject, fun(Msg) -> Msg end)
    end,
    Binary = gleam_erlang_ffi:select(Selector),
    case Decoder(Binary) of
        {ok, Message} ->
            case Handler(Message, State) of
                {continue, New_state} ->
                    global_loop(New_state, Subject, Decoder, Handler);

                stop ->
                    nil;

                {stop_abnormal, _} ->
                    nil
            end;

        {error, _} ->
            global_loop(State, Subject, Decoder, Handler)
    end.

-file("src/distribute/receiver.gleam", 212).
?DOC(
    " Start a global receiver actor.\n"
    "\n"
    " This actor is compatible with `registry.register_typed` and `registry.whereis_typed`.\n"
    " It accepts messages with a `Nil` tag, which allows it to be addressed globally\n"
    " without sharing a specific Reference tag.\n"
    "\n"
    " Use this function if you intend to register the process using `registry.register_typed`.\n"
).
-spec start_global_receiver(
    GYI,
    fun((bitstring()) -> {ok, GYJ} | {error, distribute@codec:decode_error()}),
    fun((GYJ, GYI) -> next(GYI))
) -> gleam@erlang@process:subject(bitstring()).
start_global_receiver(Initial_state, Decoder, Handler) ->
    Pid = proc_lib:spawn_link(
        fun() ->
            Subject = gleam@erlang@process:unsafely_create_subject(
                erlang:self(),
                gleam@dynamic:nil()
            ),
            global_loop(Initial_state, Subject, Decoder, Handler)
        end
    ),
    gleam@erlang@process:unsafely_create_subject(Pid, gleam@dynamic:nil()).