open-standards/map/map-rs/crates/map-index/src/bridge.rs

//! Bridge from S2 geometry to the standard's address space (`map-schema`).
//!
//! The geometry crate speaks in [`crate::S2CellId`] (a rich, Hilbert-encoded
//! cell with face/level/neighbour operations). The wire standard speaks in
//! [`map_schema::S2CellId`] — an opaque, orderable `u64`. The two share the same
//! bit layout: the S2 sentinel encoding puts the level at bit `2·(30−level)`,
//! which is exactly what `map_schema::S2CellId::level()` decodes. So the bridge
//! is a zero-cost reinterpretation plus the convenience of building whole
//! `CellKey`s and covering viewports.

use map_schema::{AltBand, CellKey, Lod, TimeBucket};

use crate::{volume, S2CellId, S2Region, S2RegionCoverer, S2Rect};

const MS_PER_SEC: i64 = 1000;

/// Reinterpret a geometry cell id as a schema cell id (same `u64` bits).
#[inline]
pub fn schema_cell(cell: S2CellId) -> map_schema::S2CellId {
    map_schema::S2CellId(cell.raw())
}

/// Build a [`CellKey`] from real coordinates: take the S2 leaf cell at
/// `(lat, lng)`, walk up to `level` (the horizontal LOD), and attach the
/// vertical/temporal/detail axes.
pub fn cell_key(
    lat_deg: f64,
    lng_deg: f64,
    level: u8,
    alt: AltBand,
    time: TimeBucket,
    lod: Lod,
) -> CellKey {
    let cell = S2CellId::from_lat_lng(lat_deg, lng_deg).parent_at_level(level);
    CellKey { s2: schema_cell(cell), alt, time, lod }
}

/// The cell keys a viewport resolves to, sorted and de-duplicated — ready to
/// feed [`map_schema::Directory::ranges_for`].
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ViewportCover {
    pub cells: Vec<CellKey>,
}

/// Cover **any** S2 region — a cap, a rectangle, or a polygon (a boundary, an
/// airspace footprint, a geofence) — into the set of [`CellKey`]s a reader must
/// fetch (SPEC §4). The coverer chooses cells between `min_level` and
/// `max_level`, bounded by `max_cells`; each is stamped with the shared
/// vertical/temporal/detail axes. The result is sorted (the directory's
/// canonical order), so the range planner coalesces it into contiguous byte
/// spans.
#[allow(clippy::too_many_arguments)]
pub fn cover_region(
    region: &dyn S2Region,
    min_level: u8,
    max_level: u8,
    max_cells: usize,
    alt: AltBand,
    time: TimeBucket,
    lod: Lod,
) -> ViewportCover {
    let coverer = S2RegionCoverer::new(min_level, max_level, max_cells);
    let mut cells: Vec<CellKey> = coverer
        .get_covering(region)
        .into_iter()
        .map(|c| CellKey { s2: schema_cell(c), alt, time, lod })
        .collect();
    cells.sort();
    cells.dedup();
    ViewportCover { cells }
}

/// Cover a lat/lng rectangle viewport — the common case. Delegates to
/// [`cover_region`].
#[allow(clippy::too_many_arguments)]
pub fn cover_viewport(
    rect: &S2Rect,
    min_level: u8,
    max_level: u8,
    max_cells: usize,
    alt: AltBand,
    time: TimeBucket,
    lod: Lod,
) -> ViewportCover {
    cover_region(rect, min_level, max_level, max_cells, alt, time, lod)
}

// ── The 4D axes: the recommended altitude-band + time-bucket scheme ─────────

/// The signed schema [`AltBand`] for a recommended [`volume::AltBand`]. Band 0
/// is the datum surface; bathymetry is −1; higher bands climb to cislunar.
pub fn schema_band(band: volume::AltBand) -> AltBand {
    use volume::AltBand as V;
    AltBand(match band {
        V::Bathymetry => -1,
        V::Surface => 0,
        V::LowAlt => 1,
        V::MidAlt => 2,
        V::HighAlt => 3,
        V::Leo => 4,
        V::MeoGeo => 5,
        V::BeyondGeo => 6,
    })
}

