"""Notification outbox — admin API.""" from __future__ import annotations from datetime import UTC, datetime, timedelta, timezone from uuid import UUID from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, Field from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import ( get_current_user, require_admin, require_admin_read, ) from app.core.dependencies import ( require_active_employee_for_mutations as _leaver_runtime_gate, ) from app.core.masking import redact_money_in_text from app.core.response import paginated_response, success_response from app.notifications.models import ( ACK_NOTE_PREFIX, STATUS_DEAD, STATUS_DELIVERED, STATUS_DELIVERING, STATUS_DELIVERY_UNCERTAIN, STATUS_FAILED, STATUS_PARTIAL, STATUS_PENDING, NotificationOutbox, ) from app.notifications.service import ( SECRET_TEMPLATE_KEYS, NotificationEnvelope, NotificationService, ) # Round-51 S1-P1: leaver-runtime gate previously absent on this router. # All mutating endpoints (outbox cancel / retry, inbox mark-read) must # refuse leavers — symmetric with cost_router / lifecycle_router / # apikey_router. Read endpoints (list_outbox, inbox list) still gate # via require_admin_read / get_current_user; the leaver gate adds # defence-in-depth on the mutate surface. router = APIRouter( prefix="/api/v1/notifications", tags=["notifications"], dependencies=[Depends(_leaver_runtime_gate)], ) def _is_acknowledged(row: NotificationOutbox) -> bool: return row.status == STATUS_DEAD and ( row.last_error or "" ).startswith(ACK_NOTE_PREFIX) def _acked_dead_clause(): """SQL predicate matching acknowledged dead rows.""" return ( (NotificationOutbox.status == STATUS_DEAD) & NotificationOutbox.last_error.like(f"{ACK_NOTE_PREFIX}%") ) def _unacked_dead_clause(): """SQL predicate matching dead rows NO admin has acknowledged — the "needs action" triage bucket.""" from sqlalchemy import or_ return (NotificationOutbox.status == STATUS_DEAD) & or_( NotificationOutbox.last_error.is_(None), ~NotificationOutbox.last_error.like(f"{ACK_NOTE_PREFIX}%"), ) class OutboxRow(BaseModel): id: UUID recipient_person_id: UUID | None = None recipient_email: str | None = None template_key: str | None = None title: str body_text: str # Rich HTML body when the mail carried one (the Day-1 onboarding # welcome, admin-authored HTML template overrides). ``None`` for # plain-text-only mails. The admin detail drawer renders this in a # sandboxed iframe so the reviewer sees exactly what the recipient got. body_html: str | None = None url: str | None = None channels: list[str] delivered_channels: list[str] | None = None failed_channels: list[str] | None = None status: str # Triage helpers derived server-side so the FE never re-types the # ack-prefix string match (it used to, and the two could drift): # ``acknowledged`` — dead row an admin consciously closed; # ``inbox_only`` — delivered with zero push channels (every push # channel was admin-disabled; the in-app bell row was the delivery # surface). The admin table renders these as 已关闭 / 仅站内 tags. acknowledged: bool = False inbox_only: bool = False attempts: int max_attempts: int last_error: str | None = None next_attempt_at: str | None = None delivered_at: str | None = None delivery_claimed_at: str | None = None delivery_lease_expires_at: str | None = None delivery_claimed_channels: list[str] | None = None delivery_claim_count: int = 0 provider_delivery_keys: dict[str, str] | None = None related_resource_type: str | None = None related_resource_id: UUID | None = None created_at: str updated_at: str # Secret-body rows (Google first-login password) — the body is revealed # ONLY to a global_admin (IT); ``/outbox`` is also readable by ``auditor`` / # ``management`` (require_admin_read), who must never see cleartext # credentials. Canonical key set (SECRET_TEMPLATE_KEYS) lives in the service # layer and is imported at the top of this module. _SECRET_BODY_MASK = ( "🔒 [首登临时密码仅 IT 管理员(global_admin)可见 — 如需兜底," "请以 IT 管理员身份查看,或在 Google 管理控制台重置后转交]" ) def _reader_in_invoice_circle(user: AuthUser) -> bool: """Is this outbox READER inside the invoice-measure circle (finance / cloud_admin / global_admin)? Lazy import (same pattern as ``audit/router.py::_can_view_invoice``) — notifications is core infrastructure and must not import a domain module at module scope. """ from app.domains.cmdb.cost_view import can_view_invoice_measure return can_view_invoice_measure(user.role_names) def _serialize( row: NotificationOutbox, *, reveal_secret: bool = False, mask_cloud_money: bool = False, ) -> dict: body_text = row.body_text body_html = row.body_html if row.template_key in SECRET_TEMPLATE_KEYS and not reveal_secret: body_text = _SECRET_BODY_MASK # Secret-body mails are plain-text (they carry a cleartext # password), so ``body_html`` is normally None — but null it # unconditionally so a future HTML-bodied secret can never leak # the credential past the mask to a non-IT reader. body_html = None elif mask_cloud_money and row.related_resource_type in { "cmdb_cloud_scope", "finance.budget", }: # 保密红线（cost-measure-surface-sweep.md 表四 body_text 行，修法 d）： # cloud billing alert cards write absolute amounts into the body # for CIRCLE recipients; this admin read surface is gated # ``require_admin_read`` (auditor / management — outside the # circle). Per-recipient rendering (2026-07-30) keeps amounts # off NON-circle recipients' rows, but circle-recipient rows and # every historical row still carry them — so the read path # strips currency-amount fragments for out-of-circle readers. # Keyed on related_resource_type because these envelopes carry # no template_key. ``finance.budget`` is deliberately fail-closed: # the outbox does not persist a budget's cost_type, so a historical # row may be cloud or wildcard/mixed. Over-redacting a software-only # budget in this wider admin-read surface is safer than leaking a # cloud invoice amount; its direct employee notification remains # unredacted because the writer knows that budget is non-cloud. body_text, _ = redact_money_in_text(body_text) body_html, _ = redact_money_in_text(body_html) return OutboxRow( id=row.id, recipient_person_id=row.recipient_person_id, recipient_email=row.recipient_email, template_key=row.template_key, title=row.title, body_text=body_text, body_html=body_html, url=row.url, channels=list(row.channels or []), delivered_channels=list(row.delivered_channels or []), failed_channels=list(row.failed_channels or []), status=row.status, acknowledged=_is_acknowledged(row), inbox_only=( row.status in (STATUS_DELIVERED, STATUS_PARTIAL) and not row.delivered_channels ), attempts=row.attempts, max_attempts=row.max_attempts, last_error=row.last_error, next_attempt_at=row.next_attempt_at.isoformat() if row.next_attempt_at else None, delivered_at=row.delivered_at.isoformat() if row.delivered_at else None, delivery_claimed_at=( claimed_at.isoformat() if (claimed_at := getattr(row, "delivery_claimed_at", None)) else None ), delivery_lease_expires_at=( lease_expires_at.isoformat() if ( lease_expires_at := getattr( row, "delivery_lease_expires_at", None, ) ) else None ), delivery_claimed_channels=list( getattr(row, "delivery_claimed_channels", None) or [] ), delivery_claim_count=getattr(row, "delivery_claim_count", 0) or 0, provider_delivery_keys=dict( getattr(row, "provider_delivery_keys", None) or {} ), related_resource_type=row.related_resource_type, related_resource_id=row.related_resource_id, created_at=row.created_at.isoformat(), updated_at=row.updated_at.isoformat(), ).model_dump(mode="json") def _outbox_filtered_stmt( *, status: str | None, q: str | None, date_from: datetime | None, date_to: datetime | None, ): """Build the filtered (unordered, unpaginated) outbox SELECT. Single source of truth for the WHERE logic shared by the list endpoint and the XLSX export — a filter fix on one surface applies to the other automatically. ``status`` accepts the triage buckets (needs_action / in_flight / notified / acked), the legacy synthetic ``unhealthy``, or a raw status; ``q`` is the case-insensitive substring search over recipient_email / title / template_key. """ stmt = select(NotificationOutbox) if date_from is not None: stmt = stmt.where(NotificationOutbox.created_at >= date_from) if date_to is not None: stmt = stmt.where(NotificationOutbox.created_at <= date_to) if status == "needs_action": from sqlalchemy import or_ stmt = stmt.where(or_( _unacked_dead_clause(), NotificationOutbox.status == STATUS_DELIVERY_UNCERTAIN, )) elif status == "in_flight": stmt = stmt.where(NotificationOutbox.status.in_( [STATUS_PENDING, STATUS_FAILED, STATUS_DELIVERING] )) elif status == "notified": stmt = stmt.where(NotificationOutbox.status.in_( [STATUS_DELIVERED, STATUS_PARTIAL] )) elif status == "acked": stmt = stmt.where(_acked_dead_clause()) elif status == "unhealthy": stmt = stmt.where(NotificationOutbox.status.in_( [ STATUS_PENDING, STATUS_PARTIAL, STATUS_FAILED, STATUS_DEAD, STATUS_DELIVERING, STATUS_DELIVERY_UNCERTAIN, ] )) elif status: stmt = stmt.where(NotificationOutbox.status == status) if q and q.strip(): # ILIKE-equivalent on three fields. The ``escape=`` arg only # declares the escape char — the user-supplied substring still # has to be escaped manually, otherwise typing a literal ``%`` # or ``_`` (uncommon for emails but possible for template_key) # would silently match every row. raw = q.strip() escaped = ( raw.replace("\\", "\\\\") .replace("%", "\\%") .replace("_", "\\_") ) like = f"%{escaped}%" from sqlalchemy import or_ stmt = stmt.where(or_( NotificationOutbox.recipient_email.ilike(like, escape="\\"), NotificationOutbox.title.ilike(like, escape="\\"), NotificationOutbox.template_key.ilike(like, escape="\\"), )) return stmt @router.get("/outbox", summary="List notification outbox rows") async def list_outbox( request: Request, current_user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), status: str | None = Query( default=None, description=( "Triage buckets (what the admin page uses): 'needs_action' " "= dead rows no admin has acknowledged + provider outcomes " "awaiting reconciliation; 'in_flight' = pending + failed + " "delivering (the retry worker is still on them); " "'notified' = delivered + partial (the recipient WAS " "notified on at least one surface); 'acked' = dead rows an " "admin closed. Raw statuses (pending / delivered / partial " "/ failed / dead / delivering / delivery_uncertain) and the " "legacy synthetic 'unhealthy' (all non-delivered operational " "statuses plus partial) still match exactly. " "Resolved server-side so the row set AND the pagination " "total stay honest." ), ), q: str | None = Query( default=None, max_length=200, description=( "Free-text search. Matched case-insensitively against " "recipient_email, title, and template_key — any one hit " "qualifies. Empty / whitespace-only is ignored so the UI " "can bind the box to a single state without a special " "'cleared' branch." ), ), date_from: datetime | None = Query( default=None, description=( "Inclusive lower bound on created_at (ISO datetime). The " "table has indefinite retention — this is how an operator " "reaches records older than the default recent pages " "without paginating through everything newer." ), ), date_to: datetime | None = Query( default=None, description="Inclusive upper bound on created_at (ISO datetime).", ), page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=200), ): stmt = _outbox_filtered_stmt( status=status, q=q, date_from=date_from, date_to=date_to, ) count_stmt = select(func.count()).select_from(stmt.subquery()) total = (await db.execute(count_stmt)).scalar() or 0 rows = list(( await db.execute( stmt .order_by(NotificationOutbox.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) ).scalars().all()) return paginated_response( data=[ _serialize( r, reveal_secret=current_user.is_admin, mask_cloud_money=not _reader_in_invoice_circle(current_user), ) for r in rows ], total=total, page=page, page_size=page_size, request=request, ) @router.get("/outbox/export.xlsx", summary="Export the filtered outbox as XLSX") async def export_outbox( request: Request, current_user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), status: str | None = Query( default=None, description=( "Same triage buckets / raw statuses as the list endpoint — " "the UI passes the exact filter the table is showing." ), ), q: str | None = Query(default=None, max_length=200), date_from: datetime | None = Query(default=None), date_to: datetime | None = Query(default=None), ): """Export the filtered notification-outbox list as XLSX. Filter signature mirrors :func:`list_outbox` minus pagination, and RBAC mirrors it too (``require_admin_read``). Reuses :func:`_outbox_filtered_stmt` so the WHERE logic cannot drift from the list. Capped at :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows; nested JSON payloads (content / metadata) are not exported — only the scalar triage fields the admin table shows. """ from app.core.exporting import EXPORT_ROW_LIMIT, xlsx_export_response stmt = _outbox_filtered_stmt( status=status, q=q, date_from=date_from, date_to=date_to, ) count_stmt = select(func.count()).select_from(stmt.subquery()) total = (await db.execute(count_stmt)).scalar() or 0 rows = list(( await db.execute( stmt .order_by(NotificationOutbox.created_at.desc()) .limit(EXPORT_ROW_LIMIT) ) ).scalars().all()) columns = [ ("recipient_email", "收件人邮箱"), ("recipient_person_id", "收件人ID"), ("title", "标题"), ("template_key", "模板键"), ("status", "状态"), ("acknowledged", "已确认关闭"), ("inbox_only", "仅站内"), ("channels", "渠道"), ("delivered_channels", "已送达渠道"), ("failed_channels", "失败渠道"), ("attempts", "尝试次数"), ("max_attempts", "最大尝试次数"), ("last_error", "最后错误"), ("next_attempt_at", "下次尝试时间"), ("delivered_at", "送达时间"), ("created_at", "创建时间"), ("id", "ID"), ] return await xlsx_export_response( db, request, current_user, columns=columns, rows=[ _serialize( r, mask_cloud_money=not _reader_in_invoice_circle(current_user), ) for r in rows ], total_matched=total, filename_stem="notification_outbox", sheet_title="Notification Outbox", audit_action="notifications.outbox.export", audit_resource_type="notification_outbox", filters={ "status": status, "q": q, "date_from": date_from, "date_to": date_to, }, ) @router.get("/outbox/summary", summary="Counts by status + channel health") async def outbox_summary( request: Request, current_user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): rows = ( await db.execute( select(NotificationOutbox.status, func.count(NotificationOutbox.id)) .group_by(NotificationOutbox.status) ) ).all() counts = {status: int(count) for status, count in rows} # Legacy bucket kept for API compat (old clients / MCP callers). unhealthy = sum( counts.get(s, 0) for s in ( STATUS_PENDING, STATUS_FAILED, STATUS_PARTIAL, STATUS_DEAD, STATUS_DELIVERING, STATUS_DELIVERY_UNCERTAIN, ) ) # Triage buckets — the numbers the admin page actually renders. # Raw ``by_status.dead`` is dominated by acknowledged history (prod: # 4 895 of 4 896 dead rows were already consciously closed), so the # page counting it stayed permanently red and the operator stopped # trusting it. ``needs_action`` is the inbox-zero number. acked = ( await db.execute( select(func.count(NotificationOutbox.id)) .where(_acked_dead_clause()) ) ).scalar() or 0 buckets = { "needs_action": ( counts.get(STATUS_DEAD, 0) - int(acked) + counts.get(STATUS_DELIVERY_UNCERTAIN, 0) ), "in_flight": ( counts.get(STATUS_PENDING, 0) + counts.get(STATUS_FAILED, 0) + counts.get(STATUS_DELIVERING, 0) ), "notified": ( counts.get(STATUS_DELIVERED, 0) + counts.get(STATUS_PARTIAL, 0) ), "acked": int(acked), } # Recent throughput beats an all-time delivered total (which only # ever grows and says nothing about whether delivery works TODAY). delivered_7d = ( await db.execute( select(func.count(NotificationOutbox.id)).where( NotificationOutbox.status.in_( [STATUS_DELIVERED, STATUS_PARTIAL] ), NotificationOutbox.delivered_at >= datetime.now(UTC) - timedelta(days=7), ) ) ).scalar() or 0 # Dead ops-critical alerts (guaranteed-delivery templates) — the # loudest number on the page: each one is an alert about a broken # thing that itself failed to reach anyone. Counted over template # prefixes via OR'd LIKEs (the prefix list is short and static). # Acknowledged rows ('cancelled by admin' note — a human looked # and consciously closed it) are excluded; without the carve-out # the banner would stay red forever, since a dead row's only other # exit is a successful retry, which a structurally-failed send # never produces (2026-06-10 dead-letter ack). from sqlalchemy import or_ from app.notifications.service import _CRITICAL_TEMPLATE_PREFIXES critical_dead = ( await db.execute( select(func.count(NotificationOutbox.id)).where( _unacked_dead_clause(), or_(*[ NotificationOutbox.template_key.like(f"{p}%") for p in _CRITICAL_TEMPLATE_PREFIXES ]), ) ) ).scalar() or 0 # Channel health: which push channels are actually usable right # now. The state mirrors the dispatcher's own skip logic exactly # (available → sends; disabled → skipped; unconfigured → attempted # and fails loudly), so the banner can never disagree with what # delivery will actually do. from app.notifications.service import _CHANNEL_INTEGRATION_KIND svc = NotificationService(db) availability = await svc.channel_availability() channels = { channel: { "integration_kind": kind, "state": availability.get(channel, "available"), } for channel, kind in _CHANNEL_INTEGRATION_KIND.items() } return success_response( data={ "by_status": counts, "unhealthy": unhealthy, "buckets": buckets, "delivered_7d": int(delivered_7d), "critical_dead": int(critical_dead), "channels": channels, }, request=request, ) @router.get( "/outbox/employee-reach", summary="How much this platform pushes at employees, per template", ) async def outbox_employee_reach( request: Request, current_user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), days: int = Query(default=30, ge=1, le=180), limit: int = Query(default=25, ge=1, le=100), ): """「别给员工发太多」的可证伪版本。 The sibling ``/outbox/report`` answers *did delivery work*. This one answers *how much are we asking of people*, and reports the **per-recipient** average rather than the total — because a template that sent 500 cards to 500 people is fine, one that sent 500 to 30 people is not, and a total cannot tell them apart. That is precisely how a 16.7-per-person-per-month reminder sat unnoticed underneath a headline that looked like ordinary volume. Scoped to ACTIVE employees so operational admin paging (which is genuinely high-volume and genuinely wanted) cannot dilute the number that is supposed to describe employee experience. Admin-read gated. """ from app.notifications import cadence return success_response( data=await cadence.volume_report(db, window_days=days, limit=limit), request=request, ) @router.get("/outbox/report", summary="Aggregated outbox report") async def outbox_report( request: Request, current_user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), days: int = Query(default=30, ge=1, le=365), tz_offset_minutes: int = Query( default=0, ge=-840, le=840, description=( "Viewer's offset east of UTC in minutes (JS: " "-new Date().getTimezoneOffset()). Day buckets are cut on " "the viewer's local midnight — a UTC cut would shift every " "CN-office evening notification into the next day's bar." ), ), ): """Report tab behind the admin outbox page: daily volume by status, per-channel delivery honesty, and top templates over a created-at window. Channel stats come from the per-row ``channels`` / ``delivered_channels`` / ``failed_channels`` JSON lists, which SQL can't group — so this scans the window (newest-first, capped) and aggregates in Python, same approach as the error-cluster scan. """ since = datetime.now(UTC) - timedelta(days=days) scan_cap = 50_000 rows = ( await db.execute( select( NotificationOutbox.created_at, NotificationOutbox.status, NotificationOutbox.channels, NotificationOutbox.delivered_channels, NotificationOutbox.failed_channels, NotificationOutbox.template_key, ) .where(NotificationOutbox.created_at >= since) .order_by(NotificationOutbox.created_at.desc()) .limit(scan_cap) ) ).all() tz = timezone(timedelta(minutes=tz_offset_minutes)) statuses = ( STATUS_PENDING, STATUS_DELIVERED, STATUS_PARTIAL, STATUS_FAILED, STATUS_DEAD, STATUS_DELIVERING, STATUS_DELIVERY_UNCERTAIN, ) # Pre-fill every local date in the window so the chart axis is # continuous — a zero-send day must render as a gap-free zero bar, # not silently collapse the axis. today_local = datetime.now(tz).date() daily: dict[str, dict[str, int]] = {} for back in range(days - 1, -1, -1): key = (today_local - timedelta(days=back)).isoformat() daily[key] = {s: 0 for s in statuses} | {"total": 0} by_status: dict[str, int] = {s: 0 for s in statuses} channels: dict[str, dict[str, int]] = {} templates: dict[str, dict[str, int]] = {} for created_at, status, chans, delivered, failed, template_key in rows: # MySQL DATETIME round-trips naive — re-attach UTC before # shifting into the viewer's timezone. if created_at.tzinfo is None: created_at = created_at.replace(tzinfo=UTC) day = created_at.astimezone(tz).date().isoformat() bucket = daily.get(day) if bucket is not None: bucket[status] = bucket.get(status, 0) + 1 bucket["total"] += 1 by_status[status] = by_status.get(status, 0) + 1 delivered_set = set(delivered or []) failed_set = set(failed or []) for c in chans or []: ch = channels.setdefault( c, {"attempted": 0, "delivered": 0, "failed": 0}, ) ch["attempted"] += 1 if c in delivered_set: ch["delivered"] += 1 elif c in failed_set: ch["failed"] += 1 tpl = templates.setdefault( template_key or "(none)", {"total": 0, "notified": 0, "dead": 0}, ) tpl["total"] += 1 if status in (STATUS_DELIVERED, STATUS_PARTIAL): tpl["notified"] += 1 elif status == STATUS_DEAD: tpl["dead"] += 1 top_templates = [ {"template_key": k, **v} for k, v in sorted( templates.items(), key=lambda kv: kv[1]["total"], reverse=True, )[:10] ] total = len(rows) notified = by_status[STATUS_DELIVERED] + by_status[STATUS_PARTIAL] return success_response( data={ "days": days, "since": since.isoformat(), "daily": [{"date": d, **v} for d, v in daily.items()], "by_status": by_status, "channels": channels, "top_templates": top_templates, "totals": { "total": total, "notified": notified, "in_flight": ( by_status[STATUS_PENDING] + by_status[STATUS_FAILED] + by_status[STATUS_DELIVERING] ), "dead": ( by_status[STATUS_DEAD] + by_status[STATUS_DELIVERY_UNCERTAIN] ), "notified_rate": ( round(notified / total, 4) if total else None ), }, "scanned": total, "scan_capped": total >= scan_cap, }, request=request, ) @router.get( "/outbox/error-clusters", summary="Group unhealthy outbox rows by normalised error", ) async def outbox_error_clusters( request: Request, current_user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), status: str = Query( default="unhealthy", description=( "Status scope: a concrete status (dead / partial / failed / " "pending / delivering / delivery_uncertain) or 'unhealthy' " "for the full operational set." ), ), days: int | None = Query( default=None, ge=1, le=365, description="Optional created-at window (days back from now).", ), ): """Dead-letter triage view: clusters rows whose ``last_error`` differs only in variable tokens (emails / UUIDs / request ids), so 400 'Feishu open_id not found for ' rows read as ONE line with a count — and each cluster carries enough row ids for the bulk retry / cancel buttons to act on it directly. """ from app.notifications.service import compute_error_clusters if status == "unhealthy": statuses = [ STATUS_PENDING, STATUS_PARTIAL, STATUS_FAILED, STATUS_DEAD, STATUS_DELIVERY_UNCERTAIN, ] elif status in ( STATUS_PENDING, STATUS_PARTIAL, STATUS_FAILED, STATUS_DEAD, STATUS_DELIVERY_UNCERTAIN, ): statuses = [status] else: from app.core.exceptions import AppException raise AppException( code="VALIDATION_ERROR", message=( "status must be one of pending / partial / failed / " "dead / delivery_uncertain / unhealthy" ), status_code=422, ) # Acknowledged dead rows are consciously-closed history — clustering # them buries the live problem under storms an admin already # handled, so every scope that includes ``dead`` excludes them. data = await compute_error_clusters( db, statuses=statuses, days=days, exclude_acked=True, ) return success_response(data=data, request=request) class BatchIdsBody(BaseModel): """Bulk-op payload: explicit row ids only (≤500 per call). Deliberately no filter-driven mode — a typo'd filter on a mutating endpoint can silence thousands of rows; explicit ids keep the blast radius equal to what the admin actually selected (the error-cluster endpoint hands back per-cluster ids for the storm case). """ ids: list[UUID] = Field(min_length=1, max_length=500) reason: str | None = Field(default=None, max_length=500) @router.post( "/outbox/batch-retry", summary="Re-queue up to 500 outbox rows for immediate retry", ) async def batch_retry_outbox( body: BatchIdsBody, request: Request, current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Bulk dead-letter resend. Rows are revived into the retry pool (one fresh attempt each) and a background drain starts delivering immediately; delivery itself applies the same departed-recipient and disabled-channel guards as the cron worker. Delivered rows are skipped — re-firing them would duplicate already-seen messages. """ service = NotificationService(db) result = await service.revive_outbox_rows( list(body.ids), actor_user_id=current_user.id, ) await db.commit() if result["revived"]: from app.core.background import spawn from app.notifications.service import drain_due_outbox_in_background spawn( drain_due_outbox_in_background(), name="notifications.batch_retry_drain", ) return success_response(data=result, request=request) @router.post( "/outbox/batch-cancel", summary="Cancel up to 500 outbox rows", ) async def batch_cancel_outbox( body: BatchIdsBody, request: Request, current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Bulk stop / acknowledge. Retryable rows flip to ``dead``; ALREADY-dead rows get the ``cancelled by admin`` acknowledgement note (2026-06-10 dead-letter ack — the marker the critical-dead banner excludes), so a storm of dead critical alerts can be consciously closed in one click. Reuses the single-row ``cancel_outbox`` per id so each row keeps its individual audit trail (silencing alerts is a sensitive op); delivered / already-acknowledged rows no-op idempotently. """ service = NotificationService(db) cancelled = 0 skipped_terminal = 0 for outbox_id in body.ids: row = await db.get(NotificationOutbox, outbox_id) if row is None: skipped_terminal += 1 continue if row.status in (STATUS_DELIVERED, STATUS_DELIVERING) or _is_acknowledged(row): skipped_terminal += 1 continue await service.cancel_outbox( outbox_id, actor_user_id=current_user.id, reason=body.reason, ) cancelled += 1 await db.commit() return success_response( data={"cancelled": cancelled, "skipped_terminal": skipped_terminal}, request=request, ) class CancelBody(BaseModel): reason: str | None = None class ReconcileBody(BaseModel): delivered: bool note: str | None = Field(default=None, max_length=500) @router.post("/outbox/{outbox_id}/cancel", summary="Cancel a retryable row / acknowledge a dead one") async def cancel_outbox_row( outbox_id: UUID, request: Request, body: CancelBody | None = None, current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Flip a pending/partial/failed outbox row to ``dead`` without exhausting the remaining retries. Already-delivered and already-acknowledged rows are returned unchanged — the response body mirrors the row's final state so a double-click on the UI button reads as idempotent, not an error. Un-acknowledged dead rows gain the 'cancelled by admin' note (status stays dead) — the dead-letter acknowledgement. """ service = NotificationService(db) row = await service.cancel_outbox( outbox_id, actor_user_id=current_user.id, reason=(body.reason if body else None), ) await db.commit() # ``updated_at`` carries ``onupdate=func.now()`` — the UPDATE flush # expires it, and ``_serialize`` (a sync function) touching the # expired attribute triggers a sync lazy-load → MissingGreenlet # 500 AFTER the cancel already persisted. Latent since the route # shipped (masked because dead rows used to short-circuit before # any flush); surfaced by the 2026-06-10 dead-letter ack tests. await db.refresh(row) return success_response( data=_serialize( row, reveal_secret=current_user.is_admin, mask_cloud_money=not _reader_in_invoice_circle(current_user), ), request=request, ) @router.post( "/outbox/{outbox_id}/reconcile", summary="Resolve a delivery-uncertain row after provider verification", ) async def reconcile_outbox_row( outbox_id: UUID, body: ReconcileBody, request: Request, current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Persist an explicit provider-side reconciliation decision. Confirmed delivery closes the claimed channels. Confirmed absence puts the row back into the retry pool; no automatic retry is possible until an operator makes this audited decision. """ row = await NotificationService(db).reconcile_uncertain_outbox( outbox_id, delivered=body.delivered, actor_user_id=current_user.id, note=body.note, ) await db.commit() await db.refresh(row) if not body.delivered and row.status == STATUS_FAILED: from app.core.background import spawn from app.notifications.service import drain_due_outbox_in_background spawn( drain_due_outbox_in_background(), name="notifications.reconciled_retry_drain", ) return success_response( data=_serialize( row, reveal_secret=current_user.is_admin, mask_cloud_money=not _reader_in_invoice_circle(current_user), ), request=request, ) @router.post("/outbox/{outbox_id}/retry", summary="Retry one outbox row now") async def retry_outbox_row( outbox_id: UUID, request: Request, current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): # Load under a row lock (mirrors ``cancel_outbox`` and the cron drain's # ``with_for_update(skip_locked=True)``): the ``notification.outbox_retry`` # cron and this manual retry both revive the SAME failed row via # ``_deliver``. Without the lock, a concurrent admin click + cron tick # each read the row on an unlocked snapshot and both dispatch → the # recipient gets two identical Feishu cards / emails. A blocking # ``FOR UPDATE`` here waits out the cron's SKIP-LOCKED claim, then the # status gate below re-evaluates on the freshly-committed state (a row # the cron just delivered is now DELIVERED → refused). ``populate_existing`` # bypasses the identity-map cache so we act on the locked row, not a stale copy. row = await db.get( NotificationOutbox, outbox_id, with_for_update=True, populate_existing=True, ) if row is None: from app.core.exceptions import NotFoundError raise NotFoundError("NotificationOutbox", str(outbox_id)) # Idempotency / no-op gate: retry is intended for ``pending`` / # ``partial`` / ``failed`` rows. Re-invoking ``_deliver`` on an # already-delivered or admin-cancelled (``dead``) row would re-fire # Feishu cards / emails — turning the retry endpoint into a # low-cost spam vector if a privileged token (admin PAT) is # stolen. (Round-17 audit.) Symmetric with ``cancel_outbox``'s # idempotency guard at notifications/service.py:1612. from app.notifications.models import STATUS_DEAD, STATUS_DELIVERED if row.status in ( STATUS_DELIVERED, STATUS_DEAD, STATUS_DELIVERING, STATUS_DELIVERY_UNCERTAIN, ): return success_response( data={ "status": row.status, "reason": ( "outbox row is terminal or provider outcome is " "unsettled — retry refused" ), }, request=request, ) # Same guard as the resend path: a password-bearing row (the Google # first-login temp password, recorded for IT fallback) must NEVER be # re-dispatched — retry re-fires the body through the normal pipeline → # recipient.work_email + Feishu, the exact channels the out-of-band send # avoided. A FAILED password row (personal-email bounced) would otherwise # slip past the terminal gate above. Re-deliver via a Google console # reset, not this endpoint. if row.template_key in SECRET_TEMPLATE_KEYS: return success_response( data={ "status": "skipped", "reason": ( "password-bearing notification — retry refused (reset in " "the Google admin console instead)" ), }, request=request, ) from app.identity.person_models import IdentityPerson recipient = ( await db.get(IdentityPerson, row.recipient_person_id) if row.recipient_person_id else None ) if recipient is None: return success_response( data={"status": "skipped", "reason": "recipient not found"}, request=request, ) # Round-51 R2-P0-2: admin clicks "retry" on a stale outbox row whose # recipient went leaver_pending / left after the original send. # Cron retry path already guards on this; the manual admin retry # must follow suit — otherwise one mistaken click during inventory # cleanup re-fires Feishu cards / emails to ex-employees. from app.auth.service import is_effectively_departed if recipient.deleted_at is not None or is_effectively_departed(recipient): return success_response( data={ "status": "skipped", "reason": "recipient departed / deleted — redispatch refused", }, request=request, ) service = NotificationService(db) env = NotificationEnvelope( recipient=recipient, title=row.title, body_text=row.body_text, channels=list(row.channels or []), template_key=row.template_key, context=dict(row.context or {}), body_html=row.body_html, url=row.url, related_resource_type=row.related_resource_type, related_resource_id=row.related_resource_id, ) result = await service._deliver_with_durable_claim( row, env, trace_id=request.state.trace_id, actor_user_id=current_user.id, ) await db.commit() return success_response( data={ "status": row.status, "delivered_channels": result.delivered_channels, "feishu_error": result.feishu_error, "email_error": result.email_error, }, request=request, ) @router.post( "/outbox/{outbox_id}/resend", summary="Resend one outbox row as a fresh dispatch", ) async def resend_outbox_row( outbox_id: UUID, request: Request, current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Explicit "再次发送": clone the row into a brand-new outbox dispatch and deliver it now. Unlike ``/retry`` (which re-fires the SAME row and refuses terminal rows), resend works on ANY source status — ``delivered`` included — because it leaves the original delivery record untouched and inserts a fresh row. Departed / person-less recipients are skipped (the response ``status`` is ``skipped`` with a reason) rather than re-notifying an ex-employee. """ service = NotificationService(db) outcome = await service.resend_outbox_row( outbox_id, actor_user_id=current_user.id, trace_id=request.state.trace_id, ) await db.commit() return success_response(data=outcome, request=request) @router.post( "/outbox/batch-resend", summary="Resend up to 500 outbox rows as fresh dispatches", ) async def batch_resend_outbox( body: BatchIdsBody, request: Request, current_user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Bulk "批量再次发送". Each resolvable, non-departed row is cloned into a fresh dispatch and delivered inline; person-less / departed / unknown ids are skipped with a per-id reason so the admin toast stays honest. """ service = NotificationService(db) result = await service.resend_outbox_rows( list(body.ids), actor_user_id=current_user.id, trace_id=request.state.trace_id, ) await db.commit() return success_response(data=result, request=request) # ── Personal inbox (current user only) ── # These endpoints power the header bell + dropdown. They expose only the # delivered notifications addressed to the calling user — no PII from # other rows can leak. Pagination is bounded. @router.get("/inbox", summary="Current user's notification inbox") async def list_my_inbox( request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), q: str | None = Query( default=None, max_length=100, description="Case-insensitive substring over title / template_key", ), read: str | None = Query( default=None, description="'read' / 'unread'; anything else = all", ), template_prefix: str | None = Query( default=None, max_length=50, description=( "Raw template_key prefix filter — debugging only. Use `category`; " "the key namespace is mixed, so prefix chips miss whole domains." ), ), category: str | None = Query( default=None, max_length=40, description=( "Category filter from the notification taxonomy " "(requests_approvals / account_access / assets / cloud / " "lifecycle / security / social / system / other). Classifies via " "the same registry that labels each row, so it can never drift." ), ), bucket: str | None = Query( default=None, description="Inbox bucket filter: 'action' / 'info' / 'social'", ), ): from app.notifications.inbox_service import InboxService svc = InboxService(db) result = await svc.list_for_recipient( recipient_id=current_user.id, page=page, page_size=page_size, q=q, read=read, template_prefix=template_prefix, category=category, bucket=bucket, ) return paginated_response( data=result["items"], total=result["total"], page=result["page"], page_size=result["page_size"], request=request, ) @router.get("/inbox/summary", summary="Inbox unread count for current user") async def my_inbox_summary( request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), since: str | None = Query( # noqa: ARG001 default=None, description=( "Deprecated. Kept on the schema only so an in-flight client " "with the old localStorage-driven query string doesn't 422. " "Read state is now persisted server-side in " "``notification_outbox.read_at`` (migration 086) — pass " "nothing and the count reflects DB truth." ), deprecated=True, ), ): """Counts pending vs. unread for the bell-icon badge. Source of truth is ``notification_outbox.read_at`` (NULL = unread) so the badge survives cache wipes, device switches, and parallel sessions. The legacy ``?since=`` query param from the localStorage era is silently ignored — see migration 086 for the rationale. """ from app.notifications.inbox_service import InboxService svc = InboxService(db) data = await svc.summary_for_recipient(recipient_id=current_user.id) return success_response(data=data, request=request) @router.get("/inbox/{notification_id}", summary="One inbox notification by id") async def get_my_inbox_item( notification_id: UUID, request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """A single inbox row the calling user owns, or 404. O(1) replacement for the client-side "page the whole inbox to find one deep-linked id" workaround (#1804): the mobile / desktop notification-detail pages call this directly. Owner-scoped + visibility-filtered in the service, so an id belonging to another user (or a hidden/secret row) is indistinguishable from a missing one — 404 either way, no existence oracle. Declared AFTER ``/inbox/summary`` so the static path wins over this dynamic one. """ from app.core.exceptions import NotFoundError from app.notifications.inbox_service import InboxService svc = InboxService(db) item = await svc.get_for_recipient( recipient_id=current_user.id, notification_id=notification_id, ) if item is None: raise NotFoundError("NotificationOutbox", str(notification_id)) return success_response(data=item, request=request) class MarkReadBody(BaseModel): """Either ``ids`` (mark exactly these rows) or ``all`` (mark every unread row owned by the calling user). The two modes are mutually exclusive — sending both errors out so a buggy client can't silently mark the wrong batch. ``ids`` length capped at 500 — without a cap a logged-in user could POST a 1MB list and force the server to parse 100k UUIDs + emit a `WHERE id IN (...)` of 100k elements. Mirrors the cap already present in the ``mark_my_notifications_read`` MCP tool. """ ids: list[UUID] | None = Field(default=None, max_length=500) all: bool = False @router.post( "/inbox/mark-read", summary="Mark current user's notifications as read", ) async def mark_my_inbox_read( body: MarkReadBody, request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Persist read state to ``notification_outbox.read_at``. Two modes: * ``{"all": true}`` — mark every currently-unread row owned by the caller. Used by the bell-icon "全部已读" button and by the implicit "you opened the dropdown" mark. * ``{"ids": [...]}`` — mark exactly those rows. Reserved for future per-item interactions; today the dropdown only uses the bulk path. The query is ``WHERE recipient_person_id = me`` so a malicious client can't mark another user's rows even by guessing IDs. Already-read rows are skipped — ``read_at`` is set ONLY on rows where it's currently NULL, so the original read timestamp is preserved (useful for any future "when did I first see this?" audit query). """ if body.all and body.ids: from app.core.exceptions import AppException raise AppException( code="VALIDATION_ERROR", message="Pass either `all: true` or `ids`, not both.", status_code=422, ) if not body.all and not body.ids: from app.core.exceptions import AppException raise AppException( code="VALIDATION_ERROR", message="Either `all: true` or a non-empty `ids` is required.", status_code=422, ) from app.notifications.inbox_service import InboxService svc = InboxService(db) if body.all: marked = await svc.mark_all_read(recipient_id=current_user.id) else: marked = await svc.mark_read( recipient_id=current_user.id, ids=list(body.ids or []), ) await db.commit() return success_response(data={"marked": marked}, request=request) # ── Notification preferences (self-service) ───────────────────────── # The employee's control surface over the outbound firehose: per-category # × per-channel muting + a daily quiet-hours window. The in-app inbox is # never suppressed (muting only quiets PUSH); critical templates always # push. See app/notifications/preferences_service.py + categories.py. class PreferencesBody(BaseModel): """``muted_channels`` maps a category key to the push channels muted for it, e.g. ``{"social": ["feishu", "email"], "cloud": ["email"]}``. Quiet hours are local-time ``HH:MM`` bounds (resolved in the recipient's own timezone); a window where start > end wraps past midnight. The service sanitises everything — unknown categories / channels and unparseable times are dropped rather than 422'd, so a stale client can't wedge the save.""" muted_channels: dict[str, list[str]] = Field(default_factory=dict) quiet_hours_enabled: bool = False quiet_hours_start: str | None = Field(default=None, max_length=5) quiet_hours_end: str | None = Field(default=None, max_length=5) @router.get( "/preferences", summary="Get my notification preferences + category catalog", ) async def get_my_preferences( request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): from app.notifications.preferences_service import ( NotificationPreferenceService, ) svc = NotificationPreferenceService(db) row = await svc.get(current_user.id) return success_response(data=svc.serialize(row), request=request) @router.put( "/preferences", summary="Update my notification preferences", ) async def update_my_preferences( body: PreferencesBody, request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): from app.notifications.preferences_service import ( NotificationPreferenceService, ) svc = NotificationPreferenceService(db) row = await svc.upsert( person_id=current_user.id, muted_channels=body.muted_channels, quiet_hours_enabled=body.quiet_hours_enabled, quiet_hours_start=body.quiet_hours_start, quiet_hours_end=body.quiet_hours_end, ) await db.commit() return success_response(data=svc.serialize(row), request=request) @router.get( "/email-templates", summary="Email-template catalog for the 邮件模板 admin page", ) async def list_email_templates( request: Request, _admin: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Enumerate every admin-editable email template with its effective content (override ∪ in-code default), placeholder variables, domain group and flags. The dedicated editor lists/edits these; the generic platform-settings page hides the ``email_template.*`` namespace and links here so there is one home for email templates. """ from app.notifications.template_catalog import build_registry items = await build_registry(db) return success_response(data=items, request=request) class EmailPreviewBody(BaseModel): """WYSIWYG preview request — the editor sends exactly what the admin has typed so the rendered card matches what recipients will get.""" subject: str = Field(default="", max_length=512) body_text: str = Field(default="", max_length=20000) lang: str = Field(default="zh-CN", max_length=16) # Readable demo values for the common placeholders so the preview reads # like a real mail instead of leaking ``{person_name}`` tokens. Unknown # placeholders fall back to a humanised version of the variable name. _PREVIEW_SAMPLES: dict[str, str] = { "person_name": "张三", "person_email": "zhangsan@pingcap.com", "recipient_name": "李四", "manager_name": "王经理", "leaver_name": "张三", "leaver_email": "zhangsan@pingcap.com", "actor_name": "李四", "company_name": "PingCAP", "department": "IT", "job_title": "Engineer", "location": "Beijing", "days_left": "3", "leaver_date": "2026-07-03", "count": "2", "asset_label": "MacBook Pro 16", "asset_type": "Laptop", "software_name": "Figma", "provider_name": "OpenAI", "portal_url": "https://helix.pingcap.net/me", } # ``cta_line`` is injected by the send path in the recipient's language, so the # preview has to vary it too — a Chinese CTA under an English preview misled # admins into "fixing" copy that was already correct. _PREVIEW_CTA_BY_LANG: dict[str, str] = { "zh": "请到 Helix 查看: https://helix.pingcap.net/me", "en": "View in Helix: https://helix.pingcap.net/me", "ja": "Helix でご確認ください: https://helix.pingcap.net/me", } @router.post( "/email-templates/preview", summary="Render an email body in the branded HTML shell (preview)", ) async def preview_email_template( body: EmailPreviewBody, request: Request, _admin: AuthUser = Depends(require_admin_read()), ): """Render ``subject`` + ``body_text`` through the production HTML shell (:func:`app.notifications.email_layout.render_email_shell`) with demo placeholder values, so admins see the actual look before saving. """ import re from app.core.locale import DEFAULT_LANG, normalize_lang from app.notifications.email_layout import render_email_shell lang = normalize_lang(body.lang) or DEFAULT_LANG samples = dict(_PREVIEW_SAMPLES) samples["cta_line"] = _PREVIEW_CTA_BY_LANG[lang] def _fill(text: str) -> str: return re.sub( r"\{([a-zA-Z_][a-zA-Z0-9_]*)\}", lambda m: samples.get(m.group(1), m.group(1).replace("_", " ")), text or "", ) html = render_email_shell( subject=_fill(body.subject), body_text=_fill(body.body_text), cta_url="https://helix.pingcap.net/me", lang=body.lang, ) return success_response(data={"html": html}, request=request)