///|
/// Aggregate outcome of PageLens's structural consistency pass.
pub(all) struct IntegrityReport {
  diagnostics : Array[Diagnostic]
  database_opened : Bool
  pages_examined : Int
  btree_pages : Int
  btree_cells : Int
  records_decoded : Int
  overflow_pages : Int
  freelist_pages : Int
  wal_frames : Int
}

///|
pub fn IntegrityReport::count(
  self : IntegrityReport,
  severity : Severity,
) -> Int {
  let mut result = 0
  for diagnostic in self.diagnostics {
    if diagnostic.severity == severity {
      result = result + 1
    }
  }
  result
}

///|
pub fn IntegrityReport::error_count(self : IntegrityReport) -> Int {
  self.count(Error)
}

///|
pub fn IntegrityReport::warning_count(self : IntegrityReport) -> Int {
  self.count(Warning)
}

///|
pub fn IntegrityReport::info_count(self : IntegrityReport) -> Int {
  self.count(Info)
}

///|
pub fn IntegrityReport::is_ok(self : IntegrityReport) -> Bool {
  self.database_opened && self.error_count() == 0
}

///|
fn append_diagnostics(
  target : Array[Diagnostic],
  additions : Array[Diagnostic],
) -> Unit {
  for diagnostic in additions {
    target.push(diagnostic)
  }
}

///|
fn new_failed_report(diagnostics : Array[Diagnostic]) -> IntegrityReport {
  {
    diagnostics,
    database_opened: false,
    pages_examined: 0,
    btree_pages: 0,
    btree_cells: 0,
    records_decoded: 0,
    overflow_pages: 0,
    freelist_pages: 0,
    wal_frames: 0,
  }
}

///|
fn record_schema_root(
  record : SqliteRecord,
  database : DatabaseImage,
) -> UInt64? {
  guard record.columns.length() >= 4 else { return None }
  match record.columns[3].value {
    Integer(value) =>
      if value <= 0L || value.reinterpret_as_uint64() > database.page_count {
        None
      } else {
        Some(value.reinterpret_as_uint64())
      }
    _ => None
  }
}

///|
fn checked_overflow_chain(
  database : DatabaseImage,
  cell : BtreeCell,
  overflow_pages : Array[UInt64],
  diagnostics : Array[Diagnostic],
) -> Unit {
  guard cell.overflow_page is Some(first_page) else { return }
  let remaining = cell.payload_size - cell.local_payload.length()
  let result : OverflowChain? = Some(
    read_overflow_chain(database, first_page, remaining),
  ) catch {
    error => {
      diagnostics.push(
        Diagnostic::error(
          "OVERFLOW_CHAIN_INVALID",
          error.message(),
          page_number=first_page,
        ),
      )
      None
    }
  }
  guard result is Some(chain) else { return }
  append_diagnostics(diagnostics, chain.diagnostics)
  for page in chain.pages {
    if page_list_contains(overflow_pages, page) {
      diagnostics.push(
        Diagnostic::error(
          "OVERFLOW_PAGE_REUSED",
          "overflow page is referenced by more than one cell",
          page_number=page,
        ),
      )
    } else {
      overflow_pages.push(page)
    }
  }
}

///|
fn check_btree_page(
  database : DatabaseImage,
  page_number : UInt64,
  discover_schema_roots : Bool,
  schema_roots : Array[UInt64],
  overflow_pages : Array[UInt64],
  diagnostics : Array[Diagnostic],
  counters : Array[Int],
) -> Unit {
  let parsed : BtreePageHeader? = Some(
    parse_btree_page_header(database, page_number),
  ) catch {
    error => {
      diagnostics.push(
        Diagnostic::error("BTREE_PAGE_INVALID", error.message(), page_number~),
      )
      None
    }
  }
  guard parsed is Some(header) else { return }
  counters[0] = counters[0] + 1
  append_diagnostics(diagnostics, validate_page_layout(database, header))
  let cells : Array[BtreeCell]? = Some(parse_btree_page_cells(database, header)) catch {
    error => {
      diagnostics.push(
        Diagnostic::error("BTREE_CELL_INVALID", error.message(), page_number~),
      )
      None
    }
  }
  guard cells is Some(values) else { return }
  counters[1] = counters[1] + values.length()
  for cell in values {
    checked_overflow_chain(database, cell, overflow_pages, diagnostics)
    if cell.payload_size == 0 {
      continue
    }
    let parsed_record : SqliteRecord? = Some(parse_cell_record(database, cell)) catch {
      error => {
        diagnostics.push(
          Diagnostic::error(
            "RECORD_INVALID",
            error.message(),
            offset=(page_number.to_int() - 1) * database.header.page_size +
              cell.offset,
            page_number~,
          ),
        )
        None
      }
    }
    if parsed_record is Some(record) {
      counters[2] = counters[2] + 1
      if discover_schema_roots &&
        record_schema_root(record, database) is Some(root) &&
        !page_list_contains(schema_roots, root) {
        schema_roots.push(root)
      }
    }
  }
}

