distribute

-module(distribute@actor).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/actor.gleam").
-export([start_typed_actor/4, start_server/4, start/3, start_global/3, child_spec_typed_actor/3, child_spec_server/3, start_typed_actor_registered/5, start_typed_actor_started/4, child_spec_typed_actor_typed/4, start_typed_actor_supervised/4, pool_supervisor/5]).
-export_type([actor_error/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 actor_error() :: {start_failed, gleam@otp@actor:start_error()} |
    {invalid_configuration, binary()}.

-file("src/distribute/actor.gleam", 176).
?DOC(
    " Start a type-safe actor with global subject (RECOMMENDED).\n"
    "\n"
    " This is the recommended way to create distributed actors. It wraps\n"
    " `receiver.start_typed_actor` for convenience and compatibility.\n"
    "\n"
    " The returned `GlobalSubject(msg)` can be:\n"
    " - Registered globally with `registry.register_typed`\n"
    " - Used for cross-node messaging with `messaging.send_typed`\n"
    " - Passed to other processes safely\n"
    "\n"
    " Malformed messages are silently ignored to prevent crashes.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `initial_state`: The initial state value\n"
    " - `encoder`: Encoder for outgoing messages (used by clients)\n"
    " - `decoder`: Decoder for incoming messages\n"
    " - `handler`: Message handler returning `Next(state)`\n"
    "\n"
    " ## Returns\n"
    "\n"
    " A `GlobalSubject(msg)` that enforces type-safe messaging.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " pub type Counter {\n"
    "   Inc\n"
    "   Dec\n"
    "   GetValue(Subject(Int))\n"
    " }\n"
    "\n"
    " let actor = actor.start_typed_actor(\n"
    "   0,\n"
    "   counter_encoder(),\n"
    "   counter_decoder(),\n"
    "   fn(msg, count) {\n"
    "     case msg {\n"
    "       Inc -> receiver.Continue(count + 1)\n"
    "       Dec -> receiver.Continue(count - 1)\n"
    "       GetValue(reply) -> {\n"
    "         process.send(reply, count)\n"
    "         receiver.Continue(count)\n"
    "       }\n"
    "     }\n"
    "   },\n"
    " )\n"
    " ```\n"
).
-spec start_typed_actor(
    HQS,
    fun((HQT) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HQT} | {error, distribute@codec:decode_error()}),
    fun((HQT, HQS) -> distribute@receiver:next(HQS))
) -> distribute@global:global_subject(HQT).
start_typed_actor(Initial_state, Encoder, Decoder, Handler) ->
    distribute@receiver:start_typed_actor(
        Initial_state,
        Encoder,
        Decoder,
        Handler
    ).

-file("src/distribute/actor.gleam", 231).
?DOC(
    " Start a request-response server actor.\n"
    "\n"
    " This is a specialized version of `start_typed_actor` optimized for\n"
    " request-response patterns. It's semantically identical but provides\n"
    " clearer intent in the API.\n"
    "\n"
    " Use this when your actor primarily processes requests and sends responses\n"
    " back to callers via reply subjects embedded in the request messages.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `initial_state`: The initial server state\n"
    " - `encoder`: Encoder for request messages\n"
    " - `decoder`: Decoder for request messages\n"
    " - `handler`: Request handler that processes requests and sends replies\n"
    "\n"
    " ## Returns\n"
    "\n"
    " A `GlobalSubject(request)` for sending requests to the server.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " pub type Request {\n"
    "   Add(Int, Int, Subject(Int))\n"
    "   Multiply(Int, Int, Subject(Int))\n"
    " }\n"
    "\n"
    " let server = actor.start_server(\n"
    "   Nil,\n"
    "   request_encoder(),\n"
    "   request_decoder(),\n"
    "   fn(req, _state) {\n"
    "     case req {\n"
    "       Add(a, b, reply) -> {\n"
    "         process.send(reply, a + b)\n"
    "         receiver.Continue(Nil)\n"
    "       }\n"
    "       Multiply(a, b, reply) -> {\n"
    "         process.send(reply, a * b)\n"
    "         receiver.Continue(Nil)\n"
    "       }\n"
    "     }\n"
    "   },\n"
    " )\n"
    " ```\n"
).
-spec start_server(
    HQY,
    fun((HQZ) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HQZ} | {error, distribute@codec:decode_error()}),
    fun((HQZ, HQY) -> distribute@receiver:next(HQY))
) -> distribute@global:global_subject(HQZ).
start_server(Initial_state, Encoder, Decoder, Handler) ->
    start_typed_actor(Initial_state, Encoder, Decoder, Handler).

-file("src/distribute/actor.gleam", 259).
?DOC(
    " Start an actor that returns a raw `Subject(BitArray)`.\n"
    "\n"
    " This is a lower-level helper that wraps `receiver.start_typed_receiver`.\n"
    " Prefer `start_typed_actor` which returns a type-safe `GlobalSubject(msg)`.\n"
    "\n"
    " Use this only when you need direct access to the underlying `Subject(BitArray)`\n"
    " for manual message encoding or integration with legacy code.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `initial_state`: The initial state\n"
    " - `decoder`: Decoder for incoming BitArray messages\n"
    " - `handler`: Message handler\n"
    "\n"
    " ## Returns\n"
    "\n"
    " `Ok(Subject(BitArray))` on success, `Error(StartFailed(err))` on failure.\n"
).
-spec start(
    HRE,
    fun((bitstring()) -> {ok, HRF} | {error, distribute@codec:decode_error()}),
    fun((HRF, HRE) -> distribute@receiver:next(HRE))
) -> {ok, gleam@erlang@process:subject(bitstring())} |
    {error, gleam@otp@actor:start_error()}.
start(Initial_state, Decoder, Handler) ->
    distribute@log:warn(
        <<"actor.start is deprecated and will be removed in v3.0.0; use actor.start_typed_actor(...) instead"/utf8>>,
        [{<<"replacement"/utf8>>, <<"actor.start_typed_actor"/utf8>>}]
    ),
    distribute@receiver:start_typed_receiver(Initial_state, Decoder, Handler).

-file("src/distribute/actor.gleam", 289).
?DOC(
    " Start a global actor returning a raw `Subject(BitArray)`.\n"
    "\n"
    " Similar to `start` but uses `receiver.start_global_receiver` which\n"
    " creates a subject compatible with global registry operations.\n"
    "\n"
    " Prefer `start_typed_actor` which wraps this and returns a type-safe\n"
    " `GlobalSubject(msg)`.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `initial_state`: The initial state\n"
    " - `decoder`: Decoder for incoming messages\n"
    " - `handler`: Message handler\n"
    "\n"
    " ## Returns\n"
    "\n"
    " A `Subject(BitArray)` with a `Nil` tag for global use.\n"
).
-spec start_global(
    HRL,
    fun((bitstring()) -> {ok, HRM} | {error, distribute@codec:decode_error()}),
    fun((HRM, HRL) -> distribute@receiver:next(HRL))
) -> gleam@erlang@process:subject(bitstring()).
start_global(Initial_state, Decoder, Handler) ->
    distribute@log:warn(
        <<"actor.start_global is deprecated and will be removed in v3.0.0; use actor.start_typed_actor(...) and registry.register_typed(...) instead"/utf8>>,
        [{<<"replacement"/utf8>>, <<"actor.start_typed_actor"/utf8>>}]
    ),
    distribute@receiver:start_global_receiver(Initial_state, Decoder, Handler).

-file("src/distribute/actor.gleam", 337).
?DOC(
    " Create a supervision child spec for a typed actor.\n"
    "\n"
    " This function returns a `ChildSpecification` that can be added to a\n"
    " `gleam/otp/supervision` supervisor tree. The supervisor will automatically\n"
    " restart the actor if it crashes.\n"
    "\n"
    " Note: Returns `Subject(BitArray)` rather than `GlobalSubject(msg)` for\n"
    " OTP compatibility. Wrap with `global.from_subject` after supervision start\n"
    " if you need the typed wrapper.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `initial_state`: The initial state for the actor\n"
    " - `decoder`: Decoder for messages\n"
    " - `handler`: Message handler function\n"
    "\n"
    " ## Returns\n"
    "\n"
    " A `ChildSpecification(Subject(BitArray))` for use with supervisors.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " import gleam/otp/supervision.{supervisor}\n"
    "\n"
    " pub fn start_supervised() {\n"
    "   supervisor.start_link(fn(children) {\n"
    "     children\n"
    "     |> supervision.add(actor.child_spec_typed_actor(\n"
    "       0,\n"
    "       my_decoder(),\n"
    "       my_handler,\n"
    "     ))\n"
    "   })\n"
    " }\n"
    " ```\n"
).
-spec child_spec_typed_actor(
    HRQ,
    fun((bitstring()) -> {ok, HRR} | {error, distribute@codec:decode_error()}),
    fun((HRR, HRQ) -> distribute@receiver:next(HRQ))
) -> gleam@otp@supervision:child_specification(gleam@erlang@process:subject(bitstring())).
child_spec_typed_actor(Initial_state, Decoder, Handler) ->
    gleam@otp@supervision:worker(
        fun() -> _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
            ),
            gleam@otp@actor:start(_pipe@1) end
    ).

-file("src/distribute/actor.gleam", 377).
?DOC(
    " Create a supervision child spec for a request-response server.\n"
    "\n"
    " Identical to `child_spec_typed_actor` but with clearer semantic intent\n"
    " for server patterns.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " supervisor.start_link(fn(children) {\n"
    "   children\n"
    "   |> supervision.add(actor.child_spec_server(\n"
    "     Nil,\n"
    "     request_decoder(),\n"
    "     handle_request,\n"
    "   ))\n"
    " })\n"
    " ```\n"
).
-spec child_spec_server(
    HRW,
    fun((bitstring()) -> {ok, HRX} | {error, distribute@codec:decode_error()}),
    fun((HRX, HRW) -> distribute@receiver:next(HRW))
) -> gleam@otp@supervision:child_specification(gleam@erlang@process:subject(bitstring())).
child_spec_server(Initial_state, Decoder, Handler) ->
    child_spec_typed_actor(Initial_state, Decoder, Handler).

-file("src/distribute/actor.gleam", 393).
?DOC(
    " Start and register a typed actor under a global name.\n"
    " \n"
    " This helper starts a typed actor and registers its subject with\n"
    " `registry.register_typed(name, subject)`. Returns `Ok(GlobalSubject(msg))`\n"
    " on success or `Error(RegisterError)` if registration fails.\n"
    "\n"
    " Note: If registration fails, the started actor process will continue running.\n"
    " The caller is responsible for cleanup in error cases if needed.\n"
).
-spec start_typed_actor_registered(
    binary(),
    HSC,
    fun((HSD) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HSD} | {error, distribute@codec:decode_error()}),
    fun((HSD, HSC) -> distribute@receiver:next(HSC))
) -> {ok, distribute@global:global_subject(HSD)} |
    {error, distribute@registry:register_error()}.
start_typed_actor_registered(Name, Initial_state, Encoder, Decoder, Handler) ->
    Gs = start_typed_actor(Initial_state, Encoder, Decoder, Handler),
    case distribute@registry:register_typed(Name, distribute@global:subject(Gs)) of
        {ok, _} ->
            {ok, Gs};

        {error, Err} ->
            {error, Err}
    end.

-file("src/distribute/actor.gleam", 410).
?DOC(
    " Start a typed actor via the OTP actor API and return the full Started\n"
    " result with `GlobalSubject(msg)` as the returned data. This is useful\n"
    " for supervision integration where the caller needs access to the pid.\n"
).
-spec start_typed_actor_started(
    HSK,
    fun((HSL) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HSL} | {error, distribute@codec:decode_error()}),
    fun((HSL, HSK) -> distribute@receiver:next(HSK))
) -> {ok, gleam@otp@actor:started(distribute@global:global_subject(HSL))} |
    {error, gleam@otp@actor:start_error()}.
start_typed_actor_started(Initial_state, Encoder, 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) ->
            Subject = erlang:element(3, Started),
            Global_subject = distribute@global:from_subject(
                Subject,
                Encoder,
                Decoder
            ),
            {started, erlang:element(2, Started), Global_subject}
        end
    ).

