-module(crew).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/crew.gleam").
-export([fixed_size/2, subscribe/1, unsubscribe/2, start/1, supervised/1, new/1, select_map/3, enqueue_many/2, work_many/3, work/3, parallel_map/4, work_ordered/3, enqueue/2]).
-export_type([builder/0, channel/1, pool_msg/0, state/0, worker/0, queue_item/0, active_worker/0, caller/0, request/0, any_/0, work/0, work_result/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(
    " A worker pool for Gleam that distributes work across a limited number of\n"
    " worker processes. Workers are pooled to avoid the overhead of\n"
    " spawning and killing processes for each task.\n"
    "\n"
    " The pool manages a queue of work items and distributes them to idle workers.\n"
    " When no workers are available, work is queued until a worker becomes free.\n"
    " The pool handles worker crashes gracefully and automatically manages the\n"
    " worker lifecycle.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " import crew\n"
    " import gleam/otp/static_supervisor as supervisor\n"
    "\n"
    " pub fn main() {\n"
    "   // Create a pool name\n"
    "   let pool_name = process.new_name(\"my_crew\")\n"
    "\n"
    "   // Start an unsupervised pool\n"
    "   let assert Ok(_) =\n"
    "       crew.new(pool_name)\n"
    "       |> crew.fixed_size(4)\n"
    "       |> crew.start\n"
    "\n"
    "   // Execute work on the pool\n"
    "   let result = crew.work(pool_name, 5000, fn() {\n"
    "     // Some expensive computation\n"
    "     expensive_computation()\n"
    "   })\n"
    " }\n"
    " ```\n"
).

-opaque builder() :: {builder, gleam@erlang@process:name(pool_msg()), integer()}.

-opaque channel(FKL) :: {channel,
        gleam@erlang@process:subject(pool_msg()),
        gleam@erlang@process:subject(work_result()),
        gleam@erlang@process:monitor()} |
    {gleam_phantom, FKL}.

-opaque pool_msg() :: {worker_started,
        gleam@erlang@process:pid_(),
        gleam@erlang@process:subject(work())} |
    {worker_idle, gleam@erlang@process:pid_()} |
    {monitored_process_exited, gleam@erlang@process:down()} |
    {subscribe, gleam@erlang@process:subject(work_result())} |
    {unsubscribe,
        gleam@erlang@process:subject(work_result()),
        gleam@erlang@process:subject(nil)} |
    {enqueue,
        gleam@erlang@process:subject(work_result()),
        list(fun(() -> any_()))}.

-type state() :: {state,
        list(worker()),
        gleam@dict:dict(gleam@erlang@process:pid_(), active_worker()),
        gleam@dict:dict(gleam@erlang@process:pid_(), caller()),
        gleam@dict:dict(gleam@erlang@process:subject(work_result()), request()),
        gleam@deque:deque(queue_item())}.

-type worker() :: {worker,
        gleam@erlang@process:pid_(),
        gleam@erlang@process:subject(work()),
        gleam@erlang@process:monitor()}.

-type queue_item() :: {queue_item,
        fun(() -> any_()),
        gleam@erlang@process:pid_(),
        gleam@erlang@process:subject(work_result())}.

-type active_worker() :: {active_worker,
        worker(),
        gleam@erlang@process:pid_(),
        gleam@erlang@process:subject(work_result())}.

-type caller() :: {caller,
        gleam@erlang@process:pid_(),
        gleam@erlang@process:monitor(),
        gleam@set:set(gleam@erlang@process:subject(work_result()))}.

-type request() :: {request,
        gleam@erlang@process:pid_(),
        gleam@set:set(gleam@erlang@process:pid_()),
        gleam@erlang@process:subject(work_result())}.

-type any_() :: any().

-type work() :: {work,
        fun(() -> any_()),
        gleam@erlang@process:subject(work_result())}.

-type work_result() :: {done, any_()} |
    {worker_exited, gleam@erlang@process:exit_reason()}.

-file("src/crew.gleam", 92).
?DOC(
    " Set the number of worker processes in the pool to a fixed number.\n"
    "\n"
    " When set to less than 1 starting the pool will fail.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " crew.new(pool_name)\n"
    " |> crew.fixed_size(8)  // Use 8 workers regardless of CPU count\n"
    " ```\n"
).
-spec fixed_size(builder(), integer()) -> builder().
fixed_size(Builder, Size) ->
    {builder, erlang:element(2, Builder), Size}.

-file("src/crew.gleam", 413).
?DOC(
    " Create a typed subscription channel to the worker pool for work submission.\n"
    "\n"
    " This function provides a lower-level interface for submitting work to the\n"
    " pool. In many cases calling the `work*` function from a separate process\n"
    " will be easier.\n"
    "\n"
    " Subscribing to the pool creates a reference to the current process in the pool\n"
    " that must be cleaned up using `unsubscribe`. The returned channel can be used\n"
    " with `select_map` and `enqueue` for custom message handling.\n"
    "\n"
    " ## Panics\n"
    " - If the pool is not running\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let channel = crew.subscribe(pool_name)\n"
    " crew.enqueue(channel, fn() { some_work() })\n"
    " // Handle results with selector...\n"
    " crew.unsubscribe(channel, selector)\n"
    " ```\n"
).
-spec subscribe(gleam@erlang@process:name(pool_msg())) -> channel(any()).
subscribe(Pool) ->
    Pool_subject = gleam@erlang@process:named_subject(Pool),
    Pool_pid@1 = case gleam@erlang@process:subject_owner(Pool_subject) of
        {ok, Pool_pid} -> Pool_pid;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Pool is not running"/utf8>>,
                        file => <<?FILEPATH/utf8>>,
                        module => <<"crew"/utf8>>,
                        function => <<"subscribe"/utf8>>,
                        line => 415,
                        value => _assert_fail,
                        start => 12515,
                        'end' => 12576,
                        pattern_start => 12526,
                        pattern_end => 12538})
    end,
    Monitor = gleam@erlang@process:monitor(Pool_pid@1),
    Receive = gleam@erlang@process:new_subject(),
    gleam@otp@actor:send(Pool_subject, {subscribe, Receive}),
    {channel, Pool_subject, Receive, Monitor}.

