"""Pre-launch hardening — seat / billing / catalog schema additions.

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

Pre-launch second-pass audit follow-on. The model layer in this branch
already references each of these columns / tables; this migration is
the missing schema half. Every change is idempotent — re-running
``alembic upgrade head`` is a no-op.

What this migration adds
========================

(1) ``software_license_plans.currency`` — ``VARCHAR(3) NOT NULL
    DEFAULT 'CNY'``. ISO 4217 code for the plan's price. Cross-currency
    aggregation is rejected at the service layer so the bill compute
    path can group by currency without branching. Existing rows are
    backfilled to ``'CNY'`` (matches the model server_default; safe
    because all pre-existing prices were assumed CNY in the legacy
    aggregator).

(2) ``software_seat_allocations.drift_observed_at`` — ``DATETIME NULL``.
    The feishu seat-sync handler stamps this the first time a seat's
    vendor-side membership is found missing; clears it the moment the
    membership reappears; revokes the seat after the configured grace
    window. NULL = currently in roster (or never observed missing).
    Existing rows default to NULL.

(3) ``software_permissions.deleted_at`` — ``DATETIME NULL``. Catalog
    soft-delete. Hard-delete would orphan FK references on historical
    UserPermission rows whose status is ``revoked`` — the catalog id is
    the only handle for "what permission did person X hold in
    2026-Q1?". NULL = active catalog row.

(4) ``software_monthly_bill_history`` — NEW TABLE. Snapshot of
    finalised ``software_monthly_bills`` rows captured by
    ``reopen_month`` BEFORE the period is cleared and re-closed. Lets
    the audit trail answer "what did we previously bill for 2026-04,
    and who reopened / re-closed it?". One reopen produces one
    ``supersedes_seq`` generation per period (first reopen = 1, second
    = 2, …); the surviving live bill row is always the most recent
    re-close.

(5) ``cmdb_billing_alerts_dedup`` — NEW TABLE. Idempotency window for
    ``maybe_send_budget_alert``. UNIQUE on ``(integration_id, month)``
    — at most one alert row per (cloud account, calendar month). Stops
    the budget-alert handler from spamming the channel when the
    monthly billing sync re-runs intra-day.

The ``integration_id`` column is stored as ``VARCHAR(36)`` rather than
an FK to ``integrations.id`` to dodge TiDB's CHAR↔VARCHAR FK strictness
(see ``CLAUDE.md`` memory ``feedback_tidb_fk_strictness``). The dedup
table is essentially a cache; an orphan row after an integration delete
is harmless and gets pruned by the natural month-rollover.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

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


_BILL_HISTORY_TABLE = "software_monthly_bill_history"
_BILLING_DEDUP_TABLE = "cmdb_billing_alerts_dedup"


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 _ensure_index(bind, table: str, name: str, columns: list[str]) -> None:
    if not _table_exists(bind, table):
        return
    if _index_exists(bind, table, name):
        return
    op.create_index(name, table, columns)


def _drop_index(bind, table: str, name: str) -> None:
    if not _table_exists(bind, table):
        return
    if _index_exists(bind, table, name):
        op.drop_index(name, table_name=table)


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

    # 1. software_license_plans.currency
    if _table_exists(bind, "software_license_plans") and not _column_exists(
        bind, "software_license_plans", "currency",
    ):
        op.add_column(
            "software_license_plans",
            sa.Column(
                "currency",
                sa.String(3),
                nullable=False,
                server_default=sa.text("'CNY'"),
            ),
        )
        # Explicit backfill — server_default covers new rows, but the
        # model also declares ``default="CNY"`` and we want the
        # at-rest data to read 'CNY' even when an older code-path
        # somehow inserted NULL (NOT NULL guarantees it didn't, but
        # belt-and-braces).
        bind.execute(sa.text(
            "UPDATE software_license_plans SET currency = 'CNY' "
            "WHERE currency IS NULL OR currency = ''",
        ))

    # 1b. software_monthly_bills.currency
    # Companion to (1) — every closed bill row carries the plan's
    # currency at close time so a multi-currency tenant doesn't
    # silently merge USD/CNY totals when reading historical periods.
    # `_build_lines_for_period` writes this on close; `list_period_bills`
    # / `list_known_periods` read it; `reopen_month` snapshot copies it.
    # Without this column, every billing endpoint 5xxes on first call
    # post-migration. (Pre-launch round 2 logic-scan P0.)
    if _table_exists(bind, "software_monthly_bills") and not _column_exists(
        bind, "software_monthly_bills", "currency",
    ):
        op.add_column(
            "software_monthly_bills",
            sa.Column(
                "currency",
                sa.String(3),
                nullable=False,
                server_default=sa.text("'CNY'"),
            ),
        )
        bind.execute(sa.text(
            "UPDATE software_monthly_bills SET currency = 'CNY' "
            "WHERE currency IS NULL OR currency = ''",
        ))

    # 2. software_seat_allocations.drift_observed_at
    if _table_exists(bind, "software_seat_allocations") and not _column_exists(
        bind, "software_seat_allocations", "drift_observed_at",
    ):
        op.add_column(
            "software_seat_allocations",
            sa.Column(
                "drift_observed_at",
                sa.DateTime(timezone=True),
                nullable=True,
            ),
        )

    # 3. software_permissions.deleted_at
    if _table_exists(bind, "software_permissions") and not _column_exists(
        bind, "software_permissions", "deleted_at",
    ):
        op.add_column(
            "software_permissions",
            sa.Column(
                "deleted_at",
                sa.DateTime(timezone=True),
                nullable=True,
            ),
        )

    # 4. software_monthly_bill_history (new table).
    if not _table_exists(bind, _BILL_HISTORY_TABLE):
        op.create_table(
            _BILL_HISTORY_TABLE,
            sa.Column("id", sa.CHAR(36), primary_key=True),
            sa.Column("period", sa.String(7), nullable=False),
            sa.Column("supersedes_seq", sa.Integer, nullable=False),
            sa.Column("original_bill_id", sa.CHAR(36), nullable=False),
            sa.Column("allocation_id", sa.CHAR(36), nullable=False),
            sa.Column("software_id", sa.CHAR(36), nullable=False),
            sa.Column("plan_id", sa.CHAR(36), nullable=False),
            sa.Column("user_id", sa.CHAR(36), nullable=False),
            sa.Column("software_name", sa.String(128), nullable=True),
            sa.Column("plan_name", sa.String(128), nullable=True),
            sa.Column("user_name", sa.String(128), nullable=True),
            sa.Column("department", sa.String(128), nullable=True),
            sa.Column("reason", sa.Text, nullable=True),
            sa.Column("price", sa.Numeric(12, 2), nullable=False),
            sa.Column("billing_cycle", sa.String(16), nullable=False),
            sa.Column(
                "currency", sa.String(3), nullable=False,
                server_default=sa.text("'CNY'"),
            ),
            sa.Column("amount", sa.Numeric(12, 2), nullable=False),
            sa.Column("allocated_at", sa.DateTime(timezone=True), nullable=False),
            sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
            sa.Column(
                "original_finalized_at", sa.DateTime(timezone=True),
                nullable=False,
            ),
            sa.Column("original_finalized_by", sa.CHAR(36), nullable=True),
            sa.Column(
                "superseded_at", sa.DateTime(timezone=True),
                nullable=False,
            ),
            sa.Column("superseded_by", sa.CHAR(36), nullable=True),
        )
    _ensure_index(
        bind, _BILL_HISTORY_TABLE,
        "ix_software_monthly_bill_history_period_seq",
        ["period", "supersedes_seq"],
    )
    _ensure_index(
        bind, _BILL_HISTORY_TABLE,
        "ix_software_monthly_bill_history_period",
        ["period"],
    )

    # 5. cmdb_billing_alerts_dedup (new table).
    if not _table_exists(bind, _BILLING_DEDUP_TABLE):
        op.create_table(
            _BILLING_DEDUP_TABLE,
            sa.Column("id", sa.CHAR(36), primary_key=True),
            # Stored as VARCHAR(36) (no FK) on purpose — TiDB rejects
            # CHAR↔VARCHAR FK pairings that local MySQL accepts (see
            # CLAUDE.md memory ``feedback_tidb_fk_strictness``). The
            # dedup row is a cache; orphan rows after an integration
            # delete are harmless and pruned by month rollover.
            sa.Column("integration_id", sa.String(36), nullable=False),
            sa.Column("month", sa.String(7), nullable=False),
            sa.Column("last_alerted_at", sa.DateTime(timezone=True), nullable=False),
            sa.Column("last_alerted_amount_usd", sa.Numeric(14, 2), nullable=False),
            sa.Column("last_alerted_budget_usd", sa.Numeric(14, 2), nullable=False),
        )
    if _table_exists(bind, _BILLING_DEDUP_TABLE) and not _index_exists(
        bind, _BILLING_DEDUP_TABLE, "uq_cmdb_billing_alerts_dedup_integration_month",
    ):
        op.create_index(
            "uq_cmdb_billing_alerts_dedup_integration_month",
            _BILLING_DEDUP_TABLE,
            ["integration_id", "month"],
            unique=True,
        )


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

    # 5. cmdb_billing_alerts_dedup
    _drop_index(
        bind, _BILLING_DEDUP_TABLE,
        "uq_cmdb_billing_alerts_dedup_integration_month",
    )
    if _table_exists(bind, _BILLING_DEDUP_TABLE):
        op.drop_table(_BILLING_DEDUP_TABLE)

    # 4. software_monthly_bill_history
    _drop_index(
        bind, _BILL_HISTORY_TABLE,
        "ix_software_monthly_bill_history_period",
    )
    _drop_index(
        bind, _BILL_HISTORY_TABLE,
        "ix_software_monthly_bill_history_period_seq",
    )
    if _table_exists(bind, _BILL_HISTORY_TABLE):
        op.drop_table(_BILL_HISTORY_TABLE)

    # 3. software_permissions.deleted_at
    if _table_exists(bind, "software_permissions") and _column_exists(
        bind, "software_permissions", "deleted_at",
    ):
        op.drop_column("software_permissions", "deleted_at")

    # 2. software_seat_allocations.drift_observed_at
    if _table_exists(bind, "software_seat_allocations") and _column_exists(
        bind, "software_seat_allocations", "drift_observed_at",
    ):
        op.drop_column("software_seat_allocations", "drift_observed_at")

    # 1b. software_monthly_bills.currency
    if _table_exists(bind, "software_monthly_bills") and _column_exists(
        bind, "software_monthly_bills", "currency",
    ):
        op.drop_column("software_monthly_bills", "currency")

    # 1. software_license_plans.currency
    if _table_exists(bind, "software_license_plans") and _column_exists(
        bind, "software_license_plans", "currency",
    ):
        op.drop_column("software_license_plans", "currency")