-file("src/distribute/actor.gleam", 440).
?DOC(
    " Create a typed child spec that returns `GlobalSubject(msg)` as the child data.\n"
    " This helper is supervision-friendly and returns a `ChildSpecification` with\n"
    " the typed data so supervisors can access the typed subject directly.\n"
).
-spec child_spec_typed_actor_typed(
    HST,
    fun((HSU) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HSU} | {error, distribute@codec:decode_error()}),
    fun((HSU, HST) -> distribute@receiver:next(HST))
) -> gleam@otp@supervision:child_specification(distribute@global:global_subject(HSU)).
child_spec_typed_actor_typed(Initial_state, Encoder, Decoder, Handler) ->
    gleam@otp@supervision:worker(
        fun() ->
            start_typed_actor_started(Initial_state, Encoder, Decoder, Handler)
        end
    ).

-file("src/distribute/actor.gleam", 496).
?DOC(
    " Start a typed actor under a new supervisor in one call.\n"
    "\n"
    " This high-level helper creates a supervisor, adds the typed actor as a child,\n"
    " and starts the supervisor tree. Returns both the supervisor PID and the\n"
    " typed actor's GlobalSubject for convenient access.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `initial_state`: The initial state for the actor\n"
    " - `encoder`: Encoder for messages\n"
    " - `decoder`: Decoder for messages\n"
    " - `handler`: Message handler function\n"
    "\n"
    " ## Returns\n"
    "\n"
    " `Ok((Pid, GlobalSubject(msg)))` on success with both handles,\n"
    " or `Error(StartError)` if supervision setup fails.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let result = actor.start_typed_actor_supervised(\n"
    "   0,\n"
    "   counter_encoder(),\n"
    "   counter_decoder(),\n"
    "   fn(msg, count) {\n"
    "     case msg {\n"
    "       Increment -> receiver.Continue(count + 1)\n"
    "       GetCount(reply) -> {\n"
    "         process.send(reply, count)\n"
    "         receiver.Continue(count)\n"
    "       }\n"
    "     }\n"
    "   },\n"
    " )\n"
    "\n"
    " case result {\n"
    "   Ok(#(sup_pid, actor_subject)) -> {\n"
    "     // Use actor_subject for messaging\n"
    "     let _ = global.send(actor_subject, Increment)\n"
    "     // Supervisor will restart actor on crashes\n"
    "   }\n"
    "   Error(err) -> // handle error\n"
    " }\n"
    " ```\n"
).
-spec start_typed_actor_supervised(
    HTA,
    fun((HTB) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HTB} | {error, distribute@codec:decode_error()}),
    fun((HTB, HTA) -> distribute@receiver:next(HTA))
) -> {ok, {gleam@erlang@process:pid_(), distribute@global:global_subject(HTB)}} |
    {error, gleam@otp@actor:start_error()}.
