feat(db): S1 迁移 0042 —— 账号全生命周期落库（登录态 + 状态机 + 事件台账）

账号不再依赖登录记录的第一步：把「账号自己的东西」搬到账号上。本提交**只加列、只读源表**，不改任何行为，单独部署安全。 accounts 新增 15 列： - 登录态：cookie / cookie_updated_at / cookie_source - CK 健康：ck_check_status / ck_check_result / ck_checked_at - 重登进度：relogin_attempts / relogin_next_at / relogin_last_error / relogin_failure_kind - 生命周期：status / status_reason / status_changed_at / disabled_at / archived_at - 索引：ck_checked_at、status、与 (status, relogin_next_at) account_events 新增 5 列（category / subject_type / subject_key / actor_type / actor_id）与台账复合索引 (account_id, event_type, subject_key, occurred_at) —— 支撑「365 天同皮肤」这类需要检索客体的规则，让事件表同时承担业务台账。 login_tasks 补 (account_id, attempt_status) 索引：CK 搬到账号后，「在途尝试」只能留在登录记录上，巡检/重登每轮都要按它过滤。回填做成生产级： - 只读 login_tasks、分块 200 账号、幂等（只在 accounts.cookie 为空时写）； - 窗口函数取「最近一条持 CK 的行」，不用 GROUP_CONCAT（受 max_len 限制会静默截断）； - 自校验 fail closed：回填不完整直接抛错让应用起不来，宁可起不来也不带半份 CK 上线； - 外加值级校验，防张冠李戴（条数相等仍可能 A 账号拿到 B 账号的 CK）。踩坑留档：cookie 是 TEXT，**不能有 server_default**（MySQL 1101： BLOB/TEXT/GEOMETRY/JSON column can not have a default value），模型与迁移两侧都要去掉。验证： - 空库整链 upgrade head + 模型对齐 + 索引存在 → 通过； - 本机开发库实跑 0041→0042：记录侧 4 个持 CK 账号 = 账号侧 4 个，值级不一致 0， CK 仍是密文（enc:v1: 前缀），状态分布 active 4 / imported 8； - downgrade 到 0041 再 upgrade：数据完整还原（这是 CK 丢失场景的安全网）； - 全量 pytest 364 passed / 6 skipped。
2026-10-04 13:45:59 +08:00
parent 3a227ffec1
commit e916a7015d
4 changed files with 432 additions and 1 deletions
@@ -26,7 +26,7 @@ TESTS_DIR = Path(__file__).resolve().parent
 PROJECT_ROOT = TESTS_DIR.parent
 VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"

-HEAD_REVISION = "20261003_0041"
+HEAD_REVISION = "20261003_0042"

 # 迁移新增、但不声明在模型里的复合查询索引。
 EXTRA_INDEXES = (
@@ -36,6 +36,9 @@ EXTRA_INDEXES = (
    "ix_login_tasks_status_finished_at_id",
    # 账号时间线按「账号 + 时间倒序」查，模型里不声明这个复合索引。
    "ix_account_events_account_occurred",
+    # CK 搬到账号后，「在途尝试」只能留在登录记录上，巡检/重登每轮按
+    # (account_id, attempt_status) 过滤 —— 模型里不声明这个索引。
+    "ix_login_tasks_account_attempt_status",
 )


@@ -150,6 +150,8 @@ _SENSITIVE_COLUMNS: tuple[tuple[str, str], ...] = (
    ("accounts", "password"),
    ("accounts", "email"),
    ("accounts", "email_password"),
+    # CK 已由 login_tasks 迁到账号上；两边都列上，回填后任一侧出现明文都能被兜住。
+    ("accounts", "cookie"),
    ("login_tasks", "cookie"),
    ("huya_accounts", "account_password"),
    ("huya_accounts", "cookie"),
@@ -0,0 +1,332 @@
+"""账号全生命周期：把登录态与生命周期状态落到 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:
+    """按登录态推导生命周期状态。
+
+    取值域见设计文档 §3.4。这里只用**已有的确定信息**推导，不做猜测：
+    有 CK 且未判失效 → active；判失效则按重登进度分流；无 CK → imported。
+    """
+    bind.execute(
+        sa.text(
+            """
+            UPDATE accounts
+            SET status = CASE
+                    WHEN cookie IS NULL OR cookie = '' THEN 'imported'
+                    WHEN ck_check_status = 'invalid'
+                         AND relogin_failure_kind IN ('captcha', 'credential')
+                        THEN 'manual'
+                    WHEN ck_check_status = 'invalid'
+                         AND relogin_next_at IS NOT NULL THEN 'relogin'
+                    WHEN ck_check_status = 'invalid' THEN 'invalid'
+                    ELSE 'active'
+                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)
@@ -127,6 +127,66 @@ class Account(Base):
        UTCDateTime, default=_utcnow, onupdate=_utcnow
    )

