videre-core 0.47.0

//! The library database: the schema it is written with, the one entry point
//! allowed to create it, and the reader path that never creates anything.
//!
//! Only [`initialize`] creates the main database. It builds the schema in a
//! unique temporary file inside the library's own state directory, closes it
//! fully checkpointed, and publishes it by rename while holding the
//! library's activity (exclusive) and init locks, so no other videre process
//! ever observes a half-built library and no run can overwrite an existing
//! one. [`open_existing`] refuses an uninitialized library outright: it opens
//! without `CREATE`, so a reader can never bring a database (or even a lock
//! file) into being, the same readers-create-nothing rule the paths and
//! config layers follow.
//!
//! An existing database is checked before it is trusted: it must carry the
//! SQLite header, and if it has tables of its own but no `file_hashes` it is
//! somebody else's database and is refused unchanged. Its schema version
//! must be supported by this build: schema 3 is upgraded once under exclusive
//! locks to preserve monotonic face IDs, while older content-key schemas are
//! refused with an instruction to rebuild. Every refusal leaves the file
//! untouched. A fresh database's required tables and
//! columns are verified after preparation, so a helper that swallowed an
//! error cannot produce a false ready marker.
//!
//! Every indexed path in an opened database is validated against the
//! canonical root's components, once per context, before anything is
//! written and before any row is served: a database holding a path outside
//! its root belongs to a different library and is refused whole, with its
//! bytes untouched, because serving half of one library plus half of
//! another is silently wrong rather than loudly incomplete. Success is
//! memoized per context via the context's `index_validated` memo; failure
//! never is.

use crate::io_timeout::STAT_TIMEOUT;
use crate::library::{bounded_op, root_cause_is_not_found, LibraryContext};
use crate::library_locks::ActivityMode;
use anyhow::{bail, Context, Result};
use rusqlite::Connection;
use std::io::Read;
use std::path::{Component, Path};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

/// Schema 3 introduced the metadata-independent content key. Version 4 makes
/// face IDs monotonic so retained learning provenance cannot refer to a
/// different face after pruning.
const SCHEMA_VERSION: i64 = 4;

/// The first sixteen bytes of every SQLite database file. A library
/// candidate without them is not a SQLite file, whatever its name.
const SQLITE_MAGIC: [u8; 16] = *b"SQLite format 3\0";

/// How long every open waits for a concurrent writer before failing, so two
/// videre processes on one library surface as a bounded error rather than
/// the immediate `database is locked` default.
const BUSY_TIMEOUT: Duration = Duration::from_secs(5);

/// The complete current `file_hashes` schema: the columns the scan writes,
/// the newer optional columns older libraries predate, and the location
/// columns the location passes add. One list is the source for both the
/// `CREATE TABLE` and the add-missing upgrade, so the two cannot drift.
const FILE_HASHES_COLUMNS: &[(&str, &str)] = &[
    ("path", "TEXT PRIMARY KEY"),
    ("hash", "TEXT NOT NULL"),
    ("meta_hash", "TEXT"),
    ("size_bytes", "INTEGER"),
    ("created_at", "TEXT"),
    ("modified_at", "TEXT"),
    ("ext", "TEXT"),
    ("mime", "TEXT"),
    ("phash", "INTEGER"),
    ("exif_date", "TEXT"),
    ("gps_lat", "REAL"),
    ("gps_lon", "REAL"),
    ("width", "INTEGER"),
    ("height", "INTEGER"),
    ("duration_secs", "REAL"),
    ("codec", "TEXT"),
    ("location_name", "TEXT"),
    ("location_cluster_id", "INTEGER"),
    // Incremental XMP reconcile: three-state per row. NULL = never reconciled
    // (reconcile once); '' = reconciled, no sidecar present at the time; an
    // rfc3339 timestamp = the sidecar's mtime at the last reconcile. Compared
    // against the sidecar's current mtime to decide whether a re-import is
    // needed, so unchanged files are never re-read.
    ("xmp_sidecar_mtime", "TEXT"),
];

/// The complete current `faces` schema, verified after preparation.
const FACES_COLUMNS: &[(&str, &str)] = &[
    ("id", "INTEGER PRIMARY KEY"),
    ("hash", "TEXT NOT NULL"),
    ("bbox", "TEXT NOT NULL"),
    ("landmark", "TEXT"),
    ("embedding", "BLOB NOT NULL"),
    ("cluster_id", "INTEGER"),
    ("person_label", "TEXT"),
    ("confirmed", "INTEGER DEFAULT 0"),
    ("is_primary", "INTEGER DEFAULT 0"),
    ("det_score", "REAL"),
    ("blur", "REAL"),
    ("oriented", "INTEGER"),
];

/// The optional tables a prepared library carries, created by the same
/// helpers the commands use rather than a parallel copy of their DDL. All
/// are optional in the sense that a command that never runs leaves them
/// empty, never absent.
const REQUIRED_TABLES: &[&str] = &[
    "file_hashes",
    "people",
    "faces",
    "faces_scanned",
    "marks",
    "photo_tags",
    "classifications",
    "location_clusters",
    "pipeline_runs",
    "face_learning_events",
    "face_learning_event_faces",
    "face_learning_profiles",
    "face_learning_questions",
    "face_learning_question_faces",
];

/// Sequence counter making the build temporary's name unique within a
/// process; the pid makes it unique across processes.
static TMP_SEQ: AtomicU64 = AtomicU64::new(0);

/// Build the `file_hashes` DDL from the one column list, so the created
/// table and the schema checks are the same source of truth. The table
/// declares the location-cluster relationship so SQLite rejects an unknown
/// cluster reference instead of storing an orphan. `table` must be a trusted
/// static name (only ever a literal here): it is interpolated, not bound.
pub(crate) fn file_hashes_ddl_for(table: &str, if_not_exists: bool) -> String {
    let columns: Vec<String> = FILE_HASHES_COLUMNS
        .iter()
        .map(|(name, decl)| format!("    {name:<20} {decl}"))
        .collect();
    let exists = if if_not_exists { "IF NOT EXISTS " } else { "" };
    format!(
        "CREATE TABLE {exists}{table} (\n{}\n,    FOREIGN KEY (location_cluster_id) REFERENCES location_clusters(id)\n        ON DELETE RESTRICT ON UPDATE RESTRICT\n);",
        columns.join(",\n")
    )
}

/// Whether `table` has a column named `column`.
fn column_exists(conn: &Connection, table: &str, column: &str) -> rusqlite::Result<bool> {
    let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
    let names = stmt.query_map([], |row| row.get::<_, String>(1))?;
    for name in names {
        if name? == column {
            return Ok(true);
        }
    }
    Ok(false)
}

/// Add the expected columns that are absent, one explicit `ALTER` each,
/// decided by inspecting `PRAGMA table_info` rather than by attempting every
/// `ALTER` and swallowing the errors. The one tolerable failure is a column
/// that appeared between the inspection and the `ALTER` (another connection
/// upgraded first): present afterwards is success, still missing is a real
/// error, so no error is dismissed wholesale.
fn add_missing_columns(
    conn: &Connection,
    table: &str,
    expected: &[(&str, &str)],
) -> rusqlite::Result<()> {
    for (name, decl) in expected {
        if column_exists(conn, table, name)? {
            continue;
        }
        let alter = format!("ALTER TABLE {table} ADD COLUMN {name} {decl}");
        if let Err(e) = conn.execute_batch(&alter) {
            if !column_exists(conn, table, name)? {
                return Err(e);
            }
        }
    }
    Ok(())
}

/// Create or upgrade the scan table, the one schema implementation the
/// scanner calls: `crates/videre/src/sqlite_output.rs` delegates here rather
/// than owning a duplicate of the DDL. Idempotent, safe on every open, and
/// additive: an existing older table keeps its rows and gains only the
/// columns it lacks.
pub fn ensure_scan_schema(conn: &Connection) -> rusqlite::Result<()> {
    conn.execute_batch(&file_hashes_ddl_for("file_hashes", true))?;
    add_missing_columns(conn, "file_hashes", FILE_HASHES_COLUMNS)?;
    ensure_hash_index(conn)?;
    // The declared location-cluster key needs its parent table to exist
    // before any row is written, including through standalone scan writers
    // that never run the full schema preparation.
    crate::location_cluster::ensure_location_clusters_table(conn)
}

/// Verify the tables and columns a prepared library must have. Runs after
/// the DDL helpers, several of which swallow their own errors by design
/// (they were written for read paths), so a swallowed error surfaces here
/// instead of producing a ready marker over a missing table or column.
fn verify_schema(conn: &Connection) -> Result<()> {
    for table in REQUIRED_TABLES {
        if !crate::db::table_exists(conn, table)? {
            bail!("required table {table} is missing after schema preparation");
        }
    }
    let mut required: Vec<(&str, &str)> = Vec::new();
    for (name, _) in FILE_HASHES_COLUMNS {
        required.push(("file_hashes", name));
    }
    for (name, _) in FACES_COLUMNS {
        required.push(("faces", name));
    }
    required.push(("classifications", "model_id"));
    required.push(("classifications", "hash"));
    for (table, column) in &required {
        if !column_exists(conn, table, column)? {
            bail!("required column {table}.{column} is missing after schema preparation");
        }
    }
    let faces_sql: String = conn.query_row(
        "SELECT sql FROM sqlite_master WHERE type='table' AND name='faces'",
        [],
        |row| row.get(0),
    )?;
    anyhow::ensure!(
        faces_sql
            .to_ascii_uppercase()
            .contains("INTEGER PRIMARY KEY AUTOINCREMENT"),
        "faces.id must use AUTOINCREMENT"
    );
    Ok(())
}

