versatiles_container 4.10.0

//! Read tiles and metadata from a `.versatiles` container.
//!
//! The `VersaTilesReader` parses the container header, decompresses the **block index**,
//! reads embedded `TileJSON` metadata, and exposes tiles via [`TileSource`]. The
//! file format organizes data into fixed **256×256 tile blocks**; each block stores
//! a Brotli-compressed tile index (byte ranges), followed by a contiguous region of
//! tile blobs. This reader lazily caches decoded tile indices for fast random access.
//!
//! ## Extracted artifacts
//! - `tilejson`: parsed `TileJSON` from the `meta_range` (if present)
//! - `parameters`: [`TileSourceMetadata`] with `tile_format`, `tile_compression`, and a
//!   **tile pyramid** computed from the block index
//! - `block_index`: lightweight structure describing all block ranges
//!
//! ## Usage
//! ```rust,no_run
//! use versatiles_container::*;
//! use versatiles_core::*;
//! use anyhow::Result;
//! use futures::StreamExt;
//! use std::path::Path;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
//!     // Open a .versatiles container (relative or absolute path)
//!     let runtime = TilesRuntime::default();
//!     let path = Path::new("./data/world.versatiles");
//!     let mut reader = VersaTilesReader::open(&path, runtime).await?;
//!
//!     // Inspect parameters & TileJSON
//!     let metadata = reader.metadata();
//!     let tj = reader.tilejson();
//!     println!("format={:?} compression={:?}", metadata.tile_format(), metadata.tile_compression());
//!
//!     // Fetch one tile
//!     if let Some(mut tile) = reader.tile(&TileCoord::new(15, 1, 4)?).await? {
//!         let _blob = tile.as_blob(metadata.tile_compression())?;
//!     }
//!
//!     // Stream a bbox (coalesces reads per block for fewer I/O calls)
//!     let pyramid = reader.tile_pyramid().await?;
//!     let bbox = pyramid.level_bbox(4);
//!     let mut stream = reader.tile_stream(bbox).await?;
//!     while let Some((coord, mut tile)) = stream.next().await {
//!         let _size = tile.as_blob(metadata.tile_compression())?.len();
//!         // use (coord, _size)
//!     }
//!     Ok(())
//! }
//! ```
//!
//! ## Errors
//! Returns errors when the file cannot be read or decompressed, when metadata/index parsing fails,
//! or when a requested tile is missing.

use std::{fmt::Debug, mem::size_of, ops::Shr, path::Path, sync::Arc, time::Instant};

use anyhow::Result;
use async_trait::async_trait;
use futures::stream::StreamExt;
use moka::future::Cache;
#[cfg(feature = "cli")]
use versatiles_core::utils::PrettyPrint;
use versatiles_core::{
	ByteRange, ConcurrencyLimits, TileBBox, TileCoord, TileJSON, TilePyramid, TileStream,
	compression::decompress,
	io::{DataReader, DataReaderFile},
};
use versatiles_derive::context;

use super::types::{BlockDefinition, BlockIndex, FileHeader, TileIndex};
use crate::{
	SharedTileSource, SourceType, Tile, TileSource, TileSourceMetadata, TilesReader, TilesRuntime, Traversal,
	TraversalOrder, TraversalSize, container::tile_chunking::Chunks,
};

/// Reader for `.versatiles` containers.
///
/// Decompresses and parses the block index, merges embedded `TileJSON`, computes a
/// per-zoom tile pyramid, and serves tiles via lazy index lookups. Tile
/// indices are cached (least-recently-used) to accelerate repeated random access.
pub struct VersaTilesReader {
	block_index: BlockIndex,
	header: FileHeader,
	metadata: TileSourceMetadata,
	reader: Arc<DataReader>,
	tile_index_cache: Cache<TileCoord, Arc<TileIndex>>,
	tilejson: TileJSON,
	runtime: TilesRuntime,
}

