shardline_server/
database_migration.rs

1use sqlx::{Error as SqlxError, PgPool, Row, postgres::PgPoolOptions, query, raw_sql};
2use thiserror::Error;
3
4/// One Shardline schema migration.
5#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6pub struct DatabaseMigration {
7    /// Monotonic migration version identifier.
8    pub version: &'static str,
9    /// Human-readable migration name.
10    pub name: &'static str,
11    /// SQL applied when migrating forward.
12    pub up_sql: &'static str,
13    /// SQL applied when reverting the migration.
14    pub down_sql: &'static str,
15}
16
17/// Requested database-migration action.
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub enum DatabaseMigrationCommand {
20    /// Apply pending migrations.
21    Up {
22        /// Maximum number of pending migrations to apply.
23        steps: Option<usize>,
24    },
25    /// Revert applied migrations from newest to oldest.
26    Down {
27        /// Maximum number of applied migrations to revert.
28        steps: usize,
29    },
30    /// Report applied and pending migrations without mutating schema state.
31    Status,
32}
33
34/// Database-migration runtime options.
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct DatabaseMigrationOptions {
37    database_url: String,
38    command: DatabaseMigrationCommand,
39}
40
41impl DatabaseMigrationOptions {
42    /// Creates database-migration options.
43    #[must_use]
44    pub const fn new(database_url: String, command: DatabaseMigrationCommand) -> Self {
45        Self {
46            database_url,
47            command,
48        }
49    }
50
51    /// Returns the Postgres connection URL.
52    #[must_use]
53    pub fn database_url(&self) -> &str {
54        &self.database_url
55    }
56
57    /// Returns the selected command.
58    #[must_use]
59    pub const fn command(&self) -> &DatabaseMigrationCommand {
60        &self.command
61    }
62}
63
64/// One migration row in the status report.
65#[derive(Debug, Clone, PartialEq, Eq)]
66pub struct DatabaseMigrationStatusEntry {
67    /// Monotonic migration version identifier.
68    pub version: String,
69    /// Human-readable migration name.
70    pub name: String,
71    /// Whether this migration is currently applied.
72    pub applied: bool,
73    /// UTC application timestamp when applied.
74    pub applied_at_utc: Option<String>,
75}
76
77/// Database-migration execution report.
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub struct DatabaseMigrationReport {
80    /// Backend identifier.
81    pub backend: String,
82    /// Requested command.
83    pub command: DatabaseMigrationCommand,
84    /// Number of migrations applied during this run.
85    pub applied_count: u64,
86    /// Number of migrations reverted during this run.
87    pub reverted_count: u64,
88    /// Number of migrations applied after this run completes.
89    pub applied_total_count: u64,
90    /// Number of migrations still pending after this run.
91    pub pending_count: u64,
92    /// Full ordered status for every bundled migration.
93    pub migrations: Vec<DatabaseMigrationStatusEntry>,
94}
95
96/// Database-migration failure.
97#[derive(Debug, Error)]
98pub enum DatabaseMigrationError {
99    /// The database URL was empty.
100    #[error("database URL must not be empty")]
101    EmptyDatabaseUrl,
102    /// Postgres access failed.
103    #[error(transparent)]
104    Sqlx(#[from] SqlxError),
105    /// Migration history contains a version unknown to the running binary.
106    #[error("database contains an unknown shardline migration version: {0}")]
107    UnknownAppliedMigration(String),
108    /// A previously applied migration no longer matches the bundled SQL.
109    #[error(
110        "bundled migration checksum mismatch for version {version}: expected {expected_checksum}, observed {observed_checksum}"
111    )]
112    ChecksumMismatch {
113        /// Bundled migration version.
114        version: String,
115        /// Hash of the bundled SQL.
116        expected_checksum: String,
117        /// Hash recorded in the database.
118        observed_checksum: String,
119    },
120}
121
122#[derive(Debug, Clone, PartialEq, Eq)]
123struct AppliedMigration {
124    version: String,
125    checksum: String,
126    applied_at_utc: String,
127}
128
129const MIGRATION_HISTORY_TABLE: &str = "shardline_schema_migrations";
130
131const SHARDLINE_MIGRATIONS: [DatabaseMigration; 7] = [
132    DatabaseMigration {
133        version: "20260417000000",
134        name: "metadata_store",
135        up_sql: include_str!("../migrations/20260417000000_metadata_store.up.sql"),
136        down_sql: include_str!("../migrations/20260417000000_metadata_store.down.sql"),
137    },
138    DatabaseMigration {
139        version: "20260417010000",
140        name: "retention_holds",
141        up_sql: include_str!("../migrations/20260417010000_retention_holds.up.sql"),
142        down_sql: include_str!("../migrations/20260417010000_retention_holds.down.sql"),
143    },
144    DatabaseMigration {
145        version: "20260418000000",
146        name: "dedupe_shards",
147        up_sql: include_str!("../migrations/20260418000000_dedupe_shards.up.sql"),
148        down_sql: include_str!("../migrations/20260418000000_dedupe_shards.down.sql"),
149    },
150    DatabaseMigration {
151        version: "20260418010000",
152        name: "webhook_deliveries",
153        up_sql: include_str!("../migrations/20260418010000_webhook_deliveries.up.sql"),
154        down_sql: include_str!("../migrations/20260418010000_webhook_deliveries.down.sql"),
155    },
156    DatabaseMigration {
157        version: "20260418020000",
158        name: "provider_repository_states",
159        up_sql: include_str!(
160            "../migrations/20260418020000_provider_repository_states.up.sql"
161        ),
162        down_sql: include_str!(
163            "../migrations/20260418020000_provider_repository_states.down.sql"
164        ),
165    },
166    DatabaseMigration {
167        version: "20260418110000",
168        name: "provider_repository_reconciliation",
169        up_sql: include_str!(
170            "../migrations/20260418110000_provider_repository_reconciliation.up.sql"
171        ),
172        down_sql: include_str!(
173            "../migrations/20260418110000_provider_repository_reconciliation.down.sql"
174        ),
175    },
176    DatabaseMigration {
177        version: "20260629000000",
178        name: "hub_api",
179        up_sql: include_str!("../migrations/20260629000000_hub_api.up.sql"),
180        down_sql: include_str!("../migrations/20260629000000_hub_api.down.sql"),
181    },
182];
183
184/// Returns the bundled Shardline migration list in application order.
185#[must_use]
186pub const fn bundled_database_migrations() -> &'static [DatabaseMigration] {
187    &SHARDLINE_MIGRATIONS
188}
189
190/// Applies pending Shardline migrations to an existing Postgres pool.
191///
192/// # Errors
193///
194/// Returns [`DatabaseMigrationError`] when the migration history is inconsistent or
195/// when Postgres rejects the schema updates.
196pub async fn apply_database_migrations(pool: &PgPool) -> Result<(), DatabaseMigrationError> {
197    ensure_migration_history_table(pool).await?;
198    verify_applied_migrations(pool).await?;
199
200    for migration in pending_migrations(pool).await? {
201        apply_one_migration(pool, migration).await?;
202    }
203
204    Ok(())
205}
206
207/// Executes a Shardline database-migration command against Postgres.
208///
209/// # Errors
210///
211/// Returns [`DatabaseMigrationError`] when connection setup, migration history
212/// verification, or SQL execution fails.
213pub async fn run_database_migration(
214    options: &DatabaseMigrationOptions,
215) -> Result<DatabaseMigrationReport, DatabaseMigrationError> {
216    if options.database_url().trim().is_empty() {
217        return Err(DatabaseMigrationError::EmptyDatabaseUrl);
218    }
219
220    let pool = PgPoolOptions::new()
221        .max_connections(5)
222        .connect(options.database_url())
223        .await?;
224    ensure_migration_history_table(&pool).await?;
225    verify_applied_migrations(&pool).await?;
226
227    let (applied_count, reverted_count) = match options.command() {
228        DatabaseMigrationCommand::Up { steps } => {
229            let pending = pending_migrations(&pool).await?;
230            let mut applied_count = 0_u64;
231            for migration in pending.into_iter().take(steps.unwrap_or(usize::MAX)) {
232                apply_one_migration(&pool, migration).await?;
233                applied_count = applied_count.saturating_add(1);
234            }
235            (applied_count, 0)
236        }
237        DatabaseMigrationCommand::Down { steps } => {
238            let applied = applied_migrations_in_order(&pool).await?;
239            let mut reverted_count = 0_u64;
240            for migration in applied.into_iter().rev().take(*steps) {
241                revert_one_migration(&pool, migration).await?;
242                reverted_count = reverted_count.saturating_add(1);
243            }
244            (0, reverted_count)
245        }
246        DatabaseMigrationCommand::Status => (0, 0),
247    };
248
249    let migrations = migration_status_entries(&pool).await?;
250    let applied_total_count =
251        u64::try_from(migrations.iter().filter(|entry| entry.applied).count()).unwrap_or(u64::MAX);
252    let pending_count =
253        u64::try_from(migrations.iter().filter(|entry| !entry.applied).count()).unwrap_or(u64::MAX);
254
255    Ok(DatabaseMigrationReport {
256        backend: "postgres".to_owned(),
257        command: options.command().clone(),
258        applied_count,
259        reverted_count,
260        applied_total_count,
261        pending_count,
262        migrations,
263    })
264}
265
266async fn ensure_migration_history_table(pool: &PgPool) -> Result<(), SqlxError> {
267    raw_sql(&format!(
268        "CREATE TABLE IF NOT EXISTS {MIGRATION_HISTORY_TABLE} (
269            version TEXT PRIMARY KEY,
270            name TEXT NOT NULL,
271            checksum TEXT NOT NULL,
272            applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
273        )"
274    ))
275    .execute(pool)
276    .await?;
277
278    Ok(())
279}
280
281async fn verify_applied_migrations(pool: &PgPool) -> Result<(), DatabaseMigrationError> {
282    for applied in load_applied_migrations(pool).await? {
283        let Some(migration) = migration_by_version(&applied.version) else {
284            return Err(DatabaseMigrationError::UnknownAppliedMigration(
285                applied.version,
286            ));
287        };
288        let expected_checksum = migration_checksum(migration);
289        if expected_checksum != applied.checksum {
290            return Err(DatabaseMigrationError::ChecksumMismatch {
291                version: migration.version.to_owned(),
292                expected_checksum,
293                observed_checksum: applied.checksum,
294            });
295        }
296    }
297
298    Ok(())
299}
300
301async fn pending_migrations(
302    pool: &PgPool,
303) -> Result<Vec<&'static DatabaseMigration>, DatabaseMigrationError> {
304    let applied = load_applied_migrations(pool).await?;
305    let pending = SHARDLINE_MIGRATIONS
306        .iter()
307        .filter(|migration| {
308            applied
309                .iter()
310                .all(|entry| entry.version != migration.version)
311        })
312        .collect();
313    Ok(pending)
314}
315
316async fn applied_migrations_in_order(
317    pool: &PgPool,
318) -> Result<Vec<&'static DatabaseMigration>, DatabaseMigrationError> {
319    let applied = load_applied_migrations(pool).await?;
320    let mut migrations = Vec::with_capacity(applied.len());
321    for entry in applied {
322        let Some(migration) = migration_by_version(&entry.version) else {
323            return Err(DatabaseMigrationError::UnknownAppliedMigration(
324                entry.version,
325            ));
326        };
327        migrations.push(migration);
328    }
329    migrations.sort_by_key(|migration| migration.version);
330    Ok(migrations)
331}
332
333async fn apply_one_migration(
334    pool: &PgPool,
335    migration: &'static DatabaseMigration,
336) -> Result<(), DatabaseMigrationError> {
337    let mut transaction = pool.begin().await?;
338    raw_sql(migration.up_sql).execute(&mut *transaction).await?;
339    query(&format!(
340        "INSERT INTO {MIGRATION_HISTORY_TABLE} (version, name, checksum)
341         VALUES ($1, $2, $3)
342         ON CONFLICT (version) DO NOTHING"
343    ))
344    .bind(migration.version)
345    .bind(migration.name)
346    .bind(migration_checksum(migration))
347    .execute(&mut *transaction)
348    .await?;
349    transaction.commit().await?;
350    Ok(())
351}
352
353async fn revert_one_migration(
354    pool: &PgPool,
355    migration: &'static DatabaseMigration,
356) -> Result<(), DatabaseMigrationError> {
357    let mut transaction = pool.begin().await?;
358    raw_sql(migration.down_sql)
359        .execute(&mut *transaction)
360        .await?;
361    query(&format!(
362        "DELETE FROM {MIGRATION_HISTORY_TABLE} WHERE version = $1"
363    ))
364    .bind(migration.version)
365    .execute(&mut *transaction)
366    .await?;
367    transaction.commit().await?;
368    Ok(())
369}
370
371async fn load_applied_migrations(pool: &PgPool) -> Result<Vec<AppliedMigration>, SqlxError> {
372    let rows = query(&format!(
373        "SELECT version, checksum,
374                to_char(applied_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"')
375                    AS applied_at_utc
376         FROM {MIGRATION_HISTORY_TABLE}
377         ORDER BY version"
378    ))
379    .fetch_all(pool)
380    .await?;
381
382    let mut migrations = Vec::with_capacity(rows.len());
383    for row in rows {
384        migrations.push(AppliedMigration {
385            version: row.try_get::<String, _>("version")?,
386            checksum: row.try_get::<String, _>("checksum")?,
387            applied_at_utc: row.try_get::<String, _>("applied_at_utc")?,
388        });
389    }
390
391    Ok(migrations)
392}
393
394async fn migration_status_entries(
395    pool: &PgPool,
396) -> Result<Vec<DatabaseMigrationStatusEntry>, DatabaseMigrationError> {
397    let applied = load_applied_migrations(pool).await?;
398    let mut statuses = Vec::with_capacity(SHARDLINE_MIGRATIONS.len());
399    for migration in SHARDLINE_MIGRATIONS {
400        let applied_entry = applied
401            .iter()
402            .find(|entry| entry.version == migration.version);
403        statuses.push(DatabaseMigrationStatusEntry {
404            version: migration.version.to_owned(),
405            name: migration.name.to_owned(),
406            applied: applied_entry.is_some(),
407            applied_at_utc: applied_entry.map(|entry| entry.applied_at_utc.clone()),
408        });
409    }
410
411    Ok(statuses)
412}
413
414fn migration_by_version(version: &str) -> Option<&'static DatabaseMigration> {
415    SHARDLINE_MIGRATIONS
416        .iter()
417        .find(|migration| migration.version == version)
418}
419
420fn migration_checksum(migration: &DatabaseMigration) -> String {
421    blake3::hash(migration.up_sql.as_bytes())
422        .to_hex()
423        .to_string()
424}
425
426#[cfg(test)]
427mod tests {
428    use std::{env::var as env_var, error::Error, process::id as process_id};
429
430    use sqlx::{PgPool, postgres::PgPoolOptions, query};
431    use url::Url;
432
433    use super::{
434        DatabaseMigrationCommand, DatabaseMigrationOptions, apply_database_migrations,
435        bundled_database_migrations, run_database_migration,
436    };
437
438    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
439    async fn database_migration_up_status_and_down_cover_full_lifecycle() {
440        let result = exercise_database_migration_up_status_and_down_cover_full_lifecycle().await;
441        let error = result.as_ref().err().map(ToString::to_string);
442        assert!(
443            result.is_ok(),
444            "database migration lifecycle failed: {error:?}"
445        );
446    }
447
448    #[test]
449    fn bundled_database_migrations_are_monotonic() {
450        let migrations = bundled_database_migrations();
451        assert!(!migrations.is_empty());
452        assert!(migrations.windows(2).all(|window| {
453            let Some(first) = window.first() else {
454                return false;
455            };
456            let Some(second) = window.get(1) else {
457                return false;
458            };
459            first.version < second.version
460        }));
461    }
462
463    async fn exercise_database_migration_up_status_and_down_cover_full_lifecycle()
464    -> Result<(), Box<dyn Error>> {
465        let Some(base_url) = env_var("DATABASE_URL").ok() else {
466            return Ok(());
467        };
468
469        let database_name = format!("shardline_db_migrate_{}", process_id());
470        let admin_url = database_url_for(&base_url, "postgres")?;
471        let admin_pool = PgPoolOptions::new()
472            .max_connections(1)
473            .connect(&admin_url)
474            .await?;
475        recreate_database(&admin_pool, &database_name).await?;
476
477        let database_url = database_url_for(&base_url, &database_name)?;
478        let status_before = run_database_migration(&DatabaseMigrationOptions::new(
479            database_url.clone(),
480            DatabaseMigrationCommand::Status,
481        ))
482        .await?;
483        assert_eq!(status_before.applied_count, 0);
484        assert_eq!(status_before.reverted_count, 0);
485        assert_eq!(status_before.applied_total_count, 0);
486        assert_eq!(
487            status_before.pending_count,
488            u64::try_from(bundled_database_migrations().len())?
489        );
490
491        let up = run_database_migration(&DatabaseMigrationOptions::new(
492            database_url.clone(),
493            DatabaseMigrationCommand::Up { steps: Some(2) },
494        ))
495        .await?;
496        assert_eq!(up.applied_count, 2);
497        assert_eq!(up.reverted_count, 0);
498        assert_eq!(up.applied_total_count, 2);
499
500        let up_rest = run_database_migration(&DatabaseMigrationOptions::new(
501            database_url.clone(),
502            DatabaseMigrationCommand::Up { steps: None },
503        ))
504        .await?;
505        assert_eq!(
506            up_rest.applied_count,
507            u64::try_from(bundled_database_migrations().len().saturating_sub(2))?
508        );
509        assert_eq!(up_rest.pending_count, 0);
510
511        let down = run_database_migration(&DatabaseMigrationOptions::new(
512            database_url.clone(),
513            DatabaseMigrationCommand::Down { steps: 1 },
514        ))
515        .await?;
516        assert_eq!(down.reverted_count, 1);
517        assert_eq!(down.pending_count, 1);
518
519        let pool = PgPoolOptions::new()
520            .max_connections(5)
521            .connect(&database_url)
522            .await?;
523        apply_database_migrations(&pool).await?;
524
525        let final_status = run_database_migration(&DatabaseMigrationOptions::new(
526            database_url,
527            DatabaseMigrationCommand::Status,
528        ))
529        .await?;
530        assert_eq!(final_status.pending_count, 0);
531        assert_eq!(
532            final_status.applied_total_count,
533            u64::try_from(bundled_database_migrations().len())?
534        );
535
536        Ok(())
537    }
538
539    async fn recreate_database(pool: &PgPool, database_name: &str) -> Result<(), Box<dyn Error>> {
540        query(&format!("DROP DATABASE IF EXISTS {database_name}"))
541            .execute(pool)
542            .await?;
543        query(&format!("CREATE DATABASE {database_name}"))
544            .execute(pool)
545            .await?;
546        Ok(())
547    }
548
549    fn database_url_for(base_url: &str, database_name: &str) -> Result<String, Box<dyn Error>> {
550        let mut url = Url::parse(base_url)?;
551        url.set_path(database_name);
552        Ok(url.to_string())
553    }
554}
shardline_server/database_migration.rs

shardline_server/
database_migration.rs