/// The declared parent-child relationships exist: a table with the right
/// columns but no key is not a valid schema.
fn verify_foreign_keys(conn: &Connection) -> Result<()> {
    for (child, from, parent, to, on_update, on_delete) in [
        (
            "faces",
            "person_label",
            "people",
            "name",
            "RESTRICT",
            "RESTRICT",
        ),
        (
            "file_hashes",
            "location_cluster_id",
            "location_clusters",
            "id",
            "RESTRICT",
            "RESTRICT",
        ),
        (
            "face_learning_event_faces",
            "event_id",
            "face_learning_events",
            "id",
            "NO ACTION",
            "NO ACTION",
        ),
        (
            "face_learning_question_faces",
            "question_id",
            "face_learning_questions",
            "id",
            "NO ACTION",
            "NO ACTION",
        ),
    ] {
        let sql = format!("PRAGMA foreign_key_list({child})");
        let mut stmt = conn.prepare(&sql)?;
        let declared = stmt
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(3)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                ))
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;
        if !declared.contains(&(
            from.to_owned(),
            parent.to_owned(),
            to.to_owned(),
            on_update.to_owned(),
            on_delete.to_owned(),
        )) {
            bail!("missing required foreign key on {child}.{from} with {on_update} updates and {on_delete} deletes");
        }
    }
    Ok(())
}

/// Whether the schema is complete: every open refuses a library whose schema
/// is not.
fn schema_complete(conn: &Connection) -> Result<bool> {
    Ok(verify_schema(conn).is_ok() && verify_foreign_keys(conn).is_ok())
}

/// Index content hashes for scan-only libraries as well as embedded ones.
pub fn ensure_hash_index(conn: &Connection) -> rusqlite::Result<()> {
    conn.execute_batch("CREATE INDEX IF NOT EXISTS idx_file_hashes_hash ON file_hashes(hash);")
}

/// Prepare the complete supported schema on a fresh database: the scan table
/// plus the optional faces/people/marks/tags/classification/location/run
/// tables, created by the existing helpers the commands already use, then
/// verified.
fn prepare_schema(conn: &Connection) -> Result<()> {
    ensure_scan_schema(conn)?;
    // people comes with the faces schema; faces_scanned stays with the face
    // writers that own it.
    crate::face_db::ensure_faces_schema(conn)?;
    crate::face_db::create_faces_scanned_table(conn)?;
    crate::marks::ensure_marks_table(conn)?;
    crate::tags::ensure_photo_tags_table(conn)?;
    crate::classify::ensure_classifications_table(conn)?;
    crate::location_cluster::ensure_location_clusters_table(conn)?;
    crate::pipeline_runs::ensure_pipeline_runs_table(conn)?;
    crate::decode_failures::ensure_table(conn)?;
    crate::face_learning::ensure_learning_tables(conn)?;
    crate::face_learning::ensure_profile_table(conn)?;
    crate::face_learning::ensure_question_tables(conn)?;
    verify_schema(conn)?;
    Ok(())
}

/// The schema version recorded in the database file.
fn user_version(conn: &Connection) -> Result<i64> {
    let version: i64 = conn
        .query_row("PRAGMA user_version", [], |r| r.get(0))
        .context("read the library schema version")?;
    Ok(version)
}

/// Rebuild the only table whose row-ID policy changes. Historical journal and
/// question references deliberately have no face FK, so reserve their maximum
/// too. A failed copy or FK check rolls the entire version change back.
fn upgrade_v3_to_v4(conn: &Connection) -> Result<()> {
    conn.execute_batch("BEGIN IMMEDIATE")?;
    let result = (|| -> Result<()> {
        anyhow::ensure!(
            user_version(conn)? == 3,
            "expected schema 3 for face ID upgrade"
        );
        let max_id: i64 = conn
            .query_row(
                "SELECT MAX(id) FROM (
                SELECT id FROM faces
                UNION ALL SELECT face_id FROM face_learning_event_faces
                UNION ALL SELECT face_id FROM face_learning_question_faces
                UNION ALL SELECT representative_face_id FROM face_learning_questions
             )",
                [],
                |row| row.get::<_, Option<i64>>(0),
            )?
            .unwrap_or(0);
        conn.execute_batch("ALTER TABLE faces RENAME TO faces_v3_old;")?;
        conn.execute_batch(&crate::face_db::faces_ddl_for("faces", false))?;
        conn.execute_batch(
            "INSERT INTO faces
             (id,hash,bbox,landmark,embedding,cluster_id,person_label,confirmed,is_primary,det_score,blur,oriented)
             SELECT id,hash,bbox,landmark,embedding,cluster_id,person_label,confirmed,is_primary,det_score,blur,oriented
             FROM faces_v3_old;
             DROP TABLE faces_v3_old;",
        )?;
        conn.execute(
            "UPDATE sqlite_sequence SET seq = MAX(seq, ?1) WHERE name = 'faces'",
            [max_id],
        )?;
        conn.execute("INSERT INTO sqlite_sequence(name,seq) SELECT 'faces',?1 WHERE NOT EXISTS (SELECT 1 FROM sqlite_sequence WHERE name='faces')", [max_id])?;
        crate::face_learning::ensure_profile_table(conn)?;
        let violations: i64 =
            conn.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| {
                row.get(0)
            })?;
        anyhow::ensure!(
            violations == 0,
            "schema upgrade found {violations} foreign key violation(s)"
        );
        verify_schema(conn)?;
        conn.pragma_update(None, "user_version", SCHEMA_VERSION)?;
        Ok(())
    })();
    match result {
        Ok(()) => match conn.execute_batch("COMMIT") {
            Ok(()) => Ok(()),
            Err(error) => {
                let _ = conn.execute_batch("ROLLBACK");
                Err(error).context("commit schema 4 upgrade")
            }
        },
        Err(error) => {
            let _ = conn.execute_batch("ROLLBACK");
            Err(error.context("upgrade library face IDs to schema 4"))
        }
    }
}

/// Whether the file begins with the SQLite header. The caller has already
/// handled existence and size; this is the supported-library check that
/// comes before anything else reads the file as a library.
fn is_sqlite_file(path: &Path) -> Result<bool> {
    let owned = path.to_path_buf();
    match bounded_op(path, "read", STAT_TIMEOUT, move || {
        let mut file = std::fs::File::open(&owned)?;
        let mut header = [0u8; 16];
        // A file shorter than the header is not a database either; the
        // mismatch is folded into the answer rather than surfaced as an
        // error, because the caller's question is exactly "is this one?".
        match file.read_exact(&mut header) {
            Ok(()) => Ok(Some(header)),
            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => Ok(None),
            Err(e) => Err(e),
        }
    }) {
        Ok(Some(header)) => Ok(header == SQLITE_MAGIC),
        Ok(None) => Ok(false),
        Err(e) => Err(e),
    }
}

/// What an inspection of the database file found.
enum DbFile {
    /// No file: nothing has been initialized.
    Absent,
    /// A zero-byte file: the debris of an interrupted creation, never a
    /// library, and never exposed as successfully initialized.
    ZeroByte,
    /// A non-empty file carrying the SQLite header.
    Present,
}

/// Inspect the database file, refusing redirected state on the way in: a
/// symlinked or multiply-linked database would silently couple two
/// libraries, and neither may be used.
fn inspect_db_file(ctx: &LibraryContext) -> Result<DbFile> {
    crate::library_locks::reject_redirect(&ctx.paths.db, "the library database")?;
    let owned = ctx.paths.db.clone();
    let len = match bounded_op(&ctx.paths.db, "read", STAT_TIMEOUT, move || {
        std::fs::symlink_metadata(&owned).map(|m| m.len())
    }) {
        Ok(len) => len,
        Err(e) if root_cause_is_not_found(&e) => return Ok(DbFile::Absent),
        Err(e) => return Err(e),
    };
    if len == 0 {
        return Ok(DbFile::ZeroByte);
    }
    if !is_sqlite_file(&ctx.paths.db)? {
        bail!(
            "{} is not a SQLite database; not a videre library",
            ctx.paths.db.display()
        );
    }
    Ok(DbFile::Present)
}

/// Open a connection without `CREATE`: opening must never bring a database
/// into being, so readers and signal handlers reach for this rather than
/// `Connection::open`. `NOFOLLOW` backs up the lstat redirect checks with
/// the kernel's own refusal to open a symlink's target; `NO_MUTEX` matches
/// the default `rusqlite::Connection::open` uses, leaving threading to the
/// caller. Foreign keys are verified on, so every write (the signal-handler
/// path's included) runs under the same enforcement as every other library
/// connection.
pub(crate) fn open_without_create(path: &Path) -> rusqlite::Result<Connection> {
    use rusqlite::OpenFlags;
    let conn = Connection::open_with_flags(
        path,
        OpenFlags::SQLITE_OPEN_READ_WRITE
            | OpenFlags::SQLITE_OPEN_NO_MUTEX
            | OpenFlags::SQLITE_OPEN_NOFOLLOW,
    )?;
    crate::db::enable_foreign_keys(&conn)?;
    Ok(conn)
}

/// Open an existing database with the busy timeout every library open uses.
fn open_existing_conn(ctx: &LibraryContext) -> Result<Connection> {
    let conn = open_without_create(&ctx.paths.db)
        .with_context(|| format!("open {}", ctx.paths.db.display()))?;
    conn.busy_timeout(BUSY_TIMEOUT)
        .context("set the library database busy timeout")?;
    Ok(conn)
}

