"""CMDB legacy-ID rescue + bare-metal integration seed.

The legacy-CMDB migration script (server/scripts/migrate_cmdb.py)
synthesized per-row instance_ids of the form

    legacy-{vendor_id}-{src_id}-{raw_iid}

so the same legacy ``instance_id`` from two different vendor rows
wouldn't collapse into one. The bug: the live cloud-provider drivers
return the raw cloud-issued ID (e.g. UCloud ``uhost-c1pa0wrhq5u``,
Aliyun ``i-xxx``), so the unique key ``(integration_id, instance_id)``
never matches between migrated rows and live-sync rows. Every live
sync after the migration created a NEW shadow row for each cloud-side
instance AND ``_safe_sweep`` marked every ``legacy-*`` row
``deleted_at=NOW()`` as a missing orphan.

Prod impact 2026-04-24:
  * UCloud sync ``f92da3de-cc..``: ``upserted=31, deleted=116`` —
    every UCloud server migrated from legacy CMDB was soft-deleted in
    one sweep. Includes ``10.9.84.114 / DLP-test`` (the user-facing
    incident that triggered this fix).
  * Ksyun-legacy (541 rows) survived only because the integration is
    ``enabled=false`` — cron skips it, no sweep ran. Still normalize
    so a future enable doesn't reproduce the same incident.

This migration does three things, all idempotent:

  1. Seed a dedicated ``legacy-bare-metal`` integration
     (``kind=cloud``, ``enabled=false``, no driver). It's the
     never-iterated parking lot for rows whose ``instance_id`` was
     synthesized because no cloud-issued ID exists (true IDC /
     bare-metal). Cron iterates only enabled cloud integrations, so
     rows parked here are safe from ``_safe_sweep`` forever.
  2. Normalize ``cmdb_servers.instance_id`` and ``cmdb_load_balancers.lb_id``:
     strip the ``legacy-{vendor}-{src}-`` prefix, recover the raw
     cloud-issued ID, clear ``deleted_at`` if the row was swept by the
     ID-shape mismatch. On collision with a live-sync row, merge
     owner_person_id / app / remark / owner_raw onto the live row when
     they're NULL there, then drop the legacy row (its id was a
     migration UUID; nothing else references it).
  3. Move pure-synthetic (no raw_iid) rows under the
     ``legacy-bare-metal`` integration with key
     ``legacy-bare-{vendor}-{src_id}``.

cmdb_dns_records and cmdb_cdn_domains are not touched by this rescue:
they use ``(integration_id, zone_name, record_id)`` /
``(integration_id, domain_name)`` keys whose legacy values DID match
the live drivers in most cases (per prod sample inspection). Any
remaining sweep damage there is caught going forward by the
``_safe_sweep`` ID-shape mismatch guard added in this PR.

Revision ID: 212
Revises: 211
Create Date: 2026-05-28
"""
from __future__ import annotations

import json
import logging
import uuid
from datetime import datetime, timezone

import sqlalchemy as sa
from alembic import op

revision = "212"
down_revision = "211"
branch_labels = None
depends_on = None

logger = logging.getLogger(__name__)

BARE_METAL_SLUG = "legacy-bare-metal"
BARE_METAL_NAME = "Legacy / Bare-metal (no driver)"


def _split_legacy_id(value: str) -> tuple[str | None, str | None, str]:
    """Parse ``legacy-{vendor_id}-{src_id}-{raw_iid}`` → (vendor, src, raw).

    Returns ``(None, None, value)`` if the string isn't a legacy
    prefix. ``raw_iid`` is everything after the 3rd ``-``, preserved
    verbatim (cloud IDs like UUIDs that contain ``-`` survive intact).
    """
    if not value or not value.startswith("legacy-"):
        return None, None, value
    parts = value.split("-", 3)
    if len(parts) < 4:
        return None, None, value
    _, vendor_id, src_id, raw = parts
    return vendor_id, src_id, raw


def _seed_bare_metal_integration(conn) -> str:
    """Insert the ``legacy-bare-metal`` cloud integration if absent.
    Returns its id (existing or new). Idempotent under re-run."""
    existing = conn.execute(
        sa.text("SELECT id FROM integrations WHERE slug = :slug"),
        {"slug": BARE_METAL_SLUG},
    ).fetchone()
    if existing:
        return existing[0]
    new_id = str(uuid.uuid4())
    now = datetime.now(timezone.utc)
    conn.execute(
        sa.text(
            "INSERT INTO integrations "
            "  (id, provider, name, description, "
            "   secrets_encrypted, secrets_nonce, config_public, "
            "   enabled, kind, slug, is_primary, capabilities, "
            "   created_at, updated_at) "
            "VALUES "
            "  (:id, :provider, :name, :description, "
            "   NULL, NULL, NULL, "
            "   :enabled, :kind, :slug, :is_primary, :capabilities, "
            "   :now, :now)"
        ),
        {
            "id": new_id,
            "provider": "none",  # sentinel — no driver class registered
            "name": BARE_METAL_NAME,
            "description": (
                "Catch-all for migrated rows without a cloud-issued "
                "instance_id (IDC / bare-metal). Disabled by design so "
                "cloud-sync cron never iterates over it and "
                "_safe_sweep can never reach the rows parked here."
            ),
            "enabled": 0,  # MUST stay disabled — see docstring
            "kind": "cloud",
            "slug": BARE_METAL_SLUG,
            "is_primary": 0,
            "capabilities": json.dumps([]),
            "now": now,
        },
    )
    logger.info("seeded legacy-bare-metal integration id=%s", new_id)
    return new_id


