feat(drive): prepare removal of user_id

this commit changes GET /api/<resources> to return resource caller has access to
    this is not anymmore resources users is owner of

migrations/20260901000002_caller_group_ids_function.sql

@@ -0,0 +1,74 @@
+-- ════════════════════════════════════════════════════════════════════════════
+-- PR-B — storage.caller_group_ids: recursive group-membership expansion in SQL
+-- ════════════════════════════════════════════════════════════════════════════
+-- Every listing surface that scopes by "drives the caller can Read" (Photos,
+-- Places, GET /api/drives, Trash, Search, root-folder listing) needs the
+-- caller's *effective subject set* = caller_id ∪ every group they belong to
+-- transitively.
+--
+-- Pre-Option-A the Rust-side `PgAclEngine::expand_subject_for_listing` did
+-- the walk once (via `WITH RECURSIVE` in `subject_group_pg_repository.rs::
+-- groups_for_user`) and cached the result in a Moka table with 30-second
+-- TTL; every listing handler then passed the two parallel arrays
+-- `subject_types` + `subject_ids` into the SQL. That leaked the expansion
+-- ceremony into every caller (7+ sites).
+--
+-- Option A pushes the walk into a `STABLE` SQL function so each listing
+-- query embeds the expansion inline:
+--
+--   WHERE (g.subject_type = 'user'  AND g.subject_id = $caller)
+--      OR (g.subject_type = 'group' AND g.subject_id IN
+--              (SELECT storage.caller_group_ids($caller)))
+--
+-- Callers pass a bare `caller_id: Uuid` — no more expand-then-bind
+-- ceremony. Postgres re-runs the walk per listing (~1-3 ms against the
+-- indexed `auth.subject_group_members` table); we lose the Moka cache
+-- benefit but gain a single audit trail for "how does group access
+-- cascade" (this function) and drop ~15 lines of Rust glue per listing.
+--
+-- Cycle safety: `subject_group_pg_repository.rs::add_member` enforces
+-- an INSERT-time cycle check via `WITH RECURSIVE descendants`, so the
+-- membership DAG is guaranteed acyclic. Depth is capped at
+-- MAX_GROUP_DEPTH by the same INSERT path. The recursion below always
+-- terminates.
+--
+-- `STABLE`: the function reads DB state but never modifies it, and the
+-- result is deterministic within a transaction. Postgres can memoise
+-- calls within a single query plan (e.g. multiple references in the
+-- same SELECT) and inline the CTE into the surrounding query where
+-- beneficial. Marking it `VOLATILE` would forbid both optimisations.
+--
+-- `LEAKPROOF` is deliberately NOT set: the function reads a private
+-- auth table, so it must not be pushed below a security barrier.
+--
+-- `SECURITY INVOKER` (the default) — runs with the calling role's
+-- permissions, so RLS on `auth.subject_group_members` (if ever added)
+-- applies consistently.
+
+CREATE OR REPLACE FUNCTION storage.caller_group_ids(caller UUID)
+RETURNS SETOF UUID
+LANGUAGE sql
+STABLE
+AS $$
+    WITH RECURSIVE user_groups AS (
+        -- Direct memberships: groups the caller is listed in as a user.
+        SELECT group_id
+          FROM auth.subject_group_members
+         WHERE member_user_id = caller
+
+        UNION
+
+        -- Transitive memberships: groups that contain a group the caller
+        -- already belongs to. Repeats until no new rows are produced.
+        SELECT m.group_id
+          FROM auth.subject_group_members m
+          JOIN user_groups ug ON m.member_group_id = ug.group_id
+    )
+    SELECT group_id FROM user_groups;
+$$;
+
+-- Backing indexes used by the recursion. Already present from
+-- 20260307000000_initial_schema.sql on
+-- `auth.subject_group_members (member_user_id)` and
+-- `auth.subject_group_members (member_group_id)` — no additional
+-- indexes needed here.

src/application/services/folder_service.rs

@@ -358,18 +358,18 @@ impl FolderUseCase for FolderService {
                 .await?;
             return self.list_folders(parent_id).await;
         }
-        // No parent → list the user's root folders.
+        // No parent → list the caller's readable root folders. The
+        // predicate scopes by drive-membership grants (post-PR-B),
+        // closing the pre-D7 gap where the legacy `user_id` filter
+        // surfaced admin-created folders that admin had no role on.
         let folders = self
             .folder_storage
-            .list_folders_by_owner(parent_id, caller_id)
+            .list_root_folders_for_caller(caller_id)
             .await
             .map_err(|e| {
                 DomainError::internal_error(
                     "FolderStorage",
-                    format!(
-                        "Failed to list folders for owner '{}' in parent {:?}: {}",
-                        caller_id, parent_id, e
-                    ),
+                    format!("Failed to list root folders for caller '{caller_id}': {e}"),
                 )
             })?;
         Ok(folders.into_iter().map(FolderDto::from).collect())
