///|
pub(all) enum SnapshotPageSource {
  DatabaseFile
  WalFrame(Int)
} derive(Eq, Debug)

///|
pub(all) struct SnapshotPage {
  page_number : UInt64
  source : SnapshotPageSource
  bytes : Bytes
} derive(Eq, Debug)

///|
/// One contiguous byte range changed by a WAL frame.
pub(all) struct ChangedRange {
  offset : Int
  length : Int
} derive(Eq, Debug)

///|
pub(all) struct PageDifference {
  page_number : UInt64
  frame_index : Int
  changed_bytes : Int
  first_changed_offset : Int?
  last_changed_offset : Int?
  ranges : Array[ChangedRange]
} derive(Eq, Debug)

///|
/// An immutable database view at one committed WAL boundary.
pub(all) struct SnapshotView {
  database : DatabaseImage
  wal : WalFile
  max_frame : Int
  logical_page_count : UInt64
}

///|
pub fn SnapshotView::page_size(self : SnapshotView) -> Int {
  self.database.header.page_size
}

///|
pub fn SnapshotView::uses_wal(self : SnapshotView) -> Bool {
  self.max_frame > 0
}

///|
fn committed_database_size(wal : WalFile, frame_index : Int) -> UInt64? {
  if frame_index <= 0 || frame_index > wal.frames.length() {
    return None
  }
  let frame = wal.frames[frame_index - 1]
  if frame.is_commit() && frame.is_valid() {
    Some(frame.database_size_after_commit)
  } else {
    None
  }
}

///|
pub fn SnapshotView::at_commit(
  database : DatabaseImage,
  wal : WalFile,
  frame_index : Int,
) -> SnapshotView raise ParseError {
  guard database.header.page_size == wal.header.page_size else {
    raise InvalidValue(
      8,
      "WAL page size",
      wal.header.page_size.to_string() +
      " differs from database page size " +
      database.header.page_size.to_string(),
    )
  }
  guard frame_index >= 0 && frame_index <= wal.frames.length() else {
    raise InvalidValue(frame_index, "snapshot frame", "outside WAL frame range")
  }
  let logical_page_count = if frame_index == 0 {
    database.page_count
  } else {
    guard committed_database_size(wal, frame_index) is Some(size) else {
      raise InvalidValue(
        frame_index, "snapshot frame", "frame is not a valid commit boundary",
      )
    }
    size
  }
  { database, wal, max_frame: frame_index, logical_page_count }
}

///|
/// Construct the newest fully committed view. Invalid and uncommitted WAL
/// frames never influence the resulting snapshot.
pub fn open_snapshot(
  database : DatabaseImage,
  wal : WalFile,
) -> SnapshotView raise ParseError {
  let commit = match wal.last_commit_frame {
    Some(value) => value
    None => 0
  }
  SnapshotView::at_commit(database, wal, commit)
}

///|
pub fn SnapshotView::read_page(
  self : SnapshotView,
  page_number : UInt64,
) -> SnapshotPage raise ParseError {
  guard page_number >= 1UL && page_number <= self.logical_page_count else {
    raise InvalidPageNumber(page_number, self.logical_page_count)
  }
  match self.wal.latest_frame_for_page(page_number, max_frame=self.max_frame) {
    Some(frame) =>
      { page_number, source: WalFrame(frame.index), bytes: frame.page_data }
    None => {
      guard page_number <= self.database.page_count else {
        raise InvalidValue(
          page_number.to_int(),
          "snapshot page",
          "page is not present in the database or committed WAL",
        )
      }
      {
        page_number,
        source: DatabaseFile,
        bytes: self.database.page_bytes(page_number),
      }
    }
  }
}

///|
pub fn SnapshotView::changed_pages(self : SnapshotView) -> Array[UInt64] {
  let pages : Array[UInt64] = []
  for frame in self.wal.frames {
    if frame.index <= self.max_frame &&
      frame.is_valid() &&
      frame.page_number <= self.logical_page_count &&
      !page_list_contains(pages, frame.page_number) {
      pages.push(frame.page_number)
    }
  }
  pages
}

///|
fn original_snapshot_page(
  snapshot : SnapshotView,
  page_number : UInt64,
) -> Bytes raise ParseError {
  if page_number <= snapshot.database.page_count {
    snapshot.database.page_bytes(page_number)
  } else {
    Bytes::make(snapshot.page_size(), b'\x00')
  }
}

///|
fn calculate_changed_ranges(
  before : Bytes,
  after : Bytes,
) -> Array[ChangedRange] {
  let ranges : Array[ChangedRange] = []
  let mut index = 0
  while index < before.length() && index < after.length() {
    if before[index] == after[index] {
      index = index + 1
      continue
    }
    let start = index
    while index < before.length() &&
          index < after.length() &&
          before[index] != after[index] {
      index = index + 1
    }
    ranges.push({ offset: start, length: index - start })
  }
  ranges
}

///|
pub fn SnapshotView::difference_for_page(
  self : SnapshotView,
  page_number : UInt64,
) -> PageDifference? raise ParseError {
  guard page_number >= 1UL && page_number <= self.logical_page_count else {
    raise InvalidPageNumber(page_number, self.logical_page_count)
  }
  guard self.wal.latest_frame_for_page(page_number, max_frame=self.max_frame)
    is Some(frame) else {
    return None
  }
  let before = original_snapshot_page(self, page_number)
  let after = frame.page_data
  let ranges = calculate_changed_ranges(before, after)
  let mut changed_bytes = 0
  for range in ranges {
    changed_bytes = changed_bytes + range.length
  }
  let first_changed_offset = if ranges.length() == 0 {
    None
  } else {
    Some(ranges[0].offset)
  }
  let last_changed_offset = if ranges.length() == 0 {
    None
  } else {
    let final_range = ranges[ranges.length() - 1]
    Some(final_range.offset + final_range.length - 1)
  }
  Some({
    page_number,
    frame_index: frame.index,
    changed_bytes,
    first_changed_offset,
    last_changed_offset,
    ranges,
  })
}

///|
pub fn SnapshotView::differences(
  self : SnapshotView,
) -> Array[PageDifference] raise ParseError {
  let differences : Array[PageDifference] = []
  for page in self.changed_pages() {
    if self.difference_for_page(page) is Some(difference) {
      differences.push(difference)
    }
  }
  differences
}