DarkRoom/core/dr-catalog/src/walk.rs

//! TRACES: FR-CAT-1 | FR-CAT-1a | FR-CAT-9 | NFR-PORT-1 | NFR-P1
//! Turning the decisions in [`scan`](crate::scan) into a catalogued library.
//!
//! [`scan`](crate::scan) knows what a changed directory *means* and holds no
//! I/O; [`dr_plat::Storage`] knows how to read a directory and holds no
//! catalog. This module is the only place the two meet, which is what keeps
//! both of them testable on their own — and what lets Android arrive as a
//! second `Storage` implementation with nothing here to change (ARCH §6.9).
//!
//! # What it costs on a library that has not changed
//!
//! One probe per directory and nothing else: no listing, no `stat` per file, no
//! row written. The saving is not incidental — 50k images in 2k folders is 2k
//! probes against 50k stats, and NFR-P1 is written against the former.
//!
//! # What pruning cannot see
//!
//! Writing to an existing file moves neither its directory's mtime nor its
//! entry count, so a folder full of rewritten photographs looks untouched and
//! is skipped. This is the price of pruning at the directory level, and it is
//! worth being plain about rather than discovering later.
//!
//! It bites less than it sounds. Almost nothing rewrites a RAW in place: an
//! export, a backup restore, `mv`, and every editor that saves safely write a
//! new file beside the old one and rename over it, which does move both the
//! mtime and — briefly — the count. Those are caught. What is missed is a
//! genuine in-place write, which for a photograph library is close to nothing;
//! and it is missed only until the folder is listed for some other reason.
//!
//! # The deletion sweep, which is the dangerous part
//!
//! Rows are deleted in two places, and both are guarded by the same principle:
//! **absence is only evidence of deletion where absence was actually
//! observed.** A folder that was listed proves its missing images are gone. A
//! folder that was pruned proves nothing about its contents, and a scan that
//! was cancelled or that failed part-way proves nothing about the folders it
//! never reached — hence [`ScanOutcome::may_prune`], and hence the folder sweep
//! running once at the end rather than as it goes.
//!
//! Get this wrong and an unplugged drive deletes the library. Every image would
//! be absent, every folder unreached, and the sweep would take all of them
//! along with their ratings and their edits.

use std::collections::HashMap;

use dr_plat::{DirRef, Node, Storage};
use dr_types::{Availability, FormatFilter, RootId, SourceRef};
use rusqlite::{Connection, OptionalExtension};

use crate::error::CatalogError;
use crate::query::availability_code;
use crate::scan::{
    classify_dir, classify_entry, DirAction, DirState, EntryAction, KnownFile, ScanOutcome,
};
use crate::trash::TRASH_DIR;

/// How deep to recurse before giving up.
///
/// A symlink loop would otherwise walk forever — the filesystem implementation
/// follows symlinks deliberately, because a photographer who symlinks last
/// year's drive into the library means it. Real libraries are nowhere near this
/// deep. The same figure `dr_sync::scan` uses, for the same reason.
pub const MAX_DEPTH: usize = 32;

/// TRACES: FR-CAT-3
/// Directory holding derived state — thumbnail shards and catalog snapshots.
///
/// Quoted rather than imported, exactly as [`TRASH_DIR`] is quoted in
/// [`crate::trash`]: `dr-catalog` does not depend on `dr-sync`, and adding that
/// dependency for one string would invert the layering. A test asserts the two
/// agree.
pub const DERIVED_DIR: &str = ".darkroom-derived";

/// Whether a directory is one the scan must stay out of.
///
/// **The trash exclusion is half of the soft delete.** Trashed images are real
/// files in a real folder under the library root (FR-CAT-15); a scan that
/// walked it would re-index them as ordinary photographs and the delete would
/// come undone on the next refresh.
///
/// Matched on the folder's own name at any depth, so a nested library moved in
/// wholesale carries its exclusions with it.
pub fn is_excluded(name: &str) -> bool {
    matches!(name, TRASH_DIR | DERIVED_DIR)
}

/// Which grant a root represents — the `roots.kind` column.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RootKind {
    /// A directory on a filesystem.
    Local,
    /// An Android persisted document tree.
    Saf,
    /// A folder on a remote server, reached through the sync layer.
    Remote,
}

impl RootKind {
    pub fn as_str(self) -> &'static str {
        match self {
            RootKind::Local => "local",
            RootKind::Saf => "saf",
            RootKind::Remote => "remote",
        }
    }
}

/// TRACES: FR-CAT-1
/// Find or create the row for a granted library location.
///
/// `label` is how the grant is spelled for this kind of root — the directory
/// path on Linux, the tree URI on Android, the remote folder on a server. It is
/// display text and identity, never something `core/` resolves: reaching the
/// files is [`Storage`]'s business and takes a [`RootId`], which is the whole
/// point of FR-CAT-1a.
///
/// Idempotent, because it has to be: a second row for the same folder would
/// fragment the library across two roots, splitting the images and comparing
/// each half against the wrong scan generation. The `UNIQUE(kind, label)` in
/// the schema is what enforces it; this reuses the row rather than failing.
pub fn ensure_root(conn: &Connection, kind: RootKind, label: &str) -> Result<RootId, CatalogError> {
    conn.execute(
        "INSERT INTO roots(kind, label) VALUES (?1, ?2) ON CONFLICT DO NOTHING",
        rusqlite::params![kind.as_str(), label],
    )?;
    let id: i64 = conn.query_row(
        "SELECT id FROM roots WHERE kind = ?1 AND label = ?2",
        rusqlite::params![kind.as_str(), label],
        |r| r.get(0),
    )?;
    Ok(RootId(id as u64))
}

/// Progress during a scan, so a large library reports rather than appears hung.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ScanProgress {
    pub directories_listed: usize,
    /// Directories proven unchanged and skipped. The value of pruning, made
    /// visible — on a healthy rescan this is nearly the whole library.
    pub directories_pruned: usize,
    pub images_found: usize,
}

/// What a scan did.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ScanReport {
    pub progress: ScanProgress,
    pub inserted: usize,
    /// Files whose size or mtime moved: metadata re-read, thumbnail rebuilt,
    /// content hash dropped.
    pub updated: usize,
    /// Files already catalogued and untouched. The overwhelming majority, and
    /// each one costs nothing but a hash lookup.
    pub unchanged: usize,
    /// Rows removed because the file was *observed* to be gone.
    pub images_removed: usize,
    pub folders_removed: usize,
    pub outcome: ScanOutcome,
}

