settled-reach/server/src/atlas/layer_proxy.rs

//! Atlas layer-stream proxy handler (#969, D-225).
//!
//! Serves a body's generation-cascade layer data to the client, compute-on-
//! demand and mod-first:
//! - **Cache hit** → serialize the cached `Layer1Output` and reply `Ready`.
//! - **Cache miss** → resolve the body's source heightmap ([`BodySourceResolver`]),
//!   enqueue an `Immediate` `AnalyzeBody` on the background queue (#968), and
//!   reply `Pending` (the client re-requests; the drain system populates the
//!   cache, so a later request hits).
//!
//! Pure handler logic; the bridge wiring (message routing) is the proxy's other
//! half. No baking — the heightmap is the only source of truth (D-225).

use serde::{Deserialize, Serialize};

use crate::atlas::body_params_reader::BodyParamsReader;
use crate::atlas::body_world_state::{BodyWorldStateCache, SimTick};
use crate::atlas::cascade::CascadeLayer;
use crate::atlas::city_context_reader::CityContextReader;
use crate::atlas::gen_queue::{GenPriority, GenWorkItem, GenerationQueue};
use crate::atlas::layer1::Layer1Output;
use crate::atlas::source_resolver::{BodySourceResolver, SourceResolveError};
use crate::seed::SeedChain;

/// Fallback sea level when the heightmap PNG carries no `sea_level` tEXt chunk
/// (the loader prefers the chunk; this is only the floor).
const DEFAULT_SEA_LEVEL: f32 = 0.3;

/// A client request for a body's generation layers (D-225).
///
/// `up_to` is a forward-compat seam that is **not yet honored**: `run_work_item`
/// currently runs the cascade through `CascadeLayer::Settlement` unconditionally,
/// ignoring this field. Wiring per-request depth (and the partial caching it
/// implies) is deferred to #1021.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AtlasLayerRequest {
    pub body_id: String,
    pub up_to: CascadeLayer,
}

/// Status of a layer response (D-225).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum AtlasLayerStatus {
    /// Layer data is ready (`layer1` is populated).
    Ready,
    /// Analysis was enqueued; the client should re-request shortly.
    Pending,
    /// The body is unknown or has no source terrain — re-requesting won't help.
    NotFound,
    /// Resolution / IO failure (message for the client log).
    Error(String),
}

/// A layer response: the computed `Layer1Output`, or a non-ready status.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AtlasLayerResponse {
    pub body_id: String,
    pub status: AtlasLayerStatus,
    pub layer1: Option<Layer1Output>,
}

