diff --git a/docs/config/env.md b/docs/config/env.md index 05febfdb..1602d4e2 100644 --- a/docs/config/env.md +++ b/docs/config/env.md @@ -8,6 +8,7 @@ Most runtime variables use the `OXICLOUD_` prefix. A few build-time or allocator |---|---|---| | `OXICLOUD_STORAGE_PATH` | `./storage` | Root storage directory | | `OXICLOUD_STATIC_PATH` | `./static` | Static files directory | +| `OXICLOUD_TEMP_DIR` | `std::env::temp_dir()` (`$TMPDIR`) | Directory for tier-1 temporary data — pure scratch, safe to lose at reboot. Backend services stream blobs here when extractors need a `&Path` (id3, mp3_duration, ffprobe, nom-exif video). Files are auto-removed after use. On Linux `/tmp` is often tmpfs (RAM-backed); point at a disk-backed dir under RAM-constrained deployments. | | `OXICLOUD_SERVER_PORT` | `8086` | Server port | | `OXICLOUD_SERVER_HOST` | `127.0.0.1` | Server bind address (IPv4 or IPv6 allowed) | | `OXICLOUD_BASE_URL` | (auto) | Public base URL for share links; defaults to `http://{host}:{port}` | diff --git a/docs/plan/derived-blobs.md b/docs/plan/derived-blobs.md new file mode 100644 index 00000000..52462057 --- /dev/null +++ b/docs/plan/derived-blobs.md @@ -0,0 +1,306 @@ +# Plan — Derived content as blobs (tier-2 refactor) + +**Status:** design captured 2026-08-02, not implemented. Follow-up to +`fix/services-use-blob-abstraction` — that PR normalised the +**read-side** (services consume blobs through `BlobStorageBackend` +uniformly). This plan tackles the **write-side**: services that +today write derived artifacts (thumbnails, transcodes) to a local +sidecar directory and would benefit from writing them through the +backend abstraction instead. + +## Context — the three-tier storage taxonomy + +Today the codebase runs three implicit tiers with no explicit +separation: + +| Tier | Purpose | Loss on reboot? | Where today | +|---|---|---|---| +| **1 — Temp** | Pure scratch, deletable at reboot | ✅ fine | `std::env::temp_dir()` (ad-hoc callers). Now unified under `OXICLOUD_TEMP_DIR` (`AppConfig::temp_dir`). | +| **2 — Persistent spool** | Caches; expensive but rebuildable | ⚠️ possible but painful | `/.thumbnails/`, `/.transcoded/`, `/.blob-cache/`, `/.search-index/`, `/.plugin-logs/` — all mixed into tier-3 storage today. | +| **3 — Persistent data** | Source of truth | ❌ never | `/.blobs/` (Local) OR S3/Azure bucket, via `BlobStorageBackend`. Already correctly configured via `OXICLOUD_STORAGE_ENTRIES`. | + +**Today's misclassification**: tier-2 sidecars live under +`` — the same directory as tier-3 source-of-truth +data. Ops resizing / moving / backing up tier-3 accidentally moves +tier-2 caches with it. Loss of tier 2 is expensive (regenerate +thumbnails for every photo) but not data loss; conflating them +means backup policies can't distinguish "must preserve" from "can +rebuild". + +## Multi-instance driver + +Single-instance: tier-2-as-local-cache works fine. Rebuild after +reboot is annoying but bounded. + +Multi-instance (2+ app servers behind a load balancer): + +- Request for thumbnail `abc123.jpg` lands on instance A → generates + it → stores locally at `.thumbnails/abc123.jpg`. +- Same-URL retry lands on instance B → cache miss → regenerates + from source. +- Every derived asset gets recomputed N times (N = instance count) + at worst. + +Wasteful compute, wasteful storage, inconsistent latency. The +long-term fix is to put derived content on tier 3 (shared) with a +local read-through cache in front. Multi-instance isn't the near- +term target, but the design should leave the door open. + +## Design decision — derived content IS a blob + +The blob storage abstraction is already: + +- Backend-agnostic (Local / S3 / Azure) +- Encrypted uniformly (`EncryptedBlobBackend` wrapper) +- Consistency-checked (`blobs_consistency`) +- Migratable (`backend_migration`) +- Rotatable (`backend_rotate`) +- Multi-instance-ready (S3/Azure natively; Local via network mount) + +Reusing it for derived artifacts means no second abstraction to +build and maintain, and all the operational surface (audit, +migration, key rotation) applies to derived content by default. + +### Keying + +Content-addressable via BLAKE3, same as source blobs. For +server-derived content the hash is over the produced bytes (not +the source), so: + +- Two files with **identical thumbnails** (e.g. same 256px WebP + crop of the same underlying image → identical bytes → identical + hash) share the physical blob. Dedup wins for free. +- Two files with **identical originals** but **different variant + specs** (256px vs 512px thumb) produce different blobs. Also + correct. + +The variant spec (what was rendered) lives in the referring DB row +alongside the blob hash — not in the storage key. Storage stays +one keyspace; ownership stays per-service. + +### Client-uploaded thumbnails + +Some clients (NC desktop, mobile apps) upload their own encoded +previews alongside the file. These are **not derivable** — losing +them means asking the client to regenerate, which may not be +possible (client offline, original file no longer present on +device). + +Same storage shape: BLAKE3 of the client-provided bytes → blob. +The DB row distinguishes `origin = 'server_derived' | 'client_provided'` +so consistency-check policy can differ (missing client-provided +thumbnail = data loss finding; missing server-derived = warning, +regenerable). + +## `BlobReferenceSource` — reference tracking abstraction + +Adding new blob-owning services without teaching the ref-count + +consistency machinery about them causes silent orphaning risk: +`dedup_gc` sees `ref_count = 0` and reaps live content. + +The extension point: + +```rust +#[async_trait] +pub trait BlobReferenceSource: Send + Sync { + /// Short stable identifier for logs / consistency finding + /// `source` fields. Suggested: `"files"`, `"chunks"`, + /// `"thumbnails"`, `"transcodes"`. + fn source_name(&self) -> &'static str; + + /// Count of references this source holds on `blob_hash`. + /// Called by `blobs_consistency` when recomputing + /// `refcount_mismatch` findings. + async fn count_references(&self, blob_hash: &str) -> Result; + + /// Iterate the source's referenced blobs, paged by the + /// implementation's natural cursor (typically a DB PK). Used + /// by `backend_consistency` to walk the backend against the + /// union of all sources. + async fn list_referenced_blobs( + &self, + cursor: Option>, + limit: usize, + ) -> Result<(Vec, Option>), DomainError>; + + /// Optional notify hook: `dedup_gc` reaped this blob. Sources + /// that maintain their own denormalised refcount table can + /// clean up here. Most sources leave this as the trait default + /// (noop). + fn on_blob_reaped(&self, _blob_hash: &str) {} +} +``` + +Wired via a `BlobReferenceRegistry`: + +```rust +pub struct BlobReferenceRegistry { + sources: Vec>, +} + +impl BlobReferenceRegistry { + pub fn register(&mut self, source: Arc); + pub async fn total_references(&self, hash: &str) -> Result; + // ... etc. +} +``` + +Current implicit sources become the first two explicit +registrations: + +- `FilesReferenceSource` — wraps `storage.files.blob_hash` +- `ChunksReferenceSource` — wraps `storage.chunk_manifests.chunk_hashes[]` + +Tier-2 migration adds: + +- `ThumbnailsReferenceSource` — wraps a new + `storage.thumbnails(hash, blob_hash, variant_spec, origin)` table +- `TranscodesReferenceSource` — wraps + `storage.transcodes(hash, blob_hash, target_format)` table + +Then: + +- **`dedup_gc`** — orphan iff `registry.total_references(hash) == 0` + (with the existing grace window). No per-service GC changes. +- **`blobs_consistency`** — `refcount_mismatch` recomputes via + `registry.total_references`. New services register → automatically + covered. +- **`backend_consistency`** — walks the backend and unions all + `list_referenced_blobs` streams for the "did we lose bytes" + check. + +## Sidecar directories after this refactor + +| Sidecar today | After | +|---|---| +| `.thumbnails/` | Persisted as derived blobs in tier 3. `.thumbnails/` becomes a pure read-through cache (tier 1-ish; ephemeral, per-instance). | +| `.transcoded/` | Same shape as thumbnails. | +| `.blob-cache/` | Already a cache; stays. Owned by `CachedBlobBackend`. | +| `.search-index/` | Open question — see non-goals. | +| `.plugin-logs/` | Ops-local; stays. | +| `.uploads/` | Tier 1 already; migrates to `OXICLOUD_TEMP_DIR`. | + +The persistent-spool env var reserved: + +- **`OXICLOUD_SPOOL_DIR`** — path for the local read-through + caches (`.thumbnails/`, `.transcoded/`, `.blob-cache/`). Default + `/spool`. Ops can point it at a different disk + than tier-3 storage; multi-instance deployments accept per- + instance rebuild OR mount a shared FS here. + +## Delivery order + +Coarse — the trait + registry ship first (empty-impl for +`FilesReferenceSource` + `ChunksReferenceSource` mirroring today's +hardcoded SQL). New sources bolt on independently. + +1. **`BlobReferenceSource` trait + registry** in + `application/ports/`. `FilesReferenceSource` and + `ChunksReferenceSource` implementations mirroring current SQL; + wire into `dedup_gc` + `blobs_consistency` behind an integration + test that proves the union equals the pre-refactor count on a + real DB. +2. **`OXICLOUD_SPOOL_DIR`** — config + `example.env` + docs + + `AppConfig::spool_dir`. Migrate `CachedBlobBackend` cache path + default to `/blob-cache/`. +3. **`ThumbnailService` writes go through the backend**. New + `storage.thumbnails` table + `ThumbnailsReferenceSource`. Local + `.thumbnails/` sidecar becomes a read-through cache pattern. +4. **`ImageTranscodeService`** — same shape as thumbnails. +5. **Client-uploaded thumbnails** — new `origin` column + upload + API path if needed. + +Each slice is independently mergeable. Delivery span: rough +estimate ~2 weeks end-to-end. + +## Naming clarifications to land alongside this refactor + +Two consumer-facing terminology issues that surfaced during the +read-side normalisation (2026-08-02). They're not code-breakers, +but they cost every new implementor a mental round-trip, so +they belong in the tier-2 sweep: + +### 1. `DedupService` name is implementation-shaped, not consumer-shaped + +From a consumer's perspective the service is "the thing that +reads and writes file content by hash." Deduplication is one +internal responsibility (alongside CDC chunking, ref-counting, +GC). The name `DedupService` narrates HOW it works, not WHAT it +is — new service authors read the name and don't realise they +should be routing every blob read through it. + +Suggested rename: **`BlobHandler`** (or `ContentStore` / +`BlobStore` — pick one and commit). Public surface stays +identical; consumers write `Arc` and call +`blob_handler.read_blob_bytes(hash)`. Internal doc-comments +document dedup + CDC + GC as strategies. + +Scope: ~35 files (grep `DedupService|dedup_service`), mechanical. +Keep as one commit inside the tier-2 refactor so reviewers see +"rename" independently from the substantive changes. + +### 2. `blob` overloaded across two scales + +Current usage: + +- `storage.blobs` — the physical storage table; rows are BYTES + written to a backend. Post-CDC, most entries are chunks + (fragments), not whole files. +- `storage.chunk_manifests` — the CDC manifest that references a + set of `storage.blobs` rows to reconstitute a file. +- `file.blob_hash` — the hash a file row points at; either a + whole-file blob (legacy) OR a chunk-manifest (post-CDC). + +The word "blob" carries two meanings: *whole-file content* (what +a user thinks of when they say "download the blob") vs +*physical byte-payload on disk* (what the storage backend +holds — may be a whole file, may be a chunk fragment). + +Proposed clarification for the tier-2 sweep: + +- **Blob** = the abstraction of "content of a file", identified + by BLAKE3 of the plaintext. Consumers work at this level. What + `DedupService`/`BlobHandler` returns. +- **Chunk** = a physical byte-payload written to the backend, + identified by its own BLAKE3. Storage-backend-internal. +- **Manifest** = the map from a Blob to one or more Chunks. + +Schema rename (deferred, requires migration): + +- `storage.blobs` → `storage.chunks` (that's what it actually holds now) +- `storage.chunk_manifests` → `storage.blob_manifests` (or keep — arguable) +- `BlobStorageBackend` trait → `ChunkStorageBackend` — reads and + writes physical chunks, not blobs + +`file.blob_hash` semantics stay — references a Blob via its +manifest OR (for pre-CDC legacy) points directly at a single-chunk +Blob whose hash equals its lone chunk's hash. + +Scope for this rename: ~23 files touch the SQL, plus a migration +for the table rename. Not free. Ship AFTER the tier-2 write-side +lands so we don't stack schema changes. + +## Non-goals + +- **Tantivy `.search-index/`** — memory-mapped by design, doesn't + fit the blob-storage abstraction. Separate future decision: + keep local, snapshot-to-backend periodically, or retire Tantivy + for PG-native full-text. +- **`.plugin-logs/`** — ops-local operational data, not user + content. Stays local. +- **Client thumbnail negotiation protocol** — the wire-level API + for how clients push their previews. Design piece for the + photo/mobile team when there's a real feature ask. + +## References + +- `docs/architecture/backend-storage.md` — the wrapper stack, header + format, consistency check, migration semantics that derived + content inherits. +- `docs/plan/storage-multi-entry.md` — tier-3 configuration model. +- `docs/plan/storage-key-rotation.md` — encryption/rotation applies + to derived blobs too. +- `src/AGENTS.md` — the read-side rule enforcing backend + abstraction (already shipped alongside this plan doc). +- Memory note `project_services_bypassing_blob_backend` — audit + history of the pre-normalisation bypasses. diff --git a/example.env b/example.env index 45b9b06e..55093d24 100644 --- a/example.env +++ b/example.env @@ -17,6 +17,17 @@ OXICLOUD_STORAGE_PATH=./storage # Path to static files directory (default: ./static) OXICLOUD_STATIC_PATH=./static +# Directory for tier-1 temporary data — pure scratch, safe to lose at reboot. +# Services that need a filesystem path (audio ID3/MP3 duration, video EXIF via +# ffprobe/nom-exif, etc.) stream blobs here from the active backend before +# handing the path to the extractor. The file is auto-removed after each use. +# +# Default: system tempdir (`std::env::temp_dir()`, honours $TMPDIR / $TMP). +# On Linux `/tmp` is often mounted as tmpfs (RAM-backed); high-concurrency +# production deployments concerned about RAM should point this at a +# disk-backed dir (e.g. /var/lib/oxicloud/tmp). +#OXICLOUD_TEMP_DIR=/var/lib/oxicloud/tmp + # Server port (default: 8086) OXICLOUD_SERVER_PORT=8086 diff --git a/justfile b/justfile index 5e48e615..91e89dd6 100644 --- a/justfile +++ b/justfile @@ -369,3 +369,7 @@ load-baseline: # just db load-seed: cargo run --bin load-seed -- --depth 5 --fanout 4 --files-per-leaf 3 + +# Check and test everything +# recommanded before pull request +pre-pull-request: check fe-check audit check-migrations test test-integration fe-test build api-test fe-build-e2e front-test diff --git a/src/AGENTS.md b/src/AGENTS.md index b2d17c22..ff268303 100644 --- a/src/AGENTS.md +++ b/src/AGENTS.md @@ -14,3 +14,11 @@ Non-obvious rules that trip up new code. Terse on purpose. - Any new endpoint that mints or consumes credentials/tokens must consult one of the `is_*_login_allowed()` helpers, not the raw allowlist. - Any new "policy-disabled" refusal must emit an `audit`-target line before returning — matches `auth.login_rejected`, `magic_link.redemption_rejected` conventions. + +## Storage backend access + +- **Read blob content through `Arc`.** It's the ONE canonical read abstraction — CDC-manifest-aware (`file.blob_hash` may reference a chunk manifest, not a blob), backend-agnostic (Local/S3/Azure), wrapper-transparent (encryption/retry/cache). Never take `Arc` directly in a service that reads content; you'll silently break on any file ≥ 64 KiB (`CDC_MIN_CHUNK`). Follow `thumbnail_service`, `audio_metadata_service`, `media_metadata_service`, `face_indexing_service`, `search_index::content_index_worker` as reference impls. +- **Reads use `DedupService` methods**: `dedup.read_blob_bytes(hash)` for byte-slice analyzers (ONNX, EXIF, ID3-via-Reader), `dedup.stream_blob_to_tempfile(hash, &temp_dir, ".ext")` for crates that only accept `&Path` (mp3_duration, ffprobe, `nom-exif` video), `dedup.read_blob_stream(hash)` for streaming to a downstream `Stream` consumer. +- **Never hand-craft blob paths.** No `blob_root: PathBuf` fields, no `/.blobs//.blob` constructions. `BlobStorageBackend::local_blob_path` returns `None` under `EncryptedBlobBackend`; do not rely on it. The three services that did this pre-2026-08 (audio/media/face) are the anti-pattern — see memory `project_services_bypassing_blob_backend`. +- **Persistent state = backend**, not `/*` sidecars. Local sidecars (`.thumbnails/`, `.transcoded/`, `.blob-cache/`, `.search-index/`, `.plugin-logs/`, `.uploads/`) are only for caches (regenerable) or truly-temp scratch (deleted on drop). Anything a user would notice losing → blob backend. Tier-2 migration plan: `docs/plan/derived-blobs.md`. +- **Temp files use `OXICLOUD_TEMP_DIR`** via the shared config path (`AppConfig::temp_dir`) — not raw `std::env::temp_dir()`. Ops point it at real disk on RAM-constrained Linux deployments (default `/tmp` = tmpfs = RAM). diff --git a/src/common/config.rs b/src/common/config.rs index 77984a0a..9856e28f 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -2252,6 +2252,19 @@ pub struct AppConfig { pub storage_path: PathBuf, /// Static files directory path pub static_path: PathBuf, + /// Directory for tier-1 temporary data — pure scratch, safe to + /// lose at reboot. Backend `stream_to_tempfile` writes here so + /// extractors that require a `&Path` (id3, mp3_duration, + /// ffprobe, nom-exif video) can operate on a local file + /// without the service ever seeing a raw blob path. + /// + /// Env: `OXICLOUD_TEMP_DIR`. Default: `std::env::temp_dir()` + /// (respects `$TMPDIR`). On Linux this is typically `/tmp`, + /// often mounted as tmpfs (RAM-backed); production + /// deployments concerned about physical RAM under + /// high concurrency should point this at a disk-backed + /// directory (e.g. `/var/lib/oxicloud/tmp`). + pub temp_dir: PathBuf, /// Server port pub server_port: u16, /// Server host @@ -2356,6 +2369,7 @@ impl Default for AppConfig { Self { storage_path: PathBuf::from("./storage"), static_path: PathBuf::from("./static"), + temp_dir: env::temp_dir(), server_port: 8086, server_host: "127.0.0.1".to_string(), cache: CacheConfig::default(), @@ -2394,6 +2408,10 @@ impl AppConfig { config.static_path = PathBuf::from(static_path); } + if let Ok(temp_dir) = env::var("OXICLOUD_TEMP_DIR") { + config.temp_dir = PathBuf::from(temp_dir); + } + if let Ok(server_port) = env::var("OXICLOUD_SERVER_PORT") && let Ok(port) = server_port.parse::() { diff --git a/src/common/di.rs b/src/common/di.rs index 3673d977..72d33ed3 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -468,11 +468,11 @@ impl AppServiceFactory { ); // Audio metadata service — created here so it can be wired into file_lifecycle. - let audio_metadata_service = self.create_audio_metadata_service(db_pool); + let audio_metadata_service = self.create_audio_metadata_service(db_pool, &dedup_service); // Image/video capture-metadata service — extracts EXIF/container capture // dates so the Photos timeline groups by real capture time, not upload time. - let media_metadata_service = self.create_media_metadata_service(db_pool); + let media_metadata_service = self.create_media_metadata_service(db_pool, &dedup_service); // ThumbnailRefreshHook: handles FileLifecycleHook events (create/update/delete). // Implemented on ThumbnailRefreshHook (not ThumbnailService) to avoid circular Arc: @@ -544,7 +544,7 @@ impl AppServiceFactory { } fls = fls.with_hook(media_metadata_service.clone()); if self.config.features.enable_faces { - fls = fls.with_hook(self.create_face_indexing_service(db_pool)); + fls = fls.with_hook(self.create_face_indexing_service(db_pool, &dedup_service)); } let file_lifecycle = Arc::new(fls); @@ -950,15 +950,16 @@ impl AppServiceFactory { pub fn create_audio_metadata_service( &self, db_pool: &Arc, + dedup: &Arc, ) -> Option> { if !self.config.features.enable_music { tracing::info!("Audio metadata service is disabled (music feature disabled)"); return None; } - let blob_root = self.storage_path.join(".blobs"); Some(Arc::new(AudioMetadataService::new( db_pool.clone(), - blob_root, + dedup.clone(), + self.config.temp_dir.clone(), ))) } @@ -967,9 +968,13 @@ impl AppServiceFactory { pub fn create_media_metadata_service( &self, db_pool: &Arc, + dedup: &Arc, ) -> Arc { - let blob_root = self.storage_path.join(".blobs"); - Arc::new(MediaMetadataService::new(db_pool.clone(), blob_root)) + Arc::new(MediaMetadataService::new( + db_pool.clone(), + dedup.clone(), + self.config.temp_dir.clone(), + )) } /// Creates the trash service @@ -1137,13 +1142,13 @@ impl AppServiceFactory { pub fn create_face_indexing_service( &self, db_pool: &Arc, + dedup: &Arc, ) -> Arc { - let blob_root = self.storage_path.join(".blobs"); let analyzer = self.build_face_analyzer(); Arc::new( crate::infrastructure::services::face_indexing_service::FaceIndexingService::new( db_pool.clone(), - blob_root, + dedup.clone(), analyzer, ), ) diff --git a/src/infrastructure/services/audio_metadata_service.rs b/src/infrastructure/services/audio_metadata_service.rs index 8591043f..c7a4a323 100644 --- a/src/infrastructure/services/audio_metadata_service.rs +++ b/src/infrastructure/services/audio_metadata_service.rs @@ -8,6 +8,7 @@ use uuid::Uuid; use crate::application::ports::file_lifecycle::FileLifecycleHook; use crate::common::errors::DomainError; +use crate::infrastructure::services::dedup_service::DedupService; #[derive(Debug, FromRow)] pub struct AudioFileRow { @@ -17,22 +18,32 @@ pub struct AudioFileRow { pub struct AudioMetadataService { pool: Arc, - 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, + /// Tier-1 scratch directory for `stream_blob_to_tempfile`. Pulled + /// from `AppConfig::temp_dir` (env `OXICLOUD_TEMP_DIR`) at DI time. + temp_dir: PathBuf, } impl AudioMetadataService { - pub fn new(pool: Arc, blob_root: PathBuf) -> Self { - Self { pool, blob_root } + pub fn new(pool: Arc, dedup: Arc, temp_dir: PathBuf) -> Self { + Self { + pool, + dedup, + temp_dir, + } } pub fn is_audio_file(mime_type: &str) -> bool { mime_type.starts_with("audio/") } - pub fn spawn_extraction_background(service: Arc, file_id: Uuid, file_path: PathBuf) { + pub fn spawn_extraction_background(service: Arc, file_id: Uuid, blob_hash: String) { tokio::spawn(async move { tracing::info!("🎵 Extracting audio metadata for: {}", file_id); - if let Err(e) = service.extract_and_save(&file_id, &file_path).await { + if let Err(e) = service.extract_and_save(&file_id, &blob_hash).await { tracing::warn!("Failed to extract audio metadata: {}", e); } }); @@ -41,22 +52,17 @@ impl AudioMetadataService { pub fn spawn_extraction_with_delete_background( service: Arc, file_id: Uuid, - file_path: PathBuf, + blob_hash: String, ) { tokio::spawn(async move { tracing::info!("🎵 Updating audio metadata for: {}", file_id); let _ = service.delete_metadata(&file_id).await; - if let Err(e) = service.extract_and_save(&file_id, &file_path).await { + if let Err(e) = service.extract_and_save(&file_id, &blob_hash).await { tracing::warn!("Failed to update audio metadata: {}", e); } }); } - fn blob_path(&self, hash: &str) -> PathBuf { - let prefix = &hash[0..2]; - self.blob_root.join(prefix).join(format!("{}.blob", hash)) - } - /// Extract ID3 tag and MP3 duration from a file. /// /// All I/O is synchronous (id3 + mp3_duration crates), so this MUST @@ -104,15 +110,26 @@ impl AudioMetadataService { pub async fn extract_and_save( &self, file_id: &Uuid, - file_path: &Path, + blob_hash: &str, ) -> Result<(), DomainError> { info!( - "AudioMetadataService: blob_root={:?}, file_id={}, file_path={:?}", - self.blob_root, file_id, file_path, + "AudioMetadataService: extracting file_id={}, blob_hash={}", + file_id, blob_hash, ); + // Stream the blob (CDC-aware — chunks concatenated on the fly for + // chunked files; wrapper stack handles encryption + retry + cache) + // to a tempfile in the configured tier-1 temp dir, then hand its + // `.path()` to the id3 + mp3_duration crates which only expose + // `from_path` APIs. Peak process-heap = one chunk (~1 MiB) + // regardless of MP3 size. Guard drops → tempfile auto-removed. + let named = self + .dedup + .stream_blob_to_tempfile(blob_hash, &self.temp_dir, ".mp3") + .await?; + let path = named.path().to_path_buf(); + // ── Sync I/O on the blocking thread pool (never stalls Tokio workers) ── - let path = file_path.to_path_buf(); let metadata = tokio::task::spawn_blocking(move || Self::extract_metadata_blocking(&path)) .await .map_err(|e| { @@ -121,6 +138,10 @@ impl AudioMetadataService { format!("spawn_blocking join error: {e}"), ) })?; + // Explicitly hold `named` alive until after the extraction — the + // `spawn_blocking` closure only borrows the raw `path`, so the + // guard must not drop while the extractor is running. + drop(named); let Some(m) = metadata else { return Ok(()); @@ -208,8 +229,10 @@ impl AudioMetadataService { let audio_file = row.map_err(|e| { DomainError::database_error(format!("Failed to fetch audio file row: {}", e)) })?; - let file_path = self.blob_path(&audio_file.blob_hash); - match self.extract_and_save(&audio_file.file_id, &file_path).await { + match self + .extract_and_save(&audio_file.file_id, &audio_file.blob_hash) + .await + { Ok(()) => processed += 1, Err(e) => { warn!( @@ -313,8 +336,7 @@ impl AudioMetadataService { } 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 { + if let Err(e) = service.extract_and_save(&new_file_id, &blob_hash).await { warn!( "Failed to extract audio metadata for {}: {}", new_file_id, e @@ -368,10 +390,11 @@ impl FileLifecycleHook for AudioMetadataService { }; let service = Arc::new(Self { pool: self.pool.clone(), - blob_root: self.blob_root.clone(), + dedup: self.dedup.clone(), + temp_dir: self.temp_dir.clone(), }); if is_new_blob { - Self::spawn_extraction_background(service, uuid, self.blob_path(blob_hash)); + Self::spawn_extraction_background(service, uuid, blob_hash.to_string()); } else { Self::clone_or_extract_background(service, uuid, blob_hash.to_string()); } @@ -400,7 +423,8 @@ impl FileLifecycleHook for AudioMetadataService { }; let service = Arc::new(Self { pool: self.pool.clone(), - blob_root: self.blob_root.clone(), + dedup: self.dedup.clone(), + temp_dir: self.temp_dir.clone(), }); Self::clone_from_source_background(service, uuid, source_uuid, blob_hash.to_string()); } @@ -415,9 +439,10 @@ impl FileLifecycleHook for AudioMetadataService { }; let service = Arc::new(Self { pool: self.pool.clone(), - blob_root: self.blob_root.clone(), + dedup: self.dedup.clone(), + temp_dir: self.temp_dir.clone(), }); - Self::spawn_extraction_with_delete_background(service, uuid, self.blob_path(blob_hash)); + Self::spawn_extraction_with_delete_background(service, uuid, blob_hash.to_string()); } fn on_file_deleted(&self, _file_id: &str) { diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 7c4e10a6..c8b84fa3 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -2056,6 +2056,56 @@ impl DedupService { Ok(Bytes::from(data)) } + /// Stream a blob to a temp file for extractors that only accept a + /// filesystem `Path` (id3, mp3_duration, ffprobe, nom-exif video). + /// CDC-aware — reads through [`Self::read_blob_stream`] so a chunked + /// file's chunks are concatenated on the fly. Peak process-heap = + /// one chunk (~1 MiB) regardless of blob size. + /// + /// `temp_dir` is the destination directory (typically + /// `AppConfig::temp_dir`, from env `OXICLOUD_TEMP_DIR`). `suffix` + /// is appended to the tempfile name (e.g. `".mp3"`, `".jpg"`) so + /// content-sniffing extractors get a hint. The returned + /// `NamedTempFile` auto-removes on drop; callers pass `.path()` + /// to the extractor, then let the guard fall out of scope. + pub async fn stream_blob_to_tempfile( + &self, + hash: &str, + temp_dir: &std::path::Path, + suffix: &str, + ) -> Result { + use tokio::io::AsyncWriteExt; + let named = tempfile::Builder::new() + .prefix("oxi-blob-") + .suffix(suffix) + .tempfile_in(temp_dir) + .map_err(|e| { + DomainError::internal_error("Dedup", format!("mktemp in {:?}: {e}", temp_dir)) + })?; + // Re-open with tokio's async File so we can await writes. + let path = named.path().to_path_buf(); + let mut file = tokio::fs::OpenOptions::new() + .write(true) + .truncate(true) + .open(&path) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("reopen temp: {e}")))?; + // CDC-aware: manifest lookup + chunk concat OR legacy backend passthrough. + let mut stream = self.read_blob_stream(hash).await?; + while let Some(chunk) = stream.next().await { + let bytes = chunk + .map_err(|e| DomainError::internal_error("Dedup", format!("stream chunk: {e}")))?; + file.write_all(&bytes) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("temp write: {e}")))?; + } + file.flush() + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("temp flush: {e}")))?; + drop(file); + Ok(named) + } + /// Stream a byte range — CDC-aware with legacy fallback. /// /// For CDC files: calculates which chunks overlap the requested range, @@ -3971,6 +4021,83 @@ mod rechunk_integration_tests { cleanup(&pool, &hash, &files).await; } + + // ─── stream_blob_to_tempfile — CDC-aware read to a filesystem path ─── + // + // Regression tests for the fix landed on `fix/services-use-blob-abstraction`: + // audio_metadata_service, media_metadata_service, and face_indexing_service + // all read blob content via DedupService (`read_blob_bytes` / + // `stream_blob_to_tempfile`), NOT the raw `BlobStorageBackend`. If someone + // reverts a service to `backend.get_blob_stream(hash)`, this test fails + // because `hash` is a chunk-manifest hash — the physical backend has no + // blob at that key. Bug returns silently otherwise; these tests catch it. + + /// Local backend: seed a > 64 KiB blob, rechunk to CDC, then call + /// `stream_blob_to_tempfile` and verify the tempfile contents match + /// the original. Proves the CDC chunk-concat path works. + #[tokio::test] + async fn stream_blob_to_tempfile_reads_cdc_chunked_local() { + let pool = test_pool().await; + let dir = TempDir::new().unwrap(); + let svc = local_svc(&pool, &dir).await; + + // 200 KiB → forced multi-chunk after rechunk_legacy_blobs. + let data = content(200 * 1024, 33); + let (hash, files) = seed_legacy(&svc, &pool, &dir, &data, 1, "cdc-local", None).await; + svc.rechunk_legacy_blobs().await.expect("sweep"); + + // Sanity: rechunk actually produced a manifest (i.e. we're on the + // CDC path, not the legacy-fallback branch of read_blob_stream). + assert!( + manifest(&pool, &hash).await.is_some(), + "expected a CDC manifest after rechunk (test wouldn't cover the bug otherwise)" + ); + + // New method — the entry point audio/media services use. + let temp_dir = TempDir::new().unwrap(); + let named = svc + .stream_blob_to_tempfile(&hash, temp_dir.path(), ".bin") + .await + .expect("stream_blob_to_tempfile must succeed on CDC-chunked blob"); + + let round_tripped = tokio::fs::read(named.path()).await.expect("read tempfile"); + assert_eq!(round_tripped, data, "tempfile content must match original"); + + cleanup(&pool, &hash, &files).await; + } + + /// Encrypted backend variant — proves the wrapper stack (decryption + /// on read) is honoured. Same regression class: if a service reads + /// raw ciphertext instead of going through DedupService, this fails. + #[tokio::test] + async fn stream_blob_to_tempfile_reads_cdc_chunked_encrypted() { + let pool = test_pool().await; + let dir = TempDir::new().unwrap(); + let svc = encrypted_svc(&pool, &dir).await; + + let data = content(150 * 1024, 77); + let (hash, files) = seed_legacy(&svc, &pool, &dir, &data, 1, "cdc-enc", None).await; + svc.rechunk_legacy_blobs().await.expect("sweep"); + + assert!( + manifest(&pool, &hash).await.is_some(), + "expected a CDC manifest after rechunk" + ); + + let temp_dir = TempDir::new().unwrap(); + let named = svc + .stream_blob_to_tempfile(&hash, temp_dir.path(), ".bin") + .await + .expect("stream_blob_to_tempfile must succeed on encrypted CDC blob"); + + let round_tripped = tokio::fs::read(named.path()).await.expect("read tempfile"); + assert_eq!( + round_tripped, data, + "tempfile content must match original plaintext (wrapper stack must decrypt transparently)" + ); + + cleanup(&pool, &hash, &files).await; + } } // ───────────────────────────────────────────────────────────────────────────── diff --git a/src/infrastructure/services/face_indexing_service.rs b/src/infrastructure/services/face_indexing_service.rs index b1700aad..44b4f504 100644 --- a/src/infrastructure/services/face_indexing_service.rs +++ b/src/infrastructure/services/face_indexing_service.rs @@ -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, repo: Arc, analyzer: Arc, - 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, /// 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, blob_root: PathBuf, analyzer: Arc) -> Self { + pub fn new( + pool: Arc, + dedup: Arc, + analyzer: Arc, + ) -> 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 { - 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, 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 = detected diff --git a/src/infrastructure/services/media_metadata_service.rs b/src/infrastructure/services/media_metadata_service.rs index 3d165711..f7feb673 100644 --- a/src/infrastructure/services/media_metadata_service.rs +++ b/src/infrastructure/services/media_metadata_service.rs @@ -32,6 +32,7 @@ use uuid::Uuid; use crate::application::ports::file_lifecycle::FileLifecycleHook; use crate::common::errors::DomainError; use crate::infrastructure::repositories::pg::file_metadata_repository::FileMetadataRepository; +use crate::infrastructure::services::dedup_service::DedupService; use crate::infrastructure::services::exif_service::{ExifMetadata, ExifService}; #[derive(Debug, FromRow)] @@ -50,12 +51,22 @@ pub struct MetadataExtractionResult { pub struct MediaMetadataService { pool: Arc, - 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, + /// Tier-1 scratch directory for `stream_blob_to_tempfile`. Pulled + /// from `AppConfig::temp_dir` (env `OXICLOUD_TEMP_DIR`) at DI time. + temp_dir: PathBuf, } impl MediaMetadataService { - pub fn new(pool: Arc, blob_root: PathBuf) -> Self { - Self { pool, blob_root } + pub fn new(pool: Arc, dedup: Arc, temp_dir: PathBuf) -> Self { + Self { + pool, + dedup, + temp_dir, + } } pub fn is_image_file(mime_type: &str) -> bool { @@ -71,15 +82,11 @@ impl MediaMetadataService { Self::is_image_file(mime_type) || Self::is_video_file(mime_type) } - fn blob_path(&self, hash: &str) -> PathBuf { - let prefix = &hash[0..2]; - self.blob_root.join(prefix).join(format!("{}.blob", hash)) - } - fn arc(&self) -> Arc { Arc::new(Self { pool: self.pool.clone(), - blob_root: self.blob_root.clone(), + dedup: self.dedup.clone(), + temp_dir: self.temp_dir.clone(), }) } @@ -130,13 +137,32 @@ impl MediaMetadataService { /// Extract metadata for one file and persist it (no-op when nothing useful /// could be extracted). + /// + /// Streams the blob (CDC-aware — chunks concatenated on the fly for + /// chunked files; wrapper stack handles encryption + retry + cache) + /// to a tempfile in the configured tier-1 temp dir, then hands its + /// `.path()` to the sync extractors (kamadak-exif via + /// `std::fs::read(path)` for images, `nom-exif` video track reader + /// for videos — both are path-based). Peak process-heap = one chunk + /// (~1 MiB) regardless of media size. pub async fn extract_and_save( &self, file_id: &Uuid, - file_path: &Path, + blob_hash: &str, mime_type: &str, ) -> Result<(), DomainError> { - let path = file_path.to_path_buf(); + // File-extension hint for the tempfile suffix — helps + // `nom-exif`'s content sniffer land the right parser branch. + let suffix = if Self::is_video_file(mime_type) { + ".mp4" + } else { + ".jpg" + }; + let named = self + .dedup + .stream_blob_to_tempfile(blob_hash, &self.temp_dir, suffix) + .await?; + let path = named.path().to_path_buf(); let mime = mime_type.to_string(); let meta = tokio::task::spawn_blocking(move || Self::extract_blocking(&path, &mime)) .await @@ -146,6 +172,9 @@ impl MediaMetadataService { format!("spawn_blocking join error: {e}"), ) })?; + // Keep the tempfile alive until after extraction — the + // spawn_blocking closure only borrowed the raw path. + drop(named); let Some(meta) = meta else { return Ok(()); @@ -175,13 +204,13 @@ impl MediaMetadataService { pub fn spawn_extraction_background( service: Arc, file_id: Uuid, - file_path: PathBuf, + blob_hash: String, mime_type: String, ) { tokio::spawn(async move { tracing::info!("📷 Extracting capture metadata for: {}", file_id); if let Err(e) = service - .extract_and_save(&file_id, &file_path, &mime_type) + .extract_and_save(&file_id, &blob_hash, &mime_type) .await { tracing::warn!("Failed to extract capture metadata: {}", e); @@ -192,14 +221,14 @@ impl MediaMetadataService { pub fn spawn_extraction_with_delete_background( service: Arc, file_id: Uuid, - file_path: PathBuf, + blob_hash: String, mime_type: String, ) { tokio::spawn(async move { tracing::info!("📷 Updating capture metadata for: {}", file_id); let _ = service.delete_metadata(&file_id).await; if let Err(e) = service - .extract_and_save(&file_id, &file_path, &mime_type) + .extract_and_save(&file_id, &blob_hash, &mime_type) .await { tracing::warn!("Failed to update capture metadata: {}", e); @@ -287,9 +316,8 @@ impl MediaMetadataService { info!("Cloned capture metadata for file {}", new_file_id); } Ok(_) => { - let file_path = service.blob_path(&blob_hash); if let Err(e) = service - .extract_and_save(&new_file_id, &file_path, &mime_type) + .extract_and_save(&new_file_id, &blob_hash, &mime_type) .await { warn!( @@ -335,9 +363,8 @@ impl MediaMetadataService { let media = row.map_err(|e| { DomainError::database_error(format!("Failed to fetch media file row: {}", e)) })?; - let file_path = self.blob_path(&media.blob_hash); match self - .extract_and_save(&media.file_id, &file_path, &media.mime_type) + .extract_and_save(&media.file_id, &media.blob_hash, &media.mime_type) .await { Ok(()) => processed += 1, @@ -530,7 +557,7 @@ impl FileLifecycleHook for MediaMetadataService { Self::spawn_extraction_background( service, uuid, - self.blob_path(blob_hash), + blob_hash.to_string(), content_type.to_string(), ); } else { @@ -584,7 +611,7 @@ impl FileLifecycleHook for MediaMetadataService { Self::spawn_extraction_with_delete_background( self.arc(), uuid, - self.blob_path(blob_hash), + blob_hash.to_string(), content_type.to_string(), ); }