ImageApi/src/clip_search.rs

//! `/photos/search?q=<text>` — CLIP semantic photo search.
//!
//! The route lives outside `files.rs` to keep that 1500+ line module
//! focused on EXIF / tag listing. The flow is:
//!
//! 1. Parse query params (`q`, `limit`, `threshold`, optional `library`).
//! 2. Call Apollo's `/api/internal/clip/encode_text` to get the query
//!    vector (L2-normalized 768-d f32 for ViT-L/14).
//! 3. Load every `(content_hash, clip_embedding)` for the scope from
//!    `image_exif` via `ExifDao::list_clip_index`. ~28–43 MB for a 14k
//!    library at ViT-L/14; loaded fresh per request — fast enough for
//!    v1, optimize via an AppState cache later if needed.
//! 4. Dot product (= cosine since both sides are L2-normalized), filter
//!    above `threshold`, top-K by score.
//! 5. Resolve each surviving hash back to a `(library_id, rel_path)` so
//!    the frontend can render the photo / hand off to the carousel.
//!
//! Response shape is intentionally minimal — paths + score — so the
//! frontend can reuse existing PhotoGrid rendering by joining against
//! `/api/photos/match` (or calling `/image/metadata` lazily). Don't
//! bake camera/EXIF metadata into this route; it would force a fan-out
//! per result and balloon the response.

use crate::AppState;
use crate::ai::clip_client::ClipError;
use crate::database::ExifDao;
use actix_web::{HttpResponse, Result as ActixResult, web};
use base64::Engine;
use serde::{Deserialize, Serialize};
use std::sync::Mutex;

#[derive(Debug, Deserialize)]
pub struct SearchQuery {
    /// Natural-language query. Required; empty triggers 400.
    pub q: String,
    /// Max results to return in this page. Capped to 200 server-side.
    /// Defaults to 20. Pair with `offset` for pagination.
    #[serde(default = "default_limit")]
    pub limit: usize,
    /// Zero-based offset into the sorted-and-filtered result set. The
    /// scoring loop still runs over the full embedding matrix on every
    /// page (cheap at personal-library scale — sub-100ms — and avoids
    /// stateful pagination cursors). Defaults to 0.
    #[serde(default)]
    pub offset: usize,
    /// Cosine-similarity floor below which results are dropped.
    /// 0.20 is the rough "this is plausibly relevant" line for OpenAI
    /// CLIP; tunable per call when sweeping. Defaults to 0.20.
    #[serde(default = "default_threshold")]
    pub threshold: f32,
    /// Optional single-library scope. Legacy param — new clients pass
    /// `library_ids` instead so multi-select scopes (Apollo's HUD library
    /// chips, FileViewer-React's library picker) actually filter. Kept
    /// for back-compat; `library_ids` wins when both are supplied.
    pub library: Option<i32>,
    /// Optional multi-library scope, comma-separated id list
    /// (`?library_ids=1,3`). Empty / omitted = every enabled library
    /// (the historical default). Apollo and FileViewer-React both send
    /// this when 2+ libraries are selected; the single-library case
    /// works through either param interchangeably.
    pub library_ids: Option<String>,
    /// Optional model-version filter. Defaults to the live engine's
    /// version (queried lazily). Forces a strict join so mid-flight
    /// model swaps can't mix geometries in a single response.
    #[serde(default)]
    pub model_version: Option<String>,
}

fn default_limit() -> usize {
    20
}

fn default_threshold() -> f32 {
    0.20
}

#[derive(Debug, Serialize)]
pub struct SearchHit {
    pub library_id: i32,
    pub rel_path: String,
    pub content_hash: String,
    /// Cosine similarity in [-1, 1]. In practice OpenAI CLIP returns
    /// 0.10–0.40 for the typical photo library.
    pub score: f32,
}

#[derive(Debug, Serialize)]
pub struct SearchResponse {
    pub query: String,
    pub model_version: String,
    pub threshold: f32,
    /// Total embeddings scored (= every photo in scope with a stored
    /// embedding). Same value across pages of the same query.
    pub considered: usize,
    /// Count of results above threshold, before pagination. Lets the
    /// client decide whether a "Load more" button is meaningful and
    /// stop fetching when ``offset + results.len() >= total_matching``.
    pub total_matching: usize,
    pub offset: usize,
    pub results: Vec<SearchHit>,
}

#[derive(Debug, Serialize)]
struct SearchError {
    error: String,
}

/// Decode a stored `clip_embedding` BLOB back into a `Vec<f32>`. Returns
/// `None` on malformed bytes — those rows get skipped rather than
/// failing the whole query.
fn decode_embedding(bytes: &[u8]) -> Option<Vec<f32>> {
    if bytes.is_empty() || !bytes.len().is_multiple_of(4) {
        return None;
    }
    let mut out = Vec::with_capacity(bytes.len() / 4);
    for chunk in bytes.chunks_exact(4) {
        out.push(f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]));
    }
    Some(out)
}

