// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

use std::{
    collections::BTreeMap,
    fs::File,
    io::{Read, Seek, SeekFrom, Write},
    path::Path,
    sync::Mutex,
};

use crate::{
    ChecksumType, CompressionType, Error, Result,
    block::{BlockBuilder, BlockHandle, parse_entries, parse_index_entries},
    checksum::checksum,
    coding,
    compression::{MAX_BLOCK_SIZE, compress, decompress},
};

const BLOCK_TRAILER_LEN: u64 = 5;
const BLOCK_BASED_MAGIC: u64 = 0x88e2_41b7_85f4_cff7;
const LEGACY_BLOCK_BASED_MAGIC: u64 = 0xdb47_7524_8b80_fb57;
const FOOTER_V0_LEN: u64 = 48;
const FOOTER_LEN: u64 = 53;
const VALUE_TYPE_VALUE: u64 = 1;

/// Options for a canonical legacy SST writer.
#[derive(Clone, Debug)]
pub struct Options {
    pub block_size: usize,
    pub block_restart_interval: usize,
    pub format_version: u32,
    pub checksum: ChecksumType,
    pub compression: CompressionType,
    pub compression_level: i32,
}

impl Default for Options {
    fn default() -> Self {
        Self {
            block_size: 32 * 1024,
            block_restart_interval: 16,
            format_version: 5,
            checksum: ChecksumType::Xxh3,
            compression: CompressionType::Lz4,
            compression_level: 0,
        }
    }
}

/// Metadata and bytes returned after finishing a legacy SST writer.
#[derive(Debug)]
pub struct WriterOutput {
    pub bytes: Vec<u8>,
    pub smallest_key: Vec<u8>,
    pub largest_key: Vec<u8>,
    pub entry_count: u64,
}

/// A sorted external SST writer compatible with RocksDB BlockBasedTable.
pub struct Writer<W = Vec<u8>> {
    options: Options,
    output: W,
    output_len: u64,
    data_block: BlockBuilder,
    data_block_last_key: Vec<u8>,
    index_entries: Vec<(Vec<u8>, BlockHandle)>,
    smallest_key: Vec<u8>,
    largest_key: Vec<u8>,
    entry_count: u64,
    raw_key_size: u64,
    raw_value_size: u64,
}

impl Writer<Vec<u8>> {
    pub fn new(options: Options) -> Result<Self> {
        Self::new_with_output(options, Vec::new())
    }

    pub fn finish(mut self) -> Result<WriterOutput> {
        self.finish_table()?;
        Ok(WriterOutput {
            bytes: self.output,
            smallest_key: self.smallest_key,
            largest_key: self.largest_key,
            entry_count: self.entry_count,
        })
    }
}

impl<W: Write> Writer<W> {
    /// Creates a writer that writes finished blocks directly to `output`.
    pub fn new_with_output(options: Options, output: W) -> Result<Self> {
        if options.format_version > 5 {
            return Err(Error::Unsupported(format!(
                "format version {}",
                options.format_version
            )));
        }
        if options.format_version == 0 && options.checksum != ChecksumType::Crc32c {
            return Err(Error::Unsupported(
                "format version 0 requires CRC32c".into(),
            ));
        }
        let restart_interval = options.block_restart_interval;
        Ok(Self {
            options,
            output,
            output_len: 0,
            data_block: BlockBuilder::new(restart_interval, false),
            data_block_last_key: Vec::new(),
            index_entries: Vec::new(),
            smallest_key: Vec::new(),
            largest_key: Vec::new(),
            entry_count: 0,
            raw_key_size: 0,
            raw_value_size: 0,
        })
    }

    /// Adds a user key/value pair. Keys must be strictly increasing.
    pub fn put(&mut self, key: &[u8], value: &[u8]) -> Result<()> {
        if self.entry_count != 0 && self.largest_key.as_slice() >= key {
            return Err(Error::Corruption(
                "keys must be added in strict ascending order".into(),
            ));
        }
        if self.entry_count == 0 {
            self.smallest_key.extend_from_slice(key);
        }
        self.largest_key.clear();
        self.largest_key.extend_from_slice(key);

        let internal_key = internal_key(key);
        self.data_block.add(&internal_key, value, None)?;
        self.data_block_last_key = internal_key;
        self.entry_count += 1;
        self.raw_key_size += (key.len() + 8) as u64;
        self.raw_value_size += value.len() as u64;
        if self.data_block.estimated_size() >= self.options.block_size {
            self.flush_data_block()?;
        }
        Ok(())
    }

