Merge pull request #657 from EdouardVanbelle/fix/services-use-blob-abstraction
fix: prevent services accessing directly to localstorage (audio, media metadata and face indexing)
This commit is contained in:
@@ -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}` |
|
||||
|
||||
@@ -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 | `<storage_path>/.thumbnails/`, `<storage_path>/.transcoded/`, `<storage_path>/.blob-cache/`, `<storage_path>/.search-index/`, `<storage_path>/.plugin-logs/` — all mixed into tier-3 storage today. |
|
||||
| **3 — Persistent data** | Source of truth | ❌ never | `<storage_path>/.blobs/` (Local) OR S3/Azure bucket, via `BlobStorageBackend`. Already correctly configured via `OXICLOUD_STORAGE_ENTRIES`. |
|
||||
|
||||
**Today's misclassification**: tier-2 sidecars live under
|
||||
`<storage_path>` — 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<u64, DomainError>;
|
||||
|
||||
/// 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<Vec<u8>>,
|
||||
limit: usize,
|
||||
) -> Result<(Vec<String>, Option<Vec<u8>>), 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<Arc<dyn BlobReferenceSource>>,
|
||||
}
|
||||
|
||||
impl BlobReferenceRegistry {
|
||||
pub fn register(&mut self, source: Arc<dyn BlobReferenceSource>);
|
||||
pub async fn total_references(&self, hash: &str) -> Result<u64, DomainError>;
|
||||
// ... 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
|
||||
`<storage_path>/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 `<spool_dir>/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<BlobHandler>` 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.
|
||||
+11
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<DedupService>`.** 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<dyn BlobStorageBackend>` 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 `<storage>/.blobs/<xx>/<hash>.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 `<storage_path>/*` 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).
|
||||
|
||||
@@ -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::<u16>()
|
||||
{
|
||||
|
||||
+14
-9
@@ -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<PgPool>,
|
||||
dedup: &Arc<crate::infrastructure::services::dedup_service::DedupService>,
|
||||
) -> Option<Arc<AudioMetadataService>> {
|
||||
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<PgPool>,
|
||||
dedup: &Arc<crate::infrastructure::services::dedup_service::DedupService>,
|
||||
) -> Arc<MediaMetadataService> {
|
||||
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<PgPool>,
|
||||
dedup: &Arc<crate::infrastructure::services::dedup_service::DedupService>,
|
||||
) -> Arc<crate::infrastructure::services::face_indexing_service::FaceIndexingService> {
|
||||
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,
|
||||
),
|
||||
)
|
||||
|
||||
@@ -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<PgPool>,
|
||||
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>,
|
||||
/// 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<PgPool>, blob_root: PathBuf) -> Self {
|
||||
Self { pool, blob_root }
|
||||
pub fn new(pool: Arc<PgPool>, dedup: Arc<DedupService>, 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<Self>, file_id: Uuid, file_path: PathBuf) {
|
||||
pub fn spawn_extraction_background(service: Arc<Self>, 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<Self>,
|
||||
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) {
|
||||
|
||||
@@ -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<tempfile::NamedTempFile, DomainError> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<PgPool>,
|
||||
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>,
|
||||
/// 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<PgPool>, blob_root: PathBuf) -> Self {
|
||||
Self { pool, blob_root }
|
||||
pub fn new(pool: Arc<PgPool>, dedup: Arc<DedupService>, 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<Self> {
|
||||
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<Self>,
|
||||
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<Self>,
|
||||
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(),
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user