settled-reach/server/src/voice/queue.rs

//! Voice pipeline work queue (D-138, Spike 2).
//!
//! Bounded crossbeam channel with priority ordering and backpressure.
//! Queue full → request dropped silently, game serves base text.

use std::cmp::Ordering;
use std::collections::BinaryHeap;
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
use std::sync::Arc;

use crossbeam_channel::{Receiver, Sender, TrySendError};

use crate::npc::tell_state::TellCategory;
use crate::voice::prompt_builder::ContentType;

/// Queue capacity — requests beyond this are dropped (backpressure).
const QUEUE_CAPACITY: usize = 256;

/// Priority levels for voice requests.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Priority {
    /// P0 — plot-critical NPC in current zone.
    Critical,
    /// P1 — current zone NPCs.
    High,
    /// P2 — adjacent zone pre-voicing.
    Standard,
    /// P3 — distant zones.
    Background,
}

impl Priority {
    /// Lower number = higher priority (for min-heap ordering).
    fn rank(self) -> u8 {
        match self {
            Priority::Critical => 0,
            Priority::High => 1,
            Priority::Standard => 2,
            Priority::Background => 3,
        }
    }
}

/// A request to re-voice a piece of content.
#[derive(Debug, Clone)]
pub struct VoiceRequest {
    pub priority: Priority,
    pub npc_stable_id: u64,
    pub zone_id: u32,
    pub culture_id: String,
    pub base_text: String,
    pub content_type: ContentType,
    pub content_index: u16,
    pub tell_state: Option<TellCategory>,
    pub seed: u64,
}

/// Wrapper for priority ordering in the heap (higher priority = dequeued first).
impl PartialEq for VoiceRequest {
    fn eq(&self, other: &Self) -> bool {
        self.priority.rank() == other.priority.rank()
    }
}

impl Eq for VoiceRequest {}

impl PartialOrd for VoiceRequest {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for VoiceRequest {
    fn cmp(&self, other: &Self) -> Ordering {
        // Reverse: lower rank = higher priority = should come first
        other.priority.rank().cmp(&self.priority.rank())
    }
}

/// Priority-ordered voice work queue with backpressure.
///
/// Uses a crossbeam bounded channel as transport between the game thread
/// and worker pool, with a priority heap on the consumer side.
pub struct VoiceQueue {
    sender: Sender<VoiceRequest>,
    receiver: Receiver<VoiceRequest>,
    paused: Arc<AtomicBool>,
}

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

impl VoiceQueue {
    pub fn new() -> Self {
        let (sender, receiver) = crossbeam_channel::bounded(QUEUE_CAPACITY);
        Self {
            sender,
            receiver,
            paused: Arc::new(AtomicBool::new(false)),
        }
    }

    /// Submit a voice request. Returns `false` if the queue is full or paused.
    pub fn submit(&self, request: VoiceRequest) -> bool {
        if self.is_paused() {
            tracing::trace!("voice queue paused — dropping submission");
            return false;
        }
        match self.sender.try_send(request) {
            Ok(()) => true,
            Err(TrySendError::Full(_)) => {
                tracing::trace!("voice queue full — dropping request");
                false
            }
            Err(TrySendError::Disconnected(_)) => {
                tracing::warn!("voice queue disconnected");
                false
            }
        }
    }

    /// Get a clone of the receiver for worker threads.
    pub fn receiver(&self) -> Receiver<VoiceRequest> {
        self.receiver.clone()
    }

    /// Get the shared pause flag for worker threads.
    ///
    /// Workers check this flag to avoid processing requests during zone
    /// transitions. The same `Arc<AtomicBool>` is shared between the queue
    /// and the worker pool.
    pub fn paused_flag(&self) -> Arc<AtomicBool> {
        Arc::clone(&self.paused)
    }

    /// Current number of pending requests in the channel.
    pub fn pending_count(&self) -> usize {
        self.sender.len()
    }

    /// Pause the queue (zone transition start).
    pub fn pause(&self) {
        self.paused.store(true, AtomicOrdering::SeqCst);
    }

    /// Resume the queue (zone transition complete).
    pub fn resume(&self) {
        self.paused.store(false, AtomicOrdering::SeqCst);
    }

    /// Check if the queue is paused.
    pub fn is_paused(&self) -> bool {
        self.paused.load(AtomicOrdering::SeqCst)
    }

