"""成就徽章 service — evaluate which badges a person has earned, persist newly
unlocked ones (idempotently), and assemble the wall the frontend renders.

Surfacing is pure-pull + zero-disturbance (the established 有温度 series
ethos): no push, no actionable red-badge. ``get_my_badges`` lazily evaluates on
read so a freshly-qualified badge shows immediately (the frontend fires the
celebration confetti on a newly-seen unlock); a daily cron also evaluates the
whole org so ``unlocked_at`` reflects the real earn date rather than first view.

Beyond earned/locked the wall carries three things that turn a grid of grey
squares into something worth opening:

* **progress** — how far along each unearned badge is (``5/7 天``), straight
  from the fact the badge is measured on.
* **rarity** — how many *other* people hold it, org-wide. Computed from the
  unlock table against the same active-headcount denominator the arcade
  leaderboard uses, so the two features never quote different numbers.
* **next up** — the handful of locked badges the caller is closest to, so the
  wall answers "what should I do next?" without the user doing the maths.
"""
from __future__ import annotations

import time
import uuid
from dataclasses import dataclass
from datetime import UTC, date, datetime
from typing import Any

from sqlalchemy import and_, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.tenure import observed_anniversary
from app.domains.gamification.catalog import BADGES, CATEGORIES, TIERS, Badge
from app.domains.gamification.facts import compute_facts, compute_facts_for
from app.domains.gamification.models import GamificationUnlock
from app.identity.person_models import IdentityPerson

#: How many "closest to unlocking" badges the wall highlights.
NEXT_UP_LIMIT = 3

#: Rarity is an org-wide aggregate that every employee's wall asks for. Without
#: a cache, each page open runs a GROUP BY over the unlock table plus a
#: headcount COUNT — work that is identical for every caller and changes maybe
#: a handful of times a day. 5 minutes is well inside "fresh enough for a
#: percentage" and turns a per-request scan into a per-5-minutes one.
_RARITY_TTL_SECONDS = 300


@dataclass(slots=True)
class _RarityCache:
    expires_at: float
    holders: dict[str, int]
    headcount: int


_rarity_cache: _RarityCache | None = None


def reset_rarity_cache() -> None:
    """Drop the cached org-wide rarity aggregate. Test seam — there is no
    admin "refresh" button wired to it; the TTL is the only production
    invalidation."""
    global _rarity_cache
    _rarity_cache = None


async def _rarity_snapshot(db: AsyncSession) -> tuple[dict[str, int], int]:
    """``(holders_by_code, active_headcount)``, cached for _RARITY_TTL_SECONDS."""
    global _rarity_cache
    now = time.monotonic()
    cached = _rarity_cache
    if cached is not None and now < cached.expires_at:
        return cached.holders, cached.headcount
    holders = await _holder_counts(db)
    headcount = await _active_headcount(db)
    _rarity_cache = _RarityCache(
        expires_at=now + _RARITY_TTL_SECONDS, holders=holders, headcount=headcount,
    )
    return holders, headcount


def _active_person_filter() -> tuple:
    """The active-headcount caliber used as the rarity denominator.

    Wide caliber (``!= 'left'``, i.e. including leaver_pending / pre_hire) to
    match the arcade leaderboard and participation board — the neighbouring
    features in the same series. Quoting a different denominator here would
    make "12% of colleagues have this" disagree with the numbers next door.
    """
    return (
        IdentityPerson.deleted_at.is_(None),
        IdentityPerson.employment_status != "left",
    )


async def prefetch_unlocks(
    db: AsyncSession, person_ids: list[uuid.UUID],
) -> dict[uuid.UUID, set[str]]:
    """``{person_id: {badge_code, …}}`` for a whole set, in one query.

    Lets the nightly scan skip a per-person SELECT (500 round-trips) while
    keeping each person's *insert* individually SAVEPOINT-isolated.
    """
    if not person_ids:
        return {}
    rows = (await db.execute(
        select(GamificationUnlock.person_id, GamificationUnlock.badge_code)
        .where(GamificationUnlock.person_id.in_(person_ids))
    )).all()
    out: dict[uuid.UUID, set[str]] = {}
    for person_id, code in rows:
        out.setdefault(person_id, set()).add(code)
    return out


