aion_flow

-module(aion@workflow@concurrency).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/aion/workflow/concurrency.gleam").
-export([all_with_default/2, all/1, race_with_default/2, race/1, all_settled_with_default/2, all_settled/1, map_settled_with_default/3, map_settled/2, map_with_default/3, map/2]).

-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(" Typed workflow concurrency combinators over homogeneous activity lists.\n").

-file("src/aion/workflow/concurrency.gleam", 226).
-spec decode_one(binary(), aion@codec:codec(FZH)) -> {ok, FZH} |
    {error, aion@error:activity_error()}.
decode_one(Payload, Output_codec) ->
    case (erlang:element(3, Output_codec))(Payload) of
        {ok, Output} ->
            {ok, Output};

        {error, Decode_error} ->
            {error, {activity_decode_failed, Decode_error}}
    end.

-file("src/aion/workflow/concurrency.gleam", 210).
-spec decode_many(binary(), aion@codec:codec(FZC)) -> {ok, list(FZC)} |
    {error, aion@error:activity_error()}.
decode_many(Payloads, Output_codec) ->
    case gleam@json:parse(
        Payloads,
        gleam@dynamic@decode:list(
            {decoder, fun gleam@dynamic@decode:decode_string/1}
        )
    ) of
        {ok, Encoded_payloads} ->
            gleam@list:try_map(
                Encoded_payloads,
                fun(Payload) -> decode_one(Payload, Output_codec) end
            );

        {error, _} ->
            {error,
                {activity_engine_failure,
                    <<"Invalid collect_all result envelope: "/utf8,
                        Payloads/binary>>}}
    end.

-file("src/aion/workflow/concurrency.gleam", 236).
-spec collection_id(binary(), list(binary())) -> binary().
collection_id(Prefix, Specs) ->
    <<<<Prefix/binary, ":"/utf8>>/binary,
        (erlang:integer_to_binary(erlang:length(Specs)))/binary>>.

-file("src/aion/workflow/concurrency.gleam", 187).
-spec activity_spec(
    aion@activity:activity(any(), any()),
    integer(),
    gleam@option:option(binary())
) -> binary().
activity_spec(Activity_value, Index, Workflow_default_task_queue) ->
    Input_codec = aion@activity:input_codec(Activity_value),
    Encoded_input = (erlang:element(2, Input_codec))(
        aion@activity:input(Activity_value)
    ),
    _pipe = gleam@json:object(
        [{<<"correlation"/utf8>>,
                gleam@json:string(
                    <<"activity-"/utf8,
                        (erlang:integer_to_binary(Index))/binary>>
                )},
            {<<"name"/utf8>>,
                gleam@json:string(aion@activity:name(Activity_value))},
            {<<"input"/utf8>>, gleam@json:string(Encoded_input)},
            {<<"config"/utf8>>,
                gleam@json:string(
                    aion@internal@activity_dispatch:config(
                        Activity_value,
                        Workflow_default_task_queue
                    )
                )}]
    ),
    gleam@json:to_string(_pipe).

-file("src/aion/workflow/concurrency.gleam", 177).
-spec activity_specs(
    list(aion@activity:activity(any(), any())),
    gleam@option:option(binary())
) -> list(binary()).
activity_specs(Activities, Workflow_default_task_queue) ->
    _pipe = Activities,
    gleam@list:index_map(
        _pipe,
        fun(Activity_value, Index) ->
            activity_spec(Activity_value, Index, Workflow_default_task_queue)
        end
    ).

-file("src/aion/workflow/concurrency.gleam", 35).
?DOC(
    " `all`, supplying the workflow-level default task queue used for any member\n"
    " that selects none. Precedence (member override > workflow default > the\n"
    " named `\"default\"` queue) is resolved once at the engine schedule seam.\n"
).
-spec all_with_default(
    list(aion@activity:activity(any(), FVQ)),
    gleam@option:option(binary())
) -> {ok, list(FVQ)} | {error, aion@error:activity_error()}.
all_with_default(Activities, Workflow_default_task_queue) ->
    case Activities of
        [] ->
            {ok, []};

        [First | _] ->
            Output_codec = aion@activity:output_codec(First),
            Specs = activity_specs(Activities, Workflow_default_task_queue),
            Id = collection_id(<<"all"/utf8>>, Specs),
            case aion@internal@pump:run(
                fun() ->
                    aion@internal@pump:shield(
                        aion_flow_ffi:collect_all(Id, Specs)
                    )
                end
            ) of
                {ok, Payloads} ->
                    decode_many(Payloads, Output_codec);

                {error, Raw_error} ->
                    {error,
                        aion@internal@activity_dispatch:parse_error(Raw_error)}
            end
    end.

