"""ensure the self-hosted demo login remains available

Revision ID: e5f6a7b8c9d0
Revises: d4e5f6a7b8c9
Create Date: 2026-08-05 15:31:00
"""

from collections.abc import Sequence
import os

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql

from common.auth.passwords import hash_password


revision: str = "e5f6a7b8c9d0"
down_revision: str | Sequence[str] | None = "d4e5f6a7b8c9"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


DISABLED_PASSWORD = "demo-login-disabled"
SEEDED_USERS = (
    (
        "00000000000000000000000001",
        "admin",
        "Admin",
        "0000000000000000000000000A",
    ),
)


def _create_users_table() -> None:
    op.create_table(
        "users",
        sa.Column("user_id", mysql.CHAR(length=26), nullable=False),
        sa.Column("username", sa.String(length=64), nullable=False),
        sa.Column("display_name", sa.String(length=100), nullable=False),
        sa.Column("password_hash", sa.String(length=255), nullable=False),
        sa.Column(
            "status",
            sa.String(length=16),
            server_default=sa.text("'active'"),
            nullable=False,
            comment="active/disabled/locked",
        ),
        sa.Column(
            "created_at",
            mysql.DATETIME(fsp=3),
            server_default=sa.text("CURRENT_TIMESTAMP(3)"),
            nullable=False,
        ),
        sa.Column(
            "updated_at",
            mysql.DATETIME(fsp=3),
            server_default=sa.text(
                "CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3)"
            ),
            nullable=False,
        ),
        sa.Column("email", sa.String(length=255), nullable=True),
        sa.Column("platform_role_id", mysql.CHAR(length=26), nullable=True),
        sa.Column("avatar_uri", sa.String(length=1000), nullable=True),
        sa.Column("last_login_at", mysql.DATETIME(fsp=3), nullable=True),
        sa.Column(
            "is_deleted",
            mysql.TINYINT(display_width=1),
            server_default=sa.text("0"),
            nullable=False,
        ),
        sa.Column("deleted_at", mysql.DATETIME(fsp=3), nullable=True),
        sa.PrimaryKeyConstraint("user_id"),
        comment="平台用户",
    )
    op.create_index("fk_users_platform_role", "users", ["platform_role_id"])
    op.create_index("idx_users_status", "users", ["status"])
    op.create_index("uk_users_email", "users", ["email"], unique=True)
    op.create_index("uk_users_username", "users", ["username"], unique=True)


def upgrade() -> None:
    connection = op.get_bind()
    if not sa.inspect(connection).has_table("users"):
        _create_users_table()

    users = sa.table(
        "users",
        sa.column("user_id", sa.String),
        sa.column("username", sa.String),
        sa.column("display_name", sa.String),
        sa.column("password_hash", sa.String),
        sa.column("status", sa.String),
        sa.column("email", sa.String),
        sa.column("platform_role_id", sa.String),
    )
    existing = {
        row.username: row.password_hash
        for row in connection.execute(
            sa.select(users.c.username, users.c.password_hash).where(
                users.c.username.in_([user[1] for user in SEEDED_USERS])
            )
        )
    }
    password_hash = hash_password(
        os.environ.get("INITIAL_ADMIN_PASSWORD", "admin12345")
    )

    for user_id, username, display_name, role_id in SEEDED_USERS:
        if username not in existing:
            connection.execute(
                users.insert().values(
                    user_id=user_id,
                    username=username,
                    display_name=display_name,
                    password_hash=password_hash,
                    status="active",
                    email=f"{username}@model-platform.local",
                    platform_role_id=role_id,
                )
            )
            continue

        if existing[username] in {None, "", DISABLED_PASSWORD}:
            connection.execute(
                users.update()
                .where(users.c.username == username)
                .values(password_hash=password_hash)
            )

    connection.execute(
        users.update()
        .where(users.c.username == "admin")
        .values(status="active")
    )


def downgrade() -> None:
    """Do not remove or disable accounts that may contain user data."""