diff --git a/src/application/ports/blob_lifecycle.rs b/src/application/ports/blob_lifecycle.rs index 4d1df014..021f89cf 100644 --- a/src/application/ports/blob_lifecycle.rs +++ b/src/application/ports/blob_lifecycle.rs @@ -1,6 +1,20 @@ use std::future::Future; use std::pin::Pin; +/// Observer notified by [`DedupService`] when a genuinely new blob is stored +/// for the first time (no dedup hit). +/// +/// Register with [`DedupService::add_blob_creation_hook`] during DI wiring. +pub trait BlobCreationHook: Send + Sync { + /// Called after the new blob's chunks and manifest have been written. + /// `blob_hash` is the BLAKE3 hex, `content_type` is the MIME type if known. + /// Must be best-effort — must not propagate errors. + fn on_blob_created<'a>( + &'a self, + blob_hash: &'a str, + content_type: Option<&'a str>, + ) -> Pin + Send + 'a>>; +} /// Observer notified by [`DedupService`] when a blob's ref_count reaches zero /// and it is permanently removed from storage. /// diff --git a/src/application/ports/file_lifecycle.rs b/src/application/ports/file_lifecycle.rs new file mode 100644 index 00000000..fe416e28 --- /dev/null +++ b/src/application/ports/file_lifecycle.rs @@ -0,0 +1,57 @@ +use std::future::Future; +use std::pin::Pin; + +/// Observer notified by [`FileUploadService`] when a new file record is created +/// (including dedup hits where the blob already exists). +/// +/// Register with [`FileUploadService::with_file_created_hook`] during DI wiring. +pub trait FileCreatedHook: Send + Sync { + /// Called after the file record has been persisted. + /// `file_id` — opaque file UUID string. + /// `blob_hash` — BLAKE3 hex of the blob (may already exist on disk for dedup hits). + /// `content_type` — MIME type of the content. + /// Must be best-effort — must not propagate errors. + fn on_file_created<'a>( + &'a self, + file_id: &'a str, + blob_hash: &'a str, + content_type: &'a str, + ) -> Pin + Send + 'a>>; +} + +/// Observer notified by [`FileUploadService`] when an existing file's blob is +/// replaced (WebDAV PUT overwrite, WOPI PutFile, Nextcloud chunked upload). +/// +/// Implement this trait on any service that needs to react to a content swap +/// (e.g. thumbnail invalidation + regeneration, search index update). +/// Register with [`FileUploadService::with_file_updated_hook`] during DI wiring. +/// +/// The boxed-future return keeps the trait dyn-compatible so multiple +/// implementations can be stored as `Vec>`. +pub trait FileUpdatedHook: Send + Sync { + /// Called after the new blob has been stored and the file record updated. + /// + /// `file_id` is an opaque file UUID string, `blob_hash` is the BLAKE3 hex + /// of the new blob, `content_type` is the MIME type of the new content. + /// Must be best-effort — must not propagate errors. + fn on_file_updated<'a>( + &'a self, + file_id: &'a str, + blob_hash: &'a str, + content_type: &'a str, + ) -> Pin + Send + 'a>>; +} + +/// Observer notified by [`FileManagementService`] when a file is permanently +/// deleted (either directly or after being emptied from trash). +/// +/// Register with [`FileManagementService::with_file_deleted_hook`] during DI wiring. +pub trait FileDeletedHook: Send + Sync { + /// Called after the file record has been removed. + /// `file_id` — opaque file UUID string. + /// Must be best-effort — must not propagate errors. + fn on_file_deleted<'a>( + &'a self, + file_id: &'a str, + ) -> Pin + Send + 'a>>; +} diff --git a/src/application/ports/mod.rs b/src/application/ports/mod.rs index 2cd8e061..d3250c1e 100644 --- a/src/application/ports/mod.rs +++ b/src/application/ports/mod.rs @@ -8,6 +8,7 @@ pub mod chunked_upload_ports; pub mod compression_ports; pub mod dedup_ports; pub mod favorites_ports; +pub mod file_lifecycle; pub mod file_ports; pub mod inbound; pub mod music_ports; diff --git a/src/application/services/file_management_service.rs b/src/application/services/file_management_service.rs index eb7a3d37..7bec7187 100644 --- a/src/application/services/file_management_service.rs +++ b/src/application/services/file_management_service.rs @@ -1,6 +1,7 @@ use std::sync::Arc; use crate::application::dtos::file_dto::FileDto; +use crate::application::ports::file_lifecycle::FileDeletedHook; use crate::application::ports::file_ports::FileManagementUseCase; use crate::application::ports::storage_ports::{CopyFolderTreeResult, FileReadPort, FileWritePort}; use crate::application::ports::trash_ports::TrashUseCase; @@ -11,7 +12,6 @@ use crate::infrastructure::repositories::pg::file_blob_read_repository::FileBlob use crate::infrastructure::repositories::pg::file_blob_write_repository::FileBlobWriteRepository; use crate::infrastructure::repositories::pg::folder_db_repository::FolderDbRepository; use crate::infrastructure::services::file_content_cache::FileContentCache; -use crate::infrastructure::services::thumbnail_service::ThumbnailService; use tracing::{error, info, warn}; use uuid::Uuid; @@ -26,8 +26,9 @@ pub struct FileManagementService { file_read: Option>, folder_repo: Option>, trash_service: Option>, - thumbnail_service: Option>, content_cache: Option>, + /// Hooks fired after a file is permanently deleted. + file_deleted_hooks: Vec>, } impl FileManagementService { @@ -38,8 +39,8 @@ impl FileManagementService { file_read: None, folder_repo: None, trash_service: None, - thumbnail_service: None, content_cache: None, + file_deleted_hooks: Vec::new(), } } @@ -49,7 +50,6 @@ impl FileManagementService { trash_service: Option>, file_read: Option>, folder_repo: Option>, - thumbnail_service: Option>, content_cache: Option>, ) -> Self { Self { @@ -57,11 +57,17 @@ impl FileManagementService { file_read, folder_repo, trash_service, - thumbnail_service, content_cache, + file_deleted_hooks: Vec::new(), } } + /// Registers a hook to fire after a file is permanently deleted. + pub fn with_file_deleted_hook(mut self, hook: Arc) -> Self { + self.file_deleted_hooks.push(hook); + self + } + /// Verifies ownership via the read repository. async fn verify_owner(&self, file_id: &str, caller_id: Uuid) -> Result<(), DomainError> { if let Some(read) = &self.file_read { @@ -230,15 +236,11 @@ impl FileManagementUseCase for FileManagementService { async fn delete_file(&self, id: &str) -> Result<(), DomainError> { self.file_repository.delete_file(id).await?; - // Invalidate content cache — file no longer exists. if let Some(cc) = &self.content_cache { cc.invalidate(id).await; } - // Best-effort thumbnail cleanup - if let Some(thumb) = &self.thumbnail_service - && let Err(e) = thumb.delete_thumbnails(id).await - { - warn!("Failed to delete thumbnails for file {}: {}", id, e); + for hook in &self.file_deleted_hooks { + hook.on_file_deleted(id).await; } Ok(()) } @@ -283,15 +285,11 @@ impl FileManagementUseCase for FileManagementService { // Step 2: Permanent delete — trigger handles blob ref_count warn!("Permanently deleting file: {}", id); self.file_repository.delete_file(id).await?; - // Invalidate content cache — file permanently removed. if let Some(cc) = &self.content_cache { cc.invalidate(id).await; } - // Best-effort thumbnail cleanup - if let Some(thumb) = &self.thumbnail_service - && let Err(e) = thumb.delete_thumbnails(id).await - { - warn!("Failed to delete thumbnails for file {}: {}", id, e); + for hook in &self.file_deleted_hooks { + hook.on_file_deleted(id).await; } info!("File permanently deleted: {}", id); diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index 5249789d..e160f7ab 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -2,6 +2,7 @@ use std::path::Path; use std::sync::Arc; use crate::application::dtos::file_dto::FileDto; +use crate::application::ports::file_lifecycle::{FileCreatedHook, FileUpdatedHook}; use crate::application::ports::file_ports::FileUploadUseCase; use crate::application::ports::storage_ports::{FileReadPort, FileWritePort}; use crate::application::services::storage_usage_service::StorageUsageService; @@ -52,6 +53,10 @@ pub struct FileUploadService { storage_usage_service: Option>, /// Content cache — invalidated on file update so stale content is never served. content_cache: Option>, + /// Hooks fired after a new file record is created. + file_created_hooks: Vec>, + /// Hooks fired after a file's blob is replaced (e.g. thumbnail refresh). + file_updated_hooks: Vec>, } impl FileUploadService { @@ -62,6 +67,8 @@ impl FileUploadService { file_read: None, storage_usage_service: None, content_cache: None, + file_created_hooks: Vec::new(), + file_updated_hooks: Vec::new(), } } @@ -75,6 +82,8 @@ impl FileUploadService { file_read: Some(file_read), storage_usage_service: None, content_cache: None, + file_created_hooks: Vec::new(), + file_updated_hooks: Vec::new(), } } @@ -84,6 +93,18 @@ impl FileUploadService { self } + /// Registers a hook to fire after a new file record is created. + pub fn with_file_created_hook(mut self, hook: Arc) -> Self { + self.file_created_hooks.push(hook); + self + } + + /// Registers a hook to fire after a file's blob is replaced. + pub fn with_file_updated_hook(mut self, hook: Arc) -> Self { + self.file_updated_hooks.push(hook); + self + } + /// Configures the storage usage service pub fn with_storage_usage_service( mut self, @@ -149,6 +170,10 @@ impl FileUploadUseCase for FileUploadService { name, size, dto.id ); self.maybe_update_storage_usage(&dto); + for hook in &self.file_created_hooks { + hook.on_file_created(&dto.id, &dto.etag, &dto.mime_type) + .await; + } Ok(dto) } @@ -301,7 +326,12 @@ impl FileUploadUseCase for FileUploadService { } // Re-read to get fresh DTO with updated etag and timestamps. let updated = file_read.get_file(&file_id).await?; - return Ok(FileDto::from(updated)); + let dto = FileDto::from(updated); + for hook in &self.file_updated_hooks { + hook.on_file_updated(&file_id, &dto.etag, content_type) + .await; + } + return Ok(dto); } // File doesn't exist — create it via streaming upload diff --git a/src/common/di.rs b/src/common/di.rs index a0a82bc7..d8b773e6 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -61,7 +61,7 @@ use crate::infrastructure::services::image_transcode_service::ImageTranscodeServ use crate::infrastructure::services::jwt_service::JwtTokenService; use crate::infrastructure::services::password_hasher::Argon2PasswordHasher; use crate::infrastructure::services::path_resolver_service::PathResolverService; -use crate::infrastructure::services::thumbnail_service::ThumbnailService; +use crate::infrastructure::services::thumbnail_service::{ThumbnailRefreshHook, ThumbnailService}; use crate::infrastructure::services::wopi_discovery_service::WopiDiscoveryService; use crate::infrastructure::services::zip_service::ZipService; @@ -355,12 +355,18 @@ impl AppServiceFactory { // Refactored services with all infrastructure ports // In blob model, dedup is handled by the repository — no separate write-behind needed + let thumbnail_refresh_hook = Arc::new(ThumbnailRefreshHook::new( + core.thumbnail_service.clone(), + core.dedup_service.clone(), + )); let file_upload_service = Arc::new( FileUploadService::new_with_read( repos.file_write_repository.clone(), repos.file_read_repository.clone(), ) - .with_content_cache(core.file_content_cache.clone()), + .with_content_cache(core.file_content_cache.clone()) + .with_file_created_hook(thumbnail_refresh_hook.clone()) + .with_file_updated_hook(thumbnail_refresh_hook), ); let file_retrieval_service = Arc::new(FileRetrievalService::new_with_cache( @@ -370,14 +376,16 @@ impl AppServiceFactory { )); // FileManagementService — ref_count handled by PG trigger, no dedup port needed - let file_management_service = Arc::new(FileManagementService::with_trash( - repos.file_write_repository.clone(), - trash_service.clone(), - Some(repos.file_read_repository.clone()), - Some(repos.folder_repository.clone()), - Some(core.thumbnail_service.clone()), - Some(core.file_content_cache.clone()), - )); + let file_management_service = Arc::new( + FileManagementService::with_trash( + repos.file_write_repository.clone(), + trash_service.clone(), + Some(repos.file_read_repository.clone()), + Some(repos.folder_repository.clone()), + Some(core.file_content_cache.clone()), + ) + .with_file_deleted_hook(core.thumbnail_service.clone()), + ); let file_use_case_factory = Arc::new(AppFileUseCaseFactory::new( repos.file_read_repository.clone(), @@ -983,48 +991,6 @@ pub struct CoreServices { pub config: AppConfig, } -impl CoreServices { - /// Invalidate a file's moka thumbnail cache and kick off background regeneration. - /// - /// Call this after any write that swaps the blob for an existing file. - /// Safe to call for new files too (no-op on empty cache). - /// Skips everything if the MIME type is not a supported image. - pub async fn refresh_thumbnails_after_update( - &self, - file_id: String, - blob_hash: String, - content_type: &str, - ) { - if !ThumbnailService::is_supported_image(content_type) { - return; - } - if let Err(e) = self.thumbnail_service.delete_thumbnails(&file_id).await { - tracing::warn!( - "Failed to invalidate thumbnail cache for {}: {}", - file_id, - e - ); - } - let ts = self.thumbnail_service.clone(); - let ds = self.dedup_service.clone(); - let hash = blob_hash.clone(); - 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 regeneration {}: {}", - file_id, - e - ); - } - } - }); - } -} - /// Container for repository services #[derive(Clone)] pub struct RepositoryServices { diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index defa2295..082fdcf3 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -44,7 +44,7 @@ use std::sync::Arc; use tokio::fs; use tokio::io::{AsyncReadExt, AsyncSeekExt}; -use crate::application::ports::blob_lifecycle::BlobDeletionHook; +use crate::application::ports::blob_lifecycle::{BlobCreationHook, BlobDeletionHook}; use crate::application::ports::blob_storage_ports::BlobStorageBackend; use crate::application::ports::dedup_ports::{ BlobMetadataDto, DedupPort, DedupResultDto, DedupStatsDto, @@ -84,6 +84,8 @@ pub struct DedupService { /// Isolated maintenance pool for long-running operations /// (verify_integrity, garbage_collect) that must never starve the primary. maintenance_pool: Arc, + /// Hooks notified when a genuinely new blob is stored (no dedup hit). + blob_creation_hooks: Vec>, /// Hooks notified when a blob's ref_count reaches zero and it is deleted. blob_hooks: Vec>, } @@ -103,10 +105,18 @@ impl DedupService { backend, pool, maintenance_pool, + blob_creation_hooks: vec![], blob_hooks: vec![], } } + /// Register a [`BlobCreationHook`] to be called whenever a genuinely new + /// blob is stored. Hooks are called in registration order. + pub fn add_blob_creation_hook(mut self, hook: Arc) -> Self { + self.blob_creation_hooks.push(hook); + self + } + /// Register a [`BlobDeletionHook`] to be called whenever a blob's /// ref_count reaches zero. Hooks are called in registration order. pub fn add_blob_hook(mut self, hook: Arc) -> Self { @@ -114,6 +124,13 @@ impl DedupService { self } + /// Fire all registered creation hooks for a new blob. + async fn fire_blob_creation_hooks(&self, hash: &str, content_type: Option<&str>) { + for hook in &self.blob_creation_hooks { + hook.on_blob_created(hash, content_type).await; + } + } + /// Fire all registered hooks for a deleted blob. async fn fire_blob_hooks(&self, hash: &str) { for hook in &self.blob_hooks { @@ -135,6 +152,7 @@ impl DedupService { backend: Arc::new(LocalBlobBackend::new(Path::new("/tmp/oxicloud_stub_blobs"))), pool: stub_pool.clone(), maintenance_pool: stub_pool, + blob_creation_hooks: vec![], blob_hooks: vec![], } } @@ -353,6 +371,9 @@ impl DedupService { chunk_hashes.len() ); + self.fire_blob_creation_hooks(&file_hash, content_type.as_deref()) + .await; + Ok(DedupResultDto::NewBlob { hash: file_hash, size: file_size, diff --git a/src/infrastructure/services/thumbnail_service.rs b/src/infrastructure/services/thumbnail_service.rs index abaa8916..5f934aa3 100644 --- a/src/infrastructure/services/thumbnail_service.rs +++ b/src/infrastructure/services/thumbnail_service.rs @@ -995,6 +995,111 @@ impl crate::application::ports::blob_lifecycle::BlobDeletionHook for ThumbnailSe } } +// ─── FileUpdatedHook ───────────────────────────────────────────────────────── + +/// Wires thumbnail invalidation + regeneration into the file-update lifecycle. +/// +/// Registered on [`FileUploadService`] during DI. Fires whenever a file's blob +/// is replaced (WebDAV PUT overwrite, WOPI PutFile, Nextcloud chunked upload). +pub struct ThumbnailRefreshHook { + thumbnail: Arc, + dedup: Arc, +} + +impl ThumbnailRefreshHook { + pub fn new(thumbnail: Arc, dedup: Arc) -> Self { + Self { thumbnail, dedup } + } +} + +impl crate::application::ports::file_lifecycle::FileUpdatedHook for ThumbnailRefreshHook { + fn on_file_updated<'a>( + &'a self, + file_id: &'a str, + blob_hash: &'a str, + content_type: &'a str, + ) -> std::pin::Pin + Send + 'a>> { + Box::pin(async move { + if !ThumbnailService::is_supported_image(content_type) { + return; + } + if let Err(e) = self.thumbnail.delete_thumbnails(file_id).await { + tracing::warn!( + "Failed to invalidate thumbnail cache for {}: {}", + file_id, + e + ); + } + Self::spawn_thumbnail_generation( + self.thumbnail.clone(), + self.dedup.clone(), + file_id.to_string(), + blob_hash.to_string(), + ); + }) + } +} + +impl crate::application::ports::file_lifecycle::FileCreatedHook for ThumbnailRefreshHook { + fn on_file_created<'a>( + &'a self, + file_id: &'a str, + blob_hash: &'a str, + content_type: &'a str, + ) -> std::pin::Pin + Send + 'a>> { + Box::pin(async move { + if !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(), + ); + }) + } +} + +impl ThumbnailRefreshHook { + fn spawn_thumbnail_generation( + ts: Arc, + ds: Arc, + 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 + ); + } + } + }); + } +} + +// ─── FileDeletedHook ───────────────────────────────────────────────────────── + +impl crate::application::ports::file_lifecycle::FileDeletedHook for ThumbnailService { + fn on_file_deleted<'a>( + &'a self, + file_id: &'a str, + ) -> std::pin::Pin + Send + 'a>> { + Box::pin(async move { + if let Err(e) = self.delete_thumbnails(file_id).await { + tracing::warn!("Failed to delete thumbnails for file {}: {}", file_id, e); + } + }) + } +} + // ─── Port implementation ───────────────────────────────────────────────────── /// Convert port ThumbnailSize to infra ThumbnailSize. diff --git a/src/interfaces/api/handlers/webdav_handler.rs b/src/interfaces/api/handlers/webdav_handler.rs index 906a533b..826c8664 100644 --- a/src/interfaces/api/handlers/webdav_handler.rs +++ b/src/interfaces/api/handlers/webdav_handler.rs @@ -972,15 +972,6 @@ async fn handle_put( ); } - state - .core - .refresh_thumbnails_after_update( - file_dto.id.clone(), - file_dto.etag.clone(), - &content_type, - ) - .await; - Ok(Response::builder() .status(StatusCode::NO_CONTENT) .body(Body::empty()) diff --git a/src/interfaces/api/handlers/wopi_handler.rs b/src/interfaces/api/handlers/wopi_handler.rs index 5a2501bf..fc5cd595 100644 --- a/src/interfaces/api/handlers/wopi_handler.rs +++ b/src/interfaces/api/handlers/wopi_handler.rs @@ -288,19 +288,7 @@ async fn put_file( let _ = tokio::fs::remove_file(&temp_path).await; match result { - Ok(file_dto) => { - state - .app_state - .core - .refresh_thumbnails_after_update( - file_dto.id.clone(), - file_dto.etag.clone(), - &content_type, - ) - .await; - - StatusCode::OK.into_response() - } + Ok(_file_dto) => StatusCode::OK.into_response(), Err(e) => { tracing::error!("WOPI PutFile failed: {}", e); StatusCode::INTERNAL_SERVER_ERROR.into_response() diff --git a/src/interfaces/nextcloud/uploads_handler.rs b/src/interfaces/nextcloud/uploads_handler.rs index 1bde93d4..e49522a9 100644 --- a/src/interfaces/nextcloud/uploads_handler.rs +++ b/src/interfaces/nextcloud/uploads_handler.rs @@ -164,11 +164,6 @@ async fn handle_assemble( .await .map_err(|e| AppError::internal_error(format!("Failed to update file: {}", e)))?; - state - .core - .refresh_thumbnails_after_update(dto.id.clone(), dto.etag.clone(), &content_type) - .await; - Some(dto.etag) } else { // For new files we still need to read the temp file since create_file takes &[u8]. diff --git a/src/interfaces/nextcloud/webdav_handler.rs b/src/interfaces/nextcloud/webdav_handler.rs index a7994a16..a383ad05 100644 --- a/src/interfaces/nextcloud/webdav_handler.rs +++ b/src/interfaces/nextcloud/webdav_handler.rs @@ -566,16 +566,6 @@ async fn handle_put( } } - // Bug 1 & 2 fix: invalidate stale thumbnail and regenerate from new blob. - state - .core - .refresh_thumbnails_after_update( - updated.id.clone(), - updated.etag.clone(), - &content_type, - ) - .await; - return Ok(Response::builder() .status(StatusCode::NO_CONTENT) .header(header::ETAG, format!("\"{}\"", updated.etag))