"""Backfill: revoke orphan pending software_seat_allocations.

Migration 119 step (b) attempted to revoke seat rows whose Feishu
approval was already cancelled / rejected, but its JOIN used
``software_seat_allocations.approval_request_id = approval_requests.id``.
That column is declared on the model but is NEVER populated by
``request_seat`` — the linkage is one-directional via
``approval_requests.source_payload->>'$.seat_allocation_id'`` instead.
So 119's backfill matched zero rows on TiDB and the orphan pending
seats remained, blocking re-applications via ``check_no_active_seat_request``
(409 ``DUPLICATE_SEAT_REQUEST`` — "You already have a pending seat
for X").

This migration walks the correct linkage and finishes the job that
119 intended:

  • Find ``software_seat_allocations`` rows where
    ``status = 'pending'``.
  • Join to ``approval_requests`` via the JSON path
    ``source_payload->>'$.seat_allocation_id'``.
  • Where the approval landed in a terminal non-approve state
    (``rejected`` / ``cancelled``), flip the seat to ``revoked``
    with ``revoked_at = allocated_at`` so the row's billable window
    collapses to zero (mirrors ``_revoke_pending_seat_on_rejection``
    and the new cancel-cascade in ``cancel_employee_request``).

Idempotent: status filter ``= 'pending'`` keeps re-runs as no-ops.

Bug reported 2026-05-06 by employee on /my self-service portal: a
cancelled seat request continued to block fresh requests for the
same plan because the orphan ``pending`` seat row was never reaped.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "121"
down_revision = "120_iam_user_sub_account_unique"
branch_labels = None
depends_on = None


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 upgrade() -> None:
    bind = op.get_bind()

    if not _table_exists(bind, "software_seat_allocations"):
        return
    if not _table_exists(bind, "approval_requests"):
        return

    # MySQL / TiDB JSON path: ``source_payload->>'$.seat_allocation_id'``
    # is shorthand for ``JSON_UNQUOTE(JSON_EXTRACT(source_payload,
    # '$.seat_allocation_id'))`` — produces the raw UUID string we
    # compare against the seat's id (also stored as a UUID-shaped CHAR
    # in TiDB). Subquery form rather than UPDATE-JOIN so TiDB's optimizer
    # can use the ``approval_requests.status`` predicate first (small
    # set) before evaluating the JSON extract.
    bind.execute(sa.text(
        """
        UPDATE software_seat_allocations
        SET status = 'revoked',
            revoked_at = allocated_at,
            reason = CONCAT(
                COALESCE(reason, ''),
                '\n[migration_121: orphan pending; linked approval ',
                'cancelled/rejected; revoked_at aligned to ',
                'allocated_at so the row never enters a bill]'
            )
        WHERE status = 'pending'
          AND id IN (
              SELECT seat_id FROM (
                  SELECT
                      JSON_UNQUOTE(
                          JSON_EXTRACT(source_payload, '$.seat_allocation_id')
                      ) AS seat_id
                  FROM approval_requests
                  WHERE approval_type = 'software_seat'
                    AND status IN ('rejected', 'cancelled')
                    AND JSON_EXTRACT(source_payload, '$.seat_allocation_id')
                        IS NOT NULL
              ) AS terminal_seat_ids
              WHERE seat_id IS NOT NULL
          )
        """,
    ))


def downgrade() -> None:
    # No-op: the upgrade revokes orphan seat rows that were stuck
    # in ``pending`` solely because of an upstream code bug. There's
    # no value in resurrecting them — the linked approvals are still
    # in terminal cancelled/rejected state, so a "downgrade restore"
    # would just re-create the same orphan condition.
    pass