versatiles_container 4.4.0

//! Converts tile data between formats, compressions, and coordinate conventions.
//!
//! This module provides:
//! - [`TilesConverterParameters`]: declarative knobs (bbox filter, compression override, `flip_y`, `swap_xy`)
//! - [`TilesConvertReader`]: an adapter that applies those conversions while reading
//! - [`convert_tiles_container`]: a convenience function to convert and write to a target path using a [`ContainerRegistry`]
//!
//! ## Coordinate transforms
//! - `flip_y`: inverts Y within the zoom level (useful to switch between TMS and XYZ-like schemes)
//! - `swap_xy`: swaps X and Y (occasionally useful for sources with unconventional axis ordering)
//!
//! ## Example
//! ```rust
//! use versatiles_container::*;
//! use versatiles_core::*;
//! use std::sync::Arc;
//!
//! #[tokio::main]
//! async fn main() -> anyhow::Result<()> {
//!
//!     // Create runtime with default settings
//!     let runtime = TilesRuntime::default();
//!
//!     // Open the source
//!     let reader = runtime.reader_from_str("../testdata/berlin.mbtiles").await?;
//!
//!     // Limit to a tile pyramid and keep source compression;
//!     // you could also set `tile_compression: Some(TileCompression::Brotli)` to re-encode.
//!     let converter_params = TilesConverterParameters {
//!         tile_pyramid: Some(TilePyramid::new_full_up_to(8)),
//!         ..Default::default()
//!     };
//!
//!     // Convert and write
//!     let path_versatiles = std::env::temp_dir().join("temp2.versatiles");
//!     convert_tiles_container(reader, converter_params, &path_versatiles.as_path(), runtime).await?;
//!
//!     println!("Wrote {:?}", path_versatiles);
//!     Ok(())
//! }
//! ```

use crate::{SharedTileSource, SourceType, Tile, TileSource, TileSourceMetadata, TilesRuntime};
use anyhow::{Result, bail};
use async_trait::async_trait;
use std::{path::Path, sync::Arc};
use versatiles_core::{GeoBBox, TileBBox, TileCompression, TileCoord, TileFormat, TileJSON, TilePyramid, TileStream};
use versatiles_derive::context;

/// Parameters that control how tiles are transformed during reading/conversion.
///
/// These options affect coordinate handling, the subset of tiles traversed, and
/// whether tile payloads are re-encoded with a different compression.
///
/// Use with [`TilesConvertReader::new_from_reader`] or the helper
/// [`convert_tiles_container`].
#[derive(Debug)]
pub struct TilesConverterParameters {
	/// Optional spatial/zoom restriction. When set, only tiles inside the given
	/// [`TilePyramid`] are read/streamed. Existing bounds are intersected with this.
	pub tile_pyramid: Option<TilePyramid>,
	/// Optional geographic bounding box filter. When set, existing tilejson bounds are
	/// intersected with this bbox rather than being recalculated from the pyramid.
	pub geo_bbox: Option<GeoBBox>,
	/// Optional compression override. When set, tile payloads are re-encoded to this
	/// [`TileCompression`] (e.g., Gzip → Brotli). If `None`, the source compression is kept.
	pub tile_compression: Option<TileCompression>,
	/// Optional tile format override. When set, tiles are re-encoded to this format
	/// (e.g., PNG → WEBP). Only raster formats (AVIF, JPG, PNG, WEBP) are supported.
	pub tile_format: Option<TileFormat>,
	/// Optional quality parameter (0–100) for format conversion.
	pub format_quality: Option<u8>,
	/// Optional effort parameter (0–100) for format conversion.
	pub format_effort: Option<u8>,
	/// If `true`, flip the Y coordinate within the zoom level (TMS ↔ XYZ-like).
	pub flip_y: bool,
	/// If `true`, swap X and Y coordinates.
	pub swap_xy: bool,
}

impl Default for TilesConverterParameters {
	/// Returns parameters that perform no geometric change and keep source compression.
	fn default() -> Self {
		TilesConverterParameters {
			tile_pyramid: None,
			geo_bbox: None,
			tile_compression: None,
			tile_format: None,
			format_quality: None,
			format_effort: None,
			flip_y: false,
			swap_xy: false,
		}
	}
}

