mnml-rs 0.2.14

//! Native WebSocket pane state + worker. Uses the existing
//! `tungstenite` dep (already in tree for CDP). A persistent
//! connection runs on a background thread; messages flow in
//! both directions over channels.
//!
//! Scope:
//!  - Single connection per pane (one URL, one socket).
//!  - Text and binary frames as messages.
//!  - User types into a single-line input + Enter to send.
//!  - Esc closes the connection.
//!  - Subprotocol negotiation via `Sec-WebSocket-Protocol`.
//!  - Keepalive Ping every N seconds (tungstenite auto-replies to Pong).
//!  - Auto-reconnect with exponential backoff on drop.

use std::sync::mpsc::{Receiver, Sender};
use std::time::{Duration, Instant};

/// Runtime options for a WebSocket connection. Threaded through
/// `WebsocketPane::connect` to the worker so subprotocol
/// negotiation, ping keepalive, and auto-reconnect can be
/// configured from `[ws]` in mnml config without touching the
/// worker signature every time.
#[derive(Debug, Clone)]
pub struct WsConnectOpts {
    /// Sec-WebSocket-Protocol header values (in preference order).
    /// Empty = no subprotocol negotiation.
    pub subprotocols: Vec<String>,
    /// Send a Ping frame every N seconds while the connection is
    /// open. 0 = disabled. tungstenite handles the incoming Pong
    /// reply automatically; this just keeps NATed / load-balanced
    /// connections from timing out.
    pub ping_interval_secs: u32,
    /// Reconnect up to N times on a non-user drop, with 1s → 2s →
    /// 4s → 8s → 16s backoff (capped at 16s). 0 = disabled.
    pub reconnect_max_attempts: u32,
}

impl Default for WsConnectOpts {
    fn default() -> Self {
        Self {
            subprotocols: Vec::new(),
            ping_interval_secs: 30,
            reconnect_max_attempts: 3,
        }
    }
}

pub enum WsMsg {
    /// Server → client message (or our own echo of a send).
    Recv {
        ts: Instant,
        text: String,
        outgoing: bool,
    },
    /// Connection state changed.
    State(WsState),
    /// Error from the worker (transport / parse / etc).
    Error(String),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WsState {
    Connecting,
    Open,
    Closing,
    Closed,
}

pub struct WebsocketPane {
    pub url: String,
    pub state: WsState,
    pub log: Vec<LogEntry>,
    /// Pending input the user is typing. Enter → send + clear.
    pub input: String,
    pub input_cursor: usize,
    pub rx: Receiver<WsMsg>,
    /// Channel to the worker for sending messages or close.
    pub tx_out: Sender<OutMsg>,
    /// Scroll offset within the log (rows from the bottom — 0 =
    /// follow tail). Bumped by wheel + PgUp/PgDn.
    pub scroll: usize,
}

pub enum OutMsg {
    Send(String),
    Close,
}

pub struct LogEntry {
    pub ts: Instant,
    pub outgoing: bool,
    pub text: String,
}

impl WebsocketPane {
    /// Tab title — `ws://host` or `wss://host`, schema implied by
    /// the URL scheme. Truncates long URLs.
    pub fn tab_title(&self) -> String {
        let host = host_of_url(&self.url);
        let badge = match self.state {
            WsState::Connecting => "…",
            WsState::Open => "●",
            WsState::Closing => "▼",
            WsState::Closed => "·",
        };
        format!("ws {badge} {host}")
    }

    pub fn connect(url: String, opts: WsConnectOpts) -> Self {
        let (msg_tx, rx) = std::sync::mpsc::channel::<WsMsg>();
        let (tx_out, out_rx) = std::sync::mpsc::channel::<OutMsg>();
        let url_clone = url.clone();
        let msg_tx_send = msg_tx.clone();
        std::thread::spawn(move || worker(url_clone, opts, msg_tx_send, out_rx));
        Self {
            url,
            state: WsState::Connecting,
            log: Vec::new(),
            input: String::new(),
            input_cursor: 0,
            rx,
            tx_out,
            scroll: 0,
        }
    }

