"""Pre-launch hardening: stop CASCADE deletes from wiping AI/Bailian
billing + audit history when an ``integrations`` row is deleted.

Problem (pre-launch audit 2026-05-06): every Bailian + AI-spending
table chains its history straight to the parent ``integrations`` row
via ``ondelete=CASCADE``:

  * ``cmdb_ai_resources.integration_id`` → integrations  (CASCADE)
  * ``cmdb_ai_resource_cost_daily.ai_resource_id`` → cmdb_ai_resources
    (CASCADE — fine if parent is RESTRICT'd)
  * ``bailian_workspaces.integration_id`` → integrations  (CASCADE)
  * ``bailian_api_keys.integration_id`` → integrations  (CASCADE)
  * ``bailian_usage_daily.integration_id`` → integrations  (CASCADE)
  * ``bailian_usage_daily.workspace_pk`` → bailian_workspaces
    (CASCADE — but we want history preserved if a workspace is
    purged, since usage rows are immutable billing records)

A single hand-typed ``DELETE FROM integrations WHERE id=?`` (or a
future buggy "cleanup" handler) wipes every historical AI/Bailian
chargeback ledger row tied to that integration. This is the exact
"历史记录被清光" failure mode flagged in the pre-launch audit.

Fix: convert the parent FKs to ``ON DELETE RESTRICT`` so a hard
delete of an integration row is REFUSED while billing history
references it. Operators must instead disable / soft-delete the
integration row (``Integration.enabled=False`` plus the soft-delete
column where present). For the workspace_pk → workspaces FK we drop
to ``ON DELETE SET NULL`` so usage rows become orphaned-but-preserved
if a workspace is ever pruned.

The migration is defensive: it discovers each FK constraint name
via ``information_schema.KEY_COLUMN_USAGE`` since 074's inline
``sa.ForeignKey`` column definitions auto-generate names that vary
between MySQL local and TiDB (and between repeated migrations).

The grandchild FK (``cmdb_ai_resource_cost_daily.ai_resource_id``)
stays as CASCADE because the parent ``cmdb_ai_resources`` is now
itself protected from hard delete via the integration RESTRICT chain
— the only way for ``cmdb_ai_resources`` to disappear is via the
upcoming soft-delete path (``deleted_at``), which leaves cost_daily
orphaned-but-preserved.
"""
from __future__ import annotations

import logging

import sqlalchemy as sa

from alembic import op

logger = logging.getLogger(__name__)

revision = "124"
down_revision = "123_identity_persons_cost_center"
branch_labels = None
depends_on = None


_FK_RETARGET_RESTRICT: list[tuple[str, str, str]] = [
    # (table, column, parent_table) — drop existing FK on (table.column),
    # recreate with ON DELETE RESTRICT pointing at parent_table.id.
    ("cmdb_ai_resources", "integration_id", "integrations"),
    ("bailian_workspaces", "integration_id", "integrations"),
    ("bailian_api_keys", "integration_id", "integrations"),
    ("bailian_usage_daily", "integration_id", "integrations"),
]

_FK_RETARGET_SET_NULL: list[tuple[str, str, str]] = [
    # (table, column, parent_table) — convert CASCADE to SET NULL.
    # Used when the child row is a billing/usage record we want to
    # preserve even if the parent (workspace) is pruned.
    ("bailian_usage_daily", "workspace_pk", "bailian_workspaces"),
]


def _table_exists(bind, table: str) -> bool:
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.tables "
            "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1",
        ),
        {"t": table},
    ).scalar())


def _column_exists(bind, table: str, column: str) -> bool:
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.columns "
            "WHERE table_schema = DATABASE() "
            "AND table_name = :t AND column_name = :c LIMIT 1",
        ),
        {"t": table, "c": column},
    ).scalar())


def _column_is_nullable(bind, table: str, column: str) -> bool:
    val = bind.execute(
        sa.text(
            "SELECT IS_NULLABLE FROM information_schema.columns "
            "WHERE table_schema = DATABASE() "
            "AND table_name = :t AND column_name = :c LIMIT 1",
        ),
        {"t": table, "c": column},
    ).scalar()
    return val == "YES"


def _find_fk_name(bind, table: str, column: str) -> str | None:
    """Return the constraint name of any FK on ``table.column``, or
    None if no FK exists. Uses KEY_COLUMN_USAGE which is portable
    across MySQL 8.0 and TiDB (REFERENTIAL_CONSTRAINTS doesn't carry
    the column on all TiDB versions)."""
    return bind.execute(
        sa.text(
            "SELECT constraint_name "
            "FROM information_schema.key_column_usage "
            "WHERE table_schema = DATABASE() "
            "AND table_name = :t AND column_name = :c "
            "AND referenced_table_name IS NOT NULL "
            "LIMIT 1",
        ),
        {"t": table, "c": column},
    ).scalar()


