"""Half-yearly owner confirmation for cloud servers. Pair to ``owner_confirmation_service.py`` (which handles cloud SCOPES) but operating on individual SERVERS aggregated by owner. See migration 213 + ``CmdbServerOwnerConfirmation`` model for the data shape. Lifecycle: - cron lists eligible servers (skip rules below) → groups by owner → one ``cmdb_server_owner_confirmations`` row per owner per day → Feishu card carrying the per-owner server list - owner clicks "确认全部仍在使用" → POST /server-confirmations/{id}/respond with kind=confirmed_all → service stamps each snapshot server's ``last_owner_confirmed_at = NOW()`` and the confirmation row's ``status=responded`` Skip rules (per user direction in PR planning): * Server.owner_person_id IS NULL * Owner.employment_status IN ('left', 'leaver_pending') — leaver sweep handles those servers separately via offboarding's force_transfer * Server.first_seen_at >= NOW() - 90 days — newborn guard, don't ask the owner to confirm a machine that arrived yesterday * Server.status IN ('stopped', 'decommissioned') — no point pinging about idle / dead VMs, the long-idle-Top10 panel already surfaces them * Server.deleted_at IS NOT NULL — soft-deleted * Server under an ARCHIVAL tenant (e.g. ksyun / 金山云 历史) — the company no longer runs anything there; these are read-only migrated rows. Confirming "still in use" is meaningless and just nags owners about machines Helix can't act on. Sourced from ``registry.archival_providers()`` so a future archival tenant is excluded automatically without touching this service. Idempotency: ``UniqueConstraint(owner_person_id, period_start)`` — running the cron twice on the same day is a no-op, the second tick short-circuits in ``create_pending_for_owner``. """ from __future__ import annotations import logging import uuid from datetime import UTC, date, datetime, timedelta from typing import Any from sqlalchemy import func as sql_func from sqlalchemy import or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.locale import is_en as locale_is_en from app.domains.cmdb.models import ( SERVER_OWNER_CONFIRMATION_RESPONSE_CONFIRMED_ALL, SERVER_OWNER_CONFIRMATION_STATUS_AUTO_RESOLVED, SERVER_OWNER_CONFIRMATION_STATUS_PENDING, SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED, CmdbServer, CmdbServerOwnerConfirmation, ) from app.identity.person_models import IdentityPerson from app.notifications.service import ( CardAction, CardSection, CardStat, LocalizedText, NotificationEnvelope, NotificationService, ) logger = logging.getLogger(__name__) # Half-yearly cadence — owners are prompted to confirm every 180 days # since their last confirmation (or first_seen_at if never confirmed). SEMIANNUAL_DAYS = 180 # Newborn guard — skip servers younger than this so a freshly-synced # VM doesn't get a card the same day it lands. 90 days matches the # user's explicit direction in the planning round. NEWBORN_GUARD_DAYS = 90 # Excluded server statuses (already addressed via other surfaces). EXCLUDED_STATUSES = ("stopped", "decommissioned") # Excluded employment statuses (already addressed via leaver sweep). EXCLUDED_EMPLOYMENT_STATUSES = ("left", "leaver_pending") # Per-owner DoS cap. A single owner who happens to hold 50K+ machines # (legacy bulk-import, large script-managed fleet) would otherwise land # a 50K-element ``IN (...)`` clause in three different queries # (snapshot insert, card-row hydration, _all_snapshot_confirmed) and a # 50K-element JSON payload in cmdb_server_owner_confirmations.server_ids. # 500 is generous (card UI caps visible to 10 with a "+N more" hint) and # safely fits in MySQL JSON column / IN-list planner budget. Owners with # more get a card for the first 500 and a "+N truncated" hint; the # remainder is picked up on the next weekly tick after they respond. MAX_SERVERS_PER_CARD = 500 _CARD_BUTTON_CONFIRM_ZH = "✅ 确认全部仍在使用" _CARD_BUTTON_CONFIRM_EN = "✅ Confirm all still in use" _CARD_BUTTON_CONFIRM_JA = "✅ すべて利用中として確認" _CARD_BUTTON_MANAGE_ZH = "📋 前往逐台管理" _CARD_BUTTON_MANAGE_EN = "📋 Manage individually" _CARD_BUTTON_MANAGE_JA = "📋 1 台ずつ管理する" class ServerOwnerConfirmationService: """Owner-aggregated half-yearly confirmation for cloud servers.""" def __init__(self, db: AsyncSession) -> None: self.db = db self.notifications = NotificationService(db) async def list_due_owners( self, *, today: date | None = None, ) -> dict[str, list[str]]: """Return ``{owner_person_id: [server_uuid_str, ...]}`` of due rows. Eligible = NOT soft-deleted, status NOT in (stopped, decommissioned), first_seen_at <= NOW() - 90 days, owner_person_id IS NOT NULL, and (last_owner_confirmed_at IS NULL OR last_owner_confirmed_at <= NOW() - 180 days), and the owner has NO pending confirmation row already open (interlock — without this a weekly cron would re-prompt the same unresponded owner every Monday until they click). Owner leaver-status filtering happens in the caller (cron) so this method doesn't need to JOIN identity_persons for an early narrowing — keeps the query simple, and the owner-employment filter only saves cards for ~5% of due rows in practice. Per-owner list is capped at ``MAX_SERVERS_PER_CARD`` to bound the snapshot row, the IN-list at hydration, and the Feishu card body. Truncated owners pick up the remainder on the next tick after they respond. """ ref = today or datetime.now(UTC).date() newborn_cutoff = ref - timedelta(days=NEWBORN_GUARD_DAYS) semiannual_cutoff = ref - timedelta(days=SEMIANNUAL_DAYS) # Re-prompt interlock — owners with a still-pending row from a # prior tick are skipped entirely. Without this guard a weekly # cron keeps minting fresh cards (each with a new period_start) # for the same owner until they finally click. Spam. pending_owners_q = select( CmdbServerOwnerConfirmation.owner_person_id, ).where( CmdbServerOwnerConfirmation.status == SERVER_OWNER_CONFIRMATION_STATUS_PENDING, ) pending_rows = await self.db.execute(pending_owners_q) pending_owner_ids = {str(pid) for (pid,) in pending_rows.all()} q = select(CmdbServer.id, CmdbServer.owner_person_id).where( CmdbServer.deleted_at.is_(None), CmdbServer.owner_person_id.is_not(None), CmdbServer.status.notin_(EXCLUDED_STATUSES), CmdbServer.first_seen_at <= newborn_cutoff, sql_func.coalesce( CmdbServer.last_owner_confirmed_at, # Treat "never confirmed" as eligible — a value far # enough in the past that the <= comparison passes. # Without this COALESCE NULL would fail the <= test # silently and every never-confirmed server (i.e. the # entire fleet on day 1) would be skipped. datetime(1970, 1, 1, tzinfo=UTC), ) <= semiannual_cutoff, ) # Archival-tenant exclusion — servers whose integration is a # decommissioned/read-only tenant (ksyun 金山云 历史, …) must # never be surfaced for confirmation. Resolved from the driver # registry so this is not a hardcoded provider list. A server # with a NULL integration_id (hand-added, no tenant) is kept — # only rows explicitly under an archival integration are dropped. from app.integrations import registry from app.integrations.models import Integration archival = registry.archival_providers() if archival: archival_integ_ids = select(Integration.id).where( Integration.provider.in_(archival), ) q = q.where( or_( CmdbServer.integration_id.is_(None), CmdbServer.integration_id.not_in(archival_integ_ids), ) ) rows = await self.db.execute(q) grouped: dict[str, list[str]] = {} for server_id, owner_id in rows.all(): owner_str = str(owner_id) if owner_str in pending_owner_ids: continue bucket = grouped.setdefault(owner_str, []) if len(bucket) >= MAX_SERVERS_PER_CARD: continue bucket.append(str(server_id)) return grouped async def create_pending_for_owner( self, *, owner: IdentityPerson, server_ids: list[str], period_start: date | None = None, trace_id: str | None = None, ) -> tuple[CmdbServerOwnerConfirmation, bool]: """Persist a pending row and dispatch the Feishu card. Returns ``(row, deduplicated)``. ``deduplicated=True`` means a row already existed for (owner, period_start) and no card was sent — the cron's "second tick today" short-circuit. The existing row is still returned so callers can include the id in telemetry. """ ref = period_start or datetime.now(UTC).date() # Race-safe dedup: SELECT first; if a row exists the cron is # re-running same-day. Hard guard via the unique constraint # catches the SELECT-then-INSERT race (mirrors the scope # service's pattern, with the same caveat about TiDB partial # unique indexes being unavailable). existing_q = select(CmdbServerOwnerConfirmation).where( CmdbServerOwnerConfirmation.owner_person_id == str(owner.id), CmdbServerOwnerConfirmation.period_start == ref, ) existing_result = await self.db.execute(existing_q) existing = existing_result.scalar_one_or_none() if existing is not None: return existing, True row = CmdbServerOwnerConfirmation( id=uuid.uuid4(), owner_person_id=str(owner.id), period_start=ref, server_ids=server_ids, status=SERVER_OWNER_CONFIRMATION_STATUS_PENDING, notified_at=datetime.now(UTC), ) self.db.add(row) try: await self.db.flush() except IntegrityError: # Lost the (owner, period_start) UNIQUE race against # another concurrent cron tick. Roll back, refetch the # winner, return as deduplicated so the cron's # ``owners_deduped`` counter classifies this correctly # instead of bumping ``errors`` (which would over-trip # the Round-22 majority-failed gate on legitimate races). await self.db.rollback() winner = (await self.db.execute(existing_q)).scalar_one_or_none() if winner is None: # Genuinely surprising — the unique violation must # have been a different constraint. Re-raise so the # cron's per-owner catch logs the original cause. raise return winner, True # Compose the card. We need the server names + regions to give # the owner enough context — the snapshot of UUIDs alone is a # security checkbox, the displayed list is what makes the card # actionable. One extra SELECT per owner per cron tick. server_rows = await self.db.execute( select(CmdbServer).where(CmdbServer.id.in_(server_ids)) ) servers = list(server_rows.scalars().all()) # Resolve integration → provider once per cron tick (cheap # secondary fetch). Without this the card body printed a # hardcoded "—" for every row instead of aws/aliyun/ucloud/…, # making the inventory list look broken. from app.integrations.models import Integration integ_ids = {str(s.integration_id) for s in servers if s.integration_id} provider_by_integ: dict[str, str] = {} if integ_ids: integ_rows = await self.db.execute( select(Integration.id, Integration.provider).where( Integration.id.in_(integ_ids), ) ) provider_by_integ = { str(iid): (provider or "") for iid, provider in integ_rows.all() } envelope = self._build_card( owner=owner, confirmation_id=str(row.id), servers=servers, provider_by_integration=provider_by_integ, ) await self.notifications.dispatch(envelope, trace_id=trace_id or uuid.uuid4().hex) return row, False def _build_card( self, *, owner: IdentityPerson, confirmation_id: str, servers: list[CmdbServer], provider_by_integration: dict[str, str] | None = None, ) -> NotificationEnvelope: # Shared resolver: the raw ``startswith("en")`` test sent Japanese # readers the Chinese card and mis-read free-text SF locale values. is_en = locale_is_en(owner) count = len(servers) title = ( f"☁️ Half-yearly cloud server check — {count} server(s)" if is_en else f"☁️ 云服务器半年盘点 — 共 {count} 台" ) subtitle = LocalizedText( zh="请确认以下机器是否仍在使用,长期闲置的服务器会持续产生云成本。", en=( "Confirm whether the following machines are still needed; " "untouched servers continue to incur cost." ), ja=( "以下のマシンをまだ使用しているかご確認ください。" "長期間使われていないサーバーもクラウド費用が発生し続けます。" ), ) # Compact list — Feishu rich-card body grows tall fast, so cap # the visible rows at 10 and add a "...还有 N 台" hint when # the owner has more. The /server-confirmations/mine endpoint # carries the full list for the management drawer. provider_map = provider_by_integration or {} VISIBLE = 10 body_lines: list[str] = [] for s in servers[:VISIBLE]: label = s.name or s.instance_id region = s.region or "-" provider = provider_map.get(str(s.integration_id), "") or "-" body_lines.append(f"• [b]{label}[/b] · {provider} · {region}") # The "+N more" tail is the only language-dependent part of the # section body — the server rows above it are pure data, identical # in every language. Build one body per language so a single # ``CardSection`` covers all three readers. hint_zh: str | None = None hint_en: str | None = None hint_ja: str | None = None if count > VISIBLE: extra = count - VISIBLE hint_zh = f"…还有 {extra} 台,完整列表在管理页查看。" hint_en = ( f"…and {extra} more — see the management page for the full list." ) hint_ja = ( f"…ほかに {extra} 台あります。全リストは管理ページでご確認ください。" ) def _section_body(hint: str | None) -> str: if hint is None: return "\n".join(body_lines) return "\n".join([*body_lines, "", hint]) sections = [ CardSection( title=LocalizedText( zh="待确认的服务器", en="Servers awaiting confirmation", ja="確認待ちのサーバー", ), body=LocalizedText( zh=_section_body(hint_zh), en=_section_body(hint_en), ja=_section_body(hint_ja), ), tone="info", ), ] # The single-click "confirm all" URL — the management view # exposes per-server actions, but a one-button confirm is # the 80% path for owners who just want to OK their inventory. confirm_url = ( f"/my/cloud/servers?confirmationId={confirmation_id}" f"&action=confirm_all" ) manage_url = f"/my/cloud/servers?confirmationId={confirmation_id}" actions = [ CardAction( label=LocalizedText( zh=_CARD_BUTTON_CONFIRM_ZH, en=_CARD_BUTTON_CONFIRM_EN, ja=_CARD_BUTTON_CONFIRM_JA, ), url=confirm_url, style="primary", ), CardAction( label=LocalizedText( zh=_CARD_BUTTON_MANAGE_ZH, en=_CARD_BUTTON_MANAGE_EN, ja=_CARD_BUTTON_MANAGE_JA, ), url=manage_url, style="default", ), ] stats = [ CardStat( label=LocalizedText( zh="待确认台数", en="Servers", ja="確認待ち台数", ), value=str(count), ), ] # ``title`` / ``body_text`` are only the no-template fallback now: # ``cloud.server_owner_semiannual_confirm`` is registered in the # template catalog, so ``_render_for_recipient`` renders the # zh / en / **ja** variant from there and these two strings never # reach a recipient. Kept (rather than blanked) because a stored # outbox row is read back by the bell renderer, which degrades to # the envelope if the key is ever unregistered. body_text = ( f"You have {count} cloud server(s) awaiting half-yearly confirmation. " "Click below to confirm all or visit the management page for per-server actions." if is_en else f"你有 {count} 台云服务器需要做半年盘点确认,请确认是否仍在使用。" ) return NotificationEnvelope( recipient=owner, title=title, body_text=body_text, context={"count": count}, card_subtitle=subtitle, card_stats=stats, card_sections=sections, card_actions=actions, template_key="cloud.server_owner_semiannual_confirm", related_resource_type="cmdb_server_owner_confirmation", related_resource_id=uuid.UUID(confirmation_id), url=manage_url, ) async def respond_confirm_all( self, *, confirmation_id: str, actor_person_id: str, ) -> CmdbServerOwnerConfirmation: """Owner clicks "确认全部仍在使用" — stamp every snapshot server.""" row = await self._get_pending(confirmation_id) if row.owner_person_id != actor_person_id: raise PermissionError("not the owner of this confirmation") now = datetime.now(UTC) # Bump last_owner_confirmed_at on every snapshot server that # still has THIS owner attached. If the owner has been # reassigned in the period, skip that server — the new owner # will get their own card on the next cron tick. servers_q = select(CmdbServer).where( CmdbServer.id.in_(row.server_ids), CmdbServer.owner_person_id == row.owner_person_id, CmdbServer.deleted_at.is_(None), ) result = await self.db.execute(servers_q) for server in result.scalars().all(): server.last_owner_confirmed_at = now row.status = SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED row.response_kind = SERVER_OWNER_CONFIRMATION_RESPONSE_CONFIRMED_ALL row.responded_at = now await self.db.flush() return row async def respond_per_server( self, *, confirmation_id: str, actor_person_id: str, confirmed_server_ids: list[str], ) -> CmdbServerOwnerConfirmation: """Owner confirmed a subset (per-server "still in use" in the drawer). Stamps ``last_owner_confirmed_at`` on each id in ``confirmed_server_ids`` that the owner still owns. The confirmation row stays in ``pending`` if the snapshot isn't fully covered; transitions to ``responded`` when the union of confirmed servers covers every snapshot server that still exists (vanished / soft-deleted servers count as satisfied — see ``_all_snapshot_confirmed``). """ row = await self._get_pending(confirmation_id) if row.owner_person_id != actor_person_id: raise PermissionError("not the owner of this confirmation") now = datetime.now(UTC) snapshot_set = set(row.server_ids) actually_confirm = [s for s in confirmed_server_ids if s in snapshot_set] if not actually_confirm: return row servers_q = select(CmdbServer).where( CmdbServer.id.in_(actually_confirm), CmdbServer.owner_person_id == row.owner_person_id, CmdbServer.deleted_at.is_(None), ) result = await self.db.execute(servers_q) confirmed_uuids: set[str] = set() for server in result.scalars().all(): server.last_owner_confirmed_at = now confirmed_uuids.add(str(server.id)) # If every snapshot server has been confirmed, the row is # done. Otherwise it remains pending so the owner can come # back for the rest. already_done = await self._all_snapshot_confirmed(row, now) if already_done: row.status = SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED # ``_all_snapshot_confirmed=True`` already proves the # cumulative confirmation set covers the snapshot, so the # row finished as a full confirmation regardless of how # many clicks it took. Comparing against the THIS-call # ``confirmed_uuids`` would mis-label a multi-click # journey's final tick as "partial" even though everything # is confirmed — misleading on the audit timeline. row.response_kind = SERVER_OWNER_CONFIRMATION_RESPONSE_CONFIRMED_ALL row.responded_at = now await self.db.flush() return row async def _all_snapshot_confirmed( self, row: CmdbServerOwnerConfirmation, ref: datetime, ) -> bool: """Return True if every *still-existing* server in ``row.server_ids`` has a ``last_owner_confirmed_at >= row.created_at`` AND is still owned by the confirmation's owner. The required set is the intersection of the snapshot with servers that still exist and are NOT soft-deleted. A snapshot server that vanished after card issuance (hard-deleted by a sync sweep, or soft-deleted) can never be confirmed — counting it as required would wedge a ``kind=partial`` row in ``pending`` forever, and the pending-interlock in ``list_due_owners`` would then exclude that owner from EVERY future semi-annual card. Vanished servers count as satisfied; an empty intersection means the row is trivially complete. The owner filter is defense-in-depth: a snapshot server whose ``last_owner_confirmed_at`` was bumped by some other path (admin manual SQL, another owner's confirmation after re-assign) must NOT count toward "this owner finished their card", or we'd flip the row to RESPONDED without the actor having actually clicked through every snapshot entry. Re-uses the same owner predicate ``respond_per_server`` applies to the UPDATE query. """ existing_result = await self.db.execute( select(sql_func.count(CmdbServer.id)).where( CmdbServer.id.in_(row.server_ids), CmdbServer.deleted_at.is_(None), ) ) required_count = int(existing_result.scalar() or 0) if required_count == 0: # Every snapshot server is gone — nothing left to confirm. return True result = await self.db.execute( select(sql_func.count(CmdbServer.id)).where( CmdbServer.id.in_(row.server_ids), CmdbServer.deleted_at.is_(None), CmdbServer.owner_person_id == row.owner_person_id, CmdbServer.last_owner_confirmed_at >= row.created_at, ) ) confirmed_count = int(result.scalar() or 0) return confirmed_count >= required_count async def reconcile_pending( self, *, trace_id: str | None = None, ) -> dict[str, Any]: """Drop now-ineligible servers from pending snapshots. A confirmation snapshot is frozen at card time, but the fleet keeps moving: the next provider sync can stop / soft-delete / re-assign / entirely replace snapshot servers (the 2026-07-14 incident — a scan-seeded fleet was swept by the first real UCloud sync 52 minutes after the cards went out, leaving 18 owners with pending cards whose every referenced server was dead). Without reconciliation those rows are unfinishable: * the per-server response path can never stamp a deleted / re-assigned server, so ``_all_snapshot_confirmed`` never covers the snapshot and the row stays ``pending`` forever; * the re-prompt interlock in ``list_due_owners`` sees the stuck pending row and silently suppresses ALL future half-yearly cards for that owner; * the owner keeps a "pending check" TODO badge for machines they cannot act on. For each pending row this re-applies the scan-time eligibility rules to the snapshot and: * shrinks ``server_ids`` to the still-eligible subset; * flips the row to ``auto_resolved`` when nothing eligible remains (no ``last_owner_confirmed_at`` stamp — a stopped machine that comes back becomes due again on the next scan); * flips the row to ``responded`` when every REMAINING server was already confirmed by the owner since the card went out (partial journey whose unconfirmed tail died mid-period). Ineligibility reasons mirror the scan skip rules: snapshot id vanished, soft-deleted, status in ``EXCLUDED_STATUSES``, owner re-assigned, or the server's integration became archival. Every mutated row gets a system audit entry carrying the dropped ids + reasons; the caller (cron handler / executor) owns the commit. """ from app.audit.service import AuditService from app.integrations import registry from app.integrations.models import Integration result = await self.db.execute( select(CmdbServerOwnerConfirmation).where( CmdbServerOwnerConfirmation.status == SERVER_OWNER_CONFIRMATION_STATUS_PENDING, ) ) pending_rows = list(result.scalars().all()) summary: dict[str, Any] = { "rows_scanned": len(pending_rows), "rows_shrunk": 0, "rows_auto_resolved": 0, "rows_completed": 0, "servers_dropped": 0, } if not pending_rows: return summary all_ids: set[str] = set() for row in pending_rows: all_ids.update(str(sid) for sid in row.server_ids) servers_by_id: dict[str, CmdbServer] = {} if all_ids: server_rows = await self.db.execute( select(CmdbServer).where(CmdbServer.id.in_(list(all_ids))) ) servers_by_id = { str(s.id): s for s in server_rows.scalars().all() } # Archival integrations — same exclusion the scan applies, so a # tenant flipped to archival mid-period drops out of open cards. archival_integ_ids: set[str] = set() archival = registry.archival_providers() if archival: integ_rows = await self.db.execute( select(Integration.id).where( Integration.provider.in_(archival), ) ) archival_integ_ids = {str(iid) for (iid,) in integ_rows.all()} def _ineligible_reason( row: CmdbServerOwnerConfirmation, sid: str, ) -> str | None: server = servers_by_id.get(sid) if server is None: return "missing" if server.deleted_at is not None: return "soft_deleted" if str(server.owner_person_id or "") != row.owner_person_id: return "owner_changed" if server.status in EXCLUDED_STATUSES: return str(server.status) if ( server.integration_id and str(server.integration_id) in archival_integ_ids ): return "archival_tenant" return None audit = AuditService(self.db) now = datetime.now(UTC) for row in pending_rows: eligible: list[str] = [] dropped: dict[str, str] = {} for sid in (str(s) for s in row.server_ids): reason = _ineligible_reason(row, sid) if reason is None: eligible.append(sid) else: dropped[sid] = reason if not dropped: continue row.server_ids = eligible summary["rows_shrunk"] += 1 summary["servers_dropped"] += len(dropped) if not eligible: row.status = SERVER_OWNER_CONFIRMATION_STATUS_AUTO_RESOLVED summary["rows_auto_resolved"] += 1 new_status = row.status elif await self._all_snapshot_confirmed(row, now): # The owner already confirmed everything that is still # alive — the only thing keeping the row pending was # the now-dropped dead tail. Close it as a completed # response so their earlier clicks aren't lost. row.status = SERVER_OWNER_CONFIRMATION_STATUS_RESPONDED row.response_kind = ( SERVER_OWNER_CONFIRMATION_RESPONSE_CONFIRMED_ALL ) row.responded_at = now summary["rows_completed"] += 1 new_status = row.status else: new_status = SERVER_OWNER_CONFIRMATION_STATUS_PENDING await audit.log( trace_id=trace_id or uuid.uuid4().hex, actor_type="system", actor_user_id=None, action="cmdb.server_owner_confirmation.reconcile", status="success", resource_type="cmdb_server_owner_confirmation", resource_id=str(row.id), detail={ "owner_person_id": row.owner_person_id, "dropped": dropped, "remaining": len(eligible), "new_status": new_status, }, ) await self.db.flush() return summary async def list_pending_for_owner( self, person_id: str, ) -> list[CmdbServerOwnerConfirmation]: result = await self.db.execute( select(CmdbServerOwnerConfirmation).where( CmdbServerOwnerConfirmation.owner_person_id == person_id, CmdbServerOwnerConfirmation.status == SERVER_OWNER_CONFIRMATION_STATUS_PENDING, ).order_by(CmdbServerOwnerConfirmation.created_at.desc()) ) return list(result.scalars().all()) async def get_for_owner( self, confirmation_id: str, person_id: str, ) -> CmdbServerOwnerConfirmation | None: result = await self.db.execute( select(CmdbServerOwnerConfirmation).where( CmdbServerOwnerConfirmation.id == uuid.UUID(confirmation_id), CmdbServerOwnerConfirmation.owner_person_id == person_id, ) ) return result.scalar_one_or_none() async def _get_pending( self, confirmation_id: str, ) -> CmdbServerOwnerConfirmation: result = await self.db.execute( select(CmdbServerOwnerConfirmation).where( CmdbServerOwnerConfirmation.id == uuid.UUID(confirmation_id), ) # Lock the row so two concurrent responds (double-click, or a # Feishu card + web submit racing) serialize on the pending # check below instead of both passing and double-processing — # matching the cloud-scope twin ``OwnerConfirmationService. # _get_pending``. .with_for_update() ) row = result.scalar_one_or_none() if row is None: raise LookupError(f"confirmation {confirmation_id} not found") if row.status != SERVER_OWNER_CONFIRMATION_STATUS_PENDING: raise ValueError( f"confirmation {confirmation_id} status={row.status} " "is not pending — already responded or escalated" ) return row