lightspeed

//// Large data-plane contracts with windowed queries and incremental updates.

import gleam/int
import gleam/list
import lightspeed/diff

/// Dataset profile used for large-data workloads.
pub type Dataset {
  ListDataset
  GridDataset(columns: Int)
  ChartDataset(series_key: String)
}

/// One keyed data-plane row.
pub type Row {
  Row(id: String, payload: String)
}

/// One window request over a large dataset.
pub type WindowRequest {
  WindowRequest(offset: Int, limit: Int)
}

/// Window query result.
pub type Window {
  Window(offset: Int, limit: Int, total: Int, rows: List(Row))
}

/// Incremental updates applied to the large-data plane.
pub type Update {
  Upsert(row: Row)
  Delete(id: String)
}

/// Window query validation failures.
pub type QueryError {
  InvalidWindow(offset: Int, limit: Int)
}

/// Large data-plane runtime.
pub opaque type Plane {
  Plane(target: String, dataset: Dataset, rows: List(Row))
}

/// Build one large-data plane.
pub fn new(target: String, dataset: Dataset, rows: List(Row)) -> Plane {
  Plane(target: target, dataset: dataset, rows: rows)
}

/// Build one row.
pub fn row(id: String, payload: String) -> Row {
  Row(id: id, payload: payload)
}

/// Build one window request.
pub fn request(offset: Int, limit: Int) -> WindowRequest {
  WindowRequest(offset: offset, limit: limit)
}

/// Validate one large-data plane contract.
pub fn valid(plane: Plane) -> Bool {
  plane.target != ""
  && valid_dataset(plane.dataset)
  && unique_ids(plane.rows, [])
}

/// Execute one windowed query.
pub fn query_window(
  plane: Plane,
  request: WindowRequest,
) -> Result(Window, QueryError) {
  case request.offset < 0 || request.limit <= 0 {
    True -> Error(InvalidWindow(offset: request.offset, limit: request.limit))
    False -> {
      let total = list.length(plane.rows)
      let start = int.clamp(request.offset, min: 0, max: total)
      let rows = take(drop(plane.rows, start), request.limit)
      Ok(Window(offset: start, limit: request.limit, total: total, rows: rows))
    }
  }
}

/// Apply incremental updates and return window patches for one request.
pub fn apply_updates(
  plane: Plane,
  request: WindowRequest,
  updates: List(Update),
) -> #(Plane, Result(Window, QueryError), List(diff.Patch)) {
  let previous_window = query_window(plane, request)
  let next_rows = apply_all(plane.rows, updates)
  let next_plane = Plane(..plane, rows: next_rows)
  let next_window = query_window(next_plane, request)
  let patches = case previous_window, next_window {
    Ok(previous), Ok(current) ->
      diff.keyed_patch_plan(
        plane.target,
        to_keyed_nodes(previous.rows),
        to_keyed_nodes(current.rows),
      )
    _, _ -> []
  }

  #(next_plane, next_window, patches)
}

/// True when a patch set includes full-root churn operations.
pub fn contains_full_root_churn(patches: List(diff.Patch)) -> Bool {
  case patches {
    [] -> False
    [patch, ..rest] ->
      case patch {
        diff.UpsertKeyed(_, _, _) -> contains_full_root_churn(rest)
        diff.RemoveKeyed(_, _) -> contains_full_root_churn(rest)
        _ -> True
      }
  }
}

/// True when a patch set is steady-state incremental (keyed-only or empty).
pub fn steady_state_incremental(patches: List(diff.Patch)) -> Bool {
  !contains_full_root_churn(patches)
}

/// Stable dataset label.
pub fn dataset_label(dataset: Dataset) -> String {
  case dataset {
    ListDataset -> "list"
    GridDataset(columns) -> "grid:" <> int.to_string(columns)
    ChartDataset(series_key) -> "chart:" <> series_key
  }
}

/// Stable query-error label.
pub fn query_error_label(error: QueryError) -> String {
  case error {
    InvalidWindow(offset, limit) ->
      "invalid_window:" <> int.to_string(offset) <> ":" <> int.to_string(limit)
  }
}

/// Stable window signature for fixture assertions.
pub fn window_signature(window: Window) -> String {
  "offset="
  <> int.to_string(window.offset)
  <> "|limit="
  <> int.to_string(window.limit)
  <> "|total="
  <> int.to_string(window.total)
  <> "|ids="
  <> join_with(",", list.map(window.rows, fn(row) { row.id }))
}

/// Stable plane signature for deterministic fixtures.
pub fn signature(plane: Plane) -> String {
  "target="
  <> plane.target
  <> "|dataset="
  <> dataset_label(plane.dataset)
  <> "|rows="
  <> int.to_string(list.length(plane.rows))
}

/// Plane target selector.
pub fn target(plane: Plane) -> String {
  plane.target
}

/// Plane dataset profile.
pub fn dataset(plane: Plane) -> Dataset {
  plane.dataset
}

/// Plane rows in stored order.
pub fn rows(plane: Plane) -> List(Row) {
  plane.rows
}

fn valid_dataset(dataset: Dataset) -> Bool {
  case dataset {
    ListDataset -> True
    GridDataset(columns) -> columns > 0
    ChartDataset(series_key) -> series_key != ""
  }
}

fn unique_ids(rows: List(Row), seen: List(String)) -> Bool {
  case rows {
    [] -> True
    [entry, ..rest] ->
      case contains_string(seen, entry.id) {
        True -> False
        False -> unique_ids(rest, [entry.id, ..seen])
      }
  }
}

fn contains_string(values: List(String), target: String) -> Bool {
  case values {
    [] -> False
    [value, ..rest] ->
      case value == target {
        True -> True
        False -> contains_string(rest, target)
      }
  }
}

fn apply_all(rows: List(Row), updates: List(Update)) -> List(Row) {
  case updates {
    [] -> rows
    [update, ..rest] ->
      case update {
        Upsert(row) -> apply_all(upsert(rows, row), rest)
        Delete(id) -> apply_all(remove(rows, id), rest)
      }
  }
}

fn upsert(rows: List(Row), next: Row) -> List(Row) {
  case rows {
    [] -> [next]
    [row, ..rest] ->
      case row.id == next.id {
        True -> [next, ..rest]
        False -> [row, ..upsert(rest, next)]
      }
  }
}

fn remove(rows: List(Row), target_id: String) -> List(Row) {
  case rows {
    [] -> []
    [row, ..rest] ->
      case row.id == target_id {
        True -> remove(rest, target_id)
        False -> [row, ..remove(rest, target_id)]
      }
  }
}

fn drop(rows: List(Row), count: Int) -> List(Row) {
  case count <= 0 {
    True -> rows
    False ->
      case rows {
        [] -> []
        [_row, ..rest] -> drop(rest, count - 1)
      }
  }
}

fn take(rows: List(Row), count: Int) -> List(Row) {
  case count <= 0 {
    True -> []
    False ->
      case rows {
        [] -> []
        [row, ..rest] -> [row, ..take(rest, count - 1)]
      }
  }
}

fn to_keyed_nodes(rows: List(Row)) -> List(diff.KeyedNode) {
  case rows {
    [] -> []
    [row, ..rest] -> [
      diff.keyed_node(row.id, row.payload),
      ..to_keyed_nodes(rest)
    ]
  }
}

fn join_with(separator: String, values: List(String)) -> String {
  case values {
    [] -> ""
    [value] -> value
    [value, ..rest] -> value <> separator <> join_with(separator, rest)
  }
}