@@ -432,8 +432,7 @@ impl FolderUseCase for FolderService {
         } else {
             let (folders, total_items) = self
             .folder_storage
-            .list_folders_by_owner_paginated(
-                parent_id,
+            .list_root_folders_for_caller_paginated(
                 owner_id,
                 pagination.offset(),
                 pagination.limit(),
@@ -444,8 +443,8 @@ impl FolderUseCase for FolderService {
                 DomainError::internal_error(
                     "FolderStorage",
                     format!(
-                        "Failed to list folders for owner '{}' with pagination in parent {:?}: {}",
-                        owner_id, parent_id, e
+                        "Failed to list root folders for caller '{}' with pagination: {}",
+                        owner_id, e
                     ),
                 )
             })?;

src/application/services/places_service.rs

@@ -4,29 +4,23 @@ use uuid::Uuid;
 
 use crate::application::dtos::geo_dto::{GeoBounds, GeoCluster};
 use crate::common::errors::DomainError;
-use crate::domain::services::authorization::Subject;
 use crate::infrastructure::repositories::pg::FileBlobReadRepository;
-use crate::infrastructure::services::pg_acl_engine::PgAclEngine;
 
 /// "Places" use case: the caller's geotagged photos aggregated into map
 /// clusters.
 ///
 /// Post-§15 the surface follows the Photos scope: default personal drive
 /// + drives where `policies.include_in_photo_index = true` AND caller
-/// has Read. The repository query joins `role_grants` on the drive
-/// resource type; group-mediated grants are honoured via the caller
-/// expansion done here.
+/// has Read. Group-membership expansion is handled inline by
+/// `storage.caller_group_ids(caller)` inside the repo's SQL, so this
+/// service is a thin coordinate-math wrapper — no engine dependency.
 pub struct PlacesService {
     file_read: Arc<FileBlobReadRepository>,
-    authorization: Arc<PgAclEngine>,
 }
 
 impl PlacesService {
-    pub fn new(file_read: Arc<FileBlobReadRepository>, authorization: Arc<PgAclEngine>) -> Self {
-        Self {
-            file_read,
-            authorization,
-        }
+    pub fn new(file_read: Arc<FileBlobReadRepository>) -> Self {
+        Self { file_read }
     }
 
     /// Aggregation cell side, in degrees, for a slippy-map zoom level. The
@@ -46,12 +40,8 @@ impl PlacesService {
         zoom: u8,
     ) -> Result<Vec<GeoCluster>, DomainError> {
         let cell = Self::cell_for_zoom(zoom);
-        let (subject_types, subject_ids) = self
-            .authorization
-            .expand_subject_for_listing(Subject::User(caller_id))
-            .await?;
         self.file_read
-            .list_geo_clusters(&subject_types, &subject_ids, bounds, cell)
+            .list_geo_clusters(caller_id, bounds, cell)
             .await
     }
 }

src/application/services/search_service.rs

@@ -283,20 +283,10 @@ impl SearchService {
             return Vec::new();
         };
 
-        // Resolve the caller's accessible drive set via the engine
-        // (handles group-mediated drive grants) + the repo lookup.
-        let caller = Subject::User(user_id);
-        let (subject_types, subject_ids) = match authz.expand_subject_for_listing(caller).await {
-            Ok(pair) => pair,
-            Err(e) => {
-                tracing::warn!("Content-index: subject expansion failed — degrading to empty: {e}");
-                return Vec::new();
-            }
-        };
-        let accessible_drives: Vec<Uuid> = match drive_repo
-            .list_for_subjects(&subject_types, &subject_ids)
-            .await
-        {
+        // Resolve the caller's accessible drive set. Group-mediated
+        // grants are honoured inline by `storage.caller_group_ids` on
+        // the SQL side, so no Rust-side subject expansion here.
+        let accessible_drives: Vec<Uuid> = match drive_repo.list_readable_by(user_id).await {
             Ok(drives) => drives.into_iter().map(|d| d.drive.id).collect(),
             Err(e) => {
                 tracing::warn!("Content-index: drive lookup failed — degrading to empty: {e}");
@@ -338,7 +328,7 @@ impl SearchService {
                 }
             };
             match authz
-                .check(caller, Permission::Read, Resource::File(file_uuid))
+                .check(Subject::User(user_id), Permission::Read, Resource::File(file_uuid))
                 .await
             {
                 Ok(true) => verified.push(hit),

src/application/services/share_service.rs

@@ -983,10 +983,9 @@ mod tests {
             unimplemented!()
         }
 
-        async fn list_folders_by_owner(
+        async fn list_root_folders_for_caller(
             &self,
-            _parent_id: Option<&str>,
-            _owner_id: Uuid,
+            _caller_id: Uuid,
         ) -> Result<Vec<crate::domain::entities::folder::Folder>, DomainError> {
             unimplemented!()
         }
@@ -1002,10 +1001,9 @@ mod tests {
             unimplemented!()
         }
 
-        async fn list_folders_by_owner_paginated(
+        async fn list_root_folders_for_caller_paginated(
             &self,
-            _parent_id: Option<&str>,
-            _owner_id: Uuid,
+            _caller_id: Uuid,
             _offset: usize,
             _limit: usize,
             _include_total: bool,

src/application/services/trash_service.rs

@@ -786,13 +786,9 @@ impl TrashService {
     /// keeps the two HTTP surfaces semantically consistent and avoids
     /// duplicating the subject-expansion plumbing.
     async fn drives_with_delete_for(&self, user_id: Uuid) -> Result<Vec<Uuid>> {
-        let (subject_types, subject_ids) = self
-            .authz
-            .expand_subject_for_listing(Subject::User(user_id))
-            .await?;
         let drives = self
             .drive_repo
-            .list_for_subjects(&subject_types, &subject_ids)
+            .list_readable_by(user_id)
             .await
             .map_err(|e| {
                 DomainError::internal_error(
@@ -900,15 +896,7 @@ impl TrashService {
         // D2b: scope by drives the caller can read (resolved through
         // role_grants on resource_type='drive', including group-mediated
         // grants). Empty set → empty page without a SQL round-trip.
-        let (subject_types, subject_ids) = self
-            .authz
-            .expand_subject_for_listing(Subject::User(user_id))
-            .await?;
-        let drive_ids: Vec<Uuid> = match self
-            .drive_repo
-            .list_for_subjects(&subject_types, &subject_ids)
-            .await
-        {
+        let drive_ids: Vec<Uuid> = match self.drive_repo.list_readable_by(user_id).await {
             Ok(drives) => drives.into_iter().map(|d| d.drive.id).collect(),
             Err(e) => {
                 return Err(DomainError::internal_error(

src/application/services/trash_service_test.rs

@@ -745,10 +745,9 @@ impl FolderRepository for MockFolderRepository {
         Ok(vec![])
     }
 
-    async fn list_folders_by_owner(
+    async fn list_root_folders_for_caller(
         &self,
-        _parent_id: Option<&str>,
-        _owner_id: Uuid,
+        _caller_id: Uuid,
     ) -> std::result::Result<Vec<Folder>, DomainError> {
         Ok(vec![])
     }
@@ -763,10 +762,9 @@ impl FolderRepository for MockFolderRepository {
         Ok((vec![], Some(0)))
     }
 
-    async fn list_folders_by_owner_paginated(
+    async fn list_root_folders_for_caller_paginated(
         &self,
-        _parent_id: Option<&str>,
-        _owner_id: Uuid,
+        _caller_id: Uuid,
         _offset: usize,
         _limit: usize,
         _include_total: bool,

src/common/di.rs

@@ -915,17 +915,14 @@ impl AppServiceFactory {
     /// repository — the data is the caller's Photos-scope geotagged photos
     /// (§15: default personal drive + drives with
     /// `include_in_photo_index = true` AND caller has Read).
-    /// `authorization` is used for the same subject expansion that
-    /// `photos_handler::list_photos` runs.
+    /// Group-membership expansion is inline in the SQL via
+    /// `storage.caller_group_ids`, so the service needs no AuthZ engine
+    /// handle.
     pub fn create_places_service(
         &self,
         file_read: &Arc<FileBlobReadRepository>,
-        authorization: &Arc<crate::infrastructure::services::pg_acl_engine::PgAclEngine>,
     ) -> Arc<PlacesService> {
-        let service = Arc::new(PlacesService::new(
-            file_read.clone(),
-            authorization.clone(),
-        ));
+        let service = Arc::new(PlacesService::new(file_read.clone()));
         tracing::info!("Places service initialized");
         service
     }
@@ -1293,7 +1290,7 @@ impl AppServiceFactory {
             apps.recent_service = Some(recent_service_eager.clone());
 
             places_service = if core.config.features.enable_places {
-                Some(self.create_places_service(&repos.file_read_repository, &authorization))
+                Some(self.create_places_service(&repos.file_read_repository))
             } else {
                 None
             };

src/common/stubs.rs

@@ -274,10 +274,9 @@ impl FolderRepository for StubFolderStoragePort {
         Ok(Vec::new())
     }
 
-    async fn list_folders_by_owner(
+    async fn list_root_folders_for_caller(
         &self,
-        _parent_id: Option<&str>,
-        _owner_id: Uuid,
+        _caller_id: Uuid,
     ) -> Result<Vec<Folder>, DomainError> {
         Ok(Vec::new())
     }
@@ -292,10 +291,9 @@ impl FolderRepository for StubFolderStoragePort {
         Ok((Vec::new(), Some(0)))
     }
 
-    async fn list_folders_by_owner_paginated(
+    async fn list_root_folders_for_caller_paginated(
         &self,
-        _parent_id: Option<&str>,
-        _owner_id: Uuid,
+        _caller_id: Uuid,
         _offset: usize,
         _limit: usize,
         _include_total: bool,

src/domain/repositories/drive_repository.rs

@@ -52,7 +52,7 @@ pub struct DriveWithRootName {
     /// of the root folder via JOIN at read time.
     pub root_folder_name: String,
     /// Highest role the calling user holds on this drive (direct OR
-    /// group-mediated). Populated by `list_for_subjects` (which already
+    /// group-mediated). Populated by `list_readable_by` (which already
     /// JOINs `role_grants` for accessibility, so the role is in scope at
     /// query time). `None` for repo methods called without a caller
     /// context (`get_by_id`, `get_by_ids`, `find_default_for_user`,
@@ -164,17 +164,17 @@ pub trait DriveRepository: Send + Sync + 'static {
     }
 
     /// List drives the caller can read, resolved via `role_grants` for
-    /// `resource_type='drive'`. The caller's group memberships are
-    /// expanded by the engine's `subject_match_set`; that expanded set
-    /// is what this method's `subject_ids` argument carries.
+    /// `resource_type='drive'`. Group memberships (direct + transitive)
+    /// are expanded inline by the `storage.caller_group_ids(caller)`
+    /// SQL function — callers pass only the caller's uuid, no
+    /// expansion ceremony.
     ///
     /// Returns rows in a stable order: default drive first (if any),
     /// then by display name. The `/api/drives` handler relies on that
     /// order for the picker UI without a follow-up sort.
-    async fn list_for_subjects(
+    async fn list_readable_by(
         &self,
-        subject_types: &[&str],
-        subject_ids: &[Uuid],
+        caller_id: Uuid,
     ) -> Result<Vec<DriveWithRootName>, DriveRepositoryError>;
 
     /// `true` when the drive holds no live (non-trashed) folders other
@@ -193,7 +193,7 @@ pub trait DriveRepository: Send + Sync + 'static {
     /// List every drive on the system, regardless of caller membership.
     ///
     /// Used by the admin panel's `GET /api/admin/drives`. Distinct from
-    /// `list_for_subjects` (which filters by `role_grants`) because an
+    /// `list_readable_by` (which filters by `role_grants`) because an
     /// admin who creates a shared drive for someone else has no grant
     /// on it — but still needs to see, audit, and manage it. The HTTP
     /// gate (admin-only middleware) is what makes the unrestricted

src/domain/repositories/folder_repository.rs

@@ -13,6 +13,17 @@ use crate::domain::entities::folder::Folder;
 use crate::domain::services::path_service::StoragePath;
 use uuid::Uuid;
 
+// NOTE on `caller_role` for the two listing methods below:
+// We deliberately do NOT compute or return the caller's role per row.
+// The frontend already fetches `/api/drives` (which surfaces
+// `caller_role` per drive) and cross-references by `folder.drive_id` —
+// see `MoveDialog.svelte` and the config/drive page. Adding
+// `caller_role` to `FolderDto` would either (a) mean redundant
+// server-side work for a client-side concern the client already
+// handles, or (b) drag folder-level grant cascades into the query
+// which is real cost for a rare edge case. Punted; see
+// `project_caller_role_on_file_folder_dto` memory.
+
 /// Domain port for folder persistence.
 ///
 /// Defines the CRUD and management operations required for
@@ -51,13 +62,20 @@ pub trait FolderRepository: Send + Sync + 'static {
     /// Lists folders within a parent folder
     async fn list_folders(&self, parent_id: Option<&str>) -> Result<Vec<Folder>, DomainError>;
 
-    /// Lists root-level folders owned by a specific user.
-    /// For non-root queries (parent_id is Some), ownership is implicit
-    /// because the parent already belongs to the user.
-    async fn list_folders_by_owner(
+    /// Lists root-level folders the caller can read — scoped through
+    /// drive-membership grants (`role_grants` on `resource_type='drive'`)
+    /// rather than the legacy `folders.user_id` column. Group memberships
+    /// are expanded inline by `storage.caller_group_ids($caller)` in the
+    /// SQL. Closes [[bug-root-folder-listing-legacy-user-id]] — root
+    /// folders admin created for other users but has no role on no
+    /// longer surface in the admin's `GET /api/folders`.
+    ///
+    /// Non-root queries (parent_id != None) go through `list_folders`
+    /// with the parent already permission-checked at the service layer,
+    /// so this method carries no `parent_id` parameter.
+    async fn list_root_folders_for_caller(
         &self,
-        parent_id: Option<&str>,
-        owner_id: Uuid,
+        caller_id: Uuid,
     ) -> Result<Vec<Folder>, DomainError>;
 
     /// Lists folders with pagination
@@ -69,13 +87,12 @@ pub trait FolderRepository: Send + Sync + 'static {
         include_total: bool,
     ) -> Result<(Vec<Folder>, Option<usize>), DomainError>;
 
-    /// Lists folders with pagination, scoped to a specific owner.
-    /// Combines the owner filtering of `list_folders_by_owner` with
-    /// the pagination of `list_folders_paginated`.
-    async fn list_folders_by_owner_paginated(
+    /// Paginated companion to `list_root_folders_for_caller` — same
+    /// drive-scoped predicate, adds LIMIT/OFFSET + optional
+    /// window-function COUNT.
+    async fn list_root_folders_for_caller_paginated(
         &self,
-        parent_id: Option<&str>,
-        owner_id: Uuid,
+        caller_id: Uuid,
         offset: usize,
         limit: usize,
         include_total: bool,

src/domain/repositories/trash_repository.rs

@@ -27,7 +27,7 @@ pub trait TrashRepository: Send + Sync {
     ///
     /// **Caller contract**: pass only drive UUIDs the caller has
     /// `Permission::Delete` on (resolved by the service via
-    /// `DriveRepository::list_for_subjects` + role-bundle filter). This
+    /// `DriveRepository::list_readable_by` + role-bundle filter). This
     /// repository performs no authorization — see
     /// `TrashService::empty_trash` for the canonical call site.
     async fn clear_trash(&self, drive_ids: &[Uuid]) -> Result<()>;

src/infrastructure/repositories/pg/drive_pg_repository.rs

@@ -3,7 +3,7 @@
 //! The repo deals only with the `storage.drives` table itself. Drive
 //! membership lives in `storage.role_grants` (`resource_type='drive'`)
 //! and is queried through the engine's existing grant paths;
-//! `list_for_subjects` below resolves `role_grants` → `storage.drives`
+//! `list_readable_by` below resolves `role_grants` → `storage.drives`
 //! via a single join.
 //!
 //! See `migrations/20260802000000_drives_schema_additive.sql` for the
@@ -75,7 +75,7 @@ impl DrivePgRepository {
     /// is declared owner→viewer (strongest→weakest), so `MIN` picks the
     /// strongest of the caller's grants on the drive (direct +
     /// group-mediated collapsed by GROUP BY). Used only by
-    /// `list_for_subjects`.
+    /// `list_readable_by`.
     fn row_to_drive_with_name_and_role(
         row: &sqlx::postgres::PgRow,
     ) -> Result<DriveWithRootName, DriveRepositoryError> {
@@ -463,13 +463,15 @@ impl DriveRepository for DrivePgRepository {
         Self::row_to_drive_with_name(&row)
     }
 
-    async fn list_for_subjects(
+    async fn list_readable_by(
         &self,
-        subject_types: &[&str],
-        subject_ids: &[Uuid],
+        caller_id: Uuid,
     ) -> Result<Vec<DriveWithRootName>, DriveRepositoryError> {
         // Joining role_grants → drives → folders returns every drive the
-        // expanded subject set can read, paired with its display name.
+        // caller can read, paired with its display name. Group
+        // memberships (direct + transitive) are expanded inline by
+        // `storage.caller_group_ids($caller)` — no Rust-side ceremony.
+        //
         // ORDER BY puts default drives first (so the picker UI doesn't
         // need a follow-up sort), then alphabetical by name. GROUP BY
         // collapses duplicate role_grants on the same drive (direct +
@@ -481,8 +483,6 @@ impl DriveRepository for DrivePgRepository {
         // weakest), so MIN returns the strongest. Cast `::text` matches
         // the codebase convention for reading enum columns into Rust
         // (see `pg_acl_engine.rs`); `Role::parse` handles the trip back.
-        // Collapses direct + group-mediated grants on the same drive
-        // into one row alongside the existing GROUP BY.
         let rows = sqlx::query(
             r#"
             SELECT d.id, d.kind, d.default_for_user, d.root_folder_id,
@@ -495,8 +495,11 @@ impl DriveRepository for DrivePgRepository {
               JOIN storage.role_grants g
                 ON g.resource_type = 'drive'
                AND g.resource_id   = d.id
-             WHERE g.subject_type = ANY($1)
-               AND g.subject_id   = ANY($2)
+             WHERE (
+                     (g.subject_type = 'user'  AND g.subject_id = $1)
+                  OR (g.subject_type = 'group' AND g.subject_id IN
+                          (SELECT storage.caller_group_ids($1)))
+                   )
                AND (g.expires_at IS NULL OR g.expires_at > NOW())
              GROUP BY d.id, d.kind, d.default_for_user, d.root_folder_id,
                       d.quota_bytes, d.used_bytes, d.policies,
@@ -505,16 +508,10 @@ impl DriveRepository for DrivePgRepository {
                       LOWER(f.name) ASC
             "#,
         )
-        .bind(
-            subject_types
-                .iter()
-                .map(|s| s.to_string())
-                .collect::<Vec<_>>(),
-        )
-        .bind(subject_ids)
+        .bind(caller_id)
         .fetch_all(self.pool.as_ref())
         .await
-        .map_err(|e| Self::map_sqlx_err("list_for_subjects", e))?;
+        .map_err(|e| Self::map_sqlx_err("list_readable_by", e))?;
 
         rows.iter()
             .map(Self::row_to_drive_with_name_and_role)

src/infrastructure/repositories/pg/file_blob_read_repository.rs

@@ -452,29 +452,28 @@ impl FileBlobReadRepository {
     /// drive_id already ordered by capture date, so LIMIT stops the scan
     /// early. Same O(LIMIT) shape as the pre-D7 `user_id`-keyed hot path.
     ///
-    /// Scope (`docs/plan/drive.md` §15): the caller's *effective subjects*
-    /// × drives with `policies.include_in_photo_index = true`. Default
-    /// personal drives always match because the flag is materialised to
-    /// `true` at drive creation (see
+    /// Scope (`docs/plan/drive.md` §15): drives with
+    /// `policies.include_in_photo_index = true` where the caller has a
+    /// direct grant (`subject_type = 'user'`) OR a grant on a group they
+    /// belong to transitively. Group membership is expanded inline by the
+    /// `storage.caller_group_ids(caller)` SQL function (migration
+    /// `20260901000002_caller_group_ids_function.sql`) — no ceremony at
+    /// the handler layer, no cross-space ambiguity from the earlier
+    /// parallel-arrays pattern.
+    ///
+    /// Default personal drives always match because the flag is
+    /// materialised to `true` at drive creation (see
     /// `DriveRepository::create_personal_drive_atomic` + the backfill
     /// migration `20260901000000_default_personal_photo_music_flags.sql`)
     /// — no per-kind carve-out needed. Non-default drives (secondary
     /// personals, shared drives) surface here only after their owner
     /// flips the flag on via the admin "Manage policies" modal.
-    ///
-    /// `subject_types` / `subject_ids` are the caller expanded through
-    /// their group memberships (`AuthorizationEngine::
-    /// expand_subject_for_listing`); the arrays reach into the ANY()
-    /// predicates so a group-mediated grant on a drive counts too.
     pub async fn list_media_files(
         &self,
-        subject_types: &[&str],
-        subject_ids: &[Uuid],
+        caller_id: Uuid,
         before: Option<i64>,
         limit: i64,
     ) -> Result<(Vec<File>, Vec<i64>, Vec<(Option<i32>, Option<i32>)>), DomainError> {
-        let subject_types_owned: Vec<String> =
-            subject_types.iter().map(|s| s.to_string()).collect();
         let rows: Vec<MediaFileRow> = sqlx::query_as(
             r#"
             SELECT fi.id::text, fi.name, fi.folder_id::text, fo.path,
@@ -495,21 +494,23 @@ impl FileBlobReadRepository {
                        JOIN storage.role_grants g
                          ON g.resource_type = 'drive'
                         AND g.resource_id   = d.id
-                      WHERE g.subject_type = ANY($1)
-                        AND g.subject_id   = ANY($2)
+                      WHERE (
+                              (g.subject_type = 'user'  AND g.subject_id = $1)
+                           OR (g.subject_type = 'group' AND g.subject_id IN
+                                   (SELECT storage.caller_group_ids($1)))
+                            )
                         AND (g.expires_at IS NULL OR g.expires_at > NOW())
                         AND (d.policies->>'include_in_photo_index')::boolean = true
                    )
                AND NOT fi.is_trashed
                AND (fi.mime_type LIKE 'image/%' OR fi.mime_type LIKE 'video/%')
-               AND ($3::bigint IS NULL
-                    OR EXTRACT(EPOCH FROM fi.media_sort_date)::bigint < $3::bigint)
+               AND ($2::bigint IS NULL
+                    OR EXTRACT(EPOCH FROM fi.media_sort_date)::bigint < $2::bigint)
              ORDER BY fi.media_sort_date DESC
-             LIMIT $4
+             LIMIT $3
             "#,
         )
-        .bind(&subject_types_owned)
-        .bind(subject_ids)
+        .bind(caller_id)
         .bind(before)
         .bind(limit)
         .fetch_all(self.pool.as_ref())
@@ -537,21 +538,18 @@ impl FileBlobReadRepository {
     /// Scope: same `include_in_photo_index` predicate as
     /// `list_media_files` (§15). Places is the map view over the same
     /// content set the Photos timeline shows, so the two surfaces MUST
-    /// agree on drive scope. If a drive is opt-out for Photos its
-    /// geotagged files never appear on the map either.
+    /// agree on drive scope. Group membership is expanded inline by
+    /// `storage.caller_group_ids(caller)`.
     ///
     /// This query is a per-cell aggregate (group by rounded lat/lng
     /// bucket) rather than an ORDER BY / LIMIT hot path — the plain
     /// `idx_files_drive_id` is sufficient to seek by drive.
     pub async fn list_geo_clusters(
         &self,
-        subject_types: &[&str],
-        subject_ids: &[Uuid],
+        caller_id: Uuid,
         bounds: GeoBounds,
         cell: f64,
     ) -> Result<Vec<GeoCluster>, DomainError> {
-        let subject_types_owned: Vec<String> =
-            subject_types.iter().map(|s| s.to_string()).collect();
         let rows: Vec<(i64, f64, f64, String)> = sqlx::query_as(
             r#"
             SELECT count(*)              AS n,
@@ -566,21 +564,23 @@ impl FileBlobReadRepository {
                        JOIN storage.role_grants g
                          ON g.resource_type = 'drive'
                         AND g.resource_id   = d.id
-                      WHERE g.subject_type = ANY($1)
-                        AND g.subject_id   = ANY($2)
+                      WHERE (
+                              (g.subject_type = 'user'  AND g.subject_id = $1)
+                           OR (g.subject_type = 'group' AND g.subject_id IN
+                                   (SELECT storage.caller_group_ids($1)))
+                            )
                         AND (g.expires_at IS NULL OR g.expires_at > NOW())
                         AND (d.policies->>'include_in_photo_index')::boolean = true
                    )
                AND NOT fi.is_trashed
                AND fm.latitude IS NOT NULL
                AND fm.longitude IS NOT NULL
-               AND fm.longitude BETWEEN $3 AND $4
-               AND fm.latitude  BETWEEN $5 AND $6
-             GROUP BY round(fm.longitude / $7), round(fm.latitude / $7)
+               AND fm.longitude BETWEEN $2 AND $3
+               AND fm.latitude  BETWEEN $4 AND $5
+             GROUP BY round(fm.longitude / $6), round(fm.latitude / $6)
             "#,
         )
-        .bind(&subject_types_owned)
-        .bind(subject_ids)
+        .bind(caller_id)
         .bind(bounds.west)
         .bind(bounds.east)
         .bind(bounds.south)

src/infrastructure/repositories/pg/folder_db_repository.rs

@@ -392,47 +392,55 @@ impl FolderRepository for FolderDbRepository {
             .collect()
     }
 
-    #[allow(clippy::type_complexity)]
-    async fn list_folders_by_owner(
+    async fn list_root_folders_for_caller(
         &self,
-        parent_id: Option<&str>,
-        owner_id: Uuid,
+        caller_id: Uuid,
     ) -> Result<Vec<Folder>, DomainError> {
-        let rows: Vec<FolderRow> = if let Some(pid) = parent_id {
-            sqlx::query_as(
-                r#"
-                SELECT id::text, name, path, parent_id::text, user_id, drive_id,
-                       EXTRACT(EPOCH FROM created_at)::bigint,
-                       EXTRACT(EPOCH FROM updated_at)::bigint,
-                       EXTRACT(EPOCH FROM tree_modified_at)::bigint,
-                         created_by, updated_by
-                  FROM storage.folders
-                 WHERE parent_id = $1::uuid AND user_id = $2 AND NOT is_trashed
-                 ORDER BY name
-                "#,
-            )
-            .bind(pid)
-            .bind(owner_id)
-            .fetch_all(self.pool())
-            .await
-        } else {
-            sqlx::query_as(
-                r#"
-                SELECT id::text, name, path, parent_id::text, user_id, drive_id,
-                       EXTRACT(EPOCH FROM created_at)::bigint,
-                       EXTRACT(EPOCH FROM updated_at)::bigint,
-                       EXTRACT(EPOCH FROM tree_modified_at)::bigint,
-                         created_by, updated_by
-                  FROM storage.folders
-                 WHERE parent_id IS NULL AND user_id = $1 AND NOT is_trashed
-                 ORDER BY name
-                "#,
-            )
-            .bind(owner_id)
-            .fetch_all(self.pool())
-            .await
-        }
-        .map_err(|e| DomainError::internal_error("FolderDb", format!("list_by_owner: {e}")))?;
+        // Drive-scoped root-folder listing: return every root folder
+        // whose drive the caller has any role_grant on. Group
+        // memberships (direct + transitive) resolve inline via
+        // `storage.caller_group_ids($1)`.
+        //
+        // Closes `bug_root_folder_listing_legacy_user_id`: pre-D7 this
+        // query filtered on `folders.user_id = $caller`, which returned
+        // rows admin had created for other users' drives without ever
+        // getting a role on them. The drive-membership predicate below
+        // makes the "admin's own listing" correct without a separate
+        // filter.
+        //
+        // `caller_role` is NOT surfaced here — see the memory
+        // `project_caller_role_on_file_folder_dto` and the note at the
+        // top of `folder_repository.rs`. Frontend cross-references
+        // `/api/drives::caller_role` via `folder.drive_id`.
+        let rows: Vec<FolderRow> = sqlx::query_as(
+            r#"
+            SELECT f.id::text, f.name, f.path, f.parent_id::text, f.user_id, f.drive_id,
+                   EXTRACT(EPOCH FROM f.created_at)::bigint,
+                   EXTRACT(EPOCH FROM f.updated_at)::bigint,
+                   EXTRACT(EPOCH FROM f.tree_modified_at)::bigint,
+                   f.created_by, f.updated_by
+              FROM storage.folders f
+             WHERE f.parent_id IS NULL
+               AND NOT f.is_trashed
+               AND EXISTS (
+                     SELECT 1
+                       FROM storage.role_grants g
+                      WHERE g.resource_type = 'drive'
+                        AND g.resource_id   = f.drive_id
+                        AND (g.expires_at IS NULL OR g.expires_at > NOW())
+                        AND (
+                              (g.subject_type = 'user'  AND g.subject_id = $1)
+                           OR (g.subject_type = 'group' AND g.subject_id IN
+                                   (SELECT storage.caller_group_ids($1)))
+                            )
+                   )
+             ORDER BY f.name
+            "#,
+        )
+        .bind(caller_id)
+        .fetch_all(self.pool())
+        .await
+        .map_err(|e| DomainError::internal_error("FolderDb", format!("list_root_folders: {e}")))?;
 
         rows.into_iter()
             .map(|(id, name, path, pid, uid, did, ca, ma, tma, cb, ub)| {
@@ -512,60 +520,52 @@ impl FolderRepository for FolderDbRepository {
         Ok((folders?, total))
     }
 
-    /// Paginated folder listing filtered by owner — single query with
-    /// `COUNT(*) OVER()` to avoid a separate COUNT round-trip.
-    #[allow(clippy::type_complexity)]
-    async fn list_folders_by_owner_paginated(
+    /// Paginated companion to `list_root_folders_for_caller` — same
+    /// drive-membership predicate, adds LIMIT/OFFSET and an optional
+    /// window-function COUNT so total pages can be surfaced without a
+    /// second round-trip.
+    async fn list_root_folders_for_caller_paginated(
         &self,
-        parent_id: Option<&str>,
-        owner_id: Uuid,
+        caller_id: Uuid,
         offset: usize,
         limit: usize,
         include_total: bool,
     ) -> Result<(Vec<Folder>, Option<usize>), DomainError> {
-        let rows: Vec<FolderRowPaginated> = if let Some(pid) = parent_id {
-            sqlx::query_as(
-                r#"
-                SELECT id::text, name, path, parent_id::text, user_id, drive_id,
-                       EXTRACT(EPOCH FROM created_at)::bigint,
-                       EXTRACT(EPOCH FROM updated_at)::bigint,
-                       EXTRACT(EPOCH FROM tree_modified_at)::bigint,
-                       created_by, updated_by,
-                       COUNT(*) OVER() AS total_count
-                  FROM storage.folders
-                 WHERE parent_id = $1::uuid AND user_id = $2 AND NOT is_trashed
-                 ORDER BY name
-                 LIMIT $3 OFFSET $4
-                "#,
-            )
-            .bind(pid)
-            .bind(owner_id)
-            .bind(limit as i64)
-            .bind(offset as i64)
-            .fetch_all(self.pool())
-            .await
-        } else {
-            sqlx::query_as(
-                r#"
-                SELECT id::text, name, path, parent_id::text, user_id, drive_id,
-                       EXTRACT(EPOCH FROM created_at)::bigint,
-                       EXTRACT(EPOCH FROM updated_at)::bigint,
-                       EXTRACT(EPOCH FROM tree_modified_at)::bigint,
-                       created_by, updated_by,
-                       COUNT(*) OVER() AS total_count
-                  FROM storage.folders
-                 WHERE parent_id IS NULL AND user_id = $1 AND NOT is_trashed
-                 ORDER BY name
-                 LIMIT $2 OFFSET $3
-                "#,
-            )
-            .bind(owner_id)
-            .bind(limit as i64)
-            .bind(offset as i64)
-            .fetch_all(self.pool())
-            .await
-        }
-        .map_err(|e| DomainError::internal_error("FolderDb", format!("paginate_by_owner: {e}")))?;
+        let rows: Vec<FolderRowPaginated> = sqlx::query_as(
+            r#"
+            SELECT f.id::text, f.name, f.path, f.parent_id::text, f.user_id, f.drive_id,
+                   EXTRACT(EPOCH FROM f.created_at)::bigint,
+                   EXTRACT(EPOCH FROM f.updated_at)::bigint,
+                   EXTRACT(EPOCH FROM f.tree_modified_at)::bigint,
+                   f.created_by, f.updated_by,
+                   COUNT(*) OVER() AS total_count
+              FROM storage.folders f
+             WHERE f.parent_id IS NULL
+               AND NOT f.is_trashed
+               AND EXISTS (
+                     SELECT 1
+                       FROM storage.role_grants g
+                      WHERE g.resource_type = 'drive'
+                        AND g.resource_id   = f.drive_id
+                        AND (g.expires_at IS NULL OR g.expires_at > NOW())
+                        AND (
+                              (g.subject_type = 'user'  AND g.subject_id = $1)
+                           OR (g.subject_type = 'group' AND g.subject_id IN
+                                   (SELECT storage.caller_group_ids($1)))
+                            )
+                   )
+             ORDER BY f.name
+             LIMIT $2 OFFSET $3
+            "#,
+        )
+        .bind(caller_id)
+        .bind(limit as i64)
+        .bind(offset as i64)
+        .fetch_all(self.pool())
+        .await
+        .map_err(|e| {
+            DomainError::internal_error("FolderDb", format!("list_root_folders_paginated: {e}"))
+        })?;
 
         let total = if include_total {
             Some(rows.first().map_or(0, |r| r.11) as usize)

src/infrastructure/services/pg_acl_engine.rs

@@ -377,10 +377,15 @@ impl PgAclEngine {
 
     /// Public wrapper around `subject_match_set` for callers that need
     /// the expanded `(subject_types, subject_ids)` pair without invoking
-    /// the engine's full `check`/`require` pipeline. Used by
-    /// `GET /api/drives` (and future drive-aware listing surfaces) to
-    /// ask the `DriveRepository` for every drive the caller can read,
-    /// reusing the engine's cached group-expansion logic.
+    /// the engine's full `check`/`require` pipeline.
+    ///
+    /// **Retained for legacy callers only** — new listing queries embed
+    /// the `storage.caller_group_ids` PostgreSQL function inline (see
+    /// migration `20260901000002_caller_group_ids_function.sql`) and
+    /// take a bare `caller_id: Uuid` instead of the pre-expanded arrays.
+    /// The engine's Moka cache still backs the fast path for per-request
+    /// AuthZ decisions (`check_inner`, `drive_role_cache`) where the
+    /// same subject is looked up repeatedly.
     pub async fn expand_subject_for_listing(
         &self,
         subject: Subject,

src/interfaces/api/handlers/drive_handler.rs

@@ -48,23 +48,7 @@ pub async fn list_drives(
 ) -> impl IntoResponse {
     let caller_id = auth_user.id;
 
-    let (subject_types, subject_ids) = match state
-        .authorization
-        .expand_subject_for_listing(Subject::User(caller_id))
-        .await
-    {
-        Ok(pair) => pair,
-        Err(e) => {
-            error!("list_drives: subject expansion failed: {e}");
-            return AppError::from(e).into_response();
-        }
-    };
-
-    match state
-        .drive_repo
-        .list_for_subjects(&subject_types, &subject_ids)
-        .await
-    {
+    match state.drive_repo.list_readable_by(caller_id).await {
         Ok(drives) => {
             let dtos: Vec<DriveDto> = drives.into_iter().map(DriveDto::from).collect();
             (StatusCode::OK, Json(dtos)).into_response()

src/interfaces/api/handlers/photos_handler.rs

@@ -12,8 +12,6 @@ use tracing::{error, info};
 use crate::application::dtos::file_dto::FileDto;
 use crate::application::dtos::geo_dto::GeoBounds;
 use crate::common::di::AppState;
-use crate::domain::services::authorization::Subject;
-use crate::interfaces::errors::AppError;
 use crate::interfaces::middleware::auth::AuthUser;
 
 /// Query parameters for the photos timeline endpoint.
@@ -68,26 +66,10 @@ pub async fn list_photos(
     let caller_id = auth_user.id;
     let limit = params.limit.unwrap_or(200).clamp(1, 500);
 
-    // Expand the caller into (subject_types, subject_ids) so group-mediated
-    // drive memberships surface in the Photos timeline too. Mirrors what
-    // `drive_handler::list_drives` and `trash_service::list_resources_paged`
-    // already do — one call to the AuthZ engine per request.
-    let (subject_types, subject_ids) = match state
-        .authorization
-        .expand_subject_for_listing(Subject::User(caller_id))
-        .await
-    {
-        Ok(pair) => pair,
-        Err(e) => {
-            error!("list_photos: subject expansion failed: {e}");
-            return AppError::from(e).into_response();
-        }
-    };
-
     let file_read = &state.repositories.file_read_repository;
 
     match file_read
-        .list_media_files(&subject_types, &subject_ids, params.before, limit)
+        .list_media_files(caller_id, params.before, limit)
         .await
     {
         Ok((files, sort_dates, dims)) => {