//! PMTiles v3 export — the seam into the cloud-native tile ecosystem.
//!
//! The open geospatial stack converged on a single-file, range-requestable
//! archive: **PMTiles** (a `.pmtiles` blob served straight from object storage,
//! read by MapLibre, the `pmtiles` CLIs, and Overture tooling). map's "a tileset
//! is a file, not a service" doctrine is the same idea; this crate emits the
//! incumbent's format so map output is consumable without a map-specific reader.
//!
//! What's here is the **PMTiles v3 writer**, fully self-contained and
//! round-trip-validated: Hilbert tile-id ordering, the varint clustered
//! directory, and the 127-byte header. Directories and metadata are written
//! uncompressed (`internal compression = none`), so the crate needs no gzip
//! dependency and the bytes are deterministic.
//!
//! The map adapter ([`from_map_tiles`]) filters a tileset to its **surface**
//! tiles — `alt == 0`, atemporal, base LOD — because PMTiles' 2-D z/x/y grid
//! cannot carry map's altitude / time / LOD axes (that 4-D coverage is what
//! GeoParquet is for). The S2 → z/x/y projection is supplied by the caller (it
//! lives in map's substrate), keeping this crate a pure, testable serializer.

use map_schema::{CellKey, S2CellId, Tile};
use serde_json::Value;

/// A PMTiles tile type byte (header field).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum TileType {
    /// Unknown / opaque.
    Unknown = 0,
    /// Mapbox Vector Tile.
    Mvt = 1,
    /// PNG.
    Png = 2,
    /// JPEG.
    Jpeg = 3,
    /// WebP.
    Webp = 4,
}

impl TileType {
    /// Best-effort tile type from a map codec tag.
    pub fn from_codec(codec: &str) -> Self {
        let c = codec.to_ascii_lowercase();
        if c.starts_with("mvt") {
            TileType::Mvt
        } else if c.starts_with("png") {
            TileType::Png
        } else if c.starts_with("jpeg") || c.starts_with("jpg") {
            TileType::Jpeg
        } else if c.starts_with("webp") {
            TileType::Webp
        } else {
            TileType::Unknown
        }
    }
}

/// One directory entry: a tile id, its byte range in the tile-data section, and
/// a run length (how many consecutive tile ids share these bytes).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Entry {
    tile_id: u64,
    offset: u64,
    length: u64,
    run_length: u64,
}

/// Write a PMTiles v3 archive from `(z, x, y, bytes)` tiles and a metadata JSON
/// value. Tiles are sorted into Hilbert order; the whole directory is the root
/// (suitable up to ~a few thousand tiles — large tilesets need leaf directories,
/// noted but not split here).
pub fn write(
    tiles: impl IntoIterator<Item = (u8, u32, u32, Vec<u8>)>,
    metadata: &Value,
    tile_type: TileType,
) -> Vec<u8> {
    let mut rows: Vec<(u8, u64, Vec<u8>)> =
        tiles.into_iter().map(|(z, x, y, d)| (z, tile_id(z, x, y), d)).collect();
    rows.sort_by_key(|(_, id, _)| *id);

    let (mut min_zoom, mut max_zoom) = (u8::MAX, 0u8);
    let mut tiledata: Vec<u8> = Vec::new();
    let mut entries: Vec<Entry> = Vec::with_capacity(rows.len());
    for (z, id, data) in &rows {
        min_zoom = min_zoom.min(*z);
        max_zoom = max_zoom.max(*z);
        entries.push(Entry {
            tile_id: *id,
            offset: tiledata.len() as u64,
            length: data.len() as u64,
            run_length: 1,
        });
        tiledata.extend_from_slice(data);
    }
    if rows.is_empty() {
        min_zoom = 0;
    }

    let root_dir = serialize_directory(&entries);
    let meta_bytes = serde_json::to_vec(metadata).unwrap_or_else(|_| b"{}".to_vec());

    let header_len = 127u64;
    let root_off = header_len;
    let meta_off = root_off + root_dir.len() as u64;
    let leaf_off = meta_off + meta_bytes.len() as u64;
    let tiledata_off = leaf_off; // no leaf directories

    let n = entries.len() as u64;
    let header = Header {
        root_offset: root_off,
        root_length: root_dir.len() as u64,
        metadata_offset: meta_off,
        metadata_length: meta_bytes.len() as u64,
        leaf_offset: leaf_off,
        leaf_length: 0,
        tile_data_offset: tiledata_off,
        tile_data_length: tiledata.len() as u64,
        num_addressed_tiles: n,
        num_tile_entries: n,
        num_tile_contents: n,
        clustered: true,
        internal_compression: 1, // none
        tile_compression: 1,     // none
        tile_type: tile_type as u8,
        min_zoom,
        max_zoom,
    };

    let mut out = header.to_bytes().to_vec();
    out.extend_from_slice(&root_dir);
    out.extend_from_slice(&meta_bytes);
    out.extend_from_slice(&tiledata);
    out
}

