rustial-engine 0.0.1

//! HTTP vector tile source with MVT binary decoding.
//!
//! [`HttpVectorTileSource`] is a [`TileSource`] implementation that
//! fetches binary Mapbox Vector Tiles (MVT/PBF) over HTTP, decodes
//! them using the engine's built-in MVT decoder, and produces
//! [`TileData::Vector`] payloads containing per-source-layer feature
//! collections.
//!
//! ## Architecture
//!
//! This is the engine-side equivalent of MapLibre's
//! `VectorTileSource` + `WorkerTile` pipeline.  In MapLibre:
//!
//! 1. `VectorTileSource.loadTile()` sends PBF bytes to a web worker.
//! 2. `WorkerTile.parse()` decodes the PBF and builds render buckets.
//! 3. The result is sent back to the main thread.
//!
//! In Rustial the same logical pipeline runs on the main thread during
//! [`TileSource::poll`]:
//!
//! 1. `HttpClient` fetches the PBF bytes.
//! 2. `decode_mvt()` decodes the protobuf into per-layer features.
//! 3. The result is wrapped in [`TileData::Vector`] and returned.
//!
//! ## URL template
//!
//! The source is constructed with a URL template containing `{z}`,
//! `{x}`, and `{y}` placeholders:
//!
//! ```text
//! https://demotiles.maplibre.org/tiles/{z}/{x}/{y}.pbf
//! ```
//!
//! ## TileJSON integration
//!
//! The source can optionally be configured with a [`TileJson`] metadata
//! object that provides source zoom range and bounds filtering, matching
//! MapLibre's `loadTileJSON()` ? source metadata flow.
//!
//! ## Thread safety
//!
//! `HttpVectorTileSource` is `Send + Sync`.
//!
//! [`TileSource`]: crate::tile_source::TileSource
//! [`TileJson`]: crate::tilejson::TileJson

use crate::io::{HttpClient, HttpRequest, HttpResponse};
use crate::mvt::{decode_mvt, MvtDecodeOptions};
use crate::tile_source::{
    RevalidationHint, TileData, TileError, TileFreshness, TileResponse, TileSource, VectorTileData,
};
use crate::tilejson::TileJson;
use rustial_math::TileId;
use std::collections::HashMap;
use std::sync::Mutex;
use std::time::{Duration, SystemTime};

// ---------------------------------------------------------------------------
// Freshness parsing (shared with HttpTileSource)
// ---------------------------------------------------------------------------

fn parse_cache_control_max_age(value: &str) -> Option<u64> {
    for directive in value.split(',') {
        let directive = directive.trim();
        if let Some(rest) = directive.strip_prefix("max-age=") {
            if let Ok(seconds) = rest.trim_matches('"').parse::<u64>() {
                return Some(seconds);
            }
        }
    }
    None
}

fn parse_age_seconds(response: &HttpResponse) -> u64 {
    response
        .header("age")
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or(0)
}

fn parse_http_freshness(response: &HttpResponse) -> TileFreshness {
    let now = SystemTime::now();
    let age = parse_age_seconds(response);

    let expires_at = response
        .header("cache-control")
        .and_then(parse_cache_control_max_age)
        .map(|max_age| max_age.saturating_sub(age))
        .map(Duration::from_secs)
        .and_then(|ttl| now.checked_add(ttl))
        .or_else(|| {
            response
                .header("expires")
                .and_then(|value| httpdate::parse_http_date(value).ok())
        });

    TileFreshness {
        expires_at,
        etag: response.header("etag").map(ToOwned::to_owned),
        last_modified: response.header("last-modified").map(ToOwned::to_owned),
    }
}

// ---------------------------------------------------------------------------
// HttpVectorTileSource
// ---------------------------------------------------------------------------

/// A [`TileSource`] that fetches and decodes Mapbox Vector Tiles (PBF)
/// over HTTP.
///
/// See the [module-level documentation](self) for the full pipeline
/// description.
pub struct HttpVectorTileSource {
    /// URL template with `{z}`, `{x}`, `{y}` placeholders.
    url_template: String,

    /// The HTTP client provided by the host application.
    client: Box<dyn HttpClient>,

    /// Extra headers added to every outgoing request.
    default_headers: Vec<(String, String)>,

    /// MVT decode options (e.g. layer filter).
    decode_options: MvtDecodeOptions,