fn open_existing_read_only_conn(ctx: &LibraryContext) -> Result<Connection> {
    use rusqlite::OpenFlags;
    let conn = Connection::open_with_flags(
        &ctx.paths.db,
        OpenFlags::SQLITE_OPEN_READ_ONLY
            | OpenFlags::SQLITE_OPEN_NO_MUTEX
            | OpenFlags::SQLITE_OPEN_NOFOLLOW,
    )
    .with_context(|| format!("open {} read-only", ctx.paths.db.display()))?;
    conn.busy_timeout(BUSY_TIMEOUT)
        .context("set the library database busy timeout")?;
    // Enforcement is a per-connection switch and works on read-only handles;
    // the pragma is never persisted, so nothing about the file changes.
    crate::db::enable_foreign_keys(&conn)
        .context("enable foreign keys on the read-only library connection")?;
    Ok(conn)
}

/// The WAL behaviour every library open applies, idempotent and persistent
/// in the file (see `db::open_wal` for why it matters).
fn set_wal(conn: &Connection) -> Result<()> {
    conn.pragma_update(None, "journal_mode", "WAL")
        .context("set the library database to WAL journal mode")
}

/// Refuse a SQLite file that is not a videre library: tables of its own but
/// no `file_hashes` means the database belongs to some other program, and
/// preparing videre's schema inside it would corrupt both.
fn require_supported_library(conn: &Connection, db: &Path) -> Result<()> {
    if crate::db::table_exists(conn, "file_hashes")? {
        return Ok(());
    }
    let user_tables: i64 = conn.query_row(
        "SELECT count(*) FROM sqlite_master
         WHERE type = 'table' AND name NOT LIKE 'sqlite_%'",
        [],
        |r| r.get(0),
    )?;
    anyhow::ensure!(
        user_tables == 0,
        "{} is not a videre library: a SQLite database with tables of its own and no file_hashes",
        db.display()
    );
    Ok(())
}

// Counts rows the containment validation has judged, in tests, so a scale
// test can prove the validation really ran over every row while the guard
// layer's resolution counter stays flat: zero per-row filesystem probes.
// Thread-local for the same reason as library_guard's counter: libtest runs
// each test on its own thread, so a process-global counter would attribute
// neighbouring tests' validations to whoever read it.
#[cfg(test)]
thread_local! {
    static ROWS_VALIDATED: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}

/// Test-only read of the containment row counter.
#[cfg(test)]
pub(crate) fn count_rows_validated() -> u64 {
    ROWS_VALIDATED.with(std::cell::Cell::get)
}

/// Test-only bump of the containment row counter.
#[cfg(test)]
fn note_row_validated() {
    ROWS_VALIDATED.with(|c| c.set(c.get() + 1));
}

/// Validate every indexed path against the canonical root, once per context,
/// before any write and before any row is served.
///
/// Containment is by path components, not string prefix: `<root>-sibling`
/// starts with the same bytes as `<root>` but is a different directory, and
/// a prefix comparison would let it through. A `.` or `..` component is
/// refused the same way, because `<root>/../outside` does start with
/// `<root>` lexically and videre never writes such a path, so a row carrying
/// one is hand-injected and foreign. A file that no longer exists
/// under the root remains valid, because the question is which library the
/// row belongs to, not whether its media is currently mounted. A foreign row
/// refuses the whole library: serving it would mix two libraries silently.
/// Success is memoized via the context's `index_validated` memo, never
/// failure, so a refused library is rechecked on every attempt.
fn validate_row_containment(ctx: &LibraryContext, conn: &Connection) -> Result<()> {
    if ctx.index_validated() {
        return Ok(());
    }
    if crate::db::table_exists(conn, "file_hashes")? {
        let root = ctx.paths.root.clone();
        let mut stmt = conn
            .prepare("SELECT path FROM file_hashes")
            .context("read the indexed paths")?;
        let mut rows = stmt.query([])?;
        while let Some(row) = rows.next()? {
            let path: String = row.get(0)?;
            #[cfg(test)]
            note_row_validated();
            let stored = Path::new(&path);
            let dot_component = stored
                .components()
                .any(|component| matches!(component, Component::CurDir | Component::ParentDir));
            if dot_component || !stored.starts_with(&root) {
                bail!(
                    "the database at {} indexes {}, which is outside the library root {}; it belongs to a different library, so nothing was read or changed",
                    ctx.paths.db.display(),
                    path,
                    root.display()
                );
            }
        }
    }
    ctx.mark_index_validated();
    Ok(())
}

/// The refusal for a database written by a newer videre, shared by every
/// open path so the wording and its kind cannot drift apart.
fn newer_schema(ctx: &LibraryContext, version: i64) -> anyhow::Error {
    anyhow::anyhow!(
        "the library at {} was written by a newer videre (schema version {version}, this build understands up to {SCHEMA_VERSION}); upgrade videre to open it",
        ctx.paths.root.display()
    )
    .context(crate::error_kind::ErrorKind::LibrarySchema)
}

/// Refuse a library this build cannot use as it is. Libraries written before
/// the content key (schema 3) are not migrated: their stored keys were
/// computed over whole files, so nothing in them matches a new scan, and the
/// library is built again from its photos. An incomplete schema is refused
/// the same way; videre only ever publishes a complete one.
fn require_current(ctx: &LibraryContext, conn: &Connection) -> Result<()> {
    let version = user_version(conn)?;
    if version > SCHEMA_VERSION {
        return Err(newer_schema(ctx, version));
    }
    if version == 3 {
        return Err(anyhow::anyhow!(
            "the library at {} uses schema 3; run `videre scan` or another writable command to upgrade face IDs before opening it read-only",
            ctx.paths.root.display()
        ).context(crate::error_kind::ErrorKind::LibrarySchema));
    }
    if version < SCHEMA_VERSION || !schema_complete(conn)? {
        return Err(anyhow::anyhow!(
            "the library at {} was created by an older videre, which identified files differently; \
             remove {} and run `videre scan` to build it again",
            ctx.paths.root.display(),
            ctx.paths.state.display()
        )
        .context(crate::error_kind::ErrorKind::LibrarySchema));
    }
    Ok(())
}

/// Finish opening an existing database while the caller holds the
/// appropriate locks: refuse foreign databases, foreign rows and any schema
/// but the current one (all read-only checks, so a refusal leaves the file
/// untouched), then enable WAL and foreign keys.
fn open_prepared(ctx: &LibraryContext, conn: &Connection) -> Result<()> {
    require_supported_library(conn, &ctx.paths.db)?;
    validate_row_containment(ctx, conn)?;
    require_current(ctx, conn)?;
    set_wal(conn)?;
    crate::db::enable_foreign_keys(conn)
        .context("enable foreign keys on the library connection")?;
    Ok(())
}

/// Remove one path, bounded, naming what it was in the error.
fn remove_file_bounded(path: &Path, what: &str) -> Result<()> {
    let owned = path.to_path_buf();
    bounded_op(path, "remove", STAT_TIMEOUT, move || {
        std::fs::remove_file(&owned)
    })
    .with_context(|| format!("remove {what} {}", path.display()))
}

/// Sync a directory so a rename inside it is durable, the step after the
/// rename that a crash would otherwise undo. Shared with the config layer,
/// whose scratch rename needs the same crash durability.
pub(crate) fn sync_dir(path: &Path) -> Result<()> {
    let owned = path.to_path_buf();
    bounded_op(path, "sync", STAT_TIMEOUT, move || {
        let dir = std::fs::File::open(&owned)?;
        dir.sync_all()
    })
    .with_context(|| format!("sync {}", path.display()))
}

/// Remove build temporaries left behind by failed or crashed
/// initializations. Safe under the activity and init locks, which every
/// caller of [`publish_fresh`] holds: no other videre process can be
/// building one at the same moment, so a matching file is by definition
/// dead. A crashed build can leave not only the temporary database but its
/// WAL sidecars (`.tmp-wal`, `.tmp-shm`), so every name carrying a temp
/// build stem is swept, not only the one ending in `.tmp`. Best-effort: a
/// temporary that cannot be removed must not fail the initialization that
/// is about to publish a fresh database.
fn sweep_stale_builds(ctx: &LibraryContext) {
    let state = ctx.paths.state.clone();
    let names = bounded_op(&ctx.paths.state, "read", STAT_TIMEOUT, move || {
        let mut out = Vec::new();
        for entry in std::fs::read_dir(&state)? {
            out.push(entry?.file_name());
        }
        Ok(out)
    });
    if let Ok(names) = names {
        for name in names {
            let text = name.to_string_lossy();
            let stale = text.starts_with("hashes.db.")
                && (text.ends_with(".tmp")
                    || text.ends_with(".tmp-wal")
                    || text.ends_with(".tmp-shm"));
            if stale {
                let _ =
                    remove_file_bounded(&ctx.paths.state.join(&name), "a stale build temporary");
            }
        }
    }
}

/// Publish one completed database build into its final path.
///
/// Kept as one operation so the no-overwrite contract can be tested at the
/// same boundary initialization uses.
fn publish_database(from: &Path, to: &Path) -> Result<()> {
    let rename_from = from.to_path_buf();
    let rename_to = to.to_path_buf();
    bounded_op(to, "publish", STAT_TIMEOUT, move || {
        rustix::fs::renameat_with(
            rustix::fs::CWD,
            &rename_from,
            rustix::fs::CWD,
            &rename_to,
            rustix::fs::RenameFlags::NOREPLACE,
        )
        .map_err(Into::into)
    })
    .with_context(|| format!("publish {}", to.display()))
}