    pub fn file_size(&self) -> u64 {
        self.output_len + self.data_block.estimated_size() as u64
    }

    /// Finishes the SST and returns the underlying output sink.
    pub fn finish_into_inner(mut self) -> Result<W> {
        self.finish_table()?;
        Ok(self.output)
    }

    fn finish_table(&mut self) -> Result<()> {
        if self.entry_count == 0 {
            return Err(Error::Corruption("cannot create an empty SST".into()));
        }
        self.flush_data_block()?;
        let data_size = self.output_len;

        let value_delta = self.options.format_version >= 4;
        let mut index = BlockBuilder::new(1, value_delta);
        let mut previous_handle: Option<BlockHandle> = None;
        for (key, handle) in &self.index_entries {
            let index_key = if self.options.format_version <= 2 {
                key.as_slice()
            } else {
                user_key(key)?
            };
            let mut encoded = Vec::new();
            handle.encode(&mut encoded);
            let mut delta = Vec::new();
            if let Some(previous) = previous_handle {
                let size_delta = i64::try_from(handle.size)
                    .and_then(|size| i64::try_from(previous.size).map(|previous| size - previous))
                    .map_err(|_| Error::Unsupported("index block size exceeds i64".into()))?;
                coding::put_zigzag64(&mut delta, size_delta);
            }
            index.add(index_key, &encoded, Some(&delta))?;
            previous_handle = Some(*handle);
        }
        let index_block = index.finish();
        let index_handle = self.write_block(&index_block, true)?;

        let properties = self.build_properties(data_size, index_handle);
        let properties_handle = self.write_block(&properties, false)?;
        let mut metaindex = BlockBuilder::new(1, false);
        let mut properties_handle_value = Vec::new();
        properties_handle.encode(&mut properties_handle_value);
        metaindex.add(b"rocksdb.properties", &properties_handle_value, None)?;
        let metaindex = metaindex.finish();
        let metaindex_handle = self.write_block(&metaindex, false)?;
        self.write_footer(metaindex_handle, index_handle)?;
        self.output.flush()?;
        Ok(())
    }

    fn flush_data_block(&mut self) -> Result<()> {
        if self.data_block_last_key.is_empty() {
            return Ok(());
        }
        let restart_interval = self.options.block_restart_interval;
        let block = std::mem::replace(
            &mut self.data_block,
            BlockBuilder::new(restart_interval, false),
        )
        .finish();
        let handle = self.write_block(&block, true)?;
        self.index_entries
            .push((std::mem::take(&mut self.data_block_last_key), handle));
        Ok(())
    }

    fn write_block(&mut self, raw: &[u8], allow_compression: bool) -> Result<BlockHandle> {
        let mut compression = if allow_compression {
            self.options.compression
        } else {
            CompressionType::None
        };
        let mut stored = compress(
            compression,
            self.options.format_version,
            raw,
            self.options.compression_level,
        )?;
        // RocksDB defaults to requiring at least 12.5% space saving.
        if compression != CompressionType::None && stored.len() * 8 > raw.len() * 7 {
            compression = CompressionType::None;
            stored.clear();
            stored.extend_from_slice(raw);
        }
        let handle = BlockHandle {
            offset: self.output_len,
            size: stored.len() as u64,
        };
        let block_checksum = checksum(self.options.checksum, &stored, compression as u8);
        self.output.write_all(&stored)?;
        self.output.write_all(&[compression as u8])?;
        self.output.write_all(&block_checksum.to_le_bytes())?;
        self.output_len = self
            .output_len
            .checked_add(handle.size)
            .and_then(|len| len.checked_add(BLOCK_TRAILER_LEN))
            .ok_or_else(|| Error::Unsupported("SST size exceeds u64".into()))?;
        Ok(handle)
    }

