"""放松一下 service — overview, round lifecycle (start → submit), ledger. All reads/writes are self-scoped (the caller's own wallet/sessions). The wallet row lock serializes each person's balance math + daily cap; game scores are computed exclusively server-side (see catalog.py). """ from __future__ import annotations import logging import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException, NotFoundError from app.domains.arcade import economy from app.domains.arcade.catalog import GAMES, GAMES_BY_KEY, SUBMIT_GRACE_MS from app.domains.arcade.models import ( SESSION_ACTIVE, SESSION_COMPLETED, SESSION_EXPIRED, ArcadeGameSession, ArcadePointsLedger, ) from app.identity.person_models import IdentityOrgMembership, IdentityPerson logger = logging.getLogger(__name__) # An active round older than this is dead (browser closed mid-game). It is # lazily marked expired on the next read; the entry fee stays consumed — # the round was issued. Communicated in the UI before starting. SESSION_TTL = timedelta(minutes=15) def _now() -> datetime: return datetime.now(UTC) def _aware(dt: datetime | None) -> datetime | None: """DB DATETIME columns come back naive (UTC by convention) — normalise so age math never mixes naive/aware.""" if dt is None: return None return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC) async def _expire_stale_sessions(db: AsyncSession, person: IdentityPerson) -> None: """Lazily close overdue rounds so 'active' never dangles. A round is dead once its OWN clock ran out (time_limit + submit grace + a small buffer) — not only after the generic 15-min TTL. Returning a zero-remaining round as 'active' put the frontend in a loop (adopt → instant time-up → submit rejected as expired → reload overview → adopt again…), caught in the 2026-07-18 preview walk. The TTL stays as the ceiling for game keys missing from the registry. """ now = _now() active = (await db.execute( select(ArcadeGameSession).where( ArcadeGameSession.person_id == person.id, ArcadeGameSession.status == SESSION_ACTIVE, ) )).scalars().all() for s in active: started = _aware(s.started_at) if started is None: continue spec = GAMES_BY_KEY.get(s.game_key) if spec is not None: deadline = started + timedelta( milliseconds=spec.time_limit_ms + SUBMIT_GRACE_MS + 5_000 ) else: deadline = started + SESSION_TTL if now > min(deadline, started + SESSION_TTL): s.status = SESSION_EXPIRED s.score = 0 s.finished_at = now def _serialize_session_public( session: ArcadeGameSession, spec_time_limit_ms: int, ) -> dict[str, Any]: """Session shape for the player. NEVER includes the raw payload — active rounds get the public projection, finished rounds the result.""" started = _aware(session.started_at) out: dict[str, Any] = { "session_id": str(session.id), "game_key": session.game_key, "variant": session.variant, "status": session.status, "cost_points": session.cost_points, "time_limit_ms": spec_time_limit_ms, "started_at": started.isoformat() if started else None, "score": session.score, "duration_ms": session.duration_ms, } if session.status == SESSION_ACTIVE: elapsed = int((_now() - started).total_seconds() * 1000) if started else 0 out["public"] = (session.payload or {}).get("_public", {}) out["remaining_ms"] = max(0, spec_time_limit_ms - elapsed) elif session.status == SESSION_COMPLETED: out["result"] = session.result return out async def get_overview(db: AsyncSession, person: IdentityPerson) -> dict[str, Any]: """The arcade home payload: wallet (after lazy quiz-credit reconcile), today's play budget, and the game wall with the caller's personal bests.""" cfg = await economy.get_economy_config(db) wallet = await economy.get_wallet_locked(db, person) earned_now = await economy.reconcile_quiz_credits(db, person, wallet, cfg) await _expire_stale_sessions(db, person) today = economy.local_today(person) plays_used = int((await db.scalar( select(func.count()).select_from(ArcadeGameSession).where( ArcadeGameSession.person_id == person.id, ArcadeGameSession.play_date == today, ) )) or 0) # Personal bests + play counts per game, one grouped query. rows = (await db.execute( select( ArcadeGameSession.game_key, func.max(ArcadeGameSession.score), func.count(), ) .where( ArcadeGameSession.person_id == person.id, ArcadeGameSession.status == SESSION_COMPLETED, ) .group_by(ArcadeGameSession.game_key) )).all() my_stats = {k: {"best_score": int(b or 0), "plays": int(c or 0)} for k, b, c in rows} active = (await db.execute( select(ArcadeGameSession).where( ArcadeGameSession.person_id == person.id, ArcadeGameSession.status == SESSION_ACTIVE, ).order_by(ArcadeGameSession.started_at.desc()).limit(1) )).scalars().first() from app.domains.arcade import leaderboard as _lb my_titles = await _lb.get_my_titles(db, person) return { "wallet": economy.serialize_wallet(wallet), "earned_on_open": earned_now, "entry_cost": cfg.entry_cost, "my_titles": my_titles, "today": { "date": today.isoformat(), "plays_used": plays_used, "plays_left": max(0, cfg.daily_play_cap - plays_used), "cap": cfg.daily_play_cap, }, "active_session": ( _serialize_session_public(active, GAMES_BY_KEY[active.game_key].time_limit_ms) if active and active.game_key in GAMES_BY_KEY else None ), "games": [ { "key": g.key, "category": g.category, "icon": g.icon, "time_limit_ms": g.time_limit_ms, "variants": list(g.variants), "competitive": g.competitive, "my_best_score": my_stats.get(g.key, {}).get("best_score", 0), "my_plays": my_stats.get(g.key, {}).get("plays", 0), } for g in GAMES ], "earn_rules": { "quiz_correct": cfg.points_correct, "quiz_wrong": cfg.points_wrong, "quiz_correct_streak": cfg.points_correct_streak, "streak_threshold": cfg.streak_threshold, }, } async def get_ledger( db: AsyncSession, person: IdentityPerson, *, limit: int = 20, ) -> dict[str, Any]: rows = (await db.execute( select(ArcadePointsLedger) .where(ArcadePointsLedger.person_id == person.id) .order_by(ArcadePointsLedger.created_at.desc(), ArcadePointsLedger.id.desc()) .limit(max(1, min(limit, 100))) )).scalars().all() return { "entries": [ { "id": str(r.id), "delta": r.delta, "balance_after": r.balance_after, "reason": r.reason, "ref_id": r.ref_id, "created_at": ( aware.isoformat() if (aware := _aware(r.created_at)) else None ), } for r in rows ], } def _serialize_log_row(session: ArcadeGameSession) -> dict[str, Any]: """One finished-round row for the history / audit log (never the secret payload — only the graded, publicly-safe facts).""" started = _aware(session.started_at) finished = _aware(session.finished_at) return { "session_id": str(session.id), "game_key": session.game_key, "variant": session.variant, "status": session.status, "score": session.score, "correct_count": session.correct_count, "duration_ms": session.duration_ms, "cost_points": session.cost_points, "play_date": session.play_date.isoformat() if session.play_date else None, "started_at": started.isoformat() if started else None, "finished_at": finished.isoformat() if finished else None, } async def get_history( db: AsyncSession, person: IdentityPerson, *, limit: int = 30, ) -> dict[str, Any]: """The caller's own finished game rounds (completed + expired), newest first. Self-scoped only; active rounds are excluded (not a played record yet — the overview/resume banner owns the live one).""" await _expire_stale_sessions(db, person) rows = (await db.execute( select(ArcadeGameSession) .where( ArcadeGameSession.person_id == person.id, ArcadeGameSession.status != SESSION_ACTIVE, ) .order_by(ArcadeGameSession.started_at.desc(), ArcadeGameSession.id.desc()) .limit(max(1, min(limit, 100))) )).scalars().all() return {"entries": [_serialize_log_row(s) for s in rows]} async def admin_list_sessions( db: AsyncSession, *, person_id: uuid.UUID | None = None, game_key: str | None = None, status: str | None = None, limit: int = 50, offset: int = 0, ) -> dict[str, Any]: """Cross-person round audit for admins/management: every finished round with the player's name + primary department, newest first, filterable by person / game / status and paginated. Departed people stay visible (an auditor needs the historical trail), but active (unfinished) rounds are excluded — they are not a played record yet.""" limit = max(1, min(limit, 200)) offset = max(0, offset) dept_sub = ( select( IdentityOrgMembership.person_id.label("pid"), func.max(IdentityOrgMembership.org_node_name).label("dept"), ) .where( IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.membership_status == "active", ) .group_by(IdentityOrgMembership.person_id) .subquery() ) # A specific finished-status filter narrows to exactly it; otherwise the # default is "any finished round" (everything but the still-active one). if status in (SESSION_COMPLETED, SESSION_EXPIRED): conds = [ArcadeGameSession.status == status] else: conds = [ArcadeGameSession.status != SESSION_ACTIVE] if person_id is not None: conds.append(ArcadeGameSession.person_id == person_id) if game_key: conds.append(ArcadeGameSession.game_key == game_key) total = int((await db.scalar( select(func.count()).select_from(ArcadeGameSession).where(*conds) )) or 0) rows = (await db.execute( select( ArcadeGameSession, IdentityPerson.display_name, dept_sub.c.dept, ) .join(IdentityPerson, IdentityPerson.id == ArcadeGameSession.person_id) .join(dept_sub, dept_sub.c.pid == ArcadeGameSession.person_id, isouter=True) .where(*conds) .order_by(ArcadeGameSession.started_at.desc(), ArcadeGameSession.id.desc()) .limit(limit) .offset(offset) )).all() entries = [] for session, name, dept in rows: row = _serialize_log_row(session) row["person_id"] = str(session.person_id) row["display_name"] = name or "—" row["department"] = dept or None entries.append(row) return { "entries": entries, "total": total, "limit": limit, "offset": offset, } async def start_session( db: AsyncSession, person: IdentityPerson, game_key: str, variant: str | None = None, ) -> dict[str, Any]: """Charge the entry fee and issue a fresh server-built round. Order inside one wallet-locked transaction: 1. reconcile quiz credits (a just-answered daily question counts) 2. daily-cap check (lock-serialized; unique(person,date,seq) backstops) 3. engine build (can fail → nothing charged) 4. charge + insert session """ spec = GAMES_BY_KEY.get(game_key) if spec is None: raise NotFoundError("arcade_game", game_key) # Validate the requested sub-mode against the game's declared variants # (defence-in-depth; the engine also resolves unknowns to a default). if variant is not None and spec.variants and variant not in spec.variants: raise AppException( code="ARCADE_BAD_VARIANT", message="Unknown game mode.", status_code=422, ) cfg = await economy.get_economy_config(db) wallet = await economy.get_wallet_locked(db, person) await economy.reconcile_quiz_credits(db, person, wallet, cfg) await _expire_stale_sessions(db, person) today = economy.local_today(person) plays_used = int((await db.scalar( select(func.count()).select_from(ArcadeGameSession).where( ArcadeGameSession.person_id == person.id, ArcadeGameSession.play_date == today, ) )) or 0) if plays_used >= cfg.daily_play_cap: raise AppException( code="ARCADE_DAILY_CAP_REACHED", message="Daily play limit reached — come back tomorrow.", status_code=422, ) # Refuse a second concurrent round: finish (or let expire) the current # one first. Keeps the 'one live round' invariant simple for clients. # Fetch the live round's game so the error can NAME it (and the client # can deep-link the player straight to it) — a bare "finish it first" # is useless when you don't know which game holds the round. active_round = (await db.execute( select(ArcadeGameSession).where( ArcadeGameSession.person_id == person.id, ArcadeGameSession.status == SESSION_ACTIVE, ).order_by(ArcadeGameSession.started_at.desc()).limit(1) )).scalars().first() if active_round is not None: raise AppException( code="ARCADE_ROUND_IN_PROGRESS", message="You already have a round in progress — finish it first.", status_code=409, field=active_round.game_key, ) built = await spec.build(db, person, variant) session_id = uuid.uuid4() await economy.spend_game_entry( db, wallet, session_id=session_id, cost=cfg.entry_cost, ) session = ArcadeGameSession( id=session_id, person_id=person.id, game_key=game_key, variant=built.variant, status=SESSION_ACTIVE, play_date=today, play_seq=plays_used + 1, cost_points=cfg.entry_cost, # Store the public projection alongside the secret state so a page # refresh can resume the round without re-deriving it. payload={**built.payload, "_public": built.public}, started_at=_now(), ) try: async with db.begin_nested(): db.add(session) await db.flush() except IntegrityError as exc: # uq_arcade_sess_seq collision — a concurrent start slipped past the # count (shouldn't happen under the wallet lock, but the unique key # is the backstop). Surface as the cap error; the wallet charge # rolls back with the request transaction. raise AppException( code="ARCADE_DAILY_CAP_REACHED", message="Daily play limit reached — come back tomorrow.", status_code=422, ) from exc return { "session_id": str(session_id), "game_key": game_key, "variant": built.variant, "status": SESSION_ACTIVE, "time_limit_ms": spec.time_limit_ms, "cost_points": cfg.entry_cost, "public": built.public, "balance_after": wallet.balance, "plays_left": max(0, cfg.daily_play_cap - (plays_used + 1)), "started_at": (_aware(session.started_at) or _now()).isoformat(), } async def get_session( db: AsyncSession, person: IdentityPerson, session_id: uuid.UUID, ) -> dict[str, Any]: session = await db.get(ArcadeGameSession, session_id) if session is None or session.person_id != person.id: # Same 404 for both — do not leak other people's session ids. raise NotFoundError("arcade_session", str(session_id)) spec = GAMES_BY_KEY.get(session.game_key) time_limit = spec.time_limit_ms if spec else 0 if session.status == SESSION_ACTIVE: started = _aware(session.started_at) # Expire on the game's OWN deadline (time_limit + grace), not only # the 15-min TTL — otherwise this endpoint returns a zero-remaining # round as still-active, the exact inconsistency _expire_stale_sessions # was changed to avoid (the 2026-07-18 adopt→time-up loop). if started is not None: if spec is not None: deadline = started + timedelta( milliseconds=spec.time_limit_ms + SUBMIT_GRACE_MS + 5_000 ) else: deadline = started + SESSION_TTL if _now() > min(deadline, started + SESSION_TTL): session.status = SESSION_EXPIRED session.score = 0 session.finished_at = _now() return _serialize_session_public(session, time_limit) async def submit_session( db: AsyncSession, person: IdentityPerson, session_id: uuid.UUID, submission: dict[str, Any], ) -> dict[str, Any]: """Grade a round server-side. Idempotent: re-submitting a completed round returns the stored result unchanged (first submission wins).""" session = (await db.execute( select(ArcadeGameSession) .where(ArcadeGameSession.id == session_id) .with_for_update() )).scalar_one_or_none() if session is None or session.person_id != person.id: raise NotFoundError("arcade_session", str(session_id)) spec = GAMES_BY_KEY.get(session.game_key) if spec is None: raise NotFoundError("arcade_game", session.game_key) if session.status == SESSION_COMPLETED: return _serialize_session_public(session, spec.time_limit_ms) if session.status == SESSION_EXPIRED: raise AppException( code="ARCADE_SESSION_EXPIRED", message="This round has expired.", status_code=422, ) started = _aware(session.started_at) or _now() elapsed_ms = max(0, int((_now() - started).total_seconds() * 1000)) if elapsed_ms > spec.time_limit_ms + SUBMIT_GRACE_MS: session.status = SESSION_EXPIRED session.score = 0 session.finished_at = _now() raise AppException( code="ARCADE_SESSION_EXPIRED", message="This round has expired.", status_code=422, ) payload = dict(session.payload or {}) payload.pop("_public", None) graded = spec.grade(payload, submission or {}, elapsed_ms) session.status = SESSION_COMPLETED session.score = int(graded.score) session.duration_ms = min(elapsed_ms, spec.time_limit_ms) session.result = graded.detail # Denormalize the right-answer count (games that have one) for the # 科举 knowledge leaderboard's lifetime SUM. cc = graded.detail.get("correct_count") if isinstance(graded.detail, dict) else None session.correct_count = int(cc) if isinstance(cc, int) else None session.finished_at = _now() prev_best = int((await db.scalar( select(func.max(ArcadeGameSession.score)).where( ArcadeGameSession.person_id == person.id, ArcadeGameSession.game_key == session.game_key, ArcadeGameSession.status == SESSION_COMPLETED, ArcadeGameSession.id != session.id, ) )) or 0) out = _serialize_session_public(session, spec.time_limit_ms) out["is_personal_best"] = session.score > prev_best and session.score > 0 out["previous_best"] = prev_best return out __all__ = [ "get_overview", "get_ledger", "get_history", "admin_list_sessions", "start_session", "get_session", "submit_session", "SESSION_TTL", ]