feat(roles): prepare migration ReBAC to roles

prepare migration of permission to roles
    this simplify drastically database (permission are now simply defined in code)
    and will permit reuse of the same ReBAC engine to define owners of drives

    mapping:

    ```
        Role::Viewer => &[Permission::Read],
        Role::Commenter => &[Permission::Read, Permission::Comment],
        Role::Contributor => &[Permission::Read, Permission::Create],
        Role::Editor => &[
            Permission::Read,
            Permission::Comment,
            Permission::Create,
            Permission::Update,
        ],
        Role::Owner => &[
            Permission::Read,
            Permission::Comment,
            Permission::Create,
            Permission::Update,
            Permission::Share,
            Permission::Delete,
            Permission::Manage,
        ],
    ```

src/application/dtos/grant_dto.rs

@@ -95,6 +95,7 @@ pub enum PermissionDto {
     Comment,
     Delete,
     Update,
+    Manage,
 }
 
 impl From<PermissionDto> for Permission {
@@ -106,6 +107,7 @@ impl From<PermissionDto> for Permission {
             PermissionDto::Comment => Permission::Comment,
             PermissionDto::Delete => Permission::Delete,
             PermissionDto::Update => Permission::Update,
+            PermissionDto::Manage => Permission::Manage,
         }
     }
 }
@@ -119,59 +121,176 @@ impl From<Permission> for PermissionDto {
             Permission::Comment => PermissionDto::Comment,
             Permission::Delete => PermissionDto::Delete,
             Permission::Update => PermissionDto::Update,
+            Permission::Manage => PermissionDto::Manage,
         }
     }
 }
 
 // ════════════════════════════════════════════════════════════════════════════
-// Roles (DTO-layer sugar)
+// Roles — the load-bearing model for ReBAC grants
 // ════════════════════════════════════════════════════════════════════════════
+//
+// Today a role is "DTO-layer sugar" — every grant write expands a role into
+// N rows in `storage.access_grants`. The D-Prep refactor (see
+// `docs/plan/drive.md` §Prerequisite + migration `20260730000000_role_grants.sql`)
+// pushes the role down to storage (`storage.role_grants.role TEXT`); the
+// engine reads the role and expands the bundle at query time via this same
+// `expand()` function. Adding a role is now schema-free — one variant + one
+// match arm here.
 
