distribute

-module(distribute@actor).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/distribute/actor.gleam").
-export([start_observed/5, start_registered_observed/5, start/4, start_default/3, start_registered/4, start_registered_default/3, child_spec/4, child_spec_default/3, start_registered_error_to_string/1, start_supervised/4, start_supervised_default/3, pool/5, pool_default/4, start_resource_owner/3]).
-export_type([start_registered_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.

?MODULEDOC(
    " Named actor lifecycle: start, register, supervise, pool.\n"
    "\n"
    " ## Resource cleanup and the `{terminate, Reason}` gap\n"
    "\n"
    " `gleam/otp/actor` 1.x does **not** implement OTP's\n"
    " `{terminate, Reason}` system message\n"
    " (see `gleam_otp_external.erl`; there is an explicit `TODO`).\n"
    " That has two consequences for any actor that owns external\n"
    " resources (file handles, ETS tables, ports, locks, sockets):\n"
    "\n"
    " 1. **Handler-controlled exits run user code first.** If your\n"
    "    handler returns `receiver.Stop` or `receiver.StopAbnormal`,\n"
    "    you have full control: release every resource you own *before*\n"
    "    returning the stop variant. The actor will then exit and the\n"
    "    BEAM frees its mailbox.\n"
    " 2. **External terminations skip the actor.** When the actor is\n"
    "    `process.kill`-ed (uncatchable), or when the supervisor sends\n"
    "    `exit(Pid, shutdown)`, gleam/otp/actor does not currently\n"
    "    convert it into a callback. Resources owned only on the\n"
    "    actor's heap are released by the BEAM (memory), but any\n"
    "    *external* handle (a file, a TCP connection, an ETS table you\n"
    "    forgot to make public) leaks.\n"
    "\n"
    " **The OTP-pure mitigation** is the \"linked resource owner\"\n"
    " pattern: spawn a tiny dedicated process to own the external\n"
    " resource, monitor the actor, and run a `close` callback when the\n"
    " actor dies for any reason. The library ships\n"
    " `actor.start_resource_owner/3` as a ready-made helper for this\n"
    " pattern. See `docs/recipes.md` for the underlying recipe.\n"
    "\n"
    " When upstream gleam/otp ships termination callbacks (tracking\n"
    " issue: gleam-lang/otp#126) the helper becomes redundant for the\n"
    " actor case. Its API is shaped so the migration is a one-line\n"
    " change in user code and a deprecation here, no semantics shift.\n"
).

-type start_registered_error() :: {actor_start_failed,
        gleam@otp@actor:start_error()} |
    {global_register_failed, distribute@registry:register_error()}.

-file("src/distribute/actor.gleam", 84).
?DOC(
    " Two-phase orphan termination, mirroring `supervisor:terminate_child/2`.\n"
    "\n"
    " 1. Unlink so the brutal phase (if reached) does not propagate to the\n"
    "    caller via the default actor link.\n"
    " 2. Send `exit(Pid, shutdown)`. Non-trapping actors die immediately\n"
    "    with reason `shutdown`; the monitor we set fires within the grace\n"
    "    window and we return cleanly. Trapping actors receive\n"
    "    `{'EXIT', From, shutdown}` as a mailbox message.\n"
    " 3. If the orphan is still alive after `orphan_shutdown_grace_ms`,\n"
    "    escalate to `process.kill` (uncatchable).\n"
    "\n"
    " ## Gleam OTP caveat\n"
    "\n"
    " `gleam/otp/actor` (as of 1.x) does not yet implement the\n"
    " `{terminate, Reason}` system message (see TODO in\n"
    " `gleam_otp_external.erl`), so a true OTP-style graceful shutdown via\n"
    " `sys:terminate/2` or `gen:stop/3` is currently a no-op for those\n"
    " actors. The system message is logged as \"unexpected\" and the actor\n"
    " keeps running until the kill phase. We use the simpler\n"
    " `exit(Pid, shutdown)` path: non-trapping actors die immediately,\n"
    " trapping actors fall through to brutal kill within the grace window.\n"
    " When upstream `gleam/otp` ships terminate support, switching to\n"
    " `gen:stop` here will give trapping actors a real `terminate` callback\n"
    " invocation without changing this function's contract.\n"
    "\n"
    " Used by `start_registered`, `start_registered_observed`, and `child_spec`\n"
    " when global registration fails after a successful actor init.\n"
).
-spec terminate_orphan_gracefully(gleam@erlang@process:pid_()) -> nil.
terminate_orphan_gracefully(Pid) ->
    gleam@erlang@process:unlink(Pid),
    Mon = gleam@erlang@process:monitor(Pid),
    _ = distribute_ffi_utils:exit_shutdown(Pid),
    Selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        gleam@erlang@process:select_specific_monitor(
            _pipe,
            Mon,
            fun(_) -> nil end
        )
    end,
    case gleam_erlang_ffi:select(Selector, 100) of
        {ok, nil} ->
            nil;

        {error, nil} ->
            gleam@erlang@process:kill(Pid),
            gleam@erlang@process:demonitor_process(Mon),
            distribute@telemetry:emit({orphan_kill_escalated, Pid})
    end.

-file("src/distribute/actor.gleam", 110).
?DOC(
    " Start a named actor, with a callback for decode errors.\n"
    "\n"
    " Useful for logging or metering malformed messages across nodes\n"
    " (e.g. during rolling deploys with mismatched codec versions).\n"
).
-spec start_observed(
    distribute@registry:typed_name(HBH),
    HBJ,
    fun((HBH, HBJ) -> distribute@receiver:handler_step(HBJ)),
    integer(),
    fun((distribute@codec:decode_error()) -> nil)
) -> {ok, distribute@global:global_subject(HBH)} |
    {error, gleam@otp@actor:start_error()}.
start_observed(
    Typed_name,
    Initial_state,
    Handler,
    Init_timeout_ms,
    On_decode_error
) ->
    case distribute@receiver:start_distributed_worker_observed(
        distribute@registry:typed_name_to_string(Typed_name),
        Initial_state,
        distribute@registry:typed_name_encoder(Typed_name),
        distribute@registry:typed_name_decoder(Typed_name),
        Handler,
        On_decode_error,
        Init_timeout_ms
    ) of
        {ok, Started} ->
            {ok, erlang:element(3, Started)};

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

-file("src/distribute/actor.gleam", 134).
?DOC(" Start a named actor and register it globally, with a callback for decode errors.\n").
-spec start_registered_observed(
    distribute@registry:typed_name(HBO),
    HBQ,
    fun((HBO, HBQ) -> distribute@receiver:handler_step(HBQ)),
    integer(),
    fun((distribute@codec:decode_error()) -> nil)
) -> {ok, distribute@global:global_subject(HBO)} |
    {error, start_registered_error()}.
start_registered_observed(
    Typed_name,
    Initial_state,
    Handler,
    Init_timeout_ms,
    On_decode_error
) ->
    Name = distribute@registry:typed_name_to_string(Typed_name),
    case distribute@receiver:start_distributed_worker_observed(
        Name,
        Initial_state,
        distribute@registry:typed_name_encoder(Typed_name),
        distribute@registry:typed_name_decoder(Typed_name),
        Handler,
        On_decode_error,
        Init_timeout_ms
    ) of
        {ok, Started} ->
            case distribute@registry:register_global(
                Typed_name,
                erlang:element(3, Started)
            ) of
                {ok, _} ->
                    {ok, erlang:element(3, Started)};

                {error, Err} ->
                    terminate_orphan_gracefully(erlang:element(2, Started)),
                    {error, {global_register_failed, Err}}
            end;

        {error, Err@1} ->
            {error, {actor_start_failed, Err@1}}
    end.

-file("src/distribute/actor.gleam", 167).
?DOC(" Start a named actor.\n").
-spec start(
    distribute@registry:typed_name(HBV),
    HBX,
    fun((HBV, HBX) -> distribute@receiver:handler_step(HBX)),
    integer()
) -> {ok, distribute@global:global_subject(HBV)} |
    {error, gleam@otp@actor:start_error()}.
start(Typed_name, Initial_state, Handler, Init_timeout_ms) ->
    case distribute@receiver:start_distributed_worker(
        distribute@registry:typed_name_to_string(Typed_name),
        Initial_state,
        distribute@registry:typed_name_encoder(Typed_name),
        distribute@registry:typed_name_decoder(Typed_name),
        Handler,
        Init_timeout_ms
    ) of
        {ok, Started} ->
            {ok, erlang:element(3, Started)};

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

-file("src/distribute/actor.gleam", 189).
?DOC(" Like `start`, but uses `config.get().default_init_timeout_ms` as the init timeout.\n").
-spec start_default(
    distribute@registry:typed_name(HCC),
    HCE,
    fun((HCC, HCE) -> distribute@receiver:handler_step(HCE))
) -> {ok, distribute@global:global_subject(HCC)} |
    {error, gleam@otp@actor:start_error()}.
start_default(Typed_name, Initial_state, Handler) ->
    start(
        Typed_name,
        Initial_state,
        Handler,
        erlang:element(3, distribute@config:get())
    ).

-file("src/distribute/actor.gleam", 306).
?DOC(
    " Start an actor and register it globally. Kills the actor if\n"
    " registration fails.\n"
    "\n"
    " See also: `start_registered_default/3` (configured init timeout),\n"
    " `start_registered_observed/5` (decode-error hook),\n"
    " `start_supervised/4` (auto-restart on crash).\n"
).
-spec start_registered(
    distribute@registry:typed_name(HDC),
    HDE,
    fun((HDC, HDE) -> distribute@receiver:handler_step(HDE)),
    integer()
) -> {ok, distribute@global:global_subject(HDC)} |
    {error, start_registered_error()}.
start_registered(Typed_name, Initial_state, Handler, Init_timeout_ms) ->
    Name = distribute@registry:typed_name_to_string(Typed_name),
    case distribute@receiver:start_distributed_worker(
        Name,
        Initial_state,
        distribute@registry:typed_name_encoder(Typed_name),
        distribute@registry:typed_name_decoder(Typed_name),
        Handler,
        Init_timeout_ms
    ) of
        {ok, Started} ->
            case distribute@registry:register_global(
                Typed_name,
                erlang:element(3, Started)
            ) of
                {ok, _} ->
                    {ok, erlang:element(3, Started)};

                {error, Err} ->
                    terminate_orphan_gracefully(erlang:element(2, Started)),
                    {error, {global_register_failed, Err}}
            end;

        {error, Err@1} ->
            {error, {actor_start_failed, Err@1}}
    end.

-file("src/distribute/actor.gleam", 203).
?DOC(" Like `start_registered`, but uses `config.get().default_init_timeout_ms` as the init timeout.\n").
-spec start_registered_default(
    distribute@registry:typed_name(HCJ),
    HCL,
    fun((HCJ, HCL) -> distribute@receiver:handler_step(HCL))
) -> {ok, distribute@global:global_subject(HCJ)} |
    {error, start_registered_error()}.
start_registered_default(Typed_name, Initial_state, Handler) ->
    start_registered(
        Typed_name,
        Initial_state,
        Handler,
        erlang:element(3, distribute@config:get())
    ).

-file("src/distribute/actor.gleam", 231).
?DOC(" OTP child spec for a named actor that auto-registers on (re)start.\n").
-spec child_spec(
    distribute@registry:typed_name(HCW),
    HCY,
    fun((HCW, HCY) -> distribute@receiver:handler_step(HCY)),
    integer()
) -> gleam@otp@supervision:child_specification(distribute@global:global_subject(HCW)).
child_spec(Typed_name, Initial_state, Handler, Init_timeout_ms) ->
    Name = distribute@registry:typed_name_to_string(Typed_name),
    gleam@otp@supervision:worker(
        fun() ->
            case distribute@receiver:start_distributed_worker(
                Name,
                Initial_state,
                distribute@registry:typed_name_encoder(Typed_name),
                distribute@registry:typed_name_decoder(Typed_name),
                Handler,
                Init_timeout_ms
            ) of
                {ok, Started} ->
                    case distribute@registry:register_global(
                        Typed_name,
                        erlang:element(3, Started)
                    ) of
                        {ok, _} ->
                            {ok, Started};

                        {error, _} ->
                            terminate_orphan_gracefully(
                                erlang:element(2, Started)
                            ),
                            {error,
                                {init_failed, <<"registration_failed"/utf8>>}}
                    end;

                {error, Err} ->
                    {error, Err}
            end
        end
    ).

-file("src/distribute/actor.gleam", 217).
?DOC(" Like `child_spec`, with `config.get().default_init_timeout_ms`.\n").
-spec child_spec_default(
    distribute@registry:typed_name(HCQ),
    HCS,
    fun((HCQ, HCS) -> distribute@receiver:handler_step(HCS))
) -> gleam@otp@supervision:child_specification(distribute@global:global_subject(HCQ)).
child_spec_default(Typed_name, Initial_state, Handler) ->
    child_spec(
        Typed_name,
        Initial_state,
        Handler,
        erlang:element(3, distribute@config:get())
    ).

-file("src/distribute/actor.gleam", 289).
?DOC(
    " Render a `StartRegisteredError` as a human-readable string. Mirrors\n"
    " the `*_error_to_string` formatters published by the other error\n"
    " modules so observability paths can `io.println` an error directly\n"
    " without pattern-matching at every call site.\n"
).
-spec start_registered_error_to_string(start_registered_error()) -> binary().
start_registered_error_to_string(Err) ->
    case Err of
        {actor_start_failed, init_timeout} ->
            <<"Actor init timed out"/utf8>>;

        {actor_start_failed, {init_failed, Reason}} ->
            <<"Actor init failed: "/utf8, Reason/binary>>;

        {actor_start_failed, {init_exited, _}} ->
            <<"Actor init process exited"/utf8>>;

        {global_register_failed, Re} ->
            <<"Global registration failed: "/utf8,
                (distribute@registry:register_error_to_string(Re))/binary>>
    end.

-file("src/distribute/actor.gleam", 354).
?DOC(
    " Start a supervised actor that auto-registers on (re)start.\n"
    "\n"
    " If registration fails, the worker crashes and the supervisor retries,\n"
    " which is the correct OTP pattern for transient registration failures.\n"
).
-spec start_supervised(
    distribute@registry:typed_name(HDP),
    HDR,
    fun((HDP, HDR) -> distribute@receiver:handler_step(HDR)),
    integer()
) -> {ok, gleam@erlang@process:pid_()} | {error, gleam@otp@actor:start_error()}.
start_supervised(Typed_name, Initial_state, Handler, Init_timeout_ms) ->
    Spec = child_spec(Typed_name, Initial_state, Handler, Init_timeout_ms),
    Builder = begin
        _pipe = gleam@otp@static_supervisor:new(one_for_one),
        gleam@otp@static_supervisor:add(_pipe, Spec)
    end,
    case gleam@otp@static_supervisor:start(Builder) of
        {ok, Sup} ->
            {ok, erlang:element(2, Sup)};

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

-file("src/distribute/actor.gleam", 337).
?DOC(" Like `start_supervised`, with `config.get().default_init_timeout_ms`.\n").
-spec start_supervised_default(
    distribute@registry:typed_name(HDJ),
    HDL,
    fun((HDJ, HDL) -> distribute@receiver:handler_step(HDL))
) -> {ok, gleam@erlang@process:pid_()} | {error, gleam@otp@actor:start_error()}.
start_supervised_default(Typed_name, Initial_state, Handler) ->
    start_supervised(
        Typed_name,
        Initial_state,
        Handler,
        erlang:element(3, distribute@config:get())
    ).

-file("src/distribute/actor.gleam", 575).
-spec do_index_list(integer(), list(integer())) -> list(integer()).
do_index_list(I, Acc) ->
    case I < 1 of
        true ->
            Acc;

        false ->
            do_index_list(I - 1, [I | Acc])
    end.

-file("src/distribute/actor.gleam", 571).
?DOC(
    " Build `[1, 2, ..., n]` for `n >= 1`. Replacement for `list.range`,\n"
    " which was removed in `gleam_stdlib` 1.0.\n"
).
-spec index_list(integer()) -> list(integer()).
index_list(N) ->
    do_index_list(N, []).

-file("src/distribute/actor.gleam", 404).
?DOC(
    " Start N supervised actors, registered as `name_1` .. `name_N`.\n"
    "\n"
    " Returns `Error(actor.InitFailed(...))` if `size < 1`. `list.range`\n"
    " would otherwise produce a degenerate list of weird worker names\n"
    " (e.g. `name_0`, `name_-1`) and silently start a useless supervisor.\n"
    "\n"
    " > [!WARNING]\n"
    " > **Cascading Failure Risk**:\n"
    " > `:global` and fixed pools do not mix perfectly. If a single worker\n"
    " > in the pool fails its global registration (e.g. because `name_4` is\n"
    " > legitimately held by a node on the other side of a network split),\n"
    " > the worker crashes. The pool supervisor will try to restart it. If the\n"
    " > conflict persists, the supervisor exhausts its `MaxR` restart intensity\n"
    " > and **crashes the entire pool**. This means a collision on 1 worker\n"
    " > brings down all N workers, causing the parent supervisor to restart the\n"
    " > whole pool. This architectural limit will be solved natively when the\n"
    " > `syn` registry backend is introduced in v4.2.0 (via `syn` process groups).\n"
).
-spec pool(
    distribute@registry:typed_name(HEB),
    integer(),
    HED,
    fun((HEB, HED) -> distribute@receiver:handler_step(HED)),
    integer()
) -> {ok, gleam@erlang@process:pid_()} | {error, gleam@otp@actor:start_error()}.
pool(Typed_name, Size, Initial_state, Handler, Init_timeout_ms) ->
    case Size < 1 of
        true ->
            {error,
                {init_failed,
                    <<"pool size must be >= 1, got "/utf8,
                        (erlang:integer_to_binary(Size))/binary>>}};

        false ->
            Specs = begin
                _pipe = index_list(Size),
                gleam@list:map(
                    _pipe,
                    fun(I) ->
                        Worker_tn = distribute@registry:pool_member(
                            Typed_name,
                            I
                        ),
                        child_spec(
                            Worker_tn,
                            Initial_state,
                            Handler,
                            Init_timeout_ms
                        )
                    end
                )
            end,
            Builder = gleam@list:fold(
                Specs,
                gleam@otp@static_supervisor:new(one_for_one),
                fun(Acc, Spec) -> gleam@otp@static_supervisor:add(Acc, Spec) end
            ),
            case gleam@otp@static_supervisor:start(Builder) of
                {ok, Sup} ->
                    {ok, erlang:element(2, Sup)};

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

-file("src/distribute/actor.gleam", 372).
?DOC(" Like `pool`, with `config.get().default_init_timeout_ms`.\n").
-spec pool_default(
    distribute@registry:typed_name(HDV),
    integer(),
    HDX,
    fun((HDV, HDX) -> distribute@receiver:handler_step(HDX))
) -> {ok, gleam@erlang@process:pid_()} | {error, gleam@otp@actor:start_error()}.
pool_default(Typed_name, Size, Initial_state, Handler) ->
    pool(
        Typed_name,
        Size,
        Initial_state,
        Handler,
        erlang:element(3, distribute@config:get())
    ).

-file("src/distribute/actor.gleam", 447).
?DOC(
    " How long the owner waits between liveness re-checks while watching\n"
    " `lifetime`. Picked to be long enough that an idle owner is essentially\n"
    " free, short enough that a `DOWN` we somehow missed (cosmic ray,\n"
    " runtime-tracing tool dropping the message) is detected within seconds.\n"
).
-spec resource_owner_poll_ms() -> integer().
resource_owner_poll_ms() ->
    erlang:element(8, distribute@config:get()).

-file("src/distribute/actor.gleam", 551).
?DOC(
    " Block the owner until the lifetime PID is dead. Long-polls the\n"
    " `DOWN` selector and re-checks `process.is_alive` on each timeout.\n"
    " Without the re-check, a buggy or instrumented runtime that loses a\n"
    " `DOWN` message would leave us waiting forever; with it, we converge\n"
    " on the truth within `resource_owner_poll_ms`.\n"
).
-spec await_lifetime_death(
    gleam@erlang@process:selector(nil),
    gleam@erlang@process:pid_()
) -> nil.
await_lifetime_death(Selector, Lifetime) ->
    case gleam_erlang_ffi:select(Selector, resource_owner_poll_ms()) of
        {ok, nil} ->
            nil;

        {error, nil} ->
            case erlang:is_process_alive(Lifetime) of
                false ->
                    nil;

                true ->
                    await_lifetime_death(Selector, Lifetime)
            end
    end.

-file("src/distribute/actor.gleam", 528).
?DOC(
    " Spawn a dedicated process that owns an external resource and runs\n"
    " `close(resource)` when `lifetime` dies for any reason. Workaround\n"
    " for the missing `{terminate, Reason}` callback in\n"
    " `gleam/otp/actor` 1.x: external resources (database connections,\n"
    " distributed locks, NIF handles, OS pipes) cannot rely on the\n"
    " actor's exit to free them, but a dedicated observer process can.\n"
    "\n"
    " Returns the owner's PID, in case you need to stop the owner\n"
    " independently of the lifetime PID.\n"
    "\n"
    " ## Why this shape\n"
    "\n"
    " The resource is opened **inside the owner**, so it is BEAM-process-\n"
    " owned by the owner. That matters for resources whose lifetime is\n"
    " tied to a specific process at the runtime level (ETS tables, ports,\n"
    " `gen_tcp` controlling process). \n"
    "\n"
    " The owner is **linked** to the caller (which is usually the actor's \n"
    " `init` function). It also traps exits (`process.trap_exits(True)`). \n"
    " This guarantees a **fail-fast** behaviour: if `open()` crashes before \n"
    " it returns, the owner dies, and because it is linked, it pulls the \n"
    " actor down with it immediately. This prevents the \"partial failure\" \n"
    " scenario where the actor survives but its critical resource failed \n"
    " to initialize.\n"
    "\n"
    " The owner uses a `process.monitor` on `lifetime`. Monitor delivery\n"
    " is asynchronous but reliable: the BEAM guarantees a `DOWN` for\n"
    " every monitored PID, and a monitor on an already-dead PID fires\n"
    " immediately. That makes the helper safe to call right after the\n"
    " actor has started: even if there is a microsecond gap before the\n"
    " monitor goes up, the BEAM resolves it correctly.\n"
    "\n"
    " ## Usage\n"
    "\n"
    " ```gleam\n"
    " import distribute\n"
    " import distribute/actor as dist_actor\n"
    " import distribute/global\n"
    "\n"
    " let assert Ok(gs) =\n"
    "   distribute.start_registered(name, init, handler)\n"
    " let assert Ok(actor_pid) = global.owner(gs)\n"
    " let _owner = dist_actor.start_resource_owner(\n"
    "   fn() { open_postgres_pool() },\n"
    "   fn(pool) { close_postgres_pool(pool) },\n"
    "   actor_pid,\n"
    " )\n"
    " ```\n"
    "\n"
    " If the actor needs to *use* the resource, have `open` build a\n"
    " `process.Subject` the owner serves and pass it through your\n"
    " actor's `init`.\n"
    "\n"
    " ## Failure modes\n"
    "\n"
    " - `open` raises: the owner dies before the resource is opened. Because\n"
    "   the owner is linked to the caller (the actor), the actor receives an\n"
    "   exit signal and crashes immediately. This fail-fast behaviour avoids\n"
    "   partial failures where the actor runs without its resource.\n"
    " - `close` raises: the owner dies after the panic, no further\n"
    "   cleanup runs. The resource may be in a partially-closed state.\n"
    " - `lifetime` is already dead at call time: monitor fires `DOWN`\n"
    "   immediately, `close` runs on the next scheduler tick.\n"
    " - The owner is `process.kill`-ed: cleanup is uncatchable, BEAM\n"
    "   reaps the resource via process death (only effective for\n"
    "   BEAM-process-tied resources). External handles leak. This is\n"
    "   the same uncatchable-kill caveat OTP itself has.\n"
    "\n"
    " ## Future\n"
    "\n"
    " When `gleam/otp/actor` ships native termination callbacks\n"
    " (gleam-lang/otp#126), this helper becomes redundant for the\n"
    " actor case. The contract will not shift: callers can replace the\n"
    " `actor.start_resource_owner(open, close, pid)` line with the\n"
    " upstream API and delete the helper call site. The function will\n"
    " stay (deprecated) for the more general \"tie cleanup to any PID\"\n"
    " pattern.\n"
).
-spec start_resource_owner(
    fun(() -> HEH),
    fun((HEH) -> nil),
    gleam@erlang@process:pid_()
) -> gleam@erlang@process:pid_().
start_resource_owner(Open, Close, Lifetime) ->
    proc_lib:spawn_link(
        fun() ->
            gleam_erlang_ffi:trap_exits(true),
            Resource = Open(),
            Mon = gleam@erlang@process:monitor(Lifetime),
            Down_selector = begin
                _pipe = gleam_erlang_ffi:new_selector(),
                gleam@erlang@process:select_specific_monitor(
                    _pipe,
                    Mon,
                    fun(_) -> nil end
                )
            end,
            await_lifetime_death(Down_selector, Lifetime),
            Close(Resource),
            nil
        end
    ).