mlt 0.1.26

use std::collections::BTreeMap;
use std::path::Path;
use std::sync::{Arc, mpsc};
use std::thread;
use std::time::Instant;

use anyhow::{Result as AnyResult, bail};
use bytes::Bytes;
use futures::TryStreamExt;
use martin_tile_utils::Encoding;
use mlt_core::encoder::EncoderConfig;
use pmtiles::{
    AsyncPmTilesReader, Compression, HashMapCache, Header, MmapBackend, PmTilesWriter, TileCoord,
    TileId, TileType,
};

use super::common::{
    EncodeCache, EncodedTile, PmTilesGeography, TileStats, encode_tile, make_encode_cache,
    make_progress_bar,
};
use super::{ContainerFormat, update_mlt_pmtiles_metadata};

/// Re-encode a `.pmtiles` input (MVT) into the requested container.
pub async fn convert(
    input: &Path,
    output: (&Path, ContainerFormat),
    cfg: EncoderConfig,
    tile_compression: Compression,
) -> AnyResult<()> {
    match output {
        (output, ContainerFormat::Pmtiles) => {
            convert_pmtiles_to_pmtiles(input, output, cfg, tile_compression).await
        }
        (output, _) => bail!(
            "Output must be a .pmtiles file when input is a .pmtiles file, got: {}",
            output.display()
        ),
    }
}

/// Mmap-backed reader over a local `.pmtiles` file.
/// The [`HashMapCache`] avoids re-decoding leaf directories on every `get_tile`.
type PmReader = AsyncPmTilesReader<MmapBackend, HashMapCache>;

/// Maps `PMTiles` tile compression to the [`Encoding`] used by `encode_tile`.
/// `PMTiles` has no zlib/deflate variant.
fn compression_to_encoding(compression: Compression) -> AnyResult<Encoding> {
    match compression {
        Compression::None => Ok(Encoding::Uncompressed),
        Compression::Gzip => Ok(Encoding::Gzip),
        Compression::Brotli => Ok(Encoding::Brotli),
        Compression::Zstd => Ok(Encoding::Zstd),
        Compression::Unknown => bail!("input .pmtiles uses an unknown tile compression"),
    }
}

/// Copy the source metadata JSON, overriding the MLT format and outer compression.
fn mlt_pmtiles_metadata(metadata: &str, tile_compression: Compression) -> AnyResult<String> {
    let mut value: serde_json::Value =
        serde_json::from_str(metadata).unwrap_or_else(|_| serde_json::json!({}));
    if let Some(obj) = value.as_object_mut() {
        update_mlt_pmtiles_metadata(obj, tile_compression);
    }
    Ok(serde_json::to_string(&value)?)
}

fn geography_from_header(source: &Header) -> PmTilesGeography {
    PmTilesGeography {
        min_zoom: Some(source.min_zoom),
        max_zoom: Some(source.max_zoom),
        bounds: Some((
            source.min_longitude,
            source.min_latitude,
            source.max_longitude,
            source.max_latitude,
        )),
        center: Some((
            source.center_longitude,
            source.center_latitude,
            source.center_zoom,
        )),
    }
}

/// Open a local `.pmtiles`, require MVT tiles, and report its tile [`Encoding`].
async fn open_mvt_pmtiles(input: &Path) -> AnyResult<(Arc<PmReader>, Encoding)> {
    let reader =
        Arc::new(AsyncPmTilesReader::new_with_cached_path(HashMapCache::default(), input).await?);
    let header = reader.get_header();
    if header.tile_type != TileType::Mvt {
        bail!(
            "Expected MVT tiles, got {:?} in {}",
            header.tile_type,
            input.display()
        );
    }
    let encoding = compression_to_encoding(header.tile_compression)?;
    Ok((reader, encoding))
}

/// Flatten the archive's run-length data entries into individual tile ids.
async fn collect_pmtiles_ids(reader: &Arc<PmReader>) -> AnyResult<Vec<TileId>> {
    let mut ids = Vec::new();
    let mut entries = reader.clone().entries();
    while let Some(entry) = entries.try_next().await? {
        ids.extend(entry.iter_coords());
    }
    Ok(ids)
}

