live-hub-py/web/backend/migrations/versions/20261003_0042_account_lifecycle.py

"""账号全生命周期：把登录态与生命周期状态落到 accounts，并扩展账号事件台账。

## 为什么

1. **CK 不该寄存在登录记录上**。`login_tasks.cookie` 是账号的**登录态**，不是「某次登录的
   产物」。删一条历史记录就等于删掉账号的登录态 —— 这是 2026-10-03 误删事故的根因。
2. **CK 健康与重登进度** 同样挂在「碰巧最后一次登录那一行」上；一个账号有多行持 CK 时
   （实测存在）谁是权威并不明确。
3. **生命周期状态根本不存在**，只能靠「有没有持 CK 的记录」反推。
4. `account_events` 除了时间线，还要承担**业务台账**（开通/积分/兑换/充值），
   需要可检索的客体键与索引来支撑「365 天同皮肤」这类规则。

设计依据：`docs/设计-账号全生命周期.md`。

## 安全性（本迁移只加列、只读源表，不删不改任何旧数据）

- **分块回填**：1.4 万账号、单账号 CK 约 1.7KB，一次性 UPDATE 会长时间持锁；
- **幂等**：只在 `accounts.cookie` 为空时写，可重跑；也给回填后新写入的 CK 让路；
- **自校验 fail closed**：回填不完整就让迁移抛错（宁可起不来，也不要带着半份 CK 上线）；
- **值级校验**：不只比条数 —— 条数相等仍可能张冠李戴（A 账号拿到 B 账号的 CK）。

旧列本次**全部保留**，下一次发版再 drop。
"""

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "20261003_0042"
down_revision: str | None = "20261003_0041"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

# 一次回填多少账号。太小则往返多，太大则单事务持锁久。
BACKFILL_CHUNK = 200

_ACCOUNT_COLUMNS: tuple[tuple[str, sa.types.TypeEngine, bool], ...] = (
    # ── 登录态：CK 本体 ──
    ("cookie", sa.Text(), False),
    ("cookie_updated_at", sa.DateTime(), True),
    ("cookie_source", sa.String(length=16), False),
    # ── 登录态：CK 健康 ──
    ("ck_check_status", sa.String(length=16), False),
    ("ck_check_result", sa.JSON(), True),
    ("ck_checked_at", sa.DateTime(), True),
    # ── 登录态：重登退避进度 ──
    ("relogin_attempts", sa.Integer(), False),
    ("relogin_next_at", sa.DateTime(), True),
    ("relogin_last_error", sa.String(length=512), False),
    ("relogin_failure_kind", sa.String(length=16), False),
    # ── 生命周期状态 ──
    ("status", sa.String(length=16), False),
    ("status_reason", sa.String(length=64), False),
    ("status_changed_at", sa.DateTime(), True),
    ("disabled_at", sa.DateTime(), True),
    ("archived_at", sa.DateTime(), True),
)

# 列名 → server_default（NOT NULL 的列必须有，否则已有行无法满足约束）
# ⚠️ `cookie` 是 TEXT，**不能**有 server_default（MySQL 报 1101：
#    BLOB/TEXT/GEOMETRY/JSON column can't have a default value）。
#    已有行由 ALTER TABLE 隐式填 ''，这里不配默认值即可。
_COLUMN_DEFAULTS: dict[str, str] = {
    "cookie_source": "",
    "ck_check_status": "",
    "relogin_attempts": "0",
    "relogin_last_error": "",
    "relogin_failure_kind": "",
    "status": "imported",
    "status_reason": "",
}

_EVENT_COLUMNS: tuple[tuple[str, sa.types.TypeEngine, bool], ...] = (
    ("category", sa.String(length=16), False),
    ("subject_type", sa.String(length=16), False),
    ("subject_key", sa.String(length=96), False),
    ("actor_type", sa.String(length=16), False),
    ("actor_id", sa.Integer(), True),
)


def _tables(bind) -> set[str]:
    return set(sa.inspect(bind).get_table_names())


def _columns(bind, table: str) -> set[str]:
    return {item["name"] for item in sa.inspect(bind).get_columns(table)}


def _indexes(bind, table: str) -> set[str]:
    return {item["name"] for item in sa.inspect(bind).get_indexes(table)}


