fix(migrations): m008 also widens checkpoint PK to (bus_name, subscriber_id)

Without this, fresh-DB setups would have bus_name added but the PK would
remain (subscriber_id), causing ON CONFLICT (bus_name, subscriber_id) to
fail at runtime. Adding the PK widening here ensures it runs in the correct
order (after the table is created by m003).

Author: Istar-Eldritch

epoch_pg/src/migrations/m008_add_bus_name_to_checkpoints.rs

@@ -1,9 +1,9 @@
-//! Migration 008: Add `bus_name` column to `epoch_event_bus_checkpoints`.
+//! Migration 008: Add `bus_name` column and widen the PK to `(bus_name, subscriber_id)`.
 //!
-//! Required for the composite checkpoint key `(bus_name, subscriber_id)` so that
-//! per-domain buses (each with their own `events_table`) maintain isolated
-//! checkpoint rows in the shared table.
-//! Existing rows are backfilled to `'epoch_events'` (the legacy default).
+//! Required for per-domain buses — each bus uses its own `events_table` as the
+//! `bus_name` key so checkpoints remain isolated in the shared table.
+//! Existing rows are backfilled to `'epoch_events'` (the legacy default), then the
+//! default is dropped so every new insert must supply an explicit value.
 
 use async_trait::async_trait;
 use sqlx::{Postgres, Transaction};
@@ -23,6 +23,7 @@ impl Migration for AddBusNameToCheckpoints {
     }
 
     async fn up<'a>(&self, tx: &mut Transaction<'a, Postgres>) -> Result<(), MigrationError> {
+        // Add bus_name with a backfill default for existing rows.
         sqlx::query(
             r#"
             ALTER TABLE epoch_event_bus_checkpoints
@@ -32,6 +33,39 @@ impl Migration for AddBusNameToCheckpoints {
         .execute(&mut **tx)
         .await?;
 
+        // Widen the PK to (bus_name, subscriber_id).
+        // Drop the legacy single-column PK constraint first (name may vary after the
+        // m004 rename, so we handle both names).
+        sqlx::query(
+            "ALTER TABLE epoch_event_bus_checkpoints \
+             DROP CONSTRAINT IF EXISTS epoch_event_bus_checkpoints_pkey",
+        )
+        .execute(&mut **tx)
+        .await?;
+        sqlx::query(
+            "ALTER TABLE epoch_event_bus_checkpoints \
+             DROP CONSTRAINT IF EXISTS event_bus_checkpoints_pkey",
+        )
+        .execute(&mut **tx)
+        .await?;
+
+        // Add the composite PK (idempotent: if it somehow already exists, this is a no-op
+        // at the migration-runner level because this migration only runs once).
+        sqlx::query(
+            "ALTER TABLE epoch_event_bus_checkpoints \
+             ADD PRIMARY KEY (bus_name, subscriber_id)",
+        )
+        .execute(&mut **tx)
+        .await?;
+
+        // Drop the column default now that all rows carry an explicit value.
+        sqlx::query(
+            "ALTER TABLE epoch_event_bus_checkpoints \
+             ALTER COLUMN bus_name DROP DEFAULT",
+        )
+        .execute(&mut **tx)
+        .await?;
+
         Ok(())
     }
 }