"""Identity-consumer periodic re-certification.

Reuses the access-certification engine (campaign_type ``identity_consumer``)
to re-attest every active consumer each quarter. This migration adds the
two tracking columns the live consumer needs and seeds the two crons:

* ``identity_consumers.last_certified_at`` — stamped when an auditor
  decides ``keep`` on the consumer's recert item.
* ``identity_consumers.recert_lapsed_at`` — set when the lapse sweep
  auto-disables a consumer whose recert went undecided past the grace
  window; cleared (and the consumer re-enabled) on a later ``keep``.

Crons (seeded into ``scheduled_jobs``, idempotent, mirrors migration 094):

* ``auditor.identity_consumer_recert_auto_open_quarterly`` —
  ``0 9 1 1,4,7,10 *`` UTC: open a quarterly ``identity_consumer``
  campaign (idempotent per quarter).
* ``auditor.identity_consumer_recert_lapse_sweep`` — ``0 10 * * *`` UTC:
  auto-disable consumers whose recert item is pending past the grace
  window (stops PII egress for un-recertified consumers).
"""
from __future__ import annotations

import json
import uuid

import sqlalchemy as sa

from alembic import op

revision = "287"
down_revision = "286"
branch_labels = None
depends_on = None

CONSUMERS = "identity_consumers"

_JOBS = [
    {
        "name": "auditor.identity_consumer_recert_auto_open_quarterly",
        "display_name": "安全员 · 身份消费方季度复核自动开盘",
        "description": (
            "每季度首日 09:00 UTC 自动开一个身份消费方访问复核活动"
            "(campaign_type=identity_consumer),把所有 active 消费方挂到 "
            "/compliance/certifications 等审计员逐个 keep/revoke;"
            "label 形如 '2026-Q2 身份消费方复核',该季度已有则幂等跳过。"
        ),
        "cron": "0 9 1 1,4,7,10 *",
        "config": {"due_after_days": 30},
    },
    {
        "name": "auditor.identity_consumer_recert_lapse_sweep",
        "display_name": "安全员 · 身份消费方复核逾期自动停用",
        "description": (
            "每日 10:00 UTC 扫描身份消费方复核项:决定仍 pending 且已超过"
            "截止+14 天宽限期的,自动把对应消费方置 disabled(停止对外 PII "
            "出口)并标记 recert_lapsed_at;审计员事后判 keep 会自动重新启用。"
        ),
        "cron": "0 10 * * *",
        "config": {},
    },
]


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 upgrade() -> None:
    bind = op.get_bind()
    for col in ("last_certified_at", "recert_lapsed_at"):
        if not _column_exists(bind, CONSUMERS, col):
            op.add_column(
                CONSUMERS, sa.Column(col, sa.DateTime(timezone=True), nullable=True),
            )

    if not _table_exists(bind, "scheduled_jobs"):
        return
    for job in _JOBS:
        exists = bind.execute(
            sa.text("SELECT 1 FROM scheduled_jobs WHERE name = :n LIMIT 1"),
            {"n": job["name"]},
        ).scalar()
        if exists:
            continue
        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) "
                "VALUES (:id, :n, :dn, :d, :hk, 'cron', :ce, 'UTC', 1, :cfg, "
                " 0, 0, NOW(), NOW())"
            ),
            {
                "id": str(uuid.uuid4()),
                "n": job["name"],
                "dn": job["display_name"],
                "d": job["description"],
                "hk": job["name"],
                "ce": job["cron"],
                "cfg": json.dumps(job["config"]),
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    for job in _JOBS:
        bind.execute(
            sa.text("DELETE FROM scheduled_jobs WHERE name = :n"),
            {"n": job["name"]},
        )
    for col in ("recert_lapsed_at", "last_certified_at"):
        if _column_exists(bind, CONSUMERS, col):
            op.drop_column(CONSUMERS, col)