distribute

-module(distribute@cluster).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/cluster.gleam").
-export([start_error_to_string/1, connect_error_to_string/1, start_node/2, start_monitor/0, subscribe/2, unsubscribe/2, connect/1, nodes/0, self_node/0, ping/1, is_distributed/0, connected_count/0, has_peers/0, is_healthy/0, stable_partition/2, health_parallelism_cap/1, health/0]).
-export_type([start_error/0, connect_error/0, cluster_health/0, proxy_result/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 start_error() :: {invalid_node_name, binary()} |
    {invalid_cookie_format, binary()} |
    already_started |
    {network_error, binary()} |
    {start_failed, binary()} |
    start_atom_budget_exceeded.

-type connect_error() :: connect_failed |
    connect_ignored |
    {invalid_node_format, binary()} |
    connect_atom_budget_exceeded.

-type cluster_health() :: {cluster_health,
        binary(),
        boolean(),
        list(binary()),
        integer(),
        list(binary()),
        list(binary())}.

-type proxy_result() :: {partition_ready, {list(binary()), list(binary())}} |
    proxy_exited.

-file("src/distribute/cluster.gleam", 57).
-spec start_error_to_string(start_error()) -> binary().
start_error_to_string(Err) ->
    case Err of
        {invalid_node_name, R} ->
            <<"Invalid node name: "/utf8, R/binary>>;

        {invalid_cookie_format, R@1} ->
            <<"Invalid cookie format: "/utf8, R@1/binary>>;

        already_started ->
            <<"Node is already started"/utf8>>;

        {network_error, R@2} ->
            <<"Network error: "/utf8, R@2/binary>>;

        {start_failed, R@3} ->
            <<"Node start failed: "/utf8, R@3/binary>>;

        start_atom_budget_exceeded ->
            <<"Atom budget exceeded (raise config.max_distribution_atoms)"/utf8>>
    end.

-file("src/distribute/cluster.gleam", 69).
-spec connect_error_to_string(connect_error()) -> binary().
connect_error_to_string(Err) ->
    case Err of
        connect_failed ->
            <<"Connect failed (peer unreachable or refused)"/utf8>>;

        connect_ignored ->
            <<"Connect ignored (local node not distributed)"/utf8>>;

        {invalid_node_format, R} ->
            <<"Invalid node format: "/utf8, R/binary>>;

        connect_atom_budget_exceeded ->
            <<"Atom budget exceeded (raise config.max_distribution_atoms)"/utf8>>
    end.

-file("src/distribute/cluster.gleam", 154).
?DOC(
    " Start a distributed BEAM node.\n"
    "\n"
    " `name` must contain `@` (e.g. `\"myapp@127.0.0.1\"`).\n"
    " Cookie length and charset are enforced byte-wise by the FFI: any\n"
    " failure surfaces as `InvalidCookieFormat`.\n"
    "\n"
    " Atom-budget exhaustion: the FFI emits\n"
    " `AtomBudgetExhausted(<offending input>, AtomBudgetOnStartNode)`\n"
    " before returning, with the actual offending input (name or\n"
    " cookie). We do not re-emit here because the public unit\n"
    " constructor `StartAtomBudgetExceeded` cannot carry that\n"
    " attribution.\n"
    "\n"
    " ## Blocking and OS-level dependencies\n"
    "\n"
    " **This call can block.** It delegates to `net_kernel:start/1`,\n"
    " which talks to `epmd` (Erlang Port Mapper Daemon) and resolves\n"
    " the host portion of `name` against the OS resolver. If `epmd`\n"
    " is not running, if DNS is misconfigured, or if the network goes\n"
    " down a moment before the call, the BEAM may hang on a libc\n"
    " resolver timeout for tens of seconds and there is no Gleam-side\n"
    " timeout the library can interpose.\n"
    "\n"
    " Mitigations callers can apply:\n"
    "\n"
    " - Run `epmd -daemon` before the process boots, and treat its\n"
    "   absence as a fatal startup condition rather than something\n"
    "   `start_node` should recover from.\n"
    " - Use IP literals (`myapp@127.0.0.1`) when the deployment allows,\n"
    "   bypassing DNS entirely.\n"
    " - In container deployments, ensure `/etc/hosts` resolves the\n"
    "   chosen host before `start_node` is called.\n"
    "\n"
    " If you cannot accept a potentially long boot wait, supervise the\n"
    " boot itself: spawn a process that calls `start_node`, monitor\n"
    " it, and treat a deadline miss as a startup failure. The library\n"
    " does not bake a timeout in because the right value is\n"
    " deployment-specific (a 2 s timeout is generous for IP literals\n"
    " but a hair-trigger for DNS-backed names).\n"
).
-spec start_node(binary(), binary()) -> {ok, nil} | {error, start_error()}.
start_node(Name, Cookie) ->
    cluster_ffi:start_node(Name, Cookie).

-file("src/distribute/cluster.gleam", 160).
?DOC(
    " Start the cluster monitor actor. It listens for Erlang node events\n"
    " and broadcasts them to all Gleam subscribers.\n"
).
-spec start_monitor() -> {ok,
        gleam@erlang@process:subject(distribute@cluster_monitor:message())} |
    {error, gleam@otp@actor:start_error()}.
start_monitor() ->
    distribute@cluster_monitor:start().

-file("src/distribute/cluster.gleam", 168).
?DOC(" Subscribe a subject to cluster events (NodeUp/NodeDown).\n").
-spec subscribe(
    gleam@erlang@process:subject(distribute@cluster_monitor:message()),
    gleam@erlang@process:subject(distribute@cluster_monitor:cluster_event())
) -> nil.
subscribe(Monitor, Listener) ->
    distribute@cluster_monitor:subscribe(Monitor, Listener).

-file("src/distribute/cluster.gleam", 176).
?DOC(" Unsubscribe from cluster events.\n").
-spec unsubscribe(
    gleam@erlang@process:subject(distribute@cluster_monitor:message()),
    gleam@erlang@process:subject(distribute@cluster_monitor:cluster_event())
) -> nil.
unsubscribe(Monitor, Listener) ->
    distribute@cluster_monitor:unsubscribe(Monitor, Listener).

-file("src/distribute/cluster.gleam", 202).
?DOC(
    " Connect to a remote node. Returns `Ok(Nil)` on success.\n"
    "\n"
    " ## Atom-table guardrail\n"
    "\n"
    " Each call with a previously-unseen node name interns one atom in\n"
    " the BEAM atom table (atoms are never garbage collected, and the\n"
    " table is capped at 1 048 576 entries by default). To prevent a\n"
    " caller, malicious or buggy, from exhausting the table by\n"
    " looping over millions of valid-looking names, every fresh atom\n"
    " creation is **counted against `config.max_distribution_atoms`**.\n"
    "\n"
    " The check is atomic (`atomics:add_get/3`) and lock-free. Once the\n"
    " budget is reached, this function returns\n"
    " `Error(ConnectAtomBudgetExceeded)` *before* `binary_to_atom/2` is\n"
    " called: the VM atom table cannot be exhausted through this path.\n"
    "\n"
    " Default budget: 10 000 fresh atoms over the process lifetime.\n"
    " 10x a generous cluster size, four orders of magnitude below the\n"
    " VM cap. Tune via `config.configure(... max_distribution_atoms:)`.\n"
).
-spec connect(binary()) -> {ok, nil} | {error, connect_error()}.
connect(Node) ->
    case cluster_ffi:connect(Node) of
        {error, connect_atom_budget_exceeded} ->
            distribute@telemetry:emit(
                {atom_budget_exhausted, Node, atom_budget_on_connect}
            ),
            {error, connect_atom_budget_exceeded};

        Other ->
            Other
    end.

-file("src/distribute/cluster.gleam", 220).
?DOC(" List all currently connected nodes.\n").
-spec nodes() -> list(binary()).
nodes() ->
    cluster_ffi:nodes().

-file("src/distribute/cluster.gleam", 225).
?DOC(" Get the current node's name.\n").
-spec self_node() -> binary().
self_node() ->
    cluster_ffi:self_node().

-file("src/distribute/cluster.gleam", 235).
?DOC(
    " Ping a remote node. Returns `True` if it responds.\n"
    "\n"
    " Subject to the same `config.max_distribution_atoms` guardrail as\n"
    " [`connect`](#connect): once the fresh-atom budget is exhausted,\n"
    " `ping` returns `False` (cannot reach) without touching the VM\n"
    " atom table.\n"
).
-spec ping(binary()) -> boolean().
ping(Node) ->
    cluster_ffi:ping(Node).

-file("src/distribute/cluster.gleam", 245).
?DOC(
    " Whether this node is running BEAM distribution.\n"
    "\n"
    " Backed by `erlang:is_alive/0`, which is the authoritative signal.\n"
    " It returns `true` iff `net_kernel` has been started. Previous versions\n"
    " compared the string form of the node name against `\"nonode@nohost\"`,\n"
    " which would lie if the runtime ever changed that placeholder.\n"
).
-spec is_distributed() -> boolean().
is_distributed() ->
    cluster_ffi:is_alive().

-file("src/distribute/cluster.gleam", 250).
?DOC(" Number of currently connected nodes.\n").
-spec connected_count() -> integer().
connected_count() ->
    erlang:length(nodes()).

-file("src/distribute/cluster.gleam", 258).
?DOC(
    " Whether this node has at least one connected peer.\n"
    "\n"
    " This is a topology check, not a health check: a single-node deployment\n"
    " is operationally fine and will return `False` here.\n"
).
-spec has_peers() -> boolean().
has_peers() ->
    is_distributed() andalso (nodes() /= []).

-file("src/distribute/cluster.gleam", 264).
?DOC(
    " Deprecated alias for `has_peers/0`, kept for compatibility with direct\n"
    " `distribute/cluster` imports from pre-facade code.\n"
).
-spec is_healthy() -> boolean().
is_healthy() ->
    has_peers().

-file("src/distribute/cluster.gleam", 499).
?DOC(" Consume a late partition tuple raced past our caller-side timeout.\n").
-spec drain_proxy_reply(
    gleam@erlang@process:subject({list(binary()), list(binary())})
) -> nil.
drain_proxy_reply(Result_subject) ->
    Drain = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        gleam@erlang@process:select_map(
            _pipe,
            Result_subject,
            fun(_) -> nil end
        )
    end,
    _ = gleam_erlang_ffi:select(Drain, 0),
    nil.

-file("src/distribute/cluster.gleam", 339).
-spec health_proxy_shutdown_grace_ms() -> integer().
health_proxy_shutdown_grace_ms() ->
    erlang:element(9, distribute@config:get()).

-file("src/distribute/cluster.gleam", 516).
?DOC(false).
-spec stable_partition(list(binary()), list(binary())) -> {list(binary()),
    list(binary())}.
stable_partition(Connected, Reachable_candidates) ->
    Reachable_set = gleam@list:fold(
        Reachable_candidates,
        maps:new(),
        fun(Acc, Node) -> gleam@dict:insert(Acc, Node, nil) end
    ),
    {Reachable_rev, Unreachable_rev} = gleam@list:fold(
        Connected,
        {[], []},
        fun(Acc@1, Node@1) ->
            {Reachable, Unreachable} = Acc@1,
            case gleam@dict:has_key(Reachable_set, Node@1) of
                true ->
                    {[Node@1 | Reachable], Unreachable};

                false ->
                    {Reachable, [Node@1 | Unreachable]}
            end
        end
    ),
    {lists:reverse(Reachable_rev), lists:reverse(Unreachable_rev)}.

-file("src/distribute/cluster.gleam", 680).
-spec terminate_ping_workers(list({binary(), gleam@erlang@process:pid_()})) -> nil.
terminate_ping_workers(Workers) ->
    gleam@list:each(
        Workers,
        fun(Entry) ->
            {_, Pid} = Entry,
            _ = distribute_ffi_utils:exit_shutdown(Pid),
            case erlang:is_process_alive(Pid) of
                true ->
                    gleam@erlang@process:kill(Pid);

                false ->
                    nil
            end
        end
    ).

-file("src/distribute/cluster.gleam", 653).
-spec maybe_spawn_next_worker(
    gleam@erlang@process:subject({binary(), boolean()}),
    list(binary()),
    list({binary(), gleam@erlang@process:pid_()})
) -> {list(binary()), list({binary(), gleam@erlang@process:pid_()})}.
maybe_spawn_next_worker(Collector, Queue, Workers) ->
    case Queue of
        [] ->
            {Queue, Workers};

        [Node | Rest] ->
            Pid = proc_lib:spawn(
                fun() ->
                    gleam@erlang@process:send(
                        Collector,
                        {Node, cluster_ffi:ping(Node)}
                    )
                end
            ),
            {Rest, [{Node, Pid} | Workers]}
    end.

-file("src/distribute/cluster.gleam", 670).
-spec drop_ping_worker(list({binary(), gleam@erlang@process:pid_()}), binary()) -> list({binary(),
    gleam@erlang@process:pid_()}).
drop_ping_worker(Workers, Node) ->
    gleam@list:filter(
        Workers,
        fun(Entry) ->
            {Worker_node, _} = Entry,
            Worker_node /= Node
        end
    ).

-file("src/distribute/cluster.gleam", 537).
-spec collect_pings(
    gleam@erlang@process:subject({binary(), boolean()}),
    list(binary()),
    gleam@dict:dict(binary(), nil),
    list({binary(), gleam@erlang@process:pid_()}),
    list(binary()),
    list(binary()),
    integer()
) -> {list(binary()), list(binary())}.
collect_pings(
    Collector,
    Queued,
    Pending,
    Workers,
    Reachable,
    Unreachable,
    Deadline
) ->
    case gleam@dict:is_empty(Pending) of
        true ->
            {Reachable, Unreachable};

        false ->
            Now = distribute_ffi_utils:monotonic_ms(),
            Timeout_ms = case Deadline > Now of
                true ->
                    Deadline - Now;

                false ->
                    0
            end,
            case gleam@erlang@process:'receive'(Collector, Timeout_ms) of
                {ok, {Node, true}} ->
                    case gleam@dict:has_key(Pending, Node) of
                        true ->
                            Next_pending = gleam@dict:delete(Pending, Node),
                            Next_workers = drop_ping_worker(Workers, Node),
                            {Next_queued, Next_workers@1} = maybe_spawn_next_worker(
                                Collector,
                                Queued,
                                Next_workers
                            ),
                            collect_pings(
                                Collector,
                                Next_queued,
                                Next_pending,
                                Next_workers@1,
                                [Node | Reachable],
                                Unreachable,
                                Deadline
                            );

                        false ->
                            collect_pings(
                                Collector,
                                Queued,
                                Pending,
                                Workers,
                                Reachable,
                                Unreachable,
                                Deadline
                            )
                    end;

                {ok, {Node@1, false}} ->
                    case gleam@dict:has_key(Pending, Node@1) of
                        true ->
                            Next_pending@1 = gleam@dict:delete(Pending, Node@1),
                            Next_workers@2 = drop_ping_worker(Workers, Node@1),
                            {Next_queued@1, Next_workers@3} = maybe_spawn_next_worker(
                                Collector,
                                Queued,
                                Next_workers@2
                            ),
                            collect_pings(
                                Collector,
                                Next_queued@1,
                                Next_pending@1,
                                Next_workers@3,
                                Reachable,
                                [Node@1 | Unreachable],
                                Deadline
                            );

                        false ->
                            collect_pings(
                                Collector,
                                Queued,
                                Pending,
                                Workers,
                                Reachable,
                                Unreachable,
                                Deadline
                            )
                    end;

                {error, nil} ->
                    terminate_ping_workers(Workers),
                    {Reachable, lists:append(Unreachable, maps:keys(Pending))}
            end
    end.

-file("src/distribute/cluster.gleam", 626).
-spec seed_ping_workers(
    gleam@erlang@process:subject({binary(), boolean()}),
    list(binary()),
    list({binary(), gleam@erlang@process:pid_()}),
    integer()
) -> {list(binary()), list({binary(), gleam@erlang@process:pid_()})}.
seed_ping_workers(Collector, Queue, Workers, Slots_left) ->
    case Slots_left =< 0 of
        true ->
            {Queue, Workers};

        false ->
            case Queue of
                [] ->
                    {Queue, Workers};

                [Node | Rest] ->
                    Pid = proc_lib:spawn(
                        fun() ->
                            gleam@erlang@process:send(
                                Collector,
                                {Node, cluster_ffi:ping(Node)}
                            )
                        end
                    ),
                    seed_ping_workers(
                        Collector,
                        Rest,
                        [{Node, Pid} | Workers],
                        Slots_left - 1
                    )
            end
    end.

-file("src/distribute/cluster.gleam", 345).
?DOC(false).
-spec health_parallelism_cap(integer()) -> integer().
health_parallelism_cap(Connected_count) ->
    case Connected_count =< 0 of
        true ->
            0;

        false ->
            case Connected_count < 32 of
                true ->
                    Connected_count;

                false ->
                    32
            end
    end.

-file("src/distribute/cluster.gleam", 391).
-spec parallel_partition_reachable(list(binary())) -> {list(binary()),
    list(binary())}.
parallel_partition_reachable(Connected) ->
    Result_subject = gleam@erlang@process:new_subject(),
    Proxy_pid = proc_lib:spawn(
        fun() ->
            Collector = gleam@erlang@process:new_subject(),
            Max_parallel = health_parallelism_cap(erlang:length(Connected)),
            {Queued, Workers} = seed_ping_workers(
                Collector,
                Connected,
                [],
                Max_parallel
            ),
            Deadline = distribute_ffi_utils:monotonic_ms() + 8000,
            Pending = gleam@list:fold(
                Connected,
                maps:new(),
                fun(Acc, Node) -> gleam@dict:insert(Acc, Node, nil) end
            ),
            Partition = collect_pings(
                Collector,
                Queued,
                Pending,
                Workers,
                [],
                [],
                Deadline
            ),
            gleam@erlang@process:send(Result_subject, Partition)
        end
    ),
    Proxy_mon = gleam@erlang@process:monitor(Proxy_pid),
    Selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        _pipe@1 = gleam@erlang@process:select_map(
            _pipe,
            Result_subject,
            fun(Field@0) -> {partition_ready, Field@0} end
        ),
        gleam@erlang@process:select_specific_monitor(
            _pipe@1,
            Proxy_mon,
            fun(_) -> proxy_exited end
        )
    end,
    case gleam_erlang_ffi:select(Selector, 8000 + 500) of
        {ok, {partition_ready, Partition@1}} ->
            gleam@erlang@process:demonitor_process(Proxy_mon),
            {Raw_reachable, _} = Partition@1,
            stable_partition(Connected, Raw_reachable);

        {ok, proxy_exited} ->
            {[], Connected};

        {error, nil} ->
            gleam@erlang@process:kill(Proxy_pid),
            Down_selector = begin
                _pipe@2 = gleam_erlang_ffi:new_selector(),
                gleam@erlang@process:select_specific_monitor(
                    _pipe@2,
                    Proxy_mon,
                    fun(_) -> nil end
                )
            end,
            case gleam_erlang_ffi:select(
                Down_selector,
                health_proxy_shutdown_grace_ms()
            ) of
                {ok, nil} ->
                    nil;

                {error, nil} ->
                    gleam@erlang@process:demonitor_process(Proxy_mon)
            end,
            drain_proxy_reply(Result_subject),
            {[], Connected}
    end.

-file("src/distribute/cluster.gleam", 299).
?DOC(
    " Perform a cluster health check, pinging each known node **in parallel**.\n"
    "\n"
    " See also: `has_peers/0` (boolean topology shortcut), `is_healthy/0`\n"
    " (compatibility alias), `is_distributed/0`,\n"
    " `ping/1` (single node).\n"
    "\n"
    " `net_adm:ping/1` is a synchronous network call with an implicit BEAM\n"
    " distribution timeout of several seconds. Pinging N nodes sequentially\n"
    " would block the caller for up to N * timeout_per_ping (e.g. 50 nodes\n"
    " during a partition = ~6 minutes). We fan out with bounded parallelism\n"
    " and collect results with a single 8 s deadline. Worst-case wall clock\n"
    " is still bounded by the deadline, not by cluster size.\n"
    "\n"
    " Output ordering is deterministic: `reachable_nodes` and\n"
    " `unreachable_nodes` are projected in the same order as\n"
    " `connected_nodes`, regardless of worker reply timing.\n"
).
-spec health() -> cluster_health().
health() ->
    Self = self_node(),
    Is_dist = is_distributed(),
    case Is_dist of
        false ->
            {cluster_health, Self, false, [], 0, [], []};

        true ->
            Connected = nodes(),
            {Reachable, Unreachable} = parallel_partition_reachable(Connected),
            {cluster_health,
                Self,
                true,
                Connected,
                erlang:length(Connected),
                Reachable,
                Unreachable}
    end.