"""放松一下 point economy — the single place balance math happens.

Earning is derived from 安全每日一题 answers and reconciled **lazily on
read** (opening the arcade / starting a game): no hook inside the quiz
domain, no cron, and re-running is idempotent (watermark + unique ledger
key). Historical answers before the wallet's first touch are granted as a
one-time capped backfill so long-time players start with tickets.

Spending (game entry) and the daily play cap are enforced here inside the
same wallet-row lock, so a person can never go negative or exceed the cap
via concurrent requests.

Every function expects to run inside the caller's transaction; callers
commit (HTTP request scope auto-commits, MCP uses mutates=True).
"""
from __future__ import annotations

import logging
import uuid
from dataclasses import asdict, dataclass
from datetime import date, 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
from app.core.working_hours import resolve_timezone
from app.domains.arcade.models import (
    REASON_DAILY_QUIZ,
    REASON_GAME_ENTRY,
    REASON_QUIZ_BACKFILL,
    ArcadePointsLedger,
    ArcadeWallet,
)
from app.domains.security_quiz.models import SecQuizAnswer
from app.identity.person_models import IdentityPerson
from app.integrations.models import PlatformSetting

logger = logging.getLogger(__name__)

# ── Economy DEFAULTS (the seed values; live values are admin-tunable via
#    the ``arcade_economy`` platform setting — see EconomyConfig below) ──
POINTS_CORRECT = 10          # daily quiz answered correctly
POINTS_WRONG = 2             # participation credit for a wrong answer
POINTS_CORRECT_STREAK = 15   # correct answer while on a >=7-day streak
STREAK_THRESHOLD = 7
BACKFILL_CAP = 100           # one-time historical grant ceiling
ENTRY_COST = 3               # per game round
DAILY_PLAY_CAP = 3           # hard cap, points can't override it

# The single platform-setting key holding the whole (JSON) economy config.
ECONOMY_SETTING_KEY = "arcade_economy"


@dataclass(frozen=True)
class EconomyConfig:
    """The live, admin-tunable point economy. Defaults mirror the module
    constants; an admin can retune any field via the ``arcade_economy``
    platform setting without a deploy (e.g. bump the daily cap for a
    test window, or make correct answers worth more during a campaign)."""

    points_correct: int = POINTS_CORRECT
    points_wrong: int = POINTS_WRONG
    points_correct_streak: int = POINTS_CORRECT_STREAK
    streak_threshold: int = STREAK_THRESHOLD
    backfill_cap: int = BACKFILL_CAP
    entry_cost: int = ENTRY_COST
    daily_play_cap: int = DAILY_PLAY_CAP


DEFAULT_ECONOMY = EconomyConfig()

# Per-field sane bounds (inclusive). daily_play_cap floors at 1 so the arcade
# can never be globally soft-bricked to 0 plays/day by a bad edit.
ECONOMY_BOUNDS: dict[str, tuple[int, int]] = {
    "points_correct": (0, 100_000),
    "points_wrong": (0, 100_000),
    "points_correct_streak": (0, 100_000),
    "streak_threshold": (1, 3_650),
    "backfill_cap": (0, 10_000_000),
    "entry_cost": (0, 100_000),
    "daily_play_cap": (1, 100_000),
}


def _coerce_in_bounds(field: str, raw: Any, fallback: int) -> int:
    """Best-effort read: out-of-range / non-int → keep the fallback. Used on
    the READ path so a malformed stored value never 500s the arcade."""
    lo, hi = ECONOMY_BOUNDS[field]
    try:
        v = int(raw)
    except (TypeError, ValueError):
        return fallback
    return v if lo <= v <= hi else fallback


def economy_from_raw(raw: Any) -> EconomyConfig:
    if not isinstance(raw, dict):
        return DEFAULT_ECONOMY
    return EconomyConfig(**{
        f: _coerce_in_bounds(f, raw.get(f), getattr(DEFAULT_ECONOMY, f))
        for f in ECONOMY_BOUNDS
    })


def economy_to_dict(cfg: EconomyConfig) -> dict[str, int]:
    return asdict(cfg)


async def get_economy_config(
    db: AsyncSession, *, session: AsyncSession | None = None,
) -> EconomyConfig:
    """The live economy (fails OPEN to defaults on missing/malformed row)."""
    s = session or db
    row = await s.get(PlatformSetting, ECONOMY_SETTING_KEY)
    return economy_from_raw(row.value if row is not None else None)


