//! Photo versions (#191): hash-keyed links from an AI restoration / upscale
//! / edit to its primary original, the global display choice, and the SQL
//! the listings use to hide a version while its original is reachable.
//!
//! The link is the blake3 `content_hash` of the original (CLAUDE.md →
//! "Multi-library data model"). A version keeps its own `image_exif` row,
//! so `/image?path=<version>` and thumbnails keep working; only listings
//! hide it. `photo_versions` is tiny next to `image_exif`, so reads that
//! need "which files are versions" load that set instead of joining per row.

use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};

use diesel::prelude::*;
use diesel::sql_query;
use diesel::sql_types::{BigInt, Integer, Nullable, Text};
use diesel::sqlite::SqliteConnection;

use crate::database::{DbError, DbErrorKind, connect};
use crate::otel::trace_db_call;

/// The `kind` values a version may carry.
pub const KINDS: &[&str] = &["restored", "upscale", "edit"];

/// `date_taken_source` on a version row whose date, GPS and camera fields
/// were copied from its original.
pub const DATE_SOURCE_VERSION_OF: &str = "version_of";

/// `photo_versions.source`: how the link was learned.
pub const SOURCE_XMP: &str = "xmp";
pub const SOURCE_NAME: &str = "name";
pub const SOURCE_BACKFILL: &str = "backfill";

pub fn is_known_kind(kind: &str) -> bool {
    KINDS.contains(&kind)
}

/// SQL predicate over an unaliased `image_exif` row: true unless the row is
/// a version that is hidden in the view (`view_library_ids` empty = every
/// library).
///
/// Same reachability rule as `duplicate_of_hash` in
/// `list_rel_paths_for_libraries`: a version hides only while a row for its
/// original exists in the view, so a version whose original was deleted, or
/// lives in a library outside the view, stays visible as a normal photo.
/// (Unlike duplicates the unscoped view still needs the existence check:
/// the original may be gone.) Library ids are `i32`, so inlining the list
/// is safe.
pub fn visible_version_clause(view_library_ids: &[i32]) -> String {
    format!(
        "NOT EXISTS (SELECT 1 FROM photo_versions pv \
         JOIN image_exif orig ON orig.content_hash = pv.original_hash \
         WHERE pv.version_hash = image_exif.content_hash AND {})",
        reachable_original_condition(view_library_ids)
    )
}

/// Condition over an `image_exif` row aliased `orig`: the row is an original
/// that is itself shown in the view. It must be in the view, and not demoted
/// by duplicates resolve, using the duplicate-listing rule: unscoped, any
/// `duplicate_of_hash` hides it; scoped, only when the survivor is in the
/// view too. Otherwise its versions would vanish along with it.
fn reachable_original_condition(view_library_ids: &[i32]) -> String {
    if view_library_ids.is_empty() {
        return "orig.duplicate_of_hash IS NULL".to_string();
    }
    let list = view_library_ids
        .iter()
        .map(i32::to_string)
        .collect::<Vec<_>>()
        .join(",");
    format!(
        "orig.library_id IN ({list}) AND (orig.duplicate_of_hash IS NULL OR NOT EXISTS \
         (SELECT 1 FROM image_exif AS survivor \
          WHERE survivor.content_hash = orig.duplicate_of_hash \
            AND survivor.library_id IN ({list})))"
    )
}

/// SQL predicate over an unaliased `image_exif` row: true unless the row's
/// hash is a version at all. For derived work (faces, CLIP, duplicates)
/// that never runs on a version, whatever the view.
pub const NOT_A_VERSION_CLAUSE: &str = "NOT EXISTS (SELECT 1 FROM photo_versions pv \
     WHERE pv.version_hash = image_exif.content_hash)";

/// Same as [`NOT_A_VERSION_CLAUSE`] for a raw-SQL query that aliases
/// `image_exif` as `alias`.
pub fn not_a_version_clause_for(alias: &str) -> String {
    format!(
        "NOT EXISTS (SELECT 1 FROM photo_versions pv WHERE pv.version_hash = {alias}.content_hash)"
    )
}

#[derive(Debug, Clone, PartialEq, Eq, QueryableByName, serde::Serialize)]
pub struct PhotoVersionRow {
    #[diesel(sql_type = Text)]
    pub version_hash: String,
    #[diesel(sql_type = Text)]
    pub original_hash: String,
    #[diesel(sql_type = Text)]
    pub kind: String,
    #[diesel(sql_type = Text)]
    pub source: String,
    #[diesel(sql_type = BigInt)]
    pub created_at: i64,
    #[diesel(sql_type = Integer)]
    pub library_id: i32,
    #[diesel(sql_type = Text)]
    pub rel_path: String,
}

/// A link to record.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NewVersionLink {
    pub version_hash: String,
    pub original_hash: String,
    pub kind: String,
    pub source: String,
    /// Back-ref to a live `image_exif` row of the version.
    pub library_id: i32,
    pub rel_path: String,
    /// Hashes of reference photos (provenance only).
    pub references: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LinkOutcome {
    /// A new link was recorded.
    Linked,
    /// The version already had a link; it was left alone.
    AlreadyLinked,
    /// The link makes no sense (self-link, unknown kind, cycle).
    Rejected(&'static str),
}

/// A live version file of an original, for `GET /image/versions`.
#[derive(Debug, Clone, QueryableByName)]
pub struct VersionFile {
    #[diesel(sql_type = Text)]
    pub version_hash: String,
    #[diesel(sql_type = Text)]
    pub kind: String,
    #[diesel(sql_type = Text)]
    pub source: String,
    #[diesel(sql_type = BigInt)]
    pub created_at: i64,
    #[diesel(sql_type = Integer)]
    pub library_id: i32,
    #[diesel(sql_type = Text)]
    pub rel_path: String,
    #[diesel(sql_type = Nullable<Integer>)]
    pub width: Option<i32>,
    #[diesel(sql_type = Nullable<Integer>)]
    pub height: Option<i32>,
}

/// One file (some `image_exif` row) for a hash.
#[derive(Debug, Clone, PartialEq, Eq, QueryableByName)]
pub struct HashedFile {
    #[diesel(sql_type = Text)]
    pub content_hash: String,
    #[diesel(sql_type = Integer)]
    pub library_id: i32,
    #[diesel(sql_type = Text)]
    pub rel_path: String,
}

/// What `/photos` adds next to an original: how many live versions it has
/// and, if the global choice isn't the original, the file to display.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PhotoVersionSummary {
    pub version_count: u32,
    /// `(library_id, rel_path)` of the chosen version file; the library is
    /// always the original's own.
    pub display: Option<(i32, String)>,
}

pub trait VersionDao: Sync + Send {
    /// Record a link (and its reference photos). Flattens chains: an
    /// `original_hash` that is itself a version is replaced by its root, and
    /// versions already pointing at the newly linked file move up to the root.
    fn link_version(&mut self, link: &NewVersionLink) -> Result<LinkOutcome, DbError>;

    /// The link row for a version hash, if any.
    fn get_link(&mut self, version_hash: &str) -> Result<Option<PhotoVersionRow>, DbError>;

    /// If `rel_path` is a linked version whose original is still indexed,
    /// the original's rel_path (same library when possible).
    fn original_path_of(&mut self, rel_path: &str) -> Result<Option<String>, DbError>;

