"""UNIQUE (process_id, resource_type) on ``onboarding_blockers``.

Review-r3 logic finding: the scanner contract is one blocker row per
dimension per process, but nothing enforced it. Two concurrent rescans
(employee ack → immediate reconcile in the HTTP session racing the
5-min cron in the worker session) could both see "unsatisfied + no
existing row" and both INSERT. The dup then wedges the process: the
scanner's ``{resource_type: row}`` index is last-wins, so only one dup
ever flips to ``provisioned`` while the other stays ``pending``
forever — pending count never hits zero, ``hard_ready_at`` never
stamps, false SLA cards, false T+7 incident.

Dedupe first (keep the most recently updated row per pair — it carries
the freshest resolution/label), then add the constraint. The re-hire
reopen path DELETEs all rows, so the constraint can't break it.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "281"
down_revision = "280"
branch_labels = None
depends_on = None

TABLE = "onboarding_blockers"
UQ = "uq_onboarding_blockers_process_resource"


def _constraint_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 = :c"
        ),
        {"t": table, "c": name},
    ).scalar() is not None


def upgrade() -> None:
    bind = op.get_bind()
    if _constraint_exists(bind, TABLE, UQ):
        return
    # Dedupe: keep the newest (updated_at, id) row per
    # (process_id, resource_type). MySQL/TiDB forbid updating/deleting
    # a table referenced in its own subquery (error 1093) — materialise
    # the keeper ids first.
    keeper_ids = [
        row[0]
        for row in bind.execute(
            sa.text(
                f"SELECT id FROM {TABLE} b "
                f"WHERE NOT EXISTS ("
                f"  SELECT 1 FROM (SELECT * FROM {TABLE}) b2 "
                f"  WHERE b2.process_id = b.process_id "
                f"    AND b2.resource_type = b.resource_type "
                f"    AND (b2.updated_at > b.updated_at "
                f"         OR (b2.updated_at = b.updated_at "
                f"             AND b2.id > b.id))"
                f")"
            )
        ).fetchall()
    ]
    if keeper_ids:
        # Chunk the NOT IN list defensively (prod table is small —
        # a few hundred rows — but stay under packet limits anyway).
        chunk = 1000
        all_ids = [
            row[0]
            for row in bind.execute(
                sa.text(f"SELECT id FROM {TABLE}")
            ).fetchall()
        ]
        to_delete = [i for i in all_ids if i not in set(keeper_ids)]
        for start in range(0, len(to_delete), chunk):
            batch = to_delete[start:start + chunk]
            placeholders = ", ".join(
                f":id{i}" for i in range(len(batch))
            )
            bind.execute(
                sa.text(
                    f"DELETE FROM {TABLE} WHERE id IN ({placeholders})"
                ),
                {f"id{i}": v for i, v in enumerate(batch)},
            )
    op.create_unique_constraint(
        UQ, TABLE, ["process_id", "resource_type"],
    )


def downgrade() -> None:
    bind = op.get_bind()
    if _constraint_exists(bind, TABLE, UQ):
        op.drop_constraint(UQ, TABLE, type_="unique")