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

//! Generation tier plugin (#968, D-206) — wires the background generation queue
//! and the per-body world-state cache into the running app.
//!
//! Registers [`GenerationQueue`] and [`BodyWorldStateCache`] as resources and
//! adds a `PreInput` system that drains completed work each tick and inserts the
//! computed [`BodyWorldState`](crate::atlas::body_world_state::BodyWorldState)
//! into the cache. The queue's *submitter* is the atlas layer-stream proxy
//! (#969, D-225); this plugin closes the submit→Rayon→cascade→drain→cache loop.

use bevy_app::prelude::*;
use bevy_ecs::prelude::*;
use bevy_ecs::schedule::IntoScheduleConfigs;

use crate::atlas::body_world_state::{BodyWorldStateCache, CACHE_CAPACITY};
use crate::atlas::gen_queue::{GenCompletion, GenerationQueue};
use crate::atlas::layer_proxy::{handle_atlas_request, AtlasLayerResponse, AtlasLayerStatus};
use crate::atlas::source_resolver::BodySourceResolverResource;
use crate::bridge::{AtlasRequestBuffer, AtlasResponseBuffer};
use crate::simulation::rng::SimRng;
use crate::simulation::time::SimulationTime;
use crate::tick_phases::TickPhase;

/// Wires the D-206 background generation tier into the app (#968).
pub struct GenerationPlugin;

impl Plugin for GenerationPlugin {
    fn build(&self, app: &mut App) {
        app.insert_resource(GenerationQueue::new())
            .insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY))
            .add_systems(
                Update,
                drain_generation_completions.in_set(TickPhase::PreInput),
            )
            .add_systems(Update, serve_atlas_requests.in_set(TickPhase::PreInput));
    }
}

/// Drain inbound atlas layer requests and serve each through the proxy (#969,
/// D-225): cache hit → Ready, miss → resolve + enqueue + Pending. Responses are
/// buffered for the bridge to flush in `PostSnapshot`.
fn serve_atlas_requests(
    mut requests: ResMut<AtlasRequestBuffer>,
    mut responses: ResMut<AtlasResponseBuffer>,
    mut cache: ResMut<BodyWorldStateCache>,
    queue: Res<GenerationQueue>,
    resolver: Option<Res<BodySourceResolverResource>>,
    rng: Option<Res<SimRng>>,
    time: Option<Res<SimulationTime>>,
) {
    if requests.0.is_empty() {
        return;
    }
    let world_seed = rng.as_ref().map(|r| r.seed()).unwrap_or(0);
    let tick = time.as_ref().map(|t| t.tick).unwrap_or(0);
    let pending: Vec<_> = requests.0.drain(..).collect();
    for req in pending {
        let resp = match resolver.as_ref() {
            Some(r) => handle_atlas_request(&req, &mut cache, &queue, &r.0, world_seed, tick),
            None => AtlasLayerResponse {
                body_id: req.body_id.clone(),
                status: AtlasLayerStatus::Error("no body source resolver".to_string()),
                layer1: None,
            },
        };
        responses.0.push(resp);
    }
}

