"""Durable Feishu approval submission intents. Revision ID: 442 Revises: 441 Create Date: 2026-08-02 #2283: persist the request fingerprint and Feishu-native UUID before the external create call so timeout/response-loss and process death can be reconciled without creating a second approval instance. """ from __future__ import annotations import sqlalchemy as sa from sqlalchemy.dialects import mysql from alembic import op from app.core.sql_types import GUID revision = "442" down_revision = "441" branch_labels = None depends_on = None TABLE = "approval_submission_intents" RECONCILE_INDEX = "ix_approval_submission_intents_reconcile" _REQUIRED_COLUMNS = { "id", "scope", "applicant_id", "idempotency_key", "payload_fingerprint", "provider_request_uuid", "state", "payload", "external_id", "employee_request_id", "approval_request_id", "attempt_count", "last_error", "created_at", "updated_at", "completed_at", } _REQUIRED_UNIQUES = { ("scope", "applicant_id", "idempotency_key"), ("provider_request_uuid",), ("external_id",), } class _TextIdMeta: __slots__ = ("data_type", "length", "charset", "collation") def __init__(self, data_type: str, length: int, charset: str, collation: str) -> None: self.data_type = data_type self.length = length self.charset = charset self.collation = collation def sqlalchemy_type(self) -> sa.types.TypeEngine: type_class = mysql.CHAR if self.data_type == "char" else mysql.VARCHAR return type_class( self.length, charset=self.charset, collation=self.collation, ) def key(self) -> tuple[str, int, str, str]: return self.data_type, self.length, self.charset, self.collation def _id_meta(table: str, column: str = "id") -> _TextIdMeta: row = ( op.get_bind() .execute( sa.text( "SELECT DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, " "CHARACTER_SET_NAME, COLLATION_NAME " "FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " "AND TABLE_NAME = :table AND COLUMN_NAME = :column" ), {"table": table, "column": column}, ) .mappings() .one_or_none() ) if row is None: raise RuntimeError(f"missing required parent column {table}.{column}") data_type = str(row["DATA_TYPE"] or "").lower() length = row["CHARACTER_MAXIMUM_LENGTH"] charset = row["CHARACTER_SET_NAME"] collation = row["COLLATION_NAME"] if ( data_type not in {"char", "varchar"} or not isinstance(length, int) or length <= 0 or not charset or not collation ): raise RuntimeError( f"unsupported or incomplete metadata for {table}.{column}: " f"type={data_type!r}, length={length!r}, " f"charset={charset!r}, collation={collation!r}" ) return _TextIdMeta(data_type, length, str(charset), str(collation)) def _create_table(parent_meta: dict[str, _TextIdMeta]) -> None: # Create the complete data shape first, without FKs. MySQL/TiDB DDL is # non-transactional, so a process interruption can then resume by adding # only the missing constraints rather than retrying CREATE TABLE forever. op.create_table( TABLE, sa.Column("id", GUID(), nullable=False), sa.Column("scope", sa.String(64), nullable=False), sa.Column("applicant_id", parent_meta["applicant_id"].sqlalchemy_type(), nullable=False), sa.Column("idempotency_key", sa.String(128), nullable=False), sa.Column("payload_fingerprint", sa.String(64), nullable=False), sa.Column("provider_request_uuid", sa.String(64), nullable=False), sa.Column("state", sa.String(32), nullable=False), sa.Column("payload", sa.JSON(), nullable=False), sa.Column("external_id", sa.String(128), nullable=True), sa.Column( "employee_request_id", parent_meta["employee_request_id"].sqlalchemy_type(), nullable=True, ), sa.Column( "approval_request_id", parent_meta["approval_request_id"].sqlalchemy_type(), nullable=True, ), sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"), sa.Column("last_error", sa.String(512), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint( "scope", "applicant_id", "idempotency_key", name="uq_approval_submission_intent_key", ), sa.UniqueConstraint( "provider_request_uuid", name="uq_approval_submission_intent_provider_uuid", ), sa.UniqueConstraint("external_id"), mysql_charset=parent_meta["applicant_id"].charset, mysql_collate=parent_meta["applicant_id"].collation, ) def _matching_fk(inspector: sa.Inspector, column: str, parent_table: str) -> dict | None: matches = [ fk for fk in inspector.get_foreign_keys(TABLE) if tuple(fk.get("constrained_columns") or ()) == (column,) and str(fk.get("referred_table") or "") == parent_table and tuple(fk.get("referred_columns") or ()) == ("id",) ] if len(matches) > 1: raise RuntimeError(f"multiple foreign keys found for {TABLE}.{column}") return matches[0] if matches else None def _normalize_and_add_fk( column: str, parent_table: str, constraint_name: str, parent_meta: _TextIdMeta, *, nullable: bool, ) -> None: inspector = sa.inspect(op.get_bind()) existing_fk = _matching_fk(inspector, column, parent_table) child_meta = _id_meta(TABLE, column) if existing_fk is not None and child_meta.key() == parent_meta.key(): return if existing_fk is not None: existing_name = existing_fk.get("name") if not existing_name: raise RuntimeError(f"cannot identify existing foreign key for {TABLE}.{column}") op.drop_constraint(str(existing_name), TABLE, type_="foreignkey") op.alter_column( TABLE, column, type_=parent_meta.sqlalchemy_type(), nullable=nullable, ) op.create_foreign_key( constraint_name, TABLE, parent_table, [column], ["id"], ) def upgrade() -> None: parents = { "applicant_id": ("identity_persons", False), "employee_request_id": ("employee_requests", True), "approval_request_id": ("approval_requests", True), } parent_meta = {column: _id_meta(parent) for column, (parent, _nullable) in parents.items()} inspector = sa.inspect(op.get_bind()) if not inspector.has_table(TABLE): _create_table(parent_meta) else: columns = {column["name"] for column in inspector.get_columns(TABLE)} unique_sets = { tuple(constraint.get("column_names") or ()) for constraint in inspector.get_unique_constraints(TABLE) } if columns != _REQUIRED_COLUMNS or not _REQUIRED_UNIQUES.issubset(unique_sets): raise RuntimeError( f"existing {TABLE} has an unexpected shape; refusing destructive repair" ) for column, (parent, nullable) in parents.items(): _normalize_and_add_fk( column, parent, f"fk_approval_submission_intents_{column.removesuffix('_id')}", parent_meta[column], nullable=nullable, ) inspector = sa.inspect(op.get_bind()) indexes = {index["name"] for index in inspector.get_indexes(TABLE)} if RECONCILE_INDEX not in indexes: op.create_index(RECONCILE_INDEX, TABLE, ["state", "updated_at"]) def downgrade() -> None: op.drop_index(RECONCILE_INDEX, table_name=TABLE) op.drop_table(TABLE)