diff --git a/src/common/di.rs b/src/common/di.rs index a31c0980..11fa4777 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1494,6 +1494,20 @@ impl AppServiceFactory { .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) .await; + // Its file-keyed twin: `ext-{file_id}.jpg` previews the user uploaded, + // which no copy path duplicates today. Separate job, separate keying — + // routing these into the content-keyed table would share one user's + // preview onto every file with identical content. + let _ = Arc::new( + crate::infrastructure::services::thumb_attached_import_service::ThumbAttachedImport::new( + std::path::Path::new(&self.storage_path).join(".thumbnails"), + core.dedup_service.clone(), + maintenance_pool.clone(), + ), + ) + .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) + .await; + // Third recoverable-run tenant. Iterates `storage.files` // and reports parent-folder-trashed cascade misses, // `missing_blob` (data-loss indicator — file references diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 8ff128d6..9ffaa836 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -57,6 +57,7 @@ pub mod session_liveness_gauges; pub mod share_unlock_cookie; pub mod smtp_email_sender; pub mod swappable_blob_backend; +pub mod thumb_attached_import_service; pub mod thumb_derived_import_service; pub mod thumbnail_service; #[cfg(test)] diff --git a/src/infrastructure/services/thumb_attached_import_service.rs b/src/infrastructure/services/thumb_attached_import_service.rs new file mode 100644 index 00000000..cb7235f7 --- /dev/null +++ b/src/infrastructure/services/thumb_attached_import_service.rs @@ -0,0 +1,375 @@ +//! `thumb_attached_import` — backfill `storage.file_attached_blobs` from the +//! `ext-{file_id}.jpg` sidecars that predate it. +//! +//! Second half of step 10's migration, and the twin of +//! `thumb_derived_import`. These are the thumbnails a *user* supplied — the +//! SPA's client-side generator, notably for PDFs, which have no server-side +//! render path at all. They live only as +//! `{thumbnails_root}/{size}/ext-{file_id}.jpg` on local disk. +//! +//! Until a row exists, a **copy of the file loses the preview**: the sidecar +//! is keyed by `file_id`, no copy path duplicates it, and the server silently +//! falls back to rendering from the source (or to nothing, for a PDF). That +//! is the bug `file_attached_blobs` closed for new uploads; this job closes +//! it for everything already on disk. +//! +//! ### File-keyed, and that is the whole point +//! +//! These bytes are **not** derivable from the file's content, so they must +//! never be content-keyed. Sharing one user's uploaded preview across every +//! file with identical content is the poisoning vector the table split +//! exists to prevent — see `docs/plan/derived-blobs.md`. `thumb_derived_import` +//! deliberately rejects `ext-` names for the same reason, and the two jobs +//! are separate so neither can drift into the other's keying. +//! +//! ### Idempotence needs care here +//! +//! Unlike the derived twin, `store_attached_blob` is `ON CONFLICT DO UPDATE`: +//! calling it for a row that already exists releases the previous reference +//! and takes a new one. Harmless once, but a job that did it on every run +//! would churn refcounts. So each file is skipped when a row is already +//! present, and the store is only reached on a genuine insert. +//! +//! ### Multi-instance caveat +//! +//! Sidecars are local, so this migrates only the instance it runs on. Phase 3 +//! must be gated on every instance reporting an empty tail. + +use std::path::PathBuf; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use sqlx::PgPool; +use tokio::fs; +use uuid::Uuid; + +use crate::application::ports::thumbnail_ports::ThumbnailSize; +use crate::infrastructure::scheduler::{ + JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, + RunStatus, record_or_log, +}; +use crate::infrastructure::services::dedup_service::DedupService; + +pub const THUMB_ATTACHED_IMPORT_JOB_NAME: &str = "thumb_attached_import"; + +/// Files handled between checkpoints — a read plus at most a blob write each. +const BATCH_SIZE: usize = 100; + +/// `uploaded_by` for imported rows. +/// +/// Disk records no uploader, and the column is deliberately `NOT NULL` with no +/// FK so provenance survives a user deletion. A sentinel says "imported, real +/// uploader unknown" honestly; inventing an owner — the file's `created_by`, +/// say — would fabricate provenance that could later be read as evidence an +/// Editor replaced someone's preview. +const IMPORTED_UPLOADER: Uuid = Uuid::nil(); + +pub struct ThumbAttachedImport { + thumbnails_root: PathBuf, + dedup: Arc, + pool: Arc, +} + +impl ThumbAttachedImport { + pub fn new(thumbnails_root: PathBuf, dedup: Arc, pool: Arc) -> Self { + Self { + thumbnails_root, + dedup, + pool, + } + } + + pub async fn register_recoverable_job( + self: Arc, + registry: &JobRegistry, + provider: &Arc, + ) -> Arc { + registry + .register_recoverable_job(self.clone(), provider.clone(), None) + .await; + self + } + + /// The file id an external sidecar names, or `None` when the file is not + /// one of ours. + /// + /// Requires a parseable UUID: the name is about to be used as a foreign + /// key, and a malformed one should be reported rather than fed to the + /// database. + fn file_id_from_sidecar_name(name: &str) -> Option { + let stem = name.strip_prefix("ext-")?.strip_suffix(".jpg")?; + Uuid::parse_str(stem).ok() + } + + /// Sorted external-sidecar filenames for one size directory. + /// + /// Sorted because the cursor resumes by skipping everything at or before + /// it, which only works over a stable order. + async fn sidecar_names(&self, size: ThumbnailSize) -> Vec { + let dir = self.thumbnails_root.join(size.dir_name()); + let Ok(mut entries) = fs::read_dir(&dir).await else { + return Vec::new(); + }; + let mut names = Vec::new(); + while let Ok(Some(entry)) = entries.next_entry().await { + if let Some(name) = entry.file_name().to_str() + && Self::file_id_from_sidecar_name(name).is_some() + { + names.push(name.to_string()); + } + } + names.sort(); + names + } + + /// Does the file still exist? Checked explicitly rather than letting the + /// foreign key reject the insert, so an orphaned sidecar is *counted* as + /// an orphan instead of surfacing as an opaque constraint error. + async fn file_exists(&self, file_id: Uuid) -> bool { + sqlx::query_scalar::<_, i64>("SELECT 1 FROM storage.files WHERE id = $1") + .bind(file_id) + .fetch_optional(self.pool.as_ref()) + .await + .ok() + .flatten() + .is_some() + } +} + +#[async_trait] +impl RecoverableJobHandler for ThumbAttachedImport { + fn name(&self) -> &str { + THUMB_ATTACHED_IMPORT_JOB_NAME + } + + async fn count_total(&self) -> Option { + let mut total = 0u64; + for size in ThumbnailSize::all() { + total += self.sidecar_names(*size).await.len() as u64; + } + Some(total) + } + + async fn run_resumable( + &self, + store: &dyn JobStore, + _args: &JobRunArgs, + resume_cursor: Option>, + ) -> RunOutcome { + // Cursor is `{size_dir}/{filename}`, matching thumb_derived_import: + // sizes walk in `ThumbnailSize::all()` order and names are sorted + // within each, so the pair totally orders the traversal. + let cursor: Option = match resume_cursor { + None => None, + Some(b) if b.is_empty() => None, + Some(b) => match String::from_utf8(b) { + Ok(s) => Some(s), + Err(e) => { + return RunOutcome::Failed { + message: format!("invalid cursor: not valid UTF-8: {e}"), + }; + } + }, + }; + + let mut imported = 0u64; + let mut already = 0u64; + let mut orphaned = 0u64; + let mut failed = 0u64; + let mut since_checkpoint = 0usize; + + for size in ThumbnailSize::all() { + let dir_name = size.dir_name().to_string(); + for name in self.sidecar_names(*size).await { + let position = format!("{dir_name}/{name}"); + + if let Some(c) = &cursor + && position.as_str() <= c.as_str() + { + continue; + } + + match store.status().await { + Ok(RunStatus::CancelRequested) => { + return RunOutcome::Paused { + cursor: position.into_bytes(), + }; + } + Ok(_) => {} + Err(e) => { + return RunOutcome::Failed { + message: format!("status poll: {e}"), + }; + } + } + + let Some(file_id) = Self::file_id_from_sidecar_name(&name) else { + continue; + }; + let file_id_str = file_id.to_string(); + + // Already mapped. Checked BEFORE storing, because + // `store_attached_blob` is ON CONFLICT DO UPDATE and would + // release then retake the reference on every run. + if self + .dedup + .find_attached_blob(&file_id_str, "preview", &dir_name) + .await + .is_some() + { + already += 1; + } else if !self.file_exists(file_id).await { + // The file is gone; the sidecar outlived it. Reported + // rather than deleted — this job imports, it does not + // reclaim, and a destructive default on a migration is + // exactly what `no silent auto-repair` forbids. + orphaned += 1; + record_or_log( + store, + THUMB_ATTACHED_IMPORT_JOB_NAME, + "attached_sidecar_orphan", + "anomaly", + None, + serde_json::json!({ + "path": position, + "file_id": file_id_str, + "note": "no storage.files row; sidecar left in place for the operator", + }), + ) + .await; + } else { + let path = self.thumbnails_root.join(&dir_name).join(&name); + match fs::read(&path).await { + Ok(data) => { + match self + .dedup + .store_attached_blob( + &file_id_str, + "preview", + &dir_name, + // store_external_thumbnail re-encodes to + // JPEG before writing, so the extension + // is authoritative here. + "image/jpeg", + Bytes::from(data), + IMPORTED_UPLOADER, + ) + .await + { + Ok(_) => imported += 1, + Err(e) => { + failed += 1; + record_or_log( + store, + THUMB_ATTACHED_IMPORT_JOB_NAME, + "attached_import_failed", + "anomaly", + None, + serde_json::json!({ + "path": position, + "file_id": file_id_str, + "error": format!("{e}"), + "note": "sidecar left in place; safe to re-run", + }), + ) + .await; + } + } + } + Err(e) => { + failed += 1; + record_or_log( + store, + THUMB_ATTACHED_IMPORT_JOB_NAME, + "attached_sidecar_unreadable", + "anomaly", + None, + serde_json::json!({ + "path": position, + "error": format!("{e}"), + }), + ) + .await; + } + } + } + + since_checkpoint += 1; + if since_checkpoint >= BATCH_SIZE { + if let Err(e) = store + .checkpoint(position.clone().into_bytes(), since_checkpoint as u64) + .await + { + return RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }; + } + since_checkpoint = 0; + } + } + } + + tracing::info!( + target: "oxicloud::dedup", + event = "thumb_attached_import.completed", + run_id = %store.run_id(), + imported = imported, + already_present = already, + orphaned = orphaned, + failed = failed, + "thumb_attached_import: {imported} imported, {already} already present, \ + {orphaned} orphaned, {failed} failed" + ); + + RunOutcome::completed() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const UUID: &str = "3f2b1c00-1111-2222-3333-444455556666"; + const HASH: &str = "0a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f9"; + + #[test] + fn accepts_an_external_sidecar_name() { + assert_eq!( + ThumbAttachedImport::file_id_from_sidecar_name(&format!("ext-{UUID}.jpg")), + Some(Uuid::parse_str(UUID).unwrap()) + ); + } + + /// The content-keyed sidecars belong to `thumb_derived_import`. Importing + /// one here would file-key bytes that are shared across every file with + /// the same content, so each such file would take its own reference to + /// content it does not own. + #[test] + fn rejects_content_keyed_and_malformed_names() { + for name in [ + format!("{HASH}.webp"), + format!("{HASH}.jpg"), + format!("ext-{UUID}.webp"), + format!("ext-{UUID}"), + "ext-not-a-uuid.jpg".to_string(), + format!("{UUID}.jpg"), + ] { + assert_eq!( + ThumbAttachedImport::file_id_from_sidecar_name(&name), + None, + "must not be imported as an attached preview: {name}" + ); + } + } + + /// The sentinel must be stable: rows carrying it are how an operator + /// tells an imported preview from one with real provenance. + #[test] + fn imported_uploader_is_the_nil_sentinel() { + assert_eq!( + IMPORTED_UPLOADER.to_string(), + "00000000-0000-0000-0000-000000000000" + ); + } +}