omni_cli/

daemon.rs

//! omni daemon — persistent-connection desktop keystroke replay (bd-1eefe5).
//!
//! `omni daemon` runs as a lightweight service on a desktop, holds a persistent
//! connection, and REPLAYS received [`InjectionCommand`]s as local OS keystrokes
//! — so `omni send --daemon <addr>` / `omni cast --daemon <addr>` can cast to
//! desktops, not just phone/watch keyboards. Unlike tendril (a new SSH
//! connection per keystroke) the client keeps one established connection.
//!
//! The daemon speaks the same wire as the relay ([`crate::transport::relay`]):
//! each line is either a token-gated `cast` [`ControlFrame`] or a bare
//! [`InjectionCommand`]. Received commands are replayed through a pluggable
//! [`Injector`], so the connection/auth/loop is testable headless and real OS
//! key synthesis is swappable. The same local port accepts `subscribe` clients;
//! every accepted command is injected once and mirrored to all subscribers.

use std::io::{BufRead, BufReader, Write};
use std::net::{Shutdown, TcpListener, TcpStream};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::{self, Receiver, SyncSender, TrySendError};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

use crate::AppError;
use crate::events::SemanticEvent;
use crate::transport::relay::ControlFrame;
use crate::wire::{InjectionCommand, WIRE_VERSION};

// ---------------------------------------------------------------------------
// injector
// ---------------------------------------------------------------------------

/// Replays an [`InjectionCommand`] as local OS keystrokes.
pub trait Injector: Send + Sync {
    /// Replay one command. A per-command failure is logged by the caller and
    /// does not tear down the daemon.
    fn inject(&self, cmd: &InjectionCommand) -> Result<(), AppError>;
    /// Short human label (for `omni daemon` startup logging).
    fn label(&self) -> &str;
    /// Additive application-event hook. Never fall back to OS injection.
    fn semantic_event(&self, _event: &SemanticEvent) -> Result<(), AppError> {
        Ok(())
    }
    fn supports_events(&self) -> bool {
        false
    }
}

/// An injector that only logs what it would replay — the safe default for
/// headless hosts and `--dry-run`. Writes one NDJSON line per command to stderr.
#[derive(Debug, Default)]
pub struct LogInjector;

impl Injector for LogInjector {
    fn inject(&self, cmd: &InjectionCommand) -> Result<(), AppError> {
        eprintln!("omni daemon: replay {}", cmd.json_line()?);
        Ok(())
    }
    fn label(&self) -> &str {
        "log"
    }
}

/// An injector that records every command it is asked to replay (for tests).
#[derive(Debug, Default, Clone)]
pub struct CaptureInjector {
    seen: Arc<Mutex<Vec<InjectionCommand>>>,
}

impl CaptureInjector {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }
    /// Snapshot of the commands replayed so far.
    #[must_use]
    pub fn captured(&self) -> Vec<InjectionCommand> {
        self.seen.lock().unwrap().clone()
    }
}

impl Injector for CaptureInjector {
    fn inject(&self, cmd: &InjectionCommand) -> Result<(), AppError> {
        self.seen.lock().unwrap().push(cmd.clone());
        Ok(())
    }
    fn label(&self) -> &str {
        "capture"
    }
}

/// Injector adapter used by relay-device mode: inject once, then mirror the
/// received command to every local stream subscriber.
pub struct BroadcastingInjector {
    inner: Arc<dyn Injector>,
    bus: CommandBus,
}

impl BroadcastingInjector {
    #[must_use]
    pub fn new(inner: Arc<dyn Injector>, bus: CommandBus) -> Self {
        Self { inner, bus }
    }
}

impl Injector for BroadcastingInjector {
    fn inject(&self, cmd: &InjectionCommand) -> Result<(), AppError> {
        // Fan out first so a slow platform injector cannot add latency to app
        // subscribers. The bus is non-blocking and isolates slow consumers.
        self.bus.publish(cmd);
        self.inner.inject(cmd)
    }

    fn label(&self) -> &str {
        self.inner.label()
    }
    fn semantic_event(&self, event: &SemanticEvent) -> Result<(), AppError> {
        self.bus.publish_event(event);
        Ok(())
    }
    fn supports_events(&self) -> bool {
        true
    }
}

/// Linux/X11 injector: replay commands via `xdotool`.
#[derive(Debug, Default)]
pub struct XdotoolInjector;

impl Injector for XdotoolInjector {
    fn inject(&self, cmd: &InjectionCommand) -> Result<(), AppError> {
        for argv in xdotool_argvs(cmd)? {
            run_argv(&argv)?;
        }
        Ok(())
    }
    fn label(&self) -> &str {
        "xdotool"
    }
}

/// Run one argv with [`std::process::Command`], mapping a non-zero exit / spawn
/// failure to an [`AppError`].
fn run_argv(argv: &[String]) -> Result<(), AppError> {
    let (prog, rest) = argv
        .split_first()
        .ok_or_else(|| AppError::internal("inject_argv", "empty argv"))?;
    let status = std::process::Command::new(prog)
        .args(rest)
        .status()
        .map_err(|e| AppError::internal("inject_spawn", format!("{prog}: {e}")))?;
    if status.success() {
        Ok(())
    } else {
        Err(AppError::internal(
            "inject_status",
            format!("{} exited with {status}", argv.join(" ")),
        ))
    }
}

/// Map a semantic key name to an X11 keysym understood by `xdotool key`.
/// Unknown names pass through unchanged (xdotool accepts many keysyms).
#[must_use]
pub fn xdotool_keysym(name: &str) -> String {
    match name {
        "enter" | "return" => "Return",
        "esc" | "escape" => "Escape",
        "tab" => "Tab",
        "space" => "space",
        "backspace" => "BackSpace",
        "delete" | "del" => "Delete",
        "left" => "Left",
        "right" => "Right",
        "up" => "Up",
        "down" => "Down",
        "home" => "Home",
        "end" => "End",
        other => other,
    }
    .to_owned()
}

/// Map one combo token to its xdotool form: modifiers are normalised, other
/// tokens go through [`xdotool_keysym`].
fn xdotool_combo_token(tok: &str) -> String {
    match tok.trim().to_ascii_lowercase().as_str() {
        "ctrl" | "control" => "ctrl".to_owned(),
        "alt" | "option" => "alt".to_owned(),
        "shift" => "shift".to_owned(),
        "super" | "cmd" | "command" | "meta" | "win" => "super".to_owned(),
        other => xdotool_keysym(other),
    }
}

