v0.3.3-pre.002

crates/ksp-store-postgres-lib/src/migration.rs

@@ -1,12 +1,15 @@
 // file: crates/ksp-store-postgres-lib/src/migration.rs
-// version: 2
+// version: 3
 
 use sha2::Digest; // rust-rules: trait-import
 
 const ADVISORY_LOCK_KEY: i64 = 0x4b53_5053_544f_5245;
-const BOOTSTRAP_MIGRATION_NAME: &str = "bootstrap";
-const BOOTSTRAP_MIGRATION_SQL: &str = include_str!("../migrations/V000__bootstrap.sql");
-const BOOTSTRAP_MIGRATION_VERSION: i64 = 0;
+const EMBEDDED_MIGRATIONS: &[EmbeddedMigration] = &[EmbeddedMigration {
+    hook: MigrationHook::None,
+    name: "bootstrap",
+    sql: include_str!("../migrations/V000__bootstrap.sql"),
+    version: 0,
+}];
 const HEX_LOWER: &[u8; 16] = b"0123456789abcdef";
 const HISTORY_INSERT_SQL: &str = "INSERT INTO ksp_store_schema_migrations (version, name, checksum, applied_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP)";
 const HISTORY_LOAD_SQL: &str = "SELECT version, name, checksum FROM ksp_store_schema_migrations ORDER BY version";
@@ -40,20 +43,44 @@ struct AppliedMigration {
     version: i64,
 }
 
