use std::path::{Path, PathBuf};
use std::sync::mpsc;
use std::time::{Duration, Instant};
use anyhow::{Context, Result};
use notify::RecursiveMode;
use notify_debouncer_full::{new_debouncer, DebouncedEvent, DebounceEventResult};
use crate::ingest::scanner::NOISE_DIRS;
use crate::ingest::parser::SUPPORTED_EXTENSIONS;
pub const COOLDOWN_QUIET_MS: u64 = 2000;
pub const COOLDOWN_DEADLINE_MS: u64 = 5000;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChangeKind {
Created,
Modified,
Deleted,
}
#[derive(Debug, Clone)]
pub struct WatchEvent {
pub paths: Vec<PathBuf>,
pub kind: ChangeKind,
}
pub fn run_watch_loop(
root: &Path,
stop_flag: std::sync::Arc<std::sync::atomic::AtomicBool>,
on_change: impl Fn(Vec<WatchEvent>) + Send + 'static,
) -> Result<()> {
let include_exts = supported_exts();
let (tx, rx) = mpsc::channel::<DebounceEventResult>();
let mut debouncer = new_debouncer(
Duration::from_millis(300),
None,
move |result: DebounceEventResult| {
let _ = tx.send(result);
},
)
.with_context(|| "创建文件防抖监听器失败")?;
let watch_roots = vec![root.to_path_buf()];
for watch_root in &watch_roots {
if !watch_root.exists() {
tracing::warn!("监听根不存在,跳过: {}", watch_root.display());
continue;
}
debouncer
.watch(watch_root.as_path(), RecursiveMode::Recursive)
.with_context(|| format!("监听目录失败: {}", watch_root.display()))?;
}
tracing::info!(
"文件监听已启动(阻塞模式): {}",
watch_roots
.iter()
.map(|p| p.display().to_string())
.collect::<Vec<_>>()
.join(", ")
);
use std::sync::atomic::Ordering;
let mut pending: Vec<(PathBuf, ChangeKind)> = Vec::new();
let mut pending_first_at: Option<Instant> = None;
let mut quiet_since: Option<Instant> = None;
loop {
if stop_flag.load(Ordering::Relaxed) {
tracing::info!("收到停止信号,文件监听退出");
return Ok(());
}
match rx.recv_timeout(Duration::from_millis(500)) {
Ok(Ok(events)) => {
let watch_events = process_batch(&events, &include_exts);
if !watch_events.is_empty() {
apply_batch(&mut pending, &watch_events);
let now = Instant::now();
pending_first_at.get_or_insert(now);
quiet_since = Some(now);
}
}
Ok(Err(errors)) => {
for e in &errors {
tracing::warn!("文件监听错误: {:?}", e);
}
}
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
if should_flush_now(&pending, &mut pending_first_at, &mut quiet_since) {
on_change(flush_events(&std::mem::take(&mut pending)));
}
}
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
}
if let Some(first_at) = pending_first_at {
let total = Instant::now().saturating_duration_since(first_at);
if should_flush(Duration::ZERO, total) {
on_change(flush_events(&std::mem::take(&mut pending)));
pending_first_at = None;
quiet_since = None;
}
}
}
Ok(())
}
fn should_flush(quiet_elapsed: Duration, total_elapsed: Duration) -> bool {
quiet_elapsed >= Duration::from_millis(COOLDOWN_QUIET_MS)
|| total_elapsed >= Duration::from_millis(COOLDOWN_DEADLINE_MS)
}
fn should_flush_now(
pending: &[(PathBuf, ChangeKind)],
pending_first_at: &mut Option<Instant>,
quiet_since: &mut Option<Instant>,
) -> bool {
if pending.is_empty() {
return false;
}
let now = Instant::now();
let quiet_elapsed = quiet_since
.map(|q| now.saturating_duration_since(q))
.unwrap_or(Duration::ZERO);
let total_elapsed = pending_first_at
.map(|f| now.saturating_duration_since(f))
.unwrap_or(Duration::ZERO);
let flush = should_flush(quiet_elapsed, total_elapsed);
if flush {
*pending_first_at = None;
*quiet_since = None;
}
flush
}
fn apply_batch(pending: &mut Vec<(PathBuf, ChangeKind)>, events: &[WatchEvent]) {
for event in events {
for p in &event.paths {
match pending.iter_mut().find(|(path, _)| path == p) {
Some(entry) => entry.1 = event.kind,
None => pending.push((p.clone(), event.kind)),
}
}
}
}
fn flush_events(pending: &[(PathBuf, ChangeKind)]) -> Vec<WatchEvent> {
let mut out: Vec<WatchEvent> = Vec::new();
for (path, kind) in pending {
match out.iter_mut().find(|e| e.kind == *kind) {
Some(ev) => ev.paths.push(path.clone()),
None => out.push(WatchEvent {
paths: vec![path.clone()],
kind: *kind,
}),
}
}
out
}
pub fn process_batch(events: &[DebouncedEvent], include_exts: &[String]) -> Vec<WatchEvent> {
fold_events(aggregate_events(events, include_exts))
}
fn fold_events(events: Vec<WatchEvent>) -> Vec<WatchEvent> {
let mut out: Vec<WatchEvent> = Vec::new();
for event in &events {
let paths: Vec<PathBuf> = event
.paths
.iter()
.filter(|p| {
if event.kind != ChangeKind::Deleted && has_path(&events, ChangeKind::Deleted, p) {
return false;
}
if event.kind == ChangeKind::Created && has_path(&events, ChangeKind::Modified, p) {
return false;
}
true
})
.cloned()
.collect();
if !paths.is_empty() {
out.push(WatchEvent { paths, kind: event.kind });
}
}
out
}
fn has_path(events: &[WatchEvent], kind: ChangeKind, path: &Path) -> bool {
events
.iter()
.any(|e| e.kind == kind && e.paths.iter().any(|p| p == path))
}
fn aggregate_events(events: &[DebouncedEvent], include_exts: &[String]) -> Vec<WatchEvent> {
let mut out: Vec<WatchEvent> = Vec::new();
for debounced in events {
let kind = change_kind_of(&debounced.event.kind);
for p in &debounced.event.paths {
if !should_report(p, include_exts) {
continue;
}
match out.iter_mut().find(|e| e.kind == kind) {
Some(ev) if !ev.paths.contains(p) => ev.paths.push(p.clone()),
Some(_) => {}
None => out.push(WatchEvent { paths: vec![p.clone()], kind }),
}
}
}
out
}
fn change_kind_of(kind: ¬ify::EventKind) -> ChangeKind {
match kind {
notify::EventKind::Create(_) => ChangeKind::Created,
notify::EventKind::Remove(_) => ChangeKind::Deleted,
_ => ChangeKind::Modified,
}
}
fn supported_exts() -> Vec<String> {
SUPPORTED_EXTENSIONS
.iter()
.map(|e| e.trim_start_matches('.').to_string())
.collect()
}
fn should_report(path: &Path, include_exts: &[String]) -> bool {
!should_ignore(path) && matches_include(path, include_exts)
}
fn should_ignore(path: &Path) -> bool {
path.components().any(|c| {
if let Some(s) = c.as_os_str().to_str() {
NOISE_DIRS.contains(&s)
} else {
false
}
})
}
fn matches_include(path: &Path, include_exts: &[String]) -> bool {
if include_exts.is_empty() {
return true;
}
match path.extension() {
Some(ext) => include_exts.iter().any(|e| ext == e.as_str()),
None => false,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_should_ignore_target_dir() {
let p = Path::new("/repo/target/debug/main.rs");
assert!(should_ignore(p));
}
#[test]
fn test_should_ignore_git_dir() {
let p = Path::new("/repo/.git/HEAD");
assert!(should_ignore(p));
}
#[test]
fn test_should_ignore_node_modules() {
let p = Path::new("/repo/node_modules/foo/index.js");
assert!(should_ignore(p));
}
#[test]
fn test_should_ignore_dist_and_venv() {
assert!(should_ignore(Path::new("/repo/dist/bundle.js")));
assert!(should_ignore(Path::new("/repo/.venv/lib/py.py")));
}
#[test]
fn test_should_not_ignore_src_dir() {
let p = Path::new("/repo/src/main.rs");
assert!(!should_ignore(p));
}
#[test]
fn test_matches_include_with_matching_ext() {
let exts = vec!["rs".to_string(), "tsx".to_string()];
assert!(matches_include(Path::new("main.rs"), &exts));
assert!(matches_include(Path::new("comp.tsx"), &exts));
}
#[test]
fn test_matches_include_with_mismatch_ext() {
let exts = vec!["rs".to_string()];
assert!(!matches_include(Path::new("main.js"), &exts));
assert!(!matches_include(Path::new("no_ext"), &exts));
}
#[test]
fn test_supported_exts_cover_all_parsers() {
let exts = supported_exts();
for expected in ["rs", "ts", "tsx", "py", "go", "js", "jsx", "mjs", "cjs", "cs", "java"] {
assert!(exts.contains(&expected.to_string()), "缺少 {expected}");
}
}
#[test]
fn test_should_report_filters_ignored_and_mismatched() {
let exts = vec!["rs".to_string()];
assert!(should_report(Path::new("/repo/src/main.rs"), &exts));
assert!(!should_report(Path::new("/repo/target/main.rs"), &exts));
assert!(!should_report(Path::new("/repo/src/main.js"), &exts));
assert!(!should_report(Path::new("/repo/src/no_ext"), &exts));
}
#[test]
fn test_watch_event_kind_preserved() {
use notify::event::{DataChange, ModifyKind, RemoveKind};
use notify::{Event, EventKind};
let mk = || {
let mut e = Event::new(EventKind::Modify(ModifyKind::Data(DataChange::Content)));
e.paths = vec![PathBuf::from("src/a.rs")];
DebouncedEvent::new(e, std::time::Instant::now())
};
let mut removed = Event::new(EventKind::Remove(RemoveKind::File));
removed.paths = vec![PathBuf::from("src/b.rs")];
let events = vec![
mk(),
DebouncedEvent::new(removed, std::time::Instant::now()),
];
let exts = vec!["rs".to_string()];
let aggregated = aggregate_events(&events, &exts);
assert_eq!(aggregated.len(), 2, "Modify 与 Remove 应各自聚合成独立事件");
let modified = aggregated
.iter()
.find(|e| e.kind == ChangeKind::Modified)
.expect("应存在 Modified 事件");
assert_eq!(modified.paths, vec![PathBuf::from("src/a.rs")]);
let deleted = aggregated
.iter()
.find(|e| e.kind == ChangeKind::Deleted)
.expect("应存在 Deleted 事件");
assert_eq!(deleted.paths, vec![PathBuf::from("src/b.rs")]);
}
#[test]
fn test_aggregate_events_dedups_same_kind_paths() {
use notify::event::{DataChange, ModifyKind};
use notify::{Event, EventKind};
let mk = || {
let mut e = Event::new(EventKind::Modify(ModifyKind::Data(DataChange::Content)));
e.paths = vec![PathBuf::from("src/a.rs")];
DebouncedEvent::new(e, std::time::Instant::now())
};
let exts = vec!["rs".to_string()];
let aggregated = aggregate_events(&[mk(), mk()], &exts);
assert_eq!(aggregated.len(), 1);
assert_eq!(aggregated[0].paths, vec![PathBuf::from("src/a.rs")]);
}
fn make_debounced(kind: notify::EventKind, path: &str) -> DebouncedEvent {
let mut e = notify::Event::new(kind);
e.paths = vec![PathBuf::from(path)];
DebouncedEvent::new(e, std::time::Instant::now())
}
#[test]
fn test_fold_modified_deleted() {
use notify::event::{DataChange, ModifyKind, RemoveKind};
let exts = vec!["rs".to_string()];
let events = vec![
make_debounced(notify::EventKind::Modify(ModifyKind::Data(DataChange::Content)), "src/a.rs"),
make_debounced(notify::EventKind::Remove(RemoveKind::File), "src/a.rs"),
];
let folded = process_batch(&events, &exts);
assert_eq!(folded.len(), 1, "同路径 Modified+Deleted 应折叠为单事件");
assert_eq!(folded[0].kind, ChangeKind::Deleted);
assert_eq!(folded[0].paths, vec![PathBuf::from("src/a.rs")]);
}
#[test]
fn test_fold_created_deleted() {
use notify::event::{CreateKind, RemoveKind};
let exts = vec!["rs".to_string()];
let events = vec![
make_debounced(notify::EventKind::Create(CreateKind::File), "src/a.rs"),
make_debounced(notify::EventKind::Remove(RemoveKind::File), "src/a.rs"),
];
let folded = process_batch(&events, &exts);
assert_eq!(folded.len(), 1);
assert_eq!(folded[0].kind, ChangeKind::Deleted);
}
#[test]
fn test_fold_created_modified() {
use notify::event::{CreateKind, DataChange, ModifyKind};
let exts = vec!["rs".to_string()];
let events = vec![
make_debounced(notify::EventKind::Create(CreateKind::File), "src/a.rs"),
make_debounced(notify::EventKind::Modify(ModifyKind::Data(DataChange::Content)), "src/a.rs"),
];
let folded = process_batch(&events, &exts);
assert_eq!(folded.len(), 1);
assert_eq!(folded[0].kind, ChangeKind::Modified);
assert_eq!(folded[0].paths, vec![PathBuf::from("src/a.rs")]);
}
#[test]
fn test_aggregate_events_same_path_cross_kind() {
use notify::event::{DataChange, ModifyKind, RemoveKind};
let exts = vec!["rs".to_string()];
let events = vec![
make_debounced(notify::EventKind::Modify(ModifyKind::Data(DataChange::Content)), "src/a.rs"),
make_debounced(notify::EventKind::Remove(RemoveKind::File), "src/a.rs"),
];
let folded = process_batch(&events, &exts);
assert_eq!(folded.len(), 1);
assert_eq!(folded[0].kind, ChangeKind::Deleted);
}
#[test]
fn test_aggregate_events_preserves_distinct_paths() {
use notify::event::{DataChange, ModifyKind, RemoveKind};
let exts = vec!["rs".to_string()];
let events = vec![
make_debounced(notify::EventKind::Modify(ModifyKind::Data(DataChange::Content)), "src/a.rs"),
make_debounced(notify::EventKind::Remove(RemoveKind::File), "src/a.rs"),
make_debounced(notify::EventKind::Modify(ModifyKind::Data(DataChange::Content)), "src/b.rs"),
];
let folded = process_batch(&events, &exts);
assert_eq!(folded.len(), 2, "a 折叠为 Deleted、b 独立 Modified,共 2 事件");
let deleted = folded
.iter()
.find(|e| e.kind == ChangeKind::Deleted)
.expect("应存在 Deleted 事件");
assert_eq!(deleted.paths, vec![PathBuf::from("src/a.rs")]);
let modified = folded
.iter()
.find(|e| e.kind == ChangeKind::Modified)
.expect("应存在 Modified 事件");
assert_eq!(modified.paths, vec![PathBuf::from("src/b.rs")]);
}
#[test]
fn test_watch_loop_exits_on_pre_set_stop_flag() {
let dir = std::env::temp_dir().join(format!("code_repo_wiki_watch_stop_{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
let stop_flag = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
let start = std::time::Instant::now();
let result = run_watch_loop(&dir, stop_flag, |_| panic!("不应触发回调"));
assert!(result.is_ok(), "优雅退出应返回 Ok: {result:?}");
assert!(
start.elapsed() < std::time::Duration::from_secs(5),
"预置停止标记应在监听启动后立即退出(无需等待事件)"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn test_should_flush_quiet_elapsed_reaches_threshold() {
assert!(
should_flush(
Duration::from_millis(COOLDOWN_QUIET_MS),
Duration::from_millis(500)
),
"安静 2s 应触发(尾沿)"
);
assert!(
should_flush(
Duration::from_millis(3000),
Duration::from_millis(3000)
),
"安静 3s 应触发"
);
}
#[test]
fn test_should_flush_deadline_forced() {
assert!(
should_flush(
Duration::from_millis(300),
Duration::from_millis(COOLDOWN_DEADLINE_MS)
),
"总时长 5s 应强制触发(编辑未停也触发)"
);
}
#[test]
fn test_should_flush_within_cooldown_does_not_trigger() {
assert!(
!should_flush(
Duration::from_millis(1500),
Duration::from_millis(1500)
),
"编辑未停且未到 5s 上限不应触发"
);
assert!(
!should_flush(Duration::ZERO, Duration::ZERO),
"刚收到事件不应触发"
);
}
#[test]
fn test_apply_batch_dedups_and_combines() {
let mut pending: Vec<(PathBuf, ChangeKind)> = Vec::new();
let batch1 = vec![WatchEvent {
paths: vec![PathBuf::from("src/a.rs")],
kind: ChangeKind::Modified,
}];
let batch2 = vec![
WatchEvent {
paths: vec![PathBuf::from("src/a.rs"), PathBuf::from("src/b.rs")],
kind: ChangeKind::Modified,
},
WatchEvent {
paths: vec![PathBuf::from("src/c.rs")],
kind: ChangeKind::Deleted,
},
];
apply_batch(&mut pending, &batch1);
apply_batch(&mut pending, &batch2);
let flushed = flush_events(&pending);
assert_eq!(flushed.len(), 2, "同 kind 合并为 1 组 + Deleted 1 组");
let modified = flushed
.iter()
.find(|e| e.kind == ChangeKind::Modified)
.expect("应存在 Modified 组");
assert_eq!(
modified.paths,
vec![PathBuf::from("src/a.rs"), PathBuf::from("src/b.rs")],
"a 去重、b 追加"
);
assert!(flushed.iter().any(|e| e.kind == ChangeKind::Deleted));
}
#[test]
fn test_apply_batch_later_kind_overwrites() {
let mut pending: Vec<(PathBuf, ChangeKind)> = Vec::new();
apply_batch(
&mut pending,
&[WatchEvent {
paths: vec![PathBuf::from("src/a.rs")],
kind: ChangeKind::Modified,
}],
);
apply_batch(
&mut pending,
&[WatchEvent {
paths: vec![PathBuf::from("src/a.rs"), PathBuf::from("src/b.rs")],
kind: ChangeKind::Deleted,
}],
);
let flushed = flush_events(&pending);
assert_eq!(flushed.len(), 1, "跨批 Modified+Deleted 收敛为单个事件");
assert_eq!(flushed[0].kind, ChangeKind::Deleted);
assert_eq!(
flushed[0].paths,
vec![PathBuf::from("src/a.rs"), PathBuf::from("src/b.rs")]
);
}
#[test]
fn test_apply_batch_delete_then_recreate_keeps_created() {
let mut pending: Vec<(PathBuf, ChangeKind)> = Vec::new();
apply_batch(
&mut pending,
&[WatchEvent {
paths: vec![PathBuf::from("src/a.rs")],
kind: ChangeKind::Deleted,
}],
);
apply_batch(
&mut pending,
&[WatchEvent {
paths: vec![PathBuf::from("src/a.rs")],
kind: ChangeKind::Created,
}],
);
let flushed = flush_events(&pending);
assert_eq!(flushed.len(), 1);
assert_eq!(
flushed[0].kind,
ChangeKind::Created,
"删除重建必须收敛为 Created(文件最终存在),否则下游误删产物页"
);
assert_eq!(flushed[0].paths, vec![PathBuf::from("src/a.rs")]);
}
#[test]
fn test_apply_batch_accumulates_across_batches() {
let mut pending: Vec<(PathBuf, ChangeKind)> = Vec::new();
apply_batch(
&mut pending,
&[WatchEvent {
paths: vec![PathBuf::from("src/a.rs")],
kind: ChangeKind::Modified,
}],
);
apply_batch(
&mut pending,
&[WatchEvent {
paths: vec![PathBuf::from("src/b.rs")],
kind: ChangeKind::Modified,
}],
);
apply_batch(
&mut pending,
&[WatchEvent {
paths: vec![PathBuf::from("src/c.rs")],
kind: ChangeKind::Modified,
}],
);
let flushed = flush_events(&pending);
assert_eq!(flushed.len(), 1);
assert_eq!(
flushed[0].paths,
vec![
PathBuf::from("src/a.rs"),
PathBuf::from("src/b.rs"),
PathBuf::from("src/c.rs")
],
"三批同 kind 路径应全部累积"
);
}
}