udb 0.4.21

//! U14 — online migration phase orchestrator.
//!
//! `MigrationPhase` already enumerates `Prepare → Backfill → Validate
//! → Switch → Cleanup` (see `migration/diff_backends.rs`). What was
//! missing — and what the upgrade doc flagged as "follow-up" — was the
//! orchestrator that actually drives a migration through them:
//!
//! - records `(run_id, phase, status, started_at, finished_at, error)`
//!   in a durable ledger;
//! - calls per-phase user hooks (the actual `prepare_*` / `backfill_*`
//!   work for the backend);
//! - **resumes** from the last incomplete phase on restart;
//! - **gates** each phase on a capability check (refuses to advance
//!   into `Switch` if the validation phase didn't pass).
//!
//! This module is pure orchestration logic — no SQL embedded — so it's
//! testable end-to-end against an in-memory ledger and a fake plan.
//! The real Postgres-backed ledger plugs in via [`PhaseLedger`].
//!
//! ## Why a separate runner vs. extending `MigrationAuditSink`?
//!
//! `MigrationAuditSink` records one row per **artifact** (a single
//! backend-resource DDL). A phase records one row per **logical phase
//! of the whole migration** and is the unit operators monitor in the
//! dashboard. Conflating them would force every artifact to repeat
//! the phase context. Keeping them separate matches how the
//! production ledger is queried.
//!
//! ## State machine
//!
//! ```text
//!   ┌─────────┐  ok    ┌─────────┐ ok   ┌──────────┐ ok  ┌────────┐ ok  ┌─────────┐
//!   │ Prepare │──────▶ │ Backfill│────▶ │ Validate │───▶ │ Switch │───▶ │ Cleanup │──▶ Completed
//!   └────┬────┘        └────┬────┘      └────┬─────┘     └───┬────┘     └────┬────┘
//!        │ err              │ err            │ err           │ err           │ err
//!        ▼                  ▼                ▼               ▼               ▼
//!     PhaseFailed       PhaseFailed       PhaseFailed     PhaseFailed     PhaseFailed
//! ```
//!
//! `PhaseFailed` is terminal for the run. The operator either rolls
//! back manually, fixes the underlying issue and retries the failed
//! phase, or marks the run abandoned. The runner does NOT auto-retry
//! — phase failures usually need human judgement, not a bare loop.

use std::collections::HashMap;

use async_trait::async_trait;
use sqlx::{PgPool, Row};

use crate::migration::diff_backends::MigrationPhase;

/// Per-phase status as recorded in the ledger.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PhaseStatus {
    /// Phase row exists but the runner hasn't started this phase yet.
    /// Useful for surfacing "next phase is X" in the dashboard.
    Pending,
    /// Runner is executing the user hook for this phase.
    Running,
    /// User hook returned `Ok(())`. Runner can advance.
    Completed,
    /// User hook returned `Err(_)`. Run is paused; operator must
    /// resolve and either resume or abandon.
    Failed,
    /// Operator marked the run abandoned after a failure. The phase
    /// stays in this state — resumes will not pick it up.
    Abandoned,
}

impl PhaseStatus {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Running => "running",
            Self::Completed => "completed",
            Self::Failed => "failed",
            Self::Abandoned => "abandoned",
        }
    }

    pub fn terminal(self) -> bool {
        matches!(self, Self::Failed | Self::Abandoned | Self::Completed)
    }
}

fn phase_from_str(value: &str) -> Result<MigrationPhase, String> {
    match value {
        "prepare" => Ok(MigrationPhase::Prepare),
        "backfill" => Ok(MigrationPhase::Backfill),
        "validate" => Ok(MigrationPhase::Validate),
        "switch" => Ok(MigrationPhase::Switch),
        "cleanup" => Ok(MigrationPhase::Cleanup),
        _ => Err(format!("unknown migration phase '{value}'")),
    }
}

fn phase_status_from_str(value: &str) -> Result<PhaseStatus, String> {
    match value {
        "pending" => Ok(PhaseStatus::Pending),
        "running" => Ok(PhaseStatus::Running),
        "completed" => Ok(PhaseStatus::Completed),
        "failed" => Ok(PhaseStatus::Failed),
        "abandoned" => Ok(PhaseStatus::Abandoned),
        _ => Err(format!("unknown migration phase status '{value}'")),
    }
}

