"""CMDB 对象存储 — owner_person_id FK (parity with LB / CDN / DNS).

Revision ID: 366
Revises: 365
Create Date: 2026-06-30

``cmdb_object_storage`` was the last cloud-resource table still storing
owner as a free-text ``owner_raw`` column only. Every sibling table
(servers + domains since migration 042, CDN since 160, load-balancers
since 173) carries an ``owner_person_id`` FK to ``identity_persons`` so
the owner renders as a PersonTag (leaver strike-through) and is edited
through the UserSelect person picker rather than a free-text input.

This migration brings OSS to parity:

* add ``owner_person_id`` (nullable FK, ondelete SET NULL) +
  ``ix_cmdb_object_storage_owner_person_id`` index.
* a CLEAN, exact backfill from ``owner_raw`` on case-insensitive
  ``work_email`` match only (same UPDATE shape as migration 173's LB
  backfill). We deliberately do NOT fuzzy-match on display_name —
  ``owner_raw`` on OSS is usually a cloud-tag team string
  ("legacy-team-finance"), not a person, so ``owner_person_id`` is the
  deliberate Helix mapping an admin sets via the picker. ``owner_raw``
  stays put as the cloud-tag hint.

FK column type follows the helper from migration 160 / 173 — it mirrors
``identity_persons.id``'s actual DATA_TYPE so TiDB doesn't reject the
constraint with error 3780 (CHAR vs VARCHAR incompatibility). Every
column / index / FK creation is guarded so a partial apply re-runs
cleanly.
"""

import sqlalchemy as sa
from sqlalchemy import text

from alembic import op

revision = "366"
down_revision = "365"
branch_labels = None
depends_on = None


# ─── helpers (mirror migration 173) ────────────────────────────────


def _fk_source_col_type(bind, target_table: str, target_column: str, *, length: int = 36):
    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()
    data_type = (row[0] or "char").lower() if row else "char"
    collation = row[1] if row else None
    type_cls = sa.CHAR if data_type == "char" else sa.String
    return type_cls(length, collation=collation) if collation else type_cls(length)


def _column_exists(bind, table: str, column: str) -> bool:
    row = 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},
    ).fetchone()
    return row is not None


def _index_exists(bind, table: str, index: str) -> bool:
    row = 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},
    ).fetchone()
    return row is not None


def _fk_exists(bind, table: str, constraint: str) -> bool:
    row = 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},
    ).fetchone()
    return row is not None


# ─── upgrade ───────────────────────────────────────────────────────


def upgrade() -> None:
    bind = op.get_bind()
    persons_id_type = _fk_source_col_type(bind, "identity_persons", "id")

    if not _column_exists(bind, "cmdb_object_storage", "owner_person_id"):
        op.add_column(
            "cmdb_object_storage",
            sa.Column("owner_person_id", persons_id_type, nullable=True),
        )
    else:
        # Repair type drift from a partial apply.
        op.alter_column(
            "cmdb_object_storage",
            "owner_person_id",
            type_=persons_id_type,
            existing_nullable=True,
        )

    # Clean, exact backfill: work_email match only (no display_name fuzzy
    # match). owner_raw is usually a cloud-tag team string, so most rows
    # will not resolve — that's intentional; owner_person_id is the
    # deliberate Helix mapping an admin sets later via the picker.
    op.execute(text("""
        UPDATE cmdb_object_storage oss
        JOIN identity_persons p
            ON LOWER(p.work_email) = LOWER(oss.owner_raw)
        SET oss.owner_person_id = CAST(p.id AS CHAR(36))
        WHERE oss.owner_raw IS NOT NULL
          AND oss.owner_raw != ''
          AND oss.owner_person_id IS NULL
          AND oss.deleted_at IS NULL
          AND p.deleted_at IS NULL
          AND p.employment_status = 'active'
    """))

    if not _index_exists(bind, "cmdb_object_storage", "ix_cmdb_object_storage_owner_person_id"):
        op.create_index(
            "ix_cmdb_object_storage_owner_person_id",
            "cmdb_object_storage",
            ["owner_person_id"],
        )
    if not _fk_exists(bind, "cmdb_object_storage", "fk_cmdb_object_storage_owner_person_id"):
        op.create_foreign_key(
            "fk_cmdb_object_storage_owner_person_id",
            "cmdb_object_storage",
            "identity_persons",
            ["owner_person_id"],
            ["id"],
            ondelete="SET NULL",
        )


# ─── downgrade ─────────────────────────────────────────────────────


def downgrade() -> None:
    bind = op.get_bind()
    if _fk_exists(bind, "cmdb_object_storage", "fk_cmdb_object_storage_owner_person_id"):
        op.drop_constraint(
            "fk_cmdb_object_storage_owner_person_id",
            "cmdb_object_storage", type_="foreignkey",
        )
    if _index_exists(bind, "cmdb_object_storage", "ix_cmdb_object_storage_owner_person_id"):
        op.drop_index(
            "ix_cmdb_object_storage_owner_person_id",
            table_name="cmdb_object_storage",
        )
    if _column_exists(bind, "cmdb_object_storage", "owner_person_id"):
        op.drop_column("cmdb_object_storage", "owner_person_id")