refactor(services): add file_lifecycle and blob_lifecycle
- complete src/application/ports/blob_lifecycle.rs with traits:
* BlobCreationHook
* BlobDeletionHook
- add src/application/ports/file_lifecycle.rs with traits:
* FileCreatedHook
* FileDeletedHook
* FileUpdatedHook
This commit is contained in:
@@ -1,6 +1,20 @@
|
|||||||
use std::future::Future;
|
use std::future::Future;
|
||||||
use std::pin::Pin;
|
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<Box<dyn Future<Output = ()> + Send + 'a>>;
|
||||||
|
}
|
||||||
/// Observer notified by [`DedupService`] when a blob's ref_count reaches zero
|
/// Observer notified by [`DedupService`] when a blob's ref_count reaches zero
|
||||||
/// and it is permanently removed from storage.
|
/// and it is permanently removed from storage.
|
||||||
///
|
///
|
||||||
|
|||||||
@@ -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<Box<dyn Future<Output = ()> + 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<Arc<dyn FileUpdatedHook>>`.
|
||||||
|
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<Box<dyn Future<Output = ()> + 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<Box<dyn Future<Output = ()> + Send + 'a>>;
|
||||||
|
}
|
||||||
@@ -8,6 +8,7 @@ pub mod chunked_upload_ports;
|
|||||||
pub mod compression_ports;
|
pub mod compression_ports;
|
||||||
pub mod dedup_ports;
|
pub mod dedup_ports;
|
||||||
pub mod favorites_ports;
|
pub mod favorites_ports;
|
||||||
|
pub mod file_lifecycle;
|
||||||
pub mod file_ports;
|
pub mod file_ports;
|
||||||
pub mod inbound;
|
pub mod inbound;
|
||||||
pub mod music_ports;
|
pub mod music_ports;
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use crate::application::dtos::file_dto::FileDto;
|
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::file_ports::FileManagementUseCase;
|
||||||
use crate::application::ports::storage_ports::{CopyFolderTreeResult, FileReadPort, FileWritePort};
|
use crate::application::ports::storage_ports::{CopyFolderTreeResult, FileReadPort, FileWritePort};
|
||||||
use crate::application::ports::trash_ports::TrashUseCase;
|
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::file_blob_write_repository::FileBlobWriteRepository;
|
||||||
use crate::infrastructure::repositories::pg::folder_db_repository::FolderDbRepository;
|
use crate::infrastructure::repositories::pg::folder_db_repository::FolderDbRepository;
|
||||||
use crate::infrastructure::services::file_content_cache::FileContentCache;
|
use crate::infrastructure::services::file_content_cache::FileContentCache;
|
||||||
use crate::infrastructure::services::thumbnail_service::ThumbnailService;
|
|
||||||
use tracing::{error, info, warn};
|
use tracing::{error, info, warn};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
@@ -26,8 +26,9 @@ pub struct FileManagementService {
|
|||||||
file_read: Option<Arc<FileBlobReadRepository>>,
|
file_read: Option<Arc<FileBlobReadRepository>>,
|
||||||
folder_repo: Option<Arc<FolderDbRepository>>,
|
folder_repo: Option<Arc<FolderDbRepository>>,
|
||||||
trash_service: Option<Arc<TrashService>>,
|
trash_service: Option<Arc<TrashService>>,
|
||||||
thumbnail_service: Option<Arc<ThumbnailService>>,
|
|
||||||
content_cache: Option<Arc<FileContentCache>>,
|
content_cache: Option<Arc<FileContentCache>>,
|
||||||
|
/// Hooks fired after a file is permanently deleted.
|
||||||
|
file_deleted_hooks: Vec<Arc<dyn FileDeletedHook>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl FileManagementService {
|
impl FileManagementService {
|
||||||
@@ -38,8 +39,8 @@ impl FileManagementService {
|
|||||||
file_read: None,
|
file_read: None,
|
||||||
folder_repo: None,
|
folder_repo: None,
|
||||||
trash_service: None,
|
trash_service: None,
|
||||||
thumbnail_service: None,
|
|
||||||
content_cache: None,
|
content_cache: None,
|
||||||
|
file_deleted_hooks: Vec::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -49,7 +50,6 @@ impl FileManagementService {
|
|||||||
trash_service: Option<Arc<TrashService>>,
|
trash_service: Option<Arc<TrashService>>,
|
||||||
file_read: Option<Arc<FileBlobReadRepository>>,
|
file_read: Option<Arc<FileBlobReadRepository>>,
|
||||||
folder_repo: Option<Arc<FolderDbRepository>>,
|
folder_repo: Option<Arc<FolderDbRepository>>,
|
||||||
thumbnail_service: Option<Arc<ThumbnailService>>,
|
|
||||||
content_cache: Option<Arc<FileContentCache>>,
|
content_cache: Option<Arc<FileContentCache>>,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
@@ -57,11 +57,17 @@ impl FileManagementService {
|
|||||||
file_read,
|
file_read,
|
||||||
folder_repo,
|
folder_repo,
|
||||||
trash_service,
|
trash_service,
|
||||||
thumbnail_service,
|
|
||||||
content_cache,
|
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<dyn FileDeletedHook>) -> Self {
|
||||||
|
self.file_deleted_hooks.push(hook);
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
/// Verifies ownership via the read repository.
|
/// Verifies ownership via the read repository.
|
||||||
async fn verify_owner(&self, file_id: &str, caller_id: Uuid) -> Result<(), DomainError> {
|
async fn verify_owner(&self, file_id: &str, caller_id: Uuid) -> Result<(), DomainError> {
|
||||||
if let Some(read) = &self.file_read {
|
if let Some(read) = &self.file_read {
|
||||||
@@ -230,15 +236,11 @@ impl FileManagementUseCase for FileManagementService {
|
|||||||
|
|
||||||
async fn delete_file(&self, id: &str) -> Result<(), DomainError> {
|
async fn delete_file(&self, id: &str) -> Result<(), DomainError> {
|
||||||
self.file_repository.delete_file(id).await?;
|
self.file_repository.delete_file(id).await?;
|
||||||
// Invalidate content cache — file no longer exists.
|
|
||||||
if let Some(cc) = &self.content_cache {
|
if let Some(cc) = &self.content_cache {
|
||||||
cc.invalidate(id).await;
|
cc.invalidate(id).await;
|
||||||
}
|
}
|
||||||
// Best-effort thumbnail cleanup
|
for hook in &self.file_deleted_hooks {
|
||||||
if let Some(thumb) = &self.thumbnail_service
|
hook.on_file_deleted(id).await;
|
||||||
&& let Err(e) = thumb.delete_thumbnails(id).await
|
|
||||||
{
|
|
||||||
warn!("Failed to delete thumbnails for file {}: {}", id, e);
|
|
||||||
}
|
}
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
@@ -283,15 +285,11 @@ impl FileManagementUseCase for FileManagementService {
|
|||||||
// Step 2: Permanent delete — trigger handles blob ref_count
|
// Step 2: Permanent delete — trigger handles blob ref_count
|
||||||
warn!("Permanently deleting file: {}", id);
|
warn!("Permanently deleting file: {}", id);
|
||||||
self.file_repository.delete_file(id).await?;
|
self.file_repository.delete_file(id).await?;
|
||||||
// Invalidate content cache — file permanently removed.
|
|
||||||
if let Some(cc) = &self.content_cache {
|
if let Some(cc) = &self.content_cache {
|
||||||
cc.invalidate(id).await;
|
cc.invalidate(id).await;
|
||||||
}
|
}
|
||||||
// Best-effort thumbnail cleanup
|
for hook in &self.file_deleted_hooks {
|
||||||
if let Some(thumb) = &self.thumbnail_service
|
hook.on_file_deleted(id).await;
|
||||||
&& let Err(e) = thumb.delete_thumbnails(id).await
|
|
||||||
{
|
|
||||||
warn!("Failed to delete thumbnails for file {}: {}", id, e);
|
|
||||||
}
|
}
|
||||||
info!("File permanently deleted: {}", id);
|
info!("File permanently deleted: {}", id);
|
||||||
|
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ use std::path::Path;
|
|||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
use crate::application::dtos::file_dto::FileDto;
|
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::file_ports::FileUploadUseCase;
|
||||||
use crate::application::ports::storage_ports::{FileReadPort, FileWritePort};
|
use crate::application::ports::storage_ports::{FileReadPort, FileWritePort};
|
||||||
use crate::application::services::storage_usage_service::StorageUsageService;
|
use crate::application::services::storage_usage_service::StorageUsageService;
|
||||||
@@ -52,6 +53,10 @@ pub struct FileUploadService {
|
|||||||
storage_usage_service: Option<Arc<StorageUsageService>>,
|
storage_usage_service: Option<Arc<StorageUsageService>>,
|
||||||
/// Content cache — invalidated on file update so stale content is never served.
|
/// Content cache — invalidated on file update so stale content is never served.
|
||||||
content_cache: Option<Arc<FileContentCache>>,
|
content_cache: Option<Arc<FileContentCache>>,
|
||||||
|
/// Hooks fired after a new file record is created.
|
||||||
|
file_created_hooks: Vec<Arc<dyn FileCreatedHook>>,
|
||||||
|
/// Hooks fired after a file's blob is replaced (e.g. thumbnail refresh).
|
||||||
|
file_updated_hooks: Vec<Arc<dyn FileUpdatedHook>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl FileUploadService {
|
impl FileUploadService {
|
||||||
@@ -62,6 +67,8 @@ impl FileUploadService {
|
|||||||
file_read: None,
|
file_read: None,
|
||||||
storage_usage_service: None,
|
storage_usage_service: None,
|
||||||
content_cache: 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),
|
file_read: Some(file_read),
|
||||||
storage_usage_service: None,
|
storage_usage_service: None,
|
||||||
content_cache: None,
|
content_cache: None,
|
||||||
|
file_created_hooks: Vec::new(),
|
||||||
|
file_updated_hooks: Vec::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -84,6 +93,18 @@ impl FileUploadService {
|
|||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Registers a hook to fire after a new file record is created.
|
||||||
|
pub fn with_file_created_hook(mut self, hook: Arc<dyn FileCreatedHook>) -> 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<dyn FileUpdatedHook>) -> Self {
|
||||||
|
self.file_updated_hooks.push(hook);
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
/// Configures the storage usage service
|
/// Configures the storage usage service
|
||||||
pub fn with_storage_usage_service(
|
pub fn with_storage_usage_service(
|
||||||
mut self,
|
mut self,
|
||||||
@@ -149,6 +170,10 @@ impl FileUploadUseCase for FileUploadService {
|
|||||||
name, size, dto.id
|
name, size, dto.id
|
||||||
);
|
);
|
||||||
self.maybe_update_storage_usage(&dto);
|
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)
|
Ok(dto)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -301,7 +326,12 @@ impl FileUploadUseCase for FileUploadService {
|
|||||||
}
|
}
|
||||||
// Re-read to get fresh DTO with updated etag and timestamps.
|
// Re-read to get fresh DTO with updated etag and timestamps.
|
||||||
let updated = file_read.get_file(&file_id).await?;
|
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
|
// File doesn't exist — create it via streaming upload
|
||||||
|
|||||||
+18
-52
@@ -61,7 +61,7 @@ use crate::infrastructure::services::image_transcode_service::ImageTranscodeServ
|
|||||||
use crate::infrastructure::services::jwt_service::JwtTokenService;
|
use crate::infrastructure::services::jwt_service::JwtTokenService;
|
||||||
use crate::infrastructure::services::password_hasher::Argon2PasswordHasher;
|
use crate::infrastructure::services::password_hasher::Argon2PasswordHasher;
|
||||||
use crate::infrastructure::services::path_resolver_service::PathResolverService;
|
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::wopi_discovery_service::WopiDiscoveryService;
|
||||||
use crate::infrastructure::services::zip_service::ZipService;
|
use crate::infrastructure::services::zip_service::ZipService;
|
||||||
|
|
||||||
@@ -355,12 +355,18 @@ impl AppServiceFactory {
|
|||||||
|
|
||||||
// Refactored services with all infrastructure ports
|
// Refactored services with all infrastructure ports
|
||||||
// In blob model, dedup is handled by the repository — no separate write-behind needed
|
// 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(
|
let file_upload_service = Arc::new(
|
||||||
FileUploadService::new_with_read(
|
FileUploadService::new_with_read(
|
||||||
repos.file_write_repository.clone(),
|
repos.file_write_repository.clone(),
|
||||||
repos.file_read_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(
|
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
|
// FileManagementService — ref_count handled by PG trigger, no dedup port needed
|
||||||
let file_management_service = Arc::new(FileManagementService::with_trash(
|
let file_management_service = Arc::new(
|
||||||
repos.file_write_repository.clone(),
|
FileManagementService::with_trash(
|
||||||
trash_service.clone(),
|
repos.file_write_repository.clone(),
|
||||||
Some(repos.file_read_repository.clone()),
|
trash_service.clone(),
|
||||||
Some(repos.folder_repository.clone()),
|
Some(repos.file_read_repository.clone()),
|
||||||
Some(core.thumbnail_service.clone()),
|
Some(repos.folder_repository.clone()),
|
||||||
Some(core.file_content_cache.clone()),
|
Some(core.file_content_cache.clone()),
|
||||||
));
|
)
|
||||||
|
.with_file_deleted_hook(core.thumbnail_service.clone()),
|
||||||
|
);
|
||||||
|
|
||||||
let file_use_case_factory = Arc::new(AppFileUseCaseFactory::new(
|
let file_use_case_factory = Arc::new(AppFileUseCaseFactory::new(
|
||||||
repos.file_read_repository.clone(),
|
repos.file_read_repository.clone(),
|
||||||
@@ -983,48 +991,6 @@ pub struct CoreServices {
|
|||||||
pub config: AppConfig,
|
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
|
/// Container for repository services
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub struct RepositoryServices {
|
pub struct RepositoryServices {
|
||||||
|
|||||||
@@ -44,7 +44,7 @@ use std::sync::Arc;
|
|||||||
use tokio::fs;
|
use tokio::fs;
|
||||||
use tokio::io::{AsyncReadExt, AsyncSeekExt};
|
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::blob_storage_ports::BlobStorageBackend;
|
||||||
use crate::application::ports::dedup_ports::{
|
use crate::application::ports::dedup_ports::{
|
||||||
BlobMetadataDto, DedupPort, DedupResultDto, DedupStatsDto,
|
BlobMetadataDto, DedupPort, DedupResultDto, DedupStatsDto,
|
||||||
@@ -84,6 +84,8 @@ pub struct DedupService {
|
|||||||
/// Isolated maintenance pool for long-running operations
|
/// Isolated maintenance pool for long-running operations
|
||||||
/// (verify_integrity, garbage_collect) that must never starve the primary.
|
/// (verify_integrity, garbage_collect) that must never starve the primary.
|
||||||
maintenance_pool: Arc<PgPool>,
|
maintenance_pool: Arc<PgPool>,
|
||||||
|
/// Hooks notified when a genuinely new blob is stored (no dedup hit).
|
||||||
|
blob_creation_hooks: Vec<Arc<dyn BlobCreationHook>>,
|
||||||
/// Hooks notified when a blob's ref_count reaches zero and it is deleted.
|
/// Hooks notified when a blob's ref_count reaches zero and it is deleted.
|
||||||
blob_hooks: Vec<Arc<dyn BlobDeletionHook>>,
|
blob_hooks: Vec<Arc<dyn BlobDeletionHook>>,
|
||||||
}
|
}
|
||||||
@@ -103,10 +105,18 @@ impl DedupService {
|
|||||||
backend,
|
backend,
|
||||||
pool,
|
pool,
|
||||||
maintenance_pool,
|
maintenance_pool,
|
||||||
|
blob_creation_hooks: vec![],
|
||||||
blob_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<dyn BlobCreationHook>) -> Self {
|
||||||
|
self.blob_creation_hooks.push(hook);
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
/// Register a [`BlobDeletionHook`] to be called whenever a blob's
|
/// Register a [`BlobDeletionHook`] to be called whenever a blob's
|
||||||
/// ref_count reaches zero. Hooks are called in registration order.
|
/// ref_count reaches zero. Hooks are called in registration order.
|
||||||
pub fn add_blob_hook(mut self, hook: Arc<dyn BlobDeletionHook>) -> Self {
|
pub fn add_blob_hook(mut self, hook: Arc<dyn BlobDeletionHook>) -> Self {
|
||||||
@@ -114,6 +124,13 @@ impl DedupService {
|
|||||||
self
|
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.
|
/// Fire all registered hooks for a deleted blob.
|
||||||
async fn fire_blob_hooks(&self, hash: &str) {
|
async fn fire_blob_hooks(&self, hash: &str) {
|
||||||
for hook in &self.blob_hooks {
|
for hook in &self.blob_hooks {
|
||||||
@@ -135,6 +152,7 @@ impl DedupService {
|
|||||||
backend: Arc::new(LocalBlobBackend::new(Path::new("/tmp/oxicloud_stub_blobs"))),
|
backend: Arc::new(LocalBlobBackend::new(Path::new("/tmp/oxicloud_stub_blobs"))),
|
||||||
pool: stub_pool.clone(),
|
pool: stub_pool.clone(),
|
||||||
maintenance_pool: stub_pool,
|
maintenance_pool: stub_pool,
|
||||||
|
blob_creation_hooks: vec![],
|
||||||
blob_hooks: vec![],
|
blob_hooks: vec![],
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -353,6 +371,9 @@ impl DedupService {
|
|||||||
chunk_hashes.len()
|
chunk_hashes.len()
|
||||||
);
|
);
|
||||||
|
|
||||||
|
self.fire_blob_creation_hooks(&file_hash, content_type.as_deref())
|
||||||
|
.await;
|
||||||
|
|
||||||
Ok(DedupResultDto::NewBlob {
|
Ok(DedupResultDto::NewBlob {
|
||||||
hash: file_hash,
|
hash: file_hash,
|
||||||
size: file_size,
|
size: file_size,
|
||||||
|
|||||||
@@ -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<ThumbnailService>,
|
||||||
|
dedup: Arc<DedupService>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ThumbnailRefreshHook {
|
||||||
|
pub fn new(thumbnail: Arc<ThumbnailService>, dedup: Arc<DedupService>) -> 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<Box<dyn std::future::Future<Output = ()> + 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<Box<dyn std::future::Future<Output = ()> + 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<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
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ─── FileDeletedHook ─────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
impl crate::application::ports::file_lifecycle::FileDeletedHook for ThumbnailService {
|
||||||
|
fn on_file_deleted<'a>(
|
||||||
|
&'a self,
|
||||||
|
file_id: &'a str,
|
||||||
|
) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + 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 ─────────────────────────────────────────────────────
|
// ─── Port implementation ─────────────────────────────────────────────────────
|
||||||
|
|
||||||
/// Convert port ThumbnailSize to infra ThumbnailSize.
|
/// Convert port ThumbnailSize to infra ThumbnailSize.
|
||||||
|
|||||||
@@ -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()
|
Ok(Response::builder()
|
||||||
.status(StatusCode::NO_CONTENT)
|
.status(StatusCode::NO_CONTENT)
|
||||||
.body(Body::empty())
|
.body(Body::empty())
|
||||||
|
|||||||
@@ -288,19 +288,7 @@ async fn put_file(
|
|||||||
let _ = tokio::fs::remove_file(&temp_path).await;
|
let _ = tokio::fs::remove_file(&temp_path).await;
|
||||||
|
|
||||||
match result {
|
match result {
|
||||||
Ok(file_dto) => {
|
Ok(_file_dto) => StatusCode::OK.into_response(),
|
||||||
state
|
|
||||||
.app_state
|
|
||||||
.core
|
|
||||||
.refresh_thumbnails_after_update(
|
|
||||||
file_dto.id.clone(),
|
|
||||||
file_dto.etag.clone(),
|
|
||||||
&content_type,
|
|
||||||
)
|
|
||||||
.await;
|
|
||||||
|
|
||||||
StatusCode::OK.into_response()
|
|
||||||
}
|
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::error!("WOPI PutFile failed: {}", e);
|
tracing::error!("WOPI PutFile failed: {}", e);
|
||||||
StatusCode::INTERNAL_SERVER_ERROR.into_response()
|
StatusCode::INTERNAL_SERVER_ERROR.into_response()
|
||||||
|
|||||||
@@ -164,11 +164,6 @@ async fn handle_assemble(
|
|||||||
.await
|
.await
|
||||||
.map_err(|e| AppError::internal_error(format!("Failed to update file: {}", e)))?;
|
.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)
|
Some(dto.etag)
|
||||||
} else {
|
} else {
|
||||||
// For new files we still need to read the temp file since create_file takes &[u8].
|
// For new files we still need to read the temp file since create_file takes &[u8].
|
||||||
|
|||||||
@@ -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()
|
return Ok(Response::builder()
|
||||||
.status(StatusCode::NO_CONTENT)
|
.status(StatusCode::NO_CONTENT)
|
||||||
.header(header::ETAG, format!("\"{}\"", updated.etag))
|
.header(header::ETAG, format!("\"{}\"", updated.etag))
|
||||||
|
|||||||
Reference in New Issue
Block a user