+#[derive(Clone, Copy)]
+struct EmbeddedMigration {
+    hook: MigrationHook,
+    name: &'static str,
+    sql: &'static str,
+    version: i64,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum MigrationHook {
+    None,
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum MigrationHookContext {
+    AppliedNow,
+    Existing,
+}
+
 /// Returns the latest migration version embedded by this backend runtime.
 #[must_use]
 pub(crate) const fn current_migration_version() -> i64 {
-    return BOOTSTRAP_MIGRATION_VERSION;
+    return EMBEDDED_MIGRATIONS[EMBEDDED_MIGRATIONS.len() - 1].version;
 }
 
 /// Runs the private bounded PostgreSQL schema bootstrap on one dedicated pooled client.
 pub(crate) async fn bootstrap(
     client: &mut deadpool_postgres::Client,
+    network: &ksp_store_api::RawNetworkId,
     auto_migrate: bool,
     migration_timeout: std::time::Duration,
     migration_lock_timeout: std::time::Duration,
 ) -> std::result::Result<(), crate::PostgresBackendError> {
-    let bounded = tokio::time::timeout(migration_timeout, bootstrap_inner(client, auto_migrate, migration_timeout, migration_lock_timeout)).await;
+    let registry_result = validate_embedded_registry(EMBEDDED_MIGRATIONS);
+    if let std::result::Result::Err(error) = registry_result {
+        return std::result::Result::Err(error);
+    }
+    let bounded = tokio::time::timeout(migration_timeout, bootstrap_inner(client, network, auto_migrate, migration_timeout, migration_lock_timeout)).await;
     return match bounded {
         std::result::Result::Ok(result) => result,
         std::result::Result::Err(_) => {
@@ -64,6 +91,7 @@ pub(crate) async fn bootstrap(
 
 async fn bootstrap_inner(
     client: &mut deadpool_postgres::Client,
+    network: &ksp_store_api::RawNetworkId,
     auto_migrate: bool,
     migration_timeout: std::time::Duration,
     migration_lock_timeout: std::time::Duration,
@@ -88,27 +116,7 @@ async fn bootstrap_inner(
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(error) => return std::result::Result::Err(error),
     };
-    let checksum = bootstrap_checksum();
-    if !metadata_exists {
-        if !auto_migrate {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_pending"));
-        }
-        let create_result = transaction.batch_execute(BOOTSTRAP_MIGRATION_SQL).await;
-        if create_result.is_err() {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_create"));
-        }
-        let shape_result = verify_metadata_shape(&transaction).await;
-        if let std::result::Result::Err(error) = shape_result {
-            return std::result::Result::Err(error);
-        }
-        let insert_result = transaction.execute(HISTORY_INSERT_SQL, &[&BOOTSTRAP_MIGRATION_VERSION, &BOOTSTRAP_MIGRATION_NAME, &checksum]).await;
-        match insert_result {
-            std::result::Result::Ok(1) => {},
-            std::result::Result::Ok(_) | std::result::Result::Err(_) => {
-                return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_insert"));
-            },
-        }
-    } else {
+    let next_index = if metadata_exists {
         let shape_result = verify_metadata_shape(&transaction).await;
         if let std::result::Result::Err(error) = shape_result {
             return std::result::Result::Err(error);
@@ -118,10 +126,24 @@ async fn bootstrap_inner(
             std::result::Result::Ok(value) => value,
             std::result::Result::Err(error) => return std::result::Result::Err(error),
         };
-        let validation_result = validate_history(history.as_slice(), checksum.as_str());
-        if let std::result::Result::Err(error) = validation_result {
-            return std::result::Result::Err(error);
+        let validation_result = validate_history(history.as_slice(), EMBEDDED_MIGRATIONS);
+        match validation_result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
         }
+    } else {
+        0
+    };
+    let existing_hook_result = run_applied_migration_hooks(&transaction, network, next_index).await;
+    if let std::result::Result::Err(error) = existing_hook_result {
+        return std::result::Result::Err(error);
+    }
+    if next_index < EMBEDDED_MIGRATIONS.len() && !auto_migrate {
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_pending"));
+    }
+    let apply_result = apply_pending_migrations(&transaction, network, next_index).await;
+    if let std::result::Result::Err(error) = apply_result {
+        return std::result::Result::Err(error);
     }
     let commit_result = transaction.commit().await;
     return match commit_result {
@@ -171,20 +193,75 @@ async fn acquire_advisory_lock(
     }
 }
 
-async fn set_statement_timeout(
+async fn apply_migration(
     transaction: &deadpool_postgres::Transaction<'_>,
-    timeout: std::time::Duration,
+    network: &ksp_store_api::RawNetworkId,
+    migration: &EmbeddedMigration,
 ) -> std::result::Result<(), crate::PostgresBackendError> {
-    let timeout_value = format!("{}ms", timeout.as_millis());
-    let result = transaction.query_one(SET_STATEMENT_TIMEOUT_SQL, &[&timeout_value]).await;
-    return match result {
-        std::result::Result::Ok(_) => std::result::Result::Ok(()),
-        std::result::Result::Err(_) => {
-            std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "statement_timeout"))
+    let execute_result = transaction.batch_execute(migration.sql).await;
+    if execute_result.is_err() {
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_apply"));
+    }
+    let shape_result = verify_metadata_shape(transaction).await;
+    if let std::result::Result::Err(error) = shape_result {
+        return std::result::Result::Err(error);
+    }
+    let hook_result = run_migration_hook(transaction, network, migration.hook, MigrationHookContext::AppliedNow).await;
+    if let std::result::Result::Err(error) = hook_result {
+        return std::result::Result::Err(error);
+    }
+    let checksum = migration_checksum(migration.sql);
+    let insert_result = transaction.execute(HISTORY_INSERT_SQL, &[&migration.version, &migration.name, &checksum]).await;
+    return match insert_result {
+        std::result::Result::Ok(1) => std::result::Result::Ok(()),
+        std::result::Result::Ok(_) | std::result::Result::Err(_) => {
+            std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_insert"))
         },
     };
 }
 
+async fn apply_pending_migrations(
+    transaction: &deadpool_postgres::Transaction<'_>,
+    network: &ksp_store_api::RawNetworkId,
+    next_index: usize,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    let mut index = next_index;
+    while index < EMBEDDED_MIGRATIONS.len() {
+        let migration = &EMBEDDED_MIGRATIONS[index];
+        let result = apply_migration(transaction, network, migration).await;
+        if let std::result::Result::Err(error) = result {
+            return std::result::Result::Err(error);
+        }
+        index += 1;
+    }
+    return std::result::Result::Ok(());
+}
+
+async fn load_history(transaction: &deadpool_postgres::Transaction<'_>) -> std::result::Result<std::vec::Vec<AppliedMigration>, crate::PostgresBackendError> {
+    let result = transaction.query(HISTORY_LOAD_SQL, &[]).await;
+    let rows = match result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_load"));
+        },
+    };
+    let mut history = std::vec::Vec::with_capacity(rows.len());
+    for row in rows {
+        let version_result = row.try_get::<usize, i64>(0);
+        let name_result = row.try_get::<usize, std::string::String>(1);
+        let checksum_result = row.try_get::<usize, std::string::String>(2);
+        match (version_result, name_result, checksum_result) {
+            (std::result::Result::Ok(version), std::result::Result::Ok(name), std::result::Result::Ok(checksum)) => {
+                history.push(AppliedMigration { checksum, name, version });
+            },
+            _ => {
+                return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_decode"));
+            },
+        }
+    }
+    return std::result::Result::Ok(history);
+}
+
 async fn metadata_exists(transaction: &deadpool_postgres::Transaction<'_>) -> std::result::Result<bool, crate::PostgresBackendError> {
     let result = transaction.query_one(METADATA_EXISTS_SQL, &[]).await;
     let row = match result {
@@ -201,6 +278,104 @@ async fn metadata_exists(transaction: &deadpool_postgres::Transaction<'_>) -> st
     };
 }
 
+fn migration_checksum(sql: &str) -> std::string::String {
+    let mut hasher = sha2::Sha256::new();
+    hasher.update(sql.as_bytes());
+    let digest = hasher.finalize();
+    let bytes = digest.as_slice();
+    let mut encoded = std::string::String::with_capacity(bytes.len() * 2);
+    for byte in bytes {
+        let value = *byte;
+        encoded.push(char::from(HEX_LOWER[(value >> 4) as usize]));
+        encoded.push(char::from(HEX_LOWER[(value & 0x0f) as usize]));
+    }
+    return encoded;
+}
+
+async fn run_applied_migration_hooks(
+    transaction: &deadpool_postgres::Transaction<'_>,
+    network: &ksp_store_api::RawNetworkId,
+    applied_count: usize,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    let mut index = 0_usize;
+    while index < applied_count {
+        let migration = &EMBEDDED_MIGRATIONS[index];
+        let result = run_migration_hook(transaction, network, migration.hook, MigrationHookContext::Existing).await;
+        if let std::result::Result::Err(error) = result {
+            return std::result::Result::Err(error);
+        }
+        index += 1;
+    }
+    return std::result::Result::Ok(());
+}
+
+async fn run_migration_hook(
+    _transaction: &deadpool_postgres::Transaction<'_>,
+    _network: &ksp_store_api::RawNetworkId,
+    hook: MigrationHook,
+    _context: MigrationHookContext,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    return match hook {
+        MigrationHook::None => std::result::Result::Ok(()),
+    };
+}
+
+async fn set_statement_timeout(
+    transaction: &deadpool_postgres::Transaction<'_>,
+    timeout: std::time::Duration,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    let timeout_value = format!("{}ms", timeout.as_millis());
+    let result = transaction.query_one(SET_STATEMENT_TIMEOUT_SQL, &[&timeout_value]).await;
+    return match result {
+        std::result::Result::Ok(_) => std::result::Result::Ok(()),
+        std::result::Result::Err(_) => {
+            std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "statement_timeout"))
+        },
+    };
+}
+
+fn validate_embedded_registry(migrations: &[EmbeddedMigration]) -> std::result::Result<(), crate::PostgresBackendError> {
+    if migrations.is_empty() {
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "registry_empty"));
+    }
+    let mut expected_version = 0_i64;
+    for migration in migrations {
+        if migration.version != expected_version || migration.name.is_empty() || migration.sql.is_empty() {
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "registry_invalid"));
+        }
+        expected_version = match expected_version.checked_add(1) {
+            std::option::Option::Some(value) => value,
+            std::option::Option::None => {
+                return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "registry_invalid"));
+            },
+        };
+    }
+    return std::result::Result::Ok(());
+}
+
+fn validate_history(history: &[AppliedMigration], migrations: &[EmbeddedMigration]) -> std::result::Result<usize, crate::PostgresBackendError> {
+    if history.is_empty() {
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_missing"));
+    }
+    let latest_version = migrations[migrations.len() - 1].version;
+    let mut index = 0_usize;
+    for applied in history {
+        if applied.version > latest_version {
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::SchemaNewer, "history_newer"));
+        }
+        if index >= migrations.len() {
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::SchemaNewer, "history_newer"));
+        }
+        let expected = &migrations[index];
+        let expected_checksum = migration_checksum(expected.sql);
+        if applied.version != expected.version || applied.name != expected.name || applied.checksum != expected_checksum {
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_diverged"));
+        }
+        index += 1;
+    }
+    return std::result::Result::Ok(index);
+}
+
 async fn verify_metadata_shape(transaction: &deadpool_postgres::Transaction<'_>) -> std::result::Result<(), crate::PostgresBackendError> {
     let result = transaction.query(METADATA_SHAPE_SQL, &[]).await;
     let rows = match result {
@@ -209,23 +384,21 @@ async fn verify_metadata_shape(transaction: &deadpool_postgres::Transaction<'_>)
             return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_shape"));
         },
     };
-    let expected = [("version", "bigint", "NO"), ("name", "text", "NO"), ("checksum", "text", "NO"), ("applied_at", "timestamp with time zone", "NO")];
-    let mut found = [false; 4];
+    const REQUIRED: [(&str, &str, &str); 4] =
+        [("version", "bigint", "NO"), ("name", "text", "NO"), ("checksum", "text", "NO"), ("applied_at", "timestamp with time zone", "NO")];
+    let mut found = [false; REQUIRED.len()];
     for row in rows {
-        let column_result = row.try_get::<usize, std::string::String>(0);
+        let name_result = row.try_get::<usize, std::string::String>(0);
         let data_type_result = row.try_get::<usize, std::string::String>(1);
         let nullable_result = row.try_get::<usize, std::string::String>(2);
-        let (column, data_type, nullable) = match (column_result, data_type_result, nullable_result) {
-            (std::result::Result::Ok(column), std::result::Result::Ok(data_type), std::result::Result::Ok(nullable)) => (column, data_type, nullable),
+        let (name, data_type, nullable) = match (name_result, data_type_result, nullable_result) {
+            (std::result::Result::Ok(name), std::result::Result::Ok(data_type), std::result::Result::Ok(nullable)) => (name, data_type, nullable),
             _ => {
                 return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_shape_decode"));
             },
         };
-        for (index, expected_row) in expected.iter().enumerate() {
-            if column == expected_row.0 {
-                if found[index] || data_type != expected_row.1 || nullable != expected_row.2 {
-                    return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "metadata_shape"));
-                }
+        for (index, required) in REQUIRED.iter().enumerate() {
+            if name == required.0 && data_type == required.1 && nullable == required.2 {
                 found[index] = true;
                 break;
             }
@@ -254,65 +427,6 @@ async fn verify_metadata_shape(transaction: &deadpool_postgres::Transaction<'_>)
     };
 }
 
-async fn load_history(transaction: &deadpool_postgres::Transaction<'_>) -> std::result::Result<std::vec::Vec<AppliedMigration>, crate::PostgresBackendError> {
-    let result = transaction.query(HISTORY_LOAD_SQL, &[]).await;
-    let rows = match result {
-        std::result::Result::Ok(value) => value,
-        std::result::Result::Err(_) => {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_load"));
-        },
-    };
-    let mut history = std::vec::Vec::with_capacity(rows.len());
-    for row in rows {
-        let version_result = row.try_get::<usize, i64>(0);
-        let name_result = row.try_get::<usize, std::string::String>(1);
-        let checksum_result = row.try_get::<usize, std::string::String>(2);
-        match (version_result, name_result, checksum_result) {
-            (std::result::Result::Ok(version), std::result::Result::Ok(name), std::result::Result::Ok(checksum)) => {
-                history.push(AppliedMigration { checksum, name, version });
-            },
-            _ => {
-                return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_decode"));
-            },
-        }
-    }
-    return std::result::Result::Ok(history);
-}
-
-fn validate_history(history: &[AppliedMigration], expected_checksum: &str) -> std::result::Result<(), crate::PostgresBackendError> {
-    let mut sentinel_found = false;
-    for applied in history {
-        if applied.version > BOOTSTRAP_MIGRATION_VERSION {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::SchemaNewer, "history_newer"));
-        }
-        if applied.version < BOOTSTRAP_MIGRATION_VERSION {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_unknown"));
-        }
-        if applied.name != BOOTSTRAP_MIGRATION_NAME || applied.checksum != expected_checksum {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_diverged"));
-        }
-        sentinel_found = true;
-    }
-    if !sentinel_found {
-        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_missing"));
-    }
-    return std::result::Result::Ok(());
-}
-
-fn bootstrap_checksum() -> std::string::String {
-    let mut hasher = sha2::Sha256::new();
-    hasher.update(BOOTSTRAP_MIGRATION_SQL.as_bytes());
-    let digest = hasher.finalize();
-    let bytes = digest.as_slice();
-    let mut encoded = std::string::String::with_capacity(bytes.len() * 2);
-    for byte in bytes {
-        let value = *byte;
-        encoded.push(char::from(HEX_LOWER[(value >> 4) as usize]));
-        encoded.push(char::from(HEX_LOWER[(value & 0x0f) as usize]));
-    }
-    return encoded;
-}
-
 #[cfg(test)]
 #[path = "../unit_tests/migration.rs"]
 mod tests;

crates/ksp-store-postgres-lib/src/runtime.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-store-postgres-lib/src/runtime.rs
-// version: 3
+// version: 4
 
 const APPLICATION_NAME: &str = "ksp-store";
 const MAX_CONNECTION_URI_BYTES: usize = 4_096;
@@ -263,7 +263,8 @@ impl PostgresBackend {
             tls_mode = settings.tls_mode().code(),
             "PostgreSQL Store backend established initial physical connection"
         );
-        let bootstrap_result = crate::bootstrap(&mut client, settings.auto_migrate, settings.migration_timeout, settings.migration_lock_timeout).await;
+        let bootstrap_result =
+            crate::bootstrap(&mut client, settings.network(), settings.auto_migrate, settings.migration_timeout, settings.migration_lock_timeout).await;
         if let std::result::Result::Err(error) = bootstrap_result {
             return std::result::Result::Err(error);
         }