/// Build a fresh database in a unique state-local temporary file, close it
/// fully checkpointed, and publish it by rename without overwriting an
/// existing destination. Only called with the library's activity (exclusive)
/// and init locks held, which is what makes the absent-destination check
/// and the rename one uninterrupted decision no other videre process can
/// interleave with. On failure the temporary is deliberately left in place:
/// removing it would touch the volume that just failed, and the next
/// successful initialization sweeps it.
fn publish_fresh(ctx: &LibraryContext) -> Result<Connection> {
    sweep_stale_builds(ctx);
    let tmp = ctx.paths.state.join(format!(
        "hashes.db.{}.{}.tmp",
        std::process::id(),
        TMP_SEQ.fetch_add(1, Ordering::Relaxed)
    ));
    let build = || -> Result<()> {
        let conn = Connection::open(&tmp).with_context(|| format!("build {}", tmp.display()))?;
        conn.busy_timeout(BUSY_TIMEOUT)?;
        prepare_schema(&conn)?;
        conn.pragma_update(None, "user_version", SCHEMA_VERSION)
            .context("record the library schema version")?;
        set_wal(&conn)?;
        // Checkpoint and close fully, so what gets published is one
        // self-contained file, never a half-written WAL pair.
        let _ = conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);");
        Ok(())
    };
    build()?;
    let len = {
        let owned = tmp.clone();
        bounded_op(&tmp, "read", STAT_TIMEOUT, move || {
            std::fs::metadata(&owned).map(|m| m.len())
        })
        .with_context(|| format!("read {}", tmp.display()))?
    };
    anyhow::ensure!(
        len > 0,
        "the built database is empty; refusing to publish it"
    );
    anyhow::ensure!(
        is_sqlite_file(&tmp)?,
        "the built database does not carry the SQLite header; refusing to publish it"
    );
    // Publish only onto an absent destination: under the locks, a
    // destination that appeared is not ours to replace.
    crate::library_locks::reject_redirect(&ctx.paths.db, "the library database")?;
    let owned = ctx.paths.db.clone();
    let exists = match bounded_op(&ctx.paths.db, "read", STAT_TIMEOUT, move || {
        std::fs::symlink_metadata(&owned).map(|_| ())
    }) {
        Ok(()) => true,
        Err(e) if root_cause_is_not_found(&e) => false,
        Err(e) => return Err(e),
    };
    anyhow::ensure!(
        !exists,
        "a database appeared at {} while the library was being initialized",
        ctx.paths.db.display()
    );
    publish_database(&tmp, &ctx.paths.db)?;
    sync_dir(&ctx.paths.state)?;
    let conn = open_existing_conn(ctx)?;
    open_prepared(ctx, &conn)?;
    Ok(conn)
}

/// Initialize the library's database: the only entry point that may create
/// it, its state directory, or its lock files.
///
/// Creates the state directories, takes the library's activity lock
/// exclusive and then its init lock (the acquisition order every writer
/// follows), and either publishes a freshly built database by rename or
/// opens an existing one, which must be current (an older one is refused). The config is
/// written last, read-before-write: exactly the six declarations of a
/// fresh library, and only when no config exists, so a repeated initialize
/// never rewrites what a user or a previous run wrote.
pub fn initialize(ctx: &LibraryContext) -> Result<Connection> {
    crate::library_locks::ensure_state_and_locks(ctx)?;
    let _activity = crate::library_locks::try_activity(ctx, ActivityMode::Exclusive)
        .with_context(|| format!("initialize {}", ctx.paths.root.display()))?;
    let _init = crate::library_locks::try_init(ctx)?;
    // Identity rechecked under the locks: the root or its .videre may have
    // been replaced while the directories were being created.
    crate::library_locks::verify_state(ctx)?;
    crate::library_locks::reject_redirect(&ctx.paths.config, "the library config")?;
    crate::library_locks::reject_dir_redirect(&ctx.paths.embeddings, "the embeddings directory")?;
    let conn = match inspect_db_file(ctx)? {
        DbFile::Absent => publish_fresh(ctx)?,
        DbFile::ZeroByte => {
            // Provably empty, so replacing it cannot lose anything, and a
            // zero-byte file must never be exposed as a library.
            remove_file_bounded(&ctx.paths.db, "the empty database")?;
            publish_fresh(ctx)?
        }
        DbFile::Present => {
            let conn = open_existing_conn(ctx)?;
            if user_version(&conn)? == 3 {
                require_supported_library(&conn, &ctx.paths.db)?;
                validate_row_containment(ctx, &conn)?;
                upgrade_v3_to_v4(&conn)?;
            }
            open_prepared(ctx, &conn)?;
            conn
        }
    };
    crate::library_config::write_initial_if_absent(ctx)?;
    Ok(conn)
}

/// Open the library's existing database, creating nothing: no database, no
/// lock file, not even the locks directory.
///
/// The open normally holds the activity lock shared while inspecting the
/// library. A schema-3 database needs one upgrade: release that lock, then
/// take activity-exclusive and init locks before reopening and migrating.
/// This is a writable handle even when called by a read-oriented command;
/// [`open_existing_read_only`] never takes this path.
pub fn open_existing(ctx: &LibraryContext) -> Result<Connection> {
    crate::library_locks::verify_state(ctx)?;
    let activity = crate::library_locks::try_activity(ctx, ActivityMode::Shared)?;
    crate::library_locks::reject_redirect(&ctx.paths.config, "the library config")?;
    crate::library_locks::reject_dir_redirect(&ctx.paths.embeddings, "the embeddings directory")?;
    match inspect_db_file(ctx)? {
        DbFile::Present => {}
        DbFile::Absent => bail!(
            "library {} has not been initialized: no database at {}",
            ctx.paths.root.display(),
            ctx.paths.db.display()
        ),
        DbFile::ZeroByte => bail!(
            "library {} was never fully initialized: {} is empty; initialize it (for example with videre scan)",
            ctx.paths.root.display(),
            ctx.paths.db.display()
        ),
    }
    // Every refusal below leaves the file exactly as it was, except for the
    // one supported schema upgrade.
    let conn = open_existing_conn(ctx)?;
    if user_version(&conn)? == 3 {
        require_supported_library(&conn, &ctx.paths.db)?;
        validate_row_containment(ctx, &conn)?;
        drop(conn);
        drop(activity);
        // Locks refuse rather than wait, so a videre still running in this
        // library (an older `watch` or gallery) stops the upgrade: say so.
        let upgrade_blocked = || {
            format!(
                "the library at {} needs a one-time upgrade to schema 4, which needs \
                 every other videre command in it stopped first; stop them and retry",
                ctx.paths.root.display()
            )
        };
        let _exclusive = crate::library_locks::try_activity(ctx, ActivityMode::Exclusive)
            .with_context(upgrade_blocked)?;
        let _init = crate::library_locks::try_init(ctx).with_context(upgrade_blocked)?;
        crate::library_locks::verify_state(ctx)?;
        let conn = open_existing_conn(ctx)?;
        require_supported_library(&conn, &ctx.paths.db)?;
        validate_row_containment(ctx, &conn)?;
        if user_version(&conn)? == 3 {
            upgrade_v3_to_v4(&conn)?;
        }
        open_prepared(ctx, &conn)?;
        return Ok(conn);
    }
    open_prepared(ctx, &conn)?;
    Ok(conn)
}

/// Opens a complete current library without creating or changing its
/// database. Anything else is refused.
pub fn open_existing_read_only(ctx: &LibraryContext) -> Result<Connection> {
    crate::library_locks::verify_state(ctx)?;
    let _activity = crate::library_locks::try_activity(ctx, ActivityMode::Shared)?;
    crate::library_locks::reject_redirect(&ctx.paths.config, "the library config")?;
    crate::library_locks::reject_dir_redirect(&ctx.paths.embeddings, "the embeddings directory")?;
    match inspect_db_file(ctx)? {
        DbFile::Present => {}
        DbFile::Absent => bail!(
            "library {} has not been initialized: no database at {}",
            ctx.paths.root.display(),
            ctx.paths.db.display()
        ),
        DbFile::ZeroByte => bail!(
            "library {} was never fully initialized: {} is empty; initialize it (for example with videre scan)",
            ctx.paths.root.display(),
            ctx.paths.db.display()
        ),
    }

    let conn = open_existing_read_only_conn(ctx)?;
    require_supported_library(&conn, &ctx.paths.db)?;
    validate_row_containment(ctx, &conn)?;
    require_current(ctx, &conn)?;
    Ok(conn)
}

#[cfg(test)]
mod tests {
    use crate::db::enable_foreign_keys;

    fn make_v3(conn: &Connection) {
        conn.execute_batch("DROP TABLE faces;").unwrap();
        let ddl = crate::face_db::faces_ddl_for("faces", false)
            .replace("INTEGER PRIMARY KEY AUTOINCREMENT", "INTEGER PRIMARY KEY");
        conn.execute_batch(&ddl).unwrap();
        conn.pragma_update(None, "user_version", 3).unwrap();
    }

