"""Cloud scope-request / grant lifecycle timeline. Reads the existing append-only ``audit_logs`` table — already populated by every state-changing call in :mod:`scope_request_service` and :mod:`lifecycle_service` — and exposes it as a typed, actor-enriched timeline for the per-record detail surfaces (admin drawer, owner scope detail, employee 我的云订阅 history). Why not a new ``cmdb_cloud_*_events`` table --------------------------------------------- 1. **Single source of truth.** ``audit_logs`` is the authoritative trail per ``CLAUDE.md`` ("Audit by design: every key action traceable with trace_id"). Adding a second event store invites double-write inconsistency — the audit row is the one operators, compliance, and forensics already trust. 2. **No retrofit risk.** Every emission point already exists (~15 distinct actions across the two domains). Building a parallel event-writer means touching every call site; missing one leaks a silent gap into the timeline. 3. **Indexability solved at the audit layer.** Migration 180 adds a composite ``(resource_type, resource_id, created_at)`` index so the per-record fetch is a tight index range scan rather than a full table scan. Action whitelists ----------------- We don't surface every audit row blindly — purely internal bookkeeping actions (e.g. ``revoke_attempt_failed`` retries that the user wouldn't recognize) and orchestration noise (``callback_ignored_state_advanced``) stay hidden by default. The whitelists below are the union of "things a reasonable owner / requester / approver would expect to see on the record's timeline". The ``event_kind`` value returned to the frontend is the i18n key suffix (e.g. ``"submitted"`` → ``cloud.timeline.submitted``); the frontend's translation table is the source of truth for human labels. """ from __future__ import annotations import uuid from dataclasses import dataclass from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.models import AuditLog from app.identity.person_models import IdentityPerson # ── Action → event_kind whitelists ──────────────────────────────── # # Keys are the exact ``audit_logs.action`` strings emitted by the # service layer; values are the stable event_kind tokens the frontend # translates via ``cloud.timeline.``. When the audit row's # ``status`` field is non-success the frontend appends a status suffix # (``failed`` / ``skipped``) so a failed ``provision`` renders # distinctly from a successful one without needing two separate # audit-action strings. SCOPE_REQUEST_TIMELINE_ACTIONS: dict[str, str] = { "cloud.scope_request.submit": "submitted", "cloud.scope_request.approve": "approved", "cloud.scope_request.approve_resume": "approve_resumed", "cloud.scope_request.cloud_created": "cloud_created", "cloud.scope_request.provision": "provisioned", "cloud.scope_request.manual_provision": "manual_provisioned", "cloud.scope_request.reject": "rejected", "cloud.scope_request.withdraw": "withdrawn", "cloud.scope_request.cancelled_via_feishu": "cancelled_via_feishu", } GRANT_TIMELINE_ACTIONS: dict[str, str] = { "cloud.grant.request": "requested", "cloud.grant.approve": "approved", "cloud.grant.reject": "rejected", "cloud.grant.owner_add": "owner_added", "cloud.grant.renew": "renewed", "cloud.grant.renew_reinstate": "renew_reinstated", "cloud.grant.revoke": "revoked", "cloud.grant.disable": "disabled", # Renewal-request lifecycle (2026-07-29). These are logged against # the SCOPE (a submission spans several grants), so the per-grant # timeline picks them up via the scope-wide ledger rather than by # resource_id — see ``list_scope_access_ledger``. "cloud.grant.renewal_requested": "renewal_requested", "cloud.grant.renewal_requested_via_token": "renewal_requested", "cloud.grant.renewal_approved": "renewal_approved", "cloud.grant.renewal_rejected": "renewal_rejected", "cloud.grant.renewal_cancelled": "renewal_cancelled", "cloud.grant.renewal_lapsed": "renewal_lapsed", "cloud.grant.renew_admin_override": "renew_admin_override", } # Audit-row ``detail`` keys that are operator-internal and should not # leak to the timeline panel. Mostly orchestration metadata that has # no meaning to a human reader. Keep this list narrow — pruning too # eagerly hides forensic context the timeline is supposed to show. _REDACT_DETAIL_KEYS: frozenset[str] = frozenset({ "feishu_instance_code", # surfaces elsewhere as a clickable link "request_group_id", # internal grouping id, not user-meaningful }) SCOPE_REQUEST_RESOURCE_TYPE = "cmdb_cloud_scope_request" GRANT_RESOURCE_TYPE = "cmdb_cloud_access_grant" @dataclass(frozen=True) class TimelineEvent: """One row on the lifecycle timeline.""" id: str event_kind: str action: str status: str occurred_at: str actor_user_id: str | None actor_label: str | None actor_type: str trace_id: str detail: dict[str, Any] def to_dict(self) -> dict[str, Any]: return { "id": self.id, "event_kind": self.event_kind, "action": self.action, "status": self.status, "occurred_at": self.occurred_at, "actor_user_id": self.actor_user_id, "actor_label": self.actor_label, "actor_type": self.actor_type, "trace_id": self.trace_id, "detail": self.detail, } def _redact_detail(raw: dict[str, Any] | None) -> dict[str, Any]: if not raw: return {} return {k: v for k, v in raw.items() if k not in _REDACT_DETAIL_KEYS} async def _fetch_timeline( db: AsyncSession, *, resource_type: str, resource_id: str, action_map: dict[str, str], ) -> list[TimelineEvent]: """Query audit_logs for one record's whitelisted lifecycle events. Uses the composite index ``ix_audit_logs_resource_lookup`` (migration 180, landed for the `/api/v1/audit/resources/history` endpoint and reused here) on ``(resource_type, resource_id, created_at)`` so the expected hot path stays a bounded range scan even on a multi-million-row audit table. """ rows = list((await db.execute( select(AuditLog).where( AuditLog.resource_type == resource_type, AuditLog.resource_id == resource_id, AuditLog.action.in_(list(action_map.keys())), ) # Stable tiebreaker on ``id`` for events with identical # ``created_at`` (high-volume / quick-fire transitions such as # the multi-row provision-then-grant burst on auto-approve). # Without this the order is whatever the storage engine # returns and can flip between requests — UI presents an # unstable timeline. The composite index includes only # ``created_at`` as the trailing column; adding ``id.asc()`` # to ORDER BY is a no-op at index-scan time and only matters # when collisions occur. .order_by(AuditLog.created_at.asc(), AuditLog.id.asc()) )).scalars().all()) return await _enrich(db, rows, action_map) async def _enrich( db: AsyncSession, rows: list[AuditLog], action_map: dict[str, str], ) -> list[TimelineEvent]: """Attach actor labels and map audit rows to timeline events. Extracted from ``_fetch_timeline`` so the scope-wide ledger can reuse it without duplicating the actor batching or the redaction rules — two copies would drift and one of them would eventually leak a key ``_REDACT_DETAIL_KEYS`` exists to hide. """ if not rows: return [] # Actor enrichment: one IN-clause fetch for all distinct actors so # we don't N+1 IdentityPerson lookups per event. Audit rows whose # ``actor_type`` is ``system`` / ``integration`` have NULL # ``actor_user_id`` and skip the join. # # Defense in depth: ``uuid.UUID(str(a))`` is wrapped in try/except # so a single malformed audit row (legacy migration leftover, # manual DB patch typo, etc.) does not crash the whole timeline # endpoint with a 500. The affected event still renders — just # without an enriched actor label. actor_ids: list[uuid.UUID] = [] for r in rows: if r.actor_user_id is None: continue try: actor_ids.append(uuid.UUID(str(r.actor_user_id))) except (TypeError, ValueError): continue actor_label_by_id: dict[str, str] = {} if actor_ids: persons = (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(actor_ids), ) )).scalars().all() for p in persons: label = p.display_name or p.work_email or str(p.id) if p.work_email and p.display_name: label = f"{p.display_name} · {p.work_email}" actor_label_by_id[str(p.id)] = label out: list[TimelineEvent] = [] for r in rows: actor_id = str(r.actor_user_id) if r.actor_user_id else None out.append(TimelineEvent( id=str(r.id), event_kind=action_map[r.action], action=r.action, status=r.status, occurred_at=r.created_at.isoformat() if r.created_at else "", actor_user_id=actor_id, actor_label=actor_label_by_id.get(actor_id) if actor_id else None, actor_type=r.actor_type, trace_id=str(r.trace_id), detail=_redact_detail(r.detail), )) return out async def fetch_scope_request_timeline( db: AsyncSession, request_id: str, ) -> list[TimelineEvent]: """Lifecycle events for one ``CmdbCloudScopeRequest`` row, oldest first. Includes submit, approve / approve_resume, cloud_created, provision (success/failure), manual_provision, reject, withdraw, cancelled_via_feishu. Returns an empty list when the record has no audit history — the caller decides whether to render "no events yet" or treat as 404 (validate the request id exists separately at the router layer). """ return await _fetch_timeline( db, resource_type=SCOPE_REQUEST_RESOURCE_TYPE, resource_id=request_id, action_map=SCOPE_REQUEST_TIMELINE_ACTIONS, ) async def fetch_grant_timeline( db: AsyncSession, grant_id: str, ) -> list[TimelineEvent]: """Lifecycle events for one ``CmdbCloudAccessGrant`` row, oldest first. Includes request, approve, reject, owner_add, renew / renew_reinstate, revoke, disable. Internal retry / dispatcher actions (``revoke_attempt_failed``, ``revoke_retry``, ``revoke_retry_skip_scope_closed``) are intentionally omitted — they're orchestration noise the owner / grant holder wouldn't recognize. The terminal status of the record (revoked / disabled) plus the ``revoke_reason`` field on the grant itself already convey the outcome. """ return await _fetch_timeline( db, resource_type=GRANT_RESOURCE_TYPE, resource_id=grant_id, action_map=GRANT_TIMELINE_ACTIONS, ) async def list_scope_access_ledger( db: AsyncSession, scope_id: str, *, grant_ids: list[str], limit: int = 300, ) -> list[dict[str, Any]]: """Scope-wide access ledger: what actually happened, in order. Replaces the old 「申请记录」 tab, which was not a ledger at all — it re-rendered the same grant rows the members table already showed, sorted by ``granted_at``. Three consequences, all measured on aws-account 385595570414 (230 grant rows): * **Renewals were invisible.** A renewal mutates ``expires_at`` / ``last_renewed_at`` on an existing row; it adds no row and does not touch ``granted_at``. 92 renewals on that scope left no trace on the tab — a row renewed three months after issue still displayed its original date and nothing else. * **``granted_at`` was mislabelled 「申请时间」.** It is the moment access became EFFECTIVE, and ``_dispatch_grant`` re-stamps it on every successful dispatch — including a revival — so for a revived grant the column showed the revival date, not any application. * **218 of 230 rows never involved an application at all.** They are ``source='discovered'`` rows that grants_sync found cloud-side, and their 「申请时间」 was the first-sync timestamp. The tab's own empty state said "员工通过申请权限提交后会出现在这里", contradicting 95% of what it displayed. Built from ``audit_logs`` rather than a new table: audit is already the append-only record of every one of these transitions, already indexed for this access pattern, and adding a parallel ledger table would create a second source of truth that can disagree. Two queries are unioned because the events live under two resource anchors: per-grant actions (request / approve / renew / revoke …) are logged against the grant, while a renewal SUBMISSION spans several grants and is logged against the scope. """ events: list[TimelineEvent] = [] if grant_ids: rows = list((await db.execute( select(AuditLog).where( AuditLog.resource_type == GRANT_RESOURCE_TYPE, AuditLog.resource_id.in_(grant_ids), AuditLog.action.in_(list(GRANT_TIMELINE_ACTIONS.keys())), ) .order_by(AuditLog.created_at.desc(), AuditLog.id.desc()) .limit(limit) )).scalars().all()) events.extend(await _enrich(db, rows, GRANT_TIMELINE_ACTIONS)) scope_rows = list((await db.execute( select(AuditLog).where( AuditLog.resource_type == "cmdb_cloud_scope", AuditLog.resource_id == scope_id, AuditLog.action.in_(list(GRANT_TIMELINE_ACTIONS.keys())), ) .order_by(AuditLog.created_at.desc(), AuditLog.id.desc()) .limit(limit) )).scalars().all()) events.extend(await _enrich(db, scope_rows, GRANT_TIMELINE_ACTIONS)) # Newest first — this is a "what changed lately" surface, unlike the # per-grant timeline drawer which reads oldest-first as a story. # # ``id`` is a tiebreaker, not decoration: ``audit_logs.created_at`` is # second-granular, and a submit-then-approve burst lands three events in # the SAME second. Sorting on the timestamp alone leaves their relative # order to whatever the two queries happened to return, so the list # visibly reshuffles between refreshes. Same reason ``_fetch_timeline`` # carries ``AuditLog.id`` in its ORDER BY. events.sort(key=lambda e: (e.occurred_at, e.id), reverse=True) return [e.to_dict() for e in events[:limit]]