"""software_monthly_bills: addon-line idempotency key (period, charge_id).

The only uniqueness guard on ``software_monthly_bills`` was
``uq_software_monthly_bills_period_alloc`` (migration 063). Migration 348
added ``line_type='addon'`` rows with ``allocation_id=NULL`` — and MySQL/
TiDB UNIQUE treats every NULL as distinct, so the alloc key gives addon
lines ZERO protection: a double ``close_month`` (admin racing the 03:00
auto-close cron; the advisory lock used to be released before commit)
inserted every addon charge twice and every downstream ``SUM(amount)``
rollup double-counted real spend.

``charge_id`` is non-NULL exactly for addon lines (seat lines carry
``charge_id=NULL`` and pass freely), so ``UNIQUE(period, charge_id)`` is
the addon twin of the alloc key. Belt-and-braces with the code fix that
now commits before releasing the close lock.

Guarded duplicate cleanup first (keep the earliest row per (period,
charge_id)) so the index build cannot fail on an already double-billed
period; the deleted duplicates are exact copies of the survivors.

Revision ID: 382
Revises: 381
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "382"
down_revision = "381"
branch_labels = None
depends_on = None

TABLE = "software_monthly_bills"
UQ = "uq_software_monthly_bills_period_charge"


def _index_exists(bind, 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 upgrade() -> None:
    bind = op.get_bind()
    if _index_exists(bind, UQ):
        return
    # Deduplicate any already double-inserted addon lines: keep the
    # earliest row (lowest finalized_at, tie-broken by id) per
    # (period, charge_id). Seat lines (charge_id IS NULL) untouched.
    bind.execute(sa.text(
        f"""
        DELETE b FROM {TABLE} b
        JOIN {TABLE} keep
          ON keep.period = b.period
         AND keep.charge_id = b.charge_id
         AND b.charge_id IS NOT NULL
         AND (
              keep.finalized_at < b.finalized_at
              OR (keep.finalized_at = b.finalized_at AND keep.id < b.id)
         )
        """
    ))
    op.create_unique_constraint(UQ, TABLE, ["period", "charge_id"])


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