///|
pub(all) struct TableDiffError {
  code : String
  message : String
} derive(Eq)

///|
pub(all) struct RowAddition {
  key : String
  row_number : Int
  values : Array[String]
} derive(Eq)

///|
pub(all) struct RowRemoval {
  key : String
  row_number : Int
  values : Array[String]
} derive(Eq)

///|
pub(all) struct CellChange {
  key : String
  old_row_number : Int
  new_row_number : Int
  column : String
  old_value : String
  new_value : String
} derive(Eq)

///|
pub(all) struct DuplicateKeyReport {
  key : String
  old_rows : Array[Int]
  new_rows : Array[Int]
} derive(Eq)

///|
pub(all) struct ColumnChangeStats {
  name : String
  changed_cells : Int
  old_only : Bool
  new_only : Bool
} derive(Eq)

///|
pub(all) struct DiffGateReport {
  passed : Bool
  risk : String
  reasons : Array[String]
} derive(Eq)

///|
pub(all) struct TableDiffReport {
  key_columns : Array[String]
  compared_columns : Array[String]
  old_only_columns : Array[String]
  new_only_columns : Array[String]
  added_rows : Array[RowAddition]
  removed_rows : Array[RowRemoval]
  changed_cells : Array[CellChange]
  unchanged_rows : Int
  duplicate_keys : Array[DuplicateKeyReport]
  missing_key_rows_old : Array[Int]
  missing_key_rows_new : Array[Int]
  change_score : Int
} derive(Eq)

///|
struct DiffKeyEntry {
  key : String
  row_index : Int
  row_number : Int
} derive(Eq)

///|
struct DiffColumnPair {
  name : String
  old_index : Int
  new_index : Int
} derive(Eq)

///|
fn diff_error(code : String, message : String) -> TableDiffError {
  { code, message }
}

///|
fn diff_clone_strings(values : Array[String]) -> Array[String] {
  let copy : Array[String] = []
  let mut i = 0
  while i < values.length() {
    copy.push(values[i])
    i = i + 1
  }
  copy
}

///|
fn diff_clone_ints(values : Array[Int]) -> Array[Int] {
  let copy : Array[Int] = []
  let mut i = 0
  while i < values.length() {
    copy.push(values[i])
    i = i + 1
  }
  copy
}

///|
fn diff_value_at(row : Array[String], index : Int) -> String {
  match row.get(index) {
    Some(value) => value
    None => ""
  }
}

///|
fn diff_string_in(values : Array[String], target : String) -> Bool {
  let mut i = 0
  while i < values.length() {
    if values[i] == target {
      return true
    }
    i = i + 1
  }
  false
}

///|
fn diff_push_unique_string(values : Array[String], value : String) -> Bool {
  if diff_string_in(values, value) {
    false
  } else {
    values.push(value)
    true
  }
}

///|
fn diff_key_columns_have_duplicate(key_columns : Array[String]) -> Bool {
  let mut i = 0
  while i < key_columns.length() {
    let mut j = i + 1
    while j < key_columns.length() {
      if key_columns[i] == key_columns[j] {
        return true
      }
      j = j + 1
    }
    i = i + 1
  }
  false
}

///|
fn diff_key_indices(
  table : Table,
  key_columns : Array[String],
  side : String,
) -> Result[Array[Int], TableDiffError] {
  if key_columns.length() == 0 {
    return Err(
      diff_error("empty_key", "table diff requires at least one key column"),
    )
  }
  if diff_key_columns_have_duplicate(key_columns) {
    return Err(diff_error("duplicate_key_column", "key columns must be unique"))
  }
  if table.header.length() == 0 {
    let msg =
      $|\{side} table has no header row
    return Err(diff_error("missing_header", msg))
  }
  let indices : Array[Int] = []
  let mut i = 0
  while i < key_columns.length() {
    match table.find_column(key_columns[i]) {
      Some(index) => indices.push(index)
      None => {
        let msg =
          $|\{side} table is missing key column '\{key_columns[i]}'
        return Err(diff_error("missing_key_column", msg))
      }
    }
    i = i + 1
  }
  Ok(indices)
}

