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

//! Background generation queue — prioritized Rayon thread pool (D-206).
//!
//! All runtime-background generation work runs through this queue. The main
//! tick thread submits work items (non-blocking) and drains completion events
//! once per tick via a `crossbeam` channel.
//!
//! **Priority levels (D-206):**
//! - `Immediate`: player arrives within 1 game-minute. Runs first.
//! - `High`:      player arrives within 5 game-minutes.
//! - `Medium`:    player is in the same system.
//! - `Low`:       player has heard of this location via NPC/news.
//!
//! **Work item types (D-206):**
//! - `AnalyzeBody`:       D8 drainage + attractor extraction for a body.
//! - `GenerateSkeleton`:  Phase 1 DistrictSkeleton for a city.
//! - `FillChunk`:         Phase 2 chunk fill for a pre-loaded district.
//!
//! Completion events are delivered to the main thread via
//! `GenerationQueue::drain_completions()`, called once per tick from a Bevy
//! system in `TickPhase::PreInput`.
//!
//! **Thread count (D-206):** `available_parallelism - 2`, minimum 1.

use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use bevy_ecs::prelude::Resource;
use crossbeam_channel::{Receiver, Sender};

use crate::atlas::body_world_state::BodyWorldState;
use crate::atlas::cascade::{run_cascade_from_heightmap, CascadeLayer};
use crate::atlas::heightmap::{load_heightmap_png, GRID_H, GRID_W};
use crate::atlas::skeleton_gen::generate_skeleton;
use crate::seed::SeedChain;
use crate::simulation::generator::{CityGenerationContext, DistrictWorldState};

// ---------------------------------------------------------------------------
// Priority
// ---------------------------------------------------------------------------

/// Work priority levels — lower discriminant = higher priority.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum GenPriority {
    /// Player arrives within ~1 game-minute. Runs before all other levels.
    Immediate = 0,
    /// Player arrives within ~5 game-minutes.
    High = 1,
    /// Player is in the same system.
    Medium = 2,
    /// Player has seen or heard of this location (NPC dialogue, news ticker).
    Low = 3,
}

// ---------------------------------------------------------------------------
// Work item types
// ---------------------------------------------------------------------------

/// A unit of background generation work (D-206).
#[derive(Debug, Clone)]
pub enum GenWorkItem {
    /// Run the Layer-1 cascade (drainage → features → sub-biome) for this body.
    /// The enqueuer resolves the inputs (D-225): `heightmap_path` is the
    /// mod-resolved source PNG, `body_seed` is this body's SeedChain position.
    /// `run_work_item` is pure compute — it does no path/DB resolution.
    AnalyzeBody {
        body_id: String,
        heightmap_path: PathBuf,
        sea_level: f32,
        body_seed: SeedChain,
    },
    /// Generate a Phase 1 DistrictSkeleton for this city.
    ///
    /// `context` is the D-199 economic read-set pre-resolved at dispatch time.
    /// All 6 required fields must be populated before this item is submitted
    /// (D-199: "Missing fields abort the task … generation does not proceed with
    /// partial context").
    ///
    /// `body_id` routes the resulting `SkeletonGenerated` completion into the
    /// correct `BodyWorldState` cache entry (D-230).
    ///
    /// `district_id` is the stable content-addressable id for the generated
    /// district (keyed by city position + world seed).
    ///
    /// `economic_role`, `population`, and `founding_age_years` are D-199 fields
    /// carried alongside the context because `generate_skeleton` accepts them as
    /// separate parameters (its signature is not changed by this ticket).
    GenerateSkeleton {
        city_id: u64,
        body_id: String,
        /// D-199 economic read-set + all other context fields.
        context: Box<CityGenerationContext>,
        /// Stable content-addressable district id (D-194/D-230).
        district_id: u64,
        /// District-level seed chain (D-224).
        chain: SeedChain,
        // D-199 raw fields passed to generate_skeleton separately.
        economic_role: String,
        population: i64,
        founding_age_years: u32,
    },
    /// Pre-fill a chunk in an existing district.
    FillChunk {
        district_id: u64,
        block_pos: (u32, u32),
    },
}