def _rescue_table(
    conn,
    *,
    table: str,
    key_column: str,
    bare_metal_id: str,
    merge_columns: list[str],
) -> dict[str, int]:
    """Walk all rows in ``table`` whose ``key_column`` looks like
    ``legacy-{vendor}-{src}-...``. Strip the prefix to recover the
    raw cloud-issued ID, then:

      * If the row has a real raw_iid AND no live row collides on
        ``(integration_id, raw_iid)``: UPDATE in-place — set
        ``key_column=raw_iid`` and ``deleted_at=NULL``.
      * If a live row collides: MERGE — fill missing ``merge_columns``
        on the live row from the legacy row, then DELETE the legacy
        row (its row_id was a migration UUID, never referenced).
        Soft-delete is not enough here — leaving the legacy row alive
        with the same effective payload would re-orphan on the next
        sync once it matches a live ID.
      * If the row has no raw_iid (true bare-metal): MOVE to the
        ``legacy-bare-metal`` integration with key
        ``legacy-bare-{vendor}-{src}`` and ``deleted_at=NULL``.

    Returns counter dict for logging.
    """
    counts = {
        "in_place": 0,    # legacy row updated with new key + undeleted
        "merged": 0,      # legacy row merged into live + deleted
        "bare_metal": 0,  # legacy row moved to bare-metal int
        "unchanged": 0,   # legacy row already a bare format, no-op
    }
    # Pull every legacy-prefixed row plus the columns we may merge.
    # Backtick every identifier — ``app`` is a MySQL reserved-ish word
    # in some contexts (the bare-token form silently produces a SQL
    # syntax error when the parser picks the keyword interpretation).
    # Same defensive quoting pattern applied to every subsequent UPDATE
    # / SELECT in this migration.
    select_cols = (
        "`id`, `integration_id`, `" + key_column + "`, `deleted_at`, "
        + ", ".join(f"`{c}`" for c in merge_columns)
    )
    rows = conn.execute(
        sa.text(
            f"SELECT {select_cols} FROM `{table}` "
            f"WHERE `{key_column}` LIKE 'legacy-%'"
        )
    ).mappings().all()
    unchanged_samples: list[tuple[str, str]] = []
    for r in rows:
        legacy_key = r[key_column]
        vendor_id, src_id, raw = _split_legacy_id(legacy_key)
        if not vendor_id or src_id is None:
            # Malformed ``legacy-`` prefix (e.g. ``legacy-X`` with no
            # src_id). Cannot recover the raw cloud-id mechanically;
            # leave the row as-is (operator must hand-resolve). Log
            # provenance so admin can grep for the affected rows
            # without re-querying the source DB later.
            counts["unchanged"] += 1
            if len(unchanged_samples) < 20:
                unchanged_samples.append((str(r["id"]), legacy_key))
            continue
        if vendor_id == "bare":
            # ``legacy-bare-{vendor}-{src_id}`` is the NEW bare-metal
            # synthetic shape written by the updated migrate_cmdb.py
            # (rows without a cloud-issued instance_id parked under
            # ``legacy-bare-metal`` integration). These rows are
            # already in their intended final state — do NOT parse
            # the ``bare`` token as a vendor and rewrite the row to
            # ``instance_id=src_id`` under the live UCloud / Aliyun /
            # etc. integration. Defensive guard against re-application
            # of this migration via ``op.run("212")`` after migrate_cmdb
            # has written bare-metal rows — alembic itself won't
            # re-run the migration, but copy-paste into a future
            # migration could hit this path.
            counts["unchanged"] += 1
            continue
        if raw:
            new_key = raw[:512]
            collision_cols = (
                "`id`, `deleted_at`, "
                + ", ".join(f"`{c}`" for c in merge_columns)
            )
            collision = conn.execute(
                sa.text(
                    f"SELECT {collision_cols} FROM `{table}` "
                    f"WHERE `integration_id` = :iid "
                    f"AND `{key_column}` = :k "
                    "AND `id` != :self_id LIMIT 1"
                ),
                {"iid": r["integration_id"], "k": new_key, "self_id": r["id"]},
            ).mappings().first()
            if collision:
                # Merge legacy row's metadata onto the live row.
                merge_updates: dict[str, object] = {}
                for col in merge_columns:
                    if collision[col] is None and r[col] is not None:
                        merge_updates[col] = r[col]
                if merge_updates:
                    assignments = ", ".join(
                        f"`{col}` = :{col}" for col in merge_updates
                    )
                    conn.execute(
                        sa.text(
                            f"UPDATE `{table}` SET {assignments} "
                            "WHERE `id` = :tid"
                        ),
                        {**merge_updates, "tid": collision["id"]},
                    )
                # Also clear deleted_at on the live row if it was
                # tombstoned (e.g. UCloud sync's collateral sweep that
                # picked up a previously-alive uhost-format row
                # because the integration's total active count dropped
                # below the floor). The live row IS the one cloud-side,
                # so it must be active.
                if collision["deleted_at"] is not None:
                    conn.execute(
                        sa.text(
                            f"UPDATE `{table}` SET `deleted_at` = NULL "
                            "WHERE `id` = :tid"
                        ),
                        {"tid": collision["id"]},
                    )
                conn.execute(
                    sa.text(f"DELETE FROM `{table}` WHERE `id` = :id"),
                    {"id": r["id"]},
                )
                counts["merged"] += 1
            else:
                # No collision — UPDATE in-place. Clears deleted_at
                # too so the row reappears immediately on the next
                # ``GET /cmdb/servers`` request.
                conn.execute(
                    sa.text(
                        f"UPDATE `{table}` SET `{key_column}` = :k, "
                        "`deleted_at` = NULL WHERE `id` = :id"
                    ),
                    {"k": new_key, "id": r["id"]},
                )
                counts["in_place"] += 1
        else:
            # No raw_iid → true bare-metal. Park under bare-metal int.
            # Build a deterministic key from vendor + src_id so re-runs
            # land on the same row.
            new_key = f"legacy-bare-{vendor_id}-{src_id}"
            # Collision check INSIDE the bare-metal integration in case
            # this rescue ran before and we're a re-run.
            existing = conn.execute(
                sa.text(
                    f"SELECT `id` FROM `{table}` "
                    f"WHERE `integration_id` = :iid "
                    f"AND `{key_column}` = :k "
                    "AND `id` != :self_id LIMIT 1"
                ),
                {"iid": bare_metal_id, "k": new_key, "self_id": r["id"]},
            ).first()
            if existing:
                # Re-run case: just drop this row, the prior pass kept the
                # canonical one. (Merge logic identical to live-collision
                # path is overkill for the second-time-around case.)
                conn.execute(
                    sa.text(f"DELETE FROM `{table}` WHERE `id` = :id"),
                    {"id": r["id"]},
                )
            else:
                conn.execute(
                    sa.text(
                        f"UPDATE `{table}` SET `integration_id` = :iid, "
                        f"`{key_column}` = :k, `deleted_at` = NULL "
                        "WHERE `id` = :id"
                    ),
                    {
                        "iid": bare_metal_id,
                        "k": new_key,
                        "id": r["id"],
                    },
                )
            counts["bare_metal"] += 1
    if unchanged_samples:
        logger.warning(
            "%s: %d row(s) with malformed legacy- prefix left "
            "untouched. First %d (id, instance_key): %s",
            table, counts["unchanged"], len(unchanged_samples),
            unchanged_samples,
        )
    return counts


