"""Asset office multi-admin + auto-derived asset_admin role + location backfill.

This migration overhauls the "资产管理员" model so that:

1. An office (``asset_office_addresses`` row) can have MULTIPLE asset
   admins instead of a single ``admin_person_id`` FK. The new
   ``asset_office_admins`` join table is the authoritative full list.
   The legacy ``admin_person_id`` column is preserved as a
   denormalised "primary contact" — existing single-admin call
   sites (offboarding handover routing, inventory ping fan-out,
   hr_lifecycle) continue to read it without change. New writes
   keep it in sync with the FIRST row in the join table.

2. The ``asset_admin`` role becomes **fully auto-derived** from
   the union of all office admins. Manual grants in
   ``auth_user_roles`` are revoked here; from this PR forward the
   role membership is the canonical projection of
   ``asset_office_admins.person_id``. The matching service
   (``app.domains.asset.office_admin_sync``) re-syncs on every
   office-admin add/remove, and a daily cron
   (``asset.asset_admin_role_reconcile``) acts as the backstop.

3. ``asset_items.location='国内'`` is a free-text legacy value that
   predates the office-held model. For laptops we can usually
   recover the real city from the assignee's
   ``identity_persons.city``; this migration runs the SQL fix in
   the same transaction so the platform never ships in an
   intermediate state.

The data backfill is idempotent — re-running the migration is a
no-op:
  * ``INSERT ... ON DUPLICATE KEY UPDATE`` for the join-table seed
  * ``DELETE + INSERT`` for the role re-sync (single transaction)
  * ``UPDATE ... WHERE location = '国内'`` is a no-op once the rows
    have been moved off the placeholder

Revision ID: 220
Revises: 219
Create Date: 2026-05-29

NOTE: originally written as 217 → bumped to 218 (PR #566 took 217) →
219 (PR #572 took 218) → 220 + ``down_revision="219"`` (PR #568 took
219). Per the ``alembic_revision_race`` rule.
"""
from __future__ import annotations

import uuid

import sqlalchemy as sa
from alembic import op


revision = "220"
down_revision = "219"
branch_labels = None
depends_on = None


# Category values stored in ``asset_items.category`` that count as
# "笔记本" for the 国内 → city backfill. ``laptop`` is the modern
# enum; the legacy import path occasionally wrote a Chinese string
# too — we handle both. ``desktop`` is intentionally excluded: the
# user's request was specifically about the 95 laptops.
_LAPTOP_CATEGORIES = ("laptop", "笔记本", "笔记本电脑")


