Initial commit

backend/src/workers/analyze.rs

@@ -0,0 +1,203 @@
+use std::{collections::HashSet, sync::Arc, time::Duration};
+
+use async_trait::async_trait;
+use diesel::prelude::*;
+use serde::Deserialize;
+use serde_json::{json, Map, Value};
+use tokio::task;
+use tracing::{error, warn};
+use uuid::Uuid;
+
+use super::ocr::{document_is_pdf, OCR_TEXT_ASSET_TYPE};
+use crate::{
+    jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT, JOB_GENERATE_OCR_TEXT, JOB_GENERATE_THUMBNAILS},
+    models::{Document, DocumentAsset, DocumentVersion},
+    schema::{document_assets, document_versions, documents},
+    state::AppState,
+};
+
+use super::{JobExecution, JobHandler};
+
+#[derive(Debug, Deserialize)]
+struct AnalyzePayload {
+    document_id: Uuid,
+    document_version_id: Uuid,
+    #[serde(default)]
+    force: bool,
+}
+
+pub struct AnalyzeDocumentJob;
+
+impl AnalyzeDocumentJob {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+#[async_trait]
+impl JobHandler for AnalyzeDocumentJob {
+    fn job_type(&self) -> &'static str {
+        JOB_ANALYZE_DOCUMENT
+    }
+
+    async fn handle(&self, state: Arc<AppState>, job: crate::models::Job) -> JobExecution {
+        let payload: AnalyzePayload = match serde_json::from_value(job.payload.clone()) {
+            Ok(payload) => payload,
+            Err(err) => {
+                return JobExecution::Failed {
+                    error: format!("invalid analyze payload: {err}"),
+                }
+            }
+        };
+
+        let state_clone = state.clone();
+        match task::spawn_blocking(move || analyze_document(state_clone, payload)).await {
+            Ok(Ok(execution)) => execution,
+            Ok(Err(err)) => {
+                warn!(job_id = %job.id, error = %err, "analyze job will retry");
+                JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: err,
+                }
+            }
+            Err(join_err) => {
+                error!(job_id = %job.id, error = %join_err, "analyze task panicked");
+                JobExecution::Retry {
+                    delay: Duration::from_secs(60),
+                    error: format!("worker panicked: {join_err}"),
+                }
+            }
+        }
+    }
+}
+
+fn analyze_document(state: Arc<AppState>, payload: AnalyzePayload) -> Result<JobExecution, String> {
+    let mut conn = state.db().map_err(|err| format!("{err:?}"))?;
+
+    let version: DocumentVersion = document_versions::table
+        .find(payload.document_version_id)
+        .first(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    if version.document_id != payload.document_id {
+        return Err("document/version mismatch".into());
+    }
+
+    let document: Document = documents::table
+        .find(payload.document_id)
+        .first(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    let (supported, reason) = determine_thumbnail_support(&document);
+    let ocr_supported = document_is_pdf(&document);
+
+    let existing_ocr: Option<DocumentAsset> = document_assets::table
+        .filter(document_assets::document_version_id.eq(payload.document_version_id))
+        .filter(document_assets::asset_type.eq(OCR_TEXT_ASSET_TYPE))
+        .first(&mut conn)
+        .optional()
+        .map_err(|err| format!("{err:?}"))?;
+
+    let skip_ocr = existing_ocr.is_some() && !payload.force;
+
+    let mut summary_map = match version.operations_summary {
+        Value::Object(map) => map,
+        _ => Map::new(),
+    };
+    summary_map.insert("thumbnail_supported".to_string(), Value::Bool(supported));
+    if let Some(reason) = reason {
+        summary_map.insert("thumbnail_reason".to_string(), Value::String(reason));
+    } else {
+        summary_map.remove("thumbnail_reason");
+    }
+
+    summary_map.insert("ocr_supported".to_string(), Value::Bool(ocr_supported));
+    if ocr_supported {
+        summary_map.remove("ocr_reason");
+    } else {
+        summary_map.insert(
+            "ocr_reason".to_string(),
+            Value::String("document is not a PDF".into()),
+        );
+    }
+
+    diesel::update(document_versions::table.find(version.id))
+        .set(document_versions::operations_summary.eq(Value::Object(summary_map)))
+        .execute(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    if supported {
+        let enqueue_result = enqueue_job(
+            &mut conn,
+            JOB_GENERATE_THUMBNAILS,
+            json!({
+                "document_id": payload.document_id,
+                "document_version_id": payload.document_version_id,
+                "force": payload.force,
+            }),
+            None,
+        );
+
+        if let Err(err) = enqueue_result {
+            return Err(err.to_string());
+        }
+    }
+
+    if ocr_supported && !skip_ocr {
+        let enqueue_result = enqueue_job(
+            &mut conn,
+            JOB_GENERATE_OCR_TEXT,
+            json!({
+                "document_id": payload.document_id,
+                "document_version_id": payload.document_version_id,
+                "force": payload.force,
+            }),
+            None,
+        );
+
+        if let Err(err) = enqueue_result {
+            return Err(err.to_string());
+        }
+    }
+
+    Ok(JobExecution::Success)
+}
+
+pub(crate) fn determine_thumbnail_support(document: &Document) -> (bool, Option<String>) {
+    let supported_mimes: HashSet<&'static str> = [
+        "image/jpeg",
+        "image/png",
+        "image/gif",
+        "image/tiff",
+        "image/bmp",
+        "image/webp",
+        "application/pdf",
+    ]
+    .into_iter()
+    .collect();
+
+    if let Some(ref content_type) = document.content_type {
+        if supported_mimes.contains(content_type.as_str()) {
+            return (true, None);
+        }
+    }
+
+    if let Some(ext) = document
+        .original_name
+        .rsplit('.')
+        .next()
+        .map(|ext| ext.to_ascii_lowercase())
+    {
+        let supported_exts = [
+            "jpg", "jpeg", "png", "gif", "tif", "tiff", "bmp", "webp", "pdf",
+        ];
+        if supported_exts.contains(&ext.as_str()) {
+            return (true, None);
+        }
+    }
+
+    (
+        false,
+        Some("content type not supported for thumbnails".into()),
+    )
+}

backend/src/workers/index.rs

@@ -0,0 +1,204 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use diesel::prelude::*;
+use reqwest::Client;
+use serde::Deserialize;
+use serde_json::json;
+use tokio::task;
+use tracing::{error, warn};
+use uuid::Uuid;
+
+use crate::{
+    jobs::JOB_INDEX_DOCUMENT_TEXT,
+    models::{Document, DocumentAsset, DocumentVersion},
+    schema::{document_assets, document_versions, documents},
+    state::AppState,
+};
+
+use super::{ocr::OCR_TEXT_ASSET_TYPE, JobExecution, JobHandler};
+
+#[derive(Debug, Deserialize)]
+struct IndexPayload {
+    document_id: Uuid,
+    document_version_id: Uuid,
+}
+
+pub struct IndexDocumentTextJob;
+
+impl IndexDocumentTextJob {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+#[async_trait]
+impl JobHandler for IndexDocumentTextJob {
+    fn job_type(&self) -> &'static str {
+        JOB_INDEX_DOCUMENT_TEXT
+    }
+
+    async fn handle(&self, state: Arc<AppState>, job: crate::models::Job) -> JobExecution {
+        let payload: IndexPayload = match serde_json::from_value(job.payload.clone()) {
+            Ok(payload) => payload,
+            Err(err) => {
+                return JobExecution::Failed {
+                    error: format!("invalid index payload: {err}"),
+                }
+            }
+        };
+
+        let quickwit_endpoint = match &state.config.quickwit_endpoint {
+            Some(endpoint) => endpoint.clone(),
+            None => {
+                warn!("quickwit endpoint missing; skipping indexing");
+                return JobExecution::Success;
+            }
+        };
+
+        let quickwit_index = match &state.config.quickwit_index {
+            Some(index) => index.clone(),
+            None => {
+                warn!("quickwit index missing; skipping indexing");
+                return JobExecution::Success;
+            }
+        };
+
+        let client = Client::new();
+
+        let state_clone = state.clone();
+        let context = match task::spawn_blocking(move || load_context(state_clone, &payload)).await
+        {
+            Ok(Ok(ctx)) => ctx,
+            Ok(Err(err)) => {
+                warn!(job_id = %job.id, error = %err, "index job will retry");
+                return JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: err,
+                };
+            }
+            Err(join_err) => {
+                error!(job_id = %job.id, error = %join_err, "index task panicked");
+                return JobExecution::Retry {
+                    delay: Duration::from_secs(60),
+                    error: format!("worker panicked: {join_err}"),
+                };
+            }
+        };
+
+        if context.text_asset.is_none() {
+            warn!(job_id = %job.id, "missing OCR text asset; failing indexing job");
+            return JobExecution::Failed {
+                error: "missing OCR text asset".into(),
+            };
+        }
+
+        let asset = context.text_asset.unwrap();
+        let text = match state.storage.get_object(&asset.s3_key).await {
+            Ok(bytes) => match String::from_utf8(bytes) {
+                Ok(text) => text,
+                Err(err) => {
+                    warn!(job_id = %job.id, error = %err, "ocr text not valid UTF-8");
+                    return JobExecution::Failed {
+                        error: "ocr text not valid UTF-8".into(),
+                    };
+                }
+            },
+            Err(err) => {
+                warn!(job_id = %job.id, error = %err, "failed to download ocr text");
+                return JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: err.to_string(),
+                };
+            }
+        };
+
+        if text.trim().is_empty() {
+            warn!(job_id = %job.id, "ocr text empty; skipping");
+            return JobExecution::Failed {
+                error: "ocr text empty".into(),
+            };
+        }
+
+        let client = client;
+        let url = format!(
+            "{}/api/v1/{}/ingest?commit=auto",
+            quickwit_endpoint, quickwit_index
+        );
+        let payload = json!({
+            "document_id": context.document.id,
+            "version_id": context.version.id,
+            "title": context.document.title.to_lowercase(),
+            "text": text.to_lowercase()
+        });
+
+        let body = serde_json::to_string(&payload).unwrap();
+
+        match client
+            .post(&url)
+            .header("content-type", "application/x-ndjson")
+            .body(format!("{}\n", body))
+            .send()
+            .await
+        {
+            Ok(response) => {
+                if response.status().is_success() {
+                    JobExecution::Success
+                } else {
+                    let status = response.status();
+                    let body = response.text().await.unwrap_or_default();
+                    warn!(job_id = %job.id, %status, %body, "quickwit ingest failed");
+                    JobExecution::Retry {
+                        delay: Duration::from_secs(30),
+                        error: format!("quickwit ingest failed with status {status}"),
+                    }
+                }
+            }
+            Err(err) => {
+                warn!(job_id = %job.id, error = %err, "quickwit request failed");
+                JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: err.to_string(),
+                }
+            }
+        }
+    }
+}
+
+struct IndexContext {
+    document: Document,
+    version: DocumentVersion,
+    text_asset: Option<DocumentAsset>,
+}
+
+fn load_context(state: Arc<AppState>, payload: &IndexPayload) -> Result<IndexContext, String> {
+    let mut conn = state.db().map_err(|err| format!("{err:?}"))?;
+
+    let version: DocumentVersion = document_versions::table
+        .find(payload.document_version_id)
+        .first(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    if version.document_id != payload.document_id {
+        return Err("document/version mismatch".into());
+    }
+
+    let document: Document = documents::table
+        .find(payload.document_id)
+        .first(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    let text_asset: Option<DocumentAsset> = document_assets::table
+        .filter(document_assets::document_version_id.eq(payload.document_version_id))
+        .filter(document_assets::asset_type.eq(OCR_TEXT_ASSET_TYPE))
+        .first(&mut conn)
+        .optional()
+        .map_err(|err| format!("{err:?}"))?;
+
+    Ok(IndexContext {
+        document,
+        version,
+        text_asset,
+    })
+}

backend/src/workers/mod.rs

@@ -0,0 +1,136 @@
+use std::{collections::HashMap, sync::Arc, time::Duration};
+
+use async_trait::async_trait;
+use tokio::time::sleep;
+use tracing::{error, info, warn};
+
+use crate::{
+    jobs::{mark_job_failed, mark_job_succeeded, reserve_job, retry_job_after, JobQueueError},
+    models::Job,
+    state::AppState,
+};
+
+pub mod analyze;
+pub mod index;
+pub mod ocr;
+pub mod thumbnails;
+
+#[derive(Debug)]
+pub enum JobExecution {
+    Success,
+    Retry { delay: Duration, error: String },
+    Failed { error: String },
+}
+
+#[async_trait]
+pub trait JobHandler: Send + Sync {
+    fn job_type(&self) -> &'static str;
+    async fn handle(&self, state: Arc<AppState>, job: Job) -> JobExecution;
+}
+
+pub struct Worker {
+    state: Arc<AppState>,
+    handlers: HashMap<&'static str, Arc<dyn JobHandler>>,
+    poll_interval: Duration,
+}
+
+impl Worker {
+    pub fn new(
+        state: Arc<AppState>,
+        handlers: Vec<Arc<dyn JobHandler>>,
+        poll_interval: Duration,
+    ) -> Self {
+        let map = handlers
+            .into_iter()
+            .map(|handler| (handler.job_type(), handler))
+            .collect();
+        Self {
+            state,
+            handlers: map,
+            poll_interval,
+        }
+    }
+
+    pub async fn run(&self) {
+        info!("worker started");
+        loop {
+            match self.tick().await {
+                Ok(true) => {}
+                Ok(false) => sleep(self.poll_interval).await,
+                Err(err) => {
+                    error!(error = %err, "worker tick failed");
+                    sleep(self.poll_interval).await;
+                }
+            }
+        }
+    }
+
+    async fn tick(&self) -> Result<bool, JobQueueError> {
+        let job_types: Vec<&str> = self.handlers.keys().copied().collect();
+        if job_types.is_empty() {
+            return Ok(false);
+        }
+
+        let mut conn = match self.state.db() {
+            Ok(conn) => conn,
+            Err(err) => {
+                error!(?err, "failed to obtain database connection in worker");
+                return Ok(false);
+            }
+        };
+
+        let job_opt = reserve_job(&mut conn, &job_types)?;
+        drop(conn);
+
+        if let Some(job) = job_opt {
+            if let Some(handler) = self.handlers.get(job.job_type.as_str()) {
+                let result = handler.handle(self.state.clone(), job.clone()).await;
+                match result {
+                    JobExecution::Success => {
+                        if let Ok(mut conn) = self.state.db() {
+                            mark_job_succeeded(&mut conn, job.id)?;
+                            info!(job_id = %job.id, job_type = %job.job_type, "job completed successfully");
+                        } else {
+                            error!("failed to mark job succeeded due to pool error");
+                        }
+                    }
+                    JobExecution::Retry { delay, error } => {
+                        warn!(job_id = %job.id, job_type = %job.job_type, %error, "job will retry");
+                        if let Ok(mut conn) = self.state.db() {
+                            retry_job_after(&mut conn, job.id, delay, &error)?;
+                        } else {
+                            error!("failed to requeue job for retry due to pool error");
+                        }
+                    }
+                    JobExecution::Failed { error } => {
+                        error!(job_id = %job.id, job_type = %job.job_type, %error, "job failed");
+                        if let Ok(mut conn) = self.state.db() {
+                            mark_job_failed(&mut conn, job.id, &error)?;
+                        } else {
+                            error!("failed to mark job failed due to pool error");
+                        }
+                    }
+                }
+            } else {
+                error!(job_type = %job.job_type, "no handler registered for job type");
+                if let Ok(mut conn) = self.state.db() {
+                    mark_job_failed(&mut conn, job.id, "no handler registered")?;
+                } else {
+                    error!("failed to mark job failed for missing handler due to pool error");
+                }
+            }
+            Ok(true)
+        } else {
+            Ok(false)
+        }
+    }
+}
+
+pub fn default_handlers() -> Vec<Arc<dyn JobHandler>> {
+    vec![
+        Arc::new(analyze::AnalyzeDocumentJob::new()),
+        Arc::new(thumbnails::GenerateThumbnailsJob::new()),
+        Arc::new(ocr::GenerateOcrTextJob::new()),
+        Arc::new(index::IndexDocumentTextJob::new()),
+    ]
+}

backend/src/workers/ocr.rs

@@ -0,0 +1,438 @@
+use std::{
+    fmt, fs,
+    io::{ErrorKind, Write},
+    process::Command,
+    sync::Arc,
+    time::Duration,
+};
+
+use async_trait::async_trait;
+use chrono::Utc;
+use diesel::{pg::upsert::excluded, prelude::*};
+use pdfium_render::prelude::*;
+use serde::Deserialize;
+use serde_json::json;
+use tempfile::NamedTempFile;
+use tokio::task;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use crate::{
+    jobs::{enqueue_job, JOB_GENERATE_OCR_TEXT, JOB_INDEX_DOCUMENT_TEXT},
+    models::{Document, DocumentAsset, DocumentVersion, NewDocumentAsset},
+    schema::{document_assets, document_versions, documents},
+    state::AppState,
+};
+
+use super::{JobExecution, JobHandler};
+
+pub const OCR_TEXT_ASSET_TYPE: &str = "ocr-text";
+const MIN_TEXT_LENGTH: usize = 50;
+
+#[derive(Clone, Debug, Deserialize)]
+struct OcrPayload {
+    document_id: Uuid,
+    document_version_id: Uuid,
+    #[serde(default)]
+    force: bool,
+}
+
+pub struct GenerateOcrTextJob;
+
+impl GenerateOcrTextJob {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+#[async_trait]
+impl JobHandler for GenerateOcrTextJob {
+    fn job_type(&self) -> &'static str {
+        JOB_GENERATE_OCR_TEXT
+    }
+
+    async fn handle(&self, state: Arc<AppState>, job: crate::models::Job) -> JobExecution {
+        let payload: OcrPayload = match serde_json::from_value(job.payload.clone()) {
+            Ok(payload) => payload,
+            Err(err) => {
+                return JobExecution::Failed {
+                    error: format!("invalid OCR payload: {err}"),
+                }
+            }
+        };
+
+        let state_clone = state.clone();
+        let payload_clone = payload.clone();
+        let context =
+            match task::spawn_blocking(move || load_ocr_context(state_clone, &payload_clone)).await
+            {
+                Ok(Ok(ctx)) => ctx,
+                Ok(Err(err)) => {
+                    warn!(job_id = %job.id, error = %err, "ocr job will retry");
+                    return JobExecution::Retry {
+                        delay: Duration::from_secs(30),
+                        error: err,
+                    };
+                }
+                Err(join_err) => {
+                    error!(job_id = %job.id, error = %join_err, "ocr task panicked");
+                    return JobExecution::Retry {
+                        delay: Duration::from_secs(60),
+                        error: format!("worker panicked: {join_err}"),
+                    };
+                }
+            };
+
+        if context.skip {
+            info!(job_id = %job.id, "ocr already present; skipping");
+            return JobExecution::Success;
+        }
+
+        let bytes = match state.storage.get_object(&context.version.s3_key).await {
+            Ok(bytes) => bytes,
+            Err(err) => {
+                warn!(job_id = %job.id, error = %err, "failed to fetch document for ocr");
+                return JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: err.to_string(),
+                };
+            }
+        };
+
+        let doc_meta = PdfDocumentMeta {
+            content_type: context.document.content_type.clone(),
+            original_name: context.document.original_name.clone(),
+        };
+
+        let generation =
+            match task::spawn_blocking(move || generate_ocr_text(&doc_meta, &bytes)).await {
+                Ok(result) => result,
+                Err(join_err) => {
+                    error!(job_id = %job.id, error = %join_err, "ocr text task panicked");
+                    return JobExecution::Retry {
+                        delay: Duration::from_secs(60),
+                        error: format!("worker panicked: {join_err}"),
+                    };
+                }
+            };
+
+        let Some(generation) = generation else {
+            warn!(job_id = %job.id, "no text extracted from document; failing job");
+            return JobExecution::Failed {
+                error: "no text extracted and OCR unavailable".into(),
+            };
+        };
+
+        let asset_id = context
+            .existing_asset
+            .as_ref()
+            .map(|asset| asset.id)
+            .unwrap_or_else(Uuid::new_v4);
+
+        let s3_key = format!(
+            "documents/{}/v{}/assets/{}/{}",
+            context.document.id, context.version.version_number, OCR_TEXT_ASSET_TYPE, asset_id
+        );
+
+        if let Err(err) = state
+            .storage
+            .put_object(
+                &s3_key,
+                generation.text.into_bytes(),
+                Some("text/plain".into()),
+                None,
+            )
+            .await
+        {
+            warn!(job_id = %job.id, error = %err, "failed to upload ocr text");
+            return JobExecution::Retry {
+                delay: Duration::from_secs(30),
+                error: err.to_string(),
+            };
+        }
+
+        let state_clone = state.clone();
+        match task::spawn_blocking(move || {
+            persist_ocr_metadata(state_clone, &context, asset_id, &s3_key, generation.source)
+        })
+        .await
+        {
+            Ok(Ok(())) => {
+                if state.config.quickwit_endpoint.is_some() && state.config.quickwit_index.is_some()
+                {
+                    if let Err(err) = enqueue_index_job(&state, &payload) {
+                        warn!(job_id = %job.id, error = %err, "failed to enqueue index job");
+                    }
+                }
+                JobExecution::Success
+            }
+            Ok(Err(err)) => {
+                warn!(job_id = %job.id, error = %err, "failed to persist ocr metadata");
+                JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: err,
+                }
+            }
+            Err(join_err) => {
+                error!(job_id = %job.id, error = %join_err, "ocr metadata task panicked");
+                JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: format!("metadata update panic: {join_err}"),
+                }
+            }
+        }
+    }
+}
+
+struct PdfDocumentMeta {
+    content_type: Option<String>,
+    original_name: String,
+}
+
+struct OcrContext {
+    document: Document,
+    version: DocumentVersion,
+    existing_asset: Option<DocumentAsset>,
+    skip: bool,
+}
+
+struct OcrGeneration {
+    text: String,
+    source: &'static str,
+}
+
+fn load_ocr_context(state: Arc<AppState>, payload: &OcrPayload) -> Result<OcrContext, String> {
+    let mut conn = state.db().map_err(|err| format!("{err:?}"))?;
+
+    let version: DocumentVersion = document_versions::table
+        .find(payload.document_version_id)
+        .first(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    if version.document_id != payload.document_id {
+        return Err("document/version mismatch".into());
+    }
+
+    let document: Document = documents::table
+        .find(payload.document_id)
+        .first(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    let existing: Option<DocumentAsset> = document_assets::table
+        .filter(document_assets::document_version_id.eq(payload.document_version_id))
+        .filter(document_assets::asset_type.eq(OCR_TEXT_ASSET_TYPE))
+        .first(&mut conn)
+        .optional()
+        .map_err(|err| format!("{err:?}"))?;
+
+    let is_pdf = document_is_pdf(&document);
+    if !is_pdf {
+        return Ok(OcrContext {
+            document,
+            version,
+            existing_asset: existing,
+            skip: true,
+        });
+    }
+
+    let skip = existing.is_some() && !payload.force;
+
+    Ok(OcrContext {
+        document,
+        version,
+        existing_asset: existing,
+        skip,
+    })
+}
+
+fn generate_ocr_text(meta: &PdfDocumentMeta, bytes: &[u8]) -> Option<OcrGeneration> {
+    if !document_meta_is_pdf(meta) {
+        return None;
+    }
+
+    if let Ok(text) = extract_pdf_text(bytes) {
+        if text.trim().chars().count() >= MIN_TEXT_LENGTH {
+            return Some(OcrGeneration {
+                text,
+                source: "pdf-text",
+            });
+        }
+    }
+
+    match run_ocr(bytes) {
+        Ok(Some(text)) => Some(OcrGeneration {
+            text,
+            source: "ocr",
+        }),
+        Ok(None) => None,
+        Err(OcrError::BinaryMissing) => {
+            warn!("ocrmypdf not installed; cannot perform OCR");
+            None
+        }
+        Err(err) => {
+            warn!(error = ?err, "ocr command failed");
+            None
+        }
+    }
+}
+
+fn extract_pdf_text(bytes: &[u8]) -> Result<String, String> {
+    let pdfium = Pdfium::default();
+    let document = pdfium
+        .load_pdf_from_byte_slice(bytes, None)
+        .map_err(|err| format!("load pdf: {err}"))?;
+
+    let mut combined = String::new();
+    let pages = document.pages();
+    for page_index in 0..pages.len() {
+        let page = pages
+            .get(page_index)
+            .map_err(|err| format!("load page {page_index}: {err}"))?;
+        if let Ok(page_text) = page.text() {
+            for segment in page_text.segments().iter() {
+                combined.push_str(&segment.text());
+                combined.push('\n');
+            }
+        };
+    }
+
+    Ok(combined)
+}
+
+#[derive(Debug)]
+enum OcrError {
+    BinaryMissing,
+    Failed(String),
+}
+
+impl fmt::Display for OcrError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            OcrError::BinaryMissing => write!(f, "ocrmypdf binary not found"),
+            OcrError::Failed(msg) => write!(f, "ocr failed: {msg}"),
+        }
+    }
+}
+
+fn run_ocr(bytes: &[u8]) -> Result<Option<String>, OcrError> {
+    let mut input = NamedTempFile::new().map_err(|err| OcrError::Failed(err.to_string()))?;
+    input
+        .write_all(bytes)
+        .map_err(|err| OcrError::Failed(err.to_string()))?;
+    input
+        .flush()
+        .map_err(|err| OcrError::Failed(err.to_string()))?;
+
+    let output_pdf = NamedTempFile::new().map_err(|err| OcrError::Failed(err.to_string()))?;
+    let sidecar = NamedTempFile::new().map_err(|err| OcrError::Failed(err.to_string()))?;
+
+    let status = Command::new("ocrmypdf")
+        .arg("--sidecar")
+        .arg(sidecar.path())
+        .arg("--skip-text")
+        .arg(input.path())
+        .arg(output_pdf.path())
+        .output();
+
+    match status {
+        Ok(output) => {
+            if !output.status.success() {
+                return Err(OcrError::Failed(format!(
+                    "ocrmypdf failed: exit={} stderr={}",
+                    output.status,
+                    String::from_utf8_lossy(&output.stderr)
+                )));
+            }
+
+            let text = fs::read_to_string(sidecar.path())
+                .map_err(|err| OcrError::Failed(err.to_string()))?;
+            if text.trim().chars().count() >= MIN_TEXT_LENGTH {
+                Ok(Some(text))
+            } else {
+                Ok(None)
+            }
+        }
+        Err(err) => {
+            if err.kind() == ErrorKind::NotFound {
+                Err(OcrError::BinaryMissing)
+            } else {
+                Err(OcrError::Failed(err.to_string()))
+            }
+        }
+    }
+}
+
+fn persist_ocr_metadata(
+    state: Arc<AppState>,
+    context: &OcrContext,
+    asset_id: Uuid,
+    s3_key: &str,
+    source: &'static str,
+) -> Result<(), String> {
+    let mut conn = state.db().map_err(|err| format!("{err:?}"))?;
+
+    let new_asset = NewDocumentAsset {
+        id: asset_id,
+        document_version_id: context.version.id,
+        asset_type: OCR_TEXT_ASSET_TYPE.to_string(),
+        s3_key: s3_key.to_string(),
+        mime_type: "text/plain".to_string(),
+        metadata: json!({
+            "generated_at": Utc::now().to_rfc3339(),
+            "source": source,
+        }),
+    };
+
+    diesel::insert_into(document_assets::table)
+        .values(&new_asset)
+        .on_conflict((
+            document_assets::document_version_id,
+            document_assets::asset_type,
+        ))
+        .do_update()
+        .set((
+            document_assets::s3_key.eq(excluded(document_assets::s3_key)),
+            document_assets::mime_type.eq(excluded(document_assets::mime_type)),
+            document_assets::metadata.eq(excluded(document_assets::metadata)),
+        ))
+        .execute(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    Ok(())
+}
+
+fn enqueue_index_job(state: &AppState, payload: &OcrPayload) -> Result<(), String> {
+    let mut conn = state.db().map_err(|err| format!("{err:?}"))?;
+    enqueue_job(
+        &mut conn,
+        JOB_INDEX_DOCUMENT_TEXT,
+        json!({
+            "document_id": payload.document_id,
+            "document_version_id": payload.document_version_id,
+        }),
+        None,
+    )
+    .map(|_| ())
+    .map_err(|err| err.to_string())
+}
+
+pub fn document_is_pdf(document: &Document) -> bool {
+    document_meta_is_pdf(&PdfDocumentMeta {
+        content_type: document.content_type.clone(),
+        original_name: document.original_name.clone(),
+    })
+}
+
+fn document_meta_is_pdf(meta: &PdfDocumentMeta) -> bool {
+    if let Some(content_type) = &meta.content_type {
+        if content_type.eq_ignore_ascii_case("application/pdf") {
+            return true;
+        }
+    }
+
+    meta.original_name
+        .rsplit('.')
+        .next()
+        .map(|ext| ext.eq_ignore_ascii_case("pdf"))
+        .unwrap_or(false)
+}

backend/src/workers/thumbnails.rs

@@ -0,0 +1,535 @@
+use std::{convert::TryInto, io::Cursor, panic, sync::Arc, time::Duration};
+
+use async_trait::async_trait;
+use chrono::Utc;
+use diesel::{pg::upsert::excluded, prelude::*};
+use image::{GenericImageView, ImageFormat, ImageReader};
+use pdfium_render::prelude::*;
+use serde::Deserialize;
+use serde_json::{json, Map, Value};
+use tokio::task;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use crate::{
+    jobs::JOB_GENERATE_THUMBNAILS,
+    models::{Document, DocumentAsset, DocumentVersion, NewDocumentAsset},
+    schema::{document_assets, document_versions, documents},
+    state::AppState,
+};
+
+use super::{analyze::determine_thumbnail_support, JobExecution, JobHandler};
+
+const THUMBNAIL_WIDTH: u32 = 512;
+const THUMBNAIL_HEIGHT: u32 = 512;
+const PREVIEW_WIDTH: u32 = THUMBNAIL_WIDTH * 4;
+const PREVIEW_HEIGHT: u32 = THUMBNAIL_HEIGHT * 4;
+const THUMBNAIL_ASSET_TYPE: &str = "thumbnail";
+const PREVIEW_ASSET_TYPE: &str = "preview";
+
+#[derive(Debug, Deserialize)]
+struct ThumbnailPayload {
+    document_id: Uuid,
+    document_version_id: Uuid,
+    #[serde(default)]
+    force: bool,
+}
+
+pub struct GenerateThumbnailsJob;
+
+impl GenerateThumbnailsJob {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+#[async_trait]
+impl JobHandler for GenerateThumbnailsJob {
+    fn job_type(&self) -> &'static str {
+        JOB_GENERATE_THUMBNAILS
+    }
+
+    async fn handle(&self, state: Arc<AppState>, job: crate::models::Job) -> JobExecution {
+        let payload: ThumbnailPayload = match serde_json::from_value(job.payload.clone()) {
+            Ok(p) => p,
+            Err(err) => {
+                return JobExecution::Failed {
+                    error: format!("invalid thumbnail payload: {err}"),
+                }
+            }
+        };
+
+        let state_clone = state.clone();
+        let initial =
+            match task::spawn_blocking(move || load_thumbnail_context(state_clone, &payload)).await
+            {
+                Ok(Ok(ctx)) => ctx,
+                Ok(Err(err)) => {
+                    warn!(job_id = %job.id, error = %err, "thumbnail job will retry");
+                    return JobExecution::Retry {
+                        delay: Duration::from_secs(30),
+                        error: err,
+                    };
+                }
+                Err(join_err) => {
+                    error!(job_id = %job.id, error = %join_err, "thumbnail task panicked");
+                    return JobExecution::Retry {
+                        delay: Duration::from_secs(60),
+                        error: format!("worker panicked: {join_err}"),
+                    };
+                }
+            };
+
+        if initial.skip {
+            info!(job_id = %job.id, "thumbnails already exist; skipping");
+            return JobExecution::Success;
+        }
+
+        let bytes = match state.storage.get_object(&initial.version.s3_key).await {
+            Ok(bytes) => bytes,
+            Err(err) => {
+                warn!(job_id = %job.id, error = %err, "thumbnail fetch failed; will retry");
+                return JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: err.to_string(),
+                };
+            }
+        };
+
+        let generation = match generate_preview_and_thumbnail(&initial.document, &bytes) {
+            Ok(result) => result,
+            Err(err) => {
+                return JobExecution::Failed { error: err };
+            }
+        };
+
+        if let Some(page_count) = generation.page_count {
+            let state_clone = state.clone();
+            let document_id = initial.document.id;
+            let version_id = initial.version.id;
+            match task::spawn_blocking(move || {
+                persist_document_page_count(state_clone, document_id, version_id, page_count)
+            })
+            .await
+            {
+                Ok(Ok(())) => {}
+                Ok(Err(err)) => {
+                    warn!(
+                        job_id = %job.id,
+                        document_id = %document_id,
+                        version_id = %version_id,
+                        error = %err,
+                        "failed to update document page count metadata; retrying"
+                    );
+                    return JobExecution::Retry {
+                        delay: Duration::from_secs(30),
+                        error: err,
+                    };
+                }
+                Err(join_err) => {
+                    error!(
+                        job_id = %job.id,
+                        document_id = %document_id,
+                        version_id = %version_id,
+                        error = %join_err,
+                        "page count metadata task panicked"
+                    );
+                    return JobExecution::Retry {
+                        delay: Duration::from_secs(60),
+                        error: format!("metadata panic: {join_err}"),
+                    };
+                }
+            }
+        }
+
+        let thumbnail_asset_id = initial
+            .existing_thumbnail
+            .as_ref()
+            .map(|asset| asset.id)
+            .unwrap_or_else(Uuid::new_v4);
+        let thumbnail_s3_key = format!(
+            "documents/{}/v{}/assets/{}/{}",
+            initial.document.id,
+            initial.version.version_number,
+            THUMBNAIL_ASSET_TYPE,
+            thumbnail_asset_id
+        );
+
+        let preview_asset_id = initial
+            .existing_preview
+            .as_ref()
+            .map(|asset| asset.id)
+            .unwrap_or_else(Uuid::new_v4);
+        let preview_s3_key = format!(
+            "documents/{}/v{}/assets/{}/{}",
+            initial.document.id,
+            initial.version.version_number,
+            PREVIEW_ASSET_TYPE,
+            preview_asset_id
+        );
+
+        if let Err(err) = state
+            .storage
+            .put_object(
+                &preview_s3_key,
+                generation.preview.image_bytes.clone(),
+                Some("image/png".into()),
+                None,
+            )
+            .await
+        {
+            warn!(job_id = %job.id, error = %err, "failed to upload preview; retrying");
+            return JobExecution::Retry {
+                delay: Duration::from_secs(30),
+                error: err.to_string(),
+            };
+        }
+
+        if let Err(err) = state
+            .storage
+            .put_object(
+                &thumbnail_s3_key,
+                generation.thumbnail.image_bytes.clone(),
+                Some("image/png".into()),
+                None,
+            )
+            .await
+        {
+            warn!(job_id = %job.id, error = %err, "failed to upload thumbnail; retrying");
+            return JobExecution::Retry {
+                delay: Duration::from_secs(30),
+                error: err.to_string(),
+            };
+        }
+
+        let state_clone = state.clone();
+        match task::spawn_blocking(move || {
+            persist_assets_metadata(
+                state_clone,
+                &initial,
+                &[
+                    AssetPersistence {
+                        asset_type: PREVIEW_ASSET_TYPE,
+                        asset_id: preview_asset_id,
+                        s3_key: &preview_s3_key,
+                        generated: &generation.preview,
+                    },
+                    AssetPersistence {
+                        asset_type: THUMBNAIL_ASSET_TYPE,
+                        asset_id: thumbnail_asset_id,
+                        s3_key: &thumbnail_s3_key,
+                        generated: &generation.thumbnail,
+                    },
+                ],
+            )
+        })
+        .await
+        {
+            Ok(Ok(())) => {}
+            Ok(Err(err)) => {
+                warn!(job_id = %job.id, error = %err, "failed to persist thumbnail metadata; retrying");
+                return JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: err,
+                };
+            }
+            Err(join_err) => {
+                error!(job_id = %job.id, error = %join_err, "thumbnail metadata update panicked");
+                return JobExecution::Retry {
+                    delay: Duration::from_secs(30),
+                    error: format!("metadata update panic: {join_err}"),
+                };
+            }
+        }
+
+        JobExecution::Success
+    }
+}
+
+struct ThumbnailContext {
+    document: Document,
+    version: DocumentVersion,
+    existing_thumbnail: Option<DocumentAsset>,
+    existing_preview: Option<DocumentAsset>,
+    skip: bool,
+}
+
+struct GeneratedImage {
+    image_bytes: Vec<u8>,
+    width: Option<i32>,
+    height: Option<i32>,
+}
+
+struct GeneratedAssets {
+    thumbnail: GeneratedImage,
+    preview: GeneratedImage,
+    page_count: Option<u32>,
+}
+
+struct AssetPersistence<'a> {
+    asset_type: &'static str,
+    asset_id: Uuid,
+    s3_key: &'a str,
+    generated: &'a GeneratedImage,
+}
+
+fn load_thumbnail_context(
+    state: Arc<AppState>,
+    payload: &ThumbnailPayload,
+) -> Result<ThumbnailContext, String> {
+    let mut conn = state.db().map_err(|err| format!("{err:?}"))?;
+
+    let version: DocumentVersion = document_versions::table
+        .find(payload.document_version_id)
+        .first(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    if version.document_id != payload.document_id {
+        return Err("document/version mismatch".into());
+    }
+
+    let document: Document = documents::table
+        .find(payload.document_id)
+        .first(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    let existing_assets: Vec<DocumentAsset> = document_assets::table
+        .filter(document_assets::document_version_id.eq(payload.document_version_id))
+        .filter(document_assets::asset_type.eq_any(vec![
+            THUMBNAIL_ASSET_TYPE.to_string(),
+            PREVIEW_ASSET_TYPE.to_string(),
+        ]))
+        .load(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    let mut existing_thumbnail = None;
+    let mut existing_preview = None;
+    for asset in existing_assets {
+        match asset.asset_type.as_str() {
+            THUMBNAIL_ASSET_TYPE => existing_thumbnail = Some(asset),
+            PREVIEW_ASSET_TYPE => existing_preview = Some(asset),
+            _ => {}
+        }
+    }
+
+    let (supported, _) = determine_thumbnail_support(&document);
+    if !supported {
+        return Err("thumbnail generation not supported for this document".into());
+    }
+
+    let skip = existing_thumbnail.is_some() && existing_preview.is_some() && !payload.force;
+
+    Ok(ThumbnailContext {
+        document,
+        version,
+        existing_thumbnail,
+        existing_preview,
+        skip,
+    })
+}
+
+fn generate_preview_and_thumbnail(
+    document: &Document,
+    bytes: &[u8],
+) -> Result<GeneratedAssets, String> {
+    let is_pdf = document
+        .content_type
+        .as_deref()
+        .map(|mime| mime == "application/pdf")
+        .unwrap_or_else(|| {
+            document
+                .original_name
+                .rsplit('.')
+                .next()
+                .map(|ext| ext.eq_ignore_ascii_case("pdf"))
+                .unwrap_or(false)
+        });
+
+    if is_pdf {
+        let pdf_assets = generate_pdf_assets(bytes)?;
+        Ok(GeneratedAssets {
+            preview: pdf_assets.preview,
+            thumbnail: pdf_assets.thumbnail,
+            page_count: Some(pdf_assets.page_count),
+        })
+    } else {
+        let (preview, thumbnail) = generate_image_assets(bytes)?;
+        Ok(GeneratedAssets {
+            preview,
+            thumbnail,
+            page_count: None,
+        })
+    }
+}
+
+fn generate_image_assets(bytes: &[u8]) -> Result<(GeneratedImage, GeneratedImage), String> {
+    let reader = ImageReader::new(Cursor::new(bytes))
+        .with_guessed_format()
+        .map_err(|err| err.to_string())?;
+    let image = reader.decode().map_err(|err| err.to_string())?;
+
+    let preview_image = if image.width() > PREVIEW_WIDTH || image.height() > PREVIEW_HEIGHT {
+        image.thumbnail(PREVIEW_WIDTH, PREVIEW_HEIGHT)
+    } else {
+        image.clone()
+    };
+
+    let thumbnail_image =
+        if preview_image.width() > THUMBNAIL_WIDTH || preview_image.height() > THUMBNAIL_HEIGHT {
+            preview_image.thumbnail(THUMBNAIL_WIDTH, THUMBNAIL_HEIGHT)
+        } else {
+            preview_image.clone()
+        };
+
+    let preview = encode_dynamic_image(preview_image)?;
+    let thumbnail = encode_dynamic_image(thumbnail_image)?;
+
+    Ok((preview, thumbnail))
+}
+
+struct PdfGeneratedAssets {
+    preview: GeneratedImage,
+    thumbnail: GeneratedImage,
+    page_count: u32,
+}
+
+fn generate_pdf_assets(bytes: &[u8]) -> Result<PdfGeneratedAssets, String> {
+    let pdfium = panic::catch_unwind(|| Pdfium::default())
+        .map_err(|_| "failed to initialize PDFium".to_string())?;
+
+    let document = pdfium
+        .load_pdf_from_byte_slice(bytes, None)
+        .map_err(|err| format!("load pdf: {err}"))?;
+
+    let pages = document.pages();
+    let total_pages = pages.len();
+
+    let page = pages
+        .get(0)
+        .map_err(|err| format!("load first page: {err}"))?;
+
+    let render_config = PdfRenderConfig::new()
+        .set_target_width(PREVIEW_WIDTH as i32)
+        .set_maximum_height(PREVIEW_HEIGHT as i32)
+        .render_form_data(true)
+        .rotate_if_landscape(PdfPageRenderRotation::None, true);
+
+    let bitmap = page
+        .render_with_config(&render_config)
+        .map_err(|err| format!("render pdf page: {err}"))?;
+
+    let preview_buffer = bitmap.as_image().to_rgb8();
+    let preview_image = image::DynamicImage::ImageRgb8(preview_buffer);
+
+    let thumbnail_image =
+        if preview_image.width() > THUMBNAIL_WIDTH || preview_image.height() > THUMBNAIL_HEIGHT {
+            preview_image.thumbnail(THUMBNAIL_WIDTH, THUMBNAIL_HEIGHT)
+        } else {
+            preview_image.clone()
+        };
+
+    let preview = encode_dynamic_image(preview_image)?;
+    let thumbnail = encode_dynamic_image(thumbnail_image)?;
+
+    let page_count: u32 = total_pages
+        .try_into()
+        .map_err(|_| "page count exceeds supported range".to_string())?;
+
+    Ok(PdfGeneratedAssets {
+        preview,
+        thumbnail,
+        page_count,
+    })
+}
+
+fn encode_dynamic_image(image: image::DynamicImage) -> Result<GeneratedImage, String> {
+    let (width, height) = image.dimensions();
+    let mut cursor = Cursor::new(Vec::new());
+    image
+        .write_to(&mut cursor, ImageFormat::Png)
+        .map_err(|err| err.to_string())?;
+    Ok(GeneratedImage {
+        image_bytes: cursor.into_inner(),
+        width: Some(width as i32),
+        height: Some(height as i32),
+    })
+}
+
+fn persist_assets_metadata(
+    state: Arc<AppState>,
+    context: &ThumbnailContext,
+    assets: &[AssetPersistence<'_>],
+) -> Result<(), String> {
+    let mut conn = state.db().map_err(|err| format!("{err:?}"))?;
+
+    for asset in assets {
+        let new_asset = NewDocumentAsset {
+            id: asset.asset_id,
+            document_version_id: context.version.id,
+            asset_type: asset.asset_type.to_string(),
+            s3_key: asset.s3_key.to_string(),
+            mime_type: "image/png".to_string(),
+            metadata: json!({
+                "generated_at": Utc::now().to_rfc3339(),
+                "width": asset.generated.width,
+                "height": asset.generated.height,
+            }),
+        };
+
+        diesel::insert_into(document_assets::table)
+            .values(&new_asset)
+            .on_conflict((
+                document_assets::document_version_id,
+                document_assets::asset_type,
+            ))
+            .do_update()
+            .set((
+                document_assets::s3_key.eq(excluded(document_assets::s3_key)),
+                document_assets::mime_type.eq(excluded(document_assets::mime_type)),
+                document_assets::metadata.eq(excluded(document_assets::metadata)),
+            ))
+            .execute(&mut conn)
+            .map_err(|err| format!("{err:?}"))?;
+    }
+
+    Ok(())
+}
+
+fn persist_document_page_count(
+    state: Arc<AppState>,
+    document_id: Uuid,
+    document_version_id: Uuid,
+    page_count: u32,
+) -> Result<(), String> {
+    let mut conn = state.db().map_err(|err| format!("{err:?}"))?;
+
+    let existing_metadata: Value = document_versions::table
+        .filter(document_versions::id.eq(document_version_id))
+        .filter(document_versions::document_id.eq(document_id))
+        .select(document_versions::metadata)
+        .first(&mut conn)
+        .map_err(|err| format!("{err:?}"))?;
+
+    let updated = match existing_metadata {
+        Value::Object(mut map) => {
+            map.insert("page_count".to_string(), Value::from(page_count));
+            Value::Object(map)
+        }
+        _ => {
+            let mut map = Map::new();
+            map.insert("page_count".to_string(), Value::from(page_count));
+            Value::Object(map)
+        }
+    };
+
+    diesel::update(
+        document_versions::table
+            .filter(document_versions::id.eq(document_version_id))
+            .filter(document_versions::document_id.eq(document_id)),
+    )
+    .set(document_versions::metadata.eq(updated))
+    .execute(&mut conn)
+    .map_err(|err| format!("{err:?}"))?;
+
+    Ok(())
+}