Merge pull request #448 from AtalayaLabs/claude/gracious-heisenberg-5u1raf

src/application/services/file_management_service.rs

@@ -151,7 +151,7 @@ impl FileManagementService {
 
         let dto = FileDto::from(copied_file);
         if let Some(hook) = &self.file_lifecycle_hook {
-            hook.on_file_copied(&dto.id, &dto.etag, &dto.mime_type, file_id);
+            hook.on_file_copied(&dto.id, &dto.content_hash, &dto.mime_type, file_id);
         }
         Ok(dto)
     }

src/application/services/file_upload_service.rs

@@ -177,7 +177,7 @@ impl FileUploadUseCase for FileUploadService {
         );
         self.maybe_update_storage_usage(&dto);
         if let Some(hook) = &self.file_lifecycle_hook {
-            hook.on_file_created(&dto.id, &dto.etag, &dto.mime_type, is_new_blob);
+            hook.on_file_created(&dto.id, &dto.content_hash, &dto.mime_type, is_new_blob);
         }
         Ok(dto)
     }
@@ -260,7 +260,7 @@ impl FileUploadUseCase for FileUploadService {
         let dto = FileDto::from(file);
         self.maybe_update_storage_usage(&dto);
         if let Some(hook) = &self.file_lifecycle_hook {
-            hook.on_file_created(&dto.id, &dto.etag, &dto.mime_type, is_new_blob);
+            hook.on_file_created(&dto.id, &dto.content_hash, &dto.mime_type, is_new_blob);
         }
         Ok(dto)
     }
@@ -356,7 +356,7 @@ impl FileUploadUseCase for FileUploadService {
             })?;
             let dto = FileDto::from(updated);
             if let Some(hook) = &self.file_lifecycle_hook {
-                hook.on_file_updated(&file_id, &dto.etag, content_type);
+                hook.on_file_updated(&file_id, &dto.content_hash, content_type);
             }
             return Ok(dto);
         }
@@ -394,7 +394,7 @@ impl FileUploadUseCase for FileUploadService {
             .await?;
         let dto = FileDto::from(created);
         if let Some(hook) = &self.file_lifecycle_hook {
-            hook.on_file_created(&dto.id, &dto.etag, content_type, is_new_blob);
+            hook.on_file_created(&dto.id, &dto.content_hash, content_type, is_new_blob);
         }
         Ok(dto)
     }

src/common/di.rs

@@ -380,6 +380,9 @@ impl AppServiceFactory {
                 db_pool.clone(),
                 core.dedup_service.clone(),
                 folder_repo_concrete.clone(),
+                // Shared blob-hash cache: the write side invalidates entries
+                // on content swaps/deletes so reads never serve stale blobs.
+                file_read_repository.blob_hash_cache(),
             ));
 
         // I18n repository — file-system backed, gated by the locale

src/infrastructure/repositories/pg/file_blob_read_repository.rs

@@ -59,8 +59,14 @@ pub struct FileBlobReadRepository {
     dedup: Arc<DedupService>,
     /// Lock-free cache: file_id → blob_hash.
     /// Populated by `get_file()` and `resolve_blob_hash()` (slow path).
-    /// Entries persist until TTI expiry (30 s idle) or capacity eviction —
-    /// safe because blob_hash is content-addressed and never mutated.
+    /// Entries persist until TTI expiry (30 s idle) or capacity eviction.
+    /// Content updates DO remap a file_id to a new hash in place
+    /// (`swap_blob_hash`), so the write repository shares this cache (see
+    /// [`Self::blob_hash_cache`]) and invalidates the entry on every
+    /// content swap and hard delete — without that, streaming downloads
+    /// kept serving the previous blob for the TTI window after a PUT
+    /// update (or 500'd once the old blob was garbage-collected), and
+    /// every read refreshed the TTI, extending the window indefinitely.
     hash_cache: Cache<String, String>,
 }
 
@@ -80,6 +86,14 @@ impl FileBlobReadRepository {
         }
     }
 