/// Converts tiles from the given reader and writes them to `path` using the provided runtime.
///
/// The conversion is applied by wrapping `reader` in a [`TilesConvertReader`] configured by `cp`.
///
/// ### Arguments
/// - `reader`: Source container reader.
/// - `cp`: Conversion parameters (bbox filter, compression override, `flip_y`, `swap_xy`).
/// - `path`: Output path; the format is inferred from its extension (or directory).
/// - `runtime`: Runtime configuration providing registry, cache, and event system.
///
/// ### Errors
/// Returns an error if reading tiles fails, if writing to the destination fails,
/// or if no suitable writer is registered for the output path.
///
/// ### Example
/// See the module-level example.
#[context("Converting tiles from reader to file")]
pub async fn convert_tiles_container(
	reader: SharedTileSource,
	cp: TilesConverterParameters,
	path: &Path,
	runtime: TilesRuntime,
) -> Result<()> {
	runtime.events().step("Starting conversion".to_string());

	let converter = TilesConvertReader::new_from_reader(reader, cp).await?;
	runtime.write_to_path(converter.into_shared(), path).await?;

	if runtime.had_errors() {
		bail!(
			"conversion completed with {} read error(s) — output may be incomplete",
			runtime.error_count()
		);
	}

	runtime.events().step("Conversion complete".to_string());
	Ok(())
}

/// Converts tiles from the given reader and writes them to a string destination.
///
/// The destination can be a file path or an SFTP URL (`sftp://...`).
#[context("Converting tiles from reader to destination")]
pub async fn convert_tiles_container_to_str(
	reader: SharedTileSource,
	cp: TilesConverterParameters,
	destination: &str,
	runtime: TilesRuntime,
) -> Result<()> {
	runtime.events().step("Starting conversion".to_string());

	let converter = TilesConvertReader::new_from_reader(reader, cp).await?;
	runtime.write_to_str(converter.into_shared(), destination).await?;

	if runtime.had_errors() {
		bail!(
			"conversion completed with {} read error(s) — output may be incomplete",
			runtime.error_count()
		);
	}

	runtime.events().step("Conversion complete".to_string());
	Ok(())
}

/// Reader adapter that applies coordinate transforms, bbox filtering, and optional
/// compression changes on-the-fly.
///
/// This type implements [`TileSource`], so it can be used anywhere a normal
/// reader is expected. Use [`TilesConvertReader::new_from_reader`] to wrap an
/// existing reader.
#[derive(Debug)]
pub struct TilesConvertReader {
	reader: SharedTileSource,
	converter_parameters: TilesConverterParameters,
	reader_metadata: TileSourceMetadata,
	tilejson: TileJSON,
	tile_pyramid: Arc<TilePyramid>,
}

impl TilesConvertReader {
	/// Wraps an existing reader so that reads are transformed according to `cp`.
	///
	/// Updates the internal [`TileSourceMetadata`] and [`TileJSON`] to reflect
	/// the chosen bbox, axis transforms, and compression.
	///
	/// ### Errors
	/// Propagates errors from querying/deriving parameters or updating metadata.
	#[context("Creating converter reader from existing reader")]
	pub async fn new_from_reader(reader: SharedTileSource, cp: TilesConverterParameters) -> Result<TilesConvertReader> {
		let mut tile_pyramid = reader.tile_pyramid().await?.as_ref().clone();
		if cp.flip_y {
			tile_pyramid.flip_y();
		}
		if cp.swap_xy {
			tile_pyramid.swap_xy();
		}

		if let Some(filter_pyramid) = &cp.tile_pyramid {
			tile_pyramid.intersect_pyramid(filter_pyramid);
		}

		let mut new_rp: TileSourceMetadata = reader.metadata().clone();

		if let Some(tile_format) = cp.tile_format {
			new_rp.set_tile_format(tile_format);
		}

		if let Some(tile_compression) = cp.tile_compression {
			new_rp.set_tile_compression(tile_compression);
		}

		// Expose the (possibly transformed/filtered) pyramid through the metadata so
		// downstream consumers (e.g. `update_tilejson`, writers) can derive bounds,
		// zoom range, and center from it.
		new_rp.set_tile_pyramid(tile_pyramid.clone());

		let mut tilejson = reader.tilejson().clone();

		if let Some(ref geo_bbox) = cp.geo_bbox {
			// Intersect existing tilejson bounds with the given geo bbox
			if let Some(ref mut bounds) = tilejson.bounds {
				bounds.intersect(geo_bbox);
			} else {
				tilejson.bounds = Some(*geo_bbox);
			}
			tilejson.center = None;
		}
		new_rp.update_tilejson(&mut tilejson);

		Ok(TilesConvertReader {
			reader,
			converter_parameters: cp,
			reader_metadata: new_rp,
			tilejson,
			tile_pyramid: Arc::new(tile_pyramid),
		})
	}
}

