ForbiddenStarsApp/backend/alembic/versions/0014_rating_inputs.py

"""Рейтинг (#23): раунд окончания, цели и миры участников, правило 9 раундов, last_standing.

Идемпотентна: на свежей БД столбцы и новый CHECK создаёт 0001 (create_all из актуальных
моделей) -> меняется только бэкфилл (на пустой БД он ничего не находит); на существующей
БД добавляет столбцы, расширяет CHECK причины победы (если он в БД есть) и проставляет
last_standing.

Столбцы nullable и задним числом не заполняются: NULL — «нет данных», рейтинг
подставляет вместо них типичные значения (docs/rating/rating-system.md, 4.8).

Revision ID: 0014_rating_inputs
Revises: 0013_user_token_version
Create Date: 2026-09-14
"""
from typing import Sequence, Union

import sqlalchemy as sa
from sqlalchemy import inspect

from alembic import op

revision: str = "0014_rating_inputs"
down_revision: Union[str, None] = "0013_user_token_version"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

CK_WIN_REASON = "ck_match_win_reason"
OLD_REASONS = "win_reason IS NULL OR win_reason IN ('objectives','worlds','plastic','resources')"
NEW_REASONS = (
    "win_reason IS NULL OR win_reason IN "
    "('objectives','worlds','plastic','resources','last_standing')"
)

# Причина last_standing ⇔ невыбывший участник ровно один (решение владельца по #22).
BACKFILL_LAST_STANDING = """
UPDATE matches SET win_reason = 'last_standing'
WHERE status = 'finished'
  AND (win_reason IS NULL OR win_reason <> 'last_standing')
  AND (SELECT COUNT(*) FROM match_participants mp
       WHERE mp.match_id = matches.id AND mp.eliminated = 0) = 1
"""


def _columns(insp, table: str) -> set[str]:
    return {c["name"] for c in insp.get_columns(table)}


def _win_reason_check(insp) -> str | None:
    for ck in insp.get_check_constraints("matches"):
        if ck.get("name") == CK_WIN_REASON:
            return ck["sqltext"]
    return None


def _recreate_matches_with_check(bind, drop_existing: bool, sqltext: str) -> None:
    """Замена CHECK в SQLite = пересоздание таблицы matches (batch copy-and-move).

    При включённых внешних ключах DROP старой таблицы выполнил бы неявный DELETE, и
    ON DELETE CASCADE унёс бы участников, вложения и черновики. Alembic из CLI работает
    без PRAGMA foreign_keys (её включает только движок приложения), а внутри транзакции
    PRAGMA не переключить — поэтому не рискуем и отказываемся с понятной ошибкой."""
    if bind.exec_driver_sql("PRAGMA foreign_keys").scalar():
        raise RuntimeError(
            "0014: PRAGMA foreign_keys=ON — пересоздание matches удалило бы участников "
            "каскадом. Запускайте миграции через `alembic upgrade head` (CLI)."
        )
    with op.batch_alter_table("matches", recreate="always") as b:
        if drop_existing:
            b.drop_constraint(CK_WIN_REASON, type_="check")
        b.create_check_constraint(CK_WIN_REASON, sqltext)


def upgrade() -> None:
    bind = op.get_bind()
    insp = inspect(bind)

    if "nine_rounds_rule" not in _columns(insp, "groups"):
        with op.batch_alter_table("groups") as b:
            b.add_column(
                sa.Column("nine_rounds_rule", sa.Boolean(), nullable=False, server_default="0")
            )

    match_cols = _columns(insp, "matches")
    with op.batch_alter_table("matches") as b:
        if "end_round" not in match_cols:
            b.add_column(sa.Column("end_round", sa.Integer(), nullable=True))
        if "nine_rounds_rule" not in match_cols:
            b.add_column(
                sa.Column("nine_rounds_rule", sa.Boolean(), nullable=False, server_default="0")
            )

    participant_cols = _columns(insp, "match_participants")
    with op.batch_alter_table("match_participants") as b:
        if "objectives" not in participant_cols:
            b.add_column(sa.Column("objectives", sa.Integer(), nullable=True))
        if "worlds" not in participant_cols:
            b.add_column(sa.Column("worlds", sa.Integer(), nullable=True))

    # БД, созданные до появления CHECK в моделях (0001 тогда был старше), ограничения
    # не имеют вовсе — last_standing им и так разрешён, пересоздавать таблицу незачем.
    check = _win_reason_check(inspect(bind))
    if check is not None and "last_standing" not in check:
        _recreate_matches_with_check(bind, drop_existing=True, sqltext=NEW_REASONS)

    op.execute(BACKFILL_LAST_STANDING)


def downgrade() -> None:
    bind = op.get_bind()
    # Прежний CHECK не знает last_standing: такие партии теряют признак (раньше их
    # записывали «по целям»).
    op.execute("UPDATE matches SET win_reason = 'objectives' WHERE win_reason = 'last_standing'")
    check = _win_reason_check(inspect(bind))
    if check is not None and "last_standing" in check:
        _recreate_matches_with_check(bind, drop_existing=True, sqltext=OLD_REASONS)

    insp = inspect(bind)
    participant_cols = _columns(insp, "match_participants")
    with op.batch_alter_table("match_participants") as b:
        for name in ("worlds", "objectives"):
            if name in participant_cols:
                b.drop_column(name)
    match_cols = _columns(insp, "matches")
    with op.batch_alter_table("matches") as b:
        for name in ("nine_rounds_rule", "end_round"):
            if name in match_cols:
                b.drop_column(name)
    if "nine_rounds_rule" in _columns(insp, "groups"):
        with op.batch_alter_table("groups") as b:
            b.drop_column("nine_rounds_rule")