///|
/// SSTable 的写入与读取：`data block* | filter block | index block | footer(48)`
///
/// 磁盘布局见 `docs/design/format.md` §3.1。
///
/// **这一层只处理 `Bytes`，不碰文件系统。** 这不是偷懒，是有意的分层：
/// - 核心（memtable / block / sstable）保持 backend 无关，
///   CI 会在 wasm-gc / js 上编译它，证明没有偷偷依赖 native 特性
/// - 文件 IO 放在边界层（`fs` / dbcli），换平台或换存储介质只改那一层
/// - 测试不需要临时目录、不需要清理、不会因为 Windows 文件锁而随机失败
///
/// 代价是一次性把整个 SST 放进内存。对 v1 的规模（几十 MB）可以接受，
/// 后面引入 mmap 或分块读时，只需替换 `SstReader` 的取字节方式。

///|
/// footer 固定 48 字节（§3.5）。
pub const SST_FOOTER_SIZE : Int32 = 48

///|
/// 当前写出的 SST 版本。读到更大的版本必须拒绝打开（§5.2）。
///
/// **2** = data block / index block 尾部带 CRC32C（ADR-0010）。
/// 写端永远写这个版本，读端同时认 1（老库）与 2。
pub const SST_VERSION : Int32 = 2

///|
/// 唯一仍然支持的旧版本：block 没有尾部 CRC。
///
/// 留着它是为了「升级不炸库」—— 读到 v1 文件照常能读，
/// 但**不做内容校验**。下一次 compaction 会把数据重写成 v2。
pub const SST_VERSION_V1 : Int32 = 1

///|
/// 文件末尾的魔数 `"MSST"`。
const SST_MAGIC : Bytes = b"MSST"

///|
/// 解析出来的 footer。
pub struct SstFooter {
  filter_offset : Int32
  filter_size : Int32
  index_offset : Int32
  index_size : Int32
  smallest_seq : UInt64
  version : Int32
}

