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(GIK) :: {typed_name,
        binary(),
        fun((GIK) -> {ok, bitstring()} |
            {error, distribute@codec:encode_error()}),
        fun((bitstring()) -> {ok, GIK} |
            {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", 41).
?DOC(
    " Create a typed name.\n"
    "\n"
    " Share this value between registration and lookup so the compiler\n"
    " can check that both sides use the same message type.\n"
).
-spec typed_name(
    binary(),
    fun((GIL) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, GIL} | {error, distribute@codec:decode_error()})
) -> typed_name(GIL).
typed_name(Name, Encoder, Decoder) ->
    {typed_name, Name, Encoder, Decoder}.

-file("src/distribute/registry.gleam", 54).
?DOC(
    " Same as `typed_name` but takes a bundled `Codec`.\n"
    "\n"
    " ```gleam\n"
    " let counter = registry.named(\"counter\", codec.int())\n"
    " ```\n"
).
-spec named(binary(), distribute@codec:codec(GIP)) -> typed_name(GIP).
named(Name, C) ->
    {typed_name, Name, erlang:element(2, C), erlang:element(3, C)}.

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

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

-file("src/distribute/registry.gleam", 77).
?DOC(
    " Create a `TypedName` for a pool member.\n"
    "\n"
    " Given a base `TypedName` with name `\"worker\"` and index `2`,\n"
    " returns a `TypedName` with name `\"worker_2\"` and the same codecs.\n"
).
-spec pool_member(typed_name(GJA), integer()) -> typed_name(GJA).
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", 177).
?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", 183).
?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", 195).
?DOC(
    " Look up a globally registered `GlobalSubject` by `TypedName`.\n"
    "\n"
    " Reconstructs the Subject with a deterministic tag derived from\n"
    " the name.\n"
).
-spec lookup(typed_name(GJS)) -> {ok, distribute@global:global_subject(GJS)} |
    {error, nil}.
lookup(Tn) ->
    case whereis(erlang:element(2, Tn)) of
        {ok, Pid} ->
            {ok,
                distribute@global:from_name(
                    erlang:element(2, Tn),
                    Pid,
                    erlang:element(3, Tn),
                    erlang:element(4, Tn)
                )};

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

-file("src/distribute/registry.gleam", 203).
?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", 223).
-spec do_lookup_with_timeout(typed_name(GKC), integer(), integer(), integer()) -> {ok,
        distribute@global:global_subject(GKC)} |
    {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", 214).
?DOC(
    " Look up a `GlobalSubject` with polling and timeout.\n"
    "\n"
    " Retries the lookup every `poll_interval_ms` until found or\n"
    " `timeout_ms` elapses.\n"
).
-spec lookup_with_timeout(typed_name(GJX), integer(), integer()) -> {ok,
        distribute@global:global_subject(GJX)} |
    {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", 248).
-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", 263).
-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", 135).
?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, _} ->
            Res = registry_ffi:register(Name, Pid),
            case registry_ffi:is_ok_atom(Res) of
                true ->
                    {ok, nil};

                false ->
                    case registry_ffi:is_already_registered(Res) of
                        true ->
                            {error, already_exists};

                        false ->
                            {error,
                                classify_error(
                                    registry_ffi:get_error_reason(Res)
                                )}
                    end
            end
    end.

-file("src/distribute/registry.gleam", 153).
?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", 167).
?DOC(
    " Register a `GlobalSubject` under a typed name.\n"
    "\n"
    " The `msg` type parameter on `TypedName(msg)` and `GlobalSubject(msg)`\n"
    " must match — the compiler enforces this.\n"
).
-spec register_global(typed_name(GJJ), distribute@global:global_subject(GJJ)) -> {ok,
        nil} |
    {error, register_error()}.
register_global(Tn, Global_subject) ->
    register_typed(
        erlang:element(2, Tn),
        distribute@global:subject(Global_subject)
    ).