async def set_economy_config(
    db: AsyncSession, updates: dict[str, Any], *, actor_id: uuid.UUID | None = None,
) -> EconomyConfig:
    """Merge ``updates`` over the CURRENT config and persist. Only the fields
    present in ``updates`` change; each provided value is STRICTLY validated
    (unknown key / non-int / out-of-range → 422) so an admin form gets a
    clear error instead of a silently-ignored edit."""
    current = await get_economy_config(db)
    merged = economy_to_dict(current)
    for key, raw in updates.items():
        if key not in ECONOMY_BOUNDS:
            raise AppException(
                code="ARCADE_ECONOMY_UNKNOWN_FIELD",
                message=f"Unknown economy field: {key}",
                status_code=422,
            )
        lo, hi = ECONOMY_BOUNDS[key]
        try:
            v = int(raw)
        except (TypeError, ValueError):
            raise AppException(
                code="ARCADE_ECONOMY_INVALID_VALUE",
                message=f"{key} must be an integer.",
                status_code=422,
            ) from None
        if not (lo <= v <= hi):
            raise AppException(
                code="ARCADE_ECONOMY_OUT_OF_RANGE",
                message=f"{key} must be between {lo} and {hi}.",
                status_code=422,
            )
        merged[key] = v

    new_cfg = EconomyConfig(**merged)
    row = await db.get(PlatformSetting, ECONOMY_SETTING_KEY)
    if row is None:
        row = PlatformSetting(
            key=ECONOMY_SETTING_KEY,
            value=economy_to_dict(new_cfg),
            description="放松一下 经济参数（答题积分 / 每日局数 / 每局入场费 …）",
        )
        db.add(row)
    else:
        row.value = economy_to_dict(new_cfg)
    if actor_id is not None:
        row.updated_by = actor_id
    await db.flush()
    return new_cfg


def local_today(person: IdentityPerson) -> date:
    return datetime.now(resolve_timezone(person)).date()


async def get_wallet_locked(db: AsyncSession, person: IdentityPerson) -> ArcadeWallet:
    """Fetch-or-create the person's wallet with a row lock held.

    The wallet row is the per-person serialization point for all balance
    math. Creation races resolve via the PK: the loser's INSERT trips the
    duplicate key inside a SAVEPOINT and re-reads (locked) the winner.
    """
    wallet = (await db.execute(
        select(ArcadeWallet)
        .where(ArcadeWallet.person_id == person.id)
        .with_for_update()
    )).scalar_one_or_none()
    if wallet is not None:
        return wallet

    try:
        async with db.begin_nested():
            wallet = ArcadeWallet(
                person_id=person.id, balance=0, total_earned=0, total_spent=0,
                quiz_credited_through=None,
            )
            db.add(wallet)
            await db.flush()
    except IntegrityError:
        wallet = (await db.execute(
            select(ArcadeWallet)
            .where(ArcadeWallet.person_id == person.id)
            .with_for_update()
        )).scalar_one_or_none()
        if wallet is None:  # genuinely not a race — surface it
            raise
        return wallet
    # Fresh insert in our own txn — already exclusively ours.
    return wallet


def _streak_ending_at(correct_dates: set[date], d: date) -> int:
    """Length of the consecutive-correct-day run ending at ``d`` inclusive
    (0 when ``d`` itself wasn't answered correctly)."""
    run = 0
    cursor = d
    while cursor in correct_dates:
        run += 1
        cursor -= timedelta(days=1)
    return run


async def _try_insert_ledger(
    db: AsyncSession, *, person_id: uuid.UUID, delta: int, balance_after: int,
    reason: str, ref_id: str,
) -> bool:
    """Insert one ledger row SAVEPOINT-isolated. Returns False when the
    unique key says this credit already exists (concurrent path won).

    The wallet object is deliberately never touched inside the SAVEPOINT —
    a nested rollback expires objects dirtied within it, which would
    silently discard (or double-apply) balance math."""
    try:
        async with db.begin_nested():
            db.add(ArcadePointsLedger(
                id=uuid.uuid4(), person_id=person_id, delta=delta,
                balance_after=balance_after, reason=reason, ref_id=ref_id,
            ))
            await db.flush()
        return True
    except IntegrityError:
        return False


