///|
/// A decoded cell from one of SQLite's four B-tree page types.
pub(all) struct BtreeCell {
  page_number : UInt64
  index : Int
  offset : Int
  page_type : PageType
  left_child_page : UInt64?
  rowid : Int64?
  payload_size : Int
  local_payload : Bytes
  overflow_page : UInt64?
  encoded_size : Int
} derive(Eq, Debug)

///|
pub fn BtreeCell::has_overflow(self : BtreeCell) -> Bool {
  self.overflow_page is Some(_)
}

///|
pub fn BtreeCell::local_payload_size(self : BtreeCell) -> Int {
  self.local_payload.length()
}

///|
fn checked_payload_size(varint : Varint, offset : Int) -> Int raise ParseError {
  guard varint.value <= 0x7fffffffUL else {
    raise InvalidValue(offset, "cell payload size", varint.value.to_string())
  }
  varint.value.to_int()
}

///|
fn read_local_payload(
  reader : BinaryReader,
  page_type : PageType,
  payload_size : Int,
  usable_size : Int,
) -> (Bytes, UInt64?) raise ParseError {
  let local_size = local_payload_size(page_type, payload_size, usable_size)
  let local_bytes = reader.read_bytes(local_size)
  let overflow = if local_size < payload_size {
    Some(reader.read_u32_be())
  } else {
    None
  }
  (local_bytes, overflow)
}

///|
pub fn parse_btree_cell(
  database : DatabaseImage,
  header : BtreePageHeader,
  index : Int,
) -> BtreeCell raise ParseError {
  guard index >= 0 && index < header.cell_count else {
    raise InvalidValue(index, "cell index", index.to_string())
  }
  let page = database.page_bytes(header.page_number)
  let offset = header.cell_pointers[index]
  let usable_size = database.header.usable_page_size()
  let reader = BinaryReader::range(
    page,
    offset,
    usable_size - offset,
    "B-tree cell " +
    index.to_string() +
    " on page " +
    header.page_number.to_string(),
  )
  let left_child_page : UInt64? = match header.page_type {
    IndexInterior | TableInterior => Some(reader.read_u32_be())
    IndexLeaf | TableLeaf => None
  }
  let mut rowid : Int64? = None
  let mut payload_size = 0
  let mut local_payload = b""
  let mut overflow_page : UInt64? = None
  match header.page_type {
    TableInterior =>
      rowid = Some(reader.read_varint().value.reinterpret_as_int64())
    TableLeaf => {
      let size = reader.read_varint()
      payload_size = checked_payload_size(size, offset)
      rowid = Some(reader.read_varint().value.reinterpret_as_int64())
      let (local_bytes, overflow) = read_local_payload(
        reader,
        header.page_type,
        payload_size,
        usable_size,
      )
      local_payload = local_bytes
      overflow_page = overflow
    }
    IndexInterior | IndexLeaf => {
      let size = reader.read_varint()
      payload_size = checked_payload_size(size, offset)
      let (local_bytes, overflow) = read_local_payload(
        reader,
        header.page_type,
        payload_size,
        usable_size,
      )
      local_payload = local_bytes
      overflow_page = overflow
    }
  }
  if left_child_page is Some(child) {
    guard child >= 1UL && child <= database.page_count else {
      raise InvalidPageNumber(child, database.page_count)
    }
  }
  if overflow_page is Some(page_number) {
    guard page_number >= 1UL && page_number <= database.page_count else {
      raise InvalidPageNumber(page_number, database.page_count)
    }
  }
  {
    page_number: header.page_number,
    index,
    offset,
    page_type: header.page_type,
    left_child_page,
    rowid,
    payload_size,
    local_payload,
    overflow_page,
    encoded_size: reader.relative_position(),
  }
}

///|
pub fn parse_btree_page_cells(
  database : DatabaseImage,
  header : BtreePageHeader,
) -> Array[BtreeCell] raise ParseError {
  let cells : Array[BtreeCell] = []
  for index = 0; index < header.cell_count; index = index + 1 {
    cells.push(parse_btree_cell(database, header, index))
  }
  cells
}

///|
pub fn materialize_cell_payload(
  database : DatabaseImage,
  cell : BtreeCell,
) -> Bytes raise ParseError {
  let missing = cell.payload_size - cell.local_payload.length()
  if missing <= 0 {
    return cell.local_payload
  }
  guard cell.overflow_page is Some(first_page) else {
    raise InvalidValue(
      database.page_offset(cell.page_number) + cell.offset,
      "cell overflow pointer",
      "missing pointer for spilled payload",
    )
  }
  let overflow = read_overflow_chain(database, first_page, missing)
  join_payload(cell.local_payload, overflow.payload)
}

///|
pub fn parse_cell_record(
  database : DatabaseImage,
  cell : BtreeCell,
) -> SqliteRecord raise ParseError {
  guard cell.payload_size > 0 else {
    raise InvalidValue(cell.offset, "record payload", "cell has no payload")
  }
  parse_record(
    materialize_cell_payload(database, cell),
    database.header.text_encoding,
  )
}

///|
pub(all) struct BtreeVisit {
  page_number : UInt64
  depth : Int
  page_type : PageType
  cell_count : Int
  child_pages : Array[UInt64]
} derive(Eq, Debug)

///|
pub(all) struct BtreeTraversal {
  root_page : UInt64
  visits : Array[BtreeVisit]
  diagnostics : Array[Diagnostic]
} derive(Eq, Debug)

///|
fn traverse_btree_page(
  database : DatabaseImage,
  page_number : UInt64,
  depth : Int,
  max_depth : Int,
  visited : Array[UInt64],
  visits : Array[BtreeVisit],
  diagnostics : Array[Diagnostic],
) -> Unit raise ParseError {
  if depth > max_depth {
    diagnostics.push(
      Diagnostic::error(
        "BTREE_DEPTH_LIMIT",
        "B-tree traversal exceeded the configured depth limit",
        page_number~,
      ),
    )
    return
  }
  if contains_page(visited, page_number) {
    diagnostics.push(
      Diagnostic::error(
        "BTREE_PAGE_REUSED",
        "B-tree page is referenced more than once",
        page_number~,
      ),
    )
    return
  }
  visited.push(page_number)
  let header = parse_btree_page_header(database, page_number)
  let cells = parse_btree_page_cells(database, header)
  let children : Array[UInt64] = []
  for cell in cells {
    if cell.left_child_page is Some(child) {
      children.push(child)
    }
  }
  if header.right_most_pointer is Some(child) {
    children.push(child)
  }
  visits.push({
    page_number,
    depth,
    page_type: header.page_type,
    cell_count: header.cell_count,
    child_pages: children,
  })
  for child in children {
    traverse_btree_page(
      database,
      child,
      depth + 1,
      max_depth,
      visited,
      visits,
      diagnostics,
    )
  }
}

///|
pub fn traverse_btree(
  database : DatabaseImage,
  root_page : UInt64,
  max_depth? : Int = 64,
) -> BtreeTraversal raise ParseError {
  database.validate_page_number(root_page)
  guard max_depth >= 0 else {
    raise InvalidValue(0, "B-tree depth limit", max_depth.to_string())
  }
  let visits : Array[BtreeVisit] = []
  let diagnostics : Array[Diagnostic] = []
  traverse_btree_page(
    database,
    root_page,
    0,
    max_depth,
    [],
    visits,
    diagnostics,
  )
  { root_page, visits, diagnostics }
}