walrust/

ltx.rs

//! LTX (Litestream Transaction) format support
//!
//! This module provides utilities for encoding and decoding LTX files,
//! which are Litestream-compatible transaction files containing SQLite pages.

use anyhow::{anyhow, Result};
use litepages::{Checksum, Decoder, Encoder, Header, HeaderFlags, PageNum, PageSize, TXID};
use std::io::{Read, Write};
use std::path::Path;
use std::time::SystemTime;

/// Create an LTX file from a SQLite database snapshot
pub fn encode_snapshot<W: Write>(
    writer: W,
    db_path: &Path,
    page_size: u32,
    txid: u64,
) -> Result<()> {
    let db_data = std::fs::read(db_path)?;
    let page_size_val = PageSize::new(page_size).map_err(|e| anyhow!("Invalid page size: {}", e))?;
    let num_pages = db_data.len() / page_size as usize;

    let header = Header {
        flags: HeaderFlags::COMPRESS_LZ4,
        page_size: page_size_val,
        commit: PageNum::new(num_pages as u32).map_err(|e| anyhow!("Invalid page count: {}", e))?,
        min_txid: TXID::ONE, // Snapshot starts at TXID 1
        max_txid: TXID::new(txid).map_err(|e| anyhow!("Invalid TXID: {}", e))?,
        timestamp: SystemTime::now(),
        pre_apply_checksum: None,
    };

    let mut encoder = Encoder::new(writer, &header)?;

    // Encode each page
    for i in 0..num_pages {
        let page_num = PageNum::new((i + 1) as u32).map_err(|e| anyhow!("Invalid page num: {}", e))?;
        let start = i * page_size as usize;
        let end = start + page_size as usize;
        let page_data = &db_data[start..end];

        encoder.encode_page(page_num, page_data)?;
    }

    // Compute final checksum and finish
    let checksum = compute_db_checksum(&db_data);
    encoder.finish(checksum)?;

    Ok(())
}

/// Result of decoding an LTX snapshot, including the post-apply checksum for chain verification
#[derive(Debug)]
pub struct DecodeResult {
    pub header: Header,
    pub post_apply_checksum: Checksum,
}

/// Decode an LTX file and reconstruct the database (full write)
///
/// Returns the header and post_apply_checksum for chain tracking.
/// The post_apply_checksum should be used as the expected pre_apply_checksum
/// for the next incremental LTX file.
///
/// Checksum verification is skipped when the NO_CHECKSUM flag is set (litestream compatibility).
pub fn decode_to_db<R: Read>(reader: R, output_path: &Path) -> Result<DecodeResult> {
    let (mut decoder, header) = Decoder::new(reader)?;

    // Check if this is a litestream file with NO_CHECKSUM flag
    let skip_checksums = header.flags.contains(HeaderFlags::NO_CHECKSUM);

    if skip_checksums {
        tracing::debug!(
            "Skipping checksum verification (NO_CHECKSUM flag set - litestream compatibility)"
        );
    }

    let page_size = header.page_size.into_inner() as usize;
    let num_pages = header.commit.into_inner() as usize;

    let mut db_data = vec![0u8; num_pages * page_size];
    let mut page_buf = vec![0u8; page_size];

    while let Some(page_num) = decoder.decode_page(&mut page_buf)? {
        let idx = (page_num.into_inner() - 1) as usize;
        let start = idx * page_size;
        db_data[start..start + page_size].copy_from_slice(&page_buf);
    }

    // Verify file checksum (internal integrity)
    let trailer = decoder.finish()?;

    // Write database file
    std::fs::write(output_path, &db_data)?;

    // Compute actual checksum from database
    let actual_checksum = compute_db_checksum(&db_data);

    // Verify post_apply_checksum matches actual written DB (skip if NO_CHECKSUM)
    if !skip_checksums {
        if trailer.post_apply_checksum != actual_checksum {
            return Err(anyhow!(
                "Post-apply checksum mismatch after decode: expected {:016x}, got {:016x}. \
                 This may indicate corruption in the LTX file.",
                trailer.post_apply_checksum.into_inner(),
                actual_checksum.into_inner()
            ));
        }
        tracing::debug!(
            "Post-apply checksum verified: {:016x}",
            actual_checksum.into_inner()
        );
    }

    tracing::debug!(
        "Decoded snapshot (TXID {}-{})",
        header.min_txid.into_inner(),
        header.max_txid.into_inner()
    );

    Ok(DecodeResult {
        header,
        post_apply_checksum: actual_checksum,
    })
}

/// Result of applying an LTX file, including the post-apply checksum for chain verification
#[derive(Debug)]
pub struct ApplyResult {
    pub header: Header,
    pub post_apply_checksum: Checksum,
}

