"""Company-wide Announcement request flow. Mirrors ``app.domains.domain_registration.service`` but with a single free-text field (description). No revoke / no leaver handover — once IT publishes the announcement there's nothing to take back. Feishu flow_key: ``announcement_request``. The external_code (template GUID) is registered via ``/admin/integrations//flows``. """ from __future__ import annotations import json import logging 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, submit_with_durable_intent, terminal_intent_resource_id, ) from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.domains.announcement_request.models import ( ANNOUNCEMENT_REQ_STATUS_APPROVED, ANNOUNCEMENT_REQ_STATUS_CANCELLED, ANNOUNCEMENT_REQ_STATUS_PENDING, ANNOUNCEMENT_REQ_STATUS_REJECTED, AnnouncementRequest, ) 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 = "announcement_request" _ANNOUNCEMENT_FIELD_KEYWORDS: dict[str, tuple[str, ...]] = { "description": ( "description", "application", "申请", "说明", "公告", "内容", ), } _ANNOUNCEMENT_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 _ANNOUNCEMENT_FIELD_KEYWORDS.get(field_key, ()) ) def _build_announcement_form_payload( widgets: list[dict[str, Any]], values: dict[str, Any], ) -> str: # Honour Feishu ``display_condition`` — hidden widgets skipped. 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 _ANNOUNCEMENT_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 _ANNOUNCEMENT_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"): # The template only has one required field — the # description — so any unmatched required free-text widget # gets the description value. if wtype in ("textarea", "input"): matched_value = values.get("description", "") 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 AnnouncementRequestService: # 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, description: str, trace_id: str, ) -> AnnouncementRequest: # Round 2b audit fix-up 2026-05-04: service-layer leaver gate # to mirror the router-level dependency for MCP-direct callers. if requester.employment_status in ("leaver_pending", "left"): raise AppException( code="ANNOUNCEMENT_REQUESTER_NOT_ACTIVE", message=( "Your account is in offboarding; new announcement " "requests are not accepted." ), status_code=403, ) clean_description = (description or "").strip() if not clean_description: raise AppException( code="ANNOUNCEMENT_DESCRIPTION_REQUIRED", message="Please describe the announcement you'd like sent.", status_code=400, ) # Atomicity: 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, flooding the approver. Mirror # the canonical vpn_application / ai_tool_application pattern: take a # per-requester distributed lock, re-check pending INSIDE the lock, # and commit the pending row before releasing so a racing submit sees # it on re-check. from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) lock_key = f"announcement_req_submit:{requester.id}" try: async with distributed_lock(lock_key, ttl=120): # 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(AnnouncementRequest.id).where( AnnouncementRequest.person_id == str(requester.id), AnnouncementRequest.status == ANNOUNCEMENT_REQ_STATUS_PENDING, ).limit(1).with_for_update() ) if existing_pending: raise AppException( code="ANNOUNCEMENT_ALREADY_PENDING", message=( "You already have a pending announcement " "request. 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 = {"description": clean_description} form_payload = _build_announcement_form_payload( widgets, field_values, ) feishu_user_id = await client.resolve_user_id_by_email( requester.work_email, ) req_id = str(uuid.uuid4()) approval_request_id = str(uuid.uuid4()) intent = await submit_with_durable_intent( self.db, client=client, scope=_FEISHU_FLOW_KEY, applicant_id=requester.id, semantic_payload={"description": clean_description}, provider_payload={ "approval_code": approval_code, "identity_type": "user_id", "identity_value": feishu_user_id, "form": form_payload, }, finalization_payload={ "request_id": req_id, "approval_request_id": approval_request_id, "description": clean_description, "approval_code": approval_code, }, trace_id=trace_id, ) return await self._finalize_submission_intent( intent.id, recovered=intent.attempt_count > 1, ) except LockBusy as exc: raise AppException( code="ANNOUNCEMENT_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, ) -> AnnouncementRequest: """Atomically materialise the domain row, mirror, audit and intent.""" 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("announcement submission intent disappeared") if intent.state == "completed": request_id = terminal_intent_resource_id( intent, resource_type="announcement_request", ) existing = await self.db.get(AnnouncementRequest, request_id) if existing is None: raise RuntimeError("completed announcement intent has no row") return existing payload = dict((intent.payload or {}).get("finalization") or {}) request_id = str(payload["request_id"]) existing = await self.db.get(AnnouncementRequest, request_id) if intent.state != "provider_created" or not intent.external_id: raise RuntimeError( f"cannot finalize announcement intent in {intent.state!r}" ) now = datetime.now(UTC) if existing is None: row = AnnouncementRequest( id=request_id, person_id=str(intent.applicant_id), description=str(payload["description"]), status=ANNOUNCEMENT_REQ_STATUS_PENDING, feishu_instance_code=intent.external_id, created_at=now, updated_at=now, ) self.db.add(row) await self.db.flush() approval_request = ApprovalRequest( id=uuid.UUID(str(payload["approval_request_id"])), external_id=intent.external_id, approval_type=_FEISHU_FLOW_KEY, applicant_id=intent.applicant_id, status=ANNOUNCEMENT_REQ_STATUS_PENDING, source_payload={ "announcement_request_id": request_id, "approval_code": payload["approval_code"], }, ) self.db.add(approval_request) await self.db.flush() row.approval_request_id = str(approval_request.id) row.updated_at = now await self.db.flush() else: row = existing intent.approval_request_id = uuid.UUID(str(payload["approval_request_id"])) intent.state = "completed" intent.completed_at = 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 else "user", actor_user_id=None if recovered else intent.applicant_id, action=( "approval.submission_intent.reconciled" if recovered else "announcement_request.submit" ), status="success", resource_type="announcement_request", resource_id=request_id, detail={ "intent_id": str(intent.id), "provider_request_uuid": intent.provider_request_uuid, "feishu_instance_code": intent.external_id, "recovered": recovered, }, ) compact_terminal_intent_payload( intent, resource_type="announcement_request", resource_id=request_id, ) await self.db.commit() await self.db.refresh(row) return row async def reconcile_submission_intents( self, *, min_age_minutes: int = 2, batch_limit: int = 100, ) -> dict[str, int]: """Recover announcement creates without requiring an HTTP retry.""" cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes) recoverable_states = ( "prepared", "submitting", "unknown", "provider_created", ) intent_ids = list((await self.db.scalars( select(ApprovalSubmissionIntent.id) .where( ApprovalSubmissionIntent.scope == _FEISHU_FLOW_KEY, ApprovalSubmissionIntent.state.in_(recoverable_states), ( (ApprovalSubmissionIntent.state == "provider_created") | (ApprovalSubmissionIntent.updated_at < cutoff) ), reconciliation_due_predicate(datetime.now(UTC)), ) .order_by(*reconciliation_order()) .limit(batch_limit) )).all()) summary = { "checked": len(intent_ids), "reconciled": 0, "still_unknown": 0, "failed": 0, } if not intent_ids: return summary client = await self._client() for intent_id in intent_ids: try: intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is None or intent.state == "completed": continue intent_payload = dict(intent.payload or {}) intent = await submit_with_durable_intent( self.db, client=client, scope=intent.scope, applicant_id=intent.applicant_id, semantic_payload=dict(intent_payload.get("semantic") or {}), provider_payload=dict(intent_payload.get("provider") or {}), finalization_payload=dict( intent_payload.get("finalization") or {} ), trace_id=str( intent_payload.get("trace_id") or f"approval-intent:{intent.id}" ), ) await self._finalize_submission_intent( intent.id, recovered=True, ) summary["reconciled"] += 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, *, req_id: str, actor: IdentityPerson, trace_id: str, ) -> AnnouncementRequest: row = await self._lock(req_id) if row.person_id != str(actor.id): raise AppException( code="ANNOUNCEMENT_NOT_OWNER", message="Only the original requester can cancel.", status_code=403, ) if row.status != ANNOUNCEMENT_REQ_STATUS_PENDING: raise AppException( code="ANNOUNCEMENT_INVALID_STATE", message=( f"Cannot cancel a request 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 announcement_request %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="announcement_request", resource_id=str(row.id), instance_code=row.feishu_instance_code, error=e, trace_id=trace_id, ) row.status = ANNOUNCEMENT_REQ_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="announcement_request.cancel", # Reflect the Feishu side-cancel outcome — logging "success" when # the upstream instance is still live hides a Helix/Feishu # split-brain from compliance reviewers. status="success" if self.last_feishu_cancel_synced else "partial", resource_type="announcement_request", resource_id=row.id, detail={"feishu_cancel_synced": self.last_feishu_cancel_synced}, ) 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 {} req_id = source.get("announcement_request_id") if not req_id: logger.warning( "announcement_request callback missing id: ar=%s", approval_request.id, ) return feishu_status = (instance_detail.get("status") or "").upper() try: row = await self._lock(req_id) except NotFoundError: logger.warning( "announcement_request callback for missing row %s — ignoring", req_id, ) return if row.status != ANNOUNCEMENT_REQ_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. Feishu approvals can sit pending for days; # don't broadcast a leaver's announcement 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 = ANNOUNCEMENT_REQ_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="announcement_request.callback_refused_leaver", status="failure", resource_type="announcement_request", 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 = ANNOUNCEMENT_REQ_STATUS_APPROVED row.granted_at = now elif feishu_status in ("REJECTED", "REFUSED"): row.status = ANNOUNCEMENT_REQ_STATUS_REJECTED elif feishu_status in ("CANCELED", "CANCELLED", "DELETED"): row.status = ANNOUNCEMENT_REQ_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"announcement_request.{row.status}", status="success", resource_type="announcement_request", resource_id=row.id, detail={ "instance_code": approval_request.external_id, "feishu_status": feishu_status, }, ) # ── Helpers ───────────────────────────────────────────────── async def _lock(self, req_id: str) -> AnnouncementRequest: # SELECT ... FOR UPDATE so a concurrent cancel + Feishu approval # webhook can't race into a split-brain. Previously this was a # plain ``db.get`` — the helper's name implied a lock that # wasn't there. r = ( await self.db.execute( select(AnnouncementRequest) .where(AnnouncementRequest.id == req_id) .with_for_update() ) ).scalar_one_or_none() if r is None: raise NotFoundError("AnnouncementRequest", req_id) return r async def _mirror_approval_request_status( self, row: AnnouncementRequest, 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)