-file("src/aion/workflow/concurrency.gleam", 26).
?DOC(
    " Spawn all activities concurrently and collect their typed outputs in input\n"
    " order.\n"
    "\n"
    " Activity inputs are encoded with each activity's input `Codec`; returned\n"
    " payloads are decoded one-by-one with the homogeneous output `Codec`. AT owns\n"
    " selective receive, fail-fast behaviour, correlation, and cancellation of\n"
    " remaining activities when any activity fails.\n"
    "\n"
    " The collect is a yield point: pending workflow queries are serviced by the\n"
    " query pump before the fan-out settles, exactly as activity awaits, signal\n"
    " receives, timers, and child awaits do.\n"
).
-spec all(list(aion@activity:activity(any(), FVI))) -> {ok, list(FVI)} |
    {error, aion@error:activity_error()}.
all(Activities) ->
    all_with_default(Activities, none).

-file("src/aion/workflow/concurrency.gleam", 71).
?DOC(
    " `race`, supplying the workflow-level default task queue used for any member\n"
    " that selects none. Precedence (member override > workflow default > the\n"
    " named `\"default\"` queue) is resolved once at the engine schedule seam.\n"
).
-spec race_with_default(
    list(aion@activity:activity(any(), FWG)),
    gleam@option:option(binary())
) -> {ok, FWG} | {error, aion@error:activity_error()}.
race_with_default(Activities, Workflow_default_task_queue) ->
    case Activities of
        [] ->
            {error,
                {activity_engine_failure,
                    <<"race requires at least one activity"/utf8>>}};

        [First | _] ->
            Output_codec = aion@activity:output_codec(First),
            Specs = activity_specs(Activities, Workflow_default_task_queue),
            Id = collection_id(<<"race"/utf8>>, Specs),
            case aion@internal@pump:run(
                fun() ->
                    aion@internal@pump:shield(
                        aion_flow_ffi:collect_race(Id, Specs)
                    )
                end
            ) of
                {ok, Payload} ->
                    decode_one(Payload, Output_codec);

                {error, Raw_error} ->
                    {error,
                        aion@internal@activity_dispatch:parse_error(Raw_error)}
            end
    end.

-file("src/aion/workflow/concurrency.gleam", 62).
?DOC(
    " Race activities and return the first settled typed result.\n"
    "\n"
    " This is FIRST SETTLE semantics, not first-success-wins: the first activity to\n"
    " finish wins whether it completes successfully or returns an `ActivityError`.\n"
    " AT records that winner and cancels the losers.\n"
    "\n"
    " Like `all`, the race is a query-pump yield point: pending workflow queries\n"
    " are serviced while the race is parked.\n"
).
-spec race(list(aion@activity:activity(any(), FVZ))) -> {ok, FVZ} |
    {error, aion@error:activity_error()}.
race(Activities) ->
    race_with_default(Activities, none).

-file("src/aion/workflow/concurrency.gleam", 124).
?DOC(
    " `all_settled`, supplying the workflow-level default task queue used for\n"
    " any member that selects none. Precedence (member override > workflow\n"
    " default > the named `\"default\"` queue) is resolved once at the engine\n"
    " schedule seam.\n"
).
-spec all_settled_with_default(
    list(aion@activity:activity(any(), FWW)),
    gleam@option:option(binary())
) -> list({ok, FWW} | {error, aion@error:activity_error()}).
all_settled_with_default(Activities, Workflow_default_task_queue) ->
    aion@internal@activity_dispatch:all_settled(
        Activities,
        Workflow_default_task_queue
    ).

