tylertoo-core 0.7.0

//! PMTiles v3 writer implementation.
//!
//! Implements the PMTiles v3 spec: https://github.com/protomaps/PMTiles/blob/main/spec/v3/spec.md
//!
//! Key design decisions:
//! - Uses Hilbert curve ordering for tile IDs (spatial locality)
//! - Delta-encoded directories for better compression
//! - Configurable compression (gzip, brotli, zstd) for both directories and tiles
//! - Clustered mode for efficient sequential reads

use crate::compression::{self, Compression};
use crate::dedup::{DeduplicationCache, DeduplicationStats, TileHasher};
use crate::tile::TileBounds;
use crate::{Error, Result};
use std::collections::{BTreeMap, HashMap};
use std::fs::File;
use std::io::{BufWriter, Write};
use std::path::Path;

/// PMTiles v3 magic number
const PMTILES_MAGIC: &[u8; 7] = b"PMTiles";
const PMTILES_VERSION: u8 = 3;

/// Tile type enumeration (byte 99 in header)
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum TileType {
    Unknown = 0,
    Mvt = 1,
    Png = 2,
    Jpeg = 3,
    Webp = 4,
    Avif = 5,
}

// Compression enum is now imported from crate::compression

/// PMTiles v3 header (127 bytes)
///
/// Layout follows the spec exactly:
/// - Bytes 0-6: Magic "PMTiles"
/// - Byte 7: Version (3)
/// - Bytes 8-95: Offsets and lengths (8 u64s)
/// - Bytes 96-99: Flags (clustered, compression, type)
/// - Bytes 100-101: Zoom levels
/// - Bytes 102-117: Bounds (min_lon, min_lat, max_lon, max_lat as i32 * 10_000_000)
/// - Bytes 118-126: Center (zoom, lon, lat)
#[derive(Debug, Clone)]
pub struct Header {
    pub root_dir_offset: u64,
    pub root_dir_length: u64,
    pub json_metadata_offset: u64,
    pub json_metadata_length: u64,
    pub leaf_dirs_offset: u64,
    pub leaf_dirs_length: u64,
    pub tile_data_offset: u64,
    pub tile_data_length: u64,
    pub addressed_tiles_count: u64,
    pub tile_entries_count: u64,
    pub tile_contents_count: u64,
    pub clustered: bool,
    pub internal_compression: Compression,
    pub tile_compression: Compression,
    pub tile_type: TileType,
    pub min_zoom: u8,
    pub max_zoom: u8,
    pub min_lon: f64,
    pub min_lat: f64,
    pub max_lon: f64,
    pub max_lat: f64,
    pub center_zoom: u8,
    pub center_lon: f64,
    pub center_lat: f64,
}

impl Default for Header {
    fn default() -> Self {
        Self {
            root_dir_offset: 127, // Immediately after header
            root_dir_length: 0,
            json_metadata_offset: 0,
            json_metadata_length: 0,
            leaf_dirs_offset: 0,
            leaf_dirs_length: 0,
            tile_data_offset: 0,
            tile_data_length: 0,
            addressed_tiles_count: 0,
            tile_entries_count: 0,
            tile_contents_count: 0,
            clustered: true,
            internal_compression: Compression::Gzip,
            tile_compression: Compression::Gzip,
            tile_type: TileType::Mvt,
            min_zoom: 0,
            max_zoom: 14,
            min_lon: -180.0,
            min_lat: -85.0,
            max_lon: 180.0,
            max_lat: 85.0,
            center_zoom: 0,
            center_lon: 0.0,
            center_lat: 0.0,
        }
    }
}

impl Header {
    /// Serialize header to exactly 127 bytes
    ///
    /// Position encoding follows the spec: multiply by 10,000,000 and store as i32 LE
    pub fn to_bytes(&self) -> [u8; 127] {
        let mut buf = [0u8; 127];

        // Magic (7 bytes) + Version (1 byte)
        buf[0..7].copy_from_slice(PMTILES_MAGIC);
        buf[7] = PMTILES_VERSION;

        // Offsets and lengths (8 bytes each, little-endian)
        buf[8..16].copy_from_slice(&self.root_dir_offset.to_le_bytes());
        buf[16..24].copy_from_slice(&self.root_dir_length.to_le_bytes());
        buf[24..32].copy_from_slice(&self.json_metadata_offset.to_le_bytes());
        buf[32..40].copy_from_slice(&self.json_metadata_length.to_le_bytes());
        buf[40..48].copy_from_slice(&self.leaf_dirs_offset.to_le_bytes());
        buf[48..56].copy_from_slice(&self.leaf_dirs_length.to_le_bytes());
        buf[56..64].copy_from_slice(&self.tile_data_offset.to_le_bytes());
        buf[64..72].copy_from_slice(&self.tile_data_length.to_le_bytes());

        // Tile counts
        buf[72..80].copy_from_slice(&self.addressed_tiles_count.to_le_bytes());
        buf[80..88].copy_from_slice(&self.tile_entries_count.to_le_bytes());
        buf[88..96].copy_from_slice(&self.tile_contents_count.to_le_bytes());

        // Clustered flag
        buf[96] = if self.clustered { 1 } else { 0 };

        // Compression and type
        buf[97] = self.internal_compression as u8;
        buf[98] = self.tile_compression as u8;
        buf[99] = self.tile_type as u8;

        // Zoom levels
        buf[100] = self.min_zoom;
        buf[101] = self.max_zoom;

        // Bounds: lon/lat as i32 * 10,000,000 (spec-compliant encoding)
        let encode_coord = |v: f64| -> [u8; 4] { ((v * 10_000_000.0) as i32).to_le_bytes() };

        buf[102..106].copy_from_slice(&encode_coord(self.min_lon));
        buf[106..110].copy_from_slice(&encode_coord(self.min_lat));
        buf[110..114].copy_from_slice(&encode_coord(self.max_lon));
        buf[114..118].copy_from_slice(&encode_coord(self.max_lat));

        // Center: zoom + lon/lat
        buf[118] = self.center_zoom;
        buf[119..123].copy_from_slice(&encode_coord(self.center_lon));
        buf[123..127].copy_from_slice(&encode_coord(self.center_lat));

        buf
    }
}

impl TileType {
    /// Parse a PMTiles spec byte code back into a tile type (byte 99).
    ///
    /// Returns `None` for codes outside the PMTiles v3 spec (0-5).
    pub fn from_code(code: u8) -> Option<Self> {
        match code {
            0 => Some(TileType::Unknown),
            1 => Some(TileType::Mvt),
            2 => Some(TileType::Png),
            3 => Some(TileType::Jpeg),
            4 => Some(TileType::Webp),
            5 => Some(TileType::Avif),
            _ => None,
        }
    }
}

impl Header {
    /// Parse a PMTiles v3 header from the first 127 bytes of an archive.
    ///
    /// Inverse of [`Header::to_bytes`]; the read side of the pipeline (issue
    /// #112) uses this to locate directories and tile data. Fails on short
    /// input, bad magic, unsupported version, or out-of-spec compression /
    /// tile-type codes.
    pub fn from_bytes(bytes: &[u8]) -> Result<Header> {
        let err = |msg: String| Error::PMTilesRead(msg);
        if bytes.len() < 127 {
            return Err(err(format!(
                "file too short for PMTiles header: {} bytes (need 127)",
                bytes.len()
            )));
        }
        if &bytes[0..7] != PMTILES_MAGIC {
            return Err(err("bad magic: not a PMTiles archive".to_string()));
        }
        if bytes[7] != PMTILES_VERSION {
            return Err(err(format!(
                "unsupported PMTiles version {} (only v3 is supported)",
                bytes[7]
            )));
        }

        let read_u64 =
            |at: usize| u64::from_le_bytes(bytes[at..at + 8].try_into().expect("8-byte slice"));
        let read_coord = |at: usize| {
            f64::from(i32::from_le_bytes(
                bytes[at..at + 4].try_into().expect("4-byte slice"),
            )) / 10_000_000.0
        };

        let internal_compression = Compression::from_code(bytes[97])
            .ok_or_else(|| err(format!("invalid internal compression code {}", bytes[97])))?;
        let tile_compression = Compression::from_code(bytes[98])
            .ok_or_else(|| err(format!("invalid tile compression code {}", bytes[98])))?;
        let tile_type = TileType::from_code(bytes[99])
            .ok_or_else(|| err(format!("invalid tile type code {}", bytes[99])))?;

        Ok(Header {
            root_dir_offset: read_u64(8),
            root_dir_length: read_u64(16),
            json_metadata_offset: read_u64(24),
            json_metadata_length: read_u64(32),
            leaf_dirs_offset: read_u64(40),
            leaf_dirs_length: read_u64(48),
            tile_data_offset: read_u64(56),
            tile_data_length: read_u64(64),
            addressed_tiles_count: read_u64(72),
            tile_entries_count: read_u64(80),
            tile_contents_count: read_u64(88),
            clustered: bytes[96] == 1,
            internal_compression,
            tile_compression,
            tile_type,
            min_zoom: bytes[100],
            max_zoom: bytes[101],
            min_lon: read_coord(102),
            min_lat: read_coord(106),
            max_lon: read_coord(110),
            max_lat: read_coord(114),
            center_zoom: bytes[118],
            center_lon: read_coord(119),
            center_lat: read_coord(123),
        })
    }
}

/// Convert tile coordinates (z, x, y) to a TileID for PMTiles
///
/// Uses Hilbert curve ordering for spatial locality. The tile ID is a cumulative
/// position on the series of Hilbert curves starting at zoom level 0.
///
/// Examples from spec:
/// - Z=0, X=0, Y=0 → TileID=0
/// - Z=1, X=0, Y=0 → TileID=1
/// - Z=1, X=0, Y=1 → TileID=2
/// - Z=1, X=1, Y=1 → TileID=3
/// - Z=1, X=1, Y=0 → TileID=4
/// - Z=2, X=0, Y=0 → TileID=5
pub fn tile_id(z: u8, x: u32, y: u32) -> u64 {
    if z == 0 {
        return 0;
    }

    // Calculate base ID: sum of all tiles in previous zoom levels
    // At zoom z, there are 4^z tiles. Base for zoom z is sum of 4^i for i in 1..z
    let base_id: u64 = (1..z as u64).map(|i| 4u64.pow(i as u32)).sum();
    let hilbert_idx = xy_to_hilbert(z, x, y);
    base_id + hilbert_idx + 1
}

/// Convert x,y coordinates to Hilbert curve index at zoom level z
///
/// Implementation follows the standard Hilbert curve algorithm:
/// https://en.wikipedia.org/wiki/Hilbert_curve
fn xy_to_hilbert(z: u8, x: u32, y: u32) -> u64 {
    let n = 1u32 << z;
    let mut rx: u32;
    let mut ry: u32;
    let mut s: u32;
    let mut d: u64 = 0;
    let mut x = x;
    let mut y = y;

    s = n / 2;
    while s > 0 {
        rx = if (x & s) > 0 { 1 } else { 0 };
        ry = if (y & s) > 0 { 1 } else { 0 };
        d += (s as u64) * (s as u64) * ((3 * rx) ^ ry) as u64;

        // Rotate quadrant - use n-1 (full grid size - 1) not s-1
        if ry == 0 {
            if rx == 1 {
                x = n - 1 - x;
                y = n - 1 - y;
            }
            std::mem::swap(&mut x, &mut y);
        }
        s /= 2;
    }
    d
}

/// Convert a PMTiles TileID back to tile coordinates (z, x, y).
///
/// Inverse of [`tile_id`]. Supports zoom levels 0-31 (the range a u64
/// cumulative Hilbert ID can address); returns an error for IDs beyond z31.
pub fn tile_id_to_zxy(id: u64) -> Result<(u8, u32, u32)> {
    let mut acc: u64 = 0;
    for z in 0u8..=31 {
        let num = 1u64 << (2 * u64::from(z));
        if id - acc < num {
            let (x, y) = hilbert_d2xy(z, id - acc);
            return Ok((z, x, y));
        }
        acc += num;
    }
    Err(Error::PMTilesRead(format!(
        "tile id {id} exceeds the zoom 31 address space"
    )))
}

/// Convert a Hilbert curve index back to x,y coordinates at zoom level z
///
/// Standard inverse Hilbert algorithm (d2xy), mirroring [`xy_to_hilbert`]:
/// https://en.wikipedia.org/wiki/Hilbert_curve
fn hilbert_d2xy(z: u8, d: u64) -> (u32, u32) {
    let n = 1u64 << z;
    let (mut x, mut y) = (0u64, 0u64);
    let mut t = d;
    let mut s = 1u64;
    while s < n {
        let rx = 1 & (t / 2);
        let ry = 1 & (t ^ rx);
        // Rotate quadrant - the inverse uses s-1 (current block size - 1),
        // where the forward transform uses n-1 (see xy_to_hilbert).
        if ry == 0 {
            if rx == 1 {
                x = s - 1 - x;
                y = s - 1 - y;
            }
            std::mem::swap(&mut x, &mut y);
        }
        x += s * rx;
        y += s * ry;
        t /= 4;
        s *= 2;
    }
    (x as u32, y as u32)
}

// ============================================================================
// Task 8: Directory Encoding
// ============================================================================

/// A directory entry pointing to tile data
///
/// In PMTiles, directories are columnar: all tile_ids are stored together,
/// then all run_lengths, then all lengths, then all offsets.
#[derive(Debug, Clone)]
pub struct DirEntry {
    pub tile_id: u64,
    pub offset: u64,
    pub length: u32,
    pub run_length: u32, // Number of consecutive tiles with same data (0 = leaf directory)
}

/// Encode a u64 as a varint (protobuf-style, little-endian)
///
/// Each byte uses 7 bits for data, MSB indicates continuation.
pub fn encode_varint(mut value: u64, buf: &mut Vec<u8>) {
    while value >= 0x80 {
        buf.push((value as u8) | 0x80);
        value >>= 7;
    }
    buf.push(value as u8);
}

/// Decode a varint from bytes
///
/// Returns (value, bytes_consumed) or None if invalid/incomplete.
pub fn decode_varint(data: &[u8]) -> Option<(u64, usize)> {
    let mut result: u64 = 0;
    let mut shift = 0;
    for (i, &byte) in data.iter().enumerate() {
        result |= ((byte & 0x7f) as u64) << shift;
        if byte & 0x80 == 0 {
            return Some((result, i + 1));
        }
        shift += 7;
        if shift >= 64 {
            return None; // Overflow
        }
    }
    None
}

/// Encode directory entries in PMTiles columnar format with delta encoding
///
/// Format: count, delta_tile_ids[], run_lengths[], lengths[], offsets[]
/// All values are varints. Tile IDs use simple delta encoding.
///
/// Offset encoding follows the PMTiles v3 spec:
/// - If offset equals expected position (contiguous), encode as 0
/// - Otherwise, encode as offset + 1
///
/// This allows efficient representation of contiguous tile data (common case).
pub fn encode_directory(entries: &[DirEntry]) -> Vec<u8> {
    let mut buf = Vec::new();

    // Number of entries
    encode_varint(entries.len() as u64, &mut buf);

    if entries.is_empty() {
        return buf;
    }

    // Delta-encoded tile IDs
    let mut last_id = 0u64;
    for entry in entries {
        encode_varint(entry.tile_id - last_id, &mut buf);
        last_id = entry.tile_id;
    }

    // Run lengths
    for entry in entries {
        encode_varint(entry.run_length as u64, &mut buf);
    }

    // Lengths
    for entry in entries {
        encode_varint(entry.length as u64, &mut buf);
    }

    // Offset encoding per PMTiles v3 spec:
    // - For contiguous entries (offset == expected_offset): encode 0
    // - Otherwise: encode offset + 1
    let mut expected_offset = 0u64;
    for (i, entry) in entries.iter().enumerate() {
        let is_contiguous = i > 0 && entry.offset == expected_offset;
        if is_contiguous {
            encode_varint(0, &mut buf);
        } else {
            encode_varint(entry.offset + 1, &mut buf);
        }

        // Update expected offset for next entry. The spec's contiguity rule
        // applies to every entry, leaf pointers (run_length == 0) included:
        // their `length` is the compressed leaf size and leaves are laid out
        // back to back in the leaf section (#377).
        expected_offset = entry.offset + entry.length as u64;
    }

    buf
}

