Archive/src-tauri/src/watcher.rs

//! 로컬 소스 파일 워처 — 변경 이벤트를 디바운스해 증분 재스캔을 트리거한다.
//! 워처는 힌트일 뿐이고 재스캔(조정 스캔)이 진실의 원천이다 (계획서 §4).

use crate::AppState;
use notify_debouncer_full::notify::RecursiveMode;
use notify_debouncer_full::{new_debouncer, DebounceEventResult, DebouncedEvent};
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::Ordering;
use std::sync::Mutex;
use std::time::Duration;
use tauri::{Emitter, Manager};

type Debouncer = notify_debouncer_full::Debouncer<
    notify_debouncer_full::notify::RecommendedWatcher,
    notify_debouncer_full::RecommendedCache,
>;

/// 워치 불가 소스의 **첫** 재스캔 간격. 네트워크 공유(SMB/NAS)는 변경 알림을 안 주는
/// 서버가 있어 폴백이 필요하다.
const POLL_MIN: Duration = Duration::from_secs(60);
/// 변경이 계속 없을 때 늘어나는 상한.
///
/// 고정 60초는 대규모 소스에서 치명적이다 — 100만 건 열거가 60초 안에 끝나지 않으므로
/// 재스캔이 끝나기 전에 다음 재스캔이 큐에 쌓이고, 앱이 켜져 있는 동안 영구히 디스크를 간다.
/// 그래서 변경이 없으면 간격을 2배씩 늘리고(상한 30분), 변경이 있으면 즉시 되돌린다.
const POLL_MAX: Duration = Duration::from_secs(30 * 60);
/// 만기 검사 주기 — 간격 자체가 아니라 '지금 만기된 소스가 있는지' 보는 틱.
const POLL_TICK: Duration = Duration::from_secs(5);

/// 폴링 소스 한 개의 상태. 소스별로 따로 늘어난다(느린 NAS 하나가 다른 소스를 늦추지 않는다).
struct PollState {
    /// 다음 재스캔 예정 시각
    due: std::time::Instant,
    /// 현재 간격 (POLL_MIN..=POLL_MAX)
    interval: Duration,
}

type PollMap = std::sync::Arc<Mutex<HashMap<i64, PollState>>>;

pub struct WatcherManager {
    watchers: Mutex<HashMap<i64, Debouncer>>,
    /// 워치 불가로 폴링에 의존하는 소스 (네트워크 공유 등)
    polled: PollMap,
    dirty_tx: crossbeam_channel::Sender<i64>,
}

/// 재스캔 결과를 폴링 간격에 반영한다. 변경이 있었으면 최소 간격으로 되돌리고,
/// 없었으면 2배로 늘린다. 폴링 대상이 아닌 소스(워처가 붙은 소스)는 아무 일도 하지 않는다.
fn report_scan(polled: &PollMap, source_id: i64, changed: bool) {
    let mut map = polled.lock().unwrap();
    if let Some(st) = map.get_mut(&source_id) {
        st.interval = if changed {
            POLL_MIN
        } else {
            (st.interval * 2).min(POLL_MAX)
        };
        st.due = std::time::Instant::now() + st.interval;
        tracing::debug!(source_id, secs = st.interval.as_secs(), changed, "폴링 간격 갱신");
    }
}

