"""放松一下 leaderboards + 称号 (titles) — the competition layer.

Three read shapes, all computed live from ``arcade_game_sessions`` (no
separate stats table; the (game_key, variant, score) and (game_key,
person_id) indexes keep them cheap at ~500 employees):

* **per-game best-score board** — top players by their best score in a
  game (optionally a variant / trivia category); powers "谁最厉害".
* **科举 knowledge board** — top players by lifetime correct trivia
  answers (SUM of correct_count over quiz_rush); the 白丁→状元 title
  ladder is derived from the same number.
* **per-category champions** — the live rank-1 of each trivia category's
  board, surfaced as "达人" titles (历史达人 …) the holder shows off.

Departed / soft-deleted people are excluded from every board and from the
rank math (same discipline as the security-quiz leaderboard).
"""
from __future__ import annotations

from typing import Any

from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.arcade import titles
from app.domains.arcade.catalog import GAMES_BY_KEY
from app.domains.arcade.models import SESSION_COMPLETED, ArcadeGameSession
from app.domains.arcade.trivia_models import TRIVIA_CATEGORIES
from app.identity.person_models import IdentityPerson

TRIVIA_GAME = "quiz_rush"


def _active_person_join(stmt):
    """Restrict to active, non-deleted people (headcount-style board)."""
    return stmt.join(
        IdentityPerson,
        and_(
            IdentityPerson.id == ArcadeGameSession.person_id,
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status != "left",
        ),
    )


async def _best_score_board(
    db: AsyncSession, person: IdentityPerson, *, game_key: str,
    variant: str | None, limit: int,
) -> dict[str, Any]:
    """Top players by best (max) completed score in game_key[/variant]."""
    best = func.max(ArcadeGameSession.score)
    base = _active_person_join(
        select(
            ArcadeGameSession.person_id.label("pid"),
            best.label("best"),
            func.count().label("plays"),
        ).where(
            ArcadeGameSession.game_key == game_key,
            ArcadeGameSession.status == SESSION_COMPLETED,
        )
    ).group_by(ArcadeGameSession.person_id)
    if variant:
        base = base.where(ArcadeGameSession.variant == variant)
    sub = base.subquery()

    # Deterministic total order (best DESC, plays ASC, pid ASC) so the
    # displayed rank and my_rank agree even on ties (mirrors the security-
    # quiz board's ordered-tiebreak rank math).
    rows = (await db.execute(
        select(
            sub.c.pid, sub.c.best, sub.c.plays,
            IdentityPerson.display_name, IdentityPerson.avatar_url,
        )
        .join(IdentityPerson, IdentityPerson.id == sub.c.pid)
        .order_by(sub.c.best.desc(), sub.c.plays.asc(), sub.c.pid.asc())
        .limit(max(1, min(limit, 100)))
    )).all()
    top = [
        {
            "rank": i + 1,
            "person_id": str(pid),
            "display_name": name or "—",
            "avatar_url": avatar,
            "score": int(best or 0),
            "plays": int(plays or 0),
            "is_me": pid == person.id,
        }
        for i, (pid, best, plays, name, avatar) in enumerate(rows)
    ]

    # Caller's own best + rank. Count players strictly AHEAD under the SAME
    # order as the displayed list (best DESC, plays ASC, pid ASC) so a tie
    # doesn't give two people the same rank while they render at different
    # positions.
    my_row = (await db.execute(
        select(sub.c.best, sub.c.plays).where(sub.c.pid == person.id)
    )).first()
    my_best = int((my_row.best if my_row else 0) or 0)
    my_plays = int((my_row.plays if my_row else 0) or 0)
    my_rank = None
    if my_best > 0:
        ahead = await db.scalar(
            select(func.count()).select_from(sub).where(
                or_(
                    sub.c.best > my_best,
                    and_(sub.c.best == my_best, sub.c.plays < my_plays),
                    and_(
                        sub.c.best == my_best,
                        sub.c.plays == my_plays,
                        sub.c.pid < person.id,
                    ),
                )
            )
        )
        my_rank = int(ahead or 0) + 1
    return {"top": top, "my_rank": my_rank, "my_best": my_best}


