"""Domain registration request flow. Mirrors ``app.domains.sso.service`` — the only differences are the Feishu flow_key (``domain_registration_request``), the form fields (domain_name + purpose, two text widgets), and the absence of an explicit "system owner" handle (the requester IS the owner; handover is the single way owner changes hands). """ from __future__ import annotations import json import logging import re import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.approval.cancel_desync import notify_feishu_cancel_desync from app.approval.feishu_client import FeishuApprovalClient from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent from app.approval.submission_intents import ( compact_terminal_intent_payload, defer_submission_intent_retry, reconciliation_due_predicate, reconciliation_order, schedule_submission_intent_retry, submit_with_durable_intent, terminal_intent_resource_id, ) from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.domains.domain_registration.models import ( DOMAIN_REG_STATUS_APPROVED, DOMAIN_REG_STATUS_CANCELLED, DOMAIN_REG_STATUS_PENDING, DOMAIN_REG_STATUS_REJECTED, DOMAIN_REG_STATUS_REVOKED, DomainRegistration, ) from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config from app.integrations.service import get_approval_flow logger = logging.getLogger(__name__) _FEISHU_FLOW_KEY = "domain_registration_request" _INTENT_RECOVERABLE_STATES = ( "prepared", "submitting", "unknown", "provider_created", "compensation_pending", ) # Loose FQDN regex; the registrar's API will do the strict check at # activation time. We just want to reject trivial typos at submit # time so a malformed value doesn't reach the approver. _DOMAIN_RE = re.compile( r"^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)" r"(\.[a-zA-Z]{2,63})+$", ) _DOMAIN_FIELD_KEYWORDS: dict[str, tuple[str, ...]] = { "domain_name": ("domain", "域名"), "purpose": ("purpose", "原因", "理由", "用途", "description"), } _DOMAIN_SKIP_WIDGET_TYPES = frozenset( {"text", "section", "image", "attachmentV2"}, ) def _match_widget(widget_name: str, field_key: str) -> bool: name_lower = (widget_name or "").lower() return any( kw.lower() in name_lower for kw in _DOMAIN_FIELD_KEYWORDS.get(field_key, ()) ) def _build_domain_form_payload( widgets: list[dict[str, Any]], values: dict[str, Any], ) -> str: # Honour Feishu ``display_condition`` — hidden widgets skipped # (no required-check, no serialisation). Shared evaluator in # ``app.integrations.feishu_form_visibility``; see that module's # docstring for schema + operator coverage. from app.integrations.feishu_form_visibility import ( evaluate_display_conditions, ) visibility = evaluate_display_conditions(widgets, {}, assume_visible_when_answers_unsourced=True) serialized: list[dict[str, Any]] = [] for widget in widgets: wtype = widget.get("type", "input") if wtype in _DOMAIN_SKIP_WIDGET_TYPES: continue wid = widget.get("id") if not wid: continue if not visibility.get(wid, True): continue wname = widget.get("name") or "" matched_value: Any = None for field_key in _DOMAIN_FIELD_KEYWORDS: if _match_widget(wname, field_key) and field_key in values: matched_value = values[field_key] break if matched_value is None and widget.get("required"): if wtype in ("textarea", "input"): matched_value = values.get("purpose", "") if matched_value in (None, "", [], {}): continue text = ( str(matched_value) if not isinstance(matched_value, (dict, list)) else json.dumps(matched_value, ensure_ascii=False) ) serialized.append({"id": wid, "type": wtype, "value": text}) return json.dumps(serialized, ensure_ascii=False) def _parse_template_widgets(raw_form: Any) -> list[dict[str, Any]]: if isinstance(raw_form, str): return json.loads(raw_form or "[]") if isinstance(raw_form, list): return raw_form return [] class DomainRegistrationService: # Outcome of the most recent ``cancel()`` call's best-effort Feishu # side-cancel. Routers / MCP tools read it after ``cancel()`` to # surface ``feishu_cancel_synced`` to the requester. last_feishu_cancel_synced: bool = True def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self._feishu_client: FeishuApprovalClient | None = None async def _client(self) -> FeishuApprovalClient: if self._feishu_client is None: cfg = await get_feishu_config(session=self.db) self._feishu_client = FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) return self._feishu_client # ── Submit ────────────────────────────────────────────────── async def submit( self, *, requester: IdentityPerson, domain_name: str, purpose: str, trace_id: str, ) -> DomainRegistration: requester_id = requester.id # Round 2b audit fix-up 2026-05-04: service-layer leaver gate # so MCP-direct callers (PAT) can't bypass the router-level # ``_leaver_runtime_gate``. if requester.employment_status in ("leaver_pending", "left"): raise AppException( code="DOMAIN_REG_REQUESTER_NOT_ACTIVE", message=( "Your account is in offboarding; new domain " "registrations are not accepted." ), status_code=403, ) clean_domain = (domain_name or "").strip().lower() if not clean_domain: raise AppException( code="DOMAIN_REG_NAME_REQUIRED", message="Please enter a domain name.", status_code=400, ) if not _DOMAIN_RE.match(clean_domain): raise AppException( code="DOMAIN_REG_NAME_INVALID", message=( "Domain name doesn't look right — please use a " "fully-qualified form like ``example.com``." ), status_code=400, ) clean_purpose = (purpose or "").strip() if not clean_purpose: raise AppException( code="DOMAIN_REG_PURPOSE_REQUIRED", message="Please describe what the domain is for.", status_code=400, ) # Atomicity (review batch 33): the duplicate-pending guard # below is check-then-act with no lock and no backing unique # constraint, so two concurrent submits (double-click / agent # retry) could both pass the check and each fire a Feishu # approval instance. Mirror the canonical ``apikey.submit`` # pattern: take a per-requester distributed lock and re-check # pending INSIDE the lock so only one submission proceeds. from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) lock_key = f"domain_reg_submit:{requester_id}" try: async with distributed_lock(lock_key, ttl=120): # The auth dependency may already have opened a repeatable-read # snapshot before we waited for the logical lock. End it, then # reload the applicant with a locking current read so the lock # winner's committed pending row and lifecycle are visible. await self.db.rollback() current_requester = await self.db.scalar( select(IdentityPerson) .where(IdentityPerson.id == requester_id) .with_for_update() .execution_options(populate_existing=True) ) if current_requester is None or current_requester.employment_status in ( "leaver_pending", "left", ): raise AppException( code="DOMAIN_REG_REQUESTER_NOT_ACTIVE", message=( "Your account is in offboarding; new domain " "registrations are not accepted." ), status_code=403, ) requester = current_requester # Pre-launch hardening 2026-05-04: reject duplicate pending # submissions by the same requester (double-click / agent retry). existing_pending = await self.db.scalar( select(DomainRegistration.id).where( DomainRegistration.person_id == str(requester.id), DomainRegistration.status == DOMAIN_REG_STATUS_PENDING, ).limit(1) ) if existing_pending: raise AppException( code="DOMAIN_REG_ALREADY_PENDING", message=( "You already have a pending domain registration. " "Cancel it first or wait for it to be processed." ), status_code=409, ) flow = await get_approval_flow(_FEISHU_FLOW_KEY, session=self.db) approval_code = flow.external_code client = await self._client() definition = await client.get_approval_definition(approval_code) widgets = _parse_template_widgets(definition.get("form", "[]")) field_values = { "domain_name": clean_domain, "purpose": clean_purpose, } form_payload = _build_domain_form_payload(widgets, field_values) feishu_user_id = await client.resolve_user_id_by_email( requester.work_email, ) reg_id = str(uuid.uuid4()) intent = await submit_with_durable_intent( self.db, client=client, scope=_FEISHU_FLOW_KEY, applicant_id=requester.id, semantic_payload={ "domain_name": clean_domain, "purpose": clean_purpose, }, provider_payload={ "approval_code": approval_code, "identity_type": "user_id", "identity_value": feishu_user_id, "form": form_payload, }, finalization_payload={ "domain_registration_id": reg_id, "approval_request_id": str(uuid.uuid4()), "approval_code": approval_code, "domain_name": clean_domain, "purpose": clean_purpose, }, trace_id=trace_id, ) row = await self._finalize_submission_intent( intent.id, recovered=( intent.attempt_count > 1 or str((intent.payload or {}).get("trace_id") or "") != trace_id ), ) if row.status == DOMAIN_REG_STATUS_CANCELLED: raise AppException( code="DOMAIN_REG_REQUESTER_NOT_ACTIVE", message=( "Your account is in offboarding; the Feishu " "approval has been withdrawn." ), status_code=403, ) return row except LockBusy as exc: raise AppException( code="DOMAIN_REG_SUBMIT_IN_FLIGHT", message=( "Another submission of yours is being processed. " "Please retry in a few seconds." ), status_code=409, ) from exc async def _finalize_submission_intent( self, intent_id: uuid.UUID, *, recovered: bool, ) -> DomainRegistration: intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None or intent.scope != _FEISHU_FLOW_KEY: raise RuntimeError("domain-registration submission intent disappeared") if intent.state in {"completed", "compensated"}: reg_id = terminal_intent_resource_id( intent, resource_type="domain_registration", ) existing = await self.db.get(DomainRegistration, reg_id) if existing is None: raise RuntimeError("terminal domain-registration intent has no row") return existing payload = dict((intent.payload or {}).get("finalization") or {}) reg_id = str(payload["domain_registration_id"]) existing = await self.db.get(DomainRegistration, reg_id) if intent.state == "compensation_pending": if existing is None: raise RuntimeError("domain-registration compensation has no row") await self.db.commit() await self._compensate_provider_intent(intent_id) await self.db.refresh(existing) return existing if intent.state != "provider_created" or not intent.external_id: raise RuntimeError( f"cannot finalize domain-registration intent in {intent.state!r}" ) requester = await self.db.scalar( select(IdentityPerson) .where(IdentityPerson.id == intent.applicant_id) .with_for_update() .execution_options(populate_existing=True) ) compensate = ( requester is None or requester.employment_status in ("leaver_pending", "left") ) local_status = ( DOMAIN_REG_STATUS_CANCELLED if compensate else DOMAIN_REG_STATUS_PENDING ) if existing is not None: raise RuntimeError( "provider-created domain-registration intent already has a row" ) now = datetime.now(UTC) row = DomainRegistration( id=reg_id, person_id=str(intent.applicant_id), domain_name=str(payload["domain_name"]), purpose=str(payload["purpose"]), status=local_status, feishu_instance_code=intent.external_id, created_at=now, updated_at=now, ) self.db.add(row) await self.db.flush() approval_request_id = uuid.UUID(str(payload["approval_request_id"])) approval_request = ApprovalRequest( id=approval_request_id, external_id=intent.external_id, approval_type=_FEISHU_FLOW_KEY, applicant_id=intent.applicant_id, status=local_status, source_payload={ "domain_registration_id": reg_id, "approval_code": payload["approval_code"], "domain_name": payload["domain_name"], "submission_intent_id": str(intent.id), }, ) self.db.add(approval_request) await self.db.flush() row.approval_request_id = str(approval_request_id) intent.approval_request_id = approval_request_id intent.state = "compensation_pending" if compensate else "completed" intent.completed_at = None if compensate else now intent.last_error = None await self.audit.log( trace_id=( f"approval-intent-reconcile:{intent.id}" if recovered else str((intent.payload or {}).get("trace_id")) ), actor_type="system" if recovered or compensate else "user", actor_user_id=None if recovered or compensate else intent.applicant_id, action=( "domain_registration.submit_compensating_leaver" if compensate else ( "approval.submission_intent.reconciled" if recovered else "domain_registration.submit" ) ), status="failure" if compensate else "success", resource_type="domain_registration", resource_id=reg_id, detail={ "intent_id": str(intent.id), "provider_request_uuid": intent.provider_request_uuid, "feishu_instance_code": intent.external_id, "domain_name": payload["domain_name"], "recovered": recovered, "compensation_required": compensate, }, ) if not compensate: compact_terminal_intent_payload( intent, resource_type="domain_registration", resource_id=reg_id, ) await self.db.commit() await self.db.refresh(row) if compensate: await self._compensate_provider_intent(intent_id) await self.db.refresh(row) return row async def _compensate_provider_intent(self, intent_id: uuid.UUID) -> bool: intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is None: raise RuntimeError("domain-registration compensation disappeared") if intent.state == "compensated": return True if intent.state != "compensation_pending" or not intent.external_id: raise RuntimeError( f"cannot compensate domain-registration intent in {intent.state!r}" ) stored = dict(intent.payload or {}) provider = dict(stored.get("provider") or {}) reg_id = str( (stored.get("finalization") or {}).get("domain_registration_id") ) instance_code = intent.external_id await self.db.rollback() client = await self._client() try: detail = await client.get_instance_detail(instance_code) except Exception: # noqa: BLE001 detail = {} provider_status = str(detail.get("status") or "").upper() already_closed = provider_status in { "CANCELED", "CANCELLED", "REJECTED", "REFUSED", "DELETED", "TIMEOUT", "RECALLED", "REVERTED", } try: if not already_closed: identity_type = str(provider["identity_type"]) identity_value = str(provider["identity_value"]) await client.cancel_instance( instance_code=instance_code, approval_code=str(provider["approval_code"]), user_id=identity_value if identity_type == "user_id" else None, open_id=identity_value if identity_type == "open_id" else None, ) except Exception as exc: # noqa: BLE001 await self.db.rollback() intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is not None and intent.state == "compensation_pending": schedule_submission_intent_retry(intent, exc) await self.audit.log( trace_id=f"approval-intent-compensate:{intent_id}", actor_type="system", actor_user_id=None, action="approval.submission_intent.compensation_pending", status="failure", resource_type="domain_registration", resource_id=reg_id, detail={ "intent_id": str(intent_id), "instance_code": instance_code, "error_type": type(exc).__name__, }, ) await notify_feishu_cancel_desync( self.db, flow_label="域名注册申请", resource_type="domain_registration_submission_compensation", resource_id=reg_id, instance_code=instance_code, error=exc, trace_id=f"approval-intent-compensate:{intent_id}", ) await self.db.commit() return False await self.db.rollback() intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None: raise RuntimeError("domain-registration compensation disappeared") if intent.state == "compensated": return True if intent.state != "compensation_pending": raise RuntimeError("domain-registration compensation state changed") intent.state = "compensated" intent.completed_at = datetime.now(UTC) intent.last_error = None await self.audit.log( trace_id=f"approval-intent-compensate:{intent_id}", actor_type="system", actor_user_id=None, action="approval.submission_intent.compensated", status="success", resource_type="domain_registration", resource_id=reg_id, detail={ "intent_id": str(intent_id), "instance_code": instance_code, "provider_status_before_cancel": provider_status or None, "already_closed": already_closed, }, ) compact_terminal_intent_payload( intent, resource_type="domain_registration", resource_id=reg_id, ) await self.db.commit() return True async def reconcile_submission_intents( self, *, min_age_minutes: int = 2, batch_limit: int = 100, ) -> dict[str, int]: cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes) refs = list((await self.db.execute( select( ApprovalSubmissionIntent.id, ApprovalSubmissionIntent.applicant_id, ) .where( ApprovalSubmissionIntent.scope == _FEISHU_FLOW_KEY, ApprovalSubmissionIntent.state.in_(_INTENT_RECOVERABLE_STATES), ( ApprovalSubmissionIntent.state.in_(( "provider_created", "compensation_pending", )) | (ApprovalSubmissionIntent.updated_at < cutoff) ), reconciliation_due_predicate(datetime.now(UTC)), ) .order_by(*reconciliation_order()) .limit(batch_limit) )).all()) await self.db.rollback() summary = { "checked": len(refs), "reconciled": 0, "compensated": 0, "still_unknown": 0, "failed": 0, } from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) for intent_id, applicant_id in refs: try: async with distributed_lock( f"domain_reg_submit:{applicant_id}", ttl=120, ): intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is None or intent.state not in _INTENT_RECOVERABLE_STATES: continue stored = dict(intent.payload or {}) if intent.state != "compensation_pending": intent = await submit_with_durable_intent( self.db, client=await self._client(), scope=intent.scope, applicant_id=intent.applicant_id, recovery_key=stored.get("recovery_key"), semantic_payload=dict(stored.get("semantic") or {}), provider_payload=dict(stored.get("provider") or {}), finalization_payload=dict( stored.get("finalization") or {} ), trace_id=str( stored.get("trace_id") or f"approval-intent:{intent.id}" ), ) row = await self._finalize_submission_intent( intent.id, recovered=True, ) refreshed = await self.db.get( ApprovalSubmissionIntent, intent.id, ) if refreshed is not None and refreshed.state == "compensated": summary["compensated"] += 1 elif ( refreshed is not None and refreshed.state == "compensation_pending" ): summary["still_unknown"] += 1 elif row.status == DOMAIN_REG_STATUS_PENDING: summary["reconciled"] += 1 except LockBusy as exc: await defer_submission_intent_retry(self.db, intent_id, exc) summary["still_unknown"] += 1 except AppException as exc: if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN": await self.db.rollback() summary["still_unknown"] += 1 else: await defer_submission_intent_retry(self.db, intent_id, exc) summary["failed"] += 1 except Exception as exc: # noqa: BLE001 await defer_submission_intent_retry(self.db, intent_id, exc) summary["failed"] += 1 return summary # ── Cancel ────────────────────────────────────────────────── async def cancel( self, *, reg_id: str, actor: IdentityPerson, trace_id: str, ) -> DomainRegistration: row = await self._lock(reg_id) if row.person_id != str(actor.id): raise AppException( code="DOMAIN_REG_NOT_OWNER", message="Only the original requester can cancel.", status_code=403, ) if row.status != DOMAIN_REG_STATUS_PENDING: raise AppException( code="DOMAIN_REG_INVALID_STATE", message=( f"Cannot cancel a registration in {row.status!r} state." ), status_code=409, ) self.last_feishu_cancel_synced = True if row.feishu_instance_code: try: client = await self._client() flow = await get_approval_flow( _FEISHU_FLOW_KEY, session=self.db, ) feishu_user_id = await client.resolve_user_id_by_email( actor.work_email, ) await client.cancel_instance( instance_code=row.feishu_instance_code, user_id=feishu_user_id, approval_code=flow.external_code, ) except Exception as e: # noqa: BLE001 logger.warning( "Feishu cancel failed for domain_registration %s " "(instance=%s): %s", row.id, row.feishu_instance_code, e, ) self.last_feishu_cancel_synced = False await notify_feishu_cancel_desync( self.db, flow_label="域名注册申请", resource_type="domain_registration", resource_id=str(row.id), instance_code=row.feishu_instance_code, error=e, trace_id=trace_id, ) row.status = DOMAIN_REG_STATUS_CANCELLED row.updated_at = datetime.now(UTC) await self._mirror_approval_request_status(row, "cancelled") await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="domain_registration.cancel", status="success", resource_type="domain_registration", resource_id=row.id, ) return row # ── Feishu callback ───────────────────────────────────────── async def handle_feishu_callback( self, *, approval_request: ApprovalRequest, instance_detail: dict[str, Any], ) -> None: source = approval_request.source_payload or {} reg_id = source.get("domain_registration_id") if not reg_id: logger.warning( "domain_registration callback missing id: ar=%s", approval_request.id, ) return feishu_status = (instance_detail.get("status") or "").upper() try: row = await self._lock(reg_id) except NotFoundError: logger.warning( "domain_registration callback for missing row %s — ignoring", reg_id, ) return if row.status != DOMAIN_REG_STATUS_PENDING: return now = datetime.now(UTC) if feishu_status in ("APPROVED", "PASS"): # Pre-launch hardening 2026-05-06: re-validate requester # is still active before stamping APPROVED. Domain # registrations carry annual cost + ownership semantics # — a leaver shouldn't have a domain provisioned to them # post-departure. requester = await self.db.get(IdentityPerson, row.person_id) if requester is None or requester.deleted_at is not None or ( requester.employment_status in ("left", "leaver_pending") ): row.status = DOMAIN_REG_STATUS_REJECTED row.updated_at = now requester_status = ( requester.employment_status if requester else None ) await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action="domain_registration.callback_refused_leaver", status="failure", resource_type="domain_registration", resource_id=row.id, detail={ "instance_code": approval_request.external_id, "feishu_status": feishu_status, "requester_status": requester_status, "reason": "requester_offboarded_during_pending", }, ) return row.status = DOMAIN_REG_STATUS_APPROVED row.granted_at = now elif feishu_status in ("REJECTED", "REFUSED"): row.status = DOMAIN_REG_STATUS_REJECTED elif feishu_status in ("CANCELED", "CANCELLED", "DELETED"): row.status = DOMAIN_REG_STATUS_CANCELLED else: return row.updated_at = now await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action=f"domain_registration.{row.status}", status="success", resource_type="domain_registration", resource_id=row.id, detail={ "instance_code": approval_request.external_id, "feishu_status": feishu_status, }, ) # ── Leaver handover ───────────────────────────────────────── async def list_pending_handovers_for_person( self, person_id: str, ) -> list[DomainRegistration]: rows = ( await self.db.execute( select(DomainRegistration).where( DomainRegistration.person_id == person_id, DomainRegistration.status == DOMAIN_REG_STATUS_APPROVED, ).order_by(DomainRegistration.granted_at.asc()) ) ).scalars().all() return list(rows) async def handover( self, *, reg_id: str, successor_person_id: str, actor: IdentityPerson, trace_id: str, ) -> DomainRegistration: row = await self._lock(reg_id) if row.status != DOMAIN_REG_STATUS_APPROVED: raise AppException( code="DOMAIN_REG_NOT_HANDOVERABLE", message=( f"Cannot handover a registration in {row.status!r} state." ), status_code=409, ) successor = await self.db.get( IdentityPerson, uuid.UUID(successor_person_id), ) if successor is None: raise NotFoundError("IdentityPerson", successor_person_id) if successor.id == uuid.UUID(row.person_id): raise AppException( code="DOMAIN_REG_SAME_OWNER", message="Successor must differ from the current holder.", status_code=400, ) # Pre-launch P1 fix (Round 45 follow-up, CR-311): refuse to # hand a domain to a leaver / deleted person. Without this, # an admin (or LLM via the MCP tool) can transfer a domain to # a leaving employee — the same offboarding cycle then needs # to immediately re-transfer it again, while the brief gap # leaves a leaver as the registered holder of a live domain. # Mirrors the SSO grant handover gate (sso/service.py). successor_status = (successor.employment_status or "").lower() if successor.deleted_at is not None or successor_status in { "left", "leaver_pending", }: raise AppException( code="DOMAIN_REG_SUCCESSOR_INACTIVE", message=( f"Cannot hand over domain to " f"{successor.work_email or successor.display_name} — " f"they are {successor_status or 'deleted'}. " "Pick an active employee." ), status_code=409, ) previous = row.person_id row.handover_from_person_id = previous row.person_id = str(successor.id) row.handover_at = datetime.now(UTC) row.handover_actor_id = str(actor.id) row.updated_at = row.handover_at await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="domain_registration.handover", status="success", resource_type="domain_registration", resource_id=row.id, detail={"from_person_id": previous, "to_person_id": row.person_id}, ) return row async def revoke( self, *, reg_id: str, actor: IdentityPerson, reason: str | None, trace_id: str, ) -> DomainRegistration: row = await self._lock(reg_id) if row.status != DOMAIN_REG_STATUS_APPROVED: raise AppException( code="DOMAIN_REG_NOT_REVOKABLE", message=( f"Cannot revoke a registration in {row.status!r} state." ), status_code=409, ) now = datetime.now(UTC) row.status = DOMAIN_REG_STATUS_REVOKED row.revoked_at = now row.revoke_reason = (reason or "").strip()[:255] or None row.updated_at = now await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="domain_registration.revoke", status="success", resource_type="domain_registration", resource_id=row.id, detail={"reason": row.revoke_reason}, ) return row # ── Helpers ───────────────────────────────────────────────── async def _lock(self, reg_id: str) -> DomainRegistration: # SELECT ... FOR UPDATE so a concurrent cancel + Feishu approval # webhook can't both observe the same status and race into a # split-brain. Previously this was a plain ``db.get`` — the # helper's name implied a lock that wasn't there. # ``populate_existing`` forces SQLAlchemy to overwrite any copy of this # row already in the session identity map with the freshly-locked DB # state. Without it, a caller that loaded the row earlier in the same # session would keep its stale attributes despite holding the lock — # the documented locked-get identity-map trap. Matches the convention # in provisioning/offboarding ``_lock`` helpers. r = ( await self.db.execute( select(DomainRegistration) .where(DomainRegistration.id == reg_id) .with_for_update() .execution_options(populate_existing=True) ) ).scalar_one_or_none() if r is None: raise NotFoundError("DomainRegistration", reg_id) return r async def _mirror_approval_request_status( self, row: DomainRegistration, new_status: str, ) -> None: if not row.feishu_instance_code: return ar = ( await self.db.execute( select(ApprovalRequest).where( ApprovalRequest.external_id == row.feishu_instance_code, ApprovalRequest.approval_type == _FEISHU_FLOW_KEY, ) ) ).scalar_one_or_none() if ar is None: return ar.status = new_status ar.updated_at = datetime.now(UTC)