    /// Optional TileJSON metadata for source zoom / bounds filtering.
    tilejson: Option<TileJson>,

    /// Mapping from request URL to the originating `TileId`.
    pending: Mutex<HashMap<String, TileId>>,

    /// When `true`, [`poll`](TileSource::poll) returns
    /// [`TileData::RawVector`] payloads instead of fully decoded
    /// [`TileData::Vector`], allowing the caller to offload the
    /// CPU-heavy MVT protobuf decode to a background thread via
    /// [`DataTaskPool::spawn_decode`](crate::async_data::DataTaskPool::spawn_decode).
    deferred_decode: bool,
}

impl std::fmt::Debug for HttpVectorTileSource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let pending_count = self.pending.lock().map(|p| p.len()).unwrap_or(0);
        f.debug_struct("HttpVectorTileSource")
            .field("url_template", &self.url_template)
            .field("has_tilejson", &self.tilejson.is_some())
            .field("pending", &pending_count)
            .finish()
    }
}

impl HttpVectorTileSource {
    /// Create a new HTTP vector tile source.
    ///
    /// The URL template should produce PBF/MVT binary responses,
    /// e.g. `"https://demotiles.maplibre.org/tiles/{z}/{x}/{y}.pbf"`.
    pub fn new(url_template: impl Into<String>, client: Box<dyn HttpClient>) -> Self {
        Self {
            url_template: url_template.into(),
            client,
            default_headers: Vec::new(),
            decode_options: MvtDecodeOptions::default(),
            tilejson: None,
            pending: Mutex::new(HashMap::new()),
            deferred_decode: false,
        }
    }

    /// Enable deferred decoding mode.
    ///
    /// When enabled, [`poll`](TileSource::poll) returns
    /// [`TileData::RawVector`] payloads carrying the raw PBF bytes
    /// instead of performing the MVT decode inline.  The caller is
    /// responsible for submitting the decode work to a background
    /// thread and promoting the tile cache entry once complete.
    pub fn with_deferred_decode(mut self, deferred: bool) -> Self {
        self.deferred_decode = deferred;
        self
    }

    /// Whether deferred decoding is enabled.
    #[inline]
    pub fn deferred_decode(&self) -> bool {
        self.deferred_decode
    }

    /// Attach TileJSON metadata to configure source zoom range and
    /// bounds filtering.
    pub fn with_tilejson(mut self, tilejson: TileJson) -> Self {
        self.tilejson = Some(tilejson);
        self
    }

    /// Set MVT decode options (e.g. layer filter).
    pub fn with_decode_options(mut self, options: MvtDecodeOptions) -> Self {
        self.decode_options = options;
        self
    }

    /// Add a default header sent with every tile request.
    pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.default_headers.push((name.into(), value.into()));
        self
    }

    /// Expand the URL template for a given tile ID.
    pub fn tile_url(&self, id: &TileId) -> String {
        self.url_template
            .replace("{z}", &id.zoom.to_string())
            .replace("{x}", &id.x.to_string())
            .replace("{y}", &id.y.to_string())
    }

    /// The URL template this source was constructed with.
    #[inline]
    pub fn url_template(&self) -> &str {
        &self.url_template
    }

    /// The number of requests currently in flight.
    pub fn pending_count(&self) -> usize {
        self.pending.lock().map(|p| p.len()).unwrap_or(0)
    }

    /// Reference to the attached TileJSON metadata, if any.
    pub fn tilejson(&self) -> Option<&TileJson> {
        self.tilejson.as_ref()
    }

    /// Decode raw MVT bytes for a tile into a `VectorTileData`.
    fn decode_tile_bytes(
        &self,
        bytes: &[u8],
        tile_id: &TileId,
    ) -> Result<VectorTileData, TileError> {
        let layers = decode_mvt(bytes, tile_id, &self.decode_options)
            .map_err(|e| TileError::Decode(format!("MVT decode: {e}")))?;

        Ok(VectorTileData { layers })
    }
}

impl TileSource for HttpVectorTileSource {
    fn request(&self, id: TileId) {
        let url = self.tile_url(&id);

        if let Ok(mut pending) = self.pending.lock() {
            pending.insert(url.clone(), id);
        }

        let mut req = HttpRequest::get(&url);
        for (name, value) in &self.default_headers {
            req = req.with_header(name.clone(), value.clone());
        }

        self.client.send(req);
    }