-file("src/aion/workflow/concurrency.gleam", 114).
?DOC(
    " Spawn all activities concurrently and settle every member independently:\n"
    " one `Result` slot per activity, in input order, with NO fail-fast and NO\n"
    " sibling cancellation.\n"
    "\n"
    " Each member dispatches through the same single-dispatch wire `run` uses\n"
    " (tier-aware: an in-VM selection crosses the arity-4 thunk wire, everything\n"
    " else the arity-3 remote wire), collecting correlation ids in input order;\n"
    " each id is then awaited in input order. Remote members run retry and timeout\n"
    " policies to their own final outcome. The current arity-4 in-VM wire applies\n"
    " neither policy, so an in-VM member declaring retry or timeout is not\n"
    " dispatched and its slot is an explicit `ActivityEngineFailure`. Policy-free\n"
    " in-VM members remain supported. A terminal failure arrives as `Error(...)`\n"
    " in that member's slot while its siblings keep their own results. Empty list\n"
    " settles to `[]`.\n"
    "\n"
    " Every await is a query-pump yield point, exactly like `run`.\n"
    "\n"
    " CAUTION for hand-authored workflows: do not wrap a settled fan-out in a\n"
    " `with_timeout` scope. Scope expiry cancels only the operation being\n"
    " awaited; members dispatched but not yet awaited when the scope expires\n"
    " have no cancellation story on this wire.\n"
).
-spec all_settled(list(aion@activity:activity(any(), FWO))) -> list({ok, FWO} |
    {error, aion@error:activity_error()}).
all_settled(Activities) ->
    all_settled_with_default(Activities, none).

-file("src/aion/workflow/concurrency.gleam", 143).
?DOC(
    " `map_settled`, supplying the workflow-level default task queue used for\n"
    " any produced activity that selects none. See `all_settled_with_default`\n"
    " for the resolution precedence.\n"
).
-spec map_settled_with_default(
    list(FXN),
    fun((FXN) -> aion@activity:activity(any(), FXQ)),
    gleam@option:option(binary())
) -> list({ok, FXQ} | {error, aion@error:activity_error()}).
map_settled_with_default(Items, To_activity, Workflow_default_task_queue) ->
    _pipe = Items,
    _pipe@1 = gleam@list:map(_pipe, To_activity),
    all_settled_with_default(_pipe@1, Workflow_default_task_queue).

-file("src/aion/workflow/concurrency.gleam", 133).
?DOC(
    " Dynamically produce one activity per input element, then settle like\n"
    " `all_settled`: one `Result` slot per item, item order, no fail-fast.\n"
).
-spec map_settled(list(FXE), fun((FXE) -> aion@activity:activity(any(), FXH))) -> list({ok,
        FXH} |
    {error, aion@error:activity_error()}).
map_settled(Items, To_activity) ->
    map_settled_with_default(Items, To_activity, none).

-file("src/aion/workflow/concurrency.gleam", 167).
?DOC(
    " `map`, supplying the workflow-level default task queue used for any produced\n"
    " activity that selects none. Precedence (activity override > workflow default\n"
    " > the named `\"default\"` queue) is resolved once at the engine schedule seam.\n"
).
-spec map_with_default(
    list(FYG),
    fun((FYG) -> aion@activity:activity(any(), FYJ)),
    gleam@option:option(binary())
) -> {ok, list(FYJ)} | {error, aion@error:activity_error()}.
map_with_default(Items, To_activity, Workflow_default_task_queue) ->
    _pipe = Items,
    _pipe@1 = gleam@list:map(_pipe, To_activity),
    all_with_default(_pipe@1, Workflow_default_task_queue).

-file("src/aion/workflow/concurrency.gleam", 157).
?DOC(
    " Dynamically produce one activity per input element, then collect like `all`.\n"
    "\n"
    " The v1 concurrency surface intentionally covers homogeneous-output list\n"
    " fan-out. Typed tuple variants such as `all2`/`all3` are deferred additions.\n"
).
-spec map(list(FXX), fun((FXX) -> aion@activity:activity(any(), FYA))) -> {ok,
        list(FYA)} |
    {error, aion@error:activity_error()}.
map(Items, To_activity) ->
    map_with_default(Items, To_activity, none).