omniterm 0.2.27

use axum::{
    Router,
    extract::{Query, State},
    response::{
        IntoResponse,
        sse::{Event, KeepAlive, Sse},
    },
    routing::get,
};
use futures_util::stream::{self, Stream};
use notify::event::CreateKind;
use notify::{Event as NotifyEvent, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use serde::Deserialize;
use std::convert::Infallible;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::time::Duration;
use tokio::sync::broadcast;
use walkdir::WalkDir;

use crate::AppState;

use super::files::{resolve_project_root, resolve_session_base};

/// 事件去抖窗口（ADR-6）：短窗口内按 (kind, path) 去重合并后再批量下发，
/// 消除「单次保存的 N 个 Modify → N 次前端全量 list」的请求放大。
const DEBOUNCE_WINDOW_MS: u64 = 100;
/// 合并窗口内 pending 表的上限：洪峰时不等待窗口，超限立即清空并下发单条
/// `resync` 让前端全量重同步（有界缓冲红线，见 docs/dev/performance-and-safety.md §P1）。
const MAX_PENDING: usize = 256;
/// 全量重同步事件：消费跟不上（broadcast Lagged）或合并缓冲超限时的降级出口，
/// 前端收到即触发一次全量刷新（ADR-6）。
const RESYNC_EVENT: &str = r#"{"kind":"resync","path":""}"#;

pub fn routes() -> Router<AppState> {
    Router::new().route("/files/watch", get(watch_files))
}

#[derive(Deserialize)]
struct WatchQuery {
    session: Option<String>,
    workspace: Option<String>,
}

/// SSE endpoint for real-time file change notifications.
/// Watches the specified directory and pushes change events to the client.
async fn watch_files(
    State(state): State<AppState>,
    Query(q): Query<WatchQuery>,
) -> impl IntoResponse {
    // Resolve the directory to watch
    let watch_path = if let Some(sid) = q.session.as_deref() {
        resolve_session_base(&state, sid).await.map(|(cwd, _)| PathBuf::from(cwd))
    } else {
        let wid = q.workspace.as_deref().unwrap_or("default");
        resolve_project_root(&state, wid).await.map(PathBuf::from)
    };

    let watch_path = match watch_path {
        Some(p) if p.exists() => p,
        _ => {
            // Return an empty stream if path can't be resolved
            let empty: Pin<Box<dyn Stream<Item = Result<Event, Infallible>> + Send>> =
                Box::pin(stream::empty());
            return Sse::new(empty).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)));
        }
    };

    let (tx, mut rx) = broadcast::channel::<String>(64);

    // Shutdown channel: uses std::sync::mpsc (not tokio::sync::watch) so the
    // blocking thread can detect sender-drop without depending on a tokio
    // runtime context.  `now_or_never()` inside `spawn_blocking` was unreliable
    // and leaked inotify instances (each watcher = 1 inotify fd, system limit
    // is 128; leaked watchers consumed ~50% → new Vite could not start).
    let (shutdown_tx, shutdown_rx) = std::sync::mpsc::channel::<()>();
    // 新目录注册通道：回调里不能碰 watcher（构造在回调外），Create(Folder) 事件
    // 把新目录路径投给消费循环，由它在同一阻塞线程内补注册（见下）。
    let (new_dir_tx, new_dir_rx) = std::sync::mpsc::channel::<PathBuf>();

    let watch_dir = watch_path.clone();
    let watch_dir_for_cb = watch_dir.clone();
    tokio::task::spawn_blocking(move || {
        let mut watcher = match RecommendedWatcher::new(
            move |res: Result<NotifyEvent, notify::Error>| {
                let event = match res {
                    Ok(e) => e,
                    Err(_) => return,
                };
                // 新目录 → 投给消费循环补注册 inotify watch（跳过 ignore 目录）。
                // 必须手动处理：`RecursiveMode::Recursive` 会让 notify 内部 WalkDir
                // 把 node_modules/.git/target 等全注册（1 万+ watch，见 collect_watch_dirs
                // 注释），且 notify 8.2 在该规模 + 持续事件下 `handle_inotify` 内层循环
                // 饿死 mio poll——notify-rx 线程 100% CPU、RSS 堆膨胀（2026-08-16 根因）。
                if let EventKind::Create(CreateKind::Folder) = event.kind {
                    for path in &event.paths {
                        if !is_ignored_dir(path, &watch_dir_for_cb) {
                            let _ = new_dir_tx.send(path.clone());
                        }
                    }
                }
                let changes = notify_event_to_changes(&event, &watch_dir_for_cb);
                for change in changes {
                    let _ = tx.send(change);
                }
            },
            notify::Config::default(),
        ) {
            Ok(w) => w,
            Err(_) => return,
        };

        // 根目录 + 全部非 ignore 子目录逐个注册（NonRecursive）。相比
        // `RecursiveMode::Recursive`（notify 内部 WalkDir 不跳过 ignore 目录，
        // 项目含 node_modules 时 watch 数可达上万），此处把 watch 数压到
        // 实际业务目录量级，从源头隔离 node_modules 等高频目录的事件风暴。
        if watcher.watch(&watch_dir, RecursiveMode::NonRecursive).is_err() {
            return;
        }
        for dir in collect_watch_dirs(&watch_dir) {
            if watcher.watch(&dir, RecursiveMode::NonRecursive).is_err() {
                break;
            }
        }

        // Park until the SSE stream drops `shutdown_tx` (client disconnect).
        // `recv_timeout` returns `Disconnected` when the sender is dropped —
        // a pure sync mechanism that works reliably from any thread.
        // 同时消费 `new_dir_rx`：新目录（含 Create 时已存在的深层子目录）补注册。
        loop {
            while let Ok(dir) = new_dir_rx.try_recv() {
                if watcher.watch(&dir, RecursiveMode::NonRecursive).is_ok() {
                    for sub in collect_watch_dirs(&dir) {
                        if watcher.watch(&sub, RecursiveMode::NonRecursive).is_err() {
                            break;
                        }
                    }
                }
            }
            if let Err(std::sync::mpsc::RecvTimeoutError::Disconnected) =
                shutdown_rx.recv_timeout(Duration::from_millis(250))
            {
                break;
                // Timeout 或 Ok：继续等待
            }
        }
        // `watcher` drops here → `inotify_rm_watch` for every registered path.
    });

    let sse_stream = async_stream::stream! {
        // `_shutdown_guard` lives as long as the generator; dropping the SSE
        // body drops the generator, which drops the guard, which drops the
        // channel sender, which wakes the blocking task via `Disconnected`.
        let _shutdown_guard = shutdown_tx;
        // 事件去抖（ADR-6）：窗口内按 (kind, path) 去重合并后批量 yield，
        // 降低「每事件一次全量 list」的请求放大。两个溢出源收敛到同一条
        // `resync` 出口（合并缓冲超限 / broadcast Lagged）。
        let mut pending: Vec<String> = Vec::new();
        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
        // Sleep 是 !Unpin，用 Pin<Box<..>> 持有以便 select 循环里重复 await
        let mut flush_timer: Option<Pin<Box<tokio::time::Sleep>>> = None;
        loop {
            tokio::select! {
                recv = rx.recv() => match recv {
                    Ok(data) => {
                        let over = merge_pending(&mut pending, &mut seen, data, MAX_PENDING);
                        if over {
                            yield Ok(Event::default().event("change").data(RESYNC_EVENT));
                            pending.clear();
                            seen.clear();
                            flush_timer = None;
                        } else if flush_timer.is_none() {
                            flush_timer = Some(Box::pin(tokio::time::sleep(Duration::from_millis(DEBOUNCE_WINDOW_MS))));
                        }
                    }
                    // 消费跟不上导致广播溢出：数据已不连续，下发 resync 全量重同步
                    Err(broadcast::error::RecvError::Lagged(_)) => {
                        yield Ok(Event::default().event("change").data(RESYNC_EVENT));
                        pending.clear();
                        seen.clear();
                        flush_timer = None;
                    }
                    Err(broadcast::error::RecvError::Closed) => break,
                },
                _ = async {
                    if let Some(t) = &mut flush_timer {
                        t.await;
                    }
                }, if flush_timer.is_some() => {
                    for change in pending.drain(..) {
                        yield Ok(Event::default().event("change").data(change));
                    }
                    seen.clear();
                    flush_timer = None;
                }
            }
        }
    };

    let boxed: Pin<Box<dyn Stream<Item = Result<Event, Infallible>> + Send>> = Box::pin(sse_stream);

    Sse::new(boxed).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)).text("ping"))
}

