| """Apply and verify the additive thread message-kind migration.""" |
|
|
| from __future__ import annotations |
|
|
| import os |
| import re |
| from pathlib import Path |
|
|
| import psycopg |
| from psycopg.rows import dict_row |
|
|
|
|
| MIGRATION = Path(__file__).parent / "migrations" / "20260814_thread_message_kind.sql" |
|
|
|
|
| def validate_migration(sql: str) -> None: |
| normalized = " ".join(sql.upper().split()) |
| required = ( |
| "ADD COLUMN IF NOT EXISTS MESSAGE_KIND TEXT NOT NULL DEFAULT 'COORDINATION'", |
| "WHERE MESSAGE_KIND = 'COORDINATION'", |
| "THREADS_MESSAGE_KIND_CHECK", |
| "IDX_THREADS_ACTIONABLE", |
| "COMMIT;", |
| ) |
| if not all(token in normalized for token in required): |
| raise RuntimeError("thread message-kind migration shape is incomplete") |
| forbidden = ( |
| r"\bDELETE\s+FROM\b", |
| r"\bTRUNCATE\b", |
| r"\bDROP\s+(TABLE|COLUMN|INDEX)\b", |
| r"\bALTER\s+COLUMN\b", |
| r"\bRENAME\b", |
| ) |
| if any(re.search(pattern, normalized) for pattern in forbidden): |
| raise RuntimeError("thread message-kind migration contains destructive mutation") |
| updates = re.findall(r"\bUPDATE\s+\w+\s+SET\b", normalized) |
| if updates != ["UPDATE THREADS SET"]: |
| raise RuntimeError("thread message-kind migration may contain only its bounded backfill") |
|
|
|
|
| def apply_thread_message_kind_migration(database_url: str, migration_path: Path = MIGRATION) -> None: |
| sql = migration_path.read_text(encoding="utf-8") |
| validate_migration(sql) |
| with psycopg.connect(database_url, autocommit=True, row_factory=dict_row) as conn: |
| with conn.cursor() as cur: |
| cur.execute(sql, prepare=False) |
| cur.execute( |
| """ |
| SELECT |
| to_regclass('public.threads') IS NULL AS threads_absent, |
| EXISTS ( |
| SELECT 1 FROM information_schema.columns |
| WHERE table_schema='public' AND table_name='threads' |
| AND column_name='message_kind' AND is_nullable='NO' |
| ) AS message_kind_present, |
| EXISTS ( |
| SELECT 1 FROM pg_constraint c |
| JOIN pg_class t ON t.oid=c.conrelid |
| JOIN pg_namespace n ON n.oid=t.relnamespace |
| WHERE c.conname='threads_message_kind_check' |
| AND t.relname='threads' AND n.nspname='public' |
| ) AS check_present, |
| to_regclass('public.idx_threads_actionable') IS NOT NULL |
| AS actionable_index_present |
| """ |
| ) |
| verification = cur.fetchone() or {} |
| if not verification.get("threads_absent") and not all( |
| verification.get(name) for name in ( |
| "message_kind_present", "check_present", "actionable_index_present", |
| ) |
| ): |
| raise RuntimeError("thread message-kind migration verification failed") |
|
|
|
|
| def main() -> None: |
| database_url = os.environ.get("DATABASE_URL") |
| if not database_url: |
| raise RuntimeError("DATABASE_URL env var is required") |
| apply_thread_message_kind_migration(database_url) |
| print("thread message-kind schema verified") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
