-module(lightspeed@pipeline@operations_surface).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/pipeline/operations_surface.gleam").
-export([profile_version_label/0, new/3, default_surface/0, audits/1, conformance_records/1, lineage_records/1, valid/1, start_run/2, pipeline_runtime/1, checkpoints/1, complete_run/2, apply_control/3, enqueue_batch/7, start_available/2, acknowledge_batch/3, fail_batch/5, certify_heavy_data_conformance/1, conformance_passed/1, name/1, adapter_profile_label/1, freshness/1, freshness_label/1, sink_idempotency_keys/1, telemetry_label/1, lifecycle_label/1, tenant_runtime/1, audit_signature/1, conformance_signature/1, lineage_signature/1, signature/1, fixture_snapshots/0, snapshot_signature/0]).
-export_type([adapter_profile/0, freshness_metadata/0, lineage_record/0, conformance_record/0, audit_record/0, surface/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(" Integrated data-pipeline and ETL runtime operations surface contracts (M60).\n").

-type adapter_profile() :: standard_profile |
    heavy_data_profile |
    burst_recovery_profile.

-type freshness_metadata() :: {freshness_metadata,
        integer(),
        integer(),
        integer(),
        boolean()}.

-type lineage_record() :: {lineage_record,
        binary(),
        binary(),
        binary(),
        binary(),
        binary(),
        integer(),
        binary()}.

-type conformance_record() :: {conformance_record,
        adapter_profile(),
        binary(),
        binary(),
        binary(),
        binary(),
        boolean()}.

-type audit_record() :: {audit_record, binary(), binary(), integer()}.

-opaque surface() :: {surface,
        binary(),
        adapter_profile(),
        lightspeed@pipeline@orchestrator:runtime(),
        lightspeed@tenant@policy:runtime(),
        freshness_metadata(),
        list(lineage_record()),
        list(conformance_record()),
        list(audit_record())}.

-file("src/lightspeed/pipeline/operations_surface.gleam", 81).
?DOC(" Stable M60 profile version label.\n").
-spec profile_version_label() -> binary().
profile_version_label() ->
    <<"m60.profile.v"/utf8, (erlang:integer_to_binary(1))/binary>>.

-file("src/lightspeed/pipeline/operations_surface.gleam", 707).
-spec default_tenant_runtime() -> lightspeed@tenant@policy:runtime().
default_tenant_runtime() ->
    lightspeed@tenant@policy:new(
        lightspeed@tenant@policy:tenant_context(
            <<"ops-bot"/utf8>>,
            <<"tenant-orders"/utf8>>,
            tenant_admin
        ),
        lightspeed@tenant@policy:expanded_budget(16, 6, 8, 12, 6, 10)
    ).

-file("src/lightspeed/pipeline/operations_surface.gleam", 714).
-spec freshness_budget_ms(adapter_profile()) -> integer().
freshness_budget_ms(Profile) ->
    case Profile of
        standard_profile ->
            120000;

        heavy_data_profile ->
            180000;

        burst_recovery_profile ->
            240000
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 696).
-spec backpressure_boundary_for_profile(adapter_profile()) -> lightspeed@async@backpressure:boundary().
backpressure_boundary_for_profile(Profile) ->
    case Profile of
        standard_profile ->
            lightspeed@async@backpressure:default_boundary();

        heavy_data_profile ->
            lightspeed@async@backpressure:boundary(push_pull, 5, 12, 400);

        burst_recovery_profile ->
            lightspeed@async@backpressure:boundary(pull_only, 6, 16, 350)
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 610).
-spec connector_plan_for_profile(adapter_profile()) -> lightspeed@pipeline@connector:connector_plan().
connector_plan_for_profile(Profile) ->
    case Profile of
        standard_profile ->
            lightspeed@pipeline@connector:connector_plan(
                <<"orders_standard_plan"/utf8>>,
                lightspeed@pipeline@connector:queue_source(
                    <<"orders_queue"/utf8>>,
                    <<"orders.raw"/utf8>>,
                    6
                ),
                lightspeed@pipeline@connector:database_sink(
                    <<"orders_store"/utf8>>,
                    <<"orders_projection"/utf8>>,
                    <<"order_id"/utf8>>
                ),
                {retry_policy, 3, 250, 2000},
                {dead_letter_policy, true, <<"orders.dead_letter"/utf8>>, 65536},
                none,
                500
            );

        heavy_data_profile ->
            lightspeed@pipeline@connector:connector_plan(
                <<"orders_heavy_data_plan"/utf8>>,
                lightspeed@pipeline@connector:database_source(
                    <<"orders_logical_replica"/utf8>>,
                    <<"select * from orders_stream where updated_at >= :cursor"/utf8>>,
                    1200
                ),
                lightspeed@pipeline@connector:database_sink(
                    <<"orders_analytics_store"/utf8>>,
                    <<"orders_projection_v2"/utf8>>,
                    <<"order_id"/utf8>>
                ),
                {retry_policy, 4, 500, 4000},
                {dead_letter_policy,
                    true,
                    <<"orders.analytics.dead_letter"/utf8>>,
                    131072},
                {some,
                    {reprocess_window,
                        1,
                        10000,
                        <<"heavy_data_replay_window"/utf8>>}},
                2000
            );

        burst_recovery_profile ->
            lightspeed@pipeline@connector:connector_plan(
                <<"orders_burst_recovery_plan"/utf8>>,
                lightspeed@pipeline@connector:file_source(
                    <<"orders_compact_snapshot"/utf8>>,
                    <<"/var/lib/lightspeed/orders.ndjson"/utf8>>,
                    <<"ndjson"/utf8>>,
                    1000
                ),
                lightspeed@pipeline@connector:pubsub_sink(
                    <<"orders_recovery_bus"/utf8>>,
                    <<"orders.recovery"/utf8>>,
                    <<"tenant_order"/utf8>>
                ),
                {retry_policy, 5, 200, 2500},
                {dead_letter_policy,
                    true,
                    <<"orders.recovery.dead_letter"/utf8>>,
                    98304},
                {some,
                    {reprocess_window,
                        1,
                        15000,
                        <<"burst_recovery_window"/utf8>>}},
                1600
            )
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 592).
-spec default_pipeline_runtime() -> lightspeed@pipeline:runtime().
default_pipeline_runtime() ->
    lightspeed@pipeline:new(
        lightspeed@pipeline:pipeline(
            <<"orders_pipeline_m60"/utf8>>,
            {interval, 1000},
            [lightspeed@pipeline:source_stage(
                    <<"extract_orders"/utf8>>,
                    <<"orders.raw.v2"/utf8>>
                ),
                lightspeed@pipeline:transform_stage(
                    <<"normalize_orders"/utf8>>,
                    <<"orders.raw.v2"/utf8>>,
                    <<"orders.normalized.v2"/utf8>>
                ),
                lightspeed@pipeline:sink_stage(
                    <<"write_orders"/utf8>>,
                    <<"orders.normalized.v2"/utf8>>
                )]
        )
    ).