    fn request_revalidate(&self, id: TileId, hint: RevalidationHint) {
        let url = self.tile_url(&id);

        if let Ok(mut pending) = self.pending.lock() {
            pending.insert(url.clone(), id);
        }

        let mut req = HttpRequest::get(&url);
        for (name, value) in &self.default_headers {
            req = req.with_header(name.clone(), value.clone());
        }

        if let Some(etag) = &hint.etag {
            req = req.with_header("If-None-Match", etag.clone());
        }
        if let Some(last_modified) = &hint.last_modified {
            req = req.with_header("If-Modified-Since", last_modified.clone());
        }

        self.client.send(req);
    }

    fn poll(&self) -> Vec<(TileId, Result<TileResponse, TileError>)> {
        let responses = self.client.poll();
        if responses.is_empty() {
            return Vec::new();
        }

        let mut pending = match self.pending.lock() {
            Ok(p) => p,
            Err(_) => return Vec::new(),
        };

        let mut results = Vec::with_capacity(responses.len());

        for (url, response) in responses {
            let tile_id = match pending.remove(&url) {
                Some(id) => id,
                None => continue,
            };

            match response {
                Ok(resp) if resp.status == 304 => {
                    let freshness = parse_http_freshness(&resp);
                    results.push((tile_id, Ok(TileResponse::not_modified(freshness))));
                }
                Ok(resp) if resp.is_success() => {
                    let freshness = parse_http_freshness(&resp);
                    if self.deferred_decode {
                        let raw = crate::tile_source::RawVectorPayload {
                            tile_id,
                            bytes: std::sync::Arc::new(resp.body),
                            decode_options: self.decode_options.clone(),
                        };
                        results.push((
                            tile_id,
                            Ok(TileResponse {
                                data: TileData::RawVector(raw),
                                freshness,
                                not_modified: false,
                            }),
                        ));
                    } else {
                        let tile_result =
                            self.decode_tile_bytes(&resp.body, &tile_id)
                                .map(|vector_data| TileResponse {
                                    data: TileData::Vector(vector_data),
                                    freshness,
                                    not_modified: false,
                                });
                        results.push((tile_id, tile_result));
                    }
                }
                Ok(resp) if resp.status == 404 => {
                    results.push((tile_id, Err(TileError::NotFound(tile_id))));
                }
                Ok(resp) => {
                    results.push((
                        tile_id,
                        Err(TileError::Network(format!("HTTP {}", resp.status))),
                    ));
                }
                Err(err) => {
                    results.push((tile_id, Err(TileError::Network(err))));
                }
            }
        }

        results
    }

