//! State shared between the pipeline threads.
//!
//! Only two things really need sharing: which turn is current, and the flags
//! that allow cutting whatever is in progress. Everything else travels over
//! channels. Keeping this surface small is what keeps interruption easy to
//! follow: cutting means bumping the turn and raising two flags.

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;

use asist_core::event::TurnId;
use asist_core::http::Cancel;

#[derive(Clone)]
pub struct Session {
    inner: Arc<Inner>,
}

struct Inner {
    /// Turn being served. Any work from an earlier turn that shows up on a
    /// channel is garbage and is thrown away.
    current: AtomicU64,
    /// The assistant is talking (or about to).
    speaking: AtomicBool,
    /// Cierre solicitado.
    stopping: AtomicBool,
    /// Cuts the model answer in progress.
    llm: Cancel,
    /// Cuts the synthesis in progress.
    tts: Cancel,
}

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

impl Session {
    pub fn new() -> Self {
        Self {
            inner: Arc::new(Inner {
                current: AtomicU64::new(0),
                speaking: AtomicBool::new(false),
                stopping: AtomicBool::new(false),
                llm: Cancel::new(),
                tts: Cancel::new(),
            }),
        }
    }

    pub fn current(&self) -> TurnId {
        TurnId(self.inner.current.load(Ordering::SeqCst))
    }

    /// Opens a new turn and returns its id.
    pub fn begin_turn(&self) -> TurnId {
        let id = TurnId(self.inner.current.fetch_add(1, Ordering::SeqCst) + 1);
        self.inner.llm.reset();
        self.inner.tts.reset();
        id
    }

    /// Is `turn` still the current turn?
    ///
    /// Threads check it before spending work: a sentence that belongs to an
    /// outdated turn must be neither synthesized nor heard.
    pub fn is_current(&self, turn: TurnId) -> bool {
        turn == self.current()
    }

    /// Cuts everything in progress for the current turn.
    pub fn interrupt(&self) {
        self.inner.llm.cancel();
        self.inner.tts.cancel();
        self.inner.speaking.store(false, Ordering::SeqCst);
    }

    pub fn llm_cancel(&self) -> &Cancel {
        &self.inner.llm
    }

    pub fn tts_cancel(&self) -> &Cancel {
        &self.inner.tts
    }

    pub fn set_speaking(&self, speaking: bool) {
        self.inner.speaking.store(speaking, Ordering::SeqCst);
    }

    pub fn is_speaking(&self) -> bool {
        self.inner.speaking.load(Ordering::SeqCst)
    }

    pub fn request_stop(&self) {
        self.inner.stopping.store(true, Ordering::SeqCst);
        self.interrupt();
    }

    pub fn is_stopping(&self) -> bool {
        self.inner.stopping.load(Ordering::SeqCst)
    }
}

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

    #[test]
    fn each_turn_gets_an_increasing_id() {
        let session = Session::new();
        assert_eq!(session.begin_turn(), TurnId(1));
        assert_eq!(session.begin_turn(), TurnId(2));
        assert_eq!(session.current(), TurnId(2));
    }

    #[test]
    fn work_from_an_old_turn_becomes_stale() {
        let session = Session::new();
        let first = session.begin_turn();
        session.begin_turn();
        assert!(!session.is_current(first), "the old turn must be discarded");
    }

    #[test]
    fn opening_a_turn_rearms_cancellation() {
        let session = Session::new();
        session.begin_turn();
        session.interrupt();
        assert!(session.llm_cancel().is_cancelled());

        session.begin_turn();
        assert!(
            !session.llm_cancel().is_cancelled(),
            "a new turn must not inherit the previous cancellation"
        );
    }

    #[test]
    fn interrupting_silences_and_cancels_both_stages() {
        let session = Session::new();
        session.begin_turn();
        session.set_speaking(true);
        session.interrupt();
        assert!(session.tts_cancel().is_cancelled());
        assert!(!session.is_speaking());
    }

    #[test]
    fn requesting_shutdown_also_interrupts() {
        let session = Session::new();
        session.request_stop();
        assert!(session.is_stopping());
        assert!(session.llm_cancel().is_cancelled());
    }
}