huskies: merge 1072

server/src/db/shadow_write.rs

@@ -11,10 +11,23 @@ use crate::slog;
 use sqlx::SqlitePool;
 use sqlx::sqlite::SqliteConnectOptions;
 use std::collections::HashMap;
+use std::collections::HashSet;
 use std::path::Path;
 use std::sync::OnceLock;
 use tokio::sync::mpsc;
 
+/// One migration row in the live database that is not in the compiled-in set.
+///
+/// Returned by [`check_schema_drift`] for each unknown migration.
+pub struct UnknownMigration {
+    /// sqlx migration version number (derived from the filename timestamp).
+    pub version: i64,
+    /// Human-readable description from the migration filename.
+    pub description: String,
+    /// When the migration was applied, as stored in `_sqlx_migrations.installed_on`.
+    pub installed_on: String,
+}
+
 /// The process-global SQLite pool, set once by [`init`].
 ///
 /// Other modules call [`get_shared_pool`] to access the pool without needing
@@ -133,3 +146,31 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> {
     let _ = PIPELINE_DB.set(PipelineDb { tx });
     Ok(())
 }
+
+/// Compare the live `_sqlx_migrations` table against the compiled-in migration
+/// set and return any rows whose version is not known to this binary.
+///
+/// A non-empty result means the database was previously opened by a newer
+/// binary that applied additional migrations.  The server must refuse to start
+/// in that state because the schema may contain tables or columns that this
+/// binary does not understand.
+pub async fn check_schema_drift(pool: &SqlitePool) -> Vec<UnknownMigration> {
+    let migrator = sqlx::migrate!("./migrations");
+    let known: HashSet<i64> = migrator.migrations.iter().map(|m| m.version).collect();
+
+    let rows: Vec<(i64, String, String)> = sqlx::query_as(
+        "SELECT version, description, installed_on FROM _sqlx_migrations ORDER BY version",
+    )
+    .fetch_all(pool)
+    .await
+    .unwrap_or_default();
+
+    rows.into_iter()
+        .filter(|(v, _, _)| !known.contains(v))
+        .map(|(version, description, installed_on)| UnknownMigration {
+            version,
+            description,
+            installed_on,
+        })
+        .collect()
+}