From ae9e806e6188a51eae39580120e18846c16cc958 Mon Sep 17 00:00:00 2001 From: LyAhn Date: Mon, 8 Jun 2026 06:50:01 +0100 Subject: [PATCH] feat(watcher): adaptive filesystem watchdog with zero CPU when idle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Monitors all registered folders using OS-native events (ReadDirectoryChangesW on Windows) so new, modified, and deleted files are reflected in the gallery automatically without manual reindexing. Key design points: - Adaptive blocking: recv() when no events pending (zero CPU), switches to recv_timeout(earliest_deadline) only when debounce timers are running — wakes exactly when the soonest event is ready, no busy-polling - 500 ms per-path debounce coalesces rapid OS event bursts into one action - Change detection preserved: build_record skips upsert if file_size + mtime unchanged, preventing spurious thumbnail/embedding re-queues and avoiding clobbering of existing metadata like thumbnail_path - Access events (reads) filtered out; only Create/Modify/Remove trigger work - Deletion path: emits watcher-deleted event with image IDs; frontend removes those images from state and clears selectedImage if it was deleted WatcherHandle stored in app state; add_folder / remove_folder / update_folder_path commands keep the watched path set in sync with the DB. --- src-tauri/Cargo.lock | 92 +++++++++++++++- src-tauri/Cargo.toml | 1 + src-tauri/src/commands.rs | 29 ++++- src-tauri/src/db.rs | 38 +++++++ src-tauri/src/indexer.rs | 221 +++++++++++++++++++++++++++++++++++++- src-tauri/src/lib.rs | 3 + src/store.ts | 19 ++++ 7 files changed, 396 insertions(+), 7 deletions(-) diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index bc2e9f4..5e8dbe0 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -1750,6 +1750,15 @@ dependencies = [ "winapi", ] +[[package]] +name = "fsevent-sys" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2" +dependencies = [ + "libc", +] + [[package]] name = "futf" version = "0.1.5" @@ -2932,6 +2941,26 @@ dependencies = [ "cfb", ] +[[package]] +name = "inotify" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8069d3ec154eb856955c1c0fbffefbf5f3c40a104ec912d4797314c1801abff" +dependencies = [ + "bitflags 1.3.2", + "inotify-sys", + "libc", +] + +[[package]] +name = "inotify-sys" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e05c02b5e89bff3b946cedeca278abc628fe811e604f027c45a8aa3cf793d0eb" +dependencies = [ + "libc", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -3133,6 +3162,26 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "kqueue" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "273c0752728918e0ac4976f2b275b6fefb9ecd400585dec929419f3844cd87b5" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087" +dependencies = [ + "bitflags 2.11.0", + "libc", +] + [[package]] name = "kuchikiki" version = "0.8.8-speedreader" @@ -3438,6 +3487,18 @@ dependencies = [ "simd-adler32", ] +[[package]] +name = "mio" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c" +dependencies = [ + "libc", + "log", + "wasi 0.11.1+wasi-snapshot-preview1", + "windows-sys 0.48.0", +] + [[package]] name = "mio" version = "1.2.0" @@ -3621,6 +3682,25 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "notify" +version = "6.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6205bd8bb1e454ad2e27422015fb5e4f2bcc7e08fa8f27058670d208324a4d2d" +dependencies = [ + "bitflags 2.11.0", + "crossbeam-channel", + "filetime", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio 0.8.11", + "walkdir", + "windows-sys 0.48.0", +] + [[package]] name = "notify-rust" version = "4.17.0" @@ -4308,6 +4388,7 @@ dependencies = [ "kamadak-exif", "log", "memmap2", + "notify", "ort", "r2d2", "r2d2_sqlite", @@ -6392,7 +6473,7 @@ checksum = "2bd1c4c0fc4a7ab90fc15ef6daaa3ec3b893f004f915f2392557ed23237820cd" dependencies = [ "bytes", "libc", - "mio", + "mio 1.2.0", "parking_lot", "pin-project-lite", "signal-hook-registry", @@ -7519,6 +7600,15 @@ dependencies = [ "windows-targets 0.42.2", ] +[[package]] +name = "windows-sys" +version = "0.48.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" +dependencies = [ + "windows-targets 0.48.5", +] + [[package]] name = "windows-sys" version = "0.52.0" diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index e79d1a3..6c201c7 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -51,6 +51,7 @@ ureq = { version = "3.3.0", default-features = false, features = ["native-tls"] zip = { version = "4.6.1", default-features = false, features = ["deflate"] } csv = "1" kamadak-exif = "0.5" +notify = "6" tauri-plugin-notification = "2" # ── Dev-mode performance ──────────────────────────────────────────────────── diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index fbeef49..1eabb3b 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -5,7 +5,7 @@ use crate::captioner::{ use crate::db::{self, DbPool, ExploreTagEntry, Folder, FolderJobProgress, ImageRecord, ImageTag}; use crate::embedder; use crate::hnsw_index; -use crate::indexer; +use crate::indexer::{self, WatcherHandle}; use crate::tagger::{self, TaggerAcceleration, TaggerModelStatus, TaggerRuntimeProbe}; use crate::vector; use serde::{Deserialize, Serialize}; @@ -202,6 +202,7 @@ pub struct GetImagesByIdsParams { pub async fn add_folder( app: AppHandle, db: State<'_, DbState>, + watcher: State<'_, WatcherHandle>, path: String, ) -> Result { let folder_path = PathBuf::from(&path); @@ -230,6 +231,7 @@ pub async fn add_folder( .find(|f| f.id == folder_id) .ok_or("Folder not found after insert")?; + watcher.add_folder(folder_path.clone(), folder_id); indexer::index_folder(app, db.inner().clone(), folder_id, folder_path); Ok(folder) @@ -250,9 +252,23 @@ pub async fn get_background_job_progress( } #[tauri::command] -pub async fn remove_folder(db: State<'_, DbState>, folder_id: i64) -> Result<(), String> { +pub async fn remove_folder( + db: State<'_, DbState>, + watcher: State<'_, WatcherHandle>, + folder_id: i64, +) -> Result<(), String> { let conn = db.get().map_err(|e| e.to_string())?; - db::delete_folder(&conn, folder_id).map_err(|e| e.to_string()) + // Capture the path before deletion so we can unregister the watcher. + let folder_path = db::get_folders(&conn) + .map_err(|e| e.to_string())? + .into_iter() + .find(|f| f.id == folder_id) + .map(|f| PathBuf::from(f.path)); + db::delete_folder(&conn, folder_id).map_err(|e| e.to_string())?; + if let Some(path) = folder_path { + watcher.remove_folder(&path); + } + Ok(()) } #[tauri::command] @@ -352,6 +368,7 @@ pub async fn rename_folder( pub async fn update_folder_path( app: AppHandle, db: State<'_, DbState>, + watcher: State<'_, WatcherHandle>, folder_id: i64, new_path: String, ) -> Result<(), String> { @@ -363,7 +380,7 @@ pub async fn update_folder_path( .file_name() .map(|n| n.to_string_lossy().to_string()) .unwrap_or_else(|| new_path.clone()); - { + let old_path = { let conn = db.get().map_err(|e| e.to_string())?; // Fetch the old path before updating so image paths can be rewritten. let old_path = db::get_folders(&conn) @@ -374,7 +391,9 @@ pub async fn update_folder_path( .ok_or("Folder not found")?; db::update_folder_path(&conn, folder_id, &old_path, &new_path, &new_name) .map_err(|e| e.to_string())?; - } + old_path + }; + watcher.update_folder(&PathBuf::from(old_path), new_path_buf.clone(), folder_id); indexer::index_folder(app, db.inner().clone(), folder_id, new_path_buf); Ok(()) } diff --git a/src-tauri/src/db.rs b/src-tauri/src/db.rs index 26f408d..e5454ea 100644 --- a/src-tauri/src/db.rs +++ b/src-tauri/src/db.rs @@ -1352,6 +1352,44 @@ pub fn update_image_details( .map_err(Into::into) } +/// Look up the lightweight indexed-media entry for a single path. +/// Used by the filesystem watcher to run change-detection before upserting. +pub fn get_indexed_entry_by_path(conn: &Connection, path: &str) -> Result> { + let result = conn.query_row( + "SELECT id, path, modified_at, file_size, media_kind FROM images WHERE path = ?1", + [path], + |row| { + Ok(IndexedMediaEntry { + id: row.get(0)?, + path: row.get(1)?, + modified_at: row.get(2)?, + file_size: row.get(3)?, + media_kind: row.get(4)?, + }) + }, + ); + match result { + Ok(entry) => Ok(Some(entry)), + Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None), + Err(e) => Err(e.into()), + } +} + +/// Look up just the image id for a path. Used by the filesystem watcher +/// to find the DB row to delete when a file is removed from disk. +pub fn get_image_id_by_path(conn: &Connection, path: &str) -> Result> { + let result = conn.query_row( + "SELECT id FROM images WHERE path = ?1", + [path], + |row| row.get(0), + ); + match result { + Ok(id) => Ok(Some(id)), + Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None), + Err(e) => Err(e.into()), + } +} + pub fn get_image_by_id(conn: &Connection, image_id: i64) -> Result { conn.query_row( "SELECT id, folder_id, path, filename, thumbnail_path, width, height, file_size, created_at, modified_at, taken_at, mime_type, diff --git a/src-tauri/src/indexer.rs b/src-tauri/src/indexer.rs index eda6d06..2ac7ed4 100644 --- a/src-tauri/src/indexer.rs +++ b/src-tauri/src/indexer.rs @@ -9,9 +9,10 @@ use crate::vector; use anyhow::Result; use rayon::prelude::*; use serde::Serialize; +use notify::{RecommendedWatcher, RecursiveMode, Watcher}; use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; -use std::sync::{Mutex, OnceLock}; +use std::sync::{Arc, Mutex, OnceLock}; use std::time::{Duration, Instant}; use tauri::{AppHandle, Emitter}; use walkdir::WalkDir; @@ -1282,3 +1283,221 @@ fn mime_for_ext(ext: &str) -> &'static str { _ => "image/jpeg", } } + +// ── Filesystem watcher ──────────────────────────────────────────────────────── + +/// How long to wait after the last event for a path before processing it. +/// Absorbs bursts of OS events that accompany a single logical file write. +const WATCHER_DEBOUNCE: Duration = Duration::from_millis(500); + +struct WatcherInner { + watcher: Mutex, + /// Maps each watched folder root → its folder_id in the DB. + folder_map: Arc>>, +} + +/// Shared handle that lets command handlers register and deregister watched +/// directories without touching the debounce thread directly. +#[derive(Clone)] +pub struct WatcherHandle { + inner: Arc, +} + +impl WatcherHandle { + pub fn add_folder(&self, path: PathBuf, folder_id: i64) { + { + let mut w = self.inner.watcher.lock().unwrap(); + if path.is_dir() { + if let Err(e) = w.watch(&path, RecursiveMode::Recursive) { + eprintln!("Watcher: failed to watch {:?}: {}", path, e); + } + } + } + self.inner.folder_map.lock().unwrap().insert(path, folder_id); + } + + pub fn remove_folder(&self, path: &Path) { + { + let mut w = self.inner.watcher.lock().unwrap(); + let _ = w.unwatch(path); + } + self.inner.folder_map.lock().unwrap().remove(path); + } + + pub fn update_folder(&self, old_path: &Path, new_path: PathBuf, folder_id: i64) { + { + let mut w = self.inner.watcher.lock().unwrap(); + let _ = w.unwatch(old_path); + if new_path.is_dir() { + if let Err(e) = w.watch(&new_path, RecursiveMode::Recursive) { + eprintln!("Watcher: failed to watch {:?}: {}", new_path, e); + } + } + } + let mut map = self.inner.folder_map.lock().unwrap(); + map.remove(old_path); + map.insert(new_path, folder_id); + } +} + +/// Start the filesystem watcher. Watches all folders currently in the DB and +/// returns a handle that command handlers can use to add/remove watched paths. +/// +/// The debounce loop uses adaptive blocking: +/// - `recv()` when no events are pending — zero CPU +/// - `recv_timeout(earliest_deadline)` when events are pending — wakes exactly +/// when the soonest debounce window expires, no busy-polling +pub fn start_watcher(app: AppHandle, pool: DbPool) -> WatcherHandle { + let (tx, rx) = std::sync::mpsc::channel::>(); + + let raw_watcher = notify::recommended_watcher(move |result| { + let _ = tx.send(result); + }) + .expect("Failed to create filesystem watcher"); + + // Seed the folder map from the DB so existing folders are watched on startup. + let folder_map: Arc>> = Arc::new(Mutex::new(HashMap::new())); + { + let conn = pool.get().expect("Watcher: failed to get DB connection for init"); + let folders = db::get_folders(&conn).unwrap_or_default(); + let mut map = folder_map.lock().unwrap(); + for f in folders { + map.insert(PathBuf::from(f.path), f.id); + } + } + + // Register each known folder with the OS watcher. + let raw_watcher = { + let mut w = raw_watcher; + let map = folder_map.lock().unwrap(); + for path in map.keys() { + if path.is_dir() { + if let Err(e) = w.watch(path, RecursiveMode::Recursive) { + eprintln!("Watcher: failed to watch {:?}: {}", path, e); + } + } + } + w + }; + + let handle = WatcherHandle { + inner: Arc::new(WatcherInner { + watcher: Mutex::new(raw_watcher), + folder_map: Arc::clone(&folder_map), + }), + }; + + // Spawn the debounce loop on its own thread. + let folder_map_thread = Arc::clone(&folder_map); + std::thread::spawn(move || { + // path → deadline: the earliest instant at which this path should be processed. + let mut pending: HashMap = HashMap::new(); + + loop { + // Adaptive blocking: block forever when idle, wake at the earliest + // deadline when events are queued. + let received = if pending.is_empty() { + match rx.recv() { + Ok(e) => Some(e), + Err(_) => break, // sender dropped — app is shutting down + } + } else { + let earliest = pending.values().copied().min().unwrap(); // safe: non-empty + let timeout = earliest.saturating_duration_since(Instant::now()); + match rx.recv_timeout(timeout) { + Ok(e) => Some(e), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => None, + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, + } + }; + + let now = Instant::now(); + + // Absorb incoming event — coalesces rapid writes into one deadline. + if let Some(Ok(event)) = received { + use notify::EventKind; + // Skip pure access events (reads); they never change file content. + if !matches!(event.kind, EventKind::Access(_)) { + for path in event.paths { + if is_supported_media(&path) { + pending.insert(path, now + WATCHER_DEBOUNCE); + } + } + } + } + + // Process all paths whose debounce window has expired. + let ready: Vec = pending + .iter() + .filter(|(_, &deadline)| deadline <= now) + .map(|(p, _)| p.clone()) + .collect(); + + for path in ready { + pending.remove(&path); + process_watcher_path(&app, &pool, &folder_map_thread, &path); + } + } + }); + + handle +} + +/// Decide what to do with a path whose debounce window just expired. +/// If the file exists → upsert (with change-detection to avoid clobbering +/// metadata like thumbnail_path). If the file is gone → delete from DB. +fn process_watcher_path( + app: &AppHandle, + pool: &DbPool, + folder_map: &Arc>>, + path: &Path, +) { + // Determine which registered folder owns this file. + let folder_id = { + let map = folder_map.lock().unwrap(); + map.iter() + .find(|(folder_path, _)| path.starts_with(folder_path.as_path())) + .map(|(_, &id)| id) + }; + let Some(folder_id) = folder_id else { return }; + + let conn = match pool.get() { + Ok(c) => c, + Err(e) => { + eprintln!("Watcher: DB pool error: {}", e); + return; + } + }; + + if path.exists() { + // File still on disk — upsert if content changed. + let path_str = path.to_string_lossy(); + let existing = db::get_indexed_entry_by_path(&conn, &path_str).unwrap_or(None); + let Some(record) = build_record(path, folder_id, existing.as_ref()) else { + return; // file unchanged (same size + mtime) — nothing to do + }; + drop(conn); // commit_batch acquires its own connection from the pool + + match commit_batch(pool, &[record]) { + Ok(committed) if !committed.is_empty() => { + emit_images(app, &IndexedImagesBatch { folder_id, images: committed }); + emit_folder_job_progress(app, pool, &[folder_id], false); + } + Ok(_) => {} + Err(e) => eprintln!("Watcher: commit error for {:?}: {}", path, e), + } + } else { + // File removed from disk — clean up DB row. + let path_str = path.to_string_lossy(); + match db::get_image_id_by_path(&conn, &path_str) { + Ok(Some(image_id)) => { + if db::delete_images_by_ids(&conn, &[image_id]).is_ok() { + db::update_folder_count(&conn, folder_id).ok(); + let _ = app.emit("watcher-deleted", vec![image_id]); + } + } + Ok(None) => {} // never indexed or already removed + Err(e) => eprintln!("Watcher: lookup error for {:?}: {}", path, e), + } + } +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 009851f..bb8a736 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -74,8 +74,11 @@ pub fn run() { // indexer::start_caption_worker(app.handle().clone(), pool.clone(), app_dir.clone()); indexer::start_tagging_worker(app.handle().clone(), pool.clone(), app_dir.clone()); + let watcher_handle = indexer::start_watcher(app.handle().clone(), pool.clone()); + app.manage(pool); app.manage(media_tools); + app.manage(watcher_handle); Ok(()) }) diff --git a/src/store.ts b/src/store.ts index af0f2dc..96fdd27 100644 --- a/src/store.ts +++ b/src/store.ts @@ -1747,6 +1747,24 @@ export const useGalleryStore = create((set, get) => ({ }); }); + const unlistenWatcherDeleted = await listen("watcher-deleted", (event) => { + const deletedIds = new Set(event.payload); + set((state) => { + const removed = state.images.filter((img) => deletedIds.has(img.id)).length; + const images = state.images.filter((img) => !deletedIds.has(img.id)); + const selectedImage = + state.selectedImage && deletedIds.has(state.selectedImage.id) + ? null + : state.selectedImage; + return { + images, + totalImages: Math.max(0, state.totalImages - removed), + loadedCount: Math.max(0, state.loadedCount - removed), + selectedImage, + }; + }); + }); + return () => { unlistenProgress(); unlistenMediaJobs(); @@ -1754,6 +1772,7 @@ export const useGalleryStore = create((set, get) => ({ unlistenTaggerModelProgress(); unlistenImages(); unlistenThumbnails(); + unlistenWatcherDeleted(); }; }, }));