-file("src/lightspeed/pipeline/operations_surface.gleam", 95).
?DOC(" Build a profile-specific surface with default pipeline and connector plans.\n").
-spec new(binary(), adapter_profile(), lightspeed@tenant@policy:runtime()) -> surface().
new(Name, Adapter_profile, Tenant_runtime) ->
    Orchestration_runtime = lightspeed@pipeline@orchestrator:new(
        default_pipeline_runtime(),
        connector_plan_for_profile(Adapter_profile),
        backpressure_boundary_for_profile(Adapter_profile)
    ),
    {surface,
        Name,
        Adapter_profile,
        Orchestration_runtime,
        Tenant_runtime,
        {freshness_metadata, freshness_budget_ms(Adapter_profile), 0, 0, true},
        [],
        [],
        []}.

-file("src/lightspeed/pipeline/operations_surface.gleam", 86).
?DOC(" Build a default M60 operations surface.\n").
-spec default_surface() -> surface().
default_surface() ->
    new(
        <<"orders_etl_operations_surface"/utf8>>,
        heavy_data_profile,
        default_tenant_runtime()
    ).

-file("src/lightspeed/pipeline/operations_surface.gleam", 453).
?DOC(" Audit records in emit order.\n").
-spec audits(surface()) -> list(audit_record()).
audits(Surface) ->
    lists:reverse(erlang:element(9, Surface)).

-file("src/lightspeed/pipeline/operations_surface.gleam", 898).
-spec audits_valid(list(audit_record())) -> boolean().
audits_valid(Records) ->
    case Records of
        [] ->
            true;

        [Record | Rest] ->
            (((erlang:element(2, Record) /= <<""/utf8>>) andalso (erlang:element(
                3,
                Record
            )
            /= <<""/utf8>>))
            andalso (erlang:element(4, Record) >= 0))
            andalso audits_valid(Rest)
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 448).
?DOC(" Conformance records in emit order.\n").
-spec conformance_records(surface()) -> list(conformance_record()).
conformance_records(Surface) ->
    lists:reverse(erlang:element(8, Surface)).

-file("src/lightspeed/pipeline/operations_surface.gleam", 886).
-spec conformance_records_valid(list(conformance_record())) -> boolean().
conformance_records_valid(Records) ->
    case Records of
        [] ->
            true;

        [Record | Rest] ->
            ((((erlang:element(3, Record) /= <<""/utf8>>) andalso (erlang:element(
                4,
                Record
            )
            /= <<""/utf8>>))
            andalso (erlang:element(5, Record) /= <<""/utf8>>))
            andalso (erlang:element(6, Record) /= <<""/utf8>>))
            andalso conformance_records_valid(Rest)
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 443).
?DOC(" Lineage records in emit order.\n").
-spec lineage_records(surface()) -> list(lineage_record()).
lineage_records(Surface) ->
    lists:reverse(erlang:element(7, Surface)).