#[async_trait]
impl TileSource for TilesConvertReader {
	fn source_type(&self) -> Arc<SourceType> {
		SourceType::new_processor("TilesConvertReader", self.reader.source_type())
	}

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

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

	async fn tile(&self, coord: &TileCoord) -> Result<Option<Tile>> {
		let mut coord = *coord;

		if self.converter_parameters.flip_y {
			coord.flip_y();
		}

		if self.converter_parameters.swap_xy {
			coord.swap_xy();
		}

		let tile = self.reader.tile(&coord).await?;

		let Some(mut tile) = tile else { return Ok(None) };

		if let Some(tile_format) = self.converter_parameters.tile_format {
			tile.change_format(
				tile_format,
				self.converter_parameters.format_quality,
				self.converter_parameters.format_effort,
			)?;
		}

		if let Some(compression) = self.converter_parameters.tile_compression {
			tile.change_compression(&compression)?;
		}

		Ok(Some(tile))
	}

	async fn tile_coord_stream(&self, mut bbox: TileBBox) -> Result<TileStream<'static, ()>> {
		if self.converter_parameters.swap_xy {
			bbox.swap_xy();
		}
		if self.converter_parameters.flip_y {
			bbox.flip_y();
		}

		let mut stream = self.reader.tile_coord_stream(bbox).await?;

		let flip_y = self.converter_parameters.flip_y;
		let swap_xy = self.converter_parameters.swap_xy;

		if flip_y || swap_xy {
			stream = stream.map_coord(move |mut coord| {
				if flip_y {
					coord.flip_y();
				}
				if swap_xy {
					coord.swap_xy();
				}
				coord
			});
		}

		Ok(stream)
	}

	async fn tile_stream(&self, mut bbox: TileBBox) -> Result<TileStream<'static, Tile>> {
		log::trace!("converter::tile_stream {bbox:?}");
		if self.converter_parameters.swap_xy {
			bbox.swap_xy();
		}
		if self.converter_parameters.flip_y {
			bbox.flip_y();
		}

		let mut stream = self.reader.tile_stream(bbox).await?;

		let flip_y = self.converter_parameters.flip_y;
		let swap_xy = self.converter_parameters.swap_xy;

		if flip_y || swap_xy {
			stream = stream.map_coord(move |mut coord| {
				if flip_y {
					coord.flip_y();
				}
				if swap_xy {
					coord.swap_xy();
				}
				coord
			});
		}

		if let Some(tile_format) = self.converter_parameters.tile_format {
			let quality = self.converter_parameters.format_quality;
			let effort = self.converter_parameters.format_effort;
			stream = stream
				.map_parallel_try(move |_coord, mut tile| {
					tile.change_format(tile_format, quality, effort)?;
					Ok(tile)
				})
				.unwrap_results();
		}

		if let Some(tile_compression) = self.converter_parameters.tile_compression {
			stream = stream
				.map_parallel_try(move |_coord, mut tile| {
					tile.change_compression(&tile_compression)?;
					Ok(tile)
				})
				.unwrap_results();
		}

		Ok(stream)
	}

	async fn tile_pyramid(&self) -> Result<Arc<TilePyramid>> {
		Ok(self.tile_pyramid.clone())
	}
}