/// TRACES: FR-CAT-1 | FR-CAT-9 | NFR-P1
/// Scan one granted root into the catalog.
///
/// `cancel` is polled once per directory — FR-CAT-1 requires a scan the user
/// can stop, and a cancelled scan leaves valid partial state: everything
/// listed is committed, nothing is pruned.
///
/// `now` is passed rather than read, so a test can assert on `added_at` and so
/// two rows written by one scan carry one timestamp.
///
/// Returns a report rather than failing on an unreachable root. That is not
/// leniency: [`ScanOutcome`] is *how* the caller is told, it is what
/// [`ScanOutcome::may_prune`] consumes, and an `Err` here would discard the
/// counts of everything the scan did manage to do. A `CatalogError` means the
/// database itself is in trouble.
pub fn scan_root(
    conn: &Connection,
    storage: &dyn Storage,
    root: RootId,
    formats: &FormatFilter,
    now: i64,
    cancel: impl Fn() -> bool,
    mut on_progress: impl FnMut(ScanProgress),
) -> Result<ScanReport, CatalogError> {
    let mut report = ScanReport::default();

    // A fresh generation per *run*, taken and stored before anything is
    // walked — not per completed scan. A cancelled run marks the folders it
    // reached, and if the next run reused the number those marks would look
    // current: a folder deleted in between would carry the generation the
    // sweep is about to compare against, and would survive.
    let generation = bump_generation(conn, root, now)?;

    let start = match storage.root_dir(root) {
        Ok(d) => d,
        // Not merely "no images": the grant is gone or the drive is out. Every
        // image under it is unreachable, not deleted (FR-CAT-9).
        Err(e) => {
            log::warn!("scan: root {} unreachable: {e}", root.0);
            mark_root_offline(conn, root)?;
            report.outcome = ScanOutcome::RootUnreachable;
            return Ok(report);
        }
    };

    // Explicit stack rather than recursion, as in `dr_sync::scan`: a
    // pathological tree must not overflow. Each item carries the catalog id of
    // its parent folder, which is what threads the folder tree together.
    let mut stack: Vec<(DirRef, usize, Option<i64>)> = vec![(start, 0, None)];
    let mut cancelled = false;
    let mut partial = false;

    while let Some((dir, depth, parent)) = stack.pop() {
        if cancel() {
            log::info!("scan: cancelled at {dir}");
            cancelled = true;
            break;
        }

        // Excluded by name, before it is probed or listed. The root itself is
        // exempt: a user who granted `.darkroom-trash` as their library meant
        // it, and refusing to scan the folder they chose would look like the
        // app doing nothing.
        if !dir.is_root() && is_excluded(dir.name()) {
            continue;
        }

        if depth > MAX_DEPTH {
            // Deeper folders exist and were not seen, so this scan cannot
            // prove anything absent.
            log::warn!("scan: depth limit at {dir}, not descending further");
            partial = true;
            continue;
        }

        let state = match storage.dir_state(&dir) {
            Ok(s) => s,
            Err(e) if dir.is_root() => {
                log::warn!("scan: root {} unreachable: {e}", root.0);
                mark_root_offline(conn, root)?;
                report.outcome = ScanOutcome::RootUnreachable;
                return Ok(report);
            }
            // One folder failed while the library as a whole is fine — a
            // permission on a subdirectory, most likely. Its images stay in
            // the catalog, and the sweep is disarmed for the whole run.
            Err(e) => {
                log::warn!("scan: {dir} could not be probed: {e}");
                partial = true;
                continue;
            }
        };

        // One transaction per directory. A single transaction for the whole
        // scan would make a 50k-image first run all-or-nothing and unresumable;
        // one per row would fsync 50k times.
        let tx = conn.unchecked_transaction()?;
        let (folder_id, stored) = touch_folder(&tx, root, &dir, parent, generation)?;

        match classify_dir(stored, state) {
            // Unchanged. Its own contents need not be read, but a change in a
            // *grandchild* moves no timestamp here — a filesystem propagates
            // nothing upward — so the known children are still walked.
            DirAction::RecurseOnly => {
                report.progress.directories_pruned += 1;
                for child in known_children(&tx, root, folder_id)? {
                    stack.push((child, depth + 1, Some(folder_id)));
                }
                tx.commit()?;
            }

            DirAction::ListAndRecurse => {
                let entries = match storage.list(&dir) {
                    Ok(e) => e,
                    Err(e) if dir.is_root() => {
                        drop(tx);
                        log::warn!("scan: root {} unreachable: {e}", root.0);
                        mark_root_offline(conn, root)?;
                        report.outcome = ScanOutcome::RootUnreachable;
                        return Ok(report);
                    }
                    Err(e) => {
                        // Rolled back rather than committed: the folder row's
                        // new generation would otherwise claim it was reached
                        // successfully.
                        drop(tx);
                        log::warn!("scan: {dir} could not be listed: {e}");
                        partial = true;
                        continue;
                    }
                };

                report.progress.directories_listed += 1;

                // Emptied as the listing is walked, so whatever is left at the
                // end is precisely what the folder no longer holds. Removing
                // rather than marking keeps the sweep O(1) per file instead of
                // a scan of the folder's rows per file.
                let mut known = catalogued_images(&tx, folder_id)?;

                for entry in &entries {
                    match &entry.node {
                        Node::Dir(child) => stack.push((child.clone(), depth + 1, Some(folder_id))),
                        Node::File(src) => {
                            // Taken out of `known` whatever is decided next:
                            // this file was seen, and the sweep at the end is
                            // about the ones that were not.
                            let existing = known.remove(src.key());
                            let action = classify_entry(
                                &entry.meta,
                                existing.map(|e| e.as_known()),
                                formats,
                            );

                            // A format the filter no longer asks for, left
                            // exactly as it is — row and all. Unticking JPEG
                            // means stop looking for new ones, not delete the
                            // ones already found: the file is sitting right
                            // there, and absence is what justifies a delete.
                            if action == EntryAction::Ignored {
                                continue;
                            }
                            report.progress.images_found += 1;

                            match action {
                                EntryAction::Unchanged => {
                                    report.unchanged += 1;
                                    // Unchanged in size and mtime, but perhaps
                                    // not in reachability: a file that was
                                    // marked offline while the drive was out is
                                    // back, and nothing else would ever notice,
                                    // because nothing about it has moved.
                                    if let Some(row) = &existing {
                                        restore_availability(&tx, row, src)?;
                                    }
                                }
                                EntryAction::Insert => {
                                    insert_image(
                                        &tx,
                                        root,
                                        folder_id,
                                        src,
                                        entry.meta.size,
                                        entry.meta.mtime,
                                        now,
                                    )?;
                                    report.inserted += 1;
                                }
                                EntryAction::Changed => {
                                    update_image(
                                        &tx,
                                        root,
                                        folder_id,
                                        src,
                                        entry.meta.size,
                                        entry.meta.mtime,
                                    )?;
                                    report.updated += 1;
                                }
                                EntryAction::Ignored => unreachable!("returned above"),
                            }
                        }
                    }
                }

                // The file-level sweep, and the only place absence was
                // observed: this folder was read, and these rows were not in
                // it. Trashed images are exempt — their file has been moved
                // into the trash on purpose, so being absent from the folder is
                // the expected state, and deleting the row would lose the way
                // back (FR-CAT-15).
                for row in known.values() {
                    if row.trashed {
                        continue;
                    }
                    tx.execute("DELETE FROM images WHERE id = ?1", [row.id])?;
                    report.images_removed += 1;
                }

                // Only now, with the listing read and reconciled, is the
                // directory's state safe to record. Storing it before would let
                // the next scan prune a folder whose contents were never
                // actually seen, hiding every image beneath it permanently —
                // the same trap `dr_sync::scan` documents for ETags.
                tx.execute(
                    "UPDATE folders SET mtime = ?1, entry_count = ?2 WHERE id = ?3",
                    rusqlite::params![state.mtime, state.entry_count, folder_id],
                )?;
                tx.commit()?;
            }
        }

        on_progress(report.progress);
    }

    report.outcome = if cancelled {
        ScanOutcome::Cancelled
    } else if partial {
        ScanOutcome::PartialFailure
    } else {
        ScanOutcome::Complete
    };

    if report.outcome.may_prune() {
        let (folders, images) = prune_unreached(conn, root, generation)?;
        report.folders_removed = folders;
        report.images_removed += images;
    }

    Ok(report)
}

