khadhroony-solana-project/crates/ksp-store-postgres-lib/src/migration.rs

// file: crates/ksp-store-postgres-lib/src/migration.rs
// version: 7

use sha2::Digest; // rust-rules: trait-import

const ADVISORY_LOCK_KEY: i64 = 0x4b53_5053_544f_5245;
const EMBEDDED_MIGRATIONS: &[EmbeddedMigration] = &[
    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",
        resources: crate::V001_RESOURCES,
        version: 1,
    },
];
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";
const IDENTITY_INSERT_SQL: &str = "INSERT INTO ksp_store_identity (singleton, network) VALUES (1, $1)";
const IDENTITY_LOAD_SQL: &str = "SELECT singleton, network FROM ksp_store_identity ORDER BY singleton LIMIT 2";
const LOCK_POLL_INTERVAL_MS: u64 = 25;
const METADATA_EXISTS_SQL: &str = r#"SELECT EXISTS (
    SELECT 1 FROM information_schema.tables
    WHERE table_schema = current_schema()
      AND table_name = 'ksp_store_schema_migrations'
      AND table_type = 'BASE TABLE'
)"#;
const SET_STATEMENT_TIMEOUT_SQL: &str = "SELECT set_config('statement_timeout', $1, true)";

struct AppliedMigration {
    checksum: std::string::String,
    name: std::string::String,
    version: i64,
}

#[derive(Clone, Copy)]
enum MigrationChecksum {
    LegacySql(&'static str),
    Resources,
}

#[derive(Clone, Copy)]
struct EmbeddedMigration {
    checksum: MigrationChecksum,
    hook: MigrationHook,
    name: &'static str,
    resources: &'static [crate::SchemaResource],
    version: i64,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum MigrationHook {
    None,
    StoreIdentity,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum MigrationHookContext {
    AppliedNow,
    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 {
    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,
    schema_autocreate: bool,
    schema_autoupdate: bool,
    migration_timeout: std::time::Duration,
    migration_lock_timeout: std::time::Duration,
) -> std::result::Result<(), crate::PostgresBackendError> {
    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, schema_autocreate, schema_autoupdate, migration_timeout, migration_lock_timeout),
    )
    .await;
    return match bounded {
        std::result::Result::Ok(result) => result,
        std::result::Result::Err(_) => {
            std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_timeout"))
        },
    };
}

async fn bootstrap_inner(
    client: &mut deadpool_postgres::Client,
    network: &ksp_store_api::RawNetworkId,
    schema_autocreate: bool,
    schema_autoupdate: bool,
    migration_timeout: std::time::Duration,
    migration_lock_timeout: std::time::Duration,
) -> std::result::Result<(), crate::PostgresBackendError> {
    let transaction_result = client.transaction().await;
    let transaction = match transaction_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => {
            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_begin"));
        },
    };
    let lock_result = acquire_advisory_lock(&transaction, migration_lock_timeout).await;
    if let std::result::Result::Err(error) = lock_result {
        return std::result::Result::Err(error);
    }
    let timeout_result = set_statement_timeout(&transaction, migration_timeout).await;
    if let std::result::Result::Err(error) = timeout_result {
        return std::result::Result::Err(error);
    }
    let exists_result = metadata_exists(&transaction).await;
    let metadata_exists = match exists_result {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => 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 {
            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(), EMBEDDED_MIGRATIONS);
        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 {
        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() && 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, mutation_mode).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 {
        std::result::Result::Ok(()) => std::result::Result::Ok(()),
        std::result::Result::Err(_) => {
            std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_commit"))
        },
    };
}

async fn acquire_advisory_lock(
    transaction: &deadpool_postgres::Transaction<'_>,
    timeout: std::time::Duration,
) -> std::result::Result<(), crate::PostgresBackendError> {
    let started = tokio::time::Instant::now();
    let deadline = match started.checked_add(timeout) {
        std::option::Option::Some(value) => value,
        std::option::Option::None => {
            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock_timeout"));
        },
    };
    loop {
        let row_result = transaction.query_one("SELECT pg_try_advisory_xact_lock($1)", &[&ADVISORY_LOCK_KEY]).await;
        let row = match row_result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock"));
            },
        };
        let acquired_result = row.try_get::<usize, bool>(0);
        let acquired = match acquired_result {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock_decode"));
            },
        };
        if acquired {
            return std::result::Result::Ok(());
        }
        let now = tokio::time::Instant::now();
        if now >= deadline {
            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock_timeout"));
        }
        let candidate = now + std::time::Duration::from_millis(LOCK_POLL_INTERVAL_MS);
        let wake = if candidate < deadline { candidate } else { deadline };
        tokio::time::sleep_until(wake).await;
    }
}

async fn apply_migration(
    transaction: &deadpool_postgres::Transaction<'_>,
    network: &ksp_store_api::RawNetworkId,
    migration: &EmbeddedMigration,
    mutation_mode: SchemaMutationMode,
) -> std::result::Result<(), crate::PostgresBackendError> {
    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 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);
    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,
    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, mutation_mode).await;
        if let std::result::Result::Err(error) = result {
            return std::result::Result::Err(error);
        }
        index += 1;
    }
    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, resource.id))
        },
        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 {
        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 {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(_) => {
            return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_probe"));
        },
    };
    return match row.try_get::<usize, bool>(0) {
        std::result::Result::Ok(value) => std::result::Result::Ok(value),
        std::result::Result::Err(_) => {
            std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_probe_decode"))
        },
    };
}

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(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;
        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(()),
        MigrationHook::StoreIdentity => bind_store_identity(transaction, network, context).await,
    };
}

async fn bind_store_identity(
    transaction: &deadpool_postgres::Transaction<'_>,
    network: &ksp_store_api::RawNetworkId,
    context: MigrationHookContext,
) -> std::result::Result<(), crate::PostgresBackendError> {
    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"));
            },
        }
        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;
    return match rows_result {
        std::result::Result::Ok(value) => std::result::Result::Ok(value),
        std::result::Result::Err(_) => {
            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"));
    }
    let row = &rows[0];
    let singleton_result = row.try_get::<usize, i16>(0);
    let network_result = row.try_get::<usize, std::string::String>(1);
    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"));
        },
    };
    if singleton != 1 {
        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"));
        },
    };
    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::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"))
        },
    };
}

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(schema_autoupdate_disabled_error());
                    }
                    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 schema_autoupdate_disabled_error() -> crate::PostgresBackendError {
    return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "schema_autoupdate_disabled");
}

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.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 => {
                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);
        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);
}

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)]
#[path = "../unit_tests/migration.rs"]
mod tests;