    /// `(library_id, rel_path)` of the original of the version at
    /// `(library_id, rel_path)`, when it is a linked version with a live
    /// original. For per-library writes that must land on the original.
    fn original_location_of(
        &mut self,
        library_id: i32,
        rel_path: &str,
    ) -> Result<Option<(i32, String)>, DbError>;

    /// Which of `hashes` are versions of something.
    fn version_hashes_among(&mut self, hashes: &[String]) -> Result<HashSet<String>, DbError>;

    /// Live versions of an original (those with an `image_exif` row),
    /// oldest first.
    fn list_versions(&mut self, original_hash: &str) -> Result<Vec<VersionFile>, DbError>;

    /// Reference photo hashes recorded for a version.
    fn list_references(&mut self, version_hash: &str) -> Result<Vec<String>, DbError>;

    /// First live `image_exif` row for each hash, preferring `prefer_library`.
    fn files_for_hashes(
        &mut self,
        hashes: &[String],
        prefer_library: Option<i32>,
    ) -> Result<HashMap<String, (i32, String)>, DbError>;

    /// The stored display choice for an original, if any.
    fn get_display(&mut self, original_hash: &str) -> Result<Option<String>, DbError>;

    /// Set (`Some`) or clear (`None`) the global display choice.
    fn set_display(
        &mut self,
        original_hash: &str,
        display_hash: Option<&str>,
    ) -> Result<(), DbError>;

    /// Per original `(library_id, rel_path)` that has at least one live
    /// version: the version count and display file. Empty when no photo has
    /// versions, so `/photos` stays byte-identical.
    fn photo_summaries(&mut self) -> Result<HashMap<(i32, String), PhotoVersionSummary>, DbError>;

    /// `(library_id, rel_path)` of every version row hidden in the view
    /// (`view_library_ids` empty = all libraries). For listings that don't
    /// go through SQL (filesystem walk, tag search, media gauges).
    fn hidden_version_rows(
        &mut self,
        view_library_ids: &[i32],
    ) -> Result<HashSet<(i32, String)>, DbError>;

    /// Of `dirs` (rel paths), those that hold indexed files and where every
    /// indexed file anywhere underneath is a version hidden in the view.
    /// Such a folder would open empty. A folder with no indexed rows at all
    /// is not reported (nothing says it is safe to hide). One indexed range
    /// scan per dir; no filesystem walk.
    fn dirs_holding_only_hidden_versions(
        &mut self,
        view_library_ids: &[i32],
        dirs: &[String],
    ) -> Result<HashSet<String>, DbError>;

    /// Rows in a library whose file name follows the version convention
    /// (`<name>.<kind>[.<n>].<ext>`), have a hash and aren't linked yet.
    /// Returns `(id, rel_path, content_hash)` ordered by row id, starting
    /// after `after_id` (the caller's resume cursor, so rows that never
    /// resolve can't starve the ones behind them).
    fn unlinked_name_candidates(
        &mut self,
        library_id: i32,
        after_id: i32,
        limit: i64,
    ) -> Result<Vec<(i32, String, String)>, DbError>;

    /// Copy the original's date, GPS and camera fields onto every version
    /// row whose copy is stale. Returns the number of rows updated.
    /// Idempotent; only converted originals (`date_utc` set) are copied, so
    /// the legacy-convention date drains never fight the copy.
    fn sync_version_metadata(&mut self) -> Result<usize, DbError>;
}

pub struct SqliteVersionDao {
    connection: Arc<Mutex<SqliteConnection>>,
}

impl Default for SqliteVersionDao {
    fn default() -> Self {
        Self::new()
    }
}

impl SqliteVersionDao {
    pub fn new() -> Self {
        SqliteVersionDao {
            connection: Arc::new(Mutex::new(connect())),
        }
    }

    /// Test-only constructor sharing an already-wrapped connection, so
    /// other DAOs see the rows this one writes.
    #[cfg(test)]
    pub fn from_shared(connection: Arc<Mutex<SqliteConnection>>) -> Self {
        SqliteVersionDao { connection }
    }
}

/// Chunk size for `IN (...)` lists, safely under SQLite's variable limit.
const CHUNK: usize = 500;

fn placeholders(n: usize) -> String {
    vec!["?"; n].join(",")
}

/// If `rel_path` is a linked version whose original has a live row, the
/// original's rel_path (same library when possible). Tag and favorite
/// writes on a version path bind to the original through this.
///
/// `library_id` scopes the version lookup to one library's row; `None` is
/// for the library-agnostic tag/favorite DAOs, which only know a rel_path.
/// Annotations are keyed by rel_path alone, so in either case a path that
/// *also* names a file in some library that is not a linked version stays
/// itself: redirecting it would hijack that photo's own tags and favorites.
pub fn canonical_rel_path(
    conn: &mut SqliteConnection,
    library_id: Option<i32>,
    rel_path: &str,
) -> QueryResult<Option<String>> {
    Ok(canonical_location(conn, library_id, rel_path)?.map(|(_, p)| p))
}

/// [`canonical_rel_path`] with the original's library too, for callers that
/// write per-library rows (the manual date lives on the `image_exif` row).
pub fn canonical_location(
    conn: &mut SqliteConnection,
    library_id: Option<i32>,
    rel_path: &str,
) -> QueryResult<Option<(i32, String)>> {
    #[derive(QueryableByName)]
    struct Row {
        #[diesel(sql_type = Integer)]
        library_id: i32,
        #[diesel(sql_type = Text)]
        rel_path: String,
    }
    let rows = sql_query(
        "SELECT o.library_id AS library_id, o.rel_path AS rel_path FROM image_exif v \
         JOIN photo_versions pv ON pv.version_hash = v.content_hash \
         JOIN image_exif o ON o.content_hash = pv.original_hash \
         WHERE v.rel_path = ?1 AND (?2 IS NULL OR v.library_id = ?2) \
           AND NOT EXISTS (SELECT 1 FROM image_exif x WHERE x.rel_path = v.rel_path \
                             AND NOT EXISTS (SELECT 1 FROM photo_versions p \
                                              WHERE p.version_hash = x.content_hash)) \
         ORDER BY (o.library_id = v.library_id) DESC, o.id \
         LIMIT 1",
    )
    .bind::<Text, _>(rel_path)
    .bind::<Nullable<Integer>, _>(library_id)
    .load::<Row>(conn)?;
    Ok(rows.into_iter().next().map(|r| (r.library_id, r.rel_path)))
}

