SupplyAgent/alembic/versions/20260730_04_add_local_auth.py

"""add local users and server-side sessions

Revision ID: 20260730_04
Revises: 20260729_03
Create Date: 2026-07-30
"""
from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "20260730_04"
down_revision: str | None = "20260729_03"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
    op.create_table(
        "auth_user",
        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
        sa.Column("username", sa.String(length=64), nullable=False),
        sa.Column("password_hash", sa.String(length=255), nullable=False),
        sa.Column("display_name", sa.String(length=128), nullable=False),
        sa.Column("role", sa.String(length=16), nullable=False),
        sa.Column("status", sa.String(length=16), nullable=False),
        sa.Column("failed_login_count", sa.Integer(), nullable=False, server_default="0"),
        sa.Column("locked_until", sa.DateTime(), nullable=True),
        sa.Column("last_login_at", sa.DateTime(), nullable=True),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.Column("updated_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("username", name="uq_auth_user_username"),
        mysql_charset="utf8mb4",
    )
    op.create_index("ix_auth_user_username", "auth_user", ["username"], unique=False)
    op.create_index(
        "idx_auth_user_role_status",
        "auth_user",
        ["role", "status"],
        unique=False,
    )

    op.create_table(
        "auth_session",
        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
        sa.Column("token_hash", sa.String(length=64), nullable=False),
        sa.Column("user_id", sa.Integer(), nullable=False),
        sa.Column("expires_at", sa.DateTime(), nullable=False),
        sa.Column("last_active_at", sa.DateTime(), nullable=False),
        sa.Column("ip_address", sa.String(length=64), nullable=True),
        sa.Column("user_agent", sa.String(length=512), nullable=True),
        sa.Column("created_at", sa.DateTime(), server_default=sa.func.now(), nullable=False),
        sa.ForeignKeyConstraint(["user_id"], ["auth_user.id"], ondelete="CASCADE"),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("token_hash", name="uq_auth_session_token_hash"),
        mysql_charset="utf8mb4",
    )
    op.create_index(
        "ix_auth_session_token_hash",
        "auth_session",
        ["token_hash"],
        unique=False,
    )
    op.create_index(
        "idx_auth_session_user",
        "auth_session",
        ["user_id", "expires_at"],
        unique=False,
    )
    op.create_index(
        "idx_auth_session_expiry",
        "auth_session",
        ["expires_at"],
        unique=False,
    )


def downgrade() -> None:
    op.drop_index("idx_auth_session_expiry", table_name="auth_session")
    op.drop_index("idx_auth_session_user", table_name="auth_session")
    op.drop_index("ix_auth_session_token_hash", table_name="auth_session")
    op.drop_table("auth_session")
    op.drop_index("idx_auth_user_role_status", table_name="auth_user")
    op.drop_index("ix_auth_user_username", table_name="auth_user")
    op.drop_table("auth_user")