-file("src/crew.gleam", 496).
-spec drop_messages_loop(gleam@erlang@process:selector(boolean())) -> nil.
drop_messages_loop(Selector) ->
    case gleam_erlang_ffi:select(Selector) of
        true ->
            nil;

        false ->
            drop_messages_loop(Selector)
    end.

-file("src/crew.gleam", 486).
-spec wait_for_unsubscribe(channel(any()), gleam@erlang@process:subject(any())) -> nil.
wait_for_unsubscribe(Channel, Unsubscribed) ->
    Unsubscribe_selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        _pipe@1 = gleam@erlang@process:select_map(
            _pipe,
            Unsubscribed,
            fun(_) -> true end
        ),
        _pipe@2 = gleam@erlang@process:select_specific_monitor(
            _pipe@1,
            erlang:element(4, Channel),
            fun(_) -> true end
        ),
        gleam@erlang@process:select_map(
            _pipe@2,
            erlang:element(3, Channel),
            fun(_) -> false end
        )
    end,
    drop_messages_loop(Unsubscribe_selector).

-file("src/crew.gleam", 471).
?DOC(
    " Remove a subscription channel from the pool and clean up resources.\n"
    "\n"
    " This function should be called when you're done using a channel obtained\n"
    " from `subscribe`. It stops any in-progress work for this channel and\n"
    " removes the channel's handlers from the selector.\n"
    "\n"
    " Any finished work that is still coming in is dropped.\n"
).
-spec unsubscribe(channel(any()), gleam@erlang@process:selector(FMG)) -> gleam@erlang@process:selector(FMG).
unsubscribe(Channel, Selector) ->
    Unsubscribed = gleam@erlang@process:new_subject(),
    gleam@otp@actor:send(
        erlang:element(2, Channel),
        {unsubscribe, erlang:element(3, Channel), Unsubscribed}
    ),
    wait_for_unsubscribe(Channel, Unsubscribed),
    gleam@erlang@process:demonitor_process(erlang:element(4, Channel)),
    _pipe = Selector,
    _pipe@1 = gleam@erlang@process:deselect_specific_monitor(
        _pipe,
        erlang:element(4, Channel)
    ),
    gleam@erlang@process:deselect(_pipe@1, erlang:element(3, Channel)).

-file("src/crew.gleam", 563).
-spec init_pool(gleam@erlang@process:subject(pool_msg())) -> {ok,
        gleam@otp@actor:initialised(state(), pool_msg(), gleam@erlang@process:subject(pool_msg()))} |
    {error, any()}.
init_pool(Self) ->
    Selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        _pipe@1 = gleam@erlang@process:select(_pipe, Self),
        gleam@erlang@process:select_monitors(
            _pipe@1,
            fun(Field@0) -> {monitored_process_exited, Field@0} end
        )
    end,
    State = {state, [], maps:new(), maps:new(), maps:new(), gleam@deque:new()},
    _pipe@2 = gleam@otp@actor:initialised(State),
    _pipe@3 = gleam@otp@actor:selecting(_pipe@2, Selector),
    _pipe@4 = gleam@otp@actor:returning(_pipe@3, Self),
    {ok, _pipe@4}.