    #[test]
    fn v3_upgrade_preserves_faces_and_reserves_historical_ids() {
        let (_temp, ctx) = library();
        let conn = initialize(&ctx).unwrap();
        make_v3(&conn);
        conn.execute_batch(
            "INSERT INTO people(name,full_name) VALUES('isil','Işıl');
             INSERT INTO faces(id,hash,bbox,embedding,person_label,confirmed)
                 VALUES(7,'h','0,0,1,1',X'00','isil',1);
             INSERT INTO face_learning_events
                 (id,action_kind,decision_kind,outcome,embedding_model_id,feature_schema_version,feature_snapshot_json,support_count)
                 VALUES(1,'assign','user','accepted','test',1,'{}',0);
             INSERT INTO face_learning_event_faces(event_id,face_id,role,ordinal)
                 VALUES(1,80,'subject',0);
             INSERT INTO face_learning_questions
                 (id,status,target_identity,profile_id,model_kind,representative_face_id,cluster_id,evidence_revision,evidence_json)
                 VALUES(1,'pending','isil',1,'test',90,1,'rev','{}');",
        ).unwrap();
        drop(conn);

        let conn = open_existing(&ctx).unwrap();
        assert_eq!(user_version(&conn).unwrap(), SCHEMA_VERSION);
        assert!(conn
            .query_row(
                "SELECT sql FROM sqlite_master WHERE name='faces'",
                [],
                |r| r.get::<_, String>(0)
            )
            .unwrap()
            .contains("AUTOINCREMENT"));
        assert_eq!(
            conn.query_row("SELECT person_label FROM faces WHERE id=7", [], |r| r
                .get::<_, String>(0))
                .unwrap(),
            "isil"
        );
        let seq: i64 = conn
            .query_row(
                "SELECT seq FROM sqlite_sequence WHERE name='faces'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(seq >= 90, "historical ID was not reserved: {seq}");
        conn.execute(
            "INSERT INTO faces(hash,bbox,embedding) VALUES('next','0,0,1,1',X'00')",
            [],
        )
        .unwrap();
        assert!(conn.last_insert_rowid() > 90);
        assert!(conn
            .prepare("PRAGMA foreign_key_check")
            .unwrap()
            .query([])
            .unwrap()
            .next()
            .unwrap()
            .is_none());
        drop(conn);
        assert_eq!(
            user_version(&open_existing(&ctx).unwrap()).unwrap(),
            SCHEMA_VERSION
        );
    }

    #[test]
    fn v3_upgrade_rolls_back_on_failure() {
        let (_temp, ctx) = library();
        let conn = initialize(&ctx).unwrap();
        make_v3(&conn);
        conn.execute_batch("PRAGMA foreign_keys=OFF;
            INSERT INTO faces(id,hash,bbox,embedding,person_label) VALUES(5,'bad','0,0,1,1',X'00','absent');
            PRAGMA foreign_keys=ON;").unwrap();
        drop(conn);
        assert!(open_existing(&ctx).is_err());
        let conn = open_without_create(&ctx.paths.db).unwrap();
        assert_eq!(user_version(&conn).unwrap(), 3);
        assert_eq!(
            conn.query_row("SELECT person_label FROM faces WHERE id=5", [], |r| r
                .get::<_, String>(0))
                .unwrap(),
            "absent"
        );
        assert!(!conn
            .query_row(
                "SELECT sql FROM sqlite_master WHERE name='faces'",
                [],
                |r| r.get::<_, String>(0)
            )
            .unwrap()
            .contains("AUTOINCREMENT"));
    }

    /// Another videre still at work in the library (here, a held shared
    /// lease) blocks the upgrade; the refusal says why and changes nothing.
    #[test]
    fn a_busy_library_explains_the_pending_upgrade() {
        let (_temp, ctx) = library();
        let conn = initialize(&ctx).unwrap();
        make_v3(&conn);
        drop(conn);
        let _other = crate::library_locks::try_activity(&ctx, ActivityMode::Shared).unwrap();
        let error = open_existing(&ctx).unwrap_err();
        assert!(
            format!("{error:#}").contains("one-time upgrade to schema 4"),
            "{error:#}"
        );
        assert_eq!(
            user_version(&open_without_create(&ctx.paths.db).unwrap()).unwrap(),
            3
        );
    }

    #[test]
    fn read_only_v3_refuses_without_writing() {
        let (_temp, ctx) = library();
        let conn = initialize(&ctx).unwrap();
        make_v3(&conn);
        conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")
            .unwrap();
        drop(conn);
        let before = std::fs::read(&ctx.paths.db).unwrap();
        let error = open_existing_read_only(&ctx).unwrap_err();
        assert!(format!("{error:#}").contains("videre scan"), "{error:#}");
        assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), before);
        assert_eq!(
            user_version(&open_without_create(&ctx.paths.db).unwrap()).unwrap(),
            3
        );
    }

    #[test]
    fn fresh_v4_face_ids_never_reuse_deleted_max() {
        let (_temp, ctx) = library();
        let conn = initialize(&ctx).unwrap();
        conn.execute(
            "INSERT INTO faces(hash,bbox,embedding) VALUES('first','0,0,1,1',X'00')",
            [],
        )
        .unwrap();
        let first = conn.last_insert_rowid();
        conn.execute("DELETE FROM faces WHERE id=?1", [first])
            .unwrap();
        conn.execute(
            "INSERT INTO faces(hash,bbox,embedding) VALUES('next','0,0,1,1',X'00')",
            [],
        )
        .unwrap();
        assert!(conn.last_insert_rowid() > first);
    }

    #[test]
    fn complete_schema_requires_declared_foreign_keys_not_only_columns() {
        let conn = Connection::open_in_memory().unwrap();
        prepare_schema(&conn).unwrap();
        conn.execute_batch(
            "DROP TABLE faces;
             CREATE TABLE faces (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 hash TEXT NOT NULL,
                 bbox TEXT NOT NULL,
                 embedding BLOB NOT NULL
             );",
        )
        .unwrap();
        add_missing_columns(&conn, "faces", FACES_COLUMNS).unwrap();

        assert!(verify_schema(&conn).is_ok(), "the columns are intact");
        assert!(
            verify_foreign_keys(&conn).is_err(),
            "the person key is absent"
        );
        assert!(!schema_complete(&conn).unwrap());
    }

    #[test]
    fn complete_schema_rejects_cascading_person_and_location_keys() {
        let conn = Connection::open_in_memory().unwrap();
        prepare_schema(&conn).unwrap();
        conn.execute_batch("DROP TABLE faces").unwrap();
        let cascading_faces = crate::face_db::faces_ddl_for("faces", false).replace(
            "ON DELETE RESTRICT ON UPDATE RESTRICT",
            "ON DELETE CASCADE ON UPDATE CASCADE",
        );
        conn.execute_batch(&cascading_faces).unwrap();
        assert!(verify_schema(&conn).is_ok());
        assert!(
            !schema_complete(&conn).unwrap(),
            "person deletes must not cascade"
        );

        conn.execute_batch("DROP TABLE faces").unwrap();
        conn.execute_batch(&crate::face_db::faces_ddl_for("faces", false))
            .unwrap();
        conn.execute_batch("DROP TABLE file_hashes").unwrap();
        let cascading_files = file_hashes_ddl_for("file_hashes", false).replace(
            "ON DELETE RESTRICT ON UPDATE RESTRICT",
            "ON DELETE CASCADE ON UPDATE CASCADE",
        );
        conn.execute_batch(&cascading_files).unwrap();
        assert!(
            !schema_complete(&conn).unwrap(),
            "location deletes must not cascade"
        );
    }

    #[test]
    fn fresh_schema_rejects_orphan_references() {
        let conn = Connection::open_in_memory().unwrap();
        prepare_schema(&conn).unwrap();
        enable_foreign_keys(&conn).unwrap();

        let err = conn
            .execute(
                "INSERT INTO faces(hash,bbox,embedding,person_label) VALUES('h','0,0,1,1',X'00','missing')",
                [],
            )
            .unwrap_err();
        assert_eq!(
            err.sqlite_error_code(),
            Some(rusqlite::ErrorCode::ConstraintViolation),
            "an unknown person label must violate the faces foreign key"
        );

        let err = conn
            .execute(
                "INSERT INTO file_hashes(path,hash,location_cluster_id) VALUES('/x','h',999)",
                [],
            )
            .unwrap_err();
        assert_eq!(
            err.sqlite_error_code(),
            Some(rusqlite::ErrorCode::ConstraintViolation),
            "an unknown location cluster must violate the file_hashes foreign key"
        );

        // Valid parents make the children insertable, and the declared keys
        // are RESTRICT: deleting a referenced parent is rejected.
        conn.execute(
            "INSERT INTO people(name, full_name) VALUES ('elena', 'Elena')",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO location_clusters(id, centroid_lat, centroid_lon, name, photo_count, radius_km, created_at)
             VALUES (7, 52.52, 13.40, 'berlin', 1, 25.0, datetime('now'))",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO faces(hash,bbox,embedding,person_label) VALUES('h','0,0,1,1',X'00','elena')",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO file_hashes(path,hash,location_cluster_id) VALUES('/x','h',7)",
            [],
        )
        .unwrap();
        assert!(conn
            .execute("DELETE FROM people WHERE name = 'elena'", [])
            .is_err());
        assert!(conn
            .execute("DELETE FROM location_clusters WHERE id = 7", [])
            .is_err());

        // The declared keys name exactly the parents the design requires.
        for (child, from, parent, to) in [
            ("faces", "person_label", "people", "name"),
            (
                "file_hashes",
                "location_cluster_id",
                "location_clusters",
                "id",
            ),
        ] {
            let sql = format!("PRAGMA foreign_key_list({child})");
            let mut stmt = conn.prepare(&sql).unwrap();
            let declared = stmt
                .query_map([], |row| {
                    Ok((
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, String>(4)?,
                    ))
                })
                .unwrap()
                .collect::<rusqlite::Result<Vec<_>>>()
                .unwrap();
            assert!(
                declared.contains(&(from.into(), parent.into(), to.into())),
                "{child} must declare {from} -> {parent}.{to}, got {declared:?}"
            );
        }
    }

