v0.3.3-pre.003-fix.001

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

@@ -1,15 +1,22 @@
 // file: crates/ksp-store-postgres-lib/src/migration.rs
-// version: 4
+// version: 5
 
 use sha2::Digest; // rust-rules: trait-import
 
 const ADVISORY_LOCK_KEY: i64 = 0x4b53_5053_544f_5245;
 const EMBEDDED_MIGRATIONS: &[EmbeddedMigration] = &[
-    EmbeddedMigration { hook: MigrationHook::None, name: "bootstrap", sql: include_str!("../migrations/V000__bootstrap.sql"), version: 0 },
     EmbeddedMigration {
+        checksum: MigrationChecksum::LegacySql(include_str!("../migrations/v000_bootstrap/tables/001_ksp_store_schema_migrations.sql")),
+        hook: MigrationHook::None,
+        name: "bootstrap",
+        resources: crate::V000_RESOURCES,
+        version: 0,
+    },
+    EmbeddedMigration {
+        checksum: MigrationChecksum::Resources,
         hook: MigrationHook::StoreIdentity,
         name: "raw_transaction",
-        sql: include_str!("../migrations/V001__raw_transaction.sql"),
+        resources: crate::V001_RESOURCES,
         version: 1,
     },
 ];
@@ -25,21 +32,6 @@ const METADATA_EXISTS_SQL: &str = r#"SELECT EXISTS (
       AND table_name = 'ksp_store_schema_migrations'
       AND table_type = 'BASE TABLE'
 )"#;