/// Fold favorites and tags recorded on a version's own path into its
/// original's, now that reads and writes on the version path canonicalise
/// to the original and could never reach them again. Set-valued data is
/// unioned (duplicates dropped), then the version path's rows are removed.
/// A no-op unless the version has a live original (`canonical_rel_path`).
fn merge_annotations_into_original(
    conn: &mut SqliteConnection,
    library_id: i32,
    version_rel_path: &str,
) -> QueryResult<()> {
    let Some(original) = canonical_rel_path(conn, Some(library_id), version_rel_path)? else {
        return Ok(());
    };
    if original == version_rel_path {
        return Ok(());
    }
    sql_query(
        "INSERT INTO favorites (userid, rel_path) \
         SELECT f.userid, ?2 FROM favorites f WHERE f.rel_path = ?1 \
           AND NOT EXISTS (SELECT 1 FROM favorites g WHERE g.userid = f.userid AND g.rel_path = ?2)",
    )
    .bind::<Text, _>(version_rel_path)
    .bind::<Text, _>(&original)
    .execute(conn)?;
    sql_query("DELETE FROM favorites WHERE rel_path = ?")
        .bind::<Text, _>(version_rel_path)
        .execute(conn)?;
    sql_query(
        "INSERT INTO tagged_photo (rel_path, tag_id, created_time, content_hash) \
         SELECT ?2, t.tag_id, t.created_time, \
                (SELECT content_hash FROM image_exif WHERE rel_path = ?2 \
                  AND content_hash IS NOT NULL LIMIT 1) \
         FROM tagged_photo t WHERE t.rel_path = ?1 \
           AND NOT EXISTS (SELECT 1 FROM tagged_photo u WHERE u.tag_id = t.tag_id AND u.rel_path = ?2)",
    )
    .bind::<Text, _>(version_rel_path)
    .bind::<Text, _>(&original)
    .execute(conn)?;
    sql_query("DELETE FROM tagged_photo WHERE rel_path = ?")
        .bind::<Text, _>(version_rel_path)
        .execute(conn)?;
    Ok(())
}

/// Root original of `hash`: follows `photo_versions` until a hash that is
/// not itself a version. `None` on a cycle.
fn root_original(conn: &mut SqliteConnection, hash: &str) -> QueryResult<Option<String>> {
    #[derive(QueryableByName)]
    struct Row {
        #[diesel(sql_type = Text)]
        original_hash: String,
    }
    let mut current = hash.to_string();
    for _ in 0..8 {
        let next = sql_query("SELECT original_hash FROM photo_versions WHERE version_hash = ?")
            .bind::<Text, _>(&current)
            .load::<Row>(conn)?
            .into_iter()
            .next();
        match next {
            Some(r) => current = r.original_hash,
            None => return Ok(Some(current)),
        }
    }
    Ok(None)
}

/// Columns the version copy takes from its original, in `UPDATE ... SET (..)`
/// order, followed by the source marker.
const COPIED_COLUMNS: &str = "date_taken, date_utc, camera_offset_minutes, date_taken_tz_source, \
     gps_latitude, gps_longitude, gps_altitude, camera_make, camera_model, lens_model, \
     focal_length, aperture, shutter_speed, iso";

fn copied_differs() -> String {
    COPIED_COLUMNS
        .split(',')
        .map(|c| c.trim())
        .map(|c| format!("v.{c} IS NOT o.{c}"))
        .collect::<Vec<_>>()
        .join(" OR ")
}

impl VersionDao for SqliteVersionDao {
    fn link_version(&mut self, link: &NewVersionLink) -> Result<LinkOutcome, DbError> {
        trace_db_call("insert", "link_version", |_span| {
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");

            if !is_known_kind(&link.kind) {
                return Ok(LinkOutcome::Rejected("unknown kind"));
            }
            if link.version_hash == link.original_hash {
                return Ok(LinkOutcome::Rejected("a file cannot be a version of itself"));
            }

            conn.immediate_transaction::<LinkOutcome, anyhow::Error, _>(|conn| {
                let existing =
                    sql_query("SELECT version_hash, original_hash, kind, source, created_at, library_id, rel_path FROM photo_versions WHERE version_hash = ?")
                        .bind::<Text, _>(&link.version_hash)
                        .load::<PhotoVersionRow>(conn)?;
                if !existing.is_empty() {
                    return Ok(LinkOutcome::AlreadyLinked);
                }

                // Flatten: link to the root, never to another version.
                let Some(root) = root_original(conn, &link.original_hash)? else {
                    return Ok(LinkOutcome::Rejected("link would form a cycle"));
                };
                if root == link.version_hash {
                    return Ok(LinkOutcome::Rejected("link would form a cycle"));
                }

                sql_query(
                    "INSERT INTO photo_versions \
                     (version_hash, original_hash, kind, source, created_at, library_id, rel_path) \
                     VALUES (?, ?, ?, ?, ?, ?, ?)",
                )
                .bind::<Text, _>(&link.version_hash)
                .bind::<Text, _>(&root)
                .bind::<Text, _>(&link.kind)
                .bind::<Text, _>(&link.source)
                .bind::<BigInt, _>(chrono::Utc::now().timestamp())
                .bind::<Integer, _>(link.library_id)
                .bind::<Text, _>(&link.rel_path)
                .execute(conn)?;

                // A version is never a face / CLIP candidate; if it was
                // ingested before the link, drop what was already derived.
                sql_query("DELETE FROM face_detections WHERE content_hash = ?")
                    .bind::<Text, _>(&link.version_hash)
                    .execute(conn)?;
                sql_query(
                    "UPDATE image_exif SET clip_embedding = NULL, clip_model_version = NULL \
                     WHERE content_hash = ?",
                )
                .bind::<Text, _>(&link.version_hash)
                .execute(conn)?;
                sql_query("DELETE FROM clip_failures WHERE content_hash = ?")
                    .bind::<Text, _>(&link.version_hash)
                    .execute(conn)?;

                // Versions that pointed at this file now point at the root.
                sql_query("UPDATE photo_versions SET original_hash = ? WHERE original_hash = ?")
                    .bind::<Text, _>(&root)
                    .bind::<Text, _>(&link.version_hash)
                    .execute(conn)?;
                // A display choice made for the file that just became a
                // version is meaningless.
                sql_query("DELETE FROM photo_version_display WHERE original_hash = ?")
                    .bind::<Text, _>(&link.version_hash)
                    .execute(conn)?;

                for reference in &link.references {
                    // The primary original is not a reference to itself.
                    if *reference == root || *reference == link.version_hash {
                        continue;
                    }
                    sql_query(
                        "INSERT OR IGNORE INTO photo_version_references \
                         (version_hash, reference_hash) VALUES (?, ?)",
                    )
                    .bind::<Text, _>(&link.version_hash)
                    .bind::<Text, _>(reference)
                    .execute(conn)?;
                }

                merge_annotations_into_original(conn, link.library_id, &link.rel_path)?;
                Ok(LinkOutcome::Linked)
            })
        })
        .map_err(|e| DbError::log(DbErrorKind::InsertError, e))
    }