    fn write_footer(&mut self, metaindex: BlockHandle, index: BlockHandle) -> Result<()> {
        let mut footer = Vec::with_capacity(FOOTER_LEN as usize);
        if self.options.format_version == 0 {
            metaindex.encode(&mut footer);
            index.encode(&mut footer);
            footer.resize(40, 0);
            footer.extend_from_slice(&LEGACY_BLOCK_BASED_MAGIC.to_le_bytes());
        } else {
            footer.push(self.options.checksum as u8);
            metaindex.encode(&mut footer);
            index.encode(&mut footer);
            footer.resize(41, 0);
            footer.extend_from_slice(&self.options.format_version.to_le_bytes());
            footer.extend_from_slice(&BLOCK_BASED_MAGIC.to_le_bytes());
        }
        self.output.write_all(&footer)?;
        self.output_len = self
            .output_len
            .checked_add(footer.len() as u64)
            .ok_or_else(|| Error::Unsupported("SST size exceeds u64".into()))?;
        Ok(())
    }

    fn build_properties(&self, data_size: u64, index_handle: BlockHandle) -> Vec<u8> {
        let mut properties = BTreeMap::<&[u8], Vec<u8>>::new();
        let mut add_u64 = |name: &'static [u8], value: u64| {
            let mut encoded = Vec::new();
            coding::put_varint64(&mut encoded, value);
            properties.insert(name, encoded);
        };
        add_u64(b"rocksdb.data.size", data_size);
        add_u64(b"rocksdb.deleted.keys", 0);
        add_u64(b"rocksdb.filter.size", 0);
        add_u64(b"rocksdb.fixed.key.length", 0);
        add_u64(
            b"rocksdb.format.version",
            self.options.format_version as u64,
        );
        add_u64(
            b"rocksdb.index.key.is.user.key",
            u64::from(self.options.format_version >= 3),
        );
        add_u64(
            b"rocksdb.index.value.is.delta.encoded",
            u64::from(self.options.format_version >= 4),
        );
        add_u64(b"rocksdb.index.size", index_handle.size + BLOCK_TRAILER_LEN);
        add_u64(b"rocksdb.merge.operands", 0);
        add_u64(b"rocksdb.num.data.blocks", self.index_entries.len() as u64);
        add_u64(b"rocksdb.num.entries", self.entry_count);
        add_u64(b"rocksdb.num.filter_entries", 0);
        add_u64(b"rocksdb.num.range-deletions", 0);
        add_u64(b"rocksdb.original.file.number", 0);
        add_u64(b"rocksdb.raw.key.size", self.raw_key_size);
        add_u64(b"rocksdb.raw.value.size", self.raw_value_size);
        properties.insert(
            b"rocksdb.comparator",
            b"leveldb.BytewiseComparator".to_vec(),
        );
        properties.insert(
            b"rocksdb.compression",
            self.options.compression.rocksdb_name().as_bytes().to_vec(),
        );
        properties.insert(
            b"rocksdb.external_sst_file.global_seqno",
            0_u64.to_le_bytes().to_vec(),
        );
        properties.insert(
            b"rocksdb.external_sst_file.version",
            2_u32.to_le_bytes().to_vec(),
        );
        properties.insert(
            b"rocksdb.property.collectors",
            b"[SstFileWriterPropertiesCollector]".to_vec(),
        );

        let mut block = BlockBuilder::new(i32::MAX as usize, false);
        for (name, value) in properties {
            block
                .add(name, &value, None)
                .expect("sorted property keys are valid");
        }
        block.finish()
    }
}

/// One decoded external SST entry.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Entry {
    pub key: Vec<u8>,
    pub value: Vec<u8>,
}

/// Aggregate entry sizes corresponding to RocksDB table properties.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct TableStats {
    pub entry_count: u64,
    pub raw_key_size: u64,
    pub raw_value_size: u64,
}

trait ReadSeek: Read + Seek {}

impl<T: Read + Seek> ReadSeek for T {}