async def reconcile_quiz_credits(
    db: AsyncSession, person: IdentityPerson, wallet: ArcadeWallet,
    cfg: EconomyConfig | None = None,
) -> int:
    """Credit any not-yet-credited daily-quiz answers. Returns points added.

    Caller must hold the wallet lock. Idempotent: the watermark bounds the
    scan and UNIQUE(person, reason, ref) bounds the writes. All wallet
    mutations happen outside SAVEPOINTs (see ``_try_insert_ledger``).

    ``cfg`` carries the live (admin-tunable) point rates; callers that
    already resolved it pass it in, otherwise defaults are used.
    """
    if cfg is None:
        cfg = DEFAULT_ECONOMY
    today = local_today(person)
    earned = 0
    running_balance = wallet.balance

    if wallet.quiz_credited_through is None:
        # First touch: one-time capped backfill over answers BEFORE today;
        # today's answer (if any) is credited by the incremental pass below.
        total_correct_before = int((await db.scalar(
            select(func.count()).select_from(SecQuizAnswer).where(
                SecQuizAnswer.person_id == person.id,
                SecQuizAnswer.quiz_date < today,
                SecQuizAnswer.is_correct.is_(True),
            )
        )) or 0)
        grant = min(cfg.backfill_cap, total_correct_before * cfg.points_correct)
        if grant > 0 and await _try_insert_ledger(
            db, person_id=wallet.person_id, delta=grant,
            balance_after=running_balance + grant,
            reason=REASON_QUIZ_BACKFILL, ref_id=today.isoformat(),
        ):
            running_balance += grant
            earned += grant
        wallet.quiz_credited_through = today - timedelta(days=1)

    # Incremental: answers newer than the watermark (normally just today's).
    pending = (await db.execute(
        select(SecQuizAnswer.quiz_date, SecQuizAnswer.is_correct)
        .where(
            SecQuizAnswer.person_id == person.id,
            SecQuizAnswer.quiz_date > wallet.quiz_credited_through,
            SecQuizAnswer.quiz_date <= today,
        )
        .order_by(SecQuizAnswer.quiz_date)
    )).all()
    if pending:
        # Streak math needs the full correct-date history once.
        correct_dates = set((await db.execute(
            select(SecQuizAnswer.quiz_date).where(
                SecQuizAnswer.person_id == person.id,
                SecQuizAnswer.is_correct.is_(True),
            )
        )).scalars().all())
        for quiz_date, is_correct in pending:
            if is_correct:
                streak = _streak_ending_at(correct_dates, quiz_date)
                delta = cfg.points_correct_streak if streak >= cfg.streak_threshold else cfg.points_correct
            else:
                delta = cfg.points_wrong
            if await _try_insert_ledger(
                db, person_id=wallet.person_id, delta=delta,
                balance_after=running_balance + delta,
                reason=REASON_DAILY_QUIZ, ref_id=quiz_date.isoformat(),
            ):
                running_balance += delta
                earned += delta
            wallet.quiz_credited_through = max(wallet.quiz_credited_through, quiz_date)

    if earned:
        wallet.balance = running_balance
        wallet.total_earned += earned
    return earned


async def spend_game_entry(
    db: AsyncSession, wallet: ArcadeWallet, *, session_id: uuid.UUID, cost: int,
) -> None:
    """Deduct the entrance fee. Caller must hold the wallet lock and have
    reconciled credits first (so a same-day quiz answer counts)."""
    if wallet.balance < cost:
        raise AppException(
            code="ARCADE_INSUFFICIENT_POINTS",
            message="Not enough points — answer the daily security quiz to earn more.",
            status_code=422,
        )
    wallet.balance -= cost
    wallet.total_spent += cost
    db.add(ArcadePointsLedger(
        id=uuid.uuid4(), person_id=wallet.person_id, delta=-cost,
        balance_after=wallet.balance, reason=REASON_GAME_ENTRY,
        ref_id=str(session_id),
    ))


def serialize_wallet(wallet: ArcadeWallet) -> dict[str, Any]:
    return {
        "balance": wallet.balance,
        "total_earned": wallet.total_earned,
        "total_spent": wallet.total_spent,
    }


__all__ = [
    "POINTS_CORRECT", "POINTS_WRONG", "POINTS_CORRECT_STREAK", "STREAK_THRESHOLD",
    "BACKFILL_CAP", "ENTRY_COST", "DAILY_PLAY_CAP",
    "ECONOMY_SETTING_KEY", "ECONOMY_BOUNDS", "EconomyConfig", "DEFAULT_ECONOMY",
    "economy_from_raw", "economy_to_dict", "get_economy_config", "set_economy_config",
    "local_today", "get_wallet_locked", "reconcile_quiz_credits",
    "spend_game_entry", "serialize_wallet",
]