    /// Drain pending messages into `log` + apply state changes.
    /// Called from App.tick. Also persists each message to
    /// `~/.mnml/ws-history/<host-slug>/history.jsonl` (2026-06-21).
    pub fn drain(&mut self) {
        while let Ok(msg) = self.rx.try_recv() {
            match msg {
                WsMsg::Recv { ts, text, outgoing } => {
                    persist_history(&self.url, outgoing, &text);
                    self.log.push(LogEntry { ts, outgoing, text });
                }
                WsMsg::State(s) => self.state = s,
                WsMsg::Error(e) => {
                    self.log.push(LogEntry {
                        ts: Instant::now(),
                        outgoing: false,
                        text: format!("ERROR: {e}"),
                    });
                }
            }
        }
    }

    /// User pressed Enter — send the pending input.
    pub fn send_input(&mut self) {
        if self.input.is_empty() {
            return;
        }
        let payload = std::mem::take(&mut self.input);
        self.input_cursor = 0;
        persist_history(&self.url, true, &payload);
        // Mirror into log first (we won't get an echo from server).
        self.log.push(LogEntry {
            ts: Instant::now(),
            outgoing: true,
            text: payload.clone(),
        });
        let _ = self.tx_out.send(OutMsg::Send(payload));
    }

    pub fn close(&mut self) {
        let _ = self.tx_out.send(OutMsg::Close);
        self.state = WsState::Closing;
    }

    /// Insert a char at the cursor. Handles UTF-8 (cursor moves by
    /// `c.len_utf8()`). 2026-06-21 power-user-ws-git SEV-3 input-
    /// cursor-dead: was push()-only, cursor was always at end.
    pub fn input_insert(&mut self, c: char) {
        self.input.insert(self.input_cursor, c);
        self.input_cursor += c.len_utf8();
    }

    /// Backspace = delete char before cursor.
    pub fn input_backspace(&mut self) {
        if self.input_cursor == 0 || self.input.is_empty() {
            return;
        }
        // Find the previous char boundary.
        let mut i = self.input_cursor.saturating_sub(1);
        while i > 0 && !self.input.is_char_boundary(i) {
            i -= 1;
        }
        self.input.replace_range(i..self.input_cursor, "");
        self.input_cursor = i;
    }

    /// Delete = remove char at cursor (vim x / VS Code Del).
    pub fn input_delete(&mut self) {
        if self.input_cursor >= self.input.len() {
            return;
        }
        let mut i = self.input_cursor + 1;
        while i < self.input.len() && !self.input.is_char_boundary(i) {
            i += 1;
        }
        self.input.replace_range(self.input_cursor..i, "");
    }

    /// Move cursor left by one char.
    pub fn input_left(&mut self) {
        if self.input_cursor == 0 {
            return;
        }
        let mut i = self.input_cursor - 1;
        while i > 0 && !self.input.is_char_boundary(i) {
            i -= 1;
        }
        self.input_cursor = i;
    }

    /// Move cursor right by one char.
    pub fn input_right(&mut self) {
        if self.input_cursor >= self.input.len() {
            return;
        }
        let mut i = self.input_cursor + 1;
        while i < self.input.len() && !self.input.is_char_boundary(i) {
            i += 1;
        }
        self.input_cursor = i;
    }

    pub fn input_home(&mut self) {
        self.input_cursor = 0;
    }

    pub fn input_end(&mut self) {
        self.input_cursor = self.input.len();
    }

    /// 2026-08-08 — Ctrl+V paste. Newlines preserved (WS send
    /// input is multi-line-friendly — JSON payloads etc.).
    pub fn input_insert_str(&mut self, s: &str) {
        self.input.insert_str(self.input_cursor, s);
        self.input_cursor += s.len();
    }

    /// 2026-08-08 — Ctrl+W kill-word-back.
    pub fn input_delete_word_back(&mut self) {
        if self.input_cursor == 0 {
            return;
        }
        let head = &self.input[..self.input_cursor];
        let trimmed = head.trim_end_matches(char::is_whitespace);
        let cut = trimmed
            .char_indices()
            .rev()
            .find(|&(_, c)| c.is_whitespace())
            .map(|(i, c)| i + c.len_utf8())
            .unwrap_or(0);
        self.input.replace_range(cut..self.input_cursor, "");
        self.input_cursor = cut;
    }