///|
fn diff_escape_key_part(text : String) -> String {
  let out = StringBuilder::StringBuilder()
  let mut i = 0
  while i < text.length() {
    match text.get_char(i) {
      Some('\\') => out.write_string("\\\\")
      Some('|') => out.write_string("\\|")
      Some('\n') => out.write_string("\\n")
      Some('\r') => out.write_string("\\r")
      Some(c) => out.write_char(c)
      None => ()
    }
    i = i + 1
  }
  out.to_string()
}

///|
fn diff_build_key(row : Array[String], key_indices : Array[Int]) -> String {
  let out = StringBuilder::StringBuilder()
  let mut i = 0
  while i < key_indices.length() {
    if i > 0 {
      out.write_char('|')
    }
    out.write_string(diff_escape_key_part(diff_value_at(row, key_indices[i])))
    i = i + 1
  }
  out.to_string()
}

///|
fn diff_row_has_missing_key(
  row : Array[String],
  key_indices : Array[Int],
) -> Bool {
  let mut i = 0
  while i < key_indices.length() {
    if diff_value_at(row, key_indices[i]).trim().is_empty() {
      return true
    }
    i = i + 1
  }
  false
}

///|
fn diff_key_entries(
  table : Table,
  key_indices : Array[Int],
  missing_rows : Array[Int],
) -> Array[DiffKeyEntry] {
  let entries : Array[DiffKeyEntry] = []
  let mut r = 0
  while r < table.rows.length() {
    let row = table.rows[r]
    if diff_row_has_missing_key(row, key_indices) {
      missing_rows.push(r + 1)
    } else {
      entries.push({
        key: diff_build_key(row, key_indices),
        row_index: r,
        row_number: r + 1,
      })
    }
    r = r + 1
  }
  entries
}

///|
fn diff_count_key(entries : Array[DiffKeyEntry], key : String) -> Int {
  let mut count = 0
  let mut i = 0
  while i < entries.length() {
    if entries[i].key == key {
      count = count + 1
    }
    i = i + 1
  }
  count
}

///|
fn diff_has_key(entries : Array[DiffKeyEntry], key : String) -> Bool {
  let mut i = 0
  while i < entries.length() {
    if entries[i].key == key {
      return true
    }
    i = i + 1
  }
  false
}

///|
fn diff_find_entry(
  entries : Array[DiffKeyEntry],
  key : String,
) -> DiffKeyEntry? {
  let mut i = 0
  while i < entries.length() {
    if entries[i].key == key {
      return Some(entries[i])
    }
    i = i + 1
  }
  None
}

///|
fn diff_rows_for_key(entries : Array[DiffKeyEntry], key : String) -> Array[Int] {
  let rows : Array[Int] = []
  let mut i = 0
  while i < entries.length() {
    if entries[i].key == key {
      rows.push(entries[i].row_number)
    }
    i = i + 1
  }
  rows
}

///|
fn diff_collect_duplicate_keys(
  entries : Array[DiffKeyEntry],
  keys : Array[String],
) -> Unit {
  let mut i = 0
  while i < entries.length() {
    let key = entries[i].key
    if diff_count_key(entries, key) > 1 {
      ignore(diff_push_unique_string(keys, key))
    }
    i = i + 1
  }
}

///|
fn diff_duplicate_reports(
  old_entries : Array[DiffKeyEntry],
  new_entries : Array[DiffKeyEntry],
) -> Array[DuplicateKeyReport] {
  let keys : Array[String] = []
  diff_collect_duplicate_keys(old_entries, keys)
  diff_collect_duplicate_keys(new_entries, keys)
  let reports : Array[DuplicateKeyReport] = []
  let mut i = 0
  while i < keys.length() {
    reports.push({
      key: keys[i],
      old_rows: diff_rows_for_key(old_entries, keys[i]),
      new_rows: diff_rows_for_key(new_entries, keys[i]),
    })
    i = i + 1
  }
  reports
}

///|
fn diff_key_is_duplicate(
  duplicates : Array[DuplicateKeyReport],
  key : String,
) -> Bool {
  let mut i = 0
  while i < duplicates.length() {
    if duplicates[i].key == key {
      return true
    }
    i = i + 1
  }
  false
}

///|
fn diff_is_key_column(name : String, key_columns : Array[String]) -> Bool {
  diff_string_in(key_columns, name)
}