impl VersaTilesReader {
	/// Open a `.versatiles` container from a filesystem path.
	///
	/// Creates a `DataReaderFile` and delegates to [`VersaTilesReader::open_data`]. The path may be
	/// relative or absolute.
	///
	/// # Errors
	/// Returns an error if the file cannot be opened.
	#[context("Failed to open versatiles file at '{path:?}'")]
	pub async fn open(path: &Path, runtime: TilesRuntime) -> Result<VersaTilesReader> {
		VersaTilesReader::open_data(DataReaderFile::open(path)?, runtime).await
	}

	/// Open a `.versatiles` container from an existing [`DataReader`].
	///
	/// Reads the header, loads and (if present) decompresses the `TileJSON` metadata, then
	/// reads and decompresses the **block index** (Brotli). Finally, computes the tile pyramid
	/// from the block index and initializes the tile-index cache.
	///
	/// # Errors
	/// Returns an error if header/metadata/index reads or decompressions fail.
	#[context("Failed to open versatiles reader")]
	pub async fn open_data(mut reader: DataReader, runtime: TilesRuntime) -> Result<VersaTilesReader> {
		let name = reader.name().to_string();
		log::trace!("versatiles: opening '{name}'");

		let phase = Instant::now();
		let header = FileHeader::from_reader(&mut reader)
			.await
			.context("Failed reading the header")?;
		log::trace!(
			"versatiles: '{name}' header read in {:.2}s",
			phase.elapsed().as_secs_f32()
		);

		let tilejson = if header.meta_range.length > 0 {
			let phase = Instant::now();
			let blob = reader
				.read_range(&header.meta_range)
				.await
				.context("Failed reading the meta data")?;
			let blob = decompress(blob, &header.compression).context("Failed decompressing the meta data")?;
			let json = TileJSON::try_from_blob_or_default(&blob);
			log::trace!(
				"versatiles: '{name}' meta ({} bytes) read in {:.2}s",
				header.meta_range.length,
				phase.elapsed().as_secs_f32()
			);
			json
		} else {
			TileJSON::default()
		};

		let phase = Instant::now();
		let block_index_blob = reader
			.read_range(&header.blocks_range)
			.await
			.context("Failed reading the block index")?;
		let block_index =
			BlockIndex::from_brotli_blob(&block_index_blob).context("Failed decompressing the block index")?;
		log::trace!(
			"versatiles: '{name}' block index ({} bytes, {} block(s)) read+decoded in {:.2}s",
			header.blocks_range.length,
			block_index.len(),
			phase.elapsed().as_secs_f32()
		);

		let metadata = TileSourceMetadata::new(
			header.tile_format,
			header.compression,
			Traversal {
				order: TraversalOrder::AnyOrder,
				size: TraversalSize::new_max(256)?,
			},
			None,
		);

		Ok(VersaTilesReader {
			block_index,
			header,
			metadata,
			reader: Arc::new(reader),
			tile_index_cache: Cache::builder()
				.max_capacity(100_000_000)
				.weigher(|_k, v: &Arc<TileIndex>| {
					let bytes = size_of::<TileCoord>() + size_of::<Arc<TileIndex>>() + v.len() * size_of::<ByteRange>();
					u32::try_from(bytes).unwrap_or(u32::MAX)
				})
				.build(),
			tilejson,
			runtime,
		})
	}

	/// Load (and cache) the tile index for a block.
	///
	/// Reads the block's index blob, decompresses it, adjusts offsets to the tiles segment,
	/// and inserts the result into a concurrent byte-budgeted cache. Concurrent callers
	/// for the same block share a single load; different blocks load in parallel.
	///
	/// # Errors
	/// Returns an error if reading or decompression fails.
	#[context("Failed to get tile index for block {block:?}")]
	async fn get_block_tile_index(&self, block: &BlockDefinition) -> Result<Arc<TileIndex>> {
		let block_coord = *block.coord();
		let reader = Arc::clone(&self.reader);
		let index_range = *block.index_range();
		let tiles_offset = block.tiles_range().offset;
		let expected_count = usize::try_from(block.count_tiles())?;

		self
			.tile_index_cache
			.try_get_with(block_coord, async move {
				let blob = reader.read_range(&index_range).await?;
				let mut tile_index = TileIndex::from_brotli_blob(&blob)?;
				tile_index.shift_by(tiles_offset);
				debug_assert_eq!(tile_index.len(), expected_count);
				anyhow::Ok(Arc::new(tile_index))
			})
			.await
			.map_err(|e: Arc<anyhow::Error>| anyhow::anyhow!("{e:#}"))
	}