def upgrade() -> None:
    conn = op.get_bind()
    bare_metal_id = _seed_bare_metal_integration(conn)

    # cmdb_servers — owner_person_id / app / remark are the
    # admin-curated fields worth preserving in a merge. owner_raw was
    # dropped from cmdb_servers in migration 042 so it's not in the list.
    server_counts = _rescue_table(
        conn,
        table="cmdb_servers",
        key_column="instance_id",
        bare_metal_id=bare_metal_id,
        merge_columns=["owner_person_id", "app", "remark"],
    )
    logger.info("cmdb_servers rescue: %s", server_counts)

    # cmdb_load_balancers — owner_person_id / owner_raw / app / remark
    lb_counts = _rescue_table(
        conn,
        table="cmdb_load_balancers",
        key_column="lb_id",
        bare_metal_id=bare_metal_id,
        merge_columns=["owner_person_id", "owner_raw", "app", "remark"],
    )
    logger.info("cmdb_load_balancers rescue: %s", lb_counts)


def downgrade() -> None:
    # Irreversible by design — we've stripped the ``legacy-{vendor}-
    # {src_id}-`` prefix from the migration provenance. The original
    # ``raw_data`` JSON still carries the legacy ``id`` field if
    # forensic recovery is ever needed.
    #
    # Re-adding the prefix in a downgrade would not be safe (we don't
    # know which rows came from the migration vs. were created by
    # live sync at this point), so we leave both tables as-is and
    # accept that this revision is forward-only.
    pass