-const METADATA_PRIMARY_KEY_SQL: &str = r#"SELECT COUNT(*)::BIGINT,
-       COUNT(*) FILTER (WHERE kcu.column_name = 'version')::BIGINT
-FROM information_schema.table_constraints tc
-JOIN information_schema.key_column_usage kcu
-  ON tc.constraint_catalog = kcu.constraint_catalog
- AND tc.constraint_schema = kcu.constraint_schema
- AND tc.constraint_name = kcu.constraint_name
-WHERE tc.table_schema = current_schema()
-  AND tc.table_name = 'ksp_store_schema_migrations'
-  AND tc.constraint_type = 'PRIMARY KEY'"#;
-const METADATA_SHAPE_SQL: &str = r#"SELECT column_name, data_type, is_nullable
-FROM information_schema.columns
-WHERE table_schema = current_schema()
-  AND table_name = 'ksp_store_schema_migrations'
-ORDER BY ordinal_position"#;
 const SET_STATEMENT_TIMEOUT_SQL: &str = "SELECT set_config('statement_timeout', $1, true)";
 
 struct AppliedMigration {
@@ -48,11 +40,18 @@ struct AppliedMigration {
     version: i64,
 }
 
+#[derive(Clone, Copy)]
+enum MigrationChecksum {
+    LegacySql(&'static str),
+    Resources,
+}
+
 #[derive(Clone, Copy)]
 struct EmbeddedMigration {
+    checksum: MigrationChecksum,
     hook: MigrationHook,
     name: &'static str,
-    sql: &'static str,
+    resources: &'static [crate::SchemaResource],
     version: i64,
 }
 
@@ -68,6 +67,12 @@ enum MigrationHookContext {
     Existing,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum SchemaMutationMode {
+    Create,
+    Update,
+}
+
 /// Returns the latest migration version embedded by this backend runtime.
 #[must_use]
 pub(crate) const fn current_migration_version() -> i64 {
@@ -78,7 +83,8 @@ pub(crate) const fn current_migration_version() -> i64 {
 pub(crate) async fn bootstrap(
     client: &mut deadpool_postgres::Client,
     network: &ksp_store_api::RawNetworkId,
-    auto_migrate: bool,
+    schema_autocreate: bool,
+    schema_autoupdate: bool,
     migration_timeout: std::time::Duration,
     migration_lock_timeout: std::time::Duration,
 ) -> std::result::Result<(), crate::PostgresBackendError> {
@@ -88,7 +94,7 @@ pub(crate) async fn bootstrap(
     }
     let bounded = tokio::time::timeout(
         migration_timeout,
-        bootstrap_inner(client, network, auto_migrate, migration_timeout, migration_lock_timeout),
+        bootstrap_inner(client, network, schema_autocreate, schema_autoupdate, migration_timeout, migration_lock_timeout),
     )
     .await;
     return match bounded {
@@ -102,7 +108,8 @@ pub(crate) async fn bootstrap(
 async fn bootstrap_inner(
     client: &mut deadpool_postgres::Client,
     network: &ksp_store_api::RawNetworkId,
-    auto_migrate: bool,
+    schema_autocreate: bool,
+    schema_autoupdate: bool,
     migration_timeout: std::time::Duration,
     migration_lock_timeout: std::time::Duration,
 ) -> std::result::Result<(), crate::PostgresBackendError> {
@@ -126,10 +133,15 @@ async fn bootstrap_inner(
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(error) => return std::result::Result::Err(error),
     };
-    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);
+    let (next_index, mutation_mode) = if metadata_exists {
+        let metadata_result = crate::inspect_resource(&transaction, &crate::V000_RESOURCES[0]).await;
+        match metadata_result {
+            std::result::Result::Ok(crate::SchemaResourceState::Compatible) => {},
+            std::result::Result::Ok(crate::SchemaResourceState::Missing | crate::SchemaResourceState::Incompatible) => {
+                log_schema_block("metadata_incompatible");
+                return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "metadata_incompatible"));
+            },
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
         }
         let history_result = load_history(&transaction).await;
         let history = match history_result {
@@ -137,21 +149,40 @@ async fn bootstrap_inner(
             std::result::Result::Err(error) => return std::result::Result::Err(error),
         };
         let validation_result = validate_history(history.as_slice(), EMBEDDED_MIGRATIONS);
-        match validation_result {
+        let index = match validation_result {
             std::result::Result::Ok(value) => value,
             std::result::Result::Err(error) => return std::result::Result::Err(error),
-        }
+        };
+        (index, SchemaMutationMode::Update)
     } else {
-        0
+        if !schema_autocreate {
+            log_schema_block("schema_autocreate_disabled");
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "schema_autocreate_disabled"));
+        }
+        let managed_result = crate::managed_v001_objects_exist(&transaction).await;
+        let managed_objects_exist = match managed_result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        if managed_objects_exist && !schema_autoupdate {
+            log_schema_block("schema_adoption_disabled");
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "schema_adoption_disabled"));
+        }
+        (0, SchemaMutationMode::Create)
     };
+    let existing_schema_result = verify_or_repair_applied_migrations(&transaction, next_index, schema_autoupdate).await;
+    if let std::result::Result::Err(error) = existing_schema_result {
+        return std::result::Result::Err(error);
+    }
     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 {
+    if next_index < EMBEDDED_MIGRATIONS.len() && mutation_mode == SchemaMutationMode::Update && !schema_autoupdate {
+        log_schema_block("migration_pending");
         return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_pending"));
     }
-    let apply_result = apply_pending_migrations(&transaction, network, next_index).await;
+    let apply_result = apply_pending_migrations(&transaction, network, next_index, mutation_mode).await;
     if let std::result::Result::Err(error) = apply_result {
         return std::result::Result::Err(error);
     }
@@ -207,20 +238,23 @@ async fn apply_migration(
     transaction: &deadpool_postgres::Transaction<'_>,
     network: &ksp_store_api::RawNetworkId,
     migration: &EmbeddedMigration,
+    mutation_mode: SchemaMutationMode,
 ) -> std::result::Result<(), crate::PostgresBackendError> {
-    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"));
+    for resource in migration.resources {
+        let result = ensure_resource(transaction, resource, mutation_mode, false).await;
+        if let std::result::Result::Err(error) = result {
+            return std::result::Result::Err(error);
+        }
     }
-    let shape_result = verify_metadata_shape(transaction).await;
-    if let std::result::Result::Err(error) = shape_result {
+    let contract_result = verify_migration_contract(transaction, migration.version).await;
+    if let std::result::Result::Err(error) = contract_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 checksum = migration_checksum(migration);
     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(()),
@@ -234,11 +268,12 @@ async fn apply_pending_migrations(
     transaction: &deadpool_postgres::Transaction<'_>,
     network: &ksp_store_api::RawNetworkId,
     next_index: usize,
+    mutation_mode: SchemaMutationMode,
 ) -> 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;
+        let result = apply_migration(transaction, network, migration, mutation_mode).await;
         if let std::result::Result::Err(error) = result {
             return std::result::Result::Err(error);
         }
@@ -247,6 +282,59 @@ async fn apply_pending_migrations(
     return std::result::Result::Ok(());
 }
 
+async fn ensure_resource(
+    transaction: &deadpool_postgres::Transaction<'_>,
+    resource: &crate::SchemaResource,
+    mutation_mode: SchemaMutationMode,
+    applied_history: bool,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    let state_result = crate::inspect_resource(transaction, resource).await;
+    let state = match state_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    match state {
+        crate::SchemaResourceState::Compatible => return std::result::Result::Ok(()),
+        crate::SchemaResourceState::Incompatible => {
+            log_schema_resource_block(resource.id, "incompatible");
+            return std::result::Result::Err(crate::PostgresBackendError::new(
+                crate::PostgresBackendErrorKind::MigrationMismatch,
+                "schema_resource_incompatible",
+            ));
+        },
+        crate::SchemaResourceState::Missing => {},
+    }
+    if applied_history && !resource.repair_existing {
+        log_schema_resource_block(resource.id, "repair_forbidden");
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "schema_resource_missing"));
+    }
+    if applied_history && mutation_mode != SchemaMutationMode::Update {
+        log_schema_resource_block(resource.id, "repair_mode_invalid");
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "schema_resource_missing"));
+    }
+    if applied_history {
+        ksp_logging_lib::warn!(
+            target: crate::TRACING_TARGET,
+            resource_id = resource.id,
+            "repairing missing PostgreSQL Store schema resource under schema_autoupdate policy"
+        );
+    }
+    let execute_result = transaction.batch_execute(resource.sql).await;
+    if execute_result.is_err() {
+        log_schema_resource_block(resource.id, "apply_failed");
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "schema_resource_apply"));
+    }
+    let verified_result = crate::inspect_resource(transaction, resource).await;
+    return match verified_result {
+        std::result::Result::Ok(crate::SchemaResourceState::Compatible) => std::result::Result::Ok(()),
+        std::result::Result::Ok(crate::SchemaResourceState::Missing | crate::SchemaResourceState::Incompatible) => {
+            log_schema_resource_block(resource.id, "post_apply_incompatible");
+            std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "schema_resource_post_apply"))
+        },
+        std::result::Result::Err(error) => std::result::Result::Err(error),
+    };
+}
+
 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 {
@@ -288,11 +376,30 @@ async fn metadata_exists(transaction: &deadpool_postgres::Transaction<'_>) -> st
     };
 }
 