/// Take the next scan generation and record it against the root.
fn bump_generation(conn: &Connection, root: RootId, now: i64) -> Result<i64, CatalogError> {
    let current: Option<i64> = conn
        .query_row(
            "SELECT scan_generation FROM roots WHERE id = ?1",
            [root.0 as i64],
            |r| r.get(0),
        )
        .optional()?;
    let current = current.ok_or(CatalogError::NoSuchRoot(root.0))?;

    let next = current + 1;
    conn.execute(
        "UPDATE roots SET scan_generation = ?1, last_seen = ?2 WHERE id = ?3",
        rusqlite::params![next, now, root.0 as i64],
    )?;
    Ok(next)
}

/// TRACES: FR-CAT-9 | FR-PLAT-AND-2
/// Mark every image under a root as unreachable.
///
/// The other half of FR-CAT-9's distinction: a source *proven absent* may leave
/// the catalog, a source merely *unreachable* is marked offline and keeps its
/// ratings and its edits. The grid then says "offline" rather than showing a
/// library that has silently lost half its photographs.
///
/// The folders forget what they last looked like, which costs a full re-listing
/// when the drive comes back. That is the price of the marking: every row under
/// the root now claims something about the files that is no longer known to be
/// true, and only reading the directories again can settle it. Pruning would
/// skip them all and leave a plugged-in library showing as offline forever.
///
/// # Why the ETag goes with the mtime
///
/// The three columns are the same fact told by three kinds of storage: a local
/// directory proves it is unchanged with its mtime and entry count, and a
/// remote one proves it with a propagating ETag (ARCH §6.6). Clearing two of
/// them and leaving the third would disarm the re-listing on exactly the
/// libraries this is most likely to be called for — a remote scan prunes on
/// the ETag alone, so a root that came back would be walked, found unchanged
/// at every level, pruned whole, and left with every row still marked offline
/// and nothing that would ever clear the mark.
///
/// # Public, because losing a root is not only the local walk's business
///
/// This began as the private end of [`scan_root`]'s root-failure branches,
/// which is the only route a library reached through [`Storage`] can take.
/// The application does not currently take that route at all: it opens
/// libraries through `dr-sync`'s connectors, so the discovery happens in a
/// crate that cannot see this one's internals, and the correct response is
/// identical (FR-PLAT-AND-2). Exported rather than reimplemented beside the
/// caller that found out — a second copy would be a second thing to remember
/// when the ETag rule below changes.
///
/// [`Storage`]: dr_plat::Storage
pub fn mark_root_offline(conn: &Connection, root: RootId) -> Result<(), CatalogError> {
    let root_id = root.0 as i64;
    conn.execute(
        "UPDATE images SET availability = ?1 WHERE root_id = ?2 AND availability != ?1",
        rusqlite::params![availability_code(Availability::Offline), root_id],
    )?;
    conn.execute(
        "UPDATE folders SET mtime = NULL, entry_count = NULL, etag = NULL WHERE root_id = ?1",
        [root_id],
    )?;
    Ok(())
}

/// Record that a folder was reached in this generation, and report what the
/// last scan saw there.
///
/// The generation is written on arrival, before the folder is read: it means
/// "reached", which is what the end-of-scan sweep asks about. The `mtime` and
/// `entry_count` that mean "read" are written separately, and only afterwards.
fn touch_folder(
    conn: &Connection,
    root: RootId,
    dir: &DirRef,
    parent: Option<i64>,
    generation: i64,
) -> Result<(i64, Option<DirState>), CatalogError> {
    conn.execute(
        "INSERT INTO folders(root_id, parent_id, path, scanned_generation)
         VALUES (?1, ?2, ?3, ?4)
         ON CONFLICT(root_id, path) DO UPDATE SET
             scanned_generation = excluded.scanned_generation,
             parent_id          = excluded.parent_id",
        rusqlite::params![root.0 as i64, parent, dir.key(), generation],
    )?;

    let (id, mtime, count): (i64, Option<i64>, Option<i64>) = conn.query_row(
        "SELECT id, mtime, entry_count FROM folders WHERE root_id = ?1 AND path = ?2",
        rusqlite::params![root.0 as i64, dir.key()],
        |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
    )?;

    // Both or neither: a folder recorded by some other path (a remote scan
    // storing an ETag, say) has no local state, and half of one must not be
    // read as "unchanged since 1970".
    let stored = match (mtime, count) {
        (Some(mtime), Some(count)) => Some(DirState {
            mtime,
            entry_count: count as u32,
        }),
        _ => None,
    };
    Ok((id, stored))
}