/// Map an altitude in metres to a schema [`AltBand`] under the recommended
/// logarithmic scheme (SPEC §2).
pub fn alt_band(alt_m: f64) -> AltBand {
    schema_band(volume::AltBand::from_altitude_m(alt_m))
}

/// Map an epoch-seconds instant to a schema [`TimeBucket`] at the given tier.
/// The reference treats the input epoch-seconds as the bucket anchor; a
/// leap-second-precise TAI mapping is the publisher's concern.
pub fn time_bucket(time_secs: i64, tier: volume::TimeTier) -> TimeBucket {
    let b = volume::TimeBucket::from_timestamp(time_secs, tier);
    TimeBucket {
        start_tai_ms: b.start_secs * MS_PER_SEC,
        width_ms: volume::TimeBucket::bucket_size_secs(tier) as u64 * MS_PER_SEC as u64,
    }
}

/// Cover a **4D volume** — a horizontal region × an altitude range × a time
/// range — into the full set of [`CellKey`]s a reader must fetch: the
/// cross-product of the S2 covering, the overlapping altitude bands, and the
/// time buckets at `tier` (SPEC §2, §4). Sorted + de-duplicated for the
/// directory range planner.
#[allow(clippy::too_many_arguments)]
pub fn cover_volume(
    region: &dyn S2Region,
    alt_min_m: f64,
    alt_max_m: f64,
    time_start_secs: i64,
    time_end_secs: i64,
    tier: volume::TimeTier,
    min_level: u8,
    max_level: u8,
    max_cells: usize,
    lod: Lod,
) -> ViewportCover {
    let cells = S2RegionCoverer::new(min_level, max_level, max_cells).get_covering(region);
    let bands = volume::AltBand::bands_for_range(alt_min_m, alt_max_m);
    let buckets = volume::TimeBucket::buckets_for_range(time_start_secs, time_end_secs, tier);
    let width_ms = volume::TimeBucket::bucket_size_secs(tier) as u64 * MS_PER_SEC as u64;

    let mut keys = Vec::with_capacity(cells.len() * bands.len().max(1) * buckets.len().max(1));
    for c in cells {
        let s2 = schema_cell(c);
        for &band in &bands {
            let alt = schema_band(band);
            for &bk in &buckets {
                keys.push(CellKey {
                    s2,
                    alt,
                    time: TimeBucket { start_tai_ms: bk.start_secs * MS_PER_SEC, width_ms },
                    lod,
                });
            }
        }
    }
    keys.sort();
    keys.dedup();
    ViewportCover { cells: keys }
}

#[cfg(test)]
mod tests {
    use super::*;
    use map_schema::TimeBucket;

    #[test]
    fn schema_cell_preserves_bits_and_level() {
        // A level-10 cell over (45, 90).
        let cell = S2CellId::from_lat_lng(45.0, 90.0).parent_at_level(10);
        let schema = schema_cell(cell);
        assert_eq!(schema.0, cell.raw());
        // The schema's level decoder agrees with the geometry crate.
        assert_eq!(schema.level(), Some(cell.level()));
        assert_eq!(cell.level(), 10);
    }

    #[test]
    fn cell_key_targets_requested_level() {
        let k = cell_key(37.7749, -122.4194, 15, AltBand(0), TimeBucket::ATEMPORAL, Lod(0));
        assert_eq!(k.s2.level(), Some(15));
        assert_eq!(k.alt, AltBand(0));
        assert!(k.time.is_atemporal());
    }

    #[test]
    fn viewport_cover_is_sorted_unique_and_nonempty() {
        // A small box around San Francisco.
        let rect = S2Rect::from_degrees(37.70, 37.82, -122.52, -122.36);
        let cover = cover_viewport(&rect, 6, 14, 32, AltBand(0), TimeBucket::ATEMPORAL, Lod(0));
        assert!(!cover.cells.is_empty(), "covering should be non-empty");
        // sorted + deduped (canonical directory order)
        assert!(cover.cells.windows(2).all(|w| w[0] < w[1]));
        // every covering cell carries a valid S2 level in range
        for c in &cover.cells {
            let lvl = c.s2.level().expect("valid cell");
            assert!((6..=14).contains(&lvl));
        }
    }