/// Serve one layer request (D-225). `current_tick` stamps the cache LRU on hit;
/// `world_seed` derives the body's `SeedChain` for the enqueued analysis.
///
/// `city_reader` supplies the body's settlements for Layer-3 placement (#955),
/// read on a cache miss. `None` (or a read failure) places no cities — the
/// cascade still runs Layer 1; the body just gets no settlement placements.
///
/// `body_params_reader` supplies the body's physical parameters for the
/// RegionProfile carrier layer (T-1032, D-239 §1), read on a cache miss.
/// `None` (or a read failure) passes `body_params: None` to the work item,
/// causing the cascade to stop at `CascadeLayer::Settlement` (pre-T-1032
/// behaviour). A successful read passes `Some(Box::new(params))`, enabling
/// the full `CascadeLayer::RegionProfile` path.
pub fn handle_atlas_request(
    req: &AtlasLayerRequest,
    cache: &mut BodyWorldStateCache,
    queue: &GenerationQueue,
    resolver: &BodySourceResolver,
    city_reader: Option<&CityContextReader>,
    body_params_reader: Option<&BodyParamsReader>,
    world_seed: u64,
    current_tick: SimTick,
) -> AtlasLayerResponse {
    // Cache hit — serve immediately.
    if let Some(state) = cache.get(&req.body_id, current_tick) {
        let layer1 = Layer1Output {
            body_id: state.body_id.clone(),
            river_network: state.river_network.clone(),
            drainage_basins: state.drainage_basins.clone(),
            attractors: state.attractors.clone(),
            // The cascade ran on the downsampled heightmap, so its dims are the
            // working grid all Layer-1 positions are expressed in (#960).
            grid_w: state.heightmap_width,
            grid_h: state.heightmap_height,
        };
        return AtlasLayerResponse {
            body_id: req.body_id.clone(),
            status: AtlasLayerStatus::Ready,
            layer1: Some(layer1),
        };
    }

    // Miss — resolve the source heightmap and enqueue background analysis.
    match resolver.resolve(&req.body_id) {
        Ok(heightmap_path) => {
            // Pre-resolve this body's settlements + system faction so the Rayon
            // work item stays DB-free (#955/#956, D-225). Read failures are
            // non-fatal: log and fall back (no cities / no faction → frontier).
            let (cities, dominant_faction) = match city_reader {
                Some(reader) => {
                    let cities = reader
                        .read_body_settlements(&req.body_id)
                        .unwrap_or_else(|e| {
                            tracing::warn!(
                                body_id = %req.body_id,
                                error = %e,
                                "settlement read failed; placing no cities"
                            );
                            Vec::new()
                        });
                    let faction = reader
                        .read_body_dominant_faction(&req.body_id)
                        .unwrap_or_else(|e| {
                            tracing::warn!(
                                body_id = %req.body_id,
                                error = %e,
                                "dominant_faction read failed; defaulting to frontier"
                            );
                            None
                        });
                    (cities, faction)
                }
                None => (Vec::new(), None),
            };
            // Pre-resolve body physical params so the Rayon work item stays
            // DB-free (D-225 pattern). Read failures are non-fatal: log and
            // fall back to None (cascade stops at Settlement, pre-T-1032
            // behaviour, rather than aborting the entire analysis).
            let body_params = match body_params_reader {
                Some(reader) => reader
                    .read_body_params(&req.body_id)
                    .map(|p| Some(Box::new(p)))
                    .unwrap_or_else(|e| {
                        tracing::warn!(
                            body_id = %req.body_id,
                            error = %e,
                            "body_params read failed; region layer skipped"
                        );
                        None
                    }),
                None => None,
            };
            queue.submit(
                GenWorkItem::AnalyzeBody {
                    body_id: req.body_id.clone(),
                    heightmap_path,
                    sea_level: DEFAULT_SEA_LEVEL,
                    body_seed: SeedChain::for_body(world_seed, &req.body_id),
                    cities,
                    dominant_faction,
                    body_params,
                },
                GenPriority::Immediate,
            );
            AtlasLayerResponse {
                body_id: req.body_id.clone(),
                status: AtlasLayerStatus::Pending,
                layer1: None,
            }
        }
        // Unknown / no terrain → re-requesting won't help.
        Err(SourceResolveError::UnknownBody(_))
        | Err(SourceResolveError::NoTerrainReference { .. }) => AtlasLayerResponse {
            body_id: req.body_id.clone(),
            status: AtlasLayerStatus::NotFound,
            layer1: None,
        },
        Err(e) => AtlasLayerResponse {
            body_id: req.body_id.clone(),
            status: AtlasLayerStatus::Error(e.to_string()),
            layer1: None,
        },
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::atlas::body_world_state::{BodyWorldState, RiverNetwork, CACHE_CAPACITY};
    use crate::atlas::gen_queue::GenCompletion;
    use rusqlite::Connection;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::time::Duration;

    static SEQ: AtomicU32 = AtomicU32::new(0);

    fn req(body_id: &str) -> AtlasLayerRequest {
        AtlasLayerRequest {
            body_id: body_id.to_string(),
            up_to: CascadeLayer::Topography,
        }
    }

    const REL: &str = "wiki/star-systems/GJ-1/bodies/GJ1c/heightmap.png";

    /// systems.db with one bodies row, + a base root containing a tiny 16-bit
    /// heightmap PNG at the body's terrain_reference. Returns (db, resolver).
    fn resolver_with_body(body_id: &str) -> (PathBuf, BodySourceResolver) {
        let n = SEQ.fetch_add(1, Ordering::Relaxed);
        let db = std::env::temp_dir().join(format!("sr_proxy_{}_{n}.db", std::process::id()));
        let _ = std::fs::remove_file(&db);
        let conn = Connection::open(&db).unwrap();
        conn.execute(
            "CREATE TABLE bodies (body_id TEXT PRIMARY KEY, terrain_reference TEXT)",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO bodies (body_id, terrain_reference) VALUES (?1, ?2)",
            rusqlite::params![body_id, REL],
        )
        .unwrap();

        let root = std::env::temp_dir().join(format!("sr_proxyroot_{}_{n}", std::process::id()));
        write_tiny_heightmap(&root.join(REL));

        let resolver = BodySourceResolver::open(&db, vec![root]).unwrap();
        (db, resolver)
    }

    fn write_tiny_heightmap(path: &Path) {
        use std::io::BufWriter;
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let file = std::fs::File::create(path).unwrap();
        let mut enc = png::Encoder::new(BufWriter::new(file), 32, 16);
        enc.set_color(png::ColorType::Grayscale);
        enc.set_depth(png::BitDepth::Sixteen);
        let mut w = enc.write_header().unwrap();
        let data: Vec<u8> = (0..32u32 * 16)
            .flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes())
            .collect();
        w.write_image_data(&data).unwrap();
    }

    fn empty_resolver() -> (PathBuf, BodySourceResolver) {
        let n = SEQ.fetch_add(1, Ordering::Relaxed);
        let db = std::env::temp_dir().join(format!("sr_proxye_{}_{n}.db", std::process::id()));
        let _ = std::fs::remove_file(&db);
        let conn = Connection::open(&db).unwrap();
        conn.execute(
            "CREATE TABLE bodies (body_id TEXT, terrain_reference TEXT)",
            [],
        )
        .unwrap();
        let resolver = BodySourceResolver::open(&db, vec![std::env::temp_dir()]).unwrap();
        (db, resolver)
    }

    #[test]
    fn cache_hit_is_ready() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        cache.insert(BodyWorldState {
            body_id: "GJ1c".into(),
            heightmap: vec![0.0; 4],
            heightmap_width: 2,
            heightmap_height: 2,
            river_network: RiverNetwork::default(),
            drainage_basins: vec![],
            attractors: vec![],
            placements: vec![],
            quarters: std::collections::BTreeMap::new(),
            regions: std::collections::BTreeMap::new(),
            last_accessed: 0,
        });
        let (_db, resolver) = empty_resolver();
        let queue = GenerationQueue::with_threads(1);

        let resp = handle_atlas_request(
            &req("GJ1c"),
            &mut cache,
            &queue,
            &resolver,
            None,
            None,
            42,
            1,
        );
        assert_eq!(resp.status, AtlasLayerStatus::Ready);
        assert_eq!(resp.layer1.expect("layer1").body_id, "GJ1c");
    }

    #[test]
    fn cache_miss_enqueues_and_pends_then_analyzes() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let (_db, resolver) = resolver_with_body("GJ1c");
        let queue = GenerationQueue::with_threads(1);

        let resp = handle_atlas_request(
            &req("GJ1c"),
            &mut cache,
            &queue,
            &resolver,
            None,
            None,
            42,
            1,
        );
        assert_eq!(resp.status, AtlasLayerStatus::Pending);
        assert!(resp.layer1.is_none());

        // The enqueued analysis runs the real cascade and completes.
        std::thread::sleep(Duration::from_millis(150));
        let completions = queue.drain_completions();
        assert!(
            completions.iter().any(
                |c| matches!(c, GenCompletion::BodyAnalyzed { body_id, .. } if body_id == "GJ1c")
            ),
            "miss should enqueue an AnalyzeBody that completes: {completions:?}"
        );
    }

    #[test]
    fn unknown_body_is_not_found() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let (_db, resolver) = empty_resolver();
        let queue = GenerationQueue::with_threads(1);

        let resp = handle_atlas_request(
            &req("ghost"),
            &mut cache,
            &queue,
            &resolver,
            None,
            None,
            42,
            1,
        );
        assert_eq!(resp.status, AtlasLayerStatus::NotFound);
    }

    /// Build a DB with the columns needed by both `BodySourceResolver` and
    /// `BodyParamsReader` for the same body, plus a tiny heightmap root.
    ///
    /// Returns (db_path, resolver, body_params_reader, _root_kept_alive).
    fn resolver_and_params_reader(
        body_id: &str,
    ) -> (
        PathBuf,
        BodySourceResolver,
        crate::atlas::body_params_reader::BodyParamsReader,
        PathBuf, // root dir — must stay alive for the test duration
    ) {
        let n = SEQ.fetch_add(1, Ordering::Relaxed);
        let db = std::env::temp_dir().join(format!("sr_proxybp_{}_{n}.db", std::process::id()));
        let _ = std::fs::remove_file(&db);
        let conn = Connection::open(&db).unwrap();

        conn.execute_batch(
            "CREATE TABLE star_systems (
                 system_id TEXT PRIMARY KEY,
                 spectral_class TEXT,
                 star_type TEXT
             );
             CREATE TABLE bodies (
                 body_id TEXT PRIMARY KEY,
                 system_id TEXT,
                 terrain_reference TEXT,
                 hydrosphere TEXT,
                 atmosphere TEXT,
                 planet_class TEXT,
                 orbital_period_days REAL,
                 axial_tilt_deg REAL
             );",
        )
        .unwrap();

        conn.execute(
            "INSERT INTO star_systems (system_id, spectral_class, star_type) VALUES ('GJ-1', 'G', 'main_sequence')",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO bodies (body_id, system_id, terrain_reference, hydrosphere, atmosphere, planet_class, orbital_period_days, axial_tilt_deg)
             VALUES (?1, 'GJ-1', ?2, 'ocean', 'breathable', 'temperate', 365.25, 23.5)",
            rusqlite::params![body_id, REL],
        )
        .unwrap();
        drop(conn);

        let root = std::env::temp_dir().join(format!("sr_proxybproot_{}_{n}", std::process::id()));
        write_tiny_heightmap(&root.join(REL));

        let resolver = BodySourceResolver::open(&db, vec![root.clone()]).unwrap();
        let params_reader = crate::atlas::body_params_reader::BodyParamsReader::open(&db).unwrap();

        (db, resolver, params_reader, root)
    }

    /// With body_params_reader wired, a cache miss enqueues an AnalyzeBody that
    /// completes with populated `regions` (RegionProfile layer ran).
    #[test]
    fn body_params_reader_wired_produces_populated_regions() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let (_db, resolver, params_reader, _root) = resolver_and_params_reader("GJ1c");
        let queue = GenerationQueue::with_threads(1);

        let resp = handle_atlas_request(
            &req("GJ1c"),
            &mut cache,
            &queue,
            &resolver,
            None,
            Some(&params_reader),
            42,
            1,
        );
        assert_eq!(resp.status, AtlasLayerStatus::Pending);

        // Wait for the Rayon work item to complete.
        std::thread::sleep(Duration::from_millis(300));
        let completions = queue.drain_completions();
        let body_state = completions
            .into_iter()
            .find_map(|c| {
                if let GenCompletion::BodyAnalyzed { body_id, state } = c {
                    if body_id == "GJ1c" {
                        return Some(state);
                    }
                }
                None
            })
            .expect("AnalyzeBody must complete for GJ1c");

        assert!(
            !body_state.regions.is_empty(),
            "regions must be populated when body_params_reader is wired (T-1032 dispatch path)"
        );
    }

    /// Without body_params_reader (None), regions is empty — pre-T-1032 behaviour.
    #[test]
    fn no_body_params_reader_leaves_regions_empty() {
        let mut cache = BodyWorldStateCache::new(CACHE_CAPACITY);
        let (_db, resolver) = resolver_with_body("GJ1c");
        let queue = GenerationQueue::with_threads(1);

        let resp = handle_atlas_request(
            &req("GJ1c"),
            &mut cache,
            &queue,
            &resolver,
            None,
            None, // no body_params_reader
            42,
            1,
        );
        assert_eq!(resp.status, AtlasLayerStatus::Pending);

        std::thread::sleep(Duration::from_millis(300));
        let completions = queue.drain_completions();
        let body_state = completions
            .into_iter()
            .find_map(|c| {
                if let GenCompletion::BodyAnalyzed { body_id, state } = c {
                    if body_id == "GJ1c" {
                        return Some(state);
                    }
                }
                None
            })
            .expect("AnalyzeBody must complete for GJ1c");

        assert!(
            body_state.regions.is_empty(),
            "regions must remain empty when no body_params_reader is wired"
        );
    }
}