-module(lightspeed@ops@huge_data_harness).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/ops/huge_data_harness.gleam").
-export([budget_version_label/0, run_scenario/1, run_matrix/0, scenario_label/1, pass_fail_label/1, signature/1, scenario/1, deterministic/1, outcomes/1, failed_scenarios/1, nondeterministic_failures/1, report_signature/1, snapshot_signature/0, snapshot_report_markdown/0]).
-export_type([scenario/0, scenario_outcome/0, report/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(" Deterministic data-plane 2.0 conformance harness for M51.\n").

-type scenario() :: million_row_steady_state_incremental |
    burst_ordering_correctness |
    compact_transport_coverage |
    heavy_budget_enforcement |
    regression_gate_determinism.

-type scenario_outcome() :: {scenario_outcome,
        scenario(),
        boolean(),
        boolean(),
        binary()}.

-type report() :: {report, list(scenario_outcome()), integer(), integer()}.

-file("src/lightspeed/ops/huge_data_harness.gleam", 391).
-spec count_nondeterministic(list(scenario_outcome())) -> integer().
count_nondeterministic(Outcomes) ->
    case Outcomes of
        [] ->
            0;

        [Outcome | Rest] ->
            case erlang:element(4, Outcome) of
                true ->
                    count_nondeterministic(Rest);

                false ->
                    1 + count_nondeterministic(Rest)
            end
    end.

-file("src/lightspeed/ops/huge_data_harness.gleam", 380).
-spec count_failed(list(scenario_outcome())) -> integer().
count_failed(Outcomes) ->
    case Outcomes of
        [] ->
            0;

        [Outcome | Rest] ->
            case erlang:element(3, Outcome) of
                true ->
                    count_failed(Rest);

                false ->
                    1 + count_failed(Rest)
            end
    end.

-file("src/lightspeed/ops/huge_data_harness.gleam", 371).
-spec window_result_signature(
    {ok, lightspeed@data_plane_huge:window()} |
        {error, lightspeed@data_plane_huge:query_error()}
) -> binary().
window_result_signature(Result) ->
    case Result of
        {ok, Window} ->
            lightspeed@data_plane_huge:window_signature(Window);

        {error, Error} ->
            <<"error:"/utf8,
                (lightspeed@data_plane_huge:query_error_label(Error))/binary>>
    end.

-file("src/lightspeed/ops/huge_data_harness.gleam", 409).
-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/ops/huge_data_harness.gleam", 332).
-spec determinism_signature() -> binary().
determinism_signature() ->
    Plane = lightspeed@data_plane_huge:million_row_plane(
        <<"#determinism"/utf8>>,
        million_row_list,
        lightspeed@data_plane_huge:default_transport_profile()
    ),
    Request = lightspeed@data_plane_huge:request(300000, 900, 60),
    Batch = lightspeed@data_plane_huge:burst_batch(
        <<"determinism-a"/utf8>>,
        360,
        190,
        125
    ),
    {Next_plane, Next_window, Patch} = lightspeed@data_plane_huge:apply_burst_updates(
        Plane,
        Request,
        Batch
    ),
    Benchmarks = join_with(
        <<";"/utf8>>,
        gleam@list:map(
            lightspeed@data_plane_huge:run_huge_benchmarks(),
            fun lightspeed@data_plane_huge:benchmark_signature/1
        )
    ),
    <<<<<<<<<<<<<<"plane="/utf8,
                                (lightspeed@data_plane_huge:signature(
                                    Next_plane
                                ))/binary>>/binary,
                            "|window="/utf8>>/binary,
                        (window_result_signature(Next_window))/binary>>/binary,
                    "|patch="/utf8>>/binary,
                (lightspeed@data_plane_huge:patch_summary_signature(Patch))/binary>>/binary,
            "|benchmarks="/utf8>>/binary,
        Benchmarks/binary>>.

-file("src/lightspeed/ops/huge_data_harness.gleam", 324).
-spec evaluate_regression_gate_determinism() -> {boolean(), binary()}.
evaluate_regression_gate_determinism() ->
    First = determinism_signature(),
    Second = determinism_signature(),
    Passed = First =:= Second,
    {Passed, First}.

-file("src/lightspeed/ops/huge_data_harness.gleam", 75).
?DOC(" M51 budget profile version label.\n").
-spec budget_version_label() -> binary().
budget_version_label() ->
    <<"m51.budget.v"/utf8, (erlang:integer_to_binary(1))/binary>>.

-file("src/lightspeed/ops/huge_data_harness.gleam", 298).
-spec evaluate_heavy_budget_enforcement() -> {boolean(), binary()}.
evaluate_heavy_budget_enforcement() ->
    Benchmarks = lightspeed@data_plane_huge:run_huge_benchmarks(),
    Budgets = lightspeed@data_plane_huge:default_huge_budgets(),
    Results = lightspeed@data_plane_huge:evaluate_huge_budgets(
        Benchmarks,
        Budgets
    ),
    Failures = lightspeed@data_plane_huge:budget_failures(Results),
    Passed = Failures =:= 0,
    {Passed,
        <<<<<<<<<<<<<<"budget="/utf8, (budget_version_label())/binary>>/binary,
                                "|benchmarks="/utf8>>/binary,
                            (join_with(
                                <<";"/utf8>>,
                                gleam@list:map(
                                    Benchmarks,
                                    fun lightspeed@data_plane_huge:benchmark_signature/1
                                )
                            ))/binary>>/binary,
                        "|results="/utf8>>/binary,
                    (join_with(
                        <<";"/utf8>>,
                        gleam@list:map(
                            Results,
                            fun lightspeed@data_plane_huge:budget_result_signature/1
                        )
                    ))/binary>>/binary,
                "|failures="/utf8>>/binary,
            (erlang:integer_to_binary(Failures))/binary>>}.

-file("src/lightspeed/ops/huge_data_harness.gleam", 274).
-spec evaluate_compact_transport_coverage() -> {boolean(), binary()}.
evaluate_compact_transport_coverage() ->
    Profiles = lightspeed@data_plane_huge:compact_transport_profiles(),
    Valid_profiles = gleam@list:map(
        Profiles,
        fun lightspeed@data_plane_huge:compact_transport_valid/1
    ),
    Labels = gleam@list:map(
        Profiles,
        fun lightspeed@data_plane_huge:transport_label/1
    ),
    Passed = ((((erlang:length(Profiles) =:= 3) andalso (Valid_profiles =:= [true,
        true,
        true]))
    andalso gleam@list:any(
        Labels,
        fun(Label) ->
            Label =:= <<"ws_compact_delta|encoding=binary|frame=4096|chunk=2000|compact=delta_keys"/utf8>>
        end
    ))
    andalso gleam@list:any(
        Labels,
        fun(Label@1) ->
            Label@1 =:= <<"sse_compact_json|encoding=json|frame=6144|chunk=1500|compact=columnar_json"/utf8>>
        end
    ))
    andalso gleam@list:any(
        Labels,
        fun(Label@2) ->
            Label@2 =:= <<"longpoll_compact_delta|encoding=json|frame=3072|chunk=1200|compact=delta_keys"/utf8>>
        end
    ),
    {Passed, <<"profiles="/utf8, (join_with(<<";"/utf8>>, Labels))/binary>>}.

-file("src/lightspeed/ops/huge_data_harness.gleam", 362).
-spec query_result_signature(
    {ok, lightspeed@data_plane_huge:window()} |
        {error, lightspeed@data_plane_huge:query_error()}
) -> binary().
query_result_signature(Result) ->
    case Result of
        {ok, Window} ->
            lightspeed@data_plane_huge:window_signature(Window);

        {error, Error} ->
            <<"error:"/utf8,
                (lightspeed@data_plane_huge:query_error_label(Error))/binary>>
    end.

-file("src/lightspeed/ops/huge_data_harness.gleam", 229).
-spec evaluate_burst_ordering_correctness() -> {boolean(), binary()}.
evaluate_burst_ordering_correctness() ->
    Plane = lightspeed@data_plane_huge:million_row_plane(
        <<"#burst-ordering"/utf8>>,
        {burst_analytics, 8},
        lightspeed@data_plane_huge:default_transport_profile()
    ),
    Request = lightspeed@data_plane_huge:request(220000, 1000, 40),
    Batch_a = lightspeed@data_plane_huge:burst_batch(
        <<"burst-a"/utf8>>,
        520,
        260,
        90
    ),
    {Plane_a, Window_a, Patch_a} = lightspeed@data_plane_huge:apply_burst_updates(
        Plane,
        Request,
        Batch_a
    ),
    Batch_b = lightspeed@data_plane_huge:burst_batch(
        <<"burst-b"/utf8>>,
        610,
        320,
        130
    ),
    {Plane_b, Window_b, Patch_b} = lightspeed@data_plane_huge:apply_burst_updates(
        Plane_a,
        Request,
        Batch_b
    ),
    Regressed = lightspeed@data_plane_huge:query_window(
        Plane_b,
        lightspeed@data_plane_huge:request(220000, 1000, 120)
    ),
    Passed = case {Window_a, Window_b, Regressed} of
        {{ok, A}, {ok, B}, {error, {sequence_regressed, _, _}}} ->
            ((((lightspeed@data_plane_huge:ordering_invariant(A) andalso lightspeed@data_plane_huge:ordering_invariant(
                B
            ))
            andalso (lightspeed@data_plane_huge:window_sequence(B) > lightspeed@data_plane_huge:window_sequence(
                A
            )))
            andalso lightspeed@data_plane_huge:steady_state_no_full_root_churn(
                Patch_a
            ))
            andalso lightspeed@data_plane_huge:steady_state_no_full_root_churn(
                Patch_b
            ))
            andalso (lightspeed@data_plane_huge:total_rows(Plane_b) =:= 1000550);

        {_, _, _} ->
            false
    end,
    {Passed,
        <<<<<<<<<<<<<<<<<<"window_a="/utf8,
                                            (window_result_signature(Window_a))/binary>>/binary,
                                        "|window_b="/utf8>>/binary,
                                    (window_result_signature(Window_b))/binary>>/binary,
                                "|regressed="/utf8>>/binary,
                            (query_result_signature(Regressed))/binary>>/binary,
                        "|patch_a="/utf8>>/binary,
                    (lightspeed@data_plane_huge:patch_summary_signature(Patch_a))/binary>>/binary,
                "|patch_b="/utf8>>/binary,
            (lightspeed@data_plane_huge:patch_summary_signature(Patch_b))/binary>>}.

-file("src/lightspeed/ops/huge_data_harness.gleam", 197).
-spec evaluate_million_row_steady_state_incremental() -> {boolean(), binary()}.
evaluate_million_row_steady_state_incremental() ->
    Plane = lightspeed@data_plane_huge:million_row_plane(
        <<"#huge-grid"/utf8>>,
        {million_row_grid, 12},
        lightspeed@data_plane_huge:default_transport_profile()
    ),
    Request = lightspeed@data_plane_huge:request(450000, 1200, 80),
    Batch = lightspeed@data_plane_huge:burst_batch(
        <<"huge-grid-a"/utf8>>,
        420,
        160,
        120
    ),
    {Next_plane, Next_window, Patch} = lightspeed@data_plane_huge:apply_burst_updates(
        Plane,
        Request,
        Batch
    ),
    Passed = case Next_window of
        {ok, Window} ->
            (((lightspeed@data_plane_huge:valid(Plane) andalso lightspeed@data_plane_huge:valid(
                Next_plane
            ))
            andalso lightspeed@data_plane_huge:steady_state_no_full_root_churn(
                Patch
            ))
            andalso lightspeed@data_plane_huge:ordering_invariant(Window))
            andalso (lightspeed@data_plane_huge:total_rows(Next_plane) =:= 1000260);

        {error, _} ->
            false
    end,
    {Passed,
        <<<<<<<<<<"patch="/utf8,
                            (lightspeed@data_plane_huge:patch_summary_signature(
                                Patch
                            ))/binary>>/binary,
                        "|window="/utf8>>/binary,
                    (window_result_signature(Next_window))/binary>>/binary,
                "|plane="/utf8>>/binary,
            (lightspeed@data_plane_huge:signature(Next_plane))/binary>>}.

-file("src/lightspeed/ops/huge_data_harness.gleam", 186).
-spec evaluate(scenario()) -> {boolean(), binary()}.
evaluate(Scenario) ->
    case Scenario of
        million_row_steady_state_incremental ->
            evaluate_million_row_steady_state_incremental();

        burst_ordering_correctness ->
            evaluate_burst_ordering_correctness();

        compact_transport_coverage ->
            evaluate_compact_transport_coverage();

        heavy_budget_enforcement ->
            evaluate_heavy_budget_enforcement();

        regression_gate_determinism ->
            evaluate_regression_gate_determinism()
    end.

-file("src/lightspeed/ops/huge_data_harness.gleam", 59).
?DOC(" Run one M51 scenario twice and require deterministic parity.\n").
-spec run_scenario(scenario()) -> scenario_outcome().
run_scenario(Scenario) ->
    {First_passed, First_signature} = evaluate(Scenario),
    {Second_passed, Second_signature} = evaluate(Scenario),
    Deterministic = (First_passed =:= Second_passed) andalso (First_signature
    =:= Second_signature),
    Passed = (First_passed andalso Second_passed) andalso Deterministic,
    {scenario_outcome, Scenario, Passed, Deterministic, First_signature}.

-file("src/lightspeed/ops/huge_data_harness.gleam", 40).
?DOC(" Run all M51 scenarios.\n").
-spec run_matrix() -> report().
run_matrix() ->
    Outcomes = begin
        _pipe = [million_row_steady_state_incremental,
            burst_ordering_correctness,
            compact_transport_coverage,
            heavy_budget_enforcement,
            regression_gate_determinism],
        gleam@list:map(_pipe, fun run_scenario/1)
    end,
    {report, Outcomes, count_failed(Outcomes), count_nondeterministic(Outcomes)}.

-file("src/lightspeed/ops/huge_data_harness.gleam", 80).
?DOC(" Scenario label.\n").
-spec scenario_label(scenario()) -> binary().
scenario_label(Scenario) ->
    case Scenario of
        million_row_steady_state_incremental ->
            <<"million_row_steady_state_incremental"/utf8>>;

        burst_ordering_correctness ->
            <<"burst_ordering_correctness"/utf8>>;

        compact_transport_coverage ->
            <<"compact_transport_coverage"/utf8>>;

        heavy_budget_enforcement ->
            <<"heavy_budget_enforcement"/utf8>>;

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

-file("src/lightspeed/ops/huge_data_harness.gleam", 91).
?DOC(" Stable pass/fail label.\n").
-spec pass_fail_label(scenario_outcome()) -> binary().
pass_fail_label(Outcome) ->
    case erlang:element(3, Outcome) of
        true ->
            <<"pass"/utf8>>;

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

-file("src/lightspeed/ops/huge_data_harness.gleam", 99).
?DOC(" Scenario signature accessor.\n").
-spec signature(scenario_outcome()) -> binary().
signature(Outcome) ->
    erlang:element(5, Outcome).

-file("src/lightspeed/ops/huge_data_harness.gleam", 104).
?DOC(" Scenario accessor.\n").
-spec scenario(scenario_outcome()) -> scenario().
scenario(Outcome) ->
    erlang:element(2, Outcome).

-file("src/lightspeed/ops/huge_data_harness.gleam", 109).
?DOC(" Determinism accessor.\n").
-spec deterministic(scenario_outcome()) -> boolean().
deterministic(Outcome) ->
    erlang:element(4, Outcome).

-file("src/lightspeed/ops/huge_data_harness.gleam", 114).
?DOC(" Report outcomes accessor.\n").
-spec outcomes(report()) -> list(scenario_outcome()).
outcomes(Report) ->
    erlang:element(2, Report).

-file("src/lightspeed/ops/huge_data_harness.gleam", 119).
?DOC(" Failed scenario count.\n").
-spec failed_scenarios(report()) -> integer().
failed_scenarios(Report) ->
    erlang:element(3, Report).

-file("src/lightspeed/ops/huge_data_harness.gleam", 124).
?DOC(" Nondeterministic scenario count.\n").
-spec nondeterministic_failures(report()) -> integer().
nondeterministic_failures(Report) ->
    erlang:element(4, Report).

-file("src/lightspeed/ops/huge_data_harness.gleam", 402).
-spec bool_label(boolean()) -> binary().
bool_label(Value) ->
    case Value of
        true ->
            <<"true"/utf8>>;

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

-file("src/lightspeed/ops/huge_data_harness.gleam", 129).
?DOC(" Stable report signature.\n").
-spec report_signature(report()) -> binary().
report_signature(Report) ->
    Entries = gleam@list:map(
        erlang:element(2, Report),
        fun(Outcome) ->
            <<<<<<<<<<<<(scenario_label(erlang:element(2, Outcome)))/binary,
                                    "="/utf8>>/binary,
                                (pass_fail_label(Outcome))/binary>>/binary,
                            ":deterministic="/utf8>>/binary,
                        (bool_label(erlang:element(4, Outcome)))/binary>>/binary,
                    ":"/utf8>>/binary,
                (erlang:element(5, Outcome))/binary>>
        end
    ),
    join_with(<<";"/utf8>>, Entries).

-file("src/lightspeed/ops/huge_data_harness.gleam", 145).
?DOC(" Deterministic snapshot signature for M51 fixture drift gates.\n").
-spec snapshot_signature() -> binary().
snapshot_signature() ->
    <<<<<<"m51.snapshot.v"/utf8, (erlang:integer_to_binary(1))/binary>>/binary,
            "|"/utf8>>/binary,
        (report_signature(run_matrix()))/binary>>.

-file("src/lightspeed/ops/huge_data_harness.gleam", 153).
?DOC(" Deterministic markdown report for M51 fixture scripts.\n").
-spec snapshot_report_markdown() -> binary().
snapshot_report_markdown() ->
    Report = run_matrix(),
    Failed = failed_scenarios(Report),
    Nondeterministic = nondeterministic_failures(Report),
    Status = case (Failed =:= 0) andalso (Nondeterministic =:= 0) of
        true ->
            <<"OK"/utf8>>;

        false ->
            <<"FAIL"/utf8>>
    end,
    <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<"# Huge Data Fixture Report\n\n"/utf8,
                                                                                        "snapshot_version: "/utf8>>/binary,
                                                                                    (erlang:integer_to_binary(
                                                                                        1
                                                                                    ))/binary>>/binary,
                                                                                "\n"/utf8>>/binary,
                                                                            "budget_version: "/utf8>>/binary,
                                                                        (budget_version_label(
                                                                            
                                                                        ))/binary>>/binary,
                                                                    "\n"/utf8>>/binary,
                                                                "status: "/utf8>>/binary,
                                                            Status/binary>>/binary,
                                                        "\n"/utf8>>/binary,
                                                    "failed_scenarios: "/utf8>>/binary,
                                                (erlang:integer_to_binary(
                                                    Failed
                                                ))/binary>>/binary,
                                            "\n"/utf8>>/binary,
                                        "nondeterministic_failures: "/utf8>>/binary,
                                    (erlang:integer_to_binary(Nondeterministic))/binary>>/binary,
                                "\n\n"/utf8>>/binary,
                            "snapshot_signature: "/utf8>>/binary,
                        (snapshot_signature())/binary>>/binary,
                    "\n\n"/utf8>>/binary,
                "report_signature: "/utf8>>/binary,
            (report_signature(Report))/binary>>/binary,
        "\n"/utf8>>.