impl GenWorkItem {
    pub fn body_id(&self) -> Option<&str> {
        if let GenWorkItem::AnalyzeBody { body_id, .. } = self {
            Some(body_id)
        } else {
            None
        }
    }
}

// ---------------------------------------------------------------------------
// Completion event
// ---------------------------------------------------------------------------

/// Sent back to the main thread when a work item finishes (D-206).
#[derive(Debug)]
pub enum GenCompletion {
    BodyAnalyzed {
        body_id: String,
        /// The computed world state, ready for `BodyWorldStateCache::insert`.
        state: BodyWorldState,
    },
    SkeletonGenerated {
        city_id: u64,
        /// The body this skeleton belongs to — used to route state into
        /// `BodyWorldState.districts` (D-230).
        body_id: String,
        /// District-level world state (skeleton + block tags) produced by the plan phase (D-230).
        /// Boxed to keep `GenCompletion` variant sizes balanced (D-230 skeleton is ~2.7 KB).
        state: Box<DistrictWorldState>,
    },
    ChunkFilled {
        district_id: u64,
        block_pos: (u32, u32),
    },
    /// Work item failed — body_id or city_id for logging.
    Failed { item: GenWorkItem, reason: String },
}

// ---------------------------------------------------------------------------
// Internal queued work
// ---------------------------------------------------------------------------

struct QueuedWork {
    priority: GenPriority,
    item: GenWorkItem,
}

// ---------------------------------------------------------------------------
// GenerationQueue — Bevy Resource
// ---------------------------------------------------------------------------

/// Bevy `Resource` managing the background generation queue (D-206).
///
/// Submit work with `submit()`. Drain completions with `drain_completions()`
/// once per tick. The Rayon thread pool runs tasks in priority order.
///
/// Priority is respected because `dispatch_next()` is gated on pool saturation
/// via `in_flight_count`: it only dispatches when fewer than `n_threads` tasks
/// are running. This applies to all work item types — `in_flight` (body-id set)
/// is only for AnalyzeBody dedup; `in_flight_count` is the general saturation gate.
#[derive(Resource)]
pub struct GenerationQueue {
    /// Pending work items, sorted by priority (index 0 = highest priority).
    pending: Arc<Mutex<Vec<QueuedWork>>>,
    /// Completions channel — background tasks send here; main thread reads.
    completion_tx: Sender<GenCompletion>,
    completion_rx: Receiver<GenCompletion>,
    /// Rayon thread pool dedicated to generation work.
    pool: rayon::ThreadPool,
    /// Set of body_ids currently in-flight — used only for AnalyzeBody dedup.
    in_flight: Arc<Mutex<std::collections::BTreeSet<String>>>,
    /// Count of all work items currently executing in the Rayon pool.
    /// This is the saturation gate — all work item types increment/decrement it.
    in_flight_count: Arc<Mutex<usize>>,
    /// Thread count — caps concurrent dispatches so pending items accumulate
    /// and priority ordering is consulted before the pool has free threads.
    n_threads: usize,
}

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

impl GenerationQueue {
    /// Create a new queue with the D-206 thread count:
    /// `available_parallelism - 2`, minimum 1.
    pub fn new() -> Self {
        let n_threads = std::thread::available_parallelism()
            .map(|p| p.get().saturating_sub(2).max(1))
            .unwrap_or(1);
        Self::with_threads(n_threads)
    }

    /// Create a queue with a specific thread count (for testing).
    pub fn with_threads(n_threads: usize) -> Self {
        let pool = rayon::ThreadPoolBuilder::new()
            .num_threads(n_threads)
            .thread_name(|i| format!("gen-worker-{i}"))
            .build()
            .expect("failed to build generation rayon pool");

        let (tx, rx) = crossbeam_channel::unbounded();

        Self {
            pending: Arc::new(Mutex::new(Vec::new())),
            completion_tx: tx,
            completion_rx: rx,
            pool,
            in_flight: Arc::new(Mutex::new(std::collections::BTreeSet::new())),
            in_flight_count: Arc::new(Mutex::new(0)),
            n_threads,
        }
    }

