"""Personal-inbox queries shared between the HTTP router and MCP tools. The router used to inline the inbox SELECT / UPDATE statements directly, which meant the MCP tool surface had to either copy-paste them (drift risk) or reach into the router (circular). This service is the single read/write path: both callers go through here. Source of truth for read state is ``notification_outbox.read_at`` (migration 086). Visibility (2026-07-07): a row surfaces on the bell when it is ``delivered`` / ``partial`` (a push channel actually landed), OR it is ``dead`` **and actionable** (non-social). ``pending`` / ``failed`` rows are still mid-flight — the retry worker may yet deliver them on a push channel, so surfacing them early would mislead the user into thinking something arrived that hasn't. But a ``dead`` row is terminal: every push channel (Feishu + email) is exhausted or permanently failed and the retry worker will never touch it again. Since the outbox row *is* the inbox row, keeping a dead actionable notification off the bell was a **permanent silent loss** of a "please act on this" item (approval, seat return, expiry) — worst for a brand-new hire whose Feishu open_id and work_email don't exist yet, so a dual-channel action item dead-letters on attempt 1. The in-app bell is the recipient's always-available channel of last resort, so dead actionable rows belong on it. Dead *social* rows (Moments blessings/likes/…) stay off — a failed birthday ping is not worth a red badge. Historical dead rows are stamped ``read_at`` at migration time (migration 389) so flipping this policy does not resurrect a company-wide backlog of unread badges; only rows that die *after* the change raise the bell. Every method takes ``recipient_id`` explicitly (never derives it from contextvars) so the logic is testable in isolation and the auth check stays at the boundary (router / MCP tool runtime), not buried in a service helper. """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from uuid import UUID from sqlalchemy import and_, func, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from app.notifications.categories import ( BUCKET_ACTION, BUCKET_INFO, BUCKET_SOCIAL, CAT_OTHER, bucket_for, category_key_for, registry_in_order, ) from app.notifications.models import ( STATUS_DEAD, STATUS_DELIVERED, STATUS_DELIVERY_UNCERTAIN, STATUS_PARTIAL, NotificationOutbox, ) from app.notifications.service import SECRET_TEMPLATE_KEYS # Statuses whose delivery landed (a push channel actually reached the # recipient). ``pending`` / ``failed`` rows are still mid-flight — the # retry worker may yet deliver them — so surfacing them early would # mislead the user. ``dead`` rows are handled separately below: they are # terminal (no push channel will ever land), so an *actionable* dead row # falls back onto the bell rather than being lost. See ``_inbox_visible_clause``. _INBOX_LANDED_STATUSES = ( STATUS_DELIVERED, STATUS_PARTIAL, # The external provider result is unknown, but the outbox row itself is # a durably landed in-app notification. Keep it visible so fencing a # possible duplicate never turns into a silent user-facing loss. STATUS_DELIVERY_UNCERTAIN, ) # Secret-body rows (Google first-login password, recipient = the new hire) # are admin-fallback artifacts, not hire-facing notifications: hide them # from EVERY personal-inbox query — the list, the unread badge count, and # the family summary — so a plaintext credential is never surfaced and no # ghost-unread badge is left for an invisible row. _INBOX_NOT_SECRET = or_( NotificationOutbox.template_key.is_(None), NotificationOutbox.template_key.notin_(SECRET_TEMPLATE_KEYS), ) # Push-only rows (``hidden_from_inbox=1``, migration 375) are delivery # records for a follow-up push whose content the recipient already has # as another inbox row (e.g. the Day-1 Feishu welcome card trailing the # Day-1 welcome email). Surfacing them would read as a duplicate-send # bug, so every personal-inbox query excludes them; admin outbox # surfaces deliberately keep them visible. _INBOX_NOT_HIDDEN = NotificationOutbox.hidden_from_inbox.is_(False) # "Social" notifications — the Moments / social-feed family. These are # friendly to keep in the inbox but they are NOT actionable: nothing breaks # if the employee never opens them. The bell-icon red badge should count # only *actionable* unread (approvals, asset returns, account/permission # expiry, …) so social engagement never manufactures false urgency. # # The family, all confirmed to write an inbox (notification_outbox) row: # * ``feed.celebration.`` — work-anniversary self-greetings + lead pings # * ``feed.blessing.`` — birthday / anniversary blessings from colleagues # * ``feed.comment.`` — someone commented on your Moments post # * ``feed.like.`` — aggregated "+N likes" hint # * ``feed.mention.`` — someone @-mentioned you in a post / comment # # Deliberately NOT social (stay actionable / visible): ``feed.announcements.*`` # (company announcements employees should see) and ``feed.attachment.*`` / # other system-config notices. Classification is by ``template_key`` prefix — # the same family-segment convention the inbox ``template_prefix`` filter # uses. Rows with a NULL ``template_key`` are treated as actionable # (fail-towards-visible: an un-classified real notification must never # silently drop off the badge). _SOCIAL_TEMPLATE_PREFIXES = ( "feed.celebration.", "feed.blessing.", "feed.comment.", "feed.like.", "feed.mention.", ) def _is_social_template_clause(): """SQLAlchemy boolean: True when ``template_key`` is a social prefix. The ``template_key IS NOT NULL`` guard is load-bearing: without it a NULL ``template_key`` makes the ``LIKE`` chain evaluate to SQL NULL, and ``~NULL`` is also NULL (filtered out by WHERE) — so the "actionable = NOT social" query would silently *drop* every un-classified row instead of counting it. Anchoring NULL to a definite ``False`` here means the negation treats NULL ``template_key`` as actionable (fail-towards- visible). One source of truth for both the social and actionable counts. """ return and_( NotificationOutbox.template_key.is_not(None), or_( *( NotificationOutbox.template_key.like(f"{prefix}%") for prefix in _SOCIAL_TEMPLATE_PREFIXES ) ), ) def _inbox_visible_clause(): """SQLAlchemy boolean: True when a row belongs on the recipient's bell. Two ways in: * ``status IN (delivered, partial)`` — a push channel actually landed the message. * ``status == dead`` **and not social** — delivery is terminally exhausted on every push channel, so the in-app bell is the channel of last resort for the *actionable* item that would otherwise be silently lost. ``~_is_social_template_clause()`` treats a NULL ``template_key`` as actionable (fail-towards- visible, same convention as the actionable-badge count) and keeps dead Moments-feed spam (blessings/likes/…) off the bell. Used by the list, summary (total / unread / actionable / per-bucket), and both mark-read writes so a dead actionable row that surfaces can also be acknowledged. Orthogonal to the secret / hidden-from-inbox filters, which stay in force regardless of status. """ return or_( NotificationOutbox.status.in_(_INBOX_LANDED_STATUSES), and_( NotificationOutbox.status == STATUS_DEAD, ~_is_social_template_clause(), ), ) def _category_clauses() -> dict[str, Any]: """``category_key`` → SQLAlchemy boolean for "this row classifies here". Built from the same ordered registry the Python ``classify`` uses: a row belongs to category *i* iff it matches category *i*'s patterns AND none of the earlier categories' patterns (first-match-wins). The ``other`` entry covers a NULL ``template_key`` or a key no category claims, matching ``classify``'s default — so every row is reachable through exactly one category and none can hide from the filter. Keeping this in lock-step with ``categories.classify`` is what lets the server-side list filter agree with the per-row ``category`` / ``bucket`` labels the client renders. """ prior: list[Any] = [] clauses: dict[str, Any] = {} for cat in registry_in_order(): likes = [ NotificationOutbox.template_key.like(f"{p}%") for p in cat.prefixes ] if cat.exact: likes.append(NotificationOutbox.template_key.in_(cat.exact)) mi = and_( NotificationOutbox.template_key.is_not(None), or_(*likes) if likes else False, ) clauses[cat.key] = and_(mi, ~or_(*prior)) if prior else mi prior.append(mi) clauses[CAT_OTHER] = or_( NotificationOutbox.template_key.is_(None), ~or_(*prior) if prior else True, ) return clauses def _category_clause(category: str): """Boolean for one category key, or ``False`` for an unknown key. An unknown key matching nothing (rather than everything) means a stale or hand-typed ``?category=`` returns an empty list instead of silently ignoring the filter and implying the inbox holds nothing else. """ return _category_clauses().get(category, False) def _bucket_clause(bucket: str): """Precedence-correct boolean for "this row's classified bucket == ``bucket``". Folds the ``other`` fallback into the action bucket, matching ``classify``'s default.""" clauses = _category_clauses() terms = [ clauses[cat.key] for cat in registry_in_order() if cat.bucket == bucket and cat.key in clauses ] if bucket == BUCKET_ACTION: terms.append(clauses[CAT_OTHER]) return or_(*terms) if terms else False def _serialize_inbox_row(row: NotificationOutbox) -> dict[str, Any]: """Wire-shape for one inbox row. Mirrors the legacy ``_serialize_inbox`` helper that previously lived in router.py — kept identical so the HTTP response shape doesn't move under existing clients, plus the ``category`` / ``bucket`` classification used by the rebuilt inbox UI. """ return { "id": str(row.id), "title": row.title, "body_text": row.body_text or "", "url": row.url, "template_key": row.template_key, "category": category_key_for(row.template_key), "bucket": bucket_for(row.template_key), "status": row.status, "related_resource_type": row.related_resource_type, "related_resource_id": ( str(row.related_resource_id) if row.related_resource_id else None ), "created_at": row.created_at.isoformat(), "delivered_at": ( row.delivered_at.isoformat() if row.delivered_at else None ), "read_at": row.read_at.isoformat() if row.read_at else None, # 回执(迁移 624)。铃铛和飞书卡片读同一张表,所以「那件事已经结了」必须 # 在这里出得来 —— 否则卡片变成绿色的「已批准」,而铃铛里那条还写着 # 「去审批」,两个面对着同一件事说不同的话。前端据此渲染 ✅ 并撤掉行动 # 按钮:一颗点下去会被拒绝的按钮读起来就是平台的 bug。 "settled_at": row.settled_at.isoformat() if row.settled_at else None, "settled_outcome": row.settled_outcome, } class InboxService: def __init__(self, db: AsyncSession) -> None: self.db = db async def list_for_recipient( self, *, recipient_id: UUID, page: int = 1, page_size: int = 20, q: str | None = None, read: str | None = None, template_prefix: str | None = None, category: str | None = None, bucket: str | None = None, ) -> dict[str, Any]: """Return a page of inbox rows for ``recipient_id``. Result shape: ``{ "items": [...], "total": N, "page": P, "page_size": S }`` Pagination is 1-indexed because that's what the HTTP routes expose; clamp at the call site if you need different bounds. Filters (2026-06-11 — the notification-center page previously filtered CLIENT-side over the current 20-row page only, so the keyword box silently missed every other page): * ``q`` — case-insensitive substring over ``title`` + ``template_key``. NOT ``body_text`` — outbox rows persist the ``(see template)`` sentinel there; the real copy is rendered per-read and never written back, so a body LIKE would match nothing useful. * ``read`` — ``"read"`` / ``"unread"`` (anything else = all). * ``category`` — one of the taxonomy keys from ``notifications.categories`` (``assets`` / ``cloud`` / ``social`` / …). This is what the FE chips use: it classifies through the SAME ordered registry that produces each row's ``category`` label, so a chip can never disagree with the labels in the list it filters. * ``template_prefix`` — raw ``template_key LIKE '%'``. Retained for API compatibility and ad-hoc debugging ONLY. It is NOT a usable end-user filter: the key namespace is mixed (flat ``password_reset`` alongside dotted ``cloud.billing_alert``), so a "family = first key segment" chip set silently returns zero rows for whole domains — the mail-group family really lives under ``mail_group_*`` / ``alias_*`` / ``dynamic_email_group_*``, none of which start with ``email``. Use ``category``. """ # Defensive clamps so a misbehaving caller can't ask for # page=0 (offset=-page_size) or a 100k page_size. page = max(1, page) page_size = max(1, min(page_size, 100)) stmt = ( select(NotificationOutbox) .where(NotificationOutbox.recipient_person_id == recipient_id) .where(_inbox_visible_clause()) .where(_INBOX_NOT_SECRET) .where(_INBOX_NOT_HIDDEN) ) q_clean = (q or "").strip()[:100] if q_clean: # Escape LIKE metacharacters so ``%`` / ``_`` match literally # (same discipline as provisioning list_tasks keyword). escaped = ( q_clean .replace("\\", "\\\\") .replace("%", "\\%") .replace("_", "\\_") ) pattern = f"%{escaped}%" stmt = stmt.where( or_( NotificationOutbox.title.ilike(pattern, escape="\\"), NotificationOutbox.template_key.ilike(pattern, escape="\\"), ) ) if read == "read": stmt = stmt.where(NotificationOutbox.read_at.is_not(None)) elif read == "unread": stmt = stmt.where(NotificationOutbox.read_at.is_(None)) prefix_clean = (template_prefix or "").strip()[:50] if prefix_clean: escaped_prefix = ( prefix_clean .replace("\\", "\\\\") .replace("%", "\\%") .replace("_", "\\_") ) stmt = stmt.where( NotificationOutbox.template_key.like( f"{escaped_prefix}%", escape="\\", ) ) category_clean = (category or "").strip() if category_clean: stmt = stmt.where(_category_clause(category_clean)) if bucket in (BUCKET_ACTION, BUCKET_INFO, BUCKET_SOCIAL): stmt = stmt.where(_bucket_clause(bucket)) count_stmt = select(func.count()).select_from(stmt.subquery()) total = (await self.db.execute(count_stmt)).scalar() or 0 rows = list(( await self.db.execute( stmt .order_by(NotificationOutbox.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) ) ).scalars().all()) return { "items": await self._serialize_rows(rows, recipient_id=recipient_id), "total": int(total), "page": page, "page_size": page_size, } async def _serialize_rows( self, rows: list[NotificationOutbox], *, recipient_id: UUID, ) -> list[dict[str, Any]]: """Serialize inbox rows, rendering the ``body_text`` for display. Outbox rows persist ``body_text`` as the ``(see template)`` sentinel (the real copy is rendered per-channel at send time and never written back), so the raw column would surface the literal sentinel in the bell. Render it here via the shared ``NotificationService`` template path, in the recipient's language, with a per-page template cache so N rows don't issue N look-ups. Fail-safe: rendering returns the stored ``body_text`` on any miss, so the worst case is the pre-fix behaviour. """ if not rows: return [] # Lazy imports: NotificationService lives in the same package and # importing it at module scope risks an import cycle (it pulls in # driver / integration modules); IdentityPerson is only needed # for the recipient's language. from app.core.locale import resolve_lang from app.identity.person_models import IdentityPerson from app.notifications.service import NotificationService # Load the person rather than just ``preferred_language``: the language # resolver also consults ``country`` / ``is_china_employee`` when no # explicit preference is stored (``app.core.locale.resolve_lang``). recipient = await self.db.scalar( select(IdentityPerson).where(IdentityPerson.id == recipient_id) ) lang = resolve_lang(recipient) svc = NotificationService(self.db) cache: dict[str, dict[str, str] | None] = {} items: list[dict[str, Any]] = [] for row in rows: data = _serialize_inbox_row(row) # BOTH halves of the card get the recipient's language. Rendering # only the body (the pre-fix behaviour) produced a bell entry whose # title was in the *producer's* hardcoded language and whose body # was in the reader's — the single biggest source of "通知中心中英 # 混排" reports. data["title"] = await svc.render_inbox_title( row, lang=lang, template_cache=cache, ) data["body_text"] = await svc.render_inbox_body( row, lang=lang, template_cache=cache, ) items.append(data) return items async def get_for_recipient( self, *, recipient_id: UUID, notification_id: UUID, ) -> dict[str, Any] | None: """One inbox row by id, owner-scoped, or None if not found / not visible to this recipient. Applies the SAME visibility clauses as ``list_for_recipient`` (recipient match + landed/visible + not-secret + not-hidden) so a row the list would never show can't be fetched by guessing its id. Renders ``body_text`` via the shared per-row path. Backs ``GET /inbox/{id}`` — the O(1) replacement for clients that used to page the whole inbox to resolve a single deep-linked id. """ row = ( await self.db.execute( select(NotificationOutbox) .where(NotificationOutbox.id == notification_id) .where(NotificationOutbox.recipient_person_id == recipient_id) .where(_inbox_visible_clause()) .where(_INBOX_NOT_SECRET) .where(_INBOX_NOT_HIDDEN) ) ).scalar_one_or_none() if row is None: return None items = await self._serialize_rows([row], recipient_id=recipient_id) return items[0] if items else None async def summary_for_recipient( self, *, recipient_id: UUID, ) -> dict[str, Any]: """Counts: total + unread + last_created_at. Source of truth for unread is ``read_at IS NULL``. Same query the HTTP /inbox/summary endpoint exposes. """ base = ( select(NotificationOutbox) .where(NotificationOutbox.recipient_person_id == recipient_id) .where(_inbox_visible_clause()) .where(_INBOX_NOT_SECRET) .where(_INBOX_NOT_HIDDEN) ) total = ( await self.db.execute( select(func.count()).select_from(base.subquery()), ) ).scalar() or 0 unread = ( await self.db.execute( select(func.count()).select_from( base.where(NotificationOutbox.read_at.is_(None)).subquery(), ) ) ).scalar() or 0 # Actionable unread = unread MINUS social/ceremonial. This is what # the red badge should reflect; ``unread`` stays the honest total so # the inbox list header / "mark all read" affordance still see every # row. ``~`` negates the social clause so a NULL template_key (= # un-classified) counts as actionable. unread_actionable = ( await self.db.execute( select(func.count()).select_from( base .where(NotificationOutbox.read_at.is_(None)) .where(~_is_social_template_clause()) # 已结案的不算待办。红点的含义是「有事等着你做」,而一件 # 已经批准/驳回/关单的事没有在等任何人 —— 员工从没点开过 # 那条通知(所以 read_at 仍是 NULL)不代表那件事还开着。 # 迁移 624。 .where(NotificationOutbox.settled_at.is_(None)) .subquery(), ) ) ).scalar() or 0 last_created = ( await self.db.execute( select(func.max(NotificationOutbox.created_at)) .where(NotificationOutbox.recipient_person_id == recipient_id) .where(_inbox_visible_clause()) .where(_INBOX_NOT_SECRET) .where(_INBOX_NOT_HIDDEN), ) ).scalar() # Per-bucket unread breakdown for the bell's grouped view. Classified # in Python (precedence-correct, same as the row labels) over just the # unread template_keys — bounded by the unread count, so cheap. # # 已结案的行同样排除 —— 必须和上面 ``unread_actionable`` 用**同一个** # 判据。两处不一致的后果是用户可见的:红点说 0,而铃铛里分组视图的 # 「待处理」仍写着 1,同一个面板自己跟自己矛盾。 unread_keys = ( await self.db.execute( select(NotificationOutbox.template_key) .where(NotificationOutbox.recipient_person_id == recipient_id) .where(_inbox_visible_clause()) .where(_INBOX_NOT_SECRET) .where(_INBOX_NOT_HIDDEN) .where(NotificationOutbox.read_at.is_(None)) .where(NotificationOutbox.settled_at.is_(None)) ) ).scalars().all() unread_by_bucket = {BUCKET_ACTION: 0, BUCKET_INFO: 0, BUCKET_SOCIAL: 0} for key in unread_keys: unread_by_bucket[bucket_for(key)] += 1 return { "total": int(total), "unread": int(unread), "unread_actionable": int(unread_actionable), "unread_by_bucket": unread_by_bucket, "last_created_at": ( last_created.isoformat() if last_created else None ), } async def mark_read( self, *, recipient_id: UUID, ids: list[UUID], ) -> int: """Flip ``read_at`` from NULL → now() for the given ``ids``. ``ids`` is required — empty list is a no-op (returns 0). The query joins ``WHERE recipient_person_id = recipient_id`` so a client passing IDs they don't own marks zero rows. Caller is expected to ``commit`` the session — same pattern as every other Helix service. We deliberately do NOT commit here because some callers (the MCP write surface) compose multiple write operations into one transaction. """ if not ids: return 0 now = datetime.now(UTC) stmt = ( update(NotificationOutbox) .where(NotificationOutbox.recipient_person_id == recipient_id) .where(NotificationOutbox.read_at.is_(None)) .where(_inbox_visible_clause()) .where(NotificationOutbox.id.in_(ids)) .values(read_at=now) ) result = await self.db.execute(stmt) return int(result.rowcount or 0) async def mark_all_read(self, *, recipient_id: UUID) -> int: """Flip every still-unread row owned by ``recipient_id`` to read.""" now = datetime.now(UTC) stmt = ( update(NotificationOutbox) .where(NotificationOutbox.recipient_person_id == recipient_id) .where(NotificationOutbox.read_at.is_(None)) .where(_inbox_visible_clause()) .values(read_at=now) ) result = await self.db.execute(stmt) return int(result.rowcount or 0) __all__ = ["InboxService"]