mousehop 0.11.0

use std::{
    cell::{Cell, RefCell},
    rc::Rc,
    time::{Duration, Instant},
};

use futures::StreamExt;
use input_capture::{
    CaptureError, CaptureEvent, CaptureHandle, InputCapture, InputCaptureError, Position,
};
use input_event::{Event, KeyboardEvent, scancode};
use local_channel::mpsc::{Receiver, Sender, channel};
use mousehop_proto::ProtoEvent;
use tokio::task::{JoinHandle, spawn_local};
use tokio_util::sync::CancellationToken;

use crate::connect::MousehopConnection;

pub(crate) struct Capture {
    cancellation_token: CancellationToken,
    request_tx: Sender<CaptureRequest>,
    task: JoinHandle<()>,
    event_rx: Receiver<ICaptureEvent>,
}

pub(crate) enum ICaptureEvent {
    /// a client was entered
    CaptureBegin(CaptureHandle),
    /// capture disabled
    CaptureDisabled,
    /// capture disabled
    CaptureEnabled,
    /// A (new) client was entered.
    /// In contrast to [`ICaptureEvent::CaptureBegin`] this
    /// event is only triggered when the capture was
    /// explicitly released in the meantime by
    /// either the remote client leaving its device region,
    /// a new device entering the screen or the release bind.
    ClientEntered(u64),
    /// The connect-side received the peer's `Hello` echo and
    /// updated `client_manager.peer_commit` for `handle`. Forwarded
    /// upward so Service can broadcast `FrontendEvent::State` and
    /// the GUI's per-row version-status indicator picks up the new
    /// value. The listen-side path independently emits
    /// [`crate::emulation::EmulationEvent::PeerHello`], but it
    /// races with `active_addr` population — when an incoming
    /// `Hello` arrives before our outbound dial completes, the
    /// listen path's `get_client(addr)` returns `None` and the
    /// commit silently goes unsurfaced. The connect-side path
    /// fires later but reliably, so it carries the broadcast as a
    /// belt-and-suspenders fallback.
    PeerCommitUpdated(CaptureHandle),
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum CaptureType {
    /// a normal input capture
    Default,
    /// A capture only interested in [`CaptureEvent::Begin`] events.
    /// The capture is released immediately, if there is no
    /// Default capture at the same position.
    EnterOnly,
}

#[derive(Clone, Debug)]
enum CaptureRequest {
    /// release because the remote peer is taking over (they sent
    /// Enter+CursorPos). Skips the host-side warp so the peer's
    /// proportional CursorPos warp doesn't get clobbered by a
    /// racing local warp computed from stale virtual_cursor state.
    ReleaseForHandover,
    /// add a capture client
    Create(CaptureHandle, Position, CaptureType),
    /// destory a capture client
    Destroy(CaptureHandle),
    /// reenable input capture
    Reenable,
    /// set release bind
    SetReleaseBind(Vec<scancode::Linux>),
    /// set the auto-release pixel threshold (macOS only). 0 disables.
    SetReleaseThreshold(u32),
}

impl Capture {
    pub(crate) fn new(
        backend: Option<input_capture::Backend>,
        conn: MousehopConnection,
        release_bind: Vec<scancode::Linux>,
        release_threshold_px: u32,
    ) -> Self {
        let (request_tx, request_rx) = channel();
        let (event_tx, event_rx) = channel();
        let cancellation_token = CancellationToken::new();
        let capture_task = CaptureTask {
            active_client: None,
            backend,
            cancellation_token: cancellation_token.clone(),
            captures: Default::default(),
            conn,
            event_tx,
            request_rx,
            release_bind: Rc::new(RefCell::new(release_bind)),
            release_threshold_px: Rc::new(RefCell::new(release_threshold_px)),
            state: Default::default(),
        };
        let task = spawn_local(capture_task.run());
        Self {
            cancellation_token,
            request_tx,
            task,
            event_rx,
        }
    }

    pub(crate) fn reenable(&self) {
        self.request_tx
            .send(CaptureRequest::Reenable)
            .expect("channel closed");
    }

    pub(crate) async fn terminate(&mut self) {
        self.cancellation_token.cancel();
        log::debug!("terminating capture");
        if let Err(e) = (&mut self.task).await {
            log::warn!("{e}");
        }
    }