-#[derive(Debug, Clone, Copy, Serialize, Deserialize, ToSchema)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, ToSchema)]
 #[serde(rename_all = "lowercase")]
 pub enum Role {
+    /// Read access only. The default "anyone can look but not touch".
     Viewer,
-    //Commenter,
+    /// Read + comment. Useful for review-only stakeholders.
+    Commenter,
+    /// Read + create. The "drop-zone" role — uploads allowed, existing
+    /// content untouchable. Common for support-ticket attachments and
+    /// photo-submission folders.
+    Contributor,
+    /// Read + create + update + comment. The standard collaboration role.
     Editor,
-    //Manager,
-    Admin,
+    /// Full bundle: read, create, update, comment, delete, share — and
+    /// `Manage` for resource types that support it (drives, groups). This
+    /// is the highest user-grantable role; renamed from the historical
+    /// `Admin` to disambiguate from `UserRole::Admin` (the user-account
+    /// privilege) and to match Drive plan terminology.
+    ///
+    /// **Wire-format compat shim** (one release): also deserialises from
+    /// the legacy `"admin"` string so cached frontend clients keep
+    /// working until they refresh. Serialisation always emits `"owner"`.
+    /// Drop the alias in the cleanup PR.
+    #[serde(alias = "admin")]
+    Owner,
 }
 
 impl Role {
-    /// Expands a role into its constituent raw permissions. Storage and
-    /// engine know nothing about roles — the server normalizes here before
-    /// writing rows.
+    /// Expand the role into its permission bundle. Single source of truth —
+    /// any code that needs "does this role include Permission X?" routes
+    /// through here (or its inverse, `roles_implying`).
+    ///
+    /// After D-Prep this is called at engine read time (1 row → bundle
+    /// expanded server-side). Pre-D-Prep it was called at API write time
+    /// (1 role → N rows fanned out).
     pub fn expand(self) -> &'static [Permission] {
         match self {
             Role::Viewer => &[Permission::Read],
-            /* reserved for future
             Role::Commenter => &[Permission::Read, Permission::Comment],
-            */
+            Role::Contributor => &[Permission::Read, Permission::Create],
             Role::Editor => &[
                 Permission::Read,
                 Permission::Comment,
                 Permission::Create,
                 Permission::Update,
             ],
-            /* reserved for future
-            Role::Manager => &[
-                Permission::Read,
-                Permission::Comment,
-                Permission::Create,
-                Permission::Update,
-                Permission::Share,
-            ],
-            */
-            Role::Admin => &[
+            Role::Owner => &[
                 Permission::Read,
                 Permission::Comment,
                 Permission::Create,
                 Permission::Update,
                 Permission::Share,
                 Permission::Delete,
+                Permission::Manage,
             ],
         }
     }
+
+    /// Lowercase string discriminator — matches the SQL `role` column values
+    /// in `storage.role_grants` and the JSON wire format.
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Role::Viewer => "viewer",
+            Role::Commenter => "commenter",
+            Role::Contributor => "contributor",
+            Role::Editor => "editor",
+            Role::Owner => "owner",
+        }
+    }
+
+    /// Parse a role from its SQL / JSON string discriminator. Returns
+    /// `None` for unknown values. Accepts the legacy `"admin"` spelling
+    /// for one release of API compat (clients that cached the old name
+    /// keep working; new responses always emit `"owner"`).
+    pub fn parse(s: &str) -> Option<Self> {
+        match s {
+            "viewer" => Some(Role::Viewer),
+            "commenter" => Some(Role::Commenter),
+            "contributor" => Some(Role::Contributor),
+            "editor" => Some(Role::Editor),
+            "owner" => Some(Role::Owner),
+            // Legacy compat: drop after one release once all clients are
+            // updated. Emits a debug log so we can track stragglers.
+            "admin" => {
+                tracing::debug!(
+                    target: "oxicloud::grants",
+                    "Role::parse: accepted legacy 'admin' string as Role::Owner"
+                );
+                Some(Role::Owner)
+            }
+            _ => None,
+        }
+    }
+
+    /// Every role, in a stable order. Used by `roles_implying` and exposed
+    /// to the UI so the share modal can render the full picker without
+    /// hardcoding the list.
+    ///
+    /// **UI scope today**: the share dialog renders only `Viewer`,
+    /// `Editor`, and `Owner` (matches the existing 3-button UX). The
+    /// `Commenter` and `Contributor` variants are implemented server-
+    /// side and accepted on the API surface, reserved for future UI
+    /// exposure when a real use case asks for them. Until then they
+    /// stay invisible to end users — no picker option, no documentation
+    /// surface.
+    ///
+    /// Any role can be granted on any resource type. Permission bundles
+    /// that include capabilities the resource type doesn't check for
+    /// (e.g. `Manage` on a folder, `Create` on a file) simply produce
+    /// harmless no-ops — no separate validation layer is needed.
+    pub const ALL: [Role; 5] = [
+        Role::Viewer,
+        Role::Commenter,
+        Role::Contributor,
+        Role::Editor,
+        Role::Owner,
+    ];
+}
+
+/// Inverse of [`Role::expand`]: returns every role whose bundle contains
+/// the given permission. Used by the engine to build the SQL
+/// `WHERE role IN (...)` filter on hot-path queries like "what drives can
+/// this caller read?":
+///
+/// ```ignore
+/// SELECT resource_id FROM role_grants
+/// WHERE subject_id = $1
+///   AND resource_type = 'drive'
+///   AND role IN (roles_implying(Permission::Read));
+/// ```
+///
+/// Precomputed in code rather than stored in the DB — `Role` and
+/// `Permission` are both small fixed enums, the table can never grow
+/// beyond a handful of rows, and keeping it in-code makes "what changes
+/// when I add a Permission?" a single grep target.
+pub fn roles_implying(permission: Permission) -> &'static [Role] {
+    use Permission::*;
+    match permission {
+        // Every role grants Read — viewer is the floor.
+        Read => &[
+            Role::Viewer,
+            Role::Commenter,
+            Role::Contributor,
+            Role::Editor,
+            Role::Owner,
+        ],
+        Comment => &[Role::Commenter, Role::Editor, Role::Owner],
+        Create => &[Role::Contributor, Role::Editor, Role::Owner],
+        Update => &[Role::Editor, Role::Owner],
+        Delete => &[Role::Owner],
+        Share => &[Role::Owner],
+        Manage => &[Role::Owner],
+    }
 }
 
 // ════════════════════════════════════════════════════════════════════════════