-file("src/crew.gleam", 712).
-spec enqueue_loop(state(), request(), list(fun(() -> any_()))) -> state().
enqueue_loop(State, Request, Work) ->
    {request, Caller, Workers, Receive} = Request,
    case {Work, erlang:element(2, State)} of
        {[Work@1 | Rest], [Worker | Idle_workers]} ->
            gleam@erlang@process:send(
                erlang:element(3, Worker),
                {work, Work@1, Receive}
            ),
            Active_worker = {active_worker, Worker, Caller, Receive},
            Active_workers = gleam@dict:insert(
                erlang:element(3, State),
                erlang:element(2, Worker),
                Active_worker
            ),
            Request@1 = {request,
                erlang:element(2, Request),
                gleam@set:insert(Workers, erlang:element(2, Worker)),
                erlang:element(4, Request)},
            _pipe = {state,
                Idle_workers,
                Active_workers,
                erlang:element(4, State),
                erlang:element(5, State),
                erlang:element(6, State)},
            enqueue_loop(_pipe, Request@1, Rest);

        {_, _} ->
            Queue@1 = gleam@list:fold(
                Work,
                erlang:element(6, State),
                fun(Queue, Work@2) ->
                    gleam@deque:push_back(
                        Queue,
                        {queue_item, Work@2, Caller, Receive}
                    )
                end
            ),
            Requests = gleam@dict:insert(
                erlang:element(5, State),
                Receive,
                Request
            ),
            {state,
                erlang:element(2, State),
                erlang:element(3, State),
                erlang:element(4, State),
                Requests,
                Queue@1}
    end.

-file("src/crew.gleam", 820).
-spec worker_loop(
    gleam@erlang@process:subject(pool_msg()),
    gleam@erlang@process:pid_(),
    gleam@erlang@process:subject(work())
) -> any().
worker_loop(Pool, Self, Subject) ->
    {work, Work, Receive} = gleam_erlang_ffi:'receive'(Subject),
    Result = Work(),
    gleam@erlang@process:send(Pool, {worker_idle, Self}),
    gleam@erlang@process:send(Receive, {done, Result}),
    worker_loop(Pool, Self, Subject).

-file("src/crew.gleam", 811).
-spec worker(gleam@erlang@process:name(pool_msg())) -> nil.
worker(Pool) ->
    Self = erlang:self(),
    Subject = gleam@erlang@process:new_subject(),
    Pool_subject = gleam@erlang@process:named_subject(Pool),
    gleam@erlang@process:send(Pool_subject, {worker_started, Self, Subject}),
    worker_loop(Pool_subject, Self, Subject).

-file("src/crew.gleam", 832).
-spec repeat(integer(), FNC, fun((FNC) -> FNC)) -> FNC.
repeat(Times, State, F) ->
    case Times > 0 of
        true ->
            repeat(Times - 1, F(State), F);

        false ->
            State
    end.

-file("src/crew.gleam", 843).
-spec 'try'({ok, FND} | {error, FNE}, fun((FNE) -> FNH), fun((FND) -> FNH)) -> FNH.
'try'(Result, Or, Then) ->
    case Result of
        {ok, X} ->
            Then(X);

        {error, X@1} ->
            Or(X@1)
    end.

-file("src/crew.gleam", 747).
-spec try_dequeue_work(state(), worker()) -> state().
try_dequeue_work(State, Worker) ->
    'try'(
        gleam@deque:pop_front(erlang:element(6, State)),
        fun(_) ->
            {state,
                [Worker | erlang:element(2, State)],
                erlang:element(3, State),
                erlang:element(4, State),
                erlang:element(5, State),
                erlang:element(6, State)}
        end,
        fun(_use0) ->
            {{queue_item, Work, Caller, Receive}, Queue} = _use0,
            'try'(
                gleam_stdlib:map_get(erlang:element(5, State), Receive),
                fun(_) ->
                    try_dequeue_work(
                        {state,
                            erlang:element(2, State),
                            erlang:element(3, State),
                            erlang:element(4, State),
                            erlang:element(5, State),
                            Queue},
                        Worker
                    )
                end,
                fun(Request) ->
                    gleam@erlang@process:send(
                        erlang:element(3, Worker),
                        {work, Work, Receive}
                    ),
                    Active_worker = {active_worker, Worker, Caller, Receive},
                    Active_workers = gleam@dict:insert(
                        erlang:element(3, State),
                        erlang:element(2, Worker),
                        Active_worker
                    ),
                    Request@1 = {request,
                        erlang:element(2, Request),
                        gleam@set:insert(
                            erlang:element(3, Request),
                            erlang:element(2, Worker)
                        ),
                        erlang:element(4, Request)},
                    Requests = gleam@dict:insert(
                        erlang:element(5, State),
                        Receive,
                        Request@1
                    ),
                    {state,
                        erlang:element(2, State),
                        Active_workers,
                        erlang:element(4, State),
                        Requests,
                        Queue}
                end
            )
        end
    ).

