"""Scheduled-job handler for 成就徽章. ``gamification.scan_unlocks`` — daily. Evaluates every active person's badges and persists newly-earned ones so ``unlocked_at`` reflects the real earn date (the read path also lazily evaluates, so a missed run never hides a badge). Two properties this handler has to keep: * **Short transactions.** Facts for the whole org are computed in one bulk pass (a fixed set of grouped queries, see ``gamification/facts.py``) rather than ~18 queries per person, and unlocks commit in chunks. A cron that holds one transaction open across 500 people is how this platform previously FK-locked ``identity_persons`` and browned out login. * **Per-item isolation.** One person's evaluation failing must not poison the batch (see [[reference_per_process_savepoint_isolation]]). A systemic failure still raises so the scheduler marks the run failed ([[feedback_handler_status_must_raise]]). """ from __future__ import annotations import logging from typing import Any from sqlalchemy import select from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession from app.domains.gamification.service import ( compute_facts_for, evaluate_and_unlock, prefetch_unlocks, ) from app.identity.person_models import IdentityPerson from app.scheduled_jobs.registry import register_scheduled_job logger = logging.getLogger(__name__) #: People per bulk-facts pass. Bounds both the IN-list width of each grouped #: query and how long any single transaction stays open. CHUNK_SIZE = 100 @register_scheduled_job( key="gamification.scan_unlocks", display_name="成就徽章 每日扫描解锁", description=( "每日评估全体在职员工的成就徽章,持久化新解锁(unlocked_at 取真实达成日)。" "按 100 人一批批量算事实 + 分批提交,逐人 SAVEPOINT 隔离," "单人失败不污染整批;读路径亦懒评估兜底。" ), ) async def run_scan_unlocks( db: AsyncSession, config: dict[str, Any], ) -> dict[str, Any]: # IDs first, rows per chunk. Holding 500 ORM objects across the whole loop # would be fragile in two ways: a ``rollback()`` on ANY chunk expires every # loaded object (rollback always expires, regardless of expire_on_commit), # and the next chunk's ``person.hire_date`` would then trigger a lazy # refresh — which under async SQLAlchemy raises MissingGreenlet rather than # quietly re-reading. Loading each chunk fresh keeps the blast radius of a # failed chunk inside that chunk. person_ids = list((await db.execute( select(IdentityPerson.id).where( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status != "left", ) )).scalars().all()) scanned = 0 unlocked = 0 errors = 0 total = len(person_ids) async def _load_chunk(ids: list[Any]) -> list[IdentityPerson]: return list((await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(ids)) )).scalars().all()) for start in range(0, total, CHUNK_SIZE): ids = person_ids[start:start + CHUNK_SIZE] facts: dict[Any, dict[str, Any]] = {} already: dict[Any, set[str]] = {} # Explicit flag, not ``already``'s truthiness: an EMPTY prefetch is the # legitimate answer when nobody in the chunk has ever unlocked anything # (a fresh org, the first-ever scan), and testing truthiness would # silently disable the prefetch exactly when it is cheapest. prefetched = False try: chunk = await _load_chunk(ids) facts = await compute_facts_for(db, chunk) already = await prefetch_unlocks(db, ids) prefetched = True except Exception: # A failed bulk pass must NOT silently skip 100 people while the # run still reports "ok" — fall back to per-person computation # (slower, but each person is individually SAVEPOINT-isolated and # genuinely gets scanned). Empty dicts below make # ``evaluate_and_unlock`` compute its own facts and re-read its own # unlocks. logger.exception( "gamification.scan_unlocks: bulk facts failed for chunk at %s " "— falling back to per-person computation", start, ) await db.rollback() facts = {} already = {} try: # Re-load AFTER the rollback: it expired the objects the failed # attempt had loaded. chunk = await _load_chunk(ids) except Exception: errors += len(ids) logger.exception( "gamification.scan_unlocks: chunk at %s unreadable — " "skipped", start, ) continue for person in chunk: try: async with db.begin_nested(): # per-person SAVEPOINT newly = await evaluate_and_unlock( db, person, facts=facts.get(person.id), # ``None`` (not an empty set) when the bulk prefetch # failed, so evaluate_and_unlock re-reads this person's # unlocks instead of assuming they have none — which # would make every insert collide. already=already.get(person.id, set()) if prefetched else None, ) unlocked += len(newly) scanned += 1 except OperationalError: # Lock-wait / deadlock poisons the session: every remaining # person in this chunk would fail against a dead transaction # and the chunk-ending commit would raise anyway. Surface it so # the scheduler's transient-infra retry re-runs on a fresh # session (same reasoning as access_certification_service). logger.exception( "gamification.scan_unlocks: infra error on person %s — " "aborting so the run can be retried", person.id, ) raise except Exception: errors += 1 logger.exception( "gamification.scan_unlocks: person %s failed", person.id, ) # Commit each chunk so the transaction never spans the whole org. await db.commit() result = {"scanned": scanned, "newly_unlocked": unlocked, "errors": errors} # A handful of per-person failures are tolerated (logged); a wholesale # failure (everyone errored) is systemic — surface it as a failed run. if total and errors == total: raise RuntimeError(f"scan_unlocks: all {errors} persons failed") return result __all__ = ["run_scan_unlocks"]