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, start_typed_actor/4]).
-export_type([receive_error/0, next/1, inbox/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.

-type receive_error() :: {decode_error, distribute@codec:decode_error()} |
    timeout.

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

-type inbox() :: {message, bitstring()} |
    {exit, gleam@erlang@process:exit_message()}.

-file("src/distribute/receiver.gleam", 28).
?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, HCK} | {error, distribute@codec:decode_error()}),
    integer()
) -> {ok, HCK} | {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", 68).
?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(HCO),
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, HCR} | {error, distribute@codec:decode_error()}),
    fun(({ok, HCR} | {error, receive_error()}) -> HCO)
) -> gleam@erlang@process:selector(HCO).
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", 111).
-spec do_receive_loop(
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, HDB} | {error, distribute@codec:decode_error()}),
    HDD,
    fun((HDB, HDD) -> HDD)
) -> 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", 102).
?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, HCX} | {error, distribute@codec:decode_error()}),
    HCZ,
    fun((HCX, HCZ) -> HCZ)
) -> nil.
receive_loop_typed(Subject, Decoder, Initial_state, Handler) ->
    do_receive_loop(Subject, Decoder, Initial_state, Handler).

-file("src/distribute/receiver.gleam", 196).
?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(
    HDE,
    fun((bitstring()) -> {ok, HDF} | {error, distribute@codec:decode_error()}),
    fun((HDF, HDE) -> next(HDE))
) -> {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", 241).
-spec global_loop(
    HDQ,
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, HDS} | {error, distribute@codec:decode_error()}),
    fun((HDS, HDQ) -> next(HDQ))
) -> nil.
global_loop(State, Subject, Decoder, Handler) ->
    Selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        _pipe@1 = gleam@erlang@process:select_map(
            _pipe,
            Subject,
            fun(Msg) -> {message, Msg} end
        ),
        gleam@erlang@process:select_trapped_exits(
            _pipe@1,
            fun(Exit_msg) -> {exit, Exit_msg} end
        )
    end,
    case gleam_erlang_ffi:select(Selector) of
        {message, Binary} ->
            case Decoder(Binary) of
                {ok, Value} ->
                    case Handler(Value, 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;

        {exit, {exit_message, Pid, Reason}} ->
            Reason_str = case Reason of
                normal ->
                    <<"normal"/utf8>>;

                killed ->
                    <<"killed"/utf8>>;

                {abnormal, R} ->
                    gleam@string:inspect(R)
            end,
            distribute@log:warn(
                <<"Linked process exited, shutting down receiver"/utf8>>,
                [{<<"pid"/utf8>>, gleam@string:inspect(Pid)},
                    {<<"reason"/utf8>>, Reason_str}]
            ),
            nil
    end.

-file("src/distribute/receiver.gleam", 225).
?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(
    HDL,
    fun((bitstring()) -> {ok, HDM} | {error, distribute@codec:decode_error()}),
    fun((HDM, HDL) -> next(HDL))
) -> gleam@erlang@process:subject(bitstring()).
start_global_receiver(Initial_state, Decoder, Handler) ->
    Pid = proc_lib:spawn_link(
        fun() ->
            gleam_erlang_ffi:trap_exits(true),
            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()).

-file("src/distribute/receiver.gleam", 328).
?DOC(
    " Start a typed actor that returns a GlobalSubject (RECOMMENDED).\n"
    "\n"
    " This is the recommended way to create type-safe actors for distributed use.\n"
    " The returned `GlobalSubject` enforces encoder/decoder usage and can be\n"
    " registered globally with `registry.register_typed`.\n"
    "\n"
    " Malformed messages are silently ignored.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `initial_state`: The initial state of the actor.\n"
    " - `encoder`: The encoder for outgoing messages (used by clients).\n"
    " - `decoder`: The decoder for incoming messages.\n"
    " - `handler`: The message handler function returning `Next(state)`.\n"
    "\n"
    " ## Returns\n"
    "\n"
    " A `GlobalSubject(msg)` that can be used for type-safe messaging.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " pub type Request {\n"
    "   GetCount\n"
    "   Increment\n"
    " }\n"
    "\n"
    " let actor = receiver.start_typed_actor(\n"
    "   0,\n"
    "   my_encoder(),\n"
    "   my_decoder(),\n"
    "   fn(msg, count) {\n"
    "     case msg {\n"
    "       GetCount -> {\n"
    "         // Send response logic here\n"
    "         receiver.Continue(count)\n"
    "       }\n"
    "       Increment -> receiver.Continue(count + 1)\n"
    "     }\n"
    "   },\n"
    " )\n"
    " ```\n"
).
-spec start_typed_actor(
    HDV,
    fun((HDW) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HDW} | {error, distribute@codec:decode_error()}),
    fun((HDW, HDV) -> next(HDV))
) -> distribute@global:global_subject(HDW).
start_typed_actor(Initial_state, Encoder, Decoder, Handler) ->
    Subject = start_global_receiver(Initial_state, Decoder, Handler),
    distribute@global:from_subject(Subject, Encoder, Decoder).