-file("src/crew.gleam", 850).
-spec try_({ok, FNJ} | {error, any()}, FNN, fun((FNJ) -> FNN)) -> FNN.
try_(Result, Or, Then) ->
    case Result of
        {ok, X} ->
            Then(X);

        {error, _} ->
            Or
    end.

-file("src/crew.gleam", 772).
-spec abort_caller(
    state(),
    gleam@erlang@process:pid_(),
    gleam@erlang@process:exit_reason()
) -> state().
abort_caller(State, Caller, Reason) ->
    try_(
        gleam_stdlib:map_get(erlang:element(4, State), Caller),
        State,
        fun(Caller@1) ->
            gleam@erlang@process:demonitor_process(erlang:element(3, Caller@1)),
            Requests@1 = gleam@set:fold(
                erlang:element(4, Caller@1),
                erlang:element(5, State),
                fun(Requests, Receive) ->
                    try_(
                        gleam_stdlib:map_get(erlang:element(5, State), Receive),
                        Requests,
                        fun(Request) ->
                            gleam@set:each(
                                erlang:element(3, Request),
                                fun gleam@erlang@process:kill/1
                            ),
                            gleam@erlang@process:send(
                                Receive,
                                {worker_exited, Reason}
                            ),
                            gleam@dict:delete(Requests, Receive)
                        end
                    )
                end
            ),
            Callers = gleam@dict:delete(
                erlang:element(4, State),
                erlang:element(2, Caller@1)
            ),
            {state,
                erlang:element(2, State),
                erlang:element(3, State),
                Callers,
                Requests@1,
                erlang:element(6, State)}
        end
    ).

-file("src/crew.gleam", 584).
-spec pool(state(), pool_msg()) -> gleam@otp@actor:next(state(), pool_msg()).
pool(State, Msg) ->
    Next_state = case Msg of
        {worker_started, Pid, Send} ->
            Monitor = gleam@erlang@process:monitor(Pid),
            Worker = {worker, Pid, Send, Monitor},
            try_dequeue_work(State, Worker);

        {monitored_process_exited, {process_down, _, Pid@1, Reason}} ->
            case {gleam_stdlib:map_get(erlang:element(4, State), Pid@1),
                gleam_stdlib:map_get(erlang:element(3, State), Pid@1)} of
                {{ok, _}, _} ->
                    abort_caller(State, Pid@1, Reason);

                {{error, _}, {ok, {active_worker, _, Caller, _}}} ->
                    Active_workers = gleam@dict:delete(
                        erlang:element(3, State),
                        Pid@1
                    ),
                    _pipe = {state,
                        erlang:element(2, State),
                        Active_workers,
                        erlang:element(4, State),
                        erlang:element(5, State),
                        erlang:element(6, State)},
                    abort_caller(_pipe, Caller, Reason);

                {{error, _}, {error, _}} ->
                    Idle_workers = gleam@list:filter(
                        erlang:element(2, State),
                        fun(Worker@1) ->
                            erlang:element(2, Worker@1) /= Pid@1
                        end
                    ),
                    {state,
                        Idle_workers,
                        erlang:element(3, State),
                        erlang:element(4, State),
                        erlang:element(5, State),
                        erlang:element(6, State)}
            end;

        {monitored_process_exited, _} ->
            State;

        {worker_idle, Pid@2} ->
            try_(
                gleam_stdlib:map_get(erlang:element(3, State), Pid@2),
                State,
                fun(_use0) ->
                    {active_worker, Worker@2, _, Receive} = _use0,
                    Active_workers@1 = gleam@dict:delete(
                        erlang:element(3, State),
                        Pid@2
                    ),
                    case gleam_stdlib:map_get(erlang:element(5, State), Receive) of
                        {ok, Request} ->
                            Request@1 = {request,
                                erlang:element(2, Request),
                                gleam@set:delete(
                                    erlang:element(3, Request),
                                    Pid@2
                                ),
                                erlang:element(4, Request)},
                            Requests = gleam@dict:insert(
                                erlang:element(5, State),
                                Receive,
                                Request@1
                            ),
                            _pipe@1 = {state,
                                erlang:element(2, State),
                                Active_workers@1,
                                erlang:element(4, State),
                                Requests,
                                erlang:element(6, State)},
                            try_dequeue_work(_pipe@1, Worker@2);

                        {error, _} ->
                            {state,
                                erlang:element(2, State),
                                Active_workers@1,
                                erlang:element(4, State),
                                erlang:element(5, State),
                                erlang:element(6, State)}
                    end
                end
            );

        {enqueue, _, []} ->
            State;

        {enqueue, Receive@1, Work} ->
            try_(
                gleam_stdlib:map_get(erlang:element(5, State), Receive@1),
                State,
                fun(Request@2) -> enqueue_loop(State, Request@2, Work) end
            );

        {subscribe, Receive@2} ->
            try_(
                gleam@erlang@process:subject_owner(Receive@2),
                State,
                fun(Pid@3) ->
                    Request@3 = begin
                        _pipe@2 = gleam_stdlib:map_get(
                            erlang:element(5, State),
                            Receive@2
                        ),
                        gleam@result:unwrap(
                            _pipe@2,
                            {request, Pid@3, gleam@set:new(), Receive@2}
                        )
                    end,
                    Caller@2 = case gleam_stdlib:map_get(
                        erlang:element(4, State),
                        Pid@3
                    ) of
                        {ok, Caller@1} ->
                            {caller,
                                erlang:element(2, Caller@1),
                                erlang:element(3, Caller@1),
                                gleam@set:insert(
                                    erlang:element(4, Caller@1),
                                    Receive@2
                                )};

                        {error, _} ->
                            Monitor@1 = gleam@erlang@process:monitor(Pid@3),
                            {caller,
                                Pid@3,
                                Monitor@1,
                                begin
                                    _pipe@3 = gleam@set:new(),
                                    gleam@set:insert(_pipe@3, Receive@2)
                                end}
                    end,
                    Callers = gleam@dict:insert(
                        erlang:element(4, State),
                        Pid@3,
                        Caller@2
                    ),
                    Requests@1 = gleam@dict:insert(
                        erlang:element(5, State),
                        Receive@2,
                        Request@3
                    ),
                    {state,
                        erlang:element(2, State),
                        erlang:element(3, State),
                        Callers,
                        Requests@1,
                        erlang:element(6, State)}
                end
            );

        {unsubscribe, Receive@3, Unsubscribed} ->
            try_(
                gleam_stdlib:map_get(erlang:element(5, State), Receive@3),
                State,
                fun(Request@4) ->
                    Requests@2 = gleam@dict:delete(
                        erlang:element(5, State),
                        Receive@3
                    ),
                    gleam@set:each(
                        erlang:element(3, Request@4),
                        fun gleam@erlang@process:kill/1
                    ),
                    try_(
                        gleam_stdlib:map_get(
                            erlang:element(4, State),
                            erlang:element(2, Request@4)
                        ),
                        {state,
                            erlang:element(2, State),
                            erlang:element(3, State),
                            erlang:element(4, State),
                            Requests@2,
                            erlang:element(6, State)},
                        fun(Caller@3) ->
                            Caller@4 = {caller,
                                erlang:element(2, Caller@3),
                                erlang:element(3, Caller@3),
                                gleam@set:delete(
                                    erlang:element(4, Caller@3),
                                    Receive@3
                                )},
                            Callers@1 = case gleam@set:is_empty(
                                erlang:element(4, Caller@4)
                            ) of
                                true ->
                                    gleam@erlang@process:demonitor_process(
                                        erlang:element(3, Caller@4)
                                    ),
                                    gleam@dict:delete(
                                        erlang:element(4, State),
                                        erlang:element(2, Caller@4)
                                    );

                                false ->
                                    gleam@dict:insert(
                                        erlang:element(4, State),
                                        erlang:element(2, Caller@4),
                                        Caller@4
                                    )
                            end,
                            gleam@erlang@process:send(Unsubscribed, nil),
                            {state,
                                erlang:element(2, State),
                                erlang:element(3, State),
                                Callers@1,
                                Requests@2,
                                erlang:element(6, State)}
                        end
                    )
                end
            )
    end,
    gleam@otp@actor:continue(Next_state).