/// Integration tests verifying bbox intersection, coordinate transforms, traversal order,
/// and compression override behavior.
#[cfg(test)]
mod tests {
	use super::*;
	use crate::{MockReader, Traversal, VersaTilesReader};
	use assert_fs::NamedTempFile;
	use rstest::rstest;
	use versatiles_core::{
		GeoBBox, TileBBox,
		TileCompression::*,
		TileFormat::{self, *},
		TilePyramid,
	};

	fn new_pyramid(b: [u32; 4]) -> TilePyramid {
		TilePyramid::from([TileBBox::from_min_and_max(3, b[0], b[1], b[2], b[3]).unwrap()].as_slice())
	}

	fn get_mock_reader(tf: TileFormat, tc: TileCompression) -> SharedTileSource {
		let tile_pyramid = TilePyramid::new_full_up_to(4);
		let reader_metadata = TileSourceMetadata::new(tf, tc, Traversal::ANY, None);
		MockReader::new_mock(tile_pyramid, reader_metadata)
			.unwrap()
			.into_shared()
	}

	#[rstest]
	#[case(false, false, [2, 3, 4, 5], "23 33 43 24 34 25 35 44 45")]
	#[case(false, true, [2, 3, 5, 4], "32 33 34 35 42 43 44 45")]
	#[case(true, false, [2, 3, 4, 6], "24 34 44 23 33 22 32 21 31 43 42 41")]
	#[case(true, true, [2, 3, 6, 4], "35 34 33 32 31 45 44 43 42 41")]
	#[tokio::test]
	async fn bbox_and_tile_order(
		#[case] flip_y: bool,
		#[case] swap_xy: bool,
		#[case] bbox_out: [u32; 4],
		#[case] tile_list: &str,
	) -> Result<()> {
		let pyramid_in = new_pyramid([0, 1, 4, 5]);
		let pyramid_convert = new_pyramid([2, 3, 7, 7]);
		let pyramid_out = new_pyramid(bbox_out);

		let reader_metadata = TileSourceMetadata::new(JSON, Uncompressed, Traversal::ANY, None);
		let reader = MockReader::new_mock(pyramid_in, reader_metadata)?.into_shared();

		let temp_file = NamedTempFile::new("test.versatiles")?;
		let runtime = TilesRuntime::default();

		let cp = TilesConverterParameters {
			tile_pyramid: Some(pyramid_convert),
			geo_bbox: None,
			flip_y,
			swap_xy,
			tile_compression: None,
			..Default::default()
		};
		convert_tiles_container(reader, cp, &temp_file, runtime.clone()).await?;

		let reader_out = VersaTilesReader::open(&temp_file, runtime).await?;
		let parameters_out = reader_out.metadata();
		let tile_compression_out = parameters_out.tile_compression();
		assert_eq!(reader_out.tile_pyramid().await?.as_ref(), &pyramid_out);

		let bbox = pyramid_out.level_ref(3).to_bbox();
		let mut tiles: Vec<String> = Vec::new();
		for coord in bbox.iter_coords_zorder() {
			let mut text = reader_out
				.tile(&coord)
				.await?
				.unwrap()
				.into_blob(tile_compression_out)?
				.to_string();
			text = text
				.replace("{\"z\":3,\"x\":", "")
				.replace(",\"y\":", "")
				.replace('}', "");
			tiles.push(text);
		}
		let tiles = tiles.join(" ");
		assert_eq!(tiles, tile_list);

		Ok(())
	}

	#[test]
	fn test_tiles_converter_parameters_new() {
		let cp = TilesConverterParameters {
			tile_pyramid: Some(TilePyramid::new_full_up_to(1)),
			geo_bbox: None,
			flip_y: true,
			swap_xy: true,
			tile_compression: None,
			..Default::default()
		};

		assert!(cp.tile_pyramid.is_some());
		assert!(cp.flip_y);
		assert!(cp.swap_xy);
	}

	#[test]
	fn test_tiles_converter_parameters_default() {
		let cp = TilesConverterParameters::default();

		assert_eq!(cp.tile_pyramid, None);
		assert!(!cp.flip_y);
		assert!(!cp.swap_xy);
	}

