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_error_to_string/1, receive_typed/3, selecting_typed/4, start_receiver_observed/4, start_receiver/3, start_distributed_worker_observed/7, start_distributed_worker/6]).
-export_type([worker_inbox/0, receive_error/0, handler_step/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 worker_inbox() :: {incoming, bitstring()} | garbage.

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

-type handler_step(GHK) :: {continue, GHK} | stop | {stop_abnormal, binary()}.

-file("src/distribute/receiver.gleam", 40).
-spec receive_error_to_string(receive_error()) -> binary().
receive_error_to_string(Err) ->
    case Err of
        {decode_error, E} ->
            <<"Receive decode failed: "/utf8,
                (distribute@codec:decode_error_to_string(E))/binary>>;

        receive_timeout ->
            <<"Receive timed out"/utf8>>
    end.

-file("src/distribute/receiver.gleam", 292).
?DOC(
    " Decode with payload size enforcement and telemetry emission.\n"
    "\n"
    " Single source of truth for \"size check before decode\". Used by every\n"
    " path that turns a raw binary into a typed message; the `payload_origin`\n"
    " and `decode_origin` arguments distinguish the call site so the\n"
    " telemetry sink can label events accurately.\n"
).
-spec decode_checked(
    bitstring(),
    fun((bitstring()) -> {ok, GJE} | {error, distribute@codec:decode_error()}),
    distribute@telemetry:payload_origin(),
    distribute@telemetry:decode_origin()
) -> {ok, GJE} | {error, distribute@codec:decode_error()}.
decode_checked(Binary, Decoder, Payload_origin, Decode_origin) ->
    Cap = erlang:element(4, distribute@config:get()),
    Size = erlang:byte_size(Binary),
    case Size > Cap of
        true ->
            distribute@telemetry:emit(
                {payload_rejected, Size, Cap, Payload_origin}
            ),
            {error, {payload_too_large, Size}};

        false ->
            case Decoder(Binary) of
                {ok, V} ->
                    {ok, V};

                {error, E} ->
                    distribute@telemetry:emit(
                        {decode_failed,
                            distribute@codec:decode_error_to_string(E),
                            Decode_origin}
                    ),
                    {error, E}
            end
    end.

-file("src/distribute/receiver.gleam", 74).
?DOC(
    " Receive and decode one message.\n"
    "\n"
    " Creates a fresh selector on each call. For tight loops, prefer building\n"
    " a selector once with `selecting_typed` and reusing it with\n"
    " `process.selector_receive`.\n"
    "\n"
    " Rejects payloads that exceed `config.get().max_payload_size_bytes` with\n"
    " `Error(DecodeError(PayloadTooLarge(size)))`. The decoder is never invoked\n"
    " on oversized binaries.\n"
    "\n"
    " Negative `timeout_ms` is clamped to `0` (poll-once-and-return). Erlang's\n"
    " `receive after Timeout -> ...` would otherwise raise `timeout_value`\n"
    " and crash the caller; a typed `ReceiveTimeout` is a better contract.\n"
).
-spec receive_typed(
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, GHM} | {error, distribute@codec:decode_error()}),
    integer()
) -> {ok, GHM} | {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,
        distribute_ffi_utils:clamp_timeout(Timeout_ms)
    ) of
        {ok, Binary} ->
            case decode_checked(
                Binary,
                Decoder,
                payload_on_receive,
                decode_on_receive
            ) of
                {ok, Value} ->
                    {ok, Value};

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

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

