"""Identity-sync per-run ledger (run items) + retention cron.

One row per vendor-touching operation a sweep / single-person push
performed: subject, outcome, the rendered request payload as sent, the
vendor's response detail / error. Backs the per-run drill-down in the
数据同步 UI and the person-by-person cutover testing workflow (Authing
replacement PR2/5).

Seeded cron ``identity_sync.run_items_cleanup`` (daily 04:30 UTC)
deletes ledger rows older than 90 days. Hyphenated UUID id — see the
no-dash-UUID attach bug (migrations 200/248/267).
"""
from __future__ import annotations

import json
import uuid

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import (
    id_type_for,
    normalize_fk_column_collation,
    table_kw_for,
)

revision = "273"
down_revision = "272"
branch_labels = None
depends_on = None

TABLE = "identity_sync_run_items"
HANDLER_KEY = "identity_sync.run_items_cleanup"


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 upgrade() -> None:
    bind = op.get_bind()
    runs_id = id_type_for("identity_sync_runs")
    targets_id = id_type_for("identity_sync_targets")
    persons_id = id_type_for("identity_persons")
    kw = table_kw_for("identity_sync_runs")

    if not _table_exists(bind, TABLE):
        op.create_table(
            TABLE,
            sa.Column("id", runs_id, primary_key=True),
            sa.Column("run_id", runs_id, nullable=False),
            sa.Column("target_id", targets_id, nullable=False),
            sa.Column("item_type", sa.String(16), nullable=False),
            sa.Column("person_id", persons_id, nullable=True),
            sa.Column("subject_display", sa.String(255), nullable=False),
            sa.Column("external_key", sa.String(255), nullable=True),
            sa.Column("outcome", sa.String(16), nullable=False),
            sa.Column("request_payload", sa.JSON(), nullable=True),
            sa.Column("response_detail", sa.JSON(), nullable=True),
            sa.Column("error", sa.Text(), nullable=True),
            sa.Column("duration_ms", sa.Integer(), nullable=True),
            sa.Column(
                "created_at", sa.DateTime(timezone=True), nullable=False,
                server_default=sa.func.now(),
            ),
            sa.ForeignKeyConstraint(
                ["run_id"], ["identity_sync_runs.id"],
                name="fk_identity_sync_run_items_run", ondelete="CASCADE",
            ),
            sa.ForeignKeyConstraint(
                ["target_id"], ["identity_sync_targets.id"],
                name="fk_identity_sync_run_items_target", ondelete="CASCADE",
            ),
            # NOTE: the person_id FK is created out-of-line below. The table
            # inherits identity_sync_runs' collation (utf8mb4_bin on prod TiDB),
            # but identity_persons.id is utf8mb4_unicode_ci — an inline FK trips
            # error 3780 ("Referencing column ... are incompatible"). run_id /
            # target_id share the table collation so their inline FKs are fine.
            **kw,
        )
        # Match person_id's charset/collation to identity_persons.id, then add
        # the FK. No-op on a uniform-collation DB (CI MySQL 8); on prod it
        # ALTERs person_id to utf8mb4_unicode_ci so the FK is accepted.
        normalize_fk_column_collation(
            TABLE, "person_id", "identity_persons", nullable=True,
        )
        op.create_foreign_key(
            "fk_identity_sync_run_items_person", TABLE, "identity_persons",
            ["person_id"], ["id"], ondelete="SET NULL",
        )
        op.create_index(
            "ix_identity_sync_run_items_run", TABLE, ["run_id"],
        )
        op.create_index(
            "ix_identity_sync_run_items_target_created", TABLE,
            ["target_id", "created_at"],
        )
        op.create_index(
            "ix_identity_sync_run_items_person", TABLE, ["person_id"],
        )

    if _table_exists(bind, "scheduled_jobs"):
        bind.execute(
            sa.text(
                "INSERT INTO scheduled_jobs "
                "(id, name, display_name, description, handler_key, "
                " trigger_type, cron_expr, timezone, enabled, config, "
                " run_count, failure_count, created_at, updated_at) "
                "SELECT :id, :name, :display, :desc, :handler, 'cron', "
                ":cron, 'UTC', 1, :config, 0, 0, NOW(), NOW() "
                "FROM DUAL WHERE NOT EXISTS "
                "(SELECT 1 FROM scheduled_jobs WHERE handler_key = :handler)"
            ),
            {
                "id": str(uuid.uuid4()),
                "name": HANDLER_KEY,
                "display": "数据同步台账清理",
                "desc": (
                    "每天 04:30 UTC 删除 90 天前的 identity_sync_run_items "
                    "逐条同步台账(retention_days 可在 config 调整)。运行行"
                    "(identity_sync_runs)与状态行不受影响。"
                ),
                "handler": HANDLER_KEY,
                "cron": "30 4 * * *",
                "config": json.dumps({"retention_days": 90}),
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _table_exists(bind, "scheduled_jobs"):
        bind.execute(
            sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :h"),
            {"h": HANDLER_KEY},
        )
    if _table_exists(bind, TABLE):
        op.drop_table(TABLE)