taskle

-module(taskle).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-define(FILEPATH, "src/taskle.gleam").
-export([async/1, await/2, cancel/1, async_unlinked/1, pid/1, parallel_map/3]).
-export_type([task/1, error/0, task_message/1, await_msg/1]).

-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 task(FFL) :: {task,
        gleam@erlang@process:pid_(),
        gleam@erlang@process:monitor(),
        gleam@erlang@process:pid_(),
        gleam@erlang@process:subject(task_message(FFL))}.

-type error() :: timeout | {crashed, binary()} | not_owner.

-type task_message(FFM) :: {task_result, FFM}.

-type await_msg(FFN) :: {task_msg, task_message(FFN)} |
    {down_msg, gleam@erlang@process:down()}.

-file("src/taskle.gleam", 46).
?DOC(
    " Creates an asynchronous task that runs the given function in a separate process.\n"
    "\n"
    " The task is unlinked from the calling process, meaning that if the task crashes,\n"
    " it won't cause the calling process to crash. Only the process that creates the\n"
    " task can await its result or cancel it.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let task = taskle.async(fn() {\n"
    "   // This runs in a separate process\n"
    "   process.sleep(1000)\n"
    "   42\n"
    " })\n"
    " ```\n"
).
-spec async(fun(() -> FFO)) -> task(FFO).
async(Fun) ->
    Owner = erlang:self(),
    Subject = gleam@erlang@process:new_subject(),
    Pid = proc_lib:spawn(
        fun() ->
            Result = Fun(),
            gleam@erlang@process:send(Subject, {task_result, Result})
        end
    ),
    Monitor = gleam@erlang@process:monitor(Pid),
    {task, Pid, Monitor, Owner, Subject}.

