sable_history: Add support for running migrations on startup

configs/history_server.conf

@@ -12,6 +12,8 @@
 
     "server": {
         "database": "postgres:///sable_history?host=/var/run/postgresql/",
+        // run migrations on startup
+        "auto_run_migrations": true,
     },
 
     "event_log": {

sable_history/Cargo.toml

@@ -22,7 +22,9 @@ tracing = "0.1"
 anyhow = "1.0"
 clap = { version = "4.5", features = [ "derive" ] }
 chrono = "0.4"
+itertools = "0.10"
 uuid = { version = "1.9.1", features = ["v7", "fast-rng", "serde"] }
 
 diesel = { version = "2.2", features = [ "postgres", "chrono", "uuid" ] }
-diesel-async = { version = "0.5", features = [ "postgres" ] }
+diesel-async = { version = "0.5", features = [ "postgres", "tokio", "async-connection-wrapper" ] }
+diesel_migrations = "2.2.0"

sable_history/build.rs

@@ -1,3 +1,6 @@
 fn main() {
+    // https://docs.rs/diesel_migrations/2.2.0/diesel_migrations/macro.embed_migrations.html#automatic-rebuilds
+    println!("cargo:rerun-if-changed=migrations/");
+
     built::write_built_file().expect("Failed to acquire build-time information");
 }

sable_history/src/server/mod.rs

@@ -1,20 +1,27 @@
 use std::convert::Infallible;
+use std::sync::Arc;
 
 use anyhow::Context;
-use sable_network::prelude::*;
-use sable_server::ServerType;
+use diesel::migration::MigrationSource;
+use diesel::prelude::*;
+use diesel_async::async_connection_wrapper::AsyncConnectionWrapper;
+use diesel_async::{AsyncConnection, AsyncPgConnection};
+use diesel_migrations::{embed_migrations, EmbeddedMigrations, MigrationHarness};
+use itertools::Itertools;
 use serde::Deserialize;
 use tokio::sync::{mpsc::UnboundedReceiver, Mutex};
 
-use std::sync::Arc;
-
-use diesel_async::{AsyncConnection, AsyncPgConnection};
+use sable_network::prelude::*;
+use sable_server::ServerType;
 
 mod update_handler;
 
+pub const MIGRATIONS: EmbeddedMigrations = embed_migrations!("./migrations");
+
 #[derive(Debug, Clone, Deserialize)]
 pub struct HistoryServerConfig {
     pub database: String,
+    pub auto_run_migrations: bool,
 }
 
 pub struct HistoryServer {
@@ -44,6 +51,38 @@ impl ServerType for HistoryServer {
             sable_network::rpc::NetworkHistoryUpdate,
         >,
     ) -> anyhow::Result<Self> {
+        let database = config.database.clone();
+        if config.auto_run_migrations {
+            tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
+                // run_pending_migrations only support sync connections
+                let mut conn = AsyncConnectionWrapper::<AsyncPgConnection>::establish(&database)
+                    .context("Couldn't connect to database")?;
+                tracing::info!("Running database migrations");
+                tracing::trace!(
+                    "Required migrations: {}",
+                    MIGRATIONS
+                        .migrations()
+                        .map_err(|e| anyhow::anyhow!("Couldn't get migrations: {e}"))?
+                        .iter()
+                        .map(diesel::migration::Migration::<diesel::pg::Pg>::name)
+                        .join(", ")
+                );
+                let migrations = conn
+                    .run_pending_migrations(MIGRATIONS)
+                    .map_err(|e| anyhow::anyhow!("Database migrations failed: {e}"))?;
+                if migrations.is_empty() {
+                    tracing::info!("No database migrations to run");
+                } else {
+                    tracing::info!(
+                        "Applied database migrations: {}",
+                        migrations.iter().map(ToString::to_string).join(", ")
+                    )
+                }
+                Ok(())
+            })
+            .await
+            .context("Couldn't join migration task")??;
+        }
         Ok(Self {
             node,
             history_receiver: Mutex::new(history_receiver),