/// Export the **surface** tiles of a map tileset (alt 0, atemporal, base LOD) to
/// a PMTiles archive. `project` maps an S2 cell to a slippy `(z, x, y)`; tiles it
/// returns `None` for, or non-surface tiles, are skipped. The tile type is taken
/// from the first tile's codec.
pub fn from_map_tiles(
    tiles: &[Tile],
    project: impl Fn(S2CellId) -> Option<(u8, u32, u32)>,
    metadata: &Value,
) -> Vec<u8> {
    let tile_type =
        tiles.first().map(|t| TileType::from_codec(&t.codec)).unwrap_or(TileType::Unknown);
    let projected: Vec<(u8, u32, u32, Vec<u8>)> = tiles
        .iter()
        .filter(|t| is_surface(&t.key))
        .filter_map(|t| project(t.key.s2).map(|(z, x, y)| (z, x, y, t.data.clone())))
        .collect();
    write(projected, metadata, tile_type)
}

/// Whether a cell is a 2-D-projectable surface tile (PMTiles can't hold altitude
/// bands, time buckets, or non-base LODs).
pub fn is_surface(key: &CellKey) -> bool {
    key.alt.0 == 0 && key.time.is_atemporal() && key.lod.0 == 0
}

/// Read a tile's bytes back out of a PMTiles archive (the root-only reader used
/// to round-trip-validate [`write`]; a production reader would also follow leaf
/// directories).
pub fn read_tile(archive: &[u8], z: u8, x: u32, y: u32) -> Option<Vec<u8>> {
    let header = Header::from_bytes(archive.get(..127)?)?;
    let root = archive.get(header.root_offset as usize..(header.root_offset + header.root_length) as usize)?;
    let entries = deserialize_directory(root)?;
    let want = tile_id(z, x, y);
    let e = entries.iter().find(|e| e.tile_id <= want && want < e.tile_id + e.run_length.max(1))?;
    let base = header.tile_data_offset + e.offset;
    archive.get(base as usize..(base + e.length) as usize).map(<[u8]>::to_vec)
}

// ── Hilbert tile ids (PMTiles convention) ────────────────────────────────────

/// The first tile id at zoom `z` — `(4^z − 1) / 3`.
fn base_id(z: u8) -> u64 {
    ((1u64 << (2 * z as u32)) - 1) / 3
}

/// `(z, x, y) → tile id`: zoom base + Hilbert distance.
pub fn tile_id(z: u8, x: u32, y: u32) -> u64 {
    base_id(z) + xy2d(z, x as u64, y as u64)
}

/// `tile id → (z, x, y)` — the inverse, for verification.
pub fn zxy(mut id: u64) -> (u8, u32, u32) {
    let mut z = 0u8;
    while z < 31 && id >= base_id(z + 1) {
        z += 1;
    }
    id -= base_id(z);
    let (x, y) = d2xy(z, id);
    (z, x as u32, y as u32)
}

fn xy2d(z: u8, mut x: u64, mut y: u64) -> u64 {
    let n = 1u64 << z;
    let mut d = 0u64;
    let mut s = n / 2;
    while s > 0 {
        let rx = u64::from((x & s) > 0);
        let ry = u64::from((y & s) > 0);
        d += s * s * ((3 * rx) ^ ry);
        // rotate (canonical Hilbert, full-grid n)
        if ry == 0 {
            if rx == 1 {
                x = n - 1 - x;
                y = n - 1 - y;
            }
            std::mem::swap(&mut x, &mut y);
        }
        s /= 2;
    }
    d
}

