use crate::{
    capture::{Capture, CaptureType, ICaptureEvent},
    client::ClientManager,
    config::{Config, ConfigClient},
    connect::MousehopConnection,
    crypto,
    discovery::{Discovery, DiscoveryEvent, PrimaryCache},
    dns::{DnsEvent, DnsResolver},
    emulation::{Emulation, EmulationEvent},
    latency::{LatencyProber, ProbeResult},
    listen::{ListenerCreationError, MousehopListener},
    network::current_network_id,
};
use futures::StreamExt;
use input_capture::clipboard::{ClipboardMonitor, SuppressionList};
use input_capture::frontmost_app;
use input_event::{ClipboardEvent, Event as InputEvent};
use local_channel::mpsc::{Receiver, channel};
use log;
use mousehop_ipc::{
    AppIdent, AsyncFrontendListener, ClientHandle, ConnectionMode, FrontendEvent, FrontendRequest,
    HostKind, IncomingPeerConfig, IpcError, IpcListenerCreationError, Position, Status,
};
use mousehop_proto::ProtoEvent;
use std::{
    collections::{HashMap, HashSet, VecDeque},
    hash::{DefaultHasher, Hash, Hasher},
    io,
    net::{IpAddr, SocketAddr},
    sync::{Arc, RwLock},
    time::{Duration, Instant},
};
use thiserror::Error;
use tokio::{process::Command, signal, sync::Notify};

