Add embedded Tantivy full-text content search

/api/search now finds files by CONTENT as well as by name: BM25-ranked
matches over extracted text (PDF, Office OOXML/ODF, plain text/code)
with typo-tolerant fuzzy terms and search-as-you-type prefix matching,
served from an embedded Tantivy index at {storage}/.search-index.

Pipeline (all off the request path, mirroring tree-etag + thumbnails):
- statement triggers on storage.files append to a durable dirty queue
  (storage.search_index_dirty) - every write surface (REST, WebDAV,
  NextCloud, WOPI, trash) is covered, crash-safe by construction
- ContentIndexWorker drains the queue on the maintenance pool, extracts
  text once per unique BLAKE3 blob (storage.blob_extracted_text cache:
  N copies = 1 extraction, renames/moves = 0 re-extraction) and applies
  batched single-writer Tantivy commits; queue rows are deleted only
  after the commit succeeds (at-least-once, idempotent upserts)
- the index is a derived artifact: a version-marker mismatch wipes and
  reseeds it from Postgres, which remains the single source of truth

SearchService merges content hits into the existing name search: hits
are hydrated through ONE SQL round-trip that re-applies user scope,
trash state and every active filter (a stale index id can never leak),
scored below name matches, and returned with a plain-text snippet and
a match_source field. Index failure or
OXICLOUD_ENABLE_CONTENT_SEARCH=false degrades to name-only search; a
discard-only janitor keeps the trigger-fed queue bounded while disabled.

The frontend renders the snippet under the file name in list view.

New dependencies: tantivy 0.26, zip 8.6 (deflate only), pdf-extract 0.10.

https://claude.ai/code/session_01Sc7F4xbo83YbFAQ4xEeDrX

src/infrastructure/services/search_index/content_index_worker.rs