-fn migration_checksum(sql: &str) -> std::string::String {
+fn migration_checksum(migration: &EmbeddedMigration) -> std::string::String {
+    return match migration.checksum {
+        MigrationChecksum::LegacySql(sql) => checksum_bytes(sql.as_bytes()),
+        MigrationChecksum::Resources => {
+            let mut hasher = sha2::Sha256::new();
+            hasher.update(b"ksp-migration-resources-v1\0");
+            for resource in migration.resources {
+                hasher.update(resource.id.as_bytes());
+                hasher.update([0]);
+                hasher.update(resource.sql.as_bytes());
+                hasher.update([0]);
+            }
+            return encode_digest(hasher.finalize().as_slice());
+        },
+    };
+}
+
+fn checksum_bytes(bytes: &[u8]) -> std::string::String {
     let mut hasher = sha2::Sha256::new();
-    hasher.update(sql.as_bytes());
-    let digest = hasher.finalize();
-    let bytes = digest.as_slice();
+    hasher.update(bytes);
+    return encode_digest(hasher.finalize().as_slice());
+}
+
+fn encode_digest(bytes: &[u8]) -> std::string::String {
     let mut encoded = std::string::String::with_capacity(bytes.len() * 2);
     for byte in bytes {
         let value = *byte;
@@ -336,33 +443,44 @@ async fn bind_store_identity(
     network: &ksp_store_api::RawNetworkId,
     context: MigrationHookContext,
 ) -> std::result::Result<(), crate::PostgresBackendError> {
-    if context == MigrationHookContext::AppliedNow {
+    let first_read = load_store_identity(transaction).await;
+    let first = match first_read {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    if first.is_empty() && context == MigrationHookContext::AppliedNow {
         let insert_result = transaction.execute(IDENTITY_INSERT_SQL, &[&network.as_str()]).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,
-                    "store_identity_insert",
-                ));
+                return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "store_identity_insert"));
             },
         }
+        let second_read = load_store_identity(transaction).await;
+        let second = match second_read {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        };
+        return validate_store_identity(second.as_slice(), network);
     }