/// A seekable reader for a RocksDB block-based external SST file.
pub struct Reader {
    input: Mutex<Box<dyn ReadSeek + Send>>,
    format_version: u32,
    checksum: ChecksumType,
    data_handles: Vec<BlockHandle>,
    meta_handles: Vec<BlockHandle>,
    index_handle: BlockHandle,
    compression_name: Option<String>,
}

impl Reader {
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        Self::new(File::open(path)?)
    }

    pub fn new<R: Read + Seek + Send + 'static>(mut input: R) -> Result<Self> {
        let file_len = input.seek(SeekFrom::End(0))?;
        if file_len < FOOTER_V0_LEN {
            return Err(Error::Corruption(
                "file is shorter than an SST footer".into(),
            ));
        }
        let footer_buf = if file_len >= FOOTER_LEN {
            input.seek(SeekFrom::End(-(FOOTER_LEN as i64)))?;
            let mut footer = vec![0; FOOTER_LEN as usize];
            input.read_exact(&mut footer)?;
            footer
        } else {
            Vec::new()
        };
        let magic = footer_buf
            .get(45..53)
            .map(|bytes| u64::from_le_bytes(bytes.try_into().unwrap()));
        let (format_version, checksum_type, footer) = if magic == Some(BLOCK_BASED_MAGIC) {
            let version = u32::from_le_bytes(footer_buf[41..45].try_into().unwrap());
            if version > 5 {
                return Err(Error::Unsupported(format!("format version {version}")));
            }
            (version, ChecksumType::try_from(footer_buf[0])?, footer_buf)
        } else {
            input.seek(SeekFrom::End(-(FOOTER_V0_LEN as i64)))?;
            let mut footer = vec![0; FOOTER_V0_LEN as usize];
            input.read_exact(&mut footer)?;
            let legacy_magic = u64::from_le_bytes(footer[40..48].try_into().unwrap());
            if legacy_magic != LEGACY_BLOCK_BASED_MAGIC {
                return Err(Error::Corruption(format!(
                    "unexpected table magic {legacy_magic:#x}"
                )));
            }
            (0, ChecksumType::Crc32c, footer)
        };
        let handles_start = usize::from(format_version != 0);
        let mut pos = handles_start;
        let blocks_end = file_len - footer.len() as u64;
        let metaindex_handle =
            validate_handle(BlockHandle::decode(&footer, &mut pos)?, blocks_end)?;
        let index_handle = validate_handle(BlockHandle::decode(&footer, &mut pos)?, blocks_end)?;

        let index_raw = read_block(&mut input, index_handle, format_version, checksum_type)?;
        let data_handles = parse_index_entries(&index_raw, format_version >= 4)?
            .into_iter()
            .map(|(_, handle)| validate_handle(handle, blocks_end))
            .collect::<Result<Vec<_>>>()?;

        let metaindex_raw =
            read_block(&mut input, metaindex_handle, format_version, checksum_type)?;
        let meta_entries = parse_entries(&metaindex_raw)?;
        let mut compression_name = None;
        let meta_handles = meta_entries
            .into_iter()
            .map(|(name, value)| {
                let mut pos = 0;
                let handle = BlockHandle::decode(&value, &mut pos)?;
                let handle = validate_handle(handle, blocks_end)?;
                if name == b"rocksdb.properties" {
                    let properties = read_block(&mut input, handle, format_version, checksum_type)?;
                    for (property_name, property_value) in parse_entries(&properties)? {
                        if property_name == b"rocksdb.compression" {
                            compression_name =
                                Some(String::from_utf8(property_value).map_err(|_| {
                                    Error::Corruption("compression property is not UTF-8".into())
                                })?);
                        }
                    }
                }
                Ok(handle)
            })
            .collect::<Result<Vec<_>>>()?;

        Ok(Self {
            input: Mutex::new(Box::new(input)),
            format_version,
            checksum: checksum_type,
            data_handles,
            meta_handles,
            index_handle,
            compression_name,
        })
    }

    pub fn format_version(&self) -> u32 {
        self.format_version
    }

    pub fn checksum_type(&self) -> ChecksumType {
        self.checksum
    }

    pub fn compression_name(&self) -> String {
        self.compression_name
            .clone()
            .unwrap_or_else(|| "Unknown".to_owned())
    }

    pub fn iter(&self) -> TableIterator<'_> {
        TableIterator {
            reader: self,
            block_index: 0,
            entries: Vec::new(),
            entry_index: 0,
            valid: false,
        }
    }

    pub fn verify_checksum(&self) -> Result<()> {
        let data_handles = self.data_handles.clone();
        let meta_handles = self.meta_handles.clone();
        let mut input = self.lock_input()?;
        for handle in data_handles
            .into_iter()
            .chain(meta_handles)
            .chain(std::iter::once(self.index_handle))
        {
            read_block(&mut **input, handle, self.format_version, self.checksum)?;
        }
        Ok(())
    }

    pub fn table_stats(&self) -> Result<TableStats> {
        let mut stats = TableStats::default();
        for index in 0..self.data_handles.len() {
            for entry in self.read_entries(index)? {
                stats.entry_count += 1;
                stats.raw_key_size += (entry.key.len() + 8) as u64;
                stats.raw_value_size += entry.value.len() as u64;
            }
        }
        Ok(stats)
    }

    fn read_entries(&self, index: usize) -> Result<Vec<Entry>> {
        let handle = *self
            .data_handles
            .get(index)
            .ok_or_else(|| Error::Corruption("data block index out of bounds".into()))?;
        let mut input = self.lock_input()?;
        parse_entries(&read_block(
            &mut **input,
            handle,
            self.format_version,
            self.checksum,
        )?)?
        .into_iter()
        .map(|(internal_key, value)| {
            Ok(Entry {
                key: user_key(&internal_key)?.to_vec(),
                value,
            })
        })
        .collect()
    }

    fn lock_input(&self) -> Result<std::sync::MutexGuard<'_, Box<dyn ReadSeek + Send>>> {
        self.input
            .lock()
            .map_err(|_| Error::Corruption("SST reader lock is poisoned".into()))
    }
}

