diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index bda99dac..6815985f 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -453,6 +453,42 @@ pub trait StorageUsagePort: Send + Sync + 'static { /// Returns (used_bytes, quota_bytes) for a user. async fn get_user_storage_info(&self, user_id: Uuid) -> Result<(i64, i64), DomainError>; + + /// Incrementally adjust one drive's cached `storage.drives.used_bytes` + /// by `delta` bytes — O(1), the per-upload counterpart to the + /// O(N) full recompute below. Mirrors `add_user_storage_usage_delta` + /// in shape: single statement, `GREATEST(0, …)` clamp so a late or + /// duplicate adjustment can never drive the counter negative. + /// Deletes/trash do not decrement here (mirroring user-quota + /// design); the periodic reconciliation sweep is the correctness + /// backstop. + async fn add_drive_storage_usage_delta( + &self, + drive_id: Uuid, + delta: i64, + ) -> Result<(), DomainError>; + + /// Reconcile every drive's cached `used_bytes` against the actual + /// sum of its non-trashed files in one set-based UPDATE. Same + /// shape as `update_all_users_storage_usage`: `LEFT JOIN` over a + /// `GROUP BY drive_id` aggregate, with an `IS DISTINCT FROM` + /// guard so idle drives don't churn dead tuples. Runs from the + /// same reconciliation ticker. + async fn update_all_drives_storage_usage(&self) -> Result<(), DomainError>; + + /// Pre-upload quota check on a single drive. + /// + /// Returns `Ok(())` when `used_bytes + additional_bytes` fits under + /// `quota_bytes`, or `Err(QuotaExceeded)` otherwise. + /// `quota_bytes IS NULL` short-circuits to `Ok(())` — unlimited + /// drive. Single read-only `SELECT` on `storage.drives`; the + /// check/write window is a soft cap by design (same semantics as + /// the user-quota path), bounded by the sweep interval. + async fn check_drive_quota( + &self, + drive_id: Uuid, + additional_bytes: u64, + ) -> Result<(), DomainError>; } /// Generic storage service interface for calendar and contact services diff --git a/src/application/services/delta_upload_service.rs b/src/application/services/delta_upload_service.rs index f67dc015..d21ca200 100644 --- a/src/application/services/delta_upload_service.rs +++ b/src/application/services/delta_upload_service.rs @@ -331,6 +331,21 @@ impl DeltaUploadService { .check_storage_quota(caller_id, total_size) .await?; + // ── Per-drive quota (D4) ───────────────────────────────── + // Mirrors the per-user check above on the same `total_size`. + // Only on CREATE — Update replaces an existing row's content; + // tight size-delta accounting on update is a follow-up (today + // the periodic sweep reconciles drift either way). The + // single-statement `check_drive_quota_by_folder` lookup is a + // PK probe; cost matches the existing per-user check. + if let CommitMode::Create { folder_id, .. } = &mode { + let folder_uuid = Uuid::parse_str(folder_id) + .map_err(|_| DomainError::not_found("Folder", folder_id.clone()))?; + self.quota + .check_drive_quota_by_folder(folder_uuid, total_size) + .await?; + } + // ── Whole-file fast path: caller already owns this exact content ── // Mirrors the instant-upload endpoint: a reference bump, no chunk // work at all. Ownership is required — an existing-but-foreign diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index 612c1e0c..16fc8554 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -299,23 +299,46 @@ impl FileUploadService { let Some(storage_service) = &self.storage_usage_service else { return; }; - let Some(owner) = file + let delta = file.size as i64; + + // Per-user delta — unchanged from `b5b80549` / `fbbae541`. + if let Some(owner) = file .owner_id .as_deref() .and_then(|s| Uuid::parse_str(s).ok()) - else { - return; - }; - let delta = file.size as i64; - let service_clone = Arc::clone(storage_service); - tokio::spawn(async move { - if let Err(e) = service_clone - .add_user_storage_usage_delta(owner, delta) - .await - { - warn!("Failed to bump storage usage for {owner}: {e}"); - } - }); + { + let service_clone = Arc::clone(storage_service); + tokio::spawn(async move { + if let Err(e) = service_clone + .add_user_storage_usage_delta(owner, delta) + .await + { + warn!("Failed to bump storage usage for {owner}: {e}"); + } + }); + } + + // Per-drive delta (D4) — same fire-and-forget shape, resolves + // the drive id from the file's parent folder in one SQL + // statement. `storage.drives.used_bytes` is what the per-drive + // quota check and the picker quota bar read; drift from + // deletes / trash is reconciled by the same sweep that handles + // user-side drift. + if let Some(folder) = file + .folder_id + .as_deref() + .and_then(|s| Uuid::parse_str(s).ok()) + { + let service_clone = Arc::clone(storage_service); + tokio::spawn(async move { + if let Err(e) = service_clone + .add_drive_storage_usage_delta_by_folder(folder, delta) + .await + { + warn!("Failed to bump drive usage for folder {folder}: {e}"); + } + }); + } } } diff --git a/src/application/services/storage_usage_service.rs b/src/application/services/storage_usage_service.rs index a5bdb115..ce48b8ca 100644 --- a/src/application/services/storage_usage_service.rs +++ b/src/application/services/storage_usage_service.rs @@ -128,6 +128,157 @@ impl StorageUsageService { Ok(()) } + /// Incrementally adjust one drive's cached `storage.drives.used_bytes` + /// by `delta` bytes — same shape as + /// [`Self::add_user_storage_usage_delta`]: single statement, no + /// read-then-write window, `GREATEST(0, …)` clamp so a late or + /// duplicate adjustment can never drive the counter negative. + /// Deletes / trash do not decrement here; the periodic reconciliation + /// sweep ([`Self::update_all_drives_storage_usage`]) remains the + /// correctness backstop. + pub async fn add_drive_storage_usage_delta( + &self, + drive_id: Uuid, + delta: i64, + ) -> Result<(), DomainError> { + sqlx::query( + "UPDATE storage.drives + SET used_bytes = GREATEST(0, used_bytes + $2) + WHERE id = $1", + ) + .bind(drive_id) + .bind(delta) + .execute(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("StorageUsage", format!("drive delta: {e}")))?; + Ok(()) + } + + /// Same as [`Self::add_drive_storage_usage_delta`] but resolves + /// the drive id from a parent folder id in a single statement. + /// Avoids a separate `SELECT drive_id FROM storage.folders` round + /// trip at the upload hook site (where the folder id is what's + /// naturally on the FileDto). The nested SELECT is point-lookup + /// on the folder PK; clamp + idempotency properties are + /// unchanged. + pub async fn add_drive_storage_usage_delta_by_folder( + &self, + folder_id: Uuid, + delta: i64, + ) -> Result<(), DomainError> { + sqlx::query( + "UPDATE storage.drives + SET used_bytes = GREATEST(0, used_bytes + $2) + WHERE id = (SELECT drive_id FROM storage.folders WHERE id = $1)", + ) + .bind(folder_id) + .bind(delta) + .execute(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("StorageUsage", format!("drive delta by folder: {e}")) + })?; + Ok(()) + } + + /// Pre-upload quota check on a single drive. + /// + /// Read-only `SELECT (used_bytes, quota_bytes) FROM storage.drives`; + /// returns `QuotaExceeded` when the projected `used_bytes + + /// additional_bytes` would breach `quota_bytes`. A `NULL` + /// `quota_bytes` short-circuits to `Ok(())` (unlimited drive — + /// admin override / future system drives). + /// + /// Soft cap by design: the check/write window matches the + /// user-quota path, bounded by the sweep interval. The clamp on + /// `add_drive_storage_usage_delta` and the set-based reconciliation + /// keep the counter honest; small over-quota slippage during the + /// window is acceptable. + pub async fn check_drive_quota( + &self, + drive_id: Uuid, + additional_bytes: u64, + ) -> Result<(), DomainError> { + let row: Option<(i64, Option)> = sqlx::query_as( + "SELECT used_bytes, quota_bytes FROM storage.drives WHERE id = $1", + ) + .bind(drive_id) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("StorageUsage", format!("drive quota lookup: {e}")))?; + + let Some((used, quota)) = row else { + // Anti-enum at the upload edge would normally map to 404, + // but at this layer we surface the typed not-found and let + // the caller decide how to react. In practice the upload + // path resolves the drive id from a folder/file lookup + // first, so this branch fires only on a deleted-drive race. + return Err(DomainError::not_found("Drive", drive_id.to_string())); + }; + let Some(quota) = quota else { + return Ok(()); // unlimited + }; + // Saturate on the i64 + u64 sum so a hostile / corrupt counter + // can't silently overflow into a negative comparison. + let projected = (used as i128) + (additional_bytes as i128); + if projected > quota as i128 { + return Err(DomainError::new( + crate::common::errors::ErrorKind::QuotaExceeded, + "Drive", + format!( + "Drive quota exceeded: {} + {} > {} bytes", + used, additional_bytes, quota + ), + )); + } + Ok(()) + } + + /// Same as [`Self::check_drive_quota`] but resolves the drive id + /// from a parent folder id. Mirrors + /// [`Self::add_drive_storage_usage_delta_by_folder`] so the upload + /// handler (which holds `folder_id` from the multipart form) can + /// gate the write in one round trip. Returns + /// `DomainError::not_found("Folder", …)` if the folder id doesn't + /// resolve — the upload pipeline would 404 on that anyway. + pub async fn check_drive_quota_by_folder( + &self, + folder_id: Uuid, + additional_bytes: u64, + ) -> Result<(), DomainError> { + let row: Option<(i64, Option)> = sqlx::query_as( + "SELECT d.used_bytes, d.quota_bytes + FROM storage.drives d + JOIN storage.folders f ON f.drive_id = d.id + WHERE f.id = $1", + ) + .bind(folder_id) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("StorageUsage", format!("drive quota by folder: {e}")) + })?; + + let Some((used, quota)) = row else { + return Err(DomainError::not_found("Folder", folder_id.to_string())); + }; + let Some(quota) = quota else { + return Ok(()); // unlimited + }; + let projected = (used as i128) + (additional_bytes as i128); + if projected > quota as i128 { + return Err(DomainError::new( + crate::common::errors::ErrorKind::QuotaExceeded, + "Drive", + format!( + "Drive quota exceeded: {} + {} > {} bytes", + used, additional_bytes, quota + ), + )); + } + Ok(()) + } + /// Spawn a background task that periodically reconciles every user's cached /// `storage_used_bytes` against the actual sum of their files. /// @@ -153,7 +304,13 @@ impl StorageUsageService { ticker.tick().await; debug!("Running scheduled storage-usage reconciliation"); if let Err(e) = service.update_all_users_storage_usage().await { - error!("Scheduled storage-usage reconciliation failed: {}", e); + error!("Scheduled user storage-usage reconciliation failed: {}", e); + } + // Drive sweep runs alongside the user sweep — same + // cadence, same maintenance pool. Failure is logged + // but doesn't skip the next tick. + if let Err(e) = service.update_all_drives_storage_usage().await { + error!("Scheduled drive storage-usage reconciliation failed: {}", e); } } }); @@ -266,6 +423,63 @@ impl StorageUsagePort for StorageUsageService { let user = self.user_repository.get_user_by_id(user_id).await?; Ok((user.storage_used_bytes(), user.storage_quota_bytes())) } + + async fn add_drive_storage_usage_delta( + &self, + drive_id: Uuid, + delta: i64, + ) -> Result<(), DomainError> { + StorageUsageService::add_drive_storage_usage_delta(self, drive_id, delta).await + } + + /// Reconcile every drive's cached `used_bytes` in ONE set-based UPDATE. + /// + /// Same shape as the per-user sweep above: `LEFT JOIN` over the + /// `storage.files` aggregate keyed on `drive_id`, `IS DISTINCT + /// FROM` guard to skip no-op rewrites so idle drives don't churn + /// dead tuples. Runs from the same reconciliation ticker as the + /// user sweep; failure is logged but doesn't stop the next tick. + async fn update_all_drives_storage_usage(&self) -> Result<(), DomainError> { + debug!("Starting drive storage-usage reconciliation sweep"); + let result = sqlx::query( + r#" + UPDATE storage.drives d + SET used_bytes = COALESCE(t.total, 0) + FROM storage.drives d2 + LEFT JOIN ( + SELECT drive_id, SUM(size)::bigint AS total + FROM storage.files + WHERE NOT is_trashed + GROUP BY drive_id + ) t ON t.drive_id = d2.id + WHERE d.id = d2.id + AND d.used_bytes IS DISTINCT FROM COALESCE(t.total, 0) + "#, + ) + .execute(self.pool.as_ref()) + .await + .map_err(|e| { + error!("Drive storage-usage reconciliation sweep failed: {}", e); + DomainError::internal_error( + "StorageUsage", + format!("drive reconciliation sweep: {e}"), + ) + })?; + + info!( + "Drive storage-usage reconciliation corrected {} drive(s)", + result.rows_affected() + ); + Ok(()) + } + + async fn check_drive_quota( + &self, + drive_id: Uuid, + additional_bytes: u64, + ) -> Result<(), DomainError> { + StorageUsageService::check_drive_quota(self, drive_id, additional_bytes).await + } } // Make StorageUsageService cloneable to support spawning concurrent tasks diff --git a/src/common/config.rs b/src/common/config.rs index 7549927f..50869c3b 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -906,6 +906,15 @@ pub struct FeaturesConfig { /// thumbnail through the same WebP pipeline as photos; otherwise videos have /// no thumbnail. Env: `OXICLOUD_ENABLE_VIDEO_THUMBNAILS`. pub enable_video_thumbnails: bool, + /// Expose `/api/admin/internal/*` test-only endpoints that trigger + /// background sweeps on demand (storage-usage reconciliation, blob + /// GC). Intended for Hurl / integration tests that need to wait + /// for these maintenance jobs deterministically rather than + /// polling the cached value. Off by default — these endpoints + /// short-circuit the operator-visible cadence, so production + /// deployments don't want them reachable. Env: + /// `OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS`. + pub enable_admin_internal_endpoints: bool, } impl Default for FeaturesConfig { @@ -921,6 +930,10 @@ impl Default for FeaturesConfig { enable_faces: false, // People/faces (biometric) — opt-in, off by default expose_system_users: true, // Expose OxiCloud users as address book by default enable_video_thumbnails: true, // Video thumbs via ffmpeg (if detected) + // Test-only sweep triggers — strictly opt-in. Production + // deployments do NOT need this; the periodic ticker handles + // reconciliation transparently. + enable_admin_internal_endpoints: false, } } } @@ -1482,6 +1495,16 @@ impl AppConfig { config.features.enable_video_thumbnails = val; } + // `/api/admin/internal/*` test-only triggers. Disabled by + // default; production deployments never need this. The Hurl + // suite flips it on via `OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS=true`. + if let Ok(enable_internal) = + env::var("OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS").map(|v| v.parse::()) + && let Ok(val) = enable_internal + { + config.features.enable_admin_internal_endpoints = val; + } + if let Ok(enable_faces) = env::var("OXICLOUD_ENABLE_FACES").map(|v| v.parse::()) && let Ok(val) = enable_faces { diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index bbe1cc0e..250436b5 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -95,6 +95,13 @@ pub fn admin_routes() -> Router> { // when `OXICLOUD_SMTP_MOCK` is off, so production deployments // can route the path freely without leaking inboxes. .route("/smtp/test/captured", get(get_captured_email)) + // Test-only sweep triggers. Routes are always registered; the + // handlers themselves short-circuit to 404 when + // `features.enable_admin_internal_endpoints` is off — matches + // the `/smtp/test/captured` convention so production + // deployments don't need a different route table. + .route("/internal/trigger-sweep", post(internal_trigger_sweep)) + .route("/internal/trigger-gc", post(internal_trigger_gc)) // Drives — admin-wide view (distinct from `/api/drives` which // is filtered to the caller's role grants). .route("/drives", get(list_all_drives)) diff --git a/src/interfaces/api/handlers/chunked_upload_handler.rs b/src/interfaces/api/handlers/chunked_upload_handler.rs index 489c2e49..ea339cc1 100644 --- a/src/interfaces/api/handlers/chunked_upload_handler.rs +++ b/src/interfaces/api/handlers/chunked_upload_handler.rs @@ -214,7 +214,7 @@ impl ChunkedUploadHandler { .await { tracing::warn!( - "⛔ CHUNKED UPLOAD REJECTED (quota): user={}, file={}, size={} — {}", + "⛔ CHUNKED UPLOAD REJECTED (user quota): user={}, file={}, size={} — {}", auth_user.username, request.filename, request.total_size, @@ -230,6 +230,37 @@ impl ChunkedUploadHandler { .into_response(); } + // ── Per-drive quota (D4) ───────────────────────────────── + // Native-chunked declares `total_size` at session creation, + // so we can refuse here before any chunk is accepted — same + // wasted-bandwidth optimisation the multipart path has via + // the post-ingest check. No folder_id means root-level which + // the folder-permission check above already rejects. + if let Some(storage_svc) = state.storage_usage_service.as_ref() + && let Some(fid_str) = request.folder_id.as_deref() + && let Ok(fid) = uuid::Uuid::parse_str(fid_str) + && let Err(err) = storage_svc + .check_drive_quota_by_folder(fid, request.total_size) + .await + { + tracing::warn!( + "⛔ CHUNKED UPLOAD REJECTED (drive quota): user={}, folder={}, file={}, size={} — {}", + auth_user.username, + fid, + request.filename, + request.total_size, + err.message + ); + return ( + StatusCode::INSUFFICIENT_STORAGE, + Json(serde_json::json!({ + "error": err.message, + "error_type": "QuotaExceeded" + })), + ) + .into_response(); + } + // Validate chunk size if provided let chunk_size = request.chunk_size.unwrap_or(DEFAULT_CHUNK_SIZE); if chunk_size < 1024 * 1024 { diff --git a/src/interfaces/api/handlers/file_handler.rs b/src/interfaces/api/handlers/file_handler.rs index 00e6312d..7347f415 100644 --- a/src/interfaces/api/handlers/file_handler.rs +++ b/src/interfaces/api/handlers/file_handler.rs @@ -267,7 +267,7 @@ impl FileHandler { { upload_ingest::discard_ingested(dedup, &ingested).await; tracing::warn!( - "⛔ UPLOAD REJECTED (quota): user={}, file={}, size={}", + "⛔ UPLOAD REJECTED (user quota): user={}, file={}, size={}", auth_user.username, filename, ingested.size @@ -275,6 +275,31 @@ impl FileHandler { return Err(Self::quota_error_response(err)); } + // ── Per-drive quota enforcement (D4) ───────────────── + // Sibling to the per-user check above: same read-only + // SELECT shape, same discard-then-507 outcome. Skipped + // when there's no folder_id (root-level upload — no + // drive to charge; folder service refuses these + // independently). Unlimited-quota drives (`NULL`) + // short-circuit inside the service. + if let Some(storage_svc) = state.storage_usage_service.as_ref() + && let Some(fid_str) = folder_id.as_deref() + && let Ok(fid) = uuid::Uuid::parse_str(fid_str) + && let Err(err) = storage_svc + .check_drive_quota_by_folder(fid, ingested.size) + .await + { + upload_ingest::discard_ingested(dedup, &ingested).await; + tracing::warn!( + "⛔ UPLOAD REJECTED (drive quota): user={}, folder={}, file={}, size={}", + auth_user.username, + fid, + filename, + ingested.size + ); + return Err(Self::quota_error_response(err)); + } + // ── Register the file row against the ingested blob ── let hash = ingested.hash.clone(); let size = ingested.size;