+    # ── 登录态：CK 本体（账号自己的凭证，不再寄存在 login_tasks 上）──────────
+    # 历史问题：CK 曾只存在于 login_tasks.cookie，删一条登录历史就等于删掉账号的
+    # 登录态（2026-10-03 误删事故的根因）。CK 是**账号级凭证**，必须挂在账号上。
+    # 注意：**不能给这列加 server_default** —— MySQL 不允许 TEXT/BLOB/JSON 列有默认值
+    # （错误 1101）。已有行由 ALTER TABLE 隐式填 ''，新建行由 Python 侧 default 兜。
+    cookie: Mapped[str] = mapped_column(EncryptedText(), nullable=False, default="")
+    cookie_updated_at: Mapped[datetime | None] = mapped_column(
+        UTCDateTime, nullable=True
+    )
+    # 这份 CK 是怎么来的：web / app / sms / import（空 = 历史回填，来源未知）
+    cookie_source: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="", server_default=""
+    )
+
+    # ── 登录态：CK 健康（巡检/人工检测的当前结论）──────────────────────────
+    ck_check_status: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="", server_default=""
+    )  # valid / invalid / error / 空 = 未检测
+    ck_check_result: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    ck_checked_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True)
+
+    # ── 登录态：自动重登的退避进度 ─────────────────────────────────────────
+    relogin_attempts: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default="0"
+    )
+    relogin_next_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True)
+    relogin_last_error: Mapped[str] = mapped_column(
+        String(512), nullable=False, default="", server_default=""
+    )
+    # 失败分型：captcha / credential 转人工，email_code / email_quota /
+    # interrupted / network 自动重试。空 = 没失败过。
+    relogin_failure_kind: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="", server_default=""
+    )
+
+    # ── 生命周期状态 ───────────────────────────────────────────────────────
+    # 取值域见 docs/设计-账号全生命周期.md §3.4。**写入必须收口到
+    # services/account_lifecycle.py**，禁止各处直接赋值 —— accounts.bind_status
+    # 就是「谁都能写、语义没人管」的反面教材（36 种取值互相覆盖）。
+    status: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="imported", server_default="imported"
+    )
+    status_reason: Mapped[str] = mapped_column(
+        String(64), nullable=False, default="", server_default=""
+    )
+    status_changed_at: Mapped[datetime | None] = mapped_column(
+        UTCDateTime, nullable=True
+    )
+    disabled_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True)
+    archived_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True)
+
+    __table_args__ = (
+        # 巡检按「最久没查的优先」选行
+        Index("ix_accounts_ck_checked_at", "ck_checked_at"),
+        # 重登按状态 + 退避时间选行
+        Index("ix_accounts_status_relogin_next", "status", "relogin_next_at"),
+        # 列表/统计按生命周期状态筛选
+        Index("ix_accounts_status", "status"),
+    )
+
    assigned_user: Mapped[User | None] = relationship(
        "User", back_populates="assigned_accounts", foreign_keys=[assigned_to]
    )
@@ -729,6 +789,10 @@ class AccountEvent(Base):
    - `amount` 有符号（充值为正、兑换消耗为负），便于页面直接算累计；
    - `source_ref` 是幂等键，可空且唯一：收集器重跑、事件重复发射都靠它兜住；
    - `detail` 只存精简字段，不放原始大 JSON（那是 douyu_tasks 的包袱）。
+
+    除了「时间线」，它还承担**业务台账**（开通 / 积分 / 兑换 / 充值记录）——
+    查询靠 `(account_id, event_type, subject_key, occurred_at)` 复合索引，
+    所以**有流程的东西才另建表**（如兑换工单），「事实」一律进这里。
    """

    __tablename__ = "account_events"
@@ -738,18 +802,48 @@ class AccountEvent(Base):
    # 前缀已覆盖按账号查，再单建一个纯属冗余。
    account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"), nullable=False)
    event_type: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
+    # 粗分类：lifecycle（状态变迁）/ auth（登录与 CK）/ business（业务）/ ops（人工）
+    category: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="", server_default=""
+    )
    occurred_at: Mapped[datetime] = mapped_column(UTCDateTime, nullable=False)
    title: Mapped[str] = mapped_column(String(256), default="", server_default="")
    amount: Mapped[int | None] = mapped_column(Integer, nullable=True)
    balance_after: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # 台账检索键：subject_type = commodity / skin / order / role，
+    # subject_key = 斗鱼 commodity_id / 归一后的 skin_key / 订单号。
+    # 只放**需要检索**的键，其余细节进 detail —— 别把这里变成第二个 detail。
+    subject_type: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="", server_default=""
+    )
+    subject_key: Mapped[str] = mapped_column(
+        String(96), nullable=False, default="", server_default=""
+    )
    detail: Mapped[dict | None] = mapped_column(JSON, nullable=True)
    source: Mapped[str] = mapped_column(String(24), default="", server_default="")
+    # 谁触发的：user / system / public（收集链接）
+    actor_type: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="", server_default=""
+    )
+    actor_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
    # 幂等键：可空且唯一（MySQL 允许多个 NULL，便于「没有幂等键」的事件共存）
    source_ref: Mapped[str | None] = mapped_column(
        String(96), nullable=True, index=True, unique=True
    )
    created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=_utcnow)

+    __table_args__ = (
+        # 台账查询：「该账号该事件类型该客体最近一次」——365 天同皮肤规则靠它。
+        # 跨账号反查（「这件皮肤被谁兑过」）吃 event_type + subject_key 前缀。
+        Index(
+            "ix_account_events_ledger",
+            "account_id",
+            "event_type",
+            "subject_key",
+            "occurred_at",
+        ),
+    )
+

 class AutomationJob(Base):
    """自动化作业定义。状态落库，进程重启不丢调度进度。"""