aion_flow

-module(aion@internal@pump).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/aion/internal/pump.gleam").
-export([run/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(
    " The workflow-side query pump loop around suspending awaits.\n"
    "\n"
    " The engine answers workflow queries at yield points (AT-007 C20): when a\n"
    " query is pending for the workflow, a suspending await returns the\n"
    " sentinel `Error(\"aion_query:\" <> json)` instead of resolving. `run`\n"
    " recognises the sentinel, services the query through\n"
    " `aion_flow_query_pump` (handler lookup, try/catch, reply), and re-enters\n"
    " the same await, which re-resolves identically — pump iterations are\n"
    " invisible to history and to replay. Every other result passes through\n"
    " untouched.\n"
).

-file("src/aion/internal/pump.gleam", 21).
?DOC(
    " Run a suspending await thunk, servicing any pending queries the engine\n"
    " surfaces as `aion_query:` sentinels before the await's own resolution.\n"
    "\n"
    " The loop is tail-recursive: each serviced query is answered exactly once,\n"
    " then the await is re-entered until it resolves with a non-sentinel\n"
    " result. A query handler raise never crashes the workflow — the Erlang\n"
    " pump converts it into a `reply_query_error` and the loop continues.\n"
).
-spec run(fun(() -> {ok, binary()} | {error, binary()})) -> {ok, binary()} |
    {error, binary()}.
run(Do) ->
    case Do() of
        {error, <<"aion_query:"/utf8, Sentinel_payload/binary>>} ->
            aion_flow_query_pump:service(Sentinel_payload),
            run(Do);

        Outcome ->
            Outcome
    end.