document-management/backend/migrations/versions/0022_login_lockout.py

"""登录失败锁定：sys_user 增加连续失败计数与锁定截止时间列。

Revision ID: 0022_login_lockout
Revises: 0021_remove_assigned_scope
Create Date: 2026-09-01

变更要点：
1. sys_user 增加 failed_login_count INT NOT NULL DEFAULT 0（连续密码错误计数，
   仅 /auth/login 密码错误计入，登录成功清零）；
2. sys_user 增加 locked_until DATETIME NULL（锁定截止时间，UTC，NULL=未锁定；
   连续5次密码错误锁定15分钟，由服务层 login_lockout 维护）；
3. 升级幂等：列已存在时跳过；降级直接删列（锁定状态为易失运行态，可随时丢弃）。
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0022_login_lockout"
down_revision: str | None = "0021_remove_assigned_scope"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

_TABLE = "sys_user"
_COLUMNS = ("failed_login_count", "locked_until")


def _existing_columns() -> set[str]:
    inspector = sa.inspect(op.get_bind())
    return {column["name"] for column in inspector.get_columns(_TABLE)}


def upgrade() -> None:
    existing = _existing_columns()
    if "failed_login_count" not in existing:
        op.add_column(
            _TABLE,
            sa.Column(
                "failed_login_count",
                sa.Integer(),
                nullable=False,
                server_default=sa.text("0"),
            ),
        )
    if "locked_until" not in existing:
        op.add_column(
            _TABLE,
            sa.Column("locked_until", sa.DateTime(), nullable=True),
        )


def downgrade() -> None:
    existing = _existing_columns()
    if "locked_until" in existing:
        op.drop_column(_TABLE, "locked_until")
    if "failed_login_count" in existing:
        op.drop_column(_TABLE, "failed_login_count")