/// Collects all entries from an iterator. Intended for tests and diagnostics.
pub fn collect_sst(mut iter: TableIterator<'_>) -> Vec<(Vec<u8>, Vec<u8>)> {
    let mut entries = Vec::new();
    while iter.valid().expect("SST iterator validity check") {
        entries.push((iter.key().to_vec(), iter.value().to_vec()));
        iter.next().expect("advance SST iterator");
    }
    entries
}

/// An iterator over user keys and values in a legacy SST reader.
pub struct TableIterator<'a> {
    reader: &'a Reader,
    block_index: usize,
    entries: Vec<Entry>,
    entry_index: usize,
    valid: bool,
}

impl TableIterator<'_> {
    pub fn seek_to_first(&mut self) -> Result<bool> {
        self.block_index = 0;
        self.entry_index = 0;
        self.load_non_empty_forward()
    }

    pub fn seek_to_last(&mut self) -> Result<bool> {
        if self.reader.data_handles.is_empty() {
            self.valid = false;
            return Ok(false);
        }
        self.block_index = self.reader.data_handles.len() - 1;
        loop {
            self.entries = self.reader.read_entries(self.block_index)?;
            if let Some(last) = self.entries.len().checked_sub(1) {
                self.entry_index = last;
                self.valid = true;
                return Ok(true);
            }
            if self.block_index == 0 {
                self.valid = false;
                return Ok(false);
            }
            self.block_index -= 1;
        }
    }

    pub fn seek(&mut self, target: &[u8]) -> Result<bool> {
        // Production uses seek only to trim an SST at a region boundary or for
        // a rare API-version compatibility check. One additional sequential
        // scan on those paths is acceptable and keeps this legacy reader small.
        if !self.seek_to_first()? {
            return Ok(false);
        }
        while self.valid && self.key() < target {
            self.next()?;
        }
        Ok(self.valid)
    }

    pub fn next(&mut self) -> Result<bool> {
        if !self.valid {
            return Ok(false);
        }
        self.entry_index += 1;
        if self.entry_index < self.entries.len() {
            return Ok(true);
        }
        self.block_index += 1;
        self.entry_index = 0;
        self.load_non_empty_forward()
    }

    pub fn valid(&self) -> Result<bool> {
        Ok(self.valid)
    }

    pub fn key(&self) -> &[u8] {
        &self.current().key
    }

    pub fn value(&self) -> &[u8] {
        &self.current().value
    }

    fn current(&self) -> &Entry {
        assert!(self.valid, "key/value called on an invalid SST iterator");
        &self.entries[self.entry_index]
    }

    fn load_non_empty_forward(&mut self) -> Result<bool> {
        while self.block_index < self.reader.data_handles.len() {
            self.entries = self.reader.read_entries(self.block_index)?;
            if !self.entries.is_empty() {
                self.entry_index = 0;
                self.valid = true;
                return Ok(true);
            }
            self.block_index += 1;
        }
        self.valid = false;
        Ok(false)
    }
}

