feat(drive): add /api/drive

- permit shared drive creation from oxicloud admin (for now)
    - prepare other personal drive creation (Not implemented), need to validate
    quota policies and strategy first
    - add hurl test to verify permissions

src/infrastructure/repositories/pg/drive_pg_repository.rs

@@ -196,6 +196,113 @@ impl DriveRepository for DrivePgRepository {
         Self::row_to_drive_with_name(&row)
     }
 
+    async fn create_shared_drive_atomic(
+        &self,
+        name: &str,
+        owner_subject: crate::domain::services::authorization::Subject,
+        quota_bytes: Option<i64>,
+        granted_by: Uuid,
+    ) -> Result<DriveWithRootName, DriveRepositoryError> {
+        // Same four-write transaction shape as `create_personal_drive_atomic`
+        // (see that method for the why-not-CTE explanation). Differences:
+        //   - `kind='shared'`, `default_for_user=NULL`.
+        //   - Root folder name is caller-supplied.
+        //   - Owner grant subject is caller-supplied — either a single
+        //     User (becomes the sole drive Owner) or a Group (transitive
+        //     members inherit Owner via subject expansion).
+        //   - `granted_by` is the OxiCloud admin who provisioned the drive;
+        //     same value goes onto the folder's `created_by`/`updated_by`
+        //     for §14 provenance.
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .map_err(|e| Self::map_sqlx_err("create_shared_drive_atomic.begin", e))?;
+
+        // 1. Drive row (root_folder_id NULL — populated in step 3).
+        let drive_id: Uuid = sqlx::query_scalar(
+            r#"
+            INSERT INTO storage.drives
+                (kind, default_for_user, quota_bytes, policies)
+            VALUES ('shared', NULL, $1, '{}'::jsonb)
+            RETURNING id
+            "#,
+        )
+        .bind(quota_bytes)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e| Self::map_sqlx_err("create_shared_drive_atomic.drive", e))?;
+
+        // 2. Root folder. The folder's `user_id` carries the admin (legacy
+        //    column still NOT NULL during the dual-write window — D7
+        //    drops it once `drive_id` is the canonical ownership signal).
+        let folder_id: Uuid = sqlx::query_scalar(
+            r#"
+            INSERT INTO storage.folders
+                (name, parent_id, user_id, drive_id, created_by, updated_by)
+            VALUES ($1, NULL, $2, $3, $2, $2)
+            RETURNING id
+            "#,
+        )
+        .bind(name)
+        .bind(granted_by)
+        .bind(drive_id)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e| Self::map_sqlx_err("create_shared_drive_atomic.folder", e))?;
+
+        // 3. Close the circular reference (drive ↔ root folder).
+        sqlx::query(r#"UPDATE storage.drives SET root_folder_id = $1 WHERE id = $2"#)
+            .bind(folder_id)
+            .bind(drive_id)
+            .execute(&mut *tx)
+            .await
+            .map_err(|e| Self::map_sqlx_err("create_shared_drive_atomic.wire", e))?;
+
+        // 4. Owner role_grant — subject_type chosen from the caller's input.
+        //    Group subjects expand transitively via `subject_match_set` so
+        //    every member inherits Owner; User subjects are the single
+        //    admin case.
+        sqlx::query(
+            r#"
+            INSERT INTO storage.role_grants
+                (subject_type, subject_id, resource_type, resource_id,
+                 role, granted_by)
+            VALUES ($1, $2, 'drive', $3, 'owner', $4)
+            "#,
+        )
+        .bind(owner_subject.type_str())
+        .bind(owner_subject.id())
+        .bind(drive_id)
+        .bind(granted_by)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| Self::map_sqlx_err("create_shared_drive_atomic.grant", e))?;
+
+        // Fetch final state so the caller sees DB-computed defaults.
+        let row = sqlx::query(
+            r#"
+            SELECT d.id, d.kind, d.default_for_user, d.root_folder_id,
+                   d.quota_bytes, d.used_bytes, d.policies,
+                   d.created_at, d.updated_at,
+                   f.name AS root_folder_name
+              FROM storage.drives d
+              JOIN storage.folders f ON f.id = d.root_folder_id
+             WHERE d.id = $1
+            "#,
+        )
+        .bind(drive_id)
+        .fetch_one(&mut *tx)
+        .await
+        .map_err(|e| Self::map_sqlx_err("create_shared_drive_atomic.read", e))?;
+
+        tx.commit()
+            .await
+            .map_err(|e| Self::map_sqlx_err("create_shared_drive_atomic.commit", e))?;
+
+        Self::row_to_drive_with_name(&row)
+    }
+
     async fn get_by_id(&self, id: Uuid) -> Result<DriveWithRootName, DriveRepositoryError> {
         let row = sqlx::query(
             r#"

src/infrastructure/services/pg_acl_engine.rs

@@ -155,6 +155,13 @@ impl PgAclEngine {
                 .time_to_live(OWNER_CACHE_TTL)
                 .build(),
             drive_role_cache: Cache::builder()
+                // `invalidate_entries_if` is the cleanup hook used by
+                // `invalidate_drive_role_cache_for_drive`. moka returns
+                // `Err(InvalidationClosuresDisabled)` from that call unless
+                // this opt-in is set on the builder, so without it the
+                // bulk invalidation silently no-ops and a freshly-promoted
+                // member keeps their stale role for the full TTL.
+                .support_invalidation_closures()
                 .max_capacity(DRIVE_ROLE_CACHE_CAPACITY)
                 .time_to_live(DRIVE_ROLE_CACHE_TTL)
                 .build(),
@@ -214,6 +221,7 @@ impl PgAclEngine {
                 .time_to_live(Duration::from_secs(1))
                 .build(),
             drive_role_cache: Cache::builder()
+                .support_invalidation_closures()
                 .max_capacity(1)
                 .time_to_live(Duration::from_secs(1))
                 .build(),
@@ -238,14 +246,35 @@ impl PgAclEngine {
     ///
     /// Uses moka's predicate-based eviction — entries are marked for
     /// removal asynchronously by the maintenance task; subsequent `get`
-    /// calls observe the eviction immediately.
+    /// calls observe the eviction. Requires
+    /// `support_invalidation_closures()` on the cache builder (see the
+    /// `drive_role_cache` initialiser above), otherwise moka returns
+    /// `InvalidationClosuresDisabled` and the mutation silently leaves
+    /// stale role rows in cache for the full TTL.
     pub async fn invalidate_drive_role_cache_for_drive(&self, drive_id: Uuid) {
         // `invalidate_entries_if` rejects predicates returning errors —
         // simple Fn(K, V) -> bool. We capture `drive_id` by value (Copy)
         // and match against the second tuple component.
-        let _ = self
+        //
+        // The result is `Err` only when the cache was built without
+        // `support_invalidation_closures()` — a wiring bug, not a runtime
+        // condition the caller can recover from. We log+continue rather
+        // than panic because the consequence is a 30 s staleness window
+        // on cached role entries, not a correctness bug at write time.
+        if let Err(err) = self
             .drive_role_cache
-            .invalidate_entries_if(move |key, _v| key.1 == drive_id);
+            .invalidate_entries_if(move |key, _v| key.1 == drive_id)
+        {
+            tracing::error!(
+                target: "oxicloud::authz",
+                event = "authz.cache_invalidation_failed",
+                cache = "drive_role_cache",
+                drive_id = %drive_id,
+                error = %err,
+                "drive_role_cache cannot be bulk-invalidated — \
+                 cache builder is missing support_invalidation_closures()",
+            );
+        }
     }
 
     /// Expand a user subject into the set of subject UUIDs that should match