peerspeak/src/game/detector.rs

//! The detector service (§5): one cancellable background worker that polls the OS
//! adapters, runs the pure matcher + debouncer, and publishes the stable detected
//! game on a watch channel — only when it changes, so a flapping detector can't
//! spam `PeerState` re-announces.
//!
//! All the OS reads (Steam files / registry, the process scan) are blocking, so
//! the worker is a dedicated `std::thread`, not a tokio task; it owns the
//! [`SteamProbe`] cache and the [`Debouncer`] across ticks. The per-tick decision
//! is factored into the pure [`poll_once`] so the wiring of resolve + match +
//! debounce is unit-tested without any I/O.

use super::scan;
use super::steam::SteamProbe;
use super::{Debouncer, DetectedGame, ManualOverride, builtin_denylist, match_processes, resolve};
use std::collections::BTreeMap;
use std::io;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::Duration;
use tokio::sync::watch;

/// How often the detector samples Steam state + the process list.
pub const POLL_INTERVAL: Duration = Duration::from_secs(3);
/// Granularity of the cancellable sleep between polls, so a stop request is
/// honored promptly instead of after a full [`POLL_INTERVAL`].
const SLEEP_TICK: Duration = Duration::from_millis(200);

/// Apply one poll's worth of inputs to the debouncer, returning the new published
/// value **iff it changed** (the signal to re-announce presence / switch the
/// background). Pure: the caller supplies the already-fetched Steam detection and
/// process list, so resolve + match + debounce are testable with zero I/O.
pub fn poll_once(
    debouncer: &mut Debouncer,
    override_: &ManualOverride,
    steam: Option<DetectedGame>,
    processes: &[String],
    process_map: &BTreeMap<String, String>,
    denylist: &std::collections::BTreeSet<&str>,
) -> Option<Option<DetectedGame>> {
    let matched = match_processes(processes, process_map, denylist);
    let res = resolve(override_, steam, &matched);
    if debouncer.observe(res.game, res.immediate) {
        Some(debouncer.current().cloned())
    } else {
        None
    }
}

/// Shared, live-updatable inputs to the detector, written by core (manual override
/// changes, config edits to the process map) and read each poll by the worker.
#[derive(Default)]
pub struct DetectorInputs {
    pub override_: Mutex<ManualOverride>,
    pub process_map: Mutex<BTreeMap<String, String>>,
}

/// A running detector service. Holds the watch receiver for detected-game changes
/// and the shared inputs; dropping it (or calling [`stop`](Self::stop)) ends the
/// worker thread.
pub struct GameDetector {
    inputs: Arc<DetectorInputs>,
    rx: watch::Receiver<Option<DetectedGame>>,
    stop: Arc<AtomicBool>,
    worker: Option<JoinHandle<()>>,
}

impl GameDetector {
    /// Spawn the detector worker. `process_map` seeds the non-Steam mappings;
    /// `override_` seeds the manual override (usually `Auto`). The worker runs
    /// until [`stop`](Self::stop) or the returned `GameDetector` is dropped.
    pub fn spawn(
        override_: ManualOverride,
        process_map: BTreeMap<String, String>,
    ) -> io::Result<Self> {
        let inputs = Arc::new(DetectorInputs {
            override_: Mutex::new(override_),
            process_map: Mutex::new(process_map),
        });
        let (tx, rx) = watch::channel(None);
        let stop = Arc::new(AtomicBool::new(false));

        let worker_inputs = inputs.clone();
        let worker_stop = stop.clone();
        let worker = std::thread::Builder::new()
            .name("game-detector".to_string())
            .spawn(move || worker_loop(worker_inputs, tx, worker_stop))?;

        Ok(Self {
            inputs,
            rx,
            stop,
            worker: Some(worker),
        })
    }

    /// A clone of the watch receiver for detected-game changes. The current value
    /// is `None` until the first non-empty detection is debounced in.
    pub fn subscribe(&self) -> watch::Receiver<Option<DetectedGame>> {
        self.rx.clone()
    }

    /// Replace the manual override (applied on the next poll, immediately,
    /// bypassing debounce).
    pub fn set_override(&self, override_: ManualOverride) {
        *self.inputs.override_.lock().unwrap() = override_;
    }

    /// Replace the user process→name mappings (e.g. after a Settings edit).
    pub fn set_process_map(&self, map: BTreeMap<String, String>) {
        *self.inputs.process_map.lock().unwrap() = map;
    }

    /// Signal the worker to exit. Idempotent; also happens on drop.
    pub fn stop(&self) {
        self.stop.store(true, Ordering::Relaxed);
    }
}

