move migrations & test-database into subcrate, migrate service-config-in-db

Cargo.toml

@@ -93,12 +93,10 @@ opentelemetry_sdk = { workspace = true }
 rustwide = { version = "0.21.0", features = ["unstable-toolchain-ci", "unstable"] }
 hostname = "0.4.0"
 base64 = "0.22"
-strum = { workspace = true }
 lol_html = "2.0.0"
 font-awesome-as-a-crate = { path = "crates/lib/font-awesome-as-a-crate" }
 getrandom = "0.3.1"
 itertools = { workspace = true }
-hex = "0.4.3"
 derive_more = { workspace = true }
 sysinfo = { version = "0.37.2", default-features = false, features = ["system"] }
 derive_builder = "0.20.2"

crates/lib/docs_rs_database/Cargo.toml

@@ -2,17 +2,31 @@
 name = "docs_rs_database"
 version = "0.1.0"
 edition = "2024"
+build = "build.rs"
 
 [dependencies]
 anyhow = { workspace = true }
 docs_rs_env_vars = { path = "../docs_rs_env_vars" }
 docs_rs_opentelemetry = { path = "../docs_rs_opentelemetry" }
 futures-util = { workspace = true }
+hex = "0.4.3"
 opentelemetry = { workspace = true }
+rand = { workspace = true, optional = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
 sqlx = { workspace = true }
+strum = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
 
+[dev-dependencies]
+rand = { workspace = true }
+docs_rs_opentelemetry = { path = "../docs_rs_opentelemetry", features = ["testing"] }
+test-case = { workspace = true }
+
 [features]
-testing = []
+testing = [
+    "dep:rand",
+    "docs_rs_opentelemetry/testing",
+]

crates/lib/docs_rs_database/build.rs

@@ -0,0 +1,4 @@
+fn main() {
+    // trigger recompilation when a new migration is added
+    println!("cargo:rerun-if-changed=migrations");
+}

crates/lib/docs_rs_database/src/lib.rs

@@ -1,8 +1,13 @@
 mod config;
 mod errors;
 mod metrics;
+mod migrations;
 mod pool;
+pub mod service_config;
+#[cfg(any(test, feature = "testing"))]
+pub mod testing;
 
 pub use config::Config;
 pub use errors::PoolError;
+pub use migrations::migrate;
 pub use pool::{AsyncPoolClient, Pool};

crates/lib/docs_rs_database/src/migrations.rs

@@ -0,0 +1,60 @@
+use anyhow::Result;
+use sqlx::migrate::{Migrate, Migrator};
+
+pub static MIGRATOR: Migrator = sqlx::migrate!(); // defaults to "./migrations"
+
+pub async fn migrate(conn: &mut sqlx::PgConnection, target: Option<i64>) -> Result<()> {
+    conn.ensure_migrations_table().await?;
+
+    // `database_versions` is the table that tracked the old `schemamama` migrations.
+    // If we find the table, and it contains records, we insert a fake record
+    // into the `_sqlx_migrations` table so the big initial migration isn't executed.
+    if sqlx::query(
+        "SELECT table_name
+         FROM information_schema.tables
+         WHERE table_schema = 'public' AND table_name = 'database_versions'",
+    )
+    .fetch_optional(&mut *conn)
+    .await?
+    .is_some()
+    {
+        let max_version: Option<i64> =
+            sqlx::query_scalar("SELECT max(version) FROM database_versions")
+                .fetch_one(&mut *conn)
+                .await?;
+
+        if max_version != Some(39) {
+            anyhow::bail!(
+                "database_versions table has unexpected version: {:?}",
+                max_version
+            );
+        }
+
+        sqlx::query(
+            "INSERT INTO _sqlx_migrations ( version, description, success, checksum, execution_time )
+             VALUES ( $1, $2, TRUE, $3, -1 )",
+        )
+        // the next two parameters relate to the filename of the initial migration file
+        .bind(20231021111635i64)
+        .bind("initial")
+        // this is the hash of the initial migration file, as sqlx requires it.
+        // if the initial migration file changes, this has to be updated with the new value,
+        // easiest to get from the `_sqlx_migrations` table when the migration was normally
+        // executed.
+        .bind(hex::decode("df802e0ec416063caadd1c06b13348cd885583c44962998886b929d5fe6ef3b70575d5101c5eb31daa989721df08d806").unwrap())
+        .execute(&mut *conn)
+        .await?;
+
+        sqlx::query("DROP TABLE database_versions")
+            .execute(&mut *conn)
+            .await?;
+    }
+
+    // when we find records
+    if let Some(target) = target {
+        MIGRATOR.undo(conn, target).await?;
+    } else {
+        MIGRATOR.run(conn).await?;
+    }
+    Ok(())
+}

crates/lib/docs_rs_database/src/pool.rs

@@ -30,7 +30,7 @@ impl Pool {
         Self::new_inner(config, DEFAULT_SCHEMA, otel_meter_provider).await
     }
 
-    #[cfg(feature = "testing")]
+    #[cfg(any(test, feature = "testing"))]
     pub async fn new_with_schema(
         config: &Config,
         schema: &str,

crates/lib/docs_rs_database/src/service_config.rs

@@ -0,0 +1,109 @@
+use anyhow::Result;
+use serde::{Serialize, de::DeserializeOwned};
+
+#[derive(strum::IntoStaticStr)]
+#[strum(serialize_all = "snake_case")]
+pub enum ConfigName {
+    RustcVersion,
+    LastSeenIndexReference,
+    QueueLocked,
+    Toolchain,
+}
+
+pub async fn set_config(
+    conn: &mut sqlx::PgConnection,
+    name: ConfigName,
+    value: impl Serialize,
+) -> anyhow::Result<()> {
+    let name: &'static str = name.into();
+    sqlx::query!(
+        "INSERT INTO config (name, value)
+         VALUES ($1, $2)
+         ON CONFLICT (name) DO UPDATE SET value = $2;",
+        name,
+        &serde_json::to_value(value)?,
+    )
+    .execute(conn)
+    .await?;
+    Ok(())
+}
+
+pub async fn get_config<T>(conn: &mut sqlx::PgConnection, name: ConfigName) -> Result<Option<T>>
+where
+    T: DeserializeOwned,
+{
+    let name: &'static str = name.into();
+    Ok(
+        match sqlx::query!("SELECT value FROM config WHERE name = $1;", name)
+            .fetch_optional(conn)
+            .await?
+        {
+            Some(row) => serde_json::from_value(row.value)?,
+            None => None,
+        },
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{Config, testing::TestDatabase};
+    use docs_rs_opentelemetry::testing::TestMetrics;
+    use serde_json::Value;
+    use test_case::test_case;
+
+    #[test_case(ConfigName::RustcVersion, "rustc_version")]
+    #[test_case(ConfigName::QueueLocked, "queue_locked")]
+    #[test_case(ConfigName::LastSeenIndexReference, "last_seen_index_reference")]
+    fn test_configname_variants(variant: ConfigName, expected: &'static str) {
+        let name: &'static str = variant.into();
+        assert_eq!(name, expected);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_get_config_empty() -> anyhow::Result<()> {
+        let test_metrics = TestMetrics::new();
+        let db = TestDatabase::new(&Config::test_config()?, test_metrics.provider()).await?;
+
+        let mut conn = db.async_conn().await;
+        sqlx::query!("DELETE FROM config")
+            .execute(&mut *conn)
+            .await?;
+
+        assert!(
+            get_config::<String>(&mut conn, ConfigName::RustcVersion)
+                .await?
+                .is_none()
+        );
+        Ok(())
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_set_and_get_config_() -> anyhow::Result<()> {
+        let test_metrics = TestMetrics::new();
+        let db = TestDatabase::new(&Config::test_config()?, test_metrics.provider()).await?;
+
+        let mut conn = db.async_conn().await;
+        sqlx::query!("DELETE FROM config")
+            .execute(&mut *conn)
+            .await?;
+
+        assert!(
+            get_config::<String>(&mut conn, ConfigName::RustcVersion)
+                .await?
+                .is_none()
+        );
+
+        set_config(
+            &mut conn,
+            ConfigName::RustcVersion,
+            Value::String("some value".into()),
+        )
+        .await?;
+        assert_eq!(
+            get_config(&mut conn, ConfigName::RustcVersion).await?,
+            Some("some value".to_string())
+        );
+        Ok(())
+    }
+}

crates/lib/docs_rs_database/src/testing/mod.rs

@@ -0,0 +1,3 @@
+mod test_env;
+
+pub use test_env::TestDatabase;