use athena_schema_heal::{
    SchemaHealAuditConfig, SchemaHealContext, SchemaHealError, SchemaHealPlan, SchemaHealer,
    SchemaMutation,
};
use sqlx::PgPool;
use std::sync::atomic::{AtomicBool, Ordering};

const STORAGE_CLIENT_NAME: &str = "athena_storage";
const STORAGE_SCHEMA: &str = "athena";

static STORAGE_RUNTIME_SCHEMA_READY: AtomicBool = AtomicBool::new(false);

pub(crate) async fn ensure_storage_runtime_schema(pool: &PgPool) -> Result<(), SchemaHealError> {
    if STORAGE_RUNTIME_SCHEMA_READY.load(Ordering::Acquire) {
        if storage_runtime_schema_ready(pool).await.unwrap_or(false) {
            return Ok(());
        }

        STORAGE_RUNTIME_SCHEMA_READY.store(false, Ordering::Release);
        tracing::warn!(
            "Athena storage schema-ready cache was set but required columns are still missing; rerunning self-heal"
        );
    }

    let report = SchemaHealer::new(pool)
        .with_audit_config(SchemaHealAuditConfig {
            enabled: true,
            actor: "athena_storage_runtime".to_string(),
        })
        .apply_plan(&storage_schema_heal_plan())
        .await?;

    if report.applied_count() > 0 {
        tracing::info!(
            plan = %report.plan_name,
            applied = report.applied_count(),
            skipped = report.skipped_count(),
            "Applied Athena storage runtime schema self-heal mutations"
        );
    }

    if !storage_runtime_schema_ready(pool).await? {
        return Err(SchemaHealError::new(
            "Athena storage schema self-heal completed but required runtime tables or columns are still unresolved",
        ));
    }

    STORAGE_RUNTIME_SCHEMA_READY.store(true, Ordering::Release);
    Ok(())
}

