///|
pub(all) struct ColumnStats {
  name : String
  index : Int
  row_count : Int
  empty_count : Int
  non_empty_count : Int
  unique_count : Int
  duplicate_count : Int
  min_length : Int
  max_length : Int
  average_length : Int
} derive(Eq)

///|
pub(all) struct TableStats {
  row_count : Int
  column_count : Int
  cell_count : Int
  empty_cells : Int
  non_empty_cells : Int
  density_score : Int
  duplicate_row_count : Int
  columns : Array[ColumnStats]
} derive(Eq)

///|
pub(all) struct RequiredColumnsReport {
  required_count : Int
  present_count : Int
  missing_count : Int
  missing_columns : Array[String]
} derive(Eq)

///|
pub(all) struct SchemaChange {
  code : String
  column : String
  message : String
  breaking : Bool
} derive(Eq)

///|
pub(all) struct SchemaCompatibilityReport {
  compatible : Bool
  breaking_changes : Int
  non_breaking_changes : Int
  changes : Array[SchemaChange]
} derive(Eq)

///|
fn audit_generated_column_name(index : Int) -> String {
  let text =
    $|column_\{index + 1}
  text
}

///|
fn audit_column_name(table : Table, index : Int) -> String {
  if index < table.header.length() {
    table.header[index]
  } else {
    audit_generated_column_name(index)
  }
}

///|
fn audit_column_count(table : Table) -> Int {
  table.column_count()
}

///|
fn audit_cell_count(row_count : Int, column_count : Int) -> Int {
  row_count * column_count
}

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