///|
fn diff_column_pairs(
  old_table : Table,
  new_table : Table,
  key_columns : Array[String],
) -> Array[DiffColumnPair] {
  let pairs : Array[DiffColumnPair] = []
  let mut i = 0
  while i < old_table.header.length() {
    let name = old_table.header[i]
    if !diff_is_key_column(name, key_columns) {
      match new_table.find_column(name) {
        Some(new_index) => pairs.push({ name, old_index: i, new_index })
        None => ()
      }
    }
    i = i + 1
  }
  pairs
}

///|
fn diff_pair_names(pairs : Array[DiffColumnPair]) -> Array[String] {
  let names : Array[String] = []
  let mut i = 0
  while i < pairs.length() {
    names.push(pairs[i].name)
    i = i + 1
  }
  names
}

///|
fn diff_columns_missing_from(
  source : Array[String],
  target : Array[String],
  key_columns : Array[String],
) -> Array[String] {
  let missing : Array[String] = []
  let mut i = 0
  while i < source.length() {
    let name = source[i]
    if !diff_is_key_column(name, key_columns) && !diff_string_in(target, name) {
      missing.push(name)
    }
    i = i + 1
  }
  missing
}

///|
fn diff_row_addition(table : Table, entry : DiffKeyEntry) -> RowAddition {
  {
    key: entry.key,
    row_number: entry.row_number,
    values: clone_row(table.rows[entry.row_index]),
  }
}

///|
fn diff_row_removal(table : Table, entry : DiffKeyEntry) -> RowRemoval {
  {
    key: entry.key,
    row_number: entry.row_number,
    values: clone_row(table.rows[entry.row_index]),
  }
}

///|
fn diff_cell_change(
  key : String,
  old_row_number : Int,
  new_row_number : Int,
  column : String,
  old_value : String,
  new_value : String,
) -> CellChange {
  { key, old_row_number, new_row_number, column, old_value, new_value }
}

///|
fn diff_compare_row_cells(
  old_row : Array[String],
  new_row : Array[String],
  old_entry : DiffKeyEntry,
  new_entry : DiffKeyEntry,
  pairs : Array[DiffColumnPair],
  changes : Array[CellChange],
) -> Bool {
  let start = changes.length()
  let mut i = 0
  while i < pairs.length() {
    let pair = pairs[i]
    let old_value = diff_value_at(old_row, pair.old_index)
    let new_value = diff_value_at(new_row, pair.new_index)
    if old_value != new_value {
      changes.push(
        diff_cell_change(
          old_entry.key,
          old_entry.row_number,
          new_entry.row_number,
          pair.name,
          old_value,
          new_value,
        ),
      )
    }
    i = i + 1
  }
  changes.length() == start
}

///|
fn diff_compare_existing_rows(
  old_table : Table,
  new_table : Table,
  old_entries : Array[DiffKeyEntry],
  new_entries : Array[DiffKeyEntry],
  pairs : Array[DiffColumnPair],
  duplicates : Array[DuplicateKeyReport],
  removed : Array[RowRemoval],
  changes : Array[CellChange],
) -> Int {
  let mut unchanged_rows = 0
  let mut i = 0
  while i < old_entries.length() {
    let old_entry = old_entries[i]
    if !diff_key_is_duplicate(duplicates, old_entry.key) {
      match diff_find_entry(new_entries, old_entry.key) {
        Some(new_entry) => {
          let old_row = old_table.rows[old_entry.row_index]
          let new_row = new_table.rows[new_entry.row_index]
          if diff_compare_row_cells(
              old_row, new_row, old_entry, new_entry, pairs, changes,
            ) {
            unchanged_rows = unchanged_rows + 1
          }
        }
        None => removed.push(diff_row_removal(old_table, old_entry))
      }
    }
    i = i + 1
  }
  unchanged_rows
}

///|
fn diff_collect_added_rows(
  new_table : Table,
  old_entries : Array[DiffKeyEntry],
  new_entries : Array[DiffKeyEntry],
  duplicates : Array[DuplicateKeyReport],
) -> Array[RowAddition] {
  let added : Array[RowAddition] = []
  let mut i = 0
  while i < new_entries.length() {
    let entry = new_entries[i]
    if !diff_key_is_duplicate(duplicates, entry.key) &&
      !diff_has_key(old_entries, entry.key) {
      added.push(diff_row_addition(new_table, entry))
    }
    i = i + 1
  }
  added
}

