Archive/src-tauri/crates/archive-indexer/src/pipeline.rs

//! 스캔 파이프라인 — 열거(jwalk) → 스냅샷 비교 → 배치 upsert → 이동 감지 → 소프트 삭제.
//!
//! 핵심 성능 설계: 스캔 시작 시 소스의 활성 행 전체를 (folder_id, name) 키의
//! 메모리 스냅샷으로 로드하고, size+mtime이 같은 파일은 **DB 쓰기를 완전히
//! 생략**한다. 무변경 재스캔의 비용은 디렉터리 열거 + 스냅샷 로드뿐이다.

use crate::model::{classify_ext, ext_of};
use crate::scan::{enumerate_items, RawEntry, WalkItem};
use archive_db::Db;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Instant;

#[derive(Debug, thiserror::Error)]
pub enum ScanError {
    #[error("DB 오류: {0}")]
    Db(#[from] archive_db::DbError),
    #[error("취소됨")]
    Cancelled,
}

#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct ScanStats {
    pub seen: u64,
    pub added: u64,
    pub changed: u64,
    pub removed: u64,
    pub moved: u64,
}

#[derive(Debug, Clone, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ScanProgress {
    pub scan_id: i64,
    pub phase: String, // enumerate | metadata | done
    pub seen: u64,
    pub current_dir: String,
    pub done: bool,
}

/// 스냅샷/이동감지용 기존 행 정보
#[derive(Clone)]
struct SnapRow {
    id: i64,
    size: i64,
    mtime_ms: i64,
    name: String,
    quick_hash: Option<i64>,
}

const BATCH: usize = 1000;

fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

/// 폴더 행을 보장하고 id를 돌려준다 (조상 폴더 포함, 캐시 사용).
fn ensure_folder(
    db: &Db,
    cache: &mut HashMap<PathBuf, i64>,
    source_id: i64,
    root: &Path,
    dir: &Path,
) -> Result<i64, ScanError> {
    if let Some(id) = cache.get(dir) {
        return Ok(*id);
    }
    let parent_id = if dir == root {
        None
    } else {
        let parent = dir.parent().unwrap_or(root);
        Some(ensure_folder(db, cache, source_id, root, parent)?)
    };

    let path_str = dir.to_string_lossy().into_owned();
    let name = dir
        .file_name()
        .map(|n| n.to_string_lossy().into_owned())
        .unwrap_or_else(|| dir.to_string_lossy().into_owned());

    let id = db.with_write(move |conn| {
        conn.execute(
            "INSERT INTO folders(source_id, parent_id, path, name) VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT(source_id, path) DO UPDATE SET name = excluded.name",
            rusqlite::params![source_id, parent_id, path_str, name],
        )?;
        conn.query_row(
            "SELECT id FROM folders WHERE source_id = ?1 AND path = ?2",
            rusqlite::params![source_id, path_str],
            |r| r.get(0),
        )
    })?;
    cache.insert(dir.to_path_buf(), id);
    Ok(id)
}

/// 소스의 활성 파일 행을 (folder_id, name) 키로 로드한다.
fn load_snapshot(
    db: &Db,
    source_id: i64,
) -> Result<HashMap<(i64, String), SnapRow>, ScanError> {
    let rows: Vec<(i64, SnapRow)> = db.with_read(|conn| {
        let mut stmt = conn.prepare_cached(
            "SELECT folder_id, id, size, mtime_ms, name, quick_hash
             FROM files WHERE source_id = ?1 AND deleted_at IS NULL",
        )?;
        let rows = stmt.query_map([source_id], |r| {
            Ok((
                r.get::<_, i64>(0)?,
                SnapRow {
                    id: r.get(1)?,
                    size: r.get(2)?,
                    mtime_ms: r.get(3)?,
                    name: r.get(4)?,
                    quick_hash: r.get(5)?,
                },
            ))
        })?;
        rows.collect()
    })?;
    Ok(rows
        .into_iter()
        .map(|(folder_id, row)| ((folder_id, row.name.clone()), row))
        .collect())
}