def _add_columns(bind, table: str, columns) -> None:
    """幂等加列：已存在就跳过。"""
    existing = _columns(bind, table)
    for name, col_type, nullable in columns:
        if name in existing:
            continue
        default = _COLUMN_DEFAULTS.get(name)
        op.add_column(
            table,
            sa.Column(
                name,
                col_type,
                nullable=nullable,
                server_default=default if not nullable else None,
            ),
        )


def _add_index(bind, table: str, name: str, columns: list[str]) -> None:
    if name in _indexes(bind, table):
        return
    op.create_index(name, table, columns)


def _backfill_login_state(bind) -> None:
    """把每个账号「最近一条持有 CK 的记录」的 CK 与状态搬到 accounts。

    只读 `login_tasks`，不删不改。分块执行。
    """
    from sqlalchemy import bindparam

    pairs = bind.execute(
        sa.text(
            """
            SELECT account_id, MAX(id) AS last_id
            FROM login_tasks
            WHERE cookie IS NOT NULL AND cookie <> ''
            GROUP BY account_id
            ORDER BY account_id
            """
        )
    ).all()
    if not pairs:
        return

    account_ids = [int(row[0]) for row in pairs]
    stmt = sa.text(
        """
        UPDATE accounts a
        JOIN (
            SELECT account_id, cookie, finished_at, ck_check_status,
                   ck_check_result, ck_checked_at, relogin_attempts,
                   relogin_next_at, relogin_last_error, relogin_failure_kind
            FROM (
                SELECT lt.account_id, lt.cookie, lt.finished_at, lt.ck_check_status,
                       lt.ck_check_result, lt.ck_checked_at, lt.relogin_attempts,
                       lt.relogin_next_at, lt.relogin_last_error,
                       lt.relogin_failure_kind,
                       ROW_NUMBER() OVER (
                           PARTITION BY lt.account_id ORDER BY lt.id DESC
                       ) AS rn
                FROM login_tasks lt
                WHERE lt.cookie IS NOT NULL AND lt.cookie <> ''
                  AND lt.account_id IN :ids
            ) ranked
            WHERE ranked.rn = 1
        ) src ON src.account_id = a.id
        SET a.cookie               = src.cookie,
            a.cookie_updated_at    = src.finished_at,
            a.ck_check_status      = COALESCE(src.ck_check_status, ''),
            a.ck_check_result      = src.ck_check_result,
            a.ck_checked_at        = src.ck_checked_at,
            a.relogin_attempts     = COALESCE(src.relogin_attempts, 0),
            a.relogin_next_at      = src.relogin_next_at,
            a.relogin_last_error   = COALESCE(src.relogin_last_error, ''),
            a.relogin_failure_kind = COALESCE(src.relogin_failure_kind, '')
        WHERE a.cookie IS NULL OR a.cookie = ''
        """
    ).bindparams(bindparam("ids", expanding=True))

    for start in range(0, len(account_ids), BACKFILL_CHUNK):
        chunk = account_ids[start : start + BACKFILL_CHUNK]
        bind.execute(stmt, {"ids": chunk})


def _backfill_lifecycle_status(bind) -> None:
    """按登录态推导生命周期状态。

    ⚠️ **这里的 CASE 必须与 `services/account_lifecycle.derive_status()` 逐字对应。**
    两边规则一旦分叉，迁移刚跑完就会有一批账号的落库状态与运行时推导不一致
    （表现为「回填说是 A、下一轮巡检又改成 B」）。

    状态取值域 6 个：imported / active / invalid / manual / disabled / archived。
    **没有 `relogin`** —— 「退避重试中」等价于 `invalid + relogin_next_at 非空`，
    是 invalid 的调度细节，不单独成状态（S3 定稿时砍掉的）。
    """
    bind.execute(
        sa.text(
            """
            UPDATE accounts
            SET status = CASE
                    WHEN cookie IS NULL OR cookie = '' THEN 'imported'
                    WHEN COALESCE(ck_check_status, '') <> 'invalid' THEN 'active'
                    WHEN COALESCE(relogin_failure_kind, '') IN ('captcha', 'credential')
                        THEN 'manual'
                    WHEN COALESCE(relogin_attempts, 0) > 0 AND relogin_next_at IS NULL
                        THEN 'manual'
                    ELSE 'invalid'
                END,
                status_changed_at = COALESCE(status_changed_at, updated_at, created_at)
            WHERE status = 'imported' OR status IS NULL
            """
        )
    )