    use super::*;
    use crate::library::LibraryContext;
    use crate::library_locks;
    use rusqlite::params;

    #[test]
    fn migration_adds_xmp_sidecar_mtime_to_an_older_file_hashes() {
        let conn = Connection::open_in_memory().unwrap();
        // An older library: file_hashes without the new column.
        conn.execute_batch("CREATE TABLE file_hashes (path TEXT PRIMARY KEY, hash TEXT NOT NULL);")
            .unwrap();
        assert!(!column_exists(&conn, "file_hashes", "xmp_sidecar_mtime").unwrap());

        ensure_scan_schema(&conn).unwrap();

        assert!(
            column_exists(&conn, "file_hashes", "xmp_sidecar_mtime").unwrap(),
            "ensure_scan_schema must add the xmp_sidecar_mtime column to an older table"
        );
    }

    #[test]
    fn scan_schema_indexes_hashes_without_an_embed_run() {
        let conn = Connection::open_in_memory().unwrap();
        ensure_scan_schema(&conn).unwrap();
        let plan: String = conn
            .query_row(
                "EXPLAIN QUERY PLAN SELECT path FROM file_hashes WHERE hash IN ('a','b')",
                [],
                |row| row.get(3),
            )
            .unwrap();
        assert!(plan.contains("idx_file_hashes_hash"), "{plan}");
    }