fn flush_batch(db: &Db, scan_id: i64, batch: Vec<(i64, RawEntry)>) -> Result<(), ScanError> {
    db.with_write(move |conn| {
        let tx = conn.transaction()?;
        {
            let mut stmt = tx.prepare_cached(
                "INSERT INTO files(source_id, folder_id, name, ext, kind, size, mtime_ms, scan_id)
                 SELECT fo.source_id, ?1, ?2, ?3, ?4, ?5, ?6, ?7 FROM folders fo WHERE fo.id = ?1
                 ON CONFLICT(folder_id, name) DO UPDATE SET
                   scan_id = excluded.scan_id,
                   deleted_at = NULL,
                   meta_state = CASE WHEN files.size = excluded.size AND files.mtime_ms = excluded.mtime_ms
                                     THEN files.meta_state ELSE 0 END,
                   thumb_state = CASE WHEN files.size = excluded.size AND files.mtime_ms = excluded.mtime_ms
                                      THEN files.thumb_state ELSE 0 END,
                   quick_hash = CASE WHEN files.size = excluded.size AND files.mtime_ms = excluded.mtime_ms
                                     THEN files.quick_hash ELSE NULL END,
                   full_hash = CASE WHEN files.size = excluded.size AND files.mtime_ms = excluded.mtime_ms
                                    THEN files.full_hash ELSE NULL END,
                   phash = CASE WHEN files.size = excluded.size AND files.mtime_ms = excluded.mtime_ms
                                THEN files.phash ELSE NULL END,
                   vhash = CASE WHEN files.size = excluded.size AND files.mtime_ms = excluded.mtime_ms
                                THEN files.vhash ELSE NULL END,
                   size = excluded.size,
                   mtime_ms = excluded.mtime_ms",
            )?;
            for (folder_id, e) in &batch {
                let name = e
                    .path
                    .file_name()
                    .map(|n| n.to_string_lossy().into_owned())
                    .unwrap_or_default();
                let ext = ext_of(&e.path).unwrap_or_default();
                let kind = classify_ext(&ext) as u8;
                stmt.execute(rusqlite::params![
                    folder_id,
                    name,
                    ext,
                    kind,
                    e.size as i64,
                    e.mtime_ms,
                    scan_id
                ])?;
            }
        }
        tx.commit()
    })?;
    Ok(())
}

/// 이동 감지: 사라진 행(missing)과 이번 스캔에서 새로 추가된 행을
/// ① (size, name) ② (size, quick_hash 저장분) 기준으로 매칭해
/// 행 id를 보존한 채 경로만 갱신한다 (태그·썸네일 보존).
/// 반환: (moved, 소프트삭제 대상 id 목록)
fn detect_moves(
    db: &Db,
    source_id: i64,
    scan_id: i64,
    scan_start: i64,
    missing: Vec<SnapRow>,
) -> Result<(u64, Vec<i64>), ScanError> {
    if missing.is_empty() {
        return Ok((0, Vec::new()));
    }

    struct NewRow {
        id: i64,
        size: i64,
        name: String,
        folder_id: i64,
        path: PathBuf,
    }

    let added: Vec<NewRow> = db.with_read(|conn| {
        let mut stmt = conn.prepare(
            "SELECT f.id, f.size, f.name, f.folder_id, fo.path FROM files f
             JOIN folders fo ON fo.id = f.folder_id
             WHERE f.source_id = ?1 AND f.scan_id = ?2 AND f.added_at >= ?3",
        )?;
        let rows = stmt.query_map(rusqlite::params![source_id, scan_id, scan_start], |r| {
            let dir: String = r.get(4)?;
            let name: String = r.get(2)?;
            Ok(NewRow {
                id: r.get(0)?,
                size: r.get(1)?,
                name: name.clone(),
                folder_id: r.get(3)?,
                path: Path::new(&dir).join(&name),
            })
        })?;
        rows.collect()
    })?;

    if added.is_empty() {
        return Ok((0, missing.into_iter().map(|m| m.id).collect()));
    }

    let mut by_size_name: HashMap<(i64, &str), &SnapRow> = HashMap::new();
    let mut by_size_hash: HashMap<(i64, i64), &SnapRow> = HashMap::new();
    for m in &missing {
        by_size_name.entry((m.size, m.name.as_str())).or_insert(m);
        if let Some(qh) = m.quick_hash {
            by_size_hash.entry((m.size, qh)).or_insert(m);
        }
    }

    let mut moves: Vec<(i64, i64, String)> = Vec::new(); // (old_id, new_folder_id, new_name)
    let mut new_ids_to_delete: Vec<i64> = Vec::new();
    let mut matched_old: std::collections::HashSet<i64> = Default::default();

    for n in &added {
        let matched = by_size_name
            .get(&(n.size, n.name.as_str()))
            .copied()
            .filter(|m| !matched_old.contains(&m.id))
            .or_else(|| {
                if by_size_hash.is_empty() {
                    return None;
                }
                let qh = crate::hash::quick_hash(&n.path).ok()? as i64;
                by_size_hash
                    .get(&(n.size, qh))
                    .copied()
                    .filter(|m| !matched_old.contains(&m.id))
            });

        if let Some(m) = matched {
            matched_old.insert(m.id);
            new_ids_to_delete.push(n.id);
            moves.push((m.id, n.folder_id, n.name.clone()));
        }
    }

    let moved = moves.len() as u64;
    if moved > 0 {
        db.with_write(move |conn| {
            let tx = conn.transaction()?;
            {
                let mut del = tx.prepare_cached("DELETE FROM files WHERE id = ?1")?;
                for id in &new_ids_to_delete {
                    del.execute([id])?;
                }
                let mut upd = tx.prepare_cached(
                    "UPDATE files SET folder_id = ?2, name = ?3, scan_id = ?4 WHERE id = ?1",
                )?;
                for (old_id, folder_id, name) in &moves {
                    upd.execute(rusqlite::params![old_id, folder_id, name, scan_id])?;
                }
            }
            tx.commit()
        })?;
    }

    let to_delete = missing
        .into_iter()
        .filter(|m| !matched_old.contains(&m.id))
        .map(|m| m.id)
        .collect();
    Ok((moved, to_delete))
}

