distribute

-module(distribute@registry).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/registry.gleam").
-export([typed_name/3, named/2, typed_name_to_string/1, typed_name_encoder/1, typed_name_decoder/1, pool_member/2, unregister/1, whereis/1, lookup/1, is_registered/1, lookup_with_timeout/3, register/2, register_typed/2, register_global/2]).
-export_type([typed_name/1, register_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.

-opaque typed_name(GKC) :: {typed_name,
        binary(),
        fun((GKC) -> {ok, bitstring()} |
            {error, distribute@codec:encode_error()}),
        fun((bitstring()) -> {ok, GKC} |
            {error, distribute@codec:decode_error()})}.

-type register_error() :: already_exists |
    invalid_process |
    {invalid_argument, binary()} |
    {network_error, binary()} |
    {registration_failed, binary()}.

-file("src/distribute/registry.gleam", 27).
?DOC(" Create a typed name from separate encoder and decoder.\n").
-spec typed_name(
    binary(),
    fun((GKD) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, GKD} | {error, distribute@codec:decode_error()})
) -> typed_name(GKD).
typed_name(Name, Encoder, Decoder) ->
    {typed_name, Name, Encoder, Decoder}.

-file("src/distribute/registry.gleam", 36).
?DOC(" Create a typed name from a bundled Codec.\n").
-spec named(binary(), distribute@codec:codec(GKH)) -> typed_name(GKH).
named(Name, C) ->
    {typed_name, Name, erlang:element(2, C), erlang:element(3, C)}.

-file("src/distribute/registry.gleam", 41).
?DOC(" Get the name string from a `TypedName`.\n").
-spec typed_name_to_string(typed_name(any())) -> binary().
typed_name_to_string(Tn) ->
    erlang:element(2, Tn).

-file("src/distribute/registry.gleam", 46).
?DOC(" Get the encoder from a `TypedName`.\n").
-spec typed_name_encoder(typed_name(GKM)) -> fun((GKM) -> {ok, bitstring()} |
    {error, distribute@codec:encode_error()}).
typed_name_encoder(Tn) ->
    erlang:element(3, Tn).

-file("src/distribute/registry.gleam", 51).
?DOC(" Get the decoder from a `TypedName`.\n").
-spec typed_name_decoder(typed_name(GKP)) -> fun((bitstring()) -> {ok, GKP} |
    {error, distribute@codec:decode_error()}).
typed_name_decoder(Tn) ->
    erlang:element(4, Tn).

-file("src/distribute/registry.gleam", 56).
?DOC(" Derive a pool member name by appending the index.\n").
-spec pool_member(typed_name(GKS), integer()) -> typed_name(GKS).
pool_member(Base, Index) ->
    {typed_name,
        <<<<(erlang:element(2, Base))/binary, "_"/utf8>>/binary,
            (erlang:integer_to_binary(Index))/binary>>,
        erlang:element(3, Base),
        erlang:element(4, Base)}.

-file("src/distribute/registry.gleam", 165).
?DOC(
    " Unregister a global name.\n"
    "\n"
    " Always succeeds — unregistering a non-existent name is a no-op.\n"
).
-spec unregister(binary()) -> {ok, nil} | {error, register_error()}.
unregister(Name) ->
    _ = registry_ffi:unregister(Name),
    {ok, nil}.

-file("src/distribute/registry.gleam", 171).
?DOC(" Look up a globally registered PID by name.\n").
-spec whereis(binary()) -> {ok, gleam@erlang@process:pid_()} | {error, nil}.
whereis(Name) ->
    Res = registry_ffi:whereis(Name),
    case registry_ffi:is_pid(Res) of
        true ->
            {ok, registry_ffi:dynamic_to_pid(Res)};

        false ->
            {error, nil}
    end.

-file("src/distribute/registry.gleam", 181).
?DOC(
    " Look up a globally registered GlobalSubject by TypedName.\n"
    " Reconstructs the Subject with a deterministic tag from the name.\n"
).
-spec lookup(typed_name(GLK)) -> {ok, distribute@global:global_subject(GLK)} |
    {error, nil}.
lookup(Tn) ->
    Start = distribute_ffi_utils:system_time_ms(),
    case whereis(erlang:element(2, Tn)) of
        {ok, Pid} ->
            distribute@internal@telemetry:emit_registry_lookup_stop(
                erlang:element(2, Tn),
                true,
                distribute_ffi_utils:system_time_ms() - Start
            ),
            {ok,
                distribute@global:from_name(
                    erlang:element(2, Tn),
                    Pid,
                    erlang:element(3, Tn),
                    erlang:element(4, Tn)
                )};

        {error, nil} ->
            distribute@internal@telemetry:emit_registry_lookup_stop(
                erlang:element(2, Tn),
                false,
                distribute_ffi_utils:system_time_ms() - Start
            ),
            {error, nil}
    end.

-file("src/distribute/registry.gleam", 204).
?DOC(" Check whether a name is currently registered.\n").
-spec is_registered(binary()) -> boolean().
is_registered(Name) ->
    case whereis(Name) of
        {ok, _} ->
            true;

        {error, _} ->
            false
    end.

-file("src/distribute/registry.gleam", 222).
-spec do_lookup_with_timeout(typed_name(GLU), integer(), integer(), integer()) -> {ok,
        distribute@global:global_subject(GLU)} |
    {error, nil}.
do_lookup_with_timeout(Tn, Start, Timeout_ms, Poll_interval_ms) ->
    case lookup(Tn) of
        {ok, Gs} ->
            {ok, Gs};

        {error, _} ->
            Elapsed = distribute_ffi_utils:system_time_ms() - Start,
            case Elapsed >= Timeout_ms of
                true ->
                    {error, nil};

                false ->
                    gleam_erlang_ffi:sleep(Poll_interval_ms),
                    do_lookup_with_timeout(
                        Tn,
                        Start,
                        Timeout_ms,
                        Poll_interval_ms
                    )
            end
    end.

-file("src/distribute/registry.gleam", 213).
?DOC(
    " Look up with polling. Retries every poll_interval_ms until\n"
    " found or timeout_ms elapses.\n"
).
-spec lookup_with_timeout(typed_name(GLP), integer(), integer()) -> {ok,
        distribute@global:global_subject(GLP)} |
    {error, nil}.
lookup_with_timeout(Tn, Timeout_ms, Poll_interval_ms) ->
    Start = distribute_ffi_utils:system_time_ms(),
    do_lookup_with_timeout(Tn, Start, Timeout_ms, Poll_interval_ms).

-file("src/distribute/registry.gleam", 247).
-spec validate_name(binary()) -> {ok, nil} | {error, register_error()}.
validate_name(Name) ->
    case string:length(Name) of
        0 ->
            {error, {invalid_argument, <<"Name cannot be empty"/utf8>>}};

        Len ->
            case Len > 255 of
                true ->
                    {error,
                        {invalid_argument,
                            <<"Name too long (max 255 chars)"/utf8>>}};

                false ->
                    case gleam_stdlib:contains_string(Name, <<" "/utf8>>) of
                        true ->
                            {error,
                                {invalid_argument,
                                    <<"Name cannot contain spaces"/utf8>>}};

                        false ->
                            {ok, nil}
                    end
            end
    end.

-file("src/distribute/registry.gleam", 262).
-spec classify_error(binary()) -> register_error().
classify_error(Reason) ->
    case Reason of
        <<"already_registered"/utf8>> ->
            already_exists;

        <<"not_a_pid"/utf8>> ->
            invalid_process;

        _ ->
            case (gleam_stdlib:contains_string(Reason, <<"partition"/utf8>>)
            orelse gleam_stdlib:contains_string(Reason, <<"network"/utf8>>))
            orelse gleam_stdlib:contains_string(Reason, <<"connection"/utf8>>) of
                true ->
                    {network_error, Reason};

                false ->
                    {registration_failed, Reason}
            end
    end.

-file("src/distribute/registry.gleam", 114).
?DOC(" Register a PID under a global name.\n").
-spec register(binary(), gleam@erlang@process:pid_()) -> {ok, nil} |
    {error, register_error()}.
register(Name, Pid) ->
    case validate_name(Name) of
        {error, E} ->
            {error, E};

        {ok, _} ->
            Start = distribute_ffi_utils:system_time_ms(),
            Res = registry_ffi:register(Name, Pid),
            Duration = distribute_ffi_utils:system_time_ms() - Start,
            case registry_ffi:is_ok_atom(Res) of
                true ->
                    distribute@internal@telemetry:emit_registry_register_stop(
                        Name,
                        true,
                        Duration
                    ),
                    {ok, nil};

                false ->
                    case registry_ffi:is_already_registered(Res) of
                        true ->
                            distribute@internal@telemetry:emit_registry_register_stop(
                                Name,
                                false,
                                Duration
                            ),
                            {error, already_exists};

                        false ->
                            distribute@internal@telemetry:emit_registry_register_stop(
                                Name,
                                false,
                                Duration
                            ),
                            {error,
                                classify_error(
                                    registry_ffi:get_error_reason(Res)
                                )}
                    end
            end
    end.

-file("src/distribute/registry.gleam", 143).
?DOC(" Register a `Subject`'s owner PID under a global name.\n").
-spec register_typed(binary(), gleam@erlang@process:subject(any())) -> {ok, nil} |
    {error, register_error()}.
register_typed(Name, Subject) ->
    case gleam@erlang@process:subject_owner(Subject) of
        {ok, Pid} ->
            register(Name, Pid);

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

-file("src/distribute/registry.gleam", 155).
?DOC(
    " Register a GlobalSubject under a typed name.\n"
    " The compiler enforces that both sides share the same msg type.\n"
).
-spec register_global(typed_name(GLB), distribute@global:global_subject(GLB)) -> {ok,
        nil} |
    {error, register_error()}.
register_global(Tn, Global_subject) ->
    register_typed(
        erlang:element(2, Tn),
        distribute@global:subject(Global_subject)
    ).