"""Add ack columns to ``scheduled_job_runs`` + dedup signature on jobs.

Round-15 (2026-05-09): a CMDB IAM-user sync that hits the partial-driver
safety floor (``provider returned 9 rows vs 20 existing — refusing to
sweep``) is a *protective* failure: data is intact, just unswept. Until
this PR there was no signal beyond the admin remembering to open the
``/admin/scheduled-jobs/runs`` page, AND no way to dismiss a known-
acknowledged failure short of waiting for the next run to succeed. Same
class of failure could repeat every cron tick and re-spam the inbox.

Two changes land together:

1. ``scheduled_job_runs`` gains ``acknowledged_at`` /
   ``acknowledged_by_id`` / ``ack_reason``. An IT admin can
   acknowledge a failed run; the run row keeps ``status='failed'`` (we
   never rewrite history — audit + post-mortems depend on the truth)
   but the UI defaults to hiding ack'd failures so the queue isn't
   permanently red.

2. ``scheduled_jobs`` gains ``last_failure_signature`` +
   ``last_failure_notified_at`` so the dispatcher can deduplicate
   notifications: the FIRST occurrence of a failure signature pushes
   one in-app notification to every ``it_admin`` / ``global_admin``;
   subsequent runs with the *same* signature stay silent until either
   the signature changes (a different error) or the previous failure
   is acknowledged (admin signalled they're aware → next failure
   re-pushes for visibility).

Index ``ix_scheduled_job_runs_status_ack`` accelerates the default UI
query: ``status='failed' AND acknowledged_at IS NULL``.

FK on ``acknowledged_by_id`` follows the same pattern as
``cmdb_provider_accounts.created_by`` (migration 044): VARCHAR(36)
matched to ``identity_persons.id``'s charset/collation via
``normalize_fk_column_collation`` to avoid TiDB 3780.

Revision ID: 137
Revises: 136
Create Date: 2026-05-09
"""
from __future__ import annotations

import sqlalchemy as sa

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

# revision identifiers, used by Alembic.
revision = "137"
down_revision = "136"
branch_labels = None
depends_on = None


_FK_NAME = "fk_scheduled_job_runs_ack_by"


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, name: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.STATISTICS "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t "
            "AND INDEX_NAME = :n"
        ),
        {"t": table, "n": name},
    ).scalar() is not None


def _fk_exists(bind, table: str, name: str) -> bool:
    return bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLE_CONSTRAINTS "
            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t "
            "AND CONSTRAINT_NAME = :n AND CONSTRAINT_TYPE = 'FOREIGN KEY'"
        ),
        {"t": table, "n": name},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()

    # ── scheduled_job_runs: ack columns + index ───────────────────
    if not _column_exists(bind, "scheduled_job_runs", "acknowledged_at"):
        op.add_column(
            "scheduled_job_runs",
            sa.Column("acknowledged_at", sa.DateTime(timezone=True), nullable=True),
        )
    if not _column_exists(bind, "scheduled_job_runs", "acknowledged_by_id"):
        op.add_column(
            "scheduled_job_runs",
            sa.Column("acknowledged_by_id", sa.String(36), nullable=True),
        )
    if not _column_exists(bind, "scheduled_job_runs", "ack_reason"):
        op.add_column(
            "scheduled_job_runs",
            sa.Column("ack_reason", sa.Text, nullable=True),
        )

    # FK to identity_persons.id — normalize collation so TiDB accepts.
    normalize_fk_column_collation(
        table="scheduled_job_runs",
        column="acknowledged_by_id",
        parent_table="identity_persons",
        nullable=True,
    )
    if not _fk_exists(bind, "scheduled_job_runs", _FK_NAME):
        op.create_foreign_key(
            _FK_NAME,
            source_table="scheduled_job_runs",
            referent_table="identity_persons",
            local_cols=["acknowledged_by_id"],
            remote_cols=["id"],
            ondelete="SET NULL",
        )

    # Default UI query is failed + unacknowledged within the last N
    # hours; the existing ``ix_scheduled_job_runs_job_started`` covers
    # the per-job drilldown but not the cross-job failure scan that
    # ``ScheduledJobRunsPage`` and the notification dedup both run.
    if not _index_exists(bind, "scheduled_job_runs", "ix_scheduled_job_runs_status_ack"):
        op.create_index(
            "ix_scheduled_job_runs_status_ack",
            "scheduled_job_runs",
            ["status", "acknowledged_at"],
        )

    # ── scheduled_jobs: dedup signature + last-notified timestamp ──
    if not _column_exists(bind, "scheduled_jobs", "last_failure_signature"):
        op.add_column(
            "scheduled_jobs",
            sa.Column("last_failure_signature", sa.String(64), nullable=True),
        )
    if not _column_exists(bind, "scheduled_jobs", "last_failure_notified_at"):
        op.add_column(
            "scheduled_jobs",
            sa.Column(
                "last_failure_notified_at", sa.DateTime(timezone=True), nullable=True,
            ),
        )


def downgrade() -> None:
    bind = op.get_bind()

    if _fk_exists(bind, "scheduled_job_runs", _FK_NAME):
        op.drop_constraint(_FK_NAME, "scheduled_job_runs", type_="foreignkey")
    if _index_exists(bind, "scheduled_job_runs", "ix_scheduled_job_runs_status_ack"):
        op.drop_index(
            "ix_scheduled_job_runs_status_ack",
            table_name="scheduled_job_runs",
        )

    for col in ("ack_reason", "acknowledged_by_id", "acknowledged_at"):
        if _column_exists(bind, "scheduled_job_runs", col):
            op.drop_column("scheduled_job_runs", col)

    for col in ("last_failure_notified_at", "last_failure_signature"):
        if _column_exists(bind, "scheduled_jobs", col):
            op.drop_column("scheduled_jobs", col)