"""core_idempotency_records: request_fingerprint column.

Revision ID: 432
Revises: 431
Create Date: 2026-07-28

Binds an ``Idempotency-Key`` to the request body it was first used with
(SHA-256 hex of the raw bytes, 64 chars). Without it, a client bug that
reuses a key across DIFFERENT payloads silently gets the first call's
cached success back — a "success" for a write that never executed. With
it, ``IdempotencyService.lookup()`` compares fingerprints on an existing
completed/pending row and answers 422 ``IDEMPOTENCY_PAYLOAD_MISMATCH``
on a mismatch (IETF idempotency-key-header semantics).

Nullable, and NULL means "skip the comparison": rows written before this
migration, and hand-wired ``IdempotencyService`` call sites that don't
pass a fingerprint, keep replaying exactly as before — the deploy needs
no backfill and no flag day. Rows age out within the 24h TTL anyway.

Pure nullable column add — TiDB online DDL, no table lock, no FK, no
index (the column is only ever read off a row already fetched by the
``(key, scope)`` unique lookup). Guarded by an information_schema
column-exists check (safe on re-run and on prod stamp-then-upgrade
recovery).
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "432"
down_revision = "431"
branch_labels = None
depends_on = None

_TABLE = "core_idempotency_records"
_COLUMN = "request_fingerprint"


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


def upgrade() -> None:
    bind = op.get_bind()
    if not _column_exists(bind, _TABLE, _COLUMN):
        op.add_column(
            _TABLE,
            sa.Column(_COLUMN, sa.String(64), nullable=True),
        )


def downgrade() -> None:
    bind = op.get_bind()
    if _column_exists(bind, _TABLE, _COLUMN):
        op.drop_column(_TABLE, _COLUMN)