    pub(crate) fn create(
        &self,
        handle: CaptureHandle,
        pos: mousehop_ipc::Position,
        capture_type: CaptureType,
    ) {
        let pos = to_capture_pos(pos);
        self.request_tx
            .send(CaptureRequest::Create(handle, pos, capture_type))
            .expect("channel closed");
    }

    pub(crate) fn destroy(&self, handle: CaptureHandle) {
        self.request_tx
            .send(CaptureRequest::Destroy(handle))
            .expect("channel closed");
    }

    pub(crate) fn release_for_handover(&self) {
        self.request_tx
            .send(CaptureRequest::ReleaseForHandover)
            .expect("channel closed");
    }

    pub(crate) async fn event(&mut self) -> ICaptureEvent {
        self.event_rx.recv().await.expect("channel closed")
    }

    pub(crate) fn set_release_bind(&mut self, bind: Vec<scancode::Linux>) {
        let _ = self.request_tx.send(CaptureRequest::SetReleaseBind(bind));
    }

    pub(crate) fn set_release_threshold(&mut self, threshold: u32) {
        let _ = self
            .request_tx
            .send(CaptureRequest::SetReleaseThreshold(threshold));
    }
}

/// debounce a statement `$st`, i.e. the statement is executed only if the
/// time since the previous execution is at least `$dur`.
/// `$prev` is used to keep track of this timestamp
macro_rules! debounce {
    ($prev:ident, $dur:expr, $st:stmt) => {
        let exec = match $prev.get() {
            None => true,
            Some(instant) if instant.elapsed() > $dur => true,
            _ => false,
        };
        if exec {
            $prev.replace(Some(Instant::now()));
            $st
        }
    };
}

struct CaptureTask {
    active_client: Option<CaptureHandle>,
    backend: Option<input_capture::Backend>,
    cancellation_token: CancellationToken,
    captures: Vec<(CaptureHandle, Position, CaptureType)>,
    conn: MousehopConnection,
    event_tx: Sender<ICaptureEvent>,
    release_bind: Rc<RefCell<Vec<scancode::Linux>>>,
    release_threshold_px: Rc<RefCell<u32>>,
    request_rx: Receiver<CaptureRequest>,
    state: State,
}

impl CaptureTask {
    fn add_capture(&mut self, handle: CaptureHandle, pos: Position, capture_type: CaptureType) {
        self.captures.push((handle, pos, capture_type));
    }

    fn remove_capture(&mut self, handle: CaptureHandle) {
        self.captures.retain(|&(h, ..)| handle != h);
    }

    fn is_default_capture_at(&self, pos: Position) -> bool {
        self.captures
            .iter()
            .any(|&(_, p, t)| p == pos && t == CaptureType::Default)
    }

    fn get_pos(&self, handle: CaptureHandle) -> Position {
        self.captures
            .iter()
            .find(|(h, ..)| *h == handle)
            .expect("no such capture")
            .1
    }

    fn get_type(&self, handle: CaptureHandle) -> CaptureType {
        self.captures
            .iter()
            .find(|(h, ..)| *h == handle)
            .expect("no such capture")
            .2
    }

    async fn run(mut self) {
        loop {
            if let Err(e) = self.do_capture().await {
                log::warn!("input capture exited: {e}");
            }
            loop {
                tokio::select! {
                    r = self.request_rx.recv() => match r.expect("channel closed") {
                        CaptureRequest::Reenable => break,
                        CaptureRequest::Create(h, p, t) => self.add_capture(h, p, t),
                        CaptureRequest::Destroy(h) => self.remove_capture(h),
                        CaptureRequest::ReleaseForHandover => { /* nothing to do */ }
                        CaptureRequest::SetReleaseBind(bind) => {
                            self.release_bind.borrow_mut().clone_from(&bind);
                        }
                        CaptureRequest::SetReleaseThreshold(threshold) => {
                            *self.release_threshold_px.borrow_mut() = threshold;
                        }
                    },
                    _ = self.cancellation_token.cancelled() => return,
                }
            }
        }
    }