///|
/// 比较 `data[pos..]` 处的若干字节是否等于 `want`。
fn bytes_match(data : Bytes, pos : Int32, want : Bytes) -> Bool {
  if pos < 0 || pos + want.length() > data.length() {
    return false
  }
  for i in 0.. Int32? {
  match read_fixed64(data, pos) {
    Some(v) => if v > 0x7FFFFFFFUL { None } else { Some(v.to_int()) }
    None => None
  }
}

///|
/// 解码索引条目里的 block handle：`offset(varint64) + size(varint64)`。
fn decode_handle(v : Bytes) -> (Int32, Int32)? {
  let (off, p1) = match read_varint64(v, 0, v.length()) {
    Some(pair) => pair
    None => return None
  }
  let (size, _) = match read_varint64(v, p1, v.length()) {
    Some(pair) => pair
    None => return None
  }
  if off > 0x7FFFFFFFUL || size > 0x7FFFFFFFUL {
    return None
  }
  Some((off.to_int(), size.to_int()))
}

///|
/// 编码一个 block handle。
fn encode_handle(offset : Int32, size : Int32) -> Bytes {
  let b = Buffer::Buffer()
  write_varint64(b, offset.to_uint64())
  write_varint64(b, size.to_uint64())
  b.to_bytes()
}

///|
/// 按文件版本挑解码器。v2 校验尾部 CRC，v1 不校验。
///
/// 把版本判断收在这一处，是为了不让「哪条路径用哪个解码器」散在五个调用点 ——
/// 漏一处就会把 v1 的 block 当 v2 解，读出来的 `num_restarts` 是 CRC 的字节。
fn decode_block_ver(raw : Bytes, version : Int32) -> Block? {
  if version >= SST_VERSION {
    decode_block(raw)
  } else {
    decode_block_v1(raw)
  }
}

///|
/// filter 也一样按版本挑解码器（v2 有尾部 CRC）。
fn decode_bloom_ver(raw : Bytes, version : Int32) -> BloomFilter? {
  if version >= SST_VERSION {
    decode_bloom(raw)
  } else {
    decode_bloom_v1(raw)
  }
}

///|
/// 逐条追加、最后 `finish()` 出整个 SST 字节。
///
/// 使用契约：`add` 的 key 必须**严格升序**（与 `BlockBuilder` 一致）。
/// SST 内部按 key 有序，这是「二分查 index」和「归并 compaction」的前提。
///
/// **已知取舍**：`keys` 把所有 key 留在内存里，因为 bloom 需要在
/// `finish` 时知道元素个数才能定位图大小。1 万条 × 10 字节 = 100 KB，
/// 可以接受；到百万级需要改成「两遍写」或「可增长的 bloom」。
/// 这条记在 `PLAN.md` 的技术债里，不是忘了。
pub struct SstWriter {
  out : Buffer
  block : BlockBuilder
  index : BlockBuilder
  mut count : Int32
  mut block_count : Int32
  mut smallest_seq : UInt64
  keys : Array[Bytes]
  /// 要写出的格式版本。生产路径恒为 `SST_VERSION`；
  /// 只有兼容性测试会把它设成 `SST_VERSION_V1` 来造老文件。
  version : Int32
}

///|
pub fn SstWriter::new() -> SstWriter {
  SstWriter::with_version(SST_VERSION)
}

///|
/// 按指定版本造 writer。生产路径只有 `new()`（v2）；
/// 兼容性测试用 `with_version(SST_VERSION_V1)` 造 v1 老文件。
fn SstWriter::with_version(version : Int32) -> SstWriter {
  {
    out: Buffer::Buffer(),
    block: BlockBuilder::new(),
    index: BlockBuilder::new(),
    count: 0,
    block_count: 0,
    smallest_seq: 0xFFFFFFFFFFFFFFFFUL,
    keys: [],
    version,
  }
}

///|
/// 已写入的条目数。
pub fn SstWriter::entry_count(self : SstWriter) -> Int32 {
  self.count
}

///|
/// 已封盘的 data block 数。
pub fn SstWriter::block_count(self : SstWriter) -> Int32 {
  self.block_count
}

///|
/// 追加一条。key 必须严格大于上一条。
pub fn SstWriter::add(self : SstWriter, entry : Entry) -> Unit raise LsmError {
  self.block.add(entry)
  self.keys.push(entry.key)
  self.count = self.count + 1
  if entry.seq < self.smallest_seq {
    self.smallest_seq = entry.seq
  }
  if self.block.should_finish() {
    self.flush_block()
  }
}

///|
/// 把当前 block 落进 `out`，并在索引里记一条 `last_key -> (offset, size)`。
fn SstWriter::flush_block(self : SstWriter) -> Unit raise LsmError {
  if self.block.is_empty() {
    return
  }
  let raw = if self.version >= SST_VERSION {
    self.block.finish()
  } else {
    self.block.finish_v1()
  }
  let offset = self.out.length()
  self.out.write_bytes(raw[:])
  // 注意顺序：last_key 必须在 reset 之前取
  self.index.add({
    key: self.block.last_key(),
    value: encode_handle(offset, raw.length()),
    seq: 0UL,
    deleted: false,
  })
  self.block.reset()
  self.block_count = self.block_count + 1
}

///|
/// 产出完整 SST 字节。
///
/// 顺序严格按 §3.1：data blocks → filter → index → footer。
/// filter 要等所有 key 都写完才能建（需要知道元素个数来定位图大小），
/// 所以它只能在 `finish` 里生成。
pub fn SstWriter::finish(self : SstWriter) -> Bytes raise LsmError {
  self.flush_block()
  let bloom = BloomFilter::new(self.keys.length())
  for k in self.keys {
    bloom.put(k)
  }
  let filter = if self.version >= SST_VERSION {
    bloom.encode()
  } else {
    bloom.encode_v1()
  }
  let filter_offset = self.out.length()
  self.out.write_bytes(filter[:])
  let index_raw = if self.version >= SST_VERSION {
    self.index.finish()
  } else {
    self.index.finish_v1()
  }
  let index_offset = self.out.length()
  self.out.write_bytes(index_raw[:])
  write_fixed64(self.out, filter_offset.to_uint64())
  write_fixed64(self.out, filter.length().to_uint64())
  write_fixed64(self.out, index_offset.to_uint64())
  write_fixed64(self.out, index_raw.length().to_uint64())
  // 空 SST 没有 seq，写 0 而不是 0xFFFF...，避免读端把「未初始化」当真实值
  write_fixed64(self.out, if self.count == 0 { 0UL } else { self.smallest_seq })
  write_fixed32(self.out, self.version)
  self.out.write_bytes(SST_MAGIC[:])
  self.out.to_bytes()
}

///|
/// 一个已打开的 SST。持有整个文件字节与解析好的 index / filter。
pub struct SstReader {
  data : Bytes
  footer : SstFooter
  index : Block
  bloom : BloomFilter
  /// 文件覆盖的 key 区间 `[lo, hi]`（闭区间，且是**真实下界/上界**）。
  ///
  /// `None` 表示「拿不到边界」—— 空文件、或首块解码失败。
  /// 拿到 `None` 时所有范围判断都必须放行，不能猜：
  /// 猜错下界的后果是「数据明明在，却被跳过」，这是不可接受的错误方向。
  range : (Bytes, Bytes)?
}

///|
/// 打开一个 SST。任何结构性问题返回 `None`，不 panic。
///
/// 校验顺序刻意从「最便宜、最能早退」开始：长度 → magic → version → 区域边界。
/// 先查 magic 再查 offset 很重要 —— 如果文件根本不是 SST，
/// 那些 offset 是垃圾值，拿它们做边界检查只是白算。
pub fn open_sst(data : Bytes) -> SstReader? {
  let n = data.length()
  if n < SST_FOOTER_SIZE {
    return None
  }
  let f = n - SST_FOOTER_SIZE
  if !bytes_match(data, f + 44, SST_MAGIC) {
    return None
  }
  let version = match read_fixed32(data, f + 40) {
    Some(v) => v
    None => return None
  }
  if version != SST_VERSION && version != SST_VERSION_V1 {
    // §5.2：读到不认识的版本必须拒绝打开，绝不猜怎么读
    return None
  }
  let filter_offset = match read_offset(data, f) {
    Some(v) => v
    None => return None
  }
  let filter_size = match read_offset(data, f + 8) {
    Some(v) => v
    None => return None
  }
  let index_offset = match read_offset(data, f + 16) {
    Some(v) => v
    None => return None
  }
  let index_size = match read_offset(data, f + 24) {
    Some(v) => v
    None => return None
  }
  let smallest_seq = match read_fixed64(data, f + 32) {
    Some(v) => v
    None => return None
  }
  // 两个区域都必须完整落在 footer 之前，且 filter 在 index 之前
  if filter_offset + filter_size > f || index_offset + index_size > f {
    return None
  }
  if index_offset < filter_offset + filter_size {
    return None
  }
  let filter_raw = data[filter_offset:filter_offset + filter_size].to_owned()
  let bloom = match decode_bloom_ver(filter_raw, version) {
    Some(b) => b
    None => return None
  }
  let index_raw = data[index_offset:index_offset + index_size].to_owned()
  let index = match decode_block_ver(index_raw, version) {
    Some(b) => b
    None => return None
  }
  // 文件的 key 上界 = 最后一个 index 条目的 key（index 存的是每块的 last_key）。
  // 下界 = 首块的第一条 key。两者都必须**真实**成立，否则范围判断会误杀数据。
  let hi = match index.last() {
    Some(e) => e.key
    None => b""
  }
  let lo = if hi.length() == 0 {
    b""
  } else {
    match index.first() {
      Some(ie) =>
        match decode_handle(ie.value) {
          Some((off, size)) =>
            if off + size > data.length() {
              b""
            } else {
              match decode_block_ver(data[off:off + size].to_owned(), version) {
                Some(b) =>
                  match b.first() {
                    Some(e) => e.key
                    None => b""
                  }
                None => b""
              }
            }
          None => b""
        }
      None => b""
    }
  }
  // 只有上下界都拿到（且不为空）才启用范围判断
  let range = if hi.length() == 0 || lo.length() == 0 {
    None
  } else {
    Some((lo, hi))
  }
  Some({
    data,
    footer: {
      filter_offset,
      filter_size,
      index_offset,
      index_size,
      smallest_seq,
      version,
    },
    index,
    bloom,
    range,
  })
}

///|
pub fn SstReader::footer(self : SstReader) -> SstFooter {
  self.footer
}

///|
/// data block 个数（等于索引条目数）。
pub fn SstReader::block_count(self : SstReader) -> Int32 raise LsmError {
  self.index.entries().length()
}

///|
/// 按 key 查找。命中墓碑时返回 `deleted = true` 的条目，
/// **调用方必须自己判断**（上层要知道「这里有一个删除」，而不是「查不到」）。
///
/// 三步：bloom 快速否定 → index 定位 block → block 内二分。
pub fn SstReader::get(self : SstReader, key : Bytes) -> Entry? {
  let (h1, h2) = bloom_hashes(key)
  self.get_hashed(key, h1, h2)
}

///|
/// 同 `get`，但复用调用方算好的双哈希。
///
/// 一次点查要探 N 个 SST 的 filter，用的是同一条 key ——
/// 让每个文件各算一遍 FNV 是纯浪费。N 个文件时哈希成本从 N 次降到 1 次。
pub fn SstReader::get_hashed(
  self : SstReader,
  key : Bytes,
  h1 : UInt64,
  h2 : UInt64,
) -> Entry? {
  // 先比 key 范围，再问 bloom。
  //
  // 顺序是有讲究的：范围判断是两次 `memcmp`，bloom 是 7 次随机位访问。
  // L2 里的文件按 key 有序且互不重叠，一次未命中在 77 个文件里
  // 有 76 个能被这一步直接排除 —— bloom 从探 77 次降到 1 次。
  match self.range {
    Some((lo, hi)) =>
      if compare_bytes(key, lo) < 0 || compare_bytes(key, hi) > 0 {
        return None
      }
    None => ()
  }
  if !self.bloom.may_contain_hashes(h1, h2) {
    return None
  }
  let idx = match self.index.seek(key) {
    Some(e) => e
    None => return None
  }
  let (off, size) = match decode_handle(idx.value) {
    Some(p) => p
    None => return None
  }
  if off + size > self.data.length() {
    return None
  }
  let blk = match
    decode_block_ver(self.data[off:off + size].to_owned(), self.footer.version) {
    Some(b) => b
    None => return None
  }
  blk.get(key)
}

///|
/// 按 key 顺序解出全部条目。用于全扫描、校验与测试。
pub fn SstReader::entries(self : SstReader) -> Array[Entry] raise LsmError {
  let out : Array[Entry] = []
  for ie in self.index.entries() {
    let (off, size) = match decode_handle(ie.value) {
      Some(p) => p
      None =>
        raise LsmError::corruption(
          "sst: 索引条目里的 block handle 解码失败",
        )
    }
    if off + size > self.data.length() {
      raise LsmError::corruption("sst: data block 越界")
    }
    let blk = match
      decode_block_ver(
        self.data[off:off + size].to_owned(),
        self.footer.version,
      ) {
      Some(b) => b
      None => raise LsmError::corruption("sst: data block 解码失败")
    }
    for e in blk.entries() {
      out.push(e)
    }
  }
  out
}

///|
/// 只解出 `[start, end)` 区间里的条目，顺序与 `entries()` 一致。
///
/// **这是 `scan` 的性能关键。** `entries()` 会把整个文件解出来 ——
/// 一个 1 MiB 的 SST 大约 1 万条，扫 500 条也要付 1 万条的解码成本。
/// 数据量一大，`scan` 就退化成 O(整个库)，而调用方往往只要一小段。
///
/// 做法是拿索引做**块级剪枝**：索引里每条记录的是该 block 的 `last_key`，
/// 于是
///
/// - 二分找到第一个 `last_key >= start` 的 block —— 它之前的所有 block
///   最大 key 都小于 start，整块可以跳过
/// - 往后逐块解，块内逐条过滤（只有首尾两块可能部分越界）
/// - 遇到 `last_key >= end` 的 block，解完它就停 —— block 之间按 key
///   严格递增，后面每一块的 first_key 都更大，必然全部越界
///
/// 块内过滤而不是块内二分：一个 block 只有 4 KiB / 一百多条，
/// 解出来再线性筛比再写一套「块内定位 + 向后遍历」简单得多，
/// 而且这部分成本与块大小同阶，不是瓶颈。
///
/// **不接 `limit`**：单看一个文件，无法判断「前 N 条」是不是全局的前 N 条
/// （多路归并要跨文件比较）。限制由调用方在归并之后施加。
pub fn SstReader::scan_range(
  self : SstReader,
  start : Bytes,
  end : Bytes,
) -> Array[Entry] raise LsmError {
  let out : Array[Entry] = []
  // 整个文件与查询区间不相交就直接返回空，连索引二分都不用做。
  // 窄区间扫描时这一步能排除掉绝大多数文件。
  match self.range {
    Some((lo, hi)) =>
      if compare_bytes(hi, start) < 0 || compare_bytes(lo, end) >= 0 {
        return out
      }
    None => ()
  }
  let ies = self.index.entries()
  // 二分：第一个 last_key >= start 的 block
  let mut lo = 0
  let mut hi = ies.length()
  while lo < hi {
    let mid = (lo + hi) / 2
    if compare_bytes(ies[mid].key, start) < 0 {
      lo = mid + 1
    } else {
      hi = mid
    }
  }
  let mut i = lo
  while i < ies.length() {
    let ie = ies[i]
    let (off, size) = match decode_handle(ie.value) {
      Some(p) => p
      None =>
        raise LsmError::corruption(
          "sst: 索引条目里的 block handle 解码失败",
        )
    }
    if off + size > self.data.length() {
      raise LsmError::corruption("sst: data block 越界")
    }
    let blk = match
      decode_block_ver(
        self.data[off:off + size].to_owned(),
        self.footer.version,
      ) {
      Some(b) => b
      None => raise LsmError::corruption("sst: data block 解码失败")
    }
    for e in blk.entries() {
      if compare_bytes(e.key, start) >= 0 && compare_bytes(e.key, end) < 0 {
        out.push(e)
      }
    }
    // 这一块的 last_key 已经 >= end，后面的块必然全部越界
    if compare_bytes(ie.key, end) >= 0 {
      break
    }
    i = i + 1
  }
  out
}

///|
/// SST 上的**惰性游标**：只在被推进到新块时才解一块。
///
/// 与 `scan_range` 只差一处，但那一处是数量级的：`scan_range` 会把整个
/// 区间的条目一次全解出来，游标则是「要一条解一块」。
/// `Db::scan` 带 `limit` 时，差别就是「要 10 条只解 1 块」对
/// 「先把整个区间解完，再丢掉后面 99%」。
///
/// 为什么不是「给 `scan_range` 加个 limit 参数」：单看一个文件无法判断
/// 「前 N 条」是不是全局的前 N 条（多路归并要跨文件比较），
/// 所以限制只能由归并层施加，而限制一旦在归并层，来源就必须能被惰性推进。
priv struct SstCursor {
  reader : SstReader
  /// 索引条目（每个 block 的 last_key 与 handle）。构造时取一次，
  /// 之后每解一块只查一次数组，不重新解索引。
  ies : Array[Entry]
  start : Bytes
  end : Bytes
  /// 下一个要解的 block 下标。解完一块就 +1。
  mut block_i : Int32
  /// 当前块里**已经按区间过滤过**的条目。块消费完才换下一块。
  mut block : Array[Entry]
  mut pos : Int32
  /// 没有更多块了。**注意**：为 true 时 `block` 里可能还有没消费完的条目，
  /// 所以判断顺序必须是「先看 pos，再看 done」。
  mut done : Bool
  /// 解码失败过。
  ///
  /// `peek` 不能抛错 —— 归并器是纯函数，`MergeIterator::next` 的签名里
  /// 没有 `raise`，为了这个改动去改它的签名会波及 compaction 与全部归并测试。
  /// 所以这里只记标志，由 `Db::scan` 在返回前检查。
  mut failed : Bool
}

///|
/// 这个文件是否**可能**与 `[start, end)` 相交。
///
/// `range` 为 `None`（拿不到边界）时一律放行 —— 猜错下界的后果是
/// 「数据明明在却被跳过」，比多扫一个文件严重得多。
///
/// 为什么要单独暴露这一步：`cursor` 内部也会做同样的判断，但那是
/// **单个文件内部**的事。归并器每取一行都要遍历所有来源，来源数直接
/// 乘在热路径上 —— 55 个文件里只有 1 个可能相交时，把另外 54 个
/// 放进归并器就是 54 倍的无谓开销。
fn SstReader::overlaps(self : SstReader, start : Bytes, end : Bytes) -> Bool {
  match self.range {
    Some((lo, hi)) =>
      compare_bytes(hi, start) >= 0 && compare_bytes(lo, end) < 0
    None => true
  }
}

///|
/// 在一个 SST 上开游标，范围 `[start, end)`。
///
/// 与 `scan_range` 一样先做**文件级剪枝**（整个文件与区间不相交就直接空游标），
/// 再用索引二分定位第一个可能相交的 block。区别只是定位完不立刻解块。
fn SstReader::cursor(self : SstReader, start : Bytes, end : Bytes) -> SstCursor {
  // `entries()` 会抛 corruption，而游标的 `peek` 不能抛（见 `failed` 字段的说明）。
  // 所以这里把错误就地转成「空且失败」的游标，由 `Db::scan` 统一检查。
  let mut idx_failed = false
  let ies = self.index.entries() catch {
    _ => {
      idx_failed = true
      []
    }
  }
  let cur = {
    reader: self,
    ies,
    start,
    end,
    block_i: 0,
    block: [],
    pos: 0,
    done: false,
    failed: false,
  }
  if idx_failed {
    cur.failed = true
    cur.done = true
    return cur
  }
  match self.range {
    Some((lo, hi)) =>
      if compare_bytes(hi, start) < 0 || compare_bytes(lo, end) >= 0 {
        cur.done = true
        return cur
      }
    None => ()
  }
  // 二分：第一个 last_key >= start 的 block
  let mut lo = 0
  let mut hi = cur.ies.length()
  while lo < hi {
    let mid = (lo + hi) / 2
    if compare_bytes(cur.ies[mid].key, start) < 0 {
      lo = mid + 1
    } else {
      hi = mid
    }
  }
  cur.block_i = lo
  if lo >= cur.ies.length() {
    cur.done = true
  }
  cur
}

///|
/// 解下一块，并把块内落在区间外的条目滤掉。
///
/// 块内线性过滤而不是块内二分：一个 block 只有 4 KiB / 一百多条，
/// 解出来再筛比再写一套「块内定位 + 向后遍历」简单，成本与块大小同阶。
fn SstCursor::load_next(self : SstCursor) -> Unit {
  self.block = []
  self.pos = 0
  if self.block_i >= self.ies.length() {
    self.done = true
    return
  }
  let ie = self.ies[self.block_i]
  self.block_i = self.block_i + 1
  let (off, size) = match decode_handle(ie.value) {
    Some(p) => p
    None => {
      self.failed = true
      self.done = true
      return
    }
  }
  if off + size > self.reader.data.length() {
    self.failed = true
    self.done = true
    return
  }
  let blk = match
    decode_block_ver(
      self.reader.data[off:off + size].to_owned(),
      self.reader.footer.version,
    ) {
    Some(b) => b
    None => {
      self.failed = true
      self.done = true
      return
    }
  }
  let mut entries_failed = false
  let es = blk.entries() catch {
    _ => {
      entries_failed = true
      []
    }
  }
  if entries_failed {
    self.failed = true
    self.done = true
    return
  }
  let kept : Array[Entry] = []
  for e in es {
    if compare_bytes(e.key, self.start) >= 0 &&
      compare_bytes(e.key, self.end) < 0 {
      kept.push(e)
    }
  }
  self.block = kept
  // 这一块的 last_key 已经 >= end，后面的块必然全部越界。
  // 这里置 done 而不是 return：当前块还得让调用方读完。
  if compare_bytes(ie.key, self.end) >= 0 {
    self.done = true
  }
}

///|
/// 看当前条目，不推进。
///
/// 块用完才解下一块 —— 这是「惰性」的全部内容。
fn SstCursor::peek(self : SstCursor) -> Entry? {
  let mut out : Entry? = None
  let mut go = true
  while go {
    if self.pos < self.block.length() {
      out = Some(self.block[self.pos])
      go = false
    } else if self.done {
      go = false
    } else {
      self.load_next()
    }
  }
  out
}

///|
fn SstCursor::advance(self : SstCursor) -> Unit {
  self.pos = self.pos + 1
}