/// Apply an incremental LTX file to an existing database (in-place page writes)
///
/// This verifies the checksum chain using chained page checksums:
/// 1. After applying: verifies post_apply_checksum matches chain_checksum(pre, decoded_pages)
///
/// No full-DB read is needed for verification — the chain is self-contained.
///
/// Checksum verification is skipped when the NO_CHECKSUM flag is set (litestream compatibility).
///
/// Returns the header and post_apply_checksum for chain tracking.
pub fn apply_ltx_to_db<R: Read>(reader: R, db_path: &Path) -> Result<ApplyResult> {
    use std::fs::OpenOptions;
    use std::io::{Seek, SeekFrom, Write as IoWrite};

    let (mut decoder, header) = Decoder::new(reader)?;

    // Check if this is a litestream file with NO_CHECKSUM flag
    let skip_checksums = header.flags.contains(HeaderFlags::NO_CHECKSUM);

    if skip_checksums {
        tracing::debug!(
            "Skipping checksum verification (NO_CHECKSUM flag set - litestream compatibility)"
        );
    }

    let page_size = header.page_size.into_inner() as usize;
    let mut page_buf = vec![0u8; page_size];

    // Open existing db file for page-level writes
    let mut file = OpenOptions::new()
        .write(true)
        .open(db_path)
        .map_err(|e| anyhow!("Failed to open database for in-place apply: {}", e))?;

    // Streaming chain hasher: hash each page during decode instead of accumulating
    // a Vec<(u32, Vec<u8>)>. Pages arrive sorted from our encoder.
    let mut chain_hasher = header.pre_apply_checksum.map(ChainHasher::new);

    while let Some(page_num) = decoder.decode_page(&mut page_buf)? {
        let offset = (page_num.into_inner() as u64 - 1) * page_size as u64;
        file.seek(SeekFrom::Start(offset))?;
        file.write_all(&page_buf)?;
        if let Some(ref mut hasher) = chain_hasher {
            hasher.update(page_num.into_inner(), &page_buf);
        }
    }

    // Ensure all writes are flushed
    file.sync_all()?;
    drop(file);

    // Verify file checksum (internal integrity)
    let trailer = decoder.finish()?;

    // Get page count before consuming the hasher
    let page_count = chain_hasher.as_ref().map(|h| h.page_count()).unwrap_or(0);

    // Verify checksums using chained page hash (skip if NO_CHECKSUM)
    let post_checksum = if !skip_checksums {
        if let Some(expected_pre) = header.pre_apply_checksum {
            tracing::debug!(
                "Pre-apply checksum: {:016x}",
                expected_pre.into_inner()
            );
        }

        if let Some(hasher) = chain_hasher {
            let expected_post = hasher.finish();
            if trailer.post_apply_checksum != expected_post {
                return Err(anyhow!(
                    "Post-apply checksum mismatch: expected {:016x}, got {:016x}. \
                     This may indicate corruption during apply.",
                    trailer.post_apply_checksum.into_inner(),
                    expected_post.into_inner()
                ));
            }
            tracing::debug!(
                "Post-apply checksum verified (chain): {:016x}",
                expected_post.into_inner()
            );
            expected_post
        } else {
            trailer.post_apply_checksum
        }
    } else {
        if let Some(hasher) = chain_hasher {
            hasher.finish()
        } else {
            compute_checksum_from_file(db_path)?
        }
    };

    tracing::debug!(
        "Applied {} pages in-place (TXID {}-{})",
        page_count,
        header.min_txid.into_inner(),
        header.max_txid.into_inner()
    );

    Ok(ApplyResult {
        header,
        post_apply_checksum: post_checksum,
    })
}

/// Compute checksum from database file (for checksum tracking)
pub fn compute_checksum_from_file(db_path: &Path) -> Result<Checksum> {
    let data = std::fs::read(db_path)?;
    Ok(compute_db_checksum(&data))
}

/// Encode WAL changes as an LTX file (incremental, not snapshot)
///
/// `pre_checksum`: Checksum of database BEFORE applying these changes (required for incrementals)
/// `post_checksum`: Checksum of database AFTER applying these changes
///
/// The caller must compute `post_checksum` by simulating the changes or reading the final state.
pub fn encode_wal_changes<W: Write>(
    writer: W,
    pages: &[(u32, Vec<u8>)], // (page_num, page_data)
    page_size: u32,
    min_txid: u64,
    max_txid: u64,
    commit_page: u32,
    pre_checksum: Option<Checksum>,
    post_checksum: Checksum,
) -> Result<Checksum> {
    let page_size_val = PageSize::new(page_size).map_err(|e| anyhow!("Invalid page size: {}", e))?;

    let header = Header {
        flags: HeaderFlags::COMPRESS_LZ4,
        page_size: page_size_val,
        commit: PageNum::new(commit_page).map_err(|e| anyhow!("Invalid commit page: {}", e))?,
        min_txid: TXID::new(min_txid).map_err(|e| anyhow!("Invalid min TXID: {}", e))?,
        max_txid: TXID::new(max_txid).map_err(|e| anyhow!("Invalid max TXID: {}", e))?,
        timestamp: SystemTime::now(),
        pre_apply_checksum: pre_checksum,
    };

    let mut encoder = Encoder::new(writer, &header)?;

    // Sort by index to avoid cloning all page data
    let mut indices: Vec<usize> = (0..pages.len()).collect();
    indices.sort_by_key(|&i| pages[i].0);

    for &i in &indices {
        let pn = PageNum::new(pages[i].0).map_err(|e| anyhow!("Invalid page num: {}", e))?;
        encoder.encode_page(pn, &pages[i].1)?;
    }

    let trailer = encoder.finish(post_checksum)?;

    Ok(trailer.post_apply_checksum)
}

/// Chained page checksum: O(changed pages), not O(entire DB).
///
/// Computes `SHA-256(pre_checksum_bytes || page1_num_be || page1_data || page2_num_be || page2_data || ...)`
/// with pages sorted by page number for determinism.
pub fn chain_checksum(pre: Checksum, pages: &[(u32, Vec<u8>)]) -> Checksum {
    use sha2::{Digest, Sha256};
    let mut hasher = Sha256::new();
    hasher.update(pre.into_inner().to_be_bytes());

    let mut sorted_indices: Vec<usize> = (0..pages.len()).collect();
    sorted_indices.sort_by_key(|&i| pages[i].0);

    for &i in &sorted_indices {
        hasher.update(pages[i].0.to_be_bytes());
        hasher.update(&pages[i].1);
    }

    let result = hasher.finalize();
    Checksum::new(u64::from_be_bytes(result[0..8].try_into().unwrap()))
}

/// Streaming chain hasher: computes chain checksum incrementally during LTX decode.
///
/// Pages MUST be fed in sorted order (by page number). Our encoder sorts pages,
/// so they arrive sorted during decode. This eliminates the need to accumulate
/// all decoded pages in a Vec just for checksum verification.
pub struct ChainHasher {
    hasher: sha2::Sha256,
    page_count: usize,
}

