feat(storage): thumb_derived_import — backfill the derived tier from sidecars

First half of step 10. Every server-rendered thumbnail written before
content_derived_blobs existed lives only as
{thumbnails_root}/{size}/{hash}.webp — local-disk state that another
instance cannot see, a backend migration does not carry, and no
consistency job covers. This walks those files into the blob store and
records the mapping, so the derived tier can become authoritative and
the sidecar can be deleted.

A registered JobRegistry tenant rather than a script: the volume is
unbounded, so it needs a cursor, resume, cooperative cancel and run
history, and an operator needs somewhere to watch it. Cursor is
{size_dir}/{filename} over a sorted walk, which totally orders the
traversal.

Idempotent by construction — each file is skipped when a row already
exists, and store_derived_blob is ON CONFLICT DO NOTHING with
release-on-conflict beneath it, so re-runs cannot inflate refcounts.
Re-running is the expected operator behaviour, since Phase 3 (deleting
the sidecars) is gated on a run reporting zero imported.

hash_from_sidecar_name deliberately rejects ext-{file_id}.jpg. Those
bytes are user-supplied and file-keyed; importing them here would
content-key them and share one user's uploaded preview onto every file
with identical content. They belong to thumb_attached_import. Both the
accept and the reject set are under test.

Unreadable files and store failures record a finding and continue: a
sidecar removed by a concurrent GC unlink between listing and read is
expected, not fatal, and the file is left in place for the next run.

Registered unconditionally rather than behind a flag — a migration
nobody can find is a migration nobody runs.

src/common/di.rs

