Find catalog duplicates and fold a group onto one copy in one transaction

The library holds the same RAW in several folders: a dated folder, a
bck/ beside it, a renamed Darktable export tree. dr_catalog::duplicates
is the catalog half of consolidating them (#67).

candidates() is one grouped query over root, camera, capture instant and
size, joined back for the rows; count() is the same grouping under
COUNT. On a copy of the reference catalog (23,582 images) both take
10-35 ms and find 1,836 groups holding 3,379 spare copies.

survivor() prefers a copy outside a backup-looking folder, then one
still named the way the camera named it, then the oldest, then the
lowest id.

consolidate() re-checks the plan against the catalog, merges the copies'
judgements onto the survivor (highest rating, keywords unioned,
collections unioned with the survivor keeping its place, a flag or label
the copies agree on, faces via faces::carry_onto_copy) and records the
copies as trashed, all in one transaction, so a failure part way leaves
the group untouched. preview() runs the same code and rolls it back.

Sameness probes are kept in dedup_probes, created on first use rather
than by a migration: a schema bump would make older builds refuse this
catalog's snapshot at sync. trash::record_trashed_within lets the trash
write share the merge's transaction.

core/dr-catalog/src/error.rs

@@ -96,6 +96,13 @@ pub enum CatalogError {
 
     #[error("io: {0}")]
     Io(String),
+
+    /// TRACES: FR-CAT-11a
+    /// A duplicate group planned earlier no longer holds: a copy was trashed,
+    /// rescanned or changed since the review was drawn. The group is left
+    /// untouched rather than consolidated on a stale plan.
+    #[error("no longer a duplicate: {0}")]
+    StaleDuplicate(String),
 }
 
 impl From<rusqlite::Error> for CatalogError {

core/dr-catalog/src/faces.rs

@@ -1599,6 +1599,170 @@ fn iou(a: (f32, f32, f32, f32), b: (f32, f32, f32, f32)) -> f32 {
     }
 }
 
+/// TRACES: FR-CAT-11a | FR-CULL-10
+/// What [`carry_onto_copy`] did with one byte-identical copy's faces.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub struct FaceCarry {
+    /// Faces moved onto the survivor outright, because it had none.
+    pub moved: usize,
+    /// Names and suggestions put on a survivor's face that lacked one.
+    pub named: usize,
+    /// Rejections added to a survivor's face.
+    pub rejections: usize,
+    /// Faces named one person on the copy and another on the survivor. The
+    /// survivor's name is kept; the copy keeps its own, in the trash.
+    pub conflicts: usize,
+    /// Named faces on the copy that match nothing on the survivor. Left with
+    /// the copy rather than mixed into another pipeline's faces.
+    pub unmatched_named: usize,
+}
+
+/// TRACES: FR-CAT-11a | FR-CULL-10
+/// Bring what one copy of a photograph knows about its faces onto another
+/// copy of the same bytes, inside the caller's transaction.
+///
+/// Faces are per image, so two copies of one file indexed separately hold
+/// two sets of the same boxes, and a name confirmed on one is invisible on
+/// the other. The rule is the one [`record_detections_within`] keeps: an
+/// image holds one pipeline's faces at a time, and the user's judgements
+/// are what must survive.
+///
+/// - **The survivor has no faces at all.** The copy's faces and its run
+///   markers move over wholesale — nothing is duplicated, and the survivor
+///   is spared a detection pass it would only repeat. Its own markers go
+///   first, because a marker saying "examined, nothing found" over an image
+///   that now holds faces is the V12 state.
+/// - **The survivor has faces.** Each of the copy's faces is paired with the
+///   survivor face its box overlaps most (IoU above one half — the bytes are
+///   the same, so the boxes coincide). A name or suggestion is carried onto
+///   a survivor face that has none, a confirmation outranks a suggestion,
+///   and two different confirmed names are a conflict the survivor wins.
+///   Rejections are unioned. The copy's faces stay where they are, with the
+///   copy.
+pub fn carry_onto_copy(
+    tx: &Connection,
+    from: ImageId,
+    to: ImageId,
+) -> Result<FaceCarry, CatalogError> {
+    let mut out = FaceCarry::default();
+    let boxes = |image: ImageId| -> Result<Vec<CopyFace>, CatalogError> {
+        let mut q = tx.prepare(
+            "SELECT f.id, f.x, f.y, f.w, f.h, fp.person_id, fp.probability, fp.confirmed
+               FROM faces f
+               LEFT JOIN face_person fp ON fp.face_id = f.id
+              WHERE f.image_id = ?1
+              ORDER BY f.id",
+        )?;
+        let rows = q.query_map([image.0 as i64], |r| {
+            let person: Option<i64> = r.get(5)?;
+            Ok(CopyFace {
+                id: r.get(0)?,
+                rect: (
+                    r.get::<_, f64>(1)? as f32,
+                    r.get::<_, f64>(2)? as f32,
+                    r.get::<_, f64>(3)? as f32,
+                    r.get::<_, f64>(4)? as f32,
+                ),
+                assignment: match person {
+                    Some(p) => Some((p, r.get::<_, f64>(6)?, r.get::<_, i64>(7)? != 0)),
+                    None => None,
+                },
+            })
+        })?;
+        Ok(rows.collect::<Result<Vec<_>, _>>()?)
+    };
+
+    let theirs = boxes(from)?;
+    if theirs.is_empty() {
+        return Ok(out);
+    }
+    let ours = boxes(to)?;
+
+    if ours.is_empty() {
+        tx.execute("DELETE FROM face_index WHERE image_id = ?1", [to.0 as i64])?;
+        tx.execute(
+            "UPDATE face_index SET image_id = ?2 WHERE image_id = ?1",
+            rusqlite::params![from.0 as i64, to.0 as i64],
+        )?;
+        out.moved = tx.execute(
+            "UPDATE faces SET image_id = ?2 WHERE image_id = ?1",
+            rusqlite::params![from.0 as i64, to.0 as i64],
+        )?;
+        return Ok(out);
+    }
+
+    let mut taken = vec![false; ours.len()];
+    for face in &theirs {
+        let best = ours
+            .iter()
+            .enumerate()
+            .filter(|(i, _)| !taken[*i])
+            .map(|(i, o)| (i, iou(face.rect, o.rect)))
+            .filter(|(_, overlap)| *overlap > 0.5)
+            .max_by(|a, b| a.1.total_cmp(&b.1));
+        let Some((at, _)) = best else {
+            if face.assignment.is_some_and(|(_, _, confirmed)| confirmed) {
+                out.unmatched_named += 1;
+            }
+            continue;
+        };
+        taken[at] = true;
+        let target = &ours[at];
+
+        out.rejections += tx.execute(
+            "INSERT OR IGNORE INTO face_person_rejected (face_id, person_id)
+             SELECT ?2, person_id FROM face_person_rejected WHERE face_id = ?1",
+            rusqlite::params![face.id, target.id],
+        )?;
+
+        let Some((person, probability, confirmed)) = face.assignment else {
+            continue;
+        };
+        let carry = match target.assignment {
+            None => true,
+            // The same person: only a confirmation upgrades a suggestion.
+            Some((p, _, theirs_confirmed)) if p == person => confirmed && !theirs_confirmed,
+            // Another person, only suggested there: the user's word wins.
+            Some((_, _, false)) => confirmed,
+            // Another person, confirmed there: the survivor keeps its name.
+            Some((_, _, true)) => {
+                if confirmed {
+                    out.conflicts += 1;
+                }
+                false
+            }
+        };
+        if carry {
+            tx.execute(
+                "INSERT INTO face_person (face_id, person_id, probability, confirmed)
+                 VALUES (?1, ?2, ?3, ?4)
+                 ON CONFLICT(face_id) DO UPDATE SET
+                     person_id = excluded.person_id,
+                     probability = excluded.probability,
+                     confirmed = excluded.confirmed",
+                rusqlite::params![target.id, person, probability, confirmed],
+            )?;
+            // A name the user gave outranks a rejection of the same pair made
+            // on the survivor — the same order `confirm` applies.
+            if confirmed {
+                tx.execute(
+                    "DELETE FROM face_person_rejected WHERE face_id = ?1 AND person_id = ?2",
+                    rusqlite::params![target.id, person],
+                )?;
+            }
+            out.named += 1;
+        }
+    }
+    Ok(out)
+}
+
+/// One face as [`carry_onto_copy`] pairs it.
+struct CopyFace {
+    id: i64,
+    rect: (f32, f32, f32, f32),
+    assignment: Option<(i64, f64, bool)>,
+}
+
 pub(crate) fn now_secs() -> i64 {
     std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)

