"""Asset reservation for incoming hires — 备机闭环 (onboarding-parity PR3).

Adds ``reserved_for_person_id`` (FK identity_persons, SET NULL) +
``reserved_at`` to ``asset_items``. A reserved asset stays in its stock
status (idle / available / held_by_office / held_by_leaver) but is
earmarked for a specific incoming hire; the assign/claim/transfer paths
refuse to hand it to anyone else until the reservation is released.

FK is created out-of-line AFTER ``normalize_fk_column_collation`` —
``asset_items`` predates the collation-discipline era and its charset
may diverge from ``identity_persons`` on prod TiDB (error 3780; see
PR #980).
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import (
    id_type_for,
    normalize_fk_column_collation,
)

revision = "279"
down_revision = "278"
branch_labels = None
depends_on = None

TABLE = "asset_items"
FK_COL = "reserved_for_person_id"
AT_COL = "reserved_at"
FK_NAME = "fk_asset_items_reserved_for_person"
IDX_NAME = "ix_asset_items_reserved_for"


def _column_exists(bind, table: str, column: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.COLUMNS "
            "WHERE TABLE_SCHEMA = DATABASE() "
            "  AND TABLE_NAME = :t AND COLUMN_NAME = :c"
        ),
        {"t": table, "c": column},
    ).scalar() is not None


def _index_exists(bind, table: str, index: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.STATISTICS "
            "WHERE TABLE_SCHEMA = DATABASE() "
            "  AND TABLE_NAME = :t AND INDEX_NAME = :i"
        ),
        {"t": table, "i": index},
    ).scalar() is not None


def _fk_exists(bind, table: str, constraint: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLE_CONSTRAINTS "
            "WHERE TABLE_SCHEMA = DATABASE() "
            "  AND TABLE_NAME = :t AND CONSTRAINT_NAME = :c "
            "  AND CONSTRAINT_TYPE = 'FOREIGN KEY'"
        ),
        {"t": table, "c": constraint},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()
    persons_id = id_type_for("identity_persons")

    if not _column_exists(bind, TABLE, FK_COL):
        op.add_column(
            TABLE, sa.Column(FK_COL, persons_id, nullable=True),
        )
    if not _column_exists(bind, TABLE, AT_COL):
        op.add_column(
            TABLE,
            sa.Column(AT_COL, sa.DateTime(timezone=True), nullable=True),
        )
    if not _index_exists(bind, TABLE, IDX_NAME):
        op.create_index(IDX_NAME, TABLE, [FK_COL])
    if not _fk_exists(bind, TABLE, FK_NAME):
        normalize_fk_column_collation(TABLE, FK_COL, "identity_persons")
        op.create_foreign_key(
            FK_NAME, TABLE, "identity_persons",
            [FK_COL], ["id"], ondelete="SET NULL",
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _fk_exists(bind, TABLE, FK_NAME):
        op.drop_constraint(FK_NAME, TABLE, type_="foreignkey")
    if _index_exists(bind, TABLE, IDX_NAME):
        op.drop_index(IDX_NAME, table_name=TABLE)
    if _column_exists(bind, TABLE, AT_COL):
        op.drop_column(TABLE, AT_COL)
    if _column_exists(bind, TABLE, FK_COL):
        op.drop_column(TABLE, FK_COL)