"""Owner-approved renewal of cloud access grants. Before 2026-07-29 renewal was self-service: the holder pressed 「一键续期」 — in the UI or straight off the expiry reminder card — and the grant's clock reset on the spot. Nobody else was involved. Measured on production that day: * 763 renewals in three months, **every one** executed by the holder alone. A "90-day access review" nobody reviews is not a control. * 839 already-recycled grants across 135 people were self-revivable, because :meth:`_renew_grant_locked` accepted any row in status ``expired`` with no age limit. Recycling therefore only downgraded re-acquisition from "request + approval" to "press a button". This module makes renewal a request that the scope owner decides, and routes revival of a recycled grant through the same gate. The mechanics deliberately mirror the initial-access-request flow in ``lifecycle_service`` — same Redis advisory locks, same self-approval refusal, same "who can actually approve this" recipient routing — so there is one approval model on this domain rather than two that drift. The owner's signature is final at every grantable TTL: the 2026-07-30 「>90 天需云管理员复核」 second signature was removed on 2026-08-26 (see DECISIONS.md) — the subscription owner is the authority on their own subscription, and the 180-day hard cap is what bounds the exposure. What lives here vs. ``lifecycle_service``: * here — request lifecycle (submit / decide / cancel / lapse), eligibility rules, approver authority, notifications about renewals; * there — the act of moving ``expires_at`` and re-dispatching to the provider (:meth:`CloudLifecycleService.renew_grant`). Approval calls into it with ``approved_request_id`` set; this module never writes ``expires_at`` itself, so "never shorten", the reinstate dispatch, and the closure-field cleanup have exactly one implementation. """ from __future__ import annotations import logging import uuid from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.distributed_lock import LockBusy, distributed_lock from app.core.exceptions import AppException, ConflictError, NotFoundError from app.domains.cmdb.lifecycle_service import ( CloudLifecycleService, _is_prod_environment, load_perpetual_person_ids, ) from app.domains.cmdb.models import ( GRANT_STATUS_ACTIVE, GRANT_STATUS_EXPIRED, GRANT_TTL_DEFAULT_DAYS, GRANT_TTL_HARD_CAP_DAYS, RENEWAL_REQUEST_LAPSE_DAYS, RENEWAL_REVIVAL_MAX_AGE_DAYS, RENEWAL_STATUS_APPROVED, RENEWAL_STATUS_CANCELLED, RENEWAL_STATUS_LAPSED, RENEWAL_STATUS_PENDING_OWNER, RENEWAL_STATUS_REJECTED, RENEWAL_STATUSES_OPEN, SCOPE_STATUS_ACTIVE, CmdbCloudAccessGrant, CmdbCloudGrantRenewalRequest, CmdbCloudScope, ) from app.identity.person_models import IdentityPerson from app.notifications import card_design from app.notifications.service import ( CardAction, CardCollapse, CardFact, CardFactBox, LocalizedText, NotificationEnvelope, NotificationService, ) from app.notifications.settlement import settle logger = logging.getLogger(__name__) # Machine-readable reasons a grant cannot be renewed. The frontend keys # its disabled-button tooltips off these, so they are part of the API # contract — do not rename without updating web/src/i18n. INELIGIBLE_SCOPE_INACTIVE = "scope_inactive" INELIGIBLE_PROD = "prod_unmanaged" INELIGIBLE_PERPETUAL = "perpetual_holder" INELIGIBLE_SCOPE_OWNER = "scope_owner_exempt" INELIGIBLE_STATUS = "status_not_renewable" INELIGIBLE_REVIVAL_TOO_OLD = "revival_window_expired" INELIGIBLE_ALREADY_OPEN = "renewal_already_pending" INELIGIBLE_HOLDER_DEPARTED = "holder_departed" @dataclass(frozen=True) class RenewalEligibility: """Whether one grant can be renewed, and why not when it can't. Returned by :func:`check_renewal_eligibility` and surfaced on the grant list payload so the UI can disable 「申请续期」 with a truthful tooltip instead of letting the user click into a 409. The audit of 2026-07-29 found the reverse pattern all over the cloud surface — buttons that were enabled and then failed, or chips that claimed a state the backend disagreed with — so eligibility is computed once, server-side, and shipped to every renderer. """ eligible: bool reason: str | None = None # True when approving would REVIVE recycled access rather than extend # live access. Approvers see this prominently. is_revival: bool = False @property def as_dict(self) -> dict[str, Any]: return { "eligible": self.eligible, "reason": self.reason, "is_revival": self.is_revival, } def normalize_requested_ttl(ttl_days: int | None) -> int: """Validate a requested TTL against the grant TTL policy. Raises 422 above :data:`GRANT_TTL_HARD_CAP_DAYS`; everything at or below it is approvable by the scope owner alone. The request form must not silently downgrade what the user asked for, so nothing is clamped here — an over-cap number is rejected, not rounded down. """ if ttl_days is None: return GRANT_TTL_DEFAULT_DAYS try: value = int(ttl_days) except (TypeError, ValueError) as exc: raise AppException( code="CLOUD_RENEWAL_TTL_INVALID", message="续期天数必须是整数。", status_code=422, ) from exc if value < 1: raise AppException( code="CLOUD_RENEWAL_TTL_INVALID", message="续期天数必须大于 0。", status_code=422, ) if value > GRANT_TTL_HARD_CAP_DAYS: raise AppException( code="CLOUD_RENEWAL_TTL_TOO_LONG", message=( f"单次续期最长 {GRANT_TTL_HARD_CAP_DAYS} 天(当前填写 " f"{value} 天)。需要长期访问请申请常设角色,而不是超长期限的" "临时授权。" ), status_code=422, ) return value async def check_renewal_eligibility( db: AsyncSession, *, grant: CmdbCloudAccessGrant, scope: CmdbCloudScope, holder: IdentityPerson | None = None, perpetual_person_ids: set[str] | None = None, open_request_grant_ids: set[str] | None = None, now: datetime | None = None, ) -> RenewalEligibility: """Decide whether ``grant`` may be renewed right now. ``perpetual_person_ids`` / ``open_request_grant_ids`` are optional prefetched sets so a list endpoint can evaluate N grants without N queries; both are loaded on demand when omitted. """ now_dt = (now or datetime.now(UTC)).replace(tzinfo=None) if scope.deleted_at is not None or scope.status != SCOPE_STATUS_ACTIVE: return RenewalEligibility(False, INELIGIBLE_SCOPE_INACTIVE) # Production cloud access is not managed by Helix lifecycle # automation at all (no reminders, no auto-revoke) and the prod write # lock is default-locked, so a renewal request would be a form that # can never be actioned. Same predicate the sweeps use. if _is_prod_environment(scope.environment): return RenewalEligibility(False, INELIGIBLE_PROD) if grant.person_id is None: # Orphan-cleanup history row — no holder to renew for. return RenewalEligibility(False, INELIGIBLE_STATUS) person_id = str(grant.person_id) if perpetual_person_ids is None: perpetual_person_ids = await load_perpetual_person_ids(db) if person_id in perpetual_person_ids: # Cloud admin — the grant never expires, so there is nothing to # renew. Before this check such a holder could file a renewal for # access that was already perpetual. return RenewalEligibility(False, INELIGIBLE_PERPETUAL) if scope.owner_person_id and str(scope.owner_person_id) == person_id: # The owner's own grant is exempt from both sweeps. return RenewalEligibility(False, INELIGIBLE_SCOPE_OWNER) if holder is not None and ( holder.employment_status in ("left", "leaver_pending") or holder.deleted_at is not None ): # A departing employee's access is the offboarding pipeline's # business. Letting them extend it — even with owner approval — # would race the leaver sweep. return RenewalEligibility(False, INELIGIBLE_HOLDER_DEPARTED) if grant.status not in (GRANT_STATUS_ACTIVE, GRANT_STATUS_EXPIRED): # ``revoked`` / ``disabled`` are deliberate closures by a human or # by the inactivity sweep — those need a fresh access request that # re-runs role governance, not a renewal. return RenewalEligibility(False, INELIGIBLE_STATUS) is_revival = grant.status == GRANT_STATUS_EXPIRED if is_revival: # Bound the revival window — see RENEWAL_REVIVAL_MAX_AGE_DAYS. closed_at = grant.revoked_at or grant.expires_at if closed_at is not None: if closed_at.tzinfo is not None: closed_at = closed_at.astimezone(UTC).replace(tzinfo=None) age_days = (now_dt - closed_at).days if age_days > RENEWAL_REVIVAL_MAX_AGE_DAYS: return RenewalEligibility( False, INELIGIBLE_REVIVAL_TOO_OLD, is_revival=True, ) if open_request_grant_ids is None: open_request_grant_ids = { str(gid) for gid in (await db.execute( select(CmdbCloudGrantRenewalRequest.grant_id) .where( CmdbCloudGrantRenewalRequest.grant_id == str(grant.id), CmdbCloudGrantRenewalRequest.status.in_( RENEWAL_STATUSES_OPEN, ), ) )).scalars().all() } if str(grant.id) in open_request_grant_ids: return RenewalEligibility( False, INELIGIBLE_ALREADY_OPEN, is_revival=is_revival, ) return RenewalEligibility(True, None, is_revival=is_revival) class CloudRenewalService: """Request/decide lifecycle for cloud grant renewals.""" def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self.notify = NotificationService(db) self.lifecycle = CloudLifecycleService(db) # ── Submit ───────────────────────────────────────────── async def submit_renewal_request( self, *, scope_id: str, person_id: str, grant_ids: list[str] | None, ttl_days: int | None, reason: str, requested_by_person_id: uuid.UUID, trace_id: str, actor_role_names: list[str] | None = None, ) -> dict[str, Any]: """Raise one renewal request per eligible grant, as one group. The unit the employee acts on is a SUBSCRIPTION, not a role: with ``grant_ids=None`` every renewable grant they hold on the scope is covered by a single submission sharing one ``request_group_id``. That is deliberate — the pre-change flow was per-grant, so a holder with 8 roles on one account received 8 reminder cards and had to click 8 times, and the owner would have seen the same person 8 times in their queue. One person, one decision. """ reason_clean = (reason or "").strip() if not reason_clean: raise AppException( code="CLOUD_RENEWAL_REASON_REQUIRED", message="请填写续期原因,审批人需要据此判断是否仍有必要。", status_code=422, ) if len(reason_clean) > 1000: reason_clean = reason_clean[:1000] ttl = normalize_requested_ttl(ttl_days) # Acting for someone else is cloud-admin-only, mirroring # ``submit_access_request``. Without this any employee could file # renewals in a colleague's name. if str(person_id) != str(requested_by_person_id): from app.core.rbac import can_manage_cloud if not can_manage_cloud(actor_role_names or []): raise AppException( code="FORBIDDEN", message=( "不能代他人提交续期申请(仅云管理员可代提)。" ), status_code=403, ) scope = await self.db.get(CmdbCloudScope, uuid.UUID(scope_id)) if scope is None or scope.deleted_at is not None: raise NotFoundError("CmdbCloudScope", scope_id) holder = await self.db.get(IdentityPerson, uuid.UUID(str(person_id))) if holder is None or holder.deleted_at is not None: raise NotFoundError("IdentityPerson", str(person_id)) candidates = await self._load_renewable_grants( scope_id=scope_id, person_id=person_id, grant_ids=grant_ids, ) if not candidates: raise NotFoundError( "CmdbCloudAccessGrant", f"scope={scope_id} person={person_id}", ) perpetual = await load_perpetual_person_ids(self.db) open_ids = await self._open_request_grant_ids( [str(g.id) for g in candidates], ) eligible: list[tuple[CmdbCloudAccessGrant, RenewalEligibility]] = [] rejected: list[dict[str, str]] = [] for grant in candidates: verdict = await check_renewal_eligibility( self.db, grant=grant, scope=scope, holder=holder, perpetual_person_ids=perpetual, open_request_grant_ids=open_ids, ) if verdict.eligible: eligible.append((grant, verdict)) else: rejected.append({ "grant_id": str(grant.id), "role_key": grant.role_key, "reason": verdict.reason or "", }) if not eligible: # Every candidate was refused. Surface the FIRST reason rather # than a generic failure so the caller can render something # actionable ("你是该订阅 Owner,权限不会过期" beats "no eligible # grants"). first = rejected[0]["reason"] if rejected else INELIGIBLE_STATUS raise ConflictError( f"No renewable grant on this scope ({first})", ) group_id = str(uuid.uuid4()) now_dt = datetime.now(UTC) lapse_at = now_dt + timedelta(days=RENEWAL_REQUEST_LAPSE_DAYS) created: list[CmdbCloudGrantRenewalRequest] = [] for grant, verdict in eligible: row = CmdbCloudGrantRenewalRequest( id=str(uuid.uuid4()), grant_id=str(grant.id), open_for_grant_id=str(grant.id), scope_id=str(scope.id), person_id=str(person_id), requested_by_person_id=str(requested_by_person_id), requested_ttl_days=ttl, reason=reason_clean, is_revival=verdict.is_revival, status=RENEWAL_STATUS_PENDING_OWNER, request_group_id=group_id, lapse_at=lapse_at, ) self.db.add(row) created.append(row) try: await self.db.flush() except IntegrityError as exc: # The UNIQUE on ``open_for_grant_id`` fired: a concurrent # submission (double-clicked card, or the same person on two # devices) already opened a request for one of these grants. # Surfaced as a 409 rather than a 500 — the DB constraint is # the intended race guard, not an unexpected failure. await self.db.rollback() raise ConflictError( "该权限已有一条待审批的续期申请,请等待审批结果。", ) from exc await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=requested_by_person_id, action="cloud.grant.renewal_requested", status="success", resource_type="cmdb_cloud_scope", resource_id=str(scope.id), detail={ "request_group_id": group_id, "person_id": str(person_id), "requested_ttl_days": ttl, "reason": reason_clean, "grant_ids": [str(g.id) for g, _ in eligible], "revival_grant_ids": [ str(g.id) for g, v in eligible if v.is_revival ], "skipped": rejected, "on_behalf_of": ( str(person_id) if str(person_id) != str(requested_by_person_id) else None ), }, ) await self.db.commit() await self._notify_approvers( scope=scope, holder=holder, requests=created, ttl_days=ttl, reason=reason_clean, group_id=group_id, trace_id=trace_id, ) return { "request_group_id": group_id, "created": [str(r.id) for r in created], "grant_count": len(created), "requested_ttl_days": ttl, "lapse_at": lapse_at, "skipped": rejected, } async def _load_renewable_grants( self, *, scope_id: str, person_id: str, grant_ids: list[str] | None, ) -> list[CmdbCloudAccessGrant]: stmt = select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id == scope_id, CmdbCloudAccessGrant.person_id == person_id, CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_EXPIRED), ), ) if grant_ids: # Explicit selection still filters on (scope, person) above, # so a caller cannot smuggle in a grant belonging to someone # else or to another scope by id. stmt = stmt.where(CmdbCloudAccessGrant.id.in_(grant_ids)) rows = await self.db.execute(stmt) return list(rows.scalars().all()) async def _open_request_grant_ids( self, grant_ids: list[str], ) -> set[str]: if not grant_ids: return set() rows = await self.db.execute( select(CmdbCloudGrantRenewalRequest.grant_id).where( CmdbCloudGrantRenewalRequest.grant_id.in_(grant_ids), CmdbCloudGrantRenewalRequest.status.in_( RENEWAL_STATUSES_OPEN, ), ) ) return {str(g) for g in rows.scalars().all()} # ── Notify ───────────────────────────────────────────── async def _notify_approvers( self, *, scope: CmdbCloudScope, holder: IdentityPerson, requests: list[CmdbCloudGrantRenewalRequest], ttl_days: int, reason: str, group_id: str, trace_id: str, ) -> None: """One card per approver per submission — never per grant. The expiry reminder used to fan out per grant, which produced 861 cards to 198 people in a single cron run (2026-07-15) and 63 cards to one person in one day. Approval traffic must not repeat that mistake, so a submission covering N roles is ONE card naming all N. Deliberately a deep link, not a no-login one-click approve button. The platform does use signed action tokens for self-service acts the holder performs on their OWN access, but approving somebody else's access is an authority decision and must happen behind authentication. Failures are logged, never raised: the request is already committed, and losing the notification must not roll it back. """ # The scope owner (or the cloud-admin team on an intentionally- # ownerless scope) is the only approver, at every TTL. try: recipients = ( await self.lifecycle._resolve_request_approval_recipients( scope, requires_admin=False, ) ) except Exception as exc: # noqa: BLE001 logger.warning( "renewal: could not resolve approvers for scope %s: %s", scope.id, exc, ) return if not recipients: logger.warning( "renewal: no actionable approver for scope %s — request " "group %s is queued but nobody was paged", scope.id, group_id, ) return scope_label = scope.display_name or scope.external_id holder_label = holder.display_name or holder.work_email or "—" revival_count = sum(1 for r in requests if r.is_revival) role_lines = "\n".join( f" • {r.role_display_name or r.role_key}" + (" (已被回收,批准即恢复访问)" if r.is_revival else "") for r in ( await self._roles_for_requests(requests) ) ) # 卡片按「一句人话标题 + 事实盒 + 事实行 + 折叠长尾」重写(2026-09-07)。 # # 旧标题是 `[待审批·云权限续期] …`:方括号前缀在卡片上是浪费的一行 —— # 那些信息现在由状态徽章(待处理)和副标题(云权限续期 · 订阅名)承担, # 而标题回归成「谁要什么」。**邮件主题仍带前缀**(邮件客户端靠它扫一眼 # 分类),那由模板层的 subject 决定,与这里无关。 # # 旧正文把「多少项、多久、什么原因、哪些角色」全塞进一段散文,读者要 # 自己解析。现在:时长进事实盒的右侧(它是决定答不答应的那个数字), # 申请人/原因/原到期进事实行,角色清单折叠(超过 3 行才折)。 title = ( f"{holder_label} 申请续期 {scope_label} 的 {len(requests)} 项权限" ) # 刻意**不**放「原到期日」那一行:它在 grant 上而不在申请行上,为一行 # 装饰性事实多查一次库不值得 —— 而正文那句「驳回后按原到期日正常回收」 # 已经把它要传达的东西说清了。 body = "若不再需要,请驳回 —— 驳回后权限会按原到期日正常回收。" if revival_count: body = ( f"其中 {revival_count} 项权限已被回收,批准将重新下发到云端。\n\n" + body ) # ``group`` names the exact submission: the landing page opens the # 续期审批 queue AND scrolls/flashes that row. Without it the # approver arrived on a page with N tabs and had to find the queue # themselves — the 2026-07-31 owner report ("跳到云订阅首页,找不到"). url = f"/cloud/scopes/{scope.id}?renewals=1&group={group_id}" for person in recipients: try: await self.notify.dispatch( NotificationEnvelope( recipient=person, title=title, body_text=body, url=url, channels=["feishu", "email"], card_status=card_design.STATUS_PENDING, card_subtitle=LocalizedText( zh=f"云权限续期 · {scope_label}", en=f"Cloud access renewal · {scope_label}", ja=f"クラウド権限の更新 · {scope_label}", ), # 事实盒:左边是这次要续的东西,右边是**决定答不答应的那个 # 数字** —— 时长。原来它埋在「…上的 N 项权限,时长 90 天」 # 这句散文的中段。 card_fact_box=CardFactBox( subject=LocalizedText( zh=f"{len(requests)} 项权限", en=f"{len(requests)} permission(s)", ja=f"{len(requests)} 件の権限", ), detail=scope_label, metric=LocalizedText( zh=f"{ttl_days} 天", en=f"{ttl_days} days", ja=f"{ttl_days} 日", ), ), card_facts=[ CardFact( label=LocalizedText( zh="申请人", en="Requester", ja="申請者", ), value=holder_label, icon="person", ), CardFact( label=LocalizedText( zh="原因", en="Reason", ja="理由", ), value=reason, icon="reason", ), ], # 角色清单是长尾:1 项时直接铺开,12 项时折起来并在标题 # 上带计数(渲染器按行数自己决定,见 CardCollapse)。 card_collapse=CardCollapse( title=LocalizedText( zh=f"涉及角色 · {len(requests)} 项", en=f"Roles · {len(requests)}", ja=f"対象ロール · {len(requests)} 件", ), body=role_lines, ), card_actions=[ CardAction( label=LocalizedText( zh="去审批", en="Go to approval", ja="承認画面へ", ), url=url, style="primary", ), ], related_resource_type=( "cmdb_cloud_grant_renewal_request" ), related_resource_id=group_id, ), trace_id=trace_id, ) except Exception as exc: # noqa: BLE001 logger.warning( "renewal: approver notification failed for %s: %s", person.id, exc, ) async def _roles_for_requests( self, requests: list[CmdbCloudGrantRenewalRequest], ) -> list[Any]: """Attach role labels to request rows for notification rendering. Returns lightweight shims carrying ``role_display_name`` / ``role_key`` / ``is_revival`` so the caller can format one line per role without a per-row DB round-trip. """ if not requests: return [] rows = await self.db.execute( select( CmdbCloudAccessGrant.id, CmdbCloudAccessGrant.role_key, CmdbCloudAccessGrant.role_display_name, ).where( CmdbCloudAccessGrant.id.in_( [r.grant_id for r in requests], ) ) ) by_id = { str(gid): (rk, rd) for gid, rk, rd in rows.all() } @dataclass(frozen=True) class _RoleLine: role_key: str role_display_name: str | None is_revival: bool out: list[Any] = [] for r in requests: rk, rd = by_id.get(str(r.grant_id), (r.grant_id, None)) out.append(_RoleLine(rk, rd, r.is_revival)) return out # ── Decide ───────────────────────────────────────────── async def decide_renewal_group( self, *, request_group_id: str, approve: bool, approver_id: uuid.UUID, approver_role_names: list[str] | None, trace_id: str, note: str | None = None, ) -> dict[str, Any]: """Approve or reject every open request in one submission. Group-scoped, matching ``approve_grant``: the owner decides on a PERSON's renewal, not on each role separately, so a partially decided group can never exist. The lock is keyed by group and shares the ``approve_group:`` namespace with the initial-access approval path, so a renewal decision and an access approval on the same submission id cannot interleave. """ lock_key = f"approve_group:renewal:{request_group_id}" try: async with distributed_lock(lock_key, ttl=900): return await self._decide_renewal_group_locked( request_group_id=request_group_id, approve=approve, approver_id=approver_id, approver_role_names=approver_role_names, trace_id=trace_id, note=note, ) except LockBusy as exc: raise ConflictError( f"Renewal group {request_group_id} is already being decided", ) from exc async def _decide_renewal_group_locked( self, *, request_group_id: str, approve: bool, approver_id: uuid.UUID, approver_role_names: list[str] | None, trace_id: str, note: str | None, ) -> dict[str, Any]: from app.core.rbac import can_manage_cloud rows = await self.db.execute( select(CmdbCloudGrantRenewalRequest) .where( CmdbCloudGrantRenewalRequest.request_group_id == request_group_id, CmdbCloudGrantRenewalRequest.status.in_( RENEWAL_STATUSES_OPEN, ), ) .order_by(CmdbCloudGrantRenewalRequest.created_at) ) requests = list(rows.scalars().all()) if not requests: raise NotFoundError( "CmdbCloudGrantRenewalRequest", request_group_id, ) scope = await self.db.get( CmdbCloudScope, uuid.UUID(str(requests[0].scope_id)), ) if scope is None or scope.deleted_at is not None: raise NotFoundError( "CmdbCloudScope", str(requests[0].scope_id), ) if scope.status != SCOPE_STATUS_ACTIVE: # Same guard as ``_approve_grant_locked``: the scope may have # been decommissioned between submission and decision, and # writing ACTIVE access onto a closed scope is worse than # leaving the request open for a human to look at. raise AppException( code="CLOUD_SCOPE_NOT_ACTIVE", message=( f"订阅 {scope.external_id} 当前状态为 {scope.status!r}," "无法审批续期。" ), status_code=409, ) # AUTHORITY FIRST. Everything below this point can write to the DB, # emit an audit row attributed to the caller, or reveal something # about the requester — so an unauthorized caller must be turned # away before any of it runs. (Written the other way round first: # the holder-departed branch below cancels the request and audits # it against ``approver_id``, which let anyone who could reach the # endpoint cause a state change and read "this person has left" out # of the 409 message.) approver_is_admin = can_manage_cloud(approver_role_names or []) is_scope_owner = ( scope.owner_person_id is not None and str(scope.owner_person_id) == str(approver_id) ) if not (is_scope_owner or approver_is_admin): raise AppException( code="FORBIDDEN", message="只有该订阅的 Owner 或云管理员可以审批续期申请。", status_code=403, ) # Re-validate the holder at DECISION time, not just at submission # time. Up to RENEWAL_REQUEST_LAPSE_DAYS elapse between the two, # which is more than enough for someone to resign: approving then # would extend a departing employee's cloud access and race the # leaver sweep. The offboarding pipeline owns their access from the # moment they are marked leaving, so the request is closed rather # than left open for a second attempt. holder = await self.db.get( IdentityPerson, uuid.UUID(str(requests[0].person_id)), ) if holder is None or holder.deleted_at is not None or ( holder.employment_status in ("left", "leaver_pending") ): now_dt = datetime.now(UTC) for req in requests: self._close( req, RENEWAL_STATUS_CANCELLED, note="holder departed before decision", approver_id=None, now_dt=now_dt, ) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=approver_id, action="cloud.grant.renewal_cancelled", status="success", resource_type="cmdb_cloud_scope", resource_id=str(scope.id), detail={ "request_group_id": request_group_id, "person_id": str(requests[0].person_id), "cascade_reason": "holder_departed_at_decision_time", "request_ids": [str(r.id) for r in requests], }, ) await self.db.commit() raise ConflictError( "申请人已离职或进入离职流程,续期申请已自动关闭 —— " "其云访问由离职流程统一回收。", ) # Four-eyes: the holder must never sign off on their own renewal, # even when they happen to own the scope or hold cloud-admin. This # is the whole point of the change — self-service renewal already # existed and was removed precisely because one pair of eyes was # not a control. if str(requests[0].person_id) == str(approver_id): raise ConflictError( "不能审批自己的续期申请,请让另一位 Owner 或云管理员审批。", ) ttl = int(requests[0].requested_ttl_days) now_dt = datetime.now(UTC) if not approve: for req in requests: self._close( req, RENEWAL_STATUS_REJECTED, note=note, approver_id=approver_id, now_dt=now_dt, ) await self._audit_decision( requests=requests, scope=scope, approve=False, approver_id=approver_id, trace_id=trace_id, note=note, applied=False, ) await self.db.commit() # 审批人手上那张「去审批」卡片现在是一张过期的邀请 —— 原地改成回执。 # 放在 commit 之后:结案是对已发生事实的记录,不该参与这次决策的事务。 await settle( self.db, resource_type="cmdb_cloud_grant_renewal_request", resource_id=request_group_id, outcome=card_design.STATUS_REJECTED, actor_person_id=approver_id, ) await self._notify_requester( requests=requests, scope=scope, approved=False, note=note, trace_id=trace_id, ) return { "request_group_id": request_group_id, "status": RENEWAL_STATUS_REJECTED, "applied": False, "decided": len(requests), } # Final approval — apply it. This is the only approval stage: # until 2026-08-26 a request over ``GRANT_TTL_DEFAULT_DAYS`` was # parked in ``pending_admin`` for a cloud admin's countersignature, # and that tier was removed, so a row still sitting in that state # (its owner already signed) is finalized here rather than being # left undecidable. applied: list[dict[str, Any]] = [] failures: list[dict[str, str]] = [] for req in requests: try: grant = await self.lifecycle.renew_grant( str(req.grant_id), actor_user_id=approver_id, trace_id=trace_id, actor_role_names=approver_role_names, ttl_days_override=ttl, approved_request_id=str(req.id), ) except Exception as exc: # noqa: BLE001 # One role failing (provider rejected the re-dispatch, # principal vanished) must not abort the rest — mirrors # the partial-failure semantics of group access approval. # The request stays OPEN so it can be retried once the # cause is fixed, rather than being silently marked # approved with no access granted. logger.exception( "renewal: applying approved request %s failed", req.id, ) failures.append({ "request_id": str(req.id), "grant_id": str(req.grant_id), "error": str(exc)[:400], }) continue self._close( req, RENEWAL_STATUS_APPROVED, note=note, approver_id=approver_id, now_dt=now_dt, ) req.applied_expires_at = grant.expires_at applied.append({ "request_id": str(req.id), "grant_id": str(req.grant_id), "expires_at": grant.expires_at, }) await self._audit_decision( requests=requests, scope=scope, approve=True, approver_id=approver_id, trace_id=trace_id, note=note, applied=True, failures=failures, ) await self.db.commit() if applied and not failures: # 只有整组都应用成功才结案:部分失败的行仍然 OPEN、还在等重试, # 把卡片改成「已批准」会告诉审批人一件没做完的事已经做完了。 await settle( self.db, resource_type="cmdb_cloud_grant_renewal_request", resource_id=request_group_id, outcome=card_design.STATUS_APPROVED, actor_person_id=approver_id, ) if applied: await self._notify_requester( requests=requests, scope=scope, approved=True, note=note, trace_id=trace_id, ) return { "request_group_id": request_group_id, # A partial failure leaves the failed rows OPEN so they can be # retried; report the status they still carry rather than # claiming ``approved`` for a group that was not fully applied. "status": ( RENEWAL_STATUS_APPROVED if not failures else next( ( r.status for r in requests if r.status in RENEWAL_STATUSES_OPEN ), RENEWAL_STATUS_PENDING_OWNER, ) ), "applied": bool(applied), "decided": len(applied), "failures": failures, } def _close( self, req: CmdbCloudGrantRenewalRequest, status: str, *, note: str | None, approver_id: uuid.UUID | None, now_dt: datetime, ) -> None: """Move a request to a terminal status. Clearing ``open_for_grant_id`` is what releases the UNIQUE slot so the holder can submit again later — miss it and the grant is permanently un-renewable with a 409 that looks like a bug. """ req.status = status req.open_for_grant_id = None if note: req.decision_note = note if approver_id is not None: req.owner_decided_by_person_id = str(approver_id) req.owner_decided_at = now_dt async def _audit_decision( self, *, requests: list[CmdbCloudGrantRenewalRequest], scope: CmdbCloudScope, approve: bool, approver_id: uuid.UUID, trace_id: str, note: str | None, applied: bool, failures: list[dict[str, str]] | None = None, ) -> None: await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=approver_id, action=( "cloud.grant.renewal_approved" if approve else "cloud.grant.renewal_rejected" ), status="success", resource_type="cmdb_cloud_scope", resource_id=str(scope.id), detail={ "request_group_id": requests[0].request_group_id, "person_id": str(requests[0].person_id), "requested_ttl_days": int(requests[0].requested_ttl_days), # Kept at a constant ``owner`` so historical audit rows — # which carry ``owner`` / ``admin`` from the retired # two-signature era — stay queryable on one shape. "stage": "owner", "applied_to_grants": applied, "note": note, "request_ids": [str(r.id) for r in requests], "failures": failures or [], }, ) async def _notify_requester( self, *, requests: list[CmdbCloudGrantRenewalRequest], scope: CmdbCloudScope, approved: bool, note: str | None, trace_id: str, ) -> None: """Tell the holder the outcome — one card for the whole group. A rejection MUST reach the person: their access is now going to be recycled on the original schedule, and before this change nothing ever told a holder that their access had lapsed or been refused. """ holder = await self.db.get( IdentityPerson, uuid.UUID(str(requests[0].person_id)), ) if holder is None or not holder.work_email: return scope_label = scope.display_name or scope.external_id if approved: applied = [ r.applied_expires_at for r in requests if r.applied_expires_at is not None ] when = ( f"新到期日 {min(applied):%Y-%m-%d}" if applied else "已延长" ) title = f"[续期已通过] {scope_label}" body = ( f"你申请的 {scope_label} 云权限续期已通过。\n\n" f"涉及 {len(requests)} 项权限,{when}。" + (f"\n\n审批备注:{note}" if note else "") ) else: title = f"[续期未通过] {scope_label}" body = ( f"你申请的 {scope_label} 云权限续期未获批准。\n\n" f"涉及 {len(requests)} 项权限,将按原到期日正常回收。" + (f"\n\n审批备注:{note}" if note else "") + "\n\n如仍确有需要,请与订阅 Owner 沟通后重新提交。" ) try: await self.notify.dispatch( NotificationEnvelope( recipient=holder, title=title, body_text=body, url=f"/my/cloud/scopes/{scope.id}", channels=["feishu", "email"], related_resource_type=( "cmdb_cloud_grant_renewal_request" ), related_resource_id=str( requests[0].request_group_id or requests[0].id, ), ), trace_id=trace_id, ) except Exception as exc: # noqa: BLE001 logger.warning( "renewal: requester notification failed for %s: %s", holder.id, exc, ) # ── Cancel / lapse / cascade ─────────────────────────── async def cancel_renewal_group( self, *, request_group_id: str, actor_id: uuid.UUID, actor_role_names: list[str] | None, trace_id: str, ) -> dict[str, Any]: """Withdraw an open submission. Allowed for the holder (they decided they don't need it after all) and for cloud admins. Not for the scope owner — an owner who does not want to grant it should REJECT, which tells the requester why; silently cancelling on someone else's behalf looks to the requester like their request vanished. """ from app.core.rbac import can_manage_cloud rows = await self.db.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.request_group_id == request_group_id, CmdbCloudGrantRenewalRequest.status.in_( RENEWAL_STATUSES_OPEN, ), ) ) requests = list(rows.scalars().all()) if not requests: raise NotFoundError( "CmdbCloudGrantRenewalRequest", request_group_id, ) is_holder = str(requests[0].person_id) == str(actor_id) if not (is_holder or can_manage_cloud(actor_role_names or [])): raise AppException( code="FORBIDDEN", message="只有申请人本人或云管理员可以撤回续期申请。", status_code=403, ) now_dt = datetime.now(UTC) for req in requests: self._close( req, RENEWAL_STATUS_CANCELLED, note=None, approver_id=None, now_dt=now_dt, ) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="cloud.grant.renewal_cancelled", status="success", resource_type="cmdb_cloud_scope", resource_id=str(requests[0].scope_id), detail={ "request_group_id": request_group_id, "request_ids": [str(r.id) for r in requests], "by_holder": is_holder, }, ) await self.db.commit() return { "request_group_id": request_group_id, "status": RENEWAL_STATUS_CANCELLED, "cancelled": len(requests), } async def close_open_requests_for_grant( self, *, grant_id: str, reason: str, trace_id: str, ) -> int: """Cascade-close open requests when their grant is hard-closed. Called from the revoke / disable / scope-decommission paths. An open renewal against access that has since been deliberately revoked must not stay approvable — otherwise an owner clicking 「批准」 next week would silently RESURRECT access that a human or the offboarding pipeline had closed on purpose. Deliberately does NOT fire on expiry: a grant reaching its ``expires_at`` while a renewal is pending is the normal case, and approving it afterwards is exactly the intended revival path. """ rows = await self.db.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.grant_id == str(grant_id), CmdbCloudGrantRenewalRequest.status.in_( RENEWAL_STATUSES_OPEN, ), ) ) requests = list(rows.scalars().all()) if not requests: return 0 now_dt = datetime.now(UTC) for req in requests: self._close( req, RENEWAL_STATUS_CANCELLED, note=f"grant closed: {reason}", approver_id=None, now_dt=now_dt, ) await self.audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="cloud.grant.renewal_cancelled", status="success", resource_type="cmdb_cloud_access_grant", resource_id=str(grant_id), detail={ "cascade_reason": reason, "request_ids": [str(r.id) for r in requests], }, ) return len(requests) # ── Queries ──────────────────────────────────────────── async def list_open_for_scope( self, scope_id: str, *, privileged_viewer: bool = False, ) -> list[dict[str, Any]]: """Open renewal submissions on one scope, grouped by submission. Powers the owner's approval queue. One entry per ``request_group_id`` — the owner decides on a person, not on each of their roles. ``privileged_viewer`` follows the same privacy rule ``_serialize_grants`` applies to grant rows: the requester's **employee number** is only emitted to admins / auditors / cloud-admins. A scope owner is often a plain employee, and the read gate on this endpoint (``_require_scope_reader``) also admits finance-class viewers — neither needs 工号 to decide, and the rest of this domain deliberately withholds it from them. Name + work email (in ``person_label``) is what identifies the requester. """ rows = await self.db.execute( select( CmdbCloudGrantRenewalRequest, CmdbCloudAccessGrant.role_key, CmdbCloudAccessGrant.role_display_name, CmdbCloudAccessGrant.expires_at, IdentityPerson.display_name, IdentityPerson.work_email, IdentityPerson.employee_no, IdentityPerson.employment_status, ) .join( CmdbCloudAccessGrant, CmdbCloudAccessGrant.id == CmdbCloudGrantRenewalRequest.grant_id, ) .join( IdentityPerson, IdentityPerson.id == CmdbCloudGrantRenewalRequest.person_id, ) .where( CmdbCloudGrantRenewalRequest.scope_id == scope_id, CmdbCloudGrantRenewalRequest.status.in_( RENEWAL_STATUSES_OPEN, ), ) .order_by(CmdbCloudGrantRenewalRequest.created_at) ) groups: dict[str, dict[str, Any]] = {} for ( req, role_key, role_display, grant_expires, display_name, work_email, employee_no, employment_status, ) in rows.all(): key = str(req.request_group_id or req.id) group = groups.get(key) if group is None: group = { "request_group_id": key, "status": req.status, "person_id": str(req.person_id), "person_label": display_name or work_email, "person_employee_no": ( employee_no if privileged_viewer else None ), "person_employment_status": employment_status, "requested_ttl_days": int(req.requested_ttl_days), "reason": req.reason, "created_at": req.created_at, "lapse_at": req.lapse_at, "owner_decided_at": req.owner_decided_at, "owner_decided_by_person_id": ( str(req.owner_decided_by_person_id) if req.owner_decided_by_person_id else None ), "requested_by_person_id": str( req.requested_by_person_id, ), "on_behalf": ( str(req.requested_by_person_id) != str(req.person_id) ), "revival_count": 0, "items": [], } groups[key] = group if req.is_revival: group["revival_count"] += 1 group["items"].append({ "request_id": str(req.id), "grant_id": str(req.grant_id), "role_key": role_key, "role_display_name": role_display, "current_expires_at": grant_expires, "is_revival": req.is_revival, }) return list(groups.values()) async def list_pending_for_approver( self, approver_id: str, *, is_cloud_admin: bool, privileged_viewer: bool = False, ) -> list[dict[str, Any]]: """Open submissions awaiting THIS approver's decision. Deliberately mirrors ``_provider_cloud_renewal_approval`` in ``app/mobile/action_inbox.py`` exactly — same authority rule, same self-exclusion — so the mobile to-do badge and this endpoint can never disagree about who owes a decision. A badge that shows work the page then says does not exist is worse than no badge. * cloud admin → every open submission org-wide; * plain scope owner → only submissions on scopes they own, excluding their own (self-approval is refused, so surfacing it would be an un-completable to-do). Both statuses in ``RENEWAL_STATUSES_OPEN`` are shown to either kind of approver. Between 2026-07-31 and 2026-08-26 a plain owner was deliberately NOT shown ``pending_admin`` rows, because back then ``decide_renewal_group`` answered them with a 403 and the row would have been an un-completable to-do. With the second signature removed, an owner can finalize such a row, so hiding it would strand it instead. ``scope_label`` / ``scope_external_id`` ride along so a cross-scope approver inbox (desktop 我的待办, mobile action inbox) can name the subscription without a second round-trip per row. """ wanted_statuses = list(RENEWAL_STATUSES_OPEN) # Start from the scopes that actually HAVE open submissions rather # than from every scope the caller can approve on: a cloud admin # owes a decision on the whole org, and looping # ``list_open_for_scope`` over hundreds of scopes to find the two # with work is what made this endpoint too expensive to prime a # to-do badge with. scopes_stmt = select( CmdbCloudScope.id, CmdbCloudScope.display_name, CmdbCloudScope.external_id, ).where( CmdbCloudScope.id.in_( select(CmdbCloudGrantRenewalRequest.scope_id).where( CmdbCloudGrantRenewalRequest.status.in_(wanted_statuses), ).distinct() ), CmdbCloudScope.deleted_at.is_(None), ) if not is_cloud_admin: scopes_stmt = scopes_stmt.where( CmdbCloudScope.owner_person_id == str(approver_id), ) scope_rows = (await self.db.execute(scopes_stmt)).all() if not scope_rows: return [] out: list[dict[str, Any]] = [] for sid, scope_display, scope_external in scope_rows: groups = await self.list_open_for_scope( str(sid), privileged_viewer=privileged_viewer, ) for g in groups: if g["status"] not in wanted_statuses: continue if not is_cloud_admin and g["person_id"] == str(approver_id): continue g["scope_id"] = str(sid) g["scope_label"] = scope_display or scope_external g["scope_external_id"] = scope_external out.append(g) out.sort(key=lambda g: str(g["created_at"]), reverse=True) return out async def list_for_person( self, person_id: str, *, limit: int = 100, ) -> list[dict[str, Any]]: """The requester's own renewal history, newest first.""" rows = await self.db.execute( select( CmdbCloudGrantRenewalRequest, CmdbCloudScope.external_id, CmdbCloudScope.display_name, CmdbCloudAccessGrant.role_key, CmdbCloudAccessGrant.role_display_name, ) .join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudGrantRenewalRequest.scope_id, ) .join( CmdbCloudAccessGrant, CmdbCloudAccessGrant.id == CmdbCloudGrantRenewalRequest.grant_id, ) .where(CmdbCloudGrantRenewalRequest.person_id == person_id) .order_by(CmdbCloudGrantRenewalRequest.created_at.desc()) .limit(max(1, min(int(limit), 500))) ) out: list[dict[str, Any]] = [] for req, ext_id, scope_name, role_key, role_display in rows.all(): out.append({ "request_id": str(req.id), "request_group_id": ( str(req.request_group_id) if req.request_group_id else None ), "scope_id": str(req.scope_id), "scope_external_id": ext_id, "scope_display_name": scope_name, "role_key": role_key, "role_display_name": role_display, "requested_ttl_days": int(req.requested_ttl_days), "reason": req.reason, "status": req.status, "is_revival": req.is_revival, "created_at": req.created_at, "lapse_at": req.lapse_at, "owner_decided_at": req.owner_decided_at, "admin_decided_at": req.admin_decided_at, "decision_note": req.decision_note, "applied_expires_at": req.applied_expires_at, }) return out async def count_open_by_scope(self) -> dict[str, int]: """``{scope_id: open_request_count}`` for badge counts.""" rows = await self.db.execute( select( CmdbCloudGrantRenewalRequest.scope_id, func.count( func.distinct( CmdbCloudGrantRenewalRequest.request_group_id, ), ), ) .where( CmdbCloudGrantRenewalRequest.status.in_( RENEWAL_STATUSES_OPEN, ), ) .group_by(CmdbCloudGrantRenewalRequest.scope_id) ) return {str(sid): int(n) for sid, n in rows.all()} # How long before ``lapse_at`` the approver gets a "this is about to # lapse" nudge. 3 days of the 14-day window: late enough that it is not # noise on a request filed yesterday, early enough to still act. RENEWAL_OWNER_NUDGE_LEAD_DAYS = 3 async def nudge_expiring_renewal_requests( session: AsyncSession, now: datetime | None = None, ) -> dict[str, Any]: """Warn approvers about submissions that are about to lapse. Without this, the most likely failure of the whole approval flow is silent: the owner misses the submission card, nothing else ever mentions it, the request lapses at T+14, and the requester's access is recycled because nobody said no — they just never said anything. One nudge near the deadline converts that into a decision. ``owner_reminded_at`` makes it exactly-once per request; a daily re-send would recreate the reminder-spam pattern this rework removed. Grouped per approver per submission, so an owner with several waiting requests gets one card naming all of them. """ now_dt = now or datetime.now(UTC) cutoff = now_dt + timedelta(days=RENEWAL_OWNER_NUDGE_LEAD_DAYS) rows = (await session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.status.in_(RENEWAL_STATUSES_OPEN), CmdbCloudGrantRenewalRequest.lapse_at <= cutoff, CmdbCloudGrantRenewalRequest.lapse_at > now_dt, CmdbCloudGrantRenewalRequest.owner_reminded_at.is_(None), ) )).scalars().all() if not rows: return {"nudged": 0, "errors": []} service = CloudRenewalService(session) errors: list[str] = [] # Group by scope so one owner hears about their whole queue once. by_scope: dict[str, list[CmdbCloudGrantRenewalRequest]] = {} for req in rows: by_scope.setdefault(str(req.scope_id), []).append(req) nudged = 0 for scope_id, reqs in by_scope.items(): try: scope = await session.get(CmdbCloudScope, uuid.UUID(scope_id)) if scope is None or scope.deleted_at is not None: # Scope gone — the decide path would refuse anyway. Mark # them reminded so this does not retry forever. for req in reqs: req.owner_reminded_at = now_dt await session.commit() continue recipients = ( await service.lifecycle._resolve_request_approval_recipients( scope, requires_admin=False, ) ) groups = {str(r.request_group_id or r.id) for r in reqs} people: dict[str, str] = {} for req in reqs: holder = await session.get( IdentityPerson, uuid.UUID(str(req.person_id)), ) if holder is not None: people[str(req.person_id)] = ( holder.display_name or holder.work_email or "—" ) scope_label = scope.display_name or scope.external_id soonest = min(r.lapse_at for r in reqs) # One submission → point at it so the queue flashes that row. # Several → the queue itself is the right landing. nudge_url = f"/cloud/scopes/{scope.id}?renewals=1" + ( f"&group={next(iter(groups))}" if len(groups) == 1 else "" ) for person in recipients: await service.notify.dispatch( NotificationEnvelope( recipient=person, title=( f"[待你审批·即将超期] {scope_label} 有 " f"{len(groups)} 条续期申请" ), body_text=( f"{scope_label} 上有 {len(groups)} 条云权限续期" f"申请仍未审批,最早将在 {soonest:%Y-%m-%d} 自动" "关闭。\n\n" "申请人:" + "、".join(sorted(people.values())) + "\n\n关闭后这些权限会按原到期日回收 —— 如果" "确实不该继续,请直接驳回(申请人会收到原因);" "如果还需要,请尽快批准。" ), url=nudge_url, channels=["feishu", "email"], card_actions=[ CardAction( label=LocalizedText( zh="去审批", en="Go to approval", ja="承認画面へ", ), url=nudge_url, style="primary", ), ], related_resource_type=( "cmdb_cloud_grant_renewal_request" ), related_resource_id=next(iter(groups)), ), trace_id="cloud.renewal_lapse_sweep", ) for req in reqs: req.owner_reminded_at = now_dt # Per-scope commit — the outbox INSERT FK-locks the # recipient's ``identity_persons`` row until COMMIT, and a # sweep-wide transaction is what caused the 2026-07-10 # platform-wide login outage. await session.commit() nudged += len(reqs) except Exception as exc: # noqa: BLE001 logger.exception("renewal nudge failed for scope %s", scope_id) errors.append(f"scope {scope_id}: {exc}") await session.rollback() return {"nudged": nudged, "errors": errors} async def lapse_stale_renewal_requests( session: AsyncSession, now: datetime | None = None, ) -> dict[str, Any]: """Close renewal requests nobody decided within the lapse window. Without this, an undecided request sits open forever: it keeps the grant's one open-request slot occupied (so the holder cannot resubmit), and it stays approvable months later against a grant whose business context has moved on. Approving a stale request is how you resurrect access that everyone had already forgotten about. Lapsing is deliberately NOT an approval and NOT a rejection — the grant expires on its original schedule, exactly as if no request had been filed. The holder is told, because the alternative is access disappearing with no explanation, which is the failure mode the whole expiry-notification rework exists to remove. """ now_dt = now or datetime.now(UTC) rows = await session.execute( select(CmdbCloudGrantRenewalRequest).where( CmdbCloudGrantRenewalRequest.status.in_(RENEWAL_STATUSES_OPEN), CmdbCloudGrantRenewalRequest.lapse_at <= now_dt, ) ) requests = list(rows.scalars().all()) if not requests: return {"lapsed": 0, "groups_notified": 0, "errors": []} service = CloudRenewalService(session) audit = AuditService(session) errors: list[str] = [] by_group: dict[str, list[CmdbCloudGrantRenewalRequest]] = {} for req in requests: by_group.setdefault( str(req.request_group_id or req.id), [], ).append(req) lapsed = 0 notified = 0 for group_id, group in by_group.items(): try: for req in group: service._close( req, RENEWAL_STATUS_LAPSED, note=None, approver_id=None, now_dt=now_dt, ) lapsed += 1 await audit.log( trace_id="cloud.renewal_lapse_sweep", actor_type="system", actor_user_id=None, action="cloud.grant.renewal_lapsed", status="success", resource_type="cmdb_cloud_scope", resource_id=str(group[0].scope_id), detail={ "request_group_id": group_id, "person_id": str(group[0].person_id), "request_ids": [str(r.id) for r in group], "lapse_days": RENEWAL_REQUEST_LAPSE_DAYS, }, ) # Per-group commit. The outbox INSERT below FK-references the # recipient's ``identity_persons`` row and TiDB holds that # parent-row lock until COMMIT — batching a whole sweep into # one transaction is what turned a mass-notification day into # a platform-wide login outage on 2026-07-10. await session.commit() scope = await session.get( CmdbCloudScope, uuid.UUID(str(group[0].scope_id)), ) if scope is not None: await _notify_lapsed(session, scope=scope, group=group) await session.commit() notified += 1 except Exception as exc: # noqa: BLE001 logger.exception("renewal lapse failed for group %s", group_id) errors.append(f"group {group_id}: {exc}") await session.rollback() return { "lapsed": lapsed, "groups_notified": notified, "errors": errors, } async def _notify_lapsed( session: AsyncSession, *, scope: CmdbCloudScope, group: list[CmdbCloudGrantRenewalRequest], ) -> None: holder = await session.get( IdentityPerson, uuid.UUID(str(group[0].person_id)), ) if holder is None or not holder.work_email: return scope_label = scope.display_name or scope.external_id notify = NotificationService(session) try: await notify.dispatch( NotificationEnvelope( recipient=holder, title=f"[续期申请已过期] {scope_label}", body_text=( f"你在 {scope_label} 上提交的云权限续期申请,已超过 " f"{RENEWAL_REQUEST_LAPSE_DAYS} 天未获审批,已自动关闭。\n\n" f"涉及 {len(group)} 项权限,将按原到期日正常回收。\n\n" "如仍需使用,请重新提交并与订阅 Owner 确认。" ), url=f"/my/cloud/scopes/{scope.id}", channels=["feishu", "email"], related_resource_type="cmdb_cloud_grant_renewal_request", related_resource_id=str( group[0].request_group_id or group[0].id, ), ), trace_id="cloud.renewal_lapse_sweep", ) except Exception as exc: # noqa: BLE001 logger.warning( "renewal: lapse notification failed for %s: %s", holder.id, exc, )