fn storage_schema_heal_plan() -> SchemaHealPlan {
    let context = SchemaHealContext::new(STORAGE_CLIENT_NAME, STORAGE_SCHEMA)
        .with_object_type("athena_storage_runtime_schema_self_heal");

    let mut plan = SchemaHealPlan::new("athena_storage_runtime", context);

    plan.push_mutation(SchemaMutation::statement(
        "ensure_pgcrypto_extension",
        "CREATE EXTENSION IF NOT EXISTS pgcrypto",
    ));
    plan.push_mutation(SchemaMutation::statement(
        "ensure_athena_schema",
        "CREATE SCHEMA IF NOT EXISTS athena",
    ));

    plan.push_mutation(SchemaMutation::ensure_table(
        "ensure_storage_s3_catalogs_table",
        STORAGE_SCHEMA,
        "s3_catalogs",
        r#"
        CREATE TABLE IF NOT EXISTS athena.s3_catalogs (
            id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
            name text NOT NULL UNIQUE,
            provider text NOT NULL,
            bucket text NOT NULL,
            region text,
            endpoint text,
            force_path_style boolean NOT NULL DEFAULT false,
            default_prefix text,
            public_base_url text,
            description text NOT NULL DEFAULT '',
            is_active boolean NOT NULL DEFAULT true,
            metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
            deleted_at timestamptz,
            created_at timestamptz NOT NULL DEFAULT now(),
            updated_at timestamptz NOT NULL DEFAULT now(),
            CONSTRAINT athena_s3_catalogs_name_nonempty CHECK (btrim(name) <> ''),
            CONSTRAINT athena_s3_catalogs_provider_nonempty CHECK (btrim(provider) <> ''),
            CONSTRAINT athena_s3_catalogs_bucket_nonempty CHECK (btrim(bucket) <> ''),
            CONSTRAINT athena_s3_catalogs_metadata_is_object CHECK (jsonb_typeof(metadata) = 'object')
        )
        "#,
    ));
    plan.push_mutation(SchemaMutation::ensure_table(
        "ensure_storage_s3_credentials_table",
        STORAGE_SCHEMA,
        "s3_credentials",
        r#"
        CREATE TABLE IF NOT EXISTS athena.s3_credentials (
            id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
            catalog_id uuid NOT NULL REFERENCES athena.s3_catalogs(id) ON DELETE CASCADE,
            access_key_id text NOT NULL,
            secret_key_encrypted text NOT NULL,
            session_token_encrypted text,
            is_active boolean NOT NULL DEFAULT true,
            metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
            created_at timestamptz NOT NULL DEFAULT now(),
            updated_at timestamptz NOT NULL DEFAULT now(),
            rotated_at timestamptz,
            CONSTRAINT athena_s3_credentials_access_key_nonempty CHECK (btrim(access_key_id) <> ''),
            CONSTRAINT athena_s3_credentials_secret_nonempty CHECK (btrim(secret_key_encrypted) <> ''),
            CONSTRAINT athena_s3_credentials_metadata_is_object CHECK (jsonb_typeof(metadata) = 'object')
        )
        "#,
    ));
    plan.push_mutation(SchemaMutation::ensure_table(
        "ensure_storage_files_table",
        STORAGE_SCHEMA,
        "files",
        r#"
        CREATE TABLE IF NOT EXISTS athena.files (
            id text PRIMARY KEY,
            organization_id text,
            created_by_user_id text NOT NULL,
            s3_id uuid NOT NULL REFERENCES athena.s3_catalogs(id),
            bucket text NOT NULL,
            storage_key text NOT NULL,
            file_name text,
            original_name text,
            content_type text,
            size_bytes bigint,
            checksum_sha256 text,
            visibility text NOT NULL DEFAULT 'private',
            status text NOT NULL DEFAULT 'pending',
            resource_id text,
            metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
            deleted_at timestamptz,
            created_at timestamptz NOT NULL DEFAULT now(),
            updated_at timestamptz NOT NULL DEFAULT now(),
            prefix_path text,
            extension text,
            CONSTRAINT files_visibility_check CHECK (visibility IN ('private', 'organization', 'public')),
            CONSTRAINT files_status_check CHECK (status IN ('pending', 'uploaded', 'failed', 'deleted')),
            CONSTRAINT files_bucket_nonempty CHECK (btrim(bucket) <> ''),
            CONSTRAINT files_storage_key_nonempty CHECK (btrim(storage_key) <> ''),
            CONSTRAINT files_created_by_nonempty CHECK (btrim(created_by_user_id) <> ''),
            CONSTRAINT files_metadata_is_object CHECK (jsonb_typeof(metadata) = 'object'),
            CONSTRAINT files_unique_object UNIQUE (s3_id, bucket, storage_key)
        )
        "#,
    ));
    plan.push_mutation(SchemaMutation::ensure_table(
        "ensure_storage_file_permissions_table",
        STORAGE_SCHEMA,
        "file_permissions",
        r#"
        CREATE TABLE IF NOT EXISTS athena.file_permissions (
            id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
            file_id text NOT NULL REFERENCES athena.files(id) ON DELETE CASCADE,
            principal_type text NOT NULL
                CHECK (principal_type IN ('user', 'organization', 'team', 'role')),
            principal_id text NOT NULL,
            permission text NOT NULL
                CHECK (permission IN ('read', 'write', 'delete', 'share', 'owner')),
            granted_by_user_id text,
            created_at timestamptz NOT NULL DEFAULT now(),
            expires_at timestamptz,
            revoked_at timestamptz,
            metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
            CONSTRAINT file_permissions_metadata_is_object CHECK (jsonb_typeof(metadata) = 'object'),
            UNIQUE (file_id, principal_type, principal_id, permission)
        )
        "#,
    ));
    plan.push_mutation(SchemaMutation::ensure_table(
        "ensure_storage_audit_events_table",
        STORAGE_SCHEMA,
        "storage_audit_events",
        r#"
        CREATE TABLE IF NOT EXISTS athena.storage_audit_events (
            id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
            operation text NOT NULL,
            file_id text,
            s3_id uuid,
            bucket text,
            storage_key text,
            actor_user_id text,
            organization_id text,
            status text NOT NULL CHECK (status IN ('success', 'error')),
            error text,
            metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
            created_at timestamptz NOT NULL DEFAULT now(),
            CONSTRAINT storage_audit_events_metadata_is_object CHECK (jsonb_typeof(metadata) = 'object')
        )
        "#,
    ));
    plan.push_mutation(SchemaMutation::ensure_table(
        "ensure_storage_s3_url_cache_table",
        STORAGE_SCHEMA,
        "s3_url_cache",
        r#"
        CREATE TABLE IF NOT EXISTS athena.s3_url_cache (
            id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
            file_id text NOT NULL REFERENCES athena.files(id) ON DELETE CASCADE,
            purpose text NOT NULL DEFAULT 'read',
            bucket text NOT NULL,
            storage_key text NOT NULL,
            url text NOT NULL,
            expires_at timestamptz NOT NULL,
            expires_at_epoch_seconds bigint NOT NULL,
            created_at timestamptz NOT NULL DEFAULT now(),
            metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
            CONSTRAINT athena_s3_url_cache_purpose_valid CHECK (purpose IN ('read', 'download', 'stream')),
            CONSTRAINT athena_s3_url_cache_metadata_is_object CHECK (jsonb_typeof(metadata) = 'object'),
            UNIQUE (file_id, purpose)
        )
        "#,
    ));

    for (table, column, sql) in [
        (
            "s3_catalogs",
            "provider",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN provider text NOT NULL DEFAULT 's3'",
        ),
        (
            "s3_catalogs",
            "bucket",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN bucket text NOT NULL DEFAULT ''",
        ),
        (
            "s3_catalogs",
            "region",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN region text",
        ),
        (
            "s3_catalogs",
            "endpoint",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN endpoint text",
        ),
        (
            "s3_catalogs",
            "force_path_style",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN force_path_style boolean NOT NULL DEFAULT false",
        ),
        (
            "s3_catalogs",
            "default_prefix",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN default_prefix text",
        ),
        (
            "s3_catalogs",
            "public_base_url",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN public_base_url text",
        ),
        (
            "s3_catalogs",
            "description",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN description text NOT NULL DEFAULT ''",
        ),
        (
            "s3_catalogs",
            "is_active",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN is_active boolean NOT NULL DEFAULT true",
        ),
        (
            "s3_catalogs",
            "metadata",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN metadata jsonb NOT NULL DEFAULT '{}'::jsonb",
        ),
        (
            "s3_catalogs",
            "deleted_at",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN deleted_at timestamptz",
        ),
        (
            "s3_catalogs",
            "created_at",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN created_at timestamptz NOT NULL DEFAULT now()",
        ),
        (
            "s3_catalogs",
            "updated_at",
            "ALTER TABLE athena.s3_catalogs ADD COLUMN updated_at timestamptz NOT NULL DEFAULT now()",
        ),
        (
            "s3_credentials",
            "catalog_id",
            "ALTER TABLE athena.s3_credentials ADD COLUMN catalog_id uuid",
        ),
        (
            "s3_credentials",
            "access_key_id",
            "ALTER TABLE athena.s3_credentials ADD COLUMN access_key_id text NOT NULL DEFAULT ''",
        ),
        (
            "s3_credentials",
            "secret_key_encrypted",
            "ALTER TABLE athena.s3_credentials ADD COLUMN secret_key_encrypted text",
        ),
        (
            "s3_credentials",
            "session_token_encrypted",
            "ALTER TABLE athena.s3_credentials ADD COLUMN session_token_encrypted text",
        ),
        (
            "s3_credentials",
            "is_active",
            "ALTER TABLE athena.s3_credentials ADD COLUMN is_active boolean NOT NULL DEFAULT true",
        ),
        (
            "s3_credentials",
            "metadata",
            "ALTER TABLE athena.s3_credentials ADD COLUMN metadata jsonb NOT NULL DEFAULT '{}'::jsonb",
        ),
        (
            "s3_credentials",
            "created_at",
            "ALTER TABLE athena.s3_credentials ADD COLUMN created_at timestamptz NOT NULL DEFAULT now()",
        ),
        (
            "s3_credentials",
            "updated_at",
            "ALTER TABLE athena.s3_credentials ADD COLUMN updated_at timestamptz NOT NULL DEFAULT now()",
        ),
        (
            "s3_credentials",
            "rotated_at",
            "ALTER TABLE athena.s3_credentials ADD COLUMN rotated_at timestamptz",
        ),
        (
            "files",
            "organization_id",
            "ALTER TABLE athena.files ADD COLUMN organization_id text",
        ),
        (
            "files",
            "created_by_user_id",
            "ALTER TABLE athena.files ADD COLUMN created_by_user_id text NOT NULL DEFAULT 'system'",
        ),
        (
            "files",
            "s3_id",
            "ALTER TABLE athena.files ADD COLUMN s3_id uuid",
        ),
        (
            "files",
            "bucket",
            "ALTER TABLE athena.files ADD COLUMN bucket text NOT NULL DEFAULT ''",
        ),
        (
            "files",
            "storage_key",
            "ALTER TABLE athena.files ADD COLUMN storage_key text NOT NULL DEFAULT ''",
        ),
        (
            "files",
            "file_name",
            "ALTER TABLE athena.files ADD COLUMN file_name text",
        ),
        (
            "files",
            "original_name",
            "ALTER TABLE athena.files ADD COLUMN original_name text",
        ),
        (
            "files",
            "content_type",
            "ALTER TABLE athena.files ADD COLUMN content_type text",
        ),
        (
            "files",
            "size_bytes",
            "ALTER TABLE athena.files ADD COLUMN size_bytes bigint",
        ),
        (
            "files",
            "checksum_sha256",
            "ALTER TABLE athena.files ADD COLUMN checksum_sha256 text",
        ),
        (
            "files",
            "visibility",
            "ALTER TABLE athena.files ADD COLUMN visibility text NOT NULL DEFAULT 'private'",
        ),
        (
            "files",
            "status",
            "ALTER TABLE athena.files ADD COLUMN status text NOT NULL DEFAULT 'pending'",
        ),
        (
            "files",
            "resource_id",
            "ALTER TABLE athena.files ADD COLUMN resource_id text",
        ),
        (
            "files",
            "metadata",
            "ALTER TABLE athena.files ADD COLUMN metadata jsonb NOT NULL DEFAULT '{}'::jsonb",
        ),
        (
            "files",
            "deleted_at",
            "ALTER TABLE athena.files ADD COLUMN deleted_at timestamptz",
        ),
        (
            "files",
            "created_at",
            "ALTER TABLE athena.files ADD COLUMN created_at timestamptz NOT NULL DEFAULT now()",
        ),
        (
            "files",
            "updated_at",
            "ALTER TABLE athena.files ADD COLUMN updated_at timestamptz NOT NULL DEFAULT now()",
        ),
        (
            "files",
            "prefix_path",
            "ALTER TABLE athena.files ADD COLUMN prefix_path text",
        ),
        (
            "files",
            "extension",
            "ALTER TABLE athena.files ADD COLUMN extension text",
        ),
        (
            "file_permissions",
            "granted_by_user_id",
            "ALTER TABLE athena.file_permissions ADD COLUMN granted_by_user_id text",
        ),
        (
            "file_permissions",
            "expires_at",
            "ALTER TABLE athena.file_permissions ADD COLUMN expires_at timestamptz",
        ),
        (
            "file_permissions",
            "revoked_at",
            "ALTER TABLE athena.file_permissions ADD COLUMN revoked_at timestamptz",
        ),
        (
            "file_permissions",
            "metadata",
            "ALTER TABLE athena.file_permissions ADD COLUMN metadata jsonb NOT NULL DEFAULT '{}'::jsonb",
        ),
        (
            "storage_audit_events",
            "operation",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN operation text NOT NULL DEFAULT 'unknown'",
        ),
        (
            "storage_audit_events",
            "file_id",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN file_id text",
        ),
        (
            "storage_audit_events",
            "s3_id",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN s3_id uuid",
        ),
        (
            "storage_audit_events",
            "bucket",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN bucket text",
        ),
        (
            "storage_audit_events",
            "storage_key",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN storage_key text",
        ),
        (
            "storage_audit_events",
            "actor_user_id",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN actor_user_id text",
        ),
        (
            "storage_audit_events",
            "organization_id",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN organization_id text",
        ),
        (
            "storage_audit_events",
            "status",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN status text NOT NULL DEFAULT 'success'",
        ),
        (
            "storage_audit_events",
            "error",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN error text",
        ),
        (
            "storage_audit_events",
            "metadata",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN metadata jsonb NOT NULL DEFAULT '{}'::jsonb",
        ),
        (
            "storage_audit_events",
            "created_at",
            "ALTER TABLE athena.storage_audit_events ADD COLUMN created_at timestamptz NOT NULL DEFAULT now()",
        ),
        (
            "s3_url_cache",
            "purpose",
            "ALTER TABLE athena.s3_url_cache ADD COLUMN purpose text NOT NULL DEFAULT 'read'",
        ),
        (
            "s3_url_cache",
            "bucket",
            "ALTER TABLE athena.s3_url_cache ADD COLUMN bucket text NOT NULL DEFAULT ''",
        ),
        (
            "s3_url_cache",
            "storage_key",
            "ALTER TABLE athena.s3_url_cache ADD COLUMN storage_key text NOT NULL DEFAULT ''",
        ),
        (
            "s3_url_cache",
            "url",
            "ALTER TABLE athena.s3_url_cache ADD COLUMN url text NOT NULL DEFAULT ''",
        ),
        (
            "s3_url_cache",
            "expires_at",
            "ALTER TABLE athena.s3_url_cache ADD COLUMN expires_at timestamptz NOT NULL DEFAULT now()",
        ),
        (
            "s3_url_cache",
            "expires_at_epoch_seconds",
            "ALTER TABLE athena.s3_url_cache ADD COLUMN expires_at_epoch_seconds bigint NOT NULL DEFAULT 0",
        ),
        (
            "s3_url_cache",
            "created_at",
            "ALTER TABLE athena.s3_url_cache ADD COLUMN created_at timestamptz NOT NULL DEFAULT now()",
        ),
        (
            "s3_url_cache",
            "metadata",
            "ALTER TABLE athena.s3_url_cache ADD COLUMN metadata jsonb NOT NULL DEFAULT '{}'::jsonb",
        ),
    ] {
        push_column_mutation(&mut plan, table, column, sql);
    }

    for (table, index_name, sql) in [
        (
            "s3_catalogs",
            "athena_s3_catalogs_active_created_idx",
            "CREATE INDEX IF NOT EXISTS athena_s3_catalogs_active_created_idx ON athena.s3_catalogs (is_active, created_at DESC) WHERE deleted_at IS NULL",
        ),
        (
            "s3_catalogs",
            "athena_s3_catalogs_bucket_created_idx",
            "CREATE INDEX IF NOT EXISTS athena_s3_catalogs_bucket_created_idx ON athena.s3_catalogs (bucket, created_at DESC) WHERE deleted_at IS NULL",
        ),
        (
            "s3_credentials",
            "s3_credentials_catalog_active_idx",
            "CREATE INDEX IF NOT EXISTS s3_credentials_catalog_active_idx ON athena.s3_credentials (catalog_id) WHERE is_active = true",
        ),
        (
            "s3_credentials",
            "athena_s3_credentials_created_idx",
            "CREATE INDEX IF NOT EXISTS athena_s3_credentials_created_idx ON athena.s3_credentials (created_at DESC)",
        ),
        (
            "files",
            "files_org_prefix_idx",
            "CREATE INDEX IF NOT EXISTS files_org_prefix_idx ON athena.files (organization_id, bucket, storage_key)",
        ),
        (
            "files",
            "files_created_by_idx",
            "CREATE INDEX IF NOT EXISTS files_created_by_idx ON athena.files (created_by_user_id)",
        ),
        (
            "files",
            "files_status_idx",
            "CREATE INDEX IF NOT EXISTS files_status_idx ON athena.files (status)",
        ),
        (
            "files",
            "files_metadata_gin_idx",
            "CREATE INDEX IF NOT EXISTS files_metadata_gin_idx ON athena.files USING gin (metadata)",
        ),
        (
            "files",
            "files_s3_storage_key_idx",
            "CREATE INDEX IF NOT EXISTS files_s3_storage_key_idx ON athena.files (s3_id, storage_key)",
        ),
        (
            "file_permissions",
            "file_permissions_lookup_idx",
            "CREATE INDEX IF NOT EXISTS file_permissions_lookup_idx ON athena.file_permissions (file_id, principal_type, principal_id, permission) WHERE revoked_at IS NULL",
        ),
        (
            "file_permissions",
            "file_permissions_principal_idx",
            "CREATE INDEX IF NOT EXISTS file_permissions_principal_idx ON athena.file_permissions (principal_type, principal_id) WHERE revoked_at IS NULL",
        ),
        (
            "storage_audit_events",
            "storage_audit_events_file_created_idx",
            "CREATE INDEX IF NOT EXISTS storage_audit_events_file_created_idx ON athena.storage_audit_events (file_id, created_at DESC) WHERE file_id IS NOT NULL",
        ),
        (
            "storage_audit_events",
            "storage_audit_events_bucket_key_created_idx",
            "CREATE INDEX IF NOT EXISTS storage_audit_events_bucket_key_created_idx ON athena.storage_audit_events (bucket, storage_key, created_at DESC) WHERE bucket IS NOT NULL AND storage_key IS NOT NULL",
        ),
        (
            "storage_audit_events",
            "storage_audit_events_actor_created_idx",
            "CREATE INDEX IF NOT EXISTS storage_audit_events_actor_created_idx ON athena.storage_audit_events (actor_user_id, created_at DESC) WHERE actor_user_id IS NOT NULL",
        ),
        (
            "s3_url_cache",
            "athena_s3_url_cache_lookup_idx",
            "CREATE INDEX IF NOT EXISTS athena_s3_url_cache_lookup_idx ON athena.s3_url_cache (file_id, purpose, expires_at)",
        ),
        (
            "s3_url_cache",
            "athena_s3_url_cache_expiry_idx",
            "CREATE INDEX IF NOT EXISTS athena_s3_url_cache_expiry_idx ON athena.s3_url_cache (expires_at)",
        ),
    ] {
        push_index_mutation(&mut plan, table, index_name, sql);
    }

    plan
}

