from uuid import UUID

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

from app.audit.service import AuditService
from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import (
    check_approval_submit_rate_limit,
    check_sensitive_endpoint_rate_limit,
    get_current_user,
    get_real_ip,
    require_asset_admin,
    require_browser_session,
)
from app.core.dependencies import (  # noqa: E402
    require_active_employee_for_mutations as _leaver_runtime_gate,
)
from app.core.exceptions import ForbiddenError
from app.core.rbac import can_manage_assets
from app.core.response import paginated_response, success_response
from app.domains.apikey.schemas import (
    ApiKeyExtendRequest,
    ApiKeyFillRequest,
    ApiKeyIamRebindRequest,
    ApiKeyMemberAddRequest,
    ApiKeyProviderCreate,
    ApiKeyProviderUpdate,
    ApiKeyReleaseDecisionBody,
    ApiKeyReleaseRequest,
    ApiKeyRequestSubmit,
    ApiKeyRevokeRequest,
    ApiKeyRotateRequest,
    ApiKeyTransferRequestSubmit,
    AssignmentTransferOwner,
    AssignmentUpdate,
)
from app.domains.apikey.service import ApiKeyService

router = APIRouter(
    prefix="/api/v1/apikeys",
    tags=["apikeys"],
    dependencies=[Depends(_leaver_runtime_gate)],
)


# ─── Provider endpoints (admin) ───

