distribute

-module(distribute@registry).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/registry.gleam").
-export([register/2, register_typed/2, register_subject/2, unregister/1, whereis/1, whereis_global/3, whereis_with_tag/2, whereis_typed/1, store_subject/2, remove_stored_subject/1, lookup_subject/1, register_global/2, register_with_strategy/3, register_with_retry/4, lookup_global/3, is_registered/1, unregister_and_remove/1, lookup_with_timeout/5]).
-export_type([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.

-type register_error() :: already_registered |
    invalid_process |
    {invalid_name, binary()} |
    {network_error, binary()} |
    {register_failed, binary()}.

-file("src/distribute/registry.gleam", 98).
?DOC(" Convert RegisterError to string for logging\n").
-spec classify_register_error_to_string(register_error()) -> binary().
classify_register_error_to_string(Error) ->
    case Error of
        already_registered ->
            <<"already_registered"/utf8>>;

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

        {invalid_name, Msg} ->
            <<"invalid_name: "/utf8, Msg/binary>>;

        {network_error, Msg@1} ->
            <<"network_error: "/utf8, Msg@1/binary>>;

        {register_failed, Msg@2} ->
            <<"register_failed: "/utf8, Msg@2/binary>>
    end.

-file("src/distribute/registry.gleam", 109).
?DOC(" Check if error is network-related\n").
-spec is_network_error(binary()) -> boolean().
is_network_error(Reason) ->
    (gleam_stdlib:contains_string(Reason, <<"partition"/utf8>>) orelse gleam_stdlib:contains_string(
        Reason,
        <<"network"/utf8>>
    ))
    orelse gleam_stdlib:contains_string(Reason, <<"connection"/utf8>>).

-file("src/distribute/registry.gleam", 84).
?DOC(" Classify error reason into structured RegisterError\n").
-spec classify_register_error(binary()) -> register_error().
classify_register_error(Reason) ->
    case Reason of
        <<"already_registered"/utf8>> ->
            already_registered;

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

        <<"invalid_name"/utf8>> ->
            {invalid_name, <<"Name contains invalid characters"/utf8>>};

        _ ->
            case is_network_error(Reason) of
                true ->
                    {network_error, Reason};

                false ->
                    {register_failed, Reason}
            end
    end.

-file("src/distribute/registry.gleam", 116).
?DOC(" Validate registry name\n").
-spec validate_name(binary()) -> {ok, nil} | {error, register_error()}.
validate_name(Name) ->
    case string:length(Name) of
        0 ->
            {error, {invalid_name, <<"Name cannot be empty"/utf8>>}};

        Len when Len > 255 ->
            {error, {invalid_name, <<"Name too long (max 255 chars)"/utf8>>}};

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

                false ->
                    {ok, nil}
            end
    end.

-file("src/distribute/registry.gleam", 159).
?DOC(" Register a process globally under the given name.\n").
-spec register(binary(), gleam@erlang@process:pid_()) -> {ok, nil} |
    {error, register_error()}.
register(Name, Pid) ->
    distribute@log:debug(
        <<"Registering global name"/utf8>>,
        [{<<"name"/utf8>>, Name}]
    ),
    case validate_name(Name) of
        {error, E} ->
            distribute@log:warn(
                <<"Name validation failed during registration"/utf8>>,
                [{<<"name"/utf8>>, Name},
                    {<<"error"/utf8>>, classify_register_error_to_string(E)}]
            ),
            {error, E};

        {ok, _} ->
            Res = registry_ffi:register(Name, Pid),
            case registry_ffi:is_ok_atom(Res) of
                true ->
                    distribute@log:info(
                        <<"Successfully registered global name"/utf8>>,
                        [{<<"name"/utf8>>, Name}]
                    ),
                    {ok, nil};

                false ->
                    case registry_ffi:is_already_registered(Res) of
                        true ->
                            distribute@log:warn(
                                <<"Name already registered"/utf8>>,
                                [{<<"name"/utf8>>, Name}]
                            ),
                            {error, already_registered};

                        false ->
                            Error = classify_register_error(
                                registry_ffi:get_error_reason(Res)
                            ),
                            distribute@log:error(
                                <<"Failed to register global name"/utf8>>,
                                [{<<"name"/utf8>>, Name},
                                    {<<"error"/utf8>>,
                                        classify_register_error_to_string(Error)}]
                            ),
                            {error, Error}
                    end
            end
    end.

-file("src/distribute/registry.gleam", 139).
?DOC(
    " Register a typed Subject globally.\n"
    "\n"
    " This registers the owner Pid of the subject.\n"
    "\n"
    " Recommended: Use global.GlobalSubject and register it with this function.\n"
    " The GlobalSubject pattern ensures type-safe messaging with encoder/decoder.\n"
    "\n"
    " For GlobalSubject: Create with global.new(encoder, decoder), then register\n"
    " with register_typed(name, global.subject(global_subject)).\n"
    "\n"
    " For custom Subject: Works but requires clients to know the message format.\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", 151).
?DOC(" Register a Subject globally (alias for register_typed).\n").
-spec register_subject(binary(), gleam@erlang@process:subject(any())) -> {ok,
        nil} |
    {error, register_error()}.
register_subject(Name, Subject) ->
    register_typed(Name, Subject).

-file("src/distribute/registry.gleam", 197).
?DOC(" Unregister a globally registered name.\n").
-spec unregister(binary()) -> {ok, nil} | {error, register_error()}.
unregister(Name) ->
    Res = registry_ffi:unregister(Name),
    case registry_ffi:is_ok_atom(Res) of
        true ->
            {ok, nil};

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

-file("src/distribute/registry.gleam", 207).
?DOC(
    " Look up a globally registered process by name.\n"
    " Returns Ok(pid) if found, Error(Nil) otherwise.\n"
).
-spec whereis(binary()) -> {ok, gleam@erlang@process:pid_()} | {error, nil}.
whereis(Name) ->
    distribute@log:debug(
        <<"Resolving global name"/utf8>>,
        [{<<"name"/utf8>>, 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", 223).
?DOC(
    " Look up a globally registered GlobalSubject (RECOMMENDED).\n"
    " \n"
    " This is the type-safe way to lookup distributed processes.\n"
    " Returns a GlobalSubject that enforces encoder/decoder usage.\n"
    "\n"
    " On success, returns a GlobalSubject that can send/receive typed messages.\n"
    " On error, returns Error(Nil) if the name is not registered.\n"
).
-spec whereis_global(
    binary(),
    fun((HJG) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HJG} | {error, distribute@codec:decode_error()})
) -> {ok, distribute@global:global_subject(HJG)} | {error, nil}.
whereis_global(Name, Encoder, Decoder) ->
    case whereis(Name) of
        {ok, Pid} ->
            {ok, distribute@global:from_pid(Pid, Encoder, Decoder)};

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

-file("src/distribute/registry.gleam", 241).
?DOC(
    " Look up a globally registered process with explicit tag.\n"
    "\n"
    " Use this when you know the tag of the remote process (e.g., for custom actors).\n"
    " For GlobalSubject, use whereis_global instead (recommended).\n"
    "\n"
    " The tag parameter should match the tag used by the remote actor.\n"
    " For most cases with Nil tags, use dynamic.nil() as the tag.\n"
).
-spec whereis_with_tag(binary(), gleam@dynamic:dynamic_()) -> {ok,
        gleam@erlang@process:subject(any())} |
    {error, nil}.
whereis_with_tag(Name, Tag) ->
    case whereis(Name) of
        {ok, Pid} ->
            {ok, gleam@erlang@process:unsafely_create_subject(Pid, Tag)};

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

-file("src/distribute/registry.gleam", 258).
?DOC(
    " Look up a globally registered process and return a typed Subject with Nil tag.\n"
    "\n"
    " ⚠️ **DEPRECATED**: Use `whereis_global` for GlobalSubject or `whereis_with_tag` \n"
    " for standard actors where you know the tag.\n"
    "\n"
    " The returned Subject has a `Nil` tag and won't work with standard gleam_otp actors.\n"
).
-spec whereis_typed(binary()) -> {ok, gleam@erlang@process:subject(any())} |
    {error, nil}.
whereis_typed(Name) ->
    whereis_with_tag(Name, gleam@dynamic:nil()).

-file("src/distribute/registry.gleam", 280).
?DOC(
    " Store a complete Subject by name.\n"
    "\n"
    " Unlike `register_typed` which only stores the Pid in Erlang's global registry,\n"
    " this function stores the entire Subject including its unique tag. This is\n"
    " essential for OTP actors where the tag is used for message pattern matching.\n"
    "\n"
    " Use this when you need to retrieve the exact same Subject later, for example\n"
    " when implementing singleton actors that need to be looked up by name.\n"
    "\n"
    " The Subject is stored using `persistent_term` which is optimized for\n"
    " read-heavy, rarely-changing data.\n"
).
-spec store_subject(binary(), gleam@erlang@process:subject(any())) -> {ok, nil} |
    {error, register_error()}.
store_subject(Name, Subject) ->
    case validate_name(Name) of
        {error, E} ->
            {error, E};

        {ok, _} ->
            _ = registry_ffi:store_subject(Name, Subject),
            {ok, nil}
    end.

-file("src/distribute/registry.gleam", 312).
?DOC(
    " Remove a stored Subject by name.\n"
    "\n"
    " This removes the Subject from persistent_term storage. Note that this\n"
    " does NOT unregister the process from Erlang's global registry - use\n"
    " `unregister` for that.\n"
).
-spec remove_stored_subject(binary()) -> nil.
remove_stored_subject(Name) ->
    _ = registry_ffi:remove_subject(Name),
    nil.

-file("src/distribute/registry.gleam", 299).
?DOC(
    " Retrieve a stored Subject by name.\n"
    "\n"
    " Returns the exact Subject that was stored with `store_subject`, preserving\n"
    " the original tag. This allows proper message routing for OTP actors.\n"
    "\n"
    " Returns Error(Nil) if no Subject is stored under this name.\n"
).
-spec lookup_subject(binary()) -> {ok, gleam@erlang@process:subject(any())} |
    {error, nil}.
lookup_subject(Name) ->
    Result = registry_ffi:get_subject(Name),
    case registry_ffi:is_ok_tuple(Result) of
        true ->
            {ok, registry_ffi:extract_subject(Result)};

        false ->
            {error, nil}
    end.

-file("src/distribute/registry.gleam", 338).
?DOC(
    " Try to register a GlobalSubject, automatically extracting its underlying Subject.\n"
    "\n"
    " This is a convenience wrapper that combines the most common pattern:\n"
    " 1. Extract the Subject(BitArray) from GlobalSubject\n"
    " 2. Register it using register_typed\n"
    "\n"
    " **Example:**\n"
    " ```gleam\n"
    " let global_subject = actor.start_typed_actor(init, loop, encoder, decoder)\n"
    " register_global(global_subject, \"my-service\")\n"
    " ```\n"
).
-spec register_global(distribute@global:global_subject(any()), binary()) -> {ok,
        nil} |
    {error, register_error()}.
register_global(Global_subject, Name) ->
    register_typed(Name, distribute@global:subject(Global_subject)).

-file("src/distribute/registry.gleam", 379).
-spec do_register_with_strategy(
    distribute@global:global_subject(any()),
    binary(),
    distribute@retry:retry_policy(),
    integer()
) -> {ok, nil} | {error, register_error()}.
do_register_with_strategy(Global_subject, Name, Policy, Attempt) ->
    case register_global(Global_subject, Name) of
        {ok, nil} ->
            case Attempt > 1 of
                true ->
                    distribute@log:info(
                        <<"Registration succeeded after retries"/utf8>>,
                        [{<<"name"/utf8>>, Name},
                            {<<"total_attempts"/utf8>>,
                                erlang:integer_to_binary(Attempt)}]
                    );

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

        {error, {network_error, Reason}} ->
            case distribute@retry:should_retry(Policy, Attempt) of
                true ->
                    Delay_result = distribute@retry:calculate_delay(
                        Policy,
                        Attempt
                    ),
                    distribute@log:warn(
                        <<"Registration attempt failed, retrying with backoff..."/utf8>>,
                        [{<<"name"/utf8>>, Name},
                            {<<"attempt"/utf8>>,
                                erlang:integer_to_binary(Attempt)},
                            {<<"max_attempts"/utf8>>,
                                erlang:integer_to_binary(
                                    distribute@retry:total_attempts(Policy)
                                )},
                            {<<"delay_ms"/utf8>>,
                                erlang:integer_to_binary(
                                    erlang:element(2, Delay_result)
                                )},
                            {<<"base_delay_ms"/utf8>>,
                                erlang:integer_to_binary(
                                    erlang:element(3, Delay_result)
                                )},
                            {<<"reason"/utf8>>, Reason}]
                    ),
                    gleam_erlang_ffi:sleep(erlang:element(2, Delay_result)),
                    do_register_with_strategy(
                        Global_subject,
                        Name,
                        Policy,
                        Attempt + 1
                    );

                false ->
                    distribute@log:error(
                        <<"Registration failed after max retries"/utf8>>,
                        [{<<"name"/utf8>>, Name},
                            {<<"attempt"/utf8>>,
                                erlang:integer_to_binary(Attempt)},
                            {<<"reason"/utf8>>, Reason}]
                    ),
                    {error, {network_error, Reason}}
            end;

        {error, Err} ->
            distribute@log:error(
                <<"Registration failed permanently"/utf8>>,
                [{<<"name"/utf8>>, Name},
                    {<<"attempt"/utf8>>, erlang:integer_to_binary(Attempt)},
                    {<<"error"/utf8>>, classify_register_error_to_string(Err)}]
            ),
            {error, Err}
    end.

-file("src/distribute/registry.gleam", 371).
?DOC(
    " Synchronous registration with retry logic and exponential backoff.\n"
    "\n"
    " Attempts to register a GlobalSubject using the provided retry policy.\n"
    " This is the recommended method for production use as it supports\n"
    " exponential backoff and jitter to prevent thundering herd problems.\n"
    "\n"
    " **Parameters:**\n"
    " - `global_subject`: The GlobalSubject to register\n"
    " - `name`: The global name to register under\n"
    " - `policy`: Retry policy with backoff configuration\n"
    "\n"
    " **Example:**\n"
    " ```gleam\n"
    " import distribute/retry\n"
    "\n"
    " // Recommended: exponential backoff with jitter\n"
    " let policy = retry.default_with_jitter()\n"
    " registry.register_with_strategy(subject, \"my-service\", policy)\n"
    "\n"
    " // Custom policy for critical services\n"
    " let critical_policy = retry.aggressive()\n"
    "   |> retry.with_max_attempts(10)\n"
    " registry.register_with_strategy(subject, \"critical-service\", critical_policy)\n"
    " ```\n"
    "\n"
    " Returns Ok(Nil) on success, Error(RegisterError) if all retries fail.\n"
).
-spec register_with_strategy(
    distribute@global:global_subject(any()),
    binary(),
    distribute@retry:retry_policy()
) -> {ok, nil} | {error, register_error()}.
register_with_strategy(Global_subject, Name, Policy) ->
    do_register_with_strategy(Global_subject, Name, Policy, 1).

-file("src/distribute/registry.gleam", 451).
?DOC(
    " Synchronous registration with simple retry logic.\n"
    "\n"
    " **DEPRECATED:** Use `register_with_strategy` with a retry policy instead.\n"
    " This function uses fixed delays which can cause thundering herd problems.\n"
    "\n"
    " Attempts to register a GlobalSubject, retrying up to `max_retries` times\n"
    " if network errors occur. Useful in distributed environments with transient\n"
    " connectivity issues.\n"
    "\n"
    " **Parameters:**\n"
    " - `global_subject`: The GlobalSubject to register\n"
    " - `name`: The global name to register under\n"
    " - `max_retries`: Maximum number of retry attempts (default: 3)\n"
    " - `retry_delay_ms`: Delay between retries in milliseconds (default: 100)\n"
    "\n"
    " Returns Ok(Nil) on success, Error(RegisterError) if all retries fail.\n"
).
-spec register_with_retry(
    distribute@global:global_subject(any()),
    binary(),
    integer(),
    integer()
) -> {ok, nil} | {error, register_error()}.
register_with_retry(Global_subject, Name, Max_retries, Retry_delay_ms) ->
    Policy = begin
        _pipe = distribute@retry:default(),
        _pipe@1 = distribute@retry:with_max_attempts(_pipe, Max_retries + 1),
        _pipe@2 = distribute@retry:with_base_delay_ms(_pipe@1, Retry_delay_ms),
        _pipe@3 = distribute@retry:with_max_delay_ms(_pipe@2, Retry_delay_ms),
        distribute@retry:with_jitter(_pipe@3, no_jitter)
    end,
    do_register_with_strategy(Global_subject, Name, Policy, 1).

-file("src/distribute/registry.gleam", 480).
?DOC(
    " Lookup a GlobalSubject by name (async-friendly version).\n"
    "\n"
    " This is a convenience wrapper that simplifies the common pattern of\n"
    " looking up a GlobalSubject with encoder/decoder.\n"
    "\n"
    " **Example:**\n"
    " ```gleam\n"
    " case lookup_global(\"my-service\", my_encoder(), my_decoder()) {\n"
    "   Ok(service) -> global.send(service, MyMessage)\n"
    "   Error(_) -> io.println(\"Service not found\")\n"
    " }\n"
    " ```\n"
).
-spec lookup_global(
    binary(),
    fun((HKU) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HKU} | {error, distribute@codec:decode_error()})
) -> {ok, distribute@global:global_subject(HKU)} | {error, nil}.
lookup_global(Name, Encoder, Decoder) ->
    whereis_global(Name, Encoder, Decoder).

-file("src/distribute/registry.gleam", 559).
?DOC(
    " Check if a name is currently registered.\n"
    "\n"
    " This is more efficient than whereis when you only need to check\n"
    " existence without creating a Subject.\n"
).
-spec is_registered(binary()) -> boolean().
is_registered(Name) ->
    case whereis(Name) of
        {ok, _} ->
            true;

        {error, _} ->
            false
    end.

-file("src/distribute/registry.gleam", 573).
?DOC(
    " Unregister and remove stored subject in one call.\n"
    "\n"
    " Convenience wrapper that combines:\n"
    " 1. Unregister from global registry\n"
    " 2. Remove from persistent_term storage\n"
    "\n"
    " Useful for complete cleanup of a named actor.\n"
).
-spec unregister_and_remove(binary()) -> {ok, nil} | {error, register_error()}.
unregister_and_remove(Name) ->
    _ = remove_stored_subject(Name),
    unregister(Name).

-file("src/distribute/registry.gleam", 585).
-spec erlang_system_time_ms() -> integer().
erlang_system_time_ms() ->
    erlang:system_time(registry_ffi:millisecond_atom()).

-file("src/distribute/registry.gleam", 519).
-spec do_lookup_with_timeout(
    binary(),
    fun((HLG) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HLG} | {error, distribute@codec:decode_error()}),
    integer(),
    integer(),
    integer()
) -> {ok, distribute@global:global_subject(HLG)} | {error, nil}.
do_lookup_with_timeout(
    Name,
    Encoder,
    Decoder,
    Start_time,
    Timeout_ms,
    Poll_interval_ms
) ->
    case lookup_global(Name, Encoder, Decoder) of
        {ok, Subject} ->
            {ok, Subject};

        {error, _} ->
            Elapsed = erlang_system_time_ms() - Start_time,
            case Elapsed >= Timeout_ms of
                true ->
                    distribute@log:warn(
                        <<"Lookup timeout exceeded"/utf8>>,
                        [{<<"name"/utf8>>, Name},
                            {<<"timeout_ms"/utf8>>,
                                erlang:integer_to_binary(Timeout_ms)}]
                    ),
                    {error, nil};

                false ->
                    gleam_erlang_ffi:sleep(Poll_interval_ms),
                    do_lookup_with_timeout(
                        Name,
                        Encoder,
                        Decoder,
                        Start_time,
                        Timeout_ms,
                        Poll_interval_ms
                    )
            end
    end.

-file("src/distribute/registry.gleam", 501).
?DOC(
    " Synchronous lookup with timeout.\n"
    "\n"
    " Attempts to lookup a GlobalSubject, retrying until found or timeout.\n"
    " Useful when waiting for a service to become available.\n"
    "\n"
    " **Parameters:**\n"
    " - `name`: The global name to lookup\n"
    " - `encoder`: Encoder for the message type\n"
    " - `decoder`: Decoder for the message type\n"
    " - `timeout_ms`: Maximum time to wait in milliseconds\n"
    " - `poll_interval_ms`: Time between lookup attempts in milliseconds\n"
    "\n"
    " Returns Ok(GlobalSubject) if found, Error(Nil) on timeout.\n"
).
-spec lookup_with_timeout(
    binary(),
    fun((HLA) -> {ok, bitstring()} | {error, distribute@codec:encode_error()}),
    fun((bitstring()) -> {ok, HLA} | {error, distribute@codec:decode_error()}),
    integer(),
    integer()
) -> {ok, distribute@global:global_subject(HLA)} | {error, nil}.
lookup_with_timeout(Name, Encoder, Decoder, Timeout_ms, Poll_interval_ms) ->
    Start_time = erlang_system_time_ms(),
    do_lookup_with_timeout(
        Name,
        Encoder,
        Decoder,
        Start_time,
        Timeout_ms,
        Poll_interval_ms
    ).