"""Drop the office → asset-location bridge column.

``asset_office_addresses.asset_location`` (migration 525) existed for
one reason: to translate a managed asset location back into the office
that owned it, because asset authority hung off the office. Migration
535 moved that authority onto the location itself, so nothing needs the
translation any more — and the last display-only reader
(``AssetService.enrich_with_users``, which showed an office name beside
an asset) now derives the same answer from ``office.city`` through the
managed-location resolver, the way every other location string in the
codebase is resolved.

Keeping it would leave a second stored copy of the office↔location
relationship that nothing maintains: an operator renaming a location
would silently desync it, and the next reader would trust it.

The column carries a UNIQUE index, which has to go first — TiDB refuses
``DROP COLUMN`` while an index covers it (error 8200), where MySQL 8
silently rewrites the index instead. Both probes are schema-qualified;
an unqualified ``information_schema`` read matches a co-hosted helix
schema and skips the DDL while alembic still stamps the revision.

``admin_person_id`` is deliberately NOT dropped here even though this PR
removes its last reader. The deploy applies migrations and *then*
restarts the containers, so a column dropped in the same release the
readers disappear is briefly missing from under the still-running old
code. It goes in the next migration, one deploy later, when no running
version reads it.

Revision ID: 536
Revises: 535
Create Date: 2026-08-24
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "536"
down_revision = "535"
branch_labels = None
depends_on = None

_TABLE = "asset_office_addresses"
_COLUMN = "asset_location"
_INDEX = "uq_asset_office_addresses_asset_location"


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 _index_names_on(bind, table: str, column: str) -> list[str]:
    """Every index covering ``column``, PRIMARY excluded.

    Looked up by column rather than by name: the unique index is
    ``uq_asset_office_addresses_asset_location`` on production, but a
    schema built by ``create_all`` (CI, local test DBs) names it from
    the model instead, and hardcoding one spelling leaves the other
    environment with an index still covering a column we are about to
    drop.
    """
    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 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
    op.add_column(_TABLE, sa.Column(_COLUMN, sa.String(128), nullable=True))
    op.create_index(_INDEX, _TABLE, [_COLUMN], unique=True)