-file("src/lightspeed/pipeline/operations_surface.gleam", 871).
-spec lineage_records_valid(list(lineage_record())) -> boolean().
lineage_records_valid(Records) ->
    case Records of
        [] ->
            true;

        [Record | Rest] ->
            (((((((erlang:element(2, Record) /= <<""/utf8>>) andalso (erlang:element(
                3,
                Record
            )
            /= <<""/utf8>>))
            andalso (erlang:element(4, Record) /= <<""/utf8>>))
            andalso (erlang:element(5, Record) /= <<""/utf8>>))
            andalso (erlang:element(6, Record) /= <<""/utf8>>))
            andalso (erlang:element(7, Record) > 0))
            andalso (erlang:element(8, Record) /= <<""/utf8>>))
            andalso lineage_records_valid(Rest)
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 865).
-spec freshness_valid(freshness_metadata()) -> boolean().
freshness_valid(Freshness) ->
    ((erlang:element(2, Freshness) >= 0) andalso (erlang:element(3, Freshness)
    >= 0))
    andalso (erlang:element(4, Freshness) >= 0).

-file("src/lightspeed/pipeline/operations_surface.gleam", 125).
?DOC(" Validate one operations-surface contract.\n").
-spec valid(surface()) -> boolean().
valid(Surface) ->
    ((((((erlang:element(2, Surface) /= <<""/utf8>>) andalso lightspeed@pipeline@orchestrator:valid(
        erlang:element(4, Surface)
    ))
    andalso lightspeed@tenant@policy:valid(erlang:element(5, Surface)))
    andalso freshness_valid(erlang:element(6, Surface)))
    andalso lineage_records_valid(lineage_records(Surface)))
    andalso conformance_records_valid(conformance_records(Surface)))
    andalso audits_valid(audits(Surface)).

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

        false ->
            Right
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 853).
-spec append_audit(surface(), binary(), binary(), integer()) -> surface().
append_audit(Surface, Action, Outcome, At_ms) ->
    {surface,
        erlang:element(2, Surface),
        erlang:element(3, Surface),
        erlang:element(4, Surface),
        erlang:element(5, Surface),
        erlang:element(6, Surface),
        erlang:element(7, Surface),
        erlang:element(8, Surface),
        [{audit_record, Action, Outcome, max(0, At_ms)} |
            erlang:element(9, Surface)]}.

-file("src/lightspeed/pipeline/operations_surface.gleam", 136).
?DOC(" Start one pipeline run with tenant policy gating.\n").
-spec start_run(surface(), integer()) -> {surface(),
    {ok, nil} | {error, binary()}}.
start_run(Surface, At_ms) ->
    Clamped_at_ms = max(0, At_ms),
    {Next_policy, Outcome} = lightspeed@tenant@policy:evaluate(
        erlang:element(5, Surface),
        {start_pipeline_run, 1}
    ),
    Policy_label = lightspeed@tenant@policy:outcome_label(Outcome),
    case Outcome of
        {allowed, _} ->
            Next_runtime = lightspeed@pipeline@orchestrator:start_run(
                erlang:element(4, Surface),
                Clamped_at_ms
            ),
            Next = begin
                _pipe = {surface,
                    erlang:element(2, Surface),
                    erlang:element(3, Surface),
                    Next_runtime,
                    Next_policy,
                    erlang:element(6, Surface),
                    erlang:element(7, Surface),
                    erlang:element(8, Surface),
                    erlang:element(9, Surface)},
                append_audit(
                    _pipe,
                    <<"start_run"/utf8>>,
                    <<"allowed:"/utf8, Policy_label/binary>>,
                    Clamped_at_ms
                )
            end,
            {Next, {ok, nil}};

        {denied, _} ->
            Next@1 = begin
                _pipe@1 = {surface,
                    erlang:element(2, Surface),
                    erlang:element(3, Surface),
                    erlang:element(4, Surface),
                    Next_policy,
                    erlang:element(6, Surface),
                    erlang:element(7, Surface),
                    erlang:element(8, Surface),
                    erlang:element(9, Surface)},
                append_audit(
                    _pipe@1,
                    <<"start_run"/utf8>>,
                    <<"denied:"/utf8, Policy_label/binary>>,
                    Clamped_at_ms
                )
            end,
            {Next@1, {error, <<"policy_denied:"/utf8, Policy_label/binary>>}}
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 480).
?DOC(" Runtime pipeline accessor.\n").
-spec pipeline_runtime(surface()) -> lightspeed@pipeline:runtime().
pipeline_runtime(Surface) ->
    lightspeed@pipeline@orchestrator:runtime_pipeline(
        erlang:element(4, Surface)
    ).

-file("src/lightspeed/pipeline/operations_surface.gleam", 458).
?DOC(" Pipeline checkpoints in emit order.\n").
-spec checkpoints(surface()) -> list(lightspeed@pipeline@checkpoint:checkpoint()).
checkpoints(Surface) ->
    lightspeed@pipeline:checkpoints(pipeline_runtime(Surface)).

-file("src/lightspeed/pipeline/operations_surface.gleam", 781).
-spec latest_sequence(list(lightspeed@pipeline@checkpoint:checkpoint())) -> integer().
latest_sequence(Entries) ->
    case Entries of
        [] ->
            0;

        [Entry | Rest] ->
            max(erlang:element(4, Entry), latest_sequence(Rest))
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 763).
-spec refresh_freshness_from_runtime(surface()) -> surface().
refresh_freshness_from_runtime(Surface) ->
    Runtime = pipeline_runtime(Surface),
    Telemetry = lightspeed@pipeline:runtime_telemetry(Runtime),
    Sequence = latest_sequence(checkpoints(Surface)),
    Max_allowed = erlang:element(2, erlang:element(6, Surface)),
    Observed_lag = erlang:element(2, Telemetry),
    Is_fresh = Observed_lag =< Max_allowed,
    Freshness = {freshness_metadata,
        Max_allowed,
        Observed_lag,
        Sequence,
        Is_fresh},
    {surface,
        erlang:element(2, Surface),
        erlang:element(3, Surface),
        erlang:element(4, Surface),
        erlang:element(5, Surface),
        Freshness,
        erlang:element(7, Surface),
        erlang:element(8, Surface),
        erlang:element(9, Surface)}.