/// Decode directory entries from PMTiles columnar format
///
/// This is the inverse of encode_directory, used for reading and testing.
pub fn decode_directory(data: &[u8]) -> Option<Vec<DirEntry>> {
    let mut offset = 0;

    // Number of entries
    let (count, consumed) = decode_varint(&data[offset..])?;
    offset += consumed;
    let count = count as usize;

    if count == 0 {
        return Some(Vec::new());
    }

    let mut entries = Vec::with_capacity(count);

    // Decode delta-encoded tile IDs
    let mut last_id = 0u64;
    for _ in 0..count {
        let (delta, consumed) = decode_varint(&data[offset..])?;
        offset += consumed;
        last_id += delta;
        entries.push(DirEntry {
            tile_id: last_id,
            offset: 0,
            length: 0,
            run_length: 0,
        });
    }

    // Decode run lengths
    for entry in entries.iter_mut() {
        let (run_length, consumed) = decode_varint(&data[offset..])?;
        offset += consumed;
        entry.run_length = run_length as u32;
    }

    // Decode lengths
    for entry in entries.iter_mut() {
        let (length, consumed) = decode_varint(&data[offset..])?;
        offset += consumed;
        entry.length = length as u32;
    }

    // Decode offsets (with contiguous encoding)
    let mut expected_offset = 0u64;
    for (i, entry) in entries.iter_mut().enumerate() {
        let (encoded_offset, consumed) = decode_varint(&data[offset..])?;
        offset += consumed;

        if encoded_offset == 0 && i > 0 {
            // Contiguous: use expected offset
            entry.offset = expected_offset;
        } else {
            // Explicit offset (stored as offset + 1)
            entry.offset = encoded_offset.saturating_sub(1);
        }

        // Update expected offset for next entry — for leaf pointers too.
        // tippecanoe and go-pmtiles encode every leaf after the first as
        // contiguous; gating this on run_length > 0 resolved them all to
        // offset 0 and `decode` failed with "incomplete deflate stream" (#377).
        // Both operands come from the archive, so the sum is checked.
        expected_offset = entry.offset.checked_add(u64::from(entry.length))?;
    }

    Some(entries)
}

// ============================================================================
// Leaf Directory Support (Issue #88)
// ============================================================================

/// Maximum size for root directory to fit in initial 16KB HTTP range request.
/// PMTiles header is 127 bytes, leaving 16384 - 127 = 16257 bytes for root directory.
const MAX_ROOT_DIR_BYTES: usize = 16384 - 127;

/// Initial leaf size when partitioning entries (matches tippecanoe)
const INITIAL_LEAF_SIZE: usize = 4096;

/// Result of building root and leaf directories
#[derive(Debug)]
pub struct DirectoryLayout {
    /// Compressed root directory (may contain leaf pointers or direct tile entries)
    pub root_bytes: Vec<u8>,
    /// Compressed leaf directories concatenated (empty if no leaves needed)
    pub leaves_bytes: Vec<u8>,
    /// Number of leaf directories (0 if all entries fit in root)
    pub num_leaves: usize,
}

/// Build leaf directories by partitioning entries into chunks.
///
/// Each chunk becomes a leaf directory. The root directory contains
/// pointers to these leaves (entries with run_length=0).
///
/// # Arguments
/// * `entries` - All tile directory entries
/// * `leaf_size` - Number of entries per leaf directory
/// * `compression` - Compression algorithm to use
fn build_root_leaves(
    entries: &[DirEntry],
    leaf_size: usize,
    compression: Compression,
) -> std::io::Result<DirectoryLayout> {
    let mut root_entries = Vec::new();
    let mut leaves_bytes = Vec::new();
    let mut num_leaves = 0;

    // Partition entries into leaf directories
    for chunk in entries.chunks(leaf_size) {
        num_leaves += 1;

        // Serialize and compress this leaf
        let leaf_encoded = encode_directory(chunk);
        let leaf_compressed = compression::compress(&leaf_encoded, compression)?;

        // Root entry points to this leaf:
        // - tile_id = first tile ID in this leaf
        // - offset = position within leaves_bytes
        // - length = size of compressed leaf
        // - run_length = 0 (indicates leaf pointer, not tile entry)
        root_entries.push(DirEntry {
            tile_id: chunk[0].tile_id,
            offset: leaves_bytes.len() as u64,
            length: leaf_compressed.len() as u32,
            run_length: 0, // CRITICAL: 0 means this is a leaf pointer
        });

        leaves_bytes.extend(leaf_compressed);
    }

    // Serialize and compress root directory
    let root_encoded = encode_directory(&root_entries);
    let root_compressed = compression::compress(&root_encoded, compression)?;

    Ok(DirectoryLayout {
        root_bytes: root_compressed,
        leaves_bytes,
        num_leaves,
    })
}

/// Create optimized directory structure, using leaf directories if needed.
///
/// Follows the tippecanoe algorithm:
/// 1. Try to fit all entries in a single root directory
/// 2. If root exceeds MAX_ROOT_DIR_BYTES, partition into leaf directories
/// 3. If root still exceeds limit, double leaf_size and retry
///
/// This ensures the root directory always fits in the initial HTTP range request,
/// which is critical for pmtiles-js and other clients that fetch 16KB initially.
///
/// # Arguments
/// * `entries` - All tile directory entries (must be sorted by tile_id)
/// * `compression` - Compression algorithm to use
pub fn make_root_leaves(
    entries: &[DirEntry],
    compression: Compression,
) -> std::io::Result<DirectoryLayout> {
    // Try single directory first (no leaves)
    let single_encoded = encode_directory(entries);
    let single_compressed = compression::compress(&single_encoded, compression)?;

    if single_compressed.len() <= MAX_ROOT_DIR_BYTES {
        // Fits in root - no leaf directories needed
        return Ok(DirectoryLayout {
            root_bytes: single_compressed,
            leaves_bytes: Vec::new(),
            num_leaves: 0,
        });
    }

    // Need leaf directories - iterate with increasing leaf_size until root fits
    let mut leaf_size = INITIAL_LEAF_SIZE;

    loop {
        let layout = build_root_leaves(entries, leaf_size, compression)?;

        if layout.root_bytes.len() <= MAX_ROOT_DIR_BYTES {
            return Ok(layout);
        }

        // Root still too big - double leaf_size (fewer, larger leaves = smaller root)
        leaf_size *= 2;

        // Safety check: if leaf_size exceeds entry count, something is wrong
        if leaf_size > entries.len() * 2 {
            // Fall back to single leaf containing everything
            // (This shouldn't happen in practice)
            return build_root_leaves(entries, entries.len(), compression);
        }
    }
}

/// Compress data with gzip (backward compatibility wrapper)
pub fn gzip_compress(data: &[u8]) -> std::io::Result<Vec<u8>> {
    compression::compress(data, Compression::Gzip)
}

// ============================================================================
// Task 9: Full PMTiles Writer
// ============================================================================

