"""Pre-launch hardening — software_seat_allocations active/pending uniqueness +
software_monthly_bill_history (period, supersedes_seq, original_bill_id) uniqueness.

Revision ID: 119
Revises: 118
Create Date: 2026-05-05

Closes two ship-blockers from the seventh 9-pass audit:

1. **No DB-level guard against duplicate active/pending seat rows for the
   same (plan_id, user_id).** Combined with the lack of a single-flight
   guard on the scheduled-job runner (see migration 119 sibling work in
   `server/app/scheduled_jobs/service.py`), two concurrent runs of the
   same seat-sync job — admin "trigger now" + APScheduler tick, multi-
   replica deploy, retry-on-failure — could each pass the
   ``check-then-insert`` race window in `*_handlers.py` and INSERT a
   second active row. Result: the next ``close_month`` bills the user
   2× for one Zoom/Slack/Jira/Feishu/Google seat. The
   ``check_no_active_seat_request`` helper closes the TOCTOU window for
   seat REQUESTS via a ``SELECT … FOR UPDATE`` on the plan row, but
   none of the seat-sync handlers acquire that lock — they're meant to
   be the source of truth, so they don't go through request flow.

2. **No uniqueness on ``software_monthly_bill_history(period,
   supersedes_seq, original_bill_id)``.** The ``ix_…_period_seq``
   index is non-unique. ``reopen_month`` had no advisory lock (fixed
   in code in this same PR) so two concurrent reopens could both read
   the same MAX(supersedes_seq), both compute next_seq=N+1, and both
   INSERT history rows with the same composite key. Belt-and-braces
   alongside the lock fix.

Pre-flight steps before the constraint is added:

a. **Dedup pre-existing duplicate active/pending rows.** Production
   shouldn't have any (the lifecycle has been audited multiple times)
   but local dev clusters with hand-edited test data might.
   ROW_NUMBER() over (plan_id, user_id) keeps the most recent
   ``allocated_at`` (id as tie-breaker); older duplicates flip to
   ``status='revoked'`` with a clearly-marked reason. ``revoked_at``
   is set to NOW() so the row joins the normal billing close-month
   flow correctly.

b. **Backfill stuck-pending allocations whose approval was already
   rejected/cancelled.** The original approval callback (pre-fix in
   this PR) only flipped ``employee_requests.status`` and
   ``approval_requests.status``; it did NOT update
   ``software_seat_allocations.status`` for the rejected branch. Result:
   an active company can have an unbounded number of permanently-stuck
   ``pending`` allocations whose approval was rejected long ago.
   These rows would block the user from re-requesting (via
   ``check_no_active_seat_request``) AND would inflate the next
   ``close_month`` bill (via ``_fetch_allocations_overlapping``,
   which had no status filter — also fixed in this PR).
   ``revoked_at = approval_requests.updated_at`` so the row's lifetime
   matches the moment of rejection (zero-billable-window) instead of
   being silently extended by however many months the row sat stuck.

The backfill is idempotent — re-running ``alembic upgrade head`` is
a no-op once it's run (no allocations remain in the broken state).
"""
from __future__ import annotations

import re

import sqlalchemy as sa
from sqlalchemy.exc import OperationalError

from alembic import op

revision = "119"
down_revision = "118"
branch_labels = None
depends_on = None


_SEAT_TABLE = "software_seat_allocations"
_BILL_HISTORY_TABLE = "software_monthly_bill_history"
_SEAT_UNIQUE_KEY_COL = "active_unique_key"
_SEAT_UNIQUE_INDEX = "uq_software_seat_active_pending"
_HISTORY_UNIQUE_INDEX = "uq_software_monthly_bill_history_period_seq_bill"


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 _column_exists(bind, table: str, column: str) -> bool:
    return bool(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())


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 _software_plan_fk_defs(bind) -> list[tuple[str, str]]:
    """Return actual software_id FK names and delete rules.

    TiDB can expose surprising FK names through information_schema on
    older schemas. SHOW CREATE TABLE is the source accepted by ALTER
    TABLE DROP FOREIGN KEY, so prefer it and keep information_schema as
    a fallback only when SHOW CREATE is unavailable.
    """
    try:
        show_rows = bind.execute(sa.text("SHOW CREATE TABLE software_license_plans")).first()
    except OperationalError:
        show_rows = None
    if show_rows is not None:
        create_sql = show_rows[1]
        if isinstance(create_sql, str):
            fks: list[tuple[str, str]] = []
            for raw_line in create_sql.splitlines():
                line = raw_line.strip().rstrip(",")
                if "FOREIGN KEY" not in line.upper():
                    continue
                if "`software_id`" not in line:
                    continue
                if "REFERENCES `asset_software_catalog`" not in line:
                    continue
                match = re.search(r"CONSTRAINT\s+`([^`]+)`\s+FOREIGN KEY", line, re.IGNORECASE)
                if match is None:
                    continue
                delete_rule = "CASCADE" if "ON DELETE CASCADE" in line.upper() else "RESTRICT"
                fks.append((match.group(1), delete_rule))
            return fks

    return list(bind.execute(
        sa.text(
            """
            SELECT k.CONSTRAINT_NAME, r.DELETE_RULE
            FROM information_schema.KEY_COLUMN_USAGE k
            JOIN information_schema.REFERENTIAL_CONSTRAINTS r
              ON k.CONSTRAINT_SCHEMA = r.CONSTRAINT_SCHEMA
             AND k.CONSTRAINT_NAME = r.CONSTRAINT_NAME
             AND k.TABLE_NAME = r.TABLE_NAME
             AND k.REFERENCED_TABLE_NAME = r.REFERENCED_TABLE_NAME
            WHERE k.TABLE_SCHEMA = DATABASE()
              AND k.TABLE_NAME = 'software_license_plans'
              AND k.COLUMN_NAME = 'software_id'
              AND k.REFERENCED_TABLE_NAME = 'asset_software_catalog'
            """,
        ),
    ).all())


