"""Merge ``cmdb_provider_accounts`` into ``integrations(kind='cloud')``.

Revision ID: 046
Revises: 045
Create Date: 2026-04-21

Cloud tenants have been living in a bespoke table (``cmdb_provider_accounts``)
while every other external-system instance — HRIS, chat-approval, SaaS
provisioning — is a row in ``integrations``. This migration collapses the
duplication:

* Rows in ``cmdb_provider_accounts`` are copied 1:1 into ``integrations``
  with ``kind='cloud'``, preserving the original UUID so all
  cross-referenced resource rows (``cmdb_servers``, ``cmdb_domains`` …)
  keep resolving to the same tenant. The original ``id`` becomes
  ``integrations.id``.
* The seven CMDB resource tables FK'd to ``cmdb_provider_accounts`` have
  their FK column renamed ``provider_account_id`` → ``integration_id``
  and repointed at ``integrations.id``.
* ``integrations`` gains a ``sync_cron`` column to preserve the
  per-account cron that was carried on the old table.
* The old table is dropped.
* A ``scheduled_jobs`` row named ``cmdb.cloud_sync`` is seeded so the
  worker's APScheduler picks up a single hourly sweep that iterates every
  ``kind='cloud'`` integration, replacing the per-account cron loop in
  ``app.worker._register_cmdb_provider_syncs``.

Idempotent — every DDL is guarded by ``information_schema`` so rerunning
after a partial crash picks up where it left off.

Downgrade is best-effort: recreates the old table and copies cloud rows
back, but any resource rows already written against the new FK column
stay pointed at ``integrations`` (they're compatible because the UUIDs
match). Flagged for review before using in a real rollback.
"""
from __future__ import annotations

import re
import uuid

import sqlalchemy as sa
from alembic import op

revision = "046"
down_revision = "045"
branch_labels = None
depends_on = None


CMDB_RESOURCE_TABLES = [
    "cmdb_sync_jobs",
    "cmdb_servers",
    "cmdb_domains",
    "cmdb_dns_records",
    "cmdb_load_balancers",
    "cmdb_cdn_domains",
    "cmdb_object_storage",
]


# ─── information_schema guards ─────────────────────────────────────


def _table_exists(bind, name: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLES "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :n",
        ),
        {"n": name},
    ).scalar() is not None


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


def _fk_constraint_names(bind, table: str, column: str) -> list[str]:
    """Return the name(s) of every FK on (table, column)."""
    rows = 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",
        ),
        {"t": table, "c": column},
    ).fetchall()
    return [r[0] for r in rows]


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