	/// Sum of all block index byte lengths.
	#[cfg(feature = "cli")]
	fn index_size(&self) -> u64 {
		self.block_index.iter().map(|b| b.index_range().length).sum()
	}

	/// Sum of all block tiles byte lengths.
	fn tiles_size(&self) -> u64 {
		self.block_index.iter().map(|b| b.tiles_range().length).sum()
	}

	/// Build read **chunks** by grouping tile ranges within the same block.
	///
	/// Coalesces nearby ranges into at most ~64 MiB chunks (with a small gap tolerance)
	/// to minimize I/O calls during streaming.
	async fn get_chunks(&self, bbox: TileBBox) -> Result<Chunks> {
		let block_coords: Vec<TileCoord> = bbox.scaled_down(256).iter_coords().collect();
		let io_bound = ConcurrencyLimits::default().io_bound;

		let stream = futures::stream::iter(block_coords)
			.map(|block_coord: TileCoord| {
				async move {
					// Get the block using the block coordinate
					let Some(block) = self.block_index.block(&block_coord) else {
						return Ok(Chunks::new_empty());
					};
					let block = block.clone();
					log::trace!("block {block:?}");

					// Get the bounding box of all tiles defined in this block
					let tiles_bbox_block = block.global_bbox();
					log::trace!("tiles_bbox_block {tiles_bbox_block:?}");

					// Get the bounding box of all tiles defined in this block
					let mut tiles_bbox_used: TileBBox = bbox;
					tiles_bbox_used.intersect_bbox(tiles_bbox_block)?;
					log::trace!("tiles_bbox_used {tiles_bbox_used:?}");

					debug_assert_eq!(bbox.level(), tiles_bbox_block.level());
					debug_assert_eq!(bbox.level(), tiles_bbox_used.level());

					// Get the tile index of this block
					let tile_index: Arc<TileIndex> = self.get_block_tile_index(&block).await?;
					log::trace!("tile_index.len() {}", tile_index.len());

					let tile_ranges: Vec<(TileCoord, ByteRange)> = tile_index
						.iter()
						.enumerate()
						.filter_map(|(index, range)| {
							let coord = tiles_bbox_block.coord_at_index(index as u64).ok()?;
							if tiles_bbox_used.includes_coord(&coord) && range.length > 0 {
								Some((coord, *range))
							} else {
								None
							}
						})
						.collect();

					Ok(Chunks::from_tile_ranges(tile_ranges))
				}
			})
			.buffer_unordered(io_bound);

		let chunks: Vec<Result<Chunks>> = stream.collect().await;

		let chunks: Chunks = chunks
			.into_iter()
			.collect::<Result<Vec<Chunks>>>()?
			.into_iter()
			.flatten()
			.collect();
		Ok(chunks)
	}
}

#[async_trait]
impl TilesReader for VersaTilesReader {
	async fn open_reader(reader: DataReader, runtime: TilesRuntime) -> Result<SharedTileSource> {
		Ok(Self::open_data(reader, runtime).await?.into_shared())
	}
}

#[async_trait]
/// [`TileSource`] implementation — provides `container_name`, `parameters`, `tilejson`,
/// on-the-fly `override_compression`, single-tile fetch via `tile`, and bbox streaming via
/// `tile_stream` (with internal read coalescing).
impl TileSource for VersaTilesReader {
	fn source_type(&self) -> Arc<SourceType> {
		SourceType::new_container("versatiles", self.reader.name())
	}

	fn tilejson(&self) -> &TileJSON {
		&self.tilejson
	}

	fn metadata(&self) -> &TileSourceMetadata {
		&self.metadata
	}

	async fn tile_pyramid(&self) -> Result<Arc<TilePyramid>> {
		self
			.metadata
			.get_or_compute_tile_pyramid(|| Ok(self.block_index.tile_pyramid()))
	}