#[derive(Debug, Error)]
pub enum ServiceError {
    #[error(transparent)]
    IpcListen(#[from] IpcListenerCreationError),
    #[error(transparent)]
    Io(#[from] io::Error),
    #[error(transparent)]
    ListenError(#[from] ListenerCreationError),
    #[error("failed to load certificate: `{0}`")]
    Certificate(#[from] crypto::Error),
}

pub struct Service {
    /// configuration
    config: Config,
    /// input capture
    capture: Capture,
    /// input emulation
    emulation: Emulation,
    /// dns resolver
    resolver: DnsResolver,
    /// background per-address latency prober. Feeds the GUI's
    /// per-candidate RTT display so the user can compare interfaces on
    /// a multi-homed peer before locking one.
    latency: LatencyProber,
    /// frontend listener
    frontend_listener: AsyncFrontendListener,
    /// authorized public key sha256 fingerprints
    authorized_keys: Arc<RwLock<HashMap<String, IncomingPeerConfig>>>,
    /// Shared mDNS browse cache. Used at DTLS-accept time to
    /// reverse-lookup the connecting peer's IP back to the system
    /// hostname they advertise via Bonjour, so the GUI can show a
    /// human-readable identity in the Incoming Connections list.
    primary_cache: PrimaryCache,
    /// (outgoing) client information
    client_manager: ClientManager,
    /// current port
    port: u16,
    /// the public key fingerprint for (D)TLS
    public_key_fingerprint: String,
    /// notify for pending frontend events
    frontend_event_pending: Notify,
    /// frontend events queued for sending
    pending_frontend_events: VecDeque<FrontendEvent>,
    /// status of input capture (enabled / disabled)
    capture_status: Status,
    /// status of input emulation (enabled / disabled)
    emulation_status: Status,
    /// keep track of registered connections to avoid duplicate barriers
    incoming_conns: HashSet<SocketAddr>,
    /// map from capture handle to connection info
    incoming_conn_info: HashMap<ClientHandle, Incoming>,
    next_trigger_handle: u64,
    /// mDNS-SD service registration + browse. Advertises our primary
    /// interface IP for peer dialers to bias toward; populates
    /// shared `PrimaryCache` (read by `MousehopConnection`) from
    /// peer announcements.
    discovery: Discovery,
    /// Stream of resolved peer advertisements from [`Discovery`]'s
    /// browse task — folded into client candidate sets + interface
    /// labels in the main loop.
    discovery_event_rx: Receiver<DiscoveryEvent>,
    /// Opaque fingerprint of the network the machine is currently on
    /// (default-gateway MAC, or subnet fallback). Drives per-network
    /// lock resolution; recomputed periodically and on interface
    /// changes. `None` when undeterminable (locks then don't apply).
    current_network: Option<String>,
    /// Outgoing connection handle to fan clipboard frames out from
    /// the capture / forwarding paths. Same handle Capture owns;
    /// cloned in `Service::new` so Service can call `send` directly
    /// without routing through the capture session loop.
    conn: MousehopConnection,
    /// Cross-platform clipboard poller. `None` when the platform
    /// clipboard couldn't be opened (headless CI, Wayland session
    /// without compositor support). Service drains it in the main
    /// loop and fans the resulting events out to peers whose
    /// `clipboard_send` is true.
    clipboard_monitor: Option<ClipboardMonitor>,
    /// Recent forwards keyed on `(originator_fingerprint, hash)`.
    /// Used to break N-peer rebroadcast cycles: when this device
    /// receives a forwarded clipboard frame and would re-fan to
    /// other peers, the entry under (origin, content_hash) blocks
    /// the duplicate. Pruned lazily — entries older than
    /// `RECENT_FORWARD_TTL` are dropped on each clipboard event.
    recent_forwarded: HashMap<(String, u64), Instant>,
    /// Shared with [`ClipboardMonitor`]; mutations to the inner
    /// `HashSet` take effect on the next clipboard poll without
    /// rebuilding the monitor.
    clipboard_suppression: SuppressionList,
    /// Per-client count of consecutive probe observations where a
    /// substantially-faster candidate beat the active path, gating the
    /// `Fastest`-mode auto-upgrade so noise can't trigger a switch.
    fastest_streak: HashMap<ClientHandle, u8>,
}

/// Consecutive probe observations a faster candidate must win before
/// `Fastest` mode actually switches to it. With the ~5 s probe cadence
/// this is ~15 s of sustained advantage — long enough that a transient
/// latency dip never causes a flap.
const FASTEST_UPGRADE_STREAK: u8 = 3;

const RECENT_FORWARD_TTL: Duration = Duration::from_secs(1);

fn clipboard_hash(content: &str) -> u64 {
    let mut hasher = DefaultHasher::new();
    content.hash(&mut hasher);
    hasher.finish()
}

#[derive(Debug)]
struct Incoming {
    fingerprint: String,
    addr: SocketAddr,
    pos: Position,
}

impl Service {
    pub async fn new(config: Config) -> Result<Self, ServiceError> {
        let client_manager = ClientManager::default();
        for client in config.clients() {
            client_manager.add_with_config(client);
        }

        // load certificate
        let cert = crypto::load_or_generate_key_and_cert(config.cert_path())?;
        let public_key_fingerprint = crypto::certificate_fingerprint(&cert);

        // create frontend communication adapter, exit if already running
        let frontend_listener = AsyncFrontendListener::new().await?;

        let authorized_keys = Arc::new(RwLock::new(config.authorized_fingerprints()));
        // listener + connection. The primary-IP cache is owned by
        // the dialer side so its references survive Discovery
        // toggles; Discovery writes peer hints into it as browse
        // events arrive.
        let listener =
            MousehopListener::new(config.port(), cert.clone(), authorized_keys.clone()).await?;
        let primary_cache: PrimaryCache = Default::default();
        let conn =
            MousehopConnection::new(cert.clone(), client_manager.clone(), primary_cache.clone());

        // input capture + emulation
        let capture_backend = config.capture_backend().map(|b| b.into());
        let conn_for_service = conn.sender_clone();
        let capture = Capture::new(
            capture_backend,
            conn,
            config.release_bind(),
            config.release_threshold_px(),
        );
        let emulation_backend = config.emulation_backend().map(|b| b.into());
        let emulation = Emulation::new(emulation_backend, listener);
        // Push the persisted authorized-peers table into the receive
        // pipeline so per-peer post-processing is applied from the
        // first incoming packet.
        emulation.set_incoming_peers(authorized_keys.read().expect("lock").clone());

        // create dns resolver
        let resolver = DnsResolver::new()?;

        // background latency prober — pulls candidate addresses from
        // the (shared) client manager itself, so no targets need to be
        // pushed in as clients/DNS change.
        let latency = LatencyProber::new(client_manager.clone());

        let port = config.port();
        let (discovery_event_tx, discovery_event_rx) = channel();
        let discovery = Discovery::new(
            port,
            config.mdns_discovery(),
            primary_cache.clone(),
            discovery_event_tx,
        );
        // Resolve the lock (if any) for the network we boot on, before
        // the first dial, so a locked client connects on the right
        // interface immediately.
        let current_network = current_network_id();
        for handle in client_manager.registered_clients() {
            client_manager.recompute_active_lock(handle, current_network.as_deref());
        }
        // ClipboardMonitor is best-effort: a headless CI environment
        // or a Wayland session without compositor support yields a
        // permanent error here. We log and proceed without clipboard
        // sync rather than tying daemon startup to clipboard
        // availability.
        let clipboard_suppression: SuppressionList = {
            let host = HostKind::current();
            let initial: HashSet<AppIdent> = config
                .clipboard_suppression()
                .host()
                .iter()
                .cloned()
                .map(|s| host.make_ident(s))
                .collect();
            Arc::new(std::sync::Mutex::new(initial))
        };
        let clipboard_monitor =
            match ClipboardMonitor::with_suppression(clipboard_suppression.clone()) {
                Ok(m) => Some(m),
                Err(e) => {
                    log::warn!("clipboard monitor unavailable: {e}; clipboard sync disabled");
                    None
                }
            };
        let service = Self {
            config,
            capture,
            emulation,
            frontend_listener,
            resolver,
            latency,
            authorized_keys,
            primary_cache,
            public_key_fingerprint,
            client_manager,
            frontend_event_pending: Default::default(),
            port,
            pending_frontend_events: Default::default(),
            capture_status: Default::default(),
            emulation_status: Default::default(),
            incoming_conn_info: Default::default(),
            incoming_conns: Default::default(),
            next_trigger_handle: 0,
            discovery,
            discovery_event_rx,
            current_network,
            conn: conn_for_service,
            clipboard_monitor,
            recent_forwarded: HashMap::new(),
            clipboard_suppression,
            fastest_streak: HashMap::new(),
        };
        Ok(service)
    }

    pub async fn run(&mut self) -> Result<(), ServiceError> {
        let active = self.client_manager.active_clients();
        for handle in active.iter() {
            // small hack: `activate_client()` checks, if the client
            // is already active in client_manager and does not create a
            // capture barrier in that case so we have to deactivate it first
            self.client_manager.deactivate_client(*handle);
        }

        for handle in active {
            self.activate_client(handle);
        }

        // Periodic refresh of the Discovery service registration so
        // its TXT record stays accurate when the OS-preferred
        // interface (default route) changes — e.g. user switches
        // off Wi-Fi and Mac falls back to Ethernet. Cheap: at most
        // one re-publish every 30s, and a no-op when the primary
        // hasn't moved. `Skip` so a long suspend doesn't backlog-
        // burst on resume.
        let mut discovery_refresh_tick = tokio::time::interval(Duration::from_secs(30));
        discovery_refresh_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        // skip the immediate-fire of the first tick — Discovery
        // already published once at startup
        discovery_refresh_tick.tick().await;

        loop {
            tokio::select! {
                request = self.frontend_listener.next() => self.handle_frontend_request(request).await,
                _ = self.frontend_event_pending.notified() => self.handle_frontend_pending().await,
                event = self.emulation.event() => self.handle_emulation_event(event).await,
                event = self.capture.event() => self.handle_capture_event(event),
                event = self.resolver.event() => self.handle_resolver_event(event),
                result = self.latency.event() => self.handle_probe_result(result).await,
                event = recv_discovery(&mut self.discovery_event_rx) => {
                    self.handle_discovery_event(event)
                }
                _ = self.config.changed() => self.handle_config_change(),
                _ = discovery_refresh_tick.tick() => self.on_discovery_refresh().await,
                event = recv_clipboard(&mut self.clipboard_monitor) => {
                    self.handle_local_clipboard_event(event).await;
                }
                r = signal::ctrl_c() => break r.expect("failed to wait for CTRL+C"),
            }
        }

        log::info!("terminating service ...");
        log::debug!("terminating capture ...");
        self.capture.terminate().await;
        log::debug!("terminating emulation ...");
        self.emulation.terminate().await;
        log::debug!("terminating dns resolver ...");
        self.resolver.terminate().await;
        log::debug!("terminating latency prober ...");
        self.latency.terminate().await;

        Ok(())
    }

    async fn handle_frontend_request(
        &mut self,
        request: Option<Result<FrontendRequest, IpcError>>,
    ) {
        let request = match request.expect("frontend listener closed") {
            Ok(r) => r,
            Err(e) => return log::error!("error receiving request: {e}"),
        };
        match request {
            FrontendRequest::Activate(handle, active) => {
                self.set_client_active(handle, active);
                self.save_config().await;
            }
            FrontendRequest::AuthorizeKey(desc, fp) => {
                self.add_authorized_key(desc, fp);
                self.save_config().await;
            }
            FrontendRequest::ChangePort(port) => self.change_port(port),
            FrontendRequest::Create => {
                self.add_client();
                self.save_config().await;
            }
            FrontendRequest::Delete(handle) => {
                self.remove_client(handle);
                self.save_config().await;
            }
            FrontendRequest::EnableCapture => self.capture.reenable(),
            FrontendRequest::EnableEmulation => self.emulation.reenable(),
            FrontendRequest::Enumerate() => self.enumerate(),
            FrontendRequest::UpdateFixIps(handle, fix_ips) => {
                self.update_fix_ips(handle, fix_ips);
                self.save_config().await;
            }
            FrontendRequest::UpdateHostname(handle, host) => {
                self.update_hostname(handle, host);
                self.save_config().await;
            }
            FrontendRequest::UpdatePort(handle, port) => {
                self.update_port(handle, port);
                self.save_config().await;
            }
            FrontendRequest::UpdatePosition(handle, pos) => {
                self.update_pos(handle, pos);
                self.save_config().await;
            }
            FrontendRequest::ResolveDns(handle) => self.resolve(handle),
            FrontendRequest::Sync => self.sync_frontend(),
            FrontendRequest::RemoveAuthorizedKey(key) => {
                self.remove_authorized_key(key);
                self.save_config().await;
            }
            FrontendRequest::UpdateEnterHook(handle, enter_hook) => {
                self.update_enter_hook(handle, enter_hook)
            }
            FrontendRequest::SaveConfiguration => self.save_config().await,
            FrontendRequest::SetReleaseThreshold(threshold) => {
                self.config.set_release_threshold_px(threshold);
                self.capture.set_release_threshold(threshold);
                self.notify_frontend(FrontendEvent::ReleaseThreshold(threshold));
                self.save_config().await;
            }
            FrontendRequest::SetReleaseBind(bind) => {
                self.config.set_release_bind(bind);
                let canonical = self.config.release_bind();
                self.capture.set_release_bind(canonical.clone());
                self.notify_frontend(FrontendEvent::ReleaseBind(canonical));
                self.save_config().await;
            }
            FrontendRequest::SetIncomingPeerNaturalScroll(fp, natural_scroll) => {
                self.set_incoming_peer_natural_scroll(fp, natural_scroll);
                self.save_config().await;
            }
            FrontendRequest::SetIncomingPeerSensitivity(fp, sensitivity) => {
                self.set_incoming_peer_sensitivity(fp, sensitivity);
                self.save_config().await;
            }
            FrontendRequest::SetMdnsDiscovery(enabled) => {
                self.config.set_mdns_discovery(enabled);
                self.discovery.set_enabled(enabled);
                self.notify_frontend(FrontendEvent::MdnsDiscovery(enabled));
                self.save_config().await;
            }
            FrontendRequest::SetClientClipboardSend(handle, enabled) => {
                if self.client_manager.set_clipboard_send(handle, enabled) {
                    self.broadcast_client(handle);
                    self.save_config().await;
                }
            }
            FrontendRequest::SetConnectionMode(handle, mode) => {
                // Resolve the live network first: the user is acting on a
                // connected machine, so a stale cached `None` (from a
                // transient blip) must not leave this network's lock in
                // place to silently re-activate later.
                self.refresh_network().await;
                let net = self.current_network.clone();
                if self.client_manager.set_mode(handle, mode, net.as_deref()) {
                    // The mode change may have cleared this network's
                    // lock; re-resolve and re-dial so the new policy
                    // takes effect immediately.
                    self.client_manager
                        .recompute_active_lock(handle, net.as_deref());
                    self.conn.reset_handle(handle).await;
                    self.broadcast_client(handle);
                    self.save_config().await;
                }
            }
            FrontendRequest::SetNetworkLock(handle, ip) => {
                // Resolve the live network first (see SetConnectionMode).
                self.refresh_network().await;
                let Some(net) = self.current_network.clone() else {
                    log::warn!("cannot lock client {handle}: current network unknown");
                    return;
                };
                if self.client_manager.set_network_lock(handle, &net, ip) {
                    self.client_manager
                        .recompute_active_lock(handle, Some(&net));
                    // Drop any live connection so the lock takes effect
                    // on the next event instead of waiting for the
                    // current (possibly wrong-interface) path to drop.
                    self.conn.reset_handle(handle).await;
                    self.broadcast_client(handle);
                    self.save_config().await;
                }
            }
            FrontendRequest::SetIncomingPeerClipboardReceive(fp, enabled) => {
                self.set_incoming_peer_clipboard_receive(fp, enabled);
                self.save_config().await;
            }
            FrontendRequest::AddSuppressedApp(value) => {
                self.add_suppressed_app(value);
                self.save_config().await;
            }
            FrontendRequest::RemoveSuppressedApp(value) => {
                self.remove_suppressed_app(value);
                self.save_config().await;
            }
            FrontendRequest::ListRunningApps => {
                let apps = frontmost_app::list_running_apps();
                self.notify_frontend(FrontendEvent::RunningApps(apps));
            }
        }
    }

    fn add_suppressed_app(&mut self, value: String) {
        let value = value.trim().to_owned();
        if value.is_empty() {
            return;
        }
        let mut suppression = self.config.clipboard_suppression();
        let host = suppression.host_mut();
        if !host.iter().any(|v| v.eq_ignore_ascii_case(&value)) {
            host.push(value);
        }
        self.commit_suppression(suppression);
    }

    fn remove_suppressed_app(&mut self, value: String) {
        let mut suppression = self.config.clipboard_suppression();
        suppression
            .host_mut()
            .retain(|v| !v.eq_ignore_ascii_case(&value));
        self.commit_suppression(suppression);
    }

    /// Persist the per-OS struct, refresh the runtime `HashSet`
    /// shared with [`ClipboardMonitor`], and push the host slot to
    /// the GUI. Centralized so add/remove can't drift apart.
    fn commit_suppression(&mut self, suppression: mousehop_ipc::ClipboardSuppression) {
        let host = HostKind::current();
        let host_list = suppression.host().clone();
        {
            let mut guard = self.clipboard_suppression.lock().expect("lock");
            guard.clear();
            for s in &host_list {
                guard.insert(host.make_ident(s.clone()));
            }
        }
        self.config.set_clipboard_suppression(suppression);
        self.notify_frontend(FrontendEvent::SuppressedAppsUpdated(host_list));
    }

    /// Refresh `last_addr` / `last_hostname` for the authorized-peer
    /// entry matching `fingerprint` whenever a DTLS connect lands.
    /// Hostname comes from a reverse-lookup against the mDNS
    /// `hostname → primary_ip` cache; falls through to addr-only
    /// if discovery isn't running on either end or the peer's
    /// announced primary differs from the IP it actually connected
    /// from. Persists the update so the GUI keeps a useful
    /// identification across restarts.
    async fn update_incoming_peer_address(&mut self, addr: SocketAddr, fingerprint: &str) {
        let ip = addr.ip().to_string();
        let hostname = self.lookup_hostname_for_ip(addr.ip());
        // Scope the write guard to this block: `save_config().await`
        // below must not run while the non-async RwLock is held, or
        // the lock would span a suspension point. The block yields
        // the cloned snapshot and drops the guard before the await.
        let snapshot = {
            let mut keys = self.authorized_keys.write().expect("lock");
            let Some(peer) = keys.get_mut(fingerprint) else {
                return; // unauthorized peer; nothing to update
            };
            let mut changed = peer.last_addr.as_deref() != Some(&ip);
            if changed {
                peer.last_addr = Some(ip);
            }
            if let Some(h) = hostname {
                if peer.last_hostname.as_deref() != Some(h.as_str()) {
                    peer.last_hostname = Some(h);
                    changed = true;
                }
            }
            if !changed {
                return;
            }
            keys.clone()
        };
        // No need to push to InputEmulation — last_addr/last_hostname
        // are display-only; per-pair scroll/sensitivity is unaffected.
        self.notify_frontend(FrontendEvent::AuthorizedUpdated(snapshot));
        self.save_config().await;
    }

    fn lookup_hostname_for_ip(&self, target: std::net::IpAddr) -> Option<String> {
        self.primary_cache
            .borrow()
            .iter()
            .find_map(|(host, ip)| (*ip == target).then(|| host.clone()))
    }

    fn set_incoming_peer_natural_scroll(&mut self, fingerprint: String, natural_scroll: bool) {
        if let Some(peer) = self
            .authorized_keys
            .write()
            .expect("lock")
            .get_mut(&fingerprint)
        {
            peer.natural_scroll = natural_scroll;
        }
        let keys = self.authorized_keys.read().expect("lock").clone();
        self.emulation.set_incoming_peers(keys.clone());
        self.notify_frontend(FrontendEvent::AuthorizedUpdated(keys));
    }

    fn set_incoming_peer_sensitivity(&mut self, fingerprint: String, sensitivity: f64) {
        if let Some(peer) = self
            .authorized_keys
            .write()
            .expect("lock")
            .get_mut(&fingerprint)
        {
            peer.mouse_sensitivity = sensitivity;
        }
        let keys = self.authorized_keys.read().expect("lock").clone();
        self.emulation.set_incoming_peers(keys.clone());
        self.notify_frontend(FrontendEvent::AuthorizedUpdated(keys));
    }

    fn set_incoming_peer_clipboard_receive(
        &mut self,
        fingerprint: String,
        clipboard_receive: bool,
    ) {
        if let Some(peer) = self
            .authorized_keys
            .write()
            .expect("lock")
            .get_mut(&fingerprint)
        {
            peer.clipboard_receive = clipboard_receive;
        }
        let keys = self.authorized_keys.read().expect("lock").clone();
        // Emulation needs to know so the receive-side gate matches
        // the new value immediately, not after a config-change cycle.
        self.emulation.set_incoming_peers(keys.clone());
        self.notify_frontend(FrontendEvent::AuthorizedUpdated(keys));
    }

    async fn save_config(&mut self) {
        let clients = self.client_manager.clients();
        let clients = clients
            .into_iter()
            .map(|(c, s)| ConfigClient {
                ips: HashSet::from_iter(c.fix_ips),
                hostname: c.hostname,
                port: c.port,
                pos: c.pos,
                active: s.active,
                enter_hook: c.cmd,
                mode: c.mode,
                network_locks: c.network_locks,
                clipboard_send: c.clipboard_send,
            })
            .collect();
        self.config.set_clients(clients);
        let authorized_keys = self.authorized_keys.read().expect("lock").clone();
        self.config.set_authorized_keys(authorized_keys);
        if let Err(e) = self.config.write_back().await {
            log::warn!("failed to write config: {e}");
        }
    }

    fn handle_config_change(&mut self) {
        for h in self.client_manager.registered_clients() {
            self.remove_client(h);
        }
        for c in self.config.clients() {
            let handle = self.client_manager.add_with_config(c);
            log::info!("added client {handle}");
            let (c, s) = self.client_manager.get_state(handle).unwrap();
            if s.active {
                self.client_manager.deactivate_client(handle);
                self.activate_client(handle);
            }
            self.notify_frontend(FrontendEvent::Created(handle, c, s));
        }
        let release_bind = self.config.release_bind();
        self.capture.set_release_bind(release_bind);
        let release_threshold = self.config.release_threshold_px();
        self.capture.set_release_threshold(release_threshold);
        self.notify_frontend(FrontendEvent::ReleaseThreshold(release_threshold));
        let authorized_keys = self.config.authorized_fingerprints();
        self.authorized_keys
            .write()
            .unwrap()
            .clone_from(&authorized_keys);
        self.emulation.set_incoming_peers(authorized_keys);
        self.sync_frontend();
    }

    async fn handle_frontend_pending(&mut self) {
        while let Some(event) = self.pending_frontend_events.pop_front() {
            self.frontend_listener.broadcast(event).await;
        }
    }

    async fn handle_emulation_event(&mut self, event: EmulationEvent) {
        match event {
            EmulationEvent::ConnectionAttempt { fingerprint } => {
                self.notify_frontend(FrontendEvent::ConnectionAttempt { fingerprint });
            }
            EmulationEvent::Entered {
                addr,
                pos,
                fingerprint,
            } => {
                // check if already registered
                if !self.incoming_conns.contains(&addr) {
                    self.add_incoming(addr, pos, fingerprint.clone());
                    self.notify_frontend(FrontendEvent::DeviceEntered {
                        fingerprint,
                        addr,
                        pos,
                    });
                } else {
                    self.update_incoming(addr, pos, fingerprint);
                }
            }
            EmulationEvent::Disconnected { addr } => {
                if let Some(addr) = self.remove_incoming(addr) {
                    self.notify_frontend(FrontendEvent::IncomingDisconnected(addr));
                }
            }
            EmulationEvent::PortChanged(port) => match port {
                Ok(port) => {
                    self.port = port;
                    self.discovery.set_port(port);
                    self.notify_frontend(FrontendEvent::PortChanged(port, None));
                }
                Err(e) => self
                    .notify_frontend(FrontendEvent::PortChanged(self.port, Some(format!("{e}")))),
            },
            EmulationEvent::EmulationDisabled => {
                self.emulation_status = Status::Disabled;
                self.notify_frontend(FrontendEvent::EmulationStatus(self.emulation_status));
            }
            EmulationEvent::EmulationEnabled => {
                self.emulation_status = Status::Enabled;
                self.notify_frontend(FrontendEvent::EmulationStatus(self.emulation_status));
            }
            EmulationEvent::ReleaseNotify => self.capture.release_for_handover(),
            EmulationEvent::Connected { addr, fingerprint } => {
                self.update_incoming_peer_address(addr, &fingerprint).await;
                self.notify_frontend(FrontendEvent::DeviceConnected { addr, fingerprint });
            }
            EmulationEvent::PeerHello { addr, commit } => {
                // Map the peer's source addr back to its client handle
                // and stamp the commit. Skip if we don't have an
                // outgoing client configured for this peer (incoming-
                // only setup) — there's nowhere to display the version
                // in that case anyway.
                if let Some(handle) = self.client_manager.get_client(addr) {
                    self.client_manager.set_peer_commit(handle, Some(commit));
                    self.broadcast_client(handle);
                }
            }
            EmulationEvent::ClipboardReceived {
                addr,
                from_fingerprint,
                content,
            } => {
                self.handle_clipboard_received(addr, from_fingerprint, content)
                    .await;
            }
        }
    }

    /// Local clipboard change picked up by the polling
    /// [`ClipboardMonitor`]. Stamp the originator fingerprint on the
    /// wire frame and fan out to every active outgoing client whose
    /// `clipboard_send` is true. Records `(self_fp, hash)` in
    /// `recent_forwarded` so a later forwarded copy of the same
    /// content (re-arriving via another peer in an N-peer ring)
    /// won't be redundantly re-broadcast.
    async fn handle_local_clipboard_event(&mut self, event: Option<input_capture::CaptureEvent>) {
        let Some(event) = event else {
            return;
        };
        let input_capture::CaptureEvent::Input(InputEvent::Clipboard(ClipboardEvent::Text(
            content,
        ))) = event
        else {
            return;
        };
        let targets = self.client_manager.clipboard_send_targets();
        if targets.is_empty() {
            log::trace!(
                "clipboard captured locally ({} bytes) but no peer has clipboard_send=true; skipping fan-out",
                content.len()
            );
            return;
        }
        let from_fingerprint = self.public_key_fingerprint.clone();
        let hash = clipboard_hash(&content);
        self.prune_recent_forwarded();
        self.recent_forwarded
            .insert((from_fingerprint.clone(), hash), Instant::now());
        log::info!(
            "broadcasting local clipboard ({} bytes) to {} peer(s)",
            content.len(),
            targets.len()
        );
        for handle in targets {
            let event = ProtoEvent::Clipboard {
                from_fingerprint: from_fingerprint.clone(),
                content: content.clone(),
            };
            if let Err(e) = self.conn.send(event, handle).await {
                log::debug!("clipboard send to client {handle} failed: {e}");
            }
        }
    }

    /// Forwarded clipboard frame just landed via the listen side
    /// (the local clipboard has already been updated by
    /// `emulation::ListenTask`). Refresh the
    /// [`ClipboardMonitor`]'s last-known content so the next 500ms
    /// poll doesn't see this as a fresh local change and bounce it
    /// back, then forward to other peers honoring the recent-
    /// forwarded gate.
    async fn handle_clipboard_received(
        &mut self,
        from_addr: SocketAddr,
        from_fingerprint: String,
        content: String,
    ) {
        if let Some(monitor) = self.clipboard_monitor.as_ref() {
            monitor.update_last_content(content.clone());
        }
        let hash = clipboard_hash(&content);
        self.prune_recent_forwarded();
        let key = (from_fingerprint.clone(), hash);
        if self.recent_forwarded.contains_key(&key) {
            log::debug!(
                "skipping clipboard re-fan-out: already forwarded ({}, {} bytes) within {}ms",
                &from_fingerprint[..from_fingerprint.len().min(8)],
                content.len(),
                RECENT_FORWARD_TTL.as_millis()
            );
            return;
        }
        let targets = self.client_manager.clipboard_send_targets();
        let forward_targets: Vec<ClientHandle> = targets
            .into_iter()
            .filter(|h| {
                // Skip the client we just received from. Identified
                // by IP rather than full SocketAddr so a peer's
                // ephemeral source port (which differs between its
                // outgoing and our cached active_addr) doesn't
                // accidentally include them.
                self.client_manager
                    .active_addr(*h)
                    .map(|a| a.ip() != from_addr.ip())
                    .unwrap_or(true)
            })
            .collect();
        if forward_targets.is_empty() {
            return;
        }
        self.recent_forwarded.insert(key, Instant::now());
        log::info!(
            "forwarding clipboard ({} bytes, originator {}) to {} peer(s)",
            content.len(),
            &from_fingerprint[..from_fingerprint.len().min(8)],
            forward_targets.len()
        );
        for handle in forward_targets {
            let event = ProtoEvent::Clipboard {
                from_fingerprint: from_fingerprint.clone(),
                content: content.clone(),
            };
            if let Err(e) = self.conn.send(event, handle).await {
                log::debug!("clipboard forward to client {handle} failed: {e}");
            }
        }
    }

    fn prune_recent_forwarded(&mut self) {
        self.recent_forwarded
            .retain(|_, ts| ts.elapsed() < RECENT_FORWARD_TTL);
    }

    fn handle_capture_event(&mut self, event: ICaptureEvent) {
        match event {
            ICaptureEvent::CaptureBegin(handle) => {
                // we entered the capture zone for an incoming connection
                // => notify it that its capture should be released
                if let Some(incoming) = self.incoming_conn_info.get(&handle) {
                    self.emulation.send_leave_event(incoming.addr);
                }
            }
            ICaptureEvent::CaptureDisabled => {
                self.capture_status = Status::Disabled;
                self.notify_frontend(FrontendEvent::CaptureStatus(self.capture_status));
            }
            ICaptureEvent::CaptureEnabled => {
                self.capture_status = Status::Enabled;
                self.notify_frontend(FrontendEvent::CaptureStatus(self.capture_status));
            }
            ICaptureEvent::ClientEntered(handle) => {
                log::info!("entering client {handle} ...");
                self.spawn_hook_command(handle);
            }
            ICaptureEvent::PeerCommitUpdated(handle) => {
                self.broadcast_client(handle);
            }
        }
    }

    fn handle_resolver_event(&mut self, event: DnsEvent) {
        let handle = match event {
            DnsEvent::Resolving(handle) => {
                self.client_manager.set_resolving(handle, true);
                handle
            }
            DnsEvent::Resolved(handle, hostname, ips) => {
                self.client_manager.set_resolving(handle, false);
                if let Err(e) = &ips {
                    log::warn!("could not resolve {hostname}: {e}");
                }
                let ips = ips.unwrap_or_default();
                self.client_manager.set_dns_ips(handle, ips);
                handle
            }
        };
        self.broadcast_client(handle);
    }

    fn resolve(&self, handle: ClientHandle) {
        if let Some(hostname) = self.client_manager.get_hostname(handle) {
            self.resolver.resolve(handle, hostname);
        }
    }

    /// Apply a background latency sample and re-broadcast the client
    /// only when the stored value actually moved. The reading is
    /// quantized to 100 µs buckets so ordinary sub-microsecond jitter
    /// on a stable link doesn't churn the frontend every probe cycle.
    /// Also feeds the `Fastest`-mode convergence check.
    async fn handle_probe_result(&mut self, result: ProbeResult) {
        let ProbeResult {
            handle,
            ip,
            rtt_micros,
        } = result;
        let bucketed = rtt_micros.map(|us| us - (us % 100));
        if self.client_manager.set_latency(handle, ip, bucketed) {
            self.broadcast_client(handle);
        }
        self.maybe_upgrade_fastest(handle).await;
    }

    /// In `Fastest` mode (and only when not locked), nudge the live
    /// connection toward the lowest-latency candidate — but only once a
    /// candidate has been *sustainedly and substantially* faster than
    /// the active path (see [`ClientManager::fastest_upgrade_candidate`]
    /// and [`FASTEST_UPGRADE_STREAK`]), so probe noise can't cause
    /// flapping. Tearing down the connection makes the next send
    /// re-dial, which picks the new fastest.
    async fn maybe_upgrade_fastest(&mut self, handle: ClientHandle) {
        let eligible = self.client_manager.get_active_lock(handle).is_none()
            && self.client_manager.get_mode(handle) == ConnectionMode::Fastest;
        if !eligible {
            self.fastest_streak.remove(&handle);
            return;
        }
        if self
            .client_manager
            .fastest_upgrade_candidate(handle)
            .is_some()
        {
            let streak = self.fastest_streak.entry(handle).or_insert(0);
            *streak += 1;
            if *streak >= FASTEST_UPGRADE_STREAK {
                self.fastest_streak.remove(&handle);
                log::info!("client {handle}: switching to a sustainedly-faster interface");
                self.conn.reset_handle(handle).await;
            }
        } else {
            self.fastest_streak.remove(&handle);
        }
    }

    /// Fold a resolved peer advertisement into every client whose
    /// configured hostname matches the mDNS instance: add the
    /// discovered addresses to the candidate set and record per-address
    /// interface kinds. Re-broadcasts only the clients that changed.
    fn handle_discovery_event(&mut self, event: DiscoveryEvent) {
        let DiscoveryEvent::Resolved {
            instance,
            addresses,
            interfaces,
        } = event;
        for handle in self.client_manager.registered_clients() {
            let matches = self
                .client_manager
                .get_hostname(handle)
                .is_some_and(|h| crate::discovery::normalize_mdns_name(&h) == instance);
            if !matches {
                continue;
            }
            // Discovered ips must land before interface labels so the
            // label filter (which keeps only current candidates) sees
            // the freshly-added addresses.
            let discovered_changed = self
                .client_manager
                .set_discovered_ips(handle, addresses.clone());
            let interfaces_changed = self
                .client_manager
                .set_interfaces(handle, interfaces.clone());
            if discovered_changed {
                // The candidate set moved, so any in-progress
                // Fastest-upgrade streak no longer describes the
                // current candidates — start it fresh.
                self.fastest_streak.remove(&handle);
            }
            if discovered_changed || interfaces_changed {
                self.broadcast_client(handle);
            }
        }
    }

    /// Periodic discovery housekeeping: re-publish our advertisement
    /// (primary interface may have changed) and re-evaluate which
    /// network we're on for per-network lock resolution.
    async fn on_discovery_refresh(&mut self) {
        self.discovery.refresh();
        self.refresh_network().await;
    }

    /// Recompute the current-network fingerprint. On change, re-resolve
    /// every client's active lock and re-dial those whose lock moved so
    /// the new network's pin (or its absence) takes effect at once.
    async fn refresh_network(&mut self) {
        let net = current_network_id();
        if net == self.current_network {
            return;
        }
        log::info!("network changed: {:?} -> {:?}", self.current_network, net);
        self.current_network = net;
        for handle in self.client_manager.registered_clients() {
            if self
                .client_manager
                .recompute_active_lock(handle, self.current_network.as_deref())
            {
                self.conn.reset_handle(handle).await;
                self.broadcast_client(handle);
            }
        }
    }

    fn sync_frontend(&mut self) {
        self.enumerate();
        self.notify_frontend(FrontendEvent::EmulationStatus(self.emulation_status));
        self.notify_frontend(FrontendEvent::CaptureStatus(self.capture_status));
        self.notify_frontend(FrontendEvent::PortChanged(self.port, None));
        self.notify_frontend(FrontendEvent::PublicKeyFingerprint(
            self.public_key_fingerprint.clone(),
        ));
        self.notify_frontend(FrontendEvent::ReleaseThreshold(
            self.config.release_threshold_px(),
        ));
        self.notify_frontend(FrontendEvent::ReleaseBind(self.config.release_bind()));
        self.notify_frontend(FrontendEvent::MdnsDiscovery(self.config.mdns_discovery()));
        let keys = self.authorized_keys.read().expect("lock").clone();
        self.notify_frontend(FrontendEvent::AuthorizedUpdated(keys));
        let host_list = self.config.clipboard_suppression().host().clone();
        self.notify_frontend(FrontendEvent::SuppressedAppsUpdated(host_list));
    }

    const ENTER_HANDLE_BEGIN: u64 = u64::MAX / 2 + 1;

    fn add_incoming(&mut self, addr: SocketAddr, pos: Position, fingerprint: String) {
        let handle = Self::ENTER_HANDLE_BEGIN + self.next_trigger_handle;
        self.next_trigger_handle += 1;
        self.capture.create(handle, pos, CaptureType::EnterOnly);
        self.incoming_conns.insert(addr);
        self.incoming_conn_info.insert(
            handle,
            Incoming {
                fingerprint,
                addr,
                pos,
            },
        );
    }

    fn update_incoming(&mut self, addr: SocketAddr, pos: Position, fingerprint: String) {
        let incoming = self
            .incoming_conn_info
            .iter_mut()
            .find(|(_, i)| i.addr == addr)
            .map(|(_, i)| i)
            .expect("no such client");
        let mut changed = false;
        if incoming.fingerprint != fingerprint {
            incoming.fingerprint = fingerprint.clone();
            changed = true;
        }
        if incoming.pos != pos {
            incoming.pos = pos;
            changed = true;
        }
        if changed {
            self.remove_incoming(addr);
            self.add_incoming(addr, pos, fingerprint.clone());
            self.notify_frontend(FrontendEvent::IncomingDisconnected(addr));
            self.notify_frontend(FrontendEvent::DeviceEntered {
                fingerprint,
                addr,
                pos,
            });
        }
    }

    fn remove_incoming(&mut self, addr: SocketAddr) -> Option<SocketAddr> {
        let handle = self
            .incoming_conn_info
            .iter()
            .find(|(_, incoming)| incoming.addr == addr)
            .map(|(k, _)| *k)?;
        self.capture.destroy(handle);
        self.incoming_conns.remove(&addr);
        self.incoming_conn_info
            .remove(&handle)
            .map(|incoming| incoming.addr)
    }

    fn notify_frontend(&mut self, event: FrontendEvent) {
        self.pending_frontend_events.push_back(event);
        self.frontend_event_pending.notify_one();
    }

    fn add_authorized_key(&mut self, desc: String, fp: String) {
        // New authorizations land with default post-processing; the
        // user can tune natural-scroll / sensitivity from the
        // expanded row in the Incoming Connections list.
        let entry = IncomingPeerConfig {
            description: desc,
            ..IncomingPeerConfig::default()
        };
        self.authorized_keys
            .write()
            .expect("lock")
            .insert(fp, entry);
        let keys = self.authorized_keys.read().expect("lock").clone();
        self.emulation.set_incoming_peers(keys.clone());
        self.notify_frontend(FrontendEvent::AuthorizedUpdated(keys));
    }

    fn remove_authorized_key(&mut self, fp: String) {
        self.authorized_keys.write().expect("lock").remove(&fp);
        let keys = self.authorized_keys.read().expect("lock").clone();
        self.emulation.set_incoming_peers(keys.clone());
        self.notify_frontend(FrontendEvent::AuthorizedUpdated(keys));
    }

    fn enumerate(&mut self) {
        let clients = self.client_manager.get_client_states();
        self.notify_frontend(FrontendEvent::Enumerate(clients));
    }

    fn add_client(&mut self) {
        let handle = self.client_manager.add_client();
        log::info!("added client {handle}");
        let (c, s) = self.client_manager.get_state(handle).unwrap();
        self.notify_frontend(FrontendEvent::Created(handle, c, s));
    }

    fn set_client_active(&mut self, handle: ClientHandle, active: bool) {
        if active {
            self.activate_client(handle);
        } else {
            self.deactivate_client(handle);
        }
    }

    fn deactivate_client(&mut self, handle: ClientHandle) {
        log::debug!("deactivating client {handle}");
        if self.client_manager.deactivate_client(handle) {
            self.capture.destroy(handle);
            self.broadcast_client(handle);
            log::info!("deactivated client {handle}");
        }
    }

    fn activate_client(&mut self, handle: ClientHandle) {
        log::debug!("activating client {handle}");

        /* resolve dns on activate */
        self.resolve(handle);

        /* deactivate potential other client at this position */
        let Some(pos) = self.client_manager.get_pos(handle) else {
            return;
        };

        if let Some(other) = self.client_manager.client_at(pos) {
            if other != handle {
                self.deactivate_client(other);
            }
        }

        /* activate the client */
        if self.client_manager.activate_client(handle) {
            /* notify capture and frontends */
            self.capture.create(handle, pos, CaptureType::Default);
            self.broadcast_client(handle);
            log::info!("activated client {handle} ({pos})");
        }
    }

    fn change_port(&mut self, port: u16) {
        if self.port != port {
            self.emulation.request_port_change(port);
        } else {
            self.notify_frontend(FrontendEvent::PortChanged(self.port, None));
        }
    }

    fn remove_client(&mut self, handle: ClientHandle) {
        if self
            .client_manager
            .remove_client(handle)
            .map(|(_, s)| s.active)
            .unwrap_or(false)
        {
            self.capture.destroy(handle);
        }
        self.notify_frontend(FrontendEvent::Deleted(handle));
    }

    fn update_fix_ips(&mut self, handle: ClientHandle, fix_ips: Vec<IpAddr>) {
        self.client_manager.set_fix_ips(handle, fix_ips);
        self.broadcast_client(handle);
    }

    fn update_hostname(&mut self, handle: ClientHandle, hostname: Option<String>) {
        log::info!("hostname changed: {hostname:?}");
        if self.client_manager.set_hostname(handle, hostname.clone()) {
            self.resolve(handle);
        }
        self.broadcast_client(handle);
    }

    fn update_port(&mut self, handle: ClientHandle, port: u16) {
        self.client_manager.set_port(handle, port);
        self.broadcast_client(handle);
    }

    fn update_pos(&mut self, handle: ClientHandle, pos: Position) {
        // update state in event input emulator & input capture
        if self.client_manager.set_pos(handle, pos) {
            self.deactivate_client(handle);
            self.activate_client(handle);
        }
        self.broadcast_client(handle);
    }

    fn update_enter_hook(&mut self, handle: ClientHandle, enter_hook: Option<String>) {
        self.client_manager.set_enter_hook(handle, enter_hook);
        self.broadcast_client(handle);
    }

    fn broadcast_client(&mut self, handle: ClientHandle) {
        let event = self
            .client_manager
            .get_state(handle)
            .map(|(c, s)| FrontendEvent::State(handle, c, s))
            .unwrap_or(FrontendEvent::NoSuchClient(handle));
        self.notify_frontend(event);
    }

    fn spawn_hook_command(&self, handle: ClientHandle) {
        let Some(cmd) = self.client_manager.get_enter_cmd(handle) else {
            return;
        };
        tokio::task::spawn_local(async move {
            log::info!("spawning command!");
            let mut child = match Command::new("sh").arg("-c").arg(cmd.as_str()).spawn() {
                Ok(c) => c,
                Err(e) => {
                    log::warn!("could not execute cmd: {e}");
                    return;
                }
            };
            match child.wait().await {
                Ok(s) => {
                    if s.success() {
                        log::info!("{cmd} exited successfully");
                    } else {
                        log::warn!("{cmd} exited with {s}");
                    }
                }
                Err(e) => log::warn!("{cmd}: {e}"),
            }
        });
    }
}

/// `tokio::select!` arm helper for the optional [`ClipboardMonitor`].
/// Resolves to `Some(event)` when the monitor surfaces a change and
/// to a never-completing future when no monitor is alive — keeping
/// the surrounding `select!` from busy-spinning when clipboard sync
/// is unavailable on the host.
async fn recv_clipboard(
    monitor: &mut Option<ClipboardMonitor>,
) -> Option<input_capture::CaptureEvent> {
    match monitor.as_mut() {
        Some(m) => m.recv().await,
        None => std::future::pending().await,
    }
}

/// `tokio::select!` arm helper for the discovery event stream. Pends
/// forever once the channel is closed so the `select!` doesn't
/// busy-spin during shutdown.
async fn recv_discovery(rx: &mut Receiver<DiscoveryEvent>) -> DiscoveryEvent {
    match rx.recv().await {
        Some(event) => event,
        None => std::future::pending().await,
    }
}

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

    #[test]
    fn clipboard_hash_is_deterministic_within_run() {
        let h1 = clipboard_hash("hello, world");
        let h2 = clipboard_hash("hello, world");
        assert_eq!(h1, h2);
    }

    #[test]
    fn clipboard_hash_distinguishes_different_inputs() {
        assert_ne!(clipboard_hash("foo"), clipboard_hash("bar"));
        assert_ne!(clipboard_hash(""), clipboard_hash("\0"));
    }

    #[test]
    fn recent_forwarded_prune_evicts_expired_entries() {
        // Mirrors `Service::prune_recent_forwarded` so the eviction
        // contract is documented as code rather than implicit in
        // `HashMap::retain`.
        let mut map: HashMap<(String, u64), Instant> = HashMap::new();
        let now = Instant::now();
        let stale = now
            .checked_sub(Duration::from_secs(2))
            .expect("clock far enough from epoch for the test to subtract 2s");
        map.insert(("fp_a".into(), 1), stale);
        map.insert(("fp_b".into(), 2), now);
        map.retain(|_, ts| ts.elapsed() < RECENT_FORWARD_TTL);
        assert!(!map.contains_key(&("fp_a".to_string(), 1)));
        assert!(map.contains_key(&("fp_b".to_string(), 2)));
    }
}
