smriti/db/face_repo/

read.rs

//! Read/query methods for FaceRepo.

use rusqlite::{params, params_from_iter, Result as SqliteResult};

use crate::ml::FaceEmbedding;

use super::{FaceClusterRecord, FaceDetail, FaceRepo, FaceStatus, GalleryEmbedding, ReviewItem};

type FacePathRow = (i64, String, i32, f32, f32, f32, f32);
/// (id, file_path, orientation, taken_ts, file_hash)
///
/// `file_hash` is plumbed through so the face worker can locate the
/// pre-generated Large thumbnail on disk and decode that instead of the
/// full image — typically a 5-10× speedup on libraries that have
/// already been thumbnailed.
type UnprocessedPhotoRow = (i64, String, i32, Option<i64>, String);

impl<'a> FaceRepo<'a> {
    /// Count of photos that haven't yet had face detection run. Drives
    /// the "Resume face detection" banner on the People page.
    pub fn count_pending_face_processing(&self) -> SqliteResult<i64> {
        self.conn.query_row(
            "SELECT COUNT(*) FROM photos WHERE faces_processed = FALSE AND is_trashed = FALSE",
            [],
            |row| row.get(0),
        )
    }

    /// Get unclustered faces with photo_id and embeddings.
    pub fn get_unclustered_faces_with_photo_embeddings(
        &self,
    ) -> SqliteResult<Vec<(i64, i64, FaceEmbedding)>> {
        let mut stmt = self.conn.prepare(
            "SELECT f.id, f.photo_id, f.embedding
                 FROM faces f
                 JOIN photos p ON p.id = f.photo_id
                 WHERE f.cluster_id IS NULL
                   AND f.user_confirmed >= 0
                   AND p.is_trashed = FALSE",
        )?;

        let rows = stmt.query_map([], |row| {
            let id: i64 = row.get(0)?;
            let photo_id: i64 = row.get(1)?;
            let bytes: Vec<u8> = row.get(2)?;
            Ok((id, photo_id, bytes))
        })?;

        let mut faces = Vec::new();
        for row in rows {
            let (id, photo_id, bytes) = row?;
            match FaceEmbedding::from_bytes(&bytes) {
                Some(emb) => faces.push((id, photo_id, emb)),
                None => tracing::warn!(
                    "Corrupted face embedding for face_id={}: {} bytes",
                    id,
                    bytes.len()
                ),
            }
        }

        Ok(faces)
    }

    /// Get all clusters, ordered by photo count descending
    pub fn get_all_clusters(&self) -> SqliteResult<Vec<FaceClusterRecord>> {
        let mut stmt = self.conn.prepare(
            r#"
            SELECT id, name, representative_face_id, photo_count
            FROM face_clusters
            ORDER BY photo_count DESC, face_count DESC
            "#,
        )?;

        let rows = stmt.query_map([], |row| {
            Ok(FaceClusterRecord {
                id: row.get(0)?,
                name: row.get(1)?,
                representative_face_id: row.get(2)?,
                photo_count: row.get(3)?,
                face_thumbnail_path: None, // Set after query using drive path
            })
        })?;

        let mut clusters = Vec::new();
        for row in rows {
            clusters.push(row?);
        }

        Ok(clusters)
    }