@router.get("/providers")
async def list_providers(
    request: Request,
    include_inactive: bool = Query(False),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = ApiKeyService(db)
    # Employees can only see active providers
    show_inactive = include_inactive and can_manage_assets(current_user.role_names)
    providers = await service.list_providers(include_inactive=show_inactive)
    return success_response(data=providers, request=request)


@router.post("/providers")
async def create_provider(
    request: Request,
    body: ApiKeyProviderCreate,
    admin: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    service = ApiKeyService(db)
    provider = await service.create_provider(body)
    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=admin.id,
        actor_type="user",
        action="apikey.provider.create",
        resource_type="apikey_provider",
        resource_id=str(provider.id),
        status="success",
        detail=body.model_dump(mode="json", exclude_unset=True),
        ip_address=get_real_ip(request),
    )
    await db.commit()
    return success_response(data=provider, request=request)


@router.patch("/providers/{provider_id}")
async def update_provider(
    request: Request,
    provider_id: UUID,
    body: ApiKeyProviderUpdate,
    admin: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    service = ApiKeyService(db)
    provider = await service.update_provider(provider_id, body)
    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=admin.id,
        actor_type="user",
        action="apikey.provider.update",
        resource_type="apikey_provider",
        resource_id=str(provider_id),
        status="success",
        detail=body.model_dump(mode="json", exclude_unset=True),
        ip_address=get_real_ip(request),
    )
    await db.commit()
    return success_response(data=provider, request=request)


@router.delete("/providers/{provider_id}")
async def delete_provider(
    request: Request,
    provider_id: UUID,
    admin: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    service = ApiKeyService(db)
    await service.delete_provider(provider_id)
    audit = AuditService(db)
    await audit.log(
        trace_id=request.state.trace_id,
        actor_user_id=admin.id,
        actor_type="user",
        action="apikey.provider.delete",
        resource_type="apikey_provider",
        resource_id=str(provider_id),
        status="success",
        ip_address=get_real_ip(request),
    )
    await db.commit()
    return success_response(data=None, request=request)


# ─── Assignment endpoints (employee + admin) ───

@router.get("/my")
async def get_my_api_keys(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = ApiKeyService(db)
    keys = await service.get_my_api_keys(current_user)
    return success_response(data=keys, request=request)


@router.get("/assignments")
async def list_assignments(
    request: Request,
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    status: str | None = Query(None),
    provider_id: UUID | None = Query(None),
    assigned_to: UUID | None = Query(None),
    keyword: str | None = Query(
        None,
        description="Fuzzy match request_number / project_name / assignee name / assignee email",
    ),
    start_date: str | None = Query(None),
    end_date: str | None = Query(None),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = ApiKeyService(db)
    assignments, total = await service.list_assignments(
        current_user=current_user,
        page=page,
        page_size=page_size,
        status=status,
        provider_id=provider_id,
        assigned_to=assigned_to,
        keyword=keyword,
        start_date=start_date,
        end_date=end_date,
    )
    return paginated_response(
        data=assignments, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/assignments/export.xlsx")
async def export_assignments(
    request: Request,
    status: str | None = Query(None),
    provider_id: UUID | None = Query(None),
    assigned_to: UUID | None = Query(None),
    keyword: str | None = Query(
        None,
        description="Fuzzy match request_number / project_name / assignee name / assignee email",
    ),
    start_date: str | None = Query(None),
    end_date: str | None = Query(None),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Export the filtered API-key assignment list as XLSX.

    Filter signature mirrors :func:`list_assignments` minus pagination —
    the UI passes the exact query params the admin table is showing, so
    the download matches the screen. RBAC mirrors the list endpoint too:
    ``current_user`` is threaded into ``ApiKeyService.list_assignments``
    whose internal scope branch gives admins / management / auditor the
    full table and everyone else only their own (+ shared-roster /
    subordinate) rows — an employee's export naturally contains only
    their own assignments. Key material stays masked exactly as on the
    list (``key_masked``); the plaintext never leaves the reveal flow.
    Capped at :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows.
    """
    from app.core.exporting import (  # noqa: PLC0415
        EXPORT_ROW_LIMIT,
        xlsx_export_response,
    )

    service = ApiKeyService(db)
    assignments, total = await service.list_assignments(
        current_user=current_user,
        page=1,
        page_size=EXPORT_ROW_LIMIT,
        status=status,
        provider_id=provider_id,
        assigned_to=assigned_to,
        keyword=keyword,
        start_date=start_date,
        end_date=end_date,
    )
    columns = [
        ("request_number", "申请号"),
        ("provider_name", "提供方"),
        ("assigned_to_name", "分配对象"),
        ("assigned_to_email", "邮箱"),
        ("project_name", "项目"),
        ("reason", "用途"),
        ("status", "状态"),
        ("monthly_budget_usd", "月预算(USD)"),
        ("key_masked", "密钥掩码"),
        ("cloud_scope_display_name", "云范围"),
        ("valid_from", "生效日"),
        ("valid_until", "到期日"),
        ("created_at", "分配时间"),
        ("revoked_at", "撤销时间"),
        ("id", "ID"),
    ]
    rows = [a.model_dump() for a in assignments]
    return await xlsx_export_response(
        db,
        request,
        current_user,
        columns=columns,
        rows=rows,
        total_matched=total,
        filename_stem="apikey_assignments",
        sheet_title="API Key Assignments",
        audit_action="apikey.assignments.export",
        audit_resource_type="apikey_assignment",
        filters={
            "status": status,
            "provider_id": provider_id,
            "assigned_to": assigned_to,
            "keyword": keyword,
            "start_date": start_date,
            "end_date": end_date,
        },
    )


@router.get("/analytics")
async def get_apikey_analytics(
    request: Request,
    month: str | None = Query(
        None,
        description="Billing month YYYY-MM. Defaults to current month.",
        pattern=r"^\d{4}-(0[1-9]|1[0-2])$",
    ),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Org-wide API-Key analytics for the admin Reports tab.

    Returns distribution (by provider / status / cost-attribution /
    department / owner), the committed monthly-budget pool, actual
    month-to-date cloud spend (using a bounded-query batch projection
    of the same attribution rules as the per-key panel + budget cron),
    and lifecycle/volume metrics (created/revoked trend, validity
    buckets, revoke breakdown, shared-roster stats).

    Auth: read-side admin — ``can_manage_assets`` (asset/global admin)
    OR ``management`` (read-only leadership) OR ``auditor``. Anyone
    else: 403. Read-only — no audit row, same as the ``/spend`` family.
    """
    from app.core.exceptions import ForbiddenError
    from app.core.rbac import AUDITOR_ROLE, MANAGEMENT_ROLE, has_role

    if not (
        can_manage_assets(current_user.role_names)
        or has_role(current_user.role_names, MANAGEMENT_ROLE, AUDITOR_ROLE)
    ):
        raise ForbiddenError(
            "API Key analytics is restricted to asset/global admins, "
            "management, and auditors.",
        )

    from app.domains.apikey.analytics import compute_apikey_analytics

    result = await compute_apikey_analytics(db, month=month)
    return success_response(data=result, request=request)


@router.post("/requests")
async def submit_api_key_request(
    request: Request,
    body: ApiKeyRequestSubmit,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Round-31 audit S1 (2026-05-18): 30 req/h/user. Without this an
    # authenticated employee could script-spam Feishu approval cards
    # into IT's inbox and write thousands of ``api_key_assignments``
    # rows. Same posture as the DNS / domain-registration submit
    # endpoints (already gated), shared Redis sliding-window helper.
    _rl: None = Depends(check_approval_submit_rate_limit),
):
    service = ApiKeyService(db)
    result = await service.submit_api_key_request(
        body=body,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/fill-key")
async def fill_key(
    request: Request,
    assignment_id: UUID,
    body: ApiKeyFillRequest,
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    service = ApiKeyService(db)
    result = await service.fill_key(
        assignment_id, body,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/bailian/reissue")
async def reissue_bailian(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
    # Round-49: mint mints fresh DashScope sk- keys, the only path
    # that produces revealable plaintext. /reveal admin-on-behalf-of
    # is already browser-gated (line 337-338); closing this PAT lane
    # mirrors the same posture so a stolen asset_admin PAT cannot
    # mint + exfiltrate fresh credentials.
    _gate: None = Depends(require_browser_session),
):
    """Re-mint the DashScope sk- key on an already-granted Bailian
    assignment.

    Migration helper for assignments granted under the legacy
    AK/SK-bundle code path (pre-2026-05-18 fix): the DashScope SDK
    rejects RAM AK/SK with ``No api key provided``, so those rows have
    a key_value that decrypts but isn't usable. This endpoint mints a
    fresh sk- API key via ModelStudio CreateApiKey and replaces the
    bundle; the next reveal hands the employee a working credential
    without forcing them to re-apply.

    Admin-only. Each call mints a brand-new sk- (the operation is NOT
    idempotent on the Aliyun side), so the admin UI should confirm
    before triggering. The legacy api_key_id (if any) is preserved on
    ``detail.bailian.legacy_api_key_id`` for audit.
    """
    service = ApiKeyService(db)
    result = await service.reissue_bailian_api_key(
        assignment_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/reveal")
async def reveal_key(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _rate: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Decrypt + return one assignment's key plaintext.

    Three callers:
      * The owner (``assigned_to == current_user.id``) — self-reveal.
        Allowed via PAT (the AI client is showing the user their own
        cloud key) and via JWT.
      * A roster *member* the Owner has added under
        ``apikey_assignment_members`` (migration 200) — same access
        surface as the Owner for shared-access usage. PAT-allowed:
        a teammate's CLI / script is the canonical caller and forcing
        browser-session would defeat the shared-credential workflow.
      * An asset-admin revealing someone else's assignment (and who is
        NOT a roster member). Require a Google-SSO browser session: a
        stolen IT-Admin PAT must NOT be able to dump every employee's
        cloud-IAM credential set. Same posture as PII reveal + role
        mutation.
    """
    # Determine ownership BEFORE the service so we can require browser
    # auth on the admin-on-behalf-of branch only. ``_get_assignment`` is
    # a private helper but we replicate its PK lookup here to keep the
    # gate at the controller.
    #
    # Security note (Round-31 audit B12): collapse the "row not found"
    # path with the "row found but you can't see it" path for non-admin
    # callers — emitting distinct 404 vs 403 lets an attacker probe
    # which assignment UUIDs exist on the platform. UUIDs are
    # unguessable in practice, but the principle is the same as the
    # ``reveal`` doc: a stolen non-admin PAT should not be able to
    # enumerate. Admins keep the 404 signal because they have
    # legitimate "did this row ever exist?" use cases (e.g. checking
    # whether a deletion was applied).
    from app.core.rbac import can_manage_assets
    from app.domains.apikey.models import ApiKeyAssignment

    is_admin = can_manage_assets(current_user.role_names)
    target = await db.get(ApiKeyAssignment, assignment_id)
    if target is None:
        if is_admin:
            from app.core.exceptions import NotFoundError
            raise NotFoundError("ApiKeyAssignment", str(assignment_id))
        # Non-admin probe of a non-existent UUID: return the same 403
        # they'd get for a row they don't own (so existence isn't
        # disclosed), AND write an audit row so the probe leaves a
        # forensic trail. The service layer's ``reveal_denied`` audit
        # only fires when the row exists; without this branch a
        # scripted UUID-enumeration attack leaves NO audit evidence.
        audit = AuditService(db)
        await audit.log(
            trace_id=request.state.trace_id,
            actor_type="user",
            actor_user_id=current_user.id,
            action="apikey.key.reveal_denied",
            resource_type="api_key_assignment",
            resource_id=str(assignment_id),
            status="failure",
            detail={
                "reason": "not-found-non-admin-probe",
            },
            ip_address=get_real_ip(request),
            user_agent=request.headers.get("User-Agent"),
        )
        await db.commit()
        from app.core.exceptions import ForbiddenError
        raise ForbiddenError(
            "Only the assignment owner or an asset admin may reveal "
            "this key.",
        )
    if target.assigned_to != current_user.id:
        # Members of the shared-access roster reveal without browser-
        # session (the Owner explicitly granted them this surface; a
        # PAT call from their CLI is the legitimate flow). Admin-on-
        # behalf-of (non-member admin) still requires a browser
        # session — same threat-model as before.
        service_peek = ApiKeyService(db)
        is_member = await service_peek.is_member(
            target.id, current_user.id,
        )
        if not is_member:
            from app.core.dependencies import require_browser_session
            require_browser_session(request)

    service = ApiKeyService(db)
    key_value = await service.reveal_key(
        assignment_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data={"key_value": key_value}, request=request)


@router.get("/assignments/{assignment_id}/usage-hints")
async def get_assignment_usage_hints(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Non-secret "how to use this key" hints for the reveal UI.

    Returns the vendor kind, the 存量纳管 (adopted) provenance flag, and —
    for Azure OpenAI — the live list of model deployments IT created on
    the Foundry account so the employee sees the real deployment name to
    call. Returns NO key material, so (unlike ``/reveal``) it needs no
    browser-session step-up; owner / roster-member / asset-admin only.

    Existence is not disclosed to unauthorised callers: a non-admin who
    is neither owner nor member gets the same 403 whether or not the row
    exists (mirrors ``/reveal``).
    """
    from app.core.exceptions import ForbiddenError, NotFoundError
    from app.core.rbac import can_manage_assets
    from app.domains.apikey.models import ApiKeyAssignment

    is_admin = can_manage_assets(current_user.role_names)
    target = await db.get(ApiKeyAssignment, assignment_id)
    if target is None:
        if is_admin:
            raise NotFoundError("ApiKeyAssignment", str(assignment_id))
        raise ForbiddenError(
            "Only the assignment owner or an asset admin may view this "
            "key's usage hints.",
        )
    if target.assigned_to != current_user.id and not is_admin:
        service_peek = ApiKeyService(db)
        if not await service_peek.is_member(target.id, current_user.id):
            raise ForbiddenError(
                "Only the assignment owner or an asset admin may view "
                "this key's usage hints.",
            )

    service = ApiKeyService(db)
    result = await service.usage_hints(
        assignment_id,
        current_user=current_user,
    )
    return success_response(data=result, request=request)


@router.get("/assignments/{assignment_id}/spend")
async def get_assignment_spend(
    request: Request,
    assignment_id: UUID,
    month: str | None = Query(
        None,
        description="Billing month YYYY-MM. Defaults to current month.",
        pattern=r"^\d{4}-(0[1-9]|1[0-2])$",
    ),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Return month-to-date cloud spend for the assignment's applicant.

    Auth: applicant themselves OR an asset_admin / it_admin. Anyone
    else: 403 (see :meth:`ApiKeyService.get_assignment_spend`).

    Same query the budget-alert cron uses, so the dashboard's
    progress bar agrees with the alerts to the cent. Reads — does
    not mutate — so no audit row.
    """
    service = ApiKeyService(db)
    result = await service.get_assignment_spend(
        assignment_id,
        current_user=current_user,
        month=month,
    )
    return success_response(data=result, request=request)


@router.get("/assignments/{assignment_id}/spend/detail")
async def get_assignment_spend_detail(
    request: Request,
    assignment_id: UUID,
    month: str | None = Query(
        None,
        description="Billing month YYYY-MM. Defaults to current month.",
        pattern=r"^\d{4}-(0[1-9]|1[0-2])$",
    ),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Detail-page payload — superset of the basic ``/spend`` endpoint
    plus a 6-month MTD series (bar chart) and a per-``usage_type``
    breakdown (model × spend × quantity × effective unit price) for
    the requested month.

    Same filter recipe / auth model as
    :meth:`ApiKeyService.get_assignment_spend`. Bailian / missing-IAM
    / missing-RG buckets surface empty ``by_model`` and zero-filled
    ``monthly_series`` so the UI still has a domain to draw on.
    """
    service = ApiKeyService(db)
    result = await service.get_assignment_spend_detail(
        assignment_id,
        current_user=current_user,
        month=month,
    )
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/extend")
async def extend_validity(
    request: Request,
    assignment_id: UUID,
    body: ApiKeyExtendRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = ApiKeyService(db)
    result = await service.extend_validity(
        assignment_id, body,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.get("/assignments/{assignment_id}/notify-candidates")
async def list_notify_candidates(
    request: Request,
    assignment_id: UUID,
    q: str | None = Query(
        None,
        description=(
            "Admin-only free-text filter against identity_persons "
            "(display_name / work_email). Ignored for non-admin callers "
            "— they always get the curated manager-chain list. Min 2 "
            "chars to engage the search join."
        ),
        max_length=120,
    ),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Recipient candidates for the per-assignment edit modal.

    Auth: applicant themselves OR an asset_admin / it_admin. Non-admin
    owners receive only the curated ``applicant + manager chain`` list
    (matches the server-side restriction on ``update_assignment``);
    admin callers may additionally search the full ``identity_persons``
    table via ``?q=``.

    Replaces the modal's previous ``/identity-sync/lookups/persons``
    call which was ``require_admin``-gated — non-admin owners silently
    got a 403 + empty Select and could not self-edit their own AK
    notification list.
    """
    service = ApiKeyService(db)
    result = await service.list_notify_candidates(
        assignment_id,
        current_user=current_user,
        q=q,
        trace_id=request.state.trace_id,
    )
    return success_response(data=result, request=request)


@router.patch("/assignments/{assignment_id}")
async def update_assignment(
    request: Request,
    assignment_id: UUID,
    body: AssignmentUpdate,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Edit ``monthly_budget_usd`` and/or ``notify_recipients`` on an
    existing assignment.

    Authz: asset_admin / it_admin OR the assignment's applicant —
    enforced in the service. ``notify_recipients`` is tri-state:
    omit / null = default (applicant + manager), ``[]`` = suppress
    alerts, ``[<uuid>, ...]`` = exactly these persons.

    Distinct from the existing ``PATCH .../iam-binding`` (admin-only,
    re-targets the IAM/RAM chargeback owner) — that endpoint stays as-is.
    """
    service = ApiKeyService(db)
    result = await service.update_assignment(
        assignment_id, body,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.patch("/assignments/{assignment_id}/iam-binding")
async def rebind_iam_user(
    request: Request,
    assignment_id: UUID,
    body: ApiKeyIamRebindRequest,
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Re-target the IAM/RAM user this assignment is bound to (admin).

    Fully described on the schema (``ApiKeyIamRebindRequest``); the
    short version is: pick a different ``cmdb_iam_users`` row to be
    the canonical IAM/RAM user for this Helix-issued key. Pure
    book-keeping — does not rotate keys or call the cloud.
    """
    service = ApiKeyService(db)
    result = await service.rebind_iam_user(
        assignment_id, body,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/retry-iam-retag")
async def retry_iam_retag(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
    # Rate-limit: an admin retry loop on a permanently-broken
    # integration would otherwise burn AWS API quota and pile up
    # noise audit rows. Same sensitive-endpoint bucket as
    # ``/reveal`` and ``/bailian/reissue``.
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Re-run the AWS IAM tag rewrite when a transfer's best-effort
    tag-push failed.

    Reads ``detail.last_transfer.cloud_tag_summary``; if any bucket has
    ``status='failed'`` the admin sees a "retry" button in
    ``ApiKeyManagementPage`` which POSTs here. The endpoint re-invokes
    :meth:`ApiKeyService._retag_cloud_resources_on_transfer` for the
    current ``assigned_to`` and re-stamps the summary. Idempotent on
    success (AWS ``iam.tag_user`` is an upsert).

    Returns 200 with the refreshed assignment on success;
    ``IAM_RETAG_FAILED`` 502 when the underlying boto3 call still
    fails — the audit row + ``detail.last_transfer.retag_attempts``
    list capture the per-attempt history.
    """
    service = ApiKeyService(db)
    result = await service.retry_iam_retag(
        assignment_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/transfer-owner")
async def transfer_assignment_owner(
    request: Request,
    assignment_id: UUID,
    body: AssignmentTransferOwner,
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Admin-direct ownership transfer (reconciliation ownership /
    leaver-residual fix).

    Re-homes a key whose ``assigned_to`` points at a departed or wrong
    owner, reusing the full ``apply_transfer`` rewrite (local flip +
    cost-allocation rule swap + AWS owner-tag rewrite). Distinct from the
    self-service ``transfer-request`` peer-accept flow — this is the
    one-click admin path the reconciliation center invokes. The
    recipient must be an active employee (enforced in the service).
    """
    service = ApiKeyService(db)
    result = await service.transfer_assignment_owner(
        assignment_id, body.assigned_to,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/cost-rules/re-enable")
async def re_enable_cost_rules(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Re-enable the cost-allocation rules a prior revoke disabled
    (reconciliation cost lens).

    Inverse of the revoke-time auto-disable. Idempotent — rules already
    enabled report ``already_enabled``; missing rules report
    ``not_found``. Returns ``{rules_re_enabled, count, results}``.
    """
    service = ApiKeyService(db)
    result = await service.re_enable_allocation_rules(
        assignment_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/revoke")
async def revoke_key(
    request: Request,
    assignment_id: UUID,
    body: ApiKeyRevokeRequest | None = None,
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    service = ApiKeyService(db)
    result = await service.revoke_key(
        assignment_id,
        reason=body.reason if body else None,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
        reason_code=body.reason_code if body else None,
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/rotate")
async def rotate_key(
    request: Request,
    assignment_id: UUID,
    body: ApiKeyRotateRequest | None = None,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Rotation mints a real cloud credential — rate-limit the same
    # sliding window the other credential-touching endpoints use so a
    # script can't churn IAM keys / Bailian sk- / Azure regenerate.
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Self-service credential rotation (dual-active) for an
    auto-granted cloud API key.

    Owner-only (or asset / IT admin). Mints a fresh credential that goes
    live immediately; the old one keeps working for ``grace_days`` (0 /
    3 / 7 / 14), then the ``apikey.rotation_retire`` cron retires it.

    Does NOT return the new secret — reveal it through the standard
    ``/reveal`` endpoint (which keeps the browser-session gate for
    admin-on-behalf-of as the single plaintext choke point). See
    :meth:`ApiKeyService.rotate_key`.
    """
    service = ApiKeyService(db)
    result = await service.rotate_key(
        assignment_id,
        grace_days=body.grace_days if body else 7,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/rotate/complete")
async def complete_rotation(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Retire the old credential NOW ("我已切换,立即停用旧 key").

    Owner-only (or admin). Shortens the dual-active window.
    """
    service = ApiKeyService(db)
    result = await service.force_complete_rotation(
        assignment_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/rotate/cancel")
async def cancel_rotation(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Abort an in-flight rotation and revert to the prior credential.

    Owner-only (or admin). Invalidates the freshly-minted credential and
    restores the original ``key_value``.
    """
    service = ApiKeyService(db)
    result = await service.cancel_rotation(
        assignment_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/release-request")
async def submit_release_request(
    request: Request,
    assignment_id: UUID,
    body: ApiKeyReleaseRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Rate-limit the same Redis sliding window the original request
    # uses — release requests push a notification to the manager and
    # a script-spam pattern on this endpoint is identical to spam on
    # /requests.
    _rl: None = Depends(check_approval_submit_rate_limit),
):
    """Employee asks their manager to revoke a key (Helix-internal approval).

    See :meth:`ApiKeyService.submit_release_request` for the full
    state machine — TL;DR: ``active`` / ``expired`` →
    ``pending_release`` → (manager clicks Approve / Reject in Helix
    Web UI) → ``revoked`` or back to the prior status. Reveal is
    blocked while parked in ``pending_release``.

    Body requires ``reason_code`` (enum); ``reason_note`` required only
    when ``reason_code='other'`` (validated at the schema layer).
    """
    service = ApiKeyService(db)
    result = await service.submit_release_request(
        assignment_id, body,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.get("/release-requests")
async def list_release_requests(
    request: Request,
    role: str = Query(
        "approver",
        pattern="^(approver|requester)$",
        description=(
            "Whose list to show. 'approver' (default) returns "
            "requests where current_user is the manager / IT admin "
            "approver; 'requester' returns requests submitted by "
            "current_user."
        ),
    ),
    status: str | None = Query(
        None,
        description=(
            "Filter by status (pending / approved / rejected / "
            "cancelled). Default: pending for 'approver' role, "
            "no filter for 'requester' role."
        ),
    ),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """List API-Key release requests.

    Two views, selected by ``role`` query param:

    * ``role=approver`` (default) — pending requests waiting on
      *current_user*'s decision. Returns requests whose requester
      reports to ``current_user``; if ``current_user`` holds
      ``global_admin``, returns every pending request.

    * ``role=requester`` — requests submitted *by* current_user
      (any status), so the requester-side UI can show "my open
      releases" and offer cancel.
    """
    service = ApiKeyService(db)
    if role == "approver":
        effective_status = status if status is not None else "pending"
        rows = await service.list_release_requests_for_approver(
            current_user=current_user,
            status=effective_status,
        )
    else:
        rows = await service.list_release_requests_for_requester(
            current_user=current_user,
            status=status,
        )
    return success_response(data=rows, request=request)


@router.get("/apply-approvals")
async def list_apply_approvals(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Pending AI API-Key *apply* (申请) requests waiting on
    *current_user*'s approval in Feishu — the "待我审批" list backing the
    「API Key 申请审批」 todo tab.

    Unlike API-Key *release* (decided inside Helix), the apply approval is
    decided in the Feishu「审批」app. This endpoint only SURFACES the
    pending applies the caller is the current Feishu approver of, so the
    item is visible in Helix instead of buried in Feishu; the decision
    itself is still made in飞书审批 (no approve/reject here, per the
    no-third-party-approval decision).
    """
    service = ApiKeyService(db)
    rows = await service.list_pending_apply_approvals_for_approver(
        current_user=current_user,
    )
    return success_response(data=rows, request=request)


@router.get("/release-requests/{request_id}")
async def get_release_request(
    request: Request,
    request_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Fetch one release request. Visible to requester, requester's
    manager, or any IT Admin.
    """
    service = ApiKeyService(db)
    result = await service.get_release_request(
        request_id, current_user=current_user,
    )
    return success_response(data=result, request=request)


@router.post("/release-requests/{request_id}/approve")
async def approve_release_request(
    request: Request,
    request_id: UUID,
    body: ApiKeyReleaseDecisionBody | None = None,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Defence-in-depth rate limit on the approval surface. Operationally
    # managers approve ~1-2 per hour, so the sensitive-endpoint window
    # (same one used by /reveal, /retry-iam-retag) has plenty of
    # headroom. Adding this guards against a compromised-token replay
    # script approving every pending release in bulk.
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Manager / IT Admin approves a pending release request.

    Authz: caller must be the requester's direct manager OR hold
    ``global_admin``. On approve we synchronously call
    :meth:`ApiKeyService.revoke_key` with
    ``actor_role=self_with_approval`` so the cloud-side teardown
    runs immediately. The request row flips to ``approved``.
    """
    service = ApiKeyService(db)
    result = await service.approve_release_request(
        request_id,
        current_user=current_user,
        decision_note=body.decision_note if body else None,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/release-requests/{request_id}/reject")
async def reject_release_request(
    request: Request,
    request_id: UUID,
    body: ApiKeyReleaseDecisionBody | None = None,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Same rate-limit posture as approve — operationally light, but
    # protects the decision surface against bot replay.
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Manager / IT Admin rejects a pending release request.

    Reverts the assignment to its pre-release status. Same authz
    as :func:`approve_release_request`.
    """
    service = ApiKeyService(db)
    result = await service.reject_release_request(
        request_id,
        current_user=current_user,
        decision_note=body.decision_note if body else None,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


# ─── Shared-access roster (migration 200) ───
#
# An API Key's *Owner* (``ApiKeyAssignment.assigned_to``, the applicant)
# maintains a small list of teammates who get read-access to the same
# spend / detail / reveal surfaces. The three endpoints below are the
# Owner-facing surface. See :class:`ApiKeyAssignmentMember` for the
# schema, :meth:`ApiKeyService.is_owner_or_member` for the unified
# read-side authz gate, and :meth:`ApiKeyService.add_member` /
# ``remove_member`` / ``list_members`` for the per-action authz.


@router.get("/assignments/{assignment_id}/members")
async def list_assignment_members(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """List the roster of teammates with shared access to this key.

    Auth: Owner, any roster member, or any admin / management /
    auditor. Non-admin non-owner non-member callers get the same 403
    posture as the rest of the apikey read-side endpoints (not-found
    + foreign assignment surface identically — no UUID enumeration
    oracle).
    """
    service = ApiKeyService(db)
    rows = await service.list_members(
        assignment_id, current_user=current_user,
    )
    return success_response(data=rows, request=request)


@router.post("/assignments/{assignment_id}/members")
async def add_assignment_member(
    request: Request,
    assignment_id: UUID,
    body: ApiKeyMemberAddRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Same sliding-window cap as the budget-edit / extend endpoints
    # — bot-protecting the Owner's roster surface. An Owner spamming
    # add → notify-bomb a colleague would otherwise be unbounded.
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Owner adds a teammate to the shared-access roster.

    See :meth:`ApiKeyService.add_member` for the full state machine:
    Owner-only authz, "person is active" precondition, idempotent on
    duplicate (assignment, person), best-effort in-app + Feishu
    notification to the new member.
    """
    service = ApiKeyService(db)
    result = await service.add_member(
        assignment_id,
        person_id=body.person_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.delete("/assignments/{assignment_id}/members/{person_id}")
async def remove_assignment_member(
    request: Request,
    assignment_id: UUID,
    person_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
):
    """Owner removes a teammate from the roster.

    Idempotent on a non-existent membership row — returns 204-style
    empty success rather than 404 so the panel's bulk-remove UX
    doesn't have to special-case "already gone". The ex-member's
    next read attempt 403s; their next list call drops the row.
    """
    service = ApiKeyService(db)
    await service.remove_member(
        assignment_id,
        person_id=person_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=None, request=request)


@router.post("/assignments/{assignment_id}/transfer-request")
async def submit_transfer_request(
    request: Request,
    assignment_id: UUID,
    body: ApiKeyTransferRequestSubmit,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
    # Same rate-limit bucket as release / api_key request — the
    # downstream side-effects (creates an AssetTransferRequest, fires
    # a notification card to the recipient) are equivalent in spam
    # potential to a release submission.
    _rl: None = Depends(check_approval_submit_rate_limit),
):
    """Owner asks a colleague to take over an API Key.

    Unlike release, transfer does NOT tear down the cloud credential.
    The key keeps working through the handover. Delegates to
    :meth:`OffboardingService.create_transfer` which enforces "recipient
    active + not initiator" + "no duplicate pending xfer for this AK".

    The recipient sees the pending transfer at ``/my/transfers`` (or
    receives a Feishu / email card via the standard transfer
    notification path) and either accepts — which fires
    :meth:`ApiKeyService.apply_transfer` to rewrite the row, swap the
    cost-allocation rule, and push the new owner's email into the AWS
    IAM user's ``Owner`` tag — or rejects (key stays with the original
    owner).
    """
    service = ApiKeyService(db)
    result = await service.submit_transfer_request(
        assignment_id, body,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    await service.dispatch_transfer_notifications_after_commit(
        trace_id=str(request.state.trace_id),
    )
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/transfer-request/cancel")
async def cancel_transfer_request(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Owner withdraws their own pending transfer request.

    Idempotent: a key with no pending transfer (already accepted /
    rejected / cancelled) returns current state with no side effect.
    """
    service = ApiKeyService(db)
    result = await service.cancel_transfer_request(
        assignment_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


@router.post("/assignments/{assignment_id}/release-request/cancel")
async def cancel_release_request(
    request: Request,
    assignment_id: UUID,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Employee withdraws their own pending release request.

    Idempotent: a row that has already moved on (leader approved /
    rejected / employee cancelled previously) returns the current
    state with no side effects.
    """
    service = ApiKeyService(db)
    result = await service.cancel_release_request(
        assignment_id,
        current_user=current_user,
        trace_id=request.state.trace_id,
        ip_address=get_real_ip(request),
        user_agent=request.headers.get("User-Agent"),
    )
    await db.commit()
    return success_response(data=result, request=request)


# ─── Idle keys (migration 603) ───


class IdleKeepBody(BaseModel):
    """Holder / member says "still needed" — a reason is mandatory."""
    reason: str = Field(min_length=3, max_length=500)


class IdleConfigBody(BaseModel):
    """Partial update; unset fields keep their current value."""
    enabled_for_employees: bool | None = None
    min_age_days: int | None = Field(default=None, ge=7, le=365)
    keep_months: int | None = Field(default=None, ge=1, le=24)


@router.get("/idle")
async def list_idle_api_keys(
    request: Request,
    months: int = Query(2, ge=1, le=12),
    _user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Admin ledger of keys evaluated idle ($0 last calendar month) in the
    most recent ``months`` periods, longest streak first, with keep state."""
    from app.domains.apikey.idle_alerts import list_idle_assignments

    rows = await list_idle_assignments(db, months=months)
    return success_response(data=rows, request=request, meta={"total": len(rows)})


@router.get("/idle/config")
async def get_idle_config(
    request: Request,
    _user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    from app.domains.apikey.idle_alerts import load_idle_settings

    settings = await load_idle_settings(db)
    return success_response(data=settings.to_setting(), request=request)


@router.put("/idle/config")
async def put_idle_config(
    request: Request,
    body: IdleConfigBody,
    current_user: AuthUser = Depends(require_asset_admin()),
    db: AsyncSession = Depends(get_db),
):
    """The one real switch for paging holders about idle keys (IT digests
    are unconditional). Audited — flipping employee reach on is a decision."""
    from app.domains.apikey.idle_alerts import (
        IdleSettings,
        load_idle_settings,
        save_idle_settings,
    )

    current = await load_idle_settings(db)
    updated = IdleSettings(
        enabled_for_employees=(
            body.enabled_for_employees
            if body.enabled_for_employees is not None else current.enabled_for_employees
        ),
        min_age_days=body.min_age_days if body.min_age_days is not None else current.min_age_days,
        keep_months=body.keep_months if body.keep_months is not None else current.keep_months,
    )
    await save_idle_settings(db, updated)
    await AuditService(db).log(
        trace_id=request.state.trace_id,
        actor_type="user",
        actor_user_id=current_user.id,
        action="apikey.idle.config_updated",
        resource_type="platform_setting",
        resource_id="apikey.idle_alert",
        status="success",
        detail={"before": current.to_setting(), "after": updated.to_setting()},
    )
    await db.commit()
    return success_response(data=updated.to_setting(), request=request)


@router.post("/assignments/{assignment_id}/idle/keep")
async def keep_idle_api_key(
    request: Request,
    assignment_id: UUID,
    body: IdleKeepBody,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Owner or shared-access member keeps an idle key: silences the monthly
    question for ``keep_months`` and records the reason on the row."""
    from app.domains.apikey.idle_alerts import keep_idle_assignment, load_idle_settings
    from app.identity.person_models import IdentityPerson

    service = ApiKeyService(db)
    assignment = await service._get_assignment(assignment_id, lock=True)
    actor = await db.get(IdentityPerson, current_user.id)
    if actor is None:
        raise ForbiddenError("Unknown actor.")
    is_member = await service.is_member(assignment.id, current_user.id)
    settings = await load_idle_settings(db)
    result = await keep_idle_assignment(
        db, assignment=assignment, actor=actor, reason=body.reason,
        settings=settings, trace_id=request.state.trace_id, is_member=is_member,
    )
    await db.commit()
    return success_response(data=result, request=request)