"""Admin HTTP surface for role-change handling (issue #623).

Three endpoints, all admin-gated:

  * ``GET  /api/v1/role-changes`` — list the last 30 days of
    ``HrLifecycleEvent(event_type='role_change')`` rows joined to
    the underlying person + a count of ``pending_review`` grants
    held by that person. Server-side pagination and SQL-level search /
    ``only_pending`` filters keep query count and response size bounded.

  * ``GET  /api/v1/role-changes/{person_id}/grants`` — list every
    ``cmdb_cloud_access_grants`` row for that person and its
    ``review_status``. The admin page renders this in a per-row
    table with a 「保留 / 撤销」 decision button.

  * ``PUT  /api/v1/role-changes/grants/{grant_id}/review`` —
    flip a grant's ``review_status`` to ``confirmed_keep`` or
    ``confirmed_revoke`` and stamp ``review_status_changed_at``.
    When ``confirmed_revoke`` is chosen, the grant is also marked
    ``status='revoked'`` + ``revoked_at=NOW()`` so the cloud-side
    cleanup cron picks it up.
"""

from __future__ import annotations

from datetime import UTC, datetime
from typing import Any, Literal
from uuid import UUID

from fastapi import APIRouter, Depends, HTTPException, Query, Request
from pydantic import BaseModel, Field
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.auth.models import AuthUser
from app.core.coordinators.role_change_coordinator import (
    REVIEW_STATUS_CONFIRMED_KEEP,
    REVIEW_STATUS_CONFIRMED_REVOKE,
    REVIEW_STATUS_PENDING,
)
from app.core.database import get_db
from app.core.dependencies import (
    get_current_user,
    require_admin,
    require_admin_read,
)
from app.core.exceptions import AppException, NotFoundError
from app.core.response import success_response
from app.domains.cmdb.models import CmdbCloudAccessGrant, CmdbCloudScope
from app.domains.provisioning.models import ProvisionPolicy
from app.hr_lifecycle.mover_review_settings import (
    DEFAULT_PUSH_BASELINE_SOURCES,
    SETTING_KEY,
    RoleChangeReviewSettings,
    load_review_settings,
    save_review_settings,
)
from app.hr_lifecycle.role_change_service import build_role_change_list
from app.identity.person_models import IdentityOrgMembership, IdentityPerson

router = APIRouter(prefix="/api/v1/role-changes", tags=["role-changes"])