/// The subfolders the catalog already knows about, as references to walk.
///
/// This is what makes pruning safe on a filesystem. A directory's mtime moves
/// when its own entries change and not when a grandchild's do, so an unchanged
/// folder says nothing about the tree below it; the stored keys are how the
/// walk carries on downward without listing.
fn known_children(
    conn: &Connection,
    root: RootId,
    folder_id: i64,
) -> Result<Vec<DirRef>, CatalogError> {
    let mut stmt = conn.prepare("SELECT path FROM folders WHERE parent_id = ?1")?;
    let rows = stmt.query_map([folder_id], |r| r.get::<_, String>(0))?;
    Ok(rows
        .flatten()
        .map(|path| DirRef::from_parts(root, path))
        .collect())
}

/// What the catalog holds for one image, as the scan needs it.
#[derive(Clone, Copy)]
struct Catalogued {
    id: i64,
    size: Option<u64>,
    mtime: Option<i64>,
    availability: i64,
    trashed: bool,
}

impl Catalogued {
    /// What [`classify_entry`] compares against.
    ///
    /// A row with no recorded size or mtime — inserted by a remote scan, or by
    /// an older build — is given values no real file can match, so it is
    /// classified `Changed` and re-read. The alternative, treating it as
    /// unknown, would insert a duplicate.
    fn as_known(&self) -> KnownFile {
        KnownFile {
            size: self.size.unwrap_or(u64::MAX),
            mtime: self.mtime.unwrap_or(i64::MIN),
        }
    }
}

/// Every image the catalog has in one folder, keyed by its stored reference.
///
/// Read once per folder rather than queried per file: a folder of 2,000 frames
/// is one statement and one hash lookup each, not 2,000 indexed selects.
fn catalogued_images(
    conn: &Connection,
    folder_id: i64,
) -> Result<HashMap<String, Catalogued>, CatalogError> {
    let mut stmt = conn.prepare(
        "SELECT id, source_ref, file_size, file_mtime, availability, trashed_at
           FROM images WHERE folder_id = ?1",
    )?;
    let rows = stmt.query_map([folder_id], |r| {
        Ok((
            r.get::<_, String>(1)?,
            Catalogued {
                id: r.get(0)?,
                size: r.get::<_, Option<i64>>(2)?.map(|v| v as u64),
                mtime: r.get(3)?,
                availability: r.get(4)?,
                trashed: r.get::<_, Option<i64>>(5)?.is_some(),
            },
        ))
    })?;
    Ok(rows.flatten().collect())
}

/// Put an unchanged file's availability back where the evidence says it should
/// be, and write nothing if it is already there.
///
/// The write has to be conditional: an unchanged file is the overwhelmingly
/// common case, and an `UPDATE` per image per scan would make a no-op rescan
/// write the whole library.
fn restore_availability(
    conn: &Connection,
    row: &Catalogued,
    src: &SourceRef,
) -> Result<(), CatalogError> {
    let should_be = availability_code(availability_of(src));
    if row.availability == should_be {
        return Ok(());
    }
    conn.execute(
        "UPDATE images SET availability = ?1 WHERE id = ?2",
        rusqlite::params![should_be, row.id],
    )?;
    Ok(())
}

/// How much of this image is on hand.
///
/// A local file is the original — unless it is a Nextcloud VFS placeholder, a
/// one-byte stub standing in for a dehydrated file whose bytes are on the
/// server (ARCH §9.0). Reading such a stub does not trigger a fetch on Linux,
/// so calling it `Original` would promise pixels that are not there. Catalogued
/// as the image it stands for, marked offline: honest, and FR-NC-6c's whole
/// point.
fn availability_of(src: &SourceRef) -> Availability {
    if src.is_placeholder() {
        Availability::Offline
    } else {
        Availability::Original
    }
}

fn insert_image(
    conn: &Connection,
    root: RootId,
    folder_id: i64,
    src: &SourceRef,
    size: u64,
    mtime: i64,
    now: i64,
) -> Result<(), CatalogError> {
    // `metadata_state = 1`: the scan knows the name, the size and the mtime,
    // and has read no EXIF. Claiming otherwise would make a date filter
    // silently wrong on a freshly scanned library.
    conn.execute(
        "INSERT INTO images(root_id, folder_id, source_ref, format, file_size, file_mtime,
                            availability, metadata_state, added_at)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 1, ?8)
         ON CONFLICT(root_id, source_ref) DO UPDATE SET
             folder_id    = excluded.folder_id,
             file_size    = excluded.file_size,
             file_mtime   = excluded.file_mtime,
             availability = excluded.availability",
        rusqlite::params![
            root.0 as i64,
            folder_id,
            src.key(),
            src.extension(),
            size as i64,
            mtime,
            availability_code(availability_of(src)),
            now,
        ],
    )?;
    Ok(())
}

fn update_image(
    conn: &Connection,
    root: RootId,
    folder_id: i64,
    src: &SourceRef,
    size: u64,
    mtime: i64,
) -> Result<(), CatalogError> {
    // The content hash is dropped, not recomputed: it described bytes that no
    // longer exist, and leaving it would let reconnect-by-hash match this image
    // to a file it is no longer a copy of. `metadata_state` goes back to 1 for
    // the same reason — the EXIF on record was read from the old file.
    conn.execute(
        "UPDATE images SET folder_id = ?1, file_size = ?2, file_mtime = ?3,
                           content_hash = NULL, metadata_state = 1, availability = ?4
          WHERE root_id = ?5 AND source_ref = ?6",
        rusqlite::params![
            folder_id,
            size as i64,
            mtime,
            availability_code(availability_of(src)),
            root.0 as i64,
            src.key(),
        ],
    )?;
    Ok(())
}