-file("src/crew.gleam", 146).
-spec start_tree(gleam@erlang@process:name(pool_msg()), integer()) -> {ok,
        gleam@otp@actor:started(gleam@otp@static_supervisor:supervisor())} |
    {error, gleam@otp@actor:start_error()}.
start_tree(Name, Size) ->
    gleam@bool:guard(
        Size =< 0,
        {error, {init_failed, <<"pool size must be greater than zero"/utf8>>}},
        fun() ->
            Main_supervisor = gleam@otp@static_supervisor:new(rest_for_one),
            Worker_supervisor = gleam@otp@static_supervisor:new(one_for_one),
            Pool_spec = begin
                gleam@otp@supervision:worker(
                    fun() ->
                        _pipe = gleam@otp@actor:new_with_initialiser(
                            1000,
                            fun init_pool/1
                        ),
                        _pipe@1 = gleam@otp@actor:named(_pipe, Name),
                        _pipe@2 = gleam@otp@actor:on_message(
                            _pipe@1,
                            fun pool/2
                        ),
                        gleam@otp@actor:start(_pipe@2)
                    end
                )
            end,
            Worker_spec = begin
                gleam@otp@supervision:worker(
                    fun() ->
                        Pid = proc_lib:spawn_link(fun() -> worker(Name) end),
                        {ok, {started, Pid, nil}}
                    end
                )
            end,
            Worker_supervisor_spec = begin
                gleam@otp@supervision:supervisor(
                    fun() -> _pipe@3 = Worker_supervisor,
                        _pipe@4 = repeat(
                            Size,
                            _pipe@3,
                            fun(_capture) ->
                                gleam@otp@static_supervisor:add(
                                    _capture,
                                    Worker_spec
                                )
                            end
                        ),
                        gleam@otp@static_supervisor:start(_pipe@4) end
                )
            end,
            _pipe@5 = Main_supervisor,
            _pipe@6 = gleam@otp@static_supervisor:add(_pipe@5, Pool_spec),
            _pipe@7 = gleam@otp@static_supervisor:add(
                _pipe@6,
                Worker_supervisor_spec
            ),
            gleam@otp@static_supervisor:start(_pipe@7)
        end
    ).

