"""api_key expiry-alert dedup table + seed the daily expiry_check cron. Revision ID: 191 Revises: 190 Create Date: 2026-05-19 Why this exists --------------- PR3-2 of the API-Key lifecycle closure: surface impending expiry to the assignee BEFORE the key actually goes ``expired``. Today ``check_and_update_expired`` flips ``status='expired'`` silently the moment the cron sees ``valid_until < today``; the employee's production code finds out by 401-ing. New cron ``apikey.expiry_check`` (daily 09:00 Asia/Shanghai) sweeps every ``status='active'`` assignment, computes ``days_left = valid_until - today``, and fires: * a **7-day-out reminder** (``threshold_days=7``) — "your key expires next week, click here to extend" * a **1-day-out final warning** (``threshold_days=1``) — "your key expires tomorrow; if you still need it, extend NOW" Dedup contract is the same shape as :class:`ApiKeyBudgetAlertsDedup`: UNIQUE on ``(assignment_id, valid_until_iso, threshold_days)`` so each window fires at most once per validity period per threshold. Admin extends the key → ``valid_until`` changes → fresh dedup row needed → cron re-fires (correct). Idempotency ----------- Both the DDL and the INSERT use ``IF NOT EXISTS``-equivalent guards (``information_schema`` probes + ``WHERE NOT EXISTS`` on the INSERT). Safe to re-run. """ from __future__ import annotations import uuid import sqlalchemy as sa from alembic import op revision = "191" down_revision = "190" branch_labels = None depends_on = None def _table_exists(bind, table: str) -> bool: return bind.execute( sa.text( "SELECT 1 FROM information_schema.TABLES " "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t LIMIT 1" ), {"t": table}, ).scalar() is not None def _fk_source_col_type( bind, target_table: str, target_column: str, *, fallback_type: str = "char", fallback_collation: str | None = None, length: int = 36, ): """Mirror the target FK column's data type + collation. Hotfix (2026-05-20): the original migration declared ``assignment_id`` as ``sa.CHAR(36)`` with the table's default charset/collation. Production's ``api_key_assignments.id`` is ``char(36)`` collated ``utf8mb4_unicode_ci``; the new table's default inherits whatever the connection's charset is at ``CREATE TABLE`` time (often ``utf8mb4_0900_ai_ci`` on newer MySQL or ``utf8mb4_bin`` on TiDB), producing a collation mismatch that TiDB rejects as: (3780, "Referencing column 'assignment_id' and referenced column 'id' in foreign key constraint 'fk_1' are incompatible.") Same pattern PR #74 (bailian tables) and later migrations use: introspect target column at runtime and mirror its type + collation exactly. """ row = bind.execute( sa.text( "SELECT DATA_TYPE, COLLATION_NAME " "FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE() " " AND TABLE_NAME = :t AND COLUMN_NAME = :c" ), {"t": target_table, "c": target_column}, ).fetchone() if row: data_type = (row[0] or fallback_type).lower() collation = row[1] or fallback_collation else: data_type = fallback_type.lower() collation = fallback_collation type_cls = sa.CHAR if data_type == "char" else sa.String if collation: return type_cls(length, collation=collation) return type_cls(length) def upgrade() -> None: bind = op.get_bind() if not _table_exists(bind, "apikey_expiry_alerts_dedup"): # Mirror api_key_assignments.id's exact type + collation so # TiDB's FK strict-mode accepts the constraint. See the # ``_fk_source_col_type`` docstring above for the failure mode # this guards against. assignment_id_col = _fk_source_col_type( bind, "api_key_assignments", "id", fallback_type="char", fallback_collation="utf8mb4_unicode_ci", ) op.create_table( "apikey_expiry_alerts_dedup", sa.Column("id", sa.CHAR(36), primary_key=True), sa.Column( "assignment_id", assignment_id_col, sa.ForeignKey( "api_key_assignments.id", ondelete="CASCADE", ), nullable=False, ), sa.Column("valid_until_iso", sa.Date(), nullable=False), sa.Column("threshold_days", sa.Integer(), nullable=False), sa.Column( "last_alerted_at", sa.DateTime(timezone=True), nullable=False, ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.UniqueConstraint( "assignment_id", "valid_until_iso", "threshold_days", name="uq_apikey_expiry_alerts_dedup_assign_until_threshold", ), sa.Index( "ix_apikey_expiry_alerts_dedup_valid_until", "valid_until_iso", ), ) # Seed the cron only if the scheduled_jobs table is present (some # ancient test schemas may not have it yet — same defensive shape as # migration 167 / 169). if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text( "INSERT INTO scheduled_jobs " "(id, name, display_name, description, handler_key, " "trigger_type, cron_expr, timezone, enabled, config, " "run_count, failure_count, created_at, updated_at) " "SELECT :id, :name, :display, :desc, :handler, 'cron', " ":cron, 'Asia/Shanghai', 1, :config, 0, 0, NOW(), NOW() " "FROM DUAL WHERE NOT EXISTS " "(SELECT 1 FROM scheduled_jobs WHERE name = :name)" ), { "id": str(uuid.uuid4()), "name": "apikey.expiry_check", "display": "API Key 到期提醒", "desc": ( "每日 09:00 (Asia/Shanghai) 扫描每个 status='active' " "的 ApiKeyAssignment;在 days_left == 7 或 days_left " "== 1 的边界给申请人 + 直属上级各发一条 in-app + " "Feishu 提醒,促其在到期前点击「续期」延长 " "valid_until。dedup 表 (assignment_id, " "valid_until_iso, threshold_days) 保证同一 validity " "窗口内每档只发一次;admin 延长后 valid_until 改变 " "→ 自然产生新 dedup key → 下一窗口照常发。已过期的 " "key 由现有 check_and_update_expired 处理,这个 cron " "只做事前预警。" ), "handler": "apikey.expiry_check", "cron": "0 9 * * *", "config": "{}", }, ) def downgrade() -> None: bind = op.get_bind() if _table_exists(bind, "scheduled_jobs"): bind.execute( sa.text( "DELETE FROM scheduled_jobs " "WHERE handler_key = 'apikey.expiry_check'" ) ) if _table_exists(bind, "apikey_expiry_alerts_dedup"): op.drop_table("apikey_expiry_alerts_dedup")