///|
fn diff_change_score(
  added : Int,
  removed : Int,
  changed_cells : Int,
  duplicate_keys : Int,
  missing_key_rows : Int,
  old_only_columns : Int,
  new_only_columns : Int,
) -> Int {
  let penalty = added * 12 +
    removed * 14 +
    changed_cells * 4 +
    duplicate_keys * 18 +
    missing_key_rows * 18 +
    old_only_columns * 10 +
    new_only_columns * 6
  clamp_percent(100 - penalty)
}

///|
fn diff_build_report(
  old_table : Table,
  new_table : Table,
  key_columns : Array[String],
  old_entries : Array[DiffKeyEntry],
  new_entries : Array[DiffKeyEntry],
  missing_old : Array[Int],
  missing_new : Array[Int],
) -> TableDiffReport {
  let pairs = diff_column_pairs(old_table, new_table, key_columns)
  let duplicate_keys = diff_duplicate_reports(old_entries, new_entries)
  let removed_rows : Array[RowRemoval] = []
  let changed_cells : Array[CellChange] = []
  let unchanged_rows = diff_compare_existing_rows(
    old_table, new_table, old_entries, new_entries, pairs, duplicate_keys, removed_rows,
    changed_cells,
  )
  let added_rows = diff_collect_added_rows(
    new_table, old_entries, new_entries, duplicate_keys,
  )
  let old_only = diff_columns_missing_from(
    old_table.header,
    new_table.header,
    key_columns,
  )
  let new_only = diff_columns_missing_from(
    new_table.header,
    old_table.header,
    key_columns,
  )
  {
    key_columns: diff_clone_strings(key_columns),
    compared_columns: diff_pair_names(pairs),
    old_only_columns: old_only,
    new_only_columns: new_only,
    added_rows,
    removed_rows,
    changed_cells,
    unchanged_rows,
    duplicate_keys,
    missing_key_rows_old: diff_clone_ints(missing_old),
    missing_key_rows_new: diff_clone_ints(missing_new),
    change_score: diff_change_score(
      added_rows.length(),
      removed_rows.length(),
      changed_cells.length(),
      duplicate_keys.length(),
      missing_old.length() + missing_new.length(),
      old_only.length(),
      new_only.length(),
    ),
  }
}

///|
/// Compare two header-based tables by one or more key columns.
pub fn table_diff_by_key(
  old_table : Table,
  new_table : Table,
  key_columns : Array[String],
) -> Result[TableDiffReport, TableDiffError] {
  match diff_key_indices(old_table, key_columns, "old") {
    Ok(old_key_indices) =>
      match diff_key_indices(new_table, key_columns, "new") {
        Ok(new_key_indices) => {
          let missing_old : Array[Int] = []
          let missing_new : Array[Int] = []
          let old_entries = diff_key_entries(
            old_table, old_key_indices, missing_old,
          )
          let new_entries = diff_key_entries(
            new_table, new_key_indices, missing_new,
          )
          Ok(
            diff_build_report(
              old_table, new_table, key_columns, old_entries, new_entries, missing_old,
              missing_new,
            ),
          )
        }
        Err(err) => Err(err)
      }
    Err(err) => Err(err)
  }
}

///|
/// Alias for callers that prefer action-first naming.
pub fn diff_tables_by_key(
  old_table : Table,
  new_table : Table,
  key_columns : Array[String],
) -> Result[TableDiffReport, TableDiffError] {
  table_diff_by_key(old_table, new_table, key_columns)
}

///|
pub fn TableDiffReport::change_count(self : TableDiffReport) -> Int {
  self.added_rows.length() +
  self.removed_rows.length() +
  self.changed_cells.length() +
  self.duplicate_keys.length() +
  self.missing_key_rows_old.length() +
  self.missing_key_rows_new.length() +
  self.old_only_columns.length() +
  self.new_only_columns.length()
}

///|
pub fn TableDiffReport::has_changes(self : TableDiffReport) -> Bool {
  self.change_count() > 0
}