    #[test]
    fn opening_an_existing_library_does_not_add_a_missing_hash_index() {
        let (_temp, ctx) = library();
        let conn = initialize(&ctx).unwrap();
        conn.execute_batch("DROP INDEX IF EXISTS idx_file_hashes_hash;")
            .unwrap();
        drop(conn);
        drop(open_existing_read_only(&ctx).unwrap());
        let conn = open_existing(&ctx).unwrap();
        let index_count: i64 = conn
            .query_row(
                "SELECT count(*) FROM sqlite_master WHERE type='index' AND name='idx_file_hashes_hash'",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(index_count, 0, "opening a library must not upgrade it");
    }

    /// One library root plus a context on it. All path expectations are built
    /// from the context: construction canonicalizes the root, and on macOS a
    /// tempdir resolves under /private, so the spelling the test created is
    /// not where the state directory lives.
    fn library() -> (tempfile::TempDir, LibraryContext) {
        let temp = tempfile::tempdir().unwrap();
        let root = temp.path().join("photos");
        std::fs::create_dir(&root).unwrap();
        let ctx = LibraryContext::new(&root, &temp.path().join("cache")).unwrap();
        (temp, ctx)
    }

    /// The regression this module exists for: a reader never brings a
    /// library into being, and initializing twice never disturbs the config
    /// the first initialization wrote.
    #[test]
    fn readers_never_initialize_and_repeated_initialization_preserves_config() {
        let temp = tempfile::tempdir().unwrap();
        let root = temp.path().join("photos");
        std::fs::create_dir(&root).unwrap();
        let ctx = crate::library::LibraryContext::new(&root, &temp.path().join("cache")).unwrap();
        assert!(open_existing(&ctx).is_err());
        assert!(!ctx.paths.state.exists());
        drop(initialize(&ctx).unwrap());
        let before = std::fs::read(&ctx.paths.config).unwrap();
        drop(initialize(&ctx).unwrap());
        assert_eq!(std::fs::read(&ctx.paths.config).unwrap(), before);
        let conn = open_existing(&ctx).unwrap();
        let count: i64 = conn
            .query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
            .unwrap();
        assert_eq!(count, 0);
    }

    #[test]
    fn open_existing_read_only_refuses_an_older_schema_without_changing_bytes() {
        let (_temp, ctx) = library();
        let conn = initialize(&ctx).unwrap();
        conn.pragma_update(None, "user_version", 0).unwrap();
        conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")
            .unwrap();
        drop(conn);
        let before = std::fs::read(&ctx.paths.db).unwrap();

        let error = open_existing_read_only(&ctx).unwrap_err();

        assert!(format!("{error:#}").contains("created by an older videre"));
        assert_eq!(
            crate::error_kind::ErrorKind::in_chain(&error),
            Some(crate::error_kind::ErrorKind::LibrarySchema)
        );
        assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), before);
        let conn = open_without_create(&ctx.paths.db).unwrap();
        assert_eq!(user_version(&conn).unwrap(), 0);
    }

    #[test]
    fn a_library_from_a_newer_videre_is_refused_with_the_schema_kind() {
        let (_temp, ctx) = library();
        let conn = initialize(&ctx).unwrap();
        conn.pragma_update(None, "user_version", SCHEMA_VERSION + 1)
            .unwrap();
        drop(conn);
        for error in [
            open_existing(&ctx).unwrap_err(),
            open_existing_read_only(&ctx).unwrap_err(),
        ] {
            assert!(format!("{error:#}").contains("newer videre"), "{error:#}");
            assert_eq!(
                crate::error_kind::ErrorKind::in_chain(&error),
                Some(crate::error_kind::ErrorKind::LibrarySchema)
            );
        }
    }

    #[test]
    fn open_existing_read_only_opens_current_schema_and_rejects_writes() {
        let (_temp, ctx) = library();
        drop(initialize(&ctx).unwrap());

        let conn = open_existing_read_only(&ctx).unwrap();
        let error = conn
            .execute("INSERT INTO file_hashes (path, hash) VALUES ('x', 'h')", [])
            .unwrap_err();

        assert_eq!(
            error.sqlite_error_code(),
            Some(rusqlite::ErrorCode::ReadOnly)
        );
    }

    #[test]
    fn fresh_database_publication_never_replaces_an_existing_destination() {
        let temp = tempfile::tempdir().unwrap();
        let built = temp.path().join("built.db");
        let destination = temp.path().join("hashes.db");
        std::fs::write(&built, b"completed build").unwrap();
        std::fs::write(&destination, b"newer initializer").unwrap();

        let err = publish_database(&built, &destination).unwrap_err();

        assert!(format!("{err:#}").contains("publish"), "{err:#}");
        assert_eq!(std::fs::read(&destination).unwrap(), b"newer initializer");
        assert!(built.exists(), "the refused build remains recoverable");
    }

    #[test]
    fn corrupt_database_bytes_are_refused_unchanged() {
        let (_t, ctx) = library();
        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
        let garbage = b"definitely not a sqlite database at all";
        std::fs::write(&ctx.paths.db, garbage).unwrap();
        assert!(open_existing(&ctx).is_err());
        let err = initialize(&ctx).unwrap_err();
        assert!(
            format!("{err:#}").contains("not a SQLite database"),
            "{err:#}"
        );
        // Refusal leaves the bytes as they were; whatever the file is, it is
        // not videre's to repair.
        assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), garbage);
    }

    #[test]
    fn a_zero_byte_database_is_never_exposed_as_a_library() {
        let (_t, ctx) = library();
        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
        std::fs::write(&ctx.paths.db, b"").unwrap();
        let err = open_existing(&ctx).unwrap_err();
        assert!(format!("{err:#}").contains("empty"), "{err:#}");
        // initialize may replace it: a zero-byte file is the debris of an
        // interrupted creation and provably holds nothing.
        let conn = initialize(&ctx).unwrap();
        let count: i64 = conn
            .query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
            .unwrap();
        assert_eq!(count, 0);
        let version: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert_eq!(version, SCHEMA_VERSION);
    }

    #[test]
    fn a_foreign_sqlite_database_is_refused_unchanged() {
        let (_t, ctx) = library();
        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
        {
            let conn = rusqlite::Connection::open(&ctx.paths.db).unwrap();
            conn.execute_batch(
                "CREATE TABLE somebody_elses (id INTEGER PRIMARY KEY);
                 INSERT INTO somebody_elses VALUES (1);",
            )
            .unwrap();
        }
        let before = std::fs::read(&ctx.paths.db).unwrap();
        let err = open_existing(&ctx).unwrap_err();
        assert!(
            format!("{err:#}").contains("not a videre library"),
            "{err:#}"
        );
        let err = initialize(&ctx).unwrap_err();
        assert!(
            format!("{err:#}").contains("not a videre library"),
            "{err:#}"
        );
        assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), before);
    }

    #[test]
    fn a_config_without_a_database_is_not_a_library_and_initialize_keeps_the_config() {
        let (_t, ctx) = library();
        std::fs::create_dir_all(&ctx.paths.state).unwrap();
        std::fs::write(&ctx.paths.config, "custom = \"keep\"\n").unwrap();
        assert!(open_existing(&ctx).is_err());
        drop(initialize(&ctx).unwrap());
        assert_eq!(
            std::fs::read_to_string(&ctx.paths.config).unwrap(),
            "custom = \"keep\"\n"
        );
    }

    #[test]
    fn a_database_without_a_config_reads_and_initialize_writes_six_declarations_once() {
        let (_t, ctx) = library();
        drop(initialize(&ctx).unwrap());
        std::fs::remove_file(&ctx.paths.config).unwrap();
        // A reader serves the library with its built-in settings and does not
        // conjure the config back.
        drop(open_existing(&ctx).unwrap());
        assert!(!ctx.paths.config.exists());
        drop(initialize(&ctx).unwrap());
        let text = std::fs::read_to_string(&ctx.paths.config).unwrap();
        let table: toml::Table = toml::from_str(&text).unwrap();
        // Exactly the six fixed config declarations.
        assert_eq!(table.len(), 6, "{text}");
        assert_eq!(table["db"].as_str(), Some("hashes.db"));
        assert_eq!(table["jsonl"].as_str(), Some("hashes.jsonl"));
        assert_eq!(
            table["default_model"].as_str(),
            Some(crate::embeddings::DEFAULT_MODEL_ID)
        );
        assert_eq!(table["xmp_precedence"].as_str(), Some("db"));
        assert_eq!(table["export_xmp_on_watch"].as_bool(), Some(false));
        // And it is never rewritten on a repeated initialize.
        let before = std::fs::read(&ctx.paths.config).unwrap();
        drop(initialize(&ctx).unwrap());
        assert_eq!(std::fs::read(&ctx.paths.config).unwrap(), before);
    }

    #[test]
    fn two_simultaneous_initializers_produce_one_valid_library() {
        let temp = tempfile::tempdir().unwrap();
        let root = temp.path().join("photos");
        std::fs::create_dir(&root).unwrap();
        let ctx = LibraryContext::new(&root, &temp.path().join("cache")).unwrap();
        let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
        let handles: Vec<_> = (0..2)
            .map(|_| {
                let ctx = ctx.clone();
                let barrier = barrier.clone();
                std::thread::spawn(move || {
                    barrier.wait();
                    initialize(&ctx).map(|conn| {
                        conn.query_row::<i64, _, _>("SELECT count(*) FROM file_hashes", [], |r| {
                            r.get(0)
                        })
                        .unwrap()
                    })
                })
            })
            .collect();
        let results: Vec<_> = handles.into_iter().map(|h| h.join().unwrap()).collect();
        let winners = results.iter().filter(|r| r.is_ok()).count();
        assert_eq!(winners, 1, "exactly one initializer must win: {results:?}");
        for result in &results {
            match result {
                Ok(count) => assert_eq!(*count, 0),
                // The loser fails on the lock coordination, the same wording
                // a held activity lock produces, not on a half-built library
                // and not silently. A panic would have failed the join.
                Err(e) => {
                    let msg = format!("{e:#}");
                    assert!(
                        msg.contains("in use"),
                        "the loser must be refused by the lock, not by anything else: {msg}"
                    );
                }
            }
        }
        // The config is never truncated: exactly the six declarations.
        let table: toml::Table =
            toml::from_str(&std::fs::read_to_string(&ctx.paths.config).unwrap()).unwrap();
        assert_eq!(table.len(), 6);
        drop(open_existing(&ctx).unwrap());
    }

    #[test]
    fn a_held_activity_lock_blocks_initialization_cleanly() {
        let (_t, ctx) = library();
        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
        let _held =
            library_locks::try_activity(&ctx, library_locks::ActivityMode::Exclusive).unwrap();
        let err = initialize(&ctx).unwrap_err();
        assert!(format!("{err:#}").contains("in use"), "{err:#}");
        drop(_held);
        drop(initialize(&ctx).unwrap());
    }

    #[test]
    fn a_held_exclusive_activity_lock_blocks_readers_too() {
        let (_t, ctx) = library();
        drop(initialize(&ctx).unwrap());
        let _held =
            library_locks::try_activity(&ctx, library_locks::ActivityMode::Exclusive).unwrap();
        assert!(open_existing(&ctx).is_err());
    }

    #[test]
    fn redirected_state_database_and_config_are_refused() {
        let (temp, ctx) = library();
        // An externally linked .videre must not be used: nothing is created
        // through it.
        let elsewhere = temp.path().join("elsewhere");
        std::fs::create_dir(&elsewhere).unwrap();
        std::os::unix::fs::symlink(&elsewhere, &ctx.paths.state).unwrap();
        let err = initialize(&ctx).unwrap_err();
        assert!(format!("{err:#}").contains("symlink"), "{err:#}");
        assert!(!elsewhere.join("locks").exists());
        std::fs::remove_file(&ctx.paths.state).unwrap();

        // A redirected database must not be used either.
        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
        let outside_db = temp.path().join("outside.db");
        std::fs::write(&outside_db, b"").unwrap();
        std::os::unix::fs::symlink(&outside_db, &ctx.paths.db).unwrap();
        let err = initialize(&ctx).unwrap_err();
        assert!(format!("{err:#}").contains("symlink"), "{err:#}");
        assert_eq!(std::fs::read(&outside_db).unwrap(), b"");
        let err = open_existing(&ctx).unwrap_err();
        assert!(format!("{err:#}").contains("symlink"), "{err:#}");

        // Nor a redirected config. It is valid TOML, but context construction
        // must refuse it before any settings are loaded through the link.
        std::fs::remove_file(&ctx.paths.db).unwrap();
        let outside_cfg = temp.path().join("outside.toml");
        std::fs::write(&outside_cfg, b"custom = \"x\"\n").unwrap();
        std::os::unix::fs::symlink(&outside_cfg, &ctx.paths.config).unwrap();
        let err = LibraryContext::new(&ctx.paths.root, &temp.path().join("cache")).unwrap_err();
        assert!(format!("{err:#}").contains("symlink"), "{err:#}");
        assert_eq!(std::fs::read(&outside_cfg).unwrap(), b"custom = \"x\"\n");
    }

    #[test]
    fn multiply_linked_state_files_are_refused() {
        let (temp, ctx) = library();
        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
        // A database hard-linked into two names is two libraries silently
        // coupled into one; neither may use it.
        let twin_db = temp.path().join("twin.db");
        {
            let conn = rusqlite::Connection::open(&twin_db).unwrap();
            conn.execute_batch(
                "CREATE TABLE file_hashes (path TEXT PRIMARY KEY, hash TEXT NOT NULL);",
            )
            .unwrap();
        }
        std::fs::hard_link(&twin_db, &ctx.paths.db).unwrap();
        let err = initialize(&ctx).unwrap_err();
        assert!(format!("{err:#}").contains("hard-linked"), "{err:#}");
        let err = open_existing(&ctx).unwrap_err();
        assert!(format!("{err:#}").contains("hard-linked"), "{err:#}");
        assert!(twin_db.exists());

        // The same holds for the config.
        std::fs::remove_file(&ctx.paths.db).unwrap();
        let twin_cfg = temp.path().join("twin.toml");
        std::fs::write(&twin_cfg, "custom = \"x\"\n").unwrap();
        std::fs::hard_link(&twin_cfg, &ctx.paths.config).unwrap();
        let err = LibraryContext::new(&ctx.paths.root, &temp.path().join("cache")).unwrap_err();
        assert!(format!("{err:#}").contains("hard-linked"), "{err:#}");
    }

    #[test]
    fn a_symlinked_embeddings_directory_is_refused() {
        let (temp, ctx) = library();
        std::fs::create_dir_all(&ctx.paths.locks).unwrap();
        // The embeddings directory is a symlink to a directory outside the
        // state directory: opening through it would write this library's
        // embeddings into whatever the link points at.
        let outside = temp.path().join("outside-embeddings");
        std::fs::create_dir(&outside).unwrap();
        std::os::unix::fs::symlink(&outside, &ctx.paths.embeddings).unwrap();
        let err = initialize(&ctx).unwrap_err();
        assert!(format!("{err:#}").contains("symlink"), "{err:#}");
        // The refusal came before any build, and the outside directory was
        // not populated through the link.
        assert!(!ctx.paths.db.exists());
        assert!(std::fs::read_dir(&outside).unwrap().next().is_none());
        // A reader refuses the same redirect; the state the failed
        // initialize left behind (directories, lock files) is all it sees.
        let err = open_existing(&ctx).unwrap_err();
        assert!(format!("{err:#}").contains("symlink"), "{err:#}");
        // An ordinary in-place embeddings directory passes: it is state the
        // embeddings layer itself creates, and a previously initialized
        // library commonly already has one.
        std::fs::remove_file(&ctx.paths.embeddings).unwrap();
        std::fs::create_dir(&ctx.paths.embeddings).unwrap();
        drop(initialize(&ctx).unwrap());
        drop(open_existing(&ctx).unwrap());
    }

    #[test]
    fn a_foreign_row_refuses_the_whole_library_without_touching_it() {
        let (_t, ctx) = library();
        {
            let conn = initialize(&ctx).unwrap();
            // A string prefix of the root is not containment by components:
            // <root>-sibling starts with the same bytes but is another
            // directory.
            let adjacent = format!("{}-sibling/x.jpg", ctx.paths.root.to_str().unwrap());
            conn.execute(
                "INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
                params![adjacent],
            )
            .unwrap();
        }
        let before = std::fs::read(&ctx.paths.db).unwrap();
        let before_cfg = std::fs::read(&ctx.paths.config).unwrap();
        // A fresh context has no memoized validation, so the row is checked.
        let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
        let err = open_existing(&fresh).unwrap_err();
        let msg = format!("{err:#}");
        assert!(msg.contains("outside the library root"), "{msg}");
        assert!(msg.contains("-sibling"), "{msg}");
        assert!(initialize(&fresh).is_err());
        assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), before);
        assert_eq!(std::fs::read(&ctx.paths.config).unwrap(), before_cfg);
    }

    #[test]
    fn a_row_with_a_dot_component_is_refused_as_foreign() {
        let (_t, ctx) = library();
        {
            let conn = initialize(&ctx).unwrap();
            conn.execute(
                "INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
                params![ctx.paths.root.join("plain.jpg").to_str().unwrap()],
            )
            .unwrap();
            // `<root>/../outside/x.jpg` starts with `<root>` lexically, so
            // only the `.`/`..` component check can refuse it.
            let escaped = format!("{}/../outside/x.jpg", ctx.paths.root.to_str().unwrap());
            conn.execute(
                "INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
                params![escaped],
            )
            .unwrap();
        }
        let before = std::fs::read(&ctx.paths.db).unwrap();
        let before_cfg = std::fs::read(&ctx.paths.config).unwrap();
        let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
        let err = open_existing(&fresh).unwrap_err();
        let msg = format!("{err:#}");
        assert!(msg.contains("outside the library root"), "{msg}");
        assert!(msg.contains(".."), "{msg}");
        assert!(initialize(&fresh).is_err());
        assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), before);
        assert_eq!(std::fs::read(&ctx.paths.config).unwrap(), before_cfg);
    }

    #[test]
    fn known_missing_media_under_the_root_remains_a_valid_library() {
        let (_t, ctx) = library();
        {
            let conn = initialize(&ctx).unwrap();
            conn.execute(
                "INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
                params![ctx.paths.root.join("long-gone.jpg").to_str().unwrap()],
            )
            .unwrap();
        }
        // The file does not exist; the row is still under the root, so the
        // library stays usable (containment, not existence).
        let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
        let conn = open_existing(&fresh).unwrap();
        let count: i64 = conn
            .query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
            .unwrap();
        assert_eq!(count, 1);
    }

    #[test]
    fn root_aliases_open_the_same_initialized_library() {
        let (temp, ctx) = library();
        drop(initialize(&ctx).unwrap());
        let alias = temp.path().join("alias");
        std::os::unix::fs::symlink(&ctx.paths.root, &alias).unwrap();
        let via_alias = LibraryContext::new(&alias, &temp.path().join("cache")).unwrap();
        assert_eq!(via_alias.paths.db, ctx.paths.db);
        let conn = open_existing(&via_alias).unwrap();
        let count: i64 = conn
            .query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
            .unwrap();
        assert_eq!(count, 0);
    }

    fn assert_keys_on(conn: &Connection) {
        let enabled: i64 = conn
            .query_row("PRAGMA foreign_keys", [], |r| r.get(0))
            .unwrap();
        assert_eq!(enabled, 1);
    }

    #[test]
    fn every_library_open_enforces_foreign_keys() {
        let (_t, ctx) = library();

        // Fresh initialize.
        let conn = initialize(&ctx).unwrap();
        assert_keys_on(&conn);
        drop(conn);

        // Current open_existing (fast path).
        let conn = open_existing(&ctx).unwrap();
        assert_keys_on(&conn);
        drop(conn);

        // Read-only open.
        let conn = open_existing_read_only(&ctx).unwrap();
        assert_keys_on(&conn);
        drop(conn);

        // The no-create open used by the signal-handler path.
        let conn = open_without_create(&ctx.paths.db).unwrap();
        assert_keys_on(&conn);
    }

    #[test]
    fn a_future_schema_version_is_refused_with_an_actionable_error() {
        let (_t, ctx) = library();
        drop(initialize(&ctx).unwrap());
        {
            let conn = rusqlite::Connection::open(&ctx.paths.db).unwrap();
            conn.pragma_update(None, "user_version", SCHEMA_VERSION + 1)
                .unwrap();
        }
        let err = open_existing(&ctx).unwrap_err();
        let msg = format!("{err:#}");
        assert!(msg.contains("newer"), "{msg}");
        assert!(
            msg.contains(&format!("version {}", SCHEMA_VERSION + 1)),
            "{msg}"
        );
        assert!(initialize(&ctx).is_err());
    }

    #[test]
    fn the_published_database_is_self_contained_and_leaves_no_temporaries() {
        let (_t, ctx) = library();
        drop(initialize(&ctx).unwrap());
        let conn = open_existing(&ctx).unwrap();
        // WAL everywhere is the existing behaviour; it persisted in the file
        // when the fresh database was built.
        let mode: String = conn
            .query_row("PRAGMA journal_mode", [], |r| r.get(0))
            .unwrap();
        assert_eq!(mode.to_lowercase(), "wal");
        drop(conn);
        // A clean close leaves no sidecars and no build temporaries behind.
        let entries: Vec<String> = std::fs::read_dir(&ctx.paths.state)
            .unwrap()
            .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
            .collect();
        for entry in &entries {
            assert!(
                !entry.contains(".tmp") && !entry.ends_with("-wal") && !entry.ends_with("-shm"),
                "leftover {entry}: {entries:?}"
            );
        }
    }

    #[test]
    fn stale_build_wal_sidecars_are_swept_with_their_temporary() {
        let (_t, ctx) = library();
        std::fs::create_dir_all(&ctx.paths.state).unwrap();
        // The debris of a build that crashed mid-write: its temporary
        // database plus the WAL sidecars SQLite had opened beside it.
        let stem = "hashes.db.4242.0.tmp";
        for suffix in ["", "-wal", "-shm"] {
            std::fs::write(ctx.paths.state.join(format!("{stem}{suffix}")), b"debris").unwrap();
        }
        // An unrelated file, and the live database's own name, must not be
        // swept: the rule matches temp build stems, not any suffix.
        std::fs::write(ctx.paths.state.join("notes.txt"), b"keep").unwrap();
        drop(initialize(&ctx).unwrap());
        for suffix in ["", "-wal", "-shm"] {
            assert!(
                !ctx.paths.state.join(format!("{stem}{suffix}")).exists(),
                "sweeping must remove {stem}{suffix}"
            );
        }
        assert!(ctx.paths.state.join("notes.txt").exists());
        assert!(ctx.paths.db.exists());
    }

    #[test]
    fn a_refused_foreign_row_validation_is_never_memoized_as_success() {
        let (_t, ctx) = library();
        // A reader context separate from the initializing one, so its
        // validation memo starts clear: initialize marks its own context's
        // memo, and the refusal below must be observed by a context that has
        // not yet validated anything.
        let reader = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
        {
            let conn = initialize(&ctx).unwrap();
            conn.execute(
                "INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
                params!["/elsewhere-not-this-library/x.jpg"],
            )
            .unwrap();
        }
        let err = open_existing(&reader).unwrap_err();
        assert!(
            format!("{err:#}").contains("outside the library root"),
            "{err:#}"
        );
        // The refusal left no memo. Had the failure been recorded the way a
        // success is, the next open on this same context would skip
        // validation and quietly serve the mixed library.
        assert!(
            !reader.index_validated(),
            "a refused validation must not be memoized"
        );
        // The operator repair: the foreign row removed directly via SQL, no
        // videre involvement, exactly the fix a copied database needs.
        {
            let conn = rusqlite::Connection::open(&ctx.paths.db).unwrap();
            let removed = conn
                .execute(
                    "DELETE FROM file_hashes WHERE path = '/elsewhere-not-this-library/x.jpg'",
                    [],
                )
                .unwrap();
            assert_eq!(removed, 1);
        }
        // The same context, which has now both refused and could have
        // remembered the refusal, revalidates and succeeds; and so does a
        // context built after the repair.
        drop(open_existing(&reader).unwrap());
        assert!(reader.index_validated());
        let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
        drop(open_existing(&fresh).unwrap());
    }

    #[test]
    fn opening_a_multi_row_library_validates_every_row_without_filesystem_probes() {
        let (_t, ctx) = library();
        // A multi-row library, enough to prove the validation generalises over
        // many rows rather than a single one. Deliberately small: the invariant
        // under test is scale-independent (see the two counters below), so N
        // does not need to mirror a real library. This was once 70,000 rows -
        // not for the invariant, but purely to give a wall-clock quadratic-walk
        // canary teeth. That canary was doubly bad: a timing bound on shared CI
        // is inherently flaky, and its large single-transaction write
        // intermittently tripped SQLITE_IOERR_WRITE on GitHub's macOS runner
        // (whose virtual disk fails writes under load). Both the clock and the
        // heavy write are gone; the deterministic counters prove the same thing
        // at any N, and the *expensive* form of a quadratic regression - per-row
        // filesystem work - is caught outright by the zero-probe assertion.
        const ROWS: u32 = 10_000;
        {
            let mut conn = initialize(&ctx).unwrap();
            // Bare path strings under the root, batched in one transaction. No
            // media files exist, which is the point; containment judges which
            // library a row belongs to, not whether its media is mounted.
            let tx = conn.transaction().unwrap();
            {
                let mut stmt = tx
                    .prepare("INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')")
                    .unwrap();
                for i in 0..ROWS {
                    let path =
                        ctx.paths
                            .root
                            .join(format!("Trips/album-{:02}/img-{:05}.jpg", i / 100, i));
                    stmt.execute(params![path.to_str().unwrap()]).unwrap();
                }
            }
            tx.commit().unwrap();
        }
        // A fresh context has no memoized validation, so this open validates
        // every row. Two deterministic instruments answer the two halves of the
        // claim, at any row count: the guard layer's resolution counter is the
        // filesystem-probe instrument and must not move at all (no per-row
        // stat/canonicalise - the costly quadratic form), and the containment
        // row counter must have judged exactly ROWS rows (each visited once, a
        // linear pass), so a flat probe count cannot be explained by a
        // validation that never ran.
        let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
        let probes_before = crate::library_guard::count_resolutions();
        let rows_before = count_rows_validated();
        let conn = open_existing(&fresh).unwrap();
        assert!(
            fresh.index_validated(),
            "a successful open records the memo"
        );
        assert_eq!(
            crate::library_guard::count_resolutions() - probes_before,
            0,
            "containment validation must not touch the filesystem per row"
        );
        assert_eq!(
            count_rows_validated() - rows_before,
            u64::from(ROWS),
            "every row must have been judged exactly once (a linear pass)"
        );
        let count: i64 = conn
            .query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
            .unwrap();
        assert_eq!(count, i64::from(ROWS));
    }
}