@@ -430,12 +549,34 @@ pub struct SharedWithMeItemDto {
 }
 
 /// Derive the closest-matching role label from a set of permissions.
-/// Maps the permission set to `"admin"`, `"editor"`, or `"viewer"`.
+///
+/// **Legacy helper for the dual-write window.** Once D-Prep ships and the
+/// engine reads `role_grants.role` directly, this function becomes unused
+/// and is dropped in the cleanup PR. Kept here so callers that still hit
+/// `access_grants` and reconstruct a role for display can stay working
+/// during the transition.
+///
+/// Emits the new five-role roster on output (`"viewer"` / `"commenter"` /
+/// `"contributor"` / `"editor"` / `"owner"`). Note this is **lossy** for
+/// permission sets that don't match a bundle exactly — but D-Prep's
+/// pre-flight refuses to migrate any such cluster, so post-migration data
+/// only contains bundle-shaped sets.
 pub fn role_from_permissions(perms: &[Permission]) -> &'static str {
-    if perms.contains(&Permission::Delete) && perms.contains(&Permission::Share) {
-        "admin"
-    } else if perms.contains(&Permission::Create) || perms.contains(&Permission::Update) {
+    let has_read = perms.contains(&Permission::Read);
+    let has_comment = perms.contains(&Permission::Comment);
+    let has_create = perms.contains(&Permission::Create);
+    let has_update = perms.contains(&Permission::Update);
+    let has_delete = perms.contains(&Permission::Delete);
+    let has_share = perms.contains(&Permission::Share);
+
+    if has_delete && has_share {
+        "owner"
+    } else if has_create && has_update {
         "editor"
+    } else if has_read && has_create && !has_update {
+        "contributor"
+    } else if has_read && has_comment && !has_create && !has_update {
+        "commenter"
     } else {
         "viewer"
     }
@@ -457,7 +598,12 @@ pub struct OutgoingResourceGrantDto {
     pub subject_id: Uuid,
     /// Human-readable label (username for users, share name for tokens).
     pub subject_display: String,
-    /// Derived role label: `"viewer"` | `"editor"` | `"admin"`.
+    /// Role label: `"viewer"` | `"commenter"` | `"contributor"` | `"editor"`
+    /// | `"owner"`. Emitted by `role_from_permissions()` during the dual-write
+    /// window; once D-Prep cleanup lands this is read directly from
+    /// `storage.role_grants.role`. The legacy `"admin"` spelling is no longer
+    /// emitted — clients that cached it must accept `"owner"` too (the API
+    /// `Role::parse` still accepts `"admin"` on input for one release).
     pub role: String,
     pub granted_at: chrono::DateTime<chrono::Utc>,
     #[serde(skip_serializing_if = "Option::is_none")]

src/application/ports/authorization_ports.rs

@@ -181,4 +181,40 @@ pub trait AuthorizationEngine: Send + Sync + 'static {
     /// Removes every grant whose `subject` matches. Called when a user/token
     /// /group is deleted. Returns the count of rows removed.
     async fn revoke_all_for_subject(&self, subject: Subject) -> Result<usize, DomainError>;
+
+    // ── Role-keyed grant operations (D-Prep dual-write) ────────────────────
+    // These manage `storage.role_grants`, the role-keyed table introduced
+    // by the D-Prep refactor (see `docs/plan/drive.md` §Prerequisite).
+    //
+    // During the dual-write window both tables stay populated; the engine
+    // reads from `access_grants` until the read-path pivot lands. After
+    // the cleanup PR (which drops `access_grants`), these two methods
+    // become the ONLY grant write path — the per-permission `grant` /
+    // `revoke` above are removed at that point.
+    //
+    // The handler layer drives these (it knows the Role); lifecycle hook
+    // bulk-deletes (`revoke_all_for_*` above) wipe role_grants in lockstep
+    // inside their own implementation, so callers using those paths don't
+    // need to invoke `clear_role` separately.
+
+    /// Set the role for a `(subject, resource)` pair. Idempotent via the
+    /// UNIQUE `(subject_type, subject_id, resource_type, resource_id)`
+    /// constraint — `ON CONFLICT` updates the role + expires_at if they
+    /// changed, which is exactly the right semantics for an atomic role
+    /// change (e.g. promoting Viewer → Editor in one UPDATE with no race
+    /// window, no DELETE+INSERT).
+    async fn set_role(
+        &self,
+        granted_by: Uuid,
+        subject: Subject,
+        role: crate::application::dtos::grant_dto::Role,
+        resource: Resource,
+        expires_at: Option<chrono::DateTime<chrono::Utc>>,
+    ) -> Result<(), DomainError>;
+
+    /// Remove the role for a `(subject, resource)` pair. Idempotent —
+    /// succeeds whether or not the row existed. Called after `revoke`
+    /// succeeds to keep the two tables in sync during dual-write; after
+    /// cleanup this is the canonical role-revocation entry point.
+    async fn clear_role(&self, subject: Subject, resource: Resource) -> Result<(), DomainError>;
 }

src/domain/services/authorization.rs

@@ -140,18 +140,29 @@ pub enum Permission {
     Delete,
     /// Modify the resource (rename, move, edit content).
     Update,
+    /// Configure the resource's settings, add/remove members, change role
+    /// assignments. Used by:
+    /// - Drive owners managing drive membership and policies.
+    /// - Group owners managing the group itself (Group-as-Resource, future).
+    ///
+    /// Folder and file resources do not currently surface a `Manage` check;
+    /// the permission lives in the enum because the role bundle (`Owner`)
+    /// includes it, and the resource types that DO check it (`Drive`,
+    /// `Group`) are added in subsequent PRs (see `docs/plan/drive.md`).
+    Manage,
 }
 
 impl Permission {
     /// Every permission, in a stable order. Used by `Role::expand()` and SQL
     /// `permission = ANY(...)` lookups.
-    pub const ALL: [Permission; 6] = [
+    pub const ALL: [Permission; 7] = [
         Permission::Read,
         Permission::Create,
         Permission::Share,
         Permission::Comment,
         Permission::Delete,
         Permission::Update,
+        Permission::Manage,
     ];
 
     pub fn as_str(&self) -> &'static str {
@@ -162,6 +173,7 @@ impl Permission {
             Permission::Comment => "comment",
             Permission::Delete => "delete",
             Permission::Update => "update",
+            Permission::Manage => "manage",
         }
     }
 
@@ -175,6 +187,7 @@ impl Permission {
             "comment" => Some(Permission::Comment),
             "delete" => Some(Permission::Delete),
             "update" => Some(Permission::Update),
+            "manage" => Some(Permission::Manage),
             _ => None,
         }
     }

src/infrastructure/services/pg_acl_engine.rs

@@ -37,6 +37,7 @@ use uuid::Uuid;
 use moka::future::Cache;
 use sqlx::PgPool;
 
+use crate::application::dtos::grant_dto::roles_implying;
 use crate::application::ports::authorization_ports::AuthorizationEngine;
 use crate::common::errors::DomainError;
 use crate::domain::entities::subject_group::INTERNAL_GROUP_ID;
@@ -237,6 +238,22 @@ impl PgAclEngine {
         }
     }
 