    /// Submit a work item at the given priority.
    ///
    /// If an `AnalyzeBody` item for the same body_id is already in-flight or
    /// pending, the submission is silently ignored (idempotent).
    pub fn submit(&self, item: GenWorkItem, priority: GenPriority) {
        // Dedup AnalyzeBody submissions.
        if let Some(body_id) = item.body_id() {
            let in_flight = self.in_flight.lock().unwrap();
            if in_flight.contains(body_id) {
                return;
            }
            drop(in_flight);
            // Check pending list.
            let pending = self.pending.lock().unwrap();
            if pending.iter().any(|q| q.item.body_id() == Some(body_id)) {
                return;
            }
            drop(pending);
        }

        let mut pending = self.pending.lock().unwrap();
        let pos = pending
            .iter()
            .position(|q| q.priority > priority)
            .unwrap_or(pending.len());
        pending.insert(pos, QueuedWork { priority, item });
        drop(pending);

        self.dispatch_next();
    }

    /// Drain all completed items from the channel and dispatch pending work.
    ///
    /// Call once per tick from the main thread. Returns all completions
    /// available without blocking. After draining, dispatches as many pending
    /// items as there are free thread slots — this is the point where priority
    /// ordering matters, since the pool was saturated when items were submitted.
    pub fn drain_completions(&self) -> Vec<GenCompletion> {
        let mut out = Vec::new();
        while let Ok(c) = self.completion_rx.try_recv() {
            out.push(c);
        }
        // Fill any newly-freed slots.
        for _ in 0..out.len() {
            self.dispatch_next();
        }
        out
    }

    /// Number of items waiting in the pending queue.
    pub fn pending_count(&self) -> usize {
        self.pending.lock().unwrap().len()
    }

    // Dispatch the highest-priority pending item to the Rayon pool.
    //
    // Gated on in_flight_count < n_threads — applies to all work item types,
    // not just AnalyzeBody. When the pool is full, items stay in the sorted
    // pending Vec so priority ordering is consulted on the next free slot.
    fn dispatch_next(&self) {
        let item = {
            let count = self.in_flight_count.lock().unwrap();
            if *count >= self.n_threads {
                return;
            }
            drop(count);

            let mut pending = self.pending.lock().unwrap();
            if pending.is_empty() {
                return;
            }
            pending.remove(0).item
        };

        // Mark body as in-flight (AnalyzeBody dedup).
        if let Some(body_id) = item.body_id() {
            self.in_flight.lock().unwrap().insert(body_id.to_string());
        }
        // Increment general in-flight counter for all item types.
        *self.in_flight_count.lock().unwrap() += 1;

        let tx = self.completion_tx.clone();
        let in_flight = Arc::clone(&self.in_flight);
        let in_flight_count = Arc::clone(&self.in_flight_count);

        self.pool.spawn(move || {
            let completion = run_work_item(&item);

            // Un-mark body dedup set (AnalyzeBody only).
            if let Some(body_id) = item.body_id() {
                in_flight.lock().unwrap().remove(body_id);
            }
            // Decrement general counter for all item types.
            *in_flight_count.lock().unwrap() -= 1;

            let _ = tx.send(completion);
        });
    }
}