    fn cancel(&self, id: TileId) {
        if let Ok(mut pending) = self.pending.lock() {
            let url = self.tile_url(&id);
            pending.remove(&url);
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::io::HttpResponse;
    use std::sync::{Arc, Mutex as StdMutex};

    struct MockClient {
        sent: StdMutex<Vec<HttpRequest>>,
        responses: StdMutex<Vec<(String, Result<HttpResponse, String>)>>,
    }

    impl MockClient {
        fn new() -> Self {
            Self {
                sent: StdMutex::new(Vec::new()),
                responses: StdMutex::new(Vec::new()),
            }
        }

        fn queue_response(&self, url: &str, status: u16, body: Vec<u8>) {
            self.responses.lock().unwrap().push((
                url.to_string(),
                Ok(HttpResponse {
                    status,
                    body,
                    headers: vec![],
                }),
            ));
        }
    }

    impl HttpClient for MockClient {
        fn send(&self, request: HttpRequest) {
            self.sent.lock().unwrap().push(request);
        }

        fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
            std::mem::take(&mut *self.responses.lock().unwrap())
        }
    }

    const TEMPLATE: &str = "https://tiles.example.com/{z}/{x}/{y}.pbf";

    // Helper to build a minimal valid MVT tile with one point feature.
    fn build_test_mvt() -> Vec<u8> {
        fn encode_varint(mut val: u64) -> Vec<u8> {
            let mut buf = Vec::new();
            loop {
                let mut byte = (val & 0x7F) as u8;
                val >>= 7;
                if val != 0 {
                    byte |= 0x80;
                }
                buf.push(byte);
                if val == 0 {
                    break;
                }
            }
            buf
        }
        fn encode_tag(field: u32, wt: u8) -> Vec<u8> {
            encode_varint(((field as u64) << 3) | wt as u64)
        }
        fn encode_ld(field: u32, data: &[u8]) -> Vec<u8> {
            let mut b = encode_tag(field, 2);
            b.extend(encode_varint(data.len() as u64));
            b.extend_from_slice(data);
            b
        }
        fn encode_vi(field: u32, val: u64) -> Vec<u8> {
            let mut b = encode_tag(field, 0);
            b.extend(encode_varint(val));
            b
        }
        fn zigzag(n: i32) -> u32 {
            ((n << 1) ^ (n >> 31)) as u32
        }

        // Geometry: MoveTo(2048, 2048) - center of tile
        let mut geom = Vec::new();
        geom.extend(encode_varint(((1u64) << 3) | 1)); // MoveTo, count=1
        geom.extend(encode_varint(zigzag(2048) as u64));
        geom.extend(encode_varint(zigzag(2048) as u64));

        // Feature: type=POINT, geometry
        let mut feat = Vec::new();
        feat.extend(encode_vi(3, 1)); // POINT
        feat.extend(encode_ld(4, &geom));

        // Layer: name="test", feature, extent=4096, version=2
        let mut layer = Vec::new();
        layer.extend(encode_ld(1, b"test"));
        layer.extend(encode_ld(2, &feat));
        layer.extend(encode_vi(5, 4096));
        layer.extend(encode_vi(15, 2));

        // Tile: layer
        encode_ld(3, &layer)
    }

    #[test]
    fn url_template_substitution() {
        let client = MockClient::new();
        let source = HttpVectorTileSource::new(TEMPLATE, Box::new(client));
        let url = source.tile_url(&TileId::new(10, 512, 340));
        assert_eq!(url, "https://tiles.example.com/10/512/340.pbf");
    }

    #[test]
    fn request_sends_http_get() {
        let _client = MockClient::new();
        let sent = Arc::new(StdMutex::new(Vec::new()));
        let sent_clone = sent.clone();

        struct TrackingClient {
            sent: Arc<StdMutex<Vec<String>>>,
        }
        impl HttpClient for TrackingClient {
            fn send(&self, request: HttpRequest) {
                self.sent.lock().unwrap().push(request.url);
            }
            fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
                Vec::new()
            }
        }

        let source =
            HttpVectorTileSource::new(TEMPLATE, Box::new(TrackingClient { sent: sent_clone }));
        source.request(TileId::new(5, 10, 20));

        let urls = sent.lock().unwrap().clone();
        assert_eq!(urls, vec!["https://tiles.example.com/5/10/20.pbf"]);
    }

    #[test]
    fn successful_fetch_decodes_mvt() {
        let client = MockClient::new();
        let url = "https://tiles.example.com/0/0/0.pbf";
        let mvt_bytes = build_test_mvt();
        client.queue_response(url, 200, mvt_bytes);

        let source = HttpVectorTileSource::new(TEMPLATE, Box::new(client));
        source.request(TileId::new(0, 0, 0));

        let results = source.poll();
        assert_eq!(results.len(), 1);
        let (id, result) = &results[0];
        assert_eq!(*id, TileId::new(0, 0, 0));

        let response = result.as_ref().expect("should succeed");
        match &response.data {
            TileData::Vector(vt) => {
                assert!(vt.layers.contains_key("test"));
                assert_eq!(vt.layers["test"].len(), 1);
            }
            other => panic!("expected Vector tile data, got {:?}", other),
        }
    }

    #[test]
    fn http_404_returns_not_found() {
        let client = MockClient::new();
        client.queue_response("https://tiles.example.com/0/0/0.pbf", 404, vec![]);

        let source = HttpVectorTileSource::new(TEMPLATE, Box::new(client));
        source.request(TileId::new(0, 0, 0));

        let results = source.poll();
        assert_eq!(results.len(), 1);
        assert!(matches!(results[0].1, Err(TileError::NotFound(_))));
    }

    #[test]
    fn cancel_removes_pending() {
        let client = MockClient::new();
        client.queue_response("https://tiles.example.com/0/0/0.pbf", 200, build_test_mvt());

        let source = HttpVectorTileSource::new(TEMPLATE, Box::new(client));
        source.request(TileId::new(0, 0, 0));
        assert_eq!(source.pending_count(), 1);

        source.cancel(TileId::new(0, 0, 0));
        assert_eq!(source.pending_count(), 0);

        let results = source.poll();
        assert!(results.is_empty());
    }