    async fn do_capture(&mut self) -> Result<(), InputCaptureError> {
        /* allow cancelling capture request */
        let mut capture = tokio::select! {
            r = InputCapture::new(self.backend) => r?,
            _ = self.cancellation_token.cancelled() => return Ok(()),
        };

        let _capture_guard = DropGuard::new(
            self.event_tx.clone(),
            ICaptureEvent::CaptureEnabled,
            ICaptureEvent::CaptureDisabled,
        );

        /* create barriers for active clients */
        let r = self.create_captures(&mut capture).await;
        if let Err(e) = r {
            capture.terminate().await?;
            return Err(e.into());
        }

        // Push the configured auto-release threshold to the freshly
        // created InputCapture. The wall-press detection is
        // cross-platform — every backend benefits.
        capture.set_release_threshold(*self.release_threshold_px.borrow());

        let r = self.do_capture_session(&mut capture).await;

        // FIXME replace with async drop when stabilized
        capture.terminate().await?;

        r
    }

    async fn create_captures(&mut self, capture: &mut InputCapture) -> Result<(), CaptureError> {
        let captures = self.captures.clone();
        for (handle, pos, _type) in captures {
            tokio::select! {
                r = capture.create(handle, pos) => r?,
                _ = self.cancellation_token.cancelled() => return Ok(()),
            }
        }
        Ok(())
    }

    async fn do_capture_session(
        &mut self,
        capture: &mut InputCapture,
    ) -> Result<(), InputCaptureError> {
        loop {
            tokio::select! {
                event = capture.next() => match event {
                    Some(event) => self.handle_capture_event(capture, event?).await?,
                    None => return Ok(()),
                },
                (handle, event) = self.conn.recv() => {
                    if let Some(active) = self.active_client {
                        if handle != active {
                            // we only care about events coming from the client we are currently connected to
                            // only `Ack` and `Leave` are relevant
                            continue
                        }
                    }

                    match event {
                        // connection acknowlegded => set state to Sending
                        ProtoEvent::Ack(_) => {
                            log::info!("client {handle} acknowledged the connection!");
                            self.state = State::Sending;
                        }
                        // Peer sent Leave — either they just released
                        // their own outbound capture, or they're
                        // taking over and want us to stop sending to
                        // them. The "taking over" case is the common
                        // one (CaptureBegin on the peer triggers a
                        // send_leave_event to every incoming address)
                        // and is followed by Enter+CursorPos from the
                        // peer. Use the handover release so the local
                        // host warp doesn't race against the peer's
                        // upcoming proportional CursorPos warp on our
                        // shared cursor.
                        ProtoEvent::Leave(_) => {
                            log::info!("releasing capture: left remote client device region");
                            self.release_capture_handover(capture).await?;
                        },
                        // Peer reported its display geometry — cache it
                        // so the wall-press model has a real upper
                        // clamp on virtual_pos for this position.
                        ProtoEvent::Bounds { width, height } => {
                            let pos = self.get_pos(handle);
                            capture.set_peer_bounds(pos, width, height);
                        }
                        // Peer reported its per-pair receive-side
                        // sensitivity multiplier — feed it into the
                        // wall-press model so its delta accumulator
                        // tracks the receiver's actual cursor advance.
                        // Without this, a sub-1.0 multiplier on the
                        // receiver makes the host's auto-release model
                        // fire before the cursor reaches the wall.
                        ProtoEvent::ReceiverSensitivity { mouse_sensitivity } => {
                            let pos = self.get_pos(handle);
                            capture.set_peer_sensitivity(pos, mouse_sensitivity);
                        }
                        // Peer's commit hash arrived on the outgoing
                        // DTLS connection. The connect-side
                        // receive_loop already wrote it to
                        // `client_manager`; bubble up to Service so
                        // the GUI's version-status row refreshes.
                        ProtoEvent::Hello { .. } => {
                            self.event_tx
                                .send(ICaptureEvent::PeerCommitUpdated(handle))
                                .expect("channel closed");
                        }
                        _ => {}
                    }
                },
                e = self.request_rx.recv() => match e.expect("channel closed") {
                    CaptureRequest::Reenable => { /* already active */ },
                    CaptureRequest::ReleaseForHandover => self.release_capture_handover(capture).await?,
                    CaptureRequest::Create(h, p, t) => {
                        self.add_capture(h, p, t);
                        capture.create(h, p).await?;
                    }
                    CaptureRequest::Destroy(h) => {
                        let pos = self.get_pos(h);
                        self.remove_capture(h);
                        capture.destroy(h).await?;
                        // Drop the cached geometry — the next client
                        // added at this position may report different
                        // bounds.
                        capture.clear_peer_bounds(pos);
                        // Same lifecycle for the cached sensitivity
                        // — re-add starts at the 1.0 default until a
                        // fresh ReceiverSensitivity arrives.
                        capture.clear_peer_sensitivity(pos);
                    }
                    CaptureRequest::SetReleaseBind(bind) => {
                        self.release_bind.borrow_mut().clone_from(&bind);
                    }
                    CaptureRequest::SetReleaseThreshold(threshold) => {
                        *self.release_threshold_px.borrow_mut() = threshold;
                        capture.set_release_threshold(threshold);
                    }
                },
                _ = self.cancellation_token.cancelled() => break,
            }
        }
        Ok(())
    }

