diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index cf3be3dc..10310626 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/src/application/ports/blob_storage_ports.rs @@ -95,7 +95,10 @@ pub trait BlobStorageBackend: Send + Sync + 'static { /// Stream the full blob content in chunks. fn get_blob_stream(&self, hash: &str) -> BoxFut<'_, Result>; - /// Stream a byte range of the blob (for HTTP Range requests / video seek). + /// Stream the byte range `[start, end)` of the blob (for HTTP Range + /// requests / video seek). `end` is **exclusive**; `None` means "to the + /// end of the blob". Callers translating inclusive HTTP Range headers + /// must pass `last_byte + 1`. fn get_blob_range_stream( &self, hash: &str, diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index e43d6715..213cb759 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -286,6 +286,10 @@ pub trait FileWritePort: Send + Sync + 'static { /// When `pre_computed_hash` is provided, the dedup service skips the /// hash re-read — zero extra I/O beyond the initial spool. /// Peak RAM: ~256 KB regardless of file size. + /// + /// Returns `(new_blob_hash, updated_at_epoch)` — everything a caller + /// needs to rebuild the fresh entity/ETag from a `File` it already + /// holds, without re-reading the row it just updated. async fn update_file_content_from_temp( &self, file_id: &str, @@ -294,7 +298,7 @@ pub trait FileWritePort: Send + Sync + 'static { content_type: Option, pre_computed_hash: Option, modified_at: Option, - ) -> Result; + ) -> Result<(String, i64), DomainError>; /// Registers file metadata WITHOUT writing content to disk (write-behind). /// diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index 9490f7e6..30b064dd 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -319,7 +319,8 @@ impl FileUploadUseCase for FileUploadService { && let Some(file) = file_read.find_file_by_path(path).await? { let file_id = file.id().to_string(); - self.file_write + let (new_hash, updated_at) = self + .file_write .update_file_content_from_temp( &file_id, temp_path, @@ -333,8 +334,26 @@ impl FileUploadUseCase for FileUploadService { if let Some(cc) = &self.content_cache { cc.invalidate(&file_id).await; } - // Re-read to get fresh DTO with updated etag and timestamps. - let updated = file_read.get_file(&file_id).await?; + // Rebuild the fresh DTO from the entity already in hand plus the + // values the UPDATE just returned — a re-read would only fetch + // what we already know, at one extra round-trip per overwrite + // (WebDAV sync clients overwrite constantly). + let parts = file.into_parts(); + let updated = crate::domain::entities::file::File::with_timestamps_and_blob_hash( + parts.id, + parts.name, + parts.storage_path, + size, + parts.mime_type, + parts.folder_id, + parts.created_at, + updated_at as u64, + parts.owner_id, + new_hash, + ) + .map_err(|e| { + DomainError::internal_error("FileUpload", format!("rebuild entity: {e}")) + })?; let dto = FileDto::from(updated); if let Some(hook) = &self.file_lifecycle_hook { hook.on_file_updated(&file_id, &dto.etag, content_type); diff --git a/src/application/services/idor_protection_test.rs b/src/application/services/idor_protection_test.rs index 19dd3412..8682ae28 100644 --- a/src/application/services/idor_protection_test.rs +++ b/src/application/services/idor_protection_test.rs @@ -213,8 +213,8 @@ impl FileWritePort for MockFileWritePort { _content_type: Option, _pre_computed_hash: Option, _modified_at: Option, - ) -> Result { - Ok(String::new()) + ) -> Result<(String, i64), DomainError> { + Ok((String::new(), 0)) } async fn register_file_deferred( diff --git a/src/application/services/storage_usage_service.rs b/src/application/services/storage_usage_service.rs index ced34892..20157a3c 100644 --- a/src/application/services/storage_usage_service.rs +++ b/src/application/services/storage_usage_service.rs @@ -31,19 +31,36 @@ impl StorageUsageService { } } - /// Calculates and updates storage usage for a specific user + /// Recalculates and stores one user's usage in a single statement. + /// + /// The correlated `SUM(size)` over the user's non-trashed files is + /// O(number of files) but runs as an index-only scan on the + /// `idx_files_user_size_active` covering partial index. One round-trip + /// (was three: user lookup + SUM + UPDATE). NOT called on the request + /// path — only by the per-upload background update and the sweep. pub async fn update_user_storage_usage(&self, user_id: Uuid) -> Result { - info!("Updating storage usage for user: {}", user_id); + let total_usage: Option = sqlx::query_scalar( + r#" + UPDATE auth.users u + SET storage_used_bytes = COALESCE(( + SELECT SUM(f.size)::bigint + FROM storage.files f + WHERE f.user_id = u.id AND NOT f.is_trashed), 0) + WHERE u.id = $1 + RETURNING u.storage_used_bytes + "#, + ) + .bind(user_id) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("StorageUsage", format!("Failed to update usage: {e}")) + })?; - // Calculate storage usage directly from database - let total_usage = self.calculate_user_storage_usage(user_id).await?; + let total_usage = total_usage + .ok_or_else(|| DomainError::not_found("User", format!("User ID: {user_id}")))?; - // Update the user's storage usage in the database - self.user_repository - .update_storage_usage(user_id, total_usage) - .await?; - - info!( + debug!( "Updated storage usage for user {} to {} bytes", user_id, total_usage ); @@ -51,61 +68,35 @@ impl StorageUsageService { Ok(total_usage) } - /// Calculates a user's storage usage by summing all their file sizes. - /// - /// This is `SUM(size)` over the user's non-trashed files — O(number of - /// files), backed by the `idx_files_user_size_active` covering partial - /// index so it runs as an index-only scan. It is NOT called on the request - /// path; only by the per-upload update and the background reconciliation - /// sweep. - async fn calculate_user_storage_usage(&self, user_id: Uuid) -> Result { - debug!("Calculating storage for user: {}", user_id); - - // Direct SQL query to sum all file sizes for this user - // This is much more efficient than recursively walking folders - let total_size: i64 = sqlx::query_scalar( - r#" - SELECT COALESCE(SUM(size), 0)::bigint - FROM storage.files - WHERE user_id = $1 AND NOT is_trashed - "#, - ) - .bind(user_id) - .fetch_one(self.pool.as_ref()) - .await - .map_err(|e| { - DomainError::internal_error("StorageUsage", format!("Failed to calculate usage: {e}")) - })?; - - debug!( - "Calculated storage for user {}: {} bytes", - user_id, total_size - ); - - Ok(total_size) - } - - /// Calculates and updates storage usage for a user identified by username. + /// Same as [`Self::update_user_storage_usage`], keyed by username. pub async fn update_user_storage_usage_by_username( &self, username: &str, ) -> Result { - info!("Updating storage usage for username: {}", username); + let total_usage: Option = sqlx::query_scalar( + r#" + UPDATE auth.users u + SET storage_used_bytes = COALESCE(( + SELECT SUM(f.size)::bigint + FROM storage.files f + WHERE f.user_id = u.id AND NOT f.is_trashed), 0) + WHERE u.username = $1 + RETURNING u.storage_used_bytes + "#, + ) + .bind(username) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("StorageUsage", format!("Failed to update usage: {e}")) + })?; - let user = self.user_repository.get_user_by_username(username).await?; - let user_id = user.id(); + let total_usage = + total_usage.ok_or_else(|| DomainError::not_found("User", username.to_string()))?; - // Reuse the existing calculation logic - let total_usage = self.calculate_user_storage_usage(user_id).await?; - - // Update the user's storage usage in the database - self.user_repository - .update_storage_usage(user_id, total_usage) - .await?; - - info!( - "Updated storage usage for username {} (id={}) to {} bytes", - username, user_id, total_usage + debug!( + "Updated storage usage for username {} to {} bytes", + username, total_usage ); Ok(total_usage) @@ -159,49 +150,49 @@ impl StorageUsagePort for StorageUsageService { StorageUsageService::update_user_storage_usage_by_username(self, username).await } + /// Reconcile every internal user's cached usage in ONE set-based UPDATE. + /// + /// Replaces the previous shape (paginated user list + one spawned task + /// per user, each issuing SUM + UPDATE — up to 2N queries and N + /// concurrent tasks fighting for pool connections). A single GROUP BY + /// over the covering index feeds all users at once, and the + /// `IS DISTINCT FROM` guard skips rewriting rows whose value didn't + /// change (no dead-tuple churn for idle users). This also removes the + /// old `LIMIT 1000` page cap, which silently left users beyond the + /// first thousand unreconciled. + /// + /// External users are excluded — they carry no storage by construction + /// (DB CHECK `users_external_no_storage`). async fn update_all_users_storage_usage(&self) -> Result<(), DomainError> { - info!("Starting batch update of all users' storage usage"); + debug!("Starting storage-usage reconciliation sweep"); - // Get the list of all users - // include_external=false — external users carry no storage by - // construction (DB CHECK `users_external_no_storage`), so there's - // nothing to compute for them. - let users = self.user_repository.list_users(1000, 0, false).await?; + let result = sqlx::query( + r#" + UPDATE auth.users u + SET storage_used_bytes = COALESCE(t.total, 0) + FROM auth.users u2 + LEFT JOIN ( + SELECT user_id, SUM(size)::bigint AS total + FROM storage.files + WHERE NOT is_trashed + GROUP BY user_id + ) t ON t.user_id = u2.id + WHERE u.id = u2.id + AND NOT u2.is_external + AND u.storage_used_bytes IS DISTINCT FROM COALESCE(t.total, 0) + "#, + ) + .execute(self.pool.as_ref()) + .await + .map_err(|e| { + error!("Storage-usage reconciliation sweep failed: {}", e); + DomainError::internal_error("StorageUsage", format!("reconciliation sweep: {e}")) + })?; - let mut update_tasks = Vec::new(); - - // Process users in parallel - for user in users { - let user_id = user.id(); - let service_clone = self.clone(); - - // Spawn a background task for each user - let task = task::spawn(async move { - match service_clone.update_user_storage_usage(user_id).await { - Ok(usage) => { - debug!( - "Updated storage usage for user {}: {} bytes", - user_id, usage - ); - Ok(()) - } - Err(e) => { - error!("Failed to update storage for user {}: {}", user_id, e); - Err(e) - } - } - }); - - update_tasks.push(task); - } - - // Wait for all tasks to complete - for task in update_tasks { - // We don't propagate errors from individual users to avoid failing the entire batch - let _ = task.await; - } - - info!("Completed batch update of all users' storage usage"); + info!( + "Storage-usage reconciliation corrected {} user(s)", + result.rows_affected() + ); Ok(()) } diff --git a/src/application/services/trash_service_test.rs b/src/application/services/trash_service_test.rs index dbe1c660..951abf39 100644 --- a/src/application/services/trash_service_test.rs +++ b/src/application/services/trash_service_test.rs @@ -594,8 +594,8 @@ impl FileWritePort for MockFileRepository { _content_type: Option, _pre_computed_hash: Option, _modified_at: Option, - ) -> std::result::Result { - Ok(String::new()) + ) -> std::result::Result<(String, i64), DomainError> { + Ok((String::new(), 0)) } async fn register_file_deferred( diff --git a/src/common/stubs.rs b/src/common/stubs.rs index be8b7cbd..c2c3957e 100644 --- a/src/common/stubs.rs +++ b/src/common/stubs.rs @@ -193,8 +193,8 @@ impl FileWritePort for StubFileWritePort { _content_type: Option, _pre_computed_hash: Option, _modified_at: Option, - ) -> Result { - Ok(String::new()) + ) -> Result<(String, i64), DomainError> { + Ok((String::new(), 0)) } async fn register_file_deferred( diff --git a/src/infrastructure/repositories/pg/file_blob_write_repository.rs b/src/infrastructure/repositories/pg/file_blob_write_repository.rs index 7819f939..4bc7ba04 100644 --- a/src/infrastructure/repositories/pg/file_blob_write_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_write_repository.rs @@ -137,16 +137,19 @@ impl FileBlobWriteRepository { /// removing the new blob reference. /// /// `modified_at`: if `Some`, sets `updated_at` to that Unix timestamp; - /// if `None`, uses `NOW()` (server time). Returns the new hash on success. + /// if `None`, uses `NOW()` (server time). Returns + /// `(new_hash, updated_at_epoch)` on success — the effective timestamp + /// is returned so callers can rebuild the fresh entity without + /// re-reading the row. async fn swap_blob_hash( &self, file_id: &str, new_hash: &str, new_size: i64, modified_at: Option, - ) -> Result { + ) -> Result<(String, i64), DomainError> { // Atomic CTE: capture old hash then update in one round-trip, no TOCTOU. - let old_hash = match sqlx::query_scalar::<_, String>( + let (old_hash, updated_at) = match sqlx::query_as::<_, (String, i64)>( r#" WITH old AS ( SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE @@ -156,7 +159,7 @@ impl FileBlobWriteRepository { updated_at = COALESCE(to_timestamp($4), NOW()) FROM old WHERE f.id = old.id - RETURNING old.blob_hash + RETURNING old.blob_hash, EXTRACT(EPOCH FROM f.updated_at)::bigint "#, ) .bind(new_hash) @@ -166,7 +169,7 @@ impl FileBlobWriteRepository { .fetch_optional(self.pool.as_ref()) .await { - Ok(Some(old)) => old, + Ok(Some(row)) => row, Ok(None) => { // File not found — compensate: remove the new blob ref if let Err(e) = self.dedup.remove_reference(new_hash).await { @@ -201,7 +204,7 @@ impl FileBlobWriteRepository { ); } - Ok(new_hash.to_string()) + Ok((new_hash.to_string(), updated_at)) } /// Like [`FileWritePort::save_file_from_temp`] but also returns whether the @@ -513,7 +516,7 @@ impl FileWritePort for FileBlobWriteRepository { content_type: Option, pre_computed_hash: Option, modified_at: Option, - ) -> Result { + ) -> Result<(String, i64), DomainError> { // Streaming: pass pre-computed hash so dedup skips re-reading the file. let dedup_result = self .dedup diff --git a/src/infrastructure/services/encrypted_blob_backend.rs b/src/infrastructure/services/encrypted_blob_backend.rs index e690d3b0..a16411c7 100644 --- a/src/infrastructure/services/encrypted_blob_backend.rs +++ b/src/infrastructure/services/encrypted_blob_backend.rs @@ -10,16 +10,31 @@ //! dedup still works correctly. //! //! Layout on disk/S3: `[12-byte nonce][ciphertext + 16-byte GCM tag]` +//! +//! ## Runtime & memory characteristics +//! +//! GCM is all-or-nothing per blob: a blob can only be decrypted whole, so +//! every read materializes the full plaintext. This stays bounded because +//! `DedupService` stores all new content as CDC chunks (≤ 1 MiB each) and +//! resolves Range requests to the overlapping chunks *before* calling this +//! backend — an encrypted seek in a large video decrypts a handful of +//! chunks, never the file. The unbounded case is **legacy whole-file +//! blobs** written before CDC chunking: a range read of one still decrypts +//! the entire blob (re-uploading the file re-stores it chunked). +//! +//! Crypto work for payloads ≥ 64 KiB runs on the blocking pool so AES-GCM +//! never stalls the async runtime, and decryption happens **in place** — +//! the ciphertext buffer is reused for the plaintext instead of allocating +//! a second copy. use std::path::{Path, PathBuf}; use std::pin::Pin; -use aes_gcm::aead::{Aead, KeyInit, OsRng}; +use aes_gcm::aead::{Aead, AeadInPlace, KeyInit, OsRng}; use aes_gcm::{AeadCore, Aes256Gcm, Nonce}; use bytes::Bytes; use std::sync::Arc; use tokio::fs; -use tokio::io::AsyncWriteExt; use crate::application::ports::blob_storage_ports::{ BlobStorageBackend, BlobStream, StorageHealthStatus, @@ -29,6 +44,15 @@ use crate::domain::errors::DomainError; /// Nonce size for AES-256-GCM (96 bits = 12 bytes). const NONCE_SIZE: usize = 12; +/// Payloads at or above this size run crypto on the blocking pool; below +/// it the `spawn_blocking` round-trip costs more than the AES work itself. +const CRYPTO_OFFLOAD_THRESHOLD: usize = 64 * 1024; + +/// Emission size for decrypted payloads — matches the 64 KiB chunks the +/// unencrypted backends stream, so downstream consumers (HTTP bodies, +/// hashers) see the same backpressure shape either way. +const PLAINTEXT_EMIT_SIZE: usize = 64 * 1024; + /// `BlobStorageBackend` decorator that encrypts blobs at rest. pub struct EncryptedBlobBackend { inner: Arc, @@ -66,6 +90,51 @@ fn encrypt_bytes(cipher: &Aes256Gcm, data: &[u8]) -> Result Ok(Bytes::from(encrypted)) } +/// Decrypt the on-disk layout `[nonce][ciphertext + tag]` **in place**. +/// +/// Consumes the encrypted buffer and reuses it for the plaintext, so peak +/// RAM is one buffer — not ciphertext + plaintext side by side (which for +/// legacy whole-file blobs would double a multi-hundred-MB allocation). +fn decrypt_bytes(cipher: &Aes256Gcm, mut encrypted: Vec) -> Result { + if encrypted.len() < NONCE_SIZE { + return Err(DomainError::internal_error( + "Encryption", + "encrypted blob too short (missing nonce)", + )); + } + let mut ciphertext = encrypted.split_off(NONCE_SIZE); // `encrypted` keeps the nonce + let nonce = Nonce::from_slice(&encrypted); + cipher + .decrypt_in_place(nonce, b"", &mut ciphertext) + .map_err(|e| DomainError::internal_error("Encryption", format!("decrypt failed: {e}")))?; + Ok(Bytes::from(ciphertext)) +} + +/// Run a crypto closure inline for small payloads, on the blocking pool for +/// large ones — AES-GCM over megabytes must not stall async workers. +async fn offload_crypto(work_len: usize, job: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> Result + Send + 'static, +{ + if work_len < CRYPTO_OFFLOAD_THRESHOLD { + return job(); + } + tokio::task::spawn_blocking(job) + .await + .map_err(|e| DomainError::internal_error("Encryption", format!("crypto task join: {e}")))? +} + +/// Turn a decrypted payload into a stream of bounded, zero-copy slices. +fn plaintext_stream(data: Bytes) -> BlobStream { + let len = data.len(); + let slices: Vec> = (0..len) + .step_by(PLAINTEXT_EMIT_SIZE) + .map(|off| Ok(data.slice(off..len.min(off + PLAINTEXT_EMIT_SIZE)))) + .collect(); + Box::pin(futures::stream::iter(slices)) +} + impl BlobStorageBackend for EncryptedBlobBackend { fn initialize( &self, @@ -81,7 +150,6 @@ impl BlobStorageBackend for EncryptedBlobBackend { let inner = self.inner.clone(); let hash = hash.to_string(); let source = source_path.to_path_buf(); - // Clone cipher key material (Aes256Gcm is not Send-safe to move across await) let cipher = self.cipher.clone(); Box::pin(async move { // Read plaintext from source @@ -89,31 +157,14 @@ impl BlobStorageBackend for EncryptedBlobBackend { DomainError::internal_error("Encryption", format!("read source: {e}")) })?; - // Encrypt: nonce || ciphertext (includes GCM tag) - let nonce = Aes256Gcm::generate_nonce(&mut OsRng); - let ciphertext = cipher.encrypt(&nonce, plaintext.as_ref()).map_err(|e| { - DomainError::internal_error("Encryption", format!("encrypt failed: {e}")) - })?; + let len = plaintext.len(); + let encrypted = offload_crypto(len, move || encrypt_bytes(&cipher, &plaintext)).await?; - // Write encrypted blob to a temp file - let tmp = source.with_extension("enc.tmp"); - let mut file = fs::File::create(&tmp).await.map_err(|e| { - DomainError::internal_error("Encryption", format!("create tmp: {e}")) - })?; - file.write_all(nonce.as_slice()).await.map_err(|e| { - DomainError::internal_error("Encryption", format!("write nonce: {e}")) - })?; - file.write_all(&ciphertext).await.map_err(|e| { - DomainError::internal_error("Encryption", format!("write ciphertext: {e}")) - })?; - file.flush() - .await - .map_err(|e| DomainError::internal_error("Encryption", format!("flush: {e}")))?; - drop(file); - - let result = inner.put_blob(&hash, &tmp).await; - let _ = fs::remove_file(&tmp).await; - result + // Hand the ciphertext straight to the inner backend. The previous + // implementation spooled it to a `.enc.tmp` file only for the + // inner backend to read it back — a full extra write + read of + // every blob that came through this path. + inner.put_blob_from_bytes(&hash, encrypted).await }) } @@ -126,7 +177,8 @@ impl BlobStorageBackend for EncryptedBlobBackend { let hash = hash.to_string(); let cipher = self.cipher.clone(); Box::pin(async move { - let encrypted = encrypt_bytes(&cipher, data.as_ref())?; + let encrypted = + offload_crypto(data.len(), move || encrypt_bytes(&cipher, data.as_ref())).await?; inner.put_blob_from_bytes(&hash, encrypted).await }) } @@ -140,7 +192,8 @@ impl BlobStorageBackend for EncryptedBlobBackend { let hash = hash.to_string(); let cipher = self.cipher.clone(); Box::pin(async move { - let encrypted = encrypt_bytes(&cipher, data.as_ref())?; + let encrypted = + offload_crypto(data.len(), move || encrypt_bytes(&cipher, data.as_ref())).await?; inner.put_blob_from_bytes_unsynced(&hash, encrypted).await }) } @@ -163,28 +216,13 @@ impl BlobStorageBackend for EncryptedBlobBackend { let hash = hash.to_string(); let cipher = self.cipher.clone(); Box::pin(async move { - // Read entire encrypted blob (nonce + ciphertext) into memory for decryption + // GCM must see the whole message: collect ciphertext, decrypt in + // place off the runtime, then stream zero-copy plaintext slices. let enc_stream = inner.get_blob_stream(&hash).await?; let encrypted = collect_stream(enc_stream).await?; - - if encrypted.len() < NONCE_SIZE { - return Err(DomainError::internal_error( - "Encryption", - "encrypted blob too short (missing nonce)", - )); - } - - let (nonce_bytes, ciphertext) = encrypted.split_at(NONCE_SIZE); - let nonce = Nonce::from_slice(nonce_bytes); - let plaintext = cipher.decrypt(nonce, ciphertext).map_err(|e| { - DomainError::internal_error("Encryption", format!("decrypt failed: {e}")) - })?; - - let stream: BlobStream = - Box::pin(futures::stream::once( - async move { Ok(Bytes::from(plaintext)) }, - )); - Ok(stream) + let len = encrypted.len(); + let plaintext = offload_crypto(len, move || decrypt_bytes(&cipher, encrypted)).await?; + Ok(plaintext_stream(plaintext)) }) } @@ -199,31 +237,25 @@ impl BlobStorageBackend for EncryptedBlobBackend { let hash = hash.to_string(); let cipher = self.cipher.clone(); Box::pin(async move { - // Must decrypt the full blob then slice the plaintext range + // Decrypt the full blob, then slice the plaintext range without + // copying. For CDC chunks (every blob written since chunking + // landed) this is ≤ 1 MiB; only legacy whole-file blobs pay a + // full-blob decrypt here — see the module docs. let enc_stream = inner.get_blob_stream(&hash).await?; let encrypted = collect_stream(enc_stream).await?; + let len = encrypted.len(); + let plaintext = offload_crypto(len, move || decrypt_bytes(&cipher, encrypted)).await?; - if encrypted.len() < NONCE_SIZE { - return Err(DomainError::internal_error( - "Encryption", - "encrypted blob too short", - )); - } + // `end` is exclusive — same contract as `LocalBlobBackend`, whose + // implementation reads `end - start` bytes. The previous version + // here treated it as inclusive and returned one extra byte on + // every bounded range, corrupting 206 responses when encryption + // was enabled. + let total = plaintext.len(); + let end_excl = end.map(|e| e as usize).unwrap_or(total).min(total); + let start = (start as usize).min(end_excl); - let (nonce_bytes, ciphertext) = encrypted.split_at(NONCE_SIZE); - let nonce = Nonce::from_slice(nonce_bytes); - let plaintext = cipher.decrypt(nonce, ciphertext).map_err(|e| { - DomainError::internal_error("Encryption", format!("decrypt failed: {e}")) - })?; - - let start = start as usize; - let end = end.map(|e| (e as usize) + 1).unwrap_or(plaintext.len()); - let end = end.min(plaintext.len()); - let start = start.min(end); - - let slice = Bytes::from(plaintext[start..end].to_vec()); - let stream: BlobStream = Box::pin(futures::stream::once(async move { Ok(slice) })); - Ok(stream) + Ok(plaintext_stream(plaintext.slice(start..end_excl))) }) } @@ -326,9 +358,9 @@ mod tests { let decrypted = collect_stream(stream).await.unwrap(); assert_eq!(decrypted, data); - // Read range + // Read range — `end` is exclusive, matching LocalBlobBackend let range_stream = encrypted - .get_blob_range_stream(hash, 7, Some(15)) + .get_blob_range_stream(hash, 7, Some(16)) .await .unwrap(); let range_data = collect_stream(range_stream).await.unwrap(); @@ -345,4 +377,103 @@ mod tests { encrypted.delete_blob(hash).await.unwrap(); assert!(!encrypted.blob_exists(hash).await.unwrap()); } + + /// Payloads above `CRYPTO_OFFLOAD_THRESHOLD` take the spawn_blocking + /// path and are emitted as multiple bounded slices — the roundtrip and + /// range semantics must be identical to the inline path. + #[tokio::test] + async fn test_large_blob_offloaded_roundtrip_and_ranges() { + let tmp = TempDir::new().unwrap(); + let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs"))); + local.initialize().await.unwrap(); + + let key = EncryptedBlobBackend::generate_key(); + let encrypted = EncryptedBlobBackend::new(local, &key); + + // 300 KiB of a repeating pattern — crosses the offload threshold and + // spans several PLAINTEXT_EMIT_SIZE slices. + let data: Vec = (0..300 * 1024).map(|i| (i % 251) as u8).collect(); + let hash = "feedbeef1234567890feedbeef1234567890feedbeef1234567890feedbeef12"; + encrypted + .put_blob_from_bytes(hash, Bytes::from(data.clone())) + .await + .unwrap(); + + // Full roundtrip + let stream = encrypted.get_blob_stream(hash).await.unwrap(); + let decrypted = collect_stream(stream).await.unwrap(); + assert_eq!(decrypted, data); + + // Mid-file range crossing an emission boundary (`end` exclusive) + let (start, end) = (60_000u64, 200_000u64); + let stream = encrypted + .get_blob_range_stream(hash, start, Some(end)) + .await + .unwrap(); + let ranged = collect_stream(stream).await.unwrap(); + assert_eq!(ranged, &data[start as usize..end as usize]); + + // Open-ended suffix range + let stream = encrypted + .get_blob_range_stream(hash, 299 * 1024, None) + .await + .unwrap(); + let suffix = collect_stream(stream).await.unwrap(); + assert_eq!(suffix, &data[299 * 1024..]); + + // Range entirely past EOF yields empty content + let stream = encrypted + .get_blob_range_stream(hash, data.len() as u64 + 10, None) + .await + .unwrap(); + assert!(collect_stream(stream).await.unwrap().is_empty()); + + // Plaintext size reported + assert_eq!(encrypted.blob_size(hash).await.unwrap(), data.len() as u64); + } + + /// A flipped ciphertext byte must fail GCM authentication, never return + /// corrupted plaintext. + #[tokio::test] + async fn test_tampered_ciphertext_fails_decrypt() { + let tmp = TempDir::new().unwrap(); + let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs"))); + local.initialize().await.unwrap(); + + let key = EncryptedBlobBackend::generate_key(); + let encrypted = EncryptedBlobBackend::new(local.clone(), &key); + + let hash = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + encrypted + .put_blob_from_bytes(hash, Bytes::from_static(b"sensitive payload")) + .await + .unwrap(); + + // Corrupt one ciphertext byte on disk (past the 12-byte nonce). + let path = local.local_blob_path(hash).expect("local path"); + let mut raw = std::fs::read(&path).unwrap(); + raw[NONCE_SIZE] ^= 0xFF; + std::fs::write(&path, raw).unwrap(); + + assert!(encrypted.get_blob_stream(hash).await.is_err()); + } + + /// Decrypting with a different key must fail authentication. + #[tokio::test] + async fn test_wrong_key_fails_decrypt() { + let tmp = TempDir::new().unwrap(); + let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs"))); + local.initialize().await.unwrap(); + + let hash = "aaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccdddd"; + let writer = + EncryptedBlobBackend::new(local.clone(), &EncryptedBlobBackend::generate_key()); + writer + .put_blob_from_bytes(hash, Bytes::from_static(b"locked")) + .await + .unwrap(); + + let reader = EncryptedBlobBackend::new(local, &EncryptedBlobBackend::generate_key()); + assert!(reader.get_blob_stream(hash).await.is_err()); + } } diff --git a/static/js/features/library/photosLightbox.js b/static/js/features/library/photosLightbox.js index 2dc58076..098c4e0b 100644 --- a/static/js/features/library/photosLightbox.js +++ b/static/js/features/library/photosLightbox.js @@ -1,6 +1,13 @@ /** * OxiCloud - Photos Lightbox * Full-screen image/video viewer with prev/next navigation. + * + * Media is never buffered in page memory: videos stream straight from the + * API (the element's `src` is same-origin, so auth cookies travel + * automatically and the browser issues Range requests — playback starts + * progressively and seeking works without downloading the whole file). + * Photos open with the server-cached `large` thumbnail; the full-resolution + * original streams in only on demand via the toolbar expand button. */ import { getCsrfHeaders } from '../../core/csrf.js'; @@ -16,12 +23,17 @@ export const photosLightbox = { index: -1, /** @type {HTMLElement|null} */ _overlay: null, - /** @type {string|null} Current blob URL to revoke */ - _blobUrl: null, /** @type {(ev: KeyboardEvent) => any|null} */ _keyHandler: null, /** @type {PhotosView|null} Reference to photosView, set after both modules load */ _photosView: null, + /** + * Monotonic token identifying the most recent {@link photosLightbox._show} + * call. Image load/error callbacks fire asynchronously, so a rapid + * prev/next must not let a superseded item commit its (stale) content + * over the newer one. + */ + _showGeneration: 0, /** * Register the photosView reference (called from photos.js to avoid circular imports). @@ -36,6 +48,25 @@ export const photosLightbox = { return getCsrfHeaders(); }, + /** + * Streaming URL of the original file. Same-origin, so media elements + * send the auth cookie automatically and the browser handles Range. + * @param {FileItem} item + * @returns {string} + */ + _originalUrl(item) { + return `/api/files/${item.id}?inline=true`; + }, + + /** + * URL of the server-cached `large` thumbnail (immutable, browser-cached). + * @param {FileItem} item + * @returns {string} + */ + _thumbUrl(item) { + return `/api/files/${item.id}/thumbnail/large`; + }, + /** * Open lightbox at given index * @param {FileItem[]} items @@ -60,7 +91,6 @@ export const photosLightbox = { } }, 200); } - this._revokeBlob(); this._unbindKeys(); }, @@ -96,6 +126,7 @@ export const photosLightbox = {
+ @@ -117,7 +148,7 @@ export const photosLightbox = { } }); - // Toolbar actions + // Toolbar actions (`.lb-fullres` is wired per-item in `_show`) /** @type {HTMLButtonElement} */ (el.querySelector('.lb-download')).onclick = () => this._download(); /** @type {HTMLButtonElement} */ (el.querySelector('.lb-favorite')).onclick = () => this._toggleFavorite(); /** @type {HTMLButtonElement} */ (el.querySelector('.lb-delete')).onclick = () => this._delete(); @@ -127,8 +158,9 @@ export const photosLightbox = { }, /** Display the current item */ - async _show() { + _show() { if (!this._overlay || this.index < 0) return; + const generation = ++this._showGeneration; const item = this.items[this.index]; const content = this._overlay.querySelector('.lightbox-content'); @@ -154,31 +186,75 @@ export const photosLightbox = { /** @type {HTMLButtonElement} */ (this._overlay.querySelector('.lightbox-prev')).classList.toggle('hidden', !(this.index > 0)); /** @type {HTMLButtonElement} */ (this._overlay.querySelector('.lightbox-next')).classList.toggle('hidden', !(this.index < this.items.length - 1)); + // Reset the full-resolution button for the new item + const fullResBtn = /** @type {HTMLButtonElement} */ (this._overlay.querySelector('.lb-fullres')); + fullResBtn.classList.add('hidden'); + fullResBtn.disabled = false; + const fullResIcon = fullResBtn.querySelector('i'); + if (fullResIcon) fullResIcon.className = 'fas fa-expand'; + // Load content - this._revokeBlob(); content.innerHTML = '
'; - try { - const isVideo = item.mime_type?.startsWith('video/'); - const res = await fetch(`/api/files/${item.id}`, { - credentials: 'include', - headers: this._headers() + if (item.mime_type?.startsWith('video/')) { + const video = document.createElement('video'); + video.controls = true; + video.autoplay = true; + // Instant first frame while metadata loads; a 204 (no cached + // thumbnail yet) simply leaves the poster blank. + video.poster = this._thumbUrl(item); + video.src = this._originalUrl(item); + video.addEventListener('error', () => { + if (generation !== this._showGeneration) return; + content.innerHTML = '
Failed to load
'; }); - - if (!res.ok) throw new Error(`HTTP ${res.status}`); - const blob = await res.blob(); - this._blobUrl = URL.createObjectURL(blob); - - if (isVideo) { - content.innerHTML = ``; - } else { - content.innerHTML = `${this._escAttr(item.name)}`; - } - } catch (err) { - console.error('Lightbox load error:', err); - content.innerHTML = '
Failed to load
'; + // The native player has its own buffering UI — drop the spinner now. + content.replaceChildren(video); + this._loadMetadata(item.id, meta, dateStr, item.size_formatted || ''); + return; } + // Photo: thumbnail first, original on demand. GIFs go straight to + // the original — the thumbnail is a static JPEG and would lose the + // animation. + const isGif = item.mime_type === 'image/gif'; + let showingOriginal = isGif; + const img = document.createElement('img'); + img.alt = item.name; + + img.addEventListener('load', () => { + if (generation !== this._showGeneration) return; + // First load replaces the spinner; the on-demand swap reuses the + // already-attached element. + if (!img.isConnected) content.replaceChildren(img); + fullResBtn.classList.toggle('hidden', showingOriginal); + fullResBtn.disabled = false; + if (fullResIcon) fullResIcon.className = 'fas fa-expand'; + }); + + img.addEventListener('error', () => { + if (generation !== this._showGeneration) return; + if (!showingOriginal) { + // No server thumbnail (unsupported format or generation + // failed) — fall back to the original. + showingOriginal = true; + img.src = this._originalUrl(item); + } else { + content.innerHTML = '
Failed to load
'; + fullResBtn.classList.add('hidden'); + } + }); + + fullResBtn.onclick = () => { + if (generation !== this._showGeneration || showingOriginal) return; + showingOriginal = true; + fullResBtn.disabled = true; + if (fullResIcon) fullResIcon.className = 'fas fa-spinner fa-spin'; + img.src = this._originalUrl(item); + }; + + img.src = showingOriginal ? this._originalUrl(item) : this._thumbUrl(item); + // Load EXIF metadata this._loadMetadata(item.id, meta, dateStr, item.size_formatted || ''); }, @@ -295,19 +371,5 @@ export const photosLightbox = { document.removeEventListener('keydown', this._keyHandler); this._keyHandler = null; } - }, - - _revokeBlob() { - if (this._blobUrl) { - URL.revokeObjectURL(this._blobUrl); - this._blobUrl = null; - } - }, - - /** @param {any} s */ - _escAttr(s) { - return String(s || '') - .replace(/"/g, '"') - .replace(/