-file("src/lightspeed/pipeline/operations_surface.gleam", 169).
?DOC(" Complete one pipeline run.\n").
-spec complete_run(surface(), integer()) -> surface().
complete_run(Surface, At_ms) ->
    Clamped_at_ms = max(0, At_ms),
    Next_runtime = lightspeed@pipeline@orchestrator:complete_run(
        erlang:element(4, Surface),
        Clamped_at_ms
    ),
    _pipe = {surface,
        erlang:element(2, Surface),
        erlang:element(3, Surface),
        Next_runtime,
        erlang:element(5, Surface),
        erlang:element(6, Surface),
        erlang:element(7, Surface),
        erlang:element(8, Surface),
        erlang:element(9, Surface)},
    _pipe@1 = refresh_freshness_from_runtime(_pipe),
    append_audit(
        _pipe@1,
        <<"complete_run"/utf8>>,
        <<"completed"/utf8>>,
        Clamped_at_ms
    ).

-file("src/lightspeed/pipeline/operations_surface.gleam", 816).
-spec control_policy_action(lightspeed@pipeline@operator:action()) -> gleam@option:option(lightspeed@tenant@policy:action()).
control_policy_action(Action) ->
    case Action of
        {pause, _} ->
            {some, {apply_mitigation, pause_pipelines, 1}};

        {drain, _} ->
            {some, {apply_mitigation, pause_pipelines, 1}};

        {replay, _, _} ->
            {some, {replay_pipeline_run, 1}};

        {resume, _} ->
            none
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 797).
-spec evaluate_control_policy(
    lightspeed@tenant@policy:runtime(),
    lightspeed@pipeline@operator:action()
) -> {lightspeed@tenant@policy:runtime(), {ok, nil} | {error, binary()}}.
evaluate_control_policy(Runtime, Action) ->
    case control_policy_action(Action) of
        none ->
            {Runtime, {ok, nil}};

        {some, Policy_action} ->
            {Next_runtime, Outcome} = lightspeed@tenant@policy:evaluate(
                Runtime,
                Policy_action
            ),
            case Outcome of
                {allowed, _} ->
                    {Next_runtime, {ok, nil}};

                {denied, _} ->
                    {Next_runtime,
                        {error, lightspeed@tenant@policy:outcome_label(Outcome)}}
            end
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 180).
?DOC(" Apply one operator control action with tenant-aware replay/mitigation checks.\n").
-spec apply_control(surface(), lightspeed@pipeline@operator:action(), integer()) -> {surface(),
    {ok, nil} | {error, binary()}}.