fn internal_key(user_key: &[u8]) -> Vec<u8> {
    let mut key = Vec::with_capacity(user_key.len() + 8);
    key.extend_from_slice(user_key);
    key.extend_from_slice(&VALUE_TYPE_VALUE.to_le_bytes());
    key
}

fn user_key(internal_key: &[u8]) -> Result<&[u8]> {
    let trailer = internal_key
        .len()
        .checked_sub(8)
        .ok_or_else(|| Error::Corruption("internal key has no trailer".into()))?;
    let tag = u64::from_le_bytes(internal_key[trailer..].try_into().unwrap());
    if tag >> 8 != 0 || tag as u8 != VALUE_TYPE_VALUE as u8 {
        return Err(Error::Unsupported(format!(
            "internal key sequence/type tag {tag:#x}"
        )));
    }
    Ok(&internal_key[..trailer])
}

fn validate_handle(handle: BlockHandle, blocks_end: u64) -> Result<BlockHandle> {
    if handle.size > MAX_BLOCK_SIZE as u64 {
        return Err(Error::Corruption(format!(
            "block size {} exceeds limit {MAX_BLOCK_SIZE}",
            handle.size
        )));
    }
    let end = handle
        .offset
        .checked_add(handle.size)
        .and_then(|end| end.checked_add(BLOCK_TRAILER_LEN))
        .ok_or_else(|| Error::Corruption("block handle overflow".into()))?;
    if end > blocks_end {
        return Err(Error::Corruption(format!(
            "block at {}..{end} exceeds block-data boundary {blocks_end}",
            handle.offset
        )));
    }
    Ok(handle)
}

fn read_block<R: Read + Seek + ?Sized>(
    input: &mut R,
    handle: BlockHandle,
    format_version: u32,
    checksum_type: ChecksumType,
) -> Result<Vec<u8>> {
    if handle.size > MAX_BLOCK_SIZE as u64 {
        return Err(Error::Corruption(format!(
            "block size {} exceeds limit {MAX_BLOCK_SIZE}",
            handle.size
        )));
    }
    let len = usize::try_from(handle.size)
        .map_err(|_| Error::Unsupported("block does not fit in memory".into()))?;
    let mut stored = vec![0; len];
    input.seek(SeekFrom::Start(handle.offset))?;
    input.read_exact(&mut stored)?;
    let mut trailer = [0; BLOCK_TRAILER_LEN as usize];
    input.read_exact(&mut trailer)?;
    let compression = CompressionType::try_from(trailer[0])?;
    let expected = u32::from_le_bytes(trailer[1..].try_into().unwrap());
    let actual = checksum(checksum_type, &stored, trailer[0]);
    if actual != expected {
        return Err(Error::Corruption(format!(
            "block checksum mismatch at {}: expected {expected:#x}, got {actual:#x}",
            handle.offset
        )));
    }
    decompress(compression, format_version, &stored)
}

#[cfg(test)]
mod tests {
    use std::{
        io::Cursor,
        sync::{Arc, Mutex},
    };

    use super::*;

    #[derive(Clone, Default)]
    struct SharedOutput(Arc<Mutex<Vec<u8>>>);