fn prune_stale_folders(
    db: &Db,
    source_id: i64,
    seen: &HashMap<PathBuf, i64>,
) -> Result<(), ScanError> {
    let seen_ids: std::collections::HashSet<i64> = seen.values().copied().collect();
    let all: Vec<(i64, String)> = db.with_read(|conn| {
        let mut stmt =
            conn.prepare_cached("SELECT id, path FROM folders WHERE source_id = ?1")?;
        let rows = stmt.query_map([source_id], |r| Ok((r.get(0)?, r.get(1)?)))?;
        rows.collect()
    })?;

    let mut stale: Vec<(i64, String)> = all
        .into_iter()
        .filter(|(id, _)| !seen_ids.contains(id))
        .collect();
    if stale.is_empty() {
        return Ok(());
    }
    // 자식(더 긴 경로) 먼저 삭제 — parent_id FK 위반 방지
    stale.sort_by_key(|(_, p)| std::cmp::Reverse(p.len()));

    db.with_write(move |conn| {
        let mut stmt = conn.prepare_cached(
            "DELETE FROM folders WHERE id = ?1
               AND NOT EXISTS (SELECT 1 FROM files WHERE folder_id = ?1)
               AND NOT EXISTS (SELECT 1 FROM folders c WHERE c.parent_id = ?1)",
        )?;
        for (id, _) in &stale {
            let _ = stmt.execute([id]);
        }
        Ok(())
    })?;
    Ok(())
}

fn soft_delete(db: &Db, ids: Vec<i64>) -> Result<u64, ScanError> {
    if ids.is_empty() {
        return Ok(0);
    }
    let count = ids.len() as u64;
    let ts = now_ms();
    db.with_write(move |conn| {
        let tx = conn.transaction()?;
        {
            let mut stmt = tx.prepare_cached("UPDATE files SET deleted_at = ?2 WHERE id = ?1")?;
            for id in &ids {
                stmt.execute(rusqlite::params![id, ts])?;
            }
        }
        tx.commit()
    })?;
    Ok(count)
}