+    return validate_store_identity(first.as_slice(), network);
+}
+
+async fn load_store_identity(
+    transaction: &deadpool_postgres::Transaction<'_>,
+) -> std::result::Result<std::vec::Vec<tokio_postgres::Row>, crate::PostgresBackendError> {
     let rows_result = transaction.query(IDENTITY_LOAD_SQL, &[]).await;
-    let rows = match rows_result {
-        std::result::Result::Ok(value) => value,
+    return match rows_result {
+        std::result::Result::Ok(value) => std::result::Result::Ok(value),
         std::result::Result::Err(_) => {
-            return std::result::Result::Err(crate::PostgresBackendError::new(
-                crate::PostgresBackendErrorKind::MigrationMismatch,
-                "store_identity_read",
-            ));
+            std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "store_identity_read"))
         },
     };
+}
+
+fn validate_store_identity(rows: &[tokio_postgres::Row], network: &ksp_store_api::RawNetworkId) -> std::result::Result<(), crate::PostgresBackendError> {
     if rows.len() != 1 {
-        return std::result::Result::Err(crate::PostgresBackendError::new(
-            crate::PostgresBackendErrorKind::MigrationMismatch,
-            "store_identity_count",
-        ));
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "store_identity_count"));
     }
     let row = &rows[0];
     let singleton_result = row.try_get::<usize, i16>(0);
@@ -370,32 +488,20 @@ async fn bind_store_identity(
     let (singleton, stored_network) = match (singleton_result, network_result) {
         (std::result::Result::Ok(singleton), std::result::Result::Ok(stored_network)) => (singleton, stored_network),
         _ => {
-            return std::result::Result::Err(crate::PostgresBackendError::new(
-                crate::PostgresBackendErrorKind::MigrationMismatch,
-                "store_identity_decode",
-            ));
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "store_identity_decode"));
         },
     };
     if singleton != 1 {
-        return std::result::Result::Err(crate::PostgresBackendError::new(
-            crate::PostgresBackendErrorKind::MigrationMismatch,
-            "store_identity_singleton",
-        ));
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "store_identity_singleton"));
     }
     let stored_network = match ksp_store_api::RawNetworkId::new(stored_network) {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(_) => {
-            return std::result::Result::Err(crate::PostgresBackendError::new(
-                crate::PostgresBackendErrorKind::MigrationMismatch,
-                "store_identity_network",
-            ));
+            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "store_identity_network"));
         },
     };
     if stored_network.as_str() != network.as_str() {
-        return std::result::Result::Err(crate::PostgresBackendError::new(
-            crate::PostgresBackendErrorKind::MigrationMismatch,
-            "store_identity_network",
-        ));
+        return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "store_identity_network"));
     }
     return std::result::Result::Ok(());
 }
@@ -414,15 +520,83 @@ async fn set_statement_timeout(
     };
 }
 
