distribute

-module(distribute@cluster@monitor).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/cluster/monitor.gleam").
-export([subscribe/1, unsubscribe/1]).
-export_type([cluster_event/0, control_message/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 cluster_event() :: {node_up, binary()} | {node_down, binary()}.

-type control_message() :: stop | {internal_event, cluster_event()}.

-file("src/distribute/cluster/monitor.gleam", 35).
?DOC(
    " Subscribe to cluster topology events.\n"
    " Events are forwarded to the provided user_subject.\n"
    " Returns a control subject to stop the subscription.\n"
).
-spec subscribe(gleam@erlang@process:subject(cluster_event())) -> {ok,
        gleam@erlang@process:subject(control_message())} |
    {error, gleam@otp@actor:start_error()}.
subscribe(User_subject) ->
    _pipe@5 = gleam@otp@actor:new_with_initialiser(
        5000,
        fun(Self_subject) ->
            cluster_ffi:monitor_nodes(true),
            Selector = begin
                _pipe = gleam_erlang_ffi:new_selector(),
                _pipe@1 = gleam@erlang@process:select(_pipe, Self_subject),
                _pipe@2 = gleam@erlang@process:select_record(
                    _pipe@1,
                    cluster_ffi:nodeup_atom(),
                    1,
                    fun(Msg) ->
                        Node_atom = cluster_ffi:get_node_from_tuple(Msg),
                        {internal_event,
                            {node_up, cluster_ffi:atom_to_string(Node_atom)}}
                    end
                ),
                gleam@erlang@process:select_record(
                    _pipe@2,
                    cluster_ffi:nodedown_atom(),
                    1,
                    fun(Msg@1) ->
                        Node_atom@1 = cluster_ffi:get_node_from_tuple(Msg@1),
                        {internal_event,
                            {node_down, cluster_ffi:atom_to_string(Node_atom@1)}}
                    end
                )
            end,
            {ok,
                begin
                    _pipe@3 = gleam@otp@actor:initialised(User_subject),
                    _pipe@4 = gleam@otp@actor:selecting(_pipe@3, Selector),
                    gleam@otp@actor:returning(_pipe@4, Self_subject)
                end}
        end
    ),
    _pipe@6 = gleam@otp@actor:on_message(
        _pipe@5,
        fun(State, Msg@2) -> case Msg@2 of
                {internal_event, Event} ->
                    case Event of
                        {node_up, Node} ->
                            distribute@internal@telemetry:emit_node_event(
                                Node,
                                <<"up"/utf8>>
                            );

                        {node_down, Node@1} ->
                            distribute@internal@telemetry:emit_node_event(
                                Node@1,
                                <<"down"/utf8>>
                            )
                    end,
                    gleam@erlang@process:send(State, Event),
                    gleam@otp@actor:continue(State);

                stop ->
                    gleam@otp@actor:stop()
            end end
    ),
    _pipe@7 = gleam@otp@actor:start(_pipe@6),
    gleam@result:map(_pipe@7, fun(Started) -> erlang:element(3, Started) end).

-file("src/distribute/cluster/monitor.gleam", 79).
?DOC(" Stop the monitoring for a specific subscription.\n").
-spec unsubscribe(gleam@erlang@process:subject(control_message())) -> nil.
unsubscribe(Monitor_subject) ->
    gleam@erlang@process:send(Monitor_subject, stop).