-file("src/distribute/receiver.gleam", 116).
?DOC(
    " Add a typed handler to a `Selector`.\n"
    "\n"
    " Rejects payloads exceeding `config.get().max_payload_size_bytes` with\n"
    " `Error(DecodeError(PayloadTooLarge(size)))`. The decoder is never invoked\n"
    " on oversized binaries.\n"
).
-spec selecting_typed(
    gleam@erlang@process:selector(GHQ),
    gleam@erlang@process:subject(bitstring()),
    fun((bitstring()) -> {ok, GHT} | {error, distribute@codec:decode_error()}),
    fun(({ok, GHT} | {error, receive_error()}) -> GHQ)
) -> gleam@erlang@process:selector(GHQ).
selecting_typed(Selector, Subject, Decoder, Mapper) ->
    gleam@erlang@process:select_map(
        Selector,
        Subject,
        fun(Binary) ->
            case decode_checked(
                Binary,
                Decoder,
                payload_on_receive,
                decode_on_receive
            ) of
                {ok, Value} ->
                    Mapper({ok, Value});

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

-file("src/distribute/receiver.gleam", 345).
-spec translate_next(handler_step(GJM)) -> gleam@otp@actor:next(GJM, any()).
translate_next(Step) ->
    case Step 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", 324).
?DOC(
    " Decode with size enforcement plus an observability hook.\n"
    "\n"
    " On any error (oversized OR decode failure), calls `on_decode_error` and\n"
    " returns `Error(Nil)`. The actor handler uses this to drop bad messages\n"
    " while still notifying instrumentation.\n"
).
-spec decode_observed(
    bitstring(),
    fun((bitstring()) -> {ok, GJI} | {error, distribute@codec:decode_error()}),
    fun((distribute@codec:decode_error()) -> nil)
) -> {ok, GJI} | {error, nil}.
decode_observed(Binary, Decoder, On_decode_error) ->
    case decode_checked(
        Binary,
        Decoder,
        payload_on_actor_inbound,
        decode_on_actor_inbound
    ) of
        {ok, Value} ->
            {ok, Value};

        {error, Err} ->
            On_decode_error(Err),
            {error, nil}
    end.

-file("src/distribute/receiver.gleam", 164).
?DOC(
    " Like `start_receiver` but calls `on_decode_error` for every binary\n"
    " that fails decoding, before continuing with the current state.\n"
    "\n"
    " Oversized payloads (over `config.get().max_payload_size_bytes`) trigger\n"
    " `on_decode_error(PayloadTooLarge(size))` and are NOT forwarded to the\n"
    " decoder. This protects the actor from OOM on malicious or buggy senders.\n"
    "\n"
    " Non-Subject mailbox terms (raw `Pid ! garbage` from anywhere) are\n"
    " silently dropped via `select_other` so they cannot accumulate and\n"
    " pay the selective-receive penalty.\n"
).
-spec start_receiver_observed(
    GIF,
    fun((bitstring()) -> {ok, GIG} | {error, distribute@codec:decode_error()}),
    fun((GIG, GIF) -> handler_step(GIF)),
    fun((distribute@codec:decode_error()) -> nil)
) -> {ok, gleam@erlang@process:subject(bitstring())} |
    {error, gleam@otp@actor:start_error()}.
start_receiver_observed(Initial_state, Decoder, Handler, On_decode_error) ->
    Started = begin
        _pipe@4 = gleam@otp@actor:new_with_initialiser(
            erlang:element(3, distribute@config:get()),
            fun(_) ->
                My_subject = gleam@erlang@process:new_subject(),
                Selector = begin
                    _pipe = gleam_erlang_ffi:new_selector(),
                    _pipe@1 = gleam@erlang@process:select_map(
                        _pipe,
                        My_subject,
                        fun(Field@0) -> {incoming, Field@0} end
                    ),
                    gleam@erlang@process:select_other(
                        _pipe@1,
                        fun(_) -> garbage end
                    )
                end,
                {ok,
                    begin
                        _pipe@2 = gleam@otp@actor:initialised(Initial_state),
                        _pipe@3 = gleam@otp@actor:selecting(_pipe@2, Selector),
                        gleam@otp@actor:returning(_pipe@3, My_subject)
                    end}
            end
        ),
        _pipe@5 = gleam@otp@actor:on_message(
            _pipe@4,
            fun(State, Msg) -> case Msg of
                    garbage ->
                        gleam@otp@actor:continue(State);

                    {incoming, Binary} ->
                        case decode_observed(Binary, Decoder, On_decode_error) of
                            {ok, Message} ->
                                translate_next(Handler(Message, State));

                            {error, nil} ->
                                gleam@otp@actor:continue(State)
                        end
                end end
        ),
        gleam@otp@actor:start(_pipe@5)
    end,
    gleam@result:map(Started, fun(S) -> erlang:element(3, S) end).

-file("src/distribute/receiver.gleam", 146).
?DOC(
    " Start an OTP actor that decodes binary messages and forwards\n"
    " them to `handler`. Returns the raw `Subject(BitArray)`.\n"
    "\n"
    " Decode errors (including `PayloadTooLarge`) are silently dropped.\n"
    " Use `start_receiver_observed` to log or meter malformed messages.\n"
).
-spec start_receiver(
    GHY,
    fun((bitstring()) -> {ok, GHZ} | {error, distribute@codec:decode_error()}),
    fun((GHZ, GHY) -> handler_step(GHY))
) -> {ok, gleam@erlang@process:subject(bitstring())} |
    {error, gleam@otp@actor:start_error()}.
start_receiver(Initial_state, Decoder, Handler) ->
    start_receiver_observed(Initial_state, Decoder, Handler, fun(_) -> nil end).

-file("src/distribute/receiver.gleam", 248).
?DOC(
    " Like `start_distributed_worker` but calls `on_decode_error` for every\n"
    " binary that fails decoding. Use this to log codec mismatches across nodes\n"
    " (e.g. during rolling deploys).\n"
    "\n"
    " Oversized payloads (over `config.get().max_payload_size_bytes`) trigger\n"
    " `on_decode_error(PayloadTooLarge(size))` and are NOT forwarded to the\n"
    " decoder. Critical defence against cross-node OOM attacks.\n"
    "\n"
    " Non-Subject mailbox terms (raw `erlang:send(global:whereis_name(...),\n"
    " garbage)` from any cluster node) are silently dropped via\n"
    " `select_other`. Without this drain the unmatched terms would\n"
    " accumulate in the mailbox forever and pay the linear selective-\n"
    " receive penalty on every subsequent message. A remote DoS vector\n"
    " open to any peer that knows the registered name.\n"
).
-spec start_distributed_worker_observed(
    binary(),
    GIV,
    fun((GIW) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, GIW} | {error, distribute@codec:decode_error()}),
    fun((GIW, GIV) -> handler_step(GIV)),
    fun((distribute@codec:decode_error()) -> nil),
    integer()
) -> {ok, gleam@otp@actor:started(distribute@global:global_subject(GIW))} |
    {error, gleam@otp@actor:start_error()}.