impl Drop for GameDetector {
    fn drop(&mut self) {
        self.stop();
        if let Some(worker) = self.worker.take() {
            let _ = worker.join();
        }
    }
}

/// The blocking worker loop: probe, decide, publish on change, sleep (cancellably).
fn worker_loop(
    inputs: Arc<DetectorInputs>,
    tx: watch::Sender<Option<DetectedGame>>,
    stop: Arc<AtomicBool>,
) {
    let denylist = builtin_denylist();
    let mut steam = SteamProbe::new();
    let mut debouncer = Debouncer::default();

    while !stop.load(Ordering::Relaxed) {
        let override_ = inputs.override_.lock().unwrap().clone();
        let process_map = inputs.process_map.lock().unwrap().clone();

        let steam_game = steam.detect();
        let processes = scan::running_executables();

        if let Some(new_current) = poll_once(
            &mut debouncer,
            &override_,
            steam_game,
            &processes,
            &process_map,
            &denylist,
        ) {
            // A closed receiver means core shut down; stop quietly.
            if tx.send(new_current).is_err() {
                return;
            }
        }

        // Cancellable sleep: wake promptly on a stop request.
        let mut slept = Duration::ZERO;
        while slept < POLL_INTERVAL && !stop.load(Ordering::Relaxed) {
            std::thread::sleep(SLEEP_TICK);
            slept += SLEEP_TICK;
        }
    }
}

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

    fn game(id: &str, name: &str, source: GameSource) -> DetectedGame {
        DetectedGame {
            id: id.into(),
            name: Some(name.into()),
            source,
        }
    }

    fn map(pairs: &[(&str, &str)]) -> BTreeMap<String, String> {
        pairs
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect()
    }

    #[test]
    fn poll_once_debounces_steam_detection() {
        let deny = builtin_denylist();
        let mut d = Debouncer::default();
        let steam = game("steam:730", "CS2", GameSource::Steam);
        let empty = BTreeMap::new();

        // First poll: detected but not yet published (needs two hits).
        assert_eq!(
            poll_once(
                &mut d,
                &ManualOverride::Auto,
                Some(steam.clone()),
                &[],
                &empty,
                &deny
            ),
            None
        );
        // Second poll: published.
        assert_eq!(
            poll_once(
                &mut d,
                &ManualOverride::Auto,
                Some(steam.clone()),
                &[],
                &empty,
                &deny
            ),
            Some(Some(steam))
        );
        // Third identical poll: no change event.
        assert_eq!(
            poll_once(
                &mut d,
                &ManualOverride::Auto,
                Some(game("steam:730", "CS2", GameSource::Steam)),
                &[],
                &empty,
                &deny
            ),
            None
        );
    }

    #[test]
    fn poll_once_matches_process_when_no_steam() {
        let deny = builtin_denylist();
        let mut d = Debouncer::default();
        let procs = vec!["/games/hl2_linux".to_string()];
        let user = map(&[("hl2_linux", "Half-Life 2")]);

        poll_once(&mut d, &ManualOverride::Auto, None, &procs, &user, &deny);
        let change = poll_once(&mut d, &ManualOverride::Auto, None, &procs, &user, &deny);
        let published = change
            .expect("should publish on second hit")
            .expect("a game");
        assert_eq!(published.id, "exe:hl2_linux");
        assert_eq!(published.name.as_deref(), Some("Half-Life 2"));
    }

    #[test]
    fn poll_once_manual_override_is_immediate() {
        let deny = builtin_denylist();
        let mut d = Debouncer::default();
        let forced = game("steam:220", "HL2", GameSource::Steam);
        // Even with a live Steam detection of something else, the override wins now.
        let other = game("steam:730", "CS2", GameSource::Steam);
        let change = poll_once(
            &mut d,
            &ManualOverride::Force(forced.clone()),
            Some(other),
            &[],
            &BTreeMap::new(),
            &deny,
        );
        assert_eq!(change, Some(Some(forced)));
    }

    #[test]
    fn spawn_and_stop_is_clean() {
        // Smoke test the lifecycle: spawning and stopping must not panic, and the
        // initial published value is None.
        let det = GameDetector::spawn(ManualOverride::Auto, BTreeMap::new()).unwrap();
        assert_eq!(*det.subscribe().borrow(), None);
        det.set_override(ManualOverride::ForceNone);
        det.set_process_map(map(&[("x", "X")]));
        det.stop();
        // Dropping also stops; no hang/panic.
        drop(det);
    }
}