def _drop_and_recreate_fk(
    bind,
    *,
    table: str,
    column: str,
    parent_table: str,
    ondelete: str,
    new_name: str,
) -> None:
    if not _table_exists(bind, table):
        logger.info("122: table %s does not exist; skip", table)
        return
    if not _column_exists(bind, table, column):
        logger.info(
            "122: column %s.%s does not exist; skip", table, column,
        )
        return

    existing = _find_fk_name(bind, table, column)
    if existing:
        try:
            op.drop_constraint(existing, table, type_="foreignkey")
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "122: drop FK %s on %s.%s failed: %s",
                existing, table, column, exc,
            )

    try:
        op.create_foreign_key(
            new_name,
            table,
            parent_table,
            [column],
            ["id"],
            ondelete=ondelete,
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "122: create FK %s (%s.%s -> %s.id ondelete=%s) failed: %s. "
            "Existing FK still dropped — manual repair may be needed.",
            new_name, table, column, parent_table, ondelete, exc,
        )


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

    # ── Convert CASCADE → RESTRICT on integration parent FKs.
    # This refuses any direct DELETE FROM integrations row while
    # AI/Bailian history exists — operators MUST soft-delete via
    # ``Integration.enabled=False`` (or the upcoming integration
    # soft-delete column). The application layer never hard-deletes
    # integrations today; this is defense in depth against a future
    # bug or a hand-typed admin DELETE.
    for table, column, parent in _FK_RETARGET_RESTRICT:
        new_name = f"fk_{table}_{column}_restrict"
        _drop_and_recreate_fk(
            bind,
            table=table,
            column=column,
            parent_table=parent,
            ondelete="RESTRICT",
            new_name=new_name,
        )

    # ── Convert CASCADE → SET NULL for usage_daily.workspace_pk.
    # bailian_usage_daily is an immutable billing ledger — if we ever
    # prune a workspace row, the per-(day × account × model) usage
    # rows MUST stay (auditors / chargeback / dispute resolution all
    # depend on them). Make the column nullable first if needed.
    for table, column, parent in _FK_RETARGET_SET_NULL:
        if _column_exists(bind, table, column) and not _column_is_nullable(
            bind, table, column,
        ):
            try:
                op.alter_column(
                    table, column,
                    existing_type=sa.String(36),
                    nullable=True,
                )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "122: alter %s.%s nullable=True failed: %s",
                    table, column, exc,
                )
        new_name = f"fk_{table}_{column}_set_null"
        _drop_and_recreate_fk(
            bind,
            table=table,
            column=column,
            parent_table=parent,
            ondelete="SET NULL",
            new_name=new_name,
        )

    # ── Audit-log lookup index. AuditLog.actor_user_id was previously
    # FK-less + index-less; ``WHERE actor_user_id = :id`` lookups
    # full-scanned the table. Index name uses the standard
    # ``ix_audit_logs_actor_user_id`` shape so future migrations don't
    # collide.
    if _table_exists(bind, "audit_logs") and _column_exists(
        bind, "audit_logs", "actor_user_id",
    ):
        existing_idx = bind.execute(
            sa.text(
                "SELECT 1 FROM information_schema.statistics "
                "WHERE table_schema = DATABASE() "
                "AND table_name = 'audit_logs' "
                "AND index_name = 'ix_audit_logs_actor_user_id' "
                "LIMIT 1",
            ),
        ).scalar()
        if not existing_idx:
            try:
                op.create_index(
                    "ix_audit_logs_actor_user_id",
                    "audit_logs",
                    ["actor_user_id"],
                )
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "122: create ix_audit_logs_actor_user_id failed: %s",
                    exc,
                )


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

    # Reverse the RESTRICT/SET NULL changes back to CASCADE — note this
    # restores the historical history-wipe risk; only run downgrade if
    # you've also verified no CASCADE-triggering DELETE will fire.
    for table, column, parent in _FK_RETARGET_RESTRICT:
        existing = _find_fk_name(bind, table, column)
        if existing:
            try:
                op.drop_constraint(existing, table, type_="foreignkey")
            except Exception:  # noqa: BLE001
                pass
        try:
            op.create_foreign_key(
                f"fk_{table}_{column}_cascade",
                table,
                parent,
                [column],
                ["id"],
                ondelete="CASCADE",
            )
        except Exception:  # noqa: BLE001
            pass

    for table, column, parent in _FK_RETARGET_SET_NULL:
        existing = _find_fk_name(bind, table, column)
        if existing:
            try:
                op.drop_constraint(existing, table, type_="foreignkey")
            except Exception:  # noqa: BLE001
                pass
        try:
            op.create_foreign_key(
                f"fk_{table}_{column}_cascade",
                table,
                parent,
                [column],
                ["id"],
                ondelete="CASCADE",
            )
        except Exception:  # noqa: BLE001
            pass

    # Drop the audit-log lookup index.
    bind = op.get_bind()
    if _table_exists(bind, "audit_logs"):
        try:
            op.drop_index(
                "ix_audit_logs_actor_user_id",
                table_name="audit_logs",
            )
        except Exception:  # noqa: BLE001
            pass