fn d2xy(z: u8, mut t: u64) -> (u64, u64) {
    let n = 1u64 << z;
    let (mut x, mut y) = (0u64, 0u64);
    let mut s = 1u64;
    while s < n {
        let rx = 1 & (t / 2);
        let ry = 1 & (t ^ rx);
        // rotate (per-level s)
        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, y)
}

// ── Clustered directory (PMTiles v3) ─────────────────────────────────────────

fn serialize_directory(entries: &[Entry]) -> Vec<u8> {
    let mut buf = Vec::new();
    write_uvarint(&mut buf, entries.len() as u64);
    // column: tile-id deltas
    let mut last = 0u64;
    for e in entries {
        write_uvarint(&mut buf, e.tile_id - last);
        last = e.tile_id;
    }
    // column: run lengths
    for e in entries {
        write_uvarint(&mut buf, e.run_length);
    }
    // column: lengths
    for e in entries {
        write_uvarint(&mut buf, e.length);
    }
    // column: offsets (0 = immediately follows previous; else offset + 1)
    for (i, e) in entries.iter().enumerate() {
        if i > 0 && e.offset == entries[i - 1].offset + entries[i - 1].length {
            write_uvarint(&mut buf, 0);
        } else {
            write_uvarint(&mut buf, e.offset + 1);
        }
    }
    buf
}

fn deserialize_directory(buf: &[u8]) -> Option<Vec<Entry>> {
    let mut pos = 0usize;
    let n = read_uvarint(buf, &mut pos)? as usize;
    let mut entries = vec![Entry { tile_id: 0, offset: 0, length: 0, run_length: 0 }; n];
    let mut last = 0u64;
    for e in entries.iter_mut() {
        last += read_uvarint(buf, &mut pos)?;
        e.tile_id = last;
    }
    for e in entries.iter_mut() {
        e.run_length = read_uvarint(buf, &mut pos)?;
    }
    for e in entries.iter_mut() {
        e.length = read_uvarint(buf, &mut pos)?;
    }
    for i in 0..n {
        let v = read_uvarint(buf, &mut pos)?;
        entries[i].offset =
            if v == 0 { entries[i - 1].offset + entries[i - 1].length } else { v - 1 };
    }
    Some(entries)
}

fn write_uvarint(buf: &mut Vec<u8>, mut v: u64) {
    while v >= 0x80 {
        buf.push((v as u8) | 0x80);
        v >>= 7;
    }
    buf.push(v as u8);
}

fn read_uvarint(buf: &[u8], pos: &mut usize) -> Option<u64> {
    let mut result = 0u64;
    let mut shift = 0u32;
    loop {
        let b = *buf.get(*pos)?;
        *pos += 1;
        result |= ((b & 0x7f) as u64) << shift;
        if b & 0x80 == 0 {
            return Some(result);
        }
        shift += 7;
        if shift >= 64 {
            return None;
        }
    }
}

// ── 127-byte header ──────────────────────────────────────────────────────────

struct Header {
    root_offset: u64,
    root_length: u64,
    metadata_offset: u64,
    metadata_length: u64,
    leaf_offset: u64,
    leaf_length: u64,
    tile_data_offset: u64,
    tile_data_length: u64,
    num_addressed_tiles: u64,
    num_tile_entries: u64,
    num_tile_contents: u64,
    clustered: bool,
    internal_compression: u8,
    tile_compression: u8,
    tile_type: u8,
    min_zoom: u8,
    max_zoom: u8,
}

impl Header {
    fn to_bytes(&self) -> [u8; 127] {
        let mut b = [0u8; 127];
        b[..7].copy_from_slice(b"PMTiles");
        b[7] = 3; // spec version
        let mut put = |off: usize, v: u64| b[off..off + 8].copy_from_slice(&v.to_le_bytes());
        put(8, self.root_offset);
        put(16, self.root_length);
        put(24, self.metadata_offset);
        put(32, self.metadata_length);
        put(40, self.leaf_offset);
        put(48, self.leaf_length);
        put(56, self.tile_data_offset);
        put(64, self.tile_data_length);
        put(72, self.num_addressed_tiles);
        put(80, self.num_tile_entries);
        put(88, self.num_tile_contents);
        b[96] = self.clustered as u8;
        b[97] = self.internal_compression;
        b[98] = self.tile_compression;
        b[99] = self.tile_type;
        b[100] = self.min_zoom;
        b[101] = self.max_zoom;
        // bytes 102..127: bounds (i32 E7 ×4), center zoom, center (i32 ×2) — left
        // as world default (0). A reader keys on the directory, not these.
        b
    }