    /// 2026-08-08 — Ctrl+U kill to input start.
    pub fn input_delete_to_start(&mut self) {
        if self.input_cursor == 0 {
            return;
        }
        self.input.replace_range(..self.input_cursor, "");
        self.input_cursor = 0;
    }

    /// 2026-08-08 — Ctrl+K kill to input end.
    pub fn input_delete_to_end(&mut self) {
        self.input.truncate(self.input_cursor);
    }
}

/// Toggle non-blocking on the underlying TCP stream. tungstenite's
/// `MaybeTlsStream<TcpStream>` is what `connect()` returns; we
/// reach the raw TcpStream regardless of TLS wrapping.
fn set_socket_nonblocking(
    stream: &mut tungstenite::stream::MaybeTlsStream<std::net::TcpStream>,
    on: bool,
) {
    use tungstenite::stream::MaybeTlsStream;
    // mnml's tungstenite is built without TLS features (no wss://
    // support), so only Plain is reachable. The TLS arms are
    // gated so they don't compile-warn here while keeping
    // forward-compat if we ever enable wss.
    let MaybeTlsStream::Plain(tcp) = stream else {
        return;
    };
    let _ = tcp.set_nonblocking(on);
}

fn host_of_url(url: &str) -> String {
    let trimmed = url
        .strip_prefix("wss://")
        .or_else(|| url.strip_prefix("ws://"))
        .unwrap_or(url);
    let host = trimmed.split(['/', '?']).next().unwrap_or(trimmed);
    let max = 32usize;
    if host.chars().count() <= max {
        host.to_string()
    } else {
        let cut: String = host.chars().take(max).collect();
        format!("{cut}…")
    }
}

/// Outcome of one attempted connection lifecycle. Feeds the
/// reconnect-loop wrapper around `run_connection`.
enum ConnResult {
    /// User asked to close (`OutMsg::Close`). Do not reconnect.
    UserClose,
    /// Connection dropped by peer or transport. Eligible for
    /// reconnect if the caller's budget allows.
    Dropped,
    /// Setup-time failure (invalid URL, connect refused). Do not
    /// reconnect — probably not going to succeed on retry.
    Failed,
}

fn worker(url: String, opts: WsConnectOpts, tx: Sender<WsMsg>, out_rx: Receiver<OutMsg>) {
    let mut attempt: u32 = 0;
    loop {
        match run_connection(&url, &opts, &tx, &out_rx) {
            ConnResult::UserClose | ConnResult::Failed => {
                let _ = tx.send(WsMsg::State(WsState::Closed));
                return;
            }
            ConnResult::Dropped => {
                if attempt >= opts.reconnect_max_attempts {
                    let _ = tx.send(WsMsg::State(WsState::Closed));
                    return;
                }
                attempt += 1;
                let backoff_secs = 1u64 << (attempt - 1).min(4);
                let _ = tx.send(WsMsg::Error(format!(
                    "dropped — reconnecting in {backoff_secs}s (attempt {attempt}/{})",
                    opts.reconnect_max_attempts
                )));
                let _ = tx.send(WsMsg::State(WsState::Connecting));
                std::thread::sleep(Duration::from_secs(backoff_secs));
            }
        }
    }
}

fn run_connection(
    url: &str,
    opts: &WsConnectOpts,
    tx: &Sender<WsMsg>,
    out_rx: &Receiver<OutMsg>,
) -> ConnResult {
    use tungstenite::Message;
    use tungstenite::client::IntoClientRequest;

    // Build the client request. tungstenite accepts URL-string OR
    // a full http::Request<()>; the second form is required to
    // pass a Sec-WebSocket-Protocol header.
    let mut request = match url.into_client_request() {
        Ok(r) => r,
        Err(e) => {
            let _ = tx.send(WsMsg::Error(format!("invalid url: {e}")));
            return ConnResult::Failed;
        }
    };
    if !opts.subprotocols.is_empty() {
        let joined = opts.subprotocols.join(", ");
        if let Ok(val) = joined.parse::<tungstenite::http::HeaderValue>() {
            request.headers_mut().insert("Sec-WebSocket-Protocol", val);
        }
    }

    let (mut socket, _resp) = match tungstenite::connect(request) {
        Ok(t) => t,
        Err(e) => {
            let _ = tx.send(WsMsg::Error(format!("connect failed: {e}")));
            // Peer-side reject (401/404/reset) counts as a drop:
            // let the reconnect budget decide whether to try
            // again. Truly-bad URLs are already caught above.
            return ConnResult::Dropped;
        }
    };
    // 2026-06-21 — Set the underlying TCP socket non-blocking so
    // `socket.read()` returns WouldBlock instead of stalling the
    // worker. Was: read blocked until the server spoke, which
    // meant `OutMsg::Send`/`Close` queued via out_rx were stuck
    // behind it — first `:ws.send_message` to a quiet echo/RPC
    // server deadlocked (the SEV-1 power-user-ws-git finding).
    set_socket_nonblocking(socket.get_mut(), true);
    let _ = tx.send(WsMsg::State(WsState::Open));

    let ping_every = if opts.ping_interval_secs == 0 {
        None
    } else {
        Some(Duration::from_secs(opts.ping_interval_secs as u64))
    };
    let mut last_ping = Instant::now();

    loop {
        // Drain pending outgoing first so user-initiated sends
        // don't sit behind a slow read.
        while let Ok(out) = out_rx.try_recv() {
            match out {
                OutMsg::Send(text) => {
                    if let Err(e) = socket.send(Message::Text(text.into())) {
                        let _ = tx.send(WsMsg::Error(format!("send failed: {e}")));
                    }
                }
                OutMsg::Close => {
                    let _ = socket.close(None);
                    return ConnResult::UserClose;
                }
            }
        }
        // Time to send a keepalive?
        if let Some(interval) = ping_every
            && last_ping.elapsed() >= interval
        {
            let _ = socket.send(Message::Ping(Vec::new().into()));
            last_ping = Instant::now();
        }
        match socket.read() {
            Ok(Message::Text(s)) => {
                let _ = tx.send(WsMsg::Recv {
                    ts: Instant::now(),
                    text: s.to_string(),
                    outgoing: false,
                });
            }
            Ok(Message::Binary(b)) => {
                let _ = tx.send(WsMsg::Recv {
                    ts: Instant::now(),
                    text: format!("(binary {} bytes)", b.len()),
                    outgoing: false,
                });
            }
            Ok(Message::Close(_)) => return ConnResult::Dropped,
            Ok(_) => {} // ping / pong handled by tungstenite
            Err(tungstenite::Error::ConnectionClosed) | Err(tungstenite::Error::AlreadyClosed) => {
                return ConnResult::Dropped;
            }
            Err(tungstenite::Error::Io(io)) if io.kind() == std::io::ErrorKind::WouldBlock => {
                // No data — sleep briefly so the loop doesn't burn
                // CPU. 25ms gives a UI-imperceptible round-trip
                // between out_rx drain and read.
                std::thread::sleep(Duration::from_millis(25));
            }
            Err(e) => {
                let _ = tx.send(WsMsg::Error(format!("read error: {e}")));
                return ConnResult::Dropped;
            }
        }
    }
}

/// 2026-06-21 — best-effort persist a single message to the
/// per-host history file. Location: `<mnml-data>/ws-history/…` in
/// portable mode; `~/.mnml/ws-history/…` in HOME mode (kept for
/// backwards-compat — this scheme predates data_root and the tree
/// exists on many users' machines). Silently no-ops on failure —
/// informational, not load-bearing.
fn persist_history(url: &str, outgoing: bool, text: &str) {
    let host = host_of_url(url);
    let slug = host.replace(['/', ':'], "_").replace(
        |c: char| !c.is_ascii_alphanumeric() && c != '_' && c != '.' && c != '-',
        "_",
    );
    if slug.is_empty() {
        return;
    }
    let base = if crate::data_root::data_root_kind() == crate::data_root::DataRootKind::Portable {
        crate::data_root::data_root().join("ws-history")
    } else {
        let Some(home) = std::env::var_os("HOME") else {
            return;
        };
        std::path::PathBuf::from(home).join(".mnml/ws-history")
    };
    let dir = base.join(&slug);
    let _ = std::fs::create_dir_all(&dir);
    let path = dir.join("history.jsonl");
    let ts_ms = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis())
        .unwrap_or(0);
    // Encode text as a JSON string literal (escape control chars
    // + quote + backslash). Newlines flattened to \n.
    let escaped = text
        .replace('\\', "\\\\")
        .replace('"', "\\\"")
        .replace('\n', "\\n")
        .replace('\r', "\\r")
        .replace('\t', "\\t");
    let line = format!(
        "{{\"ts\":{ts_ms},\"url\":\"{url}\",\"outgoing\":{outgoing},\"text\":\"{escaped}\"}}\n"
    );
    use std::io::Write;
    if let Ok(mut f) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
    {
        let _ = f.write_all(line.as_bytes());
    }
}

