-module(lightspeed@pipeline@operator).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/pipeline/operator.gleam").
-export([new/0, audits/1, valid/1, action_label/1, apply/3, can_enqueue/1, can_start_work/1, state/1, state_label/1, audit_event_label/1, signature/1]).
-export_type([control_state/0, action/0, audit_event/0, runtime/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(" Operator control workflows for pause/resume/drain/replay orchestration.\n").

-type control_state() :: running |
    {paused, binary(), integer()} |
    {draining, binary(), integer()} |
    {replaying, binary(), binary(), integer()}.

-type action() :: {pause, binary()} |
    {resume, binary()} |
    {drain, binary()} |
    {replay, binary(), binary()}.

-type audit_event() :: {audit_event,
        action(),
        control_state(),
        control_state(),
        integer()}.

-opaque runtime() :: {runtime, control_state(), list(audit_event())}.

-file("src/lightspeed/pipeline/operator.gleam", 38).
?DOC(" Build one operator runtime.\n").
-spec new() -> runtime().
new() ->
    {runtime, running, []}.

-file("src/lightspeed/pipeline/operator.gleam", 93).
?DOC(" Audit events in application order.\n").
-spec audits(runtime()) -> list(audit_event()).
audits(Runtime) ->
    lists:reverse(erlang:element(3, Runtime)).

-file("src/lightspeed/pipeline/operator.gleam", 196).
-spec state_valid(control_state()) -> boolean().
state_valid(State) ->
    case State of
        running ->
            true;

        {paused, Reason, At_ms} ->
            (Reason /= <<""/utf8>>) andalso (At_ms >= 0);

        {draining, Reason@1, At_ms@1} ->
            (Reason@1 /= <<""/utf8>>) andalso (At_ms@1 >= 0);

        {replaying, From_stage, Reason@2, At_ms@2} ->
            ((From_stage /= <<""/utf8>>) andalso (Reason@2 /= <<""/utf8>>))
            andalso (At_ms@2 >= 0)
    end.

-file("src/lightspeed/pipeline/operator.gleam", 206).
-spec audit_events_valid(list(audit_event())) -> boolean().
audit_events_valid(Events) ->
    case Events of
        [] ->
            true;

        [Event | Rest] ->
            (((erlang:element(5, Event) >= 0) andalso state_valid(
                erlang:element(3, Event)
            ))
            andalso state_valid(erlang:element(4, Event)))
            andalso audit_events_valid(Rest)
    end.

-file("src/lightspeed/pipeline/operator.gleam", 43).
?DOC(" Validate operator runtime invariants.\n").
-spec valid(runtime()) -> boolean().
valid(Runtime) ->
    state_valid(erlang:element(2, Runtime)) andalso audit_events_valid(
        audits(Runtime)
    ).

-file("src/lightspeed/pipeline/operator.gleam", 110).
?DOC(" Action label.\n").
-spec action_label(action()) -> binary().
action_label(Action) ->
    case Action of
        {pause, Reason} ->
            <<"pause:"/utf8, Reason/binary>>;

        {resume, Reason@1} ->
            <<"resume:"/utf8, Reason@1/binary>>;

        {drain, Reason@2} ->
            <<"drain:"/utf8, Reason@2/binary>>;

        {replay, From_stage, Reason@3} ->
            <<<<<<"replay:"/utf8, From_stage/binary>>/binary, ":"/utf8>>/binary,
                Reason@3/binary>>
    end.

-file("src/lightspeed/pipeline/operator.gleam", 139).
-spec next_state(control_state(), action(), integer()) -> {ok, control_state()} |
    {error, binary()}.
next_state(Current, Action, At_ms) ->
    case {Current, Action} of
        {running, {pause, Reason}} ->
            {ok, {paused, Reason, At_ms}};

        {running, {drain, Reason@1}} ->
            {ok, {draining, Reason@1, At_ms}};

        {running, {replay, From_stage, Reason@2}} ->
            case From_stage /= <<""/utf8>> of
                true ->
                    {ok, {replaying, From_stage, Reason@2, At_ms}};

                false ->
                    {error, <<"invalid_replay_stage"/utf8>>}
            end;

        {running, {resume, Reason@3}} ->
            {error,
                <<"invalid_transition:running:"/utf8,
                    (action_label({resume, Reason@3}))/binary>>};

        {{paused, _, _}, {resume, _}} ->
            {ok, running};

        {{paused, _, _}, {pause, Reason@4}} ->
            {error,
                <<"invalid_transition:paused:"/utf8,
                    (action_label({pause, Reason@4}))/binary>>};

        {{paused, _, _}, {drain, Reason@5}} ->
            {ok, {draining, Reason@5, At_ms}};

        {{paused, _, _}, {replay, From_stage@1, Reason@6}} ->
            case From_stage@1 /= <<""/utf8>> of
                true ->
                    {ok, {replaying, From_stage@1, Reason@6, At_ms}};

                false ->
                    {error, <<"invalid_replay_stage"/utf8>>}
            end;

        {{draining, _, _}, {resume, _}} ->
            {ok, running};

        {{draining, _, _}, {pause, Reason@7}} ->
            {ok, {paused, Reason@7, At_ms}};

        {{draining, _, _}, {drain, Reason@8}} ->
            {error,
                <<"invalid_transition:draining:"/utf8,
                    (action_label({drain, Reason@8}))/binary>>};

        {{draining, _, _}, {replay, From_stage@2, Reason@9}} ->
            case From_stage@2 /= <<""/utf8>> of
                true ->
                    {ok, {replaying, From_stage@2, Reason@9, At_ms}};

                false ->
                    {error, <<"invalid_replay_stage"/utf8>>}
            end;

        {{replaying, _, _, _}, {resume, _}} ->
            {ok, running};

        {{replaying, _, _, _}, {pause, Reason@10}} ->
            {ok, {paused, Reason@10, At_ms}};

        {{replaying, _, _, _}, {drain, Reason@11}} ->
            {ok, {draining, Reason@11, At_ms}};

        {{replaying, _, _, _}, {replay, From_stage@3, Reason@12}} ->
            case From_stage@3 /= <<""/utf8>> of
                true ->
                    {ok, {replaying, From_stage@3, Reason@12, At_ms}};

                false ->
                    {error, <<"invalid_replay_stage"/utf8>>}
            end
    end.

-file("src/lightspeed/pipeline/operator.gleam", 225).
-spec max(integer(), integer()) -> integer().
max(Left, Right) ->
    case Left >= Right of
        true ->
            Left;

        false ->
            Right
    end.

-file("src/lightspeed/pipeline/operator.gleam", 48).
?DOC(" Apply one operator action.\n").
-spec apply(runtime(), action(), integer()) -> {ok, runtime()} |
    {error, binary()}.
apply(Runtime, Action, At_ms) ->
    Clamped_at_ms = max(0, At_ms),
    case next_state(erlang:element(2, Runtime), Action, Clamped_at_ms) of
        {error, Reason} ->
            {error, Reason};

        {ok, Next} ->
            Event = {audit_event,
                Action,
                erlang:element(2, Runtime),
                Next,
                Clamped_at_ms},
            {ok, {runtime, Next, [Event | erlang:element(3, Runtime)]}}
    end.

-file("src/lightspeed/pipeline/operator.gleam", 71).
?DOC(" Whether new work can be enqueued.\n").
-spec can_enqueue(runtime()) -> boolean().
can_enqueue(Runtime) ->
    case erlang:element(2, Runtime) of
        running ->
            true;

        _ ->
            false
    end.

-file("src/lightspeed/pipeline/operator.gleam", 79).
?DOC(" Whether queued work can be started.\n").
-spec can_start_work(runtime()) -> boolean().
can_start_work(Runtime) ->
    case erlang:element(2, Runtime) of
        running ->
            true;

        {replaying, _, _, _} ->
            true;

        _ ->
            false
    end.

-file("src/lightspeed/pipeline/operator.gleam", 88).
?DOC(" Current control state.\n").
-spec state(runtime()) -> control_state().
state(Runtime) ->
    erlang:element(2, Runtime).

-file("src/lightspeed/pipeline/operator.gleam", 98).
?DOC(" Control-state label.\n").
-spec state_label(control_state()) -> binary().
state_label(State) ->
    case State of
        running ->
            <<"running"/utf8>>;

        {paused, Reason, At_ms} ->
            <<<<<<"paused:"/utf8, Reason/binary>>/binary, ":"/utf8>>/binary,
                (erlang:integer_to_binary(At_ms))/binary>>;

        {draining, Reason@1, At_ms@1} ->
            <<<<<<"draining:"/utf8, Reason@1/binary>>/binary, ":"/utf8>>/binary,
                (erlang:integer_to_binary(At_ms@1))/binary>>;

        {replaying, From_stage, Reason@2, At_ms@2} ->
            <<<<<<<<<<"replaying:"/utf8, From_stage/binary>>/binary, ":"/utf8>>/binary,
                        Reason@2/binary>>/binary,
                    ":"/utf8>>/binary,
                (erlang:integer_to_binary(At_ms@2))/binary>>
    end.

-file("src/lightspeed/pipeline/operator.gleam", 120).
?DOC(" Audit-event label.\n").
-spec audit_event_label(audit_event()) -> binary().
audit_event_label(Event) ->
    <<<<<<<<<<<<<<"action="/utf8,
                                (action_label(erlang:element(2, Event)))/binary>>/binary,
                            "|before="/utf8>>/binary,
                        (state_label(erlang:element(3, Event)))/binary>>/binary,
                    "|after="/utf8>>/binary,
                (state_label(erlang:element(4, Event)))/binary>>/binary,
            "|at_ms="/utf8>>/binary,
        (erlang:integer_to_binary(erlang:element(5, Event)))/binary>>.

-file("src/lightspeed/pipeline/operator.gleam", 217).
-spec join_with(binary(), list(binary())) -> binary().
join_with(Separator, Values) ->
    case Values of
        [] ->
            <<""/utf8>>;

        [Value] ->
            Value;

        [Value@1 | Rest] ->
            <<<<Value@1/binary, Separator/binary>>/binary,
                (join_with(Separator, Rest))/binary>>
    end.

-file("src/lightspeed/pipeline/operator.gleam", 132).
?DOC(" Stable runtime signature.\n").
-spec signature(runtime()) -> binary().
signature(Runtime) ->
    <<<<<<"state="/utf8, (state_label(erlang:element(2, Runtime)))/binary>>/binary,
            "|audits="/utf8>>/binary,
        (join_with(
            <<";"/utf8>>,
            gleam@list:map(audits(Runtime), fun audit_event_label/1)
        ))/binary>>.