mimick/
monitor.rs

1//! Provides live filesystem monitoring, file-settling checks, and checksum generation.
2//!
3//! Uses `notify` watchers (inotify on Linux, polling under Flatpak) to
4//! detect new or modified files in watch-path directories. Files are
5//! settled via a debounce window before being queued, and checksums are
6//! computed using BLAKE3 with chunked streaming to avoid large allocations.
7
8use crate::config::{FolderSyncMethod, WatchPathEntry, best_matching_watch_entry};
9use crate::media_kinds;
10use crate::watch_path_display::display_watch_path;
11use notify::{Config as NotifyConfig, EventKind, PollWatcher, RecursiveMode, Watcher};
12use sha1::{Digest, Sha1};
13use std::collections::HashMap;
14use std::fs;
15use std::io::{self, BufReader, Read};
16use std::path::{Path, PathBuf};
17use std::time::{Duration, Instant};
18use tokio::sync::mpsc;
19
20/// Number of consecutive stable size checks required before a file is considered complete.
21const REQUIRED_STABLE_COUNTS: u32 = 3;
22const CHECK_INTERVAL_MS: u64 = 1000;
23const IDLE_TIMEOUT_SECS: u64 = 300;
24const FLATPAK_POLL_INTERVAL_MS: u64 = 2000;
25
26/// File monitor that coordinates watching directories and reporting events.
27pub struct Monitor {
28    /// Watch folders entry list configuration.
29    watch_paths: Vec<WatchPathEntry>,
30    /// State flag whether background sync is active.
31    background_sync_enabled: bool,
32}
33
34#[derive(Debug, Clone)]
35pub enum MonitorEvent {
36    /// Emitted when a file is fully populated and checksummed.
37    Ready {
38        /// Absolute path of the stabilized file.
39        path: String,
40        /// File SHA-1 checksum.
41        checksum: String,
42        /// Companion XMP sidecar path, if one was found on disk.
43        sidecar_path: Option<String>,
44    },
45    /// Emitted when a watched file is deleted.
46    Deleted {
47        /// Absolute path of the deleted file.
48        path: String,
49    },
50}
51
52/// Monitor commands passed down the main event loop.
53enum MonitorCommand {
54    /// Replace the live watch folders and settings configuration.
55    ReplaceWatchPaths {
56        /// Updated watch folder entry list.
57        watch_paths: Vec<WatchPathEntry>,
58        /// State flag whether background sync is active.
59        background_sync_enabled: bool,
60    },
61}
62
63#[derive(Clone)]
64pub struct MonitorHandle {
65    /// Command sender channel pointing at the main loop.
66    command_tx: std::sync::mpsc::Sender<MonitorCommand>,
67}
68
69impl Monitor {
70    /// Initialize a new file Monitor.
71    pub fn new(watch_paths: Vec<WatchPathEntry>, background_sync_enabled: bool) -> Self {
72        Self {
73            watch_paths,
74            background_sync_enabled,
75        }
76    }
77
78    /// Start the watcher thread and emit file-ready and deletion events.
79    ///
80    /// Hashing is offloaded to a bounded worker pool (N = num_cpus / 2, capped at 4)
81    /// so bursts of file events don't serialise on hash latency.
82    pub fn start(&self, tx: mpsc::Sender<MonitorEvent>) -> MonitorHandle {
83        let watch_paths = self.watch_paths.clone();
84        let background_sync_enabled = self.background_sync_enabled;
85        let handle = tokio::runtime::Handle::current();
86        let (command_tx, command_rx) = std::sync::mpsc::channel();
87
88        let worker_count = (num_cpus::get() / 2).clamp(1, 4);
89        let (work_tx, work_rx) = tokio::sync::mpsc::channel::<String>(32);
90
91        let active_tasks: std::sync::Arc<parking_lot::Mutex<std::collections::HashSet<String>>> =
92            std::sync::Arc::new(parking_lot::Mutex::new(std::collections::HashSet::new()));
93
94        spawn_hash_workers(
95            &handle,
96            worker_count,
97            work_rx,
98            tx.clone(),
99            active_tasks.clone(),
100        );
101        log::info!("Started {} hash worker(s) for file monitor", worker_count);
102
103        std::thread::spawn(move || {
104            let (notify_tx, notify_rx) = std::sync::mpsc::channel();
105            let mut watcher = match create_watcher(notify_tx) {
106                Ok(w) => w,
107                Err(e) => {
108                    log::error!("Failed to create file watcher: {:?}", e);
109                    return;
110                }
111            };
112
113            let mut watch_paths = watch_paths;
114            let mut background_sync_enabled = background_sync_enabled;
115            let mut watched_roots = Vec::<PathBuf>::new();
116            replace_watches(
117                &mut *watcher,
118                &mut watched_roots,
119                &watch_paths,
120                background_sync_enabled,
121            );
122
123            let mut debounce_map: HashMap<String, Instant> = HashMap::new();
124
125            loop {
126                while let Ok(command) = command_rx.try_recv() {
127                    match command {
128                        MonitorCommand::ReplaceWatchPaths {
129                            watch_paths: new_paths,
130                            background_sync_enabled: new_background_sync_enabled,
131                        } => {
132                            watch_paths = new_paths;
133                            background_sync_enabled = new_background_sync_enabled;
134                            replace_watches(
135                                &mut *watcher,
136                                &mut watched_roots,
137                                &watch_paths,
138                                background_sync_enabled,
139                            );
140                        }
141                    }
142                }
143
144                let res = match notify_rx.recv_timeout(Duration::from_millis(500)) {
145                    Ok(res) => res,
146                    Err(std::sync::mpsc::RecvTimeoutError::Timeout) => continue,
147                    Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
148                };
149
150                match res {
151                    Ok(event) => handle_notify_event(
152                        event,
153                        &watch_paths,
154                        &tx,
155                        &work_tx,
156                        &active_tasks,
157                        &mut debounce_map,
158                    ),
159                    Err(e) => log::error!("Watch error: {:?}", e),
160                }
161            }
162
163            log::warn!("File watcher thread exiting.");
164        });
165
166        MonitorHandle { command_tx }
167    }
168}
169
170fn spawn_hash_workers(
171    handle: &tokio::runtime::Handle,
172    worker_count: usize,
173    work_rx: tokio::sync::mpsc::Receiver<String>,
174    tx_out: mpsc::Sender<MonitorEvent>,
175    active_tasks: std::sync::Arc<parking_lot::Mutex<std::collections::HashSet<String>>>,
176) {
177    let work_rx = std::sync::Arc::new(tokio::sync::Mutex::new(work_rx));
178    for i in 0..worker_count {
179        let rx = work_rx.clone();
180        let tx_out = tx_out.clone();
181        let active = active_tasks.clone();
182        handle.spawn(async move {
183            loop {
184                let path_str = {
185                    let mut guard = rx.lock().await;
186                    match guard.recv().await {
187                        Some(p) => p,
188                        None => break,
189                    }
190                };
191
192                if wait_for_file_completion(&path_str).await {
193                    hash_and_emit(&path_str, &tx_out).await;
194                } else {
195                    log::warn!("File never stabilised, skipping: {}", path_str);
196                }
197
198                active.lock().remove(&path_str);
199            }
200            log::debug!("Hash worker {} exiting", i);
201        });
202    }
203}
204
205async fn hash_and_emit(path_str: &str, tx_out: &mpsc::Sender<MonitorEvent>) {
206    let p_clone = path_str.to_owned();
207    match tokio::task::spawn_blocking(move || compute_sha1_chunked(&p_clone)).await {
208        Ok(Ok(checksum)) => {
209            let sidecar_path = crate::sidecar::find_sidecar(std::path::Path::new(path_str))
210                .map(|p| p.to_string_lossy().into_owned());
211            log::info!("File ready: {} (sha1={})", path_str, checksum);
212            let _ = tx_out
213                .send(MonitorEvent::Ready {
214                    path: path_str.to_owned(),
215                    checksum,
216                    sidecar_path,
217                })
218                .await;
219        }
220        Ok(Err(e)) => {
221            log::error!("Checksum error for {}: {}", path_str, e);
222        }
223        Err(e) => {
224            log::error!("Checksum task panicked for {}: {}", path_str, e);
225        }
226    }
227}
228
229fn handle_notify_event(
230    event: notify::Event,
231    watch_paths: &[WatchPathEntry],
232    tx: &mpsc::Sender<MonitorEvent>,
233    work_tx: &tokio::sync::mpsc::Sender<String>,
234    active_tasks: &std::sync::Arc<parking_lot::Mutex<std::collections::HashSet<String>>>,
235    debounce_map: &mut HashMap<String, Instant>,
236) {
237    let Some(kind) = monitor_event_kind(&event.kind) else {
238        return;
239    };
240
241    for path in event.paths {
242        if should_skip_path(&path, kind) {
243            continue;
244        }
245
246        let path_str = path.to_string_lossy().into_owned();
247        let Some(matched_entry) = best_matching_watch_entry(&path, watch_paths) else {
248            continue;
249        };
250        let rules = matched_entry.rules();
251
252        match kind {
253            MonitorFsEvent::Delete => process_delete_path(&path, path_str, &rules, tx),
254            MonitorFsEvent::Upsert => {
255                process_upsert_path(&path, path_str, &rules, active_tasks, debounce_map, work_tx)
256            }
257        }
258    }
259}
260
261#[derive(Clone, Copy, PartialEq, Eq)]
262enum MonitorFsEvent {
263    Upsert,
264    Delete,
265}
266
267fn monitor_event_kind(kind: &EventKind) -> Option<MonitorFsEvent> {
268    if matches!(kind, EventKind::Create(_) | EventKind::Modify(_)) {
269        Some(MonitorFsEvent::Upsert)
270    } else if matches!(kind, EventKind::Remove(_)) {
271        Some(MonitorFsEvent::Delete)
272    } else {
273        None
274    }
275}
276
277fn should_skip_path(path: &std::path::Path, kind: MonitorFsEvent) -> bool {
278    (kind == MonitorFsEvent::Upsert && path.is_dir()) || !is_supported_media_path(path)
279}
280
281fn process_delete_path(
282    path: &std::path::Path,
283    path_str: String,
284    rules: &crate::config::FolderRules,
285    tx: &mpsc::Sender<MonitorEvent>,
286) {
287    if rules.delete_folder_to_album && !is_temporary_file(path) {
288        log::info!("Deleted file event: {}", path_str);
289        let _ = tx.blocking_send(MonitorEvent::Deleted { path: path_str });
290    }
291}
292
293fn process_upsert_path(
294    path: &Path,
295    path_str: String,
296    rules: &crate::config::FolderRules,
297    active_tasks: &std::sync::Arc<parking_lot::Mutex<std::collections::HashSet<String>>>,
298    debounce_map: &mut HashMap<String, Instant>,
299    work_tx: &tokio::sync::mpsc::Sender<String>,
300) {
301    if rules.sync_method == FolderSyncMethod::DownloadOnly
302        || is_temporary_file(path)
303        || !rules.matches(path)
304    {
305        return;
306    }
307
308    if active_tasks.lock().contains(&path_str) {
309        return;
310    }
311
312    let now = Instant::now();
313    let debounce_ok = debounce_map
314        .get(&path_str)
315        .map(|last| now.duration_since(*last) > Duration::from_secs(2))
316        .unwrap_or(true);
317
318    if !debounce_ok {
319        return;
320    }
321
322    if debounce_map.len() > 1000 {
323        let cutoff = now - Duration::from_secs(60);
324        debounce_map.retain(|_, last| *last > cutoff);
325    }
326
327    log::info!("New file event: {}", path_str);
328    debounce_map.insert(path_str.clone(), now);
329    active_tasks.lock().insert(path_str.clone());
330
331    if let Err(err) = work_tx.blocking_send(path_str) {
332        log::warn!("Failed to send work to hash pool: {}", err);
333    }
334}
335
336impl MonitorHandle {
337    pub fn replace_watch_paths(
338        &self,
339        watch_paths: Vec<WatchPathEntry>,
340        background_sync_enabled: bool,
341    ) {
342        if let Err(err) = self.command_tx.send(MonitorCommand::ReplaceWatchPaths {
343            watch_paths,
344            background_sync_enabled,
345        }) {
346            log::warn!("Could not update watch paths on the live monitor: {}", err);
347        }
348    }
349}
350
351/// Replace live directories being monitored by the given watcher.
352fn replace_watches(
353    watcher: &mut dyn Watcher,
354    watched_roots: &mut Vec<PathBuf>,
355    watch_paths: &[WatchPathEntry],
356    background_sync_enabled: bool,
357) {
358    for path in watched_roots.drain(..) {
359        if let Err(err) = watcher.unwatch(&path) {
360            log::debug!("Could not unwatch '{}': {:?}", path.display(), err);
361        }
362    }
363
364    let mut any_watching = false;
365    for entry in watch_paths {
366        let p = Path::new(entry.path());
367        if p.exists() {
368            match watcher.watch(p, RecursiveMode::Recursive) {
369                Ok(_) => {
370                    log::info!("Watching: {}", display_watch_path(entry.path()));
371                    watched_roots.push(p.to_path_buf());
372                    any_watching = true;
373                }
374                Err(e) => log::warn!("Failed to watch '{}': {:?}", entry.path(), e),
375            }
376        } else {
377            log::warn!("Watch path does not exist, skipping: {}", entry.path());
378        }
379    }
380
381    if !any_watching {
382        if background_sync_enabled {
383            log::warn!("No valid watch paths. File monitoring is idle until a folder is added.");
384        } else {
385            log::info!("Background sync disabled. File monitoring is idle until it is enabled.");
386        }
387    }
388}
389
390/// Create recommended watcher fallback or poll watcher depending on flatpak containerisation.
391fn create_watcher(
392    notify_tx: std::sync::mpsc::Sender<notify::Result<notify::Event>>,
393) -> notify::Result<Box<dyn Watcher>> {
394    if is_flatpak_sandbox() {
395        log::info!(
396            "Using polling file watcher in Flatpak for portal-selected folders ({}ms interval)",
397            FLATPAK_POLL_INTERVAL_MS
398        );
399        let config = NotifyConfig::default()
400            .with_poll_interval(Duration::from_millis(FLATPAK_POLL_INTERVAL_MS));
401        Ok(Box::new(PollWatcher::new(notify_tx, config)?))
402    } else {
403        Ok(Box::new(notify::recommended_watcher(notify_tx)?))
404    }
405}
406
407/// Check if running inside Flatpak sandbox environment.
408fn is_flatpak_sandbox() -> bool {
409    Path::new("/.flatpak-info").exists()
410}
411
412/// Wait for a file's size to stop changing before treating it as upload-ready.
413async fn wait_for_file_completion(path: &str) -> bool {
414    let mut last_size: i64 = -1;
415    let mut stable_count: u32 = 0;
416    let mut last_change = Instant::now();
417
418    loop {
419        if last_change.elapsed().as_secs() >= IDLE_TIMEOUT_SECS {
420            log::warn!(
421                "Timeout: file stayed inactive for {}s: {}",
422                IDLE_TIMEOUT_SECS,
423                path
424            );
425            return false;
426        }
427
428        match tokio::fs::metadata(path).await {
429            Ok(meta) => {
430                let size = meta.len() as i64;
431                if size == last_size && size > 0 {
432                    stable_count += 1;
433                    last_change = Instant::now();
434                    if stable_count >= REQUIRED_STABLE_COUNTS {
435                        return true;
436                    }
437                } else {
438                    if size != last_size {
439                        last_change = Instant::now(); // file is still growing
440                    }
441                    stable_count = 0;
442                    last_size = size;
443                }
444            }
445            Err(_) => return false,
446        }
447
448        tokio::time::sleep(Duration::from_millis(CHECK_INTERVAL_MS)).await;
449    }
450}
451
452/// Compute SHA-1 in chunks so large media files never need to be read fully into memory.
453pub(crate) fn compute_sha1_chunked(path: &str) -> io::Result<String> {
454    const BUF_SIZE: usize = 65536;
455    let file = fs::File::open(path)?;
456    let mut reader = BufReader::with_capacity(BUF_SIZE, file);
457    let mut hasher = Sha1::new();
458    let mut buf = vec![0u8; BUF_SIZE];
459    loop {
460        let n = reader.read(&mut buf)?;
461        if n == 0 {
462            break;
463        }
464        hasher.update(&buf[..n]);
465    }
466    let digest = hasher.finalize();
467    Ok(digest.iter().map(|byte| format!("{byte:02x}")).collect())
468}
469
470/// Return whether a path points to a supported media file rather than a directory.
471pub(crate) fn is_supported_media_path(path: &Path) -> bool {
472    if media_kinds::is_supported_path(path) {
473        return true;
474    }
475    let ext = path.extension().map(|e| e.to_string_lossy().to_lowercase());
476    media_kinds::is_supported_ext(ext.as_deref().unwrap_or(""))
477}
478
479pub(crate) fn is_temporary_file(path: &Path) -> bool {
480    path.file_name()
481        .and_then(|name| name.to_str())
482        .map(|name| {
483            let name = name.to_ascii_lowercase();
484            name.ends_with(".tmp")
485                || name.ends_with(".part")
486                || name.ends_with(".crdownload")
487                || name.ends_with('~')
488        })
489        .unwrap_or(false)
490}
491
492#[cfg(test)]
493mod tests {
494    use super::{
495        compute_sha1_chunked, is_flatpak_sandbox, is_supported_media_path, is_temporary_file,
496    };
497    use std::io::{BufReader, Read, Write};
498    use std::path::Path;
499    use tempfile::NamedTempFile;
500
501    /// Compute BLAKE3 in chunks. Prepared for Phase 1.5 when hashing switches
502    /// from SHA-1 to BLAKE3 for local content identity. Will be promoted to
503    /// `pub(crate)` once it replaces SHA-1 in the production path.
504    fn compute_blake3_chunked(path: &str) -> std::io::Result<String> {
505        const BUF_SIZE: usize = 65536;
506        let file = std::fs::File::open(path)?;
507        let mut reader = BufReader::with_capacity(BUF_SIZE, file);
508        let mut hasher = blake3::Hasher::new();
509        let mut buf = vec![0u8; BUF_SIZE];
510        loop {
511            let n = reader.read(&mut buf)?;
512            if n == 0 {
513                break;
514            }
515            hasher.update(&buf[..n]);
516        }
517        Ok(hasher.finalize().to_hex().to_string())
518    }
519
520    #[test]
521    fn test_compute_sha1_chunked() {
522        let mut file = NamedTempFile::new().unwrap();
523        // SHA1 of "hello world" is 2aae6c35c94fcfb415dbe95f408b9ce91ee846ed
524        file.write_all(b"hello world").unwrap();
525
526        let hash = compute_sha1_chunked(file.path().to_str().unwrap()).unwrap();
527        assert_eq!(hash, "2aae6c35c94fcfb415dbe95f408b9ce91ee846ed");
528    }
529
530    #[test]
531    fn test_compute_blake3_chunked() {
532        let mut file = NamedTempFile::new().unwrap();
533        file.write_all(b"hello world").unwrap();
534
535        let hash = compute_blake3_chunked(file.path().to_str().unwrap()).unwrap();
536        // Known BLAKE3 hash of "hello world"
537        let expected = blake3::hash(b"hello world").to_hex().to_string();
538        assert_eq!(hash, expected);
539    }
540
541    #[test]
542    fn test_flatpak_detection_is_false_in_unit_tests() {
543        assert!(!is_flatpak_sandbox());
544    }
545
546    #[test]
547    fn test_temporary_file_detection() {
548        assert!(is_temporary_file(Path::new("/tmp/video.mp4.part")));
549        assert!(is_temporary_file(Path::new("/tmp/upload.jpg.tmp")));
550        assert!(is_temporary_file(Path::new("/tmp/image.png~")));
551        assert!(!is_temporary_file(Path::new("/tmp/final.jpg")));
552    }
553
554    #[test]
555    fn test_supported_media_extensions_include_new_immich_formats() {
556        assert!(is_supported_media_path(Path::new("photo.avif")));
557        assert!(is_supported_media_path(Path::new("photo.heif")));
558        assert!(is_supported_media_path(Path::new("photo.jp2")));
559        assert!(is_supported_media_path(Path::new("photo.jxl")));
560        assert!(is_supported_media_path(Path::new("photo.psd")));
561        assert!(is_supported_media_path(Path::new("photo.svg")));
562        assert!(is_supported_media_path(Path::new("video.3gp")));
563        assert!(is_supported_media_path(Path::new("video.avi")));
564        assert!(is_supported_media_path(Path::new("video.mkv")));
565        assert!(is_supported_media_path(Path::new("video.mxf")));
566    }
567}
mimick/monitor.rs

mimick/
monitor.rs