/// Max in-flight tiles per CPU, bounding memory while keeping every core fed.
/// Encode time varies wildly per tile, so the window must be deep to hide stragglers.
const PIPELINE_DEPTH_PER_CORE: usize = 32;

/// How often to log progress when the live bar is hidden (non-terminal stderr).
const PROGRESS_LOG_EVERY: u64 = 5_000_000;

/// Emit one plain-text progress line (used when the live bar is hidden).
#[expect(
    clippy::cast_precision_loss,
    reason = "tile counts and rates are approximate progress reporting"
)]
fn log_progress_line(done: u64, total: u64, elapsed: std::time::Duration) {
    let rate = done as f64 / elapsed.as_secs_f64().max(f64::EPSILON);
    let eta_min = total.saturating_sub(done) as f64 / rate / 60.0;
    eprintln!("  {done}/{total} tiles ({rate:.0}/s, eta {eta_min:.0} min)");
}

/// Re-encodes every tile MVT -> MLT and emits results in ascending tile-id order,
/// so [`PmTilesWriter`] keeps the archive clustered and run-length encoded.
/// Three thread roles: a reader pulls raw MVT off the mmap, a worker pool encodes
/// in parallel via a lock-free MPMC channel, and an emitter reorders results back
/// into id order. A `cap`-sized permit pool backpressures the reader.
fn spawn_encode_pipeline(
    reader: Arc<PmReader>,
    ids: Vec<TileId>,
    encoding: Encoding,
    cfg: EncoderConfig,
    cache: EncodeCache,
) -> tokio::sync::mpsc::Receiver<AnyResult<EncodedTile>> {
    let parallelism = thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get);
    let cap = (parallelism * PIPELINE_DEPTH_PER_CORE).max(8);

    // Ordered output to the (async) consumer. The real backpressure is the
    // permit pool below, so this buffer only needs to cover one wave of cores.
    let (out_tx, out_rx) = tokio::sync::mpsc::channel(parallelism);

    thread::spawn(move || {
        // Permits cap how far the reader can run ahead of in-order emission.
        let (tok_tx, tok_rx) = mpsc::channel::<()>();
        for _ in 0..cap {
            tok_tx.send(()).expect("permit receiver alive");
        }
        // Raw tiles: reader -> encoder pool, via a lock-free MPMC channel.
        // (`par_bridge` funnels pulls through one mutex, capping parallelism.)
        let (raw_tx, raw_rx) = crossbeam_channel::unbounded::<(usize, TileCoord, Bytes)>();
        // Encoded tiles: encoders -> the in-order emitter (this thread).
        let (res_tx, res_rx) = mpsc::channel::<AnyResult<(usize, EncodedTile)>>();

        // Reader: one sequential pass over ascending ids.
        // Permits are taken only for tiles that exist, so `seq` stays gap-free.
        let reader_thread = {
            let res_tx = res_tx.clone();
            thread::spawn(move || {
                let rt = match tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                {
                    Ok(rt) => rt,
                    Err(e) => {
                        let _ = res_tx.send(Err(e.into()));
                        return;
                    }
                };
                rt.block_on(async move {
                    let mut seq = 0usize;
                    for id in ids {
                        let coord: TileCoord = id.into();
                        match reader.get_tile(id).await {
                            Ok(Some(data)) => {
                                // Throttle before handing the tile downstream.
                                if tok_rx.recv().is_err() {
                                    return; // consumer gone
                                }
                                if raw_tx.send((seq, coord, data)).is_err() {
                                    return;
                                }
                                seq += 1;
                            }
                            Ok(None) => {} // no tile for this id; nothing to emit
                            Err(e) => {
                                let _ = res_tx.send(Err(e.into()));
                                return;
                            }
                        }
                    }
                });
            })
        };

        // Encoders: `parallelism` workers draining the shared MPMC channel.
        let encoder_threads: Vec<_> = (0..parallelism)
            .map(|_| {
                let raw_rx = raw_rx.clone();
                let res_tx = res_tx.clone();
                let cache = cache.clone();
                thread::spawn(move || {
                    for (seq, coord, data) in raw_rx {
                        let result = encode_tile(&cache, &data, encoding, cfg).map(
                            |(data, raw_mvt_size, hit)| {
                                (
                                    seq,
                                    EncodedTile {
                                        coord,
                                        data,
                                        raw_mvt_size,
                                        hit,
                                    },
                                )
                            },
                        );
                        if res_tx.send(result).is_err() {
                            break; // emitter gone
                        }
                    }
                })
            })
            .collect();
        // Only the workers' clones should keep the raw channel open.
        drop(raw_rx);
        // Drop our sender so `res_rx` closes once reader and workers finish.
        drop(res_tx);

        // Emitter: reorders by `seq`, forwards in order, returns a permit per tile sent.
        let mut next = 0usize;
        let mut buffer: BTreeMap<usize, EncodedTile> = BTreeMap::new();
        for msg in res_rx {
            let send = match msg {
                Ok((seq, tile)) => {
                    buffer.insert(seq, tile);
                    let mut consumer_gone = false;
                    while let Some(tile) = buffer.remove(&next) {
                        if out_tx.blocking_send(Ok(tile)).is_err() {
                            consumer_gone = true;
                            break;
                        }
                        let _ = tok_tx.send(());
                        next += 1;
                    }
                    !consumer_gone
                }
                Err(e) => out_tx.blocking_send(Err(e)).is_ok(),
            };
            if !send {
                break; // consumer dropped the receiver; tear down
            }
        }

        // Release the reader (it may be parked on a permit) before joining.
        drop(tok_tx);
        let _ = reader_thread.join();
        for t in encoder_threads {
            let _ = t.join();
        }
    });

    out_rx
}