///|
fn audit_contains_string(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 audit_push_unique(values : Array[String], value : String) -> Bool {
  if audit_contains_string(values, value) {
    false
  } else {
    values.push(value)
    true
  }
}

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

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

///|
fn audit_min_length(current : Int, candidate : Int, seen : Bool) -> Int {
  if !seen {
    candidate
  } else if candidate < current {
    candidate
  } else {
    current
  }
}

///|
fn audit_average(total : Int, count : Int) -> Int {
  if count <= 0 {
    0
  } else {
    total / count
  }
}

///|
fn audit_density(cell_count : Int, empty_cells : Int) -> Int {
  if cell_count <= 0 {
    100
  } else {
    clamp_percent(100 - empty_cells * 100 / cell_count)
  }
}

///|
fn audit_duplicate_row_count(rows : Array[Array[String]]) -> Int {
  let seen : Array[Array[String]] = []
  let mut duplicates = 0
  let mut i = 0
  while i < rows.length() {
    if audit_seen_row(seen, rows[i]) {
      duplicates = duplicates + 1
    } else {
      seen.push(rows[i])
    }
    i = i + 1
  }
  duplicates
}

///|
fn audit_column_stats(table : Table, index : Int) -> ColumnStats {
  let unique : Array[String] = []
  let mut empty_count = 0
  let mut non_empty_count = 0
  let mut duplicate_count = 0
  let mut min_length = 0
  let mut max_length = 0
  let mut total_length = 0
  let mut seen_non_empty = false
  let mut row = 0
  while row < table.rows.length() {
    let value = audit_value_at(table.rows[row], index)
    let length = value.length()
    total_length = total_length + length
    if value.trim().is_empty() {
      empty_count = empty_count + 1
    } else {
      non_empty_count = non_empty_count + 1
      if !audit_push_unique(unique, value) {
        duplicate_count = duplicate_count + 1
      }
      min_length = audit_min_length(min_length, length, seen_non_empty)
      if length > max_length {
        max_length = length
      }
      seen_non_empty = true
    }
    row = row + 1
  }
  {
    name: audit_column_name(table, index),
    index,
    row_count: table.rows.length(),
    empty_count,
    non_empty_count,
    unique_count: unique.length(),
    duplicate_count,
    min_length,
    max_length,
    average_length: audit_average(total_length, table.rows.length()),
  }
}

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

///|
fn audit_rule_exists(columns : Array[ColumnRule], name : String) -> Bool {
  match audit_rule_index(columns, name) {
    Some(_) => true
    None => false
  }
}

///|
fn audit_kind_label(kind : FieldKind) -> String {
  kind_name(kind)
}

///|
fn audit_kind_accepts(source : FieldKind, target : FieldKind) -> Bool {
  if source == target {
    true
  } else {
    match target {
      Text => true
      NonEmpty => source != Text
      Decimal => source == Integer
      Integer => false
      Boolean => false
    }
  }
}

///|
fn audit_change(
  code : String,
  column : String,
  message : String,
  breaking : Bool,
) -> SchemaChange {
  { code, column, message, breaking }
}

///|
fn audit_push_removed_column(
  changes : Array[SchemaChange],
  rule : ColumnRule,
) -> Unit {
  let msg =
    $|column '\{rule.name}' was removed from schema
  changes.push(audit_change("removed_column", rule.name, msg, true))
}

///|
fn audit_push_added_column(
  changes : Array[SchemaChange],
  rule : ColumnRule,
) -> Unit {
  if rule.required {
    let msg =
      $|required column '\{rule.name}' was added to schema
    changes.push(audit_change("added_required_column", rule.name, msg, true))
  } else {
    let msg =
      $|optional column '\{rule.name}' was added to schema
    changes.push(audit_change("added_optional_column", rule.name, msg, false))
  }
}

///|
fn audit_push_type_change(
  changes : Array[SchemaChange],
  old_rule : ColumnRule,
  new_rule : ColumnRule,
) -> Unit {
  if old_rule.kind != new_rule.kind {
    let breaking = !audit_kind_accepts(old_rule.kind, new_rule.kind)
    let old_kind = audit_kind_label(old_rule.kind)
    let new_kind = audit_kind_label(new_rule.kind)
    let msg =
      $|column '\{old_rule.name}' changed type from \{old_kind} to \{new_kind}
    changes.push(audit_change("type_changed", old_rule.name, msg, breaking))
  }
}

///|
fn audit_push_required_change(
  changes : Array[SchemaChange],
  old_rule : ColumnRule,
  new_rule : ColumnRule,
) -> Unit {
  if old_rule.required != new_rule.required {
    if new_rule.required {
      let msg =
        $|column '\{old_rule.name}' changed from optional to required
      changes.push(audit_change("required_tightened", old_rule.name, msg, true))
    } else {
      let msg =
        $|column '\{old_rule.name}' changed from required to optional
      changes.push(audit_change("required_relaxed", old_rule.name, msg, false))
    }
  }
}

///|
fn audit_push_extra_policy_change(
  changes : Array[SchemaChange],
  old_schema : Schema,
  new_schema : Schema,
) -> Unit {
  if old_schema.allow_extra_columns != new_schema.allow_extra_columns {
    if new_schema.allow_extra_columns {
      changes.push(
        audit_change(
          "extra_policy_relaxed", "", "schema now allows extra columns", false,
        ),
      )
    } else {
      changes.push(
        audit_change(
          "extra_policy_tightened", "", "schema no longer allows extra columns",
          true,
        ),
      )
    }
  }
}

///|
fn audit_count_breaking(changes : Array[SchemaChange]) -> Int {
  let mut count = 0
  let mut i = 0
  while i < changes.length() {
    if changes[i].breaking {
      count = count + 1
    }
    i = i + 1
  }
  count
}

///|
fn audit_count_non_breaking(changes : Array[SchemaChange]) -> Int {
  let mut count = 0
  let mut i = 0
  while i < changes.length() {
    if !changes[i].breaking {
      count = count + 1
    }
    i = i + 1
  }
  count
}

///|
fn audit_missing_required_columns(
  table : Table,
  schema : Schema,
) -> Array[String] {
  let missing : Array[String] = []
  let mut i = 0
  while i < schema.columns.length() {
    let rule = schema.columns[i]
    if rule.required && table.find_column(rule.name) is None {
      missing.push(rule.name)
    }
    i = i + 1
  }
  missing
}

///|
fn audit_present_required_count(table : Table, schema : Schema) -> Int {
  let mut count = 0
  let mut i = 0
  while i < schema.columns.length() {
    let rule = schema.columns[i]
    if rule.required && !(table.find_column(rule.name) is None) {
      count = count + 1
    }
    i = i + 1
  }
  count
}

///|
fn audit_required_count(schema : Schema) -> Int {
  let mut count = 0
  let mut i = 0
  while i < schema.columns.length() {
    if schema.columns[i].required {
      count = count + 1
    }
    i = i + 1
  }
  count
}

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

///|
fn audit_empty_column_names(stats : TableStats) -> Array[String] {
  let names : Array[String] = []
  let mut i = 0
  while i < stats.columns.length() {
    if stats.columns[i].non_empty_count == 0 {
      names.push(stats.columns[i].name)
    }
    i = i + 1
  }
  names
}

///|
fn audit_constant_column_names(stats : TableStats) -> Array[String] {
  let names : Array[String] = []
  let mut i = 0
  while i < stats.columns.length() {
    let column = stats.columns[i]
    if column.non_empty_count > 1 && column.unique_count == 1 {
      names.push(column.name)
    }
    i = i + 1
  }
  names
}

///|
pub fn table_stats(table : Table) -> TableStats {
  let row_count = table.row_count()
  let column_count = audit_column_count(table)
  let columns : Array[ColumnStats] = []
  let mut empty_cells = 0
  let mut c = 0
  while c < column_count {
    let stats = audit_column_stats(table, c)
    empty_cells = empty_cells + stats.empty_count
    columns.push(stats)
    c = c + 1
  }
  let cell_count = audit_cell_count(row_count, column_count)
  let non_empty_cells = cell_count - empty_cells
  {
    row_count,
    column_count,
    cell_count,
    empty_cells,
    non_empty_cells,
    density_score: audit_density(cell_count, empty_cells),
    duplicate_row_count: audit_duplicate_row_count(table.rows),
    columns,
  }
}

///|
pub fn ColumnStats::summary(self : ColumnStats) -> String {
  let text =
    $|\{self.name}: empty=\{self.empty_count}, unique=\{self.unique_count}, min_len=\{self.min_length}, max_len=\{self.max_length}
  text
}

///|
pub fn TableStats::summary(self : TableStats) -> String {
  let text =
    $|rows=\{self.row_count}, columns=\{self.column_count}, density=\{self.density_score}, duplicate_rows=\{self.duplicate_row_count}
  text
}

///|
pub fn required_columns_report(
  table : Table,
  schema : Schema,
) -> RequiredColumnsReport {
  let required_count = audit_required_count(schema)
  let present_count = audit_present_required_count(table, schema)
  let missing = audit_missing_required_columns(table, schema)
  {
    required_count,
    present_count,
    missing_count: missing.length(),
    missing_columns: missing,
  }
}

///|
pub fn RequiredColumnsReport::summary(self : RequiredColumnsReport) -> String {
  if self.missing_count == 0 {
    let text =
      $|\{self.present_count}/\{self.required_count} required columns present
    text
  } else {
    let missing = audit_join_names(self.missing_columns)
    let text =
      $|\{self.present_count}/\{self.required_count} required columns present, missing: \{missing}
    text
  }
}

///|
pub fn schema_diff(
  previous : Schema,
  next : Schema,
) -> SchemaCompatibilityReport {
  let changes : Array[SchemaChange] = []
  let mut i = 0
  while i < previous.columns.length() {
    let old_rule = previous.columns[i]
    match audit_rule_index(next.columns, old_rule.name) {
      Some(index) => {
        let new_rule = next.columns[index]
        audit_push_type_change(changes, old_rule, new_rule)
        audit_push_required_change(changes, old_rule, new_rule)
      }
      None => audit_push_removed_column(changes, old_rule)
    }
    i = i + 1
  }
  let mut j = 0
  while j < next.columns.length() {
    let new_rule = next.columns[j]
    if !audit_rule_exists(previous.columns, new_rule.name) {
      audit_push_added_column(changes, new_rule)
    }
    j = j + 1
  }
  audit_push_extra_policy_change(changes, previous, next)
  let breaking = audit_count_breaking(changes)
  let non_breaking = audit_count_non_breaking(changes)
  {
    compatible: breaking == 0,
    breaking_changes: breaking,
    non_breaking_changes: non_breaking,
    changes,
  }
}

///|
pub fn SchemaCompatibilityReport::is_compatible(
  self : SchemaCompatibilityReport,
) -> Bool {
  self.compatible
}

///|
pub fn SchemaCompatibilityReport::summary(
  self : SchemaCompatibilityReport,
) -> String {
  if self.compatible {
    let text =
      $|compatible, non_breaking_changes=\{self.non_breaking_changes}
    text
  } else {
    let text =
      $|incompatible, breaking_changes=\{self.breaking_changes}, non_breaking_changes=\{self.non_breaking_changes}
    text
  }
}

///|
pub fn Table::duplicate_row_numbers(self : Table) -> Array[Int] {
  let seen : Array[Array[String]] = []
  let duplicates : Array[Int] = []
  let mut i = 0
  while i < self.rows.length() {
    if audit_seen_row(seen, self.rows[i]) {
      duplicates.push(i + 1)
    } else {
      seen.push(self.rows[i])
    }
    i = i + 1
  }
  duplicates
}

///|
pub fn Table::empty_column_names(self : Table) -> Array[String] {
  audit_empty_column_names(table_stats(self))
}

///|
pub fn Table::constant_column_names(self : Table) -> Array[String] {
  audit_constant_column_names(table_stats(self))
}

///|
pub fn Table::has_columns(self : Table, names : Array[String]) -> Bool {
  let mut i = 0
  while i < names.length() {
    if self.find_column(names[i]) is None {
      return false
    }
    i = i + 1
  }
  true
}

///|
pub fn Table::missing_columns(
  self : Table,
  names : Array[String],
) -> Array[String] {
  let missing : Array[String] = []
  let mut i = 0
  while i < names.length() {
    if self.find_column(names[i]) is None {
      missing.push(names[i])
    }
    i = i + 1
  }
  missing
}