代码提交

offline-backend-20260512/backend/scripts/migrate_maf_roundtable_conversation_v2.py

@@ -136,6 +136,56 @@ def _indexes_ddl(dialect: str) -> list[str]:
     return stmts
 
 
+def _events_table_has_source_event_id(inspector: Any) -> bool:
+    """Check if the events table already has the source_event_id column."""
+    if EVENTS_TABLE not in inspector.get_table_names():
+        return False
+    cols = {c["name"] for c in inspector.get_columns(EVENTS_TABLE)}
+    return "source_event_id" in cols
+
+
+def _events_table_has_source_unique(inspector: Any) -> bool:
+    """Check if the events table already has the (conversation_id, source_event_id) unique constraint or index.
+
+    SQLite exposes ``CREATE UNIQUE INDEX`` via ``get_indexes()``, not
+    ``get_unique_constraints()``. PostgreSQL and MySQL may use either path.
+    Both must be checked so the migration is truly idempotent on every
+    supported dialect.
+    """
+    if EVENTS_TABLE not in inspector.get_table_names():
+        return False
+    target = {"conversation_id", "source_event_id"}
+    uq_cols = [set(u.get("column_names", [])) for u in inspector.get_unique_constraints(EVENTS_TABLE)]
+    if target in uq_cols:
+        return True
+    idx_cols = [
+        set(idx.get("column_names", []))
+        for idx in inspector.get_indexes(EVENTS_TABLE)
+        if idx.get("unique")
+    ]
+    return target in idx_cols
+
+
+def _alter_add_source_event_id_sql(dialect: str) -> str:
+    """ALTER TABLE to add source_event_id column, portable across dialects."""
+    col_type = "VARCHAR(191)" if dialect in {"postgresql", "mysql", "mariadb"} else "VARCHAR(191)"
+    return f"ALTER TABLE {EVENTS_TABLE} ADD COLUMN source_event_id {col_type}"
+
+
+def _create_source_unique_index_sql(dialect: str) -> str:
+    """Create a unique index on (conversation_id, source_event_id).
+
+    Using CREATE UNIQUE INDEX instead of ALTER TABLE ADD CONSTRAINT because:
+    - SQLite only supports adding constraints via CREATE INDEX (no ALTER TABLE ADD CONSTRAINT)
+    - PostgreSQL and MySQL both accept CREATE UNIQUE INDEX
+    - Existing rows with NULL source_event_id are not deduped (SQL NULLs are distinct)
+    """
+    return (
+        f"CREATE UNIQUE INDEX IF NOT EXISTS uq_maf_conv_events_conv_source "
+        f"ON {EVENTS_TABLE} (conversation_id, source_event_id)"
+    )
+
+
 def validate_schema(inspector: Any) -> list[str]:
     """Return contract violations found by SQLAlchemy inspector."""
     issues: list[str] = []
@@ -179,9 +229,17 @@ def validate_schema(inspector: Any) -> list[str]:
                 issues.append(f"{EVENTS_TABLE}: missing column {required}")
         uqs = inspector.get_unique_constraints(EVENTS_TABLE)
         uq_cols = [set(u.get("column_names", [])) for u in uqs]
-        if {"conversation_id", "seq"} not in uq_cols:
+        # Also check unique indexes — SQLite exposes CREATE UNIQUE INDEX via
+        # get_indexes(), not get_unique_constraints().
+        idx_cols = [
+            set(idx.get("column_names", []))
+            for idx in inspector.get_indexes(EVENTS_TABLE)
+            if idx.get("unique")
+        ]
+        all_uq = uq_cols + idx_cols
+        if {"conversation_id", "seq"} not in all_uq:
             issues.append(f"{EVENTS_TABLE}: missing UNIQUE(conversation_id, seq)")
-        if {"conversation_id", "source_event_id"} not in uq_cols:
+        if {"conversation_id", "source_event_id"} not in all_uq:
             issues.append(f"{EVENTS_TABLE}: missing UNIQUE(conversation_id, source_event_id)")
 
     return issues
@@ -306,6 +364,40 @@ async def run_migration(database_url: str, *, dry_run: bool = False) -> None:
         else:
             logger.info("Already exists: %s", EVENTS_TABLE)
 
+        # ── In-place upgrade for existing events table ──
+        # Old V2 deployments created the events table WITHOUT source_event_id.
+        # Detect and add the column + unique index so dedup works.
+        if EVENTS_TABLE in existing:
+            def _check_source_col(sync_conn: Any) -> bool:
+                return _events_table_has_source_event_id(inspect(sync_conn))
+
+            has_source_col = await conn.run_sync(_check_source_col)
+            if not has_source_col:
+                alter_sql = _alter_add_source_event_id_sql(dialect)
+                logger.info("Upgrading %s: adding source_event_id column", EVENTS_TABLE)
+                if dry_run:
+                    logger.info("[DRY RUN] Would execute: %s", alter_sql)
+                else:
+                    await conn.execute(text(alter_sql))
+
+            def _check_source_unique(sync_conn: Any) -> bool:
+                return _events_table_has_source_unique(inspect(sync_conn))
+
+            has_source_uq = await conn.run_sync(_check_source_unique)
+            if not has_source_uq:
+                uq_sql = _create_source_unique_index_sql(dialect)
+                logger.info("Upgrading %s: adding unique index (conversation_id, source_event_id)", EVENTS_TABLE)
+                if dry_run:
+                    logger.info("[DRY RUN] Would execute: %s", uq_sql)
+                else:
+                    try:
+                        await conn.execute(text(uq_sql))
+                    except Exception as exc:
+                        if "already exists" in str(exc).lower():
+                            pass
+                        else:
+                            raise
+
         for idx_sql in _indexes_ddl(dialect):
             ddls.append(idx_sql)