apply_control(Surface, Action, At_ms) ->
    Clamped_at_ms = max(0, At_ms),
    {Policy_runtime, Policy_result} = evaluate_control_policy(
        erlang:element(5, Surface),
        Action
    ),
    Base_surface = {surface,
        erlang:element(2, Surface),
        erlang:element(3, Surface),
        erlang:element(4, Surface),
        Policy_runtime,
        erlang:element(6, Surface),
        erlang:element(7, Surface),
        erlang:element(8, Surface),
        erlang:element(9, Surface)},
    Action_label = <<"control:"/utf8,
        (lightspeed@pipeline@operator:action_label(Action))/binary>>,
    case Policy_result of
        {error, Reason} ->
            Next = begin
                _pipe = Base_surface,
                append_audit(
                    _pipe,
                    Action_label,
                    <<"denied:"/utf8, Reason/binary>>,
                    Clamped_at_ms
                )
            end,
            {Next, {error, <<"policy_denied:"/utf8, Reason/binary>>}};

        {ok, nil} ->
            case lightspeed@pipeline@orchestrator:apply_action(
                erlang:element(4, Base_surface),
                Action,
                Clamped_at_ms
            ) of
                {Next_runtime, {ok, _}} ->
                    Next@1 = begin
                        _pipe@1 = {surface,
                            erlang:element(2, Base_surface),
                            erlang:element(3, Base_surface),
                            Next_runtime,
                            erlang:element(5, Base_surface),
                            erlang:element(6, Base_surface),
                            erlang:element(7, Base_surface),
                            erlang:element(8, Base_surface),
                            erlang:element(9, Base_surface)},
                        _pipe@2 = refresh_freshness_from_runtime(_pipe@1),
                        append_audit(
                            _pipe@2,
                            Action_label,
                            <<"applied"/utf8>>,
                            Clamped_at_ms
                        )
                    end,
                    {Next@1, {ok, nil}};

                {_, {error, Reason@1}} ->
                    Next@2 = begin
                        _pipe@3 = Base_surface,
                        append_audit(
                            _pipe@3,
                            Action_label,
                            <<"rejected:"/utf8, Reason@1/binary>>,
                            Clamped_at_ms
                        )
                    end,
                    {Next@2, {error, Reason@1}}
            end
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 224).
?DOC(" Enqueue one batch into the ETL runtime operations surface.\n").
-spec enqueue_batch(
    surface(),
    binary(),
    binary(),
    integer(),
    integer(),
    binary(),
    integer()
) -> {surface(), {ok, nil} | {error, binary()}}.
enqueue_batch(Surface, Key, Stage, Records, Lag_ms, Idempotency_key, Sequence) ->
    {Next_runtime, Result} = lightspeed@pipeline@orchestrator:enqueue_batch(
        erlang:element(4, Surface),
        Key,
        Stage,
        Records,
        Lag_ms,
        Idempotency_key,
        Sequence
    ),
    Next = {surface,
        erlang:element(2, Surface),
        erlang:element(3, Surface),
        Next_runtime,
        erlang:element(5, Surface),
        erlang:element(6, Surface),
        erlang:element(7, Surface),
        erlang:element(8, Surface),
        erlang:element(9, Surface)},
    case Result of
        {ok, _} ->
            {begin
                    _pipe = Next,
                    append_audit(
                        _pipe,
                        <<"enqueue:"/utf8, Key/binary>>,
                        <<<<<<"queued:"/utf8, Stage/binary>>/binary, ":"/utf8>>/binary,
                            (erlang:integer_to_binary(Records))/binary>>,
                        max(0, Sequence)
                    )
                end,
                {ok, nil}};

        {error, Reason} ->
            {begin
                    _pipe@1 = Next,
                    append_audit(
                        _pipe@1,
                        <<"enqueue:"/utf8, Key/binary>>,
                        <<"rejected:"/utf8, Reason/binary>>,
                        max(0, Sequence)
                    )
                end,
                {error, Reason}}
    end.

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

        [First | Rest] ->
            gleam@list:fold(
                Rest,
                First,
                fun(Accumulator, Value) ->
                    <<<<Accumulator/binary, Separator/binary>>/binary,
                        Value/binary>>
                end
            )
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 268).
?DOC(" Start as many queued batches as allowed by backpressure boundaries.\n").
-spec start_available(surface(), integer()) -> {surface(), list(binary())}.
start_available(Surface, At_ms) ->
    Clamped_at_ms = max(0, At_ms),
    {Next_runtime, Started} = lightspeed@pipeline@orchestrator:start_available(
        erlang:element(4, Surface),
        Clamped_at_ms
    ),
    Next = begin
        _pipe = {surface,
            erlang:element(2, Surface),
            erlang:element(3, Surface),
            Next_runtime,
            erlang:element(5, Surface),
            erlang:element(6, Surface),
            erlang:element(7, Surface),
            erlang:element(8, Surface),
            erlang:element(9, Surface)},
        append_audit(
            _pipe,
            <<"start_available"/utf8>>,
            <<"started="/utf8, (join_with(<<","/utf8>>, Started))/binary>>,
            Clamped_at_ms
        )
    end,
    {Next, Started}.

-file("src/lightspeed/pipeline/operations_surface.gleam", 788).
-spec latest_checkpoint(list(lightspeed@pipeline@checkpoint:checkpoint())) -> gleam@option:option(lightspeed@pipeline@checkpoint:checkpoint()).
latest_checkpoint(Entries) ->
    case lists:reverse(Entries) of
        [] ->
            none;

        [Entry | _] ->
            {some, Entry}
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 722).
-spec apply_outcome_metadata(
    surface(),
    lightspeed@pipeline@orchestrator:process_outcome()
) -> surface().
apply_outcome_metadata(Surface, Outcome) ->
    case Outcome of
        {processed, Result} ->
            case gleam_stdlib:string_starts_with(
                lightspeed@pipeline:process_result_label(Result),
                <<"applied:"/utf8>>
            ) of
                false ->
                    Surface;

                true ->
                    case latest_checkpoint(checkpoints(Surface)) of
                        none ->
                            Surface;

                        {some, Entry} ->
                            Plan = lightspeed@pipeline@orchestrator:runtime_connector_plan(
                                erlang:element(4, Surface)
                            ),
                            Record = {lineage_record,
                                erlang:element(2, Entry),
                                erlang:element(3, Entry),
                                lightspeed@pipeline@connector:source_label(
                                    lightspeed@pipeline@connector:source(Plan)
                                ),
                                lightspeed@pipeline@connector:sink_label(
                                    lightspeed@pipeline@connector:sink(Plan)
                                ),
                                lightspeed@pipeline@checkpoint:watermark_label(
                                    erlang:element(5, Entry)
                                ),
                                erlang:element(4, Entry),
                                lightspeed@pipeline@checkpoint:checkpoint_label(
                                    Entry
                                )},
                            {surface,
                                erlang:element(2, Surface),
                                erlang:element(3, Surface),
                                erlang:element(4, Surface),
                                erlang:element(5, Surface),
                                erlang:element(6, Surface),
                                [Record | erlang:element(7, Surface)],
                                erlang:element(8, Surface),
                                erlang:element(9, Surface)}
                    end
            end;

        _ ->
            Surface
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 287).
?DOC(" Acknowledge one processed batch and update lineage/freshness metadata.\n").
-spec acknowledge_batch(surface(), binary(), integer()) -> {surface(),
    {ok, lightspeed@pipeline@orchestrator:process_outcome()} | {error, binary()}}.
