"""Celebration scanner — the engine behind the daily 动态广场 banners. Responsibilities: * Walk every active employee. Compute their "today" in their own ``resolve_timezone(person)`` local TZ. * If today's MM-DD matches ``hire_date`` (and N >= 1) → insert a ``feed_celebrations`` row of type ``anniversary``. * If today's MM-DD matches ``birthday`` (and the person's ``feed.birthday_visibility`` is NOT ``hidden``) → insert a ``feed_celebrations`` row of type ``birthday``. * For each newly-inserted celebration, create the matching ``feed_posts`` row (``system_birthday`` / ``system_anniversary``). * Refresh the day's aggregate banner posts (``aggregate_birthday`` / ``aggregate_anniversary``) by deleting the old banner and inserting a fresh one whose body mentions the current celebrant count. This keeps the banner numbers honest across re-runs. UNIQUE(person_id, occasion_date, celebration_type) on ``feed_celebrations`` makes the whole scan idempotent — re-running on the same day inserts zero new rows. Leap-day handling: a person born / hired on Feb 29 is celebrated on Feb 28 in non-leap years. (Match SF's pattern + WeChat moments.) Language -------- The 2026-06-01 "store the rendered English prose in ``feed_posts.body``" decision is **reverted**. A Chinese reader saw ``1 teammate celebrating their work anniversary today!`` sitting in an otherwise-Chinese feed, and the front-end's own three-language copy was dead code because the server always supplied a non-empty ``body``. System posts now carry ``body_i18n_key`` + ``body_params`` (migration 456) and the READER's language decides what they see. ``body`` is still written — it is the fallback for clients that predate the columns and the haystack the feed search matches on — but nothing user-facing should prefer it over the key when the key is present. Aggregate banners additionally carry their own ``occasion_date`` in ``body_params``. Do NOT re-derive it from ``created_at``: the banner is created by the cron on the first tick of an Asia/Shanghai day, and SHA 00:00–07:59 is the PREVIOUS UTC day, so ``created_at.date()`` is off-by-one for eight hours out of every twenty-four. """ from __future__ import annotations import hashlib import uuid from datetime import UTC, date, datetime from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import is_transaction_aborting_db_error from app.core.safe_logging import get_safe_logger from app.core.tenure import completed_years, is_anniversary from app.core.working_hours import resolve_timezone from app.domains.feed.models import ( BIRTHDAY_VISIBILITY_HIDDEN, CELEBRATION_OPT_OUT_ALL, CELEBRATION_OPT_OUT_ANNIVERSARY, CELEBRATION_OPT_OUT_BIRTHDAY, CELEBRATION_OPT_OUT_NONE, CELEBRATION_OPT_OUT_VALUES, CELEBRATION_TYPE_ANNIVERSARY, CELEBRATION_TYPE_BIRTHDAY, POST_TYPE_AGGREGATE_ANNIVERSARY, POST_TYPE_AGGREGATE_BIRTHDAY, POST_TYPE_SYSTEM_ANNIVERSARY, POST_TYPE_SYSTEM_BIRTHDAY, PREF_BIRTHDAY_VISIBILITY, PREF_CELEBRATION_OPT_OUT, FeedCelebration, FeedPost, IdentityPersonPreference, ) from app.identity.person_models import IdentityPerson logger = get_safe_logger(__name__) # Illustration / sticker rotation pools. Front-end maps these keys to # inline SVG / emoji combinations. Keeping the pool here means the # user's same-day artwork is stable across reloads + retries. BIRTHDAY_ILLUSTRATION_KEYS = ( "cake_balloons", "gift_party", "confetti_band", "candle_star", "donut_disco", "popper_burst", "rainbow_treat", "ribbon_shower", ) ANNIVERSARY_ILLUSTRATION_KEYS = ( "trophy_jump", "medal_sky", "rocket_launch", "stars_shine", "ribbon_gold", "team_high_five", "milestone_flag", "growth_arrow", ) # ``feed_posts.body_i18n_key`` values for system posts. These are the keys # the front-end bundles already ship (``web/src/i18n/*.ts``); do not invent # parallel ones — a key with no bundle entry renders as raw text. BODY_KEY_SINGLE_BIRTHDAY = "feed.single.birthday_sentence" BODY_KEY_SINGLE_ANNIVERSARY = "feed.single.anniversary_sentence" BODY_KEY_AGGREGATE_BIRTHDAY = "feed.aggregate.birthday_count" BODY_KEY_AGGREGATE_ANNIVERSARY = "feed.aggregate.anniversary_count" def _pick_illustration( person_id: str, occasion: date, celebration_type: str, ) -> str: pool = ( BIRTHDAY_ILLUSTRATION_KEYS if celebration_type == CELEBRATION_TYPE_BIRTHDAY else ANNIVERSARY_ILLUSTRATION_KEYS ) seed = f"{person_id}|{occasion.isoformat()}|{celebration_type}".encode() # Deterministic presentation bucket, not a privacy/security boundary. # Keep it stable so existing celebration illustrations do not reshuffle. digest = hashlib.sha1( seed, usedforsecurity=False, ).hexdigest() idx = int(digest, 16) % len(pool) return pool[idx] def _matches_md(today: date, anchor: date) -> bool: """True iff ``today`` is the day ``anchor``'s anniversary is observed. Feb-29 anchor: observed on Feb 28 in common years, on the real 29th in leap years (so it fires exactly once either way). """ return is_anniversary(anchor, today) def _years_since(today: date, anchor: date) -> int: """How many years ``anchor`` completes on ``today``. Shares ``observed_anniversary`` with ``_matches_md`` on purpose. When the two used independent rules they disagreed for Feb-29 anchors: the scanner fired on Feb 28 of a common year while this returned one year fewer, so the ``years >= 1`` gate below swallowed the person's FIRST work anniversary entirely and shipped every later one a year short. """ return completed_years(anchor, today) async def _person_birthday_visible( db: AsyncSession, person_id: str, ) -> bool: row = (await db.execute( select(IdentityPersonPreference.pref_value).where( IdentityPersonPreference.person_id == person_id, IdentityPersonPreference.pref_key == PREF_BIRTHDAY_VISIBILITY, ) )).scalar_one_or_none() if row is None: return True # default is month_day visible return row != BIRTHDAY_VISIBILITY_HIDDEN async def _celebration_opt_out(db: AsyncSession, person_id: str) -> str: """This person's ``feed.celebration_opt_out`` choice. Anything unrecognised (a hand-edited pref row, a value from a future front-end) degrades to ``none`` — the historical behaviour — rather than silently suppressing someone's celebration. """ row = (await db.execute( select(IdentityPersonPreference.pref_value).where( IdentityPersonPreference.person_id == person_id, IdentityPersonPreference.pref_key == PREF_CELEBRATION_OPT_OUT, ) )).scalar_one_or_none() if isinstance(row, str) and row in CELEBRATION_OPT_OUT_VALUES: return row return CELEBRATION_OPT_OUT_NONE def _opted_out_of(opt_out: str, celebration_type: str) -> bool: if opt_out == CELEBRATION_OPT_OUT_ALL: return True if celebration_type == CELEBRATION_TYPE_BIRTHDAY: return opt_out == CELEBRATION_OPT_OUT_BIRTHDAY return opt_out == CELEBRATION_OPT_OUT_ANNIVERSARY def _display_name(person: IdentityPerson) -> str: """Language-neutral display name for a celebration. The name is a ``body_params`` VALUE interpolated into whichever language the reader has selected, so the fallback must not be a word in any one language — the old ``or "员工"`` spliced Chinese into ``Today is @员工's birthday``. Degrade to the work-email local part, then to empty. """ name = (person.display_name or "").strip() if name: return name email = (person.work_email or "").strip() return email.split("@", 1)[0] if email else "" def _format_anniv_body(name: str, years: int) -> str: # FALLBACK prose only — the reader-facing text comes from # ``body_i18n_key`` + ``body_params`` (see the module docstring). Kept # in English because it is also the search haystack and what # pre-migration-456 clients render. return ( f"Today is @{name}'s {years}-year work anniversary — come wish them well!" ) def _format_birthday_body(name: str) -> str: return f"Today is @{name}'s birthday — come wish them well!" async def scan_today_celebrations(db: AsyncSession) -> dict[str, Any]: """Walk active employees, upsert today's celebration rows + posts. Returns a summary dict for the scheduled-job ``last_result_summary``. """ rows = (await db.execute( select(IdentityPerson).where( IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), ) )).scalars().all() new_birthdays: list[FeedCelebration] = [] new_anniversaries: list[FeedCelebration] = [] skipped_hidden = 0 skipped_opt_out = 0 skipped_no_name = 0 for person in rows: tz = resolve_timezone(person) today = datetime.now(tz).date() person_id_s = str(person.id) display_name = _display_name(person) dept = person.primary_department if not display_name: # No display_name AND no work_email — a broken identity record. # Publishing anyway produced 「今天是 @ 的生日」 with a NULL name # snapshot: a post nobody can address a blessing to, on the # company feed, naming nobody. Counted in the job result so the # broken record is visible and can be fixed upstream, which is the # only place it CAN be fixed. # # Counted PER OCCASION, matching every sibling counter — a person # with both a birthday and an anniversary today used to add 1 here # while the other counters would have added 2. Opt-out and # birthday-visibility are deliberately NOT consulted: those # branches decide whether a celebration the platform CAN render # should be shown, and this one cannot be rendered at all, so # attributing it to a preference the person expressed would send # whoever reads this counter to the wrong place. occasions = 0 if ( person.hire_date is not None and _matches_md(today, person.hire_date) and _years_since(today, person.hire_date) >= 1 ): occasions += 1 if person.birthday is not None and _matches_md(today, person.birthday): occasions += 1 # Counted, NOT logged. ``get_safe_logger`` sanitises ``extra`` — # but ``SafeLogger._emit`` also discards the message and every arg, # emitting only opaque ``*_token_v1`` values, and the worker # formatter renders ``%(message)s`` alone and never ``extra``. Two # different people therefore produced byte-identical lines with no # person_id, no count, and nothing saying "celebration" — an hourly # cron writing a line that cannot identify anything is pure noise. # ``skipped_no_name`` in the returned summary is what an operator # actually reads, and it survives. skipped_no_name += occasions continue # Loaded lazily and at most once per person: the vast majority of # employees match nothing today, and a pref SELECT per employee per # hourly tick is 500 pointless round-trips. opt_out: str | None = None # Anniversary if person.hire_date is not None and _matches_md(today, person.hire_date): years = _years_since(today, person.hire_date) if years >= 1: if opt_out is None: opt_out = await _celebration_opt_out(db, person_id_s) if _opted_out_of(opt_out, CELEBRATION_TYPE_ANNIVERSARY): # The work anniversary used to be unconditional — and it # additionally pinged the manager + HRBP, so a person who # didn't want the attention had no lever at all. skipped_opt_out += 1 else: cel = await _upsert_celebration( db, person=person, celebration_type=CELEBRATION_TYPE_ANNIVERSARY, occasion_date=today, years=years, display_name_snapshot=display_name or None, department_snapshot=dept, avatar_url_snapshot=person.avatar_url, ) if cel is not None: new_anniversaries.append(cel) await _ensure_system_post( db, celebration=cel, post_type=POST_TYPE_SYSTEM_ANNIVERSARY, body=_format_anniv_body(display_name, years), body_i18n_key=BODY_KEY_SINGLE_ANNIVERSARY, body_params={"name": display_name, "count": years}, ) # Birthday if person.birthday is not None and _matches_md(today, person.birthday): if opt_out is None: opt_out = await _celebration_opt_out(db, person_id_s) if _opted_out_of(opt_out, CELEBRATION_TYPE_BIRTHDAY): skipped_opt_out += 1 elif not await _person_birthday_visible(db, person_id_s): skipped_hidden += 1 else: cel = await _upsert_celebration( db, person=person, celebration_type=CELEBRATION_TYPE_BIRTHDAY, occasion_date=today, years=0, display_name_snapshot=display_name or None, department_snapshot=dept, avatar_url_snapshot=person.avatar_url, ) if cel is not None: new_birthdays.append(cel) await _ensure_system_post( db, celebration=cel, post_type=POST_TYPE_SYSTEM_BIRTHDAY, body=_format_birthday_body(display_name), body_i18n_key=BODY_KEY_SINGLE_BIRTHDAY, body_params={"name": display_name}, ) # Aggregate banners — one per type per day. We use the Asia/Shanghai # "today" for the banner date so a single banner row is the # canonical "today's celebrants" anchor (per-person rows already # carry their own per-TZ occasion_date). banner_date = datetime.now(ZoneInfo("Asia/Shanghai")).date() await _refresh_aggregate_banner( db, post_type=POST_TYPE_AGGREGATE_BIRTHDAY, celebration_type=CELEBRATION_TYPE_BIRTHDAY, occasion_date=banner_date, ) await _refresh_aggregate_banner( db, post_type=POST_TYPE_AGGREGATE_ANNIVERSARY, celebration_type=CELEBRATION_TYPE_ANNIVERSARY, occasion_date=banner_date, ) return { "scanned": len(rows), "new_birthdays": len(new_birthdays), "new_anniversaries": len(new_anniversaries), "skipped_hidden": skipped_hidden, "skipped_opt_out": skipped_opt_out, "skipped_no_name": skipped_no_name, } async def _upsert_celebration( db: AsyncSession, *, person: IdentityPerson, celebration_type: str, occasion_date: date, years: int, display_name_snapshot: str | None, department_snapshot: str | None, avatar_url_snapshot: str | None, ) -> FeedCelebration | None: """Return the row if newly inserted, None if it already existed.""" existing = (await db.execute( select(FeedCelebration).where( FeedCelebration.person_id == str(person.id), FeedCelebration.occasion_date == occasion_date, FeedCelebration.celebration_type == celebration_type, ) )).scalar_one_or_none() if existing is not None: return None illustration_key = _pick_illustration( str(person.id), occasion_date, celebration_type, ) row = FeedCelebration( person_id=str(person.id), celebration_type=celebration_type, occasion_date=occasion_date, years=years, display_name_snapshot=display_name_snapshot, department_snapshot=department_snapshot, avatar_url_snapshot=avatar_url_snapshot, illustration_key=illustration_key, notify_state_json=None, ) try: # SAVEPOINT, not a bare ``db.rollback()``. ``scan_today_celebrations`` # accumulates every celebration + system post + aggregate banner in # ONE session that the handler commits exactly once at the end, so a # full ``db.rollback()`` on a single concurrent-insert collision would # discard the entire scan batch flushed so far. The savepoint rolls # back only this pending row (added inside the block) and leaves the # outer scan transaction intact. Mirrors ``FeedService.toggle_like`` / # ``send_blessing`` and the backfill script. async with db.begin_nested(): db.add(row) await db.flush() except IntegrityError: # Concurrent insert from another scanner — treat as already-existed. return None return row async def _ensure_system_post( db: AsyncSession, *, celebration: FeedCelebration, post_type: str, body: str, body_i18n_key: str, body_params: dict[str, Any], ) -> FeedPost: """One system post per celebration. Idempotent on celebration_id.""" existing = (await db.execute( select(FeedPost).where( FeedPost.celebration_id == celebration.id, FeedPost.post_type == post_type, ) )).scalar_one_or_none() if existing is not None: # Never rewritten: a system post is created once, on the tick that # created its celebration, and its wording can't drift afterwards. # Posts written before migration 456 keep ``body_i18n_key = NULL`` # and render from the stored English ``body`` — the front-end's # documented fallback — until they age out of the feed. return existing post = FeedPost( author_person_id=None, post_type=post_type, visibility="public", category="general", body=body, body_i18n_key=body_i18n_key, body_params=body_params, celebration_id=celebration.id, is_pinned=False, ) db.add(post) await db.flush() return post async def _refresh_aggregate_banner( db: AsyncSession, *, post_type: str, celebration_type: str, occasion_date: date, ) -> FeedPost | None: """Idempotent aggregate banner for today. Strategy: one row per (post_type, day). On every scan we look up the current count of celebrants and either insert / update the body. ``celebration_id`` stays NULL on aggregate banners. """ count = (await db.execute( select(FeedCelebration).where( FeedCelebration.occasion_date == occasion_date, FeedCelebration.celebration_type == celebration_type, ) )).scalars().all() n = len(count) # Locate today's existing banner. The cron runs in Asia/Shanghai # wall-clock days (matches ``banner_date`` semantics) but # ``FeedPost.created_at`` is UTC; an SHA-day spans # ``[SHA-day 00:00 UTC+8, SHA-day+1 00:00 UTC+8) == # [(date-1) 16:00 UTC, date 16:00 UTC)``. Using a naive UTC # day-window misses banners created during the SHA-evening hours # (UTC pre-midnight) and the cron then double-inserts. sha_tz = ZoneInfo("Asia/Shanghai") sha_start = datetime.combine( occasion_date, datetime.min.time(), tzinfo=sha_tz, ).astimezone(UTC) sha_end = datetime.combine( occasion_date, datetime.max.time(), tzinfo=sha_tz, ).astimezone(UTC) # ``.first()`` on an ordered query, NOT ``scalar_one_or_none`` # (#2187). There is no unique index on (post_type, SHA-day) — and # there cannot easily be one, because a banner is SOFT-deleted when # the day's celebrant count drops to zero and must be re-creatable # afterwards. So a second row is structurally possible, and the # comment right above records that this cron DID double-insert once, # before the SHA-day window was fixed. # # With ``scalar_one_or_none`` a single leftover pair would raise # ``MultipleResultsFound`` → the handler re-raises → the hourly job # goes red EVERY hour, and because the exception lands after the # per-celebration flush, that day's birthdays and anniversaries roll # back too. Failing the whole day's celebrations over a duplicate # BANNER is the wrong trade: take the earliest row deterministically, # keep going, and make the anomaly loud instead of fatal. # # Checked against production while fixing this: 0 duplicate # (post_type, SHA-day) aggregate banners today, so this is # hardening, not an outage repair. dupes = (await db.execute( select(FeedPost).where( FeedPost.post_type == post_type, FeedPost.deleted_at.is_(None), FeedPost.created_at >= sha_start, FeedPost.created_at <= sha_end, ).order_by(FeedPost.created_at.asc(), FeedPost.id.asc()) )).scalars().all() if len(dupes) > 1: logger.warning( "feed.celebration: %d duplicate %s banners for %s — using the " "earliest; the extras need a manual soft-delete", len(dupes), post_type, occasion_date, ) existing = dupes[0] if dupes else None if n == 0: if existing is not None: existing.deleted_at = datetime.now(UTC) await db.flush() return None # Fallback prose (see ``_format_*_body``) — the reader-facing text comes # from ``body_i18n_key`` + ``body_params``. plural = "s" if n != 1 else "" if celebration_type == CELEBRATION_TYPE_BIRTHDAY: body = ( f"{n} teammate{plural} celebrating their birthday today — " f"happy birthday!" ) body_i18n_key = BODY_KEY_AGGREGATE_BIRTHDAY else: body = ( f"{n} teammate{plural} celebrating their work anniversary " f"today!" ) body_i18n_key = BODY_KEY_AGGREGATE_ANNIVERSARY # ``occasion_date`` travels WITH the banner. It is the SHA-local day this # banner speaks for, and it is the key readers must use to look the day's # celebrants up in ``feed_celebrations.occasion_date``. Deriving it from # ``created_at`` instead is wrong for the eight hours of every SHA day # that fall on the previous UTC day (SHA 00:00–07:59 == UTC 16:00–23:59 # of D-1) — precisely the hours this cron creates the banner in, which is # why the celebrant lookup used to come back empty and the banner # rendered with no avatars and no "view" affordance. body_params: dict[str, Any] = { # ``count``, not ``n``: i18next pluralises on the reserved ``count`` # param. With ``n`` the English key can't inflect, and at ~500 headcount # most days have exactly ONE celebrant — so the common case rendered # "1 teammates celebrating today". The server's old f-string got the # plural right; the i18n key has to as well. "count": n, "occasion_date": occasion_date.isoformat(), } if existing is not None: existing.body = body existing.body_i18n_key = body_i18n_key existing.body_params = body_params await db.flush() return existing post = FeedPost( author_person_id=None, post_type=post_type, visibility="public", category="general", body=body, body_i18n_key=body_i18n_key, body_params=body_params, celebration_id=None, is_pinned=False, ) db.add(post) await db.flush() return post # ────────────────────────────────────────────────────────────────── # Notifier — runs after the scan; emits per-celebration push events. # ────────────────────────────────────────────────────────────────── async def notify_today_celebrations(db: AsyncSession) -> dict[str, Any]: """Walk today's celebrations and emit Feishu+inbox notifications. Idempotent via ``notify_state_json``: each (celebration, role) pair is recorded so the hourly tick re-runs don't double-fire. """ from app.domains.feed import notifications as fn today = datetime.now(ZoneInfo("Asia/Shanghai")).date() rows = (await db.execute( select(FeedCelebration).where( FeedCelebration.occasion_date == today, ) )).scalars().all() sent_self = 0 sent_manager = 0 sent_hrbp = 0 skipped_opt_out = 0 now_iso = datetime.now(UTC).isoformat() for c in rows: # Generation already honours the opt-out, so today's rows normally # can't belong to an opted-out person — but a row created BEFORE the # person set the preference still exists, and the manager + HRBP # pings are exactly what they opted out of. Read it here too rather # than trusting the generator's gate. opted_out = _opted_out_of( await _celebration_opt_out(db, str(c.person_id)), c.celebration_type, ) # Per-row SAVEPOINT isolation: the loop runs on ONE shared # session, and a dispatch's DB write (outbox flush / audit log) # can abort the transaction (lock-wait 1205, constraint). Without # isolation that poisons the whole batch — the next db.get() and # the final commit raise PendingRollbackError, losing the # in-memory notify_state_json marks of rows already notified this # tick, so they get re-pinged next tick. Wrapping each row in # begin_nested + a per-row flush confines a DB-abort to that row. try: async with db.begin_nested(): state = dict(c.notify_state_json or {}) if opted_out: # Permanent skip for all three roles. ``setdefault`` so an # already-sent mark (opted out mid-day) is not rewritten # into a skip — that would misreport what was delivered. before = len(state) for role in ("self", "manager", "hrbp"): state.setdefault(role, f"skipped:opt_out:{now_iso}") if len(state) != before: skipped_opt_out += 1 # Self — mark state on EVERY terminal outcome (sent OR # permanent-skip = person inactive / missing). Without the # skip-marker the hourly cron would re-attempt the lookup + # short-circuit for every leaver / on-leave employee 24 # ticks a day — waste, and noisy in the run log. if "self" not in state: try: ok = await fn.notify_celebration_self(db, celebration=c) state["self"] = now_iso if ok else f"skipped:{now_iso}" if ok: sent_self += 1 except Exception as exc: if is_transaction_aborting_db_error(exc): raise # let the row savepoint roll back logger.exception("celebration.self dispatch failed cid=%s", c.id) # Transient — DON'T mark; retry next tick. # Manager person = await db.get(IdentityPerson, uuid.UUID(str(c.person_id))) if person is not None and "manager" not in state: mgr_id = person.manager_person_id if mgr_id is None or str(mgr_id) == str(person.id): # Permanent skip — leaver with no manager / # self-managed CEO. Mark so cron stops looking. state["manager"] = f"skipped:no_manager:{now_iso}" else: manager = await db.get(IdentityPerson, mgr_id) if manager is None or manager.employment_status != "active": state["manager"] = f"skipped:manager_inactive:{now_iso}" else: try: await fn.notify_celebration_lead( db, celebration=c, recipient=manager, role="manager", ) state["manager"] = now_iso sent_manager += 1 except Exception as exc: if is_transaction_aborting_db_error(exc): raise logger.exception( "celebration.manager dispatch failed cid=%s mgr=%s", c.id, mgr_id, ) # HRBP if person is not None and "hrbp" not in state: hrbp_id = person.hrbp_person_id if hrbp_id is None or str(hrbp_id) == str(person.id): state["hrbp"] = f"skipped:no_hrbp:{now_iso}" else: hrbp = await db.get(IdentityPerson, hrbp_id) if hrbp is None or hrbp.employment_status != "active": state["hrbp"] = f"skipped:hrbp_inactive:{now_iso}" else: try: await fn.notify_celebration_lead( db, celebration=c, recipient=hrbp, role="hrbp", ) state["hrbp"] = now_iso sent_hrbp += 1 except Exception as exc: if is_transaction_aborting_db_error(exc): raise logger.exception( "celebration.hrbp dispatch failed cid=%s hrbp=%s", c.id, hrbp_id, ) c.notify_state_json = state await db.flush() except Exception as exc: # A server-side deadlock / lock-wait (1205/1213/1305) aborts the # WHOLE transaction, not just this savepoint — continuing would # poison every subsequent db.get()/flush() with PendingRollbackError # and mask the real errno. Re-raise so the handler fails loudly # (and a fresh-session retry can take over) instead of silently # skipping the rest of the batch. if is_transaction_aborting_db_error(exc): raise # Otherwise it's a row-local failure — its savepoint rolled back # (this row's marks are discarded and retried next tick); the # batch and other rows' marks survive. logger.exception( "celebration row failed cid=%s — savepoint rolled back, " "retry next tick", c.id, ) continue # Per-row commit. notify_celebration_* → dispatch enqueues an # outbox row whose recipient_person_id FK takes a shared lock on # identity_persons, and dispatch then does the external Feishu / # email send in THIS transaction. Deferring the commit to # end-of-run holds those FK S-locks (celebrant + manager + HRBP) # across every external send; a concurrent SSO login's # UPDATE identity_persons ... SET last_login_at then blocks behind # them → 1205 login brown-out (same class as asset.borrow_overdue # / #1837, smaller blast radius). Committing per row bounds the # lock hold to one celebration and makes its notify_state marks # durable so a later-row crash can't re-ping earlier celebrants. await db.commit() await db.flush() return { "celebrations": len(rows), "sent_self": sent_self, "sent_manager": sent_manager, "sent_hrbp": sent_hrbp, "skipped_opt_out": skipped_opt_out, } # ────────────────────────────────────────────────────────────────── # Announcement pin / expiry # ────────────────────────────────────────────────────────────────── async def reconcile_announcement_publication(db: AsyncSession) -> dict[str, Any]: """Flip is_pinned / status based on publish_at / expire_at.""" from app.domains.feed.models import ( ANNOUNCEMENT_STATUS_ARCHIVED, ANNOUNCEMENT_STATUS_PUBLISHED, FeedAnnouncement, ) now = datetime.now(UTC) # Auto-publish drafts whose publish_at has passed auto_published = 0 drafts = (await db.execute( select(FeedAnnouncement).where( FeedAnnouncement.deleted_at.is_(None), FeedAnnouncement.status == "draft", FeedAnnouncement.publish_at.is_not(None), FeedAnnouncement.publish_at <= now, ) )).scalars().all() for ann in drafts: ann.status = ANNOUNCEMENT_STATUS_PUBLISHED auto_published += 1 # Auto-archive expired auto_archived = 0 expired = (await db.execute( select(FeedAnnouncement).where( FeedAnnouncement.deleted_at.is_(None), FeedAnnouncement.status == ANNOUNCEMENT_STATUS_PUBLISHED, FeedAnnouncement.expire_at.is_not(None), FeedAnnouncement.expire_at <= now, ) )).scalars().all() for ann in expired: ann.status = ANNOUNCEMENT_STATUS_ARCHIVED ann.is_pinned = False auto_archived += 1 await db.flush() return { "auto_published": auto_published, "auto_archived": auto_archived, } __all__ = [ "ANNIVERSARY_ILLUSTRATION_KEYS", "BIRTHDAY_ILLUSTRATION_KEYS", "BODY_KEY_AGGREGATE_ANNIVERSARY", "BODY_KEY_AGGREGATE_BIRTHDAY", "BODY_KEY_SINGLE_ANNIVERSARY", "BODY_KEY_SINGLE_BIRTHDAY", "notify_today_celebrations", "reconcile_announcement_publication", "scan_today_celebrations", ]