///|
pub fn TableDiffReport::risk_level(self : TableDiffReport) -> String {
  if self.duplicate_keys.length() > 0 ||
    self.missing_key_rows_old.length() > 0 ||
    self.missing_key_rows_new.length() > 0 {
    "critical"
  } else if self.removed_rows.length() > 0 ||
    self.old_only_columns.length() > 0 ||
    self.change_score < 70 {
    "high"
  } else if self.added_rows.length() > 0 ||
    self.changed_cells.length() > 0 ||
    self.new_only_columns.length() > 0 {
    "medium"
  } else {
    "low"
  }
}

///|
pub fn TableDiffReport::summary(self : TableDiffReport) -> String {
  let text =
    $|added=\{self.added_rows.length()}, removed=\{self.removed_rows.length()}, changed_cells=\{self.changed_cells.length()}, unchanged_rows=\{self.unchanged_rows}, duplicate_keys=\{self.duplicate_keys.length()}, risk=\{self.risk_level()}, score=\{self.change_score}
  text
}

///|
pub fn TableDiffReport::changed_keys(self : TableDiffReport) -> Array[String] {
  let keys : Array[String] = []
  let mut i = 0
  while i < self.changed_cells.length() {
    ignore(diff_push_unique_string(keys, self.changed_cells[i].key))
    i = i + 1
  }
  keys
}

///|
pub fn TableDiffReport::affected_row_count(self : TableDiffReport) -> Int {
  self.added_rows.length() +
  self.removed_rows.length() +
  self.changed_keys().length() +
  self.duplicate_keys.length() +
  self.missing_key_rows_old.length() +
  self.missing_key_rows_new.length()
}

///|
fn diff_column_stat_index(
  stats : Array[ColumnChangeStats],
  name : String,
) -> Int? {
  let mut i = 0
  while i < stats.length() {
    if stats[i].name == name {
      return Some(i)
    }
    i = i + 1
  }
  None
}

///|
fn diff_push_column_stat(
  stats : Array[ColumnChangeStats],
  name : String,
  changed_cells : Int,
  old_only : Bool,
  new_only : Bool,
) -> Unit {
  match diff_column_stat_index(stats, name) {
    Some(index) => {
      let current = stats[index]
      stats[index] = {
        name: current.name,
        changed_cells: current.changed_cells + changed_cells,
        old_only: current.old_only || old_only,
        new_only: current.new_only || new_only,
      }
    }
    None => stats.push({ name, changed_cells, old_only, new_only })
  }
}

///|
fn diff_column_stats_from_report(
  report : TableDiffReport,
) -> Array[ColumnChangeStats] {
  let stats : Array[ColumnChangeStats] = []
  let mut compared = 0
  while compared < report.compared_columns.length() {
    diff_push_column_stat(
      stats,
      report.compared_columns[compared],
      0,
      false,
      false,
    )
    compared = compared + 1
  }
  let mut old_only = 0
  while old_only < report.old_only_columns.length() {
    diff_push_column_stat(
      stats,
      report.old_only_columns[old_only],
      0,
      true,
      false,
    )
    old_only = old_only + 1
  }
  let mut new_only = 0
  while new_only < report.new_only_columns.length() {
    diff_push_column_stat(
      stats,
      report.new_only_columns[new_only],
      0,
      false,
      true,
    )
    new_only = new_only + 1
  }
  let mut change = 0
  while change < report.changed_cells.length() {
    diff_push_column_stat(
      stats,
      report.changed_cells[change].column,
      1,
      false,
      false,
    )
    change = change + 1
  }
  stats
}

///|
pub fn ColumnChangeStats::impact_score(self : ColumnChangeStats) -> Int {
  let drift = if self.old_only && self.new_only {
    10
  } else if self.old_only {
    8
  } else if self.new_only {
    4
  } else {
    0
  }
  self.changed_cells * 10 + drift
}

///|
pub fn ColumnChangeStats::summary(self : ColumnChangeStats) -> String {
  let drift = if self.old_only && self.new_only {
    "old_only,new_only"
  } else if self.old_only {
    "old_only"
  } else if self.new_only {
    "new_only"
  } else {
    "shared"
  }
  let text =
    $|\{self.name}: changed_cells=\{self.changed_cells}, drift=\{drift}, impact=\{self.impact_score()}
  text
}

///|
pub fn CellChange::summary(self : CellChange) -> String {
  let text =
    $|\{self.key}.\{self.column}: '\{self.old_value}' -> '\{self.new_value}'
  text
}