/// 소스 하나를 스캔한다. 블로킹 — 전용 스레드에서 호출할 것.
pub fn scan_source(
    db: &Arc<Db>,
    source_id: i64,
    root: &Path,
    cancel: &AtomicBool,
    mut on_progress: impl FnMut(ScanProgress),
) -> Result<ScanStats, ScanError> {
    let scan_start = now_ms();
    let scan_id: i64 = db.with_write(move |conn| {
        conn.execute(
            "INSERT INTO scans(source_id, started_at, status) VALUES (?1, ?2, 0)",
            rusqlite::params![source_id, scan_start],
        )?;
        Ok(conn.last_insert_rowid())
    })?;

    let mut stats = ScanStats::default();
    let mut snapshot = load_snapshot(db, source_id)?;
    let mut folder_cache: HashMap<PathBuf, i64> = HashMap::new();
    let mut batch: Vec<(i64, RawEntry)> = Vec::with_capacity(BATCH);
    let mut last_emit = Instant::now();

    for item in enumerate_items(root) {
        if cancel.load(Ordering::Relaxed) {
            db.with_write(move |conn| {
                conn.execute(
                    "UPDATE scans SET status = 2, finished_at = ?2 WHERE id = ?1",
                    rusqlite::params![scan_id, now_ms()],
                )
            })?;
            return Err(ScanError::Cancelled);
        }

        let entry = match item {
            WalkItem::Dir(dir) => {
                // 빈 폴더도 트리에 포함 (이동 대상으로 쓰인다)
                ensure_folder(db, &mut folder_cache, source_id, root, &dir)?;
                continue;
            }
            WalkItem::File(e) => e,
        };

        let dir = entry.path.parent().unwrap_or(root).to_path_buf();
        let folder_id = ensure_folder(db, &mut folder_cache, source_id, root, &dir)?;
        stats.seen += 1;

        let name = entry
            .path
            .file_name()
            .map(|n| n.to_string_lossy().into_owned())
            .unwrap_or_default();

        // 스냅샷 비교 — 무변경이면 DB 쓰기 생략 (remove로 '발견됨' 표시)
        match snapshot.remove(&(folder_id, name)) {
            Some(prev) if prev.size == entry.size as i64 && prev.mtime_ms == entry.mtime_ms => {
                // unchanged
            }
            Some(_) => {
                stats.changed += 1;
                batch.push((folder_id, entry));
            }
            None => {
                stats.added += 1;
                batch.push((folder_id, entry));
            }
        }

        if batch.len() >= BATCH {
            flush_batch(db, scan_id, std::mem::take(&mut batch))?;
        }
        if last_emit.elapsed().as_millis() >= 100 {
            last_emit = Instant::now();
            on_progress(ScanProgress {
                scan_id,
                phase: "enumerate".into(),
                seen: stats.seen,
                current_dir: dir.to_string_lossy().into_owned(),
                done: false,
            });
        }
    }
    if !batch.is_empty() {
        flush_batch(db, scan_id, std::mem::take(&mut batch))?;
    }

    // 스냅샷에 남은 행 = 디스크에서 사라진 파일 → 이동 감지 후 소프트 삭제
    let missing: Vec<SnapRow> = snapshot.into_values().collect();
    let (moved, to_delete) = detect_moves(db, source_id, scan_id, scan_start, missing)?;
    stats.moved = moved;
    stats.removed = soft_delete(db, to_delete)?;

    // 디스크에서 사라진 폴더 행 정리 (파일 참조가 없는 것만, 자식 먼저)
    prune_stale_folders(db, source_id, &folder_cache)?;

    let stats_json = serde_json::to_string(&stats).unwrap_or_default();
    db.with_write(move |conn| {
        conn.execute(
            "UPDATE scans SET status = 1, finished_at = ?2, stats = ?3 WHERE id = ?1",
            rusqlite::params![scan_id, now_ms(), stats_json],
        )
    })?;

    on_progress(ScanProgress {
        scan_id,
        phase: "done".into(),
        seen: stats.seen,
        current_dir: String::new(),
        done: true,
    });
    Ok(stats)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn setup() -> (tempfile::TempDir, tempfile::TempDir, Arc<Db>, i64) {
        let dbdir = tempfile::tempdir().unwrap();
        let media = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::open(&dbdir.path().join("t.db")).unwrap());
        let root = media.path().to_string_lossy().into_owned();
        let source_id = db
            .with_write(move |conn| {
                conn.execute(
                    "INSERT INTO sources(kind, name, root) VALUES ('local', '테스트', ?1)",
                    [&root],
                )?;
                Ok(conn.last_insert_rowid())
            })
            .unwrap();
        (dbdir, media, db, source_id)
    }

    fn count_active(db: &Db) -> i64 {
        db.with_read(|c| {
            c.query_row("SELECT count(*) FROM files WHERE deleted_at IS NULL", [], |r| r.get(0))
        })
        .unwrap()
    }

    #[test]
    fn full_scan_then_incremental() {
        let (_d, media, db, sid) = setup();
        let root = media.path().to_path_buf();
        std::fs::create_dir_all(root.join("여행/제주")).unwrap();
        std::fs::write(root.join("a.jpg"), b"aaa").unwrap();
        std::fs::write(root.join("여행/b.png"), b"bbbb").unwrap();
        std::fs::write(root.join("여행/제주/c.mp4"), b"ccccc").unwrap();

        let cancel = AtomicBool::new(false);
        let stats = scan_source(&db, sid, &root, &cancel, |_| {}).unwrap();
        assert_eq!(stats.seen, 3);
        assert_eq!(stats.added, 3);
        assert_eq!(count_active(&db), 3);

        // 변경 없는 재스캔: added/changed/removed 전부 0
        let stats2 = scan_source(&db, sid, &root, &cancel, |_| {}).unwrap();
        assert_eq!(stats2.seen, 3);
        assert_eq!(stats2.added, 0);
        assert_eq!(stats2.changed, 0);
        assert_eq!(stats2.removed, 0);
        assert_eq!(count_active(&db), 3);

        // 내용 변경(크기 변화) → changed
        std::fs::write(root.join("a.jpg"), b"aaaa-longer").unwrap();
        let stats3 = scan_source(&db, sid, &root, &cancel, |_| {}).unwrap();
        assert_eq!(stats3.changed, 1);
        assert_eq!(stats3.added, 0);

        // 파일 삭제 후 재스캔 → 소프트 삭제
        std::fs::remove_file(root.join("a.jpg")).unwrap();
        let stats4 = scan_source(&db, sid, &root, &cancel, |_| {}).unwrap();
        assert_eq!(stats4.removed, 1);
        assert_eq!(count_active(&db), 2);
    }

    #[test]
    fn move_detection_preserves_row_id() {
        let (_d, media, db, sid) = setup();
        let root = media.path().to_path_buf();
        std::fs::create_dir_all(root.join("old")).unwrap();
        std::fs::create_dir_all(root.join("new")).unwrap();
        std::fs::write(root.join("old/사진.jpg"), b"unique-content-123").unwrap();

        let cancel = AtomicBool::new(false);
        scan_source(&db, sid, &root, &cancel, |_| {}).unwrap();
        let old_id: i64 = db
            .with_read(|c| c.query_row("SELECT id FROM files WHERE name='사진.jpg'", [], |r| r.get(0)))
            .unwrap();

        std::fs::rename(root.join("old/사진.jpg"), root.join("new/사진.jpg")).unwrap();
        let stats = scan_source(&db, sid, &root, &cancel, |_| {}).unwrap();
        assert_eq!(stats.moved, 1);
        assert_eq!(stats.removed, 0);

        let (new_id, folder_path): (i64, String) = db
            .with_read(|c| {
                c.query_row(
                    "SELECT f.id, fo.path FROM files f JOIN folders fo ON fo.id=f.folder_id
                     WHERE f.name='사진.jpg' AND f.deleted_at IS NULL",
                    [],
                    |r| Ok((r.get(0)?, r.get(1)?)),
                )
            })
            .unwrap();
        assert_eq!(new_id, old_id, "행 id가 보존되어야 태그/썸네일이 유지된다");
        assert!(folder_path.ends_with("new"));
        assert_eq!(count_active(&db), 1);
    }

    #[test]
    fn soft_deleted_file_resurrects_on_return() {
        let (_d, media, db, sid) = setup();
        let root = media.path().to_path_buf();
        std::fs::write(root.join("x.jpg"), b"data").unwrap();
        let cancel = AtomicBool::new(false);

        scan_source(&db, sid, &root, &cancel, |_| {}).unwrap();
        std::fs::remove_file(root.join("x.jpg")).unwrap();
        scan_source(&db, sid, &root, &cancel, |_| {}).unwrap();
        assert_eq!(count_active(&db), 0);

        // 같은 이름 파일이 돌아옴 → deleted_at 해제
        std::fs::write(root.join("x.jpg"), b"data").unwrap();
        let stats = scan_source(&db, sid, &root, &cancel, |_| {}).unwrap();
        assert_eq!(stats.added, 1);
        assert_eq!(count_active(&db), 1);
    }
}