    fn from_bytes(b: &[u8]) -> Option<Self> {
        if b.len() < 127 || &b[..7] != b"PMTiles" || b[7] != 3 {
            return None;
        }
        let get = |off: usize| u64::from_le_bytes(b[off..off + 8].try_into().unwrap());
        Some(Header {
            root_offset: get(8),
            root_length: get(16),
            metadata_offset: get(24),
            metadata_length: get(32),
            leaf_offset: get(40),
            leaf_length: get(48),
            tile_data_offset: get(56),
            tile_data_length: get(64),
            num_addressed_tiles: get(72),
            num_tile_entries: get(80),
            num_tile_contents: get(88),
            clustered: b[96] != 0,
            internal_compression: b[97],
            tile_compression: b[98],
            tile_type: b[99],
            min_zoom: b[100],
            max_zoom: b[101],
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn hilbert_anchors_and_round_trip() {
        // The z=1 Hilbert order PMTiles uses.
        assert_eq!(tile_id(0, 0, 0), 0);
        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);
        // round-trip across zooms
        for (z, x, y) in [(0u8, 0u32, 0u32), (3, 5, 2), (8, 100, 200), (14, 8000, 100)] {
            assert_eq!(zxy(tile_id(z, x, y)), (z, x, y), "rt {z}/{x}/{y}");
        }
    }

    #[test]
    fn directory_round_trips() {
        let entries = vec![
            Entry { tile_id: 1, offset: 0, length: 10, run_length: 1 },
            Entry { tile_id: 2, offset: 10, length: 20, run_length: 1 }, // consecutive
            Entry { tile_id: 9, offset: 100, length: 5, run_length: 1 },
        ];
        let bytes = serialize_directory(&entries);
        assert_eq!(deserialize_directory(&bytes).unwrap(), entries);
    }

    #[test]
    fn header_round_trips() {
        let h = Header {
            root_offset: 127,
            root_length: 40,
            metadata_offset: 167,
            metadata_length: 2,
            leaf_offset: 169,
            leaf_length: 0,
            tile_data_offset: 169,
            tile_data_length: 35,
            num_addressed_tiles: 3,
            num_tile_entries: 3,
            num_tile_contents: 3,
            clustered: true,
            internal_compression: 1,
            tile_compression: 1,
            tile_type: TileType::Mvt as u8,
            min_zoom: 0,
            max_zoom: 14,
        };
        let b = h.to_bytes();
        assert_eq!(&b[..7], b"PMTiles");
        assert_eq!(b[7], 3);
        let h2 = Header::from_bytes(&b).unwrap();
        assert_eq!(h2.tile_data_offset, 169);
        assert_eq!(h2.max_zoom, 14);
        assert_eq!(h2.tile_type, TileType::Mvt as u8);
    }

    #[test]
    fn archive_write_then_read_tile() {
        let tiles = vec![
            (0u8, 0u32, 0u32, b"root-tile".to_vec()),
            (2, 1, 1, b"a-tile".to_vec()),
            (2, 3, 2, b"another".to_vec()),
        ];
        let archive = write(tiles, &json!({ "name": "demo" }), TileType::Mvt);
        // header is intact
        assert_eq!(&archive[..7], b"PMTiles");
        // every tile reads back byte-identical, by z/x/y
        assert_eq!(read_tile(&archive, 0, 0, 0).unwrap(), b"root-tile");
        assert_eq!(read_tile(&archive, 2, 1, 1).unwrap(), b"a-tile");
        assert_eq!(read_tile(&archive, 2, 3, 2).unwrap(), b"another");
        // a tile that isn't there
        assert!(read_tile(&archive, 5, 0, 0).is_none());
    }
}