///|
fn check_btrees(
  database : DatabaseImage,
  diagnostics : Array[Diagnostic],
) -> (Int, Int, Int, Int, Array[UInt64]) {
  let schema_roots : Array[UInt64] = [1UL]
  let examined_pages : Array[UInt64] = []
  let overflow_pages : Array[UInt64] = []
  // b-tree pages, cells, records
  let counters : Array[Int] = [0, 0, 0]
  let mut root_index = 0
  while root_index < schema_roots.length() {
    let root = schema_roots[root_index]
    let traversal : BtreeTraversal? = Some(traverse_btree(database, root)) catch {
      error => {
        diagnostics.push(
          Diagnostic::error(
            "BTREE_TRAVERSAL_FAILED",
            error.message(),
            page_number=root,
          ),
        )
        None
      }
    }
    if traversal is Some(tree) {
      append_diagnostics(diagnostics, tree.diagnostics)
      for visit in tree.visits {
        if page_list_contains(examined_pages, visit.page_number) {
          if visit.page_number != root {
            diagnostics.push(
              Diagnostic::error(
                "BTREE_PAGE_SHARED",
                "B-tree page appears in multiple tree traversals",
                page_number=visit.page_number,
              ),
            )
          }
        } else {
          examined_pages.push(visit.page_number)
          check_btree_page(
            database,
            visit.page_number,
            root_index == 0,
            schema_roots,
            overflow_pages,
            diagnostics,
            counters,
          )
        }
      }
    }
    root_index = root_index + 1
  }
  (
    examined_pages.length(),
    counters[0],
    counters[1],
    counters[2],
    overflow_pages,
  )
}

///|
fn check_page_ownership(
  btree_pages : Int,
  database : DatabaseImage,
  freelist : FreelistReport,
  overflow_pages : Array[UInt64],
  diagnostics : Array[Diagnostic],
) -> Unit {
  for page in overflow_pages {
    if page_list_contains(freelist.trunk_pages, page) ||
      page_list_contains(freelist.leaf_pages, page) {
      diagnostics.push(
        Diagnostic::error(
          "PAGE_ROLE_CONFLICT",
          "page is referenced by both an overflow chain and the freelist",
          page_number=page,
        ),
      )
    }
  }
  let classified = btree_pages +
    freelist.observed_page_count() +
    overflow_pages.length()
  if classified < database.page_count.to_int() {
    diagnostics.push(
      Diagnostic::info(
        "UNCLASSIFIED_PAGES",
        (database.page_count.to_int() - classified).to_string() +
        " page(s) are not reachable as B-tree, freelist, or overflow pages; they may include pointer-map or lock-byte pages",
      ),
    )
  }
}

///|
fn check_wal_against_database(
  database : DatabaseImage,
  wal_data : Bytes,
  diagnostics : Array[Diagnostic],
) -> Int {
  let parsed : WalFile? = Some(parse_wal(wal_data)) catch {
    error => {
      diagnostics.push(Diagnostic::error("WAL_INVALID", error.message()))
      None
    }
  }
  guard parsed is Some(wal) else { return 0 }
  append_diagnostics(diagnostics, wal.diagnostics)
  if wal.header.page_size != database.header.page_size {
    diagnostics.push(
      Diagnostic::error(
        "WAL_PAGE_SIZE_MISMATCH",
        "WAL page size differs from the database page size",
        offset=8,
      ),
    )
  }
  for frame in wal.frames {
    if frame.database_size_after_commit > 0UL &&
      frame.page_number > frame.database_size_after_commit {
      diagnostics.push(
        Diagnostic::warning(
          "WAL_COMMIT_PAGE_RANGE",
          "commit frame page number exceeds its resulting database size",
          offset=frame.offset,
          page_number=frame.page_number,
        ),
      )
    }
  }
  wal.frames.length()
}

///|
/// Run a read-only consistency pass. Parsing failures become ERROR diagnostics
/// instead of escaping, which makes this function suitable for damaged files.
pub fn check_database(data : Bytes, wal_data? : Bytes) -> IntegrityReport {
  let diagnostics : Array[Diagnostic] = []
  let opened : DatabaseImage? = Some(DatabaseImage::open(data)) catch {
    error => {
      diagnostics.push(Diagnostic::error("DATABASE_INVALID", error.message()))
      None
    }
  }
  guard opened is Some(database) else { return new_failed_report(diagnostics) }
  append_diagnostics(
    diagnostics,
    validate_database_header(database.header, data.length()),
  )
  let freelist = analyze_freelist(database)
  append_diagnostics(diagnostics, freelist.diagnostics)
  let (pages_examined, btree_pages, btree_cells, records, overflow_pages) = check_btrees(
    database, diagnostics,
  )
  check_page_ownership(
    btree_pages, database, freelist, overflow_pages, diagnostics,
  )
  let wal_frames = match wal_data {
    Some(bytes) => check_wal_against_database(database, bytes, diagnostics)
    None => 0
  }
  {
    diagnostics,
    database_opened: true,
    pages_examined,
    btree_pages,
    btree_cells,
    records_decoded: records,
    overflow_pages: overflow_pages.length(),
    freelist_pages: freelist.observed_page_count(),
    wal_frames,
  }
}