+    /// Convert a `Permission` into the array of role strings whose bundle
+    /// includes it — used to bind the `g.role = ANY($N::text[])` filter on
+    /// every cascade / lookup query that reads `storage.role_grants`.
+    ///
+    /// This is the inverse of `Role::expand()`, precomputed via
+    /// `grant_dto::roles_implying()`. The mapping is small and static (≤5
+    /// roles per permission today); resolving it in code keeps the SQL
+    /// path simple and lets us add new roles without touching every
+    /// query site.
+    fn roles_implying_strings(permission: Permission) -> Vec<&'static str> {
+        roles_implying(permission)
+            .iter()
+            .map(|r| r.as_str())
+            .collect()
+    }
+
     /// Cascading check for folders: is there a grant on any ancestor folder
     /// (including the target itself) for any of the given subject IDs and
     /// any of the given subject types?
@@ -247,6 +264,12 @@ impl PgAclEngine {
     /// `subject_ids` is the expanded set returned by `expand_user` (or a
     /// single-element vec for non-user callers).
     ///
+    /// **D-Prep**: now reads `storage.role_grants` (1 row per role
+    /// assignment) instead of `storage.access_grants` (N rows per role).
+    /// The permission filter `g.permission = $3` becomes
+    /// `g.role = ANY($3::text[])` where the array is the set of roles whose
+    /// bundle includes the requested permission — see `roles_implying()`.
+    ///
     /// Uses the GiST index on `storage.folders.lpath` for O(log N) cascade.
     async fn folder_cascade_grant_exists(
         &self,
@@ -257,14 +280,15 @@ impl PgAclEngine {
         counters: &QueryCounters,
     ) -> Result<bool, DomainError> {
         counters.sql_queries.fetch_add(1, Ordering::Relaxed);
+        let roles = Self::roles_implying_strings(permission);
         let exists: Option<i32> = sqlx::query_scalar(
             r#"
             SELECT 1
-              FROM storage.access_grants g
+              FROM storage.role_grants g
               JOIN storage.folders gf ON gf.id = g.resource_id
              WHERE g.subject_type  = ANY($1)
                AND g.subject_id    = ANY($2)
-               AND g.permission    = $3
+               AND g.role          = ANY($3::text[])
                AND g.resource_type = 'folder'
                AND (g.expires_at IS NULL OR g.expires_at > NOW())
                AND gf.lpath @> (SELECT lpath FROM storage.folders WHERE id = $4)
@@ -273,7 +297,7 @@ impl PgAclEngine {
         )
         .bind(subject_types)
         .bind(subject_ids)
-        .bind(permission.as_str())
+        .bind(&roles)
         .bind(folder_id)
         .fetch_optional(self.pool.as_ref())
         .await
@@ -285,7 +309,7 @@ impl PgAclEngine {
     /// Cascading check for files: either a direct file grant OR a grant on
     /// any ancestor folder of the file's containing folder. See
     /// `folder_cascade_grant_exists` for the meaning of `subject_types` /
-    /// `subject_ids`.
+    /// `subject_ids` and the D-Prep role-array migration.
     async fn file_cascade_grant_exists(
         &self,
         subject_types: &[&str],
@@ -295,27 +319,28 @@ impl PgAclEngine {
         counters: &QueryCounters,
     ) -> Result<bool, DomainError> {
         counters.sql_queries.fetch_add(1, Ordering::Relaxed);
+        let roles = Self::roles_implying_strings(permission);
         let exists: Option<i32> = sqlx::query_scalar(
             r#"
             SELECT 1
               FROM (
                 -- direct file grant
                 SELECT 1
-                  FROM storage.access_grants
+                  FROM storage.role_grants
                  WHERE subject_type = ANY($1)
                    AND subject_id   = ANY($2)
-                   AND permission   = $3
+                   AND role         = ANY($3::text[])
                    AND resource_type = 'file' AND resource_id = $4
                    AND (expires_at IS NULL OR expires_at > NOW())
                 UNION ALL
                 -- cascading from any ancestor folder of the file's containing folder
                 SELECT 1
-                  FROM storage.access_grants g
+                  FROM storage.role_grants g
                   JOIN storage.folders gf     ON gf.id = g.resource_id
                   JOIN storage.files target_f ON target_f.id = $4
                  WHERE g.subject_type  = ANY($1)
                    AND g.subject_id    = ANY($2)
-                   AND g.permission    = $3
+                   AND g.role          = ANY($3::text[])
                    AND g.resource_type = 'folder'
                    AND (g.expires_at IS NULL OR g.expires_at > NOW())
                    AND target_f.folder_id IS NOT NULL
@@ -327,7 +352,7 @@ impl PgAclEngine {
         )
         .bind(subject_types)
         .bind(subject_ids)
-        .bind(permission.as_str())
+        .bind(&roles)
         .bind(file_id)
         .fetch_optional(self.pool.as_ref())
         .await
@@ -1680,6 +1705,19 @@ impl AuthorizationEngine for PgAclEngine {
     }
 
     async fn revoke_all_for_resource(&self, resource: Resource) -> Result<usize, DomainError> {
+        // D-Prep dual-write: wipe role_grants for this resource too.
+        // Idempotent — succeeds whether or not any row existed.
+        sqlx::query(
+            "DELETE FROM storage.role_grants WHERE resource_type = $1 AND resource_id = $2",
+        )
+        .bind(resource.type_str())
+        .bind(resource.id())
+        .execute(self.pool.as_ref())
+        .await
+        .map_err(|e| {
+            DomainError::internal_error("PgAcl", format!("revoke role_grants for resource: {e}"))
+        })?;
+
         let result = sqlx::query(
             "DELETE FROM storage.access_grants WHERE resource_type = $1 AND resource_id = $2",
         )
@@ -1693,6 +1731,16 @@ impl AuthorizationEngine for PgAclEngine {
     }
 
     async fn revoke_all_for_subject(&self, subject: Subject) -> Result<usize, DomainError> {
+        // D-Prep dual-write: wipe role_grants for this subject too.
+        sqlx::query("DELETE FROM storage.role_grants WHERE subject_type = $1 AND subject_id = $2")
+            .bind(subject.type_str())
+            .bind(subject.id())
+            .execute(self.pool.as_ref())
+            .await
+            .map_err(|e| {
+                DomainError::internal_error("PgAcl", format!("revoke role_grants for subject: {e}"))
+            })?;
+
         let result = sqlx::query(
             "DELETE FROM storage.access_grants WHERE subject_type = $1 AND subject_id = $2",
         )
@@ -1704,6 +1752,59 @@ impl AuthorizationEngine for PgAclEngine {
 
         Ok(result.rows_affected() as usize)
     }
+
+    // ── D-Prep role_grants writes ──────────────────────────────────────────
+
+    async fn set_role(
+        &self,
+        granted_by: Uuid,
+        subject: Subject,
+        role: crate::application::dtos::grant_dto::Role,
+        resource: Resource,
+        expires_at: Option<chrono::DateTime<chrono::Utc>>,
+    ) -> Result<(), DomainError> {
+        sqlx::query(
+            r#"
+            INSERT INTO storage.role_grants
+                (subject_type, subject_id, resource_type, resource_id,
+                 role, granted_by, expires_at)
+            VALUES ($1, $2, $3, $4, $5, $6, $7)
+            ON CONFLICT (subject_type, subject_id, resource_type, resource_id)
+            DO UPDATE SET role       = EXCLUDED.role,
+                          expires_at = EXCLUDED.expires_at,
+                          granted_by = EXCLUDED.granted_by
+            "#,
+        )
+        .bind(subject.type_str())
+        .bind(subject.id())
+        .bind(resource.type_str())
+        .bind(resource.id())
+        .bind(role.as_str())
+        .bind(granted_by)
+        .bind(expires_at)
+        .execute(self.pool.as_ref())
+        .await
+        .map_err(|e| DomainError::internal_error("PgAcl", format!("set_role: {e}")))?;
+
+        Ok(())
+    }
+
+    async fn clear_role(&self, subject: Subject, resource: Resource) -> Result<(), DomainError> {
+        sqlx::query(
+            "DELETE FROM storage.role_grants \
+             WHERE subject_type = $1 AND subject_id = $2 \
+               AND resource_type = $3 AND resource_id = $4",
+        )
+        .bind(subject.type_str())
+        .bind(subject.id())
+        .bind(resource.type_str())
+        .bind(resource.id())
+        .execute(self.pool.as_ref())
+        .await
+        .map_err(|e| DomainError::internal_error("PgAcl", format!("clear_role: {e}")))?;
+
+        Ok(())
+    }
 }
 
 // ─────────────────────────────────────────────────────────────────────────────

src/interfaces/api/handlers/grant_handler.rs

@@ -14,7 +14,7 @@ use axum::{
 use futures::future::join_all;
 use serde::Deserialize;
 use std::sync::Arc;
-use tracing::{error, info, warn};
+use tracing::{error, warn};
 use utoipa::IntoParams;
 use uuid::Uuid;
 
@@ -22,7 +22,7 @@ use crate::application::dtos::cursor::PageCursor;
 use crate::application::dtos::grant_dto::{
     CreateGrantDto, CreateGrantResponseDto, GrantDto, MySharesDto, NotifyOutcomeSetDto,
     OutgoingResourceGrantDto, OutgoingResourceItemDto, PermissionDto, ResourceContentDto,
-    ResourceDto, ResourceTypeDto, SharedWithMeDto, SharedWithMeItemDto, SharedWithMeQuery,
+    ResourceDto, ResourceTypeDto, Role, SharedWithMeDto, SharedWithMeItemDto, SharedWithMeQuery,
     SubjectDto, SubjectInputDto, UpdateRoleDto, role_from_permissions,
 };
 use crate::application::ports::authorization_ports::AuthorizationEngine;
@@ -66,10 +66,23 @@ pub async fn create_grant(
     let authz = &state.authorization;
     let caller_id = auth_user.id;
 
-    // Validate: exactly one of permissions/role
-    let permissions: Vec<Permission> = match (dto.permissions, dto.role) {
-        (Some(perms), None) if !perms.is_empty() => perms.into_iter().map(Into::into).collect(),
-        (None, Some(role)) => role.expand().to_vec(),
+    // Validate: exactly one of permissions/role. Capture BOTH the
+    // permission list (for the per-permission access_grants writes that
+    // keep the old engine read path working) AND the role (for the new
+    // role_grants `set_role` dual-write that lands after the per-
+    // permission loop).
+    let (permissions, role): (Vec<Permission>, Role) = match (dto.permissions, dto.role) {
+        (Some(perms), None) if !perms.is_empty() => {
+            let perms: Vec<Permission> = perms.into_iter().map(Into::into).collect();
+            // Derive the closest matching role from the raw permission set
+            // so we have ONE role to mirror into role_grants. `Role::parse`
+            // always succeeds here because `role_from_permissions` only
+            // emits known role strings.
+            let role = Role::parse(role_from_permissions(&perms))
+                .expect("role_from_permissions returns a known role string");
+            (perms, role)
+        }
+        (None, Some(role)) => (role.expand().to_vec(), role),
         (Some(_), Some(_)) => {
             return AppError::new(
                 StatusCode::BAD_REQUEST,
@@ -154,6 +167,16 @@ pub async fn create_grant(
 
     let mut results: Vec<GrantDto> = Vec::with_capacity(permissions.len());
     for perm in permissions {
+        // `storage.access_grants.permission` CHECK constraint predates
+        // `Permission::Manage`; it accepts only the original 6 values.
+        // Manage exists in the Owner bundle for engine read-path use
+        // (via `roles_implying`) and gets persisted via the role_grants
+        // dual-write below. Skipping it here keeps the access_grants
+        // safety net populated without tripping the CHECK; the cleanup
+        // PR that drops access_grants also drops this skip.
+        if perm == Permission::Manage {
+            continue;
+        }
         match authz
             .grant(caller_id, subject, perm, resource, expires_at)
             .await
@@ -165,12 +188,31 @@ pub async fn create_grant(
             }
         }
     }
-    info!(
-        "Created {} grant(s) for subject={:?} on resource={:?} by user {}",
-        results.len(),
-        subject,
-        resource,
-        caller_id
+
+    // D-Prep dual-write: mirror the role assignment into storage.role_grants.
+    // ON CONFLICT UPDATE makes this idempotent — repeated POSTs with the
+    // same (subject, resource) update the role in place, matching the
+    // PATCH-style semantics callers will get after the engine read pivot.
+    if let Err(err) = authz
+        .set_role(caller_id, subject, role, resource, expires_at)
+        .await
+    {
+        error!("set_role dual-write failed: {err}");
+        return AppError::from(err).into_response();
+    }
+
+    tracing::info!(
+        target: "audit",
+        event = "role_grant.created",
+        caller_id = %caller_id,
+        subject_type = subject.type_str(),
+        subject_id = %subject.id(),
+        resource_type = resource.type_str(),
+        resource_id = %resource.id(),
+        role = role.as_str(),
+        permission_count = results.len(),
+        expires_at = ?expires_at,
+        "🤝 grant created with role '{}'", role.as_str(),
     );
 
     // PR N1 — route the post-grant notification through the unified
@@ -274,17 +316,20 @@ pub async fn revoke_grant(
         Err(_) => return AppError::not_found(format!("Grant {id} not found")).into_response(),
     };
 
-    // Look up the grant to find the underlying resource (and granter).
-    let on_resource = match authz.find_grant_by_id(grant_id).await {
-        Ok(Some((res, granter))) => (res, granter),
+    // Look up the grant to find the subject, resource, and granter.
+    // `find_grant_full_by_id` returns the subject too — needed for the
+    // `clear_role` dual-write below (role_grants is keyed by (subject,
+    // resource), not by access_grants id).
+    let (subject, resource, granter) = match authz.find_grant_full_by_id(grant_id).await {
+        Ok(Some(triple)) => triple,
         Ok(None) => return StatusCode::NO_CONTENT.into_response(), // idempotent
         Err(e) => return AppError::from(e).into_response(),
     };
 
     // Caller is authorized if they are the granter OR have Share on the resource.
-    if on_resource.1 != caller_id
+    if granter != caller_id
         && let Err(e) = authz
-            .require(Subject::User(caller_id), Permission::Share, on_resource.0)
+            .require(Subject::User(caller_id), Permission::Share, resource)
             .await
     {
         return AppError::from(e).into_response();
@@ -293,7 +338,36 @@ pub async fn revoke_grant(
     if let Err(e) = authz.revoke(grant_id).await {
         return AppError::from(e).into_response();
     }
-    info!("Revoked grant {grant_id} (caller {caller_id})");
+
+    // D-Prep dual-write: clear the role_grants row for this (subject,
+    // resource). Idempotent — succeeds whether or not the row existed.
+    //
+    // Today's API revokes one access_grants row by id; the role_grants
+    // row models the WHOLE (subject, resource) cluster. Calling clear_role
+    // here effectively revokes the WHOLE role assignment in role_grants,
+    // even if other per-permission access_grants rows remain. This is the
+    // correct semantics for the eventual cleanup-PR model (role_grants is
+    // role-keyed; once access_grants goes away, "revoke" means "drop the
+    // role"). During the dual-write window the two tables can drift
+    // briefly if a caller revokes only some permissions of a role, but
+    // the engine still reads access_grants so behaviour is unchanged.
+    if let Err(e) = authz.clear_role(subject, resource).await {
+        return AppError::from(e).into_response();
+    }
+
+    tracing::info!(
+        target: "audit",
+        event = "role_grant.revoked",
+        caller_id = %caller_id,
+        grant_id = %grant_id,
+        subject_type = subject.type_str(),
+        subject_id = %subject.id(),
+        resource_type = resource.type_str(),
+        resource_id = %resource.id(),
+        granter_id = %granter,
+        self_revoke = (granter == caller_id),
+        "🗑️ grant revoked",
+    );
     StatusCode::NO_CONTENT.into_response()
 }
 
@@ -509,9 +583,22 @@ pub async fn set_role(
         .map(|g| g.permission)
         .collect();
 
-    // Diff and apply.
-    let to_add: Vec<Permission> = target_perms.difference(&current_perms).copied().collect();
-    let to_remove: Vec<Permission> = current_perms.difference(&target_perms).copied().collect();
+    // Diff and apply. `Permission::Manage` is excluded from both sides
+    // because the historical `access_grants.permission` CHECK doesn't
+    // accept it — see the matching skip in `create_grant`. The role
+    // assignment captures Manage via the role_grants `set_role` call
+    // further down; the engine's read-path uses `roles_implying(Manage)`
+    // → `[Owner]` and never goes through per-permission rows.
+    let to_add: Vec<Permission> = target_perms
+        .difference(&current_perms)
+        .copied()
+        .filter(|p| *p != Permission::Manage)
+        .collect();
+    let to_remove: Vec<Permission> = current_perms
+        .difference(&target_perms)
+        .copied()
+        .filter(|p| *p != Permission::Manage)
+        .collect();
 
     for perm in &to_remove {
         if let Some(g) = current
@@ -542,6 +629,19 @@ pub async fn set_role(
         return AppError::from(e).into_response();
     }
 
+    // D-Prep dual-write: mirror the resulting role into storage.role_grants.
+    // The per-permission diff above keeps access_grants converged; this
+    // single UPSERT keeps role_grants in sync with the OVERALL outcome
+    // (one row carrying the role + expiry). After the engine read pivot
+    // and the access_grants drop, the per-permission diff above goes
+    // away and this call becomes the only mutation the handler performs.
+    if let Err(e) = authz
+        .set_role(caller_id, subject, dto.role, resource, expires_at)
+        .await
+    {
+        return AppError::from(e).into_response();
+    }
+
     // Return the new full set.
     let after = match authz.list_grants_on_resource(resource).await {
         Ok(g) => g,
@@ -553,9 +653,22 @@ pub async fn set_role(
         .map(Into::into)
         .collect();
 
-    info!(
-        "Role applied: caller={} subject={:?} resource={:?} added={:?} removed={:?}",
-        caller_id, subject, resource, to_add, to_remove
+    tracing::info!(
+        target: "audit",
+        event = "role_grant.role_set",
+        caller_id = %caller_id,
+        subject_type = subject.type_str(),
+        subject_id = %subject.id(),
+        resource_type = resource.type_str(),
+        resource_id = %resource.id(),
+        role = dto.role.as_str(),
+        permissions_added = to_add.len(),
+        permissions_removed = to_remove.len(),
+        expires_at = ?expires_at,
+        "🔁 role set to '{}' (+{} -{})",
+        dto.role.as_str(),
+        to_add.len(),
+        to_remove.len(),
     );
     (StatusCode::OK, Json(mine)).into_response()
 }