impl ChainHasher {
    /// Create a new chain hasher seeded with the pre-apply checksum.
    pub fn new(pre: Checksum) -> Self {
        use sha2::Digest;
        let mut hasher = sha2::Sha256::new();
        hasher.update(pre.into_inner().to_be_bytes());
        Self { hasher, page_count: 0 }
    }

    /// Feed a page into the hash. Pages must arrive in sorted order.
    pub fn update(&mut self, page_num: u32, page_data: &[u8]) {
        use sha2::Digest;
        self.hasher.update(page_num.to_be_bytes());
        self.hasher.update(page_data);
        self.page_count += 1;
    }

    /// Finalize and return the chain checksum.
    pub fn finish(self) -> Checksum {
        use sha2::Digest;
        let result = self.hasher.finalize();
        Checksum::new(u64::from_be_bytes(result[0..8].try_into().unwrap()))
    }

    /// Number of pages fed into the hasher.
    pub fn page_count(&self) -> usize {
        self.page_count
    }
}

/// Verify an LTX file by decoding all pages and checking the checksum
/// Returns the header on success, or an error describing the verification failure
pub fn verify_ltx<R: Read>(reader: R) -> Result<Header> {
    let (mut decoder, header) = Decoder::new(reader)?;

    let page_size = header.page_size.into_inner() as usize;
    let mut page_buf = vec![0u8; page_size];

    // Decode all pages (required to verify checksum)
    while decoder.decode_page(&mut page_buf)?.is_some() {
        // Just consume the pages
    }

    // Verify checksum - this will fail if corrupted
    decoder.finish()?;

    Ok(header)
}

/// Compute database checksum (single u64 from SHA256)
pub fn compute_db_checksum(data: &[u8]) -> Checksum {
    use sha2::{Digest, Sha256};
    let mut hasher = Sha256::new();
    hasher.update(data);
    let result = hasher.finalize();
    // Take first 8 bytes as u64
    let hash = u64::from_be_bytes(result[0..8].try_into().unwrap());
    Checksum::new(hash)
}