/// 把一条 change 消息去重合并进 pending 表。返回 `true` 表示已超 `max_pending`
/// 上限，调用方应清空 pending 并下发单条 `resync`（洪峰时不等待窗口，直接全量
/// 重同步，见 ADR-6）。同一 (kind, path) 产生相同的 JSON 字符串，以此作为去重键。
/// 纯函数，便于单测。
fn merge_pending(
    pending: &mut Vec<String>,
    seen: &mut std::collections::HashSet<String>,
    change: String,
    max_pending: usize,
) -> bool {
    if seen.insert(change.clone()) {
        pending.push(change);
    }
    pending.len() >= max_pending
}

/// Convert a notify event into one or more JSON change messages.
fn notify_event_to_changes(event: &NotifyEvent, base_dir: &std::path::Path) -> Vec<String> {
    let mut changes = Vec::new();

    for path in &event.paths {
        // Compute relative path from the watched directory
        let rel_path = path.strip_prefix(base_dir).unwrap_or(path).to_string_lossy().to_string();

        // Skip hidden files and common non-interesting directories
        if should_ignore(&rel_path) {
            continue;
        }

        let kind_str = match event.kind {
            EventKind::Create(_) => "create",
            EventKind::Modify(_) => "modify",
            EventKind::Remove(_) => "delete",
            _ => continue, // ignore access, metadata, etc.
        };

        let json = format!(r#"{{"kind":"{}","path":"{}"}}"#, kind_str, escape_json(&rel_path));
        changes.push(json);
    }

    // Handle renames specially
    if let EventKind::Modify(notify::event::ModifyKind::Name(_)) = event.kind
        && event.paths.len() == 2
    {
        let from = event.paths[0]
            .strip_prefix(base_dir)
            .unwrap_or(&event.paths[0])
            .to_string_lossy()
            .to_string();
        let to = event.paths[1]
            .strip_prefix(base_dir)
            .unwrap_or(&event.paths[1])
            .to_string_lossy()
            .to_string();

        if !should_ignore(&from) && !should_ignore(&to) {
            changes.clear(); // remove the generic modify events
            changes.push(format!(
                r#"{{"kind":"rename","path":"{}","newPath":"{}"}}"#,
                escape_json(&from),
                escape_json(&to)
            ));
        }
    }

    changes
}

/// Check if a path should be ignored (hidden files, node_modules, .git, etc.)
fn should_ignore(rel_path: &str) -> bool {
    for component in rel_path.split('/') {
        if component.starts_with('.') && !component.is_empty() {
            return true;
        }
        if component == "node_modules" || component == "target" || component == "__pycache__" {
            return true;
        }
    }
    false
}

/// 路径相对 base 的段是否命中 ignore 规则（node_modules/.git/target/隐藏项）。
/// base 自身（相对路径为空）不忽略。
fn is_ignored_dir(path: &Path, base: &Path) -> bool {
    path.strip_prefix(base).ok().is_some_and(|rel| should_ignore(&rel.to_string_lossy()))
}

/// 递归枚举需要注册 inotify watch 的目录，**跳过 ignore 目录及整棵子树**。
///
/// 对比 notify 的 `RecursiveMode::Recursive`：其内部 `WalkDir` 不跳过任何目录，
/// 项目含 node_modules 时会注册上万 watch（2026-08-16 正式版内存泄漏根因——watch
/// 数 1 万+ 后 notify 8.2 `handle_inotify` 内层 `read_events` 循环在持续事件下饿死
/// mio poll，`notify-rx` 线程 100% CPU、高频分配致堆膨胀）。这里用 walkdir
/// `filter_entry` 剪枝，watch 数压到实际业务目录量级。纯函数，便于单测。
fn collect_watch_dirs(base: &Path) -> Vec<PathBuf> {
    WalkDir::new(base)
        .into_iter()
        .filter_entry(|e| !is_ignored_dir(e.path(), base))
        .filter_map(|e| e.ok())
        .filter(|e| e.file_type().is_dir())
        .map(|e| e.into_path())
        .collect()
}

/// Escape a string for JSON embedding.
fn escape_json(s: &str) -> String {
    s.replace('\\', "\\\\")
        .replace('"', "\\\"")
        .replace('\n', "\\n")
        .replace('\r', "\\r")
        .replace('\t', "\\t")
}

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

    /// 测试临时目录：Drop 时递归清理。
    struct TempDir(PathBuf);
    impl TempDir {
        fn new() -> Self {
            let p = std::env::temp_dir().join(format!(
                "files_watch_test_{}_{}",
                std::process::id(),
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap()
                    .as_nanos()
            ));
            std::fs::create_dir_all(&p).unwrap();
            Self(p)
        }
    }
    impl Drop for TempDir {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.0);
        }
    }

    fn mkdir(p: &Path) {
        std::fs::create_dir_all(p).unwrap();
    }

    #[test]
    fn should_ignore_skips_hidden_and_build_dirs() {
        assert!(should_ignore(".git"));
        assert!(should_ignore(".git/config"));
        assert!(should_ignore("src/.hidden/x"));
        assert!(should_ignore("node_modules"));
        assert!(should_ignore("a/node_modules/b"));
        assert!(should_ignore("target"));
        assert!(should_ignore("__pycache__/x.py"));
        assert!(!should_ignore(""));
        assert!(!should_ignore("src"));
        assert!(!should_ignore("src/main.rs"));
    }

    #[test]
    fn is_ignored_dir_base_itself_is_not_ignored() {
        let base = PathBuf::from("/tmp/x");
        assert!(!is_ignored_dir(&base, &base));
        assert!(!is_ignored_dir(&base.join("src"), &base));
        assert!(is_ignored_dir(&base.join("node_modules").join("pkg"), &base));
        assert!(is_ignored_dir(&base.join(".git"), &base));
    }

    #[test]
    fn merge_pending_dedups_same_kind_path() {
        let mut pending: Vec<String> = Vec::new();
        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
        let change = r#"{"kind":"modify","path":"a.txt"}"#.to_string();
        assert!(!merge_pending(&mut pending, &mut seen, change.clone(), 10));
        assert!(!merge_pending(&mut pending, &mut seen, change, 10));
        assert_eq!(pending.len(), 1);
    }

    #[test]
    fn merge_pending_keeps_distinct_kind_or_path() {
        let mut pending: Vec<String> = Vec::new();
        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
        let a = r#"{"kind":"create","path":"b.txt"}"#.to_string();
        let b = r#"{"kind":"modify","path":"b.txt"}"#.to_string();
        let c = r#"{"kind":"modify","path":"c.txt"}"#.to_string();
        assert!(!merge_pending(&mut pending, &mut seen, a, 10));
        assert!(!merge_pending(&mut pending, &mut seen, b, 10));
        assert!(!merge_pending(&mut pending, &mut seen, c, 10));
        assert_eq!(pending.len(), 3);
    }

    #[test]
    fn merge_pending_flags_when_over_limit() {
        let mut pending: Vec<String> = Vec::new();
        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
        let mut over = false;
        for i in 0..4 {
            over = merge_pending(
                &mut pending,
                &mut seen,
                format!(r#"{{"kind":"modify","path":"f{}.txt"}}"#, i),
                3,
            );
        }
        assert!(over);
        // merge_pending 在 push 后检查超限并返回 true，调用方收到后负责清空；
        // 若未清空继续插入，则最后一条也会入列（故 len 可为 4）
        assert_eq!(pending.len(), 4);
    }

    #[test]
    fn collect_watch_dirs_skips_ignore_subtrees() {
        let tmp = TempDir::new();
        let base = tmp.0.clone();
        mkdir(&base.join("src").join("components"));
        mkdir(&base.join("node_modules").join("pkg"));
        mkdir(&base.join(".git"));
        mkdir(&base.join("target").join("debug"));
        mkdir(&base.join("public"));

        let dirs: Vec<String> = collect_watch_dirs(&base)
            .iter()
            .map(|p| p.strip_prefix(&base).unwrap().to_string_lossy().into_owned())
            .collect();

        assert!(dirs.contains(&"".into()), "根目录应注册: {dirs:?}");
        assert!(dirs.contains(&"src".into()), "src 应注册: {dirs:?}");
        assert!(dirs.contains(&"src/components".into()), "src/components 应注册: {dirs:?}");
        assert!(dirs.contains(&"public".into()), "public 应注册: {dirs:?}");
        assert!(
            !dirs.iter().any(|d| d.starts_with("node_modules")),
            "不应含 node_modules: {dirs:?}"
        );
        assert!(!dirs.iter().any(|d| d.starts_with(".git")), "不应含 .git: {dirs:?}");
        assert!(!dirs.iter().any(|d| d.starts_with("target")), "不应含 target: {dirs:?}");
    }
}