	/// Fetch a single tile by XYZ coordinate.
	///
	/// Computes the corresponding **block coordinate** (z, x>>8, y>>8), verifies membership
	/// within the block's bbox, looks up the tile's byte range from the cached index, and reads it.
	/// Returns `Ok(None)` for empty ranges or missing blocks.
	#[context("fetching tile {:?} from '{}'", coord, self.reader.name())]
	async fn tile(&self, coord: &TileCoord) -> Result<Option<Tile>> {
		// Calculate block coordinate
		let block_coord = TileCoord::new(coord.level, coord.x.shr(8), coord.y.shr(8))?;

		// Get the block using the block coordinate
		let Some(block) = self.block_index.block(&block_coord) else {
			return Ok(None);
		};
		let block = block.clone();

		// Get the block and its bounding box
		let bbox = block.global_bbox();

		// Check if the tile is within the block definition
		if !bbox.includes_coord(coord) {
			log::trace!("tile {coord:?} outside block definition");
			return Ok(None);
		}

		// Get the tile ID
		let tile_id = usize::try_from(bbox.index_of(coord)?)?;

		// Retrieve the tile index from cache or read from the reader
		let tile_index: Arc<TileIndex> = self.get_block_tile_index(&block).await?;
		let tile_range: ByteRange = *tile_index.get(tile_id);

		//  None if the tile range has zero length
		if tile_range.length == 0 {
			return Ok(None);
		}

		// Read the tile data from the reader
		let blob = self.reader.read_range(&tile_range).await?;
		Ok(Some(Tile::from_blob(
			blob,
			*self.metadata.tile_compression(),
			*self.metadata.tile_format(),
		)))
	}

	#[context("streaming tile sizes for bbox {:?}", bbox)]
	async fn tile_size_stream(&self, bbox: TileBBox) -> Result<TileStream<'static, u32>> {
		let bbox = self.metadata.intersection_bbox(&bbox);
		let block_coords: Vec<TileCoord> = bbox.scaled_down(256).iter_coords().collect();

		let mut blocks: Vec<(TileBBox, TileBBox, BlockDefinition)> = Vec::new();
		for block_coord in block_coords {
			let Some(block) = self.block_index.block(&block_coord) else {
				continue;
			};
			let block_bbox = *block.global_bbox();
			let mut used_bbox = bbox;
			used_bbox.intersect_bbox(&block_bbox)?;
			blocks.push((block_bbox, used_bbox, block.clone()));
		}

		let reader = Arc::clone(&self.reader);
		let runtime = self.runtime.clone();
		let io_bound = ConcurrencyLimits::default().io_bound;

		Ok(TileStream::from_stream(
			futures::stream::iter(blocks)
				.map(move |(block_bbox, used_bbox, block)| {
					let reader = Arc::clone(&reader);
					let runtime = runtime.clone();
					async move {
						let blob = match reader.read_range(block.index_range()).await {
							Ok(blob) => blob,
							Err(e) => {
								runtime.record_error(
									"versatiles block index",
									&e.context(format!("reading range {:?}", block.index_range())),
								);
								return futures::stream::iter(Vec::new());
							}
						};
						let tile_index = match TileIndex::from_brotli_blob(&blob) {
							Ok(idx) => idx,
							Err(e) => {
								runtime.record_error("versatiles tile index decompress", &e);
								return futures::stream::iter(Vec::new());
							}
						};

						let entries: Vec<(TileCoord, u32)> = tile_index
							.iter()
							.enumerate()
							.filter_map(|(index, range)| {
								if range.length == 0 {
									return None;
								}
								let coord = block_bbox.coord_at_index(index as u64).ok()?;
								if used_bbox.includes_coord(&coord) {
									Some((coord, u32::try_from(range.length).ok()?))
								} else {
									None
								}
							})
							.collect();

						futures::stream::iter(entries)
					}
				})
				.buffer_unordered(io_bound)
				.flatten()
				.boxed(),
		))
	}

	async fn tile_coord_stream(&self, bbox: TileBBox) -> Result<TileStream<'static, ()>> {
		Ok(self.tile_size_stream(bbox).await?.filter_map(|_, _| Some(())))
	}

	#[context("streaming tiles for bbox {:?}", bbox)]
	async fn tile_stream(&self, bbox: TileBBox) -> Result<TileStream<'static, Tile>> {
		log::trace!("versatiles::tile_stream {bbox:?}");
		let chunks = self.get_chunks(bbox).await?;
		Ok(chunks.stream(
			Arc::clone(&self.reader),
			*self.metadata.tile_compression(),
			*self.metadata.tile_format(),
		))
	}

	// deep probe of container meta
	#[cfg(feature = "cli")]
	#[context("probing versatiles container metadata")]
	async fn probe_container(&self, print: &mut PrettyPrint, _runtime: &TilesRuntime) -> Result<()> {
		print.add_key_value("tile format", &self.header.tile_format).await;
		print.add_key_value("compression", &self.header.compression).await;
		print
			.add_key_value(
				"zoom range",
				&format!("{}..{}", self.header.zoom_range[0], self.header.zoom_range[1]),
			)
			.await;
		print.add_key_value("meta size", &self.header.meta_range.length).await;
		print.add_key_value("block count", &self.block_index.len()).await;

		print
			.add_key_value("sum of block index sizes", &self.index_size())
			.await;
		print
			.add_key_value("sum of block tiles sizes", &self.tiles_size())
			.await;

		Ok(())
	}
}