    pub fn get_gallery_embeddings(&self) -> SqliteResult<Vec<GalleryEmbedding>> {
        let mut stmt = self.conn.prepare(
            "SELECT cluster_id, face_id, embedding FROM person_gallery_embeddings ORDER BY cluster_id",
        )?;

        let rows = stmt.query_map([], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, Vec<u8>>(2)?,
            ))
        })?;

        let mut result = Vec::new();
        for row in rows {
            let (cluster_id, face_id, bytes) = row?;
            match FaceEmbedding::from_bytes(&bytes) {
                Some(embedding) => {
                    result.push(GalleryEmbedding {
                        cluster_id,
                        face_id,
                        embedding,
                    });
                }
                None => tracing::warn!(
                    "Corrupted gallery embedding for cluster_id={}, face_id={}: {} bytes",
                    cluster_id,
                    face_id,
                    bytes.len()
                ),
            }
        }

        Ok(result)
    }

    pub fn get_cluster_photo_ids(&self) -> SqliteResult<Vec<(i64, i64)>> {
        let mut stmt = self.conn.prepare(
            "SELECT DISTINCT cluster_id, photo_id FROM faces WHERE cluster_id IS NOT NULL",
        )?;

        let rows = stmt.query_map([], |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)))?;
        let mut result = Vec::new();
        for row in rows {
            result.push(row?);
        }
        Ok(result)
    }

    /// Get person names for a photo (faces detected in this photo with cluster names)
    /// Get (cluster_id, display_name) pairs for all people detected in a photo.
    pub fn get_people_for_photo(&self, photo_id: i64) -> SqliteResult<Vec<(i64, String)>> {
        let mut stmt = self.conn.prepare(
            r#"
            SELECT DISTINCT cluster_id, name FROM (
                SELECT fc.id AS cluster_id, COALESCE(fc.name, 'Person ' || fc.id) AS name
                FROM faces f
                JOIN face_clusters fc ON f.cluster_id = fc.id
                WHERE f.photo_id = ?1

                UNION

                SELECT fc.id AS cluster_id, COALESCE(fc.name, 'Person ' || fc.id) AS name
                FROM photo_inferred_identities pii
                JOIN face_clusters fc ON pii.cluster_id = fc.id
                WHERE pii.photo_id = ?1
            )
            ORDER BY name
            "#,
        )?;

        let people = stmt
            .query_map(params![photo_id], |row| {
                Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?))
            })?
            .collect::<SqliteResult<Vec<_>>>()?;

        Ok(people)
    }

    /// Get all photo IDs that contain faces from a given cluster
    pub fn get_photos_for_cluster(&self, cluster_id: i64) -> SqliteResult<Vec<i64>> {
        let mut stmt = self.conn.prepare(
            r#"
            SELECT p.id
            FROM photos p
            JOIN (
                SELECT f.photo_id
                FROM faces f
                WHERE f.cluster_id = ?1
                UNION
                SELECT pii.photo_id
                FROM photo_inferred_identities pii
                WHERE pii.cluster_id = ?1
            ) matches ON matches.photo_id = p.id
            WHERE p.is_trashed = FALSE
            GROUP BY p.id
            ORDER BY p.date_taken IS NULL ASC, p.date_taken DESC, p.id DESC
            "#,
        )?;

        let rows = stmt.query_map(params![cluster_id], |row| row.get(0))?;

        let mut photo_ids = Vec::new();
        for row in rows {
            photo_ids.push(row?);
        }

        Ok(photo_ids)
    }

    /// Get all face IDs with bounding boxes and their photo file paths.
    /// Used for regenerating missing face crops.
    pub fn get_all_faces_with_paths(&self) -> SqliteResult<Vec<FacePathRow>> {
        let mut stmt = self.conn.prepare(
            r#"
            SELECT f.id, p.file_path, COALESCE(p.orientation, 1), f.bbox_x, f.bbox_y, f.bbox_width, f.bbox_height
            FROM faces f
            JOIN photos p ON f.photo_id = p.id
            "#,
        )?;

        let rows = stmt.query_map([], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, i32>(2)?,
                row.get::<_, f32>(3)?,
                row.get::<_, f32>(4)?,
                row.get::<_, f32>(5)?,
                row.get::<_, f32>(6)?,
            ))
        })?;

        let mut result = Vec::new();
        for row in rows {
            result.push(row?);
        }

        Ok(result)
    }

    /// Find contextual candidate identities from nearby photos in the same folder.
    pub fn get_contextual_cluster_candidates(
        &self,
        photo_id: i64,
        folder_prefix_like: Option<&str>,
        target_ts: i64,
        window_secs: i64,
    ) -> SqliteResult<Vec<(i64, i64, i64, String)>> {
        let sql_with_folder = r#"
            SELECT DISTINCT
                p.id,
                f.cluster_id,
                CAST(strftime('%s', p.date_taken) AS INTEGER) AS source_ts,
                p.file_path
            FROM photos p
            JOIN faces f ON f.photo_id = p.id
            WHERE p.id != ?1
                AND p.is_trashed = FALSE
                AND p.date_taken IS NOT NULL
                AND f.cluster_id IS NOT NULL
                AND p.file_path LIKE ?2
                AND ABS(CAST(strftime('%s', p.date_taken) AS INTEGER) - ?3) <= ?4
        "#;

        let sql_without_folder = r#"
            SELECT DISTINCT
                p.id,
                f.cluster_id,
                CAST(strftime('%s', p.date_taken) AS INTEGER) AS source_ts,
                p.file_path
            FROM photos p
            JOIN faces f ON f.photo_id = p.id
            WHERE p.id != ?1
                AND p.is_trashed = FALSE
                AND p.date_taken IS NOT NULL
                AND f.cluster_id IS NOT NULL
                AND ABS(CAST(strftime('%s', p.date_taken) AS INTEGER) - ?2) <= ?3
        "#;

        let mut result = Vec::new();

        if let Some(folder_like) = folder_prefix_like {
            let mut stmt = self.conn.prepare(sql_with_folder)?;
            let rows = stmt.query_map(
                params![photo_id, folder_like, target_ts, window_secs],
                |row| {
                    Ok((
                        row.get::<_, i64>(0)?,
                        row.get::<_, i64>(1)?,
                        row.get::<_, i64>(2)?,
                        row.get::<_, String>(3)?,
                    ))
                },
            )?;
            for row in rows {
                result.push(row?);
            }
        } else {
            let mut stmt = self.conn.prepare(sql_without_folder)?;
            let rows = stmt.query_map(params![photo_id, target_ts, window_secs], |row| {
                Ok((
                    row.get::<_, i64>(0)?,
                    row.get::<_, i64>(1)?,
                    row.get::<_, i64>(2)?,
                    row.get::<_, String>(3)?,
                ))
            })?;
            for row in rows {
                result.push(row?);
            }
        }

        Ok(result)
    }

    /// Get unprocessed photos with optional timestamp for contextual identity linking.
    pub fn get_unprocessed_photos_with_context(&self) -> SqliteResult<Vec<UnprocessedPhotoRow>> {
        let mut stmt = self.conn.prepare(
            r#"
            SELECT
                id,
                file_path,
                COALESCE(orientation, 1) AS orientation,
                CASE
                    WHEN date_taken IS NOT NULL THEN CAST(strftime('%s', date_taken) AS INTEGER)
                    ELSE NULL
                END AS taken_ts,
                file_hash
            FROM photos
            WHERE faces_processed = FALSE AND is_trashed = FALSE
            ORDER BY date_taken DESC
            "#,
        )?;

        let rows = stmt.query_map([], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, i32>(2)?,
                row.get::<_, Option<i64>>(3)?,
                row.get::<_, String>(4)?,
            ))
        })?;

        let mut result = Vec::new();
        for row in rows {
            result.push(row?);
        }
        Ok(result)
    }

    /// Read the cached `photos.brightness` value for a photo. Returns
    /// `None` if the photo doesn't exist OR brightness was never
    /// computed (NULL in the column). Caller is expected to fall back
    /// to on-disk recomputation in that case.
    pub fn get_photo_brightness(&self, photo_id: i64) -> SqliteResult<Option<f32>> {
        self.conn
            .query_row(
                "SELECT brightness FROM photos WHERE id = ?1",
                rusqlite::params![photo_id],
                |row| row.get::<_, Option<f32>>(0),
            )
            .or_else(|e| match e {
                rusqlite::Error::QueryReturnedNoRows => Ok(None),
                other => Err(other),
            })
    }

    /// Count of unresolved entries in the face review queue.
    pub fn review_queue_size(&self) -> SqliteResult<i64> {
        self.conn.query_row(
            "SELECT COUNT(*) FROM face_review_queue WHERE resolved_at IS NULL",
            [],
            |row| row.get(0),
        )
    }

    /// Pull the top-N most informative unresolved review items.
    ///
    /// Ordered by ambiguity ascending (closer top-2 scores first = higher
    /// information gain) then by candidate cluster size descending (resolving
    /// ambiguity on big clusters propagates to more photos).
    pub fn get_review_queue_items(&self, limit: usize) -> SqliteResult<Vec<ReviewItem>> {
        let mut stmt = self.conn.prepare(
            r#"
            SELECT
                q.id, q.face_id,
                q.candidate_cluster_id, c.name, c.face_count,
                q.score
            FROM face_review_queue q
            JOIN faces f ON f.id = q.face_id
            JOIN face_clusters c ON c.id = q.candidate_cluster_id
            WHERE q.resolved_at IS NULL
            ORDER BY COALESCE(q.ambiguity, 1.0) ASC, c.face_count DESC
            LIMIT ?1
            "#,
        )?;

        let rows = stmt.query_map(params![limit as i64], |row| {
            Ok(ReviewItem {
                queue_id: row.get(0)?,
                face_id: row.get(1)?,
                candidate_cluster_id: row.get(2)?,
                candidate_cluster_name: row.get(3)?,
                candidate_cluster_size: row.get(4)?,
                candidate_sample_face_ids: Vec::new(),
                score: row.get(5)?,
            })
        })?;

        let mut items = Vec::new();
        for row in rows {
            items.push(row?);
        }
        drop(stmt);

        // Populate candidate sample face_ids per item (up to 4 best from the cluster).
        let mut sample_stmt = self.conn.prepare(
            r#"
            SELECT id
            FROM faces
            WHERE cluster_id = ?1
            ORDER BY confidence DESC, id ASC
            LIMIT 4
            "#,
        )?;
        for item in items.iter_mut() {
            let ids = sample_stmt.query_map(params![item.candidate_cluster_id], |row| {
                row.get::<_, i64>(0)
            })?;
            let mut collected = Vec::new();
            for id in ids {
                collected.push(id?);
            }
            item.candidate_sample_face_ids = collected;
        }

        Ok(items)
    }

    /// Cluster IDs of the already-assigned faces in the given photo (other
    /// than the face we're resolving). Used as the "context" in co-occurrence.
    pub fn get_photo_other_clusters(
        &self,
        photo_id: i64,
        exclude_face_id: i64,
    ) -> SqliteResult<Vec<i64>> {
        let mut stmt = self.conn.prepare(
            "SELECT DISTINCT cluster_id FROM faces
             WHERE photo_id = ?1 AND id != ?2 AND cluster_id IS NOT NULL",
        )?;
        let rows = stmt.query_map(params![photo_id, exclude_face_id], |row| {
            row.get::<_, i64>(0)
        })?;
        let mut out = Vec::new();
        for row in rows {
            out.push(row?);
        }
        Ok(out)
    }

    /// How many photos contain both cluster_a and cluster_b.
    ///
    /// Used as the co-occurrence strength between two clusters. High values
    /// mean the two people appear together frequently and should bias
    /// retrieval toward whichever matches the rest of the photo's cast.
    pub fn cooccurrence_count(&self, cluster_a: i64, cluster_b: i64) -> SqliteResult<i64> {
        if cluster_a == cluster_b {
            return Ok(0);
        }
        self.conn.query_row(
            r#"
            SELECT COUNT(*) FROM (
                SELECT photo_id FROM faces WHERE cluster_id = ?1
                INTERSECT
                SELECT photo_id FROM faces WHERE cluster_id = ?2
            )
            "#,
            params![cluster_a, cluster_b],
            |row| row.get(0),
        )
    }

    /// Clusters assigned to faces within a time window around `date_taken`,
    /// excluding the photo itself. Returns (cluster_id, delta_seconds).
    ///
    /// Used for temporal-chain propagation: if nearby-in-time photos contain
    /// high-confidence assignments for cluster X, an ambiguous face in our
    /// photo should be biased toward X.
    pub fn temporal_neighbor_clusters(
        &self,
        photo_id: i64,
        window_secs: i64,
    ) -> SqliteResult<Vec<(i64, i64)>> {
        let mut stmt = self.conn.prepare(
            r#"
            WITH base AS (
                SELECT date_taken AS t
                FROM photos
                WHERE id = ?1 AND date_taken IS NOT NULL
            )
            SELECT DISTINCT f.cluster_id,
                CAST(ABS(
                    strftime('%s', p.date_taken) - strftime('%s', (SELECT t FROM base))
                ) AS INTEGER) AS delta_sec
            FROM faces f
            JOIN photos p ON p.id = f.photo_id
            WHERE f.cluster_id IS NOT NULL
              AND p.id != ?1
              AND p.is_trashed = FALSE
              AND p.date_taken IS NOT NULL
              AND ABS(
                  strftime('%s', p.date_taken) - strftime('%s', (SELECT t FROM base))
              ) <= ?2
            "#,
        )?;
        let rows = stmt.query_map(params![photo_id, window_secs], |row| {
            Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?))
        })?;
        let mut out = Vec::new();
        for row in rows {
            out.push(row?);
        }
        Ok(out)
    }

    /// Get faces for a cluster filtered by status.
    pub fn get_faces_by_cluster(
        &self,
        cluster_id: i64,
        status: FaceStatus,
        cursor: Option<i64>,
        limit: usize,
    ) -> SqliteResult<Vec<FaceDetail>> {
        let (where_clause, params_slice): (String, Vec<rusqlite::types::Value>) = match status {
            FaceStatus::Confirmed => (
                "f.cluster_id = ?1 AND f.user_confirmed = 1".to_string(),
                vec![rusqlite::types::Value::from(cluster_id)],
            ),
            FaceStatus::Unconfirmed => (
                "f.cluster_id = ?1 AND f.user_confirmed = 0".to_string(),
                vec![rusqlite::types::Value::from(cluster_id)],
            ),
            FaceStatus::All => (
                "f.cluster_id = ?1".to_string(),
                vec![rusqlite::types::Value::from(cluster_id)],
            ),
        };

        let mut sql = format!(
            "SELECT f.id, f.photo_id, f.cluster_id, f.confidence, f.user_confirmed
             FROM faces f
             WHERE {} ",
            where_clause
        );

        let mut params: Vec<rusqlite::types::Value> = params_slice;
        if let Some(c) = cursor {
            sql.push_str("AND f.id > ?2 ");
            params.push(rusqlite::types::Value::from(c));
            sql.push_str("ORDER BY f.id ASC LIMIT ?3");
        } else {
            sql.push_str("ORDER BY f.id ASC LIMIT ?2");
        }
        params.push(rusqlite::types::Value::from(limit as i64));

        let param_refs: Vec<&dyn rusqlite::types::ToSql> = params
            .iter()
            .map(|v| v as &dyn rusqlite::types::ToSql)
            .collect();

        let mut stmt = self.conn.prepare(&sql)?;
        let rows = stmt.query_map(param_refs.as_slice(), |row| {
            Ok(FaceDetail {
                face_id: row.get(0)?,
                photo_id: row.get(1)?,
                cluster_id: row.get(2)?,
                confidence: row.get(3)?,
                user_confirmed: row.get(4)?,
            })
        })?;

        let mut faces = Vec::new();
        for row in rows {
            faces.push(row?);
        }
        Ok(faces)
    }

    /// Recently detected faces that are not assigned to a person yet.
    /// Used only as a live, in-progress surface while the face pipeline
    /// is still clustering. The final People grid remains cluster-based.
    pub fn get_unclustered_faces(
        &self,
        cursor: Option<i64>,
        limit: usize,
    ) -> SqliteResult<Vec<FaceDetail>> {
        let mut sql = String::from(
            "SELECT f.id, f.photo_id, f.cluster_id, f.confidence, f.user_confirmed
             FROM faces f
             JOIN photos p ON p.id = f.photo_id
             WHERE f.cluster_id IS NULL
               AND p.is_trashed = FALSE ",
        );
        let mut params: Vec<rusqlite::types::Value> = Vec::new();
        if let Some(c) = cursor {
            sql.push_str("AND f.id > ?1 ORDER BY f.id ASC LIMIT ?2");
            params.push(rusqlite::types::Value::from(c));
            params.push(rusqlite::types::Value::from(limit as i64));
        } else {
            sql.push_str("ORDER BY f.id ASC LIMIT ?1");
            params.push(rusqlite::types::Value::from(limit as i64));
        }
        let param_refs: Vec<&dyn rusqlite::types::ToSql> = params
            .iter()
            .map(|v| v as &dyn rusqlite::types::ToSql)
            .collect();

        let mut stmt = self.conn.prepare(&sql)?;
        let rows = stmt.query_map(param_refs.as_slice(), |row| {
            Ok(FaceDetail {
                face_id: row.get(0)?,
                photo_id: row.get(1)?,
                cluster_id: row.get(2)?,
                confidence: row.get(3)?,
                user_confirmed: row.get(4)?,
            })
        })?;

        let mut faces = Vec::new();
        for row in rows {
            faces.push(row?);
        }
        Ok(faces)
    }

    /// Count unconfirmed faces in a cluster.
    pub fn count_unconfirmed_in_cluster(&self, cluster_id: i64) -> SqliteResult<i64> {
        self.conn.query_row(
            "SELECT COUNT(*) FROM faces WHERE cluster_id = ?1 AND user_confirmed = 0",
            params![cluster_id],
            |row| row.get(0),
        )
    }

    /// Count unconfirmed faces across all clusters.
    /// Returns (total_unconfirmed, clusters_with_unconfirmed).
    pub fn count_unconfirmed_global(&self) -> SqliteResult<(i64, i64)> {
        let total: i64 = self.conn.query_row(
            "SELECT COUNT(*) FROM faces WHERE cluster_id IS NOT NULL AND user_confirmed = 0",
            [],
            |row| row.get(0),
        )?;
        let cluster_count: i64 = self.conn.query_row(
            "SELECT COUNT(DISTINCT cluster_id) FROM faces WHERE cluster_id IS NOT NULL AND user_confirmed = 0",
            [],
            |row| row.get(0),
        )?;
        Ok((total, cluster_count))
    }

    /// Return the next batch of unconfirmed faces from the cluster with
    /// the most pending review work. This keeps the review UI O(1)
    /// instead of probing every person until it finds one with faces.
    pub fn next_unconfirmed_face_batch(&self, limit: usize) -> SqliteResult<Vec<FaceDetail>> {
        self.next_unconfirmed_face_batch_excluding(limit, &[])
    }

    /// Return the next review batch while ignoring faces the current UI
    /// session has skipped. Skips are intentionally not persisted as a
    /// reject/confirm decision, but they still need to be excluded from
    /// the server-side cluster selection or the reviewer can be sent
    /// straight back to the same skipped-only cluster.
    pub fn next_unconfirmed_face_batch_excluding(
        &self,
        limit: usize,
        excluded_face_ids: &[i64],
    ) -> SqliteResult<Vec<FaceDetail>> {
        if excluded_face_ids.is_empty() {
            return self.next_unconfirmed_face_batch_without_exclusions(limit);
        }

        let placeholders = vec!["?"; excluded_face_ids.len()].join(", ");
        let sql = format!(
            r#"
            WITH next_cluster AS (
                SELECT cluster_id
                FROM faces
                WHERE cluster_id IS NOT NULL
                  AND user_confirmed = 0
                  AND id NOT IN ({placeholders})
                GROUP BY cluster_id
                ORDER BY COUNT(*) DESC, MIN(id) ASC
                LIMIT 1
            )
            SELECT f.id, f.photo_id, f.cluster_id, f.confidence, f.user_confirmed
            FROM faces f
            JOIN next_cluster n ON n.cluster_id = f.cluster_id
            WHERE f.user_confirmed = 0
              AND f.id NOT IN ({placeholders})
            ORDER BY f.id ASC
            LIMIT ?
            "#
        );
        let mut args = Vec::with_capacity(excluded_face_ids.len() * 2 + 1);
        args.extend_from_slice(excluded_face_ids);
        args.extend_from_slice(excluded_face_ids);
        args.push(limit as i64);

        let mut stmt = self.conn.prepare(&sql)?;
        let rows = stmt.query_map(params_from_iter(args), |row| {
            Ok(FaceDetail {
                face_id: row.get(0)?,
                photo_id: row.get(1)?,
                cluster_id: row.get(2)?,
                confidence: row.get(3)?,
                user_confirmed: row.get(4)?,
            })
        })?;

        let mut faces = Vec::new();
        for row in rows {
            faces.push(row?);
        }
        Ok(faces)
    }

    fn next_unconfirmed_face_batch_without_exclusions(
        &self,
        limit: usize,
    ) -> SqliteResult<Vec<FaceDetail>> {
        let mut stmt = self.conn.prepare(
            r#"
            WITH next_cluster AS (
                SELECT cluster_id
                FROM faces
                WHERE cluster_id IS NOT NULL
                  AND user_confirmed = 0
                GROUP BY cluster_id
                ORDER BY COUNT(*) DESC, MIN(id) ASC
                LIMIT 1
            )
            SELECT f.id, f.photo_id, f.cluster_id, f.confidence, f.user_confirmed
            FROM faces f
            JOIN next_cluster n ON n.cluster_id = f.cluster_id
            WHERE f.user_confirmed = 0
            ORDER BY f.id ASC
            LIMIT ?1
            "#,
        )?;
        let rows = stmt.query_map(params![limit as i64], |row| {
            Ok(FaceDetail {
                face_id: row.get(0)?,
                photo_id: row.get(1)?,
                cluster_id: row.get(2)?,
                confidence: row.get(3)?,
                user_confirmed: row.get(4)?,
            })
        })?;

        let mut faces = Vec::new();
        for row in rows {
            faces.push(row?);
        }
        Ok(faces)
    }

    /// Get negatives for a face (clusters it should NOT be in).
    pub fn get_negatives_for_face(&self, face_id: i64) -> SqliteResult<Vec<i64>> {
        let mut stmt = self
            .conn
            .prepare("SELECT not_cluster_id FROM face_negatives WHERE face_id = ?1")?;
        let rows = stmt.query_map(params![face_id], |row| row.get::<_, i64>(0))?;
        let mut out = Vec::new();
        for row in rows {
            out.push(row?);
        }
        Ok(out)
    }

    /// K-similar: find K nearest unassigned faces to a cluster's confirmed gallery centroid.
    pub fn k_similar_to_cluster(&self, cluster_id: i64, k: usize) -> SqliteResult<Vec<(i64, f32)>> {
        if k == 0 {
            return Ok(Vec::new());
        }

        // Gather confirmed gallery embeddings for centroid.
        let mut stmt = self.conn.prepare(
            "SELECT embedding FROM person_gallery_embeddings WHERE cluster_id = ?1 AND source = 'user_confirmed'",
        )?;
        let rows = stmt.query_map(params![cluster_id], |row| row.get::<_, Vec<u8>>(0))?;
        let mut embeddings: Vec<FaceEmbedding> = Vec::new();
        for row in rows {
            let bytes = row?;
            if let Some(emb) = FaceEmbedding::from_bytes(&bytes) {
                embeddings.push(emb);
            }
        }

        // If no user-confirmed gallery, fall back to auto gallery.
        if embeddings.is_empty() {
            let mut stmt2 = self
                .conn
                .prepare("SELECT embedding FROM person_gallery_embeddings WHERE cluster_id = ?1")?;
            let rows2 = stmt2.query_map(params![cluster_id], |row| row.get::<_, Vec<u8>>(0))?;
            for row in rows2 {
                let bytes = row?;
                if let Some(emb) = FaceEmbedding::from_bytes(&bytes) {
                    embeddings.push(emb);
                }
            }
        }

        if embeddings.is_empty() {
            return Ok(Vec::new());
        }

        // Compute centroid (mean of all gallery vectors).
        let dim = embeddings[0].vector.len();
        let mut centroid = ndarray::Array1::<f32>::zeros(dim);
        for emb in &embeddings {
            centroid += &emb.vector;
        }
        centroid /= embeddings.len() as f32;

        // Normalize the centroid.
        let norm = centroid.dot(&centroid).sqrt();
        if norm > 0.0 {
            centroid /= norm;
        }
        let centroid_emb = FaceEmbedding::new(centroid);

        // Load candidate faces: unassigned (cluster_id IS NULL OR user_confirmed = 0),
        // excluding faces with a negative against this cluster.
        let mut cand_stmt = self.conn.prepare(
            r#"
            SELECT f.id, f.embedding
            FROM faces f
            JOIN photos p ON p.id = f.photo_id
            WHERE (f.cluster_id IS NULL OR (f.cluster_id IS NOT NULL AND f.user_confirmed = 0))
              AND f.user_confirmed >= 0
              AND p.is_trashed = FALSE
              AND f.id NOT IN (
                  SELECT face_id FROM face_negatives WHERE not_cluster_id = ?1
              )
            "#,
        )?;
        let cand_rows = cand_stmt.query_map(params![cluster_id], |row| {
            Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
        })?;

        let mut scored: Vec<(i64, f32)> = Vec::new();
        for row in cand_rows {
            let (face_id, bytes) = row?;
            if let Some(emb) = FaceEmbedding::from_bytes(&bytes) {
                let sim = centroid_emb.cosine_similarity(&emb);
                scored.push((face_id, sim));
            }
        }

        // Sort descending by similarity, take top K.
        scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
        scored.truncate(k);

        Ok(scored)
    }
}