/// Render the TileJSON `fields` object for a layer.
///
/// Shared by `PmtilesWriter` and `StreamingPmtilesWriter`: both emit the same
/// metadata, and a second copy of this drifted once already. Field names are
/// sorted so the output is byte-for-byte deterministic.
fn fields_json(fields: &HashMap<String, String>) -> String {
    if fields.is_empty() {
        return "{}".to_string();
    }

    let mut field_pairs: Vec<_> = fields.iter().collect();
    field_pairs.sort_by_key(|(k, _)| *k);

    let field_strings: Vec<String> = field_pairs
        .iter()
        .map(|(name, type_str)| format!(r#""{}":"{}""#, name, type_str))
        .collect();

    format!("{{{}}}", field_strings.join(","))
}

/// Render the TileJSON `tilestats` fragment, including its trailing comma.
///
/// Returns an empty string when the archive holds no features, which keeps the
/// surrounding metadata object valid. Shared with `fields_json` above.
fn tilestats_json(layer_name: &str, total_features: u64, field_count: usize) -> String {
    if total_features == 0 {
        return String::new();
    }

    format!(
        r#""tilestats":{{"layerCount":1,"layers":[{{"layer":"{}","count":{},"attributeCount":{}}}]}},"#,
        layer_name, total_features, field_count
    )
}

/// Tile entry with hash for deduplication
#[derive(Debug, Clone)]
struct TileEntry {
    /// Compressed tile data (only stored for unique tiles)
    data: Option<Vec<u8>>,
    /// Hash of uncompressed content (for deduplication)
    hash: u64,
}

/// PMTiles v3 writer
///
/// Accumulates tiles in memory (sorted by tile_id via BTreeMap),
/// then writes the complete archive on finalize.
///
/// Supports tile deduplication: identical tiles are stored once and
/// referenced via PMTiles' `run_length` feature.
pub struct PmtilesWriter {
    /// tile_id -> tile entry (data + hash)
    tiles: BTreeMap<u64, TileEntry>,
    min_zoom: u8,
    max_zoom: u8,
    bounds: TileBounds,
    layer_name: String,
    /// Field metadata: field name -> MVT type ("String", "Number", "Boolean")
    fields: HashMap<String, String>,
    /// Total feature count across all tiles
    total_features: u64,
    /// Feature count per zoom level
    features_per_zoom: HashMap<u8, u64>,
    /// Compression algorithm for tile data
    tile_compression: Compression,
    /// Compression algorithm for internal data (directories, metadata)
    internal_compression: Compression,
    /// Whether deduplication is enabled
    dedup_enabled: bool,
    /// Deduplication cache for tracking seen tiles
    dedup_cache: DeduplicationCache,
    /// Verbatim `vector_layers` array, when the archive holds more than the one
    /// layer `layer_name`/`fields` can describe (a merged band pyramid). When
    /// set it replaces the single-layer entry the writer would otherwise build.
    vector_layers_json: Option<String>,
}

impl PmtilesWriter {
    /// Create a new PMTiles writer with default gzip compression
    ///
    /// Deduplication is disabled by default for backward compatibility.
    /// Call `enable_deduplication(true)` to enable it.
    pub fn new() -> Self {
        Self {
            tiles: BTreeMap::new(),
            min_zoom: 255,
            max_zoom: 0,
            bounds: TileBounds::empty(),
            layer_name: "layer".to_string(),
            fields: HashMap::new(),
            total_features: 0,
            features_per_zoom: HashMap::new(),
            tile_compression: Compression::Gzip,
            internal_compression: Compression::Gzip,
            dedup_enabled: false,
            dedup_cache: DeduplicationCache::new(),
            vector_layers_json: None,
        }
    }

    /// Create a new PMTiles writer with specified compression
    ///
    /// Both tile data and internal data (directories, metadata) will use
    /// the same compression algorithm. Deduplication is disabled by default.
    pub fn with_compression(compression: Compression) -> Self {
        Self {
            tiles: BTreeMap::new(),
            min_zoom: 255,
            max_zoom: 0,
            bounds: TileBounds::empty(),
            layer_name: "layer".to_string(),
            fields: HashMap::new(),
            total_features: 0,
            features_per_zoom: HashMap::new(),
            tile_compression: compression,
            internal_compression: compression,
            dedup_enabled: false,
            dedup_cache: DeduplicationCache::new(),
            vector_layers_json: None,
        }
    }

    /// Enable or disable tile deduplication
    pub fn enable_deduplication(&mut self, enabled: bool) {
        self.dedup_enabled = enabled;
    }

    /// Set the compression algorithm for tile data
    pub fn set_tile_compression(&mut self, compression: Compression) {
        self.tile_compression = compression;
    }

    /// Set the compression algorithm for internal data (directories, metadata)
    pub fn set_internal_compression(&mut self, compression: Compression) {
        self.internal_compression = compression;
    }

    /// Get the current tile compression setting
    pub fn tile_compression(&self) -> Compression {
        self.tile_compression
    }

    /// Get the current internal compression setting
    pub fn internal_compression(&self) -> Compression {
        self.internal_compression
    }

    /// Check if deduplication is enabled
    pub fn is_dedup_enabled(&self) -> bool {
        self.dedup_enabled
    }

    /// Get current deduplication statistics
    pub fn dedup_stats(&self) -> &DeduplicationStats {
        self.dedup_cache.stats()
    }

    /// Set the layer name for vector_layers metadata
    pub fn set_layer_name(&mut self, name: &str) {
        self.layer_name = name.to_string();
    }

    /// Replace the whole `vector_layers` array with a verbatim JSON array.
    ///
    /// A single-layer archive is described by `layer_name` + `fields`; a merged
    /// band pyramid has several layers, each with its own zoom range and field
    /// set, which that pair cannot express.
    pub fn set_vector_layers_json(&mut self, json: String) {
        self.vector_layers_json = Some(json);
    }

    /// Set field metadata for vector_layers.fields
    ///
    /// Field types should be MVT-style: "String", "Number", or "Boolean"
    pub fn set_fields(&mut self, fields: HashMap<String, String>) {
        self.fields = fields;
    }

    /// Build the fields JSON object string
    fn build_fields_json(&self) -> String {
        fields_json(&self.fields)
    }

    /// Build the tilestats JSON fragment
    fn build_tilestats_json(&self) -> String {
        tilestats_json(&self.layer_name, self.total_features, self.fields.len())
    }

    /// Add a tile (will be gzip compressed)
    ///
    /// The tile data should be uncompressed MVT bytes.
    /// Use `add_tile_with_count` if you have feature count available.
    pub fn add_tile(&mut self, z: u8, x: u32, y: u32, data: &[u8]) -> std::io::Result<()> {
        self.add_tile_with_count(z, x, y, data, 0)
    }

    /// Add a tile with feature count for tilestats
    ///
    /// The tile data should be uncompressed MVT bytes.
    ///
    /// If deduplication is enabled, identical tiles will be stored once
    /// and referenced via PMTiles' `run_length` feature.
    pub fn add_tile_with_count(
        &mut self,
        z: u8,
        x: u32,
        y: u32,
        data: &[u8],
        feature_count: usize,
    ) -> std::io::Result<()> {
        let id = tile_id(z, x, y);
        let uncompressed_size = data.len() as u32;

        // Track zoom range
        self.min_zoom = self.min_zoom.min(z);
        self.max_zoom = self.max_zoom.max(z);

        // Track feature counts for tilestats
        self.total_features += feature_count as u64;
        *self.features_per_zoom.entry(z).or_insert(0) += feature_count as u64;

        if self.dedup_enabled {
            // Hash uncompressed data for deduplication
            let hash = TileHasher::hash(data);

            if self.dedup_cache.check(hash).is_some() {
                // Duplicate tile - store reference only (no data)
                self.dedup_cache.record_duplicate(uncompressed_size);
                self.tiles.insert(
                    id,
                    TileEntry {
                        data: None, // No data stored for duplicates
                        hash,
                    },
                );
            } else {
                // New unique tile - compress using configured algorithm and store
                let compressed = compression::compress(data, self.tile_compression)?;
                let compressed_len = compressed.len() as u32;

                // Record in cache (offset will be calculated at write time)
                self.dedup_cache
                    .record_new(hash, 0, compressed_len, uncompressed_size);

                self.tiles.insert(
                    id,
                    TileEntry {
                        data: Some(compressed),
                        hash,
                    },
                );
            }
        } else {
            // No deduplication - store every tile
            let compressed = compression::compress(data, self.tile_compression)?;
            let hash = TileHasher::hash(data);
            self.tiles.insert(
                id,
                TileEntry {
                    data: Some(compressed),
                    hash,
                },
            );
        }

        Ok(())
    }

    /// Add a pre-compressed tile
    ///
    /// Use this if the tile data is already gzip compressed.
    /// Note: Deduplication is not available for pre-compressed tiles
    /// since we cannot hash the original content.
    pub fn add_tile_compressed(
        &mut self,
        z: u8,
        x: u32,
        y: u32,
        compressed_data: Vec<u8>,
    ) -> std::io::Result<()> {
        let id = tile_id(z, x, y);
        // For pre-compressed tiles, use a unique hash based on the compressed data
        // This won't deduplicate as effectively but preserves the API
        let hash = TileHasher::hash(&compressed_data);
        self.tiles.insert(
            id,
            TileEntry {
                data: Some(compressed_data),
                hash,
            },
        );

        self.min_zoom = self.min_zoom.min(z);
        self.max_zoom = self.max_zoom.max(z);

        Ok(())
    }

    /// Set geographic bounds for the tileset
    ///
    /// Latitude values are clamped to Web Mercator bounds (±85.05°).
    pub fn set_bounds(&mut self, bounds: &TileBounds) {
        self.bounds = TileBounds::new(
            bounds.lng_min,
            bounds.lat_min.clamp(-85.05, 85.05),
            bounds.lng_max,
            bounds.lat_max.clamp(-85.05, 85.05),
        );
    }

    /// Get the number of tiles added
    pub fn tile_count(&self) -> usize {
        self.tiles.len()
    }

    /// Write the PMTiles archive to a file
    ///
    /// Layout: [Header (127)] [Root Directory] [Metadata] [Tile Data]
    ///
    /// When deduplication is enabled, identical tiles share storage and
    /// consecutive identical tiles use run_length encoding in the directory.
    pub fn write_to_file(&self, path: &Path) -> Result<()> {
        let file = File::create(path)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to create file: {}", e)))?;
        let mut writer = BufWriter::new(file);

        // Build tile data buffer and directory entries with deduplication
        let mut tile_data_buf = Vec::new();
        let mut entries = Vec::new();

        // Map hash -> (offset, length) for deduplication
        let mut hash_to_offset: HashMap<u64, (u64, u32)> = HashMap::new();
        let mut unique_contents = 0u64;

        if self.dedup_enabled {
            // With deduplication: store unique tiles, reference duplicates
            for (&id, entry) in &self.tiles {
                let (offset, length) = if let Some(ref data) = entry.data {
                    // Unique tile - write to buffer and record location
                    let offset = tile_data_buf.len() as u64;
                    let length = data.len() as u32;
                    tile_data_buf.extend_from_slice(data);
                    hash_to_offset.insert(entry.hash, (offset, length));
                    unique_contents += 1;
                    (offset, length)
                } else {
                    // Duplicate tile - look up existing location
                    *hash_to_offset.get(&entry.hash).expect("Hash must exist")
                };

                // Check if this can extend the previous entry's run_length
                // (same offset = same content, consecutive tile_id)
                if let Some(last) = entries.last_mut() {
                    let last_entry: &mut DirEntry = last;
                    if last_entry.offset == offset
                        && id == last_entry.tile_id + last_entry.run_length as u64
                    {
                        // Extend run_length instead of adding new entry
                        last_entry.run_length += 1;
                        continue;
                    }
                }

                entries.push(DirEntry {
                    tile_id: id,
                    offset,
                    length,
                    run_length: 1,
                });
            }
        } else {
            // Without deduplication: store every tile
            for (&id, entry) in &self.tiles {
                let data = entry.data.as_ref().expect("Non-dedup tiles must have data");
                entries.push(DirEntry {
                    tile_id: id,
                    offset: tile_data_buf.len() as u64,
                    length: data.len() as u32,
                    run_length: 1,
                });
                tile_data_buf.extend_from_slice(data);
                unique_contents += 1;
            }
        }

        // Split into a root directory plus leaf directories when the entries do
        // not fit the spec's 16 KiB root budget. Writing one oversized root
        // instead produces an archive that readers reject outright: go-pmtiles
        // reads the first 16 KiB and panics slicing past it. This only bites
        // above a few thousand tiles, which is why it went unnoticed while this
        // writer was exercised solely by small tests.
        let layout = make_root_leaves(&entries, self.internal_compression)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to build directories: {}", e)))?;
        let compressed_dir = layout.root_bytes;
        let leaves_bytes = layout.leaves_bytes;

        // JSON metadata with vector_layers and tilestats
        let min_z = if self.min_zoom == 255 {
            0
        } else {
            self.min_zoom
        };
        let max_z = if self.max_zoom == 0 && self.tiles.is_empty() {
            0
        } else {
            self.max_zoom
        };
        let tilestats_json = self.build_tilestats_json();
        let vector_layers = match &self.vector_layers_json {
            Some(json) => json.clone(),
            None => format!(
                r#"[{{"id":"{}","minzoom":{},"maxzoom":{},"fields":{}}}]"#,
                self.layer_name,
                min_z,
                max_z,
                self.build_fields_json()
            ),
        };
        let metadata = format!(
            r#"{{"vector_layers":{},{}"format":"pbf","generator":"tylertoo"}}"#,
            vector_layers, tilestats_json
        );
        let compressed_metadata =
            compression::compress(metadata.as_bytes(), self.internal_compression)
                .map_err(|e| Error::PMTilesWrite(format!("Failed to compress metadata: {}", e)))?;

        // Calculate section offsets
        let root_dir_offset = 127u64;
        let root_dir_length = compressed_dir.len() as u64;
        let metadata_offset = root_dir_offset + root_dir_length;
        let metadata_length = compressed_metadata.len() as u64;
        // Leaf directories sit between the metadata and the tile data, and
        // `DirEntry::offset` for a leaf is relative to leaf_dirs_offset.
        let leaf_dirs_offset = metadata_offset + metadata_length;
        let leaf_dirs_length = leaves_bytes.len() as u64;
        let tile_data_offset = leaf_dirs_offset + leaf_dirs_length;
        let tile_data_length = tile_data_buf.len() as u64;

        // Build header
        let header = Header {
            root_dir_offset,
            root_dir_length,
            json_metadata_offset: metadata_offset,
            json_metadata_length: metadata_length,
            // Always the section position, never 0 -- even with no leaves. See
            // the note on `leaf_dirs_offset` in `StreamingPmtilesWriter`.
            leaf_dirs_offset,
            leaf_dirs_length,
            tile_data_offset,
            tile_data_length,
            addressed_tiles_count: self.tiles.len() as u64,
            tile_entries_count: entries.len() as u64,
            tile_contents_count: unique_contents,
            clustered: true,
            internal_compression: self.internal_compression,
            tile_compression: self.tile_compression,
            tile_type: TileType::Mvt,
            min_zoom: if self.min_zoom == 255 {
                0
            } else {
                self.min_zoom
            },
            max_zoom: if self.max_zoom == 0 && self.tiles.is_empty() {
                0
            } else {
                self.max_zoom
            },
            min_lon: self.bounds.lng_min,
            min_lat: self.bounds.lat_min,
            max_lon: self.bounds.lng_max,
            max_lat: self.bounds.lat_max,
            center_zoom: if self.tiles.is_empty() {
                0
            } else {
                (self.min_zoom + self.max_zoom) / 2
            },
            center_lon: (self.bounds.lng_min + self.bounds.lng_max) / 2.0,
            center_lat: (self.bounds.lat_min + self.bounds.lat_max) / 2.0,
        };

        // Write all sections
        writer
            .write_all(&header.to_bytes())
            .map_err(|e| Error::PMTilesWrite(format!("Failed to write header: {}", e)))?;
        writer
            .write_all(&compressed_dir)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to write directory: {}", e)))?;
        writer
            .write_all(&compressed_metadata)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to write metadata: {}", e)))?;
        writer
            .write_all(&leaves_bytes)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to write leaf directories: {}", e)))?;
        writer
            .write_all(&tile_data_buf)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to write tile data: {}", e)))?;

        writer
            .flush()
            .map_err(|e| Error::PMTilesWrite(format!("Failed to flush: {}", e)))?;

        Ok(())
    }
}

impl Default for PmtilesWriter {
    fn default() -> Self {
        Self::new()
    }
}

// ============================================================================
// StreamingPmtilesWriter - Writes tile data to temp file immediately
// ============================================================================

use std::path::PathBuf;

/// Directory entry for streaming writer (minimal memory footprint).
/// Only stores what's needed for final directory encoding.
#[derive(Debug, Clone)]
struct StreamingDirEntry {
    tile_id: u64,
    offset: u64,
    length: u32,
}

/// Statistics about streaming write operations.
#[derive(Debug, Clone, Default)]
pub struct StreamingWriteStats {
    /// Total tiles added (including duplicates)
    pub total_tiles: u64,
    /// Unique tiles written to disk
    pub unique_tiles: u64,
    /// Bytes written to temp file
    pub bytes_written: u64,
    /// Bytes saved by deduplication
    pub bytes_saved_dedup: u64,
}

impl StreamingWriteStats {
    /// Calculate memory used by directory entries (approximate).
    /// Each StreamingDirEntry is ~24 bytes (tile_id: 8, offset: 8, length: 4 + padding).
    /// We estimate based on total_tiles since each tile gets a directory entry.
    pub fn estimated_memory_bytes(&self) -> u64 {
        // Each entry: tile_id (8) + offset (8) + length (4) = 20 bytes + padding ≈ 24 bytes
        // Plus HashMap entry overhead for dedup cache: ~40 bytes per unique
        // Plus Vec overhead: ~8 bytes
        self.total_tiles * 24 + self.unique_tiles * 40
    }
}

/// PMTiles writer that streams tile data to disk immediately.
///
/// Unlike `PmtilesWriter` which accumulates all tiles in memory,
/// `StreamingPmtilesWriter` writes compressed tile data to a temp file
/// as tiles are added. Only the small directory entries (~32 bytes each)
/// are kept in memory.
///
/// # Memory Usage
///
/// For 30,000 tiles:
/// - `PmtilesWriter`: ~1.2 GB (all tile data in memory)
/// - `StreamingPmtilesWriter`: ~2-3 MB (only directory entries)
///
/// # Example
///
/// ```no_run
/// use tylertoo_core::pmtiles_writer::StreamingPmtilesWriter;
/// use tylertoo_core::compression::Compression;
/// use std::path::Path;
///
/// let mut writer = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
/// writer.add_tile(0, 0, 0, &[0x1a, 0x00]).unwrap();
/// writer.add_tile(1, 0, 0, &[0x1a, 0x01]).unwrap();
/// writer.finalize(Path::new("output.pmtiles")).unwrap();
/// ```
pub struct StreamingPmtilesWriter {
    /// Buffered writer for temp file (tile data written immediately)
    temp_file: Option<BufWriter<File>>,
    /// Path to temp file (for cleanup and final assembly)
    temp_path: PathBuf,
    /// Directory entries (minimal memory: ~32 bytes each)
    entries: Vec<StreamingDirEntry>,
    /// Deduplication: hash → (offset, length) for detecting duplicates
    dedup_cache: HashMap<u64, (u64, u32)>,
    /// Current write offset in temp file
    current_offset: u64,
    /// Min zoom level seen
    min_zoom: u8,
    /// Max zoom level seen
    max_zoom: u8,
    /// Minimum zoom the archive *declares* even when no tile exists there
    /// (#380): the header and `vector_layers` cover `min(declared, seen)`.
    declared_min_zoom: Option<u8>,
    /// Geographic bounds
    bounds: TileBounds,
    /// Layer name for metadata
    layer_name: String,
    /// Field metadata
    fields: HashMap<String, String>,
    /// Verbatim `vector_layers` array, when the archive holds more than the one
    /// layer `layer_name`/`fields` can describe (a merged band pyramid). When
    /// set it replaces the single-layer entry the writer would otherwise build.
    vector_layers_json: Option<String>,
    /// Compression for tile data
    tile_compression: Compression,
    /// Compression for internal data (directories, metadata)
    internal_compression: Compression,
    /// Statistics
    stats: StreamingWriteStats,
    /// Total feature count
    total_features: u64,
    /// Whether finalize has been called (prevents double cleanup)
    finalized: bool,
}

impl StreamingPmtilesWriter {
    /// Create a new streaming writer with the specified compression.
    ///
    /// Creates a temp file in the system temp directory for tile data.
    pub fn new(compression: Compression) -> std::io::Result<Self> {
        Self::with_temp_dir(compression, std::env::temp_dir())
    }

    /// Create a new streaming writer with a custom temp directory.
    pub fn with_temp_dir(compression: Compression, temp_dir: PathBuf) -> std::io::Result<Self> {
        use std::time::{SystemTime, UNIX_EPOCH};

        // Generate unique temp file name with timestamp + process/thread IDs for parallel safety
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos();
        let pid = std::process::id();
        let tid = std::thread::current().id();
        let temp_path = temp_dir.join(format!("tylertoo-{}-{}-{:?}.tmp", timestamp, pid, tid));

        let file = File::create(&temp_path)?;
        let temp_file = BufWriter::with_capacity(64 * 1024, file); // 64KB buffer

        Ok(Self {
            temp_file: Some(temp_file),
            temp_path,
            entries: Vec::new(),
            dedup_cache: HashMap::new(),
            current_offset: 0,
            min_zoom: 255,
            max_zoom: 0,
            declared_min_zoom: None,
            bounds: TileBounds::empty(),
            layer_name: "layer".to_string(),
            fields: HashMap::new(),
            vector_layers_json: None,
            tile_compression: compression,
            internal_compression: compression,
            stats: StreamingWriteStats::default(),
            total_features: 0,
            finalized: false,
        })
    }

    /// Get the path to the temp file (for testing).
    pub fn temp_path(&self) -> &Path {
        &self.temp_path
    }

    /// Set the layer name for metadata.
    pub fn set_layer_name(&mut self, name: &str) {
        self.layer_name = name.to_string();
    }

    /// Declare a minimum zoom for the archive regardless of which zooms end
    /// up holding tiles (#380). The header `min_zoom` and the layer's
    /// `minzoom` become `min(declared, coarsest tile written)`: an empty zoom
    /// in a PMTiles archive is just an absent tile, so declaring z0 over a
    /// pyramid whose coarsest level generalized to nothing is honest, whereas
    /// letting the header drift to z1 hides the requested range from clients
    /// that trust it. A declared value finer than a written tile is ignored —
    /// the header can widen the range, never narrow it over real tiles.
    pub fn set_declared_min_zoom(&mut self, zoom: u8) {
        self.declared_min_zoom = Some(zoom);
    }

    /// The header's minimum zoom: the coarsest tile written, widened by any
    /// declared minimum. An archive with no tiles is z0..z0 whatever was
    /// declared — its max zoom collapses to 0, and a declared minimum above
    /// that would invert the range.
    fn header_min_zoom(&self) -> u8 {
        if self.entries.is_empty() {
            return 0;
        }
        match self.declared_min_zoom {
            Some(d) => self.min_zoom.min(d),
            None => self.min_zoom,
        }
    }

    /// Set field metadata.
    pub fn set_fields(&mut self, fields: HashMap<String, String>) {
        self.fields = fields;
    }

    /// Replace the whole `vector_layers` array with a verbatim JSON array.
    ///
    /// A single-layer archive is described by `layer_name` + `fields`; a merged
    /// band pyramid has several layers, each with its own zoom range and field
    /// set, which that pair cannot express. Honoured by the one metadata
    /// assembler (`build_metadata_json`) both `checkpoint` and `finalize` use,
    /// so a checkpointed archive and the final one cannot disagree.
    pub fn set_vector_layers_json(&mut self, json: String) {
        self.vector_layers_json = Some(json);
    }

    /// Set geographic bounds.
    ///
    /// Latitude values are clamped to Web Mercator bounds (±85.05°).
    pub fn set_bounds(&mut self, bounds: &TileBounds) {
        self.bounds = TileBounds::new(
            bounds.lng_min,
            bounds.lat_min.clamp(-85.05, 85.05),
            bounds.lng_max,
            bounds.lat_max.clamp(-85.05, 85.05),
        );
    }

    /// Get current statistics.
    pub fn stats(&self) -> &StreamingWriteStats {
        &self.stats
    }

    /// Add a tile (writes immediately to temp file if unique).
    ///
    /// Tiles are compressed and written immediately. Duplicate tiles
    /// (same content) are detected and not written again.
    pub fn add_tile(&mut self, z: u8, x: u32, y: u32, data: &[u8]) -> std::io::Result<()> {
        self.add_tile_with_count(z, x, y, data, 0)
    }

    /// Add a tile with feature count.
    pub fn add_tile_with_count(
        &mut self,
        z: u8,
        x: u32,
        y: u32,
        data: &[u8],
        feature_count: usize,
    ) -> std::io::Result<()> {
        let temp_file = self
            .temp_file
            .as_mut()
            .ok_or_else(|| std::io::Error::other("Writer already finalized"))?;

        let id = tile_id(z, x, y);
        self.stats.total_tiles += 1;
        self.total_features += feature_count as u64;

        // Track zoom range
        self.min_zoom = self.min_zoom.min(z);
        self.max_zoom = self.max_zoom.max(z);

        // Hash uncompressed data for deduplication
        let hash = crate::dedup::TileHasher::hash(data);

        // Check for duplicate
        if let Some((offset, length)) = self.dedup_cache.get(&hash) {
            // Duplicate - just add directory entry pointing to existing data
            self.entries.push(StreamingDirEntry {
                tile_id: id,
                offset: *offset,
                length: *length,
            });
            self.stats.bytes_saved_dedup += data.len() as u64;
            return Ok(());
        }

        // New unique tile - compress and write to temp file
        let compressed = compression::compress(data, self.tile_compression)?;
        let compressed_len = compressed.len() as u32;

        temp_file.write_all(&compressed)?;

        // Record in dedup cache and directory
        let offset = self.current_offset;
        self.dedup_cache.insert(hash, (offset, compressed_len));
        self.entries.push(StreamingDirEntry {
            tile_id: id,
            offset,
            length: compressed_len,
        });

        self.current_offset += compressed_len as u64;
        self.stats.unique_tiles += 1;
        self.stats.bytes_written += compressed_len as u64;

        Ok(())
    }

    /// Add a tile whose bytes are **already compressed** with this writer's
    /// `tile_compression`.
    ///
    /// This is the parallel-friendly counterpart of [`Self::add_tile_with_count`]:
    /// the caller compresses tile bytes off-thread (e.g. inside a Rayon
    /// `par_iter`) and hands the finished bytes here so the serial ordering loop
    /// never runs gzip. To keep deduplication byte-for-byte identical to the
    /// serial path, `hash` MUST be the [`TileHasher::hash`] of the tile's
    /// **uncompressed** MVT bytes (never the compressed bytes), and `raw_len` the
    /// uncompressed length (used only for the dedup byte-savings stat). Because
    /// compression is deterministic, an identical uncompressed hash implies
    /// identical compressed bytes, so the archive is bit-identical to compressing
    /// serially.
    #[allow(clippy::too_many_arguments)]
    pub fn add_tile_precompressed(
        &mut self,
        z: u8,
        x: u32,
        y: u32,
        hash: u64,
        compressed: &[u8],
        raw_len: usize,
        feature_count: usize,
    ) -> std::io::Result<()> {
        let temp_file = self
            .temp_file
            .as_mut()
            .ok_or_else(|| std::io::Error::other("Writer already finalized"))?;

        let id = tile_id(z, x, y);
        self.stats.total_tiles += 1;
        self.total_features += feature_count as u64;

        // Track zoom range
        self.min_zoom = self.min_zoom.min(z);
        self.max_zoom = self.max_zoom.max(z);

        // Dedup on the uncompressed hash (same key as add_tile_with_count).
        if let Some((offset, length)) = self.dedup_cache.get(&hash) {
            self.entries.push(StreamingDirEntry {
                tile_id: id,
                offset: *offset,
                length: *length,
            });
            self.stats.bytes_saved_dedup += raw_len as u64;
            return Ok(());
        }

        // New unique tile - write the pre-compressed bytes verbatim.
        let compressed_len = compressed.len() as u32;
        temp_file.write_all(compressed)?;

        let offset = self.current_offset;
        self.dedup_cache.insert(hash, (offset, compressed_len));
        self.entries.push(StreamingDirEntry {
            tile_id: id,
            offset,
            length: compressed_len,
        });

        self.current_offset += compressed_len as u64;
        self.stats.unique_tiles += 1;
        self.stats.bytes_written += compressed_len as u64;

        Ok(())
    }

    /// Finalize the PMTiles file.
    ///
    /// Reads tile data from temp file and assembles the final PMTiles archive
    /// with header, directory, metadata, and tile data sections.
    ///
    /// The temp file is deleted after successful finalization.
    pub fn finalize(mut self, output_path: &Path) -> Result<StreamingWriteStats> {
        // Assemble the complete archive (byte-identical to the last checkpoint,
        // if any). This flushes the temp buffer in place but leaves the handle
        // open so we can close it deterministically below.
        self.write_archive(output_path)?;

        // Close and remove the temp file now that the run is complete.
        drop(self.temp_file.take());
        let _ = std::fs::remove_file(&self.temp_path);

        // Mark as finalized so Drop doesn't try to clean up again
        self.finalized = true;

        Ok(self.stats.clone())
    }

    /// Write a valid, self-contained PMTiles archive containing every tile
    /// added so far, *without* consuming the writer or closing the temp file
    /// (Issue #229 — salvageable output).
    ///
    /// The export loop calls this after each finished level so an interrupted
    /// run still yields a valid archive capped at the last completed zoom
    /// instead of losing hours of compute. `finalize` routes through the same
    /// assembler, so the final archive is byte-identical whether or not any
    /// intermediate checkpoints were taken.
    ///
    /// The archive is written to a sibling `<output>.partial` file and then
    /// atomically renamed over `output_path`, so a kill mid-write never
    /// corrupts a previously-checkpointed archive.
    pub fn checkpoint(&mut self, output_path: &Path) -> Result<()> {
        self.write_archive(output_path)
    }

    /// Shared archive assembler backing both [`checkpoint`](Self::checkpoint)
    /// and [`finalize`](Self::finalize). Flushes the temp buffer in place (the
    /// handle stays open so the writer remains usable) and writes the header,
    /// directory, metadata and tile data to `output_path` atomically.
    fn write_archive(&mut self, output_path: &Path) -> Result<()> {
        // Flush buffered tile bytes to the temp file on disk without consuming
        // the handle — the writer must stay usable after a checkpoint.
        match self.temp_file.as_mut() {
            Some(tf) => tf
                .flush()
                .map_err(|e| Error::PMTilesWrite(format!("Failed to flush temp file: {}", e)))?,
            None => return Err(Error::PMTilesWrite("Writer already finalized".to_string())),
        }

        // Sort entries by tile_id for clustered mode. Tile ids are unique, so
        // this is deterministic and idempotent — re-sorting between checkpoints
        // and later adds yields the same final ordering as sorting once.
        self.entries.sort_by_key(|e| e.tile_id);

        // Build run-length encoded directory entries
        let dir_entries = self.build_directory_entries();

        // Build directory structure with leaf directories if needed (Issue #88)
        // This ensures root directory fits in the initial 16KB HTTP range request
        let dir_layout = make_root_leaves(&dir_entries, self.internal_compression)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to build directory: {}", e)))?;

        // Build metadata JSON
        let metadata = self.build_metadata_json();
        let compressed_metadata =
            compression::compress(metadata.as_bytes(), self.internal_compression)
                .map_err(|e| Error::PMTilesWrite(format!("Failed to compress metadata: {}", e)))?;

        // Calculate section offsets
        // Layout: Header | Root Dir | Metadata | Leaf Dirs | Tile Data
        let root_dir_offset = 127u64;
        let root_dir_length = dir_layout.root_bytes.len() as u64;
        let metadata_offset = root_dir_offset + root_dir_length;
        let metadata_length = compressed_metadata.len() as u64;
        let leaf_dirs_offset = metadata_offset + metadata_length;
        let leaf_dirs_length = dir_layout.leaves_bytes.len() as u64;
        let tile_data_offset = leaf_dirs_offset + leaf_dirs_length;
        let tile_data_length = self.current_offset;

        // Build header
        let header = Header {
            root_dir_offset,
            root_dir_length,
            json_metadata_offset: metadata_offset,
            json_metadata_length: metadata_length,
            // Always the section position, never 0.
            //
            // go-pmtiles REJECTS an archive whose leaf-directory offset is 0:
            //
            //     Failed to verify archive, Leaf directories offset=0 must not be 0
            //
            // Pointing it at the (empty) leaf section instead -- which is where
            // leaves would start, between the metadata and the tile data --
            // verifies clean. Zeroing it affected every small archive written
            // through this path, which is the production one.
            leaf_dirs_offset,
            leaf_dirs_length,
            tile_data_offset,
            tile_data_length,
            addressed_tiles_count: self.stats.total_tiles,
            tile_entries_count: dir_entries.len() as u64,
            tile_contents_count: self.stats.unique_tiles,
            clustered: true,
            internal_compression: self.internal_compression,
            tile_compression: self.tile_compression,
            tile_type: TileType::Mvt,
            min_zoom: self.header_min_zoom(),
            max_zoom: if self.max_zoom == 0 && self.entries.is_empty() {
                0
            } else {
                self.max_zoom
            },
            min_lon: self.bounds.lng_min,
            min_lat: self.bounds.lat_min,
            max_lon: self.bounds.lng_max,
            max_lat: self.bounds.lat_max,
            center_zoom: if self.entries.is_empty() {
                0
            } else {
                (self.header_min_zoom() + self.max_zoom) / 2
            },
            center_lon: (self.bounds.lng_min + self.bounds.lng_max) / 2.0,
            center_lat: (self.bounds.lat_min + self.bounds.lat_max) / 2.0,
        };

        // Assemble into a sibling `<output>.partial` file, then atomically
        // rename over `output_path`. A kill mid-write leaves any previously
        // checkpointed archive at `output_path` intact.
        let partial_path = {
            let mut os = output_path.as_os_str().to_owned();
            os.push(".partial");
            PathBuf::from(os)
        };
        let output_file = File::create(&partial_path)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to create output file: {}", e)))?;
        let mut writer = BufWriter::new(output_file);

        // Write header
        writer
            .write_all(&header.to_bytes())
            .map_err(|e| Error::PMTilesWrite(format!("Failed to write header: {}", e)))?;

        // Write root directory
        writer
            .write_all(&dir_layout.root_bytes)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to write root directory: {}", e)))?;

        // Write metadata
        writer
            .write_all(&compressed_metadata)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to write metadata: {}", e)))?;

        // Write leaf directories (if any)
        if !dir_layout.leaves_bytes.is_empty() {
            writer.write_all(&dir_layout.leaves_bytes).map_err(|e| {
                Error::PMTilesWrite(format!("Failed to write leaf directories: {}", e))
            })?;
        }

        // Copy tile data from temp file
        let mut temp_reader = File::open(&self.temp_path)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to reopen temp file: {}", e)))?;
        std::io::copy(&mut temp_reader, &mut writer)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to copy tile data: {}", e)))?;

        writer
            .flush()
            .map_err(|e| Error::PMTilesWrite(format!("Failed to flush output: {}", e)))?;
        // Close the output handle before renaming so the bytes are fully on disk.
        drop(writer);

        // Atomically publish the assembled archive.
        std::fs::rename(&partial_path, output_path)
            .map_err(|e| Error::PMTilesWrite(format!("Failed to publish archive: {}", e)))?;

        Ok(())
    }

    /// Build directory entries with run-length encoding for consecutive identical tiles.
    fn build_directory_entries(&self) -> Vec<DirEntry> {
        let mut dir_entries = Vec::new();

        for entry in &self.entries {
            // Check if this extends the previous entry's run
            if let Some(last) = dir_entries.last_mut() {
                let last_entry: &mut DirEntry = last;
                if last_entry.offset == entry.offset
                    && entry.tile_id == last_entry.tile_id + last_entry.run_length as u64
                {
                    last_entry.run_length += 1;
                    continue;
                }
            }

            dir_entries.push(DirEntry {
                tile_id: entry.tile_id,
                offset: entry.offset,
                length: entry.length,
                run_length: 1,
            });
        }

        dir_entries
    }

    /// Build metadata JSON string.
    fn build_metadata_json(&self) -> String {
        let min_z = self.header_min_zoom();
        let max_z = if self.max_zoom == 0 && self.entries.is_empty() {
            0
        } else {
            self.max_zoom
        };

        let tilestats_json = self.build_tilestats_json();
        let vector_layers = match &self.vector_layers_json {
            Some(json) => json.clone(),
            None => format!(
                r#"[{{"id":"{}","minzoom":{},"maxzoom":{},"fields":{}}}]"#,
                self.layer_name,
                min_z,
                max_z,
                self.build_fields_json()
            ),
        };

        format!(
            r#"{{"vector_layers":{},{}"format":"pbf","generator":"tylertoo"}}"#,
            vector_layers, tilestats_json
        )
    }

    fn build_fields_json(&self) -> String {
        fields_json(&self.fields)
    }

    fn build_tilestats_json(&self) -> String {
        tilestats_json(&self.layer_name, self.total_features, self.fields.len())
    }
}

impl Drop for StreamingPmtilesWriter {
    fn drop(&mut self) {
        // Clean up temp file if it still exists (e.g., if finalize wasn't called)
        if !self.finalized {
            let _ = std::fs::remove_file(&self.temp_path);
        }
    }
}

// ============================================================================
// Tests (TDD)
// ============================================================================

#[cfg(test)]
mod tests {

    /// An archive whose directory outgrows the spec's 16 KiB root budget must
    /// spill into leaf directories. Writing one oversized root instead makes a
    /// file readers reject: go-pmtiles reads the first 16 KiB of root and
    /// panics slicing past it ("slice bounds out of range [:48771] with
    /// capacity 16384" on a 23,559-tile pyramid). Only tiny archives were ever
    /// written through this path, so it stayed hidden.
    #[test]
    fn large_archive_spills_into_leaf_directories() {
        let mut writer = PmtilesWriter::new();
        // Matching the scale that exposed this: tens of thousands of tiles, so
        // the entry offsets and lengths do not delta-encode down to nothing.
        // z8 is 256x256, comfortably more than 24,000 addresses.
        // Gapped addresses, all distinct: a real pyramid's tiles are sparse, so
        // the tile-id deltas are large and the directory does not compress to
        // nothing the way a solid block of sequential ids would.
        // Irregular lengths from a tiny LCG, so the entry offsets are the
        // uneven numbers a real archive has. A regular pattern gzips down to
        // nothing and never reaches the budget this test is about.
        let mut rng = 0x2545_F491_4F6C_DD1Du64;
        for i in 0..40_000u32 {
            let (x, y) = ((i % 200) * 20, (i / 200) * 20);
            rng = rng
                .wrapping_mul(6364136223846793005)
                .wrapping_add(1442695040888963407);
            // Pre-compressed, the path a band merge uses: it also skips 40k
            // gzip calls that would make this test take half a minute.
            let len = 64 + (rng >> 33) as usize % 4096;
            writer
                .add_tile_compressed(14, x, y, vec![(i % 251) as u8; len])
                .unwrap();
        }
        let tmp = tempfile::NamedTempFile::new().unwrap();
        writer.write_to_file(tmp.path()).unwrap();

        let bytes = std::fs::read(tmp.path()).unwrap();
        let header = Header::from_bytes(&bytes).unwrap();
        assert!(
            header.root_dir_length <= 16384,
            "root directory must fit the spec budget, got {}",
            header.root_dir_length
        );
        assert!(
            header.leaf_dirs_length > 0,
            "an archive too big for one root must have leaf directories"
        );
        // Sections must not overlap: leaves sit between metadata and tile data.
        assert_eq!(
            header.leaf_dirs_offset,
            header.json_metadata_offset + header.json_metadata_length
        );
        assert_eq!(
            header.tile_data_offset,
            header.leaf_dirs_offset + header.leaf_dirs_length
        );

        // Structure is not the point -- READABILITY is. The bug this fixes
        // produced a file whose header looked perfectly reasonable, which is
        // exactly why it survived: a wrong leaf-offset base would satisfy every
        // assertion above and still hand a reader garbage. So walk the
        // directories the way a reader does and check the bytes come back.
        let read_dir = |raw: &[u8]| -> Vec<DirEntry> {
            let plain = compression::decompress(raw, header.internal_compression).unwrap();
            decode_directory(&plain).expect("directory must decode")
        };
        let root = read_dir(
            &bytes[header.root_dir_offset as usize
                ..(header.root_dir_offset + header.root_dir_length) as usize],
        );
        assert!(
            root.iter().any(|e| e.run_length == 0),
            "a spilled archive's root must contain at least one leaf pointer"
        );

        let mut found: HashMap<u64, Vec<u8>> = HashMap::new();
        for e in &root {
            let leaves = if e.run_length == 0 {
                // A leaf pointer: `offset` is relative to leaf_dirs_offset.
                let start = (header.leaf_dirs_offset + e.offset) as usize;
                read_dir(&bytes[start..start + e.length as usize])
            } else {
                vec![e.clone()]
            };
            for le in leaves {
                let start = (header.tile_data_offset + le.offset) as usize;
                found.insert(
                    le.tile_id,
                    bytes[start..start + le.length as usize].to_vec(),
                );
            }
        }
        assert_eq!(found.len(), 40_000, "every tile must be addressable");

        // Spot-check content at both ends and the middle. The payload is a run
        // of `(i % 251)` bytes, so a mis-resolved pointer gives a wrong byte.
        for i in [0u32, 1, 19_899, 39_998, 39_999] {
            let (x, y) = ((i % 200) * 20, (i / 200) * 20);
            let data = found
                .get(&tile_id(14, x, y))
                .unwrap_or_else(|| panic!("tile {i} (z14/{x}/{y}) not found"));
            let want = (i % 251) as u8;
            assert!(
                data.iter().all(|&b| b == want),
                "tile {i} (z14/{x}/{y}) resolved to the wrong bytes: \
                 expected all {want}, got {:?}..",
                &data[..data.len().min(8)]
            );
        }
    }
    use super::*;
    use std::fs;

    // -------------------------------------------------------------------------
    // Task 7: Header and Structures Tests
    // -------------------------------------------------------------------------

    #[test]
    fn test_header_size_is_127_bytes() {
        let header = Header::default();
        let bytes = header.to_bytes();
        assert_eq!(
            bytes.len(),
            127,
            "PMTiles v3 header must be exactly 127 bytes"
        );
    }

    #[test]
    fn test_header_magic_and_version() {
        let header = Header::default();
        let bytes = header.to_bytes();
        assert_eq!(&bytes[0..7], b"PMTiles", "Magic number must be 'PMTiles'");
        assert_eq!(bytes[7], 3, "Version must be 3");
    }

    #[test]
    fn test_header_default_offsets() {
        let header = Header::default();
        let bytes = header.to_bytes();

        // Root directory offset should be 127 (immediately after header)
        let root_offset = u64::from_le_bytes(bytes[8..16].try_into().unwrap());
        assert_eq!(root_offset, 127);
    }

    #[test]
    fn test_header_bounds_encoding() {
        let header = Header {
            min_lon: -122.4194, // San Francisco
            min_lat: 37.7749,
            max_lon: -122.3894,
            max_lat: 37.8049,
            ..Default::default()
        };

        let bytes = header.to_bytes();

        // Decode min_lon (bytes 102-105)
        let min_lon_encoded = i32::from_le_bytes(bytes[102..106].try_into().unwrap());
        let min_lon_decoded = min_lon_encoded as f64 / 10_000_000.0;
        assert!(
            (min_lon_decoded - header.min_lon).abs() < 0.0001,
            "Lon encoding should preserve precision to ~0.0001 degrees"
        );
    }

    #[test]
    fn test_tile_id_zoom_0() {
        // At zoom 0, there's only one tile (0,0,0) with ID 0
        assert_eq!(tile_id(0, 0, 0), 0);
    }

    #[test]
    fn test_tile_id_zoom_1_matches_spec() {
        // From PMTiles spec examples:
        // Z=1, X=0, Y=0 → TileID=1
        // Z=1, X=0, Y=1 → TileID=2
        // Z=1, X=1, Y=1 → TileID=3
        // Z=1, X=1, Y=0 → TileID=4
        assert_eq!(tile_id(1, 0, 0), 1);
        assert_eq!(tile_id(1, 0, 1), 2);
        assert_eq!(tile_id(1, 1, 1), 3);
        assert_eq!(tile_id(1, 1, 0), 4);
    }

    #[test]
    fn test_tile_id_zoom_2_base() {
        // Z=2, X=0, Y=0 → TileID=5 (base for zoom 2)
        assert_eq!(tile_id(2, 0, 0), 5);
    }

    #[test]
    fn test_tile_id_unique_at_each_zoom() {
        // All tiles at a given zoom should have unique IDs
        for z in 0..=4u8 {
            let mut ids = Vec::new();
            let n = 1u32 << z;
            for y in 0..n {
                for x in 0..n {
                    ids.push(tile_id(z, x, y));
                }
            }
            let original_len = ids.len();
            ids.sort();
            ids.dedup();
            assert_eq!(
                ids.len(),
                original_len,
                "All tile IDs at zoom {} should be unique",
                z
            );
        }
    }

    #[test]
    fn test_tile_id_to_zxy_matches_spec_examples() {
        assert_eq!(tile_id_to_zxy(0).unwrap(), (0, 0, 0));
        assert_eq!(tile_id_to_zxy(1).unwrap(), (1, 0, 0));
        assert_eq!(tile_id_to_zxy(2).unwrap(), (1, 0, 1));
        assert_eq!(tile_id_to_zxy(3).unwrap(), (1, 1, 1));
        assert_eq!(tile_id_to_zxy(4).unwrap(), (1, 1, 0));
        assert_eq!(tile_id_to_zxy(5).unwrap(), (2, 0, 0));
    }

    #[test]
    fn test_tile_id_to_zxy_inverts_tile_id_exhaustively() {
        // Exhaustive round-trip at low zooms...
        for z in 0..=5u8 {
            let n = 1u32 << z;
            for y in 0..n {
                for x in 0..n {
                    assert_eq!(
                        tile_id_to_zxy(tile_id(z, x, y)).unwrap(),
                        (z, x, y),
                        "round-trip z={z} x={x} y={y}"
                    );
                }
            }
        }
        // ...and spot checks at high zooms, including corners.
        for (z, x, y) in [
            (14u8, 4823u32, 6160u32),
            (14, 0, 0),
            (14, (1 << 14) - 1, (1 << 14) - 1),
            (20, 123_456, 654_321),
            (31, (1u32 << 31) - 1, 0),
        ] {
            assert_eq!(tile_id_to_zxy(tile_id(z, x, y)).unwrap(), (z, x, y));
        }
    }

    #[test]
    fn test_tile_id_to_zxy_rejects_out_of_range() {
        // One past the last z31 ID must error rather than wrap.
        let past_z31 = (0..=31u8).map(|z| 1u64 << (2 * u64::from(z))).sum::<u64>();
        assert!(tile_id_to_zxy(past_z31).is_err());
        assert!(tile_id_to_zxy(u64::MAX).is_err());
    }

    #[test]
    fn test_header_from_bytes_roundtrips_to_bytes() {
        let header = Header {
            root_dir_offset: 127,
            root_dir_length: 421,
            json_metadata_offset: 548,
            json_metadata_length: 33,
            leaf_dirs_offset: 581,
            leaf_dirs_length: 1290,
            tile_data_offset: 1871,
            tile_data_length: 999_999,
            addressed_tiles_count: 42,
            tile_entries_count: 40,
            tile_contents_count: 39,
            clustered: true,
            internal_compression: Compression::Gzip,
            tile_compression: Compression::Zstd,
            tile_type: TileType::Mvt,
            min_zoom: 3,
            max_zoom: 14,
            min_lon: -75.1652,
            min_lat: -33.8688,
            max_lon: 151.2093,
            max_lat: 48.8566,
            center_zoom: 8,
            center_lon: 2.3522,
            center_lat: 39.9526,
        };
        let parsed = Header::from_bytes(&header.to_bytes()).unwrap();

        assert_eq!(parsed.root_dir_offset, header.root_dir_offset);
        assert_eq!(parsed.root_dir_length, header.root_dir_length);
        assert_eq!(parsed.json_metadata_offset, header.json_metadata_offset);
        assert_eq!(parsed.json_metadata_length, header.json_metadata_length);
        assert_eq!(parsed.leaf_dirs_offset, header.leaf_dirs_offset);
        assert_eq!(parsed.leaf_dirs_length, header.leaf_dirs_length);
        assert_eq!(parsed.tile_data_offset, header.tile_data_offset);
        assert_eq!(parsed.tile_data_length, header.tile_data_length);
        assert_eq!(parsed.addressed_tiles_count, header.addressed_tiles_count);
        assert_eq!(parsed.tile_entries_count, header.tile_entries_count);
        assert_eq!(parsed.tile_contents_count, header.tile_contents_count);
        assert_eq!(parsed.clustered, header.clustered);
        assert_eq!(parsed.internal_compression, header.internal_compression);
        assert_eq!(parsed.tile_compression, header.tile_compression);
        assert_eq!(parsed.tile_type, header.tile_type);
        assert_eq!(parsed.min_zoom, header.min_zoom);
        assert_eq!(parsed.max_zoom, header.max_zoom);
        // Coordinates go through the i32 * 1e7 spec encoding: 1e-7 precision.
        for (got, want) in [
            (parsed.min_lon, header.min_lon),
            (parsed.min_lat, header.min_lat),
            (parsed.max_lon, header.max_lon),
            (parsed.max_lat, header.max_lat),
            (parsed.center_lon, header.center_lon),
            (parsed.center_lat, header.center_lat),
        ] {
            assert!((got - want).abs() < 1e-6, "{got} vs {want}");
        }
        assert_eq!(parsed.center_zoom, header.center_zoom);
    }

    #[test]
    fn test_header_from_bytes_rejects_garbage() {
        // Too short.
        assert!(Header::from_bytes(&[0u8; 50]).is_err());
        // Bad magic.
        let mut bytes = Header::default().to_bytes();
        bytes[0] = b'X';
        assert!(Header::from_bytes(&bytes).is_err());
        // Bad version.
        let mut bytes = Header::default().to_bytes();
        bytes[7] = 2;
        assert!(Header::from_bytes(&bytes).is_err());
        // Out-of-spec compression code.
        let mut bytes = Header::default().to_bytes();
        bytes[97] = 9;
        assert!(Header::from_bytes(&bytes).is_err());
        // Out-of-spec tile type code.
        let mut bytes = Header::default().to_bytes();
        bytes[99] = 9;
        assert!(Header::from_bytes(&bytes).is_err());
    }

    #[test]
    fn test_tile_id_increasing_with_zoom() {
        // Max ID at zoom z should be less than min ID at zoom z+1
        for z in 0..4u8 {
            let n = 1u32 << z;
            let max_id_at_z = (0..n)
                .flat_map(|y| (0..n).map(move |x| tile_id(z, x, y)))
                .max()
                .unwrap();

            let min_id_at_z_plus_1 = tile_id(z + 1, 0, 0);

            assert!(
                max_id_at_z < min_id_at_z_plus_1,
                "Max ID at zoom {} ({}) should be < min ID at zoom {} ({})",
                z,
                max_id_at_z,
                z + 1,
                min_id_at_z_plus_1
            );
        }
    }

    // -------------------------------------------------------------------------
    // Task 8: Directory Encoding Tests
    // -------------------------------------------------------------------------

    #[test]
    fn test_encode_varint_small_values() {
        // Values < 128 encode to single byte
        let mut buf = Vec::new();
        encode_varint(0, &mut buf);
        assert_eq!(buf, vec![0]);

        buf.clear();
        encode_varint(1, &mut buf);
        assert_eq!(buf, vec![1]);

        buf.clear();
        encode_varint(127, &mut buf);
        assert_eq!(buf, vec![127]);
    }

    #[test]
    fn test_encode_varint_128() {
        // 128 = 0x80 needs 2 bytes: [0x80, 0x01]
        let mut buf = Vec::new();
        encode_varint(128, &mut buf);
        assert_eq!(buf, vec![0x80, 0x01]);
    }

    #[test]
    fn test_encode_varint_300() {
        // 300 = 0x12C = 0b1_0010_1100
        // Low 7 bits: 0010_1100 = 0x2C, with continuation: 0xAC
        // High bits: 0000_0010 = 0x02
        let mut buf = Vec::new();
        encode_varint(300, &mut buf);
        assert_eq!(buf, vec![0xAC, 0x02]);
    }

    #[test]
    fn test_varint_roundtrip() {
        let test_values = [0u64, 1, 127, 128, 255, 256, 300, 16383, 16384, u64::MAX];

        for &value in &test_values {
            let mut buf = Vec::new();
            encode_varint(value, &mut buf);
            let (decoded, bytes_consumed) = decode_varint(&buf).expect("Should decode");
            assert_eq!(decoded, value, "Roundtrip failed for {}", value);
            assert_eq!(bytes_consumed, buf.len());
        }
    }

    #[test]
    fn test_encode_directory_empty() {
        let entries: Vec<DirEntry> = vec![];
        let encoded = encode_directory(&entries);
        // Should just be count = 0
        assert_eq!(encoded, vec![0]);
    }

    #[test]
    fn test_encode_directory_single_entry() {
        let entries = vec![DirEntry {
            tile_id: 1,
            offset: 0,
            length: 100,
            run_length: 1,
        }];
        let encoded = encode_directory(&entries);

        // Should start with count = 1
        assert!(!encoded.is_empty());
        assert_eq!(encoded[0], 1);
    }

    #[test]
    fn test_encode_directory_multiple_entries() {
        let entries = vec![
            DirEntry {
                tile_id: 5,
                offset: 0,
                length: 100,
                run_length: 1,
            },
            DirEntry {
                tile_id: 42,
                offset: 100,
                length: 200,
                run_length: 1,
            },
            DirEntry {
                tile_id: 69,
                offset: 300,
                length: 50,
                run_length: 1,
            },
        ];
        let encoded = encode_directory(&entries);

        // Should start with count = 3
        assert_eq!(encoded[0], 3);

        // The encoding should be smaller than naive (due to delta encoding)
        // Each entry would be ~24 bytes naive, but delta should compress
        assert!(encoded.len() < entries.len() * 24);
    }

    /// Three leaf pointers laid out the way tippecanoe / go-pmtiles write a
    /// root directory: every entry has run_length = 0 and every offset after
    /// the first is encoded as 0 ("contiguous with the previous entry").
    fn tippecanoe_style_leaf_root() -> Vec<u8> {
        let mut buf = Vec::new();
        encode_varint(3, &mut buf);
        // tile ids: 0, 1000, 2000 (delta-encoded)
        for delta in [0, 1000, 1000] {
            encode_varint(delta, &mut buf);
        }
        // run lengths: all 0 (leaf pointers)
        for _ in 0..3 {
            encode_varint(0, &mut buf);
        }
        // compressed leaf lengths
        for len in [100, 200, 50] {
            encode_varint(len, &mut buf);
        }
        // offsets: explicit 0 (stored as 0 + 1), then contiguous, contiguous
        for encoded_offset in [1, 0, 0] {
            encode_varint(encoded_offset, &mut buf);
        }
        buf
    }

    fn dir_tuples(entries: &[DirEntry]) -> Vec<(u64, u64, u32, u32)> {
        entries
            .iter()
            .map(|e| (e.tile_id, e.offset, e.length, e.run_length))
            .collect()
    }

    #[test]
    fn test_decode_directory_resolves_contiguous_leaf_offsets() {
        // Issue #377: the contiguous-offset rule applies to leaf pointers too.
        // Before the fix every leaf after the first decoded to offset 0, so
        // `decode` sliced the wrong bytes and gzip failed with
        // "incomplete deflate stream" on any tippecanoe archive with leaves.
        let entries = decode_directory(&tippecanoe_style_leaf_root()).unwrap();
        assert_eq!(
            dir_tuples(&entries),
            vec![(0, 0, 100, 0), (1000, 100, 200, 0), (2000, 300, 50, 0)]
        );
    }

    #[test]
    fn test_encode_directory_contiguous_leaf_entries_encode_as_zero() {
        // The encoder must apply the same rule, so our root directories are
        // as compact as the spec allows and round-trip through any reader.
        let entries = vec![
            DirEntry {
                tile_id: 0,
                offset: 0,
                length: 100,
                run_length: 0,
            },
            DirEntry {
                tile_id: 1000,
                offset: 100,
                length: 200,
                run_length: 0,
            },
            DirEntry {
                tile_id: 2000,
                offset: 300,
                length: 50,
                run_length: 0,
            },
        ];
        let encoded = encode_directory(&entries);
        assert_eq!(encoded, tippecanoe_style_leaf_root());
        assert_eq!(
            dir_tuples(&decode_directory(&encoded).unwrap()),
            dir_tuples(&entries)
        );
    }

    #[test]
    fn test_decode_directory_rejects_offset_that_overflows_the_accumulator() {
        // An explicit offset varint of u64::MAX decodes to u64::MAX - 1; the
        // contiguous accumulator (offset + length) must fail as a decode
        // error rather than overflow. Hand-encoded: two tile entries.
        let mut buf = Vec::new();
        encode_varint(2, &mut buf); // count
        encode_varint(0, &mut buf); // tile id 0
        encode_varint(1, &mut buf); // tile id 1 (delta)
        encode_varint(1, &mut buf); // run lengths
        encode_varint(1, &mut buf);
        encode_varint(5, &mut buf); // lengths
        encode_varint(5, &mut buf);
        encode_varint(u64::MAX, &mut buf); // offsets: hostile, then contiguous
        encode_varint(0, &mut buf);
        assert!(decode_directory(&buf).is_none());
    }

    #[test]
    fn test_gzip_compress_roundtrip() {
        use flate2::read::GzDecoder;
        use std::io::Read;

        let original = b"Hello, PMTiles! This is test data.";
        let compressed = gzip_compress(original).expect("Should compress");

        // Should be shorter than original (for non-trivial data)
        // Note: very small inputs might expand

        // Decompress and verify
        let mut decoder = GzDecoder::new(&compressed[..]);
        let mut decompressed = Vec::new();
        decoder
            .read_to_end(&mut decompressed)
            .expect("Should decompress");

        assert_eq!(decompressed, original);
    }

    // -------------------------------------------------------------------------
    // Task 9: Full Writer Tests
    // -------------------------------------------------------------------------

    #[test]
    fn test_writer_creation() {
        let writer = PmtilesWriter::new();
        assert_eq!(writer.tile_count(), 0);
    }

    #[test]
    fn test_writer_add_single_tile() {
        let mut writer = PmtilesWriter::new();
        let mvt_data = vec![0x1a, 0x00]; // Minimal MVT-like data

        writer.add_tile(0, 0, 0, &mvt_data).unwrap();
        assert_eq!(writer.tile_count(), 1);
    }

    #[test]
    fn test_writer_creates_valid_pmtiles_file() {
        let mut writer = PmtilesWriter::new();

        // Add a minimal tile
        let mvt_data = vec![0x1a, 0x00];
        writer.add_tile(0, 0, 0, &mvt_data).unwrap();
        writer.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));

        let path = Path::new("/tmp/test-pmtiles-writer.pmtiles");
        let _ = fs::remove_file(path);

        writer.write_to_file(path).expect("Should write file");

        // Verify file exists and has correct structure
        assert!(path.exists(), "File should exist");

        let data = fs::read(path).unwrap();

        // Check magic number and version
        assert_eq!(&data[0..7], b"PMTiles");
        assert_eq!(data[7], 3);

        // Check file is at least header size + some data
        assert!(data.len() > 127);

        // Check root directory offset points to position 127
        let root_offset = u64::from_le_bytes(data[8..16].try_into().unwrap());
        assert_eq!(root_offset, 127);

        // Clean up
        let _ = fs::remove_file(path);
    }

    #[test]
    fn test_writer_multiple_tiles_multiple_zooms() {
        let mut writer = PmtilesWriter::new();

        // Add tiles at zooms 0, 1, 2
        for z in 0..3u8 {
            let n = 1u32 << z;
            for x in 0..n {
                for y in 0..n {
                    let mvt_data = vec![0x1a, z, x as u8, y as u8];
                    writer.add_tile(z, x, y, &mvt_data).unwrap();
                }
            }
        }

        // Should have 1 + 4 + 16 = 21 tiles
        assert_eq!(writer.tile_count(), 21);

        writer.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));

        let path = Path::new("/tmp/test-pmtiles-multi.pmtiles");
        let _ = fs::remove_file(path);

        writer.write_to_file(path).expect("Should write file");

        // Verify basic structure
        let data = fs::read(path).unwrap();
        assert_eq!(&data[0..7], b"PMTiles");
        assert_eq!(data[7], 3);

        // Check tile counts in header
        let addressed_count = u64::from_le_bytes(data[72..80].try_into().unwrap());
        assert_eq!(addressed_count, 21);

        // Check zoom range
        assert_eq!(data[100], 0); // min_zoom
        assert_eq!(data[101], 2); // max_zoom

        // Clean up
        let _ = fs::remove_file(path);
    }

    #[test]
    fn test_writer_empty_tileset() {
        let writer = PmtilesWriter::new();

        let path = Path::new("/tmp/test-pmtiles-empty.pmtiles");
        let _ = fs::remove_file(path);

        writer.write_to_file(path).expect("Should write empty file");

        let data = fs::read(path).unwrap();
        assert_eq!(&data[0..7], b"PMTiles");

        // Clean up
        let _ = fs::remove_file(path);
    }

    #[test]
    fn test_writer_tile_ordering() {
        let mut writer = PmtilesWriter::new();

        // Add tiles in random order
        writer.add_tile(2, 3, 3, &[1, 2, 3]).unwrap();
        writer.add_tile(0, 0, 0, &[4, 5, 6]).unwrap();
        writer.add_tile(1, 1, 0, &[7, 8, 9]).unwrap();

        // BTreeMap should maintain Hilbert curve order
        assert_eq!(writer.tile_count(), 3);

        let path = Path::new("/tmp/test-pmtiles-ordering.pmtiles");
        let _ = fs::remove_file(path);

        writer.write_to_file(path).expect("Should write file");

        // Should succeed (clustered mode requires sorted tiles)
        assert!(path.exists());

        let _ = fs::remove_file(path);
    }

    #[test]
    fn test_writer_bounds_preserved() {
        let mut writer = PmtilesWriter::new();
        writer.add_tile(0, 0, 0, &[1, 2, 3]).unwrap();

        let bounds = TileBounds::new(-122.5, 37.7, -122.3, 37.9);
        writer.set_bounds(&bounds);

        let path = Path::new("/tmp/test-pmtiles-bounds.pmtiles");
        let _ = fs::remove_file(path);

        writer.write_to_file(path).expect("Should write file");

        let data = fs::read(path).unwrap();

        // Decode bounds from header
        let decode_coord = |offset: usize| -> f64 {
            let val = i32::from_le_bytes(data[offset..offset + 4].try_into().unwrap());
            val as f64 / 10_000_000.0
        };

        let min_lon = decode_coord(102);
        let min_lat = decode_coord(106);
        let max_lon = decode_coord(110);
        let max_lat = decode_coord(114);

        assert!((min_lon - bounds.lng_min).abs() < 0.0001);
        assert!((min_lat - bounds.lat_min).abs() < 0.0001);
        assert!((max_lon - bounds.lng_max).abs() < 0.0001);
        assert!((max_lat - bounds.lat_max).abs() < 0.0001);

        let _ = fs::remove_file(path);
    }

    // -------------------------------------------------------------------------
    // Field Metadata Tests
    // -------------------------------------------------------------------------

    #[test]
    fn test_build_fields_json_empty() {
        let writer = PmtilesWriter::new();
        assert_eq!(writer.build_fields_json(), "{}");
    }

    #[test]
    fn test_build_fields_json_with_fields() {
        let mut writer = PmtilesWriter::new();
        let mut fields = HashMap::new();
        fields.insert("name".to_string(), "String".to_string());
        fields.insert("area".to_string(), "Number".to_string());
        writer.set_fields(fields);

        let json = writer.build_fields_json();
        // Fields are sorted alphabetically
        assert_eq!(json, r#"{"area":"Number","name":"String"}"#);
    }

    #[test]
    fn test_writer_field_metadata_in_output() {
        use flate2::read::GzDecoder;
        use std::io::Read;

        let mut writer = PmtilesWriter::new();
        writer.add_tile(0, 0, 0, &[1, 2, 3]).unwrap();
        writer.set_layer_name("buildings");

        let mut fields = HashMap::new();
        fields.insert("name".to_string(), "String".to_string());
        fields.insert("height".to_string(), "Number".to_string());
        writer.set_fields(fields);

        let path = Path::new("/tmp/test-pmtiles-fields.pmtiles");
        let _ = fs::remove_file(path);

        writer.write_to_file(path).expect("Should write file");

        let data = fs::read(path).unwrap();

        // Extract metadata offset and length from header
        let metadata_offset = u64::from_le_bytes(data[24..32].try_into().unwrap()) as usize;
        let metadata_length = u64::from_le_bytes(data[32..40].try_into().unwrap()) as usize;

        // Decompress the metadata
        let compressed_metadata = &data[metadata_offset..metadata_offset + metadata_length];
        let mut decoder = GzDecoder::new(compressed_metadata);
        let mut metadata_json = String::new();
        decoder
            .read_to_string(&mut metadata_json)
            .expect("Should decompress metadata");

        // Verify fields are present
        assert!(metadata_json.contains(r#""height":"Number""#));
        assert!(metadata_json.contains(r#""name":"String""#));
        assert!(metadata_json.contains(r#""id":"buildings""#));

        let _ = fs::remove_file(path);
    }

    /// #380: an archive built for z0..z4 whose coarsest levels hold no tiles
    /// must still say z0 in the header and in `vector_layers`, or a client
    /// configured for the requested range never asks for the zoomed-out view.
    #[test]
    fn streaming_writer_declared_min_zoom_widens_header_and_layer_range() {
        use flate2::read::GzDecoder;
        use std::io::Read;

        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("declared.pmtiles");
        let mut writer = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        writer.set_layer_name("t");
        writer.set_declared_min_zoom(0);
        writer.add_tile(2, 1, 1, &[0x1a, 0x00]).unwrap();
        writer.add_tile(4, 5, 5, &[0x1a, 0x01]).unwrap();
        writer.finalize(&path).unwrap();

        let data = fs::read(&path).unwrap();
        let header = Header::from_bytes(&data[..127]).unwrap();
        assert_eq!(header.min_zoom, 0, "declared minimum wins over observed z2");
        assert_eq!(header.max_zoom, 4);

        let start = header.json_metadata_offset as usize;
        let end = start + header.json_metadata_length as usize;
        let mut json = String::new();
        GzDecoder::new(&data[start..end])
            .read_to_string(&mut json)
            .unwrap();
        assert!(
            json.contains(r#""minzoom":0"#),
            "vector_layers must advertise the declared minimum: {json}"
        );
    }

    /// A declared minimum finer than the coarsest tile present cannot narrow
    /// the range: the tiles are there, the header must cover them.
    #[test]
    fn streaming_writer_declared_min_zoom_never_hides_written_tiles() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("declared-narrow.pmtiles");
        let mut writer = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        writer.set_layer_name("t");
        writer.set_declared_min_zoom(3);
        writer.add_tile(2, 1, 1, &[0x1a, 0x00]).unwrap();
        writer.finalize(&path).unwrap();
        let data = fs::read(&path).unwrap();
        assert_eq!(Header::from_bytes(&data[..127]).unwrap().min_zoom, 2);
    }

    /// A declared minimum over an archive that got no tiles at all must not
    /// outrun the zero max zoom the empty archive collapses to: the header
    /// stays z0..z0 (min <= max), as it was before declared minimums existed.
    #[test]
    fn streaming_writer_declared_min_zoom_over_zero_tiles_stays_z0() {
        use flate2::read::GzDecoder;
        use std::io::Read;

        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("declared-empty.pmtiles");
        let mut writer = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        writer.set_layer_name("t");
        writer.set_declared_min_zoom(3);
        writer.finalize(&path).unwrap();

        let data = fs::read(&path).unwrap();
        let header = Header::from_bytes(&data[..127]).unwrap();
        assert_eq!(header.min_zoom, 0, "empty archive is z0..z0, not z3..z0");
        assert_eq!(header.max_zoom, 0);
        assert_eq!(header.center_zoom, 0);

        let start = header.json_metadata_offset as usize;
        let end = start + header.json_metadata_length as usize;
        let mut json = String::new();
        GzDecoder::new(&data[start..end])
            .read_to_string(&mut json)
            .unwrap();
        assert!(
            json.contains(r#""minzoom":0"#),
            "vector_layers of an empty archive stay at minzoom 0: {json}"
        );
    }

    // -------------------------------------------------------------------------
    // Compression Configuration Tests
    // -------------------------------------------------------------------------

    #[test]
    fn test_writer_with_compression_constructor() {
        let writer = PmtilesWriter::with_compression(Compression::Brotli);
        assert_eq!(writer.tile_compression(), Compression::Brotli);
        assert_eq!(writer.internal_compression(), Compression::Brotli);
    }

    #[test]
    fn test_writer_set_compression() {
        let mut writer = PmtilesWriter::new();
        assert_eq!(writer.tile_compression(), Compression::Gzip); // default

        writer.set_tile_compression(Compression::Zstd);
        assert_eq!(writer.tile_compression(), Compression::Zstd);

        writer.set_internal_compression(Compression::Brotli);
        assert_eq!(writer.internal_compression(), Compression::Brotli);
    }

    #[test]
    fn test_writer_brotli_compression() {
        let mut writer = PmtilesWriter::with_compression(Compression::Brotli);
        let mvt_data = vec![0x1a; 100]; // Compressible data

        writer.add_tile(0, 0, 0, &mvt_data).unwrap();
        writer.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));

        let path = Path::new("/tmp/test-pmtiles-brotli.pmtiles");
        let _ = fs::remove_file(path);

        writer
            .write_to_file(path)
            .expect("Should write file with brotli");

        let data = fs::read(path).unwrap();

        // Verify header
        assert_eq!(&data[0..7], b"PMTiles");
        assert_eq!(data[7], 3);

        // Check compression bytes in header (97 = internal, 98 = tile)
        assert_eq!(data[97], Compression::Brotli as u8);
        assert_eq!(data[98], Compression::Brotli as u8);

        let _ = fs::remove_file(path);
    }

    #[test]
    fn test_writer_zstd_compression() {
        let mut writer = PmtilesWriter::with_compression(Compression::Zstd);
        let mvt_data = vec![0x1a; 100];

        writer.add_tile(0, 0, 0, &mvt_data).unwrap();
        writer.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));

        let path = Path::new("/tmp/test-pmtiles-zstd.pmtiles");
        let _ = fs::remove_file(path);

        writer
            .write_to_file(path)
            .expect("Should write file with zstd");

        let data = fs::read(path).unwrap();

        // Check compression bytes in header
        assert_eq!(data[97], Compression::Zstd as u8);
        assert_eq!(data[98], Compression::Zstd as u8);

        let _ = fs::remove_file(path);
    }

    #[test]
    fn test_writer_no_compression() {
        let mut writer = PmtilesWriter::with_compression(Compression::None);
        let mvt_data = vec![0x1a, 0x00];

        writer.add_tile(0, 0, 0, &mvt_data).unwrap();
        writer.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));

        let path = Path::new("/tmp/test-pmtiles-none.pmtiles");
        let _ = fs::remove_file(path);

        writer
            .write_to_file(path)
            .expect("Should write file without compression");

        let data = fs::read(path).unwrap();

        // Check compression bytes in header
        assert_eq!(data[97], Compression::None as u8);
        assert_eq!(data[98], Compression::None as u8);

        let _ = fs::remove_file(path);
    }

    #[test]
    fn test_writer_mixed_compression() {
        // Test different compression for internal vs tile data
        let mut writer = PmtilesWriter::new();
        writer.set_internal_compression(Compression::Gzip);
        writer.set_tile_compression(Compression::Zstd);

        let mvt_data = vec![0x1a; 100];
        writer.add_tile(0, 0, 0, &mvt_data).unwrap();
        writer.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));

        let path = Path::new("/tmp/test-pmtiles-mixed.pmtiles");
        let _ = fs::remove_file(path);

        writer
            .write_to_file(path)
            .expect("Should write file with mixed compression");

        let data = fs::read(path).unwrap();

        // Check compression bytes in header
        assert_eq!(data[97], Compression::Gzip as u8); // internal
        assert_eq!(data[98], Compression::Zstd as u8); // tile

        let _ = fs::remove_file(path);
    }

    // -------------------------------------------------------------------------
    // Tile Deduplication Tests
    // -------------------------------------------------------------------------

    #[test]
    fn test_writer_dedup_identical_tiles() {
        let mut writer = PmtilesWriter::new();
        writer.enable_deduplication(true);

        // Add 3 identical tiles at consecutive positions
        let ocean_tile = vec![0x1a, 0x00]; // Same content
        writer.add_tile(1, 0, 0, &ocean_tile).unwrap();
        writer.add_tile(1, 0, 1, &ocean_tile).unwrap();
        writer.add_tile(1, 1, 1, &ocean_tile).unwrap();

        // 3 tiles addressed, but only 1 unique content
        assert_eq!(writer.tile_count(), 3);

        let stats = writer.dedup_stats();
        assert_eq!(stats.total_tiles, 3);
        assert_eq!(stats.unique_tiles, 1);
        assert_eq!(stats.duplicates_eliminated, 2);
    }

    #[test]
    fn test_writer_dedup_mixed_tiles() {
        let mut writer = PmtilesWriter::new();
        writer.enable_deduplication(true);

        // Add tiles: A, A, B, A, B, B
        let tile_a = vec![0x1a, 0x01];
        let tile_b = vec![0x1a, 0x02];

        writer.add_tile(0, 0, 0, &tile_a).unwrap();
        writer.add_tile(1, 0, 0, &tile_a).unwrap(); // dup
        writer.add_tile(1, 0, 1, &tile_b).unwrap();
        writer.add_tile(1, 1, 1, &tile_a).unwrap(); // dup
        writer.add_tile(1, 1, 0, &tile_b).unwrap(); // dup
        writer.add_tile(2, 0, 0, &tile_b).unwrap(); // dup

        let stats = writer.dedup_stats();
        assert_eq!(stats.total_tiles, 6);
        assert_eq!(stats.unique_tiles, 2);
        assert_eq!(stats.duplicates_eliminated, 4);
    }

    #[test]
    fn test_writer_dedup_disabled_by_default() {
        let writer = PmtilesWriter::new();
        // Deduplication should be disabled by default for backward compatibility
        assert!(!writer.is_dedup_enabled());
    }

    #[test]
    fn test_writer_dedup_file_size_reduction() {
        // Test with deduplication
        let mut writer_dedup = PmtilesWriter::new();
        writer_dedup.enable_deduplication(true);

        let ocean_tile = vec![0x1a, 0x00, 0x01, 0x02, 0x03, 0x04, 0x05];
        for z in 0..3u8 {
            let n = 1u32 << z;
            for x in 0..n {
                for y in 0..n {
                    writer_dedup.add_tile(z, x, y, &ocean_tile).unwrap();
                }
            }
        }

        let path_dedup = Path::new("/tmp/test-pmtiles-dedup-enabled.pmtiles");
        let _ = fs::remove_file(path_dedup);
        writer_dedup.write_to_file(path_dedup).unwrap();
        let size_dedup = fs::metadata(path_dedup).unwrap().len();

        // Test without deduplication
        let mut writer_no_dedup = PmtilesWriter::new();
        // Dedup disabled by default

        for z in 0..3u8 {
            let n = 1u32 << z;
            for x in 0..n {
                for y in 0..n {
                    writer_no_dedup.add_tile(z, x, y, &ocean_tile).unwrap();
                }
            }
        }

        let path_no_dedup = Path::new("/tmp/test-pmtiles-dedup-disabled.pmtiles");
        let _ = fs::remove_file(path_no_dedup);
        writer_no_dedup.write_to_file(path_no_dedup).unwrap();
        let size_no_dedup = fs::metadata(path_no_dedup).unwrap().len();

        // Deduplicated file should be smaller
        assert!(
            size_dedup < size_no_dedup,
            "Deduplicated file ({} bytes) should be smaller than non-deduplicated ({} bytes)",
            size_dedup,
            size_no_dedup
        );

        let _ = fs::remove_file(path_dedup);
        let _ = fs::remove_file(path_no_dedup);
    }

    #[test]
    fn test_writer_dedup_run_length_consecutive() {
        let mut writer = PmtilesWriter::new();
        writer.enable_deduplication(true);

        // Add consecutive tiles with same content (should use run_length)
        let tile = vec![0x1a, 0x10];
        // Zoom 1 tiles: IDs 1, 2, 3, 4 (consecutive in Hilbert order)
        writer.add_tile(1, 0, 0, &tile).unwrap(); // ID 1
        writer.add_tile(1, 0, 1, &tile).unwrap(); // ID 2
        writer.add_tile(1, 1, 1, &tile).unwrap(); // ID 3
        writer.add_tile(1, 1, 0, &tile).unwrap(); // ID 4

        let path = Path::new("/tmp/test-pmtiles-runlength.pmtiles");
        let _ = fs::remove_file(path);
        writer.write_to_file(path).unwrap();

        let data = fs::read(path).unwrap();

        // Verify header counts
        let addressed_count = u64::from_le_bytes(data[72..80].try_into().unwrap());
        let entries_count = u64::from_le_bytes(data[80..88].try_into().unwrap());
        let contents_count = u64::from_le_bytes(data[88..96].try_into().unwrap());

        // 4 tiles addressed
        assert_eq!(addressed_count, 4);
        // But only 1 directory entry (run_length = 4)
        assert_eq!(entries_count, 1);
        // And only 1 unique content
        assert_eq!(contents_count, 1);

        let _ = fs::remove_file(path);
    }

    #[test]
    fn test_writer_dedup_header_stats() {
        let mut writer = PmtilesWriter::new();
        writer.enable_deduplication(true);

        // 10 tiles, 3 unique contents
        let tile_a = vec![0x1a, 0x01];
        let tile_b = vec![0x1a, 0x02];
        let tile_c = vec![0x1a, 0x03];

        // Pattern: A A A B B C A B C C
        for _ in 0..3 {
            writer.add_tile(0, 0, 0, &tile_a).unwrap(); // Will deduplicate
        }
        // Only first A is added at z=0, rest are different coords
        // Actually, let me use different coords properly:
        let mut writer2 = PmtilesWriter::new();
        writer2.enable_deduplication(true);

        // Add 10 tiles with 3 unique contents at zoom 0-2
        writer2.add_tile(0, 0, 0, &tile_a).unwrap();
        writer2.add_tile(1, 0, 0, &tile_a).unwrap(); // dup A
        writer2.add_tile(1, 0, 1, &tile_a).unwrap(); // dup A
        writer2.add_tile(1, 1, 1, &tile_b).unwrap();
        writer2.add_tile(1, 1, 0, &tile_b).unwrap(); // dup B
        writer2.add_tile(2, 0, 0, &tile_c).unwrap();
        writer2.add_tile(2, 0, 1, &tile_a).unwrap(); // dup A
        writer2.add_tile(2, 1, 0, &tile_b).unwrap(); // dup B
        writer2.add_tile(2, 1, 1, &tile_c).unwrap(); // dup C
        writer2.add_tile(2, 2, 0, &tile_c).unwrap(); // dup C

        let path = Path::new("/tmp/test-pmtiles-header-stats.pmtiles");
        let _ = fs::remove_file(path);
        writer2.write_to_file(path).unwrap();

        let data = fs::read(path).unwrap();

        let addressed_count = u64::from_le_bytes(data[72..80].try_into().unwrap());
        let contents_count = u64::from_le_bytes(data[88..96].try_into().unwrap());

        assert_eq!(addressed_count, 10, "Should address 10 tiles");
        assert_eq!(contents_count, 3, "Should have 3 unique contents");

        let _ = fs::remove_file(path);
    }

    // =========================================================================
    // StreamingPmtilesWriter Tests (TDD)
    // =========================================================================

    #[test]
    fn test_streaming_writer_creates_temp_file() {
        let writer =
            StreamingPmtilesWriter::new(Compression::Gzip).expect("Should create streaming writer");

        // Temp file should exist
        assert!(
            writer.temp_path().exists(),
            "Temp file should be created at {:?}",
            writer.temp_path()
        );

        // Clean up happens on drop
        let temp_path = writer.temp_path().to_path_buf();
        drop(writer);
        assert!(
            !temp_path.exists(),
            "Temp file should be cleaned up on drop"
        );
    }

    #[test]
    fn test_streaming_writer_add_tile_writes_to_temp() {
        let mut writer =
            StreamingPmtilesWriter::new(Compression::Gzip).expect("Should create streaming writer");

        // Add a tile
        let mvt_data = vec![0x1a, 0x00, 0x01, 0x02];
        writer.add_tile(0, 0, 0, &mvt_data).unwrap();

        // Stats should reflect the write
        let stats = writer.stats();
        assert_eq!(stats.total_tiles, 1);
        assert_eq!(stats.unique_tiles, 1);
        assert!(
            stats.bytes_written > 0,
            "Should have written bytes to temp file"
        );

        // Note: The BufWriter may not have flushed to disk yet, so we check our
        // internal stats rather than file metadata (which requires flush)
        assert_eq!(
            stats.bytes_written, writer.current_offset,
            "bytes_written should match current_offset"
        );
    }

    #[test]
    fn test_streaming_writer_dedup_same_content() {
        let mut writer =
            StreamingPmtilesWriter::new(Compression::Gzip).expect("Should create streaming writer");

        // Add 3 tiles with identical content
        let ocean_tile = vec![0x1a, 0x00];
        writer.add_tile(1, 0, 0, &ocean_tile).unwrap();
        writer.add_tile(1, 0, 1, &ocean_tile).unwrap();
        writer.add_tile(1, 1, 1, &ocean_tile).unwrap();

        let stats = writer.stats();
        assert_eq!(stats.total_tiles, 3, "Should track 3 total tiles");
        assert_eq!(stats.unique_tiles, 1, "Should only have 1 unique tile");
        assert!(
            stats.bytes_saved_dedup > 0,
            "Should have saved bytes via deduplication"
        );
    }

    #[test]
    fn test_streaming_writer_finalize_creates_valid_pmtiles() {
        let mut writer =
            StreamingPmtilesWriter::new(Compression::Gzip).expect("Should create streaming writer");

        // Add a few tiles
        writer.add_tile(0, 0, 0, &[0x1a, 0x00]).unwrap();
        writer.add_tile(1, 0, 0, &[0x1a, 0x01]).unwrap();
        writer.add_tile(1, 0, 1, &[0x1a, 0x02]).unwrap();
        writer.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));

        let output_path = Path::new("/tmp/test-streaming-pmtiles.pmtiles");
        let _ = fs::remove_file(output_path);

        let stats = writer.finalize(output_path).expect("Should finalize");

        // Verify file was created with valid PMTiles structure
        assert!(output_path.exists(), "Output file should exist");

        let data = fs::read(output_path).unwrap();

        // Check magic and version
        assert_eq!(&data[0..7], b"PMTiles", "Should have PMTiles magic");
        assert_eq!(data[7], 3, "Should be version 3");

        // Check tile counts in header
        let addressed_count = u64::from_le_bytes(data[72..80].try_into().unwrap());
        assert_eq!(addressed_count, 3, "Should have 3 addressed tiles");

        // Verify stats
        assert_eq!(stats.total_tiles, 3);
        assert_eq!(stats.unique_tiles, 3); // All different content

        // Clean up
        let _ = fs::remove_file(output_path);
    }

    #[test]
    fn test_streaming_writer_memory_bounded() {
        let mut writer =
            StreamingPmtilesWriter::new(Compression::Gzip).expect("Should create streaming writer");

        // Add many tiles (simulating a large file scenario)
        // Even with 1000 tiles, memory should stay low
        // Use valid coordinates for each zoom level
        let mut count = 0;
        for z in 0..10u8 {
            let max_coord = 1u32 << z; // Valid range: 0 to max_coord-1
            for x in 0..max_coord.min(10) {
                for y in 0..max_coord.min(10) {
                    let data = vec![0x1a, z, (x & 0xFF) as u8, (y & 0xFF) as u8, count as u8];
                    writer.add_tile(z, x, y, &data).unwrap();
                    count += 1;
                    if count >= 1000 {
                        break;
                    }
                }
                if count >= 1000 {
                    break;
                }
            }
            if count >= 1000 {
                break;
            }
        }

        let stats = writer.stats();

        // Memory estimate should be bounded: ~64 bytes per entry (24 dir + 40 dedup)
        // 1000 tiles × 64 bytes = ~64KB (not 40MB of tile data)
        let estimated_mem = stats.estimated_memory_bytes();
        assert!(
            estimated_mem < 200_000, // Less than 200KB
            "Memory usage should be bounded, got {} bytes",
            estimated_mem
        );

        // Clean up (finalize not needed for this test)
    }

    #[test]
    fn test_streaming_writer_matches_non_streaming_output() {
        // Create identical content with both writers and compare output
        let tiles_data = vec![
            (0, 0, 0, vec![0x1a, 0x00]),
            (1, 0, 0, vec![0x1a, 0x01]),
            (1, 0, 1, vec![0x1a, 0x02]),
            (1, 1, 1, vec![0x1a, 0x00]), // Duplicate content
        ];

        // Non-streaming writer (with dedup enabled for fair comparison)
        let mut non_streaming = PmtilesWriter::with_compression(Compression::Gzip);
        non_streaming.enable_deduplication(true);
        non_streaming.set_layer_name("test");
        non_streaming.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));
        for (z, x, y, data) in &tiles_data {
            non_streaming.add_tile(*z, *x, *y, data).unwrap();
        }

        let non_streaming_path = Path::new("/tmp/test-compare-non-streaming.pmtiles");
        let _ = fs::remove_file(non_streaming_path);
        non_streaming.write_to_file(non_streaming_path).unwrap();

        // Streaming writer
        let mut streaming = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        streaming.set_layer_name("test");
        streaming.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));
        for (z, x, y, data) in &tiles_data {
            streaming.add_tile(*z, *x, *y, data).unwrap();
        }

        let streaming_path = Path::new("/tmp/test-compare-streaming.pmtiles");
        let _ = fs::remove_file(streaming_path);
        streaming.finalize(streaming_path).unwrap();

        // Compare key header fields
        let ns_data = fs::read(non_streaming_path).unwrap();
        let s_data = fs::read(streaming_path).unwrap();

        // Magic and version should match
        assert_eq!(
            &ns_data[0..8],
            &s_data[0..8],
            "Header magic/version should match"
        );

        // Addressed tiles count should match
        let ns_addressed = u64::from_le_bytes(ns_data[72..80].try_into().unwrap());
        let s_addressed = u64::from_le_bytes(s_data[72..80].try_into().unwrap());
        assert_eq!(ns_addressed, s_addressed, "Addressed tiles should match");

        // Unique contents should match
        let ns_contents = u64::from_le_bytes(ns_data[88..96].try_into().unwrap());
        let s_contents = u64::from_le_bytes(s_data[88..96].try_into().unwrap());
        assert_eq!(ns_contents, s_contents, "Unique contents should match");

        // Clean up
        let _ = fs::remove_file(non_streaming_path);
        let _ = fs::remove_file(streaming_path);
    }

    #[test]
    fn test_streaming_writer_with_feature_count() {
        use flate2::read::GzDecoder;
        use std::io::Read;

        let mut writer = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        writer.set_layer_name("buildings");

        // Add tiles with feature counts
        writer
            .add_tile_with_count(0, 0, 0, &[0x1a, 0x00], 100)
            .unwrap();
        writer
            .add_tile_with_count(1, 0, 0, &[0x1a, 0x01], 50)
            .unwrap();
        writer
            .add_tile_with_count(1, 0, 1, &[0x1a, 0x02], 75)
            .unwrap();

        let output_path = Path::new("/tmp/test-streaming-features.pmtiles");
        let _ = fs::remove_file(output_path);
        writer.finalize(output_path).unwrap();

        let data = fs::read(output_path).unwrap();

        // Extract metadata and check tilestats
        let metadata_offset = u64::from_le_bytes(data[24..32].try_into().unwrap()) as usize;
        let metadata_length = u64::from_le_bytes(data[32..40].try_into().unwrap()) as usize;
        let compressed_metadata = &data[metadata_offset..metadata_offset + metadata_length];

        let mut decoder = GzDecoder::new(compressed_metadata);
        let mut metadata_json = String::new();
        decoder.read_to_string(&mut metadata_json).unwrap();

        // Should have tilestats with total feature count
        assert!(
            metadata_json.contains("\"count\":225"),
            "Should have total feature count 225, got: {}",
            metadata_json
        );

        let _ = fs::remove_file(output_path);
    }

    // -------------------------------------------------------------------------
    // Leaf Directory Tests (Issue #88)
    // -------------------------------------------------------------------------

    /// PMTiles initial HTTP range request size (16KB)
    const INITIAL_FETCH_SIZE: usize = 16384;
    /// PMTiles header size
    const HEADER_SIZE: usize = 127;
    /// Maximum root directory size that fits in initial fetch
    const MAX_ROOT_DIR_SIZE: usize = INITIAL_FETCH_SIZE - HEADER_SIZE;

    #[test]
    fn test_large_archive_uses_leaf_directories() {
        // Create enough tiles to exceed 16KB root directory
        // gzip compresses directory entries very well (~2-5 bytes/entry compressed)
        // Need 10,000+ entries to reliably exceed the 16KB threshold
        let num_tiles = 10_000;

        let mut writer = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        writer.set_layer_name("test");
        writer.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));

        // Add many tiles at zoom 12 (distributed across tile space)
        // Using higher zoom means larger tile_ids = less compressible
        for i in 0..num_tiles {
            let x = i % 4096;
            let y = i / 4096;
            let data = vec![0x1a, (i & 0xff) as u8, ((i >> 8) & 0xff) as u8];
            writer.add_tile(12, x as u32, y as u32, &data).unwrap();
        }

        let output_path = Path::new("/tmp/test-leaf-directories.pmtiles");
        let _ = fs::remove_file(output_path);
        writer.finalize(output_path).unwrap();

        // Read the header and verify leaf directories are used
        let data = fs::read(output_path).unwrap();

        // Extract header fields
        let root_dir_length = u64::from_le_bytes(data[16..24].try_into().unwrap()) as usize;
        let leaf_dirs_offset = u64::from_le_bytes(data[40..48].try_into().unwrap());
        let leaf_dirs_length = u64::from_le_bytes(data[48..56].try_into().unwrap());

        // Debug output
        eprintln!(
            "Archive stats: root_dir_length={}, leaf_dirs_offset={}, leaf_dirs_length={}, MAX={}",
            root_dir_length, leaf_dirs_offset, leaf_dirs_length, MAX_ROOT_DIR_SIZE
        );

        // Root directory must fit in initial fetch (16KB - 127 byte header)
        assert!(
            root_dir_length <= MAX_ROOT_DIR_SIZE,
            "Root directory ({} bytes) must fit in initial fetch ({} bytes)",
            root_dir_length,
            MAX_ROOT_DIR_SIZE
        );

        // With 10,000 tiles, we MUST have leaf directories
        assert!(
            leaf_dirs_offset > 0,
            "Large archive should have leaf directories (offset={})",
            leaf_dirs_offset
        );
        assert!(
            leaf_dirs_length > 0,
            "Large archive should have leaf directories (length={})",
            leaf_dirs_length
        );

        let _ = fs::remove_file(output_path);
    }

    #[test]
    fn test_small_archive_no_leaf_directories() {
        // Small archive should NOT use leaf directories (they're overhead)
        let mut writer = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        writer.set_layer_name("test");

        // Add just a few tiles
        for i in 0..10 {
            let data = vec![0x1a, i as u8];
            writer.add_tile(0, 0, 0, &data).unwrap();
        }

        let output_path = Path::new("/tmp/test-no-leaf-directories.pmtiles");
        let _ = fs::remove_file(output_path);
        writer.finalize(output_path).unwrap();

        let data = fs::read(output_path).unwrap();
        let leaf_dirs_offset = u64::from_le_bytes(data[40..48].try_into().unwrap());
        let leaf_dirs_length = u64::from_le_bytes(data[48..56].try_into().unwrap());

        // A small archive has no leaf directories...
        assert_eq!(
            leaf_dirs_length, 0,
            "Small archive should not have leaf directories"
        );
        // ...but its OFFSET must still point at the (empty) leaf section rather
        // than being zeroed. This test previously asserted 0, which is what
        // made every small archive this writer produced fail go-pmtiles:
        //
        //     Failed to verify archive, Leaf directories offset=0 must not be 0
        //
        // The section sits between the metadata and the tile data, so with no
        // leaves it is an empty span at the end of the metadata -- which is
        // also where the tile data begins.
        let metadata_offset = u64::from_le_bytes(data[24..32].try_into().unwrap());
        let metadata_length = u64::from_le_bytes(data[32..40].try_into().unwrap());
        let tile_data_offset = u64::from_le_bytes(data[56..64].try_into().unwrap());
        assert_ne!(
            leaf_dirs_offset, 0,
            "leaf_dirs_offset must never be 0 -- go-pmtiles rejects the archive"
        );
        assert_eq!(leaf_dirs_offset, metadata_offset + metadata_length);
        assert_eq!(leaf_dirs_offset, tile_data_offset);

        let _ = fs::remove_file(output_path);
    }

    #[test]
    fn test_leaf_directory_entries_have_run_length_zero() {
        // When leaf directories are used, root entries pointing to them
        // must have run_length = 0 (per PMTiles spec)
        let num_tiles = 3000;

        let mut writer = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        writer.set_layer_name("test");

        for i in 0..num_tiles {
            let x = i % 1024;
            let y = i / 1024;
            let data = vec![0x1a, (i & 0xff) as u8];
            writer.add_tile(10, x as u32, y as u32, &data).unwrap();
        }

        let output_path = Path::new("/tmp/test-leaf-run-length.pmtiles");
        let _ = fs::remove_file(output_path);
        writer.finalize(output_path).unwrap();

        let data = fs::read(output_path).unwrap();

        // Extract and decompress root directory
        let root_dir_offset = u64::from_le_bytes(data[8..16].try_into().unwrap()) as usize;
        let root_dir_length = u64::from_le_bytes(data[16..24].try_into().unwrap()) as usize;
        let leaf_dirs_length = u64::from_le_bytes(data[48..56].try_into().unwrap());

        // Only check if we have leaf directories
        if leaf_dirs_length > 0 {
            let compressed_root = &data[root_dir_offset..root_dir_offset + root_dir_length];

            use flate2::read::GzDecoder;
            use std::io::Read;
            let mut decoder = GzDecoder::new(compressed_root);
            let mut decompressed = Vec::new();
            decoder.read_to_end(&mut decompressed).unwrap();

            // Decode directory to verify run_length = 0 for leaf pointers
            let entries = decode_directory(&decompressed).unwrap();

            // All entries in root should be leaf pointers (run_length = 0)
            for entry in &entries {
                assert_eq!(
                    entry.run_length, 0,
                    "Root directory entries pointing to leaves must have run_length=0, got {}",
                    entry.run_length
                );
            }
        }

        let _ = fs::remove_file(output_path);
    }

    #[test]
    fn add_tile_precompressed_matches_add_tile_with_count() {
        // Issue #227 moves gzip off the serial writer thread: the export loop
        // now compresses tiles inside the parallel encode section and hands the
        // finished bytes (plus the *uncompressed* hash) to
        // add_tile_precompressed. The resulting archive must be byte-identical
        // to compressing serially via add_tile_with_count, including dedup: the
        // 4th tile below repeats the 1st tile's content.
        let tiles: Vec<(u8, u32, u32, Vec<u8>)> = vec![
            (0, 0, 0, vec![0x1a, 0x05, b'h', b'e', b'l', b'l', b'o']),
            (1, 0, 0, vec![0x1a, 0x03, b'a', b'b', b'c']),
            (1, 0, 1, vec![0x1a, 0x03, b'x', b'y', b'z']),
            (1, 1, 1, vec![0x1a, 0x05, b'h', b'e', b'l', b'l', b'o']), // dup of (0,0,0)
        ];
        let dir = std::env::temp_dir();

        // Baseline: writer compresses each tile serially.
        let mut serial = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        serial.set_layer_name("test");
        serial.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));
        for (z, x, y, data) in &tiles {
            serial.add_tile_with_count(*z, *x, *y, data, 1).unwrap();
        }
        let serial_path = dir.join("gpq-227-serial.pmtiles");
        let _ = fs::remove_file(&serial_path);
        serial.finalize(&serial_path).unwrap();

        // New path: compress up front, hand bytes + uncompressed hash to writer.
        let mut parallel = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        parallel.set_layer_name("test");
        parallel.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));
        for (z, x, y, data) in &tiles {
            let hash = TileHasher::hash(data);
            let compressed = compression::compress(data, Compression::Gzip).unwrap();
            parallel
                .add_tile_precompressed(*z, *x, *y, hash, &compressed, data.len(), 1)
                .unwrap();
        }
        let parallel_path = dir.join("gpq-227-parallel.pmtiles");
        let _ = fs::remove_file(&parallel_path);
        parallel.finalize(&parallel_path).unwrap();

        let a = fs::read(&serial_path).unwrap();
        let b = fs::read(&parallel_path).unwrap();
        assert_eq!(
            a, b,
            "precompressed archive must be byte-identical to serial compression"
        );

        let _ = fs::remove_file(&serial_path);
        let _ = fs::remove_file(&parallel_path);
    }

    #[test]
    fn checkpoint_produces_valid_capped_pmtiles() {
        // Issue #229: a mid-run checkpoint must produce a fully valid PMTiles
        // archive capped at the zooms written so far, WITHOUT consuming the
        // writer — the export loop keeps adding finer levels afterwards.
        let mut writer = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        writer.set_layer_name("test");
        writer.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));

        // Two coarse levels finished (z0: 1 tile, z1: 2 tiles).
        writer.add_tile(0, 0, 0, &[0x1a, 0x00]).unwrap();
        writer.add_tile(1, 0, 0, &[0x1a, 0x01]).unwrap();
        writer.add_tile(1, 0, 1, &[0x1a, 0x02]).unwrap();

        let ckpt_path = std::env::temp_dir().join("gpq-229-checkpoint.pmtiles");
        let _ = fs::remove_file(&ckpt_path);
        writer
            .checkpoint(&ckpt_path)
            .expect("checkpoint should succeed");

        // Checkpoint is a valid, capped archive.
        let ck = fs::read(&ckpt_path).unwrap();
        assert_eq!(&ck[0..7], b"PMTiles", "checkpoint has PMTiles magic");
        assert_eq!(ck[7], 3, "checkpoint is version 3");
        assert_eq!(ck[100], 0, "checkpoint min_zoom == 0");
        assert_eq!(
            ck[101], 1,
            "checkpoint max_zoom capped at last finished level"
        );
        let ck_addressed = u64::from_le_bytes(ck[72..80].try_into().unwrap());
        assert_eq!(ck_addressed, 3, "checkpoint holds the 3 finished tiles");

        // Writer is still usable: add a finer level and finalize.
        writer.add_tile(2, 0, 0, &[0x1a, 0x03]).unwrap();
        let final_path = std::env::temp_dir().join("gpq-229-checkpoint-final.pmtiles");
        let _ = fs::remove_file(&final_path);
        writer
            .finalize(&final_path)
            .expect("finalize after checkpoint");

        let fin = fs::read(&final_path).unwrap();
        assert_eq!(fin[101], 2, "final max_zoom includes the finer level");
        let fin_addressed = u64::from_le_bytes(fin[72..80].try_into().unwrap());
        assert_eq!(fin_addressed, 4, "final holds all 4 tiles");

        let _ = fs::remove_file(&ckpt_path);
        let _ = fs::remove_file(&final_path);
    }

    #[test]
    fn checkpoint_then_finalize_byte_identical_to_no_checkpoint() {
        // Issue #229: intermediate checkpoints must NOT change the final bytes.
        // Both checkpoint and finalize route through the same archive assembler,
        // so a run that checkpoints midway must be byte-identical to one that
        // never checkpoints. The 4th tile dups the 1st to exercise dedup.
        let tiles: Vec<(u8, u32, u32, Vec<u8>)> = vec![
            (0, 0, 0, vec![0x1a, 0x05, b'h', b'e', b'l', b'l', b'o']),
            (1, 0, 0, vec![0x1a, 0x03, b'a', b'b', b'c']),
            (1, 0, 1, vec![0x1a, 0x03, b'x', b'y', b'z']),
            (1, 1, 1, vec![0x1a, 0x05, b'h', b'e', b'l', b'l', b'o']), // dup of (0,0,0)
        ];
        let dir = std::env::temp_dir();

        // Baseline: no checkpoint, single finalize.
        let mut plain = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        plain.set_layer_name("test");
        plain.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));
        for (z, x, y, data) in &tiles {
            plain.add_tile(*z, *x, *y, data).unwrap();
        }
        let plain_path = dir.join("gpq-229-plain.pmtiles");
        let _ = fs::remove_file(&plain_path);
        plain.finalize(&plain_path).unwrap();

        // Checkpointed: assemble the archive after the first tile, keep going.
        let mut ckpt = StreamingPmtilesWriter::new(Compression::Gzip).unwrap();
        ckpt.set_layer_name("test");
        ckpt.set_bounds(&TileBounds::new(-180.0, -85.0, 180.0, 85.0));
        let ckpt_path = dir.join("gpq-229-intermediate.pmtiles");
        let final_path = dir.join("gpq-229-checkpointed-final.pmtiles");
        let _ = fs::remove_file(&final_path);
        for (i, (z, x, y, data)) in tiles.iter().enumerate() {
            ckpt.add_tile(*z, *x, *y, data).unwrap();
            if i == 0 {
                let _ = fs::remove_file(&ckpt_path);
                ckpt.checkpoint(&ckpt_path).unwrap();
            }
        }
        ckpt.finalize(&final_path).unwrap();

        let a = fs::read(&plain_path).unwrap();
        let b = fs::read(&final_path).unwrap();
        assert_eq!(
            a, b,
            "checkpointed run must be byte-identical to the non-checkpointed finalize"
        );

        let _ = fs::remove_file(&plain_path);
        let _ = fs::remove_file(&ckpt_path);
        let _ = fs::remove_file(&final_path);
    }
}