start_distributed_worker_observed(
    Name,
    Initial_state,
    Encoder,
    Decoder,
    Handler,
    On_decode_error,
    Init_timeout_ms
) ->
    _pipe@4 = gleam@otp@actor:new_with_initialiser(
        Init_timeout_ms,
        fun(_) ->
            Gs = distribute@global:unsafe_from_name(
                Name,
                erlang:self(),
                Encoder,
                Decoder
            ),
            Selector = begin
                _pipe = gleam_erlang_ffi:new_selector(),
                _pipe@1 = gleam@erlang@process:select_map(
                    _pipe,
                    distribute@global:subject(Gs),
                    fun(Field@0) -> {incoming, Field@0} end
                ),
                gleam@erlang@process:select_other(
                    _pipe@1,
                    fun(_) -> garbage end
                )
            end,
            {ok,
                begin
                    _pipe@2 = gleam@otp@actor:initialised(Initial_state),
                    _pipe@3 = gleam@otp@actor:selecting(_pipe@2, Selector),
                    gleam@otp@actor:returning(_pipe@3, Gs)
                end}
        end
    ),
    _pipe@5 = gleam@otp@actor:on_message(_pipe@4, fun(State, Msg) -> case Msg of
                garbage ->
                    gleam@otp@actor:continue(State);

                {incoming, Binary} ->
                    case decode_observed(Binary, Decoder, On_decode_error) of
                        {ok, Message} ->
                            translate_next(Handler(Message, State));

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

-file("src/distribute/receiver.gleam", 215).
?DOC(
    " Start a distributed actor whose subject carries a deterministic name-based\n"
    " tag. Remote nodes can reconstruct the subject via `registry.lookup`.\n"
    "\n"
    " Decode errors (including `PayloadTooLarge`) are silently dropped; use\n"
    " `start_distributed_worker_observed` to log or meter them.\n"
    "\n"
    " `init_timeout_ms` is the OTP initialiser timeout passed to\n"
    " `actor.new_with_initialiser`.\n"
).
-spec start_distributed_worker(
    binary(),
    GIM,
    fun((GIN) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, GIN} | {error, distribute@codec:decode_error()}),
    fun((GIN, GIM) -> handler_step(GIM)),
    integer()
) -> {ok, gleam@otp@actor:started(distribute@global:global_subject(GIN))} |
    {error, gleam@otp@actor:start_error()}.
start_distributed_worker(
    Name,
    Initial_state,
    Encoder,
    Decoder,
    Handler,
    Init_timeout_ms
) ->
    start_distributed_worker_observed(
        Name,
        Initial_state,
        Encoder,
        Decoder,
        Handler,
        fun(_) -> nil end,
        Init_timeout_ms
    ).