	#[tokio::test]
	async fn test_tiles_convert_reader_new_from_reader() -> Result<()> {
		let reader = get_mock_reader(MVT, Uncompressed);
		let cp = TilesConverterParameters::default();

		let tcr = TilesConvertReader::new_from_reader(reader, cp).await?;

		assert_eq!(tcr.reader.source_type().to_string(), "container 'dummy' ('dummy')");
		assert_eq!(tcr.source_type().to_string(), "processor 'TilesConvertReader'");
		Ok(())
	}

	#[tokio::test]
	async fn test_tile() -> Result<()> {
		let reader = get_mock_reader(MVT, Uncompressed);
		let cp = TilesConverterParameters::default();
		let tcr = TilesConvertReader::new_from_reader(reader, cp).await?;

		let coord = TileCoord::new(0, 0, 0)?;
		let data = tcr.tile(&coord).await?;
		assert!(data.is_some());

		Ok(())
	}

	#[tokio::test]
	async fn test_flip_y_and_swap_xy() -> Result<()> {
		let reader = get_mock_reader(MVT, Uncompressed);
		let cp = TilesConverterParameters {
			flip_y: true,
			swap_xy: true,
			..Default::default()
		};
		let tcr = TilesConvertReader::new_from_reader(reader, cp).await?;

		let mut coord = TileCoord::new(4, 5, 6)?;
		let data = tcr.tile(&coord).await?;
		assert!(data.is_some());

		coord.flip_y();
		coord.swap_xy();
		let data_flipped = tcr.tile(&coord).await?;
		assert_eq!(data, data_flipped);

		Ok(())
	}

	#[tokio::test]
	async fn test_geo_bbox_intersects_existing_tilejson_bounds() -> Result<()> {
		// Source has bounds covering a wide area
		let source_bounds = GeoBBox::new(10.0, 50.0, 15.0, 55.0)?;
		let tile_pyramid = TilePyramid::new_full_up_to(4);
		let reader_metadata = TileSourceMetadata::new(MVT, Uncompressed, Traversal::ANY, None);
		let mut reader = MockReader::new_mock(tile_pyramid, reader_metadata)?;
		reader.tilejson_mut().bounds = Some(source_bounds);
		let reader = reader.into_shared();

		// User specifies a bbox that partially overlaps
		let filter_bbox = GeoBBox::new(12.0, 52.0, 20.0, 60.0)?;
		let mut filter_pyramid = TilePyramid::new_full();
		filter_pyramid.intersect_geo_bbox(&filter_bbox)?;

		let cp = TilesConverterParameters {
			tile_pyramid: Some(filter_pyramid),
			geo_bbox: Some(filter_bbox),
			..Default::default()
		};
		let tcr = TilesConvertReader::new_from_reader(reader, cp).await?;
		let bounds = tcr.tilejson().bounds.ok_or_else(|| anyhow::anyhow!("missing bounds"))?;

		// Bounds should be the intersection: [12.0, 52.0, 15.0, 55.0]
		assert_eq!(bounds.as_tuple(), (12.0, 52.0, 15.0, 55.0));
		Ok(())
	}

	#[tokio::test]
	async fn test_geo_bbox_without_existing_bounds_uses_pyramid() -> Result<()> {
		// Source has NO explicit bounds in tilejson
		let tile_pyramid = TilePyramid::new_full_up_to(4);
		let reader_metadata = TileSourceMetadata::new(MVT, Uncompressed, Traversal::ANY, None);
		let reader = MockReader::new_mock(tile_pyramid, reader_metadata)?.into_shared();

		let filter_bbox = GeoBBox::new(12.0, 52.0, 14.0, 54.0)?;
		let mut filter_pyramid = TilePyramid::new_full();
		filter_pyramid.intersect_geo_bbox(&filter_bbox)?;

		let cp = TilesConverterParameters {
			tile_pyramid: Some(filter_pyramid),
			geo_bbox: Some(filter_bbox),
			..Default::default()
		};
		let tcr = TilesConvertReader::new_from_reader(reader, cp).await?;

		// Bounds should be set from the pyramid (since no existing bounds to intersect)
		assert!(tcr.tilejson().bounds.is_some());
		Ok(())
	}