/// Build the `xdotool` argv(s) for a command. A command usually maps to one
/// invocation; `scroll 0` maps to none.
///
/// # Errors
/// [`AppError::validation`] if a command is missing its payload.
pub fn xdotool_argvs(cmd: &InjectionCommand) -> Result<Vec<Vec<String>>, AppError> {
    use crate::wire::Kind;
    let s = |v: &str| v.to_owned();
    match cmd.kind {
        Kind::Text => {
            let text = cmd
                .text
                .as_deref()
                .ok_or_else(|| AppError::validation("inject_text", "text command missing text"))?;
            Ok(vec![vec![
                s("xdotool"),
                s("type"),
                s("--clearmodifiers"),
                s("--"),
                s(text),
            ]])
        }
        Kind::Key => {
            let key = cmd
                .key
                .as_deref()
                .ok_or_else(|| AppError::validation("inject_key", "key command missing key"))?;
            Ok(vec![vec![
                s("xdotool"),
                s("key"),
                s("--clearmodifiers"),
                xdotool_keysym(key),
            ]])
        }
        Kind::Combo => {
            let combo = cmd.combo.as_deref().ok_or_else(|| {
                AppError::validation("inject_combo", "combo command missing combo")
            })?;
            let chord = combo
                .split(',')
                .filter(|t| !t.trim().is_empty())
                .map(xdotool_combo_token)
                .collect::<Vec<_>>()
                .join("+");
            if chord.is_empty() {
                return Err(AppError::validation("inject_combo", "empty combo"));
            }
            Ok(vec![vec![
                s("xdotool"),
                s("key"),
                s("--clearmodifiers"),
                chord,
            ]])
        }
        Kind::Scroll => {
            let amount = cmd.amount.ok_or_else(|| {
                AppError::validation("inject_scroll", "scroll command missing amount")
            })?;
            if amount == 0 {
                return Ok(vec![]);
            }
            // xdotool buttons: 4 = wheel up, 5 = wheel down. Positive = down.
            let button = if amount > 0 { "5" } else { "4" };
            let repeat = amount.unsigned_abs().to_string();
            Ok(vec![vec![
                s("xdotool"),
                s("click"),
                s("--repeat"),
                repeat,
                s(button),
            ]])
        }
    }
}

/// Linux/Wayland injector via `wtype`.
#[derive(Debug, Default)]
pub struct WtypeInjector;

impl Injector for WtypeInjector {
    fn inject(&self, cmd: &InjectionCommand) -> Result<(), AppError> {
        for argv in wtype_argvs(cmd)? {
            run_argv(&argv)?;
        }
        Ok(())
    }
    fn label(&self) -> &str {
        "wtype"
    }
}

/// wtype modifier name for a token, or `None` if it is not a modifier.
fn wtype_mod(tok: &str) -> Option<&'static str> {
    match tok.trim().to_ascii_lowercase().as_str() {
        "ctrl" | "control" => Some("ctrl"),
        "alt" | "option" => Some("alt"),
        "shift" => Some("shift"),
        "super" | "cmd" | "command" | "meta" | "win" => Some("logo"),
        _ => None,
    }
}

/// Build `wtype` argv(s) (Wayland). Reuses [`xdotool_keysym`] (wtype keys are
/// XKB keysyms too). `scroll` is unsupported.
///
/// # Errors
/// [`AppError::validation`] for a missing payload or an unsupported scroll.
pub fn wtype_argvs(cmd: &InjectionCommand) -> Result<Vec<Vec<String>>, AppError> {
    use crate::wire::Kind;
    let s = |v: &str| v.to_owned();
    match cmd.kind {
        Kind::Text => {
            let text = cmd
                .text
                .as_deref()
                .ok_or_else(|| AppError::validation("inject_text", "text command missing text"))?;
            Ok(vec![vec![s("wtype"), s("--"), s(text)]])
        }
        Kind::Key => {
            let key = cmd
                .key
                .as_deref()
                .ok_or_else(|| AppError::validation("inject_key", "key command missing key"))?;
            Ok(vec![vec![s("wtype"), s("-k"), xdotool_keysym(key)]])
        }
        Kind::Combo => {
            let combo = cmd.combo.as_deref().ok_or_else(|| {
                AppError::validation("inject_combo", "combo command missing combo")
            })?;
            let tokens: Vec<&str> = combo
                .split(',')
                .map(str::trim)
                .filter(|t| !t.is_empty())
                .collect();
            if tokens.is_empty() {
                return Err(AppError::validation("inject_combo", "empty combo"));
            }
            let mods: Vec<&str> = tokens.iter().filter_map(|t| wtype_mod(t)).collect();
            let keys: Vec<&str> = tokens
                .iter()
                .copied()
                .filter(|t| wtype_mod(t).is_none())
                .collect();
            let mut argv = vec![s("wtype")];
            for m in &mods {
                argv.push(s("-M"));
                argv.push(s(m));
            }
            for k in &keys {
                argv.push(s("-k"));
                argv.push(xdotool_keysym(k));
            }
            for m in mods.iter().rev() {
                argv.push(s("-m"));
                argv.push(s(m));
            }
            Ok(vec![argv])
        }
        Kind::Scroll => Err(AppError::validation(
            "inject_scroll_unsupported",
            "wtype backend does not support scroll",
        )),
    }
}

/// Linux/Wayland injector via `ydotool` (evdev keycodes).
#[derive(Debug, Default)]
pub struct YdotoolInjector;

impl Injector for YdotoolInjector {
    fn inject(&self, cmd: &InjectionCommand) -> Result<(), AppError> {
        for argv in ydotool_argvs(cmd)? {
            run_argv(&argv)?;
        }
        Ok(())
    }
    fn label(&self) -> &str {
        "ydotool"
    }
}

/// Linux evdev keycode for a semantic key / character (for `ydotool key`).
#[must_use]
pub fn evdev_code(name: &str) -> Option<u16> {
    let code = match name.trim().to_ascii_lowercase().as_str() {
        "enter" | "return" => 28,
        "esc" | "escape" => 1,
        "tab" => 15,
        "space" => 57,
        "backspace" => 14,
        "delete" | "del" => 111,
        "left" => 105,
        "right" => 106,
        "up" => 103,
        "down" => 108,
        "home" => 102,
        "end" => 107,
        "ctrl" | "control" => 29,
        "shift" => 42,
        "alt" | "option" => 56,
        "super" | "cmd" | "command" | "meta" | "win" => 125,
        "a" => 30,
        "b" => 48,
        "c" => 46,
        "d" => 32,
        "e" => 18,
        "f" => 33,
        "g" => 34,
        "h" => 35,
        "i" => 23,
        "j" => 36,
        "k" => 37,
        "l" => 38,
        "m" => 50,
        "n" => 49,
        "o" => 24,
        "p" => 25,
        "q" => 16,
        "r" => 19,
        "s" => 31,
        "t" => 20,
        "u" => 22,
        "v" => 47,
        "w" => 17,
        "x" => 45,
        "y" => 21,
        "z" => 44,
        "1" => 2,
        "2" => 3,
        "3" => 4,
        "4" => 5,
        "5" => 6,
        "6" => 7,
        "7" => 8,
        "8" => 9,
        "9" => 10,
        "0" => 11,
        _ => return None,
    };
    Some(code)
}