impl Default for GenerationQueue {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// Work execution stub
// ---------------------------------------------------------------------------

/// Execute one work item. This is the Rayon task body (off the tick thread).
///
/// `AnalyzeBody` runs the real Layer-1 cascade (#968, D-225). `GenerateSkeleton`
/// and `FillChunk` remain stubs — their layers (#957 / #959) are not built yet —
/// returning immediate success so the queue infrastructure stays testable.
fn run_work_item(item: &GenWorkItem) -> GenCompletion {
    match item {
        GenWorkItem::AnalyzeBody {
            body_id,
            heightmap_path,
            sea_level,
            body_seed,
        } => match load_heightmap_png(heightmap_path, body_id, *sea_level) {
            Ok(hm) => {
                // Layer 1 runs at the GRID_W×GRID_H working resolution (D-202):
                // downsample the higher-res stored heightmap first.
                let working = if hm.width > GRID_W || hm.height > GRID_H {
                    hm.downsample(GRID_W, GRID_H)
                } else {
                    hm
                };
                // Cities (&[]) get supplied once the Layer-3 settlement read is
                // wired through the work item (#955 follow-on); Topography needs none.
                let snapshot =
                    run_cascade_from_heightmap(*body_seed, working, &[], CascadeLayer::Topography);
                GenCompletion::BodyAnalyzed {
                    body_id: body_id.clone(),
                    state: snapshot.into_body_world_state(),
                }
            }
            Err(e) => GenCompletion::Failed {
                item: item.clone(),
                reason: format!("heightmap load failed: {e}"),
            },
        },
        GenWorkItem::GenerateSkeleton {
            city_id,
            body_id,
            context,
            district_id,
            chain,
            economic_role,
            population,
            founding_age_years,
        } => {
            // Build the Phase 1 skeleton from the pre-resolved D-199 context.
            // `economic_role`, `population`, and `founding_age_years` are the
            // D-199 raw fields carried alongside the context because
            // `generate_skeleton` accepts them as separate parameters.
            let skeleton = generate_skeleton(
                context,
                *population,
                economic_role,
                *district_id,
                *founding_age_years,
                *chain,
            );
            GenCompletion::SkeletonGenerated {
                city_id: *city_id,
                body_id: body_id.clone(),
                state: Box::new(DistrictWorldState {
                    skeleton,
                    block_tags: std::collections::BTreeMap::new(),
                }),
            }
        }
        GenWorkItem::FillChunk {
            district_id,
            block_pos,
        } => GenCompletion::ChunkFilled {
            district_id: *district_id,
            block_pos: *block_pos,
        },
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    fn make_queue() -> GenerationQueue {
        GenerationQueue::with_threads(2)
    }

    /// Write a tiny 16-bit grayscale heightmap PNG to a unique temp path so the
    /// real cascade can run in `run_work_item` 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_genq_{}_{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
    }

    /// Build an `AnalyzeBody` work item pointing at a tiny test heightmap.
    fn analyze(body_id: &str) -> GenWorkItem {
        GenWorkItem::AnalyzeBody {
            body_id: body_id.to_string(),
            heightmap_path: test_heightmap_path(),
            sea_level: 0.3,
            body_seed: SeedChain::for_body(42, body_id),
        }
    }

    /// Build a minimal `GenerateSkeleton` work item with a stub context.
    ///
    /// The stub context uses Commission/Regional/Urban defaults — the same
    /// values the existing skeleton_gen tests use.  These tests exercise queue
    /// mechanics (ordering, saturation, drain), not economic read-set content.
    fn gen_skeleton(city_id: u64) -> GenWorkItem {
        use crate::simulation::generator::{
            BulkClass, CityGenerationContext, FoundingOrientation, MorphologyZone,
            PoliticalArchetype, ProductionUbiquity, SettingType, WorldTier,
        };
        GenWorkItem::GenerateSkeleton {
            city_id,
            body_id: format!("TestBody{city_id}"),
            context: Box::new(CityGenerationContext {
                city_id,
                political_archetype: PoliticalArchetype::Commission,
                prosperity_baseline_bps: 6_000,
                surrounding_biome: SettingType::Urban,
                road_entry_directions: vec![],
                footprint_radius_km: 5.0,
                founding_orientation: FoundingOrientation::Cardinal,
                world_tier: WorldTier::Regional,
                morphology_zone: MorphologyZone::AlluvialPlain,
                trait_selection: vec![],
                dominant_bulk_class: BulkClass::NonPhysical,
                dominant_production_ubiquity: ProductionUbiquity::Common,
            }),
            district_id: city_id * 10,
            chain: SeedChain::root(42 + city_id),
            economic_role: "service_mixed".to_string(),
            population: 500_000,
            founding_age_years: 200,
        }
    }

    #[test]
    fn submit_and_drain() {
        let q = make_queue();
        q.submit(analyze("TestBody"), GenPriority::Medium);
        // Give Rayon time to load the heightmap and run the cascade.
        std::thread::sleep(Duration::from_millis(100));
        let completions = q.drain_completions();
        assert_eq!(completions.len(), 1);
        // The work item ran the real cascade and produced a populated state.
        assert!(matches!(
            &completions[0],
            GenCompletion::BodyAnalyzed { body_id, state }
                if body_id == "TestBody"
                    && state.heightmap_width == 32
                    && state.heightmap_height == 16
        ));
    }

    #[test]
    fn dedup_analyze_body() {
        let q = make_queue();
        // Submit the same body twice before it can complete.
        q.submit(analyze("Dup"), GenPriority::Low);
        q.submit(analyze("Dup"), GenPriority::Low);
        std::thread::sleep(Duration::from_millis(50));
        let completions = q.drain_completions();
        // Should have completed exactly once.
        assert_eq!(completions.len(), 1);
    }

    #[test]
    fn priority_ordering() {
        // Submit three items rapidly; Immediate should be dispatched first.
        // Uses 3 threads so all items can dispatch without hitting saturation.
        let q = GenerationQueue::with_threads(3);
        // Using GenerateSkeleton (no dedup logic) to test ordering directly.
        q.submit(gen_skeleton(1), GenPriority::Low);
        q.submit(gen_skeleton(2), GenPriority::Immediate);
        q.submit(gen_skeleton(3), GenPriority::Medium);
        std::thread::sleep(Duration::from_millis(100));
        let completions = q.drain_completions();
        assert_eq!(completions.len(), 3);
    }

    #[test]
    fn priority_ordering_respected_under_saturation() {
        // Single-thread queue: in_flight_count saturates at 1, so the second
        // item stays in the pending Vec and is dispatched in priority order.
        // Uses AnalyzeBody (distinct body_ids) so all paths — dedup set AND
        // in_flight_count — are exercised.
        let q = GenerationQueue::with_threads(1);
        // Submit Low first, then Immediate. With 1 thread:
        //   - "BodyA" (Low) dispatches immediately (pool empty).
        //   - "BodyB" (Immediate) is inserted at index 0 of the sorted pending
        //     Vec while "BodyA" is in-flight (in_flight_count = 1 = n_threads).
        //   - When "BodyA" completes, drain_completions() calls dispatch_next()
        //     which picks index 0 = "BodyB" (Immediate).
        q.submit(analyze("BodyA"), GenPriority::Low);
        q.submit(analyze("BodyB"), GenPriority::Immediate);
        // Wait for BodyA to complete.
        std::thread::sleep(Duration::from_millis(50));
        // drain_completions dispatches BodyB (Immediate, index 0 of pending).
        let first = q.drain_completions();
        // Wait for BodyB to complete.
        std::thread::sleep(Duration::from_millis(50));
        let second = q.drain_completions();

        assert_eq!(first.len(), 1);
        assert_eq!(second.len(), 1);
        assert!(
            matches!(&first[0], GenCompletion::BodyAnalyzed { body_id, .. } if body_id == "BodyA")
        );
        assert!(
            matches!(&second[0], GenCompletion::BodyAnalyzed { body_id, .. } if body_id == "BodyB")
        );
    }

    #[test]
    fn drain_empty_returns_empty() {
        let q = make_queue();
        let result = q.drain_completions();
        assert!(result.is_empty());
    }

    #[test]
    fn pending_count_decreases_after_completion() {
        let q = make_queue();
        q.submit(
            GenWorkItem::FillChunk {
                district_id: 99,
                block_pos: (0, 0),
            },
            GenPriority::High,
        );
        std::thread::sleep(Duration::from_millis(50));
        let completions = q.drain_completions();
        assert!(!completions.is_empty() || q.pending_count() == 0);
    }
}