def _fk_source_col_type(
    bind,
    target_table: str,
    target_column: str,
    *,
    fallback_type: str = "char",
    fallback_collation: str | None = "utf8mb4_unicode_ci",
    length: int = 36,
):
    """Mirror the target FK column's DATA_TYPE + collation (TiDB strict-FK).

    Same helper introduced in migration 210; copied locally so this
    migration is self-contained. TiDB rejects FK creation with
    ``ERROR 3780`` when DATA_TYPE differs (CHAR vs VARCHAR) — local
    MySQL silently accepts the mismatch, so the failure only surfaces
    in prod. ``asset_office_addresses.id`` and ``identity_persons.id``
    are both VARCHAR(36)+utf8mb4_unicode_ci in long-lived schemas;
    declaring this table's FK columns as ``sa.CHAR(36)`` would 3780
    in prod.
    """
    row = bind.execute(
        sa.text(
            "SELECT DATA_TYPE, COLLATION_NAME "
            "FROM information_schema.COLUMNS "
            "WHERE TABLE_SCHEMA = DATABASE() "
            "  AND TABLE_NAME = :t AND COLUMN_NAME = :c"
        ),
        {"t": target_table, "c": target_column},
    ).fetchone()
    if row:
        data_type = (row[0] or fallback_type).lower()
        collation = row[1] or fallback_collation
    else:
        data_type = fallback_type.lower()
        collation = fallback_collation
    type_cls = sa.CHAR if data_type == "char" else sa.String
    if collation:
        return type_cls(length, collation=collation)
    return type_cls(length)


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

    # 1. Multi-admin join table. FK column types mirror their target
    # columns via ``_fk_source_col_type`` to satisfy TiDB's strict FK
    # type check (CHAR vs VARCHAR mismatch → ERROR 3780, only surfaces
    # in prod — local MySQL accepts the mismatch silently).
    office_id_type = _fk_source_col_type(bind, "asset_office_addresses", "id")
    person_id_type = _fk_source_col_type(bind, "identity_persons", "id")
    op.create_table(
        "asset_office_admins",
        sa.Column("office_id", office_id_type, nullable=False),
        sa.Column("person_id", person_id_type, 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("office_id", "person_id"),
        sa.ForeignKeyConstraint(
            ["office_id"], ["asset_office_addresses.id"], ondelete="CASCADE",
        ),
        sa.ForeignKeyConstraint(
            ["person_id"], ["identity_persons.id"], ondelete="CASCADE",
        ),
    )
    op.create_index(
        "ix_asset_office_admins_person",
        "asset_office_admins",
        ["person_id"],
    )

    # 2. Seed the join table from the existing single-admin column.
    bind.execute(sa.text(
        """
        INSERT INTO asset_office_admins (office_id, person_id, is_primary, created_at)
        SELECT id, admin_person_id, 1, NOW()
        FROM asset_office_addresses
        WHERE admin_person_id IS NOT NULL
        ON DUPLICATE KEY UPDATE is_primary = 1
        """
    ))

    # 3. Re-sync the ``asset_admin`` system role.
    #
    # Step A: revoke ALL current asset_admin grants. Per user
    #         direction 2026-05-29: "资产管理员的这个角色的成员改成
    #         自动维护 ... 在这里面的人就自动是 ... 不在就不是".
    # Step B: re-grant to every person who appears in the join
    #         table (DISTINCT — a person managing 2 offices still
    #         gets exactly one role row).
    #
    # global_admin (it_admin) holders are unaffected — they carry
    # the broader role, asset_admin is additive only.
    role_row = bind.execute(sa.text(
        "SELECT id FROM auth_roles WHERE name = 'asset_admin' LIMIT 1"
    )).fetchone()
    if role_row is not None:
        role_id = role_row[0]
        bind.execute(
            sa.text("DELETE FROM auth_user_roles WHERE role_id = :rid"),
            {"rid": role_id},
        )
        bind.execute(
            sa.text(
                """
                INSERT INTO auth_user_roles (user_id, role_id, granted_at)
                SELECT DISTINCT aoa.person_id, :rid, NOW()
                FROM asset_office_admins aoa
                JOIN identity_persons p ON p.id = aoa.person_id
                WHERE p.deleted_at IS NULL
                  AND (p.employment_status IS NULL OR p.employment_status = 'active')
                """
            ),
            {"rid": role_id},
        )

    # 4. Backfill ``asset_items.location='国内'`` laptops.
    #
    # The user reported 95 affected rows. We pull the city from
    # the assignee's ``identity_persons.city``; if that's empty we
    # try ``work_location_name`` as a fallback. Rows with neither
    # are left as-is — counted + logged below so the operator can
    # see exactly how many laptops still need manual attention
    # after the migration.
    import logging  # noqa: PLC0415

    _mig_logger = logging.getLogger("alembic.runtime.migration")
    for category in _LAPTOP_CATEGORIES:
        bind.execute(
            sa.text(
                """
                UPDATE asset_items a
                JOIN identity_persons p ON p.id = a.assigned_to
                SET a.location = p.city
                WHERE a.location = '国内'
                  AND a.category = :cat
                  AND a.assigned_to IS NOT NULL
                  AND p.city IS NOT NULL
                  AND TRIM(p.city) <> ''
                """
            ),
            {"cat": category},
        )
        bind.execute(
            sa.text(
                """
                UPDATE asset_items a
                JOIN identity_persons p ON p.id = a.assigned_to
                SET a.location = p.work_location_name
                WHERE a.location = '国内'
                  AND a.category = :cat
                  AND a.assigned_to IS NOT NULL
                  AND (p.city IS NULL OR TRIM(p.city) = '')
                  AND p.work_location_name IS NOT NULL
                  AND TRIM(p.work_location_name) <> ''
                """
            ),
            {"cat": category},
        )

    # Count + surface remaining 国内 laptops so the migrator sees
    # exactly how many rows the heuristic couldn't fix and can
    # decide whether to back-fill manually.
    remaining_count = bind.execute(sa.text(
        """
        SELECT COUNT(*)
        FROM asset_items
        WHERE location = '国内'
          AND category IN ('laptop', '笔记本', '笔记本电脑')
          AND deleted_at IS NULL
        """
    )).scalar() or 0
    if remaining_count > 0:
        _mig_logger.warning(
            "migration 218: %d laptop(s) still have location='国内' after "
            "backfill — assignee has neither city nor work_location_name. "
            "Manual review required.",
            remaining_count,
        )
    else:
        _mig_logger.info(
            "migration 218: 国内 laptop backfill complete (0 rows left)."
        )

    # 5. Seed the daily reconciliation cron (backstop for the
    #    in-process role sync).
    has_jobs_table = bind.execute(sa.text(
        "SELECT 1 FROM information_schema.TABLES "
        "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'scheduled_jobs'"
    )).fetchone() is not None
    if has_jobs_table:
        bind.execute(
            sa.text(
                """
                INSERT INTO scheduled_jobs
                (id, name, display_name, description, handler_key,
                 trigger_type, cron_expr, timezone, enabled, config,
                 run_count, failure_count, created_at, updated_at)
                SELECT :id, :name, :display, :desc, :handler, 'cron',
                       :cron, 'Asia/Shanghai', 1, '{}', 0, 0, NOW(), NOW()
                FROM DUAL
                WHERE NOT EXISTS (
                    SELECT 1 FROM scheduled_jobs WHERE name = :name
                )
                """
            ),
            {
                "id": str(uuid.uuid4()),
                "name": "asset.asset_admin_role_reconcile",
                "display": "资产管理员角色对账",
                "desc": (
                    "每日 03:15(Asia/Shanghai) 对账 asset_admin 角色:"
                    "确保 auth_user_roles 与 asset_office_admins 完全一致。"
                    "正常路径在 office 管理员增删时实时同步,本 cron 仅作兜底。"
                ),
                "handler": "asset.asset_admin_role_reconcile",
                "cron": "15 3 * * *",
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    has_jobs_table = bind.execute(sa.text(
        "SELECT 1 FROM information_schema.TABLES "
        "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'scheduled_jobs'"
    )).fetchone() is not None
    if has_jobs_table:
        bind.execute(sa.text(
            "DELETE FROM scheduled_jobs WHERE handler_key = "
            "'asset.asset_admin_role_reconcile'"
        ))
    op.drop_index(
        "ix_asset_office_admins_person",
        table_name="asset_office_admins",
    )
    op.drop_table("asset_office_admins")
    # Note: location backfill and role re-sync are intentionally NOT
    # reversed — they were data-correctness fixes, not schema changes.