-file("src/crew.gleam", 115).
?DOC(
    " Start an unsupervised worker pool from the given builder.\n"
    "\n"
    " Returns a supervisor that manages the pool and its workers. In most cases,\n"
    " you should use `supervised` instead to get a child specification that can\n"
    " be added to your application's supervision tree.\n"
    "\n"
    " ## Panics\n"
    " This function will exit the process if any workers fail to start, similar\n"
    " to `static_supervisor.start`.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let assert Ok(pool_supervisor) =\n"
    "   crew.new(pool_name)\n"
    "   |> crew.start\n"
    " ```\n"
).
-spec start(builder()) -> {ok, gleam@otp@static_supervisor:supervisor()} |
    {error, gleam@otp@actor:start_error()}.
start(Builder) ->
    {builder, Name, Size} = Builder,
    gleam@result:map(
        start_tree(Name, Size),
        fun(Result) -> erlang:element(3, Result) end
    ).

-file("src/crew.gleam", 140).
?DOC(
    " Create a child specification for a supervised worker pool.\n"
    "\n"
    " This is the recommended way to start a worker pool as part of your\n"
    " application's supervision tree. The returned child specification can be\n"
    " added to a supervisor using `static_supervisor.add`.\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let pool_spec =\n"
    "   crew.new(pool_name)\n"
    "   |> crew.fixed_size(4)\n"
    "   |> crew.supervised\n"
    "\n"
    " let assert Ok(_) =\n"
    "   supervisor.new(supervisor.OneForOne)\n"
    "   |> supervisor.add(pool_spec)\n"
    "   |> supervisor.start\n"
    " ```\n"
).
-spec supervised(builder()) -> gleam@otp@supervision:child_specification(gleam@otp@static_supervisor:supervisor()).
supervised(Builder) ->
    {builder, Name, Size} = Builder,
    gleam@otp@supervision:supervisor(fun() -> start_tree(Name, Size) end).

-file("src/crew.gleam", 78).
?DOC(
    " Create a new worker pool builder with the given name.\n"
    "\n"
    " The name is used to register the pool so that work can be sent to it.\n"
    "\n"
    " By default, the pool will have a number of workers equal to the number of\n"
    " scheduler threads available on the system (typically the number of CPU cores).\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let pool_name = process.new_name(\"pool\")\n"
    " let builder = crew.new(pool_name)\n"
    " ```\n"
).
-spec new(gleam@erlang@process:name(pool_msg())) -> builder().
new(Name) ->
    {builder, Name, crew_ffi:scheduler_count()}.

-file("src/crew.gleam", 361).
-spec receive_loop(
    list(fun(() -> FLP)),
    gleam@erlang@process:selector(FLP),
    integer(),
    list(FLP)
) -> {ok, list(FLP)} | {error, nil}.
receive_loop(Work, Selector, Timeout_end, State) ->
    case Work of
        [] ->
            {ok, State};

        [_ | Work@1] ->
            Timeout = Timeout_end - crew_ffi:system_time(),
            case gleam_erlang_ffi:select(Selector, Timeout) of
                {ok, Result} ->
                    receive_loop(
                        Work@1,
                        Selector,
                        Timeout_end,
                        [Result | State]
                    );

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

