fix(versions): review findings on PR #193

- A version no longer hides behind an original that duplicates resolve
  demoted (visible_version_clause, hidden_version_rows).
- Recording a link drops the version's faces, CLIP data and clip failures
  (covers link-at-ingest and link_versions --apply).
- The display file must live in the original's library; otherwise the
  original shows, and PUT /image/versions/display answers 400.
- link_new_files looks up hashes in chunks of 500 and logs errors.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/database/version_dao.rs

@@ -47,20 +47,33 @@ pub fn is_known_kind(kind: &str) -> bool {
 /// 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 {
-    let scope = if view_library_ids.is_empty() {
-        String::new()
-    } else {
-        let list = view_library_ids
-            .iter()
-            .map(i32::to_string)
-            .collect::<Vec<_>>()
-            .join(",");
-        format!(" AND orig.library_id IN ({list})")
-    };
     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{scope})"
+         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})))"
     )
 }
 
@@ -157,7 +170,8 @@ pub struct HashedFile {
 #[derive(Debug, Clone, PartialEq, Eq, Default)]
 pub struct PhotoVersionSummary {
     pub version_count: u32,
-    /// `(library_id, rel_path)` of the chosen version file.
+    /// `(library_id, rel_path)` of the chosen version file; the library is
+    /// always the original's own.
     pub display: Option<(i32, String)>,
 }
 
@@ -368,6 +382,21 @@ impl VersionDao for SqliteVersionDao {
                 .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)
@@ -639,13 +668,12 @@ impl VersionDao for SqliteVersionDao {
                 if o.version_count == 0 {
                     continue;
                 }
-                let display = display_by_hash.get(&o.original_hash).and_then(|files| {
-                    files
-                        .iter()
-                        .find(|(lib, _)| *lib == o.library_id)
-                        .or_else(|| files.first())
-                        .cloned()
-                });
+                // 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 {
@@ -671,22 +699,13 @@ impl VersionDao for SqliteVersionDao {
                 #[diesel(sql_type = Text)]
                 rel_path: String,
             }
-            let scope = if view_library_ids.is_empty() {
-                String::new()
-            } else {
-                let list = view_library_ids
-                    .iter()
-                    .map(i32::to_string)
-                    .collect::<Vec<_>>()
-                    .join(",");
-                format!(" AND orig.library_id IN ({list})")
-            };
             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{scope})"
+                                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)
@@ -1089,4 +1108,156 @@ mod tests {
         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
+        );
+    }
 }