aion_flow

-module(aion@internal@activity_dispatch).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/aion/internal/activity_dispatch.gleam").
-export([parse_error/1, config/2, run/2, settled_policy_error/0, settled_policy_supported/1, all_settled/2]).
-export_type([dispatched/1, settled_dispatch/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.

?MODULEDOC(false).

-opaque dispatched(EWS) :: {dispatched, binary(), aion@codec:codec(EWS)}.

-type settled_dispatch(EWT) :: {settled_dispatch, dispatched(EWT)} |
    {settled_refused, aion@error:activity_error()}.

-file("src/aion/internal/activity_dispatch.gleam", 134).
?DOC(false).
-spec parse_error(binary()) -> aion@error:activity_error().
parse_error(Raw) ->
    case gleam_stdlib:string_starts_with(Raw, <<"retryable:"/utf8>>) of
        true ->
            {retryable, gleam@string:drop_start(Raw, 10), <<""/utf8>>};

        false ->
            case gleam_stdlib:string_starts_with(Raw, <<"terminal:"/utf8>>) of
                true ->
                    {terminal, gleam@string:drop_start(Raw, 9), <<""/utf8>>};

                false ->
                    case gleam_stdlib:string_starts_with(
                        Raw,
                        <<"timeout:"/utf8>>
                    ) of
                        true ->
                            {activity_timed_out,
                                {timed_out, gleam@string:drop_start(Raw, 8)}};

                        false ->
                            case gleam_stdlib:string_starts_with(
                                Raw,
                                <<"cancelled:"/utf8>>
                            ) of
                                true ->
                                    {activity_cancelled,
                                        {cancelled,
                                            gleam@string:drop_start(Raw, 10)}};

                                false ->
                                    case gleam_stdlib:string_starts_with(
                                        Raw,
                                        <<"non_determinism:"/utf8>>
                                    ) of
                                        true ->
                                            {activity_non_deterministic,
                                                {non_determinism_violation,
                                                    gleam@string:drop_start(
                                                        Raw,
                                                        16
                                                    )}};

                                        false ->
                                            {activity_engine_failure, Raw}
                                    end
                            end
                    end
            end
    end.

-file("src/aion/internal/activity_dispatch.gleam", 92).
?DOC(false).
-spec await(dispatched(EXS)) -> {ok, EXS} | {error, aion@error:activity_error()}.
await(Dispatched) ->
    {dispatched, Correlation_id, Output_codec} = Dispatched,
    case aion@internal@pump:run(
        fun() ->
            aion@internal@pump:shield(
                aion_flow_ffi:await_activity_result(Correlation_id)
            )
        end
    ) of
        {ok, Payload} ->
            case (erlang:element(3, Output_codec))(Payload) of
                {ok, Output} ->
                    {ok, Output};

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

        {error, Raw_error} ->
            {error, parse_error(Raw_error)}
    end.

-file("src/aion/internal/activity_dispatch.gleam", 178).
?DOC(false).
-spec encode_error(aion@error:activity_error()) -> binary().
encode_error(Runner_error) ->
    case Runner_error of
        {retryable, Message, _} ->
            <<"retryable:"/utf8, Message/binary>>;

        {terminal, Message@1, _} ->
            <<"terminal:"/utf8, Message@1/binary>>;

        {activity_timed_out, {timed_out, Message@2}} ->
            <<"timeout:"/utf8, Message@2/binary>>;

        {activity_cancelled, {cancelled, Reason}} ->
            <<"cancelled:"/utf8, Reason/binary>>;

        {activity_non_deterministic, {non_determinism_violation, Message@3}} ->
            <<"non_determinism:"/utf8, Message@3/binary>>;

        {activity_decode_failed, {decode_error, Reason@1, Path}} ->
            <<<<<<"terminal:activity output decode failed at "/utf8,
                        (gleam@string:join(Path, <<"."/utf8>>))/binary>>/binary,
                    ": "/utf8>>/binary,
                Reason@1/binary>>;

        {activity_engine_failure, Message@4} ->
            Message@4
    end.

-file("src/aion/internal/activity_dispatch.gleam", 164).
?DOC(false).
-spec in_vm_thunk(aion@activity:activity(any(), any())) -> fun(() -> {ok,
        binary()} |
    {error, binary()}).
in_vm_thunk(Activity_value) ->
    Runner = aion@activity:runner(Activity_value),
    Input = aion@activity:input(Activity_value),
    Output_codec = aion@activity:output_codec(Activity_value),
    fun() -> case Runner(Input) of
            {ok, Output} ->
                {ok, (erlang:element(2, Output_codec))(Output)};

            {error, Runner_error} ->
                {error, encode_error(Runner_error)}
        end end.

-file("src/aion/internal/activity_dispatch.gleam", 231).
?DOC(false).
-spec optional_string(gleam@option:option(binary())) -> gleam@json:json().
optional_string(Value) ->
    case Value of
        none ->
            gleam@json:null();

        {some, Text} ->
            gleam@json:string(Text)
    end.

-file("src/aion/internal/activity_dispatch.gleam", 238).
?DOC(false).
-spec labels_config(list({binary(), binary()})) -> gleam@json:json().
labels_config(Labels) ->
    gleam@json:object(
        gleam@list:map(
            Labels,
            fun(Pair) ->
                {erlang:element(1, Pair),
                    gleam@json:string(erlang:element(2, Pair))}
            end
        )
    ).

-file("src/aion/internal/activity_dispatch.gleam", 284).
?DOC(false).
-spec duration_json(aion@duration:duration()) -> gleam@json:json().
duration_json(Value) ->
    _pipe = Value,
    _pipe@1 = aion@duration:to_milliseconds(_pipe),
    gleam@json:int(_pipe@1).

-file("src/aion/internal/activity_dispatch.gleam", 277).
?DOC(false).
-spec optional_duration(gleam@option:option(aion@duration:duration())) -> gleam@json:json().
optional_duration(Value) ->
    case Value of
        none ->
            gleam@json:null();

        {some, Duration} ->
            duration_json(Duration)
    end.

-file("src/aion/internal/activity_dispatch.gleam", 253).
?DOC(false).
-spec backoff_config(aion@activity:backoff()) -> gleam@json:json().
backoff_config(Backoff) ->
    case Backoff of
        {exponential, Initial, Multiplier, Max} ->
            gleam@json:object(
                [{<<"kind"/utf8>>, gleam@json:string(<<"exponential"/utf8>>)},
                    {<<"initial_ms"/utf8>>, duration_json(Initial)},
                    {<<"multiplier"/utf8>>, gleam@json:float(Multiplier)},
                    {<<"max_ms"/utf8>>, duration_json(Max)}]
            );

        {linear, Initial@1, Increment, Max@1} ->
            gleam@json:object(
                [{<<"kind"/utf8>>, gleam@json:string(<<"linear"/utf8>>)},
                    {<<"initial_ms"/utf8>>, duration_json(Initial@1)},
                    {<<"increment_ms"/utf8>>, duration_json(Increment)},
                    {<<"max_ms"/utf8>>, duration_json(Max@1)}]
            );

        {fixed, Delay} ->
            gleam@json:object(
                [{<<"kind"/utf8>>, gleam@json:string(<<"fixed"/utf8>>)},
                    {<<"delay_ms"/utf8>>, duration_json(Delay)}]
            )
    end.

-file("src/aion/internal/activity_dispatch.gleam", 242).
?DOC(false).
-spec retry_config(gleam@option:option(aion@activity:retry_policy())) -> gleam@json:json().
retry_config(Policy) ->
    case Policy of
        none ->
            gleam@json:null();

        {some, {retry_policy, Attempts, Backoff}} ->
            gleam@json:object(
                [{<<"max_attempts"/utf8>>, gleam@json:int(Attempts)},
                    {<<"backoff"/utf8>>, backoff_config(Backoff)}]
            )
    end.

-file("src/aion/internal/activity_dispatch.gleam", 199).
?DOC(false).
-spec config(
    aion@activity:activity(any(), any()),
    gleam@option:option(binary())
) -> binary().
config(Activity_value, Workflow_default_task_queue) ->
    _pipe = gleam@json:object(
        [{<<"retry"/utf8>>,
                retry_config(aion@activity:retry_policy(Activity_value))},
            {<<"timeout_ms"/utf8>>,
                optional_duration(
                    aion@activity:timeout_duration(Activity_value)
                )},
            {<<"heartbeat_ms"/utf8>>,
                optional_duration(
                    aion@activity:heartbeat_interval(Activity_value)
                )},
            {<<"labels"/utf8>>,
                labels_config(aion@activity:labels(Activity_value))},
            {<<"task_queue"/utf8>>,
                optional_string(
                    aion@activity:selected_task_queue(Activity_value)
                )},
            {<<"workflow_task_queue"/utf8>>,
                optional_string(Workflow_default_task_queue)},
            {<<"node"/utf8>>,
                optional_string(aion@activity:selected_node(Activity_value))},
            {<<"tier"/utf8>>,
                optional_string(
                    gleam@option:map(
                        aion@activity:selected_tier(Activity_value),
                        fun aion@activity:tier_to_string/1
                    )
                )}]
    ),
    gleam@json:to_string(_pipe).

-file("src/aion/internal/activity_dispatch.gleam", 62).
?DOC(false).
-spec dispatch(
    aion@activity:activity(any(), EXL),
    gleam@option:option(binary())
) -> {ok, dispatched(EXL)} | {error, aion@error:activity_error()}.
dispatch(Activity_value, Workflow_default_task_queue) ->
    Input_codec = aion@activity:input_codec(Activity_value),
    Encoded_input = (erlang:element(2, Input_codec))(
        aion@activity:input(Activity_value)
    ),
    Config = config(Activity_value, Workflow_default_task_queue),
    Dispatched = case aion@activity:selected_tier(Activity_value) of
        {some, in_vm} ->
            aion_flow_ffi:dispatch_activity_in_vm(
                aion@activity:name(Activity_value),
                Encoded_input,
                Config,
                in_vm_thunk(Activity_value)
            );

        {some, remote_python} ->
            aion_flow_ffi:dispatch_activity(
                aion@activity:name(Activity_value),
                Encoded_input,
                Config
            );

        {some, remote_rust} ->
            aion_flow_ffi:dispatch_activity(
                aion@activity:name(Activity_value),
                Encoded_input,
                Config
            );

        none ->
            aion_flow_ffi:dispatch_activity(
                aion@activity:name(Activity_value),
                Encoded_input,
                Config
            )
    end,
    case Dispatched of
        {ok, Correlation_id} ->
            {ok,
                {dispatched,
                    Correlation_id,
                    aion@activity:output_codec(Activity_value)}};

        {error, Raw_error} ->
            {error, parse_error(Raw_error)}
    end.

-file("src/aion/internal/activity_dispatch.gleam", 26).
?DOC(false).
-spec run(aion@activity:activity(any(), EWV), gleam@option:option(binary())) -> {ok,
        EWV} |
    {error, aion@error:activity_error()}.
run(Activity_value, Workflow_default_task_queue) ->
    gleam@result:'try'(
        dispatch(Activity_value, Workflow_default_task_queue),
        fun(Dispatched) -> await(Dispatched) end
    ).

-file("src/aion/internal/activity_dispatch.gleam", 127).
?DOC(false).
-spec settled_policy_error() -> aion@error:activity_error().
settled_policy_error() ->
    {activity_engine_failure,
        <<"settled in-VM activities do not support retry or timeout policies on the current arity-4 wire"/utf8>>}.

-file("src/aion/internal/activity_dispatch.gleam", 112).
?DOC(false).
-spec settled_policy_supported(aion@activity:activity(any(), any())) -> boolean().
settled_policy_supported(Activity_value) ->
    case aion@activity:selected_tier(Activity_value) of
        {some, in_vm} ->
            case {aion@activity:retry_policy(Activity_value),
                aion@activity:timeout_duration(Activity_value)} of
                {none, none} ->
                    true;

                {_, _} ->
                    false
            end;

        {some, remote_python} ->
            true;

        {some, remote_rust} ->
            true;

        none ->
            true
    end.

-file("src/aion/internal/activity_dispatch.gleam", 38).
?DOC(false).
-spec all_settled(
    list(aion@activity:activity(any(), EXC)),
    gleam@option:option(binary())
) -> list({ok, EXC} | {error, aion@error:activity_error()}).
all_settled(Activities, Workflow_default_task_queue) ->
    Dispatched = gleam@list:map(
        Activities,
        fun(Activity_value) -> case settled_policy_supported(Activity_value) of
                false ->
                    {settled_refused, settled_policy_error()};

                true ->
                    case dispatch(Activity_value, Workflow_default_task_queue) of
                        {ok, Value} ->
                            {settled_dispatch, Value};

                        {error, Failure} ->
                            {settled_refused, Failure}
                    end
            end end
    ),
    gleam@list:map(Dispatched, fun(Value@1) -> case Value@1 of
                {settled_dispatch, Dispatched@1} ->
                    await(Dispatched@1);

                {settled_refused, Failure@1} ->
                    {error, Failure@1}
            end end).