///|
/// A stable, ordered view of the keys currently visible in the database.
///
/// Scans are built from the in-memory key directory and then read through the
/// normal record validation path. They therefore never expose transaction
/// markers, tombstones, or partially written records.
pub fn DB::scan_prefix(
  self : DB,
  prefix : String,
  limit : Int,
) -> Array[(String, Bytes)] raise MoonKVError {
  self.scan_after(prefix, "", limit)
}

///|
/// Returns at most `limit` keys with `prefix` after an exclusive cursor.
///
/// The returned keys are sorted lexicographically. Pass the last returned key
/// as `cursor` to fetch the next page. An empty cursor starts at the beginning.
pub fn DB::scan_after(
  self : DB,
  prefix : String,
  cursor : String,
  limit : Int,
) -> Array[(String, Bytes)] raise MoonKVError {
  if limit <= 0 {
    raise MoonKVError::InvalidRecord("scan limit must be greater than zero")
  }

  let keys = self.keydir.keys()
  keys.sort()
  let results = Array::new()
  for key in keys {
    if starts_with(key, prefix) &&
      (cursor == "" || key > cursor) &&
      self.contains(key) {
      let value = self.get(key)
      results.push((key, value))
      if results.length() >= limit {
        break
      }
    }
  }
  results
}

///|
/// A page of ordered scan results for clients that need explicit continuation.
pub struct ScanPage {
  /// Values returned in sorted key order for this page.
  rows : Array[(String, Bytes)]
  /// Cursor to pass back to `scan_page`, when another page exists.
  next_cursor : String?
  /// True when a sentinel row proved that another page exists.
  has_more : Bool
}

///|
/// Returns one page and an exclusive cursor for the following page.
///
/// Unlike inferring pagination from a short result array, `has_more` is
/// determined by reading one sentinel row. This makes an exact page at the end
/// of the keyspace unambiguous to HTTP, CLI, and replication adapters.
pub fn DB::scan_page(
  self : DB,
  prefix : String,
  cursor : String,
  limit : Int,
) -> ScanPage raise MoonKVError {
  if limit <= 0 {
    raise MoonKVError::InvalidRecord(
      "scan page limit must be greater than zero",
    )
  }
  let candidates = self.scan_after(prefix, cursor, limit + 1)
  let has_more = candidates.length() > limit
  let rows = Array::new()
  for i, row in candidates {
    if i < limit {
      rows.push(row)
    }
  }
  let next_cursor = if has_more {
    Some(rows[rows.length() - 1].0)
  } else {
    None
  }
  { rows, next_cursor, has_more }
}

///|
/// Returns keys in the half-open range `[start_key, end_key)`.
pub fn DB::scan_range(
  self : DB,
  start_key : String,
  end_key : String,
  limit : Int,
) -> Array[(String, Bytes)] raise MoonKVError {
  if limit <= 0 {
    raise MoonKVError::InvalidRecord("scan limit must be greater than zero")
  }
  if start_key > end_key {
    raise MoonKVError::InvalidRecord("scan start_key must not exceed end_key")
  }

  let keys = self.keydir.keys()
  keys.sort()
  let results = Array::new()
  for key in keys {
    if key >= start_key && key < end_key && self.contains(key) {
      let value = self.get(key)
      results.push((key, value))
      if results.length() >= limit {
        break
      }
    }
  }
  results
}

///|
/// Returns the number of currently indexed keys.
///
/// Expired entries remain lazily represented until accessed, so use
/// `verify` or a scan when a fully materialized visible count is required.
pub fn DB::key_count(self : DB) -> Int {
  self.keydir.keys().length()
}

///|
/// Returns the approximate bytes occupied by currently visible values.
///
/// This is metadata from the key directory, so it excludes record headers and
/// stale log records. It is intended for admission control and dashboards.
///
/// Values are measured in bytes and remain exact for the current keydir view.
pub fn DB::live_value_bytes(self : DB) -> Int64 {
  let mut total = 0L
  for key in self.keydir.keys() {
    if self.contains(key) {
      match self.keydir.get(key) {
        Some(entry) => total += entry.value_sz.to_int64()
        None => ()
      }
    }
  }
  total
}

///|
/// Deletes up to `limit` visible keys with the given prefix atomically.
///
/// The prefix may be empty to clear the visible keyspace. The operation first
/// collects matching keys, then commits one write batch, so an invalid limit
/// cannot leave a partial deletion.
///
/// Callers can repeat the operation with the same prefix to drain large sets.
pub fn DB::delete_prefix(
  self : DB,
  prefix : String,
  limit : Int,
) -> Int raise MoonKVError {
  if limit <= 0 {
    raise MoonKVError::InvalidRecord(
      "delete prefix limit must be greater than zero",
    )
  }
  let rows = self.scan_prefix(prefix, limit)
  let batch = self.new_write_batch()
  for row in rows {
    batch.delete(row.0)
  }
  let deleted = batch.ops.length()
  batch.commit()
  deleted
}

///|
/// Returns whether a key is currently visible.
pub fn DB::contains(self : DB, key : String) -> Bool {
  match self.keydir.get(key) {
    None => false
    Some(entry) =>
      entry.expires_at == 0L || self.logical_clock < entry.expires_at
  }
}

///|
/// A compact health and capacity snapshot for monitoring integrations.
pub struct DBStats {
  /// Number of non-deleted, non-expired keys in the in-memory index.
  key_count : Int
  /// Sum of visible value lengths, excluding record headers and stale logs.
  live_value_bytes : Int64
  /// Segment receiving the next append.
  active_file_id : Int
  /// Byte offset at which the next record will be appended.
  active_file_offset : Int
  /// Configured rollover threshold for the active segment.
  max_file_size : Int
  /// Monotonic logical timestamp used for TTL and recovery ordering.
  logical_clock : Int64
}

///|
/// Returns storage metadata without reading every value from disk.
/// This is safe to call from periodic monitoring loops.
/// The values describe the current process-local view.
/// Reopen the database to refresh it after an external writer.
pub fn DB::stats(self : DB) -> DBStats {
  {
    key_count: self.key_count(),
    live_value_bytes: self.live_value_bytes(),
    active_file_id: self.active_file_id,
    active_file_offset: self.active_file_offset,
    max_file_size: self.max_file_size,
    logical_clock: self.logical_clock,
  }
}

///|
/// Reads every visible key and returns the number successfully verified.
///
/// This is intended for startup diagnostics and maintenance tooling. It uses
/// the same bounds and checksum validation as `get`, so a corrupted value is
/// reported instead of being silently skipped.
pub fn DB::verify(self : DB) -> Int raise MoonKVError {
  let keys = self.keydir.keys()
  let mut verified = 0
  for key in keys {
    if self.contains(key) {
      let _ = self.get(key)
      verified += 1
    }
  }
  verified
}

///|
/// Computes a deterministic FNV-1a fingerprint of the visible keyspace.
///
/// The fingerprint is useful for smoke-checking replicas or detecting an
/// unexpected change between maintenance runs. It is not a cryptographic
/// digest and must not be used as an integrity or authentication boundary.
pub fn DB::fingerprint(self : DB) -> Int raise MoonKVError {
  let rows = self.scan_prefix("", self.key_count() + 1)
  let mut hash = -2128831035
  for row in rows {
    let key_bytes = string_to_bytes(row.0)
    for byte in key_bytes {
      hash = (hash ^ byte.to_int()) * 16777619
    }
    hash = (hash ^ 0) * 16777619
    for byte in row.1 {
      hash = (hash ^ byte.to_int()) * 16777619
    }
    hash = (hash ^ 255) * 16777619
  }
  hash
}