acknowledge_batch(Surface, Key, At_ms) ->
    Clamped_at_ms = max(0, At_ms),
    {Next_runtime, Result} = lightspeed@pipeline@orchestrator:ack_batch(
        erlang:element(4, Surface),
        Key,
        Clamped_at_ms
    ),
    Next = {surface,
        erlang:element(2, Surface),
        erlang:element(3, Surface),
        Next_runtime,
        erlang:element(5, Surface),
        erlang:element(6, Surface),
        erlang:element(7, Surface),
        erlang:element(8, Surface),
        erlang:element(9, Surface)},
    case Result of
        {error, Reason} ->
            {begin
                    _pipe = Next,
                    append_audit(
                        _pipe,
                        <<"ack:"/utf8, Key/binary>>,
                        <<"rejected:"/utf8, Reason/binary>>,
                        Clamped_at_ms
                    )
                end,
                {error, Reason}};

        {ok, Outcome} ->
            Next@1 = begin
                _pipe@1 = Next,
                _pipe@2 = apply_outcome_metadata(_pipe@1, Outcome),
                _pipe@3 = refresh_freshness_from_runtime(_pipe@2),
                append_audit(
                    _pipe@3,
                    <<"ack:"/utf8, Key/binary>>,
                    lightspeed@pipeline@orchestrator:process_outcome_label(
                        Outcome
                    ),
                    Clamped_at_ms
                )
            end,
            {Next@1, {ok, Outcome}}
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 320).
?DOC(" Fail one batch and apply retry/dead-letter policies.\n").
-spec fail_batch(surface(), binary(), integer(), binary(), integer()) -> {surface(),
    {ok, lightspeed@pipeline@orchestrator:process_outcome()} | {error, binary()}}.
fail_batch(Surface, Key, Attempt, Reason, At_ms) ->
    Clamped_at_ms = max(0, At_ms),
    {Next_runtime, Result} = lightspeed@pipeline@orchestrator:fail_batch(
        erlang:element(4, Surface),
        Key,
        Attempt,
        Reason,
        Clamped_at_ms
    ),
    Next = {surface,
        erlang:element(2, Surface),
        erlang:element(3, Surface),
        Next_runtime,
        erlang:element(5, Surface),
        erlang:element(6, Surface),
        erlang:element(7, Surface),
        erlang:element(8, Surface),
        erlang:element(9, Surface)},
    case Result of
        {error, Error_reason} ->
            {begin
                    _pipe = Next,
                    append_audit(
                        _pipe,
                        <<"fail:"/utf8, Key/binary>>,
                        <<"rejected:"/utf8, Error_reason/binary>>,
                        Clamped_at_ms
                    )
                end,
                {error, Error_reason}};

        {ok, Outcome} ->
            Next@1 = begin
                _pipe@1 = Next,
                _pipe@2 = refresh_freshness_from_runtime(_pipe@1),
                append_audit(
                    _pipe@2,
                    <<"fail:"/utf8, Key/binary>>,
                    lightspeed@pipeline@orchestrator:process_outcome_label(
                        Outcome
                    ),
                    Clamped_at_ms
                )
            end,
            {Next@1, {ok, Outcome}}
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 846).
-spec all_conformance_passed(list(conformance_record())) -> boolean().
all_conformance_passed(Records) ->
    case Records of
        [] ->
            true;

        [Record | Rest] ->
            erlang:element(7, Record) andalso all_conformance_passed(Rest)
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 909).
-spec bool_label(boolean()) -> binary().
bool_label(Value) ->
    case Value of
        true ->
            <<"true"/utf8>>;

        false ->
            <<"false"/utf8>>
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 827).
-spec budget_result_for_workload(
    list(lightspeed@data_plane_huge:budget_result()),
    binary()
) -> lightspeed@data_plane_huge:budget_result().
budget_result_for_workload(Results, Workload) ->
    case Results of
        [] ->
            {budget_result, Workload, false, <<"missing_budget_result"/utf8>>};

        [Result | Rest] ->
            case erlang:element(2, Result) =:= Workload of
                true ->
                    Result;

                false ->
                    budget_result_for_workload(Rest, Workload)
            end
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 364).
?DOC(" Certify source/sink conformance across heavy-data adapter profiles.\n").
-spec certify_heavy_data_conformance(surface()) -> surface().
certify_heavy_data_conformance(Surface) ->
    Plan = lightspeed@pipeline@orchestrator:runtime_connector_plan(
        erlang:element(4, Surface)
    ),
    Source_signature = lightspeed@pipeline@connector:source_label(
        lightspeed@pipeline@connector:source(Plan)
    ),
    Sink_signature = lightspeed@pipeline@connector:sink_label(
        lightspeed@pipeline@connector:sink(Plan)
    ),
    Benchmarks = lightspeed@data_plane_huge:run_huge_benchmarks(),
    Budget_results = lightspeed@data_plane_huge:evaluate_huge_budgets(
        Benchmarks,
        lightspeed@data_plane_huge:default_huge_budgets()
    ),
    Records = gleam@list:map(
        Benchmarks,
        fun(Benchmark) ->
            Budget_result = budget_result_for_workload(
                Budget_results,
                erlang:element(2, Benchmark)
            ),
            {conformance_record,
                erlang:element(3, Surface),
                Source_signature,
                Sink_signature,
                lightspeed@data_plane_huge:benchmark_signature(Benchmark),
                lightspeed@data_plane_huge:budget_result_signature(
                    Budget_result
                ),
                erlang:element(3, Budget_result)}
        end
    ),
    Passed_label = bool_label(all_conformance_passed(Records)),
    {surface,
        erlang:element(2, Surface),
        erlang:element(3, Surface),
        erlang:element(4, Surface),
        erlang:element(5, Surface),
        erlang:element(6, Surface),
        erlang:element(7, Surface),
        lists:reverse(Records),
        [{audit_record,
                <<"certify_heavy_data_conformance"/utf8>>,
                <<"passed="/utf8, Passed_label/binary>>,
                erlang:length(Records)} |
            erlang:element(9, Surface)]}.