    fn get_link(&mut self, version_hash: &str) -> Result<Option<PhotoVersionRow>, DbError> {
        trace_db_call("query", "get_version_link", |_span| {
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            sql_query(
                "SELECT version_hash, original_hash, kind, source, created_at, library_id, rel_path \
                 FROM photo_versions WHERE version_hash = ?",
            )
            .bind::<Text, _>(version_hash)
            .load::<PhotoVersionRow>(&mut *conn)
            .map(|rows| rows.into_iter().next())
            .map_err(|e| anyhow::anyhow!("Query error: {}", e))
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn original_path_of(&mut self, rel_path: &str) -> Result<Option<String>, DbError> {
        trace_db_call("query", "original_path_of", |_span| {
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            canonical_rel_path(&mut conn, None, rel_path)
                .map_err(|e| anyhow::anyhow!("Query error: {}", e))
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn original_location_of(
        &mut self,
        library_id: i32,
        rel_path: &str,
    ) -> Result<Option<(i32, String)>, DbError> {
        trace_db_call("query", "original_location_of", |_span| {
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            canonical_location(&mut conn, Some(library_id), rel_path)
                .map_err(|e| anyhow::anyhow!("Query error: {}", e))
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn version_hashes_among(&mut self, hashes: &[String]) -> Result<HashSet<String>, DbError> {
        trace_db_call("query", "version_hashes_among", |_span| {
            #[derive(QueryableByName)]
            struct Row {
                #[diesel(sql_type = Text)]
                version_hash: String,
            }
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            let mut out = HashSet::new();
            for chunk in hashes.chunks(CHUNK) {
                let sql = format!(
                    "SELECT version_hash FROM photo_versions WHERE version_hash IN ({})",
                    placeholders(chunk.len())
                );
                let mut q = sql_query(sql).into_boxed();
                for h in chunk {
                    q = q.bind::<Text, _>(h);
                }
                let rows = q
                    .load::<Row>(&mut *conn)
                    .map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
                out.extend(rows.into_iter().map(|r| r.version_hash));
            }
            Ok(out)
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn list_versions(&mut self, original_hash: &str) -> Result<Vec<VersionFile>, DbError> {
        trace_db_call("query", "list_versions", |_span| {
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            sql_query(
                "SELECT pv.version_hash AS version_hash, pv.kind AS kind, pv.source AS source, \
                        pv.created_at AS created_at, e.library_id AS library_id, \
                        e.rel_path AS rel_path, e.width AS width, e.height AS height \
                 FROM photo_versions pv \
                 JOIN image_exif e ON e.id = \
                      (SELECT MIN(id) FROM image_exif WHERE content_hash = pv.version_hash) \
                 WHERE pv.original_hash = ? \
                 ORDER BY pv.created_at, e.rel_path",
            )
            .bind::<Text, _>(original_hash)
            .load::<VersionFile>(&mut *conn)
            .map_err(|e| anyhow::anyhow!("Query error: {}", e))
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn list_references(&mut self, version_hash: &str) -> Result<Vec<String>, DbError> {
        trace_db_call("query", "list_version_references", |_span| {
            #[derive(QueryableByName)]
            struct Row {
                #[diesel(sql_type = Text)]
                reference_hash: String,
            }
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            sql_query(
                "SELECT reference_hash FROM photo_version_references \
                 WHERE version_hash = ? ORDER BY reference_hash",
            )
            .bind::<Text, _>(version_hash)
            .load::<Row>(&mut *conn)
            .map(|rows| rows.into_iter().map(|r| r.reference_hash).collect())
            .map_err(|e| anyhow::anyhow!("Query error: {}", e))
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn files_for_hashes(
        &mut self,
        hashes: &[String],
        prefer_library: Option<i32>,
    ) -> Result<HashMap<String, (i32, String)>, DbError> {
        trace_db_call("query", "version_files_for_hashes", |_span| {
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            let mut out: HashMap<String, (i32, String)> = HashMap::new();
            for chunk in hashes.chunks(CHUNK) {
                let sql = format!(
                    "SELECT content_hash, library_id, rel_path FROM image_exif \
                     WHERE content_hash IN ({}) ORDER BY id",
                    placeholders(chunk.len())
                );
                let mut q = sql_query(sql).into_boxed();
                for h in chunk {
                    q = q.bind::<Text, _>(h);
                }
                let rows = q
                    .load::<HashedFile>(&mut *conn)
                    .map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
                for r in rows {
                    let better = match out.get(&r.content_hash) {
                        None => true,
                        Some((lib, _)) => {
                            prefer_library == Some(r.library_id) && prefer_library != Some(*lib)
                        }
                    };
                    if better {
                        out.insert(r.content_hash, (r.library_id, r.rel_path));
                    }
                }
            }
            Ok(out)
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn get_display(&mut self, original_hash: &str) -> Result<Option<String>, DbError> {
        trace_db_call("query", "get_version_display", |_span| {
            #[derive(QueryableByName)]
            struct Row {
                #[diesel(sql_type = Text)]
                display_hash: String,
            }
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            sql_query("SELECT display_hash FROM photo_version_display WHERE original_hash = ?")
                .bind::<Text, _>(original_hash)
                .load::<Row>(&mut *conn)
                .map(|rows| rows.into_iter().next().map(|r| r.display_hash))
                .map_err(|e| anyhow::anyhow!("Query error: {}", e))
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn set_display(
        &mut self,
        original_hash: &str,
        display_hash: Option<&str>,
    ) -> Result<(), DbError> {
        trace_db_call("update", "set_version_display", |_span| {
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            match display_hash {
                Some(display) => sql_query(
                    "INSERT INTO photo_version_display (original_hash, display_hash, decided_at) \
                     VALUES (?, ?, ?) \
                     ON CONFLICT(original_hash) DO UPDATE SET \
                        display_hash = excluded.display_hash, decided_at = excluded.decided_at",
                )
                .bind::<Text, _>(original_hash)
                .bind::<Text, _>(display)
                .bind::<BigInt, _>(chrono::Utc::now().timestamp())
                .execute(&mut *conn),
                None => sql_query("DELETE FROM photo_version_display WHERE original_hash = ?")
                    .bind::<Text, _>(original_hash)
                    .execute(&mut *conn),
            }
            .map(|_| ())
            .map_err(|e| anyhow::anyhow!("Update error: {}", e))
        })
        .map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
    }

    fn photo_summaries(&mut self) -> Result<HashMap<(i32, String), PhotoVersionSummary>, DbError> {
        trace_db_call("query", "photo_version_summaries", |_span| {
            #[derive(QueryableByName)]
            struct OriginalRow {
                #[diesel(sql_type = Text)]
                original_hash: String,
                #[diesel(sql_type = Integer)]
                library_id: i32,
                #[diesel(sql_type = Text)]
                rel_path: String,
                #[diesel(sql_type = BigInt)]
                version_count: i64,
            }
            #[derive(QueryableByName)]
            struct DisplayRow {
                #[diesel(sql_type = Text)]
                original_hash: String,
                #[diesel(sql_type = Integer)]
                library_id: i32,
                #[diesel(sql_type = Text)]
                rel_path: String,
            }
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");

            let originals = sql_query(
                "SELECT o.content_hash AS original_hash, o.library_id AS library_id, \
                        o.rel_path AS rel_path, \
                        (SELECT COUNT(*) FROM photo_versions pv \
                          WHERE pv.original_hash = o.content_hash \
                            AND EXISTS (SELECT 1 FROM image_exif v \
                                         WHERE v.content_hash = pv.version_hash)) AS version_count \
                 FROM image_exif o \
                 WHERE o.content_hash IN (SELECT original_hash FROM photo_versions)",
            )
            .load::<OriginalRow>(&mut *conn)
            .map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
            if originals.is_empty() {
                return Ok(HashMap::new());
            }

            // Only a display choice that is still a version of that original
            // counts; anything else falls back to the original.
            let displays = sql_query(
                "SELECT d.original_hash AS original_hash, v.library_id AS library_id, \
                        v.rel_path AS rel_path \
                 FROM photo_version_display d \
                 JOIN photo_versions pv ON pv.version_hash = d.display_hash \
                                       AND pv.original_hash = d.original_hash \
                 JOIN image_exif v ON v.content_hash = d.display_hash \
                 ORDER BY v.id",
            )
            .load::<DisplayRow>(&mut *conn)
            .map_err(|e| anyhow::anyhow!("Query error: {}", e))?;
            let mut display_by_hash: HashMap<String, Vec<(i32, String)>> = HashMap::new();
            for d in displays {
                display_by_hash
                    .entry(d.original_hash)
                    .or_default()
                    .push((d.library_id, d.rel_path));
            }

            let mut out = HashMap::new();
            for o in originals {
                if o.version_count == 0 {
                    continue;
                }
                // Clients fetch the display file through the original's
                // library, so a chosen version with no row there falls back
                // to the original.
                let display = display_by_hash
                    .get(&o.original_hash)
                    .and_then(|files| files.iter().find(|(lib, _)| *lib == o.library_id).cloned());
                out.insert(
                    (o.library_id, o.rel_path),
                    PhotoVersionSummary {
                        version_count: o.version_count as u32,
                        display,
                    },
                );
            }
            Ok(out)
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn hidden_version_rows(
        &mut self,
        view_library_ids: &[i32],
    ) -> Result<HashSet<(i32, String)>, DbError> {
        trace_db_call("query", "hidden_version_rows", |_span| {
            #[derive(QueryableByName)]
            struct Row {
                #[diesel(sql_type = Integer)]
                library_id: i32,
                #[diesel(sql_type = Text)]
                rel_path: String,
            }
            let sql = format!(
                "SELECT image_exif.library_id AS library_id, image_exif.rel_path AS rel_path \
                 FROM image_exif \
                 JOIN photo_versions pv ON pv.version_hash = image_exif.content_hash \
                 WHERE EXISTS (SELECT 1 FROM image_exif orig \
                                WHERE orig.content_hash = pv.original_hash AND {})",
                reachable_original_condition(view_library_ids)
            );
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            sql_query(sql)
                .load::<Row>(&mut *conn)
                .map(|rows| {
                    rows.into_iter()
                        .map(|r| (r.library_id, r.rel_path))
                        .collect()
                })
                .map_err(|e| anyhow::anyhow!("Query error: {}", e))
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn dirs_holding_only_hidden_versions(
        &mut self,
        view_library_ids: &[i32],
        dirs: &[String],
    ) -> Result<HashSet<String>, DbError> {
        if dirs.is_empty() {
            return Ok(HashSet::new());
        }
        trace_db_call("query", "dirs_holding_only_hidden_versions", |_span| {
            #[derive(QueryableByName)]
            struct Counts {
                #[diesel(sql_type = BigInt)]
                total: i64,
                #[diesel(sql_type = BigInt)]
                visible: i64,
            }
            #[derive(QueryableByName)]
            struct Id {
                #[diesel(sql_type = Integer)]
                id: i32,
            }
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            // Scan the view's libraries (all of them when unscoped) so the
            // (library_id, rel_path) index serves the prefix range.
            let scan: Vec<i32> = if view_library_ids.is_empty() {
                sql_query("SELECT id FROM libraries")
                    .load::<Id>(&mut *conn)?
                    .into_iter()
                    .map(|r| r.id)
                    .collect()
            } else {
                view_library_ids.to_vec()
            };
            let list = scan
                .iter()
                .map(i32::to_string)
                .collect::<Vec<_>>()
                .join(",");
            let sql = format!(
                "SELECT COUNT(*) AS total, \
                        COALESCE(SUM(CASE WHEN {} THEN 1 ELSE 0 END), 0) AS visible \
                 FROM image_exif \
                 WHERE library_id IN ({list}) AND rel_path >= ? AND rel_path < ?",
                visible_version_clause(view_library_ids)
            );
            let mut out = HashSet::new();
            for dir in dirs {
                let dir = dir.trim_end_matches('/');
                if dir.is_empty() {
                    continue;
                }
                // '0' is the character after '/', closing the prefix range.
                let c = sql_query(&sql)
                    .bind::<Text, _>(format!("{dir}/"))
                    .bind::<Text, _>(format!("{dir}0"))
                    .get_result::<Counts>(&mut *conn)?;
                if c.total > 0 && c.visible == 0 {
                    out.insert(dir.to_string());
                }
            }
            Ok(out)
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn unlinked_name_candidates(
        &mut self,
        library_id: i32,
        after_id: i32,
        limit: i64,
    ) -> Result<Vec<(i32, String, String)>, DbError> {
        trace_db_call("query", "unlinked_name_candidates", |_span| {
            #[derive(QueryableByName)]
            struct Row {
                #[diesel(sql_type = Integer)]
                id: i32,
                #[diesel(sql_type = Text)]
                rel_path: String,
                #[diesel(sql_type = Text)]
                content_hash: String,
            }
            let kinds = KINDS
                .iter()
                .map(|k| format!("rel_path LIKE '%.{k}.%'"))
                .collect::<Vec<_>>()
                .join(" OR ");
            let sql = format!(
                "SELECT id, rel_path, content_hash FROM image_exif \
                 WHERE library_id = ? AND id > ? AND content_hash IS NOT NULL AND ({kinds}) \
                   AND NOT EXISTS (SELECT 1 FROM photo_versions pv \
                                    WHERE pv.version_hash = image_exif.content_hash) \
                 ORDER BY id LIMIT ?"
            );
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");
            sql_query(sql)
                .bind::<Integer, _>(library_id)
                .bind::<Integer, _>(after_id)
                .bind::<BigInt, _>(limit)
                .load::<Row>(&mut *conn)
                .map(|rows| {
                    rows.into_iter()
                        .map(|r| (r.id, r.rel_path, r.content_hash))
                        .collect()
                })
                .map_err(|e| anyhow::anyhow!("Query error: {}", e))
        })
        .map_err(|e| DbError::log(DbErrorKind::QueryError, e))
    }

    fn sync_version_metadata(&mut self) -> Result<usize, DbError> {
        trace_db_call("update", "sync_version_metadata", |_span| {
            #[derive(QueryableByName)]
            struct Pair {
                #[diesel(sql_type = Integer)]
                vid: i32,
                #[diesel(sql_type = Integer)]
                oid: i32,
            }
            let mut conn = self.connection.lock().expect("Unable to get VersionDao");

            // Version rows whose copy is missing or stale, against the
            // earliest row of a converted original. `IS NOT` compares NULLs.
            let stale = sql_query(format!(
                "SELECT v.id AS vid, o.id AS oid \
                 FROM image_exif v \
                 JOIN photo_versions pv ON pv.version_hash = v.content_hash \
                 JOIN image_exif o ON o.id = \
                      (SELECT MIN(id) FROM image_exif WHERE content_hash = pv.original_hash) \
                 WHERE o.date_taken IS NOT NULL AND o.date_utc IS NOT NULL \
                   AND (v.date_taken_source IS NOT '{DATE_SOURCE_VERSION_OF}' OR {})",
                copied_differs()
            ))
            .load::<Pair>(&mut *conn)
            .map_err(|e| anyhow::anyhow!("Query error: {}", e))?;

            for p in &stale {
                sql_query(format!(
                    "UPDATE image_exif SET ({COPIED_COLUMNS}, date_taken_source) = \
                     (SELECT {COPIED_COLUMNS}, '{DATE_SOURCE_VERSION_OF}' \
                        FROM image_exif WHERE id = ?) \
                     WHERE id = ?"
                ))
                .bind::<Integer, _>(p.oid)
                .bind::<Integer, _>(p.vid)
                .execute(&mut *conn)
                .map_err(|e| anyhow::anyhow!("Update error: {}", e))?;
            }
            Ok(stale.len())
        })
        .map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
    }
}

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

    fn dao() -> (SqliteVersionDao, Arc<Mutex<SqliteConnection>>) {
        let conn = Arc::new(Mutex::new(in_memory_db_connection()));
        (SqliteVersionDao::from_shared(conn.clone()), conn)
    }

    fn exec(conn: &Arc<Mutex<SqliteConnection>>, sql: &str) {
        sql_query(sql)
            .execute(&mut *conn.lock().unwrap())
            .unwrap_or_else(|e| panic!("{sql}: {e}"));
    }

    fn insert_exif(conn: &Arc<Mutex<SqliteConnection>>, lib: i32, rel: &str, hash: &str) {
        exec(
            conn,
            &format!(
                "INSERT OR IGNORE INTO libraries (id, name, root_path, created_at) \
                 VALUES ({lib}, 'lib{lib}', '/tmp/lib{lib}', 0)"
            ),
        );
        exec(
            conn,
            &format!(
                "INSERT INTO image_exif (library_id, rel_path, created_time, last_modified, content_hash) \
                 VALUES ({lib}, '{rel}', 0, 0, '{hash}')"
            ),
        );
    }

    fn link(vh: &str, oh: &str, rel: &str) -> NewVersionLink {
        NewVersionLink {
            version_hash: vh.into(),
            original_hash: oh.into(),
            kind: "restored".into(),
            source: SOURCE_XMP.into(),
            library_id: 1,
            rel_path: rel.into(),
            references: vec![],
        }
    }

    #[test]
    fn link_is_recorded_once_and_rejects_nonsense() {
        let (mut dao, _c) = dao();
        assert_eq!(
            dao.link_version(&link("v", "o", "a/v.png")).unwrap(),
            LinkOutcome::Linked
        );
        assert_eq!(
            dao.link_version(&link("v", "other", "a/v.png")).unwrap(),
            LinkOutcome::AlreadyLinked,
            "an existing link is never silently re-pointed"
        );
        assert_eq!(dao.get_link("v").unwrap().unwrap().original_hash, "o");
        assert!(matches!(
            dao.link_version(&link("x", "x", "a/x.png")).unwrap(),
            LinkOutcome::Rejected(_)
        ));
        let mut bad_kind = link("y", "o", "a/y.png");
        bad_kind.kind = "bogus".into();
        assert!(matches!(
            dao.link_version(&bad_kind).unwrap(),
            LinkOutcome::Rejected(_)
        ));
    }

    #[test]
    fn chains_are_flattened_to_the_root() {
        let (mut dao, _c) = dao();
        dao.link_version(&link("v1", "root", "v1.png")).unwrap();
        // A second-generation edit derived from the first version.
        dao.link_version(&link("v2", "v1", "v2.png")).unwrap();
        assert_eq!(dao.get_link("v2").unwrap().unwrap().original_hash, "root");

        // Linking an existing original underneath a new root moves its versions up.
        dao.link_version(&link("o2", "top", "o2.png")).unwrap();
        dao.link_version(&link("child", "o2", "child.png")).unwrap();
        assert_eq!(dao.get_link("child").unwrap().unwrap().original_hash, "top");

        // A cycle is refused.
        assert!(matches!(
            dao.link_version(&link("root", "v2", "root.png")).unwrap(),
            LinkOutcome::Rejected(_)
        ));
    }

    #[test]
    fn references_are_stored_but_never_the_original() {
        let (mut dao, _c) = dao();
        let mut l = link("v", "o", "v.png");
        l.references = vec!["ref1".into(), "o".into(), "ref2".into(), "ref1".into()];
        dao.link_version(&l).unwrap();
        assert_eq!(dao.list_references("v").unwrap(), vec!["ref1", "ref2"]);
    }

    #[test]
    fn visible_clause_hides_only_while_original_is_reachable() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "a/orig.jpg", "o");
        insert_exif(&c, 1, "a/orig.jpg.restored.png", "v");
        insert_exif(&c, 2, "b/other.jpg", "other");
        dao.link_version(&link("v", "o", "a/orig.jpg.restored.png"))
            .unwrap();

        let visible = |libs: &[i32]| -> Vec<String> {
            #[derive(QueryableByName)]
            struct R {
                #[diesel(sql_type = Text)]
                rel_path: String,
            }
            let mut rows: Vec<String> = sql_query(format!(
                "SELECT rel_path FROM image_exif WHERE {} ORDER BY rel_path",
                visible_version_clause(libs)
            ))
            .load::<R>(&mut *c.lock().unwrap())
            .unwrap()
            .into_iter()
            .map(|r| r.rel_path)
            .collect();
            rows.sort();
            rows
        };

        assert_eq!(visible(&[]), vec!["a/orig.jpg", "b/other.jpg"]);
        assert_eq!(visible(&[1, 2]), vec!["a/orig.jpg", "b/other.jpg"]);
        // The clause only decides hiding; the caller scopes rows to the view.
        // With the view narrowed to lib 2 the pair's original is out of
        // reach, so the version is not hidden.
        assert!(visible(&[2]).contains(&"a/orig.jpg.restored.png".to_string()));
        exec(
            &c,
            "UPDATE image_exif SET library_id = 2 WHERE rel_path = 'a/orig.jpg'",
        );
        assert!(
            visible(&[1]).contains(&"a/orig.jpg.restored.png".to_string()),
            "original lives outside library 1"
        );
        // Original deleted: the version shows up everywhere.
        exec(&c, "DELETE FROM image_exif WHERE rel_path = 'a/orig.jpg'");
        assert_eq!(visible(&[]), vec!["a/orig.jpg.restored.png", "b/other.jpg"]);
    }

    #[test]
    fn hidden_rows_and_summaries_follow_the_same_rule() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "orig.jpg", "o");
        insert_exif(&c, 1, "aiedits/orig.jpg.restored.png", "v1");
        insert_exif(&c, 1, "aiedits/orig.jpg.edit.png", "v2");
        dao.link_version(&link("v1", "o", "aiedits/orig.jpg.restored.png"))
            .unwrap();
        let mut l2 = link("v2", "o", "aiedits/orig.jpg.edit.png");
        l2.kind = "edit".into();
        dao.link_version(&l2).unwrap();

        let hidden = dao.hidden_version_rows(&[]).unwrap();
        assert_eq!(hidden.len(), 2);
        assert!(hidden.contains(&(1, "aiedits/orig.jpg.edit.png".to_string())));
        assert!(dao.hidden_version_rows(&[3]).unwrap().is_empty());

        let s = dao.photo_summaries().unwrap();
        let sum = &s[&(1, "orig.jpg".to_string())];
        assert_eq!(sum.version_count, 2);
        assert_eq!(sum.display, None);

        dao.set_display("o", Some("v2")).unwrap();
        let s = dao.photo_summaries().unwrap();
        assert_eq!(
            s[&(1, "orig.jpg".to_string())].display,
            Some((1, "aiedits/orig.jpg.edit.png".to_string()))
        );

        // A display file that vanished falls back to the original.
        exec(&c, "DELETE FROM image_exif WHERE content_hash = 'v2'");
        let s = dao.photo_summaries().unwrap();
        let sum = &s[&(1, "orig.jpg".to_string())];
        assert_eq!(sum.version_count, 1);
        assert_eq!(sum.display, None);

        dao.set_display("o", None).unwrap();
        assert_eq!(dao.get_display("o").unwrap(), None);
    }

    #[test]
    fn no_versions_means_no_summaries() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "a.jpg", "a");
        assert!(dao.photo_summaries().unwrap().is_empty());
        assert!(dao.hidden_version_rows(&[]).unwrap().is_empty());
    }

    #[test]
    fn canonical_path_resolves_a_version_to_its_original() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "orig.jpg", "o");
        insert_exif(&c, 1, "aiedits/orig.jpg.restored.png", "v");
        dao.link_version(&link("v", "o", "aiedits/orig.jpg.restored.png"))
            .unwrap();
        let mut conn = c.lock().unwrap();
        assert_eq!(
            canonical_rel_path(&mut conn, None, "aiedits/orig.jpg.restored.png").unwrap(),
            Some("orig.jpg".to_string())
        );
        assert_eq!(
            canonical_rel_path(&mut conn, None, "orig.jpg").unwrap(),
            None
        );
        drop(conn);
        // Original gone: nothing to bind to, the version is a normal photo.
        exec(&c, "DELETE FROM image_exif WHERE content_hash = 'o'");
        let mut conn = c.lock().unwrap();
        assert_eq!(
            canonical_rel_path(&mut conn, None, "aiedits/orig.jpg.restored.png").unwrap(),
            None
        );
    }

    #[test]
    fn canonical_path_is_scoped_to_the_versions_library() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "album_a/orig.jpg", "o");
        insert_exif(&c, 1, "album_a/aiedits/orig.jpg.edit.png", "v");
        dao.link_version(&link("v", "o", "album_a/aiedits/orig.jpg.edit.png"))
            .unwrap();
        // Library 2 has an unrelated, unlinked photo at the same path.
        insert_exif(&c, 2, "album_a/aiedits/orig.jpg.edit.png", "other");
        let mut conn = c.lock().unwrap();
        let p = "album_a/aiedits/orig.jpg.edit.png";
        assert_eq!(
            canonical_rel_path(&mut conn, Some(2), p).unwrap(),
            None,
            "library 2's row is not a version"
        );
        // Annotations are path-keyed, so a path that also names a
        // standalone photo is never redirected away from it.
        assert_eq!(canonical_rel_path(&mut conn, None, p).unwrap(), None);
        assert_eq!(
            canonical_rel_path(&mut conn, Some(1), p).unwrap(),
            None,
            "shared path with a standalone photo stays put"
        );
        drop(conn);
        exec(&c, "DELETE FROM image_exif WHERE content_hash = 'other'");
        let mut conn = c.lock().unwrap();
        assert_eq!(
            canonical_rel_path(&mut conn, Some(1), p).unwrap(),
            Some("album_a/orig.jpg".to_string())
        );
        assert_eq!(canonical_rel_path(&mut conn, Some(2), p).unwrap(), None);
    }

    #[test]
    fn linking_merges_annotations_from_the_version_path_into_the_original() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "album_a/orig.jpg", "o");
        insert_exif(&c, 1, "album_a/aiedits/orig.jpg.edit.png", "v");
        let v = "album_a/aiedits/orig.jpg.edit.png";
        let o = "album_a/orig.jpg";
        for t in ["t1", "t2"] {
            exec(
                &c,
                &format!("INSERT INTO tags (name, created_time) VALUES ('{t}', 0)"),
            );
        }
        // Tag 1 on both paths (duplicate), tag 2 only on the version.
        exec(
            &c,
            &format!(
                "INSERT INTO tagged_photo (rel_path, tag_id, created_time) \
             SELECT '{o}', id, 1 FROM tags WHERE name = 't1'"
            ),
        );
        exec(
            &c,
            &format!(
                "INSERT INTO tagged_photo (rel_path, tag_id, created_time) \
             SELECT '{v}', id, 2 FROM tags"
            ),
        );
        // User 1 favorited both; user 2 only the version.
        exec(
            &c,
            &format!(
                "INSERT INTO favorites (userid, rel_path) VALUES (1, '{o}'), (1, '{v}'), (2, '{v}')"
            ),
        );

        dao.link_version(&link("v", "o", v)).unwrap();

        let rows = |sql: &str| -> Vec<(String, i32)> {
            #[derive(QueryableByName)]
            struct R {
                #[diesel(sql_type = Text)]
                p: String,
                #[diesel(sql_type = Integer)]
                k: i32,
            }
            sql_query(sql)
                .load::<R>(&mut *c.lock().unwrap())
                .unwrap()
                .into_iter()
                .map(|r| (r.p, r.k))
                .collect()
        };
        assert_eq!(
            rows("SELECT rel_path AS p, userid AS k FROM favorites ORDER BY userid"),
            vec![(o.to_string(), 1), (o.to_string(), 2)]
        );
        assert_eq!(
            rows("SELECT rel_path AS p, tag_id AS k FROM tagged_photo ORDER BY tag_id"),
            vec![(o.to_string(), 1), (o.to_string(), 2)]
        );
    }

    #[test]
    fn sync_copies_date_gps_and_camera_from_a_converted_original() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "orig.jpg", "o");
        insert_exif(&c, 1, "orig.jpg.restored.png", "v");
        dao.link_version(&link("v", "o", "orig.jpg.restored.png"))
            .unwrap();

        // Unconverted original: nothing is copied (the version keeps its own date).
        exec(
            &c,
            "UPDATE image_exif SET date_taken = 1000 WHERE content_hash = 'o'",
        );
        assert_eq!(dao.sync_version_metadata().unwrap(), 0);

        exec(
            &c,
            "UPDATE image_exif SET date_taken = 1000, date_utc = 1000, camera_offset_minutes = -240, \
             date_taken_tz_source = 'offset_tag', gps_latitude = 12.5, gps_longitude = 7.25, \
             camera_make = 'Canon', camera_model = 'EOS', date_taken_source = 'exif' \
             WHERE content_hash = 'o'",
        );
        exec(
            &c,
            "UPDATE image_exif SET date_taken = 5, date_taken_source = 'fs_time' WHERE content_hash = 'v'",
        );
        assert_eq!(dao.sync_version_metadata().unwrap(), 1);
        assert_eq!(dao.sync_version_metadata().unwrap(), 0, "idempotent");

        #[derive(QueryableByName)]
        struct R {
            #[diesel(sql_type = Nullable<BigInt>)]
            date_utc: Option<i64>,
            #[diesel(sql_type = Nullable<Text>)]
            src: Option<String>,
            #[diesel(sql_type = Nullable<diesel::sql_types::Float>)]
            lat: Option<f32>,
            #[diesel(sql_type = Nullable<Text>)]
            make: Option<String>,
        }
        let row = sql_query(
            "SELECT date_utc, date_taken_source AS src, gps_latitude AS lat, camera_make AS make \
             FROM image_exif WHERE content_hash = 'v'",
        )
        .get_result::<R>(&mut *c.lock().unwrap())
        .unwrap();
        assert_eq!(row.date_utc, Some(1000));
        assert_eq!(row.src.as_deref(), Some("version_of"));
        assert_eq!(row.lat, Some(12.5));
        assert_eq!(row.make.as_deref(), Some("Canon"));

        // The original's date changes (manual override): the copy follows.
        exec(
            &c,
            "UPDATE image_exif SET date_taken = 2000, date_utc = 2000 WHERE content_hash = 'o'",
        );
        assert_eq!(dao.sync_version_metadata().unwrap(), 1);
    }

    #[test]
    fn name_candidates_skip_linked_rows() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "a/x.jpg.restored.png", "h1");
        insert_exif(&c, 1, "a/x.jpg.edit.2.png", "h2");
        insert_exif(&c, 1, "a/plain.png", "h3");
        insert_exif(&c, 2, "a/y.jpg.restored.png", "h4");
        dao.link_version(&link("h2", "ox", "a/x.jpg.edit.2.png"))
            .unwrap_or(LinkOutcome::Linked);
        let got = dao.unlinked_name_candidates(1, 0, 100).unwrap();
        assert_eq!(got.len(), 1);
        assert_eq!(
            (got[0].1.as_str(), got[0].2.as_str()),
            ("a/x.jpg.restored.png", "h1")
        );
        // The cursor skips rows at or before it.
        assert!(
            dao.unlinked_name_candidates(1, got[0].0, 100)
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn version_hashes_among_and_files_for_hashes() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "o.jpg", "o");
        insert_exif(&c, 2, "o.jpg", "o");
        dao.link_version(&link("v", "o", "v.png")).unwrap();
        let got = dao
            .version_hashes_among(&["v".to_string(), "o".to_string()])
            .unwrap();
        assert_eq!(got, HashSet::from(["v".to_string()]));
        let files = dao.files_for_hashes(&["o".to_string()], Some(2)).unwrap();
        assert_eq!(files["o"], (2, "o.jpg".to_string()));
        let files = dao.files_for_hashes(&["o".to_string()], None).unwrap();
        assert_eq!(files["o"].0, 1);
    }
    fn visible_in(c: &Arc<Mutex<SqliteConnection>>, libs: &[i32]) -> Vec<String> {
        #[derive(QueryableByName)]
        struct R {
            #[diesel(sql_type = Text)]
            rel_path: String,
        }
        let mut rows: Vec<String> = sql_query(format!(
            "SELECT rel_path FROM image_exif WHERE {} ORDER BY rel_path",
            visible_version_clause(libs)
        ))
        .load::<R>(&mut *c.lock().unwrap())
        .unwrap()
        .into_iter()
        .map(|r| r.rel_path)
        .collect();
        rows.sort();
        rows
    }

    #[test]
    fn a_version_surfaces_when_its_original_is_demoted_as_a_duplicate() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "orig.jpg", "o");
        insert_exif(&c, 1, "orig.jpg.restored.png", "v");
        insert_exif(&c, 1, "survivor.jpg", "s");
        dao.link_version(&link("v", "o", "orig.jpg.restored.png"))
            .unwrap();
        assert_eq!(visible_in(&c, &[]), vec!["orig.jpg", "survivor.jpg"]);

        exec(
            &c,
            "UPDATE image_exif SET duplicate_of_hash = 's' WHERE content_hash = 'o'",
        );
        // Unscoped: O is hidden by the duplicate rule, so V stands alone.
        let shown = visible_in(&c, &[]);
        assert!(
            shown.contains(&"orig.jpg.restored.png".to_string()),
            "{shown:?}"
        );
        assert!(dao.hidden_version_rows(&[]).unwrap().is_empty());
        // Scoped with the survivor in view: same.
        assert!(visible_in(&c, &[1]).contains(&"orig.jpg.restored.png".to_string()));
        assert!(dao.hidden_version_rows(&[1]).unwrap().is_empty());

        // Scoped to a view that lacks the survivor, O stays visible (the
        // duplicate rule), so V hides behind it again.
        insert_exif(&c, 2, "elsewhere.jpg", "e");
        exec(
            &c,
            "UPDATE image_exif SET duplicate_of_hash = 'e' WHERE content_hash = 'o'",
        );
        assert!(!visible_in(&c, &[1]).contains(&"orig.jpg.restored.png".to_string()));
        assert_eq!(dao.hidden_version_rows(&[1]).unwrap().len(), 1);
    }

    #[test]
    fn display_file_comes_from_the_originals_library_or_not_at_all() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "orig.jpg", "o");
        // Same bytes in two libraries.
        insert_exif(&c, 2, "copy/v.png", "v");
        insert_exif(&c, 1, "aiedits/v.png", "v");
        dao.link_version(&link("v", "o", "aiedits/v.png")).unwrap();
        dao.set_display("o", Some("v")).unwrap();
        let s = dao.photo_summaries().unwrap();
        assert_eq!(
            s[&(1, "orig.jpg".to_string())].display,
            Some((1, "aiedits/v.png".to_string()))
        );

        // Only the other library holds the chosen bytes: show the original.
        exec(
            &c,
            "DELETE FROM image_exif WHERE library_id = 1 AND content_hash = 'v'",
        );
        let s = dao.photo_summaries().unwrap();
        let sum = &s[&(1, "orig.jpg".to_string())];
        assert_eq!(sum.version_count, 1);
        assert_eq!(sum.display, None);
    }

    #[test]
    fn linking_drops_faces_and_clip_data_derived_before_the_link() {
        let (mut dao, c) = dao();
        insert_exif(&c, 1, "orig.jpg", "o");
        insert_exif(&c, 1, "v.png", "v");
        exec(
            &c,
            "UPDATE image_exif SET clip_embedding = x'00', clip_model_version = 'm' \
             WHERE content_hash IN ('o', 'v')",
        );
        for h in ["o", "v"] {
            exec(
                &c,
                &format!(
                    "INSERT INTO face_detections (library_id, content_hash, rel_path, bbox_x, \
                     embedding, source, status, model_version, created_at) \
                     VALUES (1, '{h}', 'x', 0.1, x'00', 'auto', 'detected', 'm', 0)"
                ),
            );
            exec(
                &c,
                &format!(
                    "INSERT INTO clip_failures (library_id, rel_path, content_hash, error, failed_at) \
                     VALUES (1, 'p-{h}', '{h}', 'boom', 0)"
                ),
            );
        }
        dao.link_version(&link("v", "o", "v.png")).unwrap();

        let n = |sql: &str| -> i64 {
            #[derive(QueryableByName)]
            struct N {
                #[diesel(sql_type = BigInt)]
                n: i64,
            }
            sql_query(sql)
                .get_result::<N>(&mut *c.lock().unwrap())
                .unwrap()
                .n
        };
        assert_eq!(
            n("SELECT COUNT(*) AS n FROM face_detections WHERE content_hash='v'"),
            0
        );
        assert_eq!(
            n("SELECT COUNT(*) AS n FROM clip_failures WHERE content_hash='v'"),
            0
        );
        assert_eq!(
            n(
                "SELECT COUNT(*) AS n FROM image_exif WHERE content_hash='v' \
               AND (clip_embedding IS NOT NULL OR clip_model_version IS NOT NULL)"
            ),
            0
        );
        // The original keeps its own.
        assert_eq!(
            n("SELECT COUNT(*) AS n FROM face_detections WHERE content_hash='o'"),
            1
        );
        assert_eq!(
            n("SELECT COUNT(*) AS n FROM clip_failures WHERE content_hash='o'"),
            1
        );
        assert_eq!(
            n(
                "SELECT COUNT(*) AS n FROM image_exif WHERE content_hash='o' AND clip_embedding IS NOT NULL"
            ),
            1
        );
    }
}