/// TRACES: FR-CAT-9
/// Delete the folders this scan did not reach, and the images inside them.
///
/// Runs only after a [`ScanOutcome::Complete`] scan, which is the guard that
/// stops an unplugged drive from taking the library with it.
///
/// Trashed images are detached first. Their folder may well be one of the ones
/// vanishing — the user deleted the whole directory — and the cascade would
/// take the row with it, orphaning a file sitting in the trash with no way back
/// (FR-CAT-15).
fn prune_unreached(
    conn: &Connection,
    root: RootId,
    generation: i64,
) -> Result<(usize, usize), CatalogError> {
    const STALE: &str = "SELECT id FROM folders WHERE root_id = ?1 AND scanned_generation < ?2";

    let tx = conn.unchecked_transaction()?;
    let root_id = root.0 as i64;
    let params = rusqlite::params![root_id, generation];

    tx.execute(
        &format!(
            "UPDATE images SET folder_id = NULL
              WHERE trashed_at IS NOT NULL AND folder_id IN ({STALE})"
        ),
        params,
    )?;

    let images: i64 = tx.query_row(
        &format!("SELECT count(*) FROM images WHERE folder_id IN ({STALE})"),
        params,
        |r| r.get(0),
    )?;
    let folders = tx.execute(
        "DELETE FROM folders WHERE root_id = ?1 AND scanned_generation < ?2",
        params,
    )?;
    tx.commit()?;

    Ok((folders, images as usize))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Catalog;
    use dr_plat::LocalStorage;
    use dr_types::Format;
    use std::cell::Cell;
    use std::fs;
    use std::path::PathBuf;

    /// A throwaway library on disk, removed when the test ends.
    ///
    /// The real [`LocalStorage`] rather than a fake, because what most needs
    /// protecting here is the seam between the two crates: a fake would agree
    /// with whatever this module expects, which is exactly the mistake a scan
    /// against a real directory cannot make.
    struct Library {
        dir: PathBuf,
        catalog: Catalog,
        root: RootId,
    }

    impl Library {
        fn new(name: &str) -> Self {
            let dir = std::env::temp_dir().join(format!(
                "dr-walk-{name}-{}-{:?}",
                std::process::id(),
                std::thread::current().id()
            ));
            let _ = fs::remove_dir_all(&dir);
            fs::create_dir_all(&dir).expect("temp dir");

            let catalog = Catalog::in_memory().expect("catalog");
            let root = ensure_root(
                catalog.connection(),
                RootKind::Local,
                &dir.display().to_string(),
            )
            .expect("root");

            Library { dir, catalog, root }
        }

        fn file(&self, rel: &str, bytes: &[u8]) -> &Self {
            let p = self.dir.join(rel);
            if let Some(parent) = p.parent() {
                fs::create_dir_all(parent).expect("mkdir");
            }
            fs::write(p, bytes).expect("write");
            self
        }

        /// Overwrite a file the way a program that cares does: write beside it,
        /// then rename over the top.
        ///
        /// Not incidental to the test — it is *why* the change is visible. A
        /// plain write moves nothing about the directory, and the scan would
        /// never look (see the module docs).
        fn resave(&self, rel: &str, bytes: &[u8]) -> &Self {
            let target = self.dir.join(rel);
            let tmp = target.with_extension("tmp");
            fs::write(&tmp, bytes).expect("write");
            fs::rename(&tmp, &target).expect("rename");
            // A scan tells a changed file by its mtime, at whole-second
            // resolution; a resave landing in the same second as the scan
            // before it looks unchanged, and the test fails when the machine
            // is fast enough. Two seconds ahead, on the file and its folder,
            // is what a real resave some time later would look like.
            let later = std::time::SystemTime::now() + std::time::Duration::from_secs(2);
            for p in [target.as_path(), target.parent().expect("parent")] {
                fs::File::open(p)
                    .and_then(|f| f.set_modified(later))
                    .expect("set mtime");
            }
            self
        }

        fn remove(&self, rel: &str) -> &Self {
            let p = self.dir.join(rel);
            if p.is_dir() {
                fs::remove_dir_all(p).expect("rmdir");
            } else {
                fs::remove_file(p).expect("rm");
            }
            self
        }

        fn storage(&self) -> LocalStorage {
            LocalStorage::with_root(self.root, self.dir.clone())
        }

        fn conn(&self) -> &Connection {
            self.catalog.connection()
        }

        fn scan(&self) -> ScanReport {
            self.scan_with(&FormatFilter::all(), &self.storage(), || false)
        }

        fn scan_with(
            &self,
            formats: &FormatFilter,
            storage: &dyn Storage,
            cancel: impl Fn() -> bool,
        ) -> ScanReport {
            scan_root(
                self.conn(),
                storage,
                self.root,
                formats,
                1_000,
                cancel,
                |_| {},
            )
            .expect("scan")
        }

        fn names(&self) -> Vec<String> {
            let mut stmt = self
                .conn()
                .prepare("SELECT source_ref FROM images ORDER BY source_ref")
                .unwrap();
            let rows = stmt.query_map([], |r| r.get::<_, String>(0)).unwrap();
            rows.flatten().collect()
        }

        fn count(&self, sql: &str) -> i64 {
            self.conn().query_row(sql, [], |r| r.get(0)).unwrap()
        }
    }

    impl Drop for Library {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.dir);
        }
    }

    #[test]
    fn a_folder_of_photographs_becomes_a_catalogued_library() {
        // The whole point of the exercise: before this existed, no image
        // reached the catalog without a Nextcloud account.
        let lib = Library::new("first-scan");
        lib.file("2026/08/IMG_0001.CR3", b"raw")
            .file("2026/08/IMG_0002.CR3", b"raw")
            .file("2025/IMG_0003.NEF", b"raw")
            .file("2025/notes.txt", b"not a photograph");

        let r = lib.scan();

        assert_eq!(r.outcome, ScanOutcome::Complete);
        assert_eq!(r.inserted, 3);
        assert_eq!(
            lib.names(),
            vec![
                "2025/IMG_0003.NEF",
                "2026/08/IMG_0001.CR3",
                "2026/08/IMG_0002.CR3"
            ]
        );
    }

    #[test]
    fn a_scanned_image_is_addressed_by_reference_not_by_path() {
        // FR-CAT-1a. The stored reference is relative to the root, so the
        // library can be moved, or reached through SAF where no path exists at
        // all, without rewriting a single row.
        let lib = Library::new("relative");
        lib.file("2026/IMG.CR3", b"raw");
        lib.scan();

        let stored = &lib.names()[0];
        assert_eq!(stored, "2026/IMG.CR3");
        assert!(
            !stored.contains(&lib.dir.display().to_string()),
            "an absolute path leaked into the catalog: {stored}"
        );
    }

    #[test]
    fn a_rescan_of_an_unchanged_library_reads_no_directory() {
        // NFR-P1 in one assertion. If this fails, every rescan is a full walk
        // and a 50k-image library takes minutes instead of seconds.
        let lib = Library::new("prune");
        lib.file("2026/08/IMG_0001.CR3", b"raw")
            .file("2025/IMG_0003.NEF", b"raw");

        let first = lib.scan();
        assert_eq!(first.progress.directories_listed, 4); // root, 2026, 2026/08, 2025

        let second = lib.scan();
        assert_eq!(second.progress.directories_listed, 0);
        assert_eq!(second.progress.directories_pruned, 4);
        assert_eq!(second.inserted, 0);
        assert_eq!(second.images_removed, 0);
        assert_eq!(second.outcome, ScanOutcome::Complete);
    }

    #[test]
    fn a_change_under_an_unchanged_parent_is_still_found() {
        // A filesystem propagates nothing upward: adding a file to `2026/08`
        // leaves `2026` and the root untouched. A scan that pruned at the first
        // unchanged folder would never see the new photograph — which is why
        // the walk carries on into known children rather than stopping.
        let lib = Library::new("grandchild");
        lib.file("2026/08/IMG_0001.CR3", b"raw");
        lib.scan();

        lib.file("2026/08/IMG_0002.CR3", b"raw");
        let r = lib.scan();

        assert_eq!(r.inserted, 1);
        assert_eq!(r.progress.directories_pruned, 2, "root and 2026");
        assert_eq!(r.progress.directories_listed, 1, "2026/08");
    }

    #[test]
    fn a_deleted_file_leaves_the_catalog() {
        let lib = Library::new("deleted");
        lib.file("a.CR3", b"raw").file("b.CR3", b"raw");
        lib.scan();

        lib.remove("b.CR3");
        let r = lib.scan();

        assert_eq!(r.images_removed, 1);
        assert_eq!(lib.names(), vec!["a.CR3"]);
    }

    #[test]
    fn a_deleted_folder_takes_its_images_with_it() {
        let lib = Library::new("deleted-folder");
        lib.file("2025/a.CR3", b"raw").file("2026/b.CR3", b"raw");
        lib.scan();

        lib.remove("2025");
        let r = lib.scan();

        assert_eq!(r.folders_removed, 1);
        assert_eq!(lib.names(), vec!["2026/b.CR3"]);
    }

    #[test]
    fn an_unreachable_root_marks_images_offline_and_deletes_nothing() {
        // The most expensive mistake this code could make. Every folder looks
        // unreached when a drive is unplugged, so a sweep would take the entire
        // library — ratings, flags and edits included (FR-CAT-9).
        let lib = Library::new("unplugged");
        lib.file("2026/IMG.CR3", b"raw");
        lib.scan();

        let unplugged = LocalStorage::with_root(lib.root, lib.dir.join("gone"));
        let r = lib.scan_with(&FormatFilter::all(), &unplugged, || false);

        assert_eq!(r.outcome, ScanOutcome::RootUnreachable);
        assert!(!r.outcome.may_prune());
        assert_eq!(lib.names().len(), 1, "the library was deleted");
        assert_eq!(
            lib.count("SELECT availability FROM images"),
            availability_code(Availability::Offline)
        );
    }

    /// TRACES: FR-PLAT-AND-2 | FR-CAT-9
    #[test]
    fn marking_a_root_offline_forgets_the_remote_validator_too() {
        // The half of the marking that only a remote library can notice, and
        // the reason it has to be here rather than beside the connector: a
        // remote scan prunes on the propagating ETag alone (ARCH §6.6). Clear
        // the local mtime and leave the ETag standing and a library that came
        // back would be walked, found unchanged at every level, pruned whole,
        // and left with every row still marked offline — with nothing that
        // would ever clear the mark, because clearing it is something only a
        // listing can do.
        //
        // Written directly because this module never writes an ETag; it is
        // `ui/dr-ui/src/library.rs`'s scan that does, against the same table.
        let lib = Library::new("etag-forgotten");
        lib.file("2026/IMG.CR3", b"raw");
        lib.scan();
        lib.conn()
            .execute(
                "UPDATE folders SET etag = 'e1' WHERE root_id = ?1",
                [lib.root.0 as i64],
            )
            .expect("etag");
        assert!(lib.count("SELECT COUNT(*) FROM folders WHERE etag IS NOT NULL") > 0);

        mark_root_offline(lib.conn(), lib.root).expect("mark");

        assert_eq!(
            lib.count("SELECT COUNT(*) FROM folders WHERE etag IS NOT NULL"),
            0,
            "an unreachable library must be re-listed, not pruned as unchanged"
        );
    }

    #[test]
    fn a_root_that_comes_back_is_available_again() {
        // The other half: a drive plugged back in must return the library to
        // usable, not leave it permanently marked offline.
        let lib = Library::new("replugged");
        lib.file("IMG.CR3", b"raw");
        lib.scan();

        let unplugged = LocalStorage::with_root(lib.root, lib.dir.join("gone"));
        lib.scan_with(&FormatFilter::all(), &unplugged, || false);
        lib.scan();

        assert_eq!(
            lib.count("SELECT availability FROM images"),
            availability_code(Availability::Original)
        );
    }

    #[test]
    fn a_cancelled_scan_deletes_nothing() {
        // Partial state is valid — jobs are resumable — but the folders that
        // were never visited must not be read as absent.
        let lib = Library::new("cancelled");
        lib.file("2025/a.CR3", b"raw")
            .file("2026/b.CR3", b"raw")
            .file("2027/c.CR3", b"raw");
        lib.scan();

        // Stop after the first directory.
        let seen = Cell::new(0);
        let r = lib.scan_with(&FormatFilter::all(), &lib.storage(), || {
            let n = seen.get();
            seen.set(n + 1);
            n > 0
        });

        assert_eq!(r.outcome, ScanOutcome::Cancelled);
        assert_eq!(r.images_removed, 0);
        assert_eq!(r.folders_removed, 0);
        assert_eq!(lib.names().len(), 3);
    }

    #[test]
    fn a_resaved_file_owes_a_reread_and_loses_its_stale_hash() {
        let lib = Library::new("resaved");
        lib.file("IMG.CR3", b"raw");
        lib.scan();

        lib.conn()
            .execute(
                "UPDATE images SET content_hash = 'abc', metadata_state = 2",
                [],
            )
            .unwrap();

        // Same name, different bytes: an export written over the original, or
        // a file restored from a backup. Both arrive by rename, which is what
        // makes them visible to an incremental scan.
        lib.resave("IMG.CR3", b"different bytes entirely");
        let r = lib.scan();

        assert_eq!(r.updated, 1);
        assert_eq!(
            lib.count("SELECT count(*) FROM images WHERE content_hash IS NULL"),
            1,
            "a hash of bytes that no longer exist would match this image to the \
             wrong file on reconnect"
        );
        assert_eq!(
            lib.count("SELECT metadata_state FROM images"),
            1,
            "EXIF must be re-read"
        );
    }

    #[test]
    fn a_file_that_has_not_moved_is_not_requeued_when_its_folder_is_reread() {
        // Adding one photograph to a folder of two thousand must cost one
        // thumbnail, not two thousand. Without this, every import rebuilds
        // everything around it.
        let lib = Library::new("no-requeue");
        lib.file("a.CR3", b"raw");
        lib.scan();
        // As if the metadata sweep had read it: what is owed is recorded in
        // `metadata_state`, and the thumbnail store answers for itself.
        lib.conn()
            .execute("UPDATE images SET metadata_state = 2", [])
            .unwrap();

        lib.file("b.CR3", b"raw");
        let r = lib.scan();

        assert_eq!(r.inserted, 1);
        assert_eq!(r.unchanged, 1);
        assert_eq!(
            lib.count("SELECT count(*) FROM images WHERE metadata_state < 2"),
            1,
            "EXIF owed for the new image, and nothing for the old one"
        );
    }

    #[test]
    fn a_file_rewritten_in_place_is_not_noticed_until_its_folder_changes() {
        // The documented limit of directory-level pruning, held by a test so it
        // stays a known trade and does not become a surprise. An in-place write
        // moves neither the directory's mtime nor its entry count, so nothing
        // says to look — and a scan that looked anyway would be a full walk of
        // the library every time.
        let lib = Library::new("in-place");
        lib.file("IMG.CR3", b"raw");
        lib.scan();

        fs::write(lib.dir.join("IMG.CR3"), b"rewritten in place, same folder").unwrap();
        assert_eq!(lib.scan().updated, 0);

        // Anything that touches the folder brings it back into view.
        lib.file("other.CR3", b"raw");
        assert_eq!(lib.scan().updated, 1);
    }

    #[test]
    fn a_new_image_owes_its_exif_and_queues_nothing() {
        // The sweeps find their work from `metadata_state` and the thumbnail
        // store. A queued job would be a second record of the same debt, and
        // no handler claims one (#73).
        let lib = Library::new("queued");
        lib.file("IMG.CR3", b"raw");
        lib.scan();

        assert_eq!(lib.count("SELECT count(*) FROM jobs"), 0);
        assert_eq!(
            lib.count("SELECT metadata_state FROM images"),
            1,
            "the scan read no EXIF and must not claim to have"
        );
    }

    #[test]
    fn the_format_filter_decides_what_is_catalogued() {
        let lib = Library::new("formats");
        lib.file("IMG.CR3", b"raw")
            .file("IMG.NEF", b"raw")
            .file("IMG.JPG", b"jpeg");

        let r = lib.scan_with(
            &FormatFilter::from_formats([Format::Cr3]),
            &lib.storage(),
            || false,
        );
        assert_eq!(r.inserted, 1);
        assert_eq!(lib.names(), vec!["IMG.CR3"]);
    }

    #[test]
    fn narrowing_the_filter_stops_finding_files_it_does_not_delete_them() {
        // Unticking JPEG is a statement about what to look for, not an
        // instruction to discard work. The files are sitting in the folder,
        // and absence is the only thing that justifies deleting a row — so a
        // tick-box must not quietly take a hundred rated photographs with it.
        let lib = Library::new("filter-narrowed");
        lib.file("IMG.CR3", b"raw").file("IMG.JPG", b"jpeg");
        lib.scan();
        assert_eq!(lib.names().len(), 2);

        // Something has to change for the folder to be listed at all.
        lib.file("NEW.CR3", b"raw");
        lib.scan_with(&FormatFilter::raw_only(), &lib.storage(), || false);

        assert!(
            lib.names().contains(&"IMG.JPG".to_string()),
            "a file that is still there was deleted by a filter change"
        );
    }

    #[test]
    fn a_dehydrated_placeholder_is_catalogued_as_the_image_it_stands_for() {
        // A Nextcloud-synced folder scanned as a local library is the common
        // case, and 121,785 of these exist in a real one (ARCH §9.0). Each is a
        // one-byte stub; reading it fetches nothing on Linux. Catalogued as a
        // CR2 and marked offline, so the grid says so rather than showing a
        // decode failure.
        let lib = Library::new("placeholder");
        lib.file("_MG_4130.CR2.nextcloud", b"\0")
            .file("_MG_4131.CR2", b"real bytes");
        lib.scan();

        assert_eq!(
            lib.count("SELECT availability FROM images WHERE source_ref LIKE '%.nextcloud'"),
            availability_code(Availability::Offline)
        );
        assert_eq!(
            lib.count("SELECT count(*) FROM images WHERE format = 'cr2'"),
            2,
            "the stub is a CR2, not an unknown '.nextcloud' type"
        );
        assert_eq!(
            lib.count("SELECT availability FROM images WHERE source_ref = '_MG_4131.CR2'"),
            availability_code(Availability::Original)
        );
    }

    #[test]
    fn the_trash_folder_is_never_scanned() {
        // The other half of the soft delete. A scan that walked the trash would
        // re-index every trashed photograph as an ordinary one, undoing the
        // delete on the next refresh (FR-CAT-15).
        let lib = Library::new("trash");
        lib.file("a.CR3", b"raw")
            .file(&format!("{TRASH_DIR}/1-b.CR3"), b"raw")
            .file(&format!("{DERIVED_DIR}/thumbs.sqlite"), b"x");
        lib.scan();

        assert_eq!(lib.names(), vec!["a.CR3"]);
        assert_eq!(
            lib.count(&format!(
                "SELECT count(*) FROM folders WHERE path LIKE '{TRASH_DIR}%'"
            )),
            0,
            "the trash must cost nothing, not merely be filtered out"
        );
    }

    #[test]
    fn a_trashed_image_survives_a_rescan_of_the_folder_it_left() {
        // Its file has been moved into the trash on purpose, so being absent
        // from its old folder is the expected state. Deleting the row would
        // lose `trashed_from` and with it the only way back (FR-CAT-15).
        let lib = Library::new("trashed-row");
        lib.file("2026/a.CR3", b"raw").file("2026/b.CR3", b"raw");
        lib.scan();

        // Trash `b` the way the UI does: move the file, then record.
        fs::create_dir_all(lib.dir.join(TRASH_DIR)).unwrap();
        fs::rename(
            lib.dir.join("2026/b.CR3"),
            lib.dir.join(format!("{TRASH_DIR}/2-b.CR3")),
        )
        .unwrap();
        lib.conn()
            .execute(
                "UPDATE images SET trashed_at = 100, trashed_from = source_ref,
                                   source_ref = ?1
                  WHERE source_ref = '2026/b.CR3'",
                [format!("{TRASH_DIR}/2-b.CR3")],
            )
            .unwrap();

        lib.scan();

        assert_eq!(
            lib.count("SELECT count(*) FROM images WHERE trashed_at IS NOT NULL"),
            1,
            "the trashed image was swept away by a rescan"
        );
    }

    #[test]
    fn a_trashed_image_outlives_the_folder_it_came_from() {
        // Same rule, reached through the folder sweep instead: the user deleted
        // the whole directory, and the cascade would have taken a row whose
        // file is sitting safely in the trash.
        let lib = Library::new("trashed-orphan");
        lib.file("2026/a.CR3", b"raw");
        lib.scan();

        fs::create_dir_all(lib.dir.join(TRASH_DIR)).unwrap();
        fs::rename(
            lib.dir.join("2026/a.CR3"),
            lib.dir.join(format!("{TRASH_DIR}/1-a.CR3")),
        )
        .unwrap();
        lib.conn()
            .execute(
                "UPDATE images SET trashed_at = 100, trashed_from = source_ref,
                                   source_ref = ?1
                  WHERE source_ref = '2026/a.CR3'",
                [format!("{TRASH_DIR}/1-a.CR3")],
            )
            .unwrap();
        lib.remove("2026");

        lib.scan();

        assert_eq!(
            lib.count("SELECT count(*) FROM images WHERE trashed_at IS NOT NULL"),
            1
        );
    }

    #[test]
    fn scanning_twice_does_not_fragment_the_library_across_two_roots() {
        // `ensure_root` is idempotent because a second row would split the
        // images between two roots and compare each half against the wrong scan
        // generation.
        let lib = Library::new("one-root");
        let again =
            ensure_root(lib.conn(), RootKind::Local, &lib.dir.display().to_string()).unwrap();

        assert_eq!(again, lib.root);
        assert_eq!(lib.count("SELECT count(*) FROM roots"), 1);
    }

    #[test]
    fn every_run_takes_a_fresh_generation() {
        // Not "once per completed scan": a cancelled run marks the folders it
        // reached, and reusing the number would let those marks look current to
        // the next sweep — so a folder deleted in between would survive it.
        let lib = Library::new("generations");
        lib.file("a.CR3", b"raw");

        lib.scan_with(&FormatFilter::all(), &lib.storage(), || true);
        let after_cancel = lib.count("SELECT scan_generation FROM roots");
        lib.scan();
        assert!(lib.count("SELECT scan_generation FROM roots") > after_cancel);
    }

    #[test]
    fn a_scan_reports_progress_as_it_goes() {
        // A first scan of a large library is minutes long; without this the UI
        // has nothing to say for all of it.
        let lib = Library::new("progress");
        lib.file("2025/a.CR3", b"raw").file("2026/b.CR3", b"raw");

        let mut updates = Vec::new();
        scan_root(
            lib.conn(),
            &lib.storage(),
            lib.root,
            &FormatFilter::all(),
            1_000,
            || false,
            |p| updates.push(p),
        )
        .unwrap();

        assert_eq!(updates.len(), 3, "one per directory visited");
        assert_eq!(updates.last().unwrap().images_found, 2);
    }

    #[test]
    fn scanning_a_root_with_no_catalog_row_is_a_typed_error() {
        // Rather than inventing a root: the label is the grant, and only the
        // caller that obtained the grant knows how to spell it.
        let lib = Library::new("no-root");
        let err = scan_root(
            lib.conn(),
            &lib.storage(),
            RootId(999),
            &FormatFilter::all(),
            0,
            || false,
            |_| {},
        );
        assert!(matches!(err, Err(CatalogError::NoSuchRoot(999))));
    }

    #[test]
    fn the_excluded_directories_match_the_ones_the_remote_scanner_excludes() {
        // Two pairs of constants in two crates that must agree, or a soft
        // delete comes undone on whichever side disagrees.
        assert!(is_excluded(TRASH_DIR));
        assert!(is_excluded(DERIVED_DIR));
        assert_eq!(DERIVED_DIR, dr_sync_derived_dir());
        assert!(!is_excluded("2026"));
        assert!(!is_excluded(".darkroom-trash-old"));
    }

    /// The scanner's constant, quoted rather than imported — `dr-catalog` does
    /// not depend on `dr-sync`, and adding that dependency for one string would
    /// invert the layering.
    fn dr_sync_derived_dir() -> &'static str {
        ".darkroom-derived"
    }
}