// Implement Debug for TilesReader
impl Debug for VersaTilesReader {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		f.debug_struct("VersaTilesReader")
			.field("parameters", &self.metadata())
			.finish()
	}
}

impl PartialEq for VersaTilesReader {
	fn eq(&self, other: &Self) -> bool {
		self.tilejson == other.tilejson && self.metadata == other.metadata && self.tiles_size() == other.tiles_size()
	}
}

#[cfg(test)]
#[allow(clippy::cast_possible_truncation)]
mod tests {
	use assert_fs::NamedTempFile;
	use versatiles_core::{Blob, TileCompression, TileFormat, TilePyramid, assert_wildcard, io::DataWriterBlob};

	use super::*;
	use crate::{MOCK_BYTES_PBF, MockReader, TilesRuntime, TilesWriter, VersaTilesWriter, make_test_file};

	// Helper to quickly create a test reader and bbox
	async fn mk_reader() -> Result<(NamedTempFile, VersaTilesReader)> {
		let temp_file = make_test_file(TileFormat::MVT, TileCompression::Gzip, 4, "versatiles").await?;
		let runtime = TilesRuntime::default();
		let reader = VersaTilesReader::open(&temp_file, runtime).await?;
		Ok((temp_file, reader))
	}

	#[tokio::test]
	async fn reader() -> Result<()> {
		let (_, reader) = mk_reader().await?;

		assert_eq!(
			format!("{reader:?}"),
			"VersaTilesReader { parameters: TileSourceMetadata { tile_compression: Gzip, tile_format: MVT, traversal: Traversal(AnyOrder,1..256), tile_pyramid: RwLock { data: None, poisoned: false, .. } } }"
		);
		assert_wildcard!(
			reader.source_type().to_string(),
			"container 'versatiles' ('*.versatiles')"
		);
		assert_eq!(
			reader.tilejson().stringify(),
			"{\"tilejson\":\"3.0.0\",\"type\":\"dummy\"}"
		);
		assert_eq!(
			format!("{:?}", reader.metadata()),
			"TileSourceMetadata { tile_compression: Gzip, tile_format: MVT, traversal: Traversal(AnyOrder,1..256), tile_pyramid: RwLock { data: None, poisoned: false, .. } }"
		);
		assert_eq!(reader.metadata().tile_compression(), &TileCompression::Gzip);
		assert_eq!(reader.metadata().tile_format(), &TileFormat::MVT);

		let blob = reader
			.tile(&TileCoord::new(4, 15, 1)?)
			.await?
			.unwrap()
			.into_blob(&TileCompression::Uncompressed)?;
		assert_eq!(blob.as_slice(), MOCK_BYTES_PBF);

		let sizes = reader
			.tile_stream(TileBBox::new_full(4)?)
			.await?
			.map_parallel_try(|_coord, mut tile| Ok(tile.as_blob(&TileCompression::Gzip)?.len()))
			.unwrap_results();
		let sizes: Vec<(TileCoord, u64)> = sizes.to_vec().await;
		assert_eq!(sizes.len(), 256);
		for (_, size) in sizes {
			assert_eq!(size, 77);
		}

		Ok(())
	}