/// 2026-06-21 — `:ws.history` reader. Walks
/// `~/.mnml/ws-history/*/history.jsonl`, returns
/// `Vec<(url, last_ts_ms, message_count)>` sorted by last_ts
/// descending. Used by the history picker.
pub fn read_ws_history() -> Vec<(String, u128, usize)> {
    let mut out: std::collections::BTreeMap<String, (u128, usize)> =
        std::collections::BTreeMap::new();
    let root = if crate::data_root::data_root_kind() == crate::data_root::DataRootKind::Portable {
        crate::data_root::data_root().join("ws-history")
    } else {
        let Some(home) = std::env::var_os("HOME") else {
            return Vec::new();
        };
        std::path::PathBuf::from(home).join(".mnml/ws-history")
    };
    let Ok(rd) = std::fs::read_dir(&root) else {
        return Vec::new();
    };
    for d in rd.flatten() {
        let p = d.path().join("history.jsonl");
        let Ok(text) = std::fs::read_to_string(&p) else {
            continue;
        };
        let mut last_ts: u128 = 0;
        let mut url_seen: Option<String> = None;
        let mut count = 0usize;
        for line in text.lines() {
            count += 1;
            // Cheap field-extract without pulling serde_json — both
            // fields are at fixed positions in the writer.
            if let Some(ts_str) = line
                .strip_prefix("{\"ts\":")
                .and_then(|s| s.split(',').next())
                && let Ok(ts) = ts_str.parse::<u128>()
            {
                last_ts = last_ts.max(ts);
            }
            if url_seen.is_none()
                && let Some(rest) = line.split("\"url\":\"").nth(1)
                && let Some(end) = rest.find('"')
            {
                url_seen = Some(rest[..end].to_string());
            }
        }
        if let Some(u) = url_seen {
            out.entry(u).or_insert((last_ts, count)).0 = last_ts;
            out.get_mut(&out.keys().next().unwrap().clone()); // no-op for borrow
        }
    }
    let mut rows: Vec<(String, u128, usize)> = out
        .into_iter()
        .map(|(url, (ts, count))| (url, ts, count))
        .collect();
    rows.sort_by_key(|b| std::cmp::Reverse(b.1));
    rows
}

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

    #[test]
    fn opts_defaults_are_sensible() {
        let o = WsConnectOpts::default();
        assert!(o.subprotocols.is_empty());
        assert_eq!(o.ping_interval_secs, 30);
        assert_eq!(o.reconnect_max_attempts, 3);
    }

    /// Backoff schedule the worker uses: `1 << (attempt-1).min(4)`.
    /// Guards against off-by-one drift + accidental unbounded
    /// growth if someone edits the shift.
    #[test]
    fn reconnect_backoff_schedule_is_capped_at_16s() {
        let schedule: Vec<u64> = (1..=8)
            .map(|attempt: u32| 1u64 << (attempt - 1).min(4))
            .collect();
        assert_eq!(schedule, vec![1, 2, 4, 8, 16, 16, 16, 16]);
    }

    #[test]
    fn ping_interval_zero_disables_keepalive() {
        let opts = WsConnectOpts {
            ping_interval_secs: 0,
            ..Default::default()
        };
        let ping_every = if opts.ping_interval_secs == 0 {
            None
        } else {
            Some(Duration::from_secs(opts.ping_interval_secs as u64))
        };
        assert!(ping_every.is_none());
    }
}