-file("src/lightspeed/pipeline/operations_surface.gleam", 403).
?DOC(" `True` when heavy-data conformance evidence is present and all records pass.\n").
-spec conformance_passed(surface()) -> boolean().
conformance_passed(Surface) ->
    Records = conformance_records(Surface),
    case Records of
        [] ->
            false;

        _ ->
            all_conformance_passed(Records)
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 412).
?DOC(" Surface name accessor.\n").
-spec name(surface()) -> binary().
name(Surface) ->
    erlang:element(2, Surface).

-file("src/lightspeed/pipeline/operations_surface.gleam", 417).
?DOC(" Adapter-profile label.\n").
-spec adapter_profile_label(adapter_profile()) -> binary().
adapter_profile_label(Profile) ->
    case Profile of
        standard_profile ->
            <<"standard"/utf8>>;

        heavy_data_profile ->
            <<"heavy_data"/utf8>>;

        burst_recovery_profile ->
            <<"burst_recovery"/utf8>>
    end.

-file("src/lightspeed/pipeline/operations_surface.gleam", 426).
?DOC(" Freshness metadata accessor.\n").
-spec freshness(surface()) -> freshness_metadata().
freshness(Surface) ->
    erlang:element(6, Surface).

-file("src/lightspeed/pipeline/operations_surface.gleam", 431).
?DOC(" Freshness metadata label.\n").
-spec freshness_label(freshness_metadata()) -> binary().
freshness_label(Freshness) ->
    <<<<<<<<<<<<<<"max_allowed_lag_ms="/utf8,
                                (erlang:integer_to_binary(
                                    erlang:element(2, Freshness)
                                ))/binary>>/binary,
                            "|observed_lag_ms="/utf8>>/binary,
                        (erlang:integer_to_binary(erlang:element(3, Freshness)))/binary>>/binary,
                    "|last_sequence="/utf8>>/binary,
                (erlang:integer_to_binary(erlang:element(4, Freshness)))/binary>>/binary,
            "|is_fresh="/utf8>>/binary,
        (bool_label(erlang:element(5, Freshness)))/binary>>.

-file("src/lightspeed/pipeline/operations_surface.gleam", 463).
?DOC(" Sink idempotency keys in emit order.\n").
-spec sink_idempotency_keys(surface()) -> list(binary()).
sink_idempotency_keys(Surface) ->
    lightspeed@pipeline:sink_idempotency_keys(pipeline_runtime(Surface)).

-file("src/lightspeed/pipeline/operations_surface.gleam", 468).
?DOC(" Pipeline runtime telemetry label.\n").
-spec telemetry_label(surface()) -> binary().
telemetry_label(Surface) ->
    lightspeed@pipeline@telemetry:label(
        lightspeed@pipeline:runtime_telemetry(pipeline_runtime(Surface))
    ).

-file("src/lightspeed/pipeline/operations_surface.gleam", 475).
?DOC(" Pipeline lifecycle label.\n").
-spec lifecycle_label(surface()) -> binary().
lifecycle_label(Surface) ->
    lightspeed@pipeline:lifecycle_label(
        lightspeed@pipeline:lifecycle(pipeline_runtime(Surface))
    ).

-file("src/lightspeed/pipeline/operations_surface.gleam", 485).
?DOC(" Runtime policy accessor.\n").
-spec tenant_runtime(surface()) -> lightspeed@tenant@policy:runtime().
tenant_runtime(Surface) ->
    erlang:element(5, Surface).

-file("src/lightspeed/pipeline/operations_surface.gleam", 545).
?DOC(" Stable audit signature.\n").
-spec audit_signature(audit_record()) -> binary().
audit_signature(Record) ->
    <<<<<<<<<<"action="/utf8, (erlang:element(2, Record))/binary>>/binary,
                    "|outcome="/utf8>>/binary,
                (erlang:element(3, Record))/binary>>/binary,
            "|at_ms="/utf8>>/binary,
        (erlang:integer_to_binary(erlang:element(4, Record)))/binary>>.

