cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: fix services accessig directly to localstorage

Browse Source 
Prevent services accessing directly to localstorage and prefer using an astraction layer
to expose full blob. The abstraction layer (dedup services) will cover backend storage
election (local, s3, ...), encryption, etc

This change permit audio_metadata_service, media_metadaa_service, face_indexing_service to handle
blobs without worring of the backend.

note: prefered way to handle blob is the streamed way. Some services may not have this possibility
This commit is contained in:
Edouard Vanbelle
2026-08-02 19:20:23 +02:00
parent 74866bea81
commit 7663f803d3
11 changed files with 621 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/services/face_indexing_service.rs
+34 -23
View File
@@ -1,14 +1,15 @@
//! Face indexing as a `FileLifecycleHook`.
//!
//! On image upload it detects + embeds faces (off the request path, in a
//! background task) and stores them. It mirrors `MediaMetadataService`: reads
//! the blob from the local `.blobs` tree, is dedup-aware (identical uploads
//! clone an existing file's faces instead of re-running inference), and is
//! completely inert when no model is configured (`FaceAnalyzerPort::is_ready()
//! == false`) — so the feature compiles and runs with the default no-op
//! analyzer until the operator wires a real ONNX model.
//! background task) and stores them. Mirrors `ThumbnailService`: reads the
//! blob through `DedupService` (CDC-manifest lookup, wrapper-stack
//! delegation, encryption transparency — the service sees none of that),
//! is dedup-aware (identical uploads clone an existing file's faces
//! instead of re-running inference), and is completely inert when no
//! model is configured (`FaceAnalyzerPort::is_ready() == false`) so the
//! feature compiles and runs with the default no-op analyzer until the
//! operator wires a real ONNX model.
use std::path::{Path, PathBuf};
use std::sync::Arc;
use chrono::Utc;
@@ -20,6 +21,7 @@ use crate::application::ports::file_lifecycle::FileLifecycleHook;
use crate::common::errors::DomainError;
use crate::domain::entities::face::Face;
use crate::infrastructure::repositories::pg::FacePgRepository;
use crate::infrastructure::services::dedup_service::DedupService;
/// Minimum detector confidence for a face to be stored.
const MIN_DET_SCORE: f32 = 0.6;
@@ -48,7 +50,10 @@ pub struct FaceIndexingService {
pool: Arc<PgPool>,
repo: Arc<FacePgRepository>,
analyzer: Arc<dyn FaceAnalyzerPort>,
blob_root: PathBuf,
/// CDC-aware blob reader. Same abstraction `thumbnail_service` uses —
/// hides both the chunk-manifest concatenation and the underlying
/// `BlobStorageBackend` wrapper stack.
dedup: Arc<DedupService>,
/// Bounds concurrent indexing tasks. The lifecycle hooks spawn one
/// task per uploaded/copied image with no ceiling, so a bulk upload
/// used to fan out N simultaneous full-image reads + decodes +
@@ -60,23 +65,21 @@ pub struct FaceIndexingService {
}
impl FaceIndexingService {
pub fn new(pool: Arc<PgPool>, blob_root: PathBuf, analyzer: Arc<dyn FaceAnalyzerPort>) -> Self {
pub fn new(
pool: Arc<PgPool>,
dedup: Arc<DedupService>,
analyzer: Arc<dyn FaceAnalyzerPort>,
) -> Self {
let repo = Arc::new(FacePgRepository::new(pool.clone()));
Self {
pool,
repo,
analyzer,
blob_root,
dedup,
index_semaphore: Arc::new(tokio::sync::Semaphore::new(max_concurrent_index())),
}
}
/// Local path of a blob: `.blobs/{prefix}/{hash}.blob`.
fn blob_path(&self, hash: &str) -> PathBuf {
let prefix = if hash.len() >= 2 { &hash[0..2] } else { hash };
self.blob_root.join(prefix).join(format!("{hash}.blob"))
}
/// Spawn a background indexing task. `reuse_dedup` clones faces from an
/// existing file with the same blob hash instead of re-running inference;
/// `delete_first` clears prior faces (used on overwrite).
@@ -84,7 +87,7 @@ impl FaceIndexingService {
let pool = self.pool.clone();
let repo = self.repo.clone();
let analyzer = self.analyzer.clone();
let blob_path = self.blob_path(&blob_hash);
let dedup = self.dedup.clone();
let semaphore = self.index_semaphore.clone();
tokio::spawn(async move {
// Queue behind the concurrency budget BEFORE touching the
@@ -102,7 +105,7 @@ impl FaceIndexingService {
&repo,
analyzer.as_ref(),
file_id,
&blob_path,
&dedup,
&blob_hash,
reuse_dedup,
)
@@ -161,7 +164,12 @@ impl FileLifecycleHook for FaceIndexingService {
}
async fn lookup_user(pool: &PgPool, file_id: Uuid) -> Result<Uuid, DomainError> {
let row: (Uuid,) = sqlx::query_as("SELECT user_id FROM storage.files WHERE id = $1")
// Post-D7: `storage.files.user_id` was dropped in
// migrations/20260904000000_drop_files_folders_user_id.sql —
// provenance moved to `created_by` / `updated_by`. For the
// faces.user_id anchor, the file's original creator is what we
// want (matches the pre-D7 semantic of the dropped column).
let row: (Uuid,) = sqlx::query_as("SELECT created_by FROM storage.files WHERE id = $1")
.bind(file_id)
.fetch_one(pool)
.await
@@ -174,7 +182,7 @@ async fn index_file(
repo: &FacePgRepository,
analyzer: &dyn FaceAnalyzerPort,
file_id: Uuid,
blob_path: &Path,
dedup: &Arc<DedupService>,
blob_hash: &str,
reuse_dedup: bool,
) -> Result<(), DomainError> {
@@ -199,9 +207,12 @@ async fn index_file(
// No peer found — fall through and analyze.
}
let bytes = tokio::fs::read(blob_path)
.await
.map_err(|e| DomainError::internal_error("Faces", format!("read blob: {e}")))?;
// CDC-aware, backend-agnostic read: `DedupService` concatenates chunks
// for CDC files, delegates straight through for legacy whole-file
// blobs, and inherits the backend wrapper stack (encryption, retry,
// cache) transparently. Peak process-heap = image size, already
// bounded by `index_semaphore` above.
let bytes = dedup.read_blob_bytes(blob_hash).await?;
let detected = analyzer.analyze(&bytes).await?;
let faces: Vec<Face> = detected