fn push_column_mutation(plan: &mut SchemaHealPlan, table: &str, column: &str, sql: &str) {
    plan.push_mutation(SchemaMutation::ensure_column(
        format!("ensure_storage_{table}_{column}_column"),
        STORAGE_SCHEMA,
        table,
        column,
        sql,
    ));
}

fn push_index_mutation(plan: &mut SchemaHealPlan, table: &str, index_name: &str, sql: &str) {
    plan.push_mutation(SchemaMutation::ensure_index(
        format!("ensure_storage_{index_name}"),
        STORAGE_SCHEMA,
        table,
        index_name,
        sql,
    ));
}

async fn storage_runtime_schema_ready(pool: &PgPool) -> Result<bool, SchemaHealError> {
    sqlx::query_scalar::<_, bool>(
        r#"
        SELECT
            to_regclass('athena.s3_catalogs') IS NOT NULL
            AND to_regclass('athena.s3_credentials') IS NOT NULL
            AND to_regclass('athena.files') IS NOT NULL
            AND to_regclass('athena.file_permissions') IS NOT NULL
            AND to_regclass('athena.storage_audit_events') IS NOT NULL
            AND to_regclass('athena.s3_url_cache') IS NOT NULL
            AND EXISTS (
                SELECT 1
                FROM information_schema.columns
                WHERE table_schema = 'athena'
                  AND table_name = 's3_credentials'
                  AND column_name = 'catalog_id'
            )
            AND EXISTS (
                SELECT 1
                FROM information_schema.columns
                WHERE table_schema = 'athena'
                  AND table_name = 's3_credentials'
                  AND column_name = 'secret_key_encrypted'
            )
            AND EXISTS (
                SELECT 1
                FROM information_schema.columns
                WHERE table_schema = 'athena'
                  AND table_name = 'files'
                  AND column_name = 'created_by_user_id'
            )
            AND EXISTS (
                SELECT 1
                FROM information_schema.columns
                WHERE table_schema = 'athena'
                  AND table_name = 'files'
                  AND column_name = 'visibility'
            )
            AND EXISTS (
                SELECT 1
                FROM information_schema.columns
                WHERE table_schema = 'athena'
                  AND table_name = 'storage_audit_events'
                  AND column_name = 'operation'
            )
            AND EXISTS (
                SELECT 1
                FROM information_schema.columns
                WHERE table_schema = 'athena'
                  AND table_name = 's3_url_cache'
                  AND column_name = 'expires_at_epoch_seconds'
            )
        "#,
    )
    .fetch_one(pool)
    .await
    .map_err(|err| {
        SchemaHealError::new(format!(
            "failed to verify Athena storage runtime schema readiness: {err}"
        ))
    })
}