	#[tokio::test]
	async fn satisfies_stream_count_invariant() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		let pyr = reader.tile_pyramid().await?;
		for z in 0..=pyr.level_max().unwrap_or(0) {
			let bbox = pyr.level_ref(z).to_bbox();
			if bbox.is_empty() {
				continue;
			}
			crate::testing::assert_stream_counts_agree(&reader, bbox).await?;
		}
		Ok(())
	}

	#[tokio::test]
	async fn tile_stream_matches_individual_blob_reads() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		let bbox = TileBBox::new_full(4)?;
		let stream = reader.tile_stream(bbox).await?;
		let mut all: Vec<(TileCoord, Blob)> = stream
			.map_parallel_try(|_coord, tile| tile.into_blob(&TileCompression::Uncompressed))
			.unwrap_results()
			.to_vec()
			.await;
		all.sort_by_key(|(c, _)| (c.y, c.x));
		assert_eq!(all.len(), bbox.count_tiles() as usize);

		// Spot check a few coordinates (corners + center)
		let probes = [
			TileCoord::new(4, 0, 0)?,
			TileCoord::new(4, 15, 0)?,
			TileCoord::new(4, 0, 15)?,
			TileCoord::new(4, 15, 15)?,
			TileCoord::new(4, 7, 8)?,
		];
		for coord in probes {
			let from_stream = all
				.iter()
				.find(|(c, _)| *c == coord)
				.map(|(_, b)| b.clone())
				.expect("present in stream");
			let from_single = reader
				.tile(&coord)
				.await?
				.expect("present via single read")
				.into_blob(&TileCompression::Uncompressed)?;
			assert_eq!(
				from_stream.as_slice(),
				from_single.as_slice(),
				"blob mismatch at {coord:?}"
			);
		}
		Ok(())
	}

	#[tokio::test]
	async fn tile_out_of_range_is_none() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		// level beyond available
		assert!(reader.tile(&TileCoord::new(5, 0, 0)?).await?.is_none());
		Ok(())
	}

	#[tokio::test]
	async fn single_tile_bbox_streams() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		let one = TileBBox::from_min_and_max(4, 15, 1, 15, 1)?;
		let blobs = reader.tile_stream(one).await?.to_vec().await;
		assert_eq!(blobs.len(), 1);
		Ok(())
	}

	#[tokio::test]
	async fn read_your_own_dog_food() -> Result<()> {
		let mut reader1 = MockReader::new_mock(
			TilePyramid::new_full_up_to(4),
			TileSourceMetadata::new(TileFormat::JSON, TileCompression::Gzip, Traversal::ANY, None),
		)?;

		let runtime = TilesRuntime::default();

		let mut data_writer1 = DataWriterBlob::new()?;
		VersaTilesWriter::write_to_writer(&mut reader1, &mut data_writer1, runtime.clone()).await?;

		let data_reader1 = data_writer1.to_reader();
		let mut reader2 = VersaTilesReader::open_data(Box::new(data_reader1), runtime.clone()).await?;

		let mut data_writer2 = DataWriterBlob::new()?;
		VersaTilesWriter::write_to_writer(&mut reader2, &mut data_writer2, runtime.clone()).await?;

		let data_reader2 = data_writer2.to_reader();
		let reader3 = VersaTilesReader::open_data(Box::new(data_reader2), runtime).await?;

		assert_eq!(reader2, reader3);

		Ok(())
	}

	#[tokio::test]
	#[cfg(feature = "cli")]
	async fn probe() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		let runtime = TilesRuntime::default();

		let mut printer = PrettyPrint::new();
		reader
			.probe_container(&mut printer.category("container").await, &runtime)
			.await?;
		assert_eq!(
			printer.stringify().await.split('\n').collect::<Vec<_>>(),
			[
				"container:",
				"  tile format: MVT",
				"  compression: Gzip",
				"  zoom range: \"0..4\"",
				"  meta size: 58",
				"  block count: 5",
				"  sum of block index sizes: 70",
				"  sum of block tiles sizes: 385",
				""
			]
		);

		Ok(())
	}

	#[tokio::test]
	async fn tile_size_stream_full_level() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		let bbox = TileBBox::new_full(4)?;
		let mut sizes: Vec<(TileCoord, u32)> = reader.tile_size_stream(bbox).await?.to_vec().await;
		sizes.sort_by_key(|(c, _)| (c.y, c.x));

		assert_eq!(sizes.len(), 256);
		for (_, size) in &sizes {
			assert_eq!(*size, 77);
		}

		Ok(())
	}

	#[tokio::test]
	async fn tile_size_stream_sub_bbox() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		let bbox = TileBBox::from_min_and_max(4, 2, 3, 5, 6)?;
		let sizes: Vec<(TileCoord, u32)> = reader.tile_size_stream(bbox).await?.to_vec().await;

		assert_eq!(sizes.len(), 16); // 4x4
		for (coord, size) in &sizes {
			assert!(bbox.includes_coord(coord), "coord {coord:?} outside requested bbox");
			assert_eq!(*size, 77);
		}

		Ok(())
	}

	#[tokio::test]
	async fn tile_size_stream_single_tile() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		let bbox = TileBBox::from_min_and_max(4, 15, 1, 15, 1)?;
		let sizes: Vec<(TileCoord, u32)> = reader.tile_size_stream(bbox).await?.to_vec().await;

		assert_eq!(sizes.len(), 1);
		assert_eq!(sizes[0].0, TileCoord::new(4, 15, 1)?);
		assert_eq!(sizes[0].1, 77);

		Ok(())
	}

	#[tokio::test]
	async fn tile_size_stream_matches_tile_stream() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		let bbox = TileBBox::new_full(4)?;
		let compression = reader.metadata().tile_compression();

		let mut sizes: Vec<(TileCoord, u32)> = reader.tile_size_stream(bbox).await?.to_vec().await;
		sizes.sort_by_key(|(c, _)| (c.level, c.y, c.x));

		let mut blob_sizes: Vec<(TileCoord, u32)> = reader
			.tile_stream(bbox)
			.await?
			.map(move |_coord, tile| {
				u32::try_from(tile.into_blob(compression).expect("tile should have blob").len()).expect("size fits u32")
			})
			.to_vec()
			.await;
		blob_sizes.sort_by_key(|(c, _)| (c.level, c.y, c.x));

		assert_eq!(sizes.len(), blob_sizes.len());
		for (a, b) in sizes.iter().zip(blob_sizes.iter()) {
			assert_eq!(a.0, b.0, "coord mismatch");
			assert_eq!(a.1, b.1, "size mismatch at {:?}", a.0);
		}

		Ok(())
	}

	#[tokio::test]
	async fn tile_size_stream_out_of_range_is_empty() -> Result<()> {
		let (_, reader) = mk_reader().await?;
		let bbox = TileBBox::new_full(5)?;
		let sizes: Vec<(TileCoord, u32)> = reader.tile_size_stream(bbox).await?.to_vec().await;

		assert!(sizes.is_empty());

		Ok(())
	}

	// --- parallelism / dedup tests ---

	#[derive(Debug, Default)]
	struct CountingState {
		in_flight: std::sync::atomic::AtomicUsize,
		max_in_flight: std::sync::atomic::AtomicUsize,
		reads: parking_lot::Mutex<Vec<ByteRange>>,
	}

	impl CountingState {
		fn reset(&self) {
			use std::sync::atomic::Ordering::SeqCst;
			self.in_flight.store(0, SeqCst);
			self.max_in_flight.store(0, SeqCst);
			self.reads.lock().clear();
		}
	}

	#[derive(Debug)]
	struct CountingReader {
		inner: DataReader,
		state: Arc<CountingState>,
		delay: std::time::Duration,
	}

	#[async_trait]
	impl versatiles_core::io::DataReaderTrait for CountingReader {
		async fn read_range(&self, range: &ByteRange) -> Result<versatiles_core::Blob> {
			use std::sync::atomic::Ordering::SeqCst;
			let n = self.state.in_flight.fetch_add(1, SeqCst) + 1;
			self.state.max_in_flight.fetch_max(n, SeqCst);
			self.state.reads.lock().push(*range);
			tokio::time::sleep(self.delay).await;
			let result = self.inner.read_range(range).await;
			self.state.in_flight.fetch_sub(1, SeqCst);
			result
		}
		async fn read_all(&self) -> Result<versatiles_core::Blob> {
			self.inner.read_all().await
		}
		fn name(&self) -> &str {
			self.inner.name()
		}
	}

	/// Build a fresh in-memory `.versatiles` blob and open a reader against it,
	/// wrapped in a `CountingReader` so we can observe the data-layer traffic.
	async fn mk_counting_reader(delay_ms: u64) -> Result<(VersaTilesReader, Arc<CountingState>)> {
		let mut src = MockReader::new_mock(
			TilePyramid::new_full_up_to(4),
			TileSourceMetadata::new(TileFormat::JSON, TileCompression::Gzip, Traversal::ANY, None),
		)?;
		let runtime = TilesRuntime::default();
		let mut writer = DataWriterBlob::new()?;
		VersaTilesWriter::write_to_writer(&mut src, &mut writer, runtime.clone()).await?;
		let inner: DataReader = Box::new(writer.to_reader());
		let state = Arc::new(CountingState::default());
		let counting = CountingReader {
			inner,
			state: Arc::clone(&state),
			delay: std::time::Duration::from_millis(delay_ms),
		};
		let reader = VersaTilesReader::open_data(Box::new(counting), runtime).await?;
		// Discard counters accumulated during open_data (header, meta, block-index).
		state.reset();
		Ok((reader, state))
	}

	#[tokio::test]
	async fn concurrent_same_block_dedups_index_read() -> Result<()> {
		use std::sync::atomic::Ordering::SeqCst;
		let (reader, state) = mk_counting_reader(20).await?;
		let reader = Arc::new(reader);
		let coord = TileCoord::new(4, 15, 1)?;

		let mut handles = Vec::new();
		for _ in 0..16 {
			let r = Arc::clone(&reader);
			handles.push(tokio::spawn(async move { r.tile(&coord).await.unwrap() }));
		}
		for h in handles {
			h.await.unwrap();
		}

		// With dedup we expect at most: 1 block-index read + 16 tile-data reads = 17.
		// Without dedup we'd see 16 + 16 = 32.
		let total = state.reads.lock().len();
		assert!(
			total <= 17,
			"expected dedup of same-block index reads, saw {total} total reads for 16 same-tile fetches"
		);
		// Sanity: at least one block-index load did happen.
		assert!(total >= 17, "expected 16 data reads + 1 index read, saw {total}");
		// And the data reads themselves should have overlapped.
		let peak = state.max_in_flight.load(SeqCst);
		assert!(peak >= 2, "expected concurrent tile-data reads, saw peak {peak}");
		Ok(())
	}

	#[tokio::test]
	async fn concurrent_different_blocks_load_in_parallel() -> Result<()> {
		use std::sync::atomic::Ordering::SeqCst;
		let (reader, state) = mk_counting_reader(40).await?;
		let reader = Arc::new(reader);

		// One tile per zoom level 0..=4 — each lives in a different block.
		let coords: Vec<TileCoord> = (0..=4u8).map(|z| TileCoord::new(z, 0, 0).unwrap()).collect();

		let mut handles = Vec::new();
		for coord in coords {
			let r = Arc::clone(&reader);
			handles.push(tokio::spawn(async move { r.tile(&coord).await.unwrap() }));
		}
		for h in handles {
			h.await.unwrap();
		}

		// 5 blocks => 5 index reads + 5 data reads = 10 total, all distinct ranges.
		assert_eq!(state.reads.lock().len(), 10);
		let peak = state.max_in_flight.load(SeqCst);
		assert!(
			peak >= 2,
			"expected concurrent block-index loads for distinct blocks, saw peak {peak}"
		);
		Ok(())
	}
}