///|
pub fn RowAddition::summary(self : RowAddition) -> String {
  let text =
    $|added key '\{self.key}' at new row \{self.row_number}
  text
}

///|
pub fn RowRemoval::summary(self : RowRemoval) -> String {
  let text =
    $|removed key '\{self.key}' from old row \{self.row_number}
  text
}

///|
pub fn DuplicateKeyReport::summary(self : DuplicateKeyReport) -> String {
  let text =
    $|duplicate key '\{self.key}', old_rows=\{diff_join_ints(self.old_rows)}, new_rows=\{diff_join_ints(self.new_rows)}
  text
}

///|
pub fn TableDiffReport::changes_for_key(
  self : TableDiffReport,
  key : String,
) -> Array[CellChange] {
  let changes : Array[CellChange] = []
  let mut i = 0
  while i < self.changed_cells.length() {
    if self.changed_cells[i].key == key {
      changes.push(self.changed_cells[i])
    }
    i = i + 1
  }
  changes
}

///|
pub fn TableDiffReport::changes_for_column(
  self : TableDiffReport,
  column : String,
) -> Array[CellChange] {
  let changes : Array[CellChange] = []
  let mut i = 0
  while i < self.changed_cells.length() {
    if self.changed_cells[i].column == column {
      changes.push(self.changed_cells[i])
    }
    i = i + 1
  }
  changes
}

///|
fn diff_has_added_key(rows : Array[RowAddition], key : String) -> Bool {
  let mut i = 0
  while i < rows.length() {
    if rows[i].key == key {
      return true
    }
    i = i + 1
  }
  false
}

///|
fn diff_has_removed_key(rows : Array[RowRemoval], key : String) -> Bool {
  let mut i = 0
  while i < rows.length() {
    if rows[i].key == key {
      return true
    }
    i = i + 1
  }
  false
}

///|
pub fn TableDiffReport::has_key_change(
  self : TableDiffReport,
  key : String,
) -> Bool {
  diff_has_added_key(self.added_rows, key) ||
  diff_has_removed_key(self.removed_rows, key) ||
  self.changes_for_key(key).length() > 0 ||
  diff_key_is_duplicate(self.duplicate_keys, key)
}

///|
pub fn TableDiffReport::keys_with_changes(
  self : TableDiffReport,
) -> Array[String] {
  let keys : Array[String] = []
  let mut added = 0
  while added < self.added_rows.length() {
    ignore(diff_push_unique_string(keys, self.added_rows[added].key))
    added = added + 1
  }
  let mut removed = 0
  while removed < self.removed_rows.length() {
    ignore(diff_push_unique_string(keys, self.removed_rows[removed].key))
    removed = removed + 1
  }
  let mut changed = 0
  while changed < self.changed_cells.length() {
    ignore(diff_push_unique_string(keys, self.changed_cells[changed].key))
    changed = changed + 1
  }
  let mut duplicate = 0
  while duplicate < self.duplicate_keys.length() {
    ignore(diff_push_unique_string(keys, self.duplicate_keys[duplicate].key))
    duplicate = duplicate + 1
  }
  keys
}

///|
pub fn TableDiffReport::column_change_stats(
  self : TableDiffReport,
) -> Array[ColumnChangeStats] {
  diff_column_stats_from_report(self)
}

///|
fn diff_stat_already_selected(
  selected : Array[ColumnChangeStats],
  name : String,
) -> Bool {
  let mut i = 0
  while i < selected.length() {
    if selected[i].name == name {
      return true
    }
    i = i + 1
  }
  false
}

///|
fn diff_best_stat_index(
  stats : Array[ColumnChangeStats],
  selected : Array[ColumnChangeStats],
) -> Int {
  let mut best_index = -1
  let mut best_score = -1
  let mut i = 0
  while i < stats.length() {
    let stat = stats[i]
    if !diff_stat_already_selected(selected, stat.name) {
      let score = stat.impact_score()
      if score > best_score {
        best_index = i
        best_score = score
      }
    }
    i = i + 1
  }
  best_index
}

///|
pub fn TableDiffReport::most_changed_columns(
  self : TableDiffReport,
  limit : Int,
) -> Array[ColumnChangeStats] {
  let stats = self.column_change_stats()
  let selected : Array[ColumnChangeStats] = []
  let target = diff_limit(limit, stats.length())
  while selected.length() < target {
    let index = diff_best_stat_index(stats, selected)
    if index < 0 {
      break
    }
    selected.push(stats[index])
  }
  selected
}

