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::AnyPool;
12use sqlx::any::{AnyPoolOptions, install_default_drivers};
13use sqlx::migrate::{Migration, Migrator};
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<AnyPool> {
30    // The `Any` driver works against both `postgres://` and `sqlite:` URLs (ADR-015), so the
31    // migrator supports either database with one code path.
32    install_default_drivers();
33    AnyPoolOptions::new()
34        .max_connections(1)
35        .connect(database_url)
36        .await
37        .context("connecting to the database")
38}
39
40/// Versions already recorded in `_sqlx_migrations`. If the table does not exist yet (no
41/// migration has ever run), this is simply empty.
42async fn applied_versions(pool: &AnyPool) -> Vec<i64> {
43    sqlx::query_scalar::<_, i64>("SELECT version FROM _sqlx_migrations ORDER BY version")
44        .fetch_all(pool)
45        .await
46        .unwrap_or_default()
47}
48
49fn label(migration: &Migration) -> String {
50    format!("{}_{}", migration.version, migration.description)
51}
52
53/// Apply all pending migrations. Returns the labels of the ones newly applied.
54pub fn run(database_url: &str, dir: &Path) -> Result<Vec<String>> {
55    runtime()?.block_on(async {
56        let migrator = Migrator::new(dir).await.context("loading migrations")?;
57        let pool = connect(database_url).await?;
58        let before = applied_versions(&pool).await;
59        migrator.run(&pool).await.context("applying migrations")?;
60        let newly = migrator
61            .iter()
62            .filter(|m| !before.contains(&m.version))
63            .map(label)
64            .collect();
65        Ok(newly)
66    })
67}
68
69/// Report which migrations are applied and which are pending.
70pub fn status(database_url: &str, dir: &Path) -> Result<Status> {
71    runtime()?.block_on(async {
72        let migrator = Migrator::new(dir).await.context("loading migrations")?;
73        let pool = connect(database_url).await?;
74        let applied = applied_versions(&pool).await;
75
76        let mut status = Status::default();
77        for migration in migrator.iter() {
78            if applied.contains(&migration.version) {
79                status.applied.push(label(migration));
80            } else {
81                status.pending.push(label(migration));
82            }
83        }
84        Ok(status)
85    })
86}
gize_db/migrate.rs

gize_db/
migrate.rs
