"""Mark stale outbox rows as ``dead`` after a notification-pipeline fix. Context (prod 2026-05-02): two regressions sat in ``notification_outbox`` for hours before the multi-bug fix PR landed: 1. **Welcome welcome welcome** — every hourly sf-eu-prod incremental sync was re-classifying CN-instance employees (P01201 udi, P01436 kevin) as ``action='create'`` because the ``_classify`` lookup was instance-scoped. Each tick produced a fresh ``welcome`` outbox row whose SMTP send failed (``email integration not configured``) and that retried 5x before going dead. The fix tightened ``_classify`` to skip cross-instance ownership, but rows already enqueued continue to retry until ``max_attempts`` is hit. This script flips them to ``dead`` so the admin Outbox stops blinking and the retry worker stops burning SMTP attempts on a known-broken target. 2. **leaver_blocked_sla_* — empty-card body** — the SLA escalation templates (``leaver_blocked_sla_warn``/``urgent``/``breach``) were missing from ``_DEFAULT_TEMPLATES``, so the Feishu card rendered ``"(no template configured)"`` instead of the SLA service's real body. Templates are now in place, but rows enqueued before the deploy carry the bad payload baked into ``title``/``body_text`` columns — re-delivering them would just re-flood the recipients with the wrong card. We dead-letter pre-cutoff rows so the next SLA tick re-fires with the new templates and the new HRBP-targeted recipient list. Scope (defensive — every WHERE clause is conjunctive so the script can NEVER touch unrelated rows): template_key IN (:keys) AND status IN ('pending', 'failed', 'partial') AND created_at < :cutoff AND ( -- For welcome: any attempt count, since the misfire produces -- ``attempts=1, status=failed`` rows that loop on the SMTP -- error. We need to clear them all. template_key = 'welcome' OR -- For SLA templates: only rows with attempts>=1 (touched at -- least once after enqueue) so a row written and not yet -- attempted by the retry worker is left alone — that one -- will deliver fresh content on its first attempt. attempts >= 1 ) Defaults to ``--dry-run``. Run with ``--commit`` to persist. """ from __future__ import annotations import argparse import asyncio from datetime import UTC, datetime from sqlalchemy import or_, select, update import app.audit.models # noqa: F401 — register tables for metadata import app.auth.models # noqa: F401 import app.domains.hr.models # noqa: F401 import app.identity.person_models # noqa: F401 import app.notifications.models # noqa: F401 from app.core.database import async_session_factory from app.notifications.models import ( STATUS_DEAD, STATUS_FAILED, STATUS_PARTIAL, STATUS_PENDING, NotificationOutbox, ) # Template keys this script will touch. ``welcome`` is the EU-sync # misfire; the three SLA keys carry the empty-card body for any row # enqueued before the deploy. Anything else is out of scope and the # script refuses to touch it. _TARGET_KEYS: tuple[str, ...] = ( "welcome", "leaver_blocked_sla_warn", "leaver_blocked_sla_urgent", "leaver_blocked_sla_breach", ) async def run(*, dry_run: bool, cutoff: datetime, keys: tuple[str, ...]) -> None: """Execute the cleanup. ``keys`` is constrained to a subset of :data:`_TARGET_KEYS` — the CLI uses the full set by default but ops can narrow to a single key if needed (``--keys welcome``). """ if not keys: raise SystemExit("--keys must include at least one template_key") if any(k not in _TARGET_KEYS for k in keys): raise SystemExit( f"--keys may only include {_TARGET_KEYS!r}; got {keys!r}" ) async with async_session_factory() as session: # Welcome path: any attempts; SLA path: only already-attempted # rows. ``or_`` keeps the predicate one-shot so the row scan is # a single index-aided pass. attempts_clause = or_( NotificationOutbox.template_key == "welcome", NotificationOutbox.attempts >= 1, ) rows = (await session.execute( select(NotificationOutbox) .where(NotificationOutbox.template_key.in_(keys)) .where(NotificationOutbox.status.in_( [STATUS_PENDING, STATUS_FAILED, STATUS_PARTIAL], )) .where(NotificationOutbox.created_at < cutoff) .where(attempts_clause) .order_by(NotificationOutbox.created_at.asc()) )).scalars().all() print( f"Found {len(rows)} stale outbox row(s) " f"(keys={list(keys)!r}, cutoff={cutoff.isoformat()})" ) for row in rows: print( f" {row.created_at.isoformat()} " f"{(row.recipient_email or str(row.recipient_person_id)):40} " f"key={row.template_key:32} " f"status={row.status:8} " f"attempts={row.attempts}/{row.max_attempts}" ) if not rows: return if dry_run: print("\n[DRY-RUN] No changes persisted. Re-run with --commit.") return await session.execute( update(NotificationOutbox) .where(NotificationOutbox.id.in_([r.id for r in rows])) .values( status=STATUS_DEAD, next_attempt_at=None, last_error=( "[manual] dead-lettered by " "dead_letter_outbox_post_deploy: stale rows from " "pre-fix HR cross-instance / SLA empty-card " "regression (fixed 2026-05-02)" ), ) ) await session.commit() print(f"\n[COMMIT] Dead-lettered {len(rows)} row(s).") def _parse_keys(raw: str) -> tuple[str, ...]: parts = tuple(p.strip() for p in raw.split(",") if p.strip()) if not parts: raise argparse.ArgumentTypeError("at least one key required") return parts def _main() -> None: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--commit", action="store_true", help="persist changes") ap.add_argument( "--cutoff", default=datetime.now(UTC).isoformat(), help=( "ISO-8601 cutoff timestamp; only rows with created_at < cutoff " "are touched. Default: now (catches everything pre-run)." ), ) ap.add_argument( "--keys", type=_parse_keys, default=_TARGET_KEYS, help=( f"Comma-separated subset of {list(_TARGET_KEYS)!r}. " f"Default: all four." ), ) args = ap.parse_args() cutoff = datetime.fromisoformat(args.cutoff) if cutoff.tzinfo is None: cutoff = cutoff.replace(tzinfo=UTC) asyncio.run(run(dry_run=not args.commit, cutoff=cutoff, keys=args.keys)) if __name__ == "__main__": _main()