"""Access review — assigned reviewer + deadline + escalation columns on
``auditor_privileged_role_holders``.

Revision ID: 232
Revises: 231
Create Date: 2026-06-01

Background
==========

PR-5 of the symmetric-onboarding series (independent track: 治理深化).
`AuditorPrivilegedRoleHolder` already implements the campaign primitive
(quarterly snapshot + per-holder decision + audit). What it doesn't
have yet:

* Per-item assignment — every holder is decidable by any
  ``security_auditor``. SOC 2 / ISO want a SPECIFIC reviewer accountable
  for each row.
* Per-item deadline — only the cycle has ``due_at``; per-row override
  lets sensitive items (e.g. AWS root) carry a tighter 7-day window
  while routine sysadmin rows stay on the cycle's 30-day window.
* Escalation tracking — 14 days past the deadline → write
  compliance_incident + push to assigned reviewer's manager so the
  row doesn't rot.

This migration adds 3 nullable columns + an index for the daily
escalation sweep. Idempotent (information_schema probes).

Companion code:
* ``PrivilegedReviewService.assign_reviewer`` — service helper.
* ``auditor.privileged_review_escalate`` cron — daily sweep.
* New ``compliance_incident`` type ``access_review_overdue``.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import id_type_for

revision = "232"
down_revision = "231"
branch_labels = None
depends_on = None

_TABLE = "auditor_privileged_role_holders"


def _column_exists(table: str, column: str) -> bool:
    row = (
        op.get_bind()
        .execute(
            sa.text(
                "SELECT 1 FROM information_schema.COLUMNS WHERE "
                "TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t "
                "AND COLUMN_NAME = :c LIMIT 1"
            ),
            {"t": table, "c": column},
        )
        .scalar()
    )
    return bool(row)


def _index_exists(table: str, index: str) -> bool:
    row = (
        op.get_bind()
        .execute(
            sa.text(
                "SELECT 1 FROM information_schema.STATISTICS WHERE "
                "TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t "
                "AND INDEX_NAME = :i LIMIT 1"
            ),
            {"t": table, "i": index},
        )
        .scalar()
    )
    return bool(row)


def upgrade() -> None:
    persons_id = id_type_for("identity_persons")

    if not _column_exists(_TABLE, "assigned_to_person_id"):
        op.add_column(
            _TABLE,
            sa.Column("assigned_to_person_id", persons_id, nullable=True),
        )
        op.create_foreign_key(
            "fk_apriv_holder_assigned",
            _TABLE,
            "identity_persons",
            ["assigned_to_person_id"],
            ["id"],
            ondelete="SET NULL",
        )

    if not _column_exists(_TABLE, "decided_at_deadline"):
        op.add_column(
            _TABLE,
            sa.Column(
                "decided_at_deadline",
                sa.DateTime(timezone=True),
                nullable=True,
            ),
        )

    if not _column_exists(_TABLE, "escalated_at"):
        op.add_column(
            _TABLE,
            sa.Column(
                "escalated_at", sa.DateTime(timezone=True), nullable=True,
            ),
        )

    # Composite index for the daily escalation sweep
    # (cycle_id, decision='pending', decided_at_deadline). The sweep
    # needs to find pending holders past their deadline cheaply.
    if not _index_exists(_TABLE, "idx_apriv_holder_pending_deadline"):
        op.create_index(
            "idx_apriv_holder_pending_deadline",
            _TABLE,
            ["decision", "decided_at_deadline"],
        )

    # Index on assigned_to_person_id so the "list_my_pending_reviews"
    # self-scoped MCP tool + manager-facing escalation lookups don't
    # full-scan.
    if not _index_exists(_TABLE, "idx_apriv_holder_assigned"):
        op.create_index(
            "idx_apriv_holder_assigned",
            _TABLE,
            ["assigned_to_person_id"],
        )

    # Seed the new daily escalation cron row.
    exists = (
        op.get_bind()
        .execute(
            sa.text(
                "SELECT 1 FROM scheduled_jobs "
                "WHERE name = 'auditor.privileged_review_escalate' LIMIT 1"
            ),
        )
        .scalar()
    )
    if not exists:
        op.get_bind().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)
                VALUES
                    (UUID(),
                     'auditor.privileged_review_escalate',
                     '特权审阅升级扫描',
                     '每天 03:30 (Asia/Shanghai) 扫描所有 pending 状态、且 '
                     'decided_at_deadline 已经过去 14 天的 AuditorPrivilegedRoleHolder, '
                     '设置 escalated_at + 写 compliance_incident '
                     '(type=access_review_overdue, severity=warning) + '
                     '推送 Feishu 卡到 assigned_to 的经理。idempotent — '
                     '一次写入 escalated_at 就不会重复。',
                     'auditor.privileged_review_escalate', 'cron', NULL,
                     '30 3 * * *', 'Asia/Shanghai',
                     1, JSON_OBJECT(), 0, 0, NOW(), NOW())
                """
            )
        )


def downgrade() -> None:
    op.get_bind().execute(
        sa.text(
            "DELETE FROM scheduled_jobs WHERE name = "
            "'auditor.privileged_review_escalate'"
        )
    )
    if _index_exists(_TABLE, "idx_apriv_holder_assigned"):
        op.drop_index("idx_apriv_holder_assigned", table_name=_TABLE)
    if _index_exists(_TABLE, "idx_apriv_holder_pending_deadline"):
        op.drop_index(
            "idx_apriv_holder_pending_deadline", table_name=_TABLE,
        )
    if _column_exists(_TABLE, "escalated_at"):
        op.drop_column(_TABLE, "escalated_at")
    if _column_exists(_TABLE, "decided_at_deadline"):
        op.drop_column(_TABLE, "decided_at_deadline")
    if _column_exists(_TABLE, "assigned_to_person_id"):
        op.drop_constraint(
            "fk_apriv_holder_assigned", _TABLE, type_="foreignkey",
        )
        op.drop_column(_TABLE, "assigned_to_person_id")