    #[test]
    fn alt_band_and_time_bucket_map_into_schema() {
        // Surface = band 0 (datum); bathymetry negative; ISS altitude climbs.
        assert_eq!(alt_band(50.0), AltBand(0));
        assert_eq!(alt_band(-500.0), AltBand(-1));
        assert_eq!(alt_band(408_000.0).0, 4); // LEO
        // An hourly bucket: 1h wide, anchored at the hour.
        let tb = time_bucket(3_600 + 5, crate::volume::TimeTier::Recent);
        assert_eq!(tb.width_ms, 3_600_000);
        assert_eq!(tb.start_tai_ms, 3_600_000);
    }

    #[test]
    fn cover_volume_spans_bands_and_buckets() {
        use crate::volume::TimeTier;
        use std::collections::BTreeSet;
        let rect = S2Rect::from_degrees(37.70, 37.74, -122.46, -122.42);
        // 0..500 m spans Surface[0,100) + LowAlt[100,1000); 0..7200 s = two hourly buckets.
        let cover = cover_volume(&rect, 0.0, 500.0, 0, 7200, TimeTier::Recent, 8, 12, 8, Lod(0));
        assert!(!cover.cells.is_empty());
        assert!(cover.cells.windows(2).all(|w| w[0] < w[1]), "sorted + unique");
        let bands: BTreeSet<i16> = cover.cells.iter().map(|c| c.alt.0).collect();
        assert!(bands.contains(&0) && bands.contains(&1), "spans Surface + LowAlt");
        let starts: BTreeSet<i64> = cover.cells.iter().map(|c| c.time.start_tai_ms).collect();
        assert_eq!(starts.len(), 2, "two hourly buckets");
        assert!(cover.cells.iter().all(|c| c.time.width_ms == 3_600_000));
    }

    #[test]
    fn cover_polygon_region_into_cell_keys() {
        use crate::{S2Point, S2Polygon};
        // A triangle near the equator — the kind of non-rectangular region
        // (a boundary / footprint / geofence) that caps and rects can't express.
        let poly = S2Polygon::from_loop(vec![
            S2Point::from_lat_lng(0.10, 0.00),
            S2Point::from_lat_lng(-0.05, 0.09),
            S2Point::from_lat_lng(-0.05, -0.09),
        ]);
        let cover = cover_region(&poly, 6, 14, 64, AltBand(0), TimeBucket::ATEMPORAL, Lod(0));
        assert!(!cover.cells.is_empty(), "a polygon must cover to some cells");
        assert!(cover.cells.windows(2).all(|w| w[0] < w[1]), "sorted + unique");
        // A point we know is inside the triangle must land under one of the
        // covered cells — i.e. the covering actually contains the region interior.
        let interior = S2CellId::from_lat_lng(0.0, 0.0).parent_at_level(14);
        let covers_interior = cover
            .cells
            .iter()
            .any(|c| S2CellId(c.s2.0).contains(&interior));
        assert!(covers_interior, "covering must include the cell over an interior point");
    }

    #[test]
    fn cover_feeds_directory_range_planner() {
        use map_schema::{ContentId, DirEntry, Directory};
        let rect = S2Rect::from_degrees(37.70, 37.78, -122.46, -122.40);
        let cover = cover_viewport(&rect, 8, 12, 16, AltBand(0), TimeBucket::ATEMPORAL, Lod(0));
        // Lay the covered cells out contiguously in a directory...
        let entries: Vec<DirEntry> = cover
            .cells
            .iter()
            .enumerate()
            .map(|(i, &key)| DirEntry {
                key,
                offset: (i as u64) * 100,
                len: 100,
                content_id: ContentId::of_bytes(&[i as u8]),
            })
            .collect();
        let dir = Directory::new(entries);
        // ...so the whole viewport coalesces into a single contiguous span.
        let ranges = dir.ranges_for(&cover.cells);
        assert_eq!(ranges.len(), 1, "contiguous layout should coalesce to one span");
    }
}