#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[test]
    fn test_snapshot_roundtrip_single_page() {
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let ltx_path = dir.path().join("test.ltx");
        let restored_path = dir.path().join("restored.db");

        // Create a simple SQLite database (4KB page size, 1 page)
        let page_size = 4096u32;
        let db_data = vec![0x42u8; page_size as usize];
        std::fs::write(&db_path, &db_data).unwrap();

        // Encode as LTX
        let ltx_file = std::fs::File::create(&ltx_path).unwrap();
        encode_snapshot(ltx_file, &db_path, page_size, 1).unwrap();

        // Decode back
        let ltx_file = std::fs::File::open(&ltx_path).unwrap();
        let result = decode_to_db(ltx_file, &restored_path).unwrap();

        // Verify
        let restored_data = std::fs::read(&restored_path).unwrap();
        assert_eq!(db_data, restored_data);
        assert_eq!(result.header.page_size.into_inner(), page_size);
        assert_eq!(result.header.min_txid.into_inner(), 1);
        assert_eq!(result.header.max_txid.into_inner(), 1);
    }

    #[test]
    fn test_snapshot_roundtrip_multiple_pages() {
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let restored_path = dir.path().join("restored.db");

        let page_size = 4096u32;
        let num_pages = 10;

        // Create database with multiple pages, each with unique content
        let mut db_data = Vec::new();
        for i in 0..num_pages {
            let mut page = vec![(i as u8).wrapping_mul(17); page_size as usize];
            // Add page number marker at start
            page[0..4].copy_from_slice(&(i as u32).to_be_bytes());
            db_data.extend(page);
        }
        std::fs::write(&db_path, &db_data).unwrap();

        // Encode as LTX to buffer
        let mut ltx_buffer = Vec::new();
        encode_snapshot(&mut ltx_buffer, &db_path, page_size, 100).unwrap();

        // Decode back
        let cursor = std::io::Cursor::new(ltx_buffer);
        let result = decode_to_db(cursor, &restored_path).unwrap();

        // Verify byte-for-byte
        let restored_data = std::fs::read(&restored_path).unwrap();
        assert_eq!(db_data.len(), restored_data.len());
        assert_eq!(db_data, restored_data);
        assert_eq!(result.header.commit.into_inner(), num_pages as u32);
        assert_eq!(result.header.max_txid.into_inner(), 100);
    }

    #[test]
    fn test_snapshot_various_page_sizes() {
        let dir = tempdir().unwrap();

        for page_size in [512u32, 1024, 2048, 4096, 8192, 16384, 32768] {
            let db_path = dir.path().join(format!("test_{}.db", page_size));
            let restored_path = dir.path().join(format!("restored_{}.db", page_size));

            // Create 3-page database
            let db_data: Vec<u8> = (0..3)
                .flat_map(|i| vec![(i * 50) as u8; page_size as usize])
                .collect();
            std::fs::write(&db_path, &db_data).unwrap();

            let mut ltx_buffer = Vec::new();
            encode_snapshot(&mut ltx_buffer, &db_path, page_size, 1).unwrap();

            let cursor = std::io::Cursor::new(ltx_buffer);
            let result = decode_to_db(cursor, &restored_path).unwrap();

            let restored_data = std::fs::read(&restored_path).unwrap();
            assert_eq!(
                db_data, restored_data,
                "Mismatch for page_size={}",
                page_size
            );
            assert_eq!(result.header.page_size.into_inner(), page_size);
        }
    }

    #[test]
    fn test_snapshot_preserves_binary_data() {
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("binary.db");
        let restored_path = dir.path().join("restored.db");

        let page_size = 4096u32;

        // Create database with all byte values (0x00-0xFF pattern)
        let mut db_data = Vec::new();
        for page_num in 0..4 {
            let mut page = vec![0u8; page_size as usize];
            for (i, byte) in page.iter_mut().enumerate() {
                *byte = ((page_num * 256 + i) % 256) as u8;
            }
            db_data.extend(page);
        }
        std::fs::write(&db_path, &db_data).unwrap();

        let mut ltx_buffer = Vec::new();
        encode_snapshot(&mut ltx_buffer, &db_path, page_size, 50).unwrap();

        let cursor = std::io::Cursor::new(ltx_buffer);
        decode_to_db(cursor, &restored_path).unwrap();

        let restored_data = std::fs::read(&restored_path).unwrap();

        // Verify every single byte
        for (i, (orig, rest)) in db_data.iter().zip(restored_data.iter()).enumerate() {
            assert_eq!(
                orig, rest,
                "Byte mismatch at offset {}: expected 0x{:02x}, got 0x{:02x}",
                i, orig, rest
            );
        }
    }

    #[test]
    fn test_incremental_ltx_encoding_with_checksum() {
        // Test encoding WAL changes as incremental LTX
        // Note: LTX format requires pre_apply_checksum for incremental files
        let page_size = 4096u32;

        // Simulate WAL changes: sequential pages (LTX requirement)
        let pages: Vec<(u32, Vec<u8>)> = vec![
            (1, vec![0xAA; page_size as usize]),
            (2, vec![0xBB; page_size as usize]),
            (3, vec![0xCC; page_size as usize]),
        ];

        // Pre-apply checksum is required for non-snapshot LTX files
        let pre_checksum = Checksum::new(0x123456789ABCDEF0);
        let expected_post = Checksum::new(0xFEDCBA9876543210);

        let mut ltx_buffer = Vec::new();
        let checksum = encode_wal_changes(
            &mut ltx_buffer,
            &pages,
            page_size,
            10,  // min_txid
            12,  // max_txid
            3,   // commit_page (db size in pages)
            Some(pre_checksum),
            expected_post,
        )
        .unwrap();

        // Verify we got the expected checksum
        assert_eq!(checksum.into_inner(), expected_post.into_inner());

        // Verify buffer is non-empty and reasonable size
        assert!(!ltx_buffer.is_empty());
        assert!(ltx_buffer.len() > 100); // At least header
    }

    #[test]
    fn test_incremental_ltx_format_rules() {
        // LTX format rules:
        // - min_txid=1 is a "snapshot" (no pre_checksum allowed)
        // - min_txid>1 is "incremental" (pre_checksum required)
        let page_size = 1024u32;
        let pre_checksum = Checksum::new(0x123456789ABCDEF0);

        // Incremental (min_txid > 1) requires pre_checksum
        let pages: Vec<(u32, Vec<u8>)> = vec![
            (1, vec![0x11; page_size as usize]),
            (2, vec![0x22; page_size as usize]),
        ];

        let expected_post = Checksum::new(0xABCDEF1234567890);

        let mut ltx_buffer = Vec::new();
        let result = encode_wal_changes(
            &mut ltx_buffer,
            &pages,
            page_size,
            10, // min_txid > 1 = incremental
            11, // max_txid
            2,
            Some(pre_checksum),
            expected_post,
        );
        assert!(
            result.is_ok(),
            "Incremental with pre_checksum should succeed: {:?}",
            result.err()
        );

        // Incremental without pre_checksum should fail
        let mut ltx_buffer2 = Vec::new();
        let result2 = encode_wal_changes(
            &mut ltx_buffer2,
            &pages,
            page_size,
            10, // min_txid > 1 = incremental
            11,
            2,
            None, // Missing pre_checksum!
            expected_post,
        );
        assert!(
            result2.is_err(),
            "Incremental without pre_checksum should fail"
        );

        // Snapshot (min_txid = 1) should not have pre_checksum
        let mut ltx_buffer3 = Vec::new();
        let result3 = encode_wal_changes(
            &mut ltx_buffer3,
            &pages,
            page_size,
            1, // min_txid = 1 = snapshot
            2,
            2,
            None, // No pre_checksum for snapshot
            expected_post,
        );
        assert!(
            result3.is_ok(),
            "Snapshot without pre_checksum should succeed: {:?}",
            result3.err()
        );
    }

    #[test]
    fn test_txid_ranges() {
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let restored_path = dir.path().join("restored.db");

        let page_size = 4096u32;
        let db_data = vec![0x42u8; page_size as usize];
        std::fs::write(&db_path, &db_data).unwrap();

        // Test various TXID values
        for txid in [1u64, 100, 1000, 999999, u32::MAX as u64] {
            let mut ltx_buffer = Vec::new();
            encode_snapshot(&mut ltx_buffer, &db_path, page_size, txid).unwrap();

            let cursor = std::io::Cursor::new(ltx_buffer);
            let result = decode_to_db(cursor, &restored_path).unwrap();

            assert_eq!(result.header.max_txid.into_inner(), txid);
        }
    }

    #[test]
    fn test_checksum_computation() {
        // Verify checksum is deterministic
        let data1 = b"hello world";
        let data2 = b"hello world";
        let data3 = b"hello worlD"; // Different

        let cs1 = compute_db_checksum(data1);
        let cs2 = compute_db_checksum(data2);
        let cs3 = compute_db_checksum(data3);

        assert_eq!(cs1.into_inner(), cs2.into_inner());
        assert_ne!(cs1.into_inner(), cs3.into_inner());
    }


    #[test]
    fn test_large_database() {
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("large.db");
        let restored_path = dir.path().join("restored.db");

        let page_size = 4096u32;
        let num_pages = 100; // 400KB database

        // Create large database with varying content
        let mut db_data = Vec::with_capacity(num_pages * page_size as usize);
        for i in 0..num_pages {
            let pattern = (i as u8).wrapping_mul(37);
            let mut page = vec![pattern; page_size as usize];
            // Mark page with its number
            let page_num_bytes = (i as u32).to_le_bytes();
            page[0..4].copy_from_slice(&page_num_bytes);
            db_data.extend(page);
        }
        std::fs::write(&db_path, &db_data).unwrap();

        let mut ltx_buffer = Vec::new();
        encode_snapshot(&mut ltx_buffer, &db_path, page_size, 1000).unwrap();

        // LTX should be compressed
        assert!(
            ltx_buffer.len() < db_data.len(),
            "LTX ({}) should be smaller than raw DB ({}) due to compression",
            ltx_buffer.len(),
            db_data.len()
        );

        let cursor = std::io::Cursor::new(ltx_buffer);
        decode_to_db(cursor, &restored_path).unwrap();

        let restored_data = std::fs::read(&restored_path).unwrap();
        assert_eq!(db_data, restored_data);
    }

    #[test]
    fn test_encode_to_memory_buffer() {
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let restored_path = dir.path().join("restored.db");

        let page_size = 4096u32;
        let db_data = vec![0x42u8; page_size as usize * 5];
        std::fs::write(&db_path, &db_data).unwrap();

        // Encode to Vec<u8> (common use case for S3 upload)
        let mut buffer: Vec<u8> = Vec::new();
        encode_snapshot(&mut buffer, &db_path, page_size, 1).unwrap();

        // Decode from Cursor (common use case for S3 download)
        let cursor = std::io::Cursor::new(buffer);
        decode_to_db(cursor, &restored_path).unwrap();

        let restored_data = std::fs::read(&restored_path).unwrap();
        assert_eq!(db_data, restored_data);
    }

    #[test]
    fn test_apply_ltx_in_place_basic() {
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        let page_size = 4096u32;
        let num_pages = 5;

        // Create initial database
        let db_data = vec![0x00u8; (page_size as usize) * num_pages];
        std::fs::write(&db_path, &db_data).unwrap();

        // Create incremental LTX that updates pages 2 and 4
        let pages: Vec<(u32, Vec<u8>)> = vec![
            (2, vec![0xAA; page_size as usize]),
            (4, vec![0xBB; page_size as usize]),
        ];

        let pre_checksum = compute_checksum_from_file(&db_path).unwrap();

        // Compute expected post_checksum after applying changes
        let expected_post = chain_checksum(pre_checksum, &pages);

        let mut ltx_buffer = Vec::new();
        encode_wal_changes(
            &mut ltx_buffer,
            &pages,
            page_size,
            2,  // min_txid
            3,  // max_txid
            num_pages as u32,
            Some(pre_checksum),
            expected_post,
        )
        .unwrap();

        // Apply in-place (verifies checksum chain)
        let cursor = std::io::Cursor::new(ltx_buffer);
        let result = apply_ltx_to_db(cursor, &db_path).unwrap();

        // Verify only changed pages were updated
        let result_data = std::fs::read(&db_path).unwrap();

        // Page 1 (index 0): unchanged
        assert_eq!(&result_data[0..page_size as usize], &vec![0x00u8; page_size as usize][..]);
        // Page 2 (index 1): updated to 0xAA
        let page2_start = page_size as usize;
        assert_eq!(&result_data[page2_start..page2_start + page_size as usize], &vec![0xAAu8; page_size as usize][..]);
        // Page 3 (index 2): unchanged
        let page3_start = 2 * page_size as usize;
        assert_eq!(&result_data[page3_start..page3_start + page_size as usize], &vec![0x00u8; page_size as usize][..]);
        // Page 4 (index 3): updated to 0xBB
        let page4_start = 3 * page_size as usize;
        assert_eq!(&result_data[page4_start..page4_start + page_size as usize], &vec![0xBBu8; page_size as usize][..]);
        // Page 5 (index 4): unchanged
        let page5_start = 4 * page_size as usize;
        assert_eq!(&result_data[page5_start..page5_start + page_size as usize], &vec![0x00u8; page_size as usize][..]);

        assert_eq!(result.header.min_txid.into_inner(), 2);
        assert_eq!(result.header.max_txid.into_inner(), 3);
    }

    #[test]
    fn test_apply_ltx_in_place_preserves_other_data() {
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        let page_size = 4096u32;

        // Create database with unique content per page
        let mut db_data = Vec::new();
        for i in 0..4u8 {
            db_data.extend(vec![i * 10; page_size as usize]);
        }
        std::fs::write(&db_path, &db_data).unwrap();

        // Update only page 3
        let pages: Vec<(u32, Vec<u8>)> = vec![
            (3, vec![0xFF; page_size as usize]),
        ];

        let pre_checksum = compute_checksum_from_file(&db_path).unwrap();

        // Compute expected post_checksum after applying changes
        let expected_post = chain_checksum(pre_checksum, &pages);

        let mut ltx_buffer = Vec::new();
        encode_wal_changes(&mut ltx_buffer, &pages, page_size, 10, 11, 4, Some(pre_checksum), expected_post).unwrap();

        let cursor = std::io::Cursor::new(ltx_buffer);
        apply_ltx_to_db(cursor, &db_path).unwrap();

        let result_data = std::fs::read(&db_path).unwrap();

        // Verify pages 1, 2, 4 unchanged
        assert_eq!(&result_data[0..page_size as usize], &vec![0u8; page_size as usize][..]);
        assert_eq!(&result_data[page_size as usize..2 * page_size as usize], &vec![10u8; page_size as usize][..]);
        // Page 3 updated
        assert_eq!(&result_data[2 * page_size as usize..3 * page_size as usize], &vec![0xFFu8; page_size as usize][..]);
        // Page 4 unchanged
        assert_eq!(&result_data[3 * page_size as usize..4 * page_size as usize], &vec![30u8; page_size as usize][..]);
    }

    #[test]
    fn test_compute_checksum_from_file() {
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        let data = vec![0x42u8; 4096];
        std::fs::write(&db_path, &data).unwrap();

        let checksum1 = compute_checksum_from_file(&db_path).unwrap();
        let checksum2 = compute_checksum_from_file(&db_path).unwrap();

        // Same file should produce same checksum
        assert_eq!(checksum1.into_inner(), checksum2.into_inner());

        // Different content should produce different checksum
        std::fs::write(&db_path, vec![0x43u8; 4096]).unwrap();
        let checksum3 = compute_checksum_from_file(&db_path).unwrap();
        assert_ne!(checksum1.into_inner(), checksum3.into_inner());
    }

    #[test]
    fn test_apply_ltx_chain_simulation() {
        // Simulate a realistic scenario: snapshot -> incremental -> incremental
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");

        let page_size = 4096u32;
        let num_pages = 3;

        // Initial database state
        let initial_data: Vec<u8> = (0..num_pages)
            .flat_map(|i| vec![(i as u8) * 10; page_size as usize])
            .collect();
        std::fs::write(&db_path, &initial_data).unwrap();

        // Snapshot (TXID 1)
        let mut snapshot_buffer = Vec::new();
        encode_snapshot(&mut snapshot_buffer, &db_path, page_size, 1).unwrap();

        // First incremental: update page 1 (TXID 2)
        let pre_checksum1 = compute_checksum_from_file(&db_path).unwrap();
        let pages1: Vec<(u32, Vec<u8>)> = vec![(1, vec![0xAA; page_size as usize])];
        let expected_post1 = chain_checksum(pre_checksum1, &pages1);

        let mut inc1_buffer = Vec::new();
        let post_checksum1 = encode_wal_changes(
            &mut inc1_buffer,
            &pages1,
            page_size,
            2, 2,
            num_pages as u32,
            Some(pre_checksum1),
            expected_post1,
        ).unwrap();

        // Apply first incremental
        let cursor1 = std::io::Cursor::new(inc1_buffer);
        let result1 = apply_ltx_to_db(cursor1, &db_path).unwrap();

        // Chain continues: post_checksum1 becomes pre_checksum2
        let pre_checksum2 = result1.post_apply_checksum;

        // Second incremental: update page 2 (TXID 3)
        let pages2: Vec<(u32, Vec<u8>)> = vec![(2, vec![0xBB; page_size as usize])];
        let expected_post2 = chain_checksum(pre_checksum2, &pages2);

        let mut inc2_buffer = Vec::new();
        encode_wal_changes(
            &mut inc2_buffer,
            &pages2,
            page_size,
            3, 3,
            num_pages as u32,
            Some(pre_checksum2),
            expected_post2,
        ).unwrap();

        // Apply second incremental
        let cursor2 = std::io::Cursor::new(inc2_buffer);
        apply_ltx_to_db(cursor2, &db_path).unwrap();

        // Final verification
        let final_data = std::fs::read(&db_path).unwrap();
        assert_eq!(&final_data[0..page_size as usize], &vec![0xAAu8; page_size as usize][..]); // Page 1 updated
        assert_eq!(&final_data[page_size as usize..2 * page_size as usize], &vec![0xBBu8; page_size as usize][..]); // Page 2 updated
        assert_eq!(&final_data[2 * page_size as usize..3 * page_size as usize], &vec![20u8; page_size as usize][..]); // Page 3 unchanged
    }

    // ============================================
    // Checksum Chain Error Tests
    // ============================================

    #[test]
    fn test_apply_ltx_post_checksum_mismatch_via_wrong_pre() {
        // With chained checksums, a wrong pre produces a self-consistent but different chain.
        // The LTX is internally consistent, so apply succeeds. Chain breaks are detected
        // at the file-linking level (previous file's post != this file's pre).
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let page_size = 4096u32;

        let initial_data = vec![0x00u8; page_size as usize * 3];
        std::fs::write(&db_path, &initial_data).unwrap();

        let pages: Vec<(u32, Vec<u8>)> = vec![(1, vec![0xAA; page_size as usize])];
        let wrong_pre_checksum = Checksum::new(0xDEADBEEF);
        let wrong_post = chain_checksum(wrong_pre_checksum, &pages);

        let mut ltx_buffer = Vec::new();
        encode_wal_changes(
            &mut ltx_buffer,
            &pages,
            page_size,
            2, 2,
            3,
            Some(wrong_pre_checksum),
            wrong_post,
        ).unwrap();

        // Self-consistent LTX applies successfully
        let cursor = std::io::Cursor::new(ltx_buffer);
        let result = apply_ltx_to_db(cursor, &db_path);
        assert!(result.is_ok(), "Self-consistent LTX should apply successfully");
    }

    #[test]
    fn test_apply_ltx_post_checksum_mismatch() {
        // Test that apply_ltx_to_db detects corruption when post_checksum doesn't match actual result
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let page_size = 4096u32;

        // Create initial database
        let initial_data = vec![0x00u8; page_size as usize * 3];
        std::fs::write(&db_path, &initial_data).unwrap();

        let pre_checksum = compute_checksum_from_file(&db_path).unwrap();

        // Create incremental with pages
        let pages: Vec<(u32, Vec<u8>)> = vec![(1, vec![0xAA; page_size as usize])];

        // Use WRONG post_checksum (not matching actual result)
        let wrong_post_checksum = Checksum::new(0xBADC0FFEE);

        let mut ltx_buffer = Vec::new();
        encode_wal_changes(
            &mut ltx_buffer,
            &pages,
            page_size,
            2, 2,
            3,
            Some(pre_checksum),
            wrong_post_checksum, // This is wrong!
        ).unwrap();

        // Applying should FAIL with post-apply checksum mismatch
        let cursor = std::io::Cursor::new(ltx_buffer);
        let result = apply_ltx_to_db(cursor, &db_path);

        assert!(result.is_err(), "Should detect post-apply checksum mismatch");
        let err_msg = result.unwrap_err().to_string();
        assert!(err_msg.contains("Post-apply checksum mismatch"), "Error should mention post-apply mismatch: {}", err_msg);
        assert!(err_msg.contains(&format!("{:016x}", wrong_post_checksum.into_inner())), "Error should show expected checksum");
    }

    #[test]
    fn test_apply_ltx_out_of_order() {
        // With chained checksums, apply_ltx_to_db only verifies internal consistency
        // (chain_checksum(pre, pages) == post). A self-consistent LTX applies successfully
        // even to the "wrong" DB state. Out-of-order detection is the caller's job:
        // each file's pre_checksum must equal the previous file's post_checksum.
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let page_size = 4096u32;

        let initial_data = vec![0x00u8; page_size as usize * 3];
        std::fs::write(&db_path, &initial_data).unwrap();

        let checksum0 = compute_checksum_from_file(&db_path).unwrap();

        // Create three chained incrementals
        let pages1: Vec<(u32, Vec<u8>)> = vec![(1, vec![0xAA; page_size as usize])];
        let post1 = chain_checksum(checksum0, &pages1);
        let mut buf1 = Vec::new();
        encode_wal_changes(&mut buf1, &pages1, page_size, 2, 2, 3, Some(checksum0), post1).unwrap();

        let pages2: Vec<(u32, Vec<u8>)> = vec![(2, vec![0xBB; page_size as usize])];
        let post2 = chain_checksum(post1, &pages2);
        let mut buf2 = Vec::new();
        encode_wal_changes(&mut buf2, &pages2, page_size, 3, 3, 3, Some(post1), post2).unwrap();

        let pages3: Vec<(u32, Vec<u8>)> = vec![(3, vec![0xCC; page_size as usize])];
        let post3 = chain_checksum(post2, &pages3);
        let mut buf3 = Vec::new();
        encode_wal_changes(&mut buf3, &pages3, page_size, 4, 4, 3, Some(post2), post3).unwrap();

        // Apply 1st only, then skip 2nd and apply 3rd
        apply_ltx_to_db(std::io::Cursor::new(&buf1), &db_path).unwrap();
        let result1 = apply_ltx_to_db(std::io::Cursor::new(&buf1), &db_path).unwrap();

        // 3rd LTX applies successfully (internally self-consistent)
        let result3 = apply_ltx_to_db(std::io::Cursor::new(&buf3), &db_path).unwrap();

        // But the chain is broken: result1.post != buf3's pre (which is post2)
        let buf3_pre = result3.header.pre_apply_checksum.unwrap();
        assert_ne!(
            result1.post_apply_checksum, buf3_pre,
            "Chain should be broken when skipping an incremental"
        );

        // Correct chain: apply all three in order
        std::fs::write(&db_path, &initial_data).unwrap();
        let r1 = apply_ltx_to_db(std::io::Cursor::new(&buf1), &db_path).unwrap();
        let r2 = apply_ltx_to_db(std::io::Cursor::new(&buf2), &db_path).unwrap();
        let r3 = apply_ltx_to_db(std::io::Cursor::new(&buf3), &db_path).unwrap();

        // Chain links match
        assert_eq!(r1.post_apply_checksum.into_inner(), r2.header.pre_apply_checksum.unwrap().into_inner());
        assert_eq!(r2.post_apply_checksum.into_inner(), r3.header.pre_apply_checksum.unwrap().into_inner());
    }

    #[test]
    fn test_decode_to_db_post_checksum_verification() {
        // Test that decode_to_db verifies post_checksum matches actual restored file
        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let restored_path = dir.path().join("restored.db");
        let page_size = 4096u32;

        // Create source database with varied content per page
        let mut db_data = Vec::new();
        for i in 0..3u8 {
            db_data.extend(vec![i * 42; page_size as usize]);
        }
        std::fs::write(&db_path, &db_data).unwrap();

        let expected_checksum = compute_checksum_from_file(&db_path).unwrap();

        // Create snapshot
        let mut snapshot_buffer = Vec::new();
        encode_snapshot(&mut snapshot_buffer, &db_path, page_size, 1).unwrap();

        // Decode successfully
        let cursor = std::io::Cursor::new(&snapshot_buffer);
        let result = decode_to_db(cursor, &restored_path);
        assert!(result.is_ok(), "Valid snapshot should decode successfully");

        // Verify the post_apply_checksum in result matches the actual restored file
        let decoded_result = result.unwrap();
        let actual_checksum = compute_checksum_from_file(&restored_path).unwrap();

        assert_eq!(
            decoded_result.post_apply_checksum.into_inner(),
            actual_checksum.into_inner(),
            "post_apply_checksum in result should match actual restored file checksum"
        );

        assert_eq!(
            actual_checksum.into_inner(),
            expected_checksum.into_inner(),
            "Restored file should match original"
        );

        // Verify restored content byte-for-byte
        let restored_data = std::fs::read(&restored_path).unwrap();
        assert_eq!(restored_data, db_data, "Restored data should match original exactly");
    }

    // ============================================
    // NO_CHECKSUM Flag Tests (Litestream Compatibility)
    // ============================================

    #[test]
    fn test_no_checksum_flag_decode() {
        // Test that files with NO_CHECKSUM flag skip checksum verification
        use litepages::Encoder;

        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let restored_path = dir.path().join("restored.db");
        let page_size = 4096u32;

        // Create source database
        let db_data = vec![0x42u8; page_size as usize * 3];
        std::fs::write(&db_path, &db_data).unwrap();

        // Create LTX file with NO_CHECKSUM flag (litestream format)
        let header = Header {
            flags: HeaderFlags::NO_CHECKSUM | HeaderFlags::COMPRESS_LZ4,
            page_size: PageSize::new(page_size).unwrap(),
            commit: PageNum::new(3).unwrap(),
            min_txid: TXID::ONE,
            max_txid: TXID::ONE,
            timestamp: SystemTime::now(),
            pre_apply_checksum: None,
        };

        let mut ltx_buffer = Vec::new();
        let mut encoder = Encoder::new(&mut ltx_buffer, &header).unwrap();

        // Encode all 3 pages
        for i in 0..3u32 {
            encoder.encode_page(PageNum::new(i + 1).unwrap(), &db_data[(i as usize * page_size as usize)..(i as usize + 1) * page_size as usize]).unwrap();
        }

        // Use zero checksum (litestream doesn't track checksums)
        encoder.finish(Checksum::new(0)).unwrap();

        // Decode should succeed even though checksum is zero
        let cursor = std::io::Cursor::new(&ltx_buffer);
        let result = decode_to_db(cursor, &restored_path);

        assert!(result.is_ok(), "Should decode successfully with NO_CHECKSUM flag");
        let decode_result = result.unwrap();

        // Walrust computes checksums internally even when NO_CHECKSUM is set (for tracking)
        // But it doesn't verify them against the LTX file's checksums
        assert!(decode_result.post_apply_checksum.into_inner() != 0, "Should compute actual checksum even with NO_CHECKSUM");

        // Verify data was restored correctly
        let restored_data = std::fs::read(&restored_path).unwrap();
        assert_eq!(restored_data, db_data, "Data should be restored correctly even with NO_CHECKSUM");

        // Verify the checksum matches the actual data
        let expected_checksum = compute_db_checksum(&db_data);
        assert_eq!(decode_result.post_apply_checksum.into_inner(), expected_checksum.into_inner());
    }

    #[test]
    fn test_no_checksum_flag_apply() {
        // Test that incremental LTX with NO_CHECKSUM flag skips checksum verification
        use litepages::Encoder;

        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let page_size = 4096u32;

        // Create initial database (3 pages of zeros)
        let initial_data = vec![0x00u8; page_size as usize * 3];
        std::fs::write(&db_path, &initial_data).unwrap();

        // Create incremental LTX with NO_CHECKSUM flag (like litestream)
        // This would normally require pre_apply_checksum, but NO_CHECKSUM skips that
        let header = Header {
            flags: HeaderFlags::NO_CHECKSUM | HeaderFlags::COMPRESS_LZ4,
            page_size: PageSize::new(page_size).unwrap(),
            commit: PageNum::new(3).unwrap(),
            min_txid: TXID::new(2).unwrap(), // Incremental (not snapshot)
            max_txid: TXID::new(2).unwrap(),
            timestamp: SystemTime::now(),
            pre_apply_checksum: Some(Checksum::new(0)), // Zero checksum (litestream doesn't track)
        };

        let mut ltx_buffer = Vec::new();
        let mut encoder = Encoder::new(&mut ltx_buffer, &header).unwrap();

        // Modify page 2
        let modified_page = vec![0xAAu8; page_size as usize];
        encoder.encode_page(PageNum::new(2).unwrap(), &modified_page).unwrap();

        // Use zero checksum
        encoder.finish(Checksum::new(0)).unwrap();

        // Apply should succeed even with zero/wrong checksums
        let cursor = std::io::Cursor::new(&ltx_buffer);
        let result = apply_ltx_to_db(cursor, &db_path);

        assert!(result.is_ok(), "Should apply successfully with NO_CHECKSUM flag");
        let apply_result = result.unwrap();

        // Walrust computes checksums internally even when NO_CHECKSUM is set (for tracking)
        // But it doesn't verify them against the LTX file's checksums
        assert!(apply_result.post_apply_checksum.into_inner() != 0, "Should compute actual checksum even with NO_CHECKSUM");

        // Verify page 2 was modified
        let result_data = std::fs::read(&db_path).unwrap();
        assert_eq!(&result_data[page_size as usize..2 * page_size as usize], &modified_page[..]);
        // Verify other pages unchanged
        assert_eq!(&result_data[0..page_size as usize], &vec![0x00u8; page_size as usize][..]);
        assert_eq!(&result_data[2 * page_size as usize..3 * page_size as usize], &vec![0x00u8; page_size as usize][..]);

        // With NO_CHECKSUM and a zero pre_checksum, the chain checksum is computed
        assert!(apply_result.post_apply_checksum.into_inner() != 0, "Should compute a chain checksum");
    }

    #[test]
    fn test_no_checksum_flag_skips_verification() {
        // Verify that NO_CHECKSUM truly skips verification by using intentionally wrong checksums
        use litepages::Encoder;

        let dir = tempdir().unwrap();
        let db_path = dir.path().join("test.db");
        let page_size = 4096u32;

        // Create database
        let db_data = vec![0x42u8; page_size as usize];
        std::fs::write(&db_path, &db_data).unwrap();

        // Create incremental with NO_CHECKSUM and WRONG pre_checksum
        // This should still succeed because NO_CHECKSUM skips verification
        let actual_checksum = compute_checksum_from_file(&db_path).unwrap();
        let wrong_checksum = Checksum::new(0xDEADBEEF); // Intentionally wrong

        assert_ne!(wrong_checksum.into_inner(), actual_checksum.into_inner(), "Checksums should be different");

        let header = Header {
            flags: HeaderFlags::NO_CHECKSUM | HeaderFlags::COMPRESS_LZ4,
            page_size: PageSize::new(page_size).unwrap(),
            commit: PageNum::new(1).unwrap(),
            min_txid: TXID::new(2).unwrap(),
            max_txid: TXID::new(2).unwrap(),
            timestamp: SystemTime::now(),
            pre_apply_checksum: Some(wrong_checksum), // WRONG on purpose!
        };

        let mut ltx_buffer = Vec::new();
        let mut encoder = Encoder::new(&mut ltx_buffer, &header).unwrap();
        encoder.encode_page(PageNum::new(1).unwrap(), &vec![0x99u8; page_size as usize]).unwrap();
        encoder.finish(Checksum::new(0xBADC0FFEE)).unwrap(); // Also wrong!

        // Should succeed because NO_CHECKSUM skips all verification
        let cursor = std::io::Cursor::new(&ltx_buffer);
        let result = apply_ltx_to_db(cursor, &db_path);

        assert!(result.is_ok(), "Should succeed with wrong checksums when NO_CHECKSUM is set");
    }
}