fix(quota): pre-check quota for COPY/MOVE

check qouta for a cross drive MOVE
    check quota for a COPY

src/application/services/file_management_service.rs

@@ -43,6 +43,13 @@ pub struct FileManagementService {
     /// that case the cross-drive move check is skipped (the policy
     /// is silently off). Production DI wires it in.
     drive_repo: Option<Arc<dyn crate::domain::repositories::drive_repository::DriveRepository>>,
+    /// Storage-usage service — used to pre-check the destination
+    /// drive's `used_bytes + delta ≤ quota_bytes` invariant on
+    /// cross-drive MOVE, matching the pre-write check the upload path
+    /// already performs. Without it, the check is silently skipped
+    /// (stub/test builders); production DI wires it in.
+    storage_usage:
+        Option<Arc<crate::application::services::storage_usage_service::StorageUsageService>>,
 }
 
 impl FileManagementService {
@@ -67,6 +74,7 @@ impl FileManagementService {
             file_lifecycle_hook: None,
             resource_access_hook: None,
             drive_repo: None,
+            storage_usage: None,
         }
     }
 
@@ -100,6 +108,18 @@ impl FileManagementService {
         self
     }
 
+    /// Wires the storage-usage service so `move_file_with_perms` can
+    /// pre-check the destination drive's quota on cross-drive moves.
+    pub fn with_storage_usage(
+        mut self,
+        storage_usage: Arc<
+            crate::application::services::storage_usage_service::StorageUsageService,
+        >,
+    ) -> Self {
+        self.storage_usage = Some(storage_usage);
+        self
+    }
+
     /// Engine check for a file resource. Parses the id into a `Uuid` and
     /// requires the specified permission.
     async fn require_file_perm(
@@ -338,6 +358,22 @@ impl FileManagementUseCase for FileManagementService {
                         dst_drive_id,
                     },
                 )?;
+                // Destination drive quota: same pre-write check the
+                // upload path already runs (`file_upload_service.rs`
+                // `check_storage_quota`), applied here so a caller
+                // can't sneak content past the drive cap via MOVE.
+                // Denial → `DomainError::QuotaExceeded` → 507
+                // Insufficient Storage. Skipped when `storage_usage`
+                // isn't wired (stub builders) — same shape as the
+                // upload path's skip semantics.
+                if let Some(storage_usage) = &self.storage_usage
+                    && let Some(size_bytes) = storage_usage.file_bytes(file_uuid).await?
+                    && let Ok(size_u64) = u64::try_from(size_bytes)
+                {
+                    storage_usage
+                        .check_drive_quota(dst_drive_id, size_u64)
+                        .await?;
+                }
                 cross_drive = Some((src_drive_id, dst_drive_id));
             }
         }
@@ -375,6 +411,31 @@ impl FileManagementUseCase for FileManagementService {
             .await?;
         self.require_target_folder_perm(target_folder_id.as_deref(), Permission::Create, caller_id)
             .await?;
+
+        // Destination drive quota: COPY creates a new file row that
+        // counts against the destination drive's `used_bytes` even
+        // though blob dedup means no new bytes hit the store. Same
+        // pre-flight shape the delta-upload path already uses.
+        // Skipped when `storage_usage` isn't wired (stub builders) or
+        // `target_folder_id` is None (root namespace — same-drive
+        // semantics inherit the source's cap coverage). Denial →
+        // `QuotaExceeded` → 507.
+        if let (Some(storage_usage), Some(target_folder)) =
+            (&self.storage_usage, target_folder_id.as_deref())
+        {
+            let file_uuid =
+                Uuid::parse_str(file_id).map_err(|_| DomainError::not_found("File", file_id))?;
+            let target_folder_uuid = Uuid::parse_str(target_folder)
+                .map_err(|_| DomainError::not_found("Folder", target_folder))?;
+            if let Some(size_bytes) = storage_usage.file_bytes(file_uuid).await?
+                && let Ok(size_u64) = u64::try_from(size_bytes)
+            {
+                storage_usage
+                    .check_drive_quota_by_folder(target_folder_uuid, size_u64)
+                    .await?;
+            }
+        }
+
         self.copy_file(file_id, target_folder_id, new_name.as_deref(), caller_id)
             .await
     }
@@ -453,6 +514,26 @@ impl FileManagementUseCase for FileManagementService {
             .await?;
         self.require_target_folder_perm(target_parent_id.as_deref(), Permission::Create, caller_id)
             .await?;
+
+        // Destination drive quota: sum the subtree's non-trashed files
+        // and refuse if the destination couldn't hold them. Skipped
+        // when `storage_usage` isn't wired or the target is root
+        // (same rationale as `copy_file_with_perms`).
+        if let (Some(storage_usage), Some(target_parent)) =
+            (&self.storage_usage, target_parent_id.as_deref())
+        {
+            let source_uuid = Uuid::parse_str(source_folder_id)
+                .map_err(|_| DomainError::not_found("Folder", source_folder_id))?;
+            let target_parent_uuid = Uuid::parse_str(target_parent)
+                .map_err(|_| DomainError::not_found("Folder", target_parent))?;
+            let subtree_bytes = storage_usage.folder_subtree_bytes(source_uuid).await?;
+            if let Ok(subtree_u64) = u64::try_from(subtree_bytes) {
+                storage_usage
+                    .check_drive_quota_by_folder(target_parent_uuid, subtree_u64)
+                    .await?;
+            }
+        }
+
         self.copy_folder_tree(source_folder_id, target_parent_id, dest_name)
             .await
     }

