Merge origin/main (Tantivy content search) into delta-sync branch
Both sides added a parameter to create_application_services and a setup step before it: this branch's storage-usage/quota service (for the instant-upload path) and main's Tantivy content index (for SearchService). The resolution keeps both — the signature takes both arguments and the build runs storage usage as step 3c and the content index as 3d. https://claude.ai/code/session_01WdNenpnujNR2sc32XVvwfS
This commit is contained in:
Claude
@@ -53,6 +53,93 @@ type FileRow = (
|
||||
Option<Uuid>,
|
||||
);
|
||||
|
||||
/// Append the optional type/date/size filters from `criteria` to
|
||||
/// `conditions`, continuing placeholder numbering from `bind_idx`. Returns
|
||||
/// the last placeholder index used. The name filter is NOT handled here —
|
||||
/// it is search-flavour specific (ILIKE for name search, absent for
|
||||
/// content-hit hydration). Mirror of [`bind_criteria_filters`]; the two
|
||||
/// must stay in sync.
|
||||
fn push_criteria_filters(
|
||||
conditions: &mut Vec<String>,
|
||||
mut bind_idx: u32,
|
||||
criteria: &SearchCriteriaDto,
|
||||
) -> u32 {
|
||||
if let Some(types) = &criteria.file_types
|
||||
&& !types.is_empty()
|
||||
{
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"LOWER(SUBSTRING(fi.name FROM '\\.([^.]+)$')) = ANY(${bind_idx})"
|
||||
));
|
||||
}
|
||||
if criteria.created_after.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"EXTRACT(EPOCH FROM fi.created_at)::bigint >= ${bind_idx}"
|
||||
));
|
||||
}
|
||||
if criteria.created_before.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"EXTRACT(EPOCH FROM fi.created_at)::bigint <= ${bind_idx}"
|
||||
));
|
||||
}
|
||||
if criteria.modified_after.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"EXTRACT(EPOCH FROM fi.updated_at)::bigint >= ${bind_idx}"
|
||||
));
|
||||
}
|
||||
if criteria.modified_before.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"EXTRACT(EPOCH FROM fi.updated_at)::bigint <= ${bind_idx}"
|
||||
));
|
||||
}
|
||||
if criteria.min_size.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!("fi.size >= ${bind_idx}"));
|
||||
}
|
||||
if criteria.max_size.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!("fi.size <= ${bind_idx}"));
|
||||
}
|
||||
bind_idx
|
||||
}
|
||||
|
||||
/// Bind the values for the filters appended by [`push_criteria_filters`],
|
||||
/// in the same order.
|
||||
fn bind_criteria_filters<'q, O>(
|
||||
mut query: sqlx::query::QueryAs<'q, sqlx::Postgres, O, sqlx::postgres::PgArguments>,
|
||||
criteria: &SearchCriteriaDto,
|
||||
) -> sqlx::query::QueryAs<'q, sqlx::Postgres, O, sqlx::postgres::PgArguments> {
|
||||
if let Some(types) = &criteria.file_types
|
||||
&& !types.is_empty()
|
||||
{
|
||||
let lower_types: Vec<String> = types.iter().map(|t| t.to_lowercase()).collect();
|
||||
query = query.bind(lower_types);
|
||||
}
|
||||
if let Some(v) = criteria.created_after {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.created_before {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.modified_after {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.modified_before {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.min_size {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.max_size {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
query
|
||||
}
|
||||
|
||||
/// File read repository backed by PostgreSQL metadata + blob storage.
|
||||
pub struct FileBlobReadRepository {
|
||||
pool: Arc<PgPool>,
|
||||
@@ -94,6 +181,80 @@ impl FileBlobReadRepository {
|
||||
self.hash_cache.clone()
|
||||
}
|
||||
|
||||
/// Hydrate content-index candidate ids into `File`s, re-applying the
|
||||
/// caller's scope and the active search filters (owner, trash state,
|
||||
/// folder scope, types, dates, sizes). The NAME filter is deliberately
|
||||
/// NOT applied — content hits don't need to match it. Ids that fail any
|
||||
/// filter (or no longer exist — the index is eventually consistent)
|
||||
/// simply drop out, so a stale index can never leak a result.
|
||||
pub async fn fetch_files_by_ids_filtered(
|
||||
&self,
|
||||
ids: &[String],
|
||||
criteria: &SearchCriteriaDto,
|
||||
user_id: Uuid,
|
||||
) -> Result<Vec<File>, DomainError> {
|
||||
// Index hits are externally produced strings — parse defensively.
|
||||
let uuid_ids: Vec<Uuid> = ids.iter().filter_map(|id| id.parse().ok()).collect();
|
||||
if uuid_ids.is_empty() {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let mut conditions: Vec<String> = vec![
|
||||
"fi.id = ANY($1)".to_string(),
|
||||
"fi.user_id = $2".to_string(),
|
||||
"fi.is_trashed = false".to_string(),
|
||||
];
|
||||
let mut bind_idx = 2u32;
|
||||
|
||||
if criteria.folder_id.is_some() {
|
||||
bind_idx += 1;
|
||||
if criteria.recursive {
|
||||
conditions.push(format!(
|
||||
"fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = ${bind_idx}::uuid)"
|
||||
));
|
||||
} else {
|
||||
conditions.push(format!("fi.folder_id = ${bind_idx}::uuid"));
|
||||
}
|
||||
}
|
||||
push_criteria_filters(&mut conditions, bind_idx, criteria);
|
||||
|
||||
let where_clause = conditions.join(" AND ");
|
||||
let sql = format!(
|
||||
"SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path, \
|
||||
fi.size, fi.mime_type, \
|
||||
EXTRACT(EPOCH FROM fi.created_at)::bigint, \
|
||||
EXTRACT(EPOCH FROM fi.updated_at)::bigint, \
|
||||
fi.blob_hash, \
|
||||
fi.user_id \
|
||||
FROM storage.files fi \
|
||||
LEFT JOIN storage.folders fo ON fo.id = fi.folder_id \
|
||||
WHERE {where_clause}"
|
||||
);
|
||||
|
||||
let mut query = sqlx::query_as::<_, FileRow>(&sql)
|
||||
.bind(uuid_ids)
|
||||
.bind(user_id);
|
||||
if let Some(folder_id) = criteria.folder_id.as_deref() {
|
||||
query = query.bind(folder_id);
|
||||
}
|
||||
query = bind_criteria_filters(query, criteria);
|
||||
|
||||
let rows = query.fetch_all(self.pool.as_ref()).await.map_err(|e| {
|
||||
DomainError::internal_error("FileBlobRead", format!("hydrate by ids: {e}"))
|
||||
})?;
|
||||
|
||||
rows.into_iter()
|
||||
.map(
|
||||
|(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)| {
|
||||
Self::row_to_file(id, name, fid, fpath, size, mime, ca, ma, blob_hash, uid)
|
||||
},
|
||||
)
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("FileBlobRead", format!("hydrate mapping: {e}"))
|
||||
})
|
||||
}
|
||||
|
||||
/// Returns the user_id (owner) for a given file ID.
|
||||
/// Mirrors `FolderDbRepository::get_folder_user_id`.
|
||||
/// Used by the AuthorizationEngine for owner short-circuit.
|
||||
@@ -1021,46 +1182,7 @@ impl FileReadPort for FileBlobReadRepository {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!("fi.name ILIKE ${bind_idx}"));
|
||||
}
|
||||
if let Some(types) = &criteria.file_types
|
||||
&& !types.is_empty()
|
||||
{
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"LOWER(SUBSTRING(fi.name FROM '\\.([^.]+)$')) = ANY(${bind_idx})"
|
||||
));
|
||||
}
|
||||
if criteria.created_after.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"EXTRACT(EPOCH FROM fi.created_at)::bigint >= ${bind_idx}"
|
||||
));
|
||||
}
|
||||
if criteria.created_before.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"EXTRACT(EPOCH FROM fi.created_at)::bigint <= ${bind_idx}"
|
||||
));
|
||||
}
|
||||
if criteria.modified_after.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"EXTRACT(EPOCH FROM fi.updated_at)::bigint >= ${bind_idx}"
|
||||
));
|
||||
}
|
||||
if criteria.modified_before.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!(
|
||||
"EXTRACT(EPOCH FROM fi.updated_at)::bigint <= ${bind_idx}"
|
||||
));
|
||||
}
|
||||
if criteria.min_size.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!("fi.size >= ${bind_idx}"));
|
||||
}
|
||||
if criteria.max_size.is_some() {
|
||||
bind_idx += 1;
|
||||
conditions.push(format!("fi.size <= ${bind_idx}"));
|
||||
}
|
||||
bind_idx = push_criteria_filters(&mut conditions, bind_idx, criteria);
|
||||
|
||||
let where_clause = conditions.join(" AND ");
|
||||
let limit_bind = bind_idx + 1;
|
||||
@@ -1107,30 +1229,7 @@ impl FileReadPort for FileBlobReadRepository {
|
||||
{
|
||||
query = query.bind(super::like_escape(name));
|
||||
}
|
||||
if let Some(types) = &criteria.file_types
|
||||
&& !types.is_empty()
|
||||
{
|
||||
let lower_types: Vec<String> = types.iter().map(|t| t.to_lowercase()).collect();
|
||||
query = query.bind(lower_types);
|
||||
}
|
||||
if let Some(v) = criteria.created_after {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.created_before {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.modified_after {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.modified_before {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.min_size {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
if let Some(v) = criteria.max_size {
|
||||
query = query.bind(v as i64);
|
||||
}
|
||||
query = bind_criteria_filters(query, criteria);
|
||||
|
||||
query = query.bind(limit).bind(offset);
|
||||
|
||||
|
||||
@@ -23,6 +23,7 @@ pub mod path_service;
|
||||
pub mod pg_acl_engine;
|
||||
pub mod retry_blob_backend;
|
||||
pub mod s3_blob_backend;
|
||||
pub mod search_index;
|
||||
pub mod share_unlock_cookie;
|
||||
pub mod smtp_email_sender;
|
||||
pub mod thumbnail_service;
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
//! Embedded full-text content index (Tantivy) and its feeding pipeline.
|
||||
//!
|
||||
//! Three pieces, mirroring the thumbnail/tree-etag architecture:
|
||||
//!
|
||||
//! * [`tantivy_content_index`] — the embedded BM25 index over file names and
|
||||
//! extracted content. Lives on local disk (`{storage}/.search-index`),
|
||||
//! single-writer, microsecond queries. A DERIVED artifact: PostgreSQL is
|
||||
//! the source of truth and the index is rebuilt (reseeded) whenever its
|
||||
//! on-disk schema version differs from the binary's.
|
||||
//! * [`text_extractor`] — pure-Rust text extraction (plain text/code, PDF,
|
||||
//! Office OOXML/ODF). CPU-bound, runs only on the background worker.
|
||||
//! * [`content_index_worker`] — drains `storage.search_index_dirty` (fed by
|
||||
//! statement triggers on `storage.files`), extracts text once per unique
|
||||
//! blob (BLAKE3-keyed cache in `storage.blob_extracted_text`), and applies
|
||||
//! batched Tantivy mutations. Never touches a request path.
|
||||
|
||||
pub mod content_index_worker;
|
||||
pub mod tantivy_content_index;
|
||||
pub mod text_extractor;
|
||||
@@ -0,0 +1,545 @@
|
||||
//! Embedded Tantivy index over file names + extracted content.
|
||||
//!
|
||||
//! Performance contract (the reason Tantivy was chosen):
|
||||
//! * queries are memory-mapped posting-list lookups — µs to low ms even at
|
||||
//! millions of documents, executed on the blocking pool (never stalls the
|
||||
//! Tokio reactor);
|
||||
//! * the single `IndexWriter` is owned by the background worker; request
|
||||
//! paths only ever touch the lock-free `IndexReader`.
|
||||
//!
|
||||
//! Index layout: one document per live file.
|
||||
//! * `file_id` — raw term, stored. Identity for upsert (delete_term + add).
|
||||
//! * `user_id` — raw term. Every query is `Must`-filtered by it, and hits
|
||||
//! are re-validated through SQL hydration afterwards (defense in depth).
|
||||
//! * `name` — tokenized file name (boosted 3x at query time).
|
||||
//! * `content` — tokenized extracted text (never stored — the index stays
|
||||
//! small; snippets come from `preview`).
|
||||
//! * `preview` — stored-only head of the extracted text used to render a
|
||||
//! snippet around the first matched term.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Mutex;
|
||||
|
||||
use tantivy::collector::TopDocs;
|
||||
use tantivy::directory::MmapDirectory;
|
||||
use tantivy::query::{BooleanQuery, BoostQuery, FuzzyTermQuery, Occur, Query, TermQuery};
|
||||
use tantivy::schema::{Field, IndexRecordOption, STORED, STRING, Schema, TEXT, Value as _};
|
||||
use tantivy::snippet::SnippetGenerator;
|
||||
use tantivy::tokenizer::TextAnalyzer;
|
||||
use tantivy::{Index, IndexReader, IndexWriter, ReloadPolicy, TantivyDocument, Term, doc};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::application::ports::content_index_ports::{ContentHitDto, ContentIndexPort};
|
||||
use crate::common::errors::DomainError;
|
||||
|
||||
/// Bump whenever the Tantivy schema OR the text extractor output changes in a
|
||||
/// way that requires re-indexing. A mismatch with the on-disk marker wipes the
|
||||
/// index directory and reseeds the dirty queue with every live file.
|
||||
pub const INDEX_SCHEMA_VERSION: &str = "1";
|
||||
|
||||
/// Recorded in `storage.blob_extracted_text.extractor`; rows from another
|
||||
/// version are dropped at worker startup (the reseed re-extracts them).
|
||||
/// Keep in lockstep with [`INDEX_SCHEMA_VERSION`].
|
||||
pub const EXTRACTOR_VERSION: &str = "rust-native-1";
|
||||
|
||||
/// Marker file inside the index directory carrying the schema version.
|
||||
const META_FILE: &str = "oxicloud-index.version";
|
||||
|
||||
/// RAM budget for the single-threaded writer. Indexing is a trickle-feed
|
||||
/// background task — one thread and a small heap keep the footprint
|
||||
/// negligible next to the request-serving process.
|
||||
const WRITER_HEAP_BYTES: usize = 64 * 1024 * 1024;
|
||||
|
||||
/// Hard cap on query tokens — a pathological query must not fan out into
|
||||
/// dozens of fuzzy automata.
|
||||
const MAX_QUERY_TOKENS: usize = 8;
|
||||
|
||||
/// Snippet length target, in characters.
|
||||
const SNIPPET_MAX_CHARS: usize = 180;
|
||||
|
||||
/// Minimum token length for typo-tolerant (edit distance 1) matching.
|
||||
/// Short tokens produce too many false positives under fuzzy matching.
|
||||
const FUZZY_MIN_CHARS: usize = 5;
|
||||
|
||||
/// Minimum token length for prefix expansion of the LAST query token
|
||||
/// (search-as-you-type behaviour).
|
||||
const PREFIX_MIN_CHARS: usize = 3;
|
||||
|
||||
/// One file to (re-)index. `content`/`preview` are `None` for files without
|
||||
/// extractable text (images, archives…) — their NAME is still indexed.
|
||||
#[derive(Debug)]
|
||||
pub struct IndexDocRecord {
|
||||
pub file_id: String,
|
||||
pub user_id: String,
|
||||
pub name: String,
|
||||
pub content: Option<String>,
|
||||
pub preview: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct IndexFields {
|
||||
file_id: Field,
|
||||
user_id: Field,
|
||||
name: Field,
|
||||
content: Field,
|
||||
preview: Field,
|
||||
}
|
||||
|
||||
pub struct TantivyContentIndex {
|
||||
/// Sole writer — owned by the background worker; the Mutex is never
|
||||
/// contended on a request path. (Writer and reader each keep the
|
||||
/// underlying `Index` alive.)
|
||||
writer: Mutex<IndexWriter>,
|
||||
reader: IndexReader,
|
||||
/// Pre-cloned analyzer for query-side tokenization (matches the index
|
||||
/// side: simple split + lowercase).
|
||||
analyzer: TextAnalyzer,
|
||||
fields: IndexFields,
|
||||
}
|
||||
|
||||
impl TantivyContentIndex {
|
||||
fn build_schema() -> (Schema, IndexFields) {
|
||||
let mut builder = Schema::builder();
|
||||
let fields = IndexFields {
|
||||
file_id: builder.add_text_field("file_id", STRING | STORED),
|
||||
user_id: builder.add_text_field("user_id", STRING),
|
||||
name: builder.add_text_field("name", TEXT),
|
||||
content: builder.add_text_field("content", TEXT),
|
||||
preview: builder.add_text_field("preview", STORED),
|
||||
};
|
||||
(builder.build(), fields)
|
||||
}
|
||||
|
||||
/// Open the index at `dir`, wiping and recreating it when the on-disk
|
||||
/// version marker is absent or stale. Returns `(index, needs_reseed)`:
|
||||
/// when `needs_reseed` is true the caller must re-enqueue every live file.
|
||||
pub fn open_or_rebuild(dir: &Path) -> Result<(Self, bool), DomainError> {
|
||||
let marker: PathBuf = dir.join(META_FILE);
|
||||
let version_ok = std::fs::read_to_string(&marker)
|
||||
.map(|v| v.trim() == INDEX_SCHEMA_VERSION)
|
||||
.unwrap_or(false);
|
||||
|
||||
if !version_ok && dir.exists() {
|
||||
std::fs::remove_dir_all(dir).map_err(|e| {
|
||||
DomainError::internal_error(
|
||||
"ContentIndex",
|
||||
format!("wiping stale index dir {}: {e}", dir.display()),
|
||||
)
|
||||
})?;
|
||||
}
|
||||
std::fs::create_dir_all(dir).map_err(|e| {
|
||||
DomainError::internal_error(
|
||||
"ContentIndex",
|
||||
format!("creating index dir {}: {e}", dir.display()),
|
||||
)
|
||||
})?;
|
||||
|
||||
let (schema, fields) = Self::build_schema();
|
||||
let mmap = MmapDirectory::open(dir)
|
||||
.map_err(|e| DomainError::internal_error("ContentIndex", format!("mmap dir: {e}")))?;
|
||||
let index = Index::open_or_create(mmap, schema)
|
||||
.map_err(|e| DomainError::internal_error("ContentIndex", format!("open: {e}")))?;
|
||||
|
||||
// Single writer thread: indexing is a background trickle, not a bulk
|
||||
// load — keep the CPU/RAM footprint minimal.
|
||||
let writer = index
|
||||
.writer_with_num_threads::<TantivyDocument>(1, WRITER_HEAP_BYTES)
|
||||
.map_err(|e| DomainError::internal_error("ContentIndex", format!("writer: {e}")))?;
|
||||
|
||||
let reader = index
|
||||
.reader_builder()
|
||||
.reload_policy(ReloadPolicy::OnCommitWithDelay)
|
||||
.try_into()
|
||||
.map_err(|e| DomainError::internal_error("ContentIndex", format!("reader: {e}")))?;
|
||||
|
||||
let analyzer = index
|
||||
.tokenizer_for_field(fields.content)
|
||||
.map_err(|e| DomainError::internal_error("ContentIndex", format!("analyzer: {e}")))?;
|
||||
|
||||
std::fs::write(&marker, INDEX_SCHEMA_VERSION).map_err(|e| {
|
||||
DomainError::internal_error("ContentIndex", format!("writing version marker: {e}"))
|
||||
})?;
|
||||
|
||||
Ok((
|
||||
Self {
|
||||
writer: Mutex::new(writer),
|
||||
reader,
|
||||
analyzer,
|
||||
fields,
|
||||
},
|
||||
!version_ok,
|
||||
))
|
||||
}
|
||||
|
||||
/// Apply one drained queue batch: deletes, then upserts, then ONE commit.
|
||||
/// Blocking (disk I/O + segment serialization) — call from the worker via
|
||||
/// `spawn_blocking`. The caller deletes the queue rows only after this
|
||||
/// returns `Ok`, so a crash in between re-processes the batch
|
||||
/// (idempotent: upsert = delete_term + add).
|
||||
pub fn apply_batch(
|
||||
&self,
|
||||
upserts: Vec<IndexDocRecord>,
|
||||
deletes: Vec<String>,
|
||||
) -> Result<(), DomainError> {
|
||||
let mut writer = self
|
||||
.writer
|
||||
.lock()
|
||||
.map_err(|_| DomainError::internal_error("ContentIndex", "writer mutex poisoned"))?;
|
||||
|
||||
for file_id in &deletes {
|
||||
writer.delete_term(Term::from_field_text(self.fields.file_id, file_id));
|
||||
}
|
||||
|
||||
for record in upserts {
|
||||
writer.delete_term(Term::from_field_text(self.fields.file_id, &record.file_id));
|
||||
let mut document = doc!(
|
||||
self.fields.file_id => record.file_id,
|
||||
self.fields.user_id => record.user_id,
|
||||
self.fields.name => record.name,
|
||||
);
|
||||
if let Some(content) = record.content {
|
||||
document.add_text(self.fields.content, content);
|
||||
}
|
||||
if let Some(preview) = record.preview {
|
||||
document.add_text(self.fields.preview, preview);
|
||||
}
|
||||
writer
|
||||
.add_document(document)
|
||||
.map_err(|e| DomainError::internal_error("ContentIndex", format!("add: {e}")))?;
|
||||
}
|
||||
|
||||
writer
|
||||
.commit()
|
||||
.map_err(|e| DomainError::internal_error("ContentIndex", format!("commit: {e}")))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Number of live documents — used by tests and the startup log line.
|
||||
pub fn num_docs(&self) -> u64 {
|
||||
self.reader.searcher().num_docs()
|
||||
}
|
||||
|
||||
/// Tokenize `raw` with the index analyzer (simple split + lowercase).
|
||||
fn query_tokens(analyzer: &TextAnalyzer, raw: &str) -> Vec<String> {
|
||||
let mut analyzer = analyzer.clone();
|
||||
let mut tokens = Vec::new();
|
||||
let mut stream = analyzer.token_stream(raw);
|
||||
while stream.advance() && tokens.len() < MAX_QUERY_TOKENS {
|
||||
tokens.push(stream.token().text.clone());
|
||||
}
|
||||
tokens
|
||||
}
|
||||
|
||||
/// Build the scored query: every token must match (in name OR content,
|
||||
/// exact OR fuzzy OR — for the last token — prefix), and the whole thing
|
||||
/// is `Must`-scoped to the user.
|
||||
fn build_query(fields: IndexFields, user_id: &str, tokens: &[String]) -> Box<dyn Query> {
|
||||
let mut clauses: Vec<(Occur, Box<dyn Query>)> = vec![(
|
||||
Occur::Must,
|
||||
Box::new(TermQuery::new(
|
||||
Term::from_field_text(fields.user_id, user_id),
|
||||
IndexRecordOption::Basic,
|
||||
)),
|
||||
)];
|
||||
|
||||
let last = tokens.len().saturating_sub(1);
|
||||
for (i, token) in tokens.iter().enumerate() {
|
||||
let name_term = Term::from_field_text(fields.name, token);
|
||||
let content_term = Term::from_field_text(fields.content, token);
|
||||
|
||||
let mut alternatives: Vec<(Occur, Box<dyn Query>)> = vec![
|
||||
(
|
||||
Occur::Should,
|
||||
// Name matches outrank content matches for the same term.
|
||||
Box::new(BoostQuery::new(
|
||||
Box::new(TermQuery::new(
|
||||
name_term.clone(),
|
||||
IndexRecordOption::WithFreqs,
|
||||
)),
|
||||
3.0,
|
||||
)),
|
||||
),
|
||||
(
|
||||
Occur::Should,
|
||||
Box::new(TermQuery::new(
|
||||
content_term.clone(),
|
||||
IndexRecordOption::WithFreqs,
|
||||
)),
|
||||
),
|
||||
];
|
||||
|
||||
if token.chars().count() >= FUZZY_MIN_CHARS {
|
||||
// Edit distance 1 absorbs typos and most singular/plural
|
||||
// morphology ("patata" ↔ "patatas") without a stemmer.
|
||||
alternatives.push((
|
||||
Occur::Should,
|
||||
Box::new(FuzzyTermQuery::new(name_term.clone(), 1, true)),
|
||||
));
|
||||
alternatives.push((
|
||||
Occur::Should,
|
||||
Box::new(FuzzyTermQuery::new(content_term.clone(), 1, true)),
|
||||
));
|
||||
}
|
||||
if i == last && token.chars().count() >= PREFIX_MIN_CHARS {
|
||||
// Search-as-you-type: the token still being typed matches as
|
||||
// a prefix ("pata" → "patatas").
|
||||
alternatives.push((
|
||||
Occur::Should,
|
||||
Box::new(FuzzyTermQuery::new_prefix(name_term, 0, true)),
|
||||
));
|
||||
alternatives.push((
|
||||
Occur::Should,
|
||||
Box::new(FuzzyTermQuery::new_prefix(content_term, 0, true)),
|
||||
));
|
||||
}
|
||||
|
||||
clauses.push((Occur::Must, Box::new(BooleanQuery::new(alternatives))));
|
||||
}
|
||||
|
||||
Box::new(BooleanQuery::new(clauses))
|
||||
}
|
||||
|
||||
/// Blocking search core — runs on the blocking pool via the port impl.
|
||||
fn search_blocking(
|
||||
searcher: tantivy::Searcher,
|
||||
analyzer: TextAnalyzer,
|
||||
fields: IndexFields,
|
||||
user_id: &str,
|
||||
raw_query: &str,
|
||||
limit: usize,
|
||||
) -> Result<Vec<ContentHitDto>, DomainError> {
|
||||
let tokens = Self::query_tokens(&analyzer, raw_query);
|
||||
if tokens.is_empty() {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let query = Self::build_query(fields, user_id, &tokens);
|
||||
let top_docs = searcher
|
||||
.search(&query, &TopDocs::with_limit(limit.max(1)).order_by_score())
|
||||
.map_err(|e| DomainError::internal_error("ContentIndex", format!("search: {e}")))?;
|
||||
|
||||
// Snippets highlight CONTENT matches; an empty fragment means the hit
|
||||
// came from the name (or a fuzzy variant) — no snippet then.
|
||||
let snippet_generator = SnippetGenerator::create(&searcher, &*query, fields.content)
|
||||
.map(|mut g| {
|
||||
g.set_max_num_chars(SNIPPET_MAX_CHARS);
|
||||
g
|
||||
})
|
||||
.ok();
|
||||
|
||||
let mut hits = Vec::with_capacity(top_docs.len());
|
||||
for (score, address) in top_docs {
|
||||
let document: TantivyDocument = searcher.doc(address).map_err(|e| {
|
||||
DomainError::internal_error("ContentIndex", format!("doc fetch: {e}"))
|
||||
})?;
|
||||
let Some(file_id) = document
|
||||
.get_first(fields.file_id)
|
||||
.and_then(|v| v.as_str())
|
||||
.map(str::to_owned)
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
|
||||
let snippet = document
|
||||
.get_first(fields.preview)
|
||||
.and_then(|v| v.as_str())
|
||||
.and_then(|preview| {
|
||||
let generator = snippet_generator.as_ref()?;
|
||||
let fragment = generator.snippet(preview).fragment().trim().to_owned();
|
||||
(!fragment.is_empty()).then_some(fragment)
|
||||
});
|
||||
|
||||
hits.push(ContentHitDto {
|
||||
file_id,
|
||||
score,
|
||||
snippet,
|
||||
});
|
||||
}
|
||||
Ok(hits)
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ContentIndexPort for TantivyContentIndex {
|
||||
async fn search_content(
|
||||
&self,
|
||||
user_id: Uuid,
|
||||
query: &str,
|
||||
limit: usize,
|
||||
) -> Result<Vec<ContentHitDto>, DomainError> {
|
||||
let searcher = self.reader.searcher();
|
||||
let analyzer = self.analyzer.clone();
|
||||
let fields = self.fields;
|
||||
let user_id = user_id.to_string();
|
||||
let query = query.to_owned();
|
||||
|
||||
tokio::task::spawn_blocking(move || {
|
||||
Self::search_blocking(searcher, analyzer, fields, &user_id, &query, limit)
|
||||
})
|
||||
.await
|
||||
.map_err(|e| DomainError::internal_error("ContentIndex", format!("join: {e}")))?
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn record(file_id: &str, user_id: &str, name: &str, content: Option<&str>) -> IndexDocRecord {
|
||||
IndexDocRecord {
|
||||
file_id: file_id.to_owned(),
|
||||
user_id: user_id.to_owned(),
|
||||
name: name.to_owned(),
|
||||
content: content.map(str::to_owned),
|
||||
preview: content.map(str::to_owned),
|
||||
}
|
||||
}
|
||||
|
||||
fn search(index: &TantivyContentIndex, user_id: &str, query: &str) -> Vec<ContentHitDto> {
|
||||
// Force a reader reload — OnCommitWithDelay is asynchronous and tests
|
||||
// must observe the commit immediately.
|
||||
index.reader.reload().unwrap();
|
||||
TantivyContentIndex::search_blocking(
|
||||
index.reader.searcher(),
|
||||
index.analyzer.clone(),
|
||||
index.fields,
|
||||
user_id,
|
||||
query,
|
||||
32,
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn index_name_and_content_with_fuzzy_prefix_and_user_isolation() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let (index, needs_reseed) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap();
|
||||
assert!(needs_reseed, "fresh dir must request a reseed");
|
||||
|
||||
index
|
||||
.apply_batch(
|
||||
vec![
|
||||
record("f1", "user-a", "patatas-fritas.jpg", None),
|
||||
record(
|
||||
"f2",
|
||||
"user-a",
|
||||
"recetas.pdf",
|
||||
Some("la mejor receta de patatas bravas del mundo"),
|
||||
),
|
||||
record(
|
||||
"f3",
|
||||
"user-b",
|
||||
"patatas-ajenas.txt",
|
||||
Some("patatas de otro usuario"),
|
||||
),
|
||||
record("f4", "user-a", "informe.txt", Some("nada relacionado aqui")),
|
||||
],
|
||||
Vec::new(),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// Exact term: name hit + content hit for user-a only.
|
||||
let hits = search(&index, "user-a", "patatas");
|
||||
let ids: Vec<&str> = hits.iter().map(|h| h.file_id.as_str()).collect();
|
||||
assert!(ids.contains(&"f1"), "name match expected: {ids:?}");
|
||||
assert!(ids.contains(&"f2"), "content match expected: {ids:?}");
|
||||
assert!(!ids.contains(&"f3"), "other user's file leaked: {ids:?}");
|
||||
assert!(!ids.contains(&"f4"), "non-matching file returned: {ids:?}");
|
||||
|
||||
// The content hit carries a snippet around the matched term.
|
||||
let content_hit = hits.iter().find(|h| h.file_id == "f2").unwrap();
|
||||
assert!(
|
||||
content_hit
|
||||
.snippet
|
||||
.as_deref()
|
||||
.unwrap_or("")
|
||||
.contains("patatas"),
|
||||
"snippet should surround the match: {:?}",
|
||||
content_hit.snippet
|
||||
);
|
||||
|
||||
// Fuzzy (distance 1): singular finds plural.
|
||||
let ids: Vec<String> = search(&index, "user-a", "patata")
|
||||
.into_iter()
|
||||
.map(|h| h.file_id)
|
||||
.collect();
|
||||
assert!(
|
||||
ids.contains(&"f2".to_owned()),
|
||||
"fuzzy match expected: {ids:?}"
|
||||
);
|
||||
|
||||
// Prefix on the last token (search-as-you-type).
|
||||
let ids: Vec<String> = search(&index, "user-a", "pata")
|
||||
.into_iter()
|
||||
.map(|h| h.file_id)
|
||||
.collect();
|
||||
assert!(
|
||||
ids.contains(&"f1".to_owned()),
|
||||
"prefix match expected: {ids:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn upsert_replaces_and_delete_removes() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let (index, _) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap();
|
||||
|
||||
index
|
||||
.apply_batch(
|
||||
vec![record(
|
||||
"f1",
|
||||
"u",
|
||||
"old-name.txt",
|
||||
Some("contenido original"),
|
||||
)],
|
||||
Vec::new(),
|
||||
)
|
||||
.unwrap();
|
||||
index
|
||||
.apply_batch(
|
||||
vec![record("f1", "u", "renamed.txt", Some("contenido original"))],
|
||||
Vec::new(),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert!(
|
||||
search(&index, "u", "old").is_empty(),
|
||||
"stale doc survived upsert"
|
||||
);
|
||||
assert_eq!(search(&index, "u", "renamed").len(), 1);
|
||||
|
||||
index
|
||||
.apply_batch(Vec::new(), vec!["f1".to_owned()])
|
||||
.unwrap();
|
||||
assert!(
|
||||
search(&index, "u", "renamed").is_empty(),
|
||||
"deleted doc still found"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reopen_preserves_documents_and_version_mismatch_wipes() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
{
|
||||
let (index, _) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap();
|
||||
index
|
||||
.apply_batch(vec![record("f1", "u", "persistente.txt", None)], Vec::new())
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
// Same version: documents survive, no reseed requested.
|
||||
{
|
||||
let (index, needs_reseed) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap();
|
||||
assert!(!needs_reseed);
|
||||
assert_eq!(index.num_docs(), 1);
|
||||
}
|
||||
|
||||
// Stale version marker: wipe + reseed.
|
||||
std::fs::write(dir.path().join(META_FILE), "0-stale").unwrap();
|
||||
let (index, needs_reseed) = TantivyContentIndex::open_or_rebuild(dir.path()).unwrap();
|
||||
assert!(needs_reseed);
|
||||
assert_eq!(index.num_docs(), 0);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,418 @@
|
||||
//! Pure-Rust text extraction for the content index.
|
||||
//!
|
||||
//! Supported: plain text/code/markup, PDF (text layer), Office OOXML
|
||||
//! (docx/xlsx/pptx) and OpenDocument (odt/ods/odp). Images, media and
|
||||
//! archives are reported as [`ExtractedText::Unsupported`] WITHOUT reading
|
||||
//! the blob (the worker checks [`supports`] first).
|
||||
//!
|
||||
//! Everything here is CPU-bound and synchronous — the worker runs it inside
|
||||
//! `spawn_blocking`, one extraction at a time, so user-facing latency is
|
||||
//! never affected. Output is whitespace-normalized and hard-capped at the
|
||||
//! caller-provided byte budget.
|
||||
|
||||
use std::io::{BufReader, Cursor};
|
||||
use std::panic::{AssertUnwindSafe, catch_unwind};
|
||||
|
||||
use quick_xml::events::Event;
|
||||
|
||||
/// Outcome of one extraction attempt.
|
||||
#[derive(Debug)]
|
||||
pub enum ExtractedText {
|
||||
/// Usable text (normalized, capped).
|
||||
Text(String),
|
||||
/// Extractor ran fine but produced no text (e.g. empty document,
|
||||
/// scanned PDF without a text layer, binary masquerading as text).
|
||||
Empty,
|
||||
/// No extractor handles this name/MIME combination.
|
||||
Unsupported,
|
||||
/// Extractor failed or panicked — terminal for this blob (recorded so it
|
||||
/// is never retried until the extractor version bumps).
|
||||
Failed(String),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq)]
|
||||
enum Kind {
|
||||
Plain,
|
||||
Pdf,
|
||||
Docx,
|
||||
Xlsx,
|
||||
Pptx,
|
||||
Odf,
|
||||
}
|
||||
|
||||
/// MIME types (beyond `text/*`) parsed as plain text.
|
||||
const TEXTUAL_MIMES: &[&str] = &[
|
||||
"application/json",
|
||||
"application/ld+json",
|
||||
"application/xml",
|
||||
"application/javascript",
|
||||
"application/x-javascript",
|
||||
"application/x-yaml",
|
||||
"application/yaml",
|
||||
"application/toml",
|
||||
"application/x-sh",
|
||||
"application/x-shellscript",
|
||||
"application/sql",
|
||||
"image/svg+xml",
|
||||
];
|
||||
|
||||
/// Extensions parsed as plain text when the MIME type is generic
|
||||
/// (`application/octet-stream` uploads are common on WebDAV clients).
|
||||
const TEXTUAL_EXTENSIONS: &[&str] = &[
|
||||
"txt", "md", "markdown", "csv", "tsv", "json", "xml", "yaml", "yml", "toml", "ini", "cfg",
|
||||
"conf", "log", "rs", "js", "mjs", "ts", "jsx", "tsx", "css", "scss", "html", "htm", "py", "rb",
|
||||
"go", "java", "c", "h", "cpp", "hpp", "cs", "php", "sh", "sql", "tex", "svg",
|
||||
];
|
||||
|
||||
fn extension_of(name: &str) -> Option<String> {
|
||||
name.rsplit_once('.').map(|(_, ext)| ext.to_lowercase())
|
||||
}
|
||||
|
||||
fn classify(name: &str, mime: &str) -> Option<Kind> {
|
||||
let mime = mime
|
||||
.split(';')
|
||||
.next()
|
||||
.unwrap_or_default()
|
||||
.trim()
|
||||
.to_lowercase();
|
||||
|
||||
if mime.starts_with("text/") || TEXTUAL_MIMES.contains(&mime.as_str()) {
|
||||
return Some(Kind::Plain);
|
||||
}
|
||||
match mime.as_str() {
|
||||
"application/pdf" => return Some(Kind::Pdf),
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document" => {
|
||||
return Some(Kind::Docx);
|
||||
}
|
||||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" => {
|
||||
return Some(Kind::Xlsx);
|
||||
}
|
||||
"application/vnd.openxmlformats-officedocument.presentationml.presentation" => {
|
||||
return Some(Kind::Pptx);
|
||||
}
|
||||
"application/vnd.oasis.opendocument.text"
|
||||
| "application/vnd.oasis.opendocument.spreadsheet"
|
||||
| "application/vnd.oasis.opendocument.presentation" => return Some(Kind::Odf),
|
||||
_ => {}
|
||||
}
|
||||
|
||||
// Generic MIME — fall back to the extension.
|
||||
match extension_of(name)?.as_str() {
|
||||
ext if TEXTUAL_EXTENSIONS.contains(&ext) => Some(Kind::Plain),
|
||||
"pdf" => Some(Kind::Pdf),
|
||||
"docx" => Some(Kind::Docx),
|
||||
"xlsx" => Some(Kind::Xlsx),
|
||||
"pptx" => Some(Kind::Pptx),
|
||||
"odt" | "ods" | "odp" => Some(Kind::Odf),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether [`extract`] has an extractor for this file — the worker calls this
|
||||
/// BEFORE reading the blob, so unsupported content (photos, video, archives)
|
||||
/// costs zero I/O.
|
||||
pub fn supports(name: &str, mime: &str) -> bool {
|
||||
classify(name, mime).is_some()
|
||||
}
|
||||
|
||||
/// Extract plain text from `bytes`, capped at `max_text_bytes` of UTF-8.
|
||||
pub fn extract(name: &str, mime: &str, bytes: &[u8], max_text_bytes: usize) -> ExtractedText {
|
||||
let Some(kind) = classify(name, mime) else {
|
||||
return ExtractedText::Unsupported;
|
||||
};
|
||||
|
||||
let result = match kind {
|
||||
Kind::Plain => extract_plain(bytes, max_text_bytes),
|
||||
Kind::Pdf => extract_pdf(bytes, max_text_bytes),
|
||||
Kind::Docx => {
|
||||
extract_zipped_xml(bytes, ZipSource::Fixed("word/document.xml"), max_text_bytes)
|
||||
}
|
||||
Kind::Xlsx => extract_zipped_xml(
|
||||
bytes,
|
||||
ZipSource::Fixed("xl/sharedStrings.xml"),
|
||||
max_text_bytes,
|
||||
),
|
||||
Kind::Pptx => extract_zipped_xml(bytes, ZipSource::Slides, max_text_bytes),
|
||||
Kind::Odf => extract_zipped_xml(bytes, ZipSource::Fixed("content.xml"), max_text_bytes),
|
||||
};
|
||||
|
||||
match result {
|
||||
Ok(text) if text.is_empty() => ExtractedText::Empty,
|
||||
Ok(text) => ExtractedText::Text(text),
|
||||
Err(reason) => ExtractedText::Failed(reason),
|
||||
}
|
||||
}
|
||||
|
||||
/// Collapse whitespace runs and cap at `max_bytes` (on a char boundary).
|
||||
/// Normalization keeps the index lean and makes stored previews readable.
|
||||
fn normalize_and_cap(text: &str, max_bytes: usize) -> String {
|
||||
let mut out = String::with_capacity(text.len().min(max_bytes));
|
||||
for word in text.split_whitespace() {
|
||||
if out.len() + word.len() + 1 > max_bytes {
|
||||
break;
|
||||
}
|
||||
if !out.is_empty() {
|
||||
out.push(' ');
|
||||
}
|
||||
out.push_str(word);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn extract_plain(bytes: &[u8], max_text_bytes: usize) -> Result<String, String> {
|
||||
// NUL byte in the head = binary masquerading under a textual name/MIME.
|
||||
if bytes.iter().take(8192).any(|&b| b == 0) {
|
||||
return Ok(String::new());
|
||||
}
|
||||
// Decode at most ~2x the budget — normalization only shrinks text, so
|
||||
// anything beyond that can never reach the output.
|
||||
let slice_end = bytes.len().min(max_text_bytes.saturating_mul(2));
|
||||
let text = String::from_utf8_lossy(&bytes[..slice_end]);
|
||||
Ok(normalize_and_cap(&text, max_text_bytes))
|
||||
}
|
||||
|
||||
fn extract_pdf(bytes: &[u8], max_text_bytes: usize) -> Result<String, String> {
|
||||
// pdf-extract is known to panic on malformed documents; a poisoned blob
|
||||
// must mark itself 'failed' instead of taking the worker down.
|
||||
let outcome = catch_unwind(AssertUnwindSafe(|| {
|
||||
pdf_extract::extract_text_from_mem(bytes)
|
||||
}));
|
||||
match outcome {
|
||||
Ok(Ok(text)) => Ok(normalize_and_cap(&text, max_text_bytes)),
|
||||
Ok(Err(e)) => Err(format!("pdf: {e}")),
|
||||
Err(_) => Err("pdf: extractor panicked".to_owned()),
|
||||
}
|
||||
}
|
||||
|
||||
enum ZipSource {
|
||||
/// One well-known entry (docx body, xlsx shared strings, ODF content).
|
||||
Fixed(&'static str),
|
||||
/// Every `ppt/slides/slideN.xml` entry.
|
||||
Slides,
|
||||
}
|
||||
|
||||
fn extract_zipped_xml(
|
||||
bytes: &[u8],
|
||||
source: ZipSource,
|
||||
max_text_bytes: usize,
|
||||
) -> Result<String, String> {
|
||||
let mut archive = zip::ZipArchive::new(Cursor::new(bytes)).map_err(|e| format!("zip: {e}"))?;
|
||||
|
||||
let entries: Vec<String> = match source {
|
||||
ZipSource::Fixed(name) => vec![name.to_owned()],
|
||||
ZipSource::Slides => {
|
||||
let mut slides: Vec<String> = archive
|
||||
.file_names()
|
||||
.filter(|n| n.starts_with("ppt/slides/slide") && n.ends_with(".xml"))
|
||||
.map(str::to_owned)
|
||||
.collect();
|
||||
slides.sort();
|
||||
slides
|
||||
}
|
||||
};
|
||||
|
||||
let mut text = String::new();
|
||||
for entry in entries {
|
||||
let Ok(file) = archive.by_name(&entry) else {
|
||||
// Tolerated: e.g. an xlsx with no shared strings table.
|
||||
continue;
|
||||
};
|
||||
collect_xml_text(BufReader::new(file), &mut text, max_text_bytes)
|
||||
.map_err(|e| format!("{entry}: {e}"))?;
|
||||
if text.len() >= max_text_bytes {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(normalize_and_cap(&text, max_text_bytes))
|
||||
}
|
||||
|
||||
/// Append every XML text node to `out` (capped). Text RUNS are concatenated
|
||||
/// without separators — OOXML splits words across `<w:t>` runs arbitrarily —
|
||||
/// while paragraph/cell boundaries insert whitespace so distinct words never
|
||||
/// fuse together.
|
||||
fn collect_xml_text<R: std::io::BufRead>(
|
||||
reader: R,
|
||||
out: &mut String,
|
||||
max_bytes: usize,
|
||||
) -> Result<(), String> {
|
||||
let mut xml = quick_xml::Reader::from_reader(reader);
|
||||
let mut buf = Vec::new();
|
||||
loop {
|
||||
if out.len() >= max_bytes {
|
||||
return Ok(());
|
||||
}
|
||||
match xml.read_event_into(&mut buf) {
|
||||
Ok(Event::Text(t)) => {
|
||||
if let Ok(decoded) = t.xml_content() {
|
||||
out.push_str(&decoded);
|
||||
}
|
||||
}
|
||||
Ok(Event::GeneralRef(r)) => {
|
||||
// quick-xml emits entity references as separate events.
|
||||
// Character refs (A) and the predefined five resolve to
|
||||
// their literal character; unknown custom entities are
|
||||
// dropped (no DTD resolution).
|
||||
if let Ok(Some(ch)) = r.resolve_char_ref() {
|
||||
out.push(ch);
|
||||
} else if let Ok(name) = r.decode() {
|
||||
match name.as_ref() {
|
||||
"amp" => out.push('&'),
|
||||
"lt" => out.push('<'),
|
||||
"gt" => out.push('>'),
|
||||
"apos" => out.push('\''),
|
||||
"quot" => out.push('"'),
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(Event::End(e)) => {
|
||||
// Paragraphs (docx w:p, pptx a:p, ODF text:p/text:h), table
|
||||
// rows and xlsx shared-string items all separate words.
|
||||
let local = e.local_name();
|
||||
if matches!(local.as_ref(), b"p" | b"h" | b"si" | b"row" | b"br") {
|
||||
out.push('\n');
|
||||
}
|
||||
}
|
||||
Ok(Event::Empty(e)) => {
|
||||
// Self-closing breaks/tabs inside a paragraph (<w:br/>, <w:tab/>).
|
||||
let local = e.local_name();
|
||||
if matches!(local.as_ref(), b"br" | b"tab") {
|
||||
out.push(' ');
|
||||
}
|
||||
}
|
||||
Ok(Event::Eof) => return Ok(()),
|
||||
Ok(_) => {}
|
||||
Err(e) => return Err(format!("xml: {e}")),
|
||||
}
|
||||
buf.clear();
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::io::Write;
|
||||
use zip::write::SimpleFileOptions;
|
||||
|
||||
fn build_zip(entries: &[(&str, &str)]) -> Vec<u8> {
|
||||
let mut writer = zip::ZipWriter::new(Cursor::new(Vec::new()));
|
||||
for (name, content) in entries {
|
||||
writer
|
||||
.start_file(*name, SimpleFileOptions::default())
|
||||
.unwrap();
|
||||
writer.write_all(content.as_bytes()).unwrap();
|
||||
}
|
||||
writer.finish().unwrap().into_inner()
|
||||
}
|
||||
|
||||
fn text_of(outcome: ExtractedText) -> String {
|
||||
match outcome {
|
||||
ExtractedText::Text(t) => t,
|
||||
other => panic!("expected Text, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plain_text_is_normalized_and_capped() {
|
||||
let out = extract(
|
||||
"notas.txt",
|
||||
"text/plain",
|
||||
b"receta de\n\npatatas bravas",
|
||||
1024,
|
||||
);
|
||||
assert_eq!(text_of(out), "receta de patatas bravas");
|
||||
|
||||
let big = "palabra ".repeat(1000);
|
||||
let out = text_of(extract("big.txt", "text/plain", big.as_bytes(), 64));
|
||||
assert!(out.len() <= 64, "cap exceeded: {}", out.len());
|
||||
assert!(out.ends_with("palabra"), "must cut on word boundary");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn binary_masquerading_as_text_yields_empty() {
|
||||
let mut bytes = b"PK\x03\x04".to_vec();
|
||||
bytes.extend_from_slice(&[0u8; 64]);
|
||||
assert!(matches!(
|
||||
extract("raro.txt", "text/plain", &bytes, 1024),
|
||||
ExtractedText::Empty
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unsupported_types_are_reported_without_reading() {
|
||||
assert!(!supports("foto.jpg", "image/jpeg"));
|
||||
assert!(matches!(
|
||||
extract("foto.jpg", "image/jpeg", &[0xFF, 0xD8], 1024),
|
||||
ExtractedText::Unsupported
|
||||
));
|
||||
assert!(supports("recetas.pdf", "application/pdf"));
|
||||
assert!(
|
||||
supports("notas", "text/plain"),
|
||||
"MIME wins without extension"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn docx_runs_concatenate_and_paragraphs_separate() {
|
||||
let body = r#"<?xml version="1.0"?>
|
||||
<w:document xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main">
|
||||
<w:body>
|
||||
<w:p><w:r><w:t>pata</w:t></w:r><w:r><w:t>tas</w:t></w:r></w:p>
|
||||
<w:p><w:r><w:t>bravas & ali oli</w:t></w:r></w:p>
|
||||
</w:body>
|
||||
</w:document>"#;
|
||||
let bytes = build_zip(&[("word/document.xml", body)]);
|
||||
let out = text_of(extract(
|
||||
"receta.docx",
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
&bytes,
|
||||
4096,
|
||||
));
|
||||
assert_eq!(out, "patatas bravas & ali oli");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn xlsx_shared_strings_extract() {
|
||||
let shared = r#"<?xml version="1.0"?>
|
||||
<sst xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" count="2">
|
||||
<si><t>patatas</t></si>
|
||||
<si><t>900 kg</t></si>
|
||||
</sst>"#;
|
||||
let bytes = build_zip(&[("xl/sharedStrings.xml", shared)]);
|
||||
let out = text_of(extract(
|
||||
"stock.xlsx",
|
||||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
&bytes,
|
||||
4096,
|
||||
));
|
||||
assert_eq!(out, "patatas 900 kg");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn odt_content_extracts_by_extension_fallback() {
|
||||
let content = r#"<?xml version="1.0"?>
|
||||
<office:document-content xmlns:office="urn:oasis:names:tc:opendocument:xmlns:office:1.0"
|
||||
xmlns:text="urn:oasis:names:tc:opendocument:xmlns:text:1.0">
|
||||
<office:body><office:text>
|
||||
<text:p>tortilla de patatas</text:p>
|
||||
</office:text></office:body>
|
||||
</office:document-content>"#;
|
||||
let bytes = build_zip(&[("content.xml", content)]);
|
||||
// Generic MIME — classification must fall back to the .odt extension.
|
||||
let out = text_of(extract(
|
||||
"receta.odt",
|
||||
"application/octet-stream",
|
||||
&bytes,
|
||||
4096,
|
||||
));
|
||||
assert_eq!(out, "tortilla de patatas");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn corrupt_pdf_fails_terminally_instead_of_panicking() {
|
||||
assert!(matches!(
|
||||
extract("roto.pdf", "application/pdf", b"definitely not a pdf", 4096),
|
||||
ExtractedText::Failed(_)
|
||||
));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user