//! 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, processes: &[String], process_map: &BTreeMap, denylist: &std::collections::BTreeSet<&str>, ) -> Option> { 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, pub process_map: Mutex>, } /// 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, rx: watch::Receiver>, stop: Arc, worker: Option>, } 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, ) -> io::Result { 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> { 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) { *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, tx: watch::Sender>, stop: Arc, ) { 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 { 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); } }