/// One row in the phase ledger.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct PhaseRecord {
    pub run_id: String,
    pub phase: MigrationPhase,
    pub status: PhaseStatus,
    /// Unix ms when the phase first transitioned to `Running`. `None`
    /// while still `Pending`.
    pub started_at_unix_ms: Option<i64>,
    /// Unix ms when the phase reached a terminal state. `None` while
    /// `Pending` or `Running`.
    pub finished_at_unix_ms: Option<i64>,
    /// Error message from the failing user hook. Empty when not failed.
    pub error: String,
    /// Free-form attempt counter — the operator can resume a failed
    /// phase and the runner bumps this before re-running. Lets the
    /// dashboard show "retry #N".
    pub attempt: u32,
}

/// Persistent ledger for phase records. The production impl writes to
/// `udb_migration_phase_ledger` (Postgres); tests use an in-memory
/// `MemoryPhaseLedger`. The trait is async-object-safe so the runner
/// can hold a `Box<dyn PhaseLedger>` if needed.
#[async_trait]
pub trait PhaseLedger: Send + Sync {
    async fn load(&self, run_id: &str) -> Result<Vec<PhaseRecord>, String>;
    async fn write(&self, record: PhaseRecord) -> Result<(), String>;
}

/// Durable Postgres-backed phase ledger used by the production
/// catalog-admin apply path.
///
/// The table is created lazily so older installations that already have
/// `udb_migration_runs` / `udb_migration_op_ledger` can start recording phase
/// progress without requiring a separate bootstrap migration first.
pub struct PostgresPhaseLedger {
    pool: PgPool,
    relation: String,
}

impl PostgresPhaseLedger {
    pub fn new(pool: PgPool, relation: impl Into<String>) -> Self {
        Self {
            pool,
            relation: relation.into(),
        }
    }

    async fn ensure_table(&self) -> Result<(), String> {
        let rel = &self.relation;
        let ddl = format!(
            "CREATE TABLE IF NOT EXISTS {rel} (
                id BIGSERIAL PRIMARY KEY,
                run_id TEXT NOT NULL,
                phase TEXT NOT NULL CHECK (phase IN ('prepare','backfill','validate','switch','cleanup')),
                status TEXT NOT NULL CHECK (status IN ('pending','running','completed','failed','abandoned')),
                started_at TIMESTAMPTZ,
                finished_at TIMESTAMPTZ,
                error TEXT NOT NULL DEFAULT '',
                attempt INTEGER NOT NULL DEFAULT 0,
                updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                UNIQUE (run_id, phase)
            )"
        );
        sqlx::query(&ddl)
            .execute(&self.pool)
            .await
            .map_err(|err| format!("ensure phase ledger table failed: {err}"))?;

        let idx = format!(
            "CREATE INDEX IF NOT EXISTS \"idx_udb_migration_phase_ledger_run\"
             ON {rel} (run_id, phase)"
        );
        sqlx::query(&idx)
            .execute(&self.pool)
            .await
            .map_err(|err| format!("ensure phase ledger index failed: {err}"))?;
        Ok(())
    }
}

#[async_trait]
impl PhaseLedger for PostgresPhaseLedger {
    async fn load(&self, run_id: &str) -> Result<Vec<PhaseRecord>, String> {
        self.ensure_table().await?;
        let rel = &self.relation;
        let sql = format!(
            "SELECT run_id, phase, status,
                    (EXTRACT(EPOCH FROM started_at) * 1000)::BIGINT AS started_at_unix_ms,
                    (EXTRACT(EPOCH FROM finished_at) * 1000)::BIGINT AS finished_at_unix_ms,
                    error, attempt
             FROM {rel}
             WHERE run_id = $1
             ORDER BY CASE phase
                 WHEN 'prepare' THEN 1
                 WHEN 'backfill' THEN 2
                 WHEN 'validate' THEN 3
                 WHEN 'switch' THEN 4
                 WHEN 'cleanup' THEN 5
                 ELSE 99
             END"
        );
        let rows = sqlx::query(&sql)
            .bind(run_id)
            .fetch_all(&self.pool)
            .await
            .map_err(|err| format!("load phase ledger failed: {err}"))?;

        let mut out = Vec::with_capacity(rows.len());
        for row in rows {
            let phase = phase_from_str(
                row.try_get::<String, _>("phase")
                    .unwrap_or_default()
                    .as_str(),
            )?;
            let status = phase_status_from_str(
                row.try_get::<String, _>("status")
                    .unwrap_or_default()
                    .as_str(),
            )?;
            out.push(PhaseRecord {
                run_id: row.try_get::<String, _>("run_id").unwrap_or_default(),
                phase,
                status,
                started_at_unix_ms: row
                    .try_get::<Option<i64>, _>("started_at_unix_ms")
                    .ok()
                    .flatten(),
                finished_at_unix_ms: row
                    .try_get::<Option<i64>, _>("finished_at_unix_ms")
                    .ok()
                    .flatten(),
                error: row.try_get::<String, _>("error").unwrap_or_default(),
                attempt: row.try_get::<i32, _>("attempt").unwrap_or_default().max(0) as u32,
            });
        }
        Ok(out)
    }

