aion_flow

-module(aion@workflow@run).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/aion/workflow/run.gleam").
-export([timestamp_to_milliseconds/1, run_with_default/2, run/1, id/0, now/0, random/0, random_int/2]).
-export_type([timestamp/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(" workflow.run (recorded activity dispatch) + now + random (determinism bindings)\n").

-opaque timestamp() :: {timestamp, integer()}.

-file("src/aion/workflow/run.gleam", 26).
?DOC(" Return the canonical millisecond representation of a deterministic timestamp.\n").
-spec timestamp_to_milliseconds(timestamp()) -> integer().
timestamp_to_milliseconds(Timestamp) ->
    erlang:element(2, Timestamp).

-file("src/aion/workflow/run.gleam", 249).
-spec activity_error(binary()) -> aion@error:activity_error().
activity_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/workflow/run.gleam", 320).
-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/workflow/run.gleam", 330).
?DOC(
    " Encode the activity's display labels as a JSON object of string values.\n"
    " The engine carries these to the worker for log and dashboard display; it\n"
    " never interprets them.\n"
).
-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/workflow/run.gleam", 376).
-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/workflow/run.gleam", 369).
-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/workflow/run.gleam", 345).
-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/workflow/run.gleam", 334).
-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/workflow/run.gleam", 279).
-spec activity_config(
    aion@activity:activity(any(), any()),
    gleam@option:option(binary())
) -> binary().
activity_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/workflow/run.gleam", 229).
?DOC(
    " Encode a runner's `ActivityError` to the prefixed reason vocabulary that\n"
    " `activity_error` below parses — the exact inverse, so kind fidelity\n"
    " (`retryable:` vs `terminal:` and friends) survives the in-VM child-process\n"
    " boundary. Detail payloads are dropped, matching what the parser\n"
    " reconstructs (`details: \"\"`) for remote failures on the same wire. An\n"
    " `ActivityEngineFailure` crosses unprefixed: the parser's fallthrough arm\n"
    " maps any unprefixed reason back to `ActivityEngineFailure`.\n"
).
-spec encode_activity_error(aion@error:activity_error()) -> binary().
encode_activity_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/workflow/run.gleam", 125).
?DOC(
    " Build the zero-argument thunk an in-VM dispatch hands the engine.\n"
    "\n"
    " The thunk closes over the typed input, the runner, and the output codec, so\n"
    " the child process needs no input decode: it runs the runner and encodes the\n"
    " outcome. Errors are encoded to the EXACT prefixed reason strings the\n"
    " `activity_error` parser below consumes, preserving `ActivityError` kind\n"
    " fidelity across the process boundary with zero new conventions.\n"
).
-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_activity_error(Runner_error)}
        end end.

-file("src/aion/workflow/run.gleam", 93).
?DOC(
    " Route one dispatch on the activity's selected execution tier.\n"
    "\n"
    " `Some(InVm)` crosses the arity-4 in-VM wire, carrying a thunk that composes\n"
    " the captured input, the runner, and the output codec — the engine records\n"
    " the same `ActivityScheduled`/`ActivityStarted` pair as a remote dispatch\n"
    " and runs the thunk in a linked child process. Every other selection\n"
    " (absence, or a remote tier) keeps today's arity-3 remote wire untouched.\n"
    " Both paths return the same correlation id and converge on the same\n"
    " `await_activity_result`, so replay is byte-identical across tiers.\n"
).
-spec dispatch(aion@activity:activity(any(), any()), binary(), binary()) -> {ok,
        binary()} |
    {error, binary()}.
dispatch(Activity_value, Encoded_input, Config) ->
    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.

-file("src/aion/workflow/run.gleam", 49).
?DOC(
    " Dispatch an activity, supplying the workflow-level default task queue used\n"
    " when the activity itself selects none.\n"
    "\n"
    " Resolution precedence (activity override > workflow default > the named\n"
    " `\"default\"` queue) is applied once at the engine schedule seam; this SDK\n"
    " only carries both unresolved selections across the boundary. A `None`\n"
    " workflow default behaves exactly as `run`.\n"
).
-spec run_with_default(
    aion@activity:activity(any(), FQX),
    gleam@option:option(binary())
) -> {ok, FQX} | {error, aion@error:activity_error()}.
run_with_default(Activity_value, Workflow_default_task_queue) ->
    Input_codec = aion@activity:input_codec(Activity_value),
    Output_codec = aion@activity:output_codec(Activity_value),
    Encoded_input = (erlang:element(2, Input_codec))(
        aion@activity:input(Activity_value)
    ),
    case dispatch(
        Activity_value,
        Encoded_input,
        activity_config(Activity_value, Workflow_default_task_queue)
    ) of
        {ok, Correlation_id} ->
            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, activity_error(Raw_error)}
            end;

        {error, Raw_error@1} ->
            {error, activity_error(Raw_error@1)}
    end.