start_typed_actor_supervised(Initial_state, Encoder, Decoder, Handler) ->
    Child_spec = child_spec_typed_actor_typed(
        Initial_state,
        Encoder,
        Decoder,
        Handler
    ),
    Builder = gleam@otp@static_supervisor:new(one_for_one),
    Builder@1 = gleam@otp@static_supervisor:add(Builder, Child_spec),
    case gleam@otp@static_supervisor:start(Builder@1) of
        {ok, Sup} ->
            case start_typed_actor_started(
                Initial_state,
                Encoder,
                Decoder,
                Handler
            ) of
                {ok, Started} ->
                    {ok, {erlang:element(2, Sup), erlang:element(3, Started)}};

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

        {error, Err@1} ->
            {error, Err@1}
    end.

-file("src/distribute/actor.gleam", 568).
?DOC(
    " Create a worker pool with N typed actors under a supervisor.\n"
    "\n"
    " This helper creates a pool of identical worker actors, all supervised\n"
    " under a single supervisor. The pool can be used for load balancing,\n"
    " parallel processing, or handling concurrent requests.\n"
    "\n"
    " ## Parameters\n"
    "\n"
    " - `pool_size`: Number of worker actors to create\n"
    " - `initial_state`: Initial state for each worker (all start with same state)\n"
    " - `encoder`: Encoder for worker messages\n"
    " - `decoder`: Decoder for worker messages  \n"
    " - `handler`: Message handler for workers\n"
    "\n"
    " ## Returns\n"
    "\n"
    " `Ok((Pid, List(GlobalSubject(msg))))` with the supervisor PID and all worker subjects,\n"
    " or `Error(StartError)` if pool creation fails.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " // Create a pool of 5 worker actors\n"
    " let result = actor.pool_supervisor(\n"
    "   5,\n"
    "   0,\n"
    "   request_encoder(),\n"
    "   request_decoder(),\n"
    "   handle_request,\n"
    " )\n"
    "\n"
    " case result {\n"
    "   Ok(#(sup_pid, workers)) -> {\n"
    "     // Round-robin distribution\n"
    "     list.index_map(tasks, fn(task, idx) {\n"
    "       let worker = list.at(workers, idx % list.length(workers))\n"
    "       case worker {\n"
    "         Ok(w) -> global.send(w, task)\n"
    "         Error(_) -> Error(Nil)\n"
    "       }\n"
    "     })\n"
    "   }\n"
    "   Error(err) -> // handle error\n"
    " }\n"
    " ```\n"
).
-spec pool_supervisor(
    integer(),
    HTI,
    fun((HTJ) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HTJ} | {error, distribute@codec:decode_error()}),
    fun((HTJ, HTI) -> distribute@receiver:next(HTI))
) -> {ok,
        {gleam@erlang@process:pid_(),
            list(distribute@global:global_subject(HTJ))}} |
    {error, gleam@otp@actor:start_error()}.
