"""放松一下 (Security Arcade) — points wallet + ledger + game sessions.

Three tables backing the employee arcade:

* ``arcade_wallets``        one row per person: transactional balance +
                            quiz-credit watermark (the per-person lock anchor)
* ``arcade_points_ledger``  append-only audit trail of every earn/spend;
                            UNIQUE(person, reason, ref) makes credits idempotent
* ``arcade_game_sessions``  server-judged game rounds (secret payload +
                            server-computed score); UNIQUE(person, date, seq)
                            backstops the daily play cap

Person FKs use ``id_type_for`` / ``table_kw_for`` so the inline FKs match
identity_persons' charset/collation on both MySQL 8 (CI) and TiDB (prod)
(error 3780 guard). No cron seeded — points reconcile lazily on read and
sessions expire lazily; there is nothing time-driven to schedule.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import id_type_for, table_kw_for

revision = "411"
down_revision = "410"
branch_labels = None
depends_on = None

WALLETS = "arcade_wallets"
LEDGER = "arcade_points_ledger"
SESSIONS = "arcade_game_sessions"


def _table_exists(bind, name: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n"
        ),
        {"n": name},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()
    pid = id_type_for("identity_persons")
    kw = table_kw_for("identity_persons")

    if not _table_exists(bind, WALLETS):
        op.create_table(
            WALLETS,
            sa.Column("person_id", pid, primary_key=True),
            sa.Column("balance", sa.Integer(), nullable=False, server_default="0"),
            sa.Column("total_earned", sa.Integer(), nullable=False, server_default="0"),
            sa.Column("total_spent", sa.Integer(), nullable=False, server_default="0"),
            sa.Column("quiz_credited_through", sa.Date(), nullable=True),
            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
            sa.ForeignKeyConstraint(["person_id"], ["identity_persons.id"], name="fk_arcade_wallet_person"),
            **kw,
        )

    if not _table_exists(bind, LEDGER):
        op.create_table(
            LEDGER,
            sa.Column("id", pid, primary_key=True),
            sa.Column("person_id", pid, nullable=False),
            sa.Column("delta", sa.Integer(), nullable=False),
            sa.Column("balance_after", sa.Integer(), nullable=False),
            sa.Column("reason", sa.String(32), nullable=False),
            sa.Column("ref_id", sa.String(64), nullable=False),
            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
            sa.ForeignKeyConstraint(["person_id"], ["identity_persons.id"], name="fk_arcade_ledger_person"),
            sa.UniqueConstraint("person_id", "reason", "ref_id", name="uq_arcade_ledger_ref"),
            **kw,
        )
        op.create_index("ix_arcade_ledger_person", LEDGER, ["person_id", "created_at"])

    if not _table_exists(bind, SESSIONS):
        op.create_table(
            SESSIONS,
            sa.Column("id", pid, primary_key=True),
            sa.Column("person_id", pid, nullable=False),
            sa.Column("game_key", sa.String(32), nullable=False),
            sa.Column("status", sa.String(16), nullable=False, server_default="active"),
            sa.Column("play_date", sa.Date(), nullable=False),
            sa.Column("play_seq", sa.Integer(), nullable=False),
            sa.Column("cost_points", sa.Integer(), nullable=False),
            sa.Column("payload", sa.JSON(), nullable=False),
            sa.Column("result", sa.JSON(), nullable=True),
            sa.Column("score", sa.Integer(), nullable=True),
            sa.Column("duration_ms", sa.Integer(), nullable=True),
            sa.Column("started_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
            sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
            sa.ForeignKeyConstraint(["person_id"], ["identity_persons.id"], name="fk_arcade_sess_person"),
            sa.UniqueConstraint("person_id", "play_date", "play_seq", name="uq_arcade_sess_seq"),
            **kw,
        )
        op.create_index("ix_arcade_sess_person", SESSIONS, ["person_id", "play_date"])
        op.create_index("ix_arcade_sess_game_score", SESSIONS, ["game_key", "score"])


def downgrade() -> None:
    bind = op.get_bind()
    for name in (SESSIONS, LEDGER, WALLETS):
        if _table_exists(bind, name):
            op.drop_table(name)