#[inline]
fn dot(a: &[f32], b: &[f32]) -> f32 {
    a.iter().zip(b.iter()).map(|(x, y)| x * y).sum()
}

/// Failure modes of [`score_photos`]. Carries enough to let each caller pick
/// an appropriate HTTP status (the CLIP service being down is a 502, a
/// disabled feature is a 503, a rejected query is a 400, a DB failure 500).
pub enum ScoreError {
    /// CLIP search isn't configured at all (no Apollo endpoint).
    Disabled,
    /// The query was rejected by the encoder (client error).
    Rejected(String),
    /// The CLIP service is transiently unavailable (upstream error).
    Unavailable(String),
    /// The encoder returned an embedding we couldn't decode.
    MalformedEmbedding,
    /// A database / index load failure.
    Internal(String),
}

/// Result of scoring the whole library against a query embedding: the
/// resolved model version, how many embeddings were considered, and every
/// `(score, content_hash)` above threshold, sorted by descending score.
/// Pagination and path resolution are the caller's job (see [`resolve_hits`])
/// so this core can be reused for both the plain search endpoint and the
/// unified endpoint (which filters by hash before paginating).
pub struct ScoredPhotos {
    pub model_version: String,
    pub considered: usize,
    /// `(cosine_score, content_hash)` pairs, descending by score.
    pub hits: Vec<(f32, String)>,
}

/// Encode `q_text` via CLIP and score it against every stored embedding in
/// the given library scope. Returns all matches above `threshold`, sorted by
/// descending similarity. Pure of HTTP concerns so it's shared by
/// `search_photos` and the unified search endpoint.
pub async fn score_photos(
    state: &AppState,
    exif_dao: &Mutex<Box<dyn ExifDao>>,
    q_text: &str,
    library_ids: &[i32],
    threshold: f32,
    model_version: Option<&str>,
) -> Result<ScoredPhotos, ScoreError> {
    if !state.clip_client.is_enabled() {
        return Err(ScoreError::Disabled);
    }

    // 1. Encode the query text. Fast — Apollo's text encoder is ~50ms on CPU.
    let query_resp = match state.clip_client.encode_text(q_text).await {
        Ok(r) => r,
        Err(ClipError::Permanent(e)) => return Err(ScoreError::Rejected(e.to_string())),
        Err(ClipError::Transient(e)) => return Err(ScoreError::Unavailable(e.to_string())),
        Err(ClipError::Disabled) => return Err(ScoreError::Disabled),
    };
    // decode_embedding works on raw bytes; the wire format is b64.
    let query_bytes = base64::engine::general_purpose::STANDARD
        .decode(query_resp.embedding.as_bytes())
        .unwrap_or_default();
    let query_vec = decode_embedding(&query_bytes).ok_or(ScoreError::MalformedEmbedding)?;

    // 2. Pull the (hash, embedding) matrix under the dao lock, release
    // before scoring. The caller-supplied `model_version` (or the live
    // engine's) forces a strict join so a mid-flight model swap can't mix
    // geometries.
    let ctx = opentelemetry::Context::current();
    let rows: Vec<(String, Vec<u8>)> = {
        let mut dao = exif_dao.lock().expect("exif dao");
        dao.list_clip_index(
            &ctx,
            library_ids,
            model_version.or(Some(&query_resp.model_version)),
        )
        .map_err(|e| {
            log::warn!("clip_search: list_clip_index failed: {:?}", e);
            ScoreError::Internal("failed to load search index".into())
        })?
    };
    let considered = rows.len();

    // 3. Score. Keep all matches and sort at the end (~microseconds at 14k).
    let mut hits: Vec<(f32, String)> = Vec::with_capacity(considered);
    for (hash, blob) in rows {
        let Some(emb) = decode_embedding(&blob) else {
            continue;
        };
        if emb.len() != query_vec.len() {
            continue;
        }
        let sim = dot(&emb, &query_vec);
        if sim < threshold {
            continue;
        }
        hits.push((sim, hash));
    }
    hits.sort_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(std::cmp::Ordering::Equal));

    Ok(ScoredPhotos {
        model_version: query_resp.model_version,
        considered,
        hits,
    })
}