+async fn verify_migration_contract(transaction: &deadpool_postgres::Transaction<'_>, version: i64) -> std::result::Result<(), crate::PostgresBackendError> {
+    if version == 1 {
+        let result = crate::verify_v001_external_compatibility(transaction).await;
+        if let std::result::Result::Err(error) = result {
+            log_schema_block(error.phase());
+            return std::result::Result::Err(error);
+        }
+    }
+    return std::result::Result::Ok(());
+}
+
+async fn verify_or_repair_applied_migrations(
+    transaction: &deadpool_postgres::Transaction<'_>,
+    applied_count: usize,
+    schema_autoupdate: bool,
+) -> std::result::Result<(), crate::PostgresBackendError> {
+    let mut index = 0_usize;
+    while index < applied_count {
+        let migration = &EMBEDDED_MIGRATIONS[index];
+        for resource in migration.resources {
+            let state_result = crate::inspect_resource(transaction, resource).await;
+            let state = match state_result {
+                std::result::Result::Ok(value) => value,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
+            match state {
+                crate::SchemaResourceState::Compatible => {},
+                crate::SchemaResourceState::Incompatible => {
+                    log_schema_resource_block(resource.id, "incompatible");
+                    return std::result::Result::Err(crate::PostgresBackendError::new(
+                        crate::PostgresBackendErrorKind::MigrationMismatch,
+                        "schema_resource_incompatible",
+                    ));
+                },
+                crate::SchemaResourceState::Missing => {
+                    if !schema_autoupdate {
+                        log_schema_resource_block(resource.id, "schema_autoupdate_disabled");
+                        return std::result::Result::Err(crate::PostgresBackendError::new(
+                            crate::PostgresBackendErrorKind::MigrationFailed,
+                            "schema_autoupdate_disabled",
+                        ));
+                    }
+                    let repair_result = ensure_resource(transaction, resource, SchemaMutationMode::Update, true).await;
+                    if let std::result::Result::Err(error) = repair_result {
+                        return std::result::Result::Err(error);
+                    }
+                },
+            }
+        }
+        let contract_result = verify_migration_contract(transaction, migration.version).await;
+        if let std::result::Result::Err(error) = contract_result {
+            return std::result::Result::Err(error);
+        }
+        index += 1;
+    }
+    return std::result::Result::Ok(());
+}
+
 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() {
+        if migration.version != expected_version || migration.name.is_empty() || migration.resources.is_empty() {
             return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "registry_invalid"));
         }
+        for (resource_index, resource) in migration.resources.iter().enumerate() {
+            if resource.id.is_empty() || resource.sql.is_empty() {
+                return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "registry_invalid"));
+            }
+            for previous in &migration.resources[..resource_index] {
+                if previous.id == resource.id {
+                    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 => {
@@ -447,7 +621,7 @@ fn validate_history(history: &[AppliedMigration], migrations: &[EmbeddedMigratio
             return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::SchemaNewer, "history_newer"));
         }
         let expected = &migrations[index];
-        let expected_checksum = migration_checksum(expected.sql);
+        let expected_checksum = migration_checksum(expected);
         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"));
         }
@@ -456,59 +630,17 @@ fn validate_history(history: &[AppliedMigration], migrations: &[EmbeddedMigratio
     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 {
-        std::result::Result::Ok(value) => value,
-        std::result::Result::Err(_) => {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_shape"));
-        },
-    };
-    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 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 (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, required) in REQUIRED.iter().enumerate() {
-            if name == required.0 && data_type == required.1 && nullable == required.2 {
-                found[index] = true;
-                break;
-            }
-        }
-    }
-    for required in found {
-        if !required {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "metadata_shape"));
-        }
-    }
-    let key_result = transaction.query_one(METADATA_PRIMARY_KEY_SQL, &[]).await;
-    let key_row = match key_result {
-        std::result::Result::Ok(value) => value,
-        std::result::Result::Err(_) => {
-            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_primary_key"));
-        },
-    };
-    let key_count = key_row.try_get::<usize, i64>(0);
-    let version_count = key_row.try_get::<usize, i64>(1);
-    return match (key_count, version_count) {
-        (std::result::Result::Ok(1), std::result::Result::Ok(1)) => std::result::Result::Ok(()),
-        (std::result::Result::Ok(_), std::result::Result::Ok(_)) => {
-            std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "metadata_primary_key"))
-        },
-        _ => std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_primary_key_decode")),
-    };
+fn log_schema_block(phase: &'static str) {
+    ksp_logging_lib::warn!(target: crate::TRACING_TARGET, phase, "PostgreSQL Store schema compatibility gate blocked automatic opening");
+}
+
+fn log_schema_resource_block(resource_id: &'static str, reason: &'static str) {
+    ksp_logging_lib::warn!(
+        target: crate::TRACING_TARGET,
+        resource_id,
+        reason,
+        "PostgreSQL Store schema resource requires manual reconciliation"
+    );
 }
 
 #[cfg(test)]