async def evaluate_and_unlock(
    db: AsyncSession,
    person: IdentityPerson,
    facts: dict[str, Any] | None = None,
    already: set[str] | None = None,
) -> list[str]:
    """Evaluate every badge for ``person`` and persist newly-earned ones.

    Idempotent — already-unlocked badges are skipped via the unique key.
    Returns the badge codes unlocked on THIS call (empty if none new). Pass
    ``facts`` / ``already`` to reuse work the caller already did in bulk (the
    nightly scan computes the whole org's facts and unlocks in one pass each).
    """
    if facts is None:
        facts = await compute_facts(db, person)
    earned = {b.code for b in BADGES if b.is_earned(facts)}
    if not earned:
        return []

    if already is None:
        already = set((await db.execute(
            select(GamificationUnlock.badge_code).where(
                GamificationUnlock.person_id == person.id,
            )
        )).scalars().all())
    newly = [c for c in earned if c not in already]
    if newly:
        # SAVEPOINT-isolate the insert: the nightly scan_unlocks cron may
        # insert the same (person, badge) in the window between the SELECT
        # above and this flush (a user opening their badge wall the very night
        # they first qualify). The unique key would then raise an
        # IntegrityError that, on this non-nested request txn, would 500 the
        # read and poison the session. Contain it and treat the rows as
        # already-unlocked (the cron / a refresh surfaces them).
        # Try the whole batch in ONE savepoint; only a collision pays for
        # per-code isolation. A long-tenured employee opening the wall for the
        # first time unlocks 15-25 badges at once, and a savepoint round-trip
        # per badge inside a synchronous GET is real latency for a case that
        # almost never collides.
        #
        # On IntegrityError — the nightly cron inserting the same badge in the
        # window between the SELECT above and this flush — retry code by code,
        # so one collision no longer discards every OTHER badge in the batch
        # (which used to silently swallow the rest of the user's confetti).
        # The failed savepoint's rollback expunges the pending rows, so the
        # retry cannot double-insert; ``test_unlock_collision_2026_08_10``
        # pins that empirically rather than trusting the ORM's semantics.
        try:
            async with db.begin_nested():
                for code in newly:
                    db.add(GamificationUnlock(person_id=person.id, badge_code=code))
                await db.flush()
        except IntegrityError:
            persisted: list[str] = []
            for code in newly:
                try:
                    async with db.begin_nested():
                        db.add(GamificationUnlock(person_id=person.id, badge_code=code))
                        await db.flush()
                    persisted.append(code)
                except IntegrityError:
                    continue  # already unlocked by the racing writer
            newly = persisted
    # Preserve catalog order in the return.
    return [b.code for b in BADGES if b.code in newly]


async def _holder_counts(db: AsyncSession) -> dict[str, int]:
    """``{badge_code: holders}`` counting only active people.

    Departed colleagues are excluded so rarity does not drift upward forever as
    the company turns over.
    """
    rows = (await db.execute(
        select(
            GamificationUnlock.badge_code,
            func.count(func.distinct(GamificationUnlock.person_id)),
        )
        .join(
            IdentityPerson,
            and_(
                IdentityPerson.id == GamificationUnlock.person_id,
                *_active_person_filter(),
            ),
        )
        .group_by(GamificationUnlock.badge_code)
    )).all()
    return {code: int(count or 0) for code, count in rows}


async def _active_headcount(db: AsyncSession) -> int:
    return int((await db.scalar(
        select(func.count()).select_from(IdentityPerson).where(*_active_person_filter())
    )) or 0)


def _holder_percent(holders: int, headcount: int) -> int | None:
    """Whole-percent share, never 0 for a badge somebody actually holds."""
    if headcount <= 0:
        return None
    if holders <= 0:
        return 0
    return max(1, round(100 * holders / headcount))


#: Anniversary badge code → the whole-year milestone it marks. Derived from
#: the catalog so a new anniversary badge cannot silently miss the true-date
#: treatment below.
_ANNIVERSARY_YEARS: dict[str, int] = {
    b.code: int(b.threshold)
    for b in BADGES
    if b.metric == "tenure_years" and b.threshold is not None
}