    async fn handle_capture_event(
        &mut self,
        capture: &mut InputCapture,
        event: (CaptureHandle, CaptureEvent),
    ) -> Result<(), CaptureError> {
        let (handle, event) = event;
        log::trace!("({handle}): {event:?}");

        if capture.keys_pressed(&self.release_bind.borrow()) {
            log::info!("releasing capture: release-bind pressed");
            return self.release_capture(capture).await;
        }

        // Backend self-released (currently only macOS, when sustained
        // back-toward-host motion crosses the configured threshold).
        // Drive the same teardown path as the release-bind chord so
        // the peer gets a Leave + key-up flush.
        if matches!(event, CaptureEvent::AutoRelease) {
            log::info!("releasing capture: backend auto-release");
            return self.release_capture(capture).await;
        }

        if matches!(event, CaptureEvent::Begin { .. }) {
            self.event_tx
                .send(ICaptureEvent::CaptureBegin(handle))
                .expect("channel closed");
        }

        // enter only capture (for incoming connections)
        if self.get_type(handle) == CaptureType::EnterOnly {
            // if there is no active outgoing connection at the current capture,
            // we release the capture
            if !self.is_default_capture_at(self.get_pos(handle)) {
                log::info!("releasing capture: no active client at this position");
                capture.release().await?;
            }
            // we dont care about events from incoming handles except for releasing the capture
            return Ok(());
        }

        // activated a new client
        if matches!(event, CaptureEvent::Begin { .. }) && Some(handle) != self.active_client {
            self.state = State::WaitingForAck;
            self.active_client.replace(handle);
            self.event_tx
                .send(ICaptureEvent::ClientEntered(handle))
                .expect("channel closed");
        }

        let opposite_pos = to_proto_pos(self.get_pos(handle).opposite());

        // If we're starting a fresh capture and the backend reported
        // a cursor position at the moment of crossing, send a
        // `CursorPos` (host-normalized fraction + entry side from the
        // peer's frame) right after Enter. The peer scales against
        // its own live bounds and pins the on-axis dimension to the
        // matching edge — self-sufficient, no prior `Bounds`
        // round-trip needed, so the very first crossing also lands
        // the cursor at the visually-corresponding point.
        let cursor_pos = if let CaptureEvent::Begin {
            cursor: Some(cursor),
        } = event
        {
            let pos = self.get_pos(handle);
            capture.host_normalized_cursor(cursor).map(|(nx, ny)| {
                let proto_pos = to_proto_pos(pos.opposite());
                (proto_pos, nx, ny)
            })
        } else {
            None
        };

        let proto_event = match &event {
            CaptureEvent::Begin { .. } => ProtoEvent::Enter(opposite_pos),
            CaptureEvent::Input(e) => match self.state {
                // connection not acknowledged, repeat `Enter` event
                State::WaitingForAck => ProtoEvent::Enter(opposite_pos),
                State::Sending => ProtoEvent::Input(e.clone()),
            },
            CaptureEvent::AutoRelease => unreachable!("handled in early return above"),
        };

        if let Err(e) = self.conn.send(proto_event, handle).await {
            const DUR: Duration = Duration::from_millis(500);
            debounce!(PREV_LOG, DUR, log::warn!("releasing capture: {e}"));
            capture.release().await?;
            return Ok(());
        }

        // Send CursorPos right after Enter so the receiver can warp
        // its cursor to the visually-corresponding point on its own
        // screen — overrides the entry-edge-midpoint warp the
        // receiver otherwise applies on Enter.
        if let Some((pos, nx, ny)) = cursor_pos {
            log::info!("[cursor-pos] send pos={pos:?} nx={nx:.3} ny={ny:.3}");
            if let Err(e) = self
                .conn
                .send(ProtoEvent::CursorPos { pos, nx, ny }, handle)
                .await
            {
                log::warn!("CursorPos send failed: {e}");
            }
        } else if matches!(event, CaptureEvent::Begin { .. }) {
            log::info!(
                "[cursor-pos] send skipped — Begin had no cursor or host_normalized_cursor returned None"
            );
        }
        Ok(())
    }