///|
fn diff_join_strings(values : Array[String]) -> String {
  let out = StringBuilder::StringBuilder()
  let mut i = 0
  while i < values.length() {
    if i > 0 {
      out.write_string(",")
    }
    out.write_string(values[i])
    i = i + 1
  }
  out.to_string()
}

///|
fn diff_reasons_summary(reasons : Array[String]) -> String {
  if reasons.length() == 0 {
    "passed"
  } else {
    diff_join_strings(reasons)
  }
}

///|
pub fn DiffGateReport::summary(self : DiffGateReport) -> String {
  let status = if self.passed { "passed" } else { "failed" }
  let text =
    $|\{status}, risk=\{self.risk}, reasons=\{diff_reasons_summary(self.reasons)}
  text
}

///|
pub fn TableDiffReport::gate(
  self : TableDiffReport,
  max_changed_cells : Int,
  allow_removed_rows : Bool,
  allow_column_drift : Bool,
) -> DiffGateReport {
  let reasons : Array[String] = []
  if self.duplicate_keys.length() > 0 {
    reasons.push("duplicate keys require manual review")
  }
  if self.missing_key_rows_old.length() > 0 ||
    self.missing_key_rows_new.length() > 0 {
    reasons.push("missing key rows require manual review")
  }
  if !allow_removed_rows && self.removed_rows.length() > 0 {
    reasons.push("removed rows are not allowed by this gate")
  }
  if max_changed_cells >= 0 && self.changed_cells.length() > max_changed_cells {
    reasons.push("changed cell count exceeds gate threshold")
  }
  if !allow_column_drift &&
    (self.old_only_columns.length() > 0 || self.new_only_columns.length() > 0) {
    reasons.push("column drift is not allowed by this gate")
  }
  { passed: reasons.length() == 0, risk: self.risk_level(), reasons }
}

///|
fn diff_join_ints(values : Array[Int]) -> String {
  let out = StringBuilder::StringBuilder()
  let mut i = 0
  while i < values.length() {
    if i > 0 {
      out.write_string(",")
    }
    out.write_string(values[i].to_string())
    i = i + 1
  }
  out.to_string()
}

///|
fn diff_write_markdown_row(out : StringBuilder, cells : Array[String]) -> Unit {
  out.write_char('|')
  let mut i = 0
  while i < cells.length() {
    out.write_char(' ')
    out.write_string(markdown_escape_cell(cells[i]))
    out.write_string(" |")
    i = i + 1
  }
  out.write_char('\n')
}

///|
fn diff_write_metric(
  out : StringBuilder,
  name : String,
  value : String,
) -> Unit {
  diff_write_markdown_row(out, [name, value])
}

///|
fn diff_limit(max_items : Int, length : Int) -> Int {
  if max_items < 0 {
    0
  } else {
    min_int(max_items, length)
  }
}

///|
fn diff_write_change_table(
  out : StringBuilder,
  changes : Array[CellChange],
  max_items : Int,
) -> Unit {
  if changes.length() == 0 {
    return
  }
  out.write_string("\n### Cell changes\n\n")
  diff_write_markdown_row(out, [
    "key", "column", "old_row", "new_row", "old_value", "new_value",
  ])
  diff_write_markdown_row(out, ["---", "---", "---", "---", "---", "---"])
  let limit = diff_limit(max_items, changes.length())
  let mut i = 0
  while i < limit {
    let change = changes[i]
    diff_write_markdown_row(out, [
      change.key,
      change.column,
      change.old_row_number.to_string(),
      change.new_row_number.to_string(),
      change.old_value,
      change.new_value,
    ])
    i = i + 1
  }
}

///|
fn diff_write_added_rows(
  out : StringBuilder,
  rows : Array[RowAddition],
  max_items : Int,
) -> Unit {
  if rows.length() == 0 {
    return
  }
  out.write_string("\n### Added rows\n\n")
  diff_write_markdown_row(out, ["key", "new_row"])
  diff_write_markdown_row(out, ["---", "---"])
  let limit = diff_limit(max_items, rows.length())
  let mut i = 0
  while i < limit {
    diff_write_markdown_row(out, [rows[i].key, rows[i].row_number.to_string()])
    i = i + 1
  }
}