def _is_missing_fk_error(exc: OperationalError) -> bool:
    orig_args = getattr(getattr(exc, "orig", None), "args", ())
    return bool(orig_args and orig_args[0] == 1091) or "1091" in str(exc)


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

    if _table_exists(bind, _SEAT_TABLE):
        # ── Step a: dedup duplicate active/pending rows ─────────
        # Pre-flight cleanup. ROW_NUMBER over (plan_id, user_id) keeps
        # the most recent allocated_at; older rows are revoked. Tie-
        # break on id DESC so the result is deterministic across re-
        # runs even when allocated_at is identical (rare; happens with
        # the same migration's bulk-insert seed data).
        # ``revoked_at = allocated_at`` (NOT NOW()) so the deduped
        # older row's billable window collapses to zero — the
        # ``_fetch_allocations_overlapping`` filter excludes any row
        # whose ``revoked_at == allocated_at``. Without this, a pre-
        # existing duplicate active row that overlapped the current
        # period would be billed for the days between its
        # allocated_at and the migration time. The kept row already
        # bills the period correctly, so deduping must NOT add a
        # second prorated line for the same (user, plan). Round-2
        # audit P2.
        bind.execute(sa.text(
            """
            UPDATE software_seat_allocations
            SET status = 'revoked',
                revoked_at = allocated_at,
                reason = CONCAT(
                    COALESCE(reason, ''),
                    '\n[migration_119: dedup duplicate active/pending; ',
                    'revoked_at aligned to allocated_at so the row ',
                    'never enters a bill]'
                )
            WHERE id IN (
                SELECT id FROM (
                    SELECT id,
                           ROW_NUMBER() OVER (
                               PARTITION BY plan_id, user_id
                               ORDER BY allocated_at DESC, id DESC
                           ) AS rn
                    FROM software_seat_allocations
                    WHERE status IN ('active', 'pending')
                ) ranked
                WHERE rn > 1
            )
            """,
        ))

        # ── Step b: backfill stuck-pending whose approval rejected ─
        # ``approval_requests`` may not exist on a pristine fresh-
        # install cluster mid-migration — guard the JOIN.
        #
        # Critical: ``revoked_at = allocated_at`` (NOT the approval's
        # updated_at) so the row's billable window collapses to zero.
        # The seat was never actually active, so no period should be
        # billed for it. The companion change in
        # ``billing_service._fetch_allocations_overlapping`` filters
        # on ``revoked_at > allocated_at`` which excludes these rows
        # from every bill computation. Without this date alignment,
        # stuck pending rows whose rejection happened mid-period would
        # be billed prorated for the days between allocation and
        # rejection — the very thing red-line #3 says we must not do.
        if _table_exists(bind, "approval_requests"):
            bind.execute(sa.text(
                """
                UPDATE software_seat_allocations sa
                JOIN approval_requests ar ON sa.approval_request_id = ar.id
                SET sa.status = 'revoked',
                    sa.revoked_at = sa.allocated_at,
                    sa.reason = CONCAT(
                        COALESCE(sa.reason, ''),
                        '\n[migration_119: stuck pending; approval ',
                        ar.status,
                        '; revoked_at aligned to allocated_at so the ',
                        'row never enters a bill]'
                    )
                WHERE sa.status = 'pending'
                  AND ar.status IN ('rejected', 'cancelled')
                """,
            ))

        # ── Step c: add VIRTUAL generated column ────────────────
        # CASE expression NULLs out non-active/pending rows so the
        # subsequent UNIQUE INDEX only constrains the rows that
        # actually represent "this user occupies a seat on this
        # plan". Multiple revoked rows for the same (plan_id, user_id)
        # remain legal (a user can request → revoke → request again
        # any number of times).
        if not _column_exists(bind, _SEAT_TABLE, _SEAT_UNIQUE_KEY_COL):
            # Note: VIRTUAL generated columns work in MySQL 5.7+ and
            # TiDB 5.x+. Concat of two CHAR(36) UUID columns yields
            # 73-char string; VARCHAR(80) is comfortable.
            bind.execute(sa.text(
                f"""
                ALTER TABLE {_SEAT_TABLE}
                ADD COLUMN {_SEAT_UNIQUE_KEY_COL} VARCHAR(80)
                GENERATED ALWAYS AS (
                    CASE WHEN status IN ('active', 'pending')
                         THEN CONCAT(plan_id, ':', user_id)
                         ELSE NULL
                    END
                ) VIRTUAL
                """,
            ))

        # ── Step d: UNIQUE INDEX on the generated column ────────
        if not _index_exists(bind, _SEAT_TABLE, _SEAT_UNIQUE_INDEX):
            op.create_index(
                _SEAT_UNIQUE_INDEX,
                _SEAT_TABLE,
                [_SEAT_UNIQUE_KEY_COL],
                unique=True,
            )

    # ── Step e: history (period, supersedes_seq, original_bill_id) ─
    # Belt-and-braces alongside the reopen_month advisory lock — even
    # if a future code path forgets the lock, two concurrent reopens
    # cannot both write the same history row.
    if _table_exists(bind, _BILL_HISTORY_TABLE) and not _index_exists(
        bind, _BILL_HISTORY_TABLE, _HISTORY_UNIQUE_INDEX,
    ):
        op.create_index(
            _HISTORY_UNIQUE_INDEX,
            _BILL_HISTORY_TABLE,
            ["period", "supersedes_seq", "original_bill_id"],
            unique=True,
        )

    # ── Step f: software_license_plans.software_id FK CASCADE→RESTRICT ─
    # The previous FK was declared with ondelete=CASCADE, so a hard
    # ``DELETE FROM asset_software_catalog`` would silently wipe every
    # license plan attached to the software — and indirectly orphan
    # every ``software_monthly_bills.plan_id`` (which is a bare UUID
    # with no FK on purpose, to isolate finance history from catalog
    # mutations). Migration 119 (this revision) tightens the FK to
    # RESTRICT so the DB refuses the delete when plans exist; the
    # application layer (``delete_software_catalog`` route) is the
    # primary guard with a friendlier 409 error.
    #
    # FK constraint names are auto-generated unless explicit. Prefer
    # SHOW CREATE TABLE for the actual droppable name, because some
    # TiDB schemas surface stale synthetic names through information_
    # schema (e.g. ``fk_1``) that ALTER TABLE cannot drop. Idempotent:
    # if a non-cascading FK already exists, leave it alone.
    if _table_exists(bind, "software_license_plans"):
        fks = _software_plan_fk_defs(bind)
        for fk_name, delete_rule in fks:
            if (delete_rule or "").upper() != "CASCADE":
                continue
            # Drop the cascading FK; re-add as RESTRICT (default).
            try:
                bind.execute(sa.text(
                    f"ALTER TABLE software_license_plans DROP FOREIGN KEY `{fk_name}`",
                ))
            except OperationalError as exc:
                if not _is_missing_fk_error(exc):
                    raise
                # Stale metadata entry; if SHOW CREATE cannot see it,
                # it is not a droppable FK on this table. Continue to
                # re-check the real table state below.
                continue

        if not _software_plan_fk_defs(bind):
            bind.execute(sa.text(
                "ALTER TABLE software_license_plans "
                "ADD CONSTRAINT fk_software_license_plans_software "
                "FOREIGN KEY (software_id) "
                "REFERENCES asset_software_catalog(id) "
                # ON DELETE RESTRICT is MySQL's default action when no
                # ON DELETE clause is given. Make it explicit so the
                # schema dump is unambiguous.
                "ON DELETE RESTRICT ON UPDATE RESTRICT",
            ))


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

    # Drop history unique
    if _table_exists(bind, _BILL_HISTORY_TABLE) and _index_exists(
        bind, _BILL_HISTORY_TABLE, _HISTORY_UNIQUE_INDEX,
    ):
        op.drop_index(_HISTORY_UNIQUE_INDEX, table_name=_BILL_HISTORY_TABLE)

    # Drop seat unique + generated column. Order matters — the index
    # must be dropped before the underlying generated column.
    if _table_exists(bind, _SEAT_TABLE):
        if _index_exists(bind, _SEAT_TABLE, _SEAT_UNIQUE_INDEX):
            op.drop_index(_SEAT_UNIQUE_INDEX, table_name=_SEAT_TABLE)
        if _column_exists(bind, _SEAT_TABLE, _SEAT_UNIQUE_KEY_COL):
            op.drop_column(_SEAT_TABLE, _SEAT_UNIQUE_KEY_COL)

    # Note: the dedup + stuck-pending backfills are NOT reversible —
    # we don't know which rows were originally pending. Downgrade
    # leaves them as 'revoked' (with the clearly-marked reason in
    # ``reason``) so an operator can re-pend them by hand if needed.