From 73f0b0fa47f7fe84e1765118d47445aa47d28e12 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Fri, 22 May 2026 13:10:47 +0200 Subject: [PATCH] refactor(lifecycle hooks): simplify integration of new services * make more coherent lifecycles * remove specific implementation on different handlers (they do not need to know existence of ThumbnailSerice nor AudioMetadataService) * reduce risk of orphean objects * ensure additional services are correctly wired (ex: Thumbnail generation was not covering all upload cases) * more details on docs/architecture/file-and-blob-lifecycle.md : ```rust // application/ports/file_lifecycle.rs pub trait FileLifecycleHook { fn on_file_created(file_id, blob_hash, content_type, is_new_blob); fn on_file_updated(file_id, blob_hash, content_type); fn on_file_copied(file_id, blob_hash, content_type, source_id) fn on_file_deleted(file_id); } // application/ports/blob_lifecycle.rs pub trait BlobLifecycleHook { fn on_blob_created(blob_hash, content_type); fn on_blob_deleted(blob_hash); } ``` --- docs/architecture/file-and-blob-lifecycle.md | 59 +++++ docs/plan/FileLifeCycle-BlobLifecycle.plan.md | 213 ++++++++++++++++++ src/application/ports/blob_lifecycle.rs | 57 ++--- src/application/ports/file_lifecycle.rs | 118 +++++----- .../services/blob_lifecycle_service.rs | 44 ++++ .../services/file_lifecycle_service.rs | 67 ++++-- .../services/file_management_service.rs | 24 +- .../services/file_upload_service.rs | 58 +++-- src/application/services/mod.rs | 1 + src/application/services/trash_service.rs | 14 +- src/common/di.rs | 55 +++-- .../pg/file_blob_write_repository.rs | 41 +++- .../services/audio_metadata_service.rs | 172 ++++++++++++++ src/infrastructure/services/dedup_service.rs | 54 ++--- .../services/thumbnail_service.rs | 146 ++++++------ src/interfaces/api/handlers/file_handler.rs | 53 +---- src/interfaces/api/handlers/webdav_handler.rs | 24 +- src/interfaces/nextcloud/uploads_handler.rs | 14 -- src/interfaces/nextcloud/webdav_handler.rs | 31 --- 19 files changed, 842 insertions(+), 403 deletions(-) create mode 100644 docs/architecture/file-and-blob-lifecycle.md create mode 100644 docs/plan/FileLifeCycle-BlobLifecycle.plan.md create mode 100644 src/application/services/blob_lifecycle_service.rs diff --git a/docs/architecture/file-and-blob-lifecycle.md b/docs/architecture/file-and-blob-lifecycle.md new file mode 100644 index 00000000..fb4ed0a2 --- /dev/null +++ b/docs/architecture/file-and-blob-lifecycle.md @@ -0,0 +1,59 @@ +# Plan: Unified lifecycle hooks (file + blob) + +## Context + +In order to help new features integration and a good hygien on objects (example: reduce risk of orphean entries) +Lifecycle traits are added + +--- + +## Design decisions + +TODO: draw a mermaid + +### Synchronous trait methods + +Hooks are fire-and-notify: every implementation either spawns a `tokio::spawn` (important to avoid blocking implementation) +internally or does nothing. Sync trait = no `Box::pin`, no `async_trait`, genuine one-liner noops. + +```rust +// application/ports/file_lifecycle.rs +pub trait FileLifecycleHook: Send + Sync { + + // fired on file created (after blob is written), is_new_blob tells if this blob was already present + fn on_file_created(&self, file_id: &str, blob_hash: &str, content_type: &str, is_new_blob: bool); + + // fired on file updated, means that blob changed + fn on_file_updated(&self, file_id: &str, blob_hash: &str, content_type: &str); + + // fired on copy, the blob remains unchanged + fn on_file_copied(&self, file_id: &str, blob_hash: &str, content_type: &str, source_id: &str) + + // fired on file deletion, not information if the blob still exists, up to the implementor to use BlobLifecycleHook if needed + fn on_file_deleted(&self, file_id: &str); +} + +// application/ports/blob_lifecycle.rs +pub trait BlobLifecycleHook: Send + Sync { + + // a blob as been created + fn on_blob_created(&self, blob_hash: &str, content_type: Option<&str>); + + // a blob as been deleted (no more refernce on it) + fn on_blob_deleted(&self, blob_hash: &str); +} +``` + +**No default methods** — explicit noops required (forces developer acknowledgement of all events, if a method is not necessary, just implement it with a noop method). + +### `is_new_blob: bool` on `on_file_created` + +Tells the implementor whether the underlying blob is genuinely new (fresh upload, no dedup hit) or already existed (copy, dedup hit on re-upload). This prevents implementors from re-scanning/re-generating work that can be shared or cloned from an existing record: + +use cases: + +- `ThumbnailRefreshHook`: if `!is_new_blob`, the `blob_hash` thumbnail already exists on disk — skip scheduling generation entirely as server side thumbnail are based on blob +- `AudioMetadataService`: if `!is_new_blob`, clone the existing metadata row for the `blob_hash` (fast DB copy) instead of re-parsing the blob. + +**Where `is_new_blob` comes from**: `FileUploadService` gets the dedup result from `FileBlobWriteRepository.save_file_from_temp()` (already computed during upload). For `copy_file()`, always `false` — same blob by definition. + diff --git a/docs/plan/FileLifeCycle-BlobLifecycle.plan.md b/docs/plan/FileLifeCycle-BlobLifecycle.plan.md new file mode 100644 index 00000000..1a387727 --- /dev/null +++ b/docs/plan/FileLifeCycle-BlobLifecycle.plan.md @@ -0,0 +1,213 @@ +# Plan: Unified lifecycle hooks (file + blob) + +## Context + +The file and blob lifecycle hook systems are partially built but inconsistently wired: +- `FileLifecycleService` only fans out `on_file_deleted`; created/updated hooks are wired directly on `FileUploadService`. +- `AudioMetadataService` implements no hook traits — called raw from 4 handler files. +- `ThumbnailRefreshHook` (file created/updated) and `ThumbnailService` (file deleted, blob deleted) are separate registrations for the same concern. +- `copy_file()` fires no hooks — copied files never get audio metadata (confirmed gap: `audio.file_metadata` is keyed by `file_id`, not `blob_hash`). +- Blob lifecycle has the same structural problem: two separate traits and two separate vecs in `DedupService`. + +Goal: one `FileLifecycleHook` + one `BlobLifecycleHook` trait, each with a composite dispatcher, all side-effects wired through them, handlers reduced to protocol translators. + +--- + +## Design decisions + +### Synchronous trait methods + +Hooks are fire-and-notify: every implementation either spawns a `tokio::spawn` internally or does nothing. Sync trait = no `Box::pin`, no `async_trait`, genuine one-liner noops. + +```rust +// application/ports/file_lifecycle.rs +pub trait FileLifecycleHook: Send + Sync { + fn on_file_created(&self, file_id: &str, blob_hash: &str, content_type: &str, is_new_blob: bool); + fn on_file_updated(&self, file_id: &str, blob_hash: &str, content_type: &str); + fn on_file_copied(&self, file_id: &str, blob_hash: &str, content_type: &str, source_id: &str) + // not information if the blob still exists, up to implementor to use BlobLifecycleHook if needed + fn on_file_deleted(&self, file_id: &str); +} + +// application/ports/blob_lifecycle.rs +pub trait BlobLifecycleHook: Send + Sync { + fn on_blob_created(&self, blob_hash: &str, content_type: Option<&str>); + fn on_blob_deleted(&self, blob_hash: &str); +} +``` + +**No default methods** — explicit noops required (forces developer acknowledgement of all events). + +### `is_new_blob: bool` on `on_file_created` + +Tells the implementor whether the underlying blob is genuinely new (fresh upload, no dedup hit) or already existed (copy, dedup hit on re-upload). This prevents implementors from re-scanning/re-generating work that can be shared or cloned from an existing record: + +- `ThumbnailRefreshHook`: if `!is_new_blob`, the `blob_hash` thumbnail already exists on disk — skip scheduling generation entirely. +- `AudioMetadataService`: if `!is_new_blob`, clone the existing metadata row for the `blob_hash` (fast DB copy) instead of re-parsing the blob. + +**Where `is_new_blob` comes from**: `FileUploadService` gets the dedup result from `FileBlobWriteRepository.save_file_from_temp()` (already computed during upload). For `copy_file()`, always `false` — same blob by definition. + +### Old traits removed entirely + +Six old traits (`FileCreatedHook`, `FileUpdatedHook`, `FileDeletedHook`, `BlobCreationHook`, `BlobDeletionHook`) removed. All implementors migrate to the two new traits. + +### Why `on_file_deleted` can be sync + +`ThumbnailService.delete_thumbnails` is currently awaited by the caller. It moves to `tokio::spawn` internally — thumbnail cleanup is best-effort, callers don't depend on it completing. + +--- + +## Thumbnail storage model (context) + +- **Disk**: keyed by `blob_hash` → `thumbnails_root/{size}/{blob_hash}.jpg` — shared between all files with the same content. +- **Moka cache**: keyed by `(file_id, size)` — cold-misses on first request for a new `file_id`, then reads from disk. +- **External thumbnails** (video frames): keyed by `file_id` → `ext-{file_id}.jpg`. + +Image copy is safe: disk thumbnail exists for the `blob_hash`, no regeneration needed (`is_new_blob = false` will skip it). + +--- + +## Files to change + +### 1. `src/application/ports/file_lifecycle.rs` +Replace three separate async traits with one sync `FileLifecycleHook` trait (3 methods + `is_new_blob` on created, no defaults). + +### 2. `src/application/ports/blob_lifecycle.rs` +Replace two separate async traits with one sync `BlobLifecycleHook` trait (2 methods, no defaults). + +### 3. `src/application/services/file_lifecycle_service.rs` +- One `Vec>`. +- One builder: `.with_hook(hook)`. +- `impl FileLifecycleHook`: plain `for` loops, no async, forwards `is_new_blob`. + +### 4. New: `src/application/services/blob_lifecycle_service.rs` +Mirror of `FileLifecycleService` for blob events: +- `Vec>`, `.with_hook()` builder, `impl BlobLifecycleHook` fan-out. + +### 5. `src/infrastructure/services/thumbnail_service.rs` +Consolidate all thumbnail hook logic into `ThumbnailRefreshHook`, implementing **both** new traits: + +**`impl FileLifecycleHook for ThumbnailRefreshHook`**: +- `on_file_created`: if `!is_new_blob` or unsupported content type → return early (blob thumbnail already on disk). Otherwise spawn generation. +- `on_file_updated`: spawn thumbnail invalidation + regeneration (existing logic). +- `on_file_deleted`: `tokio::spawn({ thumbnail.delete_thumbnails(file_id).await })`. + +**`impl BlobLifecycleHook for ThumbnailRefreshHook`**: +- `on_blob_created`: explicit noop — thumbnail gen is handled at file level via `on_file_created`. +- `on_blob_deleted`: `tokio::spawn({ thumbnail.delete_blob_thumbnails(blob_hash).await })`. + +Remove: `impl FileDeletedHook for ThumbnailService`, `impl BlobDeletionHook for ThumbnailService`. + +### 6. `src/infrastructure/services/audio_metadata_service.rs` + +**New method**: `clone_or_extract_background(service: Arc, new_file_id: Uuid, blob_hash: String)` +- Spawns a task that runs: + ```sql + INSERT INTO audio.file_metadata (file_id, title, artist, album, album_artist, + genre, track_number, disc_number, year, duration_secs, format) + SELECT $new_file_id, title, artist, album, album_artist, genre, + track_number, disc_number, year, duration_secs, format + FROM audio.file_metadata am + JOIN storage.files sf ON sf.id = am.file_id + WHERE sf.blob_hash = $blob_hash + LIMIT 1 + ON CONFLICT (file_id) DO NOTHING + ``` +- If 0 rows inserted (original not yet processed), falls back to `extract_and_save`. + +**`impl FileLifecycleHook for AudioMetadataService`**: +- `on_file_created`: if `is_audio_file(content_type)` → parse UUID, then: + - `is_new_blob = true` → `spawn_extraction_background(file_id, blob_path(blob_hash))` + - `is_new_blob = false` → `clone_or_extract_background(file_id, blob_hash)` +- `on_file_updated`: if audio → `spawn_extraction_with_delete_background`. +- `on_file_deleted`: explicit one-liner noop + comment: `audio.file_metadata` has `ON DELETE CASCADE`, DB handles cleanup. + +### 7. `src/application/services/file_upload_service.rs` +- Replace `file_created_hooks: Vec>` + `file_updated_hooks` with `file_lifecycle_hook: Option>`. +- Builder: `.with_file_lifecycle_hook(hook)`. +- Sync calls replacing async fan-out loops. Pass `is_new_blob` from the dedup result already available at this layer. + +### 8. `src/application/services/file_management_service.rs` +- Replace `Arc` with `Arc`. +- `on_file_deleted` call becomes sync. +- **Fix copy gap**: after `file_repository.copy_file()` returns the new file DTO, call `self.file_lifecycle.on_file_created(new_id, blob_hash, mime_type, false)`. + +### 9. `src/application/services/trash_service.rs` +- Replace `Arc` with `Arc`. +- `on_file_deleted` calls become sync. + +### 10. `src/infrastructure/services/dedup_service.rs` +- Replace `blob_creation_hooks: Vec>` + `blob_hooks: Vec>` with `blob_lifecycle: Option>`. +- Builder: `.with_blob_lifecycle(hook)`. +- Sync calls replacing async fan-outs. + +### 11. `src/common/di.rs` + +```rust +let thumbnail_hook = Arc::new(ThumbnailRefreshHook::new( + core.thumbnail_service.clone(), + dedup.clone(), +)); + +let file_lifecycle = Arc::new( + FileLifecycleService::new() + .with_hook(thumbnail_hook.clone()) + .with_hook(audio_metadata_service.clone()) // if Some +); + +let blob_lifecycle = Arc::new( + BlobLifecycleService::new() + .with_hook(thumbnail_hook.clone()) +); + +dedup_service.with_blob_lifecycle(blob_lifecycle) +file_upload_service.with_file_lifecycle_hook(file_lifecycle.clone()) +file_management_service.with_file_lifecycle_hook(file_lifecycle.clone()) +trash_service.with_file_lifecycle_hook(file_lifecycle.clone()) +``` + +### 12. Handler cleanup — 4 files (deletes only) + +| File | Remove | +|---|---| +| `src/interfaces/api/handlers/file_handler.rs` | direct `thumbnail_service.generate_all_sizes_background_from_bytes(...)` + `AudioMetadataService::spawn_extraction_background(...)` | +| `src/interfaces/nextcloud/webdav_handler.rs` | `AudioMetadataService::spawn_extraction_background(...)` (create) + `AudioMetadataService::spawn_extraction_with_delete_background(...)` (update) | +| `src/interfaces/nextcloud/uploads_handler.rs` | `AudioMetadataService::spawn_extraction_background(...)` | +| `src/interfaces/api/handlers/webdav_handler.rs` | `AudioMetadataService::spawn_extraction_background(...)` | + +--- + +## Execution order + +1. `file_lifecycle.rs` — new trait +2. `blob_lifecycle.rs` — new trait +3. `file_lifecycle_service.rs` — updated composite +4. New `blob_lifecycle_service.rs` +5. `thumbnail_service.rs` — merged impl of both traits +6. `audio_metadata_service.rs` — new method + `FileLifecycleHook` impl +7. `file_upload_service.rs` — unified hook field + `is_new_blob` plumbing +8. `file_management_service.rs` — type update + copy hook +9. `trash_service.rs` — type update +10. `dedup_service.rs` — unified blob hook field +11. `di.rs` — rewire +12. Handler cleanups (4 files, independent) + +--- + +## Verification + +```bash +cargo fmt --all +cargo clippy --all-features --all-targets -- -D warnings # zero warnings +cargo test --workspace # all ~208 tests green +``` + +Smoke-test manually: +- Upload an image → thumbnail appears. +- Upload same image again (dedup hit) → no thumbnail re-generation. +- Upload an audio file via Nextcloud WebDAV → audio metadata present. +- Copy an image file → copy has thumbnail served instantly from blob_hash path. +- Copy a music file → copy has audio metadata (cloned row, no blob re-parse). +- Delete a file → thumbnails cleared; audio metadata gone (DB cascade). +- Overwrite file via WebDAV PUT → thumbnail refreshes. +- Delete last copy of a blob → blob-hash thumbnail file removed from disk. diff --git a/src/application/ports/blob_lifecycle.rs b/src/application/ports/blob_lifecycle.rs index 021f89cf..8238b301 100644 --- a/src/application/ports/blob_lifecycle.rs +++ b/src/application/ports/blob_lifecycle.rs @@ -1,35 +1,26 @@ -use std::future::Future; -use std::pin::Pin; +/// Observer notified by [`DedupService`] when a blob is stored for the first +/// time or permanently removed (ref_count reaches zero). +/// +/// Register with [`BlobLifecycleService`] during DI wiring; it fans out to all +/// registered hooks. Every implementor **must** provide both methods — +/// use an explicit one-liner noop for events the implementor does not care about. +/// This forces conscious acknowledgement of every lifecycle event rather than +/// silent omission. +/// +/// All methods are synchronous. Background work must be spawned inside the +/// implementor via `tokio::spawn`; the calling service never awaits hook +/// completion. +pub trait BlobLifecycleHook: Send + Sync { + /// Called after a genuinely new blob has been written to storage (no dedup + /// hit — first time this content hash is seen). + /// + /// `blob_hash` — BLAKE3 hex identifying the blob. + /// `content_type` — MIME type if known at write time, `None` otherwise. + fn on_blob_created(&self, blob_hash: &str, content_type: Option<&str>); -/// 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. -/// -/// Implement this trait on any service that needs to react to blob deletion -/// (e.g. thumbnail cleanup, CDN invalidation, audit logging). Register with -/// [`DedupService::add_blob_hook`] during DI wiring. -/// -/// The boxed-future return keeps the trait dyn-compatible so multiple -/// implementations can be stored as `Vec>`. -pub trait BlobDeletionHook: Send + Sync { - /// Called after the blob file has been removed from disk. - /// `blob_hash` is the BLAKE3 hex string identifying the blob. - /// Must be best-effort — must not propagate errors. - fn on_blob_deleted<'a>( - &'a self, - blob_hash: &'a str, - ) -> Pin + Send + 'a>>; + /// Called after a blob's ref_count reaches zero and it has been permanently + /// removed from storage. + /// + /// `blob_hash` — BLAKE3 hex identifying the (now deleted) blob. + fn on_blob_deleted(&self, blob_hash: &str); } diff --git a/src/application/ports/file_lifecycle.rs b/src/application/ports/file_lifecycle.rs index fe416e28..dbf57d39 100644 --- a/src/application/ports/file_lifecycle.rs +++ b/src/application/ports/file_lifecycle.rs @@ -1,57 +1,71 @@ -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). +/// Observer notified by file services when a file record is created, copied, +/// updated, or permanently deleted. /// -/// 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). +/// Register with [`FileLifecycleService`] during DI wiring; it fans out to all +/// registered hooks. Every implementor **must** provide all four methods — +/// use an explicit one-liner noop for events the implementor does not care about. +/// This forces conscious acknowledgement of every lifecycle event rather than +/// silent omission. /// -/// 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. +/// All methods are synchronous. Background work must be spawned inside the +/// implementor via `tokio::spawn`; the calling service never awaits hook +/// completion. +pub trait FileLifecycleHook: Send + Sync { + /// Called after a new file record has been persisted. /// - /// `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>>; + /// `blob_hash` — BLAKE3 hex of the content blob. + /// `content_type` — MIME type. + /// `is_new_blob` — `true` if the blob was stored for the first time (no + /// dedup hit); `false` if the blob already existed (re-upload of identical + /// content). Implementors can use this to skip re-generating artefacts that + /// are keyed by `blob_hash` and already exist on disk. + /// + /// For explicit file copies use [`on_file_copied`] instead — it supplies + /// the source file id so per-file metadata can be cloned directly. + fn on_file_created( + &self, + file_id: &str, + blob_hash: &str, + content_type: &str, + is_new_blob: bool, + ); + + /// Called after a file has been created as an explicit copy of an existing file. + /// + /// `file_id` — opaque file UUID string of the **new** copy. + /// `blob_hash` — BLAKE3 hex of the shared content blob. + /// `content_type` — MIME type. + /// `source_file_id` — opaque file UUID string of the **original** file. + /// + /// Implementors may use `source_file_id` to efficiently clone per-file + /// metadata (audio tags, etc.) from the original rather than re-deriving + /// it from the blob. If the original has not yet been processed, fall back + /// to a blob-hash-based lookup or schedule a retry — the implementor owns + /// race handling. + fn on_file_copied( + &self, + file_id: &str, + blob_hash: &str, + content_type: &str, + source_file_id: &str, + ); + + /// Called after an existing file's blob has been replaced (WebDAV PUT + /// overwrite, WOPI PutFile, Nextcloud chunked upload finalization). + /// + /// `file_id` — opaque file UUID string. + /// `blob_hash` — BLAKE3 hex of the **new** blob. + /// `content_type` — MIME type of the new content. + fn on_file_updated(&self, file_id: &str, blob_hash: &str, content_type: &str); + + /// Called after a file record has been permanently removed (direct delete + /// or emptied from trash). + /// + /// NOTE: due to deduplication the blob may still exist if other files + /// reference it. Use [`BlobLifecycleHook::on_blob_deleted`] when your + /// side-effect is content-addressed (e.g. removing blob-keyed thumbnails). + /// + /// `file_id` — opaque file UUID string. + fn on_file_deleted(&self, file_id: &str); } diff --git a/src/application/services/blob_lifecycle_service.rs b/src/application/services/blob_lifecycle_service.rs new file mode 100644 index 00000000..d1fe7587 --- /dev/null +++ b/src/application/services/blob_lifecycle_service.rs @@ -0,0 +1,44 @@ +use std::sync::Arc; + +use crate::application::ports::blob_lifecycle::BlobLifecycleHook; + +/// Composite dispatcher for blob lifecycle events. +/// +/// Aggregates all [`BlobLifecycleHook`] implementations and fans out each +/// event to every registered handler. Services hold a single +/// `Arc` — new handlers are added once, in DI, without +/// touching the services themselves. +pub struct BlobLifecycleService { + hooks: Vec>, +} + +impl Default for BlobLifecycleService { + fn default() -> Self { + Self::new() + } +} + +impl BlobLifecycleService { + pub fn new() -> Self { + Self { hooks: Vec::new() } + } + + pub fn with_hook(mut self, hook: Arc) -> Self { + self.hooks.push(hook); + self + } +} + +impl BlobLifecycleHook for BlobLifecycleService { + fn on_blob_created(&self, blob_hash: &str, content_type: Option<&str>) { + for hook in &self.hooks { + hook.on_blob_created(blob_hash, content_type); + } + } + + fn on_blob_deleted(&self, blob_hash: &str) { + for hook in &self.hooks { + hook.on_blob_deleted(blob_hash); + } + } +} diff --git a/src/application/services/file_lifecycle_service.rs b/src/application/services/file_lifecycle_service.rs index 50fece64..e71b95d7 100644 --- a/src/application/services/file_lifecycle_service.rs +++ b/src/application/services/file_lifecycle_service.rs @@ -1,17 +1,15 @@ -use std::future::Future; -use std::pin::Pin; use std::sync::Arc; -use crate::application::ports::file_lifecycle::FileDeletedHook; +use crate::application::ports::file_lifecycle::FileLifecycleHook; /// Composite dispatcher for file lifecycle events. /// -/// Aggregates all `FileDeletedHook` implementations and fans out each event to -/// every registered handler. Services hold a single `Arc` -/// pointing here — new handlers are added once, in DI, without touching the -/// services themselves. +/// Aggregates all [`FileLifecycleHook`] implementations and fans out each +/// event to every registered handler. Services hold a single +/// `Arc` — new handlers are added once, in DI, without +/// touching the services themselves. pub struct FileLifecycleService { - deleted: Vec>, + hooks: Vec>, } impl Default for FileLifecycleService { @@ -22,26 +20,49 @@ impl Default for FileLifecycleService { impl FileLifecycleService { pub fn new() -> Self { - Self { - deleted: Vec::new(), - } + Self { hooks: Vec::new() } } - pub fn with_deleted_hook(mut self, hook: Arc) -> Self { - self.deleted.push(hook); + pub fn with_hook(mut self, hook: Arc) -> Self { + self.hooks.push(hook); self } } -impl FileDeletedHook for FileLifecycleService { - fn on_file_deleted<'a>( - &'a self, - file_id: &'a str, - ) -> Pin + Send + 'a>> { - Box::pin(async move { - for hook in &self.deleted { - hook.on_file_deleted(file_id).await; - } - }) +impl FileLifecycleHook for FileLifecycleService { + fn on_file_created( + &self, + file_id: &str, + blob_hash: &str, + content_type: &str, + is_new_blob: bool, + ) { + for hook in &self.hooks { + hook.on_file_created(file_id, blob_hash, content_type, is_new_blob); + } + } + + fn on_file_copied( + &self, + file_id: &str, + blob_hash: &str, + content_type: &str, + source_file_id: &str, + ) { + for hook in &self.hooks { + hook.on_file_copied(file_id, blob_hash, content_type, source_file_id); + } + } + + fn on_file_updated(&self, file_id: &str, blob_hash: &str, content_type: &str) { + for hook in &self.hooks { + hook.on_file_updated(file_id, blob_hash, content_type); + } + } + + fn on_file_deleted(&self, file_id: &str) { + for hook in &self.hooks { + hook.on_file_deleted(file_id); + } } } diff --git a/src/application/services/file_management_service.rs b/src/application/services/file_management_service.rs index 117c4752..453a3a9c 100644 --- a/src/application/services/file_management_service.rs +++ b/src/application/services/file_management_service.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use crate::application::dtos::file_dto::FileDto; use crate::application::ports::authorization_ports::AuthorizationEngine; -use crate::application::ports::file_lifecycle::FileDeletedHook; +use crate::application::ports::file_lifecycle::FileLifecycleHook; use crate::application::ports::file_ports::FileManagementUseCase; use crate::application::ports::storage_ports::{CopyFolderTreeResult, FileWritePort}; use crate::application::ports::trash_ports::TrashUseCase; @@ -29,8 +29,8 @@ pub struct FileManagementService { trash_service: Option>, content_cache: Option>, authz: Arc, - /// Hook fired after a file is permanently deleted (typically the FileLifecycleService composite). - file_deleted_hook: Option>, + /// Lifecycle hook dispatcher — fired on file created (copy) and deleted. + file_lifecycle_hook: Option>, } impl FileManagementService { @@ -52,13 +52,13 @@ impl FileManagementService { trash_service, content_cache, authz, - file_deleted_hook: None, + file_lifecycle_hook: None, } } - /// Sets the lifecycle hook fired after a file is permanently deleted. - pub fn with_file_deleted_hook(mut self, hook: Arc) -> Self { - self.file_deleted_hook = Some(hook); + /// Sets the lifecycle hook dispatcher (thumbnails, audio metadata, …). + pub fn with_file_lifecycle_hook(mut self, hook: Arc) -> Self { + self.file_lifecycle_hook = Some(hook); self } @@ -149,7 +149,11 @@ impl FileManagementService { copied_file.folder_id() ); - Ok(FileDto::from(copied_file)) + 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); + } + Ok(dto) } async fn rename_file(&self, file_id: &str, new_name: &str) -> Result { @@ -185,8 +189,8 @@ impl FileManagementService { if let Some(cc) = &self.content_cache { cc.invalidate(id).await; } - if let Some(hook) = &self.file_deleted_hook { - hook.on_file_deleted(id).await; + if let Some(hook) = &self.file_lifecycle_hook { + hook.on_file_deleted(id); } info!("File permanently deleted: {}", id); Ok(()) diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index e160f7ab..811d92d3 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -2,7 +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_lifecycle::FileLifecycleHook; use crate::application::ports::file_ports::FileUploadUseCase; use crate::application::ports::storage_ports::{FileReadPort, FileWritePort}; use crate::application::services::storage_usage_service::StorageUsageService; @@ -53,10 +53,8 @@ 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>, + /// Single lifecycle dispatcher — fires on_file_created / on_file_updated. + file_lifecycle_hook: Option>, } impl FileUploadService { @@ -67,8 +65,7 @@ impl FileUploadService { file_read: None, storage_usage_service: None, content_cache: None, - file_created_hooks: Vec::new(), - file_updated_hooks: Vec::new(), + file_lifecycle_hook: None, } } @@ -82,8 +79,7 @@ impl FileUploadService { file_read: Some(file_read), storage_usage_service: None, content_cache: None, - file_created_hooks: Vec::new(), - file_updated_hooks: Vec::new(), + file_lifecycle_hook: None, } } @@ -93,15 +89,9 @@ 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); + /// Registers the lifecycle hook dispatcher (thumbnails, audio metadata, …). + pub fn with_file_lifecycle_hook(mut self, hook: Arc) -> Self { + self.file_lifecycle_hook = Some(hook); self } @@ -153,9 +143,9 @@ impl FileUploadUseCase for FileUploadService { size: u64, pre_computed_hash: Option, ) -> Result { - let file = self + let (file, is_new_blob) = self .file_write - .save_file_from_temp( + .save_file_from_temp_with_dedup( name.clone(), folder_id, content_type, @@ -170,9 +160,8 @@ 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; + if let Some(hook) = &self.file_lifecycle_hook { + hook.on_file_created(&dto.id, &dto.etag, &dto.mime_type, is_new_blob); } Ok(dto) } @@ -222,7 +211,6 @@ impl FileUploadUseCase for FileUploadService { // Look up the folder ID by folder path let parent_id = if !parent_path.is_empty() { if let Some(file_read) = &self.file_read { - // Use get_folder_id_by_path to look up the folder directly file_read.get_folder_id_by_path(parent_path).await.ok() } else { None @@ -241,9 +229,9 @@ impl FileUploadUseCase for FileUploadService { .await .map_err(|e| DomainError::internal_error("FileUpload", format!("hash: {e}")))?; - let file = self + let (file, is_new_blob) = self .file_write - .save_file_from_temp( + .save_file_from_temp_with_dedup( filename.to_string(), parent_id, content_type.to_string(), @@ -254,6 +242,9 @@ impl FileUploadUseCase for FileUploadService { .await?; 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); + } Ok(dto) } @@ -327,9 +318,8 @@ impl FileUploadUseCase for FileUploadService { // Re-read to get fresh DTO with updated etag and timestamps. let updated = file_read.get_file(&file_id).await?; let dto = FileDto::from(updated); - for hook in &self.file_updated_hooks { - hook.on_file_updated(&file_id, &dto.etag, content_type) - .await; + if let Some(hook) = &self.file_lifecycle_hook { + hook.on_file_updated(&file_id, &dto.etag, content_type); } return Ok(dto); } @@ -354,9 +344,9 @@ impl FileUploadUseCase for FileUploadService { None }; - let created = self + let (created, is_new_blob) = self .file_write - .save_file_from_temp( + .save_file_from_temp_with_dedup( filename.to_string(), parent_id, content_type.to_string(), @@ -365,6 +355,10 @@ impl FileUploadUseCase for FileUploadService { pre_computed_hash, ) .await?; - Ok(FileDto::from(created)) + 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); + } + Ok(dto) } } diff --git a/src/application/services/mod.rs b/src/application/services/mod.rs index d3d1b072..bfa7085d 100644 --- a/src/application/services/mod.rs +++ b/src/application/services/mod.rs @@ -2,6 +2,7 @@ pub mod admin_settings_service; pub mod app_password_service; pub mod auth_application_service; pub mod batch_operations; +pub mod blob_lifecycle_service; pub mod calendar_service; pub mod contact_service; pub mod device_auth_service; diff --git a/src/application/services/trash_service.rs b/src/application/services/trash_service.rs index 911004f7..67bf1c72 100644 --- a/src/application/services/trash_service.rs +++ b/src/application/services/trash_service.rs @@ -7,7 +7,7 @@ use crate::application::dtos::display_helpers::{ }; use crate::application::dtos::trash_dto::TrashedItemDto; use crate::application::ports::authorization_ports::AuthorizationEngine; -use crate::application::ports::file_lifecycle::FileDeletedHook; +use crate::application::ports::file_lifecycle::FileLifecycleHook; use crate::application::ports::storage_ports::{FileReadPort, FileWritePort}; use crate::application::ports::trash_ports::TrashUseCase; use crate::common::errors::{DomainError, ErrorKind, Result}; @@ -53,8 +53,8 @@ pub struct TrashService { /// orphaned blob files and thumbnails that the PG trigger cannot reach. dedup_service: Arc, - /// Hook fired after a file is permanently deleted (typically the FileLifecycleService composite). - file_deleted_hook: Option>, + /// Lifecycle hook dispatcher — fired on file permanently deleted. + file_deleted_hook: Option>, /// Content cache — invalidated when files are permanently deleted from trash. content_cache: Option>, @@ -91,8 +91,8 @@ impl TrashService { } } - /// Sets the lifecycle hook fired after a file is permanently deleted. - pub fn with_file_deleted_hook(mut self, hook: Arc) -> Self { + /// Sets the lifecycle hook dispatcher (thumbnails, audio metadata, …). + pub fn with_file_deleted_hook(mut self, hook: Arc) -> Self { self.file_deleted_hook = Some(hook); self } @@ -583,7 +583,7 @@ impl TrashUseCase for TrashService { } if let Some(hook) = &self.file_deleted_hook { - hook.on_file_deleted(&file_id).await; + hook.on_file_deleted(&file_id); } } TrashedItemType::Folder => { @@ -723,7 +723,7 @@ impl TrashUseCase for TrashService { if let Some(hook) = &self.file_deleted_hook { for file_id in &trashed_file_ids { - hook.on_file_deleted(file_id).await; + hook.on_file_deleted(file_id); } } diff --git a/src/common/di.rs b/src/common/di.rs index f67f5489..65d42c01 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -42,6 +42,7 @@ use crate::infrastructure::services::pg_acl_engine::PgAclEngine; use crate::infrastructure::services::trash_cleanup_service::TrashCleanupService; use crate::application::services::app_password_service::AppPasswordService; +use crate::application::services::blob_lifecycle_service::BlobLifecycleService; use crate::application::services::calendar_service::CalendarService; use crate::application::services::device_auth_service::DeviceAuthService; use crate::application::services::file_lifecycle_service::FileLifecycleService; @@ -260,6 +261,12 @@ impl AppServiceFactory { tracing::info!("Blob storage LRU disk cache enabled"); } + // Blob lifecycle — thumbnail disk-file cleanup when blob ref_count hits zero. + // ThumbnailService (not ThumbnailRefreshHook) is used here to avoid a circular + // Arc: DedupService→BlobLifecycleService→ThumbnailRefreshHook→DedupService. + let blob_lifecycle = + Arc::new(BlobLifecycleService::new().with_hook(thumbnail_service.clone())); + // Deduplication service — PRIMARY blob storage engine (PostgreSQL-backed index) let dedup_service = Arc::new( crate::infrastructure::services::dedup_service::DedupService::new( @@ -267,7 +274,7 @@ impl AppServiceFactory { db_pool.clone(), maintenance_pool.clone(), ) - .add_blob_hook(thumbnail_service.clone()), + .with_blob_lifecycle(blob_lifecycle), ); dedup_service.initialize().await?; @@ -275,14 +282,30 @@ impl AppServiceFactory { "Core services initialized: path service, file content cache, thumbnails, chunked upload, image transcode, dedup (PRIMARY blob storage)" ); - let file_lifecycle = - Arc::new(FileLifecycleService::new().with_deleted_hook(thumbnail_service.clone())); + // Audio metadata service — created here so it can be wired into file_lifecycle. + let audio_metadata_service = self.create_audio_metadata_service(db_pool); + + // ThumbnailRefreshHook: handles FileLifecycleHook events (create/update/delete). + // Implemented on ThumbnailRefreshHook (not ThumbnailService) to avoid circular Arc: + // DedupService → BlobLifecycleService → ThumbnailRefreshHook → DedupService. + let thumbnail_refresh_hook = Arc::new(ThumbnailRefreshHook::new( + thumbnail_service.clone(), + dedup_service.clone(), + )); + + // Build the unified FileLifecycleService dispatcher. + let mut fls = FileLifecycleService::new().with_hook(thumbnail_refresh_hook); + if let Some(audio) = &audio_metadata_service { + fls = fls.with_hook(audio.clone()); + } + let file_lifecycle = Arc::new(fls); Ok(CoreServices { path_service, file_content_cache, thumbnail_service, file_lifecycle, + audio_metadata_service, chunked_upload_service, image_transcode_service, dedup_service, @@ -355,7 +378,6 @@ impl AppServiceFactory { core: &CoreServices, repos: &RepositoryServices, trash_service: Option>, - db_pool: &Arc, authz: &Arc, ) -> ApplicationServices { // Main services @@ -364,20 +386,13 @@ impl AppServiceFactory { authz.clone(), )); - // 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_file_created_hook(thumbnail_refresh_hook.clone()) - .with_file_updated_hook(thumbnail_refresh_hook), + .with_file_lifecycle_hook(core.file_lifecycle.clone()), ); let file_retrieval_service = Arc::new(FileRetrievalService::new_with_cache( @@ -397,7 +412,7 @@ impl AppServiceFactory { Some(core.file_content_cache.clone()), authz.clone(), ) - .with_file_deleted_hook(core.file_lifecycle.clone()), + .with_file_lifecycle_hook(core.file_lifecycle.clone()), ); let file_use_case_factory = Arc::new(AppFileUseCaseFactory::new( @@ -433,7 +448,7 @@ impl AppServiceFactory { share_service: None, // Configured later with create_share_service favorites_service: None, // Configured later with create_favorites_service recent_service: None, // Configured later with create_recent_service - audio_metadata_service: self.create_audio_metadata_service(db_pool), + audio_metadata_service: core.audio_metadata_service.clone(), } } @@ -633,13 +648,8 @@ impl AppServiceFactory { .await; // 4. Application services (with trash + authz already wired) - let mut apps = self.create_application_services( - &core, - &repos, - trash_service.clone(), - &pool, - &authorization, - ); + let mut apps = + self.create_application_services(&core, &repos, trash_service.clone(), &authorization); // 5. Share service let share_service = self.create_share_service(&repos, &pool); @@ -1028,8 +1038,9 @@ pub struct CoreServices { pub path_service: Arc, pub file_content_cache: Arc, pub thumbnail_service: Arc, - /// Composite lifecycle dispatcher — register new permanent-delete hooks here only. + /// Composite lifecycle dispatcher — wires thumbnails + audio metadata for all file events. pub file_lifecycle: Arc, + pub audio_metadata_service: Option>, pub chunked_upload_service: Arc, pub image_transcode_service: Arc, pub dedup_service: Arc, diff --git a/src/infrastructure/repositories/pg/file_blob_write_repository.rs b/src/infrastructure/repositories/pg/file_blob_write_repository.rs index e0c1e7a5..942f4615 100644 --- a/src/infrastructure/repositories/pg/file_blob_write_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_write_repository.rs @@ -202,10 +202,11 @@ impl FileBlobWriteRepository { Ok(new_hash.to_string()) } -} -impl FileWritePort for FileBlobWriteRepository { - async fn save_file_from_temp( + /// Like [`FileWritePort::save_file_from_temp`] but also returns whether the + /// blob was genuinely new (`true`) or a dedup hit (`false`). + /// Used by [`FileUploadService`] to pass `is_new_blob` to lifecycle hooks. + pub async fn save_file_from_temp_with_dedup( &self, name: String, folder_id: Option, @@ -213,18 +214,16 @@ impl FileWritePort for FileBlobWriteRepository { temp_path: &std::path::Path, size: u64, pre_computed_hash: Option, - ) -> Result { + ) -> Result<(File, bool), DomainError> { let user_id = self.resolve_user_id(folder_id.as_deref()).await?; - // True streaming: pass pre-computed hash (or let dedup compute it). - // When hash is pre-computed, zero extra disk reads. let dedup_result = self .dedup .store_from_file(temp_path, Some(content_type.clone()), pre_computed_hash) .await?; + let is_new_blob = !dedup_result.was_deduplicated(); let blob_hash = dedup_result.hash().to_string(); - // Insert file metadata — if this fails, compensate by removing the blob ref let row = match sqlx::query_as::<_, (String, i64, i64)>( r#" INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type) @@ -275,7 +274,7 @@ impl FileWritePort for FileBlobWriteRepository { ); let folder_path = self.lookup_folder_path(folder_id.as_deref()).await?; - Self::row_to_file( + let file = Self::row_to_file( row.0, name, folder_id, @@ -285,8 +284,32 @@ impl FileWritePort for FileBlobWriteRepository { row.1, row.2, Some(user_id), - blob_hash.clone(), + blob_hash, + )?; + Ok((file, is_new_blob)) + } +} + +impl FileWritePort for FileBlobWriteRepository { + async fn save_file_from_temp( + &self, + name: String, + folder_id: Option, + content_type: String, + temp_path: &std::path::Path, + size: u64, + pre_computed_hash: Option, + ) -> Result { + self.save_file_from_temp_with_dedup( + name, + folder_id, + content_type, + temp_path, + size, + pre_computed_hash, ) + .await + .map(|(file, _)| file) } async fn move_file( diff --git a/src/infrastructure/services/audio_metadata_service.rs b/src/infrastructure/services/audio_metadata_service.rs index 303b2e6a..8591043f 100644 --- a/src/infrastructure/services/audio_metadata_service.rs +++ b/src/infrastructure/services/audio_metadata_service.rs @@ -6,6 +6,7 @@ use std::sync::Arc; use tracing::{info, warn}; use uuid::Uuid; +use crate::application::ports::file_lifecycle::FileLifecycleHook; use crate::common::errors::DomainError; #[derive(Debug, FromRow)] @@ -231,6 +232,101 @@ impl AudioMetadataService { failed, }) } + + /// Copy audio metadata from an existing file that shares the same blob. + /// + /// Used when `is_new_blob=false` (copy/dedup hit): instead of re-parsing + /// the blob, clone the existing metadata row for `new_file_id`. Falls back + /// Clones audio metadata from a known source file, falling back to + /// [`clone_or_extract_background`] if the source has not been processed yet. + pub fn clone_from_source_background( + service: Arc, + new_file_id: Uuid, + source_file_id: Uuid, + blob_hash: String, + ) { + tokio::spawn(async move { + let result = sqlx::query( + r#" + INSERT INTO audio.file_metadata + (file_id, title, artist, album, album_artist, genre, + track_number, disc_number, year, duration_secs, format) + SELECT $1, title, artist, album, album_artist, genre, + track_number, disc_number, year, duration_secs, format + FROM audio.file_metadata + WHERE file_id = $2 + ON CONFLICT (file_id) DO NOTHING + "#, + ) + .bind(new_file_id) + .bind(source_file_id) + .execute(&*service.pool) + .await; + + match result { + Ok(r) if r.rows_affected() > 0 => { + info!( + "Cloned audio metadata from {} to {}", + source_file_id, new_file_id + ); + } + Ok(_) => { + // Source not yet processed — fall back to blob-hash lookup or extraction. + Self::clone_or_extract_background(service, new_file_id, blob_hash); + } + Err(e) => { + warn!( + "Failed to clone audio metadata from {} to {}: {}", + source_file_id, new_file_id, e + ); + } + } + }); + } + + /// to full extraction if no existing row is found (race: original not yet + /// processed). + pub fn clone_or_extract_background(service: Arc, new_file_id: Uuid, blob_hash: String) { + tokio::spawn(async move { + let rows_inserted = sqlx::query( + r#" + INSERT INTO audio.file_metadata + (file_id, title, artist, album, album_artist, genre, + track_number, disc_number, year, duration_secs, format) + SELECT $1, title, artist, album, album_artist, genre, + track_number, disc_number, year, duration_secs, format + FROM audio.file_metadata am + JOIN storage.files sf ON sf.id = am.file_id + WHERE sf.blob_hash = $2 + LIMIT 1 + ON CONFLICT (file_id) DO NOTHING + "#, + ) + .bind(new_file_id) + .bind(&blob_hash) + .execute(&*service.pool) + .await; + + match rows_inserted { + Ok(result) if result.rows_affected() > 0 => { + info!("Cloned audio metadata for file {}", new_file_id); + } + Ok(_) => { + // No existing metadata found — original not yet processed; fall back. + let file_path = service.blob_path(&blob_hash); + if let Err(e) = service.extract_and_save(&new_file_id, &file_path).await { + warn!( + "Failed to extract audio metadata for {}: {}", + new_file_id, e + ); + } + } + Err(e) => { + warn!("Failed to clone audio metadata for {}: {}", new_file_id, e); + } + } + }); + } } /// Extracted audio metadata fields transferred from the blocking thread. @@ -252,3 +348,79 @@ pub struct MetadataExtractionResult { pub processed: usize, pub failed: usize, } + +// ─── FileLifecycleHook ─────────────────────────────────────────────────────── + +impl FileLifecycleHook for AudioMetadataService { + fn on_file_created( + &self, + file_id: &str, + blob_hash: &str, + content_type: &str, + is_new_blob: bool, + ) { + if !Self::is_audio_file(content_type) { + return; + } + let Ok(uuid) = file_id.parse::() else { + warn!("on_file_created: invalid file_id UUID: {}", file_id); + return; + }; + let service = Arc::new(Self { + pool: self.pool.clone(), + blob_root: self.blob_root.clone(), + }); + if is_new_blob { + Self::spawn_extraction_background(service, uuid, self.blob_path(blob_hash)); + } else { + Self::clone_or_extract_background(service, uuid, blob_hash.to_string()); + } + } + + fn on_file_copied( + &self, + file_id: &str, + blob_hash: &str, + content_type: &str, + source_file_id: &str, + ) { + if !Self::is_audio_file(content_type) { + return; + } + let Ok(uuid) = file_id.parse::() else { + warn!("on_file_copied: invalid file_id UUID: {}", file_id); + return; + }; + let Ok(source_uuid) = source_file_id.parse::() else { + warn!( + "on_file_copied: invalid source_file_id UUID: {}", + source_file_id + ); + return; + }; + let service = Arc::new(Self { + pool: self.pool.clone(), + blob_root: self.blob_root.clone(), + }); + Self::clone_from_source_background(service, uuid, source_uuid, blob_hash.to_string()); + } + + fn on_file_updated(&self, file_id: &str, blob_hash: &str, content_type: &str) { + if !Self::is_audio_file(content_type) { + return; + } + let Ok(uuid) = file_id.parse::() else { + warn!("on_file_updated: invalid file_id UUID: {}", file_id); + return; + }; + let service = Arc::new(Self { + pool: self.pool.clone(), + blob_root: self.blob_root.clone(), + }); + Self::spawn_extraction_with_delete_background(service, uuid, self.blob_path(blob_hash)); + } + + fn on_file_deleted(&self, _file_id: &str) { + // audio.file_metadata has ON DELETE CASCADE on file_id — DB handles cleanup. + } +} diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index feef5785..e4c81a5c 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -44,11 +44,12 @@ use std::sync::Arc; use tokio::fs; use tokio::io::{AsyncReadExt, AsyncSeekExt}; -use crate::application::ports::blob_lifecycle::{BlobCreationHook, BlobDeletionHook}; +use crate::application::ports::blob_lifecycle::BlobLifecycleHook; use crate::application::ports::blob_storage_ports::BlobStorageBackend; use crate::application::ports::dedup_ports::{ BlobMetadataDto, DedupPort, DedupResultDto, DedupStatsDto, }; +use crate::application::services::blob_lifecycle_service::BlobLifecycleService; use crate::domain::errors::{DomainError, ErrorKind}; // ── CDC Constants ──────────────────────────────────────────────────────────── @@ -84,10 +85,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>, + /// Single lifecycle dispatcher — fired on blob created / deleted. + blob_lifecycle: Option>, } impl DedupService { @@ -105,36 +104,25 @@ impl DedupService { backend, pool, maintenance_pool, - blob_creation_hooks: vec![], - blob_hooks: vec![], + blob_lifecycle: None, } } - /// 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); + /// Registers the blob lifecycle dispatcher (thumbnail cleanup, …). + pub fn with_blob_lifecycle(mut self, lifecycle: Arc) -> Self { + self.blob_lifecycle = Some(lifecycle); 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 { - self.blob_hooks.push(hook); - 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; + fn fire_blob_creation_hooks(&self, hash: &str, content_type: Option<&str>) { + if let Some(lc) = &self.blob_lifecycle { + lc.on_blob_created(hash, content_type); } } - /// Fire all registered hooks for a deleted blob. - async fn fire_blob_hooks(&self, hash: &str) { - for hook in &self.blob_hooks { - hook.on_blob_deleted(hash).await; + fn fire_blob_hooks(&self, hash: &str) { + if let Some(lc) = &self.blob_lifecycle { + lc.on_blob_deleted(hash); } } @@ -152,8 +140,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![], + blob_lifecycle: None, } } @@ -371,8 +358,7 @@ impl DedupService { chunk_hashes.len() ); - self.fire_blob_creation_hooks(&file_hash, content_type.as_deref()) - .await; + self.fire_blob_creation_hooks(&file_hash, content_type.as_deref()); Ok(DedupResultDto::NewBlob { hash: file_hash, @@ -813,7 +799,7 @@ impl DedupService { } // Bug 4 fix: notify hooks — e.g. thumbnail cleanup keyed by file_hash - self.fire_blob_hooks(file_hash).await; + self.fire_blob_hooks(file_hash); tracing::info!( "MANIFEST DELETED: {} ({} chunks, {} orphan chunks removed)", @@ -893,7 +879,7 @@ impl DedupService { } // Bug 3 fix: notify hooks — e.g. thumbnail cleanup keyed by hash - self.fire_blob_hooks(hash).await; + self.fire_blob_hooks(hash); tracing::info!("BLOB DELETED: {} (no more references)", &hash[..12]); Ok(true) @@ -1000,7 +986,7 @@ impl DedupService { if let Err(e) = self.backend.delete_blob(hash).await { tracing::warn!("cleanup_if_orphaned: disk delete failed for {short}: {e}"); } - self.fire_blob_hooks(hash).await; + self.fire_blob_hooks(hash); tracing::info!("cleanup_if_orphaned: removed orphaned blob {short}"); } } @@ -1471,7 +1457,7 @@ impl DedupService { if let Err(e) = self.backend.delete_blob(hash).await { tracing::warn!("Failed to delete orphan blob {hash}: {e}"); } - self.fire_blob_hooks(hash).await; + self.fire_blob_hooks(hash); total_bytes += *size as u64; } total_deleted += batch.len() as u64; diff --git a/src/infrastructure/services/thumbnail_service.rs b/src/infrastructure/services/thumbnail_service.rs index 5f934aa3..23924f1c 100644 --- a/src/infrastructure/services/thumbnail_service.rs +++ b/src/infrastructure/services/thumbnail_service.rs @@ -984,23 +984,12 @@ impl ThumbnailService { } } -// ─── BlobDeletionHook ──────────────────────────────────────────────────────── +// ─── FileLifecycleHook + BlobLifecycleHook ─────────────────────────────────── -impl crate::application::ports::blob_lifecycle::BlobDeletionHook for ThumbnailService { - fn on_blob_deleted<'a>( - &'a self, - blob_hash: &'a str, - ) -> std::pin::Pin + Send + 'a>> { - Box::pin(async move { self.delete_blob_thumbnails(blob_hash).await }) - } -} - -// ─── FileUpdatedHook ───────────────────────────────────────────────────────── - -/// Wires thumbnail invalidation + regeneration into the file-update lifecycle. +/// Wires all thumbnail side-effects into the file and blob lifecycle. /// -/// Registered on [`FileUploadService`] during DI. Fires whenever a file's blob -/// is replaced (WebDAV PUT overwrite, WOPI PutFile, Nextcloud chunked upload). +/// Registered once on both [`FileLifecycleService`] and [`BlobLifecycleService`] +/// during DI. Handles thumbnail generation, invalidation, and cleanup. pub struct ThumbnailRefreshHook { thumbnail: Arc, dedup: Arc, @@ -1012,54 +1001,70 @@ impl ThumbnailRefreshHook { } } -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 { +impl crate::application::ports::file_lifecycle::FileLifecycleHook for ThumbnailRefreshHook { + fn on_file_created( + &self, + file_id: &str, + blob_hash: &str, + content_type: &str, + is_new_blob: bool, + ) { + // Blob-hash thumbnail already exists on disk when is_new_blob=false — skip. + 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(), + ); + } + + fn on_file_copied( + &self, + _file_id: &str, + _blob_hash: &str, + _content_type: &str, + _source_file_id: &str, + ) { + // Thumbnails are keyed by blob_hash on disk — the copy shares them automatically. + } + + fn on_file_updated(&self, file_id: &str, blob_hash: &str, content_type: &str) { + if !ThumbnailService::is_supported_image(content_type) { + return; + } + let thumbnail = self.thumbnail.clone(); + let file_id = file_id.to_string(); + let blob_hash = blob_hash.to_string(); + let dedup = self.dedup.clone(); + tokio::spawn(async move { + if let Err(e) = 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(), - ); - }) + Self::spawn_thumbnail_generation(thumbnail, dedup, file_id, blob_hash); + }); + } + + fn on_file_deleted(&self, file_id: &str) { + let thumbnail = self.thumbnail.clone(); + let file_id = file_id.to_string(); + tokio::spawn(async move { + if let Err(e) = thumbnail.delete_thumbnails(&file_id).await { + tracing::warn!("Failed to delete thumbnails for file {}: {}", file_id, e); + } + }); } } -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(), - ); - }) - } -} +// BlobLifecycleHook is implemented on ThumbnailService (not ThumbnailRefreshHook) +// to avoid a circular Arc: DedupService→BlobLifecycleService→ThumbnailRefreshHook→DedupService. +// ThumbnailService does not hold DedupService so no cycle exists. impl ThumbnailRefreshHook { fn spawn_thumbnail_generation( @@ -1085,18 +1090,31 @@ impl ThumbnailRefreshHook { } } -// ─── FileDeletedHook ───────────────────────────────────────────────────────── +// ─── BlobLifecycleHook ─────────────────────────────────────────────────────── -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); +impl crate::application::ports::blob_lifecycle::BlobLifecycleHook for ThumbnailService { + fn on_blob_created(&self, _blob_hash: &str, _content_type: Option<&str>) { + // Thumbnail generation is driven by file-level events (on_file_created). + } + + fn on_blob_deleted(&self, blob_hash: &str) { + // delete_blob_thumbnails only needs thumbnails_root — capture it to avoid Arc cycle. + let root = self.thumbnails_root.clone(); + let blob_hash = blob_hash.to_string(); + tokio::spawn(async move { + for size in ThumbnailSize::all() { + let path = root + .join(size.dir_name()) + .join(format!("{}.jpg", &blob_hash)); + if tokio::fs::metadata(&path).await.is_ok() { + let _ = tokio::fs::remove_file(&path).await; + } } - }) + tracing::debug!( + "🗑️ Deleted blob thumbnails for hash: {}…", + &blob_hash[..blob_hash.len().min(12)] + ); + }); } } diff --git a/src/interfaces/api/handlers/file_handler.rs b/src/interfaces/api/handlers/file_handler.rs index b0fef948..c664b55f 100644 --- a/src/interfaces/api/handlers/file_handler.rs +++ b/src/interfaces/api/handlers/file_handler.rs @@ -18,7 +18,6 @@ use crate::application::ports::storage_ports::{FileReadPort, StorageUsagePort}; use crate::application::ports::thumbnail_ports::ThumbnailPort; use crate::application::ports::{file_ports::OptimizedFileContent, folder_ports::FolderUseCase}; use crate::common::di::AppState; -use crate::infrastructure::services::audio_metadata_service::AudioMetadataService; use crate::interfaces::errors::AppError; use crate::interfaces::middleware::auth::AuthUser; use crate::{application::dtos::file_dto::FileDto, domain::services::authorization::Permission}; @@ -781,61 +780,11 @@ impl FileHandler { auth_user: AuthUser, multipart: Multipart, ) -> impl IntoResponse { - let (file, blob_hash) = match Self::upload_file_inner(&state, &auth_user, multipart).await { + let (file, _) = match Self::upload_file_inner(&state, &auth_user, multipart).await { Ok(pair) => pair, Err(response) => return response.into_response(), }; - // Generate thumbnails for supported images in background. - // The blob_hash was already computed during the hash-on-write spool, - // so we can reconstruct the source bytes directly from DedupService - // without an extra DB round-trip. - if state - .core - .thumbnail_service - .is_supported_image(&file.mime_type) - { - let file_id = file.id.clone(); - let thumbnail_service = state.core.thumbnail_service.clone(); - let dedup_service = state.core.dedup_service.clone(); - let blob_hash_owned = blob_hash.clone(); - - tokio::spawn(async move { - tracing::info!("🖼️ Generating thumbnails for: {}", file_id); - match dedup_service.read_blob_bytes(&blob_hash_owned).await { - Ok(original_bytes) => { - thumbnail_service.generate_all_sizes_background_from_bytes( - file_id, - blob_hash_owned, - original_bytes, - dedup_service.clone(), - ); - } - Err(err) => { - tracing::warn!( - "Failed to load source image for thumbnail generation {}: {}", - file_id, - err - ); - } - } - }); - } - - // TODO: same remark: a hook to handle easily audio service - // Extract audio metadata for supported audio files in background. - if let Some(ref audio_service) = state.applications.audio_metadata_service - && AudioMetadataService::is_audio_file(&file.mime_type) - && let Ok(file_id) = uuid::Uuid::parse_str(&file.id) - { - let file_path = state.core.dedup_service.blob_path(&blob_hash); - AudioMetadataService::spawn_extraction_background( - audio_service.clone(), - file_id, - file_path, - ); - } - Self::created_json_response(&file).into_response() } diff --git a/src/interfaces/api/handlers/webdav_handler.rs b/src/interfaces/api/handlers/webdav_handler.rs index 9d91f329..3a61d8ba 100644 --- a/src/interfaces/api/handlers/webdav_handler.rs +++ b/src/interfaces/api/handlers/webdav_handler.rs @@ -27,7 +27,6 @@ use crate::application::ports::storage_ports::StorageUsagePort; use crate::application::services::file_retrieval_service::FileRetrievalService; use crate::application::services::folder_service::FolderService; use crate::common::di::AppState; -use crate::infrastructure::services::audio_metadata_service::AudioMetadataService; use crate::infrastructure::services::path_resolver_service::ResolvedResource; use crate::interfaces::errors::AppError; use crate::interfaces::middleware::auth::{AuthUser, CurrentUser}; @@ -958,25 +957,10 @@ async fn handle_put( let _ = tokio::fs::remove_file(&temp_path).await; match result { - Ok(file_dto) => { - // Extract audio metadata for supported audio files in background. - if let Some(ref audio_service) = state.applications.audio_metadata_service - && AudioMetadataService::is_audio_file(&file_dto.mime_type) - && let Ok(file_id) = Uuid::parse_str(&file_dto.id) - { - let file_path = state.core.dedup_service.blob_path(&file_dto.etag); - AudioMetadataService::spawn_extraction_background( - audio_service.clone(), - file_id, - file_path, - ); - } - - Ok(Response::builder() - .status(StatusCode::NO_CONTENT) - .body(Body::empty()) - .unwrap()) - } + Ok(_file_dto) => Ok(Response::builder() + .status(StatusCode::NO_CONTENT) + .body(Body::empty()) + .unwrap()), Err(e) => Err(AppError::internal_error(format!( "Failed to put file: {}", e diff --git a/src/interfaces/nextcloud/uploads_handler.rs b/src/interfaces/nextcloud/uploads_handler.rs index e49522a9..52607941 100644 --- a/src/interfaces/nextcloud/uploads_handler.rs +++ b/src/interfaces/nextcloud/uploads_handler.rs @@ -8,7 +8,6 @@ use std::sync::Arc; use crate::application::ports::file_ports::{FileRetrievalUseCase, FileUploadUseCase}; use crate::common::di::AppState; use crate::common::mime_detect::{filename_from_path, refine_content_type_from_file}; -use crate::infrastructure::services::audio_metadata_service::AudioMetadataService; use crate::interfaces::errors::AppError; use crate::interfaces::middleware::auth::{AuthUser, CurrentUser}; @@ -187,19 +186,6 @@ async fn handle_assemble( .await .map_err(|e| AppError::internal_error(format!("Failed to create file: {}", e)))?; - // Extract audio metadata for supported audio files in background. - if let Some(ref audio_service) = state.applications.audio_metadata_service - && AudioMetadataService::is_audio_file(&dto.mime_type) - && let Ok(file_id) = uuid::Uuid::parse_str(&dto.id) - { - let file_path = state.core.dedup_service.blob_path(&dto.etag); - AudioMetadataService::spawn_extraction_background( - audio_service.clone(), - file_id, - file_path, - ); - } - Some(dto.etag) }; diff --git a/src/interfaces/nextcloud/webdav_handler.rs b/src/interfaces/nextcloud/webdav_handler.rs index a4a1cdb4..ca0f0ce1 100644 --- a/src/interfaces/nextcloud/webdav_handler.rs +++ b/src/interfaces/nextcloud/webdav_handler.rs @@ -21,7 +21,6 @@ use crate::application::ports::folder_ports::FolderUseCase; use crate::application::ports::trash_ports::TrashUseCase; use crate::common::di::AppState; use crate::common::mime_detect::{filename_from_path, refine_content_type}; -use crate::infrastructure::services::audio_metadata_service::AudioMetadataService; use crate::interfaces::errors::AppError; use crate::interfaces::middleware::auth::{AuthUser, CurrentUser}; @@ -550,23 +549,6 @@ async fn handle_put( .await .map_err(|e| AppError::internal_error(format!("Failed to update file: {}", e)))?; - // Update audio metadata for supported audio files. - // TODO: use notification service or hook - if let Some(ref audio_service) = state.applications.audio_metadata_service - && let Ok(file_id) = uuid::Uuid::parse_str(&updated.id) - { - let file_path = state.core.dedup_service.blob_path(&updated.etag); - let is_audio = AudioMetadataService::is_audio_file(&content_type); - - if is_audio { - AudioMetadataService::spawn_extraction_with_delete_background( - audio_service.clone(), - file_id, - file_path, - ); - } - } - return Ok(Response::builder() .status(StatusCode::NO_CONTENT) .header(header::ETAG, format!("\"{}\"", updated.etag)) @@ -588,19 +570,6 @@ async fn handle_put( .await .map_err(|e| AppError::internal_error(format!("Failed to create file: {}", e)))?; - // Extract audio metadata for supported audio files in background. - if let Some(ref audio_service) = state.applications.audio_metadata_service - && AudioMetadataService::is_audio_file(&file_dto.mime_type) - && let Ok(file_id) = uuid::Uuid::parse_str(&file_dto.id) - { - let file_path = state.core.dedup_service.blob_path(&file_dto.etag); - AudioMetadataService::spawn_extraction_background( - audio_service.clone(), - file_id, - file_path, - ); - } - let builder = Response::builder() .status(StatusCode::CREATED) .header(header::ETAG, format!("\"{}\"", file_dto.etag))