pool_supervisor(Pool_size, Initial_state, Encoder, Decoder, Handler) ->
    Child_specs = begin
        _pipe = gleam@list:range(1, Pool_size),
        gleam@list:map(
            _pipe,
            fun(_) ->
                child_spec_typed_actor_typed(
                    Initial_state,
                    Encoder,
                    Decoder,
                    Handler
                )
            end
        )
    end,
    Builder = gleam@otp@static_supervisor:new(one_for_one),
    Builder_with_children = gleam@list:fold(
        Child_specs,
        Builder,
        fun(Acc, Spec) -> gleam@otp@static_supervisor:add(Acc, Spec) end
    ),
    case gleam@otp@static_supervisor:start(Builder_with_children) of
        {ok, Sup} ->
            Workers_result = begin
                _pipe@1 = gleam@list:range(1, Pool_size),
                _pipe@2 = gleam@list:map(
                    _pipe@1,
                    fun(_) ->
                        start_typed_actor_started(
                            Initial_state,
                            Encoder,
                            Decoder,
                            Handler
                        )
                    end
                ),
                gleam@result:all(_pipe@2)
            end,
            case Workers_result of
                {ok, Started_workers} ->
                    Subjects = gleam@list:map(
                        Started_workers,
                        fun(Started) -> erlang:element(3, Started) end
                    ),
                    {ok, {erlang:element(2, Sup), Subjects}};

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

        {error, Err@1} ->
            {error, Err@1}
    end.