def _tenure_earned_date(badge_code: str, person: IdentityPerson) -> date | None:
    """The day a 同行时光 badge was actually earned, derived from hire_date.

    ``unlocked_at`` records when the system *noticed* (nightly scan or first
    wall open) — for anyone whose anniversary predates the feature, that is
    the feature's launch date, and the wall rendered it as if it were the
    anniversary: 满一周年 and 三年老兵 both stamped 2026-06-22, 启航 stamped
    "your first day: 2026-08-10" for someone hired in 2022. For these badges
    the true date is derivable, so display that instead. ``None`` (no
    hire_date, or not a tenure-date badge) falls back to ``unlocked_at``.
    """
    if person.hire_date is None:
        return None
    if badge_code == "first_day":
        return person.hire_date
    years = _ANNIVERSARY_YEARS.get(badge_code)
    if years is None:
        return None
    return observed_anniversary(person.hire_date, person.hire_date.year + years)


def _serialize(
    badge: Badge,
    facts: dict[str, Any],
    unlocked_at: datetime | date | None,
    holders: int,
    headcount: int,
) -> dict[str, Any]:
    """One badge as the EMPLOYEE wall sees it.

    Note what is deliberately absent: the raw ``holders`` count. Rarity is
    surfaced as a rounded percentage only, because the absolute number answers
    questions this surface has no business answering precisely — "how many
    people hold ≥10 cloud grants", "how many hold live API keys", "how many can
    approve". This is noise reduction, NOT a privacy boundary: the headcount
    denominator is already public to every employee (the footprint page's
    "you were colleague #N" uses the same active-headcount caliber), so a
    determined reader can still bracket the count from the percentage. The
    gated admin stats keep both figures at full precision.
    """
    earned = unlocked_at is not None
    progress = badge.progress(facts)
    # An earned badge always reads as complete: a threshold that later moves,
    # or a counter that can go down (active cloud grants, held seats), must
    # never render an already-unlocked badge as "4/5 — nearly there".
    if earned and progress is not None:
        progress = {**progress, "current": progress["target"], "percent": 100}
    return {
        "code": badge.code,
        "icon": badge.icon,
        "category": badge.category,
        "tier": badge.tier,
        "name_key": badge.name_key,
        "desc_key": badge.desc_key,
        "earned": earned,
        "unlocked_at": unlocked_at.isoformat() if unlocked_at else None,
        "progress": progress,
        # Share of active colleagues, whole percent. ``None`` when headcount
        # is unknown, so the UI hides the line rather than claiming "0% of
        # colleagues have this". Floored at 1 whenever anyone holds it: 2
        # holders in a 500-person org rounds to 0, and the person reading that
        # line is one of the two.
        "holder_percent": _holder_percent(holders, headcount),
    }


def _next_up(badges: list[dict[str, Any]]) -> list[str]:
    """Codes of the locked badges the caller is closest to earning.

    Only badges with real movement (>0%) qualify — suggesting something at 0/100
    is noise, not a nudge.
    """
    candidates = [
        b for b in badges
        if not b["earned"] and b["progress"] and b["progress"]["percent"] > 0
    ]
    candidates.sort(key=lambda b: (-b["progress"]["percent"], b["code"]))
    return [b["code"] for b in candidates[:NEXT_UP_LIMIT]]