/// Build `ydotool` argv(s) (Wayland/uinput). `scroll` is unsupported.
///
/// # Errors
/// [`AppError::validation`] for a missing payload, an unknown key/combo token
/// (no evdev mapping), or an unsupported scroll.
pub fn ydotool_argvs(cmd: &InjectionCommand) -> Result<Vec<Vec<String>>, AppError> {
    use crate::wire::Kind;
    let s = |v: &str| v.to_owned();
    match cmd.kind {
        Kind::Text => {
            let text = cmd
                .text
                .as_deref()
                .ok_or_else(|| AppError::validation("inject_text", "text command missing text"))?;
            Ok(vec![vec![s("ydotool"), s("type"), s("--"), s(text)]])
        }
        Kind::Key => {
            let key = cmd
                .key
                .as_deref()
                .ok_or_else(|| AppError::validation("inject_key", "key command missing key"))?;
            let code = evdev_code(key).ok_or_else(|| {
                AppError::validation(
                    "inject_key_code",
                    format!("no evdev keycode for key: {key}"),
                )
            })?;
            Ok(vec![vec![
                s("ydotool"),
                s("key"),
                format!("{code}:1"),
                format!("{code}:0"),
            ]])
        }
        Kind::Combo => {
            let combo = cmd.combo.as_deref().ok_or_else(|| {
                AppError::validation("inject_combo", "combo command missing combo")
            })?;
            let tokens: Vec<&str> = combo
                .split(',')
                .map(str::trim)
                .filter(|t| !t.is_empty())
                .collect();
            if tokens.is_empty() {
                return Err(AppError::validation("inject_combo", "empty combo"));
            }
            let codes: Vec<u16> = tokens
                .iter()
                .map(|t| {
                    evdev_code(t).ok_or_else(|| {
                        AppError::validation(
                            "inject_combo_code",
                            format!("no evdev keycode for combo token: {t}"),
                        )
                    })
                })
                .collect::<Result<_, _>>()?;
            let mut argv = vec![s("ydotool"), s("key")];
            for c in &codes {
                argv.push(format!("{c}:1"));
            }
            for c in codes.iter().rev() {
                argv.push(format!("{c}:0"));
            }
            Ok(vec![argv])
        }
        Kind::Scroll => Err(AppError::validation(
            "inject_scroll_unsupported",
            "ydotool backend does not support scroll",
        )),
    }
}

/// macOS injector via `osascript` (System Events).
#[derive(Debug, Default)]
pub struct OsascriptInjector;

impl Injector for OsascriptInjector {
    fn inject(&self, cmd: &InjectionCommand) -> Result<(), AppError> {
        for argv in osascript_argvs(cmd)? {
            run_argv(&argv)?;
        }
        Ok(())
    }
    fn label(&self) -> &str {
        "osascript"
    }
}

/// macOS virtual key code for a semantic key name.
#[must_use]
pub fn mac_key_code(name: &str) -> Option<u16> {
    let code = match name.trim().to_ascii_lowercase().as_str() {
        "enter" | "return" => 36,
        "esc" | "escape" => 53,
        "tab" => 48,
        "space" => 49,
        "backspace" => 51,
        "delete" | "del" => 117,
        "left" => 123,
        "right" => 124,
        "down" => 125,
        "up" => 126,
        "home" => 115,
        "end" => 119,
        _ => return None,
    };
    Some(code)
}

fn mac_modifier(tok: &str) -> Option<&'static str> {
    match tok.trim().to_ascii_lowercase().as_str() {
        "ctrl" | "control" => Some("control down"),
        "alt" | "option" => Some("option down"),
        "shift" => Some("shift down"),
        "super" | "cmd" | "command" | "meta" | "win" => Some("command down"),
        _ => None,
    }
}

fn applescript_quote(s: &str) -> String {
    format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\""))
}

/// Build `osascript -e <script>` argv(s) (macOS). `scroll` is unsupported, and
/// a combo must have exactly one non-modifier key.
///
/// # Errors
/// [`AppError::validation`] for a missing payload, an unknown special key, a
/// multi-key combo, or an unsupported scroll.
pub fn osascript_argvs(cmd: &InjectionCommand) -> Result<Vec<Vec<String>>, AppError> {
    use crate::wire::Kind;
    let wrap = |body: String| {
        vec![vec![
            "osascript".to_owned(),
            "-e".to_owned(),
            format!("tell application \"System Events\" to {body}"),
        ]]
    };
    match cmd.kind {
        Kind::Text => {
            let text = cmd
                .text
                .as_deref()
                .ok_or_else(|| AppError::validation("inject_text", "text command missing text"))?;
            Ok(wrap(format!("keystroke {}", applescript_quote(text))))
        }
        Kind::Key => {
            let key = cmd
                .key
                .as_deref()
                .ok_or_else(|| AppError::validation("inject_key", "key command missing key"))?;
            let code = mac_key_code(key).ok_or_else(|| {
                AppError::validation(
                    "inject_key_code",
                    format!("no macOS key code for key: {key}"),
                )
            })?;
            Ok(wrap(format!("key code {code}")))
        }
        Kind::Combo => {
            let combo = cmd.combo.as_deref().ok_or_else(|| {
                AppError::validation("inject_combo", "combo command missing combo")
            })?;
            let tokens: Vec<&str> = combo
                .split(',')
                .map(str::trim)
                .filter(|t| !t.is_empty())
                .collect();
            if tokens.is_empty() {
                return Err(AppError::validation("inject_combo", "empty combo"));
            }
            let mods: Vec<&str> = tokens.iter().filter_map(|t| mac_modifier(t)).collect();
            let keys: Vec<&str> = tokens
                .iter()
                .copied()
                .filter(|t| mac_modifier(t).is_none())
                .collect();
            if keys.len() != 1 {
                return Err(AppError::validation(
                    "inject_combo_macos",
                    "macOS combo needs exactly one non-modifier key",
                ));
            }
            let using = if mods.is_empty() {
                String::new()
            } else {
                format!(" using {{{}}}", mods.join(", "))
            };
            let key = keys[0];
            let body = if let Some(code) = mac_key_code(key) {
                format!("key code {code}{using}")
            } else {
                format!("keystroke {}{using}", applescript_quote(key))
            };
            Ok(wrap(body))
        }
        Kind::Scroll => Err(AppError::validation(
            "inject_scroll_unsupported",
            "osascript backend does not support scroll",
        )),
    }
}

/// Which injector backend to use.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Backend {
    /// Log-only (safe default; headless / --dry-run).
    Log,
    /// Linux/X11 via xdotool.
    Xdotool,
    /// Linux/Wayland via wtype.
    Wtype,
    /// Linux/Wayland via ydotool (uinput).
    Ydotool,
    /// macOS via osascript (System Events).
    Osascript,
}