src/application/services/folder_service.rs

@@ -31,6 +31,11 @@ pub struct FolderService {
     /// that case the cross-drive move check is skipped (the policy is
     /// silently off). Production DI wires it via `with_drive_repo`.
     drive_repo: Option<Arc<dyn crate::domain::repositories::drive_repository::DriveRepository>>,
+    /// Storage-usage service — used to pre-check the destination
+    /// drive's `used_bytes + subtree_bytes ≤ quota_bytes` invariant
+    /// on cross-drive MOVE. Silently skipped when unwired (stubs).
+    storage_usage:
+        Option<Arc<crate::application::services::storage_usage_service::StorageUsageService>>,
 }
 
 impl FolderService {
@@ -45,6 +50,7 @@ impl FolderService {
             authz,
             file_lifecycle,
             drive_repo: None,
+            storage_usage: None,
         }
     }
 
@@ -60,6 +66,19 @@ impl FolderService {
         self
     }
 
+    /// Wires the storage-usage service so `move_folder_with_perms`
+    /// can pre-check the destination drive's quota on cross-drive
+    /// folder moves.
+    pub fn with_storage_usage(
+        mut self,
+        storage_usage: Arc<
+            crate::application::services::storage_usage_service::StorageUsageService,
+        >,
+    ) -> Self {
+        self.storage_usage = Some(storage_usage);
+        self
+    }
+
     /// Batch counterpart of `get_folder`: resolve many folder ids in ONE
     /// query instead of one per id. Like `get_folder` it performs no
     /// per-folder authorization — both current callers (ACL grant listing,
@@ -593,6 +612,19 @@ impl FolderUseCase for FolderService {
                         dst_drive_id,
                     },
                 )?;
+                // Destination drive quota: sum the moved subtree's
+                // non-trashed files and refuse if the destination
+                // couldn't hold them. Same 507 shape as the file
+                // path + upload path — DomainError::QuotaExceeded
+                // maps at the AppError boundary.
+                if let Some(storage_usage) = &self.storage_usage {
+                    let subtree_bytes = storage_usage.folder_subtree_bytes(src_folder_uuid).await?;
+                    if let Ok(subtree_u64) = u64::try_from(subtree_bytes) {
+                        storage_usage
+                            .check_drive_quota(dst_drive_id, subtree_u64)
+                            .await?;
+                    }
+                }
                 cross_drive = Some((src_drive_id, dst_drive_id));
             }
         }

src/application/services/storage_usage_service.rs

@@ -213,6 +213,47 @@ impl StorageUsageService {
         Ok(())
     }
 
+    /// Return the size in bytes of a single non-trashed file. `None`
+    /// if the file is trashed or absent. Used by cross-drive MOVE to
+    /// know how many bytes will land on the destination drive so the
+    /// pre-move `check_drive_quota` call can fire.
+    pub async fn file_bytes(&self, file_id: Uuid) -> Result<Option<i64>, DomainError> {
+        let row: Option<(i64,)> = sqlx::query_as(
+            "SELECT size::bigint FROM storage.files WHERE id = $1 AND NOT is_trashed",
+        )
+        .bind(file_id)
+        .fetch_optional(self.pool.as_ref())
+        .await
+        .map_err(|e| DomainError::internal_error("StorageUsage", format!("file_bytes: {e}")))?;
+        Ok(row.map(|(s,)| s))
+    }
+
+    /// Sum the sizes of every non-trashed file whose parent folder is
+    /// `folder_id` itself or a descendant of it via the `lpath` ltree.
+    /// Used by cross-drive MOVE to know how many bytes would land on
+    /// the destination drive — necessary for the pre-move
+    /// `check_drive_quota` call.
+    ///
+    /// Returns 0 for an empty subtree AND for a non-existent
+    /// `folder_id` (the JOIN silently drops); callers that need to
+    /// distinguish those two cases must probe the folder separately.
+    pub async fn folder_subtree_bytes(&self, folder_id: Uuid) -> Result<i64, DomainError> {
+        let (bytes,): (Option<i64>,) = sqlx::query_as(
+            "SELECT COALESCE(SUM(f.size), 0)::bigint
+               FROM storage.files f
+               JOIN storage.folders fo ON fo.id = f.folder_id
+              WHERE fo.lpath <@ (SELECT lpath FROM storage.folders WHERE id = $1)
+                AND NOT f.is_trashed",
+        )
+        .bind(folder_id)
+        .fetch_one(self.pool.as_ref())
+        .await
+        .map_err(|e| {
+            DomainError::internal_error("StorageUsage", format!("folder_subtree_bytes: {e}"))
+        })?;
+        Ok(bytes.unwrap_or(0))
+    }
+
     /// 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