+    /// Shared handle to the file_id → blob_hash cache (moka clones share
+    /// the underlying storage). Handed to `FileBlobWriteRepository` at DI
+    /// time so content swaps and hard deletes invalidate the mapping the
+    /// moment they commit.
+    pub fn blob_hash_cache(&self) -> Cache<String, String> {
+        self.hash_cache.clone()
+    }
+
     /// Returns the user_id (owner) for a given file ID.
     /// Mirrors `FolderDbRepository::get_folder_user_id`.
     /// Used by the AuthorizationEngine for owner short-circuit.
@@ -157,9 +171,9 @@ impl FileBlobReadRepository {
     /// subsequent reads for the same file (e.g. Range Requests on a video,
     /// thumbnail + download, browser re-fetch) hit the cache instead of PG.
     ///
-    /// This is safe because `blob_hash` is content-addressed (SHA-256)
-    /// and never mutated — if the file's content changes, a new row with a
-    /// new `blob_hash` is created.
+    /// Staleness safety: content updates remap the file to a new hash in
+    /// place — the write repository invalidates this cache (shared via
+    /// [`Self::blob_hash_cache`]) right after every swap/delete commits.
     async fn resolve_blob_hash(&self, file_id: &str) -> Result<String, DomainError> {
         // Fast path: cached (lock-free read, refreshes TTI automatically)
         if let Some(hash) = self.hash_cache.get(file_id) {

src/infrastructure/repositories/pg/file_blob_write_repository.rs

@@ -7,6 +7,7 @@
 //! File paths are resolved by querying the materialized `storage.folders.path`
 //! column (O(1) per lookup), so no recursive CTEs are needed.
 
+use moka::sync::Cache;
 use sqlx::PgPool;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -26,6 +27,10 @@ pub struct FileBlobWriteRepository {
     pool: Arc<PgPool>,
     dedup: Arc<DedupService>,
     folder_repo: Arc<FolderDbRepository>,
+    /// Shared handle to `FileBlobReadRepository`'s file_id → blob_hash
+    /// cache. Content swaps and hard deletes invalidate the mapping here
+    /// so the read side can never serve a stale blob after a PUT update.
+    hash_cache: Cache<String, String>,
 }
 
 impl FileBlobWriteRepository {
@@ -33,11 +38,13 @@ impl FileBlobWriteRepository {
         pool: Arc<PgPool>,
         dedup: Arc<DedupService>,
         folder_repo: Arc<FolderDbRepository>,
+        hash_cache: Cache<String, String>,
     ) -> Self {
         Self {
             pool,
             dedup,
             folder_repo,
+            hash_cache,
         }
     }
 
@@ -54,6 +61,7 @@ impl FileBlobWriteRepository {
             ),
             dedup: Arc::new(DedupService::new_stub()),
             folder_repo: Arc::new(super::folder_db_repository::FolderDbRepository::new_stub()),
+            hash_cache: Cache::builder().max_capacity(10_000).build(),
         }
     }
 
@@ -505,6 +513,8 @@ impl FileWritePort for FileBlobWriteRepository {
             return Err(DomainError::not_found("File", id));
         }
 
+        // Drop the read-side file_id → blob_hash mapping for the dead row.
+        self.hash_cache.invalidate(id);
         Ok(())
     }
 
@@ -524,8 +534,14 @@ impl FileWritePort for FileBlobWriteRepository {
             .await?;
         let new_hash = dedup_result.hash().to_string();
 
-        self.swap_blob_hash(file_id, &new_hash, size as i64, modified_at)
-            .await
+        let swapped = self
+            .swap_blob_hash(file_id, &new_hash, size as i64, modified_at)
+            .await?;
+        // The file now maps to a different blob — drop the read-side cache
+        // entry so streaming downloads cannot serve the previous content
+        // for the rest of its TTI window.
+        self.hash_cache.invalidate(file_id);
+        Ok(swapped)
     }
 
     async fn register_file_deferred(

src/infrastructure/repositories/pg/trash_db_repository.rs

@@ -47,6 +47,40 @@ impl TrashDbRepository {
         }
     }
 
+    /// Runs a LIMIT-ed DELETE statement repeatedly until a round affects
+    /// fewer rows than `batch_size`, yielding to the runtime between rounds.
+    ///
+    /// `sql` must bind `$1` = cutoff timestamp and `$2` = batch size; the
+    /// candidate sub-select is served by the `idx_*_trash_expiry` partial
+    /// indexes. Each round is its own implicit transaction, so row locks,
+    /// WAL volume and the statement-trigger transition tables stay bounded
+    /// no matter how many items expired. Partial progress is fine — the
+    /// next retention sweep continues where this one stopped.
+    async fn delete_expired_batch_loop(
+        &self,
+        sql: &'static str,
+        cutoff: DateTime<Utc>,
+        batch_size: i64,
+    ) -> Result<u64> {
+        let mut total: u64 = 0;
+        loop {
+            let affected = sqlx::query(sql)
+                .bind(cutoff)
+                .bind(batch_size)
+                .execute(self.pool.as_ref())
+                .await
+                .map_err(|e| {
+                    DomainError::internal_error("TrashDb", format!("bulk delete batch: {e}"))
+                })?
+                .rows_affected();
+            total += affected;
+            if affected < batch_size as u64 {
+                return Ok(total);
+            }
+            tokio::task::yield_now().await;
+        }
+    }
+
     /// Convert a trash_items view row into a TrashedItem entity.
     fn row_to_trashed_item(
         &self,
@@ -180,40 +214,36 @@ impl TrashRepository for TrashDbRepository {
     async fn delete_expired_bulk(&self) -> Result<(u64, u64)> {
         let cutoff = Utc::now() - chrono::Duration::days(self.retention_days);
 
-        let mut tx = self
-            .pool
-            .begin()
-            .await
-            .map_err(|e| DomainError::internal_error("TrashDb", format!("begin tx: {e}")))?;
-
-        // 1. Bulk-delete expired trashed files.
+        // 1. Bulk-delete expired trashed files in batches.
         //    The PG trigger `trg_files_decrement_blob_ref` automatically
         //    decrements blob ref_count for every deleted row.
-        let files_deleted =
-            sqlx::query("DELETE FROM storage.files WHERE is_trashed = TRUE AND trashed_at < $1")
-                .bind(cutoff)
-                .execute(&mut *tx)
-                .await
-                .map_err(|e| {
-                    DomainError::internal_error("TrashDb", format!("bulk delete files: {e}"))
-                })?
-                .rows_affected();
+        let files_deleted = self
+            .delete_expired_batch_loop(
+                "DELETE FROM storage.files
+                  WHERE id IN (SELECT id FROM storage.files
+                                WHERE is_trashed = TRUE AND trashed_at < $1
+                                ORDER BY trashed_at
+                                LIMIT $2)",
+                cutoff,
+                1_000,
+            )
+            .await?;
 
-        // 2. Bulk-delete expired trashed folders.
-        //    FK ON DELETE CASCADE handles descendant folders and their files.
-        let folders_deleted =
-            sqlx::query("DELETE FROM storage.folders WHERE is_trashed = TRUE AND trashed_at < $1")
-                .bind(cutoff)
-                .execute(&mut *tx)
-                .await
-                .map_err(|e| {
-                    DomainError::internal_error("TrashDb", format!("bulk delete folders: {e}"))
-                })?
-                .rows_affected();
-
-        tx.commit()
-            .await
-            .map_err(|e| DomainError::internal_error("TrashDb", format!("commit tx: {e}")))?;
+        // 2. Bulk-delete expired trashed folders in batches.
+        //    FK ON DELETE CASCADE handles descendant folders and their
+        //    files, so each row can fan out to an entire subtree — hence
+        //    the smaller batch size.
+        let folders_deleted = self
+            .delete_expired_batch_loop(
+                "DELETE FROM storage.folders
+                  WHERE id IN (SELECT id FROM storage.folders
+                                WHERE is_trashed = TRUE AND trashed_at < $1
+                                ORDER BY trashed_at
+                                LIMIT $2)",
+                cutoff,
+                100,
+            )
+            .await?;
 
         Ok((files_deleted, folders_deleted))
     }

src/infrastructure/services/thumbnail_service.rs

@@ -258,7 +258,10 @@ impl ThumbnailService {
     /// Get a thumbnail from raw image bytes, generating it if needed.
     ///
     /// This is the storage-model-safe entrypoint for CDC/manifest-backed
-    /// blobs where no single local source file exists on disk.
+    /// blobs where no single local source file exists on disk. Prefer
+    /// [`Self::get_thumbnail_from_blob`] on request paths — it defers the
+    /// full blob read until a decode permit is held, so a stampede of
+    /// cache misses cannot stack one source image per request in RAM.
     pub async fn get_thumbnail_from_bytes(
         &self,
         file_id: &str,
@@ -287,30 +290,12 @@ impl ThumbnailService {
                     return Bytes::from(data);
                 }
 
-                tracing::info!("🎨 Generating thumbnail: {} {:?}", file_id_owned, size);
-                match Self::generate_thumbnail_from_data(
-                    original_data,
-                    size,
-                    self.generation_timeout,
-                )
-                .await
-                {
-                    Ok(bytes) => {
-                        if let Some(parent) = thumb_path.parent() {
-                            let _ = fs::create_dir_all(parent).await;
-                        }
-                        let _ = fs::write(&thumb_path, &bytes).await;
-                        bytes
-                    }
-                    Err(e) => {
-                        tracing::warn!(
-                            "Thumbnail generation failed for {} {:?}: {e}",
-                            file_id_owned,
-                            size
-                        );
-                        Bytes::new()
-                    }
-                }
+                let Ok(_permit) = self.decode_semaphore.acquire().await else {
+                    tracing::warn!("Decode semaphore closed, skipping {}", file_id_owned);
+                    return Bytes::new();
+                };
+                self.generate_and_persist(&file_id_owned, &thumb_path, size, original_data)
+                    .await
             })
             .await;
 
@@ -325,6 +310,106 @@ impl ThumbnailService {
         Ok(bytes)
     }
 
+    /// Get a thumbnail for a content-addressed blob, generating it if needed.
+    ///
+    /// Request-path entrypoint: on a memory+disk cache miss the source blob
+    /// is read **after** a decode permit is acquired, so peak RAM under a
+    /// thumbnail stampede is `permits × image size` instead of
+    /// `in-flight requests × image size`. moka's per-key init additionally
+    /// collapses concurrent requests for the same thumbnail into one read.
+    pub async fn get_thumbnail_from_blob(
+        &self,
+        file_id: &str,
+        blob_hash: &str,
+        size: ThumbnailSize,
+        dedup: Arc<DedupService>,
+    ) -> Result<Bytes, ThumbnailError> {
+        let cache_key = ThumbnailCacheKey {
+            file_id: file_id.to_string(),
+            size,
+        };
+
+        let thumb_path = self.get_thumbnail_path(blob_hash, size);
+        let file_id_owned = file_id.to_string();
+        let blob_hash_owned = blob_hash.to_string();
+
+        let entry = self
+            .cache
+            .entry(cache_key)
+            .or_insert_with(async move {
+                if let Ok(data) = fs::read(&thumb_path).await {
+                    tracing::debug!(
+                        "💾 Thumbnail loaded from disk: {} {:?}",
+                        file_id_owned,
+                        size
+                    );
+                    return Bytes::from(data);
+                }
+
+                let Ok(_permit) = self.decode_semaphore.acquire().await else {
+                    tracing::warn!("Decode semaphore closed, skipping {}", file_id_owned);
+                    return Bytes::new();
+                };
+                let original_data = match dedup.read_blob_bytes(&blob_hash_owned).await {
+                    Ok(bytes) => bytes,
+                    Err(e) => {
+                        tracing::warn!(
+                            "Failed to read blob for thumbnail {} {:?}: {e}",
+                            file_id_owned,
+                            size
+                        );
+                        return Bytes::new();
+                    }
+                };
+                self.generate_and_persist(&file_id_owned, &thumb_path, size, original_data)
+                    .await
+            })
+            .await;
+
+        let bytes = entry.into_value();
+        if bytes.is_empty() {
+            return Err(ThumbnailError::ImageError(
+                "Thumbnail generation failed".to_string(),
+            ));
+        }
+
+        tracing::debug!("🔥 Thumbnail served: {} {:?}", file_id, size);
+        Ok(bytes)
+    }
+
+    /// Decode `original_data` into one thumbnail size, persist it to its
+    /// blob-keyed disk path, and return the encoded bytes — empty `Bytes`
+    /// on failure (moka's zero-weight negative-entry convention).
+    ///
+    /// Callers must hold a `decode_semaphore` permit.
+    async fn generate_and_persist(
+        &self,
+        file_id: &str,
+        thumb_path: &Path,
+        size: ThumbnailSize,
+        original_data: Bytes,
+    ) -> Bytes {
+        tracing::info!("🎨 Generating thumbnail: {} {:?}", file_id, size);
+        match Self::generate_thumbnail_from_data(original_data, size, self.generation_timeout).await
+        {
+            Ok(bytes) => {
+                if let Some(parent) = thumb_path.parent() {
+                    let _ = fs::create_dir_all(parent).await;
+                }
+                let _ = fs::write(&thumb_path, &bytes).await;
+                bytes
+            }
+            Err(e) => {
+                tracing::warn!(
+                    "Thumbnail generation failed for {} {:?}: {e}",
+                    file_id,
+                    size
+                );
+                Bytes::new()
+            }
+        }
+    }
+
     /// Try to serve a thumbnail from cache only (memory → disk).
     ///
     /// Unlike `get_thumbnail`, this does **not** generate a new thumbnail.
@@ -823,15 +908,16 @@ impl ThumbnailService {
         });
     }
 
-    /// Generate all thumbnail sizes in the background from raw image bytes.
+    /// Generate all thumbnail sizes in the background for a content-addressed
+    /// blob (CDC/manifest-safe — no physical source file required).
     ///
-    /// This is compatible with CDC/manifest-backed blobs because it does not
-    /// require a single physical source file on disk.
-    pub fn generate_all_sizes_background_from_bytes(
+    /// The source blob is read **after** the decode permit is acquired, so N
+    /// concurrent uploads queue as N small tasks, not N full images in RAM:
+    /// peak memory is `permits × image size` regardless of upload concurrency.
+    pub fn generate_all_sizes_background_from_blob(
         self: Arc<Self>,
         file_id: String,
         blob_hash: String,
-        original_data: Bytes,
         dedup: Arc<DedupService>,
     ) {
         tokio::spawn(async move {
@@ -889,6 +975,20 @@ impl ThumbnailService {
                 }
             };
 
+            // Read the source only now that a permit bounds how many of
+            // these full-image buffers can exist at once.
+            let original_data = match dedup.read_blob_bytes(&blob_hash).await {
+                Ok(bytes) => bytes,
+                Err(e) => {
+                    tracing::warn!(
+                        "Failed to read blob for thumbnail generation {}: {}",
+                        file_id,
+                        e
+                    );
+                    return;
+                }
+            };
+
             let results = tokio::task::spawn_blocking(move || {
                 Self::render_all_thumbnails_from_data(original_data.as_ref())
             })
@@ -1013,12 +1113,13 @@ impl crate::application::ports::file_lifecycle::FileLifecycleHook for ThumbnailR
         if !is_new_blob || !ThumbnailService::is_supported_image(content_type) {
             return;
         }
-        Self::spawn_thumbnail_generation(
-            self.thumbnail.clone(),
-            self.dedup.clone(),
-            file_id.to_string(),
-            blob_hash.to_string(),
-        );
+        self.thumbnail
+            .clone()
+            .generate_all_sizes_background_from_blob(
+                file_id.to_string(),
+                blob_hash.to_string(),
+                self.dedup.clone(),
+            );
     }
 
     fn on_file_copied(
@@ -1047,7 +1148,7 @@ impl crate::application::ports::file_lifecycle::FileLifecycleHook for ThumbnailR
                     e
                 );
             }
-            Self::spawn_thumbnail_generation(thumbnail, dedup, file_id, blob_hash);
+            thumbnail.generate_all_sizes_background_from_blob(file_id, blob_hash, dedup);
         });
     }
 