@router.get(
    "",
    summary="List recent role-change events with pending_review counts",
)
async def list_role_changes(
    request: Request,
    days: int = Query(30, ge=1, le=180),
    page_size: int = Query(20, ge=1, le=100),
    cursor: str | None = Query(None, max_length=2000),
    only_pending: bool = Query(
        False,
        description="If true, only include persons with ≥1 pending_review grant.",
    ),
    q: str | None = Query(
        None,
        max_length=100,
        description=("Case-insensitive substring match on the person's display_name / work_email."),
    ),
    current_user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """One row per role-change event in the last N days.

    Returns ``{person, event, pending_count, total_count}`` per row plus
    ``page / page_size / total / next_cursor`` pagination metadata;
    the admin page uses ``pending_count > 0`` to decide whether the
    「待决断」 chip is shown.
    """
    return success_response(
        data=await build_role_change_list(
            db,
            days=days,
            page_size=page_size,
            only_pending=only_pending,
            q=q,
            cursor=cursor,
            cursor_subject=str(current_user.id),
        ),
        request=request,
    )


@router.get(
    "/{person_id}/grants",
    summary="Per-person grant list with review_status",
)
async def list_person_grants(
    person_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    person = await db.scalar(select(IdentityPerson).where(IdentityPerson.id == person_id))
    if person is None:
        raise NotFoundError("IdentityPerson", str(person_id))

    person_str = str(person_id)
    grants = list(
        (
            await db.execute(
                select(CmdbCloudAccessGrant)
                .where(CmdbCloudAccessGrant.person_id == person_str)
                .order_by(CmdbCloudAccessGrant.granted_at.desc())
            )
        )
        .scalars()
        .all()
    )

    # Resolve scope display names so the review UI shows WHICH cloud scope
    # a grant is on (e.g. "AWS Prod (123456789012)") instead of a raw UUID —
    # the admin deciding keep/revoke needs that to judge the grant.
    scope_ids = {g.scope_id for g in grants if g.scope_id}
    scope_meta: dict[str, tuple[str | None, str | None]] = {}
    if scope_ids:
        scope_rows = (
            await db.execute(
                select(
                    CmdbCloudScope.id,
                    CmdbCloudScope.display_name,
                    CmdbCloudScope.external_id,
                ).where(CmdbCloudScope.id.in_(scope_ids))
            )
        ).all()
        scope_meta = {str(r[0]): (r[1], r[2]) for r in scope_rows}

    rows = [
        {
            "id": str(g.id),
            "scope_id": g.scope_id,
            "scope_display_name": scope_meta.get(str(g.scope_id), (None, None))[0],
            "scope_external_id": scope_meta.get(str(g.scope_id), (None, None))[1],
            "role_key": g.role_key,
            "role_display_name": g.role_display_name,
            "status": g.status,
            "review_status": g.review_status,
            "review_status_changed_at": (
                g.review_status_changed_at.isoformat() if g.review_status_changed_at else None
            ),
            "granted_at": (g.granted_at.isoformat() if g.granted_at else None),
            "revoked_at": (g.revoked_at.isoformat() if g.revoked_at else None),
        }
        for g in grants
    ]
    return success_response(
        data={"person_id": str(person_id), "grants": rows},
        request=request,
    )


class _ReviewBody(BaseModel):
    decision: Literal["confirmed_keep", "confirmed_revoke"]
    note: str | None = None


async def _authorize_grant_review(
    db: AsyncSession, grant: CmdbCloudAccessGrant, user: AuthUser,
) -> None:
    """Who may resolve one ``pending_review`` cloud grant.

    Two callers, and only one of them was ever admitted:

    * ``global_admin`` — the platform operator, as before;
    * **the owner of the scope the grant sits on** — added 2026-09-05.

    Why the second: ``role_change.scope_review_sweep`` sends its card to
    the scope owner AND the admin pool, and a scope owner is frequently
    an ordinary employee. The card asked them to 「确认 keep 或 revoke」
    at an endpoint that answered 403 for them, and pointed at a console
    route their role could not open. It was unactionable for its primary
    recipient — the same defect as the mover card, in a different
    domain, and it is the reason that sweep is switched off.

    Scoped to the grant's OWN scope, never to cloud grants at large: a
    scope owner is the person who knows why access to *their* scope
    exists, which is exactly the claim being made when we ask them.

    Refusal is 403, not 404: unlike a mover packet, the existence of a
    cloud grant is not itself a disclosure about a colleague's org move,
    and the caller here reached a specific id they were told about.
    """
    from app.core.rbac import GLOBAL_ADMIN_ROLE, canonicalize_role_name

    roles = {canonicalize_role_name(r) for r in (user.role_names or [])}
    if GLOBAL_ADMIN_ROLE in roles:
        return

    owner_id: str | None = None
    try:
        scope_uuid = UUID(str(grant.scope_id))
    except (TypeError, ValueError):
        scope_uuid = None
    if scope_uuid is not None:
        owner_id = await db.scalar(
            select(CmdbCloudScope.owner_person_id).where(
                CmdbCloudScope.id == scope_uuid,
            )
        )
    # ``owner_person_id`` is a CHAR(36) string column with no GUID
    # adapter, so compare as strings (memory: a UUID bind against such a
    # column silently matches nothing — here it would silently 403 the
    # very person the card was sent to).
    if owner_id and str(owner_id) == str(user.id):
        return

    raise AppException(
        code="NOT_SCOPE_OWNER",
        message=(
            "only a global admin or the owner of this grant's cloud scope "
            "may resolve its review"
        ),
        status_code=403,
    )


@router.put(
    "/grants/{grant_id}/review",
    summary="Resolve a pending_review grant to keep or revoke",
)
async def resolve_grant_review(
    grant_id: UUID,
    body: _ReviewBody,
    request: Request,
    # NOT ``require_admin()``: the scope OWNER may decide about grants on
    # their own scope. See ``_authorize_grant_review`` for why — the card
    # that asks them was unactionable without it, for as long as the
    # sweep has existed (migration 236).
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    grant = await db.scalar(
        select(CmdbCloudAccessGrant).where(
            CmdbCloudAccessGrant.id == grant_id,
        )
    )
    if grant is None:
        raise NotFoundError("CmdbCloudAccessGrant", str(grant_id))
    await _authorize_grant_review(db, grant, current_user)
    if grant.review_status not in (
        REVIEW_STATUS_PENDING,
        REVIEW_STATUS_CONFIRMED_KEEP,
        REVIEW_STATUS_CONFIRMED_REVOKE,
    ):
        # Grant either was never flagged, or was already finalised.
        # Idempotent re-decision (keep→keep or revoke→revoke) is OK;
        # flipping keep→revoke is also OK (admin changes mind);
        # but a grant that was NEVER flagged shouldn't suddenly be
        # marked confirmed_keep here — the admin clicked the wrong
        # row. Surface 409 so the UI can refetch.
        raise HTTPException(
            status_code=409,
            detail="grant has no pending review_status to resolve",
        )
    # Cloud-side terminal state guard. Once ``status='revoked'`` the
    # cloud-side IAM binding has been torn down (or scheduled for the
    # next cleanup tick). Letting the admin flip review_status back
    # to ``confirmed_keep`` here would put the row into an incoherent
    # state — the page would render "kept" but the cloud says
    # otherwise. Refuse the flip; the admin must re-grant via the
    # standard request flow if they want access back.
    if grant.status == "revoked" and body.decision == "confirmed_keep":
        raise HTTPException(
            status_code=409,
            detail=(
                "grant is already revoked cloud-side; cannot mark "
                "confirmed_keep. Re-issue via the standard request flow."
            ),
        )

    now = datetime.now(UTC)
    new_status = (
        REVIEW_STATUS_CONFIRMED_REVOKE
        if body.decision == "confirmed_revoke"
        else REVIEW_STATUS_CONFIRMED_KEEP
    )
    grant.review_status = new_status
    grant.review_status_changed_at = now

    if body.decision == "confirmed_revoke" and grant.revoked_at is None:
        # Route through the REAL revoke path so the cloud-side IAM binding is
        # actually torn down (``driver.revoke_access``). The old code only
        # flipped ``status='revoked'`` on the ORM row and claimed a cron would
        # pick it up — but the only cron that acts on revoked rows
        # (``sweep_pending_manual_cloud_revokes``) filters on
        # ``detail.requires_manual_cloud_revoke == true``, which a bare flip
        # never set. Net effect: the person kept LIVE cloud access forever
        # while every Helix surface (+ a status='ok' audit row) said "revoked".
        # ``suppress_driver_error=True`` preserves the Helix invariant (row
        # flips to revoked even if the provider rejects us) AND makes
        # ``_revoke_grant_locked`` set ``requires_manual_cloud_revoke`` on
        # driver failure so the sweep retries the teardown.
        from app.domains.cmdb.lifecycle_service import CloudLifecycleService

        await CloudLifecycleService(db).revoke_grant(
            str(grant.id),
            reason="role_change_confirmed_revoke",
            actor_user_id=current_user.id,
            trace_id=request.headers.get("x-trace-id", "role_change_review"),
            suppress_driver_error=True,
            actor_role_names=list(current_user.role_names or []),
        )

    audit = AuditService(db)
    await audit.log(
        trace_id=request.headers.get("x-trace-id", "role_change_review"),
        actor_type="user",
        actor_user_id=current_user.id,
        action=f"role_change.grant_review_{new_status}",
        resource_type="cmdb_cloud_access_grant",
        resource_id=str(grant.id),
        status="ok",
        detail={
            "decision": body.decision,
            "note": body.note,
            "scope_id": grant.scope_id,
            "role_key": grant.role_key,
        },
        ip_address=request.client.host if request.client else None,
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()

    return success_response(
        data={
            "id": str(grant.id),
            "review_status": grant.review_status,
            "status": grant.status,
        },
        request=request,
    )


# ─── Review policy + baseline coverage ──────────────────────────────
#
# One endpoint rather than two, because the page needs both numbers to
# say anything honest. "Pushes are off" on its own reads like a setting
# somebody forgot to turn on; "pushes are off because the platform holds
# 0 baseline rules, so the only number we could put on a card would be
# an inventory count" is a statement an operator can act on.
#
# Single-segment path — no collision with ``/{person_id}/grants``.


class _ReviewConfigBody(BaseModel):
    employee_push_enabled: bool = False
    require_baseline_for_push: bool = True
    push_baseline_sources: list[str] = Field(
        default_factory=lambda: list(DEFAULT_PUSH_BASELINE_SOURCES),
    )
    min_delta_items_to_push: int = Field(default=1, ge=1, le=1000)
    stale_grant_push_enabled: bool = False
    stale_grant_days: int = Field(default=30, ge=7, le=365)


async def _baseline_coverage(db: AsyncSession) -> dict[str, object]:
    """How much of a baseline the platform actually holds.

    ``departments_with_people`` counts departments that have at least one
    active employee, because a policy for a department nobody is in
    contributes nothing to a mover review. A wildcard policy
    (``department_name IS NULL``) applies to everybody and is counted
    separately rather than folded into the coverage ratio — it is a
    company-wide default, not per-department knowledge.
    """
    total_policies = await db.scalar(
        select(func.count()).select_from(ProvisionPolicy),
    ) or 0
    active_policies = await db.scalar(
        select(func.count())
        .select_from(ProvisionPolicy)
        .where(ProvisionPolicy.is_active.is_(True)),
    ) or 0
    wildcard_policies = await db.scalar(
        select(func.count())
        .select_from(ProvisionPolicy)
        .where(
            ProvisionPolicy.is_active.is_(True),
            ProvisionPolicy.department_name.is_(None),
        ),
    ) or 0
    covered_departments = await db.scalar(
        select(func.count(func.distinct(ProvisionPolicy.department_name)))
        .where(
            ProvisionPolicy.is_active.is_(True),
            ProvisionPolicy.department_name.is_not(None),
        ),
    ) or 0
    # ``IdentityPerson.primary_department`` is a Python property over the
    # org-membership rows, not a column — so the count has to go through
    # the membership table, matching exactly what the property returns
    # (the PRIMARY, non-deleted membership's ``org_node_name``). That is
    # also the value ``provision_policies.department_name`` is compared
    # against, so the two halves of this ratio speak about the same
    # thing.
    departments_with_people = await db.scalar(
        select(func.count(func.distinct(IdentityOrgMembership.org_node_name)))
        .select_from(IdentityOrgMembership)
        .join(
            IdentityPerson,
            IdentityPerson.id == IdentityOrgMembership.person_id,
        )
        .where(
            IdentityOrgMembership.is_primary.is_(True),
            IdentityOrgMembership.deleted_at.is_(None),
            IdentityOrgMembership.org_node_name.is_not(None),
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status == "active",
        ),
    ) or 0
    return {
        "total_policies": int(total_policies),
        "active_policies": int(active_policies),
        "wildcard_policies": int(wildcard_policies),
        "covered_departments": int(covered_departments),
        "departments_with_people": int(departments_with_people),
        # The page renders this verbatim rather than recomputing: the
        # rule "a baseline exists" must have exactly one definition.
        "has_baseline": int(active_policies) > 0,
        "configure_url": "/admin/lifecycle/onboarding?tab=strategy",
    }


@router.get(
    "/review-config",
    summary="Mover-review push policy and how much baseline backs it",
)
async def get_review_config(
    request: Request,
    current_user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    settings = await load_review_settings(db)
    return success_response(
        data={
            "settings": settings.to_setting(),
            "baseline": await _baseline_coverage(db),
        },
        request=request,
    )


@router.put(
    "/review-config",
    summary="Change the mover-review push policy",
)
async def put_review_config(
    body: _ReviewConfigBody,
    request: Request,
    current_user: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Write the policy, refusing the combinations that caused the incident.

    ``RoleChangeReviewSettings.validate`` is what refuses; it is called
    inside ``save_review_settings`` so the HTTP surface and any future
    programmatic caller cannot disagree about what is allowed.
    """
    previous = await load_review_settings(db)
    settings = RoleChangeReviewSettings.from_setting(body.model_dump())
    await save_review_settings(db, settings)
    await AuditService(db).log(
        trace_id=request.headers.get("x-trace-id", "role_change_review_config"),
        actor_type="user",
        actor_user_id=current_user.id,
        action="role_change.review_config.updated",
        status="success",
        resource_type="platform_setting",
        resource_id=SETTING_KEY,
        detail={
            "before": previous.to_setting(),
            "after": settings.to_setting(),
        },
        ip_address=request.client.host if request.client else None,
        user_agent=request.headers.get("user-agent"),
    )
    await db.commit()
    return success_response(
        data={
            "settings": settings.to_setting(),
            "baseline": await _baseline_coverage(db),
        },
        request=request,
    )

#: Upper bound on the preflight's scan. Far above ``review_notify``'s
#: own per-run ``limit`` (50), so the estimate can never under-report
#: what a single tick would send, while still refusing to become an
#: unbounded table scan behind a dialog.
_PREFLIGHT_SCAN_CAP = 2000


async def _push_preflight(
    db: AsyncSession, settings: RoleChangeReviewSettings,
) -> dict[str, Any]:
    """If the push were switched on right now, what would actually go out?

    Turning notifications back on has been, until now, a leap: the only
    way to learn the blast radius was to take it. That is how nine cards
    reached a CTO and a CEO on one morning.

    So this answers the question **without sending anything**, using the
    same predicate the sender uses — ``settings.may_push`` — rather than
    a second copy of the rule that could drift into optimism. It reports
    both halves: what would be sent, and what would be held back and
    why. A preflight that only counted the sends would hide exactly the
    number an operator needs to sanity-check.

    Withheld packets are excluded from every count here, as they are in
    the sender: they are permanently out of the queue.
    """
    from app.domains.hr.models import ROLE_REVIEW_OPEN, HrRoleChangeReview

    # Bounded. An unbounded scan of the open queue is fine at today's
    # dozen packets and is a page-load timeout at a thousand — and this
    # runs from a dialog somebody opened, not a cron. The cap is
    # deliberately far above the notify job's own ``limit`` (50) so the
    # preflight can never under-report what one tick would send.
    candidates = (
        await db.execute(
            select(HrRoleChangeReview)
            .where(
                HrRoleChangeReview.status == ROLE_REVIEW_OPEN,
                HrRoleChangeReview.notified_at.is_(None),
                HrRoleChangeReview.notify_suppressed_at.is_(None),
                HrRoleChangeReview.reviewable_items > 0,
            )
            .order_by(HrRoleChangeReview.created_at.asc())
            .limit(_PREFLIGHT_SCAN_CAP)
        )
    ).scalars().all()

    would_send: list[HrRoleChangeReview] = []
    blocked_no_delta = 0
    blocked_no_baseline = 0
    blocked_source_rejected = 0
    for review in candidates:
        if settings.may_push(
            baseline_source=review.baseline_source,
            delta_items=review.delta_items,
        ):
            would_send.append(review)
        elif review.baseline_source == "none":
            blocked_no_baseline += 1
        elif review.delta_items < settings.min_delta_items_to_push:
            # Checked against a real baseline and found nothing (or too
            # little). Tested BEFORE the source bucket on purpose: a
            # packet that is both under the floor and from an unaccepted
            # source would not be sent even if the source were accepted,
            # so calling it "source rejected" would point the operator at
            # a lever that changes nothing.
            blocked_no_delta += 1
        else:
            # A real delta, from a source this org has chosen not to act
            # on — the one bucket the operator can clear from the dialog
            # they are standing in. Folding it into "no delta" would say
            # the platform found nothing, when in fact it was configured
            # not to care.
            blocked_source_rejected += 1

    # WHO would be interrupted, resolved to names. The count alone is
    # not the thing an operator needs to look at — "9 cards" reads very
    # differently from "9 cards, 5 of them to the CTO".
    reviewer_ids = {
        r.reviewer_person_id for r in would_send if r.reviewer_person_id
    }
    names: dict[Any, str] = {}
    if reviewer_ids:
        names = {
            pid: (name or "")
            for pid, name in (
                await db.execute(
                    select(IdentityPerson.id, IdentityPerson.display_name)
                    .where(IdentityPerson.id.in_(reviewer_ids))
                )
            ).all()
        }
    per_person: dict[str, int] = {}
    unrouted = 0
    for review in would_send:
        if review.reviewer_person_id:
            label = names.get(review.reviewer_person_id) or str(
                review.reviewer_person_id,
            )
            per_person[label] = per_person.get(label, 0) + 1
        else:
            # Falls back to the whole global-admin pool at send time, so
            # it is one packet but N cards. Counted separately because
            # burying it in the per-person tally would understate it.
            unrouted += 1

    return {
        "open_candidates": len(candidates),
        # True when the queue is deeper than the scan cap: the numbers
        # below are then a floor, not a total, and the dialog says so
        # rather than quietly under-reporting.
        "truncated": len(candidates) >= _PREFLIGHT_SCAN_CAP,
        "would_send": len(would_send),
        "blocked_no_baseline": blocked_no_baseline,
        "blocked_no_delta": blocked_no_delta,
        "blocked_source_rejected": blocked_source_rejected,
        "unrouted_to_admin_pool": unrouted,
        "recipients": [
            {"name": name, "packets": n}
            for name, n in sorted(
                per_person.items(), key=lambda kv: -kv[1],
            )[:20]
        ],
        # The switch's current position, so the caller can tell a
        # hypothetical ("if I turned it on") from a description of what
        # is happening now.
        "push_enabled": settings.employee_push_enabled,
    }


@router.get(
    "/review-config/preflight",
    summary="What would be sent if the employee push were switched on",
)
async def get_review_config_preflight(
    request: Request,
    current_user: AuthUser = Depends(require_admin_read()),
    db: AsyncSession = Depends(get_db),
):
    """A dry run of the send, evaluated with the push forced ON.

    The saved setting is deliberately overridden for the simulation —
    the operator is asking "what happens if I turn this on", and
    answering with the current (off) state would report zero and teach
    them nothing.
    """
    saved = await load_review_settings(db)
    simulated = RoleChangeReviewSettings.from_setting(
        {**saved.to_setting(), "employee_push_enabled": True},
    )
    return success_response(
        data={
            "settings": saved.to_setting(),
            "preflight": await _push_preflight(db, simulated),
        },
        request=request,
    )

@router.get(
    "/my-scope-grants",
    summary="pending_review grants on cloud scopes I own",
)
async def list_my_scope_grants(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """The scope owner's half of the stale-grant chase.

    ``role_change.scope_review_sweep`` has always sent its card to the
    scope OWNER as well as to IT — and a scope owner is frequently an
    ordinary employee. Until now the card pointed at an admin-gated
    console and asked them to use an endpoint that answered 403: it
    asked somebody to decide at an address where they could not decide.
    That is the same defect the mover card had, and the reason this
    sweep stays switched off.

    So the owner gets what the new manager got — the work in their own
    inbox, listing only the grants on scopes they own. Narrowing by
    ownership is the whole point: a scope owner is the person who knows
    why access to *their* scope exists, which is exactly the claim being
    made when we ask them.
    """
    owned = (
        await db.execute(
            select(CmdbCloudScope.id, CmdbCloudScope.display_name,
                   CmdbCloudScope.external_id)
            .where(
                CmdbCloudScope.owner_person_id == str(current_user.id),
                CmdbCloudScope.deleted_at.is_(None),
            )
        )
    ).all()
    if not owned:
        return success_response(data={"grants": []}, request=request)

    meta = {str(sid): (name, ext) for sid, name, ext in owned}
    grants = (
        await db.execute(
            select(CmdbCloudAccessGrant)
            .where(
                # ``scope_id`` is a CHAR(36) string column with no GUID
                # adapter — bind strings or the IN silently matches
                # nothing, which here would show the owner an empty
                # queue while the card insists there is work.
                CmdbCloudAccessGrant.scope_id.in_(list(meta.keys())),
                CmdbCloudAccessGrant.review_status == REVIEW_STATUS_PENDING,
                CmdbCloudAccessGrant.status.in_(("active", "pending")),
            )
            .order_by(CmdbCloudAccessGrant.review_status_changed_at.asc())
            .limit(500)
        )
    ).scalars().all()

    holder_ids = []
    for g in grants:
        try:
            holder_ids.append(UUID(str(g.person_id)))
        except (TypeError, ValueError):
            continue
    holders: dict[str, str] = {}
    if holder_ids:
        holders = {
            str(pid): (name or "")
            for pid, name in (
                await db.execute(
                    select(IdentityPerson.id, IdentityPerson.display_name)
                    .where(IdentityPerson.id.in_(holder_ids))
                )
            ).all()
        }

    return success_response(
        data={
            "grants": [
                {
                    "id": str(g.id),
                    "scope_id": g.scope_id,
                    "scope_display_name": meta.get(str(g.scope_id), (None, None))[0],
                    "scope_external_id": meta.get(str(g.scope_id), (None, None))[1],
                    "role_key": g.role_key,
                    "role_display_name": g.role_display_name,
                    "holder_name": holders.get(str(g.person_id)),
                    "review_status": g.review_status,
                    "review_status_changed_at": (
                        g.review_status_changed_at.isoformat()
                        if g.review_status_changed_at else None
                    ),
                    "granted_at": g.granted_at.isoformat() if g.granted_at else None,
                }
                for g in grants
            ],
        },
        request=request,
    )