async fn convert_pmtiles_to_pmtiles(
    input: &Path,
    output: &Path,
    cfg: EncoderConfig,
    tile_compression: Compression,
) -> AnyResult<()> {
    let (reader, encoding) = open_mvt_pmtiles(input).await?;
    let ids = collect_pmtiles_ids(&reader).await?;
    let input_archive_size = std::fs::metadata(input)?.len();

    eprintln!("{} -> {} (pmtiles):", input.display(), output.display());
    let start = Instant::now();
    let bar = make_progress_bar(ids.len() as u64);

    let metadata_str = mlt_pmtiles_metadata(&reader.get_metadata().await?, tile_compression)?;
    let file = std::fs::File::create(output)?;
    let mut writer = geography_from_header(reader.get_header())
        .apply(PmTilesWriter::new(TileType::Mlt))
        .tile_compression(tile_compression)
        .metadata(&metadata_str)
        .create(file)?;

    let mut tiles = spawn_encode_pipeline(reader, ids, encoding, cfg, make_encode_cache());
    let mut stats = TileStats::default();
    // The bar renders nothing when stderr isn't a terminal, so log progress periodically instead.
    let log_progress = bar.is_hidden();
    let mut done: u64 = 0;
    // Tiles arrive in ascending id order, so the writer stays clustered.
    while let Some(tile) = tiles.recv().await {
        let EncodedTile {
            coord,
            data,
            raw_mvt_size,
            hit,
        } = tile?;
        writer.add_tile(coord, &data)?;
        stats.record(data.len() as u64, raw_mvt_size, hit);
        bar.inc(1);
        done += 1;
        if log_progress && done.is_multiple_of(PROGRESS_LOG_EVERY) {
            log_progress_line(done, bar.length().unwrap_or(done), start.elapsed());
        }
    }
    writer.finalize()?;
    let output_archive_size = std::fs::metadata(output)?.len();
    bar.finish_and_clear();
    stats.print_summary(
        start,
        input_archive_size,
        output_archive_size,
        encoding,
        tile_compression,
    );

    Ok(())
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::io::Cursor;
    use std::sync::atomic::{AtomicU64, Ordering};

    use mlt_core::Parser;

    use super::*;

    const HEADER_SIZE: usize = 127;

    fn write_header(writer: PmTilesWriter) -> Vec<u8> {
        let mut bytes = Vec::new();
        writer
            .create(Cursor::new(&mut bytes))
            .expect("create PMTiles writer")
            .finalize()
            .expect("finalize PMTiles writer");
        bytes
    }

    fn parse_header(bytes: &[u8]) -> Header {
        Header::try_from_bytes(Bytes::copy_from_slice(&bytes[..HEADER_SIZE]))
            .expect("parse PMTiles header")
    }

    #[test]
    #[expect(clippy::float_cmp, reason = "bounds are copied verbatim, not computed")]
    fn copies_geographic_header_without_copying_content_encoding() {
        let source = write_header(
            PmTilesWriter::new(TileType::Mvt)
                .min_zoom(3)
                .max_zoom(12)
                .bounds(-12.345_678_9, -67.890_123_4, 98.765_432_1, 54.321_098_7)
                .center_zoom(8)
                .center(11.223_344_5, -44.556_677_8),
        );
        let source = parse_header(&source);

        let output =
            write_header(geography_from_header(&source).apply(PmTilesWriter::new(TileType::Mlt)));
        let output = parse_header(&output);

        assert_eq!(output.min_zoom, source.min_zoom);
        assert_eq!(output.max_zoom, source.max_zoom);
        assert_eq!(output.min_longitude, source.min_longitude);
        assert_eq!(output.min_latitude, source.min_latitude);
        assert_eq!(output.max_longitude, source.max_longitude);
        assert_eq!(output.max_latitude, source.max_latitude);
        assert_eq!(output.center_zoom, source.center_zoom);
        assert_eq!(output.center_longitude, source.center_longitude);
        assert_eq!(output.center_latitude, source.center_latitude);

        assert_eq!(source.tile_compression, Compression::Gzip);
        assert_eq!(output.tile_compression, Compression::None);
        assert_eq!(output.tile_type, TileType::Mlt);
    }

    static NEXT_OUTPUT_ID: AtomicU64 = AtomicU64::new(0);

    struct TempOutput(std::path::PathBuf);

    impl TempOutput {
        fn new() -> Self {
            let id = NEXT_OUTPUT_ID.fetch_add(1, Ordering::Relaxed);
            Self(std::env::temp_dir().join(format!(
                "mlt-convert-test-{}-{id}.pmtiles",
                std::process::id()
            )))
        }
    }

    impl Drop for TempOutput {
        fn drop(&mut self) {
            let _ = fs::remove_file(&self.0);
        }
    }

    #[tokio::test]
    async fn writes_gzip_compressed_mlt_pmtiles() {
        let input = Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("../../test/fixtures/omt-planet-20260112.mvt.max1.pmtiles");
        let output = TempOutput::new();

        convert_pmtiles_to_pmtiles(
            &input,
            &output.0,
            EncoderConfig::default(),
            Compression::Gzip,
        )
        .await
        .expect("conversion succeeds");

        let reader = Arc::new(
            PmReader::new_with_cached_path(HashMapCache::default(), &output.0)
                .await
                .expect("output opens"),
        );
        assert_eq!(reader.get_header().tile_type, TileType::Mlt);
        assert_eq!(reader.get_header().tile_compression, Compression::Gzip);

        let metadata: serde_json::Value =
            serde_json::from_str(&reader.get_metadata().await.expect("metadata reads"))
                .expect("metadata is JSON");
        assert_eq!(metadata["format"], "mlt");
        assert_eq!(metadata["compression"], "gzip");

        let tile_ids = collect_pmtiles_ids(&reader)
            .await
            .expect("tile directory reads");
        let raw_tile = reader
            .get_tile(tile_ids[0])
            .await
            .expect("tile reads")
            .expect("tile exists");
        assert_eq!(&raw_tile[..2], &[0x1f, 0x8b]);

        let tile = reader
            .get_tile_decompressed(tile_ids[0])
            .await
            .expect("tile decompresses")
            .expect("tile exists");
        assert!(
            !Parser::default()
                .parse_layers(&tile)
                .expect("decompressed MLT tile parses")
                .is_empty()
        );
    }
}