"""Add UNIQUE(user_id, permission_id) on user_permissions + dedup data.

Revision ID: 116
Revises: 115
Create Date: 2026-05-05

5th pre-launch 9-pass audit. The SELECT-then-INSERT in
``PermissionService.request_permission`` raced under double-click —
two concurrent POSTs for the same (user, permission) both passed the
"no existing pending/active" check and both INSERTed, producing
duplicate active rows + duplicate provisioning tasks (e.g. two Slack
invites, two Google group joins). The model now declares
``UniqueConstraint("user_id", "permission_id", name="uq_user_perm")``
and the service treats ``IntegrityError`` as ConflictError; this
migration adds the actual DB constraint and pre-cleans existing
duplicates so the constraint can land.

Dedup priority — for each (user_id, permission_id) with N>1 rows:
  active > pending > approved > revoked > expired > rejected > other
Within the same status bucket, keep the most recent ``updated_at``
(then ``created_at``, then highest ``id``); delete the rest.

Idempotent — safe to re-run; the dedupe pass is a no-op when no
duplicates exist, and the constraint-add is guarded by
information_schema.
"""
from __future__ import annotations

import sqlalchemy as sa
from alembic import op

revision = "116"
down_revision = "115"
branch_labels = None
depends_on = None


_STATUS_PRIORITY: dict[str, int] = {
    "active": 0,
    "pending": 1,
    "approved": 2,
    "revoked": 3,
    "expired": 4,
    "rejected": 5,
}


def _table_exists(bind, table: str) -> bool:
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.tables "
            "WHERE table_schema = DATABASE() AND table_name = :t LIMIT 1",
        ),
        {"t": table},
    ).scalar())


def _constraint_exists(bind, table: str, name: str) -> bool:
    return bool(bind.execute(
        sa.text(
            "SELECT 1 FROM information_schema.TABLE_CONSTRAINTS "
            "WHERE TABLE_SCHEMA = DATABASE() "
            "  AND TABLE_NAME = :t AND CONSTRAINT_NAME = :n LIMIT 1",
        ),
        {"t": table, "n": name},
    ).scalar())


def _index_exists(bind, table: str, name: str) -> bool:
    return bool(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": name},
    ).scalar())


def _dedupe_user_permissions(bind) -> int:
    """Delete duplicate rows so the new UQ can land. Returns the
    number of rows deleted (for logging visibility).
    """
    # Pull every duplicated (user_id, permission_id) group.
    rows = bind.execute(sa.text(
        """
        SELECT user_id, permission_id, COUNT(*) AS n
          FROM user_permissions
         GROUP BY user_id, permission_id
        HAVING COUNT(*) > 1
        """,
    )).fetchall()
    if not rows:
        return 0
    deleted = 0
    for user_id, permission_id, _n in rows:
        # Pull the full row set for this pair, ranked by our priority.
        candidates = bind.execute(sa.text(
            """
            SELECT id, status, updated_at, created_at
              FROM user_permissions
             WHERE user_id = :u AND permission_id = :p
            """,
        ), {"u": user_id, "p": permission_id}).fetchall()

        # MySQL/TiDB returns naive datetimes; ``dt.timestamp()`` would
        # interpret them as local time on the migration host, which
        # could flip ordering across DST boundaries on non-UTC hosts.
        # Use ``(dt - datetime.min).total_seconds()`` (negated) so we
        # get a numeric "newer wins" ordering inside each priority
        # bucket — independent of host timezone. ``datetime.min`` is
        # 0001-01-01 which any real DB row out-dates; the subtraction
        # produces a positive timedelta, which we negate to invert
        # the sort order without using ``reverse=True`` (would invert
        # priority too).
        from datetime import datetime as _dt_t
        _EPOCH = _dt_t.min

        def _rank(r):
            status = (r[1] or "").lower()
            prio = _STATUS_PRIORITY.get(status, 99)
            updated_seconds = (
                -(r[2] - _EPOCH).total_seconds() if r[2] else 0.0
            )
            created_seconds = (
                -(r[3] - _EPOCH).total_seconds() if r[3] else 0.0
            )
            return (prio, updated_seconds, created_seconds, str(r[0]))

        sorted_rows = sorted(candidates, key=_rank)
        keeper = sorted_rows[0]
        losers = [r[0] for r in sorted_rows[1:]]
        if not losers:
            continue
        # Hard-delete the losers. They were duplicates produced by the
        # race; the keeper is the authoritative record. The FK on
        # ``approval_request_id`` / ``provisioning_task_id`` lives on
        # OTHER tables pointing TO us, not us pointing to them, so
        # deleting these rows does NOT cascade. (Verified against
        # the model: user_permissions has only outgoing FKs.)
        bind.execute(sa.text(
            "DELETE FROM user_permissions WHERE id IN :ids"
        ).bindparams(sa.bindparam("ids", expanding=True)), {"ids": losers})
        deleted += len(losers)
    return deleted


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

    if not _table_exists(bind, "user_permissions"):
        # Fresh install path — table created later in initial Alembic
        # chain. Nothing to do.
        return

    # Pre-clean duplicates so the UQ can land cleanly. Logs the count
    # via op-level info so it's visible in the alembic upgrade output.
    deleted = _dedupe_user_permissions(bind)
    if deleted:
        # Surface to alembic stdout so an oncall reviewing the upgrade
        # log sees the count, not just a silent dedupe.
        print(
            f"[migration 116] pre-UQ dedupe: removed {deleted} "
            f"duplicate user_permissions row(s)"
        )

    if not _constraint_exists(bind, "user_permissions", "uq_user_perm") \
            and not _index_exists(bind, "user_permissions", "uq_user_perm"):
        op.create_unique_constraint(
            "uq_user_perm",
            "user_permissions",
            ["user_id", "permission_id"],
        )


def downgrade() -> None:
    bind = op.get_bind()
    if not _table_exists(bind, "user_permissions"):
        return
    if _constraint_exists(bind, "user_permissions", "uq_user_perm") \
            or _index_exists(bind, "user_permissions", "uq_user_perm"):
        op.drop_constraint(
            "uq_user_perm", "user_permissions", type_="unique",
        )