///|
pub(all) struct OverflowChain {
  first_page : UInt64
  pages : Array[UInt64]
  payload : Bytes
  diagnostics : Array[Diagnostic]
} derive(Eq, Debug)

///|
fn contains_page(values : Array[UInt64], page : UInt64) -> Bool {
  for value in values {
    if value == page {
      return true
    }
  }
  false
}

///|
/// Follow an overflow chain and return exactly `payload_bytes` bytes.
pub fn read_overflow_chain(
  database : DatabaseImage,
  first_page : UInt64,
  payload_bytes : Int,
) -> OverflowChain raise ParseError {
  guard payload_bytes >= 0 else {
    raise InvalidValue(0, "overflow payload length", payload_bytes.to_string())
  }
  if payload_bytes == 0 {
    return { first_page, pages: [], payload: b"", diagnostics: [] }
  }
  guard first_page != 0UL else {
    raise InvalidValue(0, "overflow page", "zero page starts a non-empty chain")
  }
  let pages : Array[UInt64] = []
  let bytes : Array[Byte] = []
  let diagnostics : Array[Diagnostic] = []
  let mut current = first_page
  let mut remaining = payload_bytes
  let capacity = database.header.usable_page_size() - 4
  while remaining > 0 {
    database.validate_page_number(current)
    guard !contains_page(pages, current) else {
      raise InvalidValue(
        database.page_offset(current),
        "overflow chain",
        "cycle at page " + current.to_string(),
      )
    }
    pages.push(current)
    let page = database.page_bytes(current)
    let reader = BinaryReader::range(
      page,
      0,
      database.header.usable_page_size(),
      "overflow page " + current.to_string(),
    )
    let next = reader.read_u32_be()
    let take = if remaining < capacity { remaining } else { capacity }
    let chunk = reader.read_bytes(take)
    for index = 0; index < chunk.length(); index = index + 1 {
      bytes.push(chunk[index])
    }
    remaining -= take
    if remaining > 0 && next == 0UL {
      raise InvalidValue(
        database.page_offset(current),
        "overflow chain",
        "chain ended with " + remaining.to_string() + " byte(s) missing",
      )
    }
    if remaining == 0 && next != 0UL {
      diagnostics.push(
        Diagnostic::warning(
          "OVERFLOW_EXTRA_PAGE",
          "overflow chain continues after the requested payload is complete",
          offset=database.page_offset(current),
          page_number=current,
        ),
      )
    }
    current = next
  }
  { first_page, pages, payload: Bytes::from_array(bytes[:]), diagnostics }
}

///|
pub fn join_payload(local_bytes : Bytes, overflow : Bytes) -> Bytes {
  Bytes::makei(local_bytes.length() + overflow.length(), fn(index) {
    if index < local_bytes.length() {
      local_bytes[index]
    } else {
      overflow[index - local_bytes.length()]
    }
  })
}