    #[test]
    fn with_tilejson_attaches_metadata() {
        let client = MockClient::new();
        let tj = TileJson::with_tiles(vec!["https://example.com/{z}/{x}/{y}.pbf".into()]);
        let source =
            HttpVectorTileSource::new(TEMPLATE, Box::new(client)).with_tilejson(tj.clone());
        assert!(source.tilejson().is_some());
        assert_eq!(source.tilejson().unwrap().tiles.len(), 1);
    }

    #[test]
    fn default_headers_are_sent() {
        #[derive(Clone)]
        struct HeaderCapture {
            last_headers: Arc<StdMutex<Vec<(String, String)>>>,
        }
        impl HttpClient for HeaderCapture {
            fn send(&self, request: HttpRequest) {
                *self.last_headers.lock().unwrap() = request.headers;
            }
            fn poll(&self) -> Vec<(String, Result<HttpResponse, String>)> {
                Vec::new()
            }
        }

        let capture = HeaderCapture {
            last_headers: Arc::new(StdMutex::new(Vec::new())),
        };
        let headers_ref = capture.last_headers.clone();

        let source = HttpVectorTileSource::new(TEMPLATE, Box::new(capture))
            .with_header("Authorization", "Bearer tok");

        source.request(TileId::new(0, 0, 0));

        let headers = headers_ref.lock().unwrap().clone();
        assert_eq!(headers.len(), 1);
        assert_eq!(headers[0].0, "Authorization");
    }

    #[test]
    fn debug_impl() {
        let client = MockClient::new();
        let source = HttpVectorTileSource::new(TEMPLATE, Box::new(client));
        let dbg = format!("{source:?}");
        assert!(dbg.contains("HttpVectorTileSource"));
    }

    #[test]
    fn deferred_decode_returns_raw_vector() {
        let client = MockClient::new();
        let url = "https://tiles.example.com/0/0/0.pbf";
        let mvt_bytes = build_test_mvt();
        client.queue_response(url, 200, mvt_bytes.clone());

        let source =
            HttpVectorTileSource::new(TEMPLATE, Box::new(client)).with_deferred_decode(true);
        assert!(source.deferred_decode());

        source.request(TileId::new(0, 0, 0));
        let results = source.poll();
        assert_eq!(results.len(), 1);

        let (id, result) = &results[0];
        assert_eq!(*id, TileId::new(0, 0, 0));
        let response = result.as_ref().expect("should succeed");
        assert!(
            response.data.is_raw_vector(),
            "deferred mode should return RawVector"
        );

        let raw = response.data.as_raw_vector().expect("should be RawVector");
        assert_eq!(raw.tile_id, TileId::new(0, 0, 0));
        assert_eq!(raw.bytes.len(), mvt_bytes.len());
    }

    #[test]
    fn deferred_decode_raw_bytes_can_be_decoded_later() {
        let client = MockClient::new();
        let url = "https://tiles.example.com/0/0/0.pbf";
        client.queue_response(url, 200, build_test_mvt());

        let source =
            HttpVectorTileSource::new(TEMPLATE, Box::new(client)).with_deferred_decode(true);
        source.request(TileId::new(0, 0, 0));
        let results = source.poll();
        let response = results[0].1.as_ref().expect("should succeed");
        let raw = response.data.as_raw_vector().expect("should be RawVector");

        // Decode the raw bytes manually (simulates what the async pipeline does).
        let layers = crate::mvt::decode_mvt(&raw.bytes, &raw.tile_id, &raw.decode_options)
            .expect("should decode");
        assert!(layers.contains_key("test"));
        assert_eq!(layers["test"].len(), 1);
    }

    #[test]
    fn deferred_decode_off_returns_vector() {
        let client = MockClient::new();
        let url = "https://tiles.example.com/0/0/0.pbf";
        client.queue_response(url, 200, build_test_mvt());

        let source =
            HttpVectorTileSource::new(TEMPLATE, Box::new(client)).with_deferred_decode(false);
        assert!(!source.deferred_decode());

        source.request(TileId::new(0, 0, 0));
        let results = source.poll();
        let response = results[0].1.as_ref().expect("should succeed");
        assert!(
            response.data.is_vector(),
            "non-deferred mode should return Vector"
        );
    }
}