gize_db/
migrate.rs

1//! Running database migrations via SQLx's runtime migrator (ADR-011).
2//!
3//! We reuse SQLx's `Migrator`, which loads the `migrations/*.sql` files at runtime, tracks
4//! applied versions in its `_sqlx_migrations` table, and applies pending ones in order.
5//! The CLI stays synchronous: these helpers own a small current-thread Tokio runtime and
6//! block on it, so the `gize` CLI needs no async plumbing.
7
8use std::path::Path;
9
10use anyhow::{Context, Result};
11use sqlx::PgPool;
12use sqlx::migrate::{Migration, Migrator};
13use sqlx::postgres::PgPoolOptions;
14
15/// Applied vs pending migrations, each labelled `<version>_<description>`.
16#[derive(Debug, Default)]
17pub struct Status {
18    pub applied: Vec<String>,
19    pub pending: Vec<String>,
20}
21
22fn runtime() -> Result<tokio::runtime::Runtime> {
23    tokio::runtime::Builder::new_current_thread()
24        .enable_all()
25        .build()
26        .context("building the async runtime")
27}
28
29async fn connect(database_url: &str) -> Result<PgPool> {
30    PgPoolOptions::new()
31        .max_connections(1)
32        .connect(database_url)
33        .await
34        .context("connecting to the database")
35}
36
37/// Versions already recorded in `_sqlx_migrations`. If the table does not exist yet (no
38/// migration has ever run), this is simply empty.
39async fn applied_versions(pool: &PgPool) -> Vec<i64> {
40    sqlx::query_scalar::<_, i64>("SELECT version FROM _sqlx_migrations ORDER BY version")
41        .fetch_all(pool)
42        .await
43        .unwrap_or_default()
44}
45
46fn label(migration: &Migration) -> String {
47    format!("{}_{}", migration.version, migration.description)
48}
49
50/// Apply all pending migrations. Returns the labels of the ones newly applied.
51pub fn run(database_url: &str, dir: &Path) -> Result<Vec<String>> {
52    runtime()?.block_on(async {
53        let migrator = Migrator::new(dir).await.context("loading migrations")?;
54        let pool = connect(database_url).await?;
55        let before = applied_versions(&pool).await;
56        migrator.run(&pool).await.context("applying migrations")?;
57        let newly = migrator
58            .iter()
59            .filter(|m| !before.contains(&m.version))
60            .map(label)
61            .collect();
62        Ok(newly)
63    })
64}
65
66/// Report which migrations are applied and which are pending.
67pub fn status(database_url: &str, dir: &Path) -> Result<Status> {
68    runtime()?.block_on(async {
69        let migrator = Migrator::new(dir).await.context("loading migrations")?;
70        let pool = connect(database_url).await?;
71        let applied = applied_versions(&pool).await;
72
73        let mut status = Status::default();
74        for migration in migrator.iter() {
75            if applied.contains(&migration.version) {
76                status.applied.push(label(migration));
77            } else {
78                status.pending.push(label(migration));
79            }
80        }
81        Ok(status)
82    })
83}
gize_db/migrate.rs

gize_db/
migrate.rs