impl WatcherManager {
    /// 워커 스레드를 시작하고 매니저를 만든다.
    pub fn new(app: tauri::AppHandle) -> std::sync::Arc<WatcherManager> {
        let (tx, rx) = crossbeam_channel::unbounded::<i64>();
        let polled: PollMap = std::sync::Arc::new(Mutex::new(HashMap::new()));
        let manager = std::sync::Arc::new(WatcherManager {
            watchers: Mutex::new(HashMap::new()),
            polled: polled.clone(),
            dirty_tx: tx.clone(),
        });

        // 워치 불가 소스 폴링 — 이벤트를 못 받는 공유도 결국 최신화된다.
        // 만기된 소스만 보낸다. 보낼 때 다음 만기를 미리 밀어 두므로, 재스캔이 오래 걸려도
        // 그 사이에 같은 소스가 다시 큐에 쌓이지 않는다(간격 갱신은 report_scan이 한다).
        {
            let polled = polled.clone();
            std::thread::Builder::new()
                .name("watch-poll".into())
                .spawn(move || loop {
                    std::thread::sleep(POLL_TICK);
                    let now = std::time::Instant::now();
                    let mut due: Vec<i64> = Vec::new();
                    {
                        let mut map = polled.lock().unwrap();
                        for (id, st) in map.iter_mut() {
                            if st.due <= now {
                                due.push(*id);
                                st.due = now + st.interval;
                            }
                        }
                    }
                    for id in due {
                        let _ = tx.send(id);
                    }
                })
                .expect("watch-poll 스레드 생성 실패");
        }
        let poll_report = polled;

        std::thread::Builder::new()
            .name("watch-rescan".into())
            .spawn(move || {
                while let Ok(first) = rx.recv() {
                    // 몰려온 이벤트를 한 번에 흡수
                    let mut dirty: std::collections::HashSet<i64> = [first].into();
                    while let Ok(more) = rx.try_recv() {
                        dirty.insert(more);
                    }

                    let state = app.state::<AppState>();
                    for source_id in dirty {
                        // 스캔 중이면 끝날 때까지 대기 (취소가 아니라 순번 대기)
                        while state.scan_running.swap(true, Ordering::SeqCst) {
                            std::thread::sleep(Duration::from_millis(500));
                        }
                        let root: Result<String, _> = state.db.with_read(|c| {
                            c.query_row(
                                "SELECT root FROM sources WHERE id = ?1 AND kind = 'local'",
                                [source_id],
                                |r| r.get(0),
                            )
                        });
                        match root {
                            Ok(root) => {
                                tracing::info!(source_id, "워처 트리거 재스캔");
                                let stats = crate::commands::run_scan_pipeline(
                                    &state.db,
                                    &state.tools,
                                    &state.thumbs,
                                    source_id,
                                    Path::new(&root),
                                    &state.scan_cancel,
                                    |_| {},
                                    || {
                                        let _ = app.emit("library-changed", source_id);
                                    },
                                );
                                // 아무것도 바뀌지 않았으면 다음 폴링을 늦춘다 —
                                // 조용한 대규모 NAS를 60초마다 다시 훑지 않게 하는 핵심이다.
                                let changed = stats
                                    .map(|s| s.added + s.changed + s.removed + s.moved > 0)
                                    .unwrap_or(false);
                                report_scan(&poll_report, source_id, changed);
                                state.scan_running.store(false, Ordering::SeqCst);
                                let _ = app.emit("library-changed", source_id);
                            }
                            Err(_) => {
                                state.scan_running.store(false, Ordering::SeqCst);
                            }
                        }
                    }
                }
            })
            .expect("watch-rescan 스레드 생성 실패");

        manager
    }

    /// 소스 워치를 해제한다 (소스 제외 시). Debouncer drop으로 감시가 멈춘다.
    pub fn remove(&self, source_id: i64) {
        if self.watchers.lock().unwrap().remove(&source_id).is_some() {
            tracing::info!(source_id, "파일 워처 해제");
        }
        self.polled.lock().unwrap().remove(&source_id);
    }

    /// DB의 로컬 소스 전체를 워치 대상으로 동기화한다 (시작 시 + 소스 추가 후 호출).
    pub fn refresh(&self, db: &archive_db::Db) {
        let sources: Vec<(i64, String)> = db
            .with_read(|conn| {
                let mut stmt =
                    conn.prepare_cached("SELECT id, root FROM sources WHERE kind = 'local'")?;
                let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?;
                rows.collect()
            })
            .unwrap_or_default();

        let mut watchers = self.watchers.lock().unwrap();
        for (source_id, root) in sources {
            if watchers.contains_key(&source_id) {
                continue;
            }
            let tx = self.dirty_tx.clone();
            let result = new_debouncer(
                Duration::from_secs(2),
                None,
                move |ev: DebounceEventResult| {
                    // 어떤 변경이든 소스 단위 재스캔 힌트로 취급
                    if let Ok(events) = ev {
                        if events.iter().any(relevant_event) {
                            let _ = tx.send(source_id);
                        }
                    }
                },
            );
            match result {
                Ok(mut debouncer) => {
                    match debouncer.watch(Path::new(&root), RecursiveMode::Recursive) {
                        Ok(()) => {
                            tracing::info!(source_id, root = %root, "파일 워처 시작");
                            watchers.insert(source_id, debouncer);
                        }
                        Err(e) => {
                            // 네트워크 드라이브 등 워치 불가 — 주기적 재스캔으로 폴백
                            tracing::warn!(
                                source_id,
                                secs = POLL_MIN.as_secs(),
                                "워치 불가 — 주기 재스캔으로 폴백(변경 없으면 간격 증가): {e}"
                            );
                            self.polled.lock().unwrap().entry(source_id).or_insert_with(|| {
                                PollState {
                                    due: std::time::Instant::now() + POLL_MIN,
                                    interval: POLL_MIN,
                                }
                            });
                        }
                    }
                }
                Err(e) => tracing::warn!("워처 생성 실패: {e}"),
            }
        }
    }
}

fn relevant_event(ev: &DebouncedEvent) -> bool {
    use notify_debouncer_full::notify::EventKind;
    matches!(
        ev.kind,
        EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_)
    )
}