@@ -1066,30 +1167,6 @@ impl crate::application::ports::file_lifecycle::FileLifecycleHook for ThumbnailR
 // to avoid a circular Arc: DedupService→BlobLifecycleService→ThumbnailRefreshHook→DedupService.
 // ThumbnailService does not hold DedupService so no cycle exists.
 
-impl ThumbnailRefreshHook {
-    fn spawn_thumbnail_generation(
-        ts: Arc<ThumbnailService>,
-        ds: Arc<DedupService>,
-        file_id: String,
-        hash: String,
-    ) {
-        tokio::spawn(async move {
-            match ds.read_blob_bytes(&hash).await {
-                Ok(bytes) => {
-                    ts.generate_all_sizes_background_from_bytes(file_id, hash, bytes, ds.clone());
-                }
-                Err(e) => {
-                    tracing::warn!(
-                        "Failed to read blob for thumbnail generation {}: {}",
-                        file_id,
-                        e
-                    );
-                }
-            }
-        });
-    }
-}
-
 // ─── BlobLifecycleHook ───────────────────────────────────────────────────────
 
 impl crate::application::ports::blob_lifecycle::BlobLifecycleHook for ThumbnailService {

src/interfaces/api/handlers/file_handler.rs

@@ -442,19 +442,13 @@ impl FileHandler {
                 .into_response();
         }
 
-        let original_bytes = match state.core.dedup_service.read_blob_bytes(&blob_hash).await {
-            Ok(bytes) => bytes,
-            Err(err) => {
-                return AppError::internal_error(format!(
-                    "Failed to load source image for thumbnail generation: {}",
-                    err
-                ))
-                .into_response();
-            }
-        };
-
         match thumbnail_service
-            .get_thumbnail_from_bytes(&id, &blob_hash, thumb_size.into(), original_bytes)
+            .get_thumbnail_from_blob(
+                &id,
+                &blob_hash,
+                thumb_size.into(),
+                state.core.dedup_service.clone(),
+            )
             .await
         {
             Ok(data) => Response::builder()

src/interfaces/nextcloud/preview_handler.rs

@@ -157,26 +157,18 @@ pub async fn handle_preview(
             .unwrap();
     }
 
-    let original_bytes = match state.core.dedup_service.read_blob_bytes(&blob_hash).await {
-        Ok(bytes) => bytes,
-        Err(err) => {
-            tracing::error!(
-                "Failed to load source image for preview {}: {}",
-                object_id,
-                err
-            );
-            return Response::builder()
-                .status(StatusCode::INTERNAL_SERVER_ERROR)
-                .body(Body::from("Failed to load preview source"))
-                .unwrap();
-        }
-    };
-
-    // Generate/get thumbnail
+    // Generate/get thumbnail — the blob is read inside the service once a
+    // decode permit is held, so preview stampedes cannot stack source
+    // images in RAM.
     match state
         .core
         .thumbnail_service
-        .get_thumbnail_from_bytes(&object_id, &blob_hash, thumb_size.into(), original_bytes)
+        .get_thumbnail_from_blob(
+            &object_id,
+            &blob_hash,
+            thumb_size.into(),
+            state.core.dedup_service.clone(),
+        )
         .await
     {
         Ok(data) => {