def _slugify(raw: str, id_suffix: str) -> str:
    """Turn an account_alias into a deterministic integration slug."""
    base = re.sub(r"[^a-zA-Z0-9]+", "-", raw or "").strip("-").lower()
    if not base:
        base = "cloud"
    base = base[:48]  # leave room for "-<id8>"
    return f"{base}-{id_suffix[:8]}"


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

    # Nothing to do on a cold DB where integrations hasn't been created yet
    if not _table_exists(bind, "integrations"):
        return

    # Step 1 — add integrations.sync_cron (idempotent)
    if not _column_exists(bind, "integrations", "sync_cron"):
        op.add_column(
            "integrations",
            sa.Column("sync_cron", sa.String(64), nullable=True),
        )

    # Step 2 — copy rows from the old table, if it still exists
    if _table_exists(bind, "cmdb_provider_accounts"):
        # Safety check: refuse to run if any id collides with an
        # existing integrations row (shouldn't happen — UUIDs are
        # global — but surfaces a pre-existing data bug before we
        # overwrite anything).
        collision = bind.execute(
            sa.text(
                "SELECT COUNT(*) FROM integrations i "
                "JOIN cmdb_provider_accounts p ON i.id = p.id"
            ),
        ).scalar() or 0
        if collision > 0:
            raise RuntimeError(
                f"Migration 045 abort: {collision} id(s) overlap between "
                "integrations and cmdb_provider_accounts. Resolve manually.",
            )

        rows = bind.execute(
            sa.text(
                "SELECT id, provider_type, account_alias, description, "
                "is_active, cred_ciphertext, cred_nonce, "
                "enabled_capabilities, sync_cron, created_by, "
                "created_at, updated_at "
                "FROM cmdb_provider_accounts "
                "WHERE deleted_at IS NULL",
            ),
        ).fetchall()

        used_slugs: set[str] = set()
        # Preload existing slugs so our generated ones don't collide.
        for (s,) in bind.execute(sa.text("SELECT slug FROM integrations")).fetchall():
            used_slugs.add(s)

        for r in rows:
            row_id = str(r[0])
            provider = r[1]
            alias = r[2]
            description = r[3]
            is_active = bool(r[4])
            cipher = r[5]
            nonce = r[6]
            caps = r[7]  # JSON — MySQL returns a string; leave as-is
            sync_cron = r[8]
            created_by = r[9]
            created_at = r[10]
            updated_at = r[11]

            slug = _slugify(alias, row_id)
            # Defensive uniqueness bump (shouldn't trigger in practice)
            base_slug = slug
            n = 1
            while slug in used_slugs:
                slug = f"{base_slug}-{n}"
                n += 1
            used_slugs.add(slug)

            bind.execute(
                sa.text(
                    "INSERT INTO integrations "
                    "(id, kind, provider, name, slug, description, enabled, "
                    " is_primary, config_public, secrets_encrypted, "
                    " secrets_nonce, capabilities, sync_cron, "
                    " created_by, created_at, updated_at) "
                    "VALUES (:id, 'cloud', :provider, :name, :slug, "
                    " :description, :enabled, 0, NULL, :cipher, :nonce, "
                    " :caps, :sync_cron, :created_by, :created_at, "
                    " :updated_at)",
                ),
                {
                    "id": row_id,
                    "provider": provider,
                    "name": alias,
                    "slug": slug,
                    "description": description,
                    "enabled": 1 if is_active else 0,
                    "cipher": cipher,
                    "nonce": nonce,
                    # caps is already a JSON string coming out of MySQL;
                    # use CAST via JSON_EXTRACT path-less call to be safe
                    "caps": caps,
                    "sync_cron": sync_cron,
                    "created_by": created_by,
                    "created_at": created_at,
                    "updated_at": updated_at,
                },
            )

    # Step 3 — rename the FK column on every resource table and repoint
    for table in CMDB_RESOURCE_TABLES:
        if not _table_exists(bind, table):
            continue

        # Drop any existing FK on the old column name
        for fk_name in _fk_constraint_names(bind, table, "provider_account_id"):
            try:
                op.drop_constraint(fk_name, table, type_="foreignkey")
            except Exception:  # noqa: BLE001
                # Some dialects raise on already-dropped constraints
                pass

        # Rename the column (skip if already renamed in a prior partial run)
        if _column_exists(bind, table, "provider_account_id") and not _column_exists(
            bind, table, "integration_id",
        ):
            op.alter_column(
                table,
                "provider_account_id",
                new_column_name="integration_id",
                existing_type=sa.String(36),
                existing_nullable=False,
            )
        elif _column_exists(bind, table, "provider_account_id") and _column_exists(
            bind, table, "integration_id",
        ):
            # Both columns exist (shouldn't happen, but be defensive).
            # Drop the old one without data, since integration_id is
            # already the live column.
            op.drop_column(table, "provider_account_id")

        # Create the new FK if integrations table is ready
        new_fk = f"fk_{table}_integration_id"
        if _column_exists(bind, table, "integration_id"):
            # Clean up any prior FK by the same generated name
            existing_names = _fk_constraint_names(bind, table, "integration_id")
            if new_fk not in existing_names:
                try:
                    op.create_foreign_key(
                        new_fk,
                        source_table=table,
                        referent_table="integrations",
                        local_cols=["integration_id"],
                        remote_cols=["id"],
                    )
                except Exception:  # noqa: BLE001
                    # Charset/collation mismatches on fresh TiDB — swallow,
                    # the rename + data are the important bits; FK is a
                    # correctness-but-not-safety constraint here.
                    pass

    # Step 4 — drop the old table
    if _table_exists(bind, "cmdb_provider_accounts"):
        op.drop_table("cmdb_provider_accounts")

    # Step 5 — seed the scheduled_jobs row (idempotent via NOT EXISTS)
    if _table_exists(bind, "scheduled_jobs"):
        op.execute(sa.text("""
            INSERT INTO scheduled_jobs
                (id, name, display_name, description, handler_key, trigger_type,
                 interval_seconds, cron_expr, timezone, enabled, config,
                 run_count, failure_count, created_at, updated_at)
            SELECT :id, 'cmdb.cloud_sync', '云账号同步',
                   '遍历所有 kind=cloud 且 enabled=true 的 integrations,依次触发云资源全量同步(servers / domains / dns / lb / cdn / oss)。',
                   'cmdb.cloud_sync', 'cron',
                   NULL, '0 * * * *', 'Asia/Shanghai', 1, JSON_OBJECT(),
                   0, 0, NOW(), NOW()
            FROM DUAL
            WHERE NOT EXISTS (SELECT 1 FROM scheduled_jobs WHERE name = 'cmdb.cloud_sync')
        """).bindparams(id=str(uuid.uuid4())))