/// Drain finished background work each tick and apply it to the cache (D-206).
///
/// Runs in `PreInput` (off the Rayon workers, on the main thread): a cheap
/// channel drain + cache insert, never the ~45 ms cascade itself.
fn drain_generation_completions(
    queue: Res<GenerationQueue>,
    mut cache: ResMut<BodyWorldStateCache>,
) {
    for completion in queue.drain_completions() {
        match completion {
            GenCompletion::BodyAnalyzed { state, .. } => cache.insert(state),
            GenCompletion::Failed { item, reason } => {
                tracing::warn!(?item, %reason, "background generation work item failed");
            }
            // Insert district world state into the matching body's cache entry (D-230).
            GenCompletion::SkeletonGenerated {
                city_id,
                body_id,
                state,
            } => {
                if !body_id.is_empty() {
                    if let Some(body_state) = cache.peek_mut(&body_id) {
                        // TODO(#957): keying by city_id is a stub — a city has multiple districts;
                        // use DistrictSkeleton.district_id as the DistrictId key when real skeleton gen lands.
                        body_state.districts.insert(city_id, state);
                    } else {
                        tracing::warn!(
                            city_id,
                            body_id,
                            "SkeletonGenerated: body not in cache — district state dropped"
                        );
                    }
                }
                // body_id empty = stub result from GenerateSkeleton stub; silently ignore.
            }
            GenCompletion::ChunkFilled { .. } => {}
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::atlas::gen_queue::{GenPriority, GenWorkItem};
    use crate::seed::SeedChain;
    use bevy_ecs::schedule::Schedule;
    use std::time::Duration;

    /// Tiny 16-bit grayscale heightmap PNG at a unique temp path, so the real
    /// cascade can run without a committed fixture.
    fn test_heightmap_path() -> std::path::PathBuf {
        use std::io::BufWriter;
        use std::sync::atomic::{AtomicU32, Ordering};
        static SEQ: AtomicU32 = AtomicU32::new(0);
        let n = SEQ.fetch_add(1, Ordering::Relaxed);
        let path =
            std::env::temp_dir().join(format!("sr_genplugin_{}_{n}.png", std::process::id()));
        let file = std::fs::File::create(&path).expect("create test heightmap");
        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().expect("png header");
        let data: Vec<u8> = (0..32u32 * 16)
            .flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes())
            .collect();
        w.write_image_data(&data).expect("png data");
        path
    }

    #[test]
    fn drain_system_populates_cache() {
        // The full loop: submit → Rayon cascade → completion → drain → cache.
        let mut world = World::new();
        world.insert_resource(GenerationQueue::new());
        world.insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY));

        world.resource::<GenerationQueue>().submit(
            GenWorkItem::AnalyzeBody {
                body_id: "PlanetX".to_string(),
                heightmap_path: test_heightmap_path(),
                sea_level: 0.3,
                body_seed: SeedChain::for_body(42, "PlanetX"),
            },
            GenPriority::Immediate,
        );

        let mut sched = Schedule::default();
        sched.add_systems(drain_generation_completions);

        // Rayon runs the cascade asynchronously; the drain runs each schedule pass.
        let mut found = false;
        for _ in 0..100 {
            sched.run(&mut world);
            if world.resource::<BodyWorldStateCache>().contains("PlanetX") {
                found = true;
                break;
            }
            std::thread::sleep(Duration::from_millis(10));
        }
        assert!(
            found,
            "drain system should insert the analyzed body into the cache"
        );
    }

    #[test]
    fn serve_drains_requests_into_responses() {
        use crate::atlas::cascade::CascadeLayer;
        use crate::atlas::layer_proxy::AtlasLayerRequest;

        let mut world = World::new();
        world.insert_resource(AtlasRequestBuffer(vec![AtlasLayerRequest {
            body_id: "GJ1c".to_string(),
            up_to: CascadeLayer::Topography,
        }]));
        world.insert_resource(AtlasResponseBuffer::default());
        world.insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY));
        world.insert_resource(GenerationQueue::with_threads(1));
        // No resolver / SimRng / SimulationTime — all optional in the system.

        let mut sched = Schedule::default();
        sched.add_systems(serve_atlas_requests);
        sched.run(&mut world);

        let responses = world.resource::<AtlasResponseBuffer>();
        assert_eq!(responses.0.len(), 1, "request should produce one response");
        assert_eq!(responses.0[0].body_id, "GJ1c");
        // No resolver wired → Error status (exercises the drain + push path).
        assert!(matches!(responses.0[0].status, AtlasLayerStatus::Error(_)));
        // The request buffer was drained.
        assert!(world.resource::<AtlasRequestBuffer>().0.is_empty());
    }
}