async def _knowledge_board(
    db: AsyncSession, person: IdentityPerson, *, limit: int,
) -> dict[str, Any]:
    """科举 board — top players by lifetime correct trivia answers, each
    carrying their imperial-exam title."""
    total = func.sum(func.coalesce(ArcadeGameSession.correct_count, 0))
    sub = _active_person_join(
        select(
            ArcadeGameSession.person_id.label("pid"),
            total.label("total"),
        ).where(
            ArcadeGameSession.game_key == TRIVIA_GAME,
            ArcadeGameSession.status == SESSION_COMPLETED,
        )
    ).group_by(ArcadeGameSession.person_id).subquery()

    # Deterministic order (total DESC, pid ASC) so displayed rank == my_rank
    # even when players are tied on total correct.
    rows = (await db.execute(
        select(
            sub.c.pid, sub.c.total,
            IdentityPerson.display_name, IdentityPerson.avatar_url,
        )
        .join(IdentityPerson, IdentityPerson.id == sub.c.pid)
        .where(sub.c.total > 0)
        .order_by(sub.c.total.desc(), sub.c.pid.asc())
        .limit(max(1, min(limit, 100)))
    )).all()
    top = [
        {
            "rank": i + 1,
            "person_id": str(pid),
            "display_name": name or "—",
            "avatar_url": avatar,
            "total_correct": int(total or 0),
            "title": titles.title_for(int(total or 0)),
            "is_me": pid == person.id,
        }
        for i, (pid, total, name, avatar) in enumerate(rows)
    ]
    my_total = int((await db.scalar(
        select(func.sum(func.coalesce(ArcadeGameSession.correct_count, 0)))
        .where(
            ArcadeGameSession.person_id == person.id,
            ArcadeGameSession.game_key == TRIVIA_GAME,
            ArcadeGameSession.status == SESSION_COMPLETED,
        )
    )) or 0)
    my_rank = None
    if my_total > 0:
        ahead = await db.scalar(
            select(func.count()).select_from(sub).where(
                or_(
                    sub.c.total > my_total,
                    and_(sub.c.total == my_total, sub.c.pid < person.id),
                )
            )
        )
        my_rank = int(ahead or 0) + 1
    return {
        "top": top,
        "my_rank": my_rank,
        "my_total_correct": my_total,
        "my_title": titles.progress(my_total),
    }


async def get_leaderboard(
    db: AsyncSession, person: IdentityPerson, *, board: str,
    game_key: str | None = None, variant: str | None = None, limit: int = 20,
) -> dict[str, Any]:
    """Dispatch: ``board='knowledge'`` → 科举 board; ``board='game'`` →
    per-game (optionally per-variant) best-score board."""
    if board == "knowledge":
        data = await _knowledge_board(db, person, limit=limit)
        data["board"] = "knowledge"
        return data
    if game_key not in GAMES_BY_KEY:
        from app.core.exceptions import NotFoundError
        raise NotFoundError("arcade_game", game_key or "")
    spec = GAMES_BY_KEY[game_key]
    if not spec.competitive:
        # Public-solution games (舒尔特/记忆翻牌/色词) ship their answer to
        # the client to render, so a cross-player best-score board would be
        # scriptable. They are personal-best only — no competitive board.
        from app.core.exceptions import AppException
        raise AppException(
            code="ARCADE_NO_LEADERBOARD",
            message="This game has no competitive leaderboard.",
            status_code=422,
        )
    if variant and spec.variants and variant not in spec.variants:
        variant = None  # ignore unknown variant → whole-game board
    data = await _best_score_board(
        db, person, game_key=game_key, variant=variant, limit=limit,
    )
    data["board"] = "game"
    data["game_key"] = game_key
    data["variant"] = variant
    return data


async def _category_champion_ids(db: AsyncSession) -> dict[str, str]:
    """The rank-1 person_id for each trivia category's best-score board
    (only categories with any play). One grouped query per category is
    fine (6 small queries); returns {category: person_id}."""
    champ: dict[str, str] = {}
    for cat in TRIVIA_CATEGORIES:
        # Tie-break EXACTLY like the displayed per-game board
        # (_best_score_board: best DESC, plays ASC, pid ASC) so the 达人
        # title always lands on whoever renders as that category's rank-1.
        # Ordering by score alone let the title flip between reads and
        # disagree with the board when two players tied on best score.
        row = (await db.execute(
            _active_person_join(
                select(
                    ArcadeGameSession.person_id.label("pid"),
                    func.max(ArcadeGameSession.score).label("best"),
                    func.count().label("plays"),
                ).where(
                    ArcadeGameSession.game_key == TRIVIA_GAME,
                    ArcadeGameSession.variant == cat,
                    ArcadeGameSession.status == SESSION_COMPLETED,
                )
            )
            .group_by(ArcadeGameSession.person_id)
            .order_by(
                func.max(ArcadeGameSession.score).desc(),
                func.count().asc(),
                ArcadeGameSession.person_id.asc(),
            )
            .limit(1)
        )).first()
        if row is not None and (row.best or 0) > 0:
            champ[cat] = str(row.pid)
    return champ


async def get_my_titles(db: AsyncSession, person: IdentityPerson) -> dict[str, Any]:
    """The caller's 科举 title + progress, plus any per-category champion
    titles they currently hold (rank-1 of that category). Powers the arcade
    home 称号 strip."""
    my_total = int((await db.scalar(
        select(func.sum(func.coalesce(ArcadeGameSession.correct_count, 0)))
        .where(
            ArcadeGameSession.person_id == person.id,
            ArcadeGameSession.game_key == TRIVIA_GAME,
            ArcadeGameSession.status == SESSION_COMPLETED,
        )
    )) or 0)
    champ = await _category_champion_ids(db)
    my_champion_categories = [c for c, pid in champ.items() if pid == str(person.id)]
    return {
        "knowledge_title": titles.progress(my_total),
        "champion_categories": my_champion_categories,
    }


__all__ = ["get_leaderboard", "get_my_titles"]