use notify::{Watcher, RecursiveMode, Event, EventKind, RecommendedWatcher};
use std::collections::{HashMap, VecDeque};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
const DEBOUNCE_MS: u64 = 500;
const TICK_MS: u64 = 100;
#[derive(Clone, Debug)]
pub struct ChangeRecord {
#[allow(dead_code)]
pub file: String,
#[allow(dead_code)]
pub ts: Instant,
#[allow(dead_code)]
pub outcome: String, }
#[derive(Clone)]
pub struct WatcherHandle {
#[allow(dead_code)]
pub last_changes: Arc<Mutex<VecDeque<ChangeRecord>>>,
pub workdir: PathBuf,
#[allow(dead_code)]
pub started_at: Instant,
}
impl WatcherHandle {
#[allow(dead_code)]
pub fn recent_changes(&self) -> Vec<ChangeRecord> {
self.last_changes.lock().map(|v| v.iter().cloned().collect()).unwrap_or_default()
}
}
fn is_supported(path: &Path) -> bool {
!reliary_search::is_likely_binary(path, 8192)
}
fn detect_project_root(start: &Path) -> Option<PathBuf> {
let mut cur = start.to_path_buf();
loop {
if cur.join(".reliary").join("index.sqlite").exists() {
return Some(cur);
}
if cur.join(".git").exists() {
return Some(cur);
}
if !cur.pop() {
return None;
}
}
}
struct Pending {
last: Instant,
removed: bool,
}
pub fn start_watcher(workdir: &Path) -> Result<WatcherHandle, String> {
let workdir = workdir.to_path_buf();
let project_root = detect_project_root(&workdir)
.ok_or_else(|| format!("no .reliary/index.sqlite in {} (run `reliary trust` first)", workdir.display()))?;
let watched_dir = project_root.clone();
let changes = Arc::new(Mutex::new(VecDeque::<ChangeRecord>::new()));
let _changes_for_handler = changes.clone();
let project_for_handler = project_root.clone();
let pending: Arc<Mutex<HashMap<PathBuf, Pending>>> = Arc::new(Mutex::new(HashMap::new()));
let pending_for_handler = pending.clone();
let mut watcher: RecommendedWatcher = match notify::recommended_watcher(move |res: notify::Result<Event>| {
let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
match res {
Ok(event) => record_event(&event, &project_for_handler, &pending_for_handler),
Err(e) => eprintln!("watcher error: {}", e),
}
}));
if outcome.is_err() {
eprintln!("watcher: callback panicked (event dropped, watcher continues)");
}
}) {
Ok(w) => w,
Err(e) => return Err(format!("create watcher: {}", e)),
};
if let Err(e) = watcher.watch(&watched_dir, RecursiveMode::Recursive) {
return Err(format!("watch {}: {}", watched_dir.display(), e));
}
eprintln!("watching {} (trailing debounce {}ms)", watched_dir.display(), DEBOUNCE_MS);
let started = Instant::now();
let project_for_worker = project_root.clone();
let changes_for_worker = changes.clone();
std::thread::spawn(move || {
let _w = watcher; loop {
std::thread::sleep(Duration::from_millis(TICK_MS));
let due: Vec<(PathBuf, bool)> = {
let mut map = pending.lock().unwrap_or_else(|p| p.into_inner());
let now = Instant::now();
let due_paths: Vec<PathBuf> = map
.iter()
.filter(|(_, p)| now.duration_since(p.last) >= Duration::from_millis(DEBOUNCE_MS))
.map(|(k, _)| k.clone())
.collect();
due_paths
.into_iter()
.filter_map(|k| map.remove(&k).map(|p| (k, p.removed)))
.collect()
};
for (path, removed) in due {
let path_str = path.to_string_lossy().to_string();
let content = if removed { String::new() } else {
match std::fs::read_to_string(&path) {
Ok(c) => c,
Err(e) => {
eprintln!("read {}: {}", path_str, e);
push_change(&changes_for_worker, &path_str, "error:read");
continue;
}
}
};
let db_path = project_for_worker.join(".reliary").join("index.sqlite");
let count = crate::reindex::reindex_single_file(
&db_path.to_string_lossy(),
&path_str,
&content,
);
let outcome = if removed {
format!("ok:removed({}phrases)", count)
} else {
format!("ok:{}phrases", count)
};
push_change(&changes_for_worker, &path_str, &outcome);
eprintln!("watch: reindexed {} ({})", path_str, outcome);
}
}
});
Ok(WatcherHandle {
last_changes: changes,
workdir: watched_dir,
started_at: started,
})
}
fn push_change(changes: &Arc<Mutex<VecDeque<ChangeRecord>>>, file: &str, outcome: &str) {
if let Ok(mut v) = changes.lock() {
v.push_back(ChangeRecord {
file: file.to_string(),
ts: Instant::now(),
outcome: outcome.to_string(),
});
if v.len() > 100 { v.pop_front(); }
}
}
fn record_event(
event: &Event,
project_root: &Path,
pending: &Arc<Mutex<HashMap<PathBuf, Pending>>>,
) {
let is_change = matches!(
event.kind,
EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_)
);
if !is_change {
return;
}
let removed = matches!(event.kind, EventKind::Remove(_));
for path in &event.paths {
if path.starts_with(project_root.join(".reliary")) {
continue;
}
if !removed
&& (!path.exists() || !is_supported(path)) {
continue;
}
let mut map = pending.lock().unwrap_or_else(|p| p.into_inner());
map.insert(path.clone(), Pending { last: Instant::now(), removed });
}
}