-module(lightspeed@data_plane).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/lightspeed/data_plane.gleam").
-export([new/3, row/2, request/2, valid/1, query_window/2, apply_updates/3, contains_full_root_churn/1, steady_state_incremental/1, dataset_label/1, query_error_label/1, window_signature/1, signature/1, target/1, dataset/1, rows/1]).
-export_type([dataset/0, row/0, window_request/0, window/0, update/0, query_error/0, plane/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(" Large data-plane contracts with windowed queries and incremental updates.\n").

-type dataset() :: list_dataset |
    {grid_dataset, integer()} |
    {chart_dataset, binary()}.

-type row() :: {row, binary(), binary()}.

-type window_request() :: {window_request, integer(), integer()}.

-type window() :: {window, integer(), integer(), integer(), list(row())}.

-type update() :: {upsert, row()} | {delete, binary()}.

-type query_error() :: {invalid_window, integer(), integer()}.

-opaque plane() :: {plane, binary(), dataset(), list(row())}.

-file("src/lightspeed/data_plane.gleam", 46).
?DOC(" Build one large-data plane.\n").
-spec new(binary(), dataset(), list(row())) -> plane().
new(Target, Dataset, Rows) ->
    {plane, Target, Dataset, Rows}.

-file("src/lightspeed/data_plane.gleam", 51).
?DOC(" Build one row.\n").
-spec row(binary(), binary()) -> row().
row(Id, Payload) ->
    {row, Id, Payload}.

-file("src/lightspeed/data_plane.gleam", 56).
?DOC(" Build one window request.\n").
-spec request(integer(), integer()) -> window_request().
request(Offset, Limit) ->
    {window_request, Offset, Limit}.

-file("src/lightspeed/data_plane.gleam", 197).
-spec contains_string(list(binary()), binary()) -> boolean().
contains_string(Values, Target) ->
    case Values of
        [] ->
            false;

        [Value | Rest] ->
            case Value =:= Target of
                true ->
                    true;

                false ->
                    contains_string(Rest, Target)
            end
    end.

-file("src/lightspeed/data_plane.gleam", 186).
-spec unique_ids(list(row()), list(binary())) -> boolean().
unique_ids(Rows, Seen) ->
    case Rows of
        [] ->
            true;

        [Entry | Rest] ->
            case contains_string(Seen, erlang:element(2, Entry)) of
                true ->
                    false;

                false ->
                    unique_ids(Rest, [erlang:element(2, Entry) | Seen])
            end
    end.

-file("src/lightspeed/data_plane.gleam", 178).
-spec valid_dataset(dataset()) -> boolean().
valid_dataset(Dataset) ->
    case Dataset of
        list_dataset ->
            true;

        {grid_dataset, Columns} ->
            Columns > 0;

        {chart_dataset, Series_key} ->
            Series_key /= <<""/utf8>>
    end.

-file("src/lightspeed/data_plane.gleam", 61).
?DOC(" Validate one large-data plane contract.\n").
-spec valid(plane()) -> boolean().
valid(Plane) ->
    ((erlang:element(2, Plane) /= <<""/utf8>>) andalso valid_dataset(
        erlang:element(3, Plane)
    ))
    andalso unique_ids(erlang:element(4, Plane), []).

-file("src/lightspeed/data_plane.gleam", 241).
-spec drop(list(row()), integer()) -> list(row()).
drop(Rows, Count) ->
    case Count =< 0 of
        true ->
            Rows;

        false ->
            case Rows of
                [] ->
                    [];

                [_ | Rest] ->
                    drop(Rest, Count - 1)
            end
    end.

-file("src/lightspeed/data_plane.gleam", 252).
-spec take(list(row()), integer()) -> list(row()).
take(Rows, Count) ->
    case Count =< 0 of
        true ->
            [];

        false ->
            case Rows of
                [] ->
                    [];

                [Row | Rest] ->
                    [Row | take(Rest, Count - 1)]
            end
    end.

-file("src/lightspeed/data_plane.gleam", 68).
?DOC(" Execute one windowed query.\n").
-spec query_window(plane(), window_request()) -> {ok, window()} |
    {error, query_error()}.
query_window(Plane, Request) ->
    case (erlang:element(2, Request) < 0) orelse (erlang:element(3, Request) =< 0) of
        true ->
            {error,
                {invalid_window,
                    erlang:element(2, Request),
                    erlang:element(3, Request)}};

        false ->
            Total = erlang:length(erlang:element(4, Plane)),
            Start = gleam@int:clamp(erlang:element(2, Request), 0, Total),
            Rows = take(
                drop(erlang:element(4, Plane), Start),
                erlang:element(3, Request)
            ),
            {ok, {window, Start, erlang:element(3, Request), Total, Rows}}
    end.

-file("src/lightspeed/data_plane.gleam", 263).
-spec to_keyed_nodes(list(row())) -> list(lightspeed@diff:keyed_node()).
to_keyed_nodes(Rows) ->
    case Rows of
        [] ->
            [];

        [Row | Rest] ->
            [lightspeed@diff:keyed_node(
                    erlang:element(2, Row),
                    erlang:element(3, Row)
                ) |
                to_keyed_nodes(Rest)]
    end.

-file("src/lightspeed/data_plane.gleam", 230).
-spec remove(list(row()), binary()) -> list(row()).
remove(Rows, Target_id) ->
    case Rows of
        [] ->
            [];

        [Row | Rest] ->
            case erlang:element(2, Row) =:= Target_id of
                true ->
                    remove(Rest, Target_id);

                false ->
                    [Row | remove(Rest, Target_id)]
            end
    end.

-file("src/lightspeed/data_plane.gleam", 219).
-spec upsert(list(row()), row()) -> list(row()).
upsert(Rows, Next) ->
    case Rows of
        [] ->
            [Next];

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

                false ->
                    [Row | upsert(Rest, Next)]
            end
    end.

-file("src/lightspeed/data_plane.gleam", 208).
-spec apply_all(list(row()), list(update())) -> list(row()).
apply_all(Rows, Updates) ->
    case Updates of
        [] ->
            Rows;

        [Update | Rest] ->
            case Update of
                {upsert, Row} ->
                    apply_all(upsert(Rows, Row), Rest);

                {delete, Id} ->
                    apply_all(remove(Rows, Id), Rest)
            end
    end.

-file("src/lightspeed/data_plane.gleam", 84).
?DOC(" Apply incremental updates and return window patches for one request.\n").
-spec apply_updates(plane(), window_request(), list(update())) -> {plane(),
    {ok, window()} | {error, query_error()},
    list(lightspeed@diff:patch())}.
apply_updates(Plane, Request, Updates) ->
    Previous_window = query_window(Plane, Request),
    Next_rows = apply_all(erlang:element(4, Plane), Updates),
    Next_plane = {plane,
        erlang:element(2, Plane),
        erlang:element(3, Plane),
        Next_rows},
    Next_window = query_window(Next_plane, Request),
    Patches = case {Previous_window, Next_window} of
        {{ok, Previous}, {ok, Current}} ->
            lightspeed@diff:keyed_patch_plan(
                erlang:element(2, Plane),
                to_keyed_nodes(erlang:element(5, Previous)),
                to_keyed_nodes(erlang:element(5, Current))
            );

        {_, _} ->
            []
    end,
    {Next_plane, Next_window, Patches}.

-file("src/lightspeed/data_plane.gleam", 107).
?DOC(" True when a patch set includes full-root churn operations.\n").
-spec contains_full_root_churn(list(lightspeed@diff:patch())) -> boolean().
contains_full_root_churn(Patches) ->
    case Patches of
        [] ->
            false;

        [Patch | Rest] ->
            case Patch of
                {upsert_keyed, _, _, _} ->
                    contains_full_root_churn(Rest);

                {remove_keyed, _, _} ->
                    contains_full_root_churn(Rest);

                _ ->
                    true
            end
    end.

-file("src/lightspeed/data_plane.gleam", 120).
?DOC(" True when a patch set is steady-state incremental (keyed-only or empty).\n").
-spec steady_state_incremental(list(lightspeed@diff:patch())) -> boolean().
steady_state_incremental(Patches) ->
    not contains_full_root_churn(Patches).

-file("src/lightspeed/data_plane.gleam", 125).
?DOC(" Stable dataset label.\n").
-spec dataset_label(dataset()) -> binary().
dataset_label(Dataset) ->
    case Dataset of
        list_dataset ->
            <<"list"/utf8>>;

        {grid_dataset, Columns} ->
            <<"grid:"/utf8, (erlang:integer_to_binary(Columns))/binary>>;

        {chart_dataset, Series_key} ->
            <<"chart:"/utf8, Series_key/binary>>
    end.

-file("src/lightspeed/data_plane.gleam", 134).
?DOC(" Stable query-error label.\n").
-spec query_error_label(query_error()) -> binary().
query_error_label(Error) ->
    case Error of
        {invalid_window, Offset, Limit} ->
            <<<<<<"invalid_window:"/utf8,
                        (erlang:integer_to_binary(Offset))/binary>>/binary,
                    ":"/utf8>>/binary,
                (erlang:integer_to_binary(Limit))/binary>>
    end.

-file("src/lightspeed/data_plane.gleam", 273).
-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/data_plane.gleam", 142).
?DOC(" Stable window signature for fixture assertions.\n").
-spec window_signature(window()) -> binary().
window_signature(Window) ->
    <<<<<<<<<<<<<<"offset="/utf8,
                                (erlang:integer_to_binary(
                                    erlang:element(2, Window)
                                ))/binary>>/binary,
                            "|limit="/utf8>>/binary,
                        (erlang:integer_to_binary(erlang:element(3, Window)))/binary>>/binary,
                    "|total="/utf8>>/binary,
                (erlang:integer_to_binary(erlang:element(4, Window)))/binary>>/binary,
            "|ids="/utf8>>/binary,
        (join_with(
            <<","/utf8>>,
            gleam@list:map(
                erlang:element(5, Window),
                fun(Row) -> erlang:element(2, Row) end
            )
        ))/binary>>.

-file("src/lightspeed/data_plane.gleam", 154).
?DOC(" Stable plane signature for deterministic fixtures.\n").
-spec signature(plane()) -> binary().
signature(Plane) ->
    <<<<<<<<<<"target="/utf8, (erlang:element(2, Plane))/binary>>/binary,
                    "|dataset="/utf8>>/binary,
                (dataset_label(erlang:element(3, Plane)))/binary>>/binary,
            "|rows="/utf8>>/binary,
        (erlang:integer_to_binary(erlang:length(erlang:element(4, Plane))))/binary>>.

-file("src/lightspeed/data_plane.gleam", 164).
?DOC(" Plane target selector.\n").
-spec target(plane()) -> binary().
target(Plane) ->
    erlang:element(2, Plane).

-file("src/lightspeed/data_plane.gleam", 169).
?DOC(" Plane dataset profile.\n").
-spec dataset(plane()) -> dataset().
dataset(Plane) ->
    erlang:element(3, Plane).

-file("src/lightspeed/data_plane.gleam", 174).
?DOC(" Plane rows in stored order.\n").
-spec rows(plane()) -> list(row()).
rows(Plane) ->
    erlang:element(4, Plane).