/// Resolve a page of `(score, content_hash)` pairs back to [`SearchHit`]s
/// (each carrying `library_id` + `rel_path`). Hashes that no longer resolve
/// to a row are skipped. Shared by both endpoints.
pub fn resolve_hits(
    exif_dao: &Mutex<Box<dyn ExifDao>>,
    scored: &[(f32, String)],
) -> Vec<SearchHit> {
    if scored.is_empty() {
        return Vec::new();
    }
    let ctx = opentelemetry::Context::current();
    let hashes: Vec<String> = scored.iter().map(|(_, h)| h.clone()).collect();
    let mut dao = exif_dao.lock().expect("exif dao");
    let path_map = dao
        .get_rel_paths_for_hashes(&ctx, &hashes)
        .unwrap_or_else(|e| {
            log::warn!("clip_search: get_rel_paths_for_hashes failed: {:?}", e);
            std::collections::HashMap::new()
        });

    let mut results = Vec::with_capacity(scored.len());
    for (score, hash) in scored {
        let row = match dao.find_by_content_hash(&ctx, hash) {
            Ok(Some(r)) => r,
            Ok(None) => continue,
            Err(e) => {
                log::warn!("clip_search: find_by_content_hash failed for {hash}: {e:?}");
                continue;
            }
        };
        // Prefer get_rel_paths_for_hashes's first entry (shares image_exif's
        // natural order), falling back to the ImageExif row.
        let rel_path = path_map
            .get(hash)
            .and_then(|paths| paths.first().cloned())
            .unwrap_or(row.file_path);
        results.push(SearchHit {
            library_id: row.library_id,
            rel_path,
            content_hash: hash.clone(),
            score: *score,
        });
    }
    results
}

/// Parse the `library_ids` (multi) / `library` (single) scope params into a
/// deduped id list. Empty = "every enabled library". Shared so the unified
/// endpoint scopes CLIP identically.
pub fn parse_library_scope(
    library_ids: Option<&str>,
    library: Option<i32>,
) -> Result<Vec<i32>, String> {
    if let Some(raw) = library_ids {
        let mut out: Vec<i32> = Vec::new();
        for piece in raw.split(',') {
            let trimmed = piece.trim();
            if trimmed.is_empty() {
                continue;
            }
            match trimmed.parse::<i32>() {
                Ok(id) => {
                    if !out.contains(&id) {
                        out.push(id);
                    }
                }
                Err(_) => return Err(format!("invalid library_ids entry: {trimmed:?}")),
            }
        }
        Ok(out)
    } else if let Some(id) = library {
        Ok(vec![id])
    } else {
        Ok(Vec::new())
    }
}

pub async fn search_photos(
    state: web::Data<AppState>,
    exif_dao: web::Data<Mutex<Box<dyn ExifDao>>>,
    query: web::Query<SearchQuery>,
) -> ActixResult<HttpResponse> {
    let q_text = query.q.trim().to_string();
    if q_text.is_empty() {
        return Ok(HttpResponse::BadRequest().json(SearchError {
            error: "query parameter `q` is required".into(),
        }));
    }

    let limit = query.limit.clamp(1, 200);
    let offset = query.offset;
    let threshold = query.threshold.clamp(-1.0, 1.0);

    let library_ids = match parse_library_scope(query.library_ids.as_deref(), query.library) {
        Ok(ids) => ids,
        Err(msg) => return Ok(HttpResponse::BadRequest().json(SearchError { error: msg })),
    };

    let scored = match score_photos(
        &state,
        &exif_dao,
        &q_text,
        &library_ids,
        threshold,
        query.model_version.as_deref(),
    )
    .await
    {
        Ok(s) => s,
        Err(e) => return Ok(score_error_response(e)),
    };

    let total_matching = scored.hits.len();
    // Pagination — slice the sorted list at `[offset, offset+limit)`. Offsets
    // past the end produce empty pages so "load more" stops naturally.
    let page: Vec<(f32, String)> = if offset >= total_matching {
        Vec::new()
    } else {
        let end = (offset + limit).min(total_matching);
        scored.hits[offset..end].to_vec()
    };
    let results = resolve_hits(&exif_dao, &page);

    Ok(HttpResponse::Ok().json(SearchResponse {
        query: q_text,
        model_version: scored.model_version,
        threshold,
        considered: scored.considered,
        total_matching,
        offset,
        results,
    }))
}

/// Map a [`ScoreError`] to the HTTP response `search_photos` historically
/// returned for each failure mode. Reused by the unified endpoint.
pub fn score_error_response(e: ScoreError) -> HttpResponse {
    match e {
        ScoreError::Disabled => HttpResponse::ServiceUnavailable().json(SearchError {
            error: "CLIP search is disabled (no Apollo CLIP endpoint configured)".into(),
        }),
        ScoreError::Rejected(msg) => HttpResponse::BadRequest().json(SearchError {
            error: format!("query rejected: {msg}"),
        }),
        ScoreError::Unavailable(msg) => HttpResponse::BadGateway().json(SearchError {
            error: format!("CLIP service unavailable: {msg}"),
        }),
        ScoreError::MalformedEmbedding => HttpResponse::BadGateway().json(SearchError {
            error: "CLIP service returned a malformed query embedding".into(),
        }),
        ScoreError::Internal(msg) => {
            HttpResponse::InternalServerError().json(SearchError { error: msg })
        }
    }
}