def _verify_backfill(bind) -> None:
    """自校验：不完整 / 张冠李戴都直接抛错，让应用起不来（fail closed）。

    `expected` 必须运行时算 —— 这个数在长（重登在跑）。
    """
    expected = int(
        bind.execute(
            sa.text(
                """
                SELECT COUNT(DISTINCT account_id) FROM login_tasks
                WHERE cookie IS NOT NULL AND cookie <> ''
                """
            )
        ).scalar()
        or 0
    )
    actual = int(
        bind.execute(
            sa.text(
                "SELECT COUNT(*) FROM accounts WHERE cookie IS NOT NULL AND cookie <> ''"
            )
        ).scalar()
        or 0
    )
    if actual < expected:
        raise RuntimeError(
            f"CK 回填不完整：期望 {expected} 个账号，实际 {actual} 个。"
            "中止迁移 —— 宁可起不来，也不要带着半份 CK 上线。"
        )

    mismatched = int(
        bind.execute(
            sa.text(
                """
                SELECT COUNT(*) FROM accounts a
                JOIN (
                    SELECT account_id, cookie,
                           ROW_NUMBER() OVER (
                               PARTITION BY account_id ORDER BY id DESC
                           ) AS rn
                    FROM login_tasks
                    WHERE cookie IS NOT NULL AND cookie <> ''
                ) src ON src.account_id = a.id AND src.rn = 1
                WHERE a.cookie <> src.cookie
                """
            )
        ).scalar()
        or 0
    )
    if mismatched:
        raise RuntimeError(
            f"CK 值级校验失败：{mismatched} 个账号的 CK 与其最近一条持 CK 记录不一致，"
            "疑似张冠李戴。中止迁移。"
        )


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

    # 1. accounts 加列 + 索引
    _add_columns(bind, "accounts", _ACCOUNT_COLUMNS)
    _add_index(bind, "accounts", "ix_accounts_ck_checked_at", ["ck_checked_at"])
    _add_index(
        bind,
        "accounts",
        "ix_accounts_status_relogin_next",
        ["status", "relogin_next_at"],
    )
    _add_index(bind, "accounts", "ix_accounts_status", ["status"])

    # 2. account_events 扩展（台账列 + 复合索引）
    if "account_events" in _tables(bind):
        _add_columns(bind, "account_events", _EVENT_COLUMNS)
        _add_index(
            bind,
            "account_events",
            "ix_account_events_ledger",
            ["account_id", "event_type", "subject_key", "occurred_at"],
        )

    # 3. login_tasks 补「在途判定」索引：CK 搬到账号后，「在途」只能留在登录记录上，
    #    巡检/重登每轮都要按 (account_id, attempt_status) 过滤。
    if "login_tasks" in _tables(bind):
        _add_index(
            bind,
            "login_tasks",
            "ix_login_tasks_account_attempt_status",
            ["account_id", "attempt_status"],
        )

    # 4. 回填 + 自校验
    _backfill_login_state(bind)
    _backfill_lifecycle_status(bind)
    _verify_backfill(bind)


def downgrade() -> None:
    bind = op.get_bind()

    for name in (
        "ix_login_tasks_account_attempt_status",
        "ix_account_events_ledger",
        "ix_accounts_status",
        "ix_accounts_status_relogin_next",
        "ix_accounts_ck_checked_at",
    ):
        table = (
            "login_tasks"
            if name == "ix_login_tasks_account_attempt_status"
            else (
                "account_events" if name == "ix_account_events_ledger" else "accounts"
            )
        )
        if table in _tables(bind) and name in _indexes(bind, table):
            op.drop_index(name, table_name=table)

    if "account_events" in _tables(bind):
        existing = _columns(bind, "account_events")
        for name, _type, _nullable in reversed(_EVENT_COLUMNS):
            if name in existing:
                op.drop_column("account_events", name)

    existing = _columns(bind, "accounts")
    for name, _type, _nullable in reversed(_ACCOUNT_COLUMNS):
        if name in existing:
            op.drop_column("accounts", name)