"""Cross-vendor external-account registry (one annotation per email).

Revision ID: 370
Revises: 369
Create Date: 2026-07-01

Every vendor's seat reconciliation flags accounts that don't match an
active Helix employee as ``software_external_holder.category='external'``.
The same email shows up across many vendors at once (service accounts,
partners, never-offboarded ex-employees), so annotating per-vendor is
redundant. ``external_account_registry`` is the ONE place an admin
classifies + notes an email; all vendors' panels read it at display time
and subtract "known/expected" classifications from their anomaly count.

Generalises the Google-only ``google_orphan_accounts`` "owner + purpose"
pattern. ``google_orphan_accounts`` stays the source of truth for Google
account presence + the offboarding ``RES_GOOGLE_ORPHAN_OWNER`` blocker
(owner lives there, untouched); ``owner_person_id`` here is only for
NON-Google emails.

``owner_person_id`` FK column type mirrors ``identity_persons.id``'s
actual DATA_TYPE so TiDB doesn't reject the constraint with error 3780
(CHAR vs VARCHAR). Every object creation is guarded so a partial apply
re-runs cleanly.
"""

import sqlalchemy as sa

from alembic import op

revision = "370"
down_revision = "369"
branch_labels = None
depends_on = None


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 _table_exists(bind, table: str) -> bool:
    row = bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t"
        ),
        {"t": table},
    ).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


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

    if not _table_exists(bind, "external_account_registry"):
        op.create_table(
            "external_account_registry",
            sa.Column("id", sa.CHAR(36), primary_key=True),
            sa.Column("email", sa.String(255), nullable=False),
            sa.Column(
                "classification", sa.String(32), nullable=False,
                server_default="unknown",
            ),
            sa.Column("note", sa.Text(), nullable=True),
            # out-of-line FK so the column type can be aligned to
            # identity_persons.id first (TiDB 3780 guard).
            sa.Column("owner_person_id", persons_id_type, nullable=True),
            sa.Column("created_by", sa.CHAR(36), nullable=True),
            sa.Column("updated_by", sa.CHAR(36), nullable=True),
            sa.Column(
                "created_at", sa.DateTime(timezone=True),
                nullable=False, server_default=sa.func.now(),
            ),
            sa.Column(
                "updated_at", sa.DateTime(timezone=True),
                nullable=False, server_default=sa.func.now(),
            ),
        )

    if not _index_exists(
        bind, "external_account_registry", "uq_external_account_registry_email",
    ):
        op.create_index(
            "uq_external_account_registry_email",
            "external_account_registry", ["email"], unique=True,
        )
    if not _index_exists(
        bind, "external_account_registry",
        "ix_external_account_registry_owner_person_id",
    ):
        op.create_index(
            "ix_external_account_registry_owner_person_id",
            "external_account_registry", ["owner_person_id"],
        )
    if not _fk_exists(
        bind, "external_account_registry",
        "fk_external_account_registry_owner_person_id",
    ):
        op.create_foreign_key(
            "fk_external_account_registry_owner_person_id",
            "external_account_registry",
            "identity_persons",
            ["owner_person_id"],
            ["id"],
            ondelete="SET NULL",
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _fk_exists(
        bind, "external_account_registry",
        "fk_external_account_registry_owner_person_id",
    ):
        op.drop_constraint(
            "fk_external_account_registry_owner_person_id",
            "external_account_registry", type_="foreignkey",
        )
    if _table_exists(bind, "external_account_registry"):
        op.drop_table("external_account_registry")