impl Backend {
    /// Parse the `--backend` value.
    ///
    /// # Errors
    /// [`AppError::validation`] for an unknown backend.
    pub fn parse(s: &str) -> Result<Self, AppError> {
        match s {
            "log" => Ok(Backend::Log),
            "xdotool" => Ok(Backend::Xdotool),
            "wtype" => Ok(Backend::Wtype),
            "ydotool" => Ok(Backend::Ydotool),
            "osascript" => Ok(Backend::Osascript),
            other => Err(AppError::validation(
                "daemon_backend",
                format!("unknown --backend {other} (expected log|xdotool|wtype|ydotool|osascript)"),
            )),
        }
    }

    /// Build the injector for this backend.
    #[must_use]
    pub fn injector(self) -> Arc<dyn Injector> {
        match self {
            Backend::Log => Arc::new(LogInjector),
            Backend::Xdotool => Arc::new(XdotoolInjector),
            Backend::Wtype => Arc::new(WtypeInjector),
            Backend::Ydotool => Arc::new(YdotoolInjector),
            Backend::Osascript => Arc::new(OsascriptInjector),
        }
    }
}

/// Whether `name` resolves to a file on `$PATH`.
#[must_use]
pub fn tool_on_path(name: &str) -> bool {
    std::env::var_os("PATH")
        .is_some_and(|paths| std::env::split_paths(&paths).any(|dir| dir.join(name).is_file()))
}

/// Resolve the `auto` backend from the environment + available tools.
#[must_use]
pub fn resolve_auto() -> Backend {
    resolve_auto_with(
        cfg!(target_os = "macos"),
        std::env::var_os("WAYLAND_DISPLAY").is_some(),
        std::env::var_os("DISPLAY").is_some(),
        &tool_on_path,
    )
}

/// Pure core of [`resolve_auto`] (OS / display / tool-presence injected for
/// testability): macOS -> osascript; Wayland -> wtype then ydotool; X11 ->
/// xdotool; else the safe log backend.
#[must_use]
pub fn resolve_auto_with(
    is_macos: bool,
    wayland: bool,
    x11: bool,
    has: &dyn Fn(&str) -> bool,
) -> Backend {
    if is_macos {
        return Backend::Osascript;
    }
    if wayland {
        if has("wtype") {
            return Backend::Wtype;
        }
        if has("ydotool") {
            return Backend::Ydotool;
        }
    }
    if x11 && has("xdotool") {
        return Backend::Xdotool;
    }
    Backend::Log
}

// ---------------------------------------------------------------------------
// daemon loop + local subscriber fan-out
// ---------------------------------------------------------------------------

/// Bounded, non-blocking fan-out of commands received by the daemon.
///
/// Every `omni listen` connection gets its own queue. A stalled subscriber can
/// drop its own events, but can never delay OS injection or other subscribers.
#[derive(Clone, Default)]
pub struct CommandBus {
    inner: Arc<Mutex<std::collections::HashMap<u64, SyncSender<InjectionCommand>>>>,
    next_id: Arc<AtomicU64>,
    events: crate::event_bus::EventBus,
}

impl CommandBus {
    const SUBSCRIBER_BUFFER: usize = 256;

    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    fn subscribe(&self) -> (u64, Receiver<InjectionCommand>) {
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let (tx, rx) = mpsc::sync_channel(Self::SUBSCRIBER_BUFFER);
        self.inner.lock().unwrap().insert(id, tx);
        (id, rx)
    }

    fn unsubscribe(&self, id: u64) {
        self.inner.lock().unwrap().remove(&id);
    }

    /// Publish without blocking the command/injection path. Full queues drop
    /// the newest event; disconnected subscribers are removed.
    pub fn publish(&self, cmd: &InjectionCommand) {
        self.inner.lock().unwrap().retain(|_, tx| {
            !matches!(tx.try_send(cmd.clone()), Err(TrySendError::Disconnected(_)))
        });
    }

    /// Non-injecting events use a separate opt-in subscriber set.
    pub fn publish_event(&self, event: &SemanticEvent) -> usize {
        self.events.publish(event)
    }
    #[must_use]
    pub fn event_subscriber_count(&self) -> usize {
        self.events.count()
    }

    /// Number of currently connected local command stream subscribers.
    #[must_use]
    pub fn subscriber_count(&self) -> usize {
        self.inner.lock().unwrap().len()
    }
}

/// Options for [`run_daemon`].
#[derive(Clone)]
pub struct DaemonOptions {
    /// Shared token. When `Some`, `cast` frames must carry a matching token and
    /// bare InjectionCommand lines are rejected. When `None`, both are accepted
    /// (loopback / local-trust mode).
    pub token: Option<String>,
    /// Stop accepting after this many connections (for tests); `None` = forever.
    pub max_conns: Option<usize>,
}

/// Run the daemon accept loop: each connection streams NDJSON; every received
/// [`InjectionCommand`] (from a `cast` frame or a bare line) is replayed through
/// `injector`. One thread per connection (mirrors `relay::serve`).
///
/// # Errors
/// [`AppError::internal`] on a fatal accept error.
pub fn run_daemon(
    listener: &TcpListener,
    injector: Arc<dyn Injector>,
    opts: &DaemonOptions,
) -> Result<(), AppError> {
    run_daemon_with_bus(listener, injector, opts, &CommandBus::new())
}

/// Run the daemon with a caller-provided command bus. Relay-device mode uses
/// this to share one bus between its outbound Helsinki connection and the local
/// listener serving any number of `omni listen` clients.
pub fn run_daemon_with_bus(
    listener: &TcpListener,
    injector: Arc<dyn Injector>,
    opts: &DaemonOptions,
    bus: &CommandBus,
) -> Result<(), AppError> {
    for (accepted, incoming) in listener.incoming().enumerate() {
        let stream = incoming.map_err(|e| AppError::internal("daemon_accept", e.to_string()))?;
        let injector = Arc::clone(&injector);
        let token = opts.token.clone();
        let bus = bus.clone();
        thread::spawn(move || {
            if let Err(error) = handle_conn(stream, injector.as_ref(), token.as_deref(), &bus) {
                eprintln!("omni daemon: connection error: {error}");
            }
        });
        if opts.max_conns.is_some_and(|max| accepted + 1 >= max) {
            break;
        }
    }
    Ok(())
}

fn token_matches(want: Option<&str>, got: &str) -> bool {
    want.is_none_or(|want| want.trim() == got.trim())
}

