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::sync::{Arc, Mutex};

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

// ---------------------------------------------------------------------------
// 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 D8 drainage analysis + attractor extraction for this body.
    AnalyzeBody { body_id: String },
    /// Generate a Phase 1 DistrictSkeleton for this city.
    GenerateSkeleton { city_id: u64 },
    /// 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,
    },
    SkeletonGenerated {
        city_id: u64,
    },
    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.
///
/// Currently a stub — real implementations will call `drainage::analyze()`,
/// the attractor pipeline, and the district skeleton generator. Stubs return
/// immediate success to allow the queue infrastructure to be tested independently.
fn run_work_item(item: &GenWorkItem) -> GenCompletion {
    match item {
        GenWorkItem::AnalyzeBody { body_id } => GenCompletion::BodyAnalyzed {
            body_id: body_id.clone(),
        },
        GenWorkItem::GenerateSkeleton { city_id } => {
            GenCompletion::SkeletonGenerated { city_id: *city_id }
        }
        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)
    }

    #[test]
    fn submit_and_drain() {
        let q = make_queue();
        q.submit(
            GenWorkItem::AnalyzeBody {
                body_id: "TestBody".to_string(),
            },
            GenPriority::Medium,
        );
        // Give Rayon time to complete the (stub) task.
        std::thread::sleep(Duration::from_millis(50));
        let completions = q.drain_completions();
        assert_eq!(completions.len(), 1);
        assert!(matches!(
            &completions[0],
            GenCompletion::BodyAnalyzed { body_id } if body_id == "TestBody"
        ));
    }

    #[test]
    fn dedup_analyze_body() {
        let q = make_queue();
        // Submit the same body twice before it can complete.
        q.submit(
            GenWorkItem::AnalyzeBody {
                body_id: "Dup".to_string(),
            },
            GenPriority::Low,
        );
        q.submit(
            GenWorkItem::AnalyzeBody {
                body_id: "Dup".to_string(),
            },
            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(
            GenWorkItem::GenerateSkeleton { city_id: 1 },
            GenPriority::Low,
        );
        q.submit(
            GenWorkItem::GenerateSkeleton { city_id: 2 },
            GenPriority::Immediate,
        );
        q.submit(
            GenWorkItem::GenerateSkeleton { city_id: 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(
            GenWorkItem::AnalyzeBody {
                body_id: "BodyA".to_string(),
            },
            GenPriority::Low,
        );
        q.submit(
            GenWorkItem::AnalyzeBody {
                body_id: "BodyB".to_string(),
            },
            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);
    }
}