    async fn release_capture(&mut self, capture: &mut InputCapture) -> Result<(), CaptureError> {
        self.notify_peer_of_leave(capture).await;
        capture.release().await
    }

    /// Release path used when the peer is taking over (they sent
    /// Enter+CursorPos). Same teardown — synthesize key-ups, reset
    /// mods, send Leave — but skip the host-side cursor warp so it
    /// doesn't race against the peer's authoritative CursorPos
    /// warp on our shared cursor.
    async fn release_capture_handover(
        &mut self,
        capture: &mut InputCapture,
    ) -> Result<(), CaptureError> {
        self.notify_peer_of_leave(capture).await;
        capture.release_no_host_warp().await
    }

    async fn notify_peer_of_leave(&mut self, capture: &mut InputCapture) {
        // If we have an active client, notify them we're leaving
        if let Some(handle) = self.active_client.take() {
            // Synthesize key-up events for every key still held in the
            // capture's pressed_keys set BEFORE sending Leave. Without
            // this, pressing the release-bind chord (typically all four
            // modifiers) leaves the peer with phantom held modifiers:
            // the down events were forwarded while capture was active,
            // but the matching up events arrive after the local tap
            // flips to passthrough and never reach the peer. The peer
            // then runs every subsequent keystroke through those held
            // mods until its watchdog times out (1+ s) or our Leave
            // arrives — and Leave can be lost over UDP/DTLS.
            for key in capture.take_pressed_keys() {
                let key_up = ProtoEvent::Input(Event::Keyboard(KeyboardEvent::Key {
                    time: 0,
                    key: key as u32,
                    state: 0,
                }));
                if let Err(e) = self.conn.send(key_up, handle).await {
                    log::warn!("failed to send key-up to client {handle}: {e}");
                }
            }
            // Reset the modifier mask too. The peer's input-emulation
            // layer keeps a separate XKB-style modifier state that's
            // updated by KeyboardEvent::Modifiers, distinct from the
            // pressed_keys set drained above. Without this, an
            // already-locked CapsLock would survive the release.
            let mods_zero = ProtoEvent::Input(Event::Keyboard(KeyboardEvent::Modifiers {
                depressed: 0,
                latched: 0,
                locked: 0,
                group: 0,
            }));
            if let Err(e) = self.conn.send(mods_zero, handle).await {
                log::warn!("failed to reset modifiers on client {handle}: {e}");
            }

            log::info!("sending Leave event to client {handle}");
            if let Err(e) = self.conn.send(ProtoEvent::Leave(0), handle).await {
                log::warn!("failed to send Leave to client {handle}: {e}");
            }
        }
    }
}

thread_local! {
    static PREV_LOG: Cell<Option<Instant>> = const { Cell::new(None) };
}

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
enum State {
    #[default]
    WaitingForAck,
    Sending,
}

fn to_capture_pos(pos: mousehop_ipc::Position) -> input_capture::Position {
    match pos {
        mousehop_ipc::Position::Left => input_capture::Position::Left,
        mousehop_ipc::Position::Right => input_capture::Position::Right,
        mousehop_ipc::Position::Top => input_capture::Position::Top,
        mousehop_ipc::Position::Bottom => input_capture::Position::Bottom,
    }
}

fn to_proto_pos(pos: input_capture::Position) -> mousehop_proto::Position {
    match pos {
        input_capture::Position::Left => mousehop_proto::Position::Left,
        input_capture::Position::Right => mousehop_proto::Position::Right,
        input_capture::Position::Top => mousehop_proto::Position::Top,
        input_capture::Position::Bottom => mousehop_proto::Position::Bottom,
    }
}

struct DropGuard<T> {
    tx: Sender<T>,
    on_drop: Option<T>,
}

impl<T> DropGuard<T> {
    fn new(tx: Sender<T>, on_new: T, on_drop: T) -> Self {
        tx.send(on_new).expect("channel closed");
        let on_drop = Some(on_drop);
        Self { tx, on_drop }
    }
}

impl<T> Drop for DropGuard<T> {
    fn drop(&mut self) {
        self.tx
            .send(self.on_drop.take().expect("item"))
            .expect("channel closed");
    }
}