fn serve_subscriber(
    mut writer: TcpStream,
    mut reader: BufReader<TcpStream>,
    bus: &CommandBus,
) -> Result<(), AppError> {
    let (id, rx) = bus.subscribe();

    // A subscriber is normally write-only after its handshake. Monitor its read
    // half so an idle client that disconnects is removed immediately rather than
    // retaining one thread + queue until the next command happens to be sent.
    let monitor_bus = bus.clone();
    let shutdown = writer
        .try_clone()
        .map_err(|e| AppError::internal("daemon_clone", e.to_string()))?;
    let monitor = thread::spawn(move || {
        let mut line = String::new();
        loop {
            line.clear();
            match reader.read_line(&mut line) {
                Ok(0) | Err(_) => break,
                Ok(_) => {}
            }
        }
        monitor_bus.unsubscribe(id);
        let _ = shutdown.shutdown(Shutdown::Both);
    });

    for cmd in rx {
        let line = cmd.json_line()?;
        if writer
            .write_all(line.as_bytes())
            .and_then(|()| writer.write_all(b"\n"))
            .and_then(|()| writer.flush())
            .is_err()
        {
            break;
        }
    }
    bus.unsubscribe(id);
    let _ = writer.shutdown(Shutdown::Both);
    let _ = monitor.join();
    Ok(())
}

fn deliver_local(cmd: &InjectionCommand, injector: &dyn Injector, bus: &CommandBus) {
    // Subscribers are a low-latency tap, independent of whether the OS
    // injection backend succeeds.
    bus.publish(cmd);
    if let Err(error) = injector.inject(cmd) {
        eprintln!("omni daemon: inject failed: {error}");
    }
}

fn handle_conn(
    stream: TcpStream,
    injector: &dyn Injector,
    token: Option<&str>,
    bus: &CommandBus,
) -> Result<(), AppError> {
    // Loopback subscriptions are local-trust even when the daemon also carries
    // OMNI_RELAY_TOKEN for its outbound relay registration. Remote subscribers
    // must authenticate when a daemon token is configured.
    let trusted_local = stream.peer_addr().is_ok_and(|peer| peer.ip().is_loopback());
    let mut writer = stream
        .try_clone()
        .map_err(|e| AppError::internal("daemon_clone", e.to_string()))?;
    let mut reader = BufReader::new(stream);
    let mut first = String::new();

    // Read the first non-blank line so we can switch the whole connection into
    // subscription mode while retaining the read half for disconnect detection.
    loop {
        first.clear();
        let n = reader
            .read_line(&mut first)
            .map_err(|e| AppError::internal("daemon_read", e.to_string()))?;
        if n == 0 {
            return Ok(());
        }
        if !first.trim().is_empty() {
            break;
        }
    }

    let subscription = match serde_json::from_str::<ControlFrame>(first.trim()) {
        Ok(ControlFrame::Subscribe { token, .. }) => Some((token, false)),
        Ok(ControlFrame::SubscribeEvents { token, .. }) => Some((token, true)),
        _ => None,
    };
    if let Some((got, event_stream)) = subscription {
        if !trusted_local && !token_matches(token, &got) {
            let mut writer = writer;
            let error = ControlFrame::Error {
                v: WIRE_VERSION,
                reason: "unauthorized".to_owned(),
            }
            .json_line()?;
            let _ = writer
                .write_all(error.as_bytes())
                .and_then(|()| writer.write_all(b"\n"))
                .and_then(|()| writer.flush());
            return Ok(());
        }
        if event_stream {
            return bus.events.serve(writer, reader);
        }
        return serve_subscriber(writer, reader, bus);
    }

    if !deliver_event_frame(first.trim(), token, bus, &mut writer)? {
        if let Some(cmd) = decode_line(first.trim(), token) {
            deliver_local(&cmd, injector, bus);
        }
    }
    for line in reader.lines() {
        let line = line.map_err(|e| AppError::internal("daemon_read", e.to_string()))?;
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        if !deliver_event_frame(trimmed, token, bus, &mut writer)? {
            if let Some(cmd) = decode_line(trimmed, token) {
                deliver_local(&cmd, injector, bus);
            }
        }
    }
    Ok(())
}

/// Returns true only for a semantic event frame, handled WITHOUT an Injector.
fn deliver_event_frame(
    line: &str,
    token: Option<&str>,
    bus: &CommandBus,
    writer: &mut TcpStream,
) -> Result<bool, AppError> {
    let (device, event, authorized) = match serde_json::from_str::<ControlFrame>(line) {
        Ok(ControlFrame::EventCast {
            v,
            device,
            event,
            token: got,
        }) => (
            device,
            event,
            v == WIRE_VERSION && token_matches(token, &got),
        ),
        _ => match SemanticEvent::decode_line(line) {
            Ok(event) => ("local".to_owned(), event, token.is_none()),
            Err(_) => return Ok(false),
        },
    };
    let response = if !authorized {
        ControlFrame::Error {
            v: WIRE_VERSION,
            reason: "unauthorized".into(),
        }
    } else if event.validate().is_err() {
        ControlFrame::Error {
            v: WIRE_VERSION,
            reason: "invalid_event".into(),
        }
    } else {
        ControlFrame::EventResult {
            v: WIRE_VERSION,
            device,
            delivered: bus.publish_event(&event),
        }
    }
    .json_line()?;
    writer
        .write_all(response.as_bytes())
        .and_then(|()| writer.write_all(b"\n"))
        .and_then(|()| writer.flush())
        .map_err(|e| AppError::internal("event_daemon_reply", e.to_string()))?;
    Ok(true)
}

/// Decode one NDJSON line into a command to replay, honouring auth. Returns
/// `None` for control frames that are not casts, auth failures, or unparseable
/// lines (logged, never fatal).
fn decode_line(line: &str, token: Option<&str>) -> Option<InjectionCommand> {
    // Prefer a control frame (the relay/sender protocol).
    if let Ok(frame) = serde_json::from_str::<ControlFrame>(line) {
        return match frame {
            ControlFrame::Cast { cmd, token: t, .. } => match token {
                Some(want) if want.trim() != t.trim() => {
                    eprintln!("omni daemon: rejected cast (bad token)");
                    None
                }
                _ => Some(cmd),
            },
            // register/ping/pong are no-ops for a local replay daemon.
            _ => None,
        };
    }
    // Otherwise a bare InjectionCommand line — only when unauthenticated.
    if token.is_some() {
        eprintln!("omni daemon: rejected bare command (token required)");
        return None;
    }
    match InjectionCommand::decode_line(line) {
        Ok(cmd) => Some(cmd),
        Err(error) => {
            eprintln!("omni daemon: skipping unparseable line: {error}");
            None
        }
    }
}

// ---------------------------------------------------------------------------
// relay-device mode (register to `omni serve`; NAT-traversal desktop delivery)
// ---------------------------------------------------------------------------

