"""Move asset-admin authority from offices to managed locations.

Before this migration the ``asset_admin`` role and every regional
write gate in the asset domain were derived from
``asset_office_admins`` — the admin list hanging off
``asset_office_addresses`` (公司办公室及主体信息). Resolving "which
admin covers this asset" therefore had to walk a four-tier ladder
(holder office → assignee city → ``office.asset_location`` → unique
country office), and a dedicated bridge column
(``asset_office_addresses.asset_location``, migration 525) existed
purely to translate a managed asset location back into an office.

The asset domain's first-class dimension is the **location**, not the
office: ``asset_items.location`` is a managed, resolver-gated column
that 100% of production assets already carry. This migration makes
locations the authority:

1. New join table ``asset_location_admins`` (location_id, person_id,
   is_primary) — the new single source of truth for ``asset_admin``
   role membership and for the regional write gate.

2. The existing office admin list is backfilled into it by mapping
   each office to its managed location (``office.asset_location``
   first, then an exact ``office.city`` name match). The mapping is
   verified to be total: if ANY office admin cannot be mapped the
   migration raises rather than silently dropping that person's
   authority. Stopping the deploy is the correct failure mode —
   a silent revoke would strip a real administrator's access and
   nobody would notice until they hit a 403.

3. ``asset_office_admins`` is renamed to ``asset_office_local_admins``.
   The office list survives, but its meaning narrows to 当地 Admin —
   the office's local IT contact, used only by the person-anchored
   HR flows (Day-1 welcome CC, onboarding hardware recipient,
   offboarding handover suggestion). Keeping the old name next to
   the new ``asset_location_admins`` would leave two similarly-named
   tables where only one grants the role.

4. Two managed locations (巴西 / 瑞典) are seeded so that every
   work location observed on an active production person resolves
   to a managed location. The matching resolver map entries ship in
   the same PR.

5. The ``asset_admin`` grant set is rebuilt from the new table in the
   same transaction, so the platform never runs with the authority
   table and the role table disagreeing.

Idempotent: every DDL step is probed first (schema-qualified — an
unqualified ``information_schema`` probe reads across every database
on the server and silently skips the DDL), and the data steps use
``INSERT ... ON DUPLICATE KEY UPDATE`` / set-difference rebuilds.

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

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import id_type_for, table_kw_for

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


# Managed locations that production person records reference but the
# master did not list. Seeded here so the resolver map entries added in
# this PR (São Paulo / Niterói → 巴西, Stockholm → 瑞典) have a target.
_NEW_LOCATIONS = (
    ("BR", "巴西"),
    ("SE", "瑞典"),
)


def _table_exists(bind, name: str) -> bool:
    return bool(
        bind.execute(
            sa.text(
                "SELECT COUNT(*) FROM information_schema.TABLES "
                "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t"
            ),
            {"t": name},
        ).scalar()
    )


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


def upgrade() -> None:
    bind = op.get_bind()

    # ── 1. The new authority table ──────────────────────────────────
    #
    # Both FK parents (asset_locations, identity_persons) collate
    # utf8mb4_unicode_ci on prod TiDB while @@collation_server is
    # utf8mb4_bin, so the table charset MUST be pinned explicitly —
    # otherwise the columns inherit the server default and the FK is
    # rejected with error 3780 on prod only (CI MySQL has one
    # collation everywhere and cannot reproduce it).
    if not _table_exists(bind, "asset_location_admins"):
        op.create_table(
            "asset_location_admins",
            sa.Column("location_id", id_type_for("asset_locations"), nullable=False),
            sa.Column("person_id", id_type_for("identity_persons"), nullable=False),
            sa.Column(
                "is_primary",
                sa.Boolean(),
                nullable=False,
                server_default=sa.text("0"),
            ),
            sa.Column(
                "created_at",
                sa.DateTime(timezone=True),
                nullable=False,
                server_default=sa.text("CURRENT_TIMESTAMP"),
            ),
            sa.PrimaryKeyConstraint("location_id", "person_id"),
            sa.ForeignKeyConstraint(
                ["location_id"], ["asset_locations.id"], ondelete="CASCADE",
            ),
            sa.ForeignKeyConstraint(
                ["person_id"], ["identity_persons.id"], ondelete="CASCADE",
            ),
            **table_kw_for("identity_persons"),
        )
    if not _index_exists(bind, "asset_location_admins", "ix_asset_location_admins_person"):
        op.create_index(
            "ix_asset_location_admins_person",
            "asset_location_admins",
            ["person_id"],
        )

    # ── 2. Seed the two missing managed locations ───────────────────
    next_sort = bind.execute(
        sa.text("SELECT COALESCE(MAX(sort_order), 0) FROM asset_locations")
    ).scalar() or 0
    for code, name in _NEW_LOCATIONS:
        exists = bind.execute(
            sa.text(
                "SELECT COUNT(*) FROM asset_locations "
                "WHERE code = :code OR name = :name"
            ),
            {"code": code, "name": name},
        ).scalar()
        if exists:
            continue
        next_sort += 1
        bind.execute(
            sa.text(
                "INSERT INTO asset_locations "
                "  (id, code, name, sort_order, is_active, created_at, updated_at) "
                "VALUES (UUID(), :code, :name, :sort, 1, NOW(), NOW())"
            ),
            {"code": code, "name": name, "sort": next_sort},
        )

    # ── 3. Backfill from the office admin list ──────────────────────
    #
    # Tier 1: office.asset_location -> asset_locations.name (the
    #         explicit operator-selected mapping, set for every
    #         production office).
    # Tier 2: office.city -> asset_locations.name (exact name match,
    #         covers a China-city office whose bridge column was never
    #         filled in).
    #
    # ``legacy`` resolves the source table name so the backfill is
    # re-runnable after step 4 has already renamed it.
    legacy = (
        "asset_office_admins"
        if _table_exists(bind, "asset_office_admins")
        else "asset_office_local_admins"
    )
    if _table_exists(bind, legacy):
        bind.execute(
            sa.text(
                f"""
                INSERT INTO asset_location_admins
                    (location_id, person_id, is_primary, created_at)
                SELECT loc.id, a.person_id, a.is_primary, NOW()
                FROM {legacy} a
                JOIN asset_office_addresses o ON o.id = a.office_id
                JOIN asset_locations loc
                  ON loc.name = COALESCE(o.asset_location, o.city)
                ON DUPLICATE KEY UPDATE
                    is_primary = VALUES(is_primary)
                """
            )
        )

        # Verify the mapping was total. A person who administers an
        # office we cannot map to a location would lose asset_admin at
        # step 5 without a trace, so refuse to continue instead.
        orphans = bind.execute(
            sa.text(
                f"""
                SELECT DISTINCT o.city, o.asset_location
                FROM {legacy} a
                JOIN asset_office_addresses o ON o.id = a.office_id
                LEFT JOIN asset_locations loc
                  ON loc.name = COALESCE(o.asset_location, o.city)
                WHERE loc.id IS NULL
                """
            )
        ).fetchall()
        if orphans:
            detail = ", ".join(
                f"{city} (asset_location={asset_location!r})"
                for city, asset_location in orphans
            )
            raise RuntimeError(
                "535: cannot map every office admin to a managed asset "
                "location — these offices have admins but no matching "
                f"asset_locations row: {detail}. Point each office's "
                "asset_location at a managed location (or create the "
                "location) before deploying, otherwise those "
                "administrators would silently lose the asset_admin role."
            )

    # ── 4. Narrow the office list to 当地 Admin ─────────────────────
    if _table_exists(bind, "asset_office_admins") and not _table_exists(
        bind, "asset_office_local_admins"
    ):
        op.rename_table("asset_office_admins", "asset_office_local_admins")

    # ── 5. Rebuild the asset_admin grant set from the new authority ──
    #
    # Same predicate the runtime reconciler uses: active, non-deleted
    # persons who either administer at least one location or hold
    # asset_global_admin. Expressed as two set-difference statements so
    # a re-run is a no-op and untouched grants keep their created_at
    # (the access-cycle ledger reads it).
    role_id = bind.execute(
        sa.text("SELECT id FROM auth_roles WHERE name = 'asset_admin'")
    ).scalar()
    if role_id is not None:
        should_hold_sql = """
            SELECT DISTINCT p.id
            FROM identity_persons p
            WHERE p.deleted_at IS NULL
              AND (p.employment_status = 'active' OR p.employment_status IS NULL)
              AND (
                    EXISTS (
                        SELECT 1 FROM asset_location_admins la
                        WHERE la.person_id = p.id
                    )
                 OR EXISTS (
                        SELECT 1 FROM auth_user_roles ur
                        JOIN auth_roles r ON r.id = ur.role_id
                        WHERE ur.user_id = p.id AND r.name = 'asset_global_admin'
                    )
              )
        """
        bind.execute(
            sa.text(
                f"""
                DELETE ur FROM auth_user_roles ur
                WHERE ur.role_id = :role_id
                  AND ur.user_id NOT IN (SELECT id FROM ({should_hold_sql}) s)
                """
            ),
            {"role_id": role_id},
        )
        bind.execute(
            sa.text(
                f"""
                INSERT INTO auth_user_roles (user_id, role_id, granted_at)
                SELECT s.id, :role_id, NOW()
                FROM ({should_hold_sql}) s
                WHERE NOT EXISTS (
                    SELECT 1 FROM auth_user_roles ur
                    WHERE ur.user_id = s.id AND ur.role_id = :role_id
                )
                """
            ),
            {"role_id": role_id},
        )

    # ── 6. Retune the daily reconciler's description ────────────────
    bind.execute(
        sa.text(
            """
            UPDATE scheduled_jobs
            SET description = :description
            WHERE handler_key = 'asset.asset_admin_role_reconcile'
            """
        ),
        {
            "description": (
                "每日 03:15(Asia/Shanghai) 对账 asset_admin 角色:"
                "确保 auth_user_roles 与 asset_location_admins 完全一致。"
                "正常路径在位置管理员增删时实时同步,本 cron 仅作兜底。"
            ),
        },
    )


def downgrade() -> None:
    bind = op.get_bind()
    if _table_exists(bind, "asset_office_local_admins") and not _table_exists(
        bind, "asset_office_admins"
    ):
        op.rename_table("asset_office_local_admins", "asset_office_admins")
    if _index_exists(bind, "asset_location_admins", "ix_asset_location_admins_person"):
        op.drop_index("ix_asset_location_admins_person", table_name="asset_location_admins")
    if _table_exists(bind, "asset_location_admins"):
        op.drop_table("asset_location_admins")