    async fn write(&self, record: PhaseRecord) -> Result<(), String> {
        self.ensure_table().await?;
        let rel = &self.relation;
        let sql = format!(
            "INSERT INTO {rel}
                (run_id, phase, status, started_at, finished_at, error, attempt, updated_at)
             VALUES (
                $1, $2, $3,
                CASE WHEN $4::BIGINT IS NULL THEN NULL ELSE to_timestamp(($4::BIGINT)::DOUBLE PRECISION / 1000.0) END,
                CASE WHEN $5::BIGINT IS NULL THEN NULL ELSE to_timestamp(($5::BIGINT)::DOUBLE PRECISION / 1000.0) END,
                $6, $7, NOW()
             )
             ON CONFLICT (run_id, phase) DO UPDATE
                SET status = EXCLUDED.status,
                    started_at = EXCLUDED.started_at,
                    finished_at = EXCLUDED.finished_at,
                    error = EXCLUDED.error,
                    attempt = EXCLUDED.attempt,
                    updated_at = NOW()"
        );
        sqlx::query(&sql)
            .bind(&record.run_id)
            .bind(record.phase.as_str())
            .bind(record.status.as_str())
            .bind(record.started_at_unix_ms)
            .bind(record.finished_at_unix_ms)
            .bind(&record.error)
            .bind(i32::try_from(record.attempt).unwrap_or(i32::MAX))
            .execute(&self.pool)
            .await
            .map_err(|err| format!("write phase ledger failed: {err}"))?;
        Ok(())
    }
}

/// The per-phase user hook the runner calls. Returns `Ok(())` to
/// advance, `Err(msg)` to pause the run on this phase.
///
/// Hooks must be **idempotent** — a resume after restart will replay
/// the last incomplete phase.
#[async_trait]
pub trait MigrationPhaseHook: Send + Sync {
    /// Called with the current phase. Implementations dispatch by
    /// `phase` to whatever backend work is needed.
    async fn run(&self, phase: MigrationPhase) -> Result<(), String>;

    /// Capability check: is this phase allowed to run on the current
    /// deployment? Default `true` for every phase. Override to refuse
    /// e.g. `Switch` when the validation step has not been signed off.
    fn capable_of(&self, _phase: MigrationPhase) -> bool {
        true
    }
}

/// What the runner returns after a `run_to_completion` call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RunnerOutcome {
    /// Every phase reached `Completed`.
    Completed { run_id: String },
    /// A phase returned `Err(_)` and the runner stopped. The dashboard
    /// shows which phase failed and the error text from the hook.
    Paused {
        run_id: String,
        phase: MigrationPhase,
        error: String,
    },
    /// The runner refused to start a phase because the hook's
    /// `capable_of` returned false. Pins that the capability check
    /// runs **before** the phase executes — no partial side effects.
    Refused {
        run_id: String,
        phase: MigrationPhase,
    },
}

