repository.py

import uuid
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any, cast

import sqlalchemy as sa
from sqlalchemy import exc as sa_exc
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession as SASession
from sqlalchemy.orm import contains_eager, selectinload

from ai.backend.common.bgtask.bgtask import BackgroundTaskManager
from ai.backend.common.contexts.user import current_user
from ai.backend.common.exception import BackendAIError
from ai.backend.common.identifier.vfolder import VFolderUUID
from ai.backend.common.metrics.metric import DomainType, LayerType
from ai.backend.common.resilience.policies.metrics import MetricArgs, MetricPolicy
from ai.backend.common.resilience.policies.retry import BackoffStrategy, RetryArgs, RetryPolicy
from ai.backend.common.resilience.resilience import Resilience
from ai.backend.common.types import (
    QuotaScopeID,
    VFolderHostPermission,
    VFolderHostPermissionMap,
    VFolderID,
)
from ai.backend.manager.data.agent.types import AgentStatus
from ai.backend.manager.data.group.types import ProjectResourceInfo
from ai.backend.manager.data.kernel.types import KernelStatus
from ai.backend.manager.data.permission.id import ObjectId, ScopeId
from ai.backend.manager.data.permission.types import (
    EntityType,
    OperationType,
    RBACElementRef,
    RBACElementType,
    RelationType,
    RoleSource,
    ScopeType,
)
from ai.backend.manager.data.vfolder.dto import UserIdentity
from ai.backend.manager.data.vfolder.types import (
    UserWithVFolderHostPermissions,
    ValidatedVFolderInfo,
    VFolderAccessInfo,
    VFolderCreateParams,
    VFolderData,
    VFolderInvitationData,
    VFolderListResult,
    VFolderMountPermission,
    VFolderPermissionData,
    VFolderSearchResult,
)
from ai.backend.manager.errors.api import InvalidAPIParameters
from ai.backend.manager.errors.auth import AuthorizationFailed
from ai.backend.manager.errors.common import ObjectNotFound
from ai.backend.manager.errors.repository import (
    ForeignKeyViolationError,
    RepositoryIntegrityError,
)
from ai.backend.manager.errors.resource import ProjectNotFound
from ai.backend.manager.errors.storage import (
    InsufficientStoragePermission,
    VFolderDeletionNotAllowed,
    VFolderFilterStatusFailed,
    VFolderHasLinkedModelCard,
    VFolderInvalidParameter,
    VFolderNotFound,
    VFolderOperationFailed,
)
from ai.backend.manager.errors.user import UserNotFound
from ai.backend.manager.models.agent import agents
from ai.backend.manager.models.group import GroupRow
from ai.backend.manager.models.kernel import kernels
from ai.backend.manager.models.keypair import KeyPairRow, keypairs
from ai.backend.manager.models.model_card.row import ModelCardRow
from ai.backend.manager.models.rbac_models.association_scopes_entities import (
    AssociationScopesEntitiesRow,
)
from ai.backend.manager.models.rbac_models.permission.permission import PermissionRow
from ai.backend.manager.models.rbac_models.role import RoleRow
from ai.backend.manager.models.rbac_models.user_role import UserRoleRow
from ai.backend.manager.models.resource_policy import keypair_resource_policies
from ai.backend.manager.models.storage import StorageSessionManager
from ai.backend.manager.models.user import (
    ACTIVE_USER_STATUSES,
    UserRole,
    UserRow,
    UserStatus,
    users,
)
from ai.backend.manager.models.utils import ExtendedAsyncSAEngine, execute_with_retry
from ai.backend.manager.models.vfolder import (
    HARD_DELETED_VFOLDER_STATUSES,
    VFolderCloneInfo,
    VFolderDeletionInfo,
    VFolderInvitationRow,
    VFolderInvitationState,
    VFolderOperationStatus,
    VFolderOwnershipType,
    VFolderPermission,
    VFolderPermissionRow,
    VFolderPermissionSetAlias,
    VFolderRow,
    VFolderStatusSet,
    delete_vfolder_relation_rows,
    ensure_host_permission_allowed,
    ensure_quota_scope_accessible_by_user,
    get_allowed_vfolder_hosts_by_group,
    get_allowed_vfolder_hosts_by_user,
    get_sessions_by_mounted_folder,
    is_unmanaged,
    query_accessible_vfolders,
    vfolder_invitations,
    vfolder_permissions,
    vfolder_status_map,
    vfolders,
)
from ai.backend.manager.models.vfolder.conditions import VFolderConditions
from ai.backend.manager.repositories.base import (
    BatchQuerier,
    IntegrityErrorCheck,
    execute_batch_querier,
)
from ai.backend.manager.repositories.base.integrity import match_integrity_error
from ai.backend.manager.repositories.base.rbac.entity_creator import (
    RBACEntityCreator,
    execute_rbac_entity_creator,
)
from ai.backend.manager.repositories.base.rbac.entity_purger import (
    RBACEntityPurger,
    execute_rbac_entity_purger,
)
from ai.backend.manager.repositories.base.rbac.granter import (
    RBACGranter,
    execute_rbac_granter,
)
from ai.backend.manager.repositories.base.rbac.revoker import (
    RBACRevoker,
    execute_rbac_revoker,
)
from ai.backend.manager.repositories.base.updater import Updater, execute_updater
from ai.backend.manager.repositories.permission_controller.role_manager import (
    RoleManager,
    UserSystemRoleSpec,
)

from ai.backend.manager.repositories.vfolder.creators import VFolderCreatorSpec
from ai.backend.manager.repositories.vfolder.types import (
    BulkVFolderPurgeResult,
    ProjectVFolderSearchScope,
    UserVFolderSearchScope,
    VFolderPurgeFailure,
)

vfolder_repository_resilience = Resilience(
    policies=[
        MetricPolicy(MetricArgs(domain=DomainType.REPOSITORY, layer=LayerType.VFOLDER_REPOSITORY)),
        RetryPolicy(
            RetryArgs(
                max_retries=10,
                retry_delay=0.1,
                backoff_strategy=BackoffStrategy.FIXED,
                non_retryable_exceptions=(BackendAIError,),
            )
        ),
    ]
)


@dataclass
class _VFolderWithLinkedModelCards:
    """A vfolder row paired with model card rows referencing it."""

    vfolder_row: VFolderRow
    model_card_rows: list[ModelCardRow]