@@ -1476,6 +1476,24 @@ impl AppServiceFactory {
         .register_recoverable_job(&core.job_registry, &job_store_provider_dyn)
         .await;
 
+        // Step 10 migration tenant: backfills `content_derived_blobs` from
+        // the on-disk thumbnail sidecars that predate it. Idempotent, so it
+        // is safe to trigger repeatedly — Phase 3 (deleting the sidecars) is
+        // gated on a run reporting zero imported. Registered unconditionally
+        // rather than behind a flag: a migration nobody can find is a
+        // migration nobody runs.
+        //
+        // `.thumbnails` lives under the storage path, matching
+        // `ThumbnailService::new(&self.storage_path, …)` above.
+        let _ = Arc::new(
+            crate::infrastructure::services::thumb_derived_import_service::ThumbDerivedImport::new(
+                std::path::Path::new(&self.storage_path).join(".thumbnails"),
+                core.dedup_service.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

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_derived_import_service;
 pub mod thumbnail_service;
 #[cfg(test)]
 mod thumbnail_service_test;

src/infrastructure/services/thumb_derived_import_service.rs

@@ -0,0 +1,313 @@
+//! `thumb_derived_import` — backfill `storage.content_derived_blobs` from the
+//! on-disk thumbnail sidecars that predate it.
+//!
+//! Step 10 of `docs/plan/derived-blobs.md`. Every server-rendered thumbnail
+//! written before `content_derived_blobs` existed lives only as
+//! `{thumbnails_root}/{size}/{hash}.webp`. That is local-disk state: another
+//! instance cannot see it, a backend migration does not carry it, and no
+//! consistency job covers it. This job moves those bytes into the blob store
+//! and records the mapping, after which the derived tier can become
+//! authoritative and the sidecar can be deleted.
+//!
+//! **Thumbnails only, and that is permanent.** The table also holds
+//! `kind = 'transcode'`, but transcoding lands *after* this migration, so
+//! transcodes are born into the table and never pass through a sidecar era.
+//! This job will not grow a transcode arm.
+//!
+//! ### Idempotent by construction
+//!
+//! Each file is skipped when a row already exists for its
+//! `(source_hash, 'thumbnail', variant)`, and `store_derived_blob` is
+//! `ON CONFLICT DO NOTHING` with a release-on-conflict underneath, so a
+//! re-run cannot inflate refcounts. Re-running is the expected operator
+//! behaviour — Phase 3 (deleting the sidecars) is gated on a run reporting
+//! zero imported.
+//!
+//! ### Multi-instance caveat
+//!
+//! Sidecars are local. Running this on one instance migrates only that
+//! instance's files, so Phase 3 must be gated on *every* instance reporting
+//! an empty tail. The run history does not aggregate across instances; that
+//! remains an operator responsibility.
+
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use tokio::fs;
+
+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_DERIVED_IMPORT_JOB_NAME: &str = "thumb_derived_import";
+
+/// Files handled between checkpoints. Each one is a read plus (at most) a
+/// blob write, so this is deliberately smaller than a pure-DB sweep's page.
+const BATCH_SIZE: usize = 100;
+
+pub struct ThumbDerivedImport {
+    thumbnails_root: PathBuf,
+    dedup: Arc<DedupService>,
+}
+
+impl ThumbDerivedImport {
+    pub fn new(thumbnails_root: PathBuf, dedup: Arc<DedupService>) -> Self {
+        Self {
+            thumbnails_root,
+            dedup,
+        }
+    }
+
+    pub async fn register_recoverable_job(
+        self: Arc<Self>,
+        registry: &JobRegistry,
+        provider: &Arc<dyn JobStoreProvider>,
+    ) -> Arc<Self> {
+        registry
+            .register_recoverable_job(self.clone(), provider.clone(), None)
+            .await;
+        self
+    }
+
+    /// The hash a sidecar filename names, or `None` when the file is not one.
+    ///
+    /// Strict, and deliberately rejects `ext-{file_id}.jpg`: those are
+    /// user-supplied, file-keyed bytes. Importing them here would content-key
+    /// them and share one user's uploaded preview onto every file with
+    /// identical content — the poisoning `file_attached_blobs` exists to
+    /// prevent. They belong to `thumb_attached_import`.
+    fn hash_from_sidecar_name(name: &str) -> Option<&str> {
+        let stem = name.strip_suffix(".webp")?;
+        if stem.len() != 64 || !stem.chars().all(|c| c.is_ascii_hexdigit()) {
+            return None;
+        }
+        Some(stem)
+    }
+
+    /// Sorted sidecar filenames for one size directory.
+    ///
+    /// Sorted so the cursor is meaningful: resume skips everything at or
+    /// before it, which only works over a stable order.
+    async fn sidecar_names(&self, size: ThumbnailSize) -> Vec<String> {
+        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::hash_from_sidecar_name(name).is_some()
+            {
+                names.push(name.to_string());
+            }
+        }
+        names.sort();
+        names
+    }
+}
+
+#[async_trait]
+impl RecoverableJobHandler for ThumbDerivedImport {
+    fn name(&self) -> &str {
+        THUMB_DERIVED_IMPORT_JOB_NAME
+    }
+
+    async fn count_total(&self) -> Option<u64> {
+        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<Vec<u8>>,
+    ) -> RunOutcome {
+        // Cursor is `{size_dir}/{filename}` — the last file completed. Sizes
+        // are walked in `ThumbnailSize::all()` order, and names are sorted
+        // within each, so the pair totally orders the walk.
+        let cursor: Option<String> = 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 failed = 0u64;
+        let mut since_checkpoint = 0usize;
+        let variant_of = |s: ThumbnailSize| s.dir_name().to_string();
+
+        for size in ThumbnailSize::all() {
+            let dir_name = variant_of(*size);
+            for name in self.sidecar_names(*size).await {
+                let position = format!("{dir_name}/{name}");
+
+                // Resume: everything at or before the cursor is done.
+                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(hash) = Self::hash_from_sidecar_name(&name) else {
+                    continue;
+                };
+
+                // Already mapped — the common case on a re-run, and the
+                // reason this job is safe to trigger repeatedly.
+                if self
+                    .dedup
+                    .find_derived_blob(hash, "thumbnail", &dir_name)
+                    .await
+                    .is_some()
+                {
+                    already += 1;
+                } else {
+                    let path = self.thumbnails_root.join(&dir_name).join(&name);
+                    match fs::read(&path).await {
+                        Ok(data) => {
+                            match self
+                                .dedup
+                                .store_derived_blob(
+                                    hash,
+                                    "thumbnail",
+                                    &dir_name,
+                                    "image/webp",
+                                    Bytes::from(data),
+                                )
+                                .await
+                            {
+                                Ok(_) => imported += 1,
+                                Err(e) => {
+                                    failed += 1;
+                                    record_or_log(
+                                        store,
+                                        THUMB_DERIVED_IMPORT_JOB_NAME,
+                                        "thumbnail_import_failed",
+                                        "anomaly",
+                                        None,
+                                        serde_json::json!({
+                                            "path":  position,
+                                            "hash":  hash,
+                                            "error": format!("{e}"),
+                                            "note":  "sidecar left in place; safe to re-run",
+                                        }),
+                                    )
+                                    .await;
+                                }
+                            }
+                        }
+                        Err(e) => {
+                            // Unreadable, or removed between listing and read
+                            // (a concurrent GC unlink). Neither is fatal.
+                            failed += 1;
+                            record_or_log(
+                                store,
+                                THUMB_DERIVED_IMPORT_JOB_NAME,
+                                "thumbnail_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_derived_import.completed",
+            run_id = %store.run_id(),
+            imported = imported,
+            already_present = already,
+            failed = failed,
+            "thumb_derived_import: {imported} imported, {already} already present, {failed} failed"
+        );
+
+        RunOutcome::completed()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const H: &str = "0a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f9";
+
+    #[test]
+    fn accepts_a_canonical_sidecar_name() {
+        assert_eq!(
+            ThumbDerivedImport::hash_from_sidecar_name(&format!("{H}.webp")),
+            Some(H)
+        );
+    }
+
+    /// `ext-` files are user-supplied and file-keyed. Importing one here
+    /// would content-key it and share it across every file with identical
+    /// content — the exact poisoning the table split prevents.
+    #[test]
+    fn rejects_external_and_malformed_names() {
+        for name in [
+            format!("ext-{H}.jpg"),
+            "ext-3f2b1c00-0000-0000-0000-000000000000.jpg".to_string(),
+            format!("{H}.jpg"),
+            format!("{}.webp", &H[..63]),
+            H.to_string(),
+            "junk.webp".to_string(),
+        ] {
+            assert_eq!(
+                ThumbDerivedImport::hash_from_sidecar_name(&name),
+                None,
+                "must not be imported as a derived thumbnail: {name}"
+            );
+        }
+    }
+}