-file("src/lightspeed/pipeline/operations_surface.gleam", 529).
?DOC(" Stable conformance signature.\n").
-spec conformance_signature(conformance_record()) -> binary().
conformance_signature(Record) ->
    <<<<<<<<<<<<<<<<<<<<<<"profile="/utf8,
                                                (adapter_profile_label(
                                                    erlang:element(2, Record)
                                                ))/binary>>/binary,
                                            "|source="/utf8>>/binary,
                                        (erlang:element(3, Record))/binary>>/binary,
                                    "|sink="/utf8>>/binary,
                                (erlang:element(4, Record))/binary>>/binary,
                            "|benchmark="/utf8>>/binary,
                        (erlang:element(5, Record))/binary>>/binary,
                    "|budget="/utf8>>/binary,
                (erlang:element(6, Record))/binary>>/binary,
            "|passed="/utf8>>/binary,
        (bool_label(erlang:element(7, Record)))/binary>>.

-file("src/lightspeed/pipeline/operations_surface.gleam", 511).
?DOC(" Stable lineage signature.\n").
-spec lineage_signature(lineage_record()) -> binary().
lineage_signature(Record) ->
    <<<<<<<<<<<<<<<<<<<<<<<<<<"run="/utf8, (erlang:element(2, Record))/binary>>/binary,
                                                    "|stage="/utf8>>/binary,
                                                (erlang:element(3, Record))/binary>>/binary,
                                            "|source="/utf8>>/binary,
                                        (erlang:element(4, Record))/binary>>/binary,
                                    "|sink="/utf8>>/binary,
                                (erlang:element(5, Record))/binary>>/binary,
                            "|watermark="/utf8>>/binary,
                        (erlang:element(6, Record))/binary>>/binary,
                    "|sequence="/utf8>>/binary,
                (erlang:integer_to_binary(erlang:element(7, Record)))/binary>>/binary,
            "|checkpoint="/utf8>>/binary,
        (erlang:element(8, Record))/binary>>.

-file("src/lightspeed/pipeline/operations_surface.gleam", 490).
?DOC(" Surface signature.\n").
-spec signature(surface()) -> binary().
signature(Surface) ->
    <<<<<<<<<<<<<<<<<<<<<<<<<<"profile="/utf8,
                                                        (adapter_profile_label(
                                                            erlang:element(
                                                                3,
                                                                Surface
                                                            )
                                                        ))/binary>>/binary,
                                                    "|orchestrator="/utf8>>/binary,
                                                (lightspeed@pipeline@orchestrator:signature(
                                                    erlang:element(4, Surface)
                                                ))/binary>>/binary,
                                            "|policy="/utf8>>/binary,
                                        (lightspeed@tenant@policy:signature(
                                            erlang:element(5, Surface)
                                        ))/binary>>/binary,
                                    "|freshness="/utf8>>/binary,
                                (freshness_label(erlang:element(6, Surface)))/binary>>/binary,
                            "|lineage="/utf8>>/binary,
                        (join_with(
                            <<";"/utf8>>,
                            gleam@list:map(
                                lineage_records(Surface),
                                fun lineage_signature/1
                            )
                        ))/binary>>/binary,
                    "|conformance="/utf8>>/binary,
                (join_with(
                    <<";"/utf8>>,
                    gleam@list:map(
                        conformance_records(Surface),
                        fun conformance_signature/1
                    )
                ))/binary>>/binary,
            "|audits="/utf8>>/binary,
        (join_with(
            <<";"/utf8>>,
            gleam@list:map(audits(Surface), fun audit_signature/1)
        ))/binary>>.

-file("src/lightspeed/pipeline/operations_surface.gleam", 555).
?DOC(" Deterministic fixture snapshots for M60 drift gates.\n").
-spec fixture_snapshots() -> list({binary(), binary()}).
fixture_snapshots() ->
    Baseline = begin
        _pipe = default_surface(),
        certify_heavy_data_conformance(_pipe)
    end,
    [{<<"default_surface"/utf8>>, signature(Baseline)},
        {<<"strict_policy_surface"/utf8>>,
            signature(
                new(
                    <<"strict_policy_surface"/utf8>>,
                    standard_profile,
                    lightspeed@tenant@policy:new(
                        lightspeed@tenant@policy:tenant_context(
                            <<"ops-strict"/utf8>>,
                            <<"tenant-orders"/utf8>>,
                            editor
                        ),
                        lightspeed@tenant@policy:expanded_budget(
                            2,
                            1,
                            1,
                            1,
                            0,
                            0
                        )
                    )
                )
            )},
        {<<"burst_recovery_surface"/utf8>>,
            signature(
                new(
                    <<"burst_recovery_surface"/utf8>>,
                    burst_recovery_profile,
                    default_tenant_runtime()
                )
            )}].

-file("src/lightspeed/pipeline/operations_surface.gleam", 582).
?DOC(" Deterministic M60 snapshot signature.\n").
-spec snapshot_signature() -> binary().
snapshot_signature() ->
    Entries = gleam@list:map(
        fixture_snapshots(),
        fun(Entry) ->
            {Label, Value} = Entry,
            <<<<Label/binary, "="/utf8>>/binary, Value/binary>>
        end
    ),
    <<<<(profile_version_label())/binary, "|"/utf8>>/binary,
        (join_with(<<";"/utf8>>, Entries))/binary>>.