"""API-key AK/SK rotation: ``rotation_retire_at`` column + retire cron seed.

Self-service credential rotation for cloud-auto-granted API keys (AWS
Bedrock IAM AK/SK, Aliyun Bailian DashScope sk-, Azure OpenAI account
key). Rotation mints a *fresh* credential that goes live immediately
while the *old* credential keeps working for a grace period (dual-
active), then the old one is retired automatically. The retire schedule
lives on ``detail.<provider>.rotation`` (provider-specific handle) but we
also denormalise the retire timestamp onto a first-class column so:

* the ``apikey.rotation_retire`` cron can scan due rotations with an
  index instead of a JSON predicate over every assignment, and
* the employee / admin UI can render the "轮换中 · 旧 key X 停用" badge
  without decrypting ``key_value`` or parsing the detail blob.

``rotation_retire_at`` is:
* ``NULL`` on every assignment not currently mid-rotation (the vast
  majority — legacy rows and post-retire rows both sit at NULL).
* A UTC timestamp = the moment the *previous* credential should be torn
  down. Set at rotation start; cleared (back to NULL) when the retire
  cron / force-complete / cancel resolves the rotation.

Also seeds the ``apikey.rotation_retire`` scheduled job (hourly). Hourly
rather than daily so a leaked-credential rotation with a short grace
(or the retire-reminder 24h before) fires with at most ~1h of extra
exposure rather than up to a day.
"""
import json as _json
import uuid

import sqlalchemy as sa

from alembic import op

revision = "401"
down_revision = "400"
branch_labels = None
depends_on = None


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


HANDLER_KEY = "apikey.rotation_retire"
JOB_NAME = "apikey.rotation_retire"

_DESC = (
    "每小时扫描处于「轮换中」(rotation_retire_at 非空且未来时刻已到) 的 "
    "ApiKeyAssignment:退役旧凭证(Bedrock delete_access_key / Bailian "
    "DeleteApiKey / Azure regenerateKey 旧槽),清空轮换状态并通知申请人"
    "「旧 key 已停用」。同时对 24h 内即将退役且尚未提醒过的轮换发送"
    "「旧 key 即将停用」提醒。退役对云侧幂等:凭证已被吊销/离职清理删除时"
    "直接 no-op。"
)


def upgrade() -> None:
    bind = op.get_bind()
    if not _table_exists(bind, "api_key_assignments"):
        return
    if not _column_exists(bind, "api_key_assignments", "rotation_retire_at"):
        op.add_column(
            "api_key_assignments",
            sa.Column("rotation_retire_at", sa.DateTime(timezone=True), nullable=True),
        )
    if not _index_exists(
        bind, "api_key_assignments", "ix_api_key_assignments_rotation_retire_at",
    ):
        op.create_index(
            "ix_api_key_assignments_rotation_retire_at",
            "api_key_assignments",
            ["rotation_retire_at"],
        )

    # Seed the retire cron (idempotent on handler_key).
    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": JOB_NAME,
                "display": "API Key · 轮换旧凭证退役",
                "desc": _DESC,
                "handler": HANDLER_KEY,
                # Hourly at :19 — offset off the top of the hour so it
                # doesn't pile onto the many :00 ticks.
                "cron": "19 * * * *",
                "config": _json.dumps({}),
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _table_exists(bind, "scheduled_jobs"):
        bind.execute(
            sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"),
            {"k": HANDLER_KEY},
        )
    if not _table_exists(bind, "api_key_assignments"):
        return
    if _index_exists(
        bind, "api_key_assignments", "ix_api_key_assignments_rotation_retire_at",
    ):
        op.drop_index(
            "ix_api_key_assignments_rotation_retire_at",
            table_name="api_key_assignments",
        )
    if _column_exists(bind, "api_key_assignments", "rotation_retire_at"):
        op.drop_column("api_key_assignments", "rotation_retire_at")