/// Drive `run_id` through every phase, starting at the first
/// non-completed phase (so a restart resumes where it left off).
///
/// Algorithm:
///
/// 1. Load every existing `PhaseRecord` for `run_id` from the ledger.
/// 2. For each phase in `MigrationPhase::all()`:
///    a. Look up the existing record (or treat as `Pending`).
///    b. If `Completed`, skip.
///    c. If `Abandoned`, return `Refused` (operator chose to stop).
///    d. Check `hook.capable_of(phase)`. If `false`, write a `Failed`
///       record and return `Refused`.
///    e. Write `Running` to the ledger, call `hook.run(phase)`.
///    f. On `Ok`, write `Completed`; on `Err`, write `Failed` and
///       return `Paused`.
/// 3. Return `Completed`.
pub async fn run_to_completion(
    run_id: &str,
    ledger: &dyn PhaseLedger,
    hook: &dyn MigrationPhaseHook,
) -> Result<RunnerOutcome, String> {
    let existing = ledger.load(run_id).await?;
    let by_phase: HashMap<MigrationPhase, PhaseRecord> =
        existing.into_iter().map(|r| (r.phase, r)).collect();

    for phase in MigrationPhase::all().iter().copied() {
        let prev = by_phase.get(&phase);
        match prev.map(|r| r.status) {
            Some(PhaseStatus::Completed) => continue,
            Some(PhaseStatus::Abandoned) => {
                return Ok(RunnerOutcome::Refused {
                    run_id: run_id.to_string(),
                    phase,
                });
            }
            _ => {}
        }

        if !hook.capable_of(phase) {
            let record = PhaseRecord {
                run_id: run_id.to_string(),
                phase,
                status: PhaseStatus::Failed,
                started_at_unix_ms: None,
                finished_at_unix_ms: Some(now_unix_ms()),
                error: format!("capability check refused phase {}", phase.as_str()),
                attempt: prev.map(|p| p.attempt).unwrap_or(0).saturating_add(1),
            };
            ledger.write(record).await?;
            return Ok(RunnerOutcome::Refused {
                run_id: run_id.to_string(),
                phase,
            });
        }

        let attempt = prev.map(|p| p.attempt).unwrap_or(0).saturating_add(1);
        let started = now_unix_ms();
        ledger
            .write(PhaseRecord {
                run_id: run_id.to_string(),
                phase,
                status: PhaseStatus::Running,
                started_at_unix_ms: Some(started),
                finished_at_unix_ms: None,
                error: String::new(),
                attempt,
            })
            .await?;
        match hook.run(phase).await {
            Ok(()) => {
                ledger
                    .write(PhaseRecord {
                        run_id: run_id.to_string(),
                        phase,
                        status: PhaseStatus::Completed,
                        started_at_unix_ms: Some(started),
                        finished_at_unix_ms: Some(now_unix_ms()),
                        error: String::new(),
                        attempt,
                    })
                    .await?;
            }
            Err(reason) => {
                ledger
                    .write(PhaseRecord {
                        run_id: run_id.to_string(),
                        phase,
                        status: PhaseStatus::Failed,
                        started_at_unix_ms: Some(started),
                        finished_at_unix_ms: Some(now_unix_ms()),
                        error: reason.clone(),
                        attempt,
                    })
                    .await?;
                return Ok(RunnerOutcome::Paused {
                    run_id: run_id.to_string(),
                    phase,
                    error: reason,
                });
            }
        }
    }

    Ok(RunnerOutcome::Completed {
        run_id: run_id.to_string(),
    })
}

/// C (2026-05-30): in-memory `PhaseLedger` for orchestration UNIT TESTS only.
/// Stores rows behind a `Mutex` so the trait methods stay `&self`.
///
/// IN-MEMORY-AUDIT [A1, no-in-memory rule]: `#[cfg(test)]`-gated so it is
/// NOT compiled into the shipped binary. Production uses
/// [`PostgresPhaseLedger`] through the catalog-admin apply path.
#[cfg(test)]
#[derive(Default, Debug)]
pub struct MemoryPhaseLedger {
    rows: std::sync::Mutex<Vec<PhaseRecord>>,
}

#[cfg(test)]
#[async_trait]
impl PhaseLedger for MemoryPhaseLedger {
    async fn load(&self, run_id: &str) -> Result<Vec<PhaseRecord>, String> {
        Ok(self
            .rows
            .lock()
            .map_err(|e| format!("MemoryPhaseLedger.rows poisoned: {e}"))?
            .iter()
            .filter(|r| r.run_id == run_id)
            .cloned()
            .collect())
    }