@@ -0,0 +1,415 @@
+//! Background drainer for `storage.search_index_dirty` — the asynchronous
+//! half of content indexing (see migration `20260701000000_content_search_index`).
+//!
+//! The statement triggers on `storage.files` only append "index me" requests
+//! to the queue, taking zero locks on user write paths. This worker turns the
+//! requests into Tantivy mutations: every `interval_ms` it drains a batch,
+//! re-reads the CURRENT file state (the queue row is a hint, not a payload),
+//! extracts text once per unique blob, applies one batched Tantivy commit and
+//! only then deletes the processed queue rows.
+//!
+//! Correctness invariants:
+//!   * At-least-once: queue rows are deleted AFTER the Tantivy commit. A
+//!     crash in between re-processes the batch — harmless, upserts are
+//!     idempotent (delete_term + add_document keyed by file_id).
+//!   * Deletes are selected by exact id (`id = ANY(...)`), never by range —
+//!     a transaction that began before our SELECT can commit a smaller id
+//!     afterwards, and a range delete would discard it unprocessed.
+//!   * Latest-op-wins per file within a batch; the authoritative state is
+//!     re-fetched from `storage.files` at drain time anyway (a file trashed
+//!     after its 'upsert' was queued simply turns into a delete).
+//!   * Extraction is keyed by blob hash (content-addressed): N files sharing
+//!     a blob cost ONE extraction, renames/moves cost zero re-extraction.
+//!     Terminal outcomes (ok/empty/failed/too_large) are cached in
+//!     `storage.blob_extracted_text`; transient blob-read errors store
+//!     nothing so the next event retries.
+//!
+//! Resource budget: single worker, one extraction at a time inside
+//! `spawn_blocking`, single-threaded Tantivy writer — the pipeline trickles
+//! along on the maintenance pool and never competes with request latency.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+use sqlx::PgPool;
+use tracing::{debug, error, info, instrument, warn};
+use uuid::Uuid;
+
+use crate::infrastructure::services::dedup_service::DedupService;
+use crate::infrastructure::services::search_index::tantivy_content_index::{
+    EXTRACTOR_VERSION, IndexDocRecord, TantivyContentIndex,
+};
+use crate::infrastructure::services::search_index::text_extractor::{self, ExtractedText};
+
+/// Queue rows drained per batch. Each row may cost a blob read + extraction,
+/// so this is far smaller than the tree-etag drain batch.
+const DRAIN_BATCH: i64 = 256;
+
+/// Max batches per tick so a huge backlog (initial reseed) cannot monopolise
+/// the maintenance connection within one tick.
+const MAX_BATCHES_PER_TICK: u32 = 4;
+
+/// Stored preview head per document (snippet source).
+const PREVIEW_BYTES: usize = 16 * 1024;
+
+/// Ticks between `blob_extracted_text` orphan sweeps (~1 h at the default
+/// 1.5 s interval).
+const ORPHAN_SWEEP_TICKS: u64 = 2400;
+
+pub struct ContentIndexWorker {
+    maintenance_pool: Arc<PgPool>,
+    dedup: Arc<DedupService>,
+    index: Arc<TantivyContentIndex>,
+    interval_ms: u64,
+    max_extract_file_bytes: u64,
+    max_text_bytes: usize,
+}
+
+impl ContentIndexWorker {
+    pub fn new(
+        maintenance_pool: Arc<PgPool>,
+        dedup: Arc<DedupService>,
+        index: Arc<TantivyContentIndex>,
+        interval_ms: u64,
+        max_extract_file_bytes: u64,
+        max_text_bytes: usize,
+    ) -> Self {
+        Self {
+            maintenance_pool,
+            dedup,
+            index,
+            // Floor the cadence so a misconfiguration can't busy-loop the
+            // maintenance pool.
+            interval_ms: interval_ms.max(200),
+            max_extract_file_bytes,
+            max_text_bytes,
+        }
+    }
+
+    /// Spawn the indexing loop. Fire-and-forget: the loop logs and survives
+    /// every error (an exited loop would silently freeze the index while the
+    /// queue grows), and the first drain runs immediately to absorb rows left
+    /// over from a previous run or the migration backfill.
+    #[instrument(skip(self))]
+    pub fn start(self, needs_reseed: bool) {
+        info!(
+            "Starting content-index worker (every {}ms, batch {}, reseed: {})",
+            self.interval_ms, DRAIN_BATCH, needs_reseed
+        );
+        tokio::spawn(async move {
+            if let Err(e) = self.prepare(needs_reseed).await {
+                error!("Content-index prepare failed (continuing with queue as-is): {e}");
+            }
+
+            let mut ticker =
+                tokio::time::interval(std::time::Duration::from_millis(self.interval_ms));
+            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+            let mut ticks: u64 = 0;
+            loop {
+                ticker.tick().await;
+                for _ in 0..MAX_BATCHES_PER_TICK {
+                    match self.drain_once().await {
+                        Ok(0) => break,
+                        Ok(drained) => {
+                            debug!("Content-index drain: processed {drained} queue row(s)");
+                            if drained < DRAIN_BATCH as usize {
+                                break;
+                            }
+                        }
+                        Err(e) => {
+                            error!("Content-index drain failed (queue preserved, will retry): {e}");
+                            break;
+                        }
+                    }
+                }
+
+                ticks += 1;
+                if ticks.is_multiple_of(ORPHAN_SWEEP_TICKS) {
+                    self.sweep_orphaned_text().await;
+                }
+            }
+        });
+    }
+
+    /// Spawn the discard-only janitor used when content search is DISABLED:
+    /// the triggers are always installed, so something must keep the queue
+    /// from growing unboundedly. Re-enabling the feature reseeds from scratch
+    /// (index version marker), so discarding here loses nothing.
+    pub fn start_drain_only_janitor(maintenance_pool: Arc<PgPool>) {
+        info!("Content search disabled — starting queue janitor (discard-only)");
+        tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));
+            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+            loop {
+                ticker.tick().await;
+                if let Err(e) = sqlx::query("DELETE FROM storage.search_index_dirty")
+                    .execute(maintenance_pool.as_ref())
+                    .await
+                {
+                    error!("Content-search queue janitor failed: {e}");
+                }
+            }
+        });
+    }
+
+    /// Startup housekeeping: drop extraction rows from other extractor
+    /// versions (the reseed re-extracts them) and, when the on-disk index was
+    /// wiped, re-enqueue every live file.
+    async fn prepare(&self, needs_reseed: bool) -> Result<(), sqlx::Error> {
+        let dropped = sqlx::query("DELETE FROM storage.blob_extracted_text WHERE extractor <> $1")
+            .bind(EXTRACTOR_VERSION)
+            .execute(self.maintenance_pool.as_ref())
+            .await?
+            .rows_affected();
+        if dropped > 0 {
+            info!("Dropped {dropped} extraction row(s) from a previous extractor version");
+        }
+
+        if needs_reseed {
+            let queued = sqlx::query(
+                "INSERT INTO storage.search_index_dirty (file_id, op)
+                 SELECT id, 'upsert' FROM storage.files WHERE NOT is_trashed",
+            )
+            .execute(self.maintenance_pool.as_ref())
+            .await?
+            .rows_affected();
+            info!("Content-index reseed: queued {queued} file(s) for indexing");
+        }
+        Ok(())
+    }
+
+    /// Drain and process one queue batch. Returns the number of queue rows
+    /// consumed (0 = queue empty).
+    async fn drain_once(&self) -> Result<usize, sqlx::Error> {
+        let rows: Vec<(i64, Uuid, String)> = sqlx::query_as(
+            "SELECT id, file_id, op FROM storage.search_index_dirty ORDER BY id LIMIT $1",
+        )
+        .bind(DRAIN_BATCH)
+        .fetch_all(self.maintenance_pool.as_ref())
+        .await?;
+
+        if rows.is_empty() {
+            return Ok(0);
+        }
+        let drained_ids: Vec<i64> = rows.iter().map(|r| r.0).collect();
+
+        // Latest op per file wins (rows are id-ordered).
+        let mut latest_op: HashMap<Uuid, bool> = HashMap::with_capacity(rows.len());
+        for (_, file_id, op) in &rows {
+            latest_op.insert(*file_id, op == "upsert");
+        }
+        let upsert_candidates: Vec<Uuid> = latest_op
+            .iter()
+            .filter_map(|(id, &upsert)| upsert.then_some(*id))
+            .collect();
+        let mut deletes: HashSet<Uuid> = latest_op
+            .iter()
+            .filter_map(|(id, &upsert)| (!upsert).then_some(*id))
+            .collect();
+
+        // Authoritative state re-read: a queued 'upsert' whose row vanished
+        // or got trashed in the meantime becomes a delete.
+        let files: Vec<(Uuid, String, String, String, String, i64)> =
+            if upsert_candidates.is_empty() {
+                Vec::new()
+            } else {
+                sqlx::query_as(
+                    "SELECT fi.id, fi.user_id::text, fi.name, fi.blob_hash, fi.mime_type, fi.size
+                   FROM storage.files fi
+                  WHERE fi.id = ANY($1) AND NOT fi.is_trashed",
+                )
+                .bind(&upsert_candidates)
+                .fetch_all(self.maintenance_pool.as_ref())
+                .await?
+            };
+        let found: HashSet<Uuid> = files.iter().map(|f| f.0).collect();
+        deletes.extend(upsert_candidates.iter().filter(|id| !found.contains(id)));
+
+        // Per-blob text: batch-read the extraction cache, extract misses.
+        let wanted_hashes: Vec<String> = files
+            .iter()
+            .filter(|(_, _, name, _, mime, size)| {
+                text_extractor::supports(name, mime) && *size as u64 <= self.max_extract_file_bytes
+            })
+            .map(|f| f.3.clone())
+            .collect();
+        let mut text_by_hash: HashMap<String, Option<String>> = HashMap::new();
+        if !wanted_hashes.is_empty() {
+            let cached: Vec<(String, Option<String>, String)> = sqlx::query_as(
+                "SELECT blob_hash, text, status FROM storage.blob_extracted_text
+                  WHERE blob_hash = ANY($1)",
+            )
+            .bind(&wanted_hashes)
+            .fetch_all(self.maintenance_pool.as_ref())
+            .await?;
+            for (hash, text, status) in cached {
+                text_by_hash.insert(hash, (status == "ok").then_some(text.unwrap_or_default()));
+            }
+        }
+
+        let mut records = Vec::with_capacity(files.len());
+        for (file_id, user_id, name, blob_hash, mime, size) in files {
+            let supported = text_extractor::supports(&name, &mime);
+            let content = if !supported {
+                None
+            } else if let Some(cached) = text_by_hash.get(&blob_hash) {
+                cached.clone()
+            } else {
+                let extracted = self
+                    .extract_and_cache(&blob_hash, &name, &mime, size as u64)
+                    .await;
+                text_by_hash.insert(blob_hash.clone(), extracted.clone());
+                extracted
+            };
+
+            let preview = content
+                .as_deref()
+                .map(|t| truncate_on_char(t, PREVIEW_BYTES));
+            records.push(IndexDocRecord {
+                file_id: file_id.to_string(),
+                user_id,
+                name,
+                content,
+                preview,
+            });
+        }
+
+        // One batched Tantivy commit, off the async runtime.
+        let index = self.index.clone();
+        let delete_ids: Vec<String> = deletes.iter().map(Uuid::to_string).collect();
+        let applied: Result<(), String> =
+            match tokio::task::spawn_blocking(move || index.apply_batch(records, delete_ids)).await
+            {
+                Ok(Ok(())) => Ok(()),
+                Ok(Err(e)) => Err(e.to_string()),
+                Err(e) => Err(format!("join: {e}")),
+            };
+        if let Err(e) = applied {
+            // Queue rows survive — the next tick retries the whole batch.
+            error!("Tantivy batch apply failed (will retry): {e}");
+            return Ok(0);
+        }
+
+        // Only now is the work durable in the index — drop the queue rows.
+        sqlx::query("DELETE FROM storage.search_index_dirty WHERE id = ANY($1)")
+            .bind(&drained_ids)
+            .execute(self.maintenance_pool.as_ref())
+            .await?;
+
+        Ok(drained_ids.len())
+    }
+
+    /// Read the blob (already size-capped), run the extractor on the blocking
+    /// pool, and persist the terminal outcome keyed by blob hash. Transient
+    /// read failures persist nothing — the next queue event retries.
+    async fn extract_and_cache(
+        &self,
+        blob_hash: &str,
+        name: &str,
+        mime: &str,
+        size: u64,
+    ) -> Option<String> {
+        if size > self.max_extract_file_bytes {
+            self.store_extraction(blob_hash, None, "too_large").await;
+            return None;
+        }
+
+        let bytes = match self.dedup.read_blob_bytes(blob_hash).await {
+            Ok(bytes) => bytes,
+            Err(e) => {
+                warn!(
+                    "Content-index blob read failed for {blob_hash} (will retry on next event): {e}"
+                );
+                return None;
+            }
+        };
+
+        let (name, mime, max_text) = (name.to_owned(), mime.to_owned(), self.max_text_bytes);
+        let outcome = tokio::task::spawn_blocking(move || {
+            text_extractor::extract(&name, &mime, &bytes, max_text)
+        })
+        .await
+        .unwrap_or_else(|e| ExtractedText::Failed(format!("join: {e}")));
+
+        match outcome {
+            ExtractedText::Text(text) => {
+                self.store_extraction(blob_hash, Some(&text), "ok").await;
+                Some(text)
+            }
+            ExtractedText::Empty => {
+                self.store_extraction(blob_hash, None, "empty").await;
+                None
+            }
+            ExtractedText::Failed(reason) => {
+                warn!("Text extraction failed for blob {blob_hash}: {reason}");
+                self.store_extraction(blob_hash, None, "failed").await;
+                None
+            }
+            ExtractedText::Unsupported => None,
+        }
+    }
+
+    async fn store_extraction(&self, blob_hash: &str, text: Option<&str>, status: &str) {
+        if let Err(e) = sqlx::query(
+            "INSERT INTO storage.blob_extracted_text (blob_hash, text, status, extractor)
+             VALUES ($1, $2, $3, $4)
+             ON CONFLICT (blob_hash) DO NOTHING",
+        )
+        .bind(blob_hash)
+        .bind(text)
+        .bind(status)
+        .bind(EXTRACTOR_VERSION)
+        .execute(self.maintenance_pool.as_ref())
+        .await
+        {
+            warn!("Failed to cache extraction for blob {blob_hash}: {e}");
+        }
+    }
+
+    /// Drop extraction rows whose blob no longer backs any live file. Uses
+    /// the `idx_files_blob_hash` index; runs hourly on the maintenance pool.
+    async fn sweep_orphaned_text(&self) {
+        match sqlx::query(
+            "DELETE FROM storage.blob_extracted_text bet
+              WHERE NOT EXISTS (SELECT 1 FROM storage.files f WHERE f.blob_hash = bet.blob_hash)",
+        )
+        .execute(self.maintenance_pool.as_ref())
+        .await
+        {
+            Ok(result) if result.rows_affected() > 0 => {
+                debug!(
+                    "Content-index sweep: dropped {} orphaned extraction row(s)",
+                    result.rows_affected()
+                );
+            }
+            Ok(_) => {}
+            Err(e) => error!("Content-index orphan sweep failed: {e}"),
+        }
+    }
+}
+
+/// Truncate on a char boundary at most `max_bytes` into `s`.
+fn truncate_on_char(s: &str, max_bytes: usize) -> String {
+    if s.len() <= max_bytes {
+        return s.to_owned();
+    }
+    let mut end = max_bytes;
+    while end > 0 && !s.is_char_boundary(end) {
+        end -= 1;
+    }
+    s[..end].to_owned()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::truncate_on_char;
+
+    #[test]
+    fn truncates_on_char_boundary() {
+        assert_eq!(truncate_on_char("patatas", 4), "pata");
+        // 'ñ' is 2 bytes — a cut landing inside it must back off ("ñoño" is
+        // ñ:0-1 o:2 ñ:3-4 o:5, so a 4-byte cut falls mid-ñ and yields "ño").
+        assert_eq!(truncate_on_char("ñoño", 4), "ño");
+        assert_eq!(truncate_on_char("ok", 10), "ok");
+    }
+}