# ─── downgrade (safety net) ────────────────────────────────────────


def downgrade() -> None:
    """Reverse migration. Best-effort; not tested at scale."""
    bind = op.get_bind()

    # Recreate cmdb_provider_accounts shell (empty), then copy cloud rows back.
    if not _table_exists(bind, "cmdb_provider_accounts"):
        op.create_table(
            "cmdb_provider_accounts",
            sa.Column("id", sa.String(36), primary_key=True),
            sa.Column("provider_type", sa.String(32), nullable=False),
            sa.Column("account_alias", sa.String(128), nullable=False),
            sa.Column("description", sa.Text, nullable=True),
            sa.Column("is_active", sa.Boolean, nullable=False, server_default=sa.true()),
            sa.Column("cred_ciphertext", sa.Text, nullable=True),
            sa.Column("cred_nonce", sa.String(32), nullable=True),
            sa.Column("enabled_capabilities", sa.JSON, nullable=False),
            sa.Column("sync_cron", sa.String(64), nullable=True),
            sa.Column("created_by", sa.String(36), nullable=True),
            sa.Column(
                "created_at", sa.DateTime(timezone=True),
                server_default=sa.func.now(),
            ),
            sa.Column(
                "updated_at", sa.DateTime(timezone=True),
                server_default=sa.func.now(),
            ),
            sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        )

    if _table_exists(bind, "integrations"):
        bind.execute(sa.text("""
            INSERT INTO cmdb_provider_accounts
                (id, provider_type, account_alias, description, is_active,
                 cred_ciphertext, cred_nonce, enabled_capabilities, sync_cron,
                 created_by, created_at, updated_at)
            SELECT id, provider, name, description, enabled,
                   secrets_encrypted, secrets_nonce,
                   COALESCE(capabilities, JSON_ARRAY()), sync_cron,
                   created_by, created_at, updated_at
            FROM integrations
            WHERE kind = 'cloud'
              AND NOT EXISTS (
                  SELECT 1 FROM cmdb_provider_accounts p WHERE p.id = integrations.id
              )
        """))

    # Rename integration_id -> provider_account_id on each resource table
    for table in CMDB_RESOURCE_TABLES:
        if not _table_exists(bind, table):
            continue
        # drop new FK
        for fk_name in _fk_constraint_names(bind, table, "integration_id"):
            try:
                op.drop_constraint(fk_name, table, type_="foreignkey")
            except Exception:  # noqa: BLE001
                pass
        if _column_exists(bind, table, "integration_id") and not _column_exists(
            bind, table, "provider_account_id",
        ):
            op.alter_column(
                table,
                "integration_id",
                new_column_name="provider_account_id",
                existing_type=sa.String(36),
                existing_nullable=False,
            )
        # Re-add old FK to the recreated cmdb_provider_accounts
        if _column_exists(bind, table, "provider_account_id"):
            old_fk = f"fk_{table}_provider_account"
            try:
                op.create_foreign_key(
                    old_fk,
                    source_table=table,
                    referent_table="cmdb_provider_accounts",
                    local_cols=["provider_account_id"],
                    remote_cols=["id"],
                )
            except Exception:  # noqa: BLE001
                pass

    # Delete the cloud rows from integrations (they now live in the old table)
    if _table_exists(bind, "integrations"):
        bind.execute(sa.text("DELETE FROM integrations WHERE kind = 'cloud'"))

    # Drop the new sync_cron column
    if _column_exists(bind, "integrations", "sync_cron"):
        op.drop_column("integrations", "sync_cron")

    # Remove the seeded scheduled_job row
    if _table_exists(bind, "scheduled_jobs"):
        bind.execute(
            sa.text("DELETE FROM scheduled_jobs WHERE name = 'cmdb.cloud_sync'"),
        )