async def get_my_badges(
    db: AsyncSession, person: IdentityPerson,
) -> dict[str, Any]:
    """The caller's full badge wall — every catalog badge with earned/locked,
    unlocked_at, progress and rarity, plus category rollups and ``next_up``.

    Lazily evaluates first so a freshly-qualified badge appears immediately.
    ``newly_unlocked`` lets the frontend fire confetti once.
    """
    facts = await compute_facts(db, person)
    newly = await evaluate_and_unlock(db, person, facts=facts)

    rows = (await db.execute(
        select(GamificationUnlock).where(GamificationUnlock.person_id == person.id)
    )).scalars().all()
    unlocked_at: dict[str, datetime | date] = {r.badge_code: r.unlocked_at for r in rows}
    # 同行时光 badges display the day they were actually earned (hire date /
    # true anniversary), not the day the scan wrote the row — see
    # ``_tenure_earned_date``. Only earned badges are touched: the row's
    # presence still decides earned/locked.
    for code in unlocked_at:
        true_date = _tenure_earned_date(code, person)
        if true_date is not None:
            unlocked_at[code] = true_date

    holders, headcount = await _rarity_snapshot(db)

    badges = [
        _serialize(b, facts, unlocked_at.get(b.code), holders.get(b.code, 0), headcount)
        for b in BADGES
    ]

    by_category = []
    for key in CATEGORIES:
        members = [b for b in badges if b["category"] == key]
        if not members:
            continue
        by_category.append({
            "key": key,
            "total": len(members),
            "earned": sum(1 for b in members if b["earned"]),
        })

    tier_counts = {
        tier: {
            "total": sum(1 for b in badges if b["tier"] == tier),
            "earned": sum(1 for b in badges if b["tier"] == tier and b["earned"]),
        }
        for tier in TIERS
    }

    return {
        "badges": badges,
        "earned_count": sum(1 for b in badges if b["earned"]),
        "total_count": len(badges),
        "categories": by_category,
        "tier_counts": tier_counts,
        "next_up": _next_up(badges),
        "newly_unlocked": newly,
        "generated_at": datetime.now(UTC).isoformat(),
    }


async def get_org_stats(db: AsyncSession) -> dict[str, Any]:
    """Org-wide unlock distribution — the admin / leadership view of whether
    any of this is landing.

    Deliberately aggregate-only: per-badge holder counts and how many people
    hold at least one badge. No named leaderboard — the wall is a personal,
    zero-pressure surface and turning it into a public ranking would change
    what it is (the arcade already owns competitive boards, opt-in by design).
    """
    # Same cached snapshot the employee wall reads. Within ONE process the two
    # therefore agree; across worker processes each holds its own 5-minute
    # snapshot, so a panel and a wall opened seconds apart can differ by
    # whatever unlocked in between. Acceptable for a slow-moving percentage,
    # and it self-heals inside the TTL — but "never differ" would be a lie.
    holders, headcount = await _rarity_snapshot(db)

    participants = int((await db.scalar(
        select(func.count(func.distinct(GamificationUnlock.person_id)))
        .select_from(GamificationUnlock)
        .join(
            IdentityPerson,
            and_(
                IdentityPerson.id == GamificationUnlock.person_id,
                *_active_person_filter(),
            ),
        )
    )) or 0)

    total_unlocks = sum(holders.values())

    badges = [
        {
            "code": b.code,
            "icon": b.icon,
            "category": b.category,
            "tier": b.tier,
            "name_key": b.name_key,
            "holders": holders.get(b.code, 0),
            # One decimal here, unlike the employee wall: the ops panel shows
            # the raw holder count beside it, so precision is useful and the
            # "never show 0% to a holder" floor is unnecessary. Keeping the
            # employee floor here would have silently changed an existing ops
            # figure (12.4% → 12%).
            "holder_percent": (
                round(100 * holders.get(b.code, 0) / headcount, 1)
                if headcount > 0 else None
            ),
        }
        for b in BADGES
    ]

    return {
        "badges": badges,
        "headcount": headcount,
        "participants": participants,
        "participation_percent": (
            round(100 * participants / headcount, 1) if headcount > 0 else None
        ),
        "total_unlocks": total_unlocks,
        "avg_per_participant": (
            round(total_unlocks / participants, 1) if participants > 0 else 0
        ),
        # Never-earned badges are the actionable signal here: a badge nobody
        # holds is either mis-tuned or measuring something that never happens.
        "unearned_codes": [b["code"] for b in badges if b["holders"] == 0],
        "generated_at": datetime.now(UTC).isoformat(),
    }


__all__ = [
    "compute_facts_for",
    "evaluate_and_unlock",
    "get_my_badges",
    "get_org_stats",
    "prefetch_unlocks",
]