-file("src/aion/workflow/run.gleam", 38).
?DOC(
    " Dispatch an activity through the single recorded side-effect boundary.\n"
    "\n"
    " Plain Gleam workflow code is re-run on replay. The only recorded\n"
    " side-effectful path exposed by this SDK is `run` (and later concurrency\n"
    " combinators over activities); there is intentionally no generic\n"
    " `side_effect(fn)` escape hatch. The activity input is encoded with the\n"
    " activity's input `Codec`, the engine dispatches and records via AD, and the\n"
    " returned payload is decoded with the output `Codec`.\n"
).
-spec run(aion@activity:activity(any(), FQR)) -> {ok, FQR} |
    {error, aion@error:activity_error()}.
run(Activity_value) ->
    run_with_default(Activity_value, none).

-file("src/aion/workflow/run.gleam", 146).
?DOC(
    " Return AD's recorded deterministic timestamp.\n"
    "\n"
    " Return the current workflow execution's unique identifier.\n"
    "\n"
    " The engine assigns a v4 UUID at workflow start and records it in the\n"
    " `WorkflowStarted` event. This NIF reads the identifier from the\n"
    " process's registered workflow handle, so it is stable across replay.\n"
).
-spec id() -> {ok, binary()} | {error, aion@error:engine_error()}.
id() ->
    case aion_flow_ffi:workflow_id() of
        {ok, Workflow_id} ->
            {ok, Workflow_id};

        {error, Raw_error} ->
            {error, {engine_failure, Raw_error}}
    end.

-file("src/aion/workflow/run.gleam", 192).
-spec parse_timestamp(binary()) -> {ok, timestamp()} |
    {error, aion@error:engine_error()}.
parse_timestamp(Raw) ->
    case gleam_stdlib:parse_int(Raw) of
        {ok, Milliseconds} ->
            {ok, {timestamp, Milliseconds}};

        {error, _} ->
            {error,
                {engine_failure,
                    <<"Invalid deterministic timestamp: "/utf8, Raw/binary>>}}
    end.

-file("src/aion/workflow/run.gleam", 156).
?DOC(
    " This is the only time source exposed to workflow code. Workflow authors must\n"
    " not call wall-clock APIs such as Gleam/Erlang clocks from workflow logic, as\n"
    " ambient time would desynchronise replay.\n"
).
-spec now() -> {ok, timestamp()} | {error, aion@error:engine_error()}.
now() ->
    case aion_flow_ffi:now() of
        {ok, Raw_timestamp} ->
            parse_timestamp(Raw_timestamp);

        {error, Raw_error} ->
            {error, {engine_failure, Raw_error}}
    end.

-file("src/aion/workflow/run.gleam", 202).
-spec parse_float(binary(), binary()) -> {ok, float()} |
    {error, aion@error:engine_error()}.
parse_float(Raw, Label) ->
    case gleam_stdlib:parse_float(Raw) of
        {ok, Value} ->
            {ok, Value};

        {error, _} ->
            {error,
                {engine_failure,
                    <<<<<<"Invalid deterministic "/utf8, Label/binary>>/binary,
                            ": "/utf8>>/binary,
                        Raw/binary>>}}
    end.

-file("src/aion/workflow/run.gleam", 167).
?DOC(
    " Draw a deterministic floating-point value from AD's seeded RNG.\n"
    "\n"
    " The engine keys the RNG seed on WorkflowId + RunId so replay observes the\n"
    " same sequence. Workflow authors must not call ambient entropy sources.\n"
).
-spec random() -> {ok, float()} | {error, aion@error:engine_error()}.
random() ->
    case aion_flow_ffi:random() of
        {ok, Raw_random} ->
            parse_float(Raw_random, <<"random"/utf8>>);

        {error, Raw_error} ->
            {error, {engine_failure, Raw_error}}
    end.

-file("src/aion/workflow/run.gleam", 212).
-spec parse_int(binary(), binary()) -> {ok, integer()} |
    {error, aion@error:engine_error()}.
parse_int(Raw, Label) ->
    case gleam_stdlib:parse_int(Raw) of
        {ok, Value} ->
            {ok, Value};

        {error, _} ->
            {error,
                {engine_failure,
                    <<<<<<"Invalid deterministic "/utf8, Label/binary>>/binary,
                            ": "/utf8>>/binary,
                        Raw/binary>>}}
    end.

-file("src/aion/workflow/run.gleam", 178).
?DOC(
    " Draw a deterministic integer in the engine-defined inclusive range.\n"
    "\n"
    " Values come from AD's seeded RNG through the FFI boundary; no wall-clock or\n"
    " ambient entropy binding is exposed by the SDK.\n"
).
-spec random_int(integer(), integer()) -> {ok, integer()} |
    {error, aion@error:engine_error()}.
random_int(Min, Max) ->
    case Min > Max of
        true ->
            {error,
                {engine_failure,
                    <<"Invalid deterministic random_int range: min is greater than max"/utf8>>}};

        false ->
            case aion_flow_ffi:random_int(
                erlang:integer_to_binary(Min),
                erlang:integer_to_binary(Max)
            ) of
                {ok, Raw_random} ->
                    parse_int(Raw_random, <<"random_int"/utf8>>);

                {error, Raw_error} ->
                    {error, {engine_failure, Raw_error}}
            end
    end.