#[cfg(test)]
mod tests {
    use rusqlite::{params, Connection};

    use super::*;
    use crate::db::create_schema;
    use ndarray::Array1;

    fn embedding(seed: f32) -> Vec<u8> {
        let mut values = vec![0.0; 512];
        values[0] = seed;
        values[1] = 1.0 - seed;
        FaceEmbedding::new(Array1::from_vec(values)).to_bytes()
    }

    fn seeded_conn() -> Connection {
        let conn = Connection::open_in_memory().unwrap();
        create_schema(&conn).unwrap();
        conn.execute(
            "INSERT INTO face_clusters (id, name, face_count, photo_count) VALUES (10, 'Person', 3, 3)",
            [],
        )
        .unwrap();

        for id in 1..=3 {
            conn.execute(
                "INSERT INTO photos (id, file_path, file_name, file_hash, file_size)
                 VALUES (?1, ?2, ?3, ?4, 100)",
                params![
                    id,
                    format!("photos/{id}.jpg"),
                    format!("{id}.jpg"),
                    format!("hash-{id}")
                ],
            )
            .unwrap();
            conn.execute(
                "INSERT INTO faces (
                    id, photo_id, bbox_x, bbox_y, bbox_width, bbox_height,
                    embedding, cluster_id, confidence, user_confirmed
                 )
                 VALUES (?1, ?2, 0.1, 0.1, 0.2, 0.2, ?3, 10, 0.99, ?4)",
                params![
                    id,
                    id,
                    embedding(id as f32 / 10.0),
                    if id == 3 { 1 } else { 0 }
                ],
            )
            .unwrap();
        }
        conn
    }

    fn insert_cluster(conn: &Connection, id: i64, name: &str) {
        conn.execute(
            "INSERT INTO face_clusters (id, name, face_count, photo_count) VALUES (?1, ?2, 0, 0)",
            params![id, name],
        )
        .unwrap();
    }

    fn insert_face(conn: &Connection, id: i64, cluster_id: i64, confirmed: i32) {
        conn.execute(
            "INSERT INTO photos (id, file_path, file_name, file_hash, file_size)
             VALUES (?1, ?2, ?3, ?4, 100)",
            params![
                id + 100,
                format!("photos/review-{id}.jpg"),
                format!("review-{id}.jpg"),
                format!("review-hash-{id}")
            ],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO faces (
                id, photo_id, bbox_x, bbox_y, bbox_width, bbox_height,
                embedding, cluster_id, confidence, user_confirmed
             )
             VALUES (?1, ?2, 0.1, 0.1, 0.2, 0.2, ?3, ?4, 0.99, ?5)",
            params![
                id,
                id + 100,
                embedding(id as f32 / 10.0),
                cluster_id,
                confirmed
            ],
        )
        .unwrap();
    }

    #[test]
    fn get_faces_by_cluster_first_page_uses_valid_limit_parameter() {
        let conn = seeded_conn();
        let repo = FaceRepo::new(&conn);

        let faces = repo
            .get_faces_by_cluster(10, FaceStatus::Unconfirmed, None, 10)
            .unwrap();

        assert_eq!(
            faces.iter().map(|f| f.face_id).collect::<Vec<_>>(),
            vec![1, 2]
        );
    }

    #[test]
    fn get_faces_by_cluster_cursor_uses_valid_limit_parameter() {
        let conn = seeded_conn();
        let repo = FaceRepo::new(&conn);

        let faces = repo
            .get_faces_by_cluster(10, FaceStatus::All, Some(1), 10)
            .unwrap();

        assert_eq!(
            faces.iter().map(|f| f.face_id).collect::<Vec<_>>(),
            vec![2, 3]
        );
    }

    #[test]
    fn next_unconfirmed_face_batch_uses_the_cluster_with_most_pending_faces() {
        let conn = Connection::open_in_memory().unwrap();
        create_schema(&conn).unwrap();
        insert_cluster(&conn, 10, "Small");
        insert_cluster(&conn, 20, "Large");
        insert_face(&conn, 1, 10, 0);
        insert_face(&conn, 2, 20, 0);
        insert_face(&conn, 3, 20, 0);
        insert_face(&conn, 4, 20, 1);
        let repo = FaceRepo::new(&conn);

        let faces = repo.next_unconfirmed_face_batch(10).unwrap();

        assert_eq!(
            faces.iter().map(|f| f.face_id).collect::<Vec<_>>(),
            vec![2, 3]
        );
        assert!(faces.iter().all(|f| f.cluster_id == Some(20)));
    }

    #[test]
    fn next_unconfirmed_face_batch_excluding_skipped_faces_moves_to_next_cluster() {
        let conn = Connection::open_in_memory().unwrap();
        create_schema(&conn).unwrap();
        insert_cluster(&conn, 10, "Small");
        insert_cluster(&conn, 20, "Large");
        insert_face(&conn, 1, 10, 0);
        insert_face(&conn, 2, 20, 0);
        insert_face(&conn, 3, 20, 0);
        let repo = FaceRepo::new(&conn);

        let faces = repo
            .next_unconfirmed_face_batch_excluding(10, &[2, 3])
            .unwrap();

        assert_eq!(faces.iter().map(|f| f.face_id).collect::<Vec<_>>(), vec![1]);
        assert!(faces.iter().all(|f| f.cluster_id == Some(10)));
    }

    #[test]
    fn unclustered_embedding_candidates_skip_trashed_and_hidden_faces() {
        let conn = Connection::open_in_memory().unwrap();
        create_schema(&conn).unwrap();
        for id in 1..=3 {
            conn.execute(
                "INSERT INTO photos (id, file_path, file_name, file_hash, file_size, is_trashed)
                 VALUES (?1, ?2, ?3, ?4, 100, ?5)",
                params![
                    id,
                    format!("photos/{id}.jpg"),
                    format!("{id}.jpg"),
                    format!("hash-{id}"),
                    if id == 2 { 1 } else { 0 }
                ],
            )
            .unwrap();
        }
        conn.execute(
            "INSERT INTO faces (
                id, photo_id, bbox_x, bbox_y, bbox_width, bbox_height,
                embedding, cluster_id, confidence, user_confirmed
             )
             VALUES
                (1, 1, 0.1, 0.1, 0.2, 0.2, ?1, NULL, 0.9, 0),
                (2, 2, 0.1, 0.1, 0.2, 0.2, ?2, NULL, 0.9, 0),
                (3, 3, 0.1, 0.1, 0.2, 0.2, ?3, NULL, 0.9, -1)",
            params![embedding(0.1), embedding(0.2), embedding(0.3)],
        )
        .unwrap();

        let faces = FaceRepo::new(&conn)
            .get_unclustered_faces_with_photo_embeddings()
            .unwrap();

        assert_eq!(
            faces.iter().map(|(id, _, _)| *id).collect::<Vec<_>>(),
            vec![1]
        );
    }

    #[test]
    fn k_similar_skips_trashed_candidates_and_zero_k_short_circuits() {
        let conn = Connection::open_in_memory().unwrap();
        create_schema(&conn).unwrap();
        insert_cluster(&conn, 10, "Person");
        for id in 1..=3 {
            conn.execute(
                "INSERT INTO photos (id, file_path, file_name, file_hash, file_size, is_trashed)
                 VALUES (?1, ?2, ?3, ?4, 100, ?5)",
                params![
                    id,
                    format!("photos/{id}.jpg"),
                    format!("{id}.jpg"),
                    format!("hash-{id}"),
                    if id == 3 { 1 } else { 0 }
                ],
            )
            .unwrap();
        }
        conn.execute(
            "INSERT INTO faces (
                id, photo_id, bbox_x, bbox_y, bbox_width, bbox_height,
                embedding, cluster_id, confidence, user_confirmed
             )
             VALUES
                (1, 1, 0.1, 0.1, 0.2, 0.2, ?1, 10, 0.9, 1),
                (2, 2, 0.1, 0.1, 0.2, 0.2, ?2, NULL, 0.9, 0),
                (3, 3, 0.1, 0.1, 0.2, 0.2, ?3, NULL, 0.9, 0)",
            params![embedding(0.9), embedding(0.8), embedding(0.9)],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO person_gallery_embeddings (cluster_id, face_id, embedding, quality_score, source)
             VALUES (10, 1, ?1, 0.9, 'user_confirmed')",
            params![embedding(0.9)],
        )
        .unwrap();
        let repo = FaceRepo::new(&conn);

        assert!(repo.k_similar_to_cluster(10, 0).unwrap().is_empty());
        let similar = repo.k_similar_to_cluster(10, 10).unwrap();

        assert_eq!(
            similar.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
            vec![2]
        );
    }
}