"""Drop the denormalised office primary-contact column.

``asset_office_addresses.admin_person_id`` was a single-FK copy of "who
is this office's primary 当地 Admin", mirrored from
``asset_office_local_admins`` by one write path. Migration 536's PR
moved the three readers — the onboarding hardware recipient, the
recycle-to-office recipient and the offboarding handover suggestion —
onto ``office_local_admins.get_office_primary_local_admin()``, which
reads the join directly and additionally filters departed contacts.

The column is dropped one release later, deliberately. The deploy runs
``alembic upgrade head`` and *then* restarts the containers, so a column
removed in the same release its readers disappear is briefly absent from
under the still-running previous version. One deploy of separation means
no running code has referenced it by the time it goes.

Both the foreign key and the index backing it have to go first, and both
are looked up by COLUMN rather than by name: production calls them
``fk_office_addresses_admin_person_id`` while a schema built by
``create_all`` gets MySQL's auto-generated ``…_ibfk_N``. Hardcoding
either spelling leaves the other environment with a constraint on a
column this migration is about to remove — which fails the DDL rather
than skipping it, so it would surface as a broken deploy.

Revision ID: 540
Revises: 539
Create Date: 2026-08-24
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "540"
down_revision = "539"
branch_labels = None
depends_on = None

_TABLE = "asset_office_addresses"
_COLUMN = "admin_person_id"


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


def _fk_names_on(bind, table: str, column: str) -> list[str]:
    rows = bind.execute(
        sa.text(
            "SELECT DISTINCT CONSTRAINT_NAME "
            "FROM information_schema.KEY_COLUMN_USAGE "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t "
            "  AND COLUMN_NAME = :c AND REFERENCED_TABLE_NAME IS NOT NULL"
        ),
        {"t": table, "c": column},
    ).all()
    return [row[0] for row in rows]


def _index_names_on(bind, table: str, column: str) -> list[str]:
    rows = bind.execute(
        sa.text(
            "SELECT DISTINCT INDEX_NAME FROM information_schema.STATISTICS "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t "
            "  AND COLUMN_NAME = :c AND INDEX_NAME <> 'PRIMARY'"
        ),
        {"t": table, "c": column},
    ).all()
    return [row[0] for row in rows]


def upgrade() -> None:
    bind = op.get_bind()
    if not _column_exists(bind, _TABLE, _COLUMN):
        return

    for fk_name in _fk_names_on(bind, _TABLE, _COLUMN):
        op.drop_constraint(fk_name, _TABLE, type_="foreignkey")

    # After the FK is gone its backing index can be dropped; TiDB refuses
    # DROP COLUMN while any index still covers it.
    for index_name in _index_names_on(bind, _TABLE, _COLUMN):
        op.drop_index(index_name, table_name=_TABLE)

    op.drop_column(_TABLE, _COLUMN)


def downgrade() -> None:
    bind = op.get_bind()
    if _column_exists(bind, _TABLE, _COLUMN):
        return
    from app.core.migration_helpers import id_type_for

    op.add_column(
        _TABLE,
        sa.Column(_COLUMN, id_type_for("identity_persons"), nullable=True),
    )
    op.create_foreign_key(
        "fk_office_addresses_admin_person_id",
        _TABLE,
        "identity_persons",
        [_COLUMN],
        ["id"],
    )
    # Restored empty on purpose: the join table is the source of truth,
    # and re-deriving a value here would recreate the drift this removal
    # exists to end.