    async fn write(&self, record: PhaseRecord) -> Result<(), String> {
        let mut rows = self
            .rows
            .lock()
            .map_err(|e| format!("MemoryPhaseLedger.rows poisoned: {e}"))?;
        // Upsert by (run_id, phase) — phases are unique per run.
        rows.retain(|r| !(r.run_id == record.run_id && r.phase == record.phase));
        rows.push(record);
        Ok(())
    }
}

fn now_unix_ms() -> i64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    // ── Test doubles ────────────────────────────────────────────────
    // The public `MemoryPhaseLedger` lives in the parent module so
    // other crates can use it; tests just alias it locally.
    type MemoryLedger = MemoryPhaseLedger;

    /// Hook that runs every phase successfully. Records the order it
    /// was called so tests can assert phase ordering.
    struct OkHook {
        seen: Mutex<Vec<MigrationPhase>>,
    }
    impl OkHook {
        fn new() -> Self {
            Self {
                seen: Mutex::new(Vec::new()),
            }
        }
    }

    #[async_trait]
    impl MigrationPhaseHook for OkHook {
        async fn run(&self, phase: MigrationPhase) -> Result<(), String> {
            self.seen.lock().unwrap().push(phase);
            Ok(())
        }
    }

    /// Hook that fails the specified phase. Pins the pause/resume
    /// contract.
    struct FailAt {
        target: MigrationPhase,
    }

    #[async_trait]
    impl MigrationPhaseHook for FailAt {
        async fn run(&self, phase: MigrationPhase) -> Result<(), String> {
            if phase == self.target {
                Err(format!("synthetic failure at {}", phase.as_str()))
            } else {
                Ok(())
            }
        }
    }

    /// Hook that refuses the Switch phase via capability check —
    /// e.g. validate hasn't been signed off.
    struct RefuseSwitch;

    #[async_trait]
    impl MigrationPhaseHook for RefuseSwitch {
        async fn run(&self, _phase: MigrationPhase) -> Result<(), String> {
            Ok(())
        }
        fn capable_of(&self, phase: MigrationPhase) -> bool {
            phase != MigrationPhase::Switch
        }
    }

    // ── Tests ───────────────────────────────────────────────────────

    /// Happy path: every phase advances in order, ledger ends with
    /// 5 `Completed` rows.
    #[tokio::test]
    async fn run_to_completion_advances_through_every_phase() {
        let ledger = MemoryLedger::default();
        let hook = OkHook::new();
        let outcome = run_to_completion("run-1", &ledger, &hook).await.unwrap();
        assert_eq!(
            outcome,
            RunnerOutcome::Completed {
                run_id: "run-1".to_string()
            }
        );
        // Hook saw every phase in order.
        let seen = hook.seen.lock().unwrap().clone();
        assert_eq!(seen, MigrationPhase::all().to_vec());
        // Ledger has 5 rows, all completed.
        let rows = ledger.load("run-1").await.unwrap();
        assert_eq!(rows.len(), 5);
        for row in &rows {
            assert_eq!(row.status, PhaseStatus::Completed, "phase {:?}", row.phase);
            assert!(row.finished_at_unix_ms.is_some());
        }
    }

    /// Failure pauses the run on the failing phase. Resuming after
    /// the hook is fixed picks up exactly where it left off — no
    /// re-running completed phases.
    #[tokio::test]
    async fn failure_pauses_then_resume_continues_without_replay() {
        let ledger = MemoryLedger::default();
        let outcome = run_to_completion(
            "run-r",
            &ledger,
            &FailAt {
                target: MigrationPhase::Backfill,
            },
        )
        .await
        .unwrap();
        match outcome {
            RunnerOutcome::Paused { phase, error, .. } => {
                assert_eq!(phase, MigrationPhase::Backfill);
                assert!(error.contains("synthetic failure at backfill"));
            }
            other => panic!("expected Paused, got {:?}", other),
        }
        // Prepare must be Completed; Backfill must be Failed; others
        // must not exist yet.
        let rows = ledger.load("run-r").await.unwrap();
        let by: HashMap<MigrationPhase, PhaseRecord> =
            rows.into_iter().map(|r| (r.phase, r)).collect();
        assert_eq!(
            by.get(&MigrationPhase::Prepare).map(|r| r.status),
            Some(PhaseStatus::Completed)
        );
        assert_eq!(
            by.get(&MigrationPhase::Backfill).map(|r| r.status),
            Some(PhaseStatus::Failed)
        );
        assert!(by.get(&MigrationPhase::Validate).is_none());

        // Resume with a hook that succeeds everywhere. The runner
        // must NOT call `prepare` again because it's already Completed.
        let resume_hook = OkHook::new();
        let outcome = run_to_completion("run-r", &ledger, &resume_hook)
            .await
            .unwrap();
        assert!(matches!(outcome, RunnerOutcome::Completed { .. }));
        let seen = resume_hook.seen.lock().unwrap().clone();
        // Prepare was skipped (already completed); the rest ran.
        assert_eq!(
            seen,
            vec![
                MigrationPhase::Backfill,
                MigrationPhase::Validate,
                MigrationPhase::Switch,
                MigrationPhase::Cleanup,
            ]
        );
        // Backfill attempt counter bumped to 2 (1 failed + 1 retry).
        let rows = ledger.load("run-r").await.unwrap();
        let backfill = rows
            .iter()
            .find(|r| r.phase == MigrationPhase::Backfill)
            .unwrap();
        assert_eq!(backfill.status, PhaseStatus::Completed);
        assert_eq!(backfill.attempt, 2);
    }

    /// Capability refusal: `capable_of(Switch) == false` blocks the
    /// switch phase **before** any side effect. Pins that the hook
    /// `run` is not called for a refused phase.
    #[tokio::test]
    async fn capability_refusal_blocks_before_side_effects() {
        let ledger = MemoryLedger::default();
        let outcome = run_to_completion("run-c", &ledger, &RefuseSwitch)
            .await
            .unwrap();
        match outcome {
            RunnerOutcome::Refused { phase, .. } => {
                assert_eq!(phase, MigrationPhase::Switch);
            }
            other => panic!("expected Refused at Switch, got {:?}", other),
        }
        // Prepare/Backfill/Validate completed; Switch failed; Cleanup
        // never written.
        let rows = ledger.load("run-c").await.unwrap();
        let by: HashMap<MigrationPhase, PhaseRecord> =
            rows.into_iter().map(|r| (r.phase, r)).collect();
        assert_eq!(
            by.get(&MigrationPhase::Validate).map(|r| r.status),
            Some(PhaseStatus::Completed)
        );
        let switch = by.get(&MigrationPhase::Switch).unwrap();
        assert_eq!(switch.status, PhaseStatus::Failed);
        assert!(
            switch.error.contains("capability check refused"),
            "got: {}",
            switch.error
        );
        assert_eq!(
            switch.started_at_unix_ms, None,
            "no started timestamp = no side effect"
        );
        assert!(by.get(&MigrationPhase::Cleanup).is_none());
    }

    /// Abandoned phase short-circuits the runner without calling
    /// the hook — operator chose to stop. Pins the workflow.
    #[tokio::test]
    async fn abandoned_phase_refuses_to_advance() {
        let ledger = MemoryLedger::default();
        // Manually plant an Abandoned row for Switch.
        ledger
            .write(PhaseRecord {
                run_id: "run-a".to_string(),
                phase: MigrationPhase::Switch,
                status: PhaseStatus::Abandoned,
                started_at_unix_ms: None,
                finished_at_unix_ms: Some(now_unix_ms()),
                error: "operator chose to stop".to_string(),
                attempt: 1,
            })
            .await
            .unwrap();
        let outcome = run_to_completion("run-a", &ledger, &OkHook::new())
            .await
            .unwrap();
        match outcome {
            RunnerOutcome::Refused { phase, .. } => {
                assert_eq!(phase, MigrationPhase::Switch);
            }
            other => panic!("expected Refused, got {:?}", other),
        }
    }

    /// Pin: PhaseStatus tokens are the wire contract. Changing one of
    /// these breaks the dashboard.
    #[test]
    fn phase_status_tokens_are_pinned() {
        assert_eq!(PhaseStatus::Pending.as_str(), "pending");
        assert_eq!(PhaseStatus::Running.as_str(), "running");
        assert_eq!(PhaseStatus::Completed.as_str(), "completed");
        assert_eq!(PhaseStatus::Failed.as_str(), "failed");
        assert_eq!(PhaseStatus::Abandoned.as_str(), "abandoned");
    }
}