-file("src/taskle.gleam", 76).
?DOC(
    " Waits for a task to complete with a timeout in milliseconds.\n"
    "\n"
    " Only the process that created the task can await its result. If called from\n"
    " a different process, returns `Error(NotOwner)`.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " case taskle.await(task, 5000) {\n"
    "   Ok(value) -> io.println(\"Success: \" <> int.to_string(value))\n"
    "   Error(taskle.Timeout) -> io.println(\"Task timed out after 5 seconds\")\n"
    "   Error(taskle.Crashed(reason)) -> io.println(\"Task failed: \" <> reason)\n"
    "   Error(taskle.NotOwner) -> io.println(\"Cannot await task from different process\")\n"
    " }\n"
    " ```\n"
).
-spec await(task(FFQ), integer()) -> {ok, FFQ} | {error, error()}.
await(Task, Timeout) ->
    {task, Task_pid, Monitor, Owner, Subject} = Task,
    gleam@bool:guard(
        erlang:self() /= Owner,
        {error, not_owner},
        fun() ->
            Selector = begin
                _pipe = gleam_erlang_ffi:new_selector(),
                _pipe@1 = gleam@erlang@process:select_specific_monitor(
                    _pipe,
                    Monitor,
                    fun(Field@0) -> {down_msg, Field@0} end
                ),
                gleam@erlang@process:select_map(
                    _pipe@1,
                    Subject,
                    fun(Field@0) -> {task_msg, Field@0} end
                )
            end,
            case gleam_erlang_ffi:select(Selector, Timeout) of
                {ok, {task_msg, {task_result, Value}}} ->
                    gleam@erlang@process:demonitor_process(Monitor),
                    {ok, Value};

                {ok, {down_msg, {process_down, _, _, Reason}}} ->
                    case Reason of
                        normal ->
                            {error, {crashed, <<"normal"/utf8>>}};

                        killed ->
                            {error, {crashed, <<"killed"/utf8>>}};

                        {abnormal, Reason@1} ->
                            {error, {crashed, gleam@string:inspect(Reason@1)}}
                    end;

                {ok, {down_msg, {port_down, _, _, _}}} ->
                    {error, {crashed, <<"port_down"/utf8>>}};

                {error, nil} ->
                    gleam@erlang@process:kill(Task_pid),
                    gleam@erlang@process:demonitor_process(Monitor),
                    {error, timeout}
            end
        end
    ).

-file("src/taskle.gleam", 128).
?DOC(
    " Cancels a running task.\n"
    "\n"
    " Only the process that created the task can cancel it. If called from\n"
    " a different process, returns `Error(NotOwner)`.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let task = taskle.async(fn() {\n"
    "   process.sleep(10_000)\n"
    "   \"done\"\n"
    " })\n"
    "\n"
    " // Cancel the task\n"
    " case taskle.cancel(task) {\n"
    "   Ok(Nil) -> io.println(\"Task cancelled\")\n"
    "   Error(taskle.NotOwner) -> io.println(\"Cannot cancel task from different process\")\n"
    " }\n"
    " ```\n"
).
-spec cancel(task(any())) -> {ok, nil} | {error, error()}.
cancel(Task) ->
    {task, Pid, Monitor, Owner, _} = Task,
    gleam@bool:guard(
        erlang:self() /= Owner,
        {error, not_owner},
        fun() ->
            gleam@erlang@process:kill(Pid),
            gleam@erlang@process:demonitor_process(Monitor),
            {ok, nil}
        end
    ).

-file("src/taskle.gleam", 220).
?DOC(
    " Creates an asynchronous task that is not linked to the calling process.\n"
    "\n"
    " This is identical to `async`, but provided for clarity when you specifically\n"
    " want to emphasize that the task is unlinked. Use this for fire-and-forget\n"
    " scenarios where you don't want the parent process to be affected if the task crashes.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let fire_and_forget = taskle.async_unlinked(fn() {\n"
    "   // This task won't affect the parent process if it crashes\n"
    "   risky_operation()\n"
    " })\n"
    " ```\n"
).
-spec async_unlinked(fun(() -> FGR)) -> task(FGR).
async_unlinked(Fun) ->
    Owner = erlang:self(),
    Subject = gleam@erlang@process:new_subject(),
    Pid = proc_lib:spawn(
        fun() ->
            Result = Fun(),
            gleam@erlang@process:send(Subject, {task_result, Result})
        end
    ),
    Monitor = gleam@erlang@process:monitor(Pid),
    {task, Pid, Monitor, Owner, Subject}.

-file("src/taskle.gleam", 246).
?DOC(
    " Returns the process ID of the task's underlying BEAM process.\n"
    "\n"
    " Useful for debugging or process monitoring.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let task = taskle.async(fn() { 42 })\n"
    " let process_id = taskle.pid(task)\n"
    " io.debug(process_id)\n"
    " ```\n"
).
-spec pid(task(any())) -> gleam@erlang@process:pid_().
pid(Task) ->
    {task, Pid, _, _, _} = Task,
    Pid.

-file("src/taskle.gleam", 177).
-spec await_all_loop(list(task(FGK)), list(FGK), integer(), integer()) -> {ok,
        list(FGK)} |
    {error, error()}.
await_all_loop(Tasks, Results, Timeout, Start_time) ->
    case Tasks of
        [] ->
            {ok, lists:reverse(Results)};

        [Task | Rest] ->
            Elapsed = erlang:system_time() - Start_time,
            Remaining_timeout = Timeout - (Elapsed div 1000000),
            gleam@bool:lazy_guard(
                Remaining_timeout =< 0,
                fun() ->
                    gleam@list:each(Rest, fun cancel/1),
                    {error, timeout}
                end,
                fun() -> case await(Task, Remaining_timeout) of
                        {ok, Result} ->
                            await_all_loop(
                                Rest,
                                [Result | Results],
                                Timeout,
                                Start_time
                            );

                        {error, Err} ->
                            gleam@list:each(Rest, fun cancel/1),
                            {error, Err}
                    end end
            )
    end.

-file("src/taskle.gleam", 167).
-spec await_all(list(task(FGE)), integer()) -> {ok, list(FGE)} |
    {error, error()}.
await_all(Tasks, Timeout) ->
    case Tasks of
        [] ->
            {ok, []};

        _ ->
            Start_time = erlang:system_time(),
            await_all_loop(Tasks, [], Timeout, Start_time)
    end.

-file("src/taskle.gleam", 157).
?DOC(
    " Processes a list of items in parallel, applying the given function to each item.\n"
    "\n"
    " Returns when all tasks complete or when any task fails/times out. If any task\n"
    " fails, all remaining tasks are cancelled.\n"
    "\n"
    " ## Examples\n"
    "\n"
    " ```gleam\n"
    " let numbers = [1, 2, 3, 4, 5]\n"
    "\n"
    " case taskle.parallel_map(numbers, fn(x) { x * x }, 5000) {\n"
    "   Ok(results) -> {\n"
    "     // results = [1, 4, 9, 16, 25]\n"
    "     io.debug(results)\n"
    "   }\n"
    "   Error(taskle.Timeout) -> io.println(\"Some tasks timed out\")\n"
    "   Error(taskle.Crashed(reason)) -> io.println(\"A task crashed: \" <> reason)\n"
    " }\n"
    " ```\n"
).
-spec parallel_map(list(FFY), fun((FFY) -> FGA), integer()) -> {ok, list(FGA)} |
    {error, error()}.
parallel_map(List, Fun, Timeout) ->
    Tasks = gleam@list:map(List, fun(Item) -> async(fun() -> Fun(Item) end) end),
    await_all(Tasks, Timeout).