    impl Write for SharedOutput {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(buf);
            Ok(buf.len())
        }

        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    fn round_trip(options: Options) {
        let mut writer = Writer::new(options).unwrap();
        for index in 0..1_000 {
            writer
                .put(
                    format!("key-{index:04}").as_bytes(),
                    format!("value-{index:04}-{}", "x".repeat(index % 80)).as_bytes(),
                )
                .unwrap();
        }
        let output = writer.finish().unwrap();
        let reader = Reader::new(Cursor::new(output.bytes)).unwrap();
        reader.verify_checksum().unwrap();
        let mut iter = reader.iter();
        assert!(iter.seek_to_first().unwrap());
        for index in 0..1_000 {
            assert_eq!(iter.key(), format!("key-{index:04}").as_bytes());
            if index != 999 {
                assert!(iter.next().unwrap());
            }
        }
        assert!(!iter.next().unwrap());
        assert!(iter.seek_to_last().unwrap());
        assert_eq!(iter.key(), b"key-0999");
        assert!(iter.seek(b"key-0412").unwrap());
        assert_eq!(iter.key(), b"key-0412");
    }

    #[test]
    fn streaming_writer_matches_memory_writer() {
        let options = Options {
            block_size: 256,
            ..Options::default()
        };
        let mut memory_writer = Writer::new(options.clone()).unwrap();
        let output = SharedOutput::default();
        let mut streaming_writer = Writer::new_with_output(options, output.clone()).unwrap();
        for index in 0..100 {
            let key = format!("key-{index:04}");
            let value = format!("value-{index:04}-{}", "x".repeat(index % 80));
            memory_writer.put(key.as_bytes(), value.as_bytes()).unwrap();
            streaming_writer
                .put(key.as_bytes(), value.as_bytes())
                .unwrap();
        }
        assert!(!output.0.lock().unwrap().is_empty());

        let memory_output = memory_writer.finish().unwrap().bytes;
        streaming_writer.finish_into_inner().unwrap();
        let streaming_output = output.0.lock().unwrap().clone();
        assert_eq!(streaming_output, memory_output);
    }

    #[test]
    fn all_backup_compressions_and_footer_versions_round_trip() {
        for format_version in 0..=5 {
            for compression in [
                CompressionType::None,
                CompressionType::Snappy,
                CompressionType::Lz4,
                CompressionType::Zstd,
            ] {
                round_trip(Options {
                    format_version,
                    compression,
                    checksum: if format_version == 0 {
                        ChecksumType::Crc32c
                    } else {
                        Options::default().checksum
                    },
                    ..Options::default()
                });
            }
        }
    }

    #[test]
    fn checksum_corruption_is_reported() {
        let mut writer = Writer::new(Options::default()).unwrap();
        writer.put(b"key", b"value").unwrap();
        let mut output = writer.finish().unwrap().bytes;
        output[0] ^= 1;
        let reader = Reader::new(Cursor::new(output)).unwrap();
        let err = reader.verify_checksum().unwrap_err();
        assert!(err.to_string().contains("checksum mismatch"), "{err}");
    }

    #[test]
    fn rejects_unsorted_keys() {
        let mut writer = Writer::new(Options::default()).unwrap();
        writer.put(b"b", b"1").unwrap();
        assert!(writer.put(b"a", b"2").is_err());
        assert!(writer.put(b"b", b"2").is_err());
    }

    #[test]
    fn rejects_oversized_footer_handle_before_allocation() {
        // Leave one empty block and its trailer before the footer so the
        // metaindex handle is valid. The oversized index handle must be
        // rejected without attempting to allocate its advertised size.
        let mut file = vec![0; BLOCK_TRAILER_LEN as usize];
        let mut footer = vec![ChecksumType::Crc32c as u8];
        BlockHandle::default().encode(&mut footer);
        BlockHandle {
            offset: 0,
            size: MAX_BLOCK_SIZE as u64 + 1,
        }
        .encode(&mut footer);
        footer.resize(41, 0);
        footer.extend_from_slice(&5_u32.to_le_bytes());
        footer.extend_from_slice(&BLOCK_BASED_MAGIC.to_le_bytes());
        file.extend_from_slice(&footer);

        let err = match Reader::new(Cursor::new(file)) {
            Ok(_) => panic!("oversized block handle was accepted"),
            Err(err) => err,
        };
        assert!(err.to_string().contains("exceeds limit"), "{err}");
    }
}