-file("src/crew.gleam", 441).
?DOC(
    " Add a worker pool channel to a selector for receiving work results.\n"
    "\n"
    " This allows you to receive work results as part of the larger message-handling\n"
    " loop. Used in conjunction with `subscribe` and `enqueue` for lower-level\n"
    " pool usage.\n"
    "\n"
    " Work results arrive in completion order. It is your responsibility to handle\n"
    " the lifecycle and ordering.\n"
    "\n"
    " This adds a specific monitor to the selector to handle the case where the pool\n"
    " goes down, so it's better to prefer few channels over many.\n"
    "\n"
    " ## Panics\n"
    " - If a worker or the pool crashes while executing work, the selector will\n"
    " panic with details about the crash.\n"
).
-spec select_map(
    gleam@erlang@process:selector(FLZ),
    channel(FMB),
    fun((FMB) -> FLZ)
) -> gleam@erlang@process:selector(FLZ).
select_map(Selector, Channel, Tagger) ->
    _pipe = Selector,
    _pipe@1 = gleam@erlang@process:select_specific_monitor(
        _pipe,
        erlang:element(4, Channel),
        fun(Down) ->
            Msg = <<"Pool exited while waiting for work to complete: "/utf8,
                (gleam@string:inspect(Down))/binary>>,
            erlang:error(#{gleam_error => panic,
                    message => Msg,
                    file => <<?FILEPATH/utf8>>,
                    module => <<"crew"/utf8>>,
                    function => <<"select_map"/utf8>>,
                    line => 451})
        end
    ),
    gleam@erlang@process:select_map(
        _pipe@1,
        erlang:element(3, Channel),
        fun(Result) -> case Result of
                {done, Value} ->
                    Tagger(crew_ffi:identity(Value));

                {worker_exited, Reason} ->
                    Msg@1 = <<"Worker exited: "/utf8,
                        (gleam@string:inspect(Reason))/binary>>,
                    erlang:error(#{gleam_error => panic,
                            message => Msg@1,
                            file => <<?FILEPATH/utf8>>,
                            module => <<"crew"/utf8>>,
                            function => <<"select_map"/utf8>>,
                            line => 458})
            end end
    ).

-file("src/crew.gleam", 517).
?DOC(
    " Submit multiple pieces of work to the pool using a subscription channel.\n"
    "\n"
    " This is a lower-level function for submitting work. Results must be handled\n"
    " using a selector with `select_map`. It is the callers responsibility to handle\n"
    " timeouts and submission order. Most users should prefer the `work*` functions.\n"
).
-spec enqueue_many(channel(FMQ), list(fun(() -> FMQ))) -> nil.
enqueue_many(Channel, Work) ->
    gleam@otp@actor:send(
        erlang:element(2, Channel),
        {enqueue, erlang:element(3, Channel), crew_ffi:identity(Work)}
    ).

-file("src/crew.gleam", 335).
-spec do_work(
    gleam@erlang@process:name(pool_msg()),
    integer(),
    list(fun(() -> FLM))
) -> list(FLM).
do_work(Pool, Timeout, Work) ->
    Channel = subscribe(Pool),
    enqueue_many(Channel, Work),
    Selector = begin
        _pipe = gleam_erlang_ffi:new_selector(),
        select_map(_pipe, Channel, fun(X) -> X end)
    end,
    Timeout_end = crew_ffi:system_time() + Timeout,
    Result = receive_loop(Work, Selector, Timeout_end, []),
    unsubscribe(Channel, Selector),
    case Result of
        {ok, Value} ->
            Value;

        {error, _} ->
            erlang:error(#{gleam_error => panic,
                    message => <<"Pool did not complete work in the alloted time"/utf8>>,
                    file => <<?FILEPATH/utf8>>,
                    module => <<"crew"/utf8>>,
                    function => <<"do_work"/utf8>>,
                    line => 357})
    end.

-file("src/crew.gleam", 246).
?DOC(
    " Execute multiple pieces of work concurrently on the pool, without\n"
    " ordering guarantees.\n"
    "\n"
    " Work is distributed among the available workers and executed concurrently.\n"
    " Results are returned in the order they complete, not the order they were\n"
    " submitted.\n"
    "\n"
    " ## Parameters\n"
    " - `pool` - The name of the pool to execute work on\n"
    " - `timeout` - Maximum time to wait for all work to complete in milliseconds\n"
    " - `work` - A list of functions containing work to be executed\n"
    "\n"
    " ## Panics\n"
    " - If the pool does not complete all work within the specified timeout\n"
    " - If the pool is not running\n"
    " - If any worker crashes while executing work\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let users = crew.work_many(pool_name, 10000, [\n"
    "   fn() { fetch_user_data(user1) },\n"
    "   fn() { fetch_user_data(user2) },\n"
    "   fn() { fetch_user_data(user3) },\n"
    " ])\n"
    " ```\n"
).
-spec work_many(
    gleam@erlang@process:name(pool_msg()),
    integer(),
    list(fun(() -> FKZ))
) -> list(FKZ).
work_many(Pool, Timeout, Work) ->
    lists:reverse(do_work(Pool, Timeout, Work)).