	#[tokio::test]
	async fn test_format_conversion() -> Result<()> {
		let reader = get_mock_reader(PNG, Uncompressed);
		let cp = TilesConverterParameters {
			tile_format: Some(WEBP),
			format_quality: Some(80),
			..Default::default()
		};
		let tcr = TilesConvertReader::new_from_reader(reader, cp).await?;

		assert_eq!(tcr.metadata().tile_format(), &WEBP);
		assert_eq!(tcr.metadata().tile_compression(), &Uncompressed);

		let coord = TileCoord::new(0, 0, 0)?;
		let tile = tcr.tile(&coord).await?.unwrap();
		assert_eq!(tile.format(), WEBP);

		Ok(())
	}

	#[tokio::test]
	async fn test_format_conversion_with_compression() -> Result<()> {
		let reader = get_mock_reader(PNG, Uncompressed);
		let cp = TilesConverterParameters {
			tile_format: Some(WEBP),
			format_quality: Some(80),
			tile_compression: Some(Gzip),
			..Default::default()
		};
		let tcr = TilesConvertReader::new_from_reader(reader, cp).await?;

		assert_eq!(tcr.metadata().tile_format(), &WEBP);
		assert_eq!(tcr.metadata().tile_compression(), &Gzip);

		let coord = TileCoord::new(0, 0, 0)?;
		let tile = tcr.tile(&coord).await?.unwrap();
		assert_eq!(tile.format(), WEBP);

		Ok(())
	}

	/// Regression: wrapping a source in a `TilesConvertReader` must not mutate the
	/// source's metadata. Previously the converter would call `set_tile_pyramid`
	/// on a clone whose `Arc<RwLock<...>>` was shared with the source, silently
	/// overwriting the source's pyramid with the swapped/flipped/filtered one.
	/// That broke any subsequent reader of `metadata.intersection_bbox`
	/// (notably `tile_size_stream` → `tile_coord_stream`), so the conversion
	/// progress count came out as 0/1 even though the actual write was correct.
	#[tokio::test]
	async fn swap_xy_does_not_corrupt_source_metadata() -> Result<()> {
		use crate::MBTilesReader;

		// `*.versatiles` is gitignored, so use the tracked mbtiles fixture instead.
		let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
			.parent()
			.unwrap()
			.join("testdata/berlin.mbtiles");
		let runtime = TilesRuntime::default();
		let reader = MBTilesReader::open(&path, runtime.clone())?;
		let shared = reader.into_shared();

		// Snapshot the source's pyramid before wrapping.
		let before = shared.tile_pyramid().await?.as_ref().clone();

		// Build a converter that swaps X/Y. This used to silently mutate the
		// shared `tile_pyramid` lock.
		let conv = TilesConvertReader::new_from_reader(
			shared.clone(),
			TilesConverterParameters {
				swap_xy: true,
				..Default::default()
			},
		)
		.await?;

		// 1) Source's pyramid must be unchanged.
		let after = shared.tile_pyramid().await?.as_ref().clone();
		assert_eq!(before, after, "source pyramid was mutated by the converter");

		// 2) `tile_coord_stream` and `tile_stream` must agree on counts at every
		//    level. Before the fix this failed for every level except z=0 (which
		//    is invariant under swap) — coord returned 0, stream returned N.
		let pyr = conv.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(&conv, bbox).await?;
		}
		Ok(())
	}

	/// Direct exercise of the [`TileSource`](crate::TileSource) stream-count
	/// invariant against the converter, with no swap/flip/filter applied —
	/// catches regressions that wouldn't be specific to the swap_xy path.
	#[tokio::test]
	async fn converter_satisfies_stream_count_invariant() -> Result<()> {
		use crate::MBTilesReader;

		let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
			.parent()
			.unwrap()
			.join("testdata/berlin.mbtiles");
		let runtime = TilesRuntime::default();
		let shared = MBTilesReader::open(&path, runtime.clone())?.into_shared();

		let conv = TilesConvertReader::new_from_reader(shared, TilesConverterParameters::default()).await?;

		let pyr = conv.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(&conv, bbox).await?;
		}
		Ok(())
	}
}