#[cfg(test)]
mod tests {
    use super::storage_schema_heal_plan;

    #[test]
    fn storage_schema_heal_plan_covers_runtime_tables() {
        let plan = storage_schema_heal_plan();
        let labels: Vec<String> = plan
            .mutations
            .iter()
            .map(|mutation| match mutation {
                athena_schema_heal::SchemaMutation::EnsureExtension(mutation) => {
                    mutation.label.clone()
                }
                athena_schema_heal::SchemaMutation::EnsureSchema(mutation) => {
                    mutation.label.clone()
                }
                athena_schema_heal::SchemaMutation::EnsureTable(mutation) => mutation.label.clone(),
                athena_schema_heal::SchemaMutation::EnsureColumn(mutation) => {
                    mutation.label.clone()
                }
                athena_schema_heal::SchemaMutation::EnsureColumnNotNull(mutation) => {
                    mutation.label.clone()
                }
                athena_schema_heal::SchemaMutation::EnsureColumnDefault(mutation) => {
                    mutation.label.clone()
                }
                athena_schema_heal::SchemaMutation::EnsureColumnGenerated(mutation) => {
                    mutation.label.clone()
                }
                athena_schema_heal::SchemaMutation::EnsureColumnType(mutation) => {
                    mutation.label.clone()
                }
                athena_schema_heal::SchemaMutation::EnsureConstraint(mutation) => {
                    mutation.label.clone()
                }
                athena_schema_heal::SchemaMutation::DropConstraint(mutation) => {
                    mutation.label.clone()
                }
                athena_schema_heal::SchemaMutation::EnsureIndex(mutation) => mutation.label.clone(),
                athena_schema_heal::SchemaMutation::DropIndex(mutation) => mutation.label.clone(),
                athena_schema_heal::SchemaMutation::DropColumn(mutation) => mutation.label.clone(),
                athena_schema_heal::SchemaMutation::Statement(mutation) => mutation.label.clone(),
            })
            .collect();

        for label in [
            "ensure_pgcrypto_extension",
            "ensure_athena_schema",
            "ensure_storage_s3_catalogs_table",
            "ensure_storage_s3_credentials_table",
            "ensure_storage_files_table",
            "ensure_storage_file_permissions_table",
            "ensure_storage_audit_events_table",
            "ensure_storage_s3_url_cache_table",
            "ensure_storage_s3_credentials_catalog_id_column",
            "ensure_storage_files_created_by_user_id_column",
            "ensure_storage_storage_audit_events_operation_column",
            "ensure_storage_athena_s3_url_cache_lookup_idx",
        ] {
            assert!(
                labels.iter().any(|candidate| candidate == label),
                "missing mutation label {label}"
            );
        }
    }
}