-file("src/crew.gleam", 211).
?DOC(
    " Execute a single piece of work on the pool and wait for the result.\n"
    "\n"
    " This function blocks until the work is completed or the timeout is reached.\n"
    " The work function is executed on one of the worker processes in the pool.\n"
    "\n"
    " ## Parameters\n"
    " - `pool` - The name of the pool to execute work on\n"
    " - `timeout` - Maximum time to wait for completion in milliseconds\n"
    " - `work` - A function containing the work to be executed\n"
    "\n"
    " ## Panics\n"
    " - If the pool does not complete the work within the specified timeout\n"
    " - If the pool is not running\n"
    " - If the worker crashes while executing the work\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let response = crew.work(pool_name, 5000, fn() {\n"
    "   httpc.send(...)\n"
    " })\n"
    " ```\n"
).
-spec work(gleam@erlang@process:name(pool_msg()), integer(), fun(() -> FKX)) -> FKX.
work(Pool, Timeout, Work) ->
    Result@1 = case work_many(Pool, Timeout, [Work]) of
        [Result] -> Result;
        _assert_fail ->
            erlang:error(#{gleam_error => let_assert,
                        message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
                        file => <<?FILEPATH/utf8>>,
                        module => <<"crew"/utf8>>,
                        function => <<"work"/utf8>>,
                        line => 216,
                        value => _assert_fail,
                        start => 6392,
                        'end' => 6446,
                        pattern_start => 6403,
                        pattern_end => 6411})
    end,
    Result@1.

-file("src/crew.gleam", 316).
?DOC(
    " Apply a function to each element of a list concurrently using the worker pool.\n"
    "\n"
    " This is similar to `list.map` but executes the mapping function\n"
    " concurrently across all workers. Results are returned in the same\n"
    " order as the input list.\n"
    "\n"
    " There's a bunch of extra overhead involved with spawning a work item\n"
    " per list element and making sure the order matches. Depending on your\n"
    " workload it might make sense to split your list into chunks first to reduce\n"
    " work queue pressure.\n"
    "\n"
    " ## Parameters\n"
    " - `list` - The list of items to map over\n"
    " - `pool` - The name of the pool to execute work on\n"
    " - `timeout` - Maximum time to wait for all mappings to complete in milliseconds\n"
    " - `fun` - The function to apply to each item\n"
    "\n"
    " ## Panics\n"
    " - If the pool does not complete all mappings within the specified timeout\n"
    " - If the pool is not running\n"
    " - If any worker crashes while executing a mapping\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let user_ids = [1, 2, 3, 4, 5]\n"
    " let users = crew.parallel_map(user_ids, pool_name, 5000, fetch_user_data)\n"
    " ```\n"
).
-spec parallel_map(
    list(FLG),
    gleam@erlang@process:name(pool_msg()),
    integer(),
    fun((FLG) -> FLJ)
) -> list(FLJ).
parallel_map(List, Pool, Timeout, Fun) ->
    Unordered = begin
        _pipe = List,
        _pipe@1 = gleam@list:index_map(
            _pipe,
            fun(Item, Index) -> fun() -> {Index, Fun(Item)} end end
        ),
        do_work(Pool, Timeout, _pipe@1)
    end,
    _pipe@2 = Unordered,
    _pipe@3 = gleam@list:sort(
        _pipe@2,
        fun(A, B) ->
            gleam@int:compare(erlang:element(1, A), erlang:element(1, B))
        end
    ),
    gleam@list:map(_pipe@3, fun(X) -> erlang:element(2, X) end).

-file("src/crew.gleam", 280).
?DOC(
    " Execute multiple pieces of work concurrently on the pool and return results\n"
    " in submission order.\n"
    "\n"
    " Work functions are distributed across available workers and executed\n"
    " concurrently, but results are reordered to match the original submission\n"
    " order before being returned.\n"
    "\n"
    " ## Parameters\n"
    " - `pool` - The name of the pool to execute work on\n"
    " - `timeout` - Maximum time to wait for all work to complete in milliseconds\n"
    " - `work` - A list of functions containing work to be executed\n"
    "\n"
    " ## Panics\n"
    " - If the pool does not complete all work within the specified timeout\n"
    " - If the pool is not running\n"
    " - If any worker crashes while executing work\n"
    "\n"
    " ## Example\n"
    "\n"
    " ```gleam\n"
    " let assert [user1, user2, user3] = crew.work_ordered(pool_name, 10000, [\n"
    "   fn() { fetch_user_data(user1) },\n"
    "   fn() { fetch_user_data(user2) },\n"
    "   fn() { fetch_user_data(user3) },\n"
    " ])\n"
    " ```\n"
).
-spec work_ordered(
    gleam@erlang@process:name(pool_msg()),
    integer(),
    list(fun(() -> FLD))
) -> list(FLD).
work_ordered(Pool, Timeout, Work) ->
    parallel_map(Work, Pool, Timeout, fun(F) -> F() end).

-file("src/crew.gleam", 508).
?DOC(
    " Submit a single piece of work to the pool using a subscription channel.\n"
    "\n"
    " This is a lower-level function for submitting work. Results must be handled\n"
    " using a selector with `select_map`. It is the callers responsibility to handle\n"
    " timeouts and submission order. Most users should prefer the `work*` functions.\n"
).
-spec enqueue(channel(FMO), fun(() -> FMO)) -> nil.
enqueue(Channel, Work) ->
    enqueue_many(Channel, [Work]).