    /// Reprioritize all pending requests after a zone change.
    ///
    /// Drains the channel, re-tags each request's priority using the
    /// provided closure, and re-submits. Requests that no longer fit
    /// (queue full after re-submission) are dropped — same backpressure
    /// rule as normal submission.
    ///
    /// Call this between `pause()` and `resume()` during zone transitions
    /// so workers don't consume stale-priority requests mid-reshuffle.
    pub fn reprioritize<F>(&self, mut classify: F)
    where
        F: FnMut(&VoiceRequest) -> Priority,
    {
        // Drain all pending requests
        let mut pending = Vec::new();
        while let Ok(req) = self.receiver.try_recv() {
            pending.push(req);
        }

        let count = pending.len();
        let mut resubmitted = 0;

        // Re-tag and re-submit (bypass pause check — reprioritize is called
        // while paused and needs to refill the channel).
        for mut req in pending {
            req.priority = classify(&req);
            match self.sender.try_send(req) {
                Ok(()) => resubmitted += 1,
                Err(TrySendError::Full(_)) => {
                    tracing::trace!("voice queue full during reprioritize — dropping request");
                }
                Err(TrySendError::Disconnected(_)) => break,
            }
        }

        if count > 0 {
            tracing::debug!(
                drained = count,
                resubmitted,
                dropped = count - resubmitted,
                "voice queue reprioritized after zone change"
            );
        }
    }
}

/// Priority drain: collect all pending items from the channel into a
/// priority-ordered heap, then drain highest-priority first.
///
/// Used by workers to process the most important requests first when
/// multiple requests are queued.
pub struct PriorityDrain {
    heap: BinaryHeap<VoiceRequest>,
}

impl PriorityDrain {
    /// Drain all currently available items from the receiver into the heap.
    pub fn from_receiver(receiver: &Receiver<VoiceRequest>) -> Self {
        let mut heap = BinaryHeap::new();
        while let Ok(req) = receiver.try_recv() {
            heap.push(req);
        }
        Self { heap }
    }

    /// Pop the highest-priority request.
    pub fn pop(&mut self) -> Option<VoiceRequest> {
        self.heap.pop()
    }

    pub fn is_empty(&self) -> bool {
        self.heap.is_empty()
    }

    pub fn len(&self) -> usize {
        self.heap.len()
    }
}

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

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

    fn make_request(priority: Priority, text: &str) -> VoiceRequest {
        VoiceRequest {
            priority,
            npc_stable_id: 1,
            zone_id: 100,
            culture_id: "van-maanens-star".into(),
            base_text: text.into(),
            content_type: ContentType::Dialogue,
            content_index: 0,
            tell_state: None,
            seed: 42,
        }
    }

    #[test]
    fn submit_and_receive() {
        let queue = VoiceQueue::new();
        assert!(queue.submit(make_request(Priority::High, "Test")));
        assert_eq!(queue.pending_count(), 1);

        let req = queue.receiver().try_recv().unwrap();
        assert_eq!(req.base_text, "Test");
    }

    #[test]
    fn backpressure_drops_when_full() {
        let (sender, _receiver) = crossbeam_channel::bounded(2);
        // Fill the channel
        sender.try_send(make_request(Priority::High, "A")).unwrap();
        sender.try_send(make_request(Priority::High, "B")).unwrap();
        // Third should fail
        assert!(sender.try_send(make_request(Priority::High, "C")).is_err());
    }

    #[test]
    fn priority_drain_orders_correctly() {
        let queue = VoiceQueue::new();
        queue.submit(make_request(Priority::Background, "low"));
        queue.submit(make_request(Priority::Critical, "high"));
        queue.submit(make_request(Priority::Standard, "mid"));

        let mut drain = PriorityDrain::from_receiver(&queue.receiver());
        assert_eq!(drain.len(), 3);

        let first = drain.pop().unwrap();
        assert_eq!(first.base_text, "high");
        assert_eq!(first.priority, Priority::Critical);

        let second = drain.pop().unwrap();
        assert_eq!(second.base_text, "mid");

        let third = drain.pop().unwrap();
        assert_eq!(third.base_text, "low");
    }

    #[test]
    fn pause_and_resume() {
        let queue = VoiceQueue::new();
        assert!(!queue.is_paused());
        queue.pause();
        assert!(queue.is_paused());
        queue.resume();
        assert!(!queue.is_paused());
    }

    #[test]
    fn submit_rejected_when_paused() {
        let queue = VoiceQueue::new();
        queue.pause();
        assert!(!queue.submit(make_request(Priority::High, "should be rejected")));
        assert_eq!(queue.pending_count(), 0);
        queue.resume();
        assert!(queue.submit(make_request(Priority::High, "should be accepted")));
        assert_eq!(queue.pending_count(), 1);
    }

    #[test]
    fn reprioritize_reshuffles_on_zone_change() {
        let queue = VoiceQueue::new();

        // Zone 100 is current, zone 200 is adjacent
        let mut req_a = make_request(Priority::High, "current zone NPC");
        req_a.zone_id = 100;
        let mut req_b = make_request(Priority::Standard, "adjacent zone NPC");
        req_b.zone_id = 200;

        queue.submit(req_a);
        queue.submit(req_b);
        assert_eq!(queue.pending_count(), 2);

        // Player moves to zone 200 — reprioritize
        queue.pause();
        queue.reprioritize(|req| {
            if req.zone_id == 200 {
                Priority::High // was adjacent, now current
            } else {
                Priority::Background // was current, now distant
            }
        });
        queue.resume();

        // Drain with priority ordering — zone 200 should come first
        let mut drain = PriorityDrain::from_receiver(&queue.receiver());
        let first = drain.pop().unwrap();
        assert_eq!(first.zone_id, 200);
        assert_eq!(first.priority, Priority::High);

        let second = drain.pop().unwrap();
        assert_eq!(second.zone_id, 100);
        assert_eq!(second.priority, Priority::Background);
    }
}