class VfolderRepository:
    _db: ExtendedAsyncSAEngine
    _role_manager: RoleManager

    def __init__(self, db: ExtendedAsyncSAEngine) -> None:
        self._db = db
        self._role_manager = RoleManager()

    @vfolder_repository_resilience.apply()
    async def get_by_id_validated(
        self, vfolder_id: uuid.UUID, user_id: uuid.UUID, domain_name: str | None
    ) -> VFolderData:
        """
        Get a VFolder by ID with ownership/permission validation.
        Returns VFolderData if user has access.
        Raises VFolderNotFound if vfolder doesn't exist or user has no access.
        """
        async with self._db.begin_readonly_session() as session:
            vfolder_row = await self._get_vfolder_by_id(session, vfolder_id)
            if not vfolder_row:
                raise VFolderNotFound()

            if vfolder_row.user == user_id:
                return self._vfolder_row_to_data(vfolder_row)

            # Check access permissions
            user_row = await session.scalar(sa.select(UserRow).where(UserRow.uuid == user_id))
            if not user_row:
                raise ObjectNotFound(object_name="User")

            # Check if user has access to this vfolder
            allowed_vfolder_types = ["user", "group"]  # TODO: get from config
            conn = await session.connection()
            vfolder_dicts = await query_accessible_vfolders(
                conn,
                user_id,
                allow_privileged_access=True,
                user_role=user_row.role,
                domain_name=domain_name,
                allowed_vfolder_types=allowed_vfolder_types,
                extra_vf_conds=(VFolderRow.id == vfolder_id),
            )

            if not vfolder_dicts:
                raise VFolderNotFound()

            return self._vfolder_row_to_data(vfolder_row)

    @vfolder_repository_resilience.apply()
    async def get_by_id(self, vfolder_id: uuid.UUID) -> VFolderData:
        """
        Get a VFolder by ID without validation.
        Returns VFolderData if found, None otherwise.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            vfolder_row = await self._get_vfolder_by_id(session, vfolder_id)
            if not vfolder_row:
                raise VFolderNotFound()
            return self._vfolder_row_to_data(vfolder_row)

    @vfolder_repository_resilience.apply()
    async def get_row_by_id(self, vfolder_id: VFolderUUID) -> Mapping[str, Any]:
        """
        Fetch a vfolder row as a plain mapping by UUID, without permission filtering.

        For use only by the REST middleware; do not call from new code.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            vfolder_row = await self._get_vfolder_by_id(session, vfolder_id)
            if not vfolder_row:
                raise VFolderNotFound(extra_data=str(vfolder_id))
            return {
                "name": vfolder_row.name,
                "id": vfolder_row.id,
                "host": vfolder_row.host,
                "quota_scope_id": vfolder_row.quota_scope_id,
                "domain_name": vfolder_row.domain_name,
                "usage_mode": vfolder_row.usage_mode,
                "created_at": vfolder_row.created_at,
                "last_used": vfolder_row.last_used,
                "max_size": vfolder_row.max_size,
                "max_files": vfolder_row.max_files,
                "ownership_type": vfolder_row.ownership_type,
                "user": str(vfolder_row.user) if vfolder_row.user else None,
                "group": str(vfolder_row.group) if vfolder_row.group else None,
                "creator": vfolder_row.creator,
                "creator_id": vfolder_row.creator_id,
                "user_email": None,
                "group_name": None,
                "is_owner": False,
                "permission": vfolder_row.permission,
                "unmanaged_path": vfolder_row.unmanaged_path,
                "cloneable": vfolder_row.cloneable,
                "status": vfolder_row.status,
                "cur_size": vfolder_row.cur_size,
            }

    @vfolder_repository_resilience.apply()
    async def batch_load_by_ids(self, ids: Sequence[uuid.UUID]) -> list[VFolderData | None]:
        """
        Batch fetch vfolders by IDs without permission validation.

        Returns a list with the same length and order as the input ids;
        entries that are not found are ``None``. Intended for GraphQL
        DataLoader use where the caller has already authorized access to a
        parent entity that references these vfolder IDs.
        """
        if not ids:
            return []
        async with self._db.begin_readonly_session() as session:
            query = sa.select(VFolderRow).where(VFolderConditions.by_ids(ids)())
            result = await session.execute(query)
            rows_by_id = {row.id: self._vfolder_row_to_data(row) for row in result.scalars().all()}
            return [rows_by_id.get(VFolderUUID(vfolder_id)) for vfolder_id in ids]

    @vfolder_repository_resilience.apply()
    async def get_allowed_vfolder_hosts(
        self, user_uuid: uuid.UUID, group_uuid: uuid.UUID | None
    ) -> VFolderHostPermissionMap:
        """
        Get the allowed VFolder hosts for a user.
        """
        async with self._db.begin_readonly_session_read_committed() as db_session:
            if group_uuid:
                group_row: GroupRow | None = await db_session.scalar(
                    sa.select(GroupRow).where(GroupRow.id == group_uuid)
                )
                if group_row is None:
                    raise ProjectNotFound(f"Project with {group_uuid} not found.")

                return group_row.allowed_vfolder_hosts

            user_row: UserRow | None = await db_session.scalar(
                sa.select(UserRow)
                .where(UserRow.uuid == user_uuid)
                .options(
                    selectinload(UserRow.main_keypair).selectinload(KeyPairRow.resource_policy_row)
                )
            )
            if user_row is None:
                raise UserNotFound(f"User with UUID {user_uuid} not found.")
            if user_row.main_keypair is None:
                raise ObjectNotFound(object_name="User keypair")
            if user_row.main_keypair.resource_policy_row is None:
                raise ObjectNotFound(object_name="User keypair resource policy")

            return user_row.main_keypair.resource_policy_row.allowed_vfolder_hosts

    @vfolder_repository_resilience.apply()
    async def get_user_with_keypair_policy_vfolder_hosts(
        self, user_uuid: uuid.UUID
    ) -> UserWithVFolderHostPermissions:
        """
        Load user data together with the merged ``allowed_vfolder_hosts`` from
        all of the user's active keypair resource policies.

        A user can hold multiple keypairs, each pointing to its own keypair
        resource policy. This method unions ``allowed_vfolder_hosts`` across
        every active keypair so that the host-permission check reflects the
        full set of hosts available to the user (rather than only the main
        keypair).

        Implementation note: a single LEFT OUTER JOIN is used (instead of an
        ORM ``selectinload`` on ``UserRow.keypairs``) so that only the columns
        needed for the host-permission check are loaded. This avoids loading
        sensitive keypair columns such as ``secret_key`` and
        ``ssh_private_key``.
        """
        async with self._db.begin_readonly_session_read_committed() as db_session:
            stmt = (
                sa.select(
                    UserRow.email,
                    UserRow.role,
                    UserRow.domain_name,
                    keypair_resource_policies.c.allowed_vfolder_hosts,
                )
                .select_from(UserRow)
                .outerjoin(
                    KeyPairRow,
                    sa.and_(
                        KeyPairRow.user == UserRow.uuid,
                        KeyPairRow.is_active.is_(True),
                    ),
                )
                .outerjoin(
                    keypair_resource_policies,
                    keypair_resource_policies.c.name == KeyPairRow.resource_policy,
                )
                .where(UserRow.uuid == user_uuid)
            )
            rows = (await db_session.execute(stmt)).all()
            if not rows:
                raise UserNotFound(f"User with UUID {user_uuid} not found.")
            email, role, domain_name, _ = rows[0]
            if role is None or domain_name is None:
                raise UserNotFound(f"User with UUID {user_uuid} has invalid role or domain data.")
            merged_hosts = VFolderHostPermissionMap()
            for row in rows:
                policy_hosts = row.allowed_vfolder_hosts
                if policy_hosts is None:
                    continue
                merged_hosts = VFolderHostPermissionMap(merged_hosts | policy_hosts)
            return UserWithVFolderHostPermissions(
                email=email,
                role=role,
                allowed_vfolder_hosts=merged_hosts,
            )

    @vfolder_repository_resilience.apply()
    async def get_max_vfolder_count(
        self, user_uuid: uuid.UUID, group_uuid: uuid.UUID | None
    ) -> int:
        """
        Get the maximum VFolder count for a user or group.
        """
        async with self._db.begin_readonly_session_read_committed() as db_session:
            if group_uuid:
                group_row: GroupRow | None = await db_session.scalar(
                    sa.select(GroupRow)
                    .where(GroupRow.id == group_uuid)
                    .options(selectinload(GroupRow.resource_policy_row))
                )
                if group_row is None:
                    raise ProjectNotFound(f"Project with {group_uuid} not found.")

                return group_row.resource_policy_row.max_vfolder_count

            user_row: UserRow | None = await db_session.scalar(
                sa.select(UserRow)
                .where(UserRow.uuid == user_uuid)
                .options(selectinload(UserRow.resource_policy_row))
            )
            if user_row is None:
                raise UserNotFound(f"User with UUID {user_uuid} not found.")

            return user_row.resource_policy_row.max_vfolder_count

    @vfolder_repository_resilience.apply()
    async def list_accessible_vfolders(
        self,
        user_id: uuid.UUID,
        user_role: UserRole,
        domain_name: str,
        allowed_vfolder_types: list[str],
        extra_conditions: sa.sql.elements.ColumnElement[bool] | None = None,
    ) -> VFolderListResult:
        """
        List all VFolders accessible to a user.
        Returns VFolderListResult with access information.
        """
        async with self._db.begin_readonly_session() as session:
            conn = await session.connection()
            vfolder_dicts = await query_accessible_vfolders(
                conn,
                user_id,
                user_role=user_role,
                domain_name=domain_name,
                allowed_vfolder_types=allowed_vfolder_types,
                extra_vf_conds=extra_conditions,
            )

            vfolder_access_infos = []
            for vfolder_dict in vfolder_dicts:
                vfolder_data = self._vfolder_dict_to_data(dict(vfolder_dict))
                is_owner = vfolder_dict.get("is_owner", False)
                permission = vfolder_dict.get("permission", VFolderPermission.READ_ONLY)

                vfolder_access_infos.append(
                    VFolderAccessInfo(
                        vfolder_data=vfolder_data,
                        is_owner=is_owner,
                        effective_permission=permission,
                    )
                )

            return VFolderListResult(vfolders=vfolder_access_infos)

    @vfolder_repository_resilience.apply()
    async def resolve_vfolder_ids_by_names(self, names: Sequence[str]) -> dict[str, uuid.UUID]:
        """Look up vfolder UUIDs for a batch of names in a single query.

        Returns only the names that match an existing row; the caller
        decides what to do with any missing names (typically raise).

        No access scoping or status filtering — callers (e.g. session-create
        paths) are responsible for validating both user access and lifecycle
        state of the resolved ids in their own downstream flow.
        """
        if not names:
            return {}
        async with self._db.begin_readonly_session() as session:
            rows = (
                await session.execute(
                    sa.select(VFolderRow.name, VFolderRow.id).where(
                        VFolderRow.name.in_(list(names))
                    )
                )
            ).all()
        return {row.name: cast(uuid.UUID, row.id) for row in rows}

    @vfolder_repository_resilience.apply()
    async def create_vfolder_with_permission(
        self, params: VFolderCreateParams, create_owner_permission: bool = False
    ) -> VFolderData:
        """
        Create a new VFolder with the given parameters and optionally create owner permission.
        Returns the created VFolderData.
        """
        async with self._db.begin_session() as session:
            # Determine scope based on ownership type
            element_type: RBACElementType
            scope_id: str
            match params.ownership_type:
                case VFolderOwnershipType.USER:
                    element_type = RBACElementType.USER
                    scope_id = str(params.user)
                case VFolderOwnershipType.GROUP:
                    element_type = RBACElementType.PROJECT
                    scope_id = str(params.group)

            # Create VFolderCreatorSpec from params
            spec = VFolderCreatorSpec(
                id=params.id,
                name=params.name,
                domain_name=params.domain_name,
                quota_scope_id=params.quota_scope_id,
                usage_mode=params.usage_mode,
                permission=params.permission,
                host=params.host,
                creator=params.creator,
                creator_id=params.creator_id,
                ownership_type=params.ownership_type,
                user=params.user,
                group=params.group,
                unmanaged_path=params.unmanaged_path,
                cloneable=params.cloneable,
                status=params.status,
            )

            # Use RBACEntityCreator for atomic entity + scope association creation
            rbac_creator = RBACEntityCreator(
                spec=spec,
                element_type=RBACElementType.VFOLDER,
                scope_ref=RBACElementRef(element_type=element_type, element_id=scope_id),
                additional_scope_refs=[],
            )
            result = await execute_rbac_entity_creator(session, rbac_creator)
            created_row = result.row

            # Create owner permission if requested (legacy compatibility)
            if create_owner_permission and params.user:
                # Get user's role_id
                user_role_id = await self._get_user_role_id(session, params.user)

                # Insert VFolderPermissionRow for legacy compatibility
                permission_insert = sa.insert(VFolderPermissionRow).values({
                    "user": params.user,
                    "vfolder": params.id.hex,
                    "permission": VFolderPermission.OWNER_PERM,
                })
                await session.execute(permission_insert)

                # Add permission to user's role using RBACGranter (entity-as-scope)
                granter = RBACGranter(
                    granted_entity_id=ObjectId(
                        entity_type=EntityType.VFOLDER,
                        entity_id=str(params.id),
                    ),
                    granted_entity_scope_type=RBACElementType.VFOLDER,
                    target_scope_id=ScopeId(
                        scope_type=ScopeType.USER,
                        scope_id=str(params.user),
                    ),
                    target_role_ids=[user_role_id],
                    operations=[OperationType.READ],
                )
                await execute_rbac_granter(session, granter)

            return created_row.to_data()

    @vfolder_repository_resilience.apply()
    async def trash_vfolder(self, updater: Updater[VFolderRow]) -> VFolderData:
        """Soft-delete a single vfolder by setting its status to DELETE_PENDING.

        Checks that no active sessions are mounting the vfolder before
        proceeding. Uses ``execute_updater`` with PK-based WHERE; returns the
        updated row or raises ``VFolderNotFound`` if no matching row exists.
        """
        async with self._db.begin_session() as session:
            # Pre-check: reject if any session is currently mounting this vfolder
            vfolder_row = await self._get_vfolder_by_id(session, uuid.UUID(str(updater.pk_value)))
            if vfolder_row is None:
                raise VFolderNotFound()
            mount_sessions = await get_sessions_by_mounted_folder(
                session, VFolderID.from_row(vfolder_row)
            )
            if mount_sessions:
                session_ids = [str(sid) for sid in mount_sessions]
                raise VFolderDeletionNotAllowed(
                    "Cannot delete the vfolder. "
                    f"The vfolder(id: {vfolder_row.id}) is mounted on sessions(ids: {session_ids})."
                )
            # Expire the pre-loaded ORM identity so execute_updater's
            # RETURNING clause produces a fresh row with updated status.
            await session.refresh(vfolder_row)
            session.expunge(vfolder_row)
            result = await execute_updater(session, updater)
            if result is None:
                raise VFolderNotFound()
            return self._vfolder_row_to_data(result.row)

    @vfolder_repository_resilience.apply()
    async def update_vfolder_attribute(self, updater: Updater[VFolderRow]) -> VFolderData:
        """
        Update VFolder attributes.
        Returns updated VFolderData.
        """
        async with self._db.begin_session() as session:
            result = await execute_updater(session, updater)
            if result is None:
                raise VFolderNotFound()
            return self._vfolder_row_to_data(result.row)

    @vfolder_repository_resilience.apply()
    async def move_vfolders_to_trash(self, vfolder_ids: list[uuid.UUID]) -> list[VFolderData]:
        """
        Move VFolders to trash.
        """

        async with self._db.begin_session() as session:
            vfolder_rows = []
            for vfolder_id in vfolder_ids:
                vfolder_row = await self._get_vfolder_by_id(session, vfolder_id)
                if vfolder_row:
                    vfolder_rows.append(vfolder_row)

            # Create deletion info for each vfolder
            deletion_infos = []
            for vfolder_row in vfolder_rows:
                vfolder_id_obj = VFolderID(
                    quota_scope_id=vfolder_row.quota_scope_id,
                    folder_id=vfolder_row.id,
                )
                deletion_info = VFolderDeletionInfo(
                    vfolder_id=vfolder_id_obj,
                    host=vfolder_row.host,
                    unmanaged_path=vfolder_row.unmanaged_path,
                )
                deletion_infos.append(deletion_info)

            # Note: initiate_vfolder_deletion requires storage_manager parameter
            # This would need to be passed to the repository method or handled differently
            # For now, we'll update the status directly instead of using the full deletion process
            for vfolder_row in vfolder_rows:
                mount_sessions = await get_sessions_by_mounted_folder(
                    session, VFolderID.from_row(vfolder_row)
                )
                if mount_sessions:
                    session_ids = [str(session_id) for session_id in mount_sessions]
                    raise VFolderDeletionNotAllowed(
                        "Cannot delete the vfolder. "
                        f"The vfolder(id: {vfolder_row.id}) is mounted on sessions(ids: {session_ids})."
                    )
                vfolder_row.status = VFolderOperationStatus.DELETE_PENDING

            await session.flush()
            for row in vfolder_rows:
                await session.refresh(row, attribute_names=["updated_at"])

            return [self._vfolder_row_to_data(row) for row in vfolder_rows]

    @vfolder_repository_resilience.apply()
    async def restore_vfolders_from_trash(self, vfolder_ids: list[uuid.UUID]) -> list[VFolderData]:
        """
        Restore VFolders from trash.
        """
        async with self._db.begin_session() as session:
            vfolder_rows = []
            for vfolder_id in vfolder_ids:
                vfolder_row = await self._get_vfolder_by_id(session, vfolder_id)
                if vfolder_row:
                    vfolder_row.status = VFolderOperationStatus.READY
                    vfolder_rows.append(vfolder_row)

            await session.flush()
            for row in vfolder_rows:
                await session.refresh(row, attribute_names=["updated_at"])
            return [self._vfolder_row_to_data(row) for row in vfolder_rows]

    async def _fetch_vfolders_with_linked_model_cards(
        self,
        session: SASession,
        vfolder_ids: Sequence[uuid.UUID],
    ) -> list[_VFolderWithLinkedModelCards]:
        """Fetch vfolder rows together with model card rows that reference them.

        Uses a single LEFT OUTER JOIN. Vfolders with no model card appear
        once with an empty ``model_card_rows`` list; vfolder IDs not present
        in the database are skipped.
        """
        if not vfolder_ids:
            return []

        stmt = (
            sa.select(VFolderRow, ModelCardRow)
            .outerjoin(ModelCardRow, ModelCardRow.vfolder == VFolderRow.id)
            .where(VFolderRow.id.in_(vfolder_ids))
        )
        rows = (await session.execute(stmt)).all()

        grouped: dict[uuid.UUID, _VFolderWithLinkedModelCards] = {}
        for vfolder_row, card_row in rows:
            record = grouped.setdefault(
                vfolder_row.id,
                _VFolderWithLinkedModelCards(
                    vfolder_row=vfolder_row,
                    model_card_rows=[],
                ),
            )
            if card_row is not None:
                record.model_card_rows.append(card_row)
        return list(grouped.values())

    @vfolder_repository_resilience.apply()
    async def delete_vfolders_forever(
        self,
        vfolder_ids: list[uuid.UUID],
        *,
        cascade_model_card: bool = False,
    ) -> BulkVFolderPurgeResult:
        """
        Delete VFolders forever with partial-success semantics.

        Each vfolder is processed independently in the same transaction:
        a vfolder with linked model card(s) becomes a failure (carrying
        ``VFolderHasLinkedModelCard``) when ``cascade_model_card`` is False;
        otherwise the cards are deleted and the vfolder transitions to
        ``DELETE_ONGOING`` like any other success.
        """

        result = BulkVFolderPurgeResult()
        async with self._db.connect() as db_conn:
            async with self._db.begin_session(db_conn) as db_session:
                records = await self._fetch_vfolders_with_linked_model_cards(
                    db_session, vfolder_ids
                )
                cards_to_delete: list[ModelCardRow] = []
                succeeded_ids: list[uuid.UUID] = []
                succeeded_rows: list[VFolderRow] = []
                for rec in records:
                    if rec.model_card_rows and not cascade_model_card:
                        result.failures.append(
                            VFolderPurgeFailure(
                                vfolder_id=rec.vfolder_row.id,
                                exception=VFolderHasLinkedModelCard(
                                    f"VFolder {rec.vfolder_row.id} is referenced by "
                                    f"{len(rec.model_card_rows)} model card(s); "
                                    "delete the model card(s) first or set "
                                    "cascade_model_card=True."
                                ),
                            )
                        )
                        continue
                    cards_to_delete.extend(rec.model_card_rows)
                    succeeded_ids.append(rec.vfolder_row.id)
                    succeeded_rows.append(rec.vfolder_row)

                if cards_to_delete:
                    for card in cards_to_delete:
                        await db_session.delete(card)
                    await db_session.flush()

                if succeeded_ids:
                    delete_stmt = (
                        sa.update(VFolderRow)
                        .where(VFolderRow.id.in_(succeeded_ids))
                        .values(status=VFolderOperationStatus.DELETE_ONGOING)
                    )
                    await db_session.execute(delete_stmt)
                    # ``onupdate=now()`` on ``updated_at`` expires the
                    # column on the in-memory rows after the UPDATE;
                    # explicitly refresh so the subsequent conversion
                    # does not need a lazy SELECT (which would fail
                    # outside the greenlet bridge).
                    for row in succeeded_rows:
                        await db_session.refresh(row, attribute_names=["updated_at"])

                succeeded_data = [self._vfolder_row_to_data(row) for row in succeeded_rows]

            if succeeded_ids:
                # Delete relation rows for succeeded vfolders only.
                await delete_vfolder_relation_rows(db_conn, self._db.begin_session, succeeded_ids)

            result.succeeded = succeeded_data
            return result

    @vfolder_repository_resilience.apply()
    async def purge_vfolder(self, purger: RBACEntityPurger[VFolderRow]) -> VFolderData:
        """
        Permanently delete a VFolder from DB.
        Only VFolders with purgable status (DELETE_PENDING, DELETE_COMPLETE) can be purged.

        Raises:
            VFolderNotFound: If the vfolder doesn't exist.
            VFolderFilterStatusFailed: If the vfolder status is not purgable.
            VFolderHasLinkedModelCard: If a model card still references the vfolder.
        """
        vfolder_uuid = cast(uuid.UUID, purger.pk_value)
        async with self._db.begin_session() as session:
            # Fetch vfolder first to validate status before purging.
            vfolder_row = await self._get_vfolder_by_id(session, vfolder_uuid)
            if vfolder_row is None:
                raise VFolderNotFound(extra_data=str(vfolder_uuid))
            if vfolder_row.status not in vfolder_status_map[VFolderStatusSet.PURGABLE]:
                raise VFolderFilterStatusFailed
            try:
                await execute_rbac_entity_purger(session, purger)
            except RepositoryIntegrityError as e:
                match_integrity_error(
                    e,
                    [
                        IntegrityErrorCheck(
                            violation_type=ForeignKeyViolationError,
                            constraint_name="fk_model_cards_vfolder_vfolders",
                            error=VFolderHasLinkedModelCard(
                                f"VFolder {vfolder_uuid} cannot be purged: it is "
                                "still referenced by one or more model card(s). "
                                "Run delete-forever (with cascade if needed) "
                                "before purge."
                            ),
                        ),
                    ],
                )
            return vfolder_row.to_data()

    @vfolder_repository_resilience.apply()
    async def get_vfolder_permissions(self, vfolder_id: uuid.UUID) -> list[VFolderPermissionData]:
        """
        Get all permissions for a VFolder.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            query = sa.select(VFolderPermissionRow).where(
                VFolderPermissionRow.vfolder == vfolder_id
            )
            result = await session.execute(query)
            permission_rows = result.scalars().all()

            return [
                VFolderPermissionData(
                    id=row.id,
                    vfolder=row.vfolder,
                    user=row.user,
                    permission=row.permission or VFolderMountPermission.READ_ONLY,
                )
                for row in permission_rows
            ]

    @vfolder_repository_resilience.apply()
    async def create_vfolder_permission(
        self,
        vfolder_id: uuid.UUID,
        user_id: uuid.UUID,
        permission: VFolderPermission,
    ) -> VFolderPermissionData:
        """
        Create a VFolder permission entry.
        """
        async with self._db.begin_session() as session:
            # Verify vfolder exists
            vfolder_row = await self._get_vfolder_by_id(session, vfolder_id)
            if vfolder_row is None:
                raise VFolderNotFound()
            # Get user's role_id
            user_role_id = await self._get_user_role_id(session, user_id)

            # Insert VFolderPermissionRow (legacy compatibility)
            permission_id = uuid.uuid4()
            insert_values = {
                "id": permission_id,
                "vfolder": vfolder_id,
                "user": user_id,
                "permission": permission,
            }
            query = sa.insert(VFolderPermissionRow).values(insert_values)
            await session.execute(query)

            # Grant permission to user's role using RBACGranter (entity-as-scope)
            granter = RBACGranter(
                granted_entity_id=ObjectId(
                    entity_type=EntityType.VFOLDER,
                    entity_id=str(vfolder_id),
                ),
                granted_entity_scope_type=RBACElementType.VFOLDER,
                target_scope_id=ScopeId(
                    scope_type=ScopeType.USER,
                    scope_id=str(user_id),
                ),
                target_role_ids=[user_role_id],
                operations=list(permission.to_rbac_operation()),
            )
            await execute_rbac_granter(session, granter)

            return VFolderPermissionData(
                id=permission_id,
                vfolder=vfolder_id,
                user=user_id,
                permission=permission,
            )

    @vfolder_repository_resilience.apply()
    async def delete_vfolder_permission(self, vfolder_id: uuid.UUID, user_id: uuid.UUID) -> None:
        """
        Delete a VFolder permission entry.
        """
        async with self._db.begin_session() as session:
            # Get user's role_id
            user_role_id = await self._get_user_role_id(session, user_id)

            # Delete VFolderPermissionRow (legacy compatibility)
            query = sa.delete(VFolderPermissionRow).where(
                (VFolderPermissionRow.vfolder == vfolder_id)
                & (VFolderPermissionRow.user == user_id)
            )
            await session.execute(query)

            # Revoke permission from user's role using RBACRevoker
            revoker = RBACRevoker(
                entity_id=ObjectId(
                    entity_type=EntityType.VFOLDER,
                    entity_id=str(vfolder_id),
                ),
                entity_scope_type=RBACElementType.VFOLDER,
                target_role_ids=[user_role_id],
                operations=None,  # Revoke all operations
            )
            await execute_rbac_revoker(session, revoker)

    @vfolder_repository_resilience.apply()
    async def get_vfolder_invitations_by_vfolder(
        self, vfolder_id: uuid.UUID
    ) -> list[VFolderInvitationData]:
        """
        Get all invitations for a VFolder.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            query = (
                sa.select(VFolderInvitationRow, UserRow.username)
                .outerjoin(UserRow, UserRow.email == VFolderInvitationRow.inviter)
                .where(VFolderInvitationRow.vfolder == vfolder_id)
            )
            result = await session.execute(query)
            rows = result.all()

            return [
                VFolderInvitationData(
                    id=row.id,
                    vfolder=row.vfolder,
                    inviter=row.inviter or "",
                    inviter_username=inviter_username,
                    invitee=row.invitee,
                    permission=row.permission or VFolderMountPermission.READ_ONLY,
                    created_at=row.created_at or datetime.now(UTC),
                    modified_at=row.modified_at,
                )
                for row, inviter_username in rows
            ]

    @vfolder_repository_resilience.apply()
    async def count_vfolders_by_user(self, user_id: uuid.UUID) -> int:
        """
        Count VFolders owned by a user (excluding hard deleted ones).
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            query = (
                sa.select(sa.func.count())
                .select_from(VFolderRow)
                .where(
                    (VFolderRow.user == user_id)
                    & (VFolderRow.status.not_in(HARD_DELETED_VFOLDER_STATUSES))
                )
            )
            result = await session.scalar(query)
            return result or 0

    @vfolder_repository_resilience.apply()
    async def count_vfolders_by_group(self, group_id: uuid.UUID) -> int:
        """
        Count VFolders owned by a group (excluding hard deleted ones).
        """

        async with self._db.begin_readonly_session_read_committed() as session:
            query = (
                sa.select(sa.func.count())
                .select_from(VFolderRow)
                .where(
                    (VFolderRow.group == group_id)
                    & (VFolderRow.status.not_in(HARD_DELETED_VFOLDER_STATUSES))
                )
            )
            result = await session.scalar(query)
            return result or 0

    @vfolder_repository_resilience.apply()
    async def check_vfolder_name_exists(
        self,
        name: str,
        user_id: uuid.UUID,
        user_role: UserRole,
        domain_name: str,
        allowed_vfolder_types: list[str],
    ) -> bool:
        """
        Check if a VFolder with the given name already exists for the user.
        """

        async with self._db.begin_readonly_session() as session:
            # Use query_accessible_vfolders to check accessible folders
            extra_vf_conds = sa.and_(
                (VFolderRow.name == name),
                (VFolderRow.status.not_in(HARD_DELETED_VFOLDER_STATUSES)),
            )

            conn = await session.connection()
            vfolder_dicts = await query_accessible_vfolders(
                conn,
                user_id,
                user_role=user_role,
                domain_name=domain_name,
                allowed_vfolder_types=allowed_vfolder_types,
                extra_vf_conds=extra_vf_conds,
            )

            return len(vfolder_dicts) > 0

    @vfolder_repository_resilience.apply()
    async def get_user_info(self, user_id: uuid.UUID) -> tuple[UserRole, str] | None:
        """
        Get user role and domain name for a user.
        Returns (role, domain_name) or None if user not found.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            user_row = await session.scalar(sa.select(UserRow).where(UserRow.uuid == user_id))
            if not user_row:
                return None
            if user_row.role is None or user_row.domain_name is None:
                return None
            return user_row.role, user_row.domain_name

    @vfolder_repository_resilience.apply()
    async def get_user_email_by_id(self, user_id: uuid.UUID) -> str | None:
        """
        Get user email by user ID.
        Returns email or None if user not found.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            user_row = await session.scalar(sa.select(UserRow).where(UserRow.uuid == user_id))
            if not user_row:
                return None
            return user_row.email

    @vfolder_repository_resilience.apply()
    async def validate_quota_scope_access(
        self,
        quota_scope_id: QuotaScopeID,
        user_identity: UserIdentity,
    ) -> None:
        """
        Validate that the user has access to the given quota scope.
        Raises InvalidAPIParameters if the user does not have access.
        """
        user_info: Mapping[str, Any] = {
            "uuid": user_identity.user_uuid,
            "role": user_identity.user_role,
            "domain_name": user_identity.domain_name,
        }
        async with self._db.begin_readonly_session_read_committed() as session:
            await ensure_quota_scope_accessible_by_user(session, quota_scope_id, user_info)

    @vfolder_repository_resilience.apply()
    async def get_users_by_ids(self, user_ids: list[uuid.UUID]) -> list[tuple[uuid.UUID, str]]:
        """
        Get user info for multiple user IDs.
        Returns list of (user_id, email) tuples.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            result = await session.execute(sa.select(UserRow).where(UserRow.uuid.in_(user_ids)))
            user_rows = result.scalars().all()
            return [(row.uuid, row.email) for row in user_rows]

    @vfolder_repository_resilience.apply()
    async def get_group_resource_info(
        self, group_id_or_name: str | uuid.UUID, domain_name: str
    ) -> ProjectResourceInfo | None:
        """Get group resource information by group ID or name."""

        async with self._db.begin_readonly_session_read_committed() as session:
            if isinstance(group_id_or_name, str):
                query = (
                    sa.select(GroupRow)
                    .where(
                        (GroupRow.domain_name == domain_name) & (GroupRow.name == group_id_or_name)
                    )
                    .options(selectinload(GroupRow.resource_policy_row))
                )
            else:  # UUID
                query = (
                    sa.select(GroupRow)
                    .where(
                        (GroupRow.domain_name == domain_name) & (GroupRow.id == group_id_or_name)
                    )
                    .options(selectinload(GroupRow.resource_policy_row))
                )

            result = await session.execute(query)
            group_row = result.scalar()

            if not group_row:
                return None

            return ProjectResourceInfo(
                project_id=group_row.id,
                max_vfolder_count=group_row.resource_policy_row.max_vfolder_count,
                max_quota_scope_size=group_row.resource_policy_row.max_quota_scope_size,
                project_type=group_row.type,
            )

    @vfolder_repository_resilience.apply()
    async def get_user_resource_info(
        self, user_id: uuid.UUID
    ) -> tuple[int, int, int | None] | None:
        """
        Get user resource information.
        Returns (max_vfolder_count, max_quota_scope_size, container_uid) or None.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            query = (
                sa.select(UserRow)
                .where(UserRow.uuid == user_id)
                .options(selectinload(UserRow.resource_policy_row))
            )
            result = await session.execute(query)
            user_row = result.scalar()

            if not user_row:
                return None

            return (
                user_row.resource_policy_row.max_vfolder_count,
                user_row.resource_policy_row.max_quota_scope_size,
                user_row.container_uid,
            )

    async def _get_vfolder_by_id(
        self, session: SASession, vfolder_id: uuid.UUID
    ) -> VFolderRow | None:
        """
        Private method to get a VFolder by ID using an existing session.
        """
        query = sa.select(VFolderRow).where(VFolderRow.id == vfolder_id)
        result = await session.execute(query)
        return result.scalar()

    async def _get_user_role_id(self, session: SASession, user_id: uuid.UUID) -> uuid.UUID:
        """
        Get the system role_id associated with a user.

        Looks up the UserRoleRow joined with RoleRow where the role source is SYSTEM.
        If no SYSTEM role exists for the user, creates one as a fallback within the
        same transaction and returns its role_id.
        """
        stmt = (
            sa.select(UserRoleRow.role_id)
            .join(RoleRow, UserRoleRow.role_id == RoleRow.id)
            .where(
                sa.and_(
                    UserRoleRow.user_id == user_id,
                    RoleRow.source == RoleSource.SYSTEM,
                )
            )
        )
        result = await session.scalar(stmt)
        if result is not None:
            return result

        # Fallback: create a system role for this user and return its id
        spec = UserSystemRoleSpec(user_id=user_id)
        role = await self._role_manager.create_system_role(session, spec)
        return role.id

    def _get_vfolder_scope(self, vfolder: VFolderData) -> ScopeId:
        """Determine scope from vfolder ownership."""
        if vfolder.ownership_type == VFolderOwnershipType.USER:
            return ScopeId(ScopeType.USER, str(vfolder.user))
        # GROUP ownership
        return ScopeId(ScopeType.PROJECT, str(vfolder.group))

    async def _validate_vfolder_ownership(
        self, session: SASession, vfolder_id: uuid.UUID, user_id: uuid.UUID
    ) -> VFolderRow:
        """
        Private method to validate VFolder ownership.
        Raises VFolderNotFound if vfolder doesn't exist or user doesn't own it.
        """
        vfolder_row = await self._get_vfolder_by_id(session, vfolder_id)
        if not vfolder_row:
            raise VFolderNotFound()

        # Check ownership
        is_owner = False
        if vfolder_row.ownership_type == VFolderOwnershipType.USER:
            is_owner = vfolder_row.user == user_id
        elif vfolder_row.ownership_type == VFolderOwnershipType.GROUP:
            # TODO: check group membership
            pass

        if not is_owner:
            raise VFolderNotFound()

        return vfolder_row

    def _vfolder_row_to_data(self, row: VFolderRow) -> VFolderData:
        """
        Convert VFolderRow to VFolderData.
        """
        return VFolderData(
            id=row.id,
            name=row.name,
            host=row.host,
            domain_name=row.domain_name,
            quota_scope_id=row.quota_scope_id,
            usage_mode=row.usage_mode,
            permission=row.permission,
            max_files=row.max_files or 0,
            max_size=row.max_size,
            num_files=row.num_files or 0,
            cur_size=row.cur_size or 0,
            created_at=row.created_at or datetime.now(UTC),
            last_used=row.last_used,
            updated_at=row.updated_at,
            creator=row.creator,
            creator_id=row.creator_id,
            unmanaged_path=row.unmanaged_path,
            ownership_type=row.ownership_type,
            user=row.user,
            group=row.group,
            cloneable=row.cloneable,
            status=row.status,
        )

    @vfolder_repository_resilience.apply()
    async def check_user_has_vfolder_permission(
        self, vfolder_id: uuid.UUID, user_ids: list[uuid.UUID]
    ) -> bool:
        """
        Check if any of the users already have permission for the vfolder.
        Returns True if any user already has permission.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            # Check direct permissions and ownership
            j = sa.join(
                VFolderPermissionRow, VFolderRow, VFolderRow.id == VFolderPermissionRow.vfolder
            )
            count_query = (
                sa.select(sa.func.count())
                .select_from(j)
                .where(
                    sa.and_(
                        sa.or_(
                            VFolderPermissionRow.user.in_(user_ids),
                            VFolderRow.user.in_(user_ids),
                        ),
                        VFolderPermissionRow.vfolder == vfolder_id,
                    )
                )
            )
            count = await session.scalar(count_query)
            return (count or 0) > 0

    @vfolder_repository_resilience.apply()
    async def get_user_by_email(self, email: str) -> tuple[uuid.UUID, str | None] | None:
        """
        Get user info by email.
        Returns (user_id, domain_name) or None if user not found.
        domain_name may be None.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            user_row = await session.scalar(sa.select(UserRow).where(UserRow.email == email))
            if not user_row:
                return None
            return user_row.uuid, user_row.domain_name

    @vfolder_repository_resilience.apply()
    async def get_users_by_emails(self, emails: list[str]) -> list[tuple[uuid.UUID, str]]:
        """
        Get user info for multiple emails.
        Returns list of (user_id, email) tuples.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            result = await session.execute(sa.select(UserRow).where(UserRow.email.in_(emails)))
            user_rows = result.scalars().all()
            return [(row.uuid, row.email) for row in user_rows]

    @vfolder_repository_resilience.apply()
    async def count_vfolder_with_name_for_user(self, user_id: uuid.UUID, vfolder_name: str) -> int:
        """
        Count VFolders with the given name accessible to the user.
        Used to check for duplicates when accepting invitations.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            j = sa.join(
                VFolderRow,
                VFolderPermissionRow,
                VFolderRow.id == VFolderPermissionRow.vfolder,
                isouter=True,
            )
            query = (
                sa.select(sa.func.count())
                .select_from(j)
                .where(
                    sa.and_(
                        sa.or_(
                            VFolderRow.user == user_id,
                            VFolderPermissionRow.user == user_id,
                        ),
                        VFolderRow.name == vfolder_name,
                        VFolderRow.status.not_in(vfolder_status_map[VFolderStatusSet.INACCESSIBLE]),
                    )
                )
            )
            result = await session.scalar(query)
            return result or 0

    @vfolder_repository_resilience.apply()
    async def check_pending_invitation_exists(
        self, vfolder_id: uuid.UUID, inviter_email: str, invitee_email: str
    ) -> bool:
        """
        Check if a pending invitation already exists.
        Returns True if a pending invitation exists.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            query = (
                sa.select(sa.func.count())
                .select_from(VFolderInvitationRow)
                .where(
                    (VFolderInvitationRow.inviter == inviter_email)
                    & (VFolderInvitationRow.invitee == invitee_email)
                    & (VFolderInvitationRow.vfolder == vfolder_id)
                    & (VFolderInvitationRow.state == VFolderInvitationState.PENDING),
                )
            )
            result = await session.scalar(query)
            return (result or 0) > 0

    @vfolder_repository_resilience.apply()
    async def create_vfolder_invitation(
        self,
        vfolder_id: uuid.UUID,
        inviter_email: str,
        invitee_email: str,
        permission: VFolderPermission,
    ) -> str | None:
        """
        Create a VFolder invitation.
        Returns the invitee email on success, None on failure.
        """
        async with self._db.begin_session() as session:
            query = sa.insert(VFolderInvitationRow).values(
                permission=permission,
                vfolder=vfolder_id,
                inviter=inviter_email,
                invitee=invitee_email,
                state=VFolderInvitationState.PENDING,
            )
            try:
                await session.execute(query)
                return invitee_email
            except sa_exc.DataError:
                return None

    @vfolder_repository_resilience.apply()
    async def get_invitation_by_id(self, invitation_id: uuid.UUID) -> VFolderInvitationData | None:
        """
        Get invitation by ID.
        Returns VFolderInvitationData or None if not found.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            query = (
                sa.select(VFolderInvitationRow, UserRow.username)
                .outerjoin(UserRow, UserRow.email == VFolderInvitationRow.inviter)
                .where(
                    (VFolderInvitationRow.id == invitation_id)
                    & (VFolderInvitationRow.state == VFolderInvitationState.PENDING),
                )
            )
            row = (await session.execute(query)).one_or_none()
            if not row:
                return None
            invitation_row, inviter_username = row

            return VFolderInvitationData(
                id=invitation_row.id,
                vfolder=invitation_row.vfolder,
                inviter=invitation_row.inviter or "",
                inviter_username=inviter_username,
                invitee=invitation_row.invitee,
                permission=invitation_row.permission or VFolderMountPermission.READ_ONLY,
                created_at=invitation_row.created_at or datetime.now(UTC),
                modified_at=invitation_row.modified_at,
            )

    @vfolder_repository_resilience.apply()
    async def update_invitation_state(
        self, invitation_id: uuid.UUID, new_state: VFolderInvitationState
    ) -> None:
        """
        Update invitation state.
        """
        async with self._db.begin_session() as session:
            query = (
                sa.update(VFolderInvitationRow)
                .where(VFolderInvitationRow.id == invitation_id)
                .values(state=new_state)
            )
            await session.execute(query)

    @vfolder_repository_resilience.apply()
    async def update_invitation_permission(
        self, invitation_id: uuid.UUID, inviter_email: str, permission: VFolderPermission
    ) -> None:
        """
        Update invitation permission (only by inviter).
        """
        async with self._db.begin_session() as session:
            query = (
                sa.update(VFolderInvitationRow)
                .values(permission=permission)
                .where(
                    sa.and_(
                        VFolderInvitationRow.id == invitation_id,
                        VFolderInvitationRow.inviter == inviter_email,
                        VFolderInvitationRow.state == VFolderInvitationState.PENDING,
                    )
                )
            )
            await session.execute(query)

    @vfolder_repository_resilience.apply()
    async def update_invited_vfolder_mount_permission(
        self, vfolder_id: uuid.UUID, user_id: uuid.UUID, permission: VFolderPermission
    ) -> None:
        """
        Update the permission of an invited user for a specific vfolder.
        """
        async with self._db.begin_session() as session:
            query = (
                sa.update(VFolderPermissionRow)
                .where(
                    sa.and_(
                        VFolderPermissionRow.vfolder == vfolder_id,
                        VFolderPermissionRow.user == user_id,
                    )
                )
                .values(permission=permission)
            )
            await session.execute(query)

    @vfolder_repository_resilience.apply()
    async def get_pending_invitations_for_user(
        self, user_email: str
    ) -> list[tuple[VFolderInvitationData, VFolderData]]:
        """
        Get all pending invitations for a user with VFolder info.
        Returns list of (invitation_data, vfolder_data) tuples.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            j = sa.join(
                VFolderInvitationRow, VFolderRow, VFolderInvitationRow.vfolder == VFolderRow.id
            ).outerjoin(UserRow, UserRow.email == VFolderInvitationRow.inviter)
            query = (
                sa.select(VFolderInvitationRow, UserRow.username)
                .select_from(j)
                .where(
                    sa.and_(
                        VFolderInvitationRow.invitee == user_email,
                        VFolderInvitationRow.state == VFolderInvitationState.PENDING,
                    )
                )
                .options(
                    contains_eager(VFolderInvitationRow.vfolder_row),
                )
            )
            result = await session.execute(query)
            rows = result.all()

            results = []
            for inv_row, inviter_username in rows:
                invitation_data = VFolderInvitationData(
                    id=inv_row.id,
                    vfolder=inv_row.vfolder,
                    inviter=inv_row.inviter or "",
                    inviter_username=inviter_username,
                    invitee=inv_row.invitee,
                    permission=inv_row.permission or VFolderMountPermission.READ_ONLY,
                    created_at=inv_row.created_at or datetime.now(UTC),
                    modified_at=inv_row.modified_at,
                )
                vfolder_data = self._vfolder_row_to_data(inv_row.vfolder_row)
                results.append((invitation_data, vfolder_data))

            return results

    @vfolder_repository_resilience.apply()
    async def get_sent_invitations_for_user(
        self, user_email: str
    ) -> list[tuple[VFolderInvitationData, VFolderData]]:
        """
        Get all pending invitations sent by a user (as inviter) with VFolder info.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            j = sa.join(
                VFolderInvitationRow, VFolderRow, VFolderInvitationRow.vfolder == VFolderRow.id
            ).outerjoin(UserRow, UserRow.email == VFolderInvitationRow.inviter)
            query = (
                sa.select(VFolderInvitationRow, UserRow.username)
                .select_from(j)
                .where(
                    sa.and_(
                        VFolderInvitationRow.inviter == user_email,
                        VFolderInvitationRow.state == VFolderInvitationState.PENDING,
                    )
                )
                .options(
                    contains_eager(VFolderInvitationRow.vfolder_row),
                )
            )
            result = await session.execute(query)
            rows = result.all()

            results = []
            for inv_row, inviter_username in rows:
                invitation_data = VFolderInvitationData(
                    id=inv_row.id,
                    vfolder=inv_row.vfolder,
                    inviter=inv_row.inviter or "",
                    inviter_username=inviter_username,
                    invitee=inv_row.invitee,
                    permission=inv_row.permission or VFolderMountPermission.READ_ONLY,
                    created_at=inv_row.created_at or datetime.now(UTC),
                    modified_at=inv_row.modified_at,
                )
                vfolder_data = self._vfolder_row_to_data(inv_row.vfolder_row)
                results.append((invitation_data, vfolder_data))

            return results

    @vfolder_repository_resilience.apply()
    async def ensure_host_permission_allowed(
        self,
        folder_host: str,
        *,
        permission: VFolderHostPermission,
        allowed_vfolder_types: Sequence[str],
        user_uuid: uuid.UUID,
        resource_policy: Mapping[str, Any],
        domain_name: str,
        group_id: uuid.UUID | None = None,
    ) -> None:
        """
        Ensure that the user has the required permission on the specified vfolder host.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            # Get connection from session
            conn = await session.connection()
            await ensure_host_permission_allowed(
                conn,
                folder_host,
                permission=permission,
                allowed_vfolder_types=allowed_vfolder_types,
                user_uuid=user_uuid,
                resource_policy=resource_policy,
                domain_name=domain_name,
                group_id=group_id,
            )

    @vfolder_repository_resilience.apply()
    async def ensure_host_permission_allowed_by_user(
        self,
        folder_host: str,
        *,
        permission: VFolderHostPermission,
        user_uuid: uuid.UUID,
        group_id: uuid.UUID | None = None,
    ) -> None:
        """Check host permission by looking up the user's resource policy from DB."""
        allowed_hosts = await self.get_allowed_vfolder_hosts(user_uuid, group_id)
        if folder_host not in allowed_hosts or permission not in allowed_hosts[folder_host]:
            raise InsufficientStoragePermission(
                f"`{permission}` Not allowed in vfolder host(`{folder_host}`)"
            )

    @vfolder_repository_resilience.apply()
    async def get_validated_vfolder_id(
        self,
        vfolder_uuid: uuid.UUID,
        *,
        permission: VFolderHostPermission,
        allowed_vfolder_types: Sequence[str],
        resource_policy: Mapping[str, Any],
    ) -> ValidatedVFolderInfo:
        """
        Resolve user from context, validate vfolder access, check host permission,
        and return validated VFolderID with storage info.
        """
        user = current_user()
        if user is None:
            raise AuthorizationFailed("User context is not available")
        vfolder_data = await self.get_by_id_validated(vfolder_uuid, user.user_id, user.domain_name)
        await self.ensure_host_permission_allowed(
            vfolder_data.host,
            permission=permission,
            allowed_vfolder_types=allowed_vfolder_types,
            user_uuid=user.user_id,
            resource_policy=resource_policy,
            domain_name=vfolder_data.domain_name,
        )
        return ValidatedVFolderInfo(
            vfolder_id=VFolderID(
                quota_scope_id=vfolder_data.quota_scope_id,
                folder_id=vfolder_data.id,
            ),
            host=vfolder_data.host,
            unmanaged_path=vfolder_data.unmanaged_path,
        )

    def _vfolder_dict_to_data(self, vfolder_dict: dict[str, Any]) -> VFolderData:
        """
        Convert vfolder dictionary from query_accessible_vfolders to VFolderData.
        """
        return VFolderData(
            id=VFolderUUID(vfolder_dict["id"]),
            name=vfolder_dict["name"],
            host=vfolder_dict["host"],
            domain_name=vfolder_dict["domain_name"],
            quota_scope_id=vfolder_dict["quota_scope_id"],
            usage_mode=vfolder_dict["usage_mode"],
            permission=vfolder_dict.get("permission"),
            max_files=vfolder_dict["max_files"],
            max_size=vfolder_dict["max_size"],
            num_files=vfolder_dict.get("num_files", 0),
            cur_size=vfolder_dict["cur_size"],
            created_at=vfolder_dict["created_at"],
            last_used=vfolder_dict["last_used"],
            updated_at=vfolder_dict["updated_at"],
            creator=vfolder_dict["creator"],
            creator_id=vfolder_dict.get("creator_id"),
            unmanaged_path=vfolder_dict["unmanaged_path"],
            ownership_type=vfolder_dict["ownership_type"],
            user=uuid.UUID(vfolder_dict["user"]) if vfolder_dict["user"] else None,
            group=uuid.UUID(vfolder_dict["group"]) if vfolder_dict["group"] else None,
            cloneable=vfolder_dict["cloneable"],
            status=vfolder_dict["status"],
        )

    @vfolder_repository_resilience.apply()
    async def initiate_vfolder_clone(
        self,
        vfolder_info: VFolderCloneInfo,
        storage_manager: StorageSessionManager,
        _background_task_manager: BackgroundTaskManager,
    ) -> tuple[uuid.UUID, uuid.UUID]:
        """
        Initiate VFolder cloning process.
        Returns (task_id, target_folder_id).
        """
        source_vf_cond = vfolders.c.id == vfolder_info.source_vfolder_id.folder_id

        async def _update_source_status() -> None:
            async with self._db.begin_session() as db_session:
                query = (
                    sa.update(vfolders)
                    .values(status=VFolderOperationStatus.CLONING)
                    .where(source_vf_cond)
                )
                await db_session.execute(query)

        await execute_with_retry(_update_source_status)

        target_proxy, target_volume = storage_manager.get_proxy_and_volume(vfolder_info.target_host)
        source_proxy, source_volume = storage_manager.get_proxy_and_volume(
            vfolder_info.source_host, is_unmanaged(vfolder_info.unmanaged_path)
        )

        if source_proxy != target_proxy:
            raise VFolderInvalidParameter(
                f"Proxy names of source and target vfolders must be equal. "
                f"Source proxy: {source_proxy}, Target proxy: {target_proxy}."
            )

        # Generate the ID of the destination vfolder.
        # TODO: If we refactor to use ORM, the folder ID will be created from the database by inserting
        #       the actual object (with RETURNING clause).  In that case, we need to temporarily
        #       mark the object to be "unusable-yet" until the storage proxy creates the destination
        #       vfolder.  After done, we need to make another transaction to clear the unusable state.
        target_folder_id = VFolderID(vfolder_info.target_quota_scope_id, uuid.uuid4())

        # Clone the vfolder contents
        manager_client = storage_manager.get_manager_facing_client(source_proxy)
        clone_response = await manager_client.clone_folder(
            source_volume,
            str(vfolder_info.source_vfolder_id),
            target_volume,
            str(target_folder_id),
        )
        task_id = clone_response.bgtask_id

        async def _insert_vfolder() -> None:
            async with self._db.begin_session() as db_session:
                insert_values = {
                    "id": target_folder_id.folder_id,
                    "name": vfolder_info.target_vfolder_name,
                    "domain_name": vfolder_info.domain_name,
                    "usage_mode": vfolder_info.usage_mode,
                    "permission": vfolder_info.permission,
                    "last_used": None,
                    "host": vfolder_info.target_host,
                    "creator": vfolder_info.email,
                    "creator_id": vfolder_info.user_id,
                    "ownership_type": VFolderOwnershipType("user"),
                    "user": vfolder_info.user_id,
                    "group": None,
                    "unmanaged_path": None,
                    "cloneable": vfolder_info.cloneable,
                    "quota_scope_id": vfolder_info.target_quota_scope_id,
                }
                query = sa.insert(vfolders).values(**insert_values)
                await db_session.execute(query)

        # Insert the new vfolder record
        await execute_with_retry(_insert_vfolder)

        return task_id, target_folder_id.folder_id

    @vfolder_repository_resilience.apply()
    async def get_logs_vfolder(
        self,
        user_id: uuid.UUID,
        user_role: UserRole,
        domain_name: str,
    ) -> VFolderData | None:
        """
        Get the accessible .logs vfolder for a user.
        Returns VFolderData if found, None otherwise.
        """
        async with self._db.begin_readonly() as conn:
            vfolder_dicts = await query_accessible_vfolders(
                conn,
                user_id,
                user_role=user_role,
                domain_name=domain_name,
                allowed_vfolder_types=["user"],
                extra_vf_conds=(vfolders.c.name == ".logs"),
            )

            if not vfolder_dicts:
                return None

            # Return the first (and should be only) matching .logs vfolder
            vfolder_dict = vfolder_dicts[0]
            return self._vfolder_dict_to_data(dict(vfolder_dict))

    @vfolder_repository_resilience.apply()
    async def share_vfolder_with_users(
        self,
        vfolder_id: uuid.UUID,
        vfolder_host: str,
        vfolder_group: uuid.UUID | None,
        requester_uuid: uuid.UUID,
        requester_email: str,
        domain_name: str,
        resource_policy: Mapping[str, Any],
        emails: list[str],
        permission: VFolderPermission,
        allowed_vfolder_types: Sequence[str],
    ) -> list[str]:
        """
        Share a group vfolder with users by granting permissions directly.
        Returns list of emails that were shared with.
        """
        async with self._db.begin_session() as session:
            conn = await session.connection()
            await ensure_host_permission_allowed(
                conn,
                vfolder_host,
                permission=VFolderHostPermission.SET_USER_PERM,
                allowed_vfolder_types=allowed_vfolder_types,
                user_uuid=requester_uuid,
                resource_policy=resource_policy,
                domain_name=domain_name,
            )

            users_table = UserRow.__table__
            j = users_table.join(
                AssociationScopesEntitiesRow,
                sa.cast(users_table.c.uuid, sa.String) == AssociationScopesEntitiesRow.entity_id,
            )
            db_query = (
                sa.select(users_table.c.uuid, users_table.c.email)
                .select_from(j)
                .where(
                    AssociationScopesEntitiesRow.scope_type == ScopeType.PROJECT,
                    AssociationScopesEntitiesRow.entity_type == EntityType.USER,
                    AssociationScopesEntitiesRow.scope_id == str(vfolder_group),
                    users_table.c.email.in_(emails),
                    users_table.c.email != requester_email,
                    users_table.c.status.in_(ACTIVE_USER_STATUSES),
                )
            )
            result = await session.execute(db_query)
            user_info = result.fetchall()
            users_to_share = [u.uuid for u in user_info]
            emails_to_share = [u.email for u in user_info]
            if len(user_info) < 1:
                raise ObjectNotFound(object_name="user")
            if len(user_info) < len(emails):
                users_not_in_group = list(set(emails) - set(emails_to_share))
                raise ObjectNotFound(
                    f"Some users do not belong to folder's group: {','.join(users_not_in_group)}",
                    object_name="user",
                )

            existing_query = sa.select(VFolderPermissionRow.user).where(
                (VFolderPermissionRow.user.in_(users_to_share))
                & (VFolderPermissionRow.vfolder == vfolder_id),
            )
            result = await session.execute(existing_query)
            users_with_existing_perm = [row.user for row in result.fetchall()]
            new_users = list(set(users_to_share) - set(users_with_existing_perm))

            for _user in new_users:
                insert_stmt = sa.insert(VFolderPermissionRow).values(
                    permission=permission,
                    vfolder=vfolder_id,
                    user=_user,
                )
                await session.execute(insert_stmt)
            for _user in users_with_existing_perm:
                update_stmt = (
                    sa.update(VFolderPermissionRow)
                    .values(permission=permission)
                    .where(VFolderPermissionRow.vfolder == vfolder_id)
                    .where(VFolderPermissionRow.user == _user)
                )
                await session.execute(update_stmt)

            return emails_to_share

    @vfolder_repository_resilience.apply()
    async def unshare_vfolder_from_users(
        self,
        vfolder_id: uuid.UUID,
        vfolder_host: str,
        requester_uuid: uuid.UUID,
        domain_name: str,
        resource_policy: Mapping[str, Any],
        emails: list[str],
        allowed_vfolder_types: Sequence[str],
    ) -> list[str]:
        """
        Revoke direct sharing permissions from users.
        Returns list of emails that were unshared.
        """
        async with self._db.begin_session() as session:
            conn = await session.connection()
            await ensure_host_permission_allowed(
                conn,
                vfolder_host,
                permission=VFolderHostPermission.SET_USER_PERM,
                allowed_vfolder_types=allowed_vfolder_types,
                user_uuid=requester_uuid,
                resource_policy=resource_policy,
                domain_name=domain_name,
            )

            users_table = UserRow.__table__
            db_query = (
                sa.select(users_table.c.uuid)
                .select_from(users_table)
                .where(users_table.c.email.in_(emails))
            )
            result = await session.execute(db_query)
            users_to_unshare = [u.uuid for u in result.fetchall()]
            if len(users_to_unshare) < 1:
                raise ObjectNotFound(object_name="user(s).")

            delete_stmt = sa.delete(VFolderPermissionRow).where(
                (VFolderPermissionRow.vfolder == vfolder_id)
                & (VFolderPermissionRow.user.in_(users_to_unshare)),
            )
            await session.execute(delete_stmt)

            return emails

    @vfolder_repository_resilience.apply()
    async def list_shared_vfolder_permissions(
        self,
        vfolder_id: uuid.UUID | None = None,
    ) -> list[dict[str, Any]]:
        """
        List all shared vfolder permission entries with vfolder and user info.
        Returns list of dicts with vfolder/permission/user fields.
        """
        async with self._db.begin_readonly_session_read_committed() as session:
            perm_table = VFolderPermissionRow.__table__
            vf_table = VFolderRow.__table__
            users_table = UserRow.__table__
            j = perm_table.join(vf_table, vf_table.c.id == perm_table.c.vfolder).join(
                users_table, users_table.c.uuid == perm_table.c.user
            )
            db_query = sa.select(
                perm_table,
                vf_table.c.id.label("vfolder_id"),
                vf_table.c.name,
                vf_table.c.group,
                vf_table.c.status,
                vf_table.c.user.label("vfolder_user"),
                users_table.c.email,
            ).select_from(j)
            if vfolder_id is not None:
                db_query = db_query.where(vf_table.c.id == vfolder_id)
            result = await session.execute(db_query)
            return [dict(row._mapping) for row in result.fetchall()]

    @vfolder_repository_resilience.apply()
    async def update_vfolder_sharing_status(
        self,
        vfolder_id: uuid.UUID,
        to_delete: list[uuid.UUID],
        to_update: list[tuple[uuid.UUID, VFolderPermission]],
    ) -> None:
        """
        Batch update and/or delete sharing permissions for a vfolder.
        """
        async with self._db.begin_session() as session:
            if to_delete:
                delete_stmt = (
                    sa.delete(VFolderPermissionRow)
                    .where(VFolderPermissionRow.vfolder == vfolder_id)
                    .where(VFolderPermissionRow.user.in_(to_delete))
                )
                await session.execute(delete_stmt)

            for user_id, perm in to_update:
                update_stmt = (
                    sa.update(VFolderPermissionRow)
                    .values(permission=perm)
                    .where(VFolderPermissionRow.vfolder == vfolder_id)
                    .where(VFolderPermissionRow.user == user_id)
                )
                await session.execute(update_stmt)

    @vfolder_repository_resilience.apply()
    async def check_vfolder_accessible(
        self,
        vfolder_id: uuid.UUID,
        user_uuid: uuid.UUID,
        user_role: UserRole,
        domain_name: str,
        allowed_vfolder_types: Sequence[str],
    ) -> bool:
        """
        Check if a vfolder is accessible by the user.
        Returns True if accessible, raises VFolderNotFound if not.
        """
        async with self._db.begin_readonly_session() as session:
            conn = await session.connection()
            extra_vf_conds = [vfolders.c.id == vfolder_id]
            entries = await query_accessible_vfolders(
                conn,
                user_uuid,
                user_role=user_role,
                domain_name=domain_name,
                allowed_vfolder_types=allowed_vfolder_types,
                extra_vf_conds=(sa.and_(*extra_vf_conds)),
            )
        if len(entries) == 0:
            raise VFolderNotFound(extra_data=vfolder_id)
        return True

    @vfolder_repository_resilience.apply()
    async def get_accessible_rows(
        self,
        user_uuid: uuid.UUID,
        user_role: UserRole,
        domain_name: str,
        is_admin: bool,
        allowed_vfolder_types: Sequence[str],
        perm: VFolderPermissionSetAlias | VFolderPermission,
        folder_id_or_name: str | uuid.UUID,
        *,
        allowed_status_set: VFolderStatusSet | None = None,
        allow_privileged_access: bool = False,
    ) -> Sequence[Mapping[str, Any]]:
        """
        Build permission conditions and query accessible vfolders.

        This is the repository-layer replacement for the legacy
        ``resolve_vfolder_rows()`` that previously lived in the API layer.
        """
        vf_user_cond = None
        vf_group_cond: sa.ColumnElement[bool] | None = None

        match perm:
            case VFolderPermissionSetAlias():
                invited_perm_cond = vfolder_permissions.c.permission.in_(list(perm.value))
                if not is_admin:
                    vf_group_cond = vfolders.c.permission.in_(list(perm.value))
            case _:
                invited_perm_cond = vfolder_permissions.c.permission == perm
                if not is_admin:
                    vf_group_cond = vfolders.c.permission == perm

        match folder_id_or_name:
            case str():
                extra_vf_conds = vfolders.c.name == folder_id_or_name
            case uuid.UUID():
                extra_vf_conds = vfolders.c.id == folder_id_or_name
            case _:
                raise RuntimeError(f"Unsupported VFolder index type {type(folder_id_or_name)}")

        async with self._db.begin_readonly() as conn:
            return await query_accessible_vfolders(
                conn,
                user_uuid,
                allow_privileged_access=allow_privileged_access,
                user_role=user_role,
                domain_name=domain_name,
                allowed_vfolder_types=allowed_vfolder_types,
                extra_vf_conds=extra_vf_conds,
                extra_invited_vf_conds=invited_perm_cond,
                extra_vf_user_conds=vf_user_cond,
                extra_vf_group_conds=vf_group_cond,
                allowed_status_set=allowed_status_set,
            )

    @vfolder_repository_resilience.apply()
    async def update_vfolder_max_size(
        self,
        vfolder_id: uuid.UUID,
        max_size_mib: int,
    ) -> None:
        """
        Update the max_size field of a vfolder in MiB.
        """
        async with self._db.begin_session() as session:
            conn = await session.connection()
            update_query = (
                sa.update(vfolders).values(max_size=max_size_mib).where(vfolders.c.id == vfolder_id)
            )
            db_result = await conn.execute(update_query)
            if db_result.rowcount != 1:
                raise VFolderOperationFailed(
                    f"Failed to update vfolder quota: expected 1 row, got {db_result.rowcount}"
                )

    @vfolder_repository_resilience.apply()
    async def get_user_storage_host_permissions(
        self,
        user_uuid: uuid.UUID,
        domain_name: str,
    ) -> VFolderHostPermissionMap:
        """
        Resolve all storage hosts and per-host permissions accessible to a user.

        Internally fetches the user's main keypair resource policy and unions
        domain/group/keypair allowed vfolder hosts. Returns the host permission
        map without filtering against currently mountable volumes — callers that
        depend on volume availability must intersect with ``StorageSessionManager``.
        """
        async with self._db.begin_readonly_session_read_committed() as db_session:
            user_row: UserRow | None = await db_session.scalar(
                sa.select(UserRow)
                .where(UserRow.uuid == user_uuid)
                .options(
                    selectinload(UserRow.main_keypair).selectinload(KeyPairRow.resource_policy_row)
                )
            )
            if user_row is None:
                raise UserNotFound(f"User with UUID {user_uuid} not found.")
            if user_row.main_keypair is None or user_row.main_keypair.resource_policy_row is None:
                resource_policy: Mapping[str, Any] = {
                    "allowed_vfolder_hosts": VFolderHostPermissionMap(),
                }
            else:
                resource_policy = {
                    "allowed_vfolder_hosts": user_row.main_keypair.resource_policy_row.allowed_vfolder_hosts,
                }
            conn = await db_session.connection()
            return await get_allowed_vfolder_hosts_by_user(
                conn=conn,
                resource_policy=resource_policy,
                domain_name=domain_name,
                user_uuid=user_uuid,
                group_id=None,
            )

    @vfolder_repository_resilience.apply()
    async def get_allowed_hosts_for_listing(
        self,
        user_uuid: uuid.UUID,
        domain_name: str,
        group_id: uuid.UUID | None,
        resource_policy: Mapping[str, Any],
        allowed_vfolder_types: Sequence[str],
    ) -> VFolderHostPermissionMap:
        """
        Get the combined allowed vfolder hosts for user and group.
        """
        async with self._db.begin_readonly_session() as session:
            conn = await session.connection()
            allowed_hosts = VFolderHostPermissionMap()
            if "user" in allowed_vfolder_types:
                allowed_hosts_by_user = await get_allowed_vfolder_hosts_by_user(
                    conn, resource_policy, domain_name, user_uuid, group_id
                )
                allowed_hosts = cast(
                    VFolderHostPermissionMap, allowed_hosts | allowed_hosts_by_user
                )
            if "group" in allowed_vfolder_types:
                allowed_hosts_by_group = await get_allowed_vfolder_hosts_by_group(
                    conn,
                    resource_policy,
                    domain_name,
                    group_id,
                )
                allowed_hosts = cast(
                    VFolderHostPermissionMap, allowed_hosts | allowed_hosts_by_group
                )
            return allowed_hosts

    @vfolder_repository_resilience.apply()
    async def change_vfolder_ownership(
        self,
        vfolder_id: uuid.UUID,
        user_email: str,
    ) -> None:
        """
        Change ownership of a user vfolder to another user.
        Validates user exists with allowed host access, updates owner,
        and removes related invitations/permissions for the new owner.
        """
        # Step 1: Get target user info and their allowed hosts
        async with self._db.begin_readonly_session() as session:
            conn = await session.connection()
            j = sa.join(users, keypairs, users.c.email == keypairs.c.user_id)
            db_query = (
                sa.select(users.c.uuid, users.c.domain_name, keypairs.c.resource_policy)
                .select_from(j)
                .where((users.c.email == user_email) & (users.c.status == UserStatus.ACTIVE))
            )
            try:
                result = await conn.execute(db_query)
            except sa.exc.DataError as e:
                raise InvalidAPIParameters from e
            user_info = result.first()
            if user_info is None:
                raise ObjectNotFound(object_name="user")

            resource_policy_name = user_info.resource_policy
            result = await conn.execute(
                sa.select(keypair_resource_policies.c.allowed_vfolder_hosts).where(
                    keypair_resource_policies.c.name == resource_policy_name
                )
            )
            resource_policy_row = result.first()
            allowed_hosts_by_user = await get_allowed_vfolder_hosts_by_user(
                conn=conn,
                resource_policy=dict(resource_policy_row._mapping) if resource_policy_row else {},
                domain_name=user_info.domain_name,
                user_uuid=user_info.uuid,
            )

        # Step 2: Check vfolder host is accessible by new owner and get old owner
        async with self._db.begin_readonly_session() as session:
            conn = await session.connection()
            db_query = (
                sa.select(vfolders.c.host, vfolders.c.user)
                .select_from(vfolders)
                .where(
                    (vfolders.c.id == vfolder_id)
                    & (vfolders.c.ownership_type == VFolderOwnershipType.USER)
                )
            )
            row = (await conn.execute(db_query)).first()
            folder_host = row.host if row else None
            old_owner_uuid = row.user if row else None
        if folder_host not in allowed_hosts_by_user:
            raise VFolderOperationFailed(
                "User to migrate vfolder needs an access to the storage host."
            )

        # Step 3: Update vfolder owner
        async def _update() -> None:
            async with self._db.begin_session() as session:
                conn = await session.connection()
                update_query = (
                    sa.update(vfolders)
                    .values(user=user_info.uuid)
                    .where(
                        (vfolders.c.id == vfolder_id)
                        & (vfolders.c.ownership_type == VFolderOwnershipType.USER)
                    )
                )
                await conn.execute(update_query)

        await execute_with_retry(_update)

        # Step 4: Delete related invitations and permissions for new owner
        async def _delete_related_rows() -> None:
            async with self._db.begin_session() as session:
                conn = await session.connection()
                del_query = sa.delete(vfolder_invitations).where(
                    (vfolder_invitations.c.invitee == user_email)
                    & (vfolder_invitations.c.vfolder == vfolder_id)
                )
                await conn.execute(del_query)

                # Also clean up new owner's RBAC records from when they were an invitee
                new_owner_role_id = await self._get_user_role_id(session, user_info.uuid)
                revoker = RBACRevoker(
                    entity_id=ObjectId(
                        entity_type=EntityType.VFOLDER,
                        entity_id=str(vfolder_id),
                    ),
                    entity_scope_type=RBACElementType.VFOLDER,
                    target_role_ids=[new_owner_role_id],
                    operations=None,
                )
                await execute_rbac_revoker(session, revoker)

        await execute_with_retry(_delete_related_rows)

        # Step 5: Clean up old owner's RBAC records for this vfolder
        if old_owner_uuid is not None and old_owner_uuid != user_info.uuid:

            async def _cleanup_old_owner_rbac() -> None:
                async with self._db.begin_session() as session:
                    user_role_id = await self._get_user_role_id(session, old_owner_uuid)
                    revoker = RBACRevoker(
                        entity_id=ObjectId(
                            entity_type=EntityType.VFOLDER,
                            entity_id=str(vfolder_id),
                        ),
                        entity_scope_type=RBACElementType.VFOLDER,
                        target_role_ids=[user_role_id],
                        operations=None,
                    )
                    await execute_rbac_revoker(session, revoker)
                    # Remove scope-entity mapping (visibility)
                    await session.execute(
                        sa.delete(AssociationScopesEntitiesRow).where(
                            sa.and_(
                                AssociationScopesEntitiesRow.scope_type == ScopeType.USER,
                                AssociationScopesEntitiesRow.scope_id == str(old_owner_uuid),
                                AssociationScopesEntitiesRow.entity_type == EntityType.VFOLDER,
                                AssociationScopesEntitiesRow.entity_id == str(vfolder_id),
                            )
                        )
                    )

            await execute_with_retry(_cleanup_old_owner_rbac)

        # Step 6: Create owner RBAC records for new owner
        async def _grant_new_owner_rbac() -> None:
            async with self._db.begin_session() as session:
                user_role_id = await self._get_user_role_id(session, user_info.uuid)

                # Upsert scope-entity mapping (AUTO relation for owner)
                # If new owner was previously invitee, upgrade REF -> AUTO
                upsert_stmt = (
                    pg_insert(AssociationScopesEntitiesRow)
                    .values(
                        scope_type=ScopeType.USER,
                        scope_id=str(user_info.uuid),
                        entity_type=EntityType.VFOLDER,
                        entity_id=str(vfolder_id),
                        relation_type=RelationType.AUTO,
                    )
                    .on_conflict_do_update(
                        constraint="uq_scope_id_entity_id",
                        set_={"relation_type": RelationType.AUTO},
                    )
                )
                await session.execute(upsert_stmt)

                # Grant owner permission (ON CONFLICT DO NOTHING in case
                # new owner already had permission as invitee)
                perm_stmt = (
                    pg_insert(PermissionRow)
                    .values(
                        role_id=user_role_id,
                        scope_type=ScopeType.VFOLDER,
                        scope_id=str(vfolder_id),
                        entity_type=EntityType.VFOLDER,
                        operation=OperationType.READ,
                    )
                    .on_conflict_do_nothing(
                        constraint="uq_permissions_role_scope_entity_op",
                    )
                )
                await session.execute(perm_stmt)

        await execute_with_retry(_grant_new_owner_rbac)

    @vfolder_repository_resilience.apply()
    async def get_alive_agent_ids(
        self,
        scaling_group: str | None = None,
    ) -> list[str]:
        """Get IDs of agents with ALIVE status, optionally filtered by scaling group."""
        async with self._db.begin_readonly_session() as session:
            conn = await session.connection()
            stmt = sa.select(agents.c.id).where(agents.c.status == AgentStatus.ALIVE)
            if scaling_group is not None:
                stmt = stmt.where(agents.c.scaling == scaling_group)
            result = await conn.execute(stmt)
            return [row.id for row in result.fetchall()]

    @vfolder_repository_resilience.apply()
    async def get_active_kernel_mount_names(self) -> set[str]:
        """Get mount names from all non-terminated kernels."""
        async with self._db.begin_readonly_session() as session:
            conn = await session.connection()
            stmt = sa.select(kernels.c.mounts).where(kernels.c.status != KernelStatus.TERMINATED)
            result = await conn.execute(stmt)
            mounted: set[str] = set()
            for row in result.fetchall():
                if row.mounts:
                    mounted.update(m[1] for m in row.mounts)
            return mounted

    @vfolder_repository_resilience.apply()
    async def search_in_project(
        self,
        querier: BatchQuerier,
        scope: ProjectVFolderSearchScope,
    ) -> VFolderSearchResult:
        """Search vfolders scoped to a project.

        Args:
            querier: BatchQuerier for filtering, ordering, and pagination
            scope: ProjectVFolderSearchScope that filters by project and validates existence

        Returns:
            VFolderSearchResult with items, total count, and pagination info
        """
        async with self._db.begin_readonly_session() as db_sess:
            query = sa.select(VFolderRow)

            result = await execute_batch_querier(
                db_sess,
                query,
                querier,
                scopes=[scope],
            )

            items = [row.VFolderRow.to_data() for row in result.rows]

            return VFolderSearchResult(
                items=items,
                total_count=result.total_count,
                has_next_page=result.has_next_page,
                has_previous_page=result.has_previous_page,
            )

    @vfolder_repository_resilience.apply()
    async def search_user_vfolders(
        self,
        querier: BatchQuerier,
        scope: UserVFolderSearchScope,
    ) -> VFolderSearchResult:
        """Search vfolders scoped to a user.

        Args:
            querier: BatchQuerier for filtering, ordering, and pagination
            scope: UserVFolderSearchScope that filters by user and validates existence

        Returns:
            VFolderSearchResult with items, total count, and pagination info
        """
        async with self._db.begin_readonly_session() as db_sess:
            query = sa.select(VFolderRow)

            result = await execute_batch_querier(
                db_sess,
                query,
                querier,
                scopes=[scope],
            )

            items = [row.VFolderRow.to_data() for row in result.rows]

            return VFolderSearchResult(
                items=items,
                total_count=result.total_count,
                has_next_page=result.has_next_page,
                has_previous_page=result.has_previous_page,
            )