core/dr-catalog/src/lib.rs

@@ -40,6 +40,7 @@ pub mod bursts;
 pub mod cache;
 pub mod collections;
 pub mod dedup;
+pub mod duplicates;
 pub mod error;
 pub mod face_shard;
 pub mod faces;

core/dr-catalog/src/trash.rs

@@ -129,28 +129,40 @@ pub fn record_trashed(
         return Ok(0);
     }
     let tx = conn.unchecked_transaction()?;
-    let mut n = 0;
-
-    {
-        let mut stmt = tx.prepare(
-            "UPDATE images
-                SET trashed_from = CASE
-                        WHEN trashed_at IS NULL THEN source_ref
-                        ELSE trashed_from
-                    END,
-                    source_ref   = ?2,
-                    trashed_at   = coalesce(trashed_at, ?3)
-              WHERE id = ?1",
-        )?;
-        for (image, path) in moved {
-            n += stmt.execute(rusqlite::params![image.0 as i64, path, now])?;
-        }
-    }
-
+    let n = record_trashed_within(&tx, moved, now)?;
     tx.commit()?;
     Ok(n)
 }
 
+/// [`record_trashed`] inside a transaction the caller owns.
+///
+/// For a caller whose trash is one half of a larger write that must land
+/// whole or not at all — consolidating duplicates (`crate::duplicates`)
+/// merges a copy's judgements onto the survivor and trashes the copy in one
+/// commit. `unchecked_transaction` cannot nest, so this is offered here
+/// rather than wrapped from above.
+pub fn record_trashed_within(
+    tx: &Connection,
+    moved: &[(ImageId, String)],
+    now: i64,
+) -> Result<usize, CatalogError> {
+    let mut n = 0;
+    let mut stmt = tx.prepare(
+        "UPDATE images
+            SET trashed_from = CASE
+                    WHEN trashed_at IS NULL THEN source_ref
+                    ELSE trashed_from
+                END,
+                source_ref   = ?2,
+                trashed_at   = coalesce(trashed_at, ?3)
+          WHERE id = ?1",
+    )?;
+    for (image, path) in moved {
+        n += stmt.execute(rusqlite::params![image.0 as i64, path, now])?;
+    }
+    Ok(n)
+}
+
 /// Record that images have been moved back out of the trash.
 ///
 /// Call after the move succeeds, for the same reason as [`record_trashed`].