/// Why a relay-device session ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SessionEnd {
    /// The relay closed the connection (EOF) — reconnect.
    Closed,
    /// The relay rejected registration (e.g. bad token) — do NOT reconnect.
    Rejected,
}

/// The platform string this build registers as (`linux` / `macos` / `windows`).
#[must_use]
pub fn current_platform() -> &'static str {
    if cfg!(target_os = "macos") {
        "macos"
    } else if cfg!(target_os = "windows") {
        "windows"
    } else {
        "linux"
    }
}

/// Register to a relay as a DEVICE over `reader`/`writer`, then drain bare
/// [`InjectionCommand`] lines, replaying each via `injector`. One pass over a
/// connected stream (no reconnect); returns how the session ended.
///
/// # Errors
/// [`AppError::internal`] on a write/read I/O error.
pub fn relay_device_session<R: BufRead, W: Write>(
    reader: R,
    mut writer: W,
    device: &str,
    platform: &str,
    token: &str,
    injector: &dyn Injector,
) -> Result<SessionEnd, AppError> {
    let register = ControlFrame::Register {
        v: WIRE_VERSION,
        device: device.to_owned(),
        platform: platform.to_owned(),
        token: token.to_owned(),
    };
    let line = if injector.supports_events() {
        // Legacy hubs ignore this additive property; new hubs opt this daemon
        // into event delivery without changing the Register public wire type.
        let mut value = serde_json::to_value(&register)
            .map_err(|e| AppError::internal("event_register", e.to_string()))?;
        value["events"] = serde_json::Value::Bool(true);
        serde_json::to_string(&value)
            .map_err(|e| AppError::internal("event_register", e.to_string()))?
    } else {
        register.json_line()?
    };
    writer
        .write_all(line.as_bytes())
        .and_then(|()| writer.write_all(b"\n"))
        .and_then(|()| writer.flush())
        .map_err(|e| AppError::internal("relay_dev_register", e.to_string()))?;

    for line in reader.lines() {
        let line = line.map_err(|e| AppError::internal("relay_dev_read", e.to_string()))?;
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        // The relay sends bare InjectionCommand lines to devices; a rejection
        // arrives as a typed ControlFrame::Error before the relay closes us.
        if let Ok(cmd) = InjectionCommand::decode_line(trimmed) {
            if let Err(error) = injector.inject(&cmd) {
                eprintln!("omni daemon: inject failed: {error}");
            }
        } else if let Ok(event) = SemanticEvent::decode_line(trimmed) {
            if injector.supports_events() {
                if let Err(error) = injector.semantic_event(&event) {
                    eprintln!("omni daemon: event delivery failed: {error}");
                }
            }
        } else if let Ok(frame) = serde_json::from_str::<ControlFrame>(trimmed) {
            match frame {
                ControlFrame::Error { reason, .. } => {
                    eprintln!("omni daemon: relay rejected registration: {reason}");
                    return Ok(SessionEnd::Rejected);
                }
                ControlFrame::Ping { .. } => {
                    // Symmetric keepalive robustness (mirrors the apple RelayClient
                    // fix): answer a relay-initiated ping with a pong so a device
                    // is never evicted if the hub adds app-level device keepalive.
                    // Best-effort: a failed pong write just means the next read
                    // ends the session, which the reconnect loop already handles.
                    let _ = writer
                        .write_all(br#"{"v":1,"op":"pong"}"#)
                        .and_then(|()| writer.write_all(b"\n"))
                        .and_then(|()| writer.flush());
                }
                // Ignore other non-command frames (e.g. a stray pong from the hub
                // answering a future device-initiated ping).
                _ => {}
            }
        }
    }
    Ok(SessionEnd::Closed)
}

/// Options for [`run_relay_device`].
#[derive(Clone)]
pub struct RelayDeviceOptions {
    /// Device id to register as (senders target it via `--device`).
    pub device: String,
    /// Platform string (`linux` / `macos` / `windows`).
    pub platform: String,
    /// Shared token for the relay register frame.
    pub token: String,
    /// Stop after this many (re)connect attempts; `None` = run forever.
    pub max_attempts: Option<usize>,
    /// Delay between reconnect attempts.
    pub backoff: Duration,
}

/// Run the relay-device loop: connect to `relay_addr`, register, drain + replay,
/// and reconnect-with-backoff on a clean drop. Stops (returns `Ok`) on a relay
/// rejection (token/config error) or when `max_attempts` is reached.
///
/// # Errors
/// [`AppError::internal`] only on an unexpected socket-clone failure; connect /
/// session failures are logged and retried.
pub fn run_relay_device(
    relay_addr: &str,
    injector: &dyn Injector,
    opts: &RelayDeviceOptions,
) -> Result<(), AppError> {
    let mut attempts = 0usize;
    loop {
        attempts += 1;
        match TcpStream::connect(relay_addr) {
            Ok(stream) => {
                let reader = match stream.try_clone() {
                    Ok(s) => BufReader::new(s),
                    Err(e) => {
                        return Err(AppError::internal("relay_dev_clone", e.to_string()));
                    }
                };
                match relay_device_session(
                    reader,
                    stream,
                    &opts.device,
                    &opts.platform,
                    &opts.token,
                    injector,
                ) {
                    Ok(SessionEnd::Rejected) => {
                        eprintln!(
                            "omni daemon: relay rejected registration — check --token; not reconnecting"
                        );
                        return Ok(());
                    }
                    Ok(SessionEnd::Closed) => {
                        eprintln!("omni daemon: relay closed connection; reconnecting");
                    }
                    Err(error) => {
                        eprintln!("omni daemon: relay session error: {error}; reconnecting");
                    }
                }
            }
            Err(e) => {
                eprintln!("omni daemon: relay connect to {relay_addr} failed: {e}; retrying");
            }
        }
        if let Some(max) = opts.max_attempts {
            if attempts >= max {
                return Ok(());
            }
        }
        thread::sleep(opts.backoff);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::transport::relay::{bind, cast_via_relay};
    use crate::wire::Kind;
    use std::net::TcpStream;
    use std::time::Duration;

    #[test]
    fn backend_parse() {
        assert_eq!(Backend::parse("log").unwrap(), Backend::Log);
        assert_eq!(Backend::parse("xdotool").unwrap(), Backend::Xdotool);
        assert_eq!(Backend::parse("wtype").unwrap(), Backend::Wtype);
        assert_eq!(Backend::parse("ydotool").unwrap(), Backend::Ydotool);
        assert_eq!(Backend::parse("osascript").unwrap(), Backend::Osascript);
        assert_eq!(Backend::parse("nope").unwrap_err().code(), "daemon_backend");
    }

    #[test]
    fn wtype_text_key_combo() {
        assert_eq!(
            wtype_argvs(&InjectionCommand::text("hi")).unwrap(),
            vec![vec!["wtype".to_owned(), "--".to_owned(), "hi".to_owned()]]
        );
        assert_eq!(
            wtype_argvs(&InjectionCommand::key("enter")).unwrap()[0],
            vec!["wtype".to_owned(), "-k".to_owned(), "Return".to_owned()]
        );
        assert_eq!(
            wtype_argvs(&InjectionCommand::combo("ctrl,a,g")).unwrap()[0],
            ["wtype", "-M", "ctrl", "-k", "a", "-k", "g", "-m", "ctrl"]
                .into_iter()
                .map(String::from)
                .collect::<Vec<_>>()
        );
        assert_eq!(
            wtype_argvs(&InjectionCommand::scroll(1))
                .unwrap_err()
                .code(),
            "inject_scroll_unsupported"
        );
    }

    #[test]
    fn ydotool_evdev_mapping() {
        assert_eq!(evdev_code("enter"), Some(28));
        assert_eq!(evdev_code("a"), Some(30));
        assert_eq!(evdev_code("f13"), None);
        assert_eq!(
            ydotool_argvs(&InjectionCommand::key("enter")).unwrap()[0],
            ["ydotool", "key", "28:1", "28:0"]
                .into_iter()
                .map(String::from)
                .collect::<Vec<_>>()
        );
        assert_eq!(
            ydotool_argvs(&InjectionCommand::combo("ctrl,a")).unwrap()[0],
            ["ydotool", "key", "29:1", "30:1", "30:0", "29:0"]
                .into_iter()
                .map(String::from)
                .collect::<Vec<_>>()
        );
        assert_eq!(
            ydotool_argvs(&InjectionCommand::key("f13"))
                .unwrap_err()
                .code(),
            "inject_key_code"
        );
        assert_eq!(
            ydotool_argvs(&InjectionCommand::scroll(1))
                .unwrap_err()
                .code(),
            "inject_scroll_unsupported"
        );
    }

    #[test]
    fn osascript_scripts() {
        let argv = osascript_argvs(&InjectionCommand::text("hello")).unwrap();
        assert_eq!(argv[0][0], "osascript");
        assert_eq!(argv[0][1], "-e");
        assert!(argv[0][2].contains("keystroke \"hello\""));
        assert!(
            osascript_argvs(&InjectionCommand::key("enter")).unwrap()[0][2].contains("key code 36")
        );
        assert!(
            osascript_argvs(&InjectionCommand::combo("cmd,a")).unwrap()[0][2]
                .contains("keystroke \"a\" using {command down}")
        );
        assert!(
            osascript_argvs(&InjectionCommand::combo("cmd,enter")).unwrap()[0][2]
                .contains("key code 36 using {command down}")
        );
        assert_eq!(
            osascript_argvs(&InjectionCommand::combo("ctrl,a,g"))
                .unwrap_err()
                .code(),
            "inject_combo_macos"
        );
        assert_eq!(
            osascript_argvs(&InjectionCommand::scroll(1))
                .unwrap_err()
                .code(),
            "inject_scroll_unsupported"
        );
    }

    #[test]
    fn auto_backend_resolution() {
        let no = |_: &str| false;
        assert_eq!(
            resolve_auto_with(true, false, false, &no),
            Backend::Osascript
        );
        assert_eq!(
            resolve_auto_with(false, true, false, &|n: &str| n == "wtype"),
            Backend::Wtype
        );
        assert_eq!(
            resolve_auto_with(false, true, false, &|n: &str| n == "ydotool"),
            Backend::Ydotool
        );
        assert_eq!(
            resolve_auto_with(false, false, true, &|n: &str| n == "xdotool"),
            Backend::Xdotool
        );
        assert_eq!(resolve_auto_with(false, false, false, &no), Backend::Log);
        assert_eq!(resolve_auto_with(false, true, false, &no), Backend::Log);
    }

    #[test]
    fn xdotool_text_argv() {
        let argvs = xdotool_argvs(&InjectionCommand::text("hello world")).unwrap();
        assert_eq!(
            argvs,
            vec![vec![
                "xdotool".to_owned(),
                "type".to_owned(),
                "--clearmodifiers".to_owned(),
                "--".to_owned(),
                "hello world".to_owned(),
            ]]
        );
    }

    #[test]
    fn xdotool_key_maps_semantic_names() {
        let argvs = xdotool_argvs(&InjectionCommand::key("enter")).unwrap();
        assert_eq!(argvs[0].last().unwrap(), "Return");
        let argvs = xdotool_argvs(&InjectionCommand::key("esc")).unwrap();
        assert_eq!(argvs[0].last().unwrap(), "Escape");
        // unknown -> passthrough
        let argvs = xdotool_argvs(&InjectionCommand::key("F5")).unwrap();
        assert_eq!(argvs[0].last().unwrap(), "F5");
    }

    #[test]
    fn xdotool_combo_chord() {
        let argvs = xdotool_argvs(&InjectionCommand::combo("ctrl,a,g")).unwrap();
        assert_eq!(argvs[0].last().unwrap(), "ctrl+a+g");
        // modifier aliases + semantic key inside a combo
        let argvs = xdotool_argvs(&InjectionCommand::combo("cmd,shift,enter")).unwrap();
        assert_eq!(argvs[0].last().unwrap(), "super+shift+Return");
    }

    #[test]
    fn xdotool_scroll_direction_and_repeat() {
        let down = xdotool_argvs(&InjectionCommand::scroll(3)).unwrap();
        assert_eq!(
            down,
            vec![vec![
                "xdotool".to_owned(),
                "click".to_owned(),
                "--repeat".to_owned(),
                "3".to_owned(),
                "5".to_owned(),
            ]]
        );
        let up = xdotool_argvs(&InjectionCommand::scroll(-2)).unwrap();
        assert_eq!(up[0].last().unwrap(), "4");
        assert_eq!(up[0][3], "2");
        // zero -> no invocation
        assert!(
            xdotool_argvs(&InjectionCommand::scroll(0))
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn capture_injector_records() {
        let inj = CaptureInjector::new();
        inj.inject(&InjectionCommand::text("hi")).unwrap();
        inj.inject(&InjectionCommand::key("enter")).unwrap();
        assert_eq!(
            inj.captured(),
            vec![InjectionCommand::text("hi"), InjectionCommand::key("enter"),]
        );
        assert_eq!(inj.captured()[0].kind, Kind::Text);
    }

    #[test]
    fn command_bus_fans_out_to_multiple_subscribers() {
        let bus = CommandBus::new();
        let (_a_id, a) = bus.subscribe();
        let (_b_id, b) = bus.subscribe();
        let cmd = InjectionCommand::key("enter");
        bus.publish(&cmd);
        assert_eq!(a.recv_timeout(Duration::from_millis(50)).unwrap(), cmd);
        assert_eq!(b.recv_timeout(Duration::from_millis(50)).unwrap(), cmd);
        assert_eq!(bus.subscriber_count(), 2);
    }

    #[test]
    fn decode_line_auth_rules() {
        let cast = ControlFrame::Cast {
            v: 1,
            device: "local".into(),
            cmd: InjectionCommand::text("hi"),
            token: "secret".into(),
        }
        .json_line()
        .unwrap();
        // matching token
        assert_eq!(
            decode_line(&cast, Some("secret")),
            Some(InjectionCommand::text("hi"))
        );
        // bad token
        assert_eq!(decode_line(&cast, Some("nope")), None);
        // no token configured -> accept
        assert_eq!(decode_line(&cast, None), Some(InjectionCommand::text("hi")));
        // bare line accepted only when unauthenticated
        let bare = InjectionCommand::scroll(2).json_line().unwrap();
        assert_eq!(decode_line(&bare, None), Some(InjectionCommand::scroll(2)));
        assert_eq!(decode_line(&bare, Some("secret")), None);
    }

    #[test]
    fn decode_line_trims_token_whitespace() {
        // A cast token carrying a trailing newline (the sops-file footgun) still
        // matches a clean configured token, and vice versa (bd-790088).
        let cast = ControlFrame::Cast {
            v: 1,
            device: "local".into(),
            cmd: InjectionCommand::text("hi"),
            token: "secret\n".into(),
        }
        .json_line()
        .unwrap();
        assert_eq!(
            decode_line(&cast, Some("secret")),
            Some(InjectionCommand::text("hi"))
        );
        let clean = ControlFrame::Cast {
            v: 1,
            device: "local".into(),
            cmd: InjectionCommand::text("hi"),
            token: "secret".into(),
        }
        .json_line()
        .unwrap();
        assert_eq!(
            decode_line(&clean, Some("secret\n")),
            Some(InjectionCommand::text("hi"))
        );
    }

    #[test]
    fn end_to_end_daemon_replays_cast() {
        let listener = bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let inj = CaptureInjector::new();
        let inj_for_daemon: Arc<dyn Injector> = Arc::new(inj.clone());
        let opts = DaemonOptions {
            token: Some("tok".into()),
            max_conns: Some(1),
        };
        thread::spawn(move || {
            let _ = run_daemon(&listener, inj_for_daemon, &opts);
        });

        // Client casts two commands over one persistent connection.
        cast_via_relay(
            addr,
            "tok",
            "local",
            &[
                InjectionCommand::text("to desktop"),
                InjectionCommand::key("enter"),
            ],
        )
        .unwrap();

        // Wait for the daemon thread to replay them.
        let mut tries = 0;
        while inj.captured().len() < 2 && tries < 50 {
            thread::sleep(Duration::from_millis(20));
            tries += 1;
        }
        assert_eq!(
            inj.captured(),
            vec![
                InjectionCommand::text("to desktop"),
                InjectionCommand::key("enter"),
            ]
        );
    }

    #[test]
    fn end_to_end_daemon_rejects_bad_token() {
        let listener = bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let inj = CaptureInjector::new();
        let inj_for_daemon: Arc<dyn Injector> = Arc::new(inj.clone());
        let opts = DaemonOptions {
            token: Some("right".into()),
            max_conns: Some(1),
        };
        thread::spawn(move || {
            let _ = run_daemon(&listener, inj_for_daemon, &opts);
        });

        let _ = cast_via_relay(addr, "wrong", "local", &[InjectionCommand::text("nope")]);
        thread::sleep(Duration::from_millis(150));
        assert!(
            inj.captured().is_empty(),
            "bad-token cast must not be replayed"
        );

        // close the connection so the daemon's max_conns loop can exit
        let _ = TcpStream::connect(addr);
    }

    #[test]
    fn platform_string_is_known() {
        assert!(["linux", "macos", "windows"].contains(&current_platform()));
    }

    #[test]
    fn relay_device_registers_and_replays() {
        let listener = bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let opts = crate::transport::relay::ServeOptions {
            token: "tok".into(),
            max_conns: Some(2),
        };
        thread::spawn(move || {
            let _ = crate::transport::relay::serve(&listener, &opts);
        });

        // The desktop daemon registers as a device and drains commands.
        let inj = CaptureInjector::new();
        let inj_for_device = inj.clone();
        let dev = TcpStream::connect(addr).unwrap();
        let reader = std::io::BufReader::new(dev.try_clone().unwrap());
        thread::spawn(move || {
            let _ = relay_device_session(reader, dev, "desktop", "linux", "tok", &inj_for_device);
        });

        thread::sleep(Duration::from_millis(100)); // let the device register
        cast_via_relay(
            addr,
            "tok",
            "desktop",
            &[
                InjectionCommand::text("via relay"),
                InjectionCommand::key("enter"),
            ],
        )
        .unwrap();

        let mut tries = 0;
        while inj.captured().len() < 2 && tries < 50 {
            thread::sleep(Duration::from_millis(20));
            tries += 1;
        }
        assert_eq!(
            inj.captured(),
            vec![
                InjectionCommand::text("via relay"),
                InjectionCommand::key("enter"),
            ]
        );
    }

    #[test]
    fn relay_device_session_rejects_bad_token() {
        let listener = bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let opts = crate::transport::relay::ServeOptions {
            token: "right".into(),
            max_conns: Some(1),
        };
        thread::spawn(move || {
            let _ = crate::transport::relay::serve(&listener, &opts);
        });

        let dev = TcpStream::connect(addr).unwrap();
        let reader = std::io::BufReader::new(dev.try_clone().unwrap());
        let inj = CaptureInjector::new();
        let end = relay_device_session(reader, dev, "d", "linux", "wrong", &inj).unwrap();
        assert_eq!(end, SessionEnd::Rejected);
        assert!(inj.captured().is_empty());
    }

    #[test]
    fn relay_device_session_answers_ping_with_pong() {
        // Symmetric keepalive robustness: a relay-initiated ping must be answered
        // with a pong so the device is never evicted. In-memory transport, so the
        // assertion is purely on the device's outbound bytes.
        let incoming = b"{\"v\":1,\"op\":\"ping\"}\n";
        let reader = std::io::BufReader::new(&incoming[..]);
        let mut writer: Vec<u8> = Vec::new();
        let inj = CaptureInjector::new();
        let end = relay_device_session(reader, &mut writer, "dev", "linux", "tok", &inj).unwrap();
        assert_eq!(end, SessionEnd::Closed);
        let sent = String::from_utf8(writer).unwrap();
        // Register frame first, then the pong answer to the ping.
        assert!(
            sent.contains(r#""op":"register""#),
            "register frame missing: {sent}"
        );
        assert!(
            sent.contains(r#""op":"pong""#),
            "pong answer missing: {sent}"
        );
        // A ping is a control frame, not a command, so nothing is injected.
        assert!(inj.captured().is_empty());
    }
}