///|
fn diff_write_removed_rows(
  out : StringBuilder,
  rows : Array[RowRemoval],
  max_items : Int,
) -> Unit {
  if rows.length() == 0 {
    return
  }
  out.write_string("\n### Removed rows\n\n")
  diff_write_markdown_row(out, ["key", "old_row"])
  diff_write_markdown_row(out, ["---", "---"])
  let limit = diff_limit(max_items, rows.length())
  let mut i = 0
  while i < limit {
    diff_write_markdown_row(out, [rows[i].key, rows[i].row_number.to_string()])
    i = i + 1
  }
}

///|
fn diff_write_duplicate_keys(
  out : StringBuilder,
  duplicates : Array[DuplicateKeyReport],
  max_items : Int,
) -> Unit {
  if duplicates.length() == 0 {
    return
  }
  out.write_string("\n### Duplicate keys\n\n")
  diff_write_markdown_row(out, ["key", "old_rows", "new_rows"])
  diff_write_markdown_row(out, ["---", "---", "---"])
  let limit = diff_limit(max_items, duplicates.length())
  let mut i = 0
  while i < limit {
    diff_write_markdown_row(out, [
      duplicates[i].key,
      diff_join_ints(duplicates[i].old_rows),
      diff_join_ints(duplicates[i].new_rows),
    ])
    i = i + 1
  }
}

///|
/// Render a compact Markdown report for review records or pull requests.
pub fn TableDiffReport::to_markdown(
  self : TableDiffReport,
  max_items : Int,
) -> String {
  let out = StringBuilder::StringBuilder()
  out.write_string("## Table diff\n\n")
  diff_write_markdown_row(out, ["metric", "value"])
  diff_write_markdown_row(out, ["---", "---"])
  diff_write_metric(out, "key_columns", diff_join_strings(self.key_columns))
  diff_write_metric(
    out,
    "compared_columns",
    diff_join_strings(self.compared_columns),
  )
  diff_write_metric(out, "summary", self.summary())
  diff_write_metric(out, "affected_rows", self.affected_row_count().to_string())
  if self.old_only_columns.length() > 0 {
    diff_write_metric(
      out,
      "old_only_columns",
      diff_join_strings(self.old_only_columns),
    )
  }
  if self.new_only_columns.length() > 0 {
    diff_write_metric(
      out,
      "new_only_columns",
      diff_join_strings(self.new_only_columns),
    )
  }
  if self.missing_key_rows_old.length() > 0 {
    diff_write_metric(
      out,
      "missing_key_rows_old",
      diff_join_ints(self.missing_key_rows_old),
    )
  }
  if self.missing_key_rows_new.length() > 0 {
    diff_write_metric(
      out,
      "missing_key_rows_new",
      diff_join_ints(self.missing_key_rows_new),
    )
  }
  diff_write_change_table(out, self.changed_cells, max_items)
  diff_write_added_rows(out, self.added_rows, max_items)
  diff_write_removed_rows(out, self.removed_rows, max_items)
  diff_write_duplicate_keys(out, self.duplicate_keys, max_items)
  out.to_string()
}

///|
/// Produce deterministic review steps for a data import or migration run.
pub fn TableDiffReport::migration_steps(
  self : TableDiffReport,
) -> Array[String] {
  let steps : Array[String] = []
  if self.duplicate_keys.length() > 0 {
    steps.push("resolve duplicate keys before applying row-level changes")
  }
  if self.missing_key_rows_old.length() > 0 ||
    self.missing_key_rows_new.length() > 0 {
    steps.push("fill missing key cells or remove rows without stable identity")
  }
  if self.old_only_columns.length() > 0 {
    steps.push("review removed columns and update downstream schema users")
  }
  if self.new_only_columns.length() > 0 {
    steps.push("review added columns and document their expected meaning")
  }
  if self.added_rows.length() > 0 {
    steps.push("append new rows after schema validation")
  }
  if self.removed_rows.length() > 0 {
    steps.push("archive removed rows before deleting them from consumers")
  }
  if self.changed_cells.length() > 0 {
    steps.push("review changed cells and re-run validation on the new table")
  }
  if steps.length() == 0 {
    steps.push("no migration action required")
  }
  steps
}