dovecote_sqlx_postgres/

schema.rs

//! PostgreSQL catalog verification for the installed Dovecote schema.

use crate::{
    error::SchemaError,
    migration::{SchemaMarker, current_migration, marker_matches_migration},
};
use sqlx::{FromRow, PgConnection, PgPool, query_as};

/// Verifies the tables and all columns required by schema version 1.
pub async fn check_schema(pool: &PgPool) -> Result<(), SchemaError> {
    let mut connection = pool
        .acquire()
        .await
        .map_err(|source| SchemaError::sql("acquire schema-check connection", source))?;
    check_schema_connection(&mut connection).await
}

/// Performs the complete schema check on an already-owned connection.
pub(crate) async fn check_schema_connection(
    connection: &mut PgConnection,
) -> Result<(), SchemaError> {
    let namespace = resolve_namespace(connection).await?;

    let marker_columns = [
        ColumnSpec::required("schema_version", "integer", None),
        ColumnSpec::required("minimum_crate_major", "smallint", None),
        ColumnSpec::required("minimum_crate_minor", "smallint", None),
        ColumnSpec::required("minimum_crate_patch", "smallint", None),
        ColumnSpec::required("rolling_compatible", "boolean", None),
    ];
    check_columns(
        connection,
        &namespace.name,
        "dovecote_schema",
        &marker_columns,
    )
    .await?;
    let marker = query_as::<_, SchemaMarker>(
        r#"
        SELECT schema_version, minimum_crate_major, minimum_crate_minor,
               minimum_crate_patch, rolling_compatible
        FROM dovecote_schema
        ORDER BY schema_version DESC
        LIMIT 1
        "#,
    )
    .fetch_optional(&mut *connection)
    .await
    .map_err(|source| SchemaError::sql("check schema marker", source))?
    .ok_or_else(|| SchemaError::MigrationMismatch {
        detail: "schema marker is missing".to_owned(),
    })?;
    let migration =
        current_migration().map_err(|detail| SchemaError::MigrationMismatch { detail })?;
    if let Err(detail) = marker_matches_migration(&marker, migration) {
        return Err(SchemaError::MigrationMismatch { detail });
    }

    let event_columns = [
        ColumnSpec::required_identity("row_id", "bigint", None),
        ColumnSpec::required("stream", "character varying", Some(255)),
        ColumnSpec::required("specversion", "character varying", Some(8)),
        ColumnSpec::required("event_id", "character varying", Some(1024)),
        ColumnSpec::required("source", "character varying", Some(2048)),
        ColumnSpec::required("event_type", "character varying", Some(1024)),
        ColumnSpec::optional("subject", "character varying", Some(2048)),
        ColumnSpec::optional("occurred_at", "timestamp with time zone", None),
        ColumnSpec::optional("datacontenttype", "character varying", Some(255)),
        ColumnSpec::optional("dataschema", "character varying", Some(2048)),
        ColumnSpec::optional("partitionkey", "character varying", Some(255)),
        ColumnSpec::required_with_default("extensions", "text", None, "'{}'::text"),
        ColumnSpec::optional("data_kind", "character varying", Some(6)),
        ColumnSpec::optional("data", "bytea", None),
        ColumnSpec::required_with_default(
            "enqueued_at",
            "timestamp with time zone",
            None,
            "current_timestamp",
        ),
    ];
    let delivery_columns = [
        ColumnSpec::required("event_row_id", "bigint", None),
        ColumnSpec::required("state", "character varying", Some(12)),
        ColumnSpec::required_with_default(
            "available_at",
            "timestamp with time zone",
            None,
            "current_timestamp",
        ),
        ColumnSpec::required_with_default("attempts", "bigint", None, "0"),
        ColumnSpec::optional("claim_token", "bytea", None),
        ColumnSpec::optional("claimed_by", "character varying", Some(255)),
        ColumnSpec::optional("claim_expires_at", "timestamp with time zone", None),
        ColumnSpec::optional("last_failure_code", "character varying", Some(128)),
        ColumnSpec::optional("last_failure_detail", "character varying", Some(2048)),
        ColumnSpec::optional("delivered_at", "timestamp with time zone", None),
        ColumnSpec::optional("quarantined_at", "timestamp with time zone", None),
        ColumnSpec::optional("quarantine_reason", "character varying", Some(2048)),
    ];
    check_columns(
        connection,
        &namespace.name,
        "dovecote_events",
        &event_columns,
    )
    .await?;
    check_columns(
        connection,
        &namespace.name,
        "dovecote_deliveries",
        &delivery_columns,
    )
    .await?;

    let expected_constraints = [
        ConstraintContract::check(
            "dovecote_schema_version_supported",
            "dovecote_schema",
            &["CHECK ((schema_version = 1))"],
        ),
        ConstraintContract::check(
            "dovecote_schema_minimum_nonnegative",
            "dovecote_schema",
            &[
                "CHECK (((minimum_crate_major >= 0) AND (minimum_crate_minor >= 0) AND (minimum_crate_patch >= 0)))",
            ],
        ),
        ConstraintContract::primary_key(
            "dovecote_schema_pkey",
            "dovecote_schema",
            &["schema_version"],
            &["PRIMARY KEY (schema_version)"],
        ),
        ConstraintContract::check(
            "dovecote_events_row_id_positive",
            "dovecote_events",
            &["CHECK ((row_id > 0))"],
        ),
        ConstraintContract::primary_key(
            "dovecote_events_pkey",
            "dovecote_events",
            &["row_id"],
            &["PRIMARY KEY (row_id)"],
        ),
        ConstraintContract::check(
            "dovecote_events_specversion",
            "dovecote_events",
            &["CHECK (((specversion) = '1.0'))"],
        ),
        ConstraintContract::check(
            "dovecote_events_stream_size",
            "dovecote_events",
            &["CHECK ((octet_length((stream)) <= 255))"],
        ),
        ConstraintContract::check(
            "dovecote_events_event_id_size",
            "dovecote_events",
            &["CHECK ((octet_length((event_id)) <= 1024))"],
        ),
        ConstraintContract::check(
            "dovecote_events_source_size",
            "dovecote_events",
            &["CHECK ((octet_length((source)) <= 2048))"],
        ),
        ConstraintContract::check(
            "dovecote_events_event_type_size",
            "dovecote_events",
            &["CHECK ((octet_length((event_type)) <= 1024))"],
        ),
        ConstraintContract::check(
            "dovecote_events_subject_size",
            "dovecote_events",
            &["CHECK (((subject IS NULL) OR (octet_length((subject)) <= 2048)))"],
        ),
        ConstraintContract::check(
            "dovecote_events_content_type_size",
            "dovecote_events",
            &["CHECK (((datacontenttype IS NULL) OR (octet_length((datacontenttype)) <= 255)))"],
        ),
        ConstraintContract::check(
            "dovecote_events_schema_size",
            "dovecote_events",
            &["CHECK (((dataschema IS NULL) OR (octet_length((dataschema)) <= 2048)))"],
        ),
        ConstraintContract::check(
            "dovecote_events_partition_size",
            "dovecote_events",
            &["CHECK (((partitionkey IS NULL) OR (octet_length((partitionkey)) <= 255)))"],
        ),
        ConstraintContract::check(
            "dovecote_events_identity_size",
            "dovecote_events",
            &["CHECK (((octet_length((source)) + octet_length((event_id))) <= 2048))"],
        ),
        ConstraintContract::check(
            "dovecote_events_data_kind",
            "dovecote_events",
            &["CHECK (((data_kind IS NULL) OR ((data_kind) = ANY ((ARRAY['json', 'binary'])))))"],
        ),
        ConstraintContract::check(
            "dovecote_events_data_pair",
            "dovecote_events",
            &["CHECK (((data_kind IS NULL) = (data IS NULL)))"],
        ),
        ConstraintContract::check(
            "dovecote_events_content_type",
            "dovecote_events",
            &[
                "CHECK (((data IS NULL) OR (octet_length(data) = 0) OR (datacontenttype IS NOT NULL)))",
            ],
        ),
        ConstraintContract::check(
            "dovecote_deliveries_state",
            "dovecote_deliveries",
            &[
                "CHECK (((state) = ANY ((ARRAY['pending', 'claimed', 'delivered', 'quarantined']))))",
            ],
        ),
        ConstraintContract::primary_key(
            "dovecote_deliveries_pkey",
            "dovecote_deliveries",
            &["event_row_id"],
            &["PRIMARY KEY (event_row_id)"],
        ),
        ConstraintContract::check(
            "dovecote_deliveries_attempts",
            "dovecote_deliveries",
            &["CHECK ((attempts >= 0))"],
        ),
        ConstraintContract::check(
            "dovecote_deliveries_token_size",
            "dovecote_deliveries",
            &["CHECK (((claim_token IS NULL) OR (octet_length(claim_token) = 16)))"],
        ),
        ConstraintContract::check(
            "dovecote_deliveries_worker_size",
            "dovecote_deliveries",
            &["CHECK (((claimed_by IS NULL) OR (octet_length((claimed_by)) <= 255)))"],
        ),
        ConstraintContract::check(
            "dovecote_deliveries_failure_code_size",
            "dovecote_deliveries",
            &[
                "CHECK (((last_failure_code IS NULL) OR (octet_length((last_failure_code)) <= 128)))",
            ],
        ),
        ConstraintContract::check(
            "dovecote_deliveries_failure_detail_size",
            "dovecote_deliveries",
            &[
                "CHECK (((last_failure_detail IS NULL) OR (octet_length((last_failure_detail)) <= 2048)))",
            ],
        ),
        ConstraintContract::check(
            "dovecote_deliveries_quarantine_size",
            "dovecote_deliveries",
            &[
                "CHECK (((quarantine_reason IS NULL) OR (octet_length((quarantine_reason)) <= 2048)))",
            ],
        ),
        ConstraintContract::check(
            "dovecote_deliveries_failure_pair",
            "dovecote_deliveries",
            &["CHECK (((last_failure_code IS NULL) = (last_failure_detail IS NULL)))"],
        ),
        ConstraintContract::check(
            "dovecote_deliveries_state_shape",
            "dovecote_deliveries",
            &[
                "CHECK (((((state) = 'pending') AND (claim_token IS NULL) AND (claimed_by IS NULL) AND (claim_expires_at IS NULL) AND (delivered_at IS NULL) AND (quarantined_at IS NULL) AND (quarantine_reason IS NULL)) OR (((state) = 'claimed') AND (claim_token IS NOT NULL) AND (claimed_by IS NOT NULL) AND (claim_expires_at IS NOT NULL) AND (delivered_at IS NULL) AND (quarantined_at IS NULL) AND (quarantine_reason IS NULL)) OR (((state) = 'delivered') AND (claim_token IS NULL) AND (claimed_by IS NULL) AND (claim_expires_at IS NULL) AND (delivered_at IS NOT NULL) AND (quarantined_at IS NULL) AND (quarantine_reason IS NULL)) OR (((state) = 'quarantined') AND (claim_token IS NULL) AND (claimed_by IS NULL) AND (claim_expires_at IS NULL) AND (delivered_at IS NULL) AND (quarantined_at IS NOT NULL) AND (quarantine_reason IS NOT NULL))))",
            ],
        ),
        ConstraintContract::foreign_key(
            "dovecote_deliveries_event_row_id_fkey",
            "dovecote_deliveries",
            &["event_row_id"],
            "dovecote_events",
            &["row_id"],
            "r",
            &["FOREIGN KEY (event_row_id) REFERENCES dovecote_events (row_id) ON DELETE RESTRICT"],
        ),
    ];

    let constraints = query_as::<_, ConstraintInfo>(
        r#"
        SELECT table_class.relname AS table_name,
               constraint_class.conname AS name,
               constraint_class.contype::text AS kind,
               ARRAY(
                   SELECT attribute.attname::text
                   FROM unnest(constraint_class.conkey) WITH ORDINALITY AS key(attnum, ordinality)
                   JOIN pg_attribute attribute
                     ON attribute.attrelid = constraint_class.conrelid
                    AND attribute.attnum = key.attnum
                   ORDER BY key.ordinality
               ) AS columns,
               parent_class.relname AS referenced_table,
               ARRAY(
                   SELECT attribute.attname::text
                   FROM unnest(constraint_class.confkey) WITH ORDINALITY AS key(attnum, ordinality)
                   JOIN pg_attribute attribute
                     ON attribute.attrelid = constraint_class.confrelid
                    AND attribute.attnum = key.attnum
                   ORDER BY key.ordinality
               ) AS referenced_columns,
               CASE WHEN constraint_class.contype = 'f'
                    THEN constraint_class.confdeltype::text END AS delete_action,
               constraint_class.convalidated AS validated,
               constraint_class.condeferrable AS deferrable,
               constraint_class.condeferred AS deferred,
               pg_get_constraintdef(constraint_class.oid) AS definition
        FROM pg_constraint constraint_class
        JOIN pg_class table_class ON table_class.oid = constraint_class.conrelid
        LEFT JOIN pg_class parent_class ON parent_class.oid = constraint_class.confrelid
        WHERE table_class.relnamespace::bigint = $1
          AND (constraint_class.confrelid = 0 OR parent_class.relnamespace::bigint = $1)
          AND constraint_class.conname = ANY($2)
        "#,
    )
    .bind(namespace.oid)
    .bind(
        expected_constraints
            .iter()
            .map(|constraint| constraint.name.to_owned())
            .collect::<Vec<_>>(),
    )
    .fetch_all(&mut *connection)
    .await
    .map_err(|source| SchemaError::sql("check constraints", source))?;
    for expected in &expected_constraints {
        let Some(actual) = constraints
            .iter()
            .find(|constraint| constraint.name == expected.name)
        else {
            return Err(SchemaError::MigrationMismatch {
                detail: format!("required constraint {} is missing", expected.name),
            });
        };

        if !actual.matches(expected) {
            return Err(SchemaError::MigrationMismatch {
                detail: format!("required constraint {} is incompatible", expected.name),
            });
        }
    }

    let expected_indexes = [
        IndexContract::new(
            "dovecote_events_source_event_id",
            "dovecote_events",
            true,
            &["source", "event_id"],
            Some(&["C", "C"]),
        ),
        IndexContract::new(
            "dovecote_deliveries_claimable",
            "dovecote_deliveries",
            false,
            &["state", "available_at", "event_row_id"],
            None,
        ),
        IndexContract::new(
            "dovecote_deliveries_expired_claims",
            "dovecote_deliveries",
            false,
            &["state", "claim_expires_at", "event_row_id"],
            None,
        ),
    ];
    let indexes = query_as::<_, IndexInfo>(
        r#"
        SELECT table_class.relname AS table_name,
               index_class.relname AS name,
               access_method.amname AS access_method,
               i.indisunique AS is_unique,
               i.indisvalid AS is_valid,
               i.indisready AS is_ready,
               i.indpred IS NOT NULL AS has_predicate,
               i.indnkeyatts AS key_columns,
               i.indnatts AS total_columns,
               COALESCE(
                   ARRAY_AGG(i.indoption[keys.ordinality::integer - 1] ORDER BY keys.ordinality)
                       FILTER (WHERE keys.ordinality <= i.indnkeyatts),
                   ARRAY[]::smallint[]
               ) AS options,
               COALESCE(
                   ARRAY_AGG(a.attname::text ORDER BY keys.ordinality)
                       FILTER (WHERE keys.ordinality <= i.indnkeyatts),
                   ARRAY[]::text[]
               ) AS columns,
               COALESCE(
                   ARRAY_AGG(COALESCE(coll.collname::text, 'default') ORDER BY keys.ordinality)
                       FILTER (WHERE keys.ordinality <= i.indnkeyatts),
                   ARRAY[]::text[]
               ) AS collations
        FROM pg_class table_class
        JOIN pg_namespace namespace ON namespace.oid = table_class.relnamespace
        JOIN pg_index i ON i.indrelid = table_class.oid
        JOIN pg_class index_class ON index_class.oid = i.indexrelid
        JOIN pg_am access_method ON access_method.oid = index_class.relam
        CROSS JOIN LATERAL unnest(i.indkey) WITH ORDINALITY AS keys(attnum, ordinality)
        JOIN pg_attribute a ON a.attrelid = table_class.oid AND a.attnum = keys.attnum
        LEFT JOIN pg_collation coll ON coll.oid = i.indcollation[keys.ordinality::integer - 1]
        WHERE table_class.relnamespace::bigint = $1
          AND index_class.relname = ANY($2)
        GROUP BY table_class.relname, index_class.relname, access_method.amname,
                 i.indisunique, i.indisvalid, i.indisready, i.indpred IS NOT NULL,
                 i.indnkeyatts, i.indnatts
        "#,
    )
    .bind(namespace.oid)
    .bind(
        expected_indexes
            .iter()
            .map(|index| index.name.to_owned())
            .collect::<Vec<_>>(),
    )
    .fetch_all(&mut *connection)
    .await
    .map_err(|source| SchemaError::sql("check indexes", source))?;
    for expected in &expected_indexes {
        let Some(actual) = indexes.iter().find(|index| index.name == expected.name) else {
            return Err(SchemaError::MigrationMismatch {
                detail: format!("required index {} is missing", expected.name),
            });
        };

        if !actual.matches(expected) {
            return Err(SchemaError::MigrationMismatch {
                detail: format!("required index {} is incompatible", expected.name),
            });
        }
    }

    Ok(())
}

#[derive(Debug, FromRow)]
struct NamespaceInfo {
    oid: i64,
    name: String,
}

async fn resolve_namespace(connection: &mut PgConnection) -> Result<NamespaceInfo, SchemaError> {
    query_as::<_, NamespaceInfo>(
        r#"
        SELECT oid::bigint AS oid, nspname AS name
        FROM pg_namespace
        WHERE nspname = current_schema()
        "#,
    )
    .fetch_optional(&mut *connection)
    .await
    .map_err(|source| SchemaError::sql("resolve current schema", source))?
    .ok_or_else(|| SchemaError::MigrationMismatch {
        detail: "the transaction has no resolvable current schema".to_owned(),
    })
}

#[derive(Clone, Copy)]
struct ColumnSpec {
    name: &'static str,
    data_type: &'static str,
    maximum_length: Option<i32>,
    nullable: bool,
    identity: bool,
    default_fragment: Option<&'static str>,
}

impl ColumnSpec {
    const fn required(
        name: &'static str,
        data_type: &'static str,
        maximum_length: Option<i32>,
    ) -> Self {
        Self {
            name,
            data_type,
            maximum_length,
            nullable: false,
            identity: false,
            default_fragment: None,
        }
    }

    const fn required_identity(
        name: &'static str,
        data_type: &'static str,
        maximum_length: Option<i32>,
    ) -> Self {
        Self {
            identity: true,
            ..Self::required(name, data_type, maximum_length)
        }
    }

    const fn optional(
        name: &'static str,
        data_type: &'static str,
        maximum_length: Option<i32>,
    ) -> Self {
        Self {
            nullable: true,
            ..Self::required(name, data_type, maximum_length)
        }
    }

    const fn required_with_default(
        name: &'static str,
        data_type: &'static str,
        maximum_length: Option<i32>,
        default_fragment: &'static str,
    ) -> Self {
        Self {
            default_fragment: Some(default_fragment),
            ..Self::required(name, data_type, maximum_length)
        }
    }
}

#[derive(Debug, FromRow)]
struct ColumnInfo {
    column_name: String,
    data_type: String,
    character_maximum_length: Option<i32>,
    is_nullable: String,
    column_default: Option<String>,
    is_identity: String,
    identity_generation: Option<String>,
}

async fn check_columns(
    connection: &mut PgConnection,
    schema_name: &str,
    table: &str,
    expected: &[ColumnSpec],
) -> Result<(), SchemaError> {
    let columns = query_as::<_, ColumnInfo>(
        r#"
        SELECT column_name, data_type, character_maximum_length,
               is_nullable, column_default, is_identity, identity_generation
        FROM information_schema.columns
        WHERE table_schema = $1 AND table_name = $2
        "#,
    )
    .bind(schema_name)
    .bind(table)
    .fetch_all(&mut *connection)
    .await
    .map_err(|source| SchemaError::sql("check columns", source))?;
    if let Some(column) = columns.iter().find(|column| {
        !expected
            .iter()
            .any(|specification| specification.name == column.column_name)
    }) {
        return Err(SchemaError::MigrationMismatch {
            detail: format!("unexpected column {}.{}", table, column.column_name),
        });
    }

    for specification in expected {
        let Some(column) = columns
            .iter()
            .find(|column| column.column_name == specification.name)
        else {
            return Err(SchemaError::MigrationMismatch {
                detail: format!(
                    "required column {}.{} is missing",
                    table, specification.name
                ),
            });
        };
        let default_matches = specification.default_fragment.is_none_or(|fragment| {
            column
                .column_default
                .as_deref()
                .is_some_and(|default| normalize_sql(default) == normalize_sql(fragment))
        });
        let identity_matches = if specification.identity {
            column.is_identity == "YES" && column.identity_generation.as_deref() == Some("ALWAYS")
        } else {
            column.is_identity == "NO"
        };
        if column.data_type != specification.data_type
            || column.character_maximum_length != specification.maximum_length
            || (column.is_nullable == "YES") != specification.nullable
            || !default_matches
            || !identity_matches
        {
            return Err(SchemaError::MigrationMismatch {
                detail: format!("column {}.{} is incompatible", table, specification.name),
            });
        }
    }
    Ok(())
}

#[derive(Debug, FromRow)]
struct ConstraintInfo {
    table_name: String,
    name: String,
    kind: String,
    columns: Vec<String>,
    referenced_table: Option<String>,
    referenced_columns: Vec<String>,
    delete_action: Option<String>,
    validated: bool,
    deferrable: bool,
    deferred: bool,
    definition: String,
}

struct ConstraintContract {
    name: &'static str,
    table_name: &'static str,
    kind: &'static str,
    columns: &'static [&'static str],
    referenced_table: Option<&'static str>,
    referenced_columns: &'static [&'static str],
    delete_action: Option<&'static str>,
    definition_variants: &'static [&'static str],
}

impl ConstraintContract {
    fn check(
        name: &'static str,
        table_name: &'static str,
        definition_variants: &'static [&'static str],
    ) -> Self {
        Self {
            name,
            table_name,
            kind: "c",
            columns: &[],
            referenced_table: None,
            referenced_columns: &[],
            delete_action: None,
            definition_variants,
        }
    }

    fn primary_key(
        name: &'static str,
        table_name: &'static str,
        columns: &'static [&'static str],
        definition_variants: &'static [&'static str],
    ) -> Self {
        Self {
            name,
            table_name,
            kind: "p",
            columns,
            referenced_table: None,
            referenced_columns: &[],
            delete_action: None,
            definition_variants,
        }
    }

    fn foreign_key(
        name: &'static str,
        table_name: &'static str,
        columns: &'static [&'static str],
        referenced_table: &'static str,
        referenced_columns: &'static [&'static str],
        delete_action: &'static str,
        definition_variants: &'static [&'static str],
    ) -> Self {
        Self {
            name,
            table_name,
            kind: "f",
            columns,
            referenced_table: Some(referenced_table),
            referenced_columns,
            delete_action: Some(delete_action),
            definition_variants,
        }
    }
}

impl ConstraintInfo {
    fn matches(&self, expected: &ConstraintContract) -> bool {
        let definition = normalize_sql(&self.definition);
        let columns_match = self.kind == "c"
            || self.columns
                == expected
                    .columns
                    .iter()
                    .map(|value| (*value).to_owned())
                    .collect::<Vec<_>>();
        self.table_name == expected.table_name
            && self.kind == expected.kind
            && columns_match
            && self.referenced_table.as_deref() == expected.referenced_table
            && self.referenced_columns
                == expected
                    .referenced_columns
                    .iter()
                    .map(|value| (*value).to_owned())
                    .collect::<Vec<_>>()
            && self.delete_action.as_deref() == expected.delete_action
            && self.validated
            && !self.deferrable
            && !self.deferred
            && expected
                .definition_variants
                .iter()
                .any(|variant| definition == normalize_sql(variant))
    }
}

#[derive(Debug, FromRow)]
struct IndexInfo {
    table_name: String,
    name: String,
    access_method: String,
    is_unique: bool,
    is_valid: bool,
    is_ready: bool,
    has_predicate: bool,
    key_columns: i16,
    total_columns: i16,
    options: Vec<i16>,
    columns: Vec<String>,
    collations: Vec<String>,
}

struct IndexContract {
    name: &'static str,
    table_name: &'static str,
    is_unique: bool,
    columns: &'static [&'static str],
    collations: Option<&'static [&'static str]>,
}

impl IndexContract {
    fn new(
        name: &'static str,
        table_name: &'static str,
        is_unique: bool,
        columns: &'static [&'static str],
        collations: Option<&'static [&'static str]>,
    ) -> Self {
        Self {
            name,
            table_name,
            is_unique,
            columns,
            collations,
        }
    }
}

impl IndexInfo {
    fn matches(&self, expected: &IndexContract) -> bool {
        self.table_name == expected.table_name
            && self.access_method == "btree"
            && self.is_unique == expected.is_unique
            && self.is_valid
            && self.is_ready
            && !self.has_predicate
            && self.key_columns == i16::try_from(expected.columns.len()).unwrap_or(i16::MAX)
            && self.total_columns == self.key_columns
            && self.options == vec![0; expected.columns.len()]
            && self.columns
                == expected
                    .columns
                    .iter()
                    .map(|value| (*value).to_owned())
                    .collect::<Vec<_>>()
            && expected.collations.is_none_or(|collations| {
                self.collations
                    == collations
                        .iter()
                        .map(|value| (*value).to_owned())
                        .collect::<Vec<_>>()
            })
    }
}

fn normalize_sql(value: &str) -> String {
    let mut value = value.to_ascii_lowercase();
    for cast in [
        "::character varying[]",
        "::character varying",
        "::timestamp with time zone",
        "::timestamp without time zone",
        "::double precision",
        "::numeric",
        "::bigint",
        "::integer",
        "::smallint",
        "::boolean",
        "::text[]",
        "::text",
    ] {
        value = value.replace(cast, "");
    }
    value
        .chars()
        .filter(|character| !character.is_ascii_whitespace())
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::migration::{
        MIGRATIONS, SCHEMA_VERSION, current_crate_version, marker_compatibility,
    };

    #[test]
    fn schema_marker_uses_the_shipped_compatibility_range() {
        let migration = MIGRATIONS
            .iter()
            .find(|migration| migration.version() == SCHEMA_VERSION)
            .expect("the v1 migration is shipped");
        let marker = SchemaMarker {
            schema_version: 1,
            minimum_crate_major: 0,
            minimum_crate_minor: 1,
            minimum_crate_patch: 0,
            rolling_compatible: false,
        };
        assert_eq!(marker_compatibility(&marker), Ok(migration.compatibility()));
        assert!(marker_matches_migration(&marker, *migration).is_ok());
        assert!(migration.compatibility().contains(current_crate_version()));

        let wrong_version = SchemaMarker {
            schema_version: 2,
            ..marker
        };
        assert!(marker_matches_migration(&wrong_version, *migration).is_err());

        let too_new = SchemaMarker {
            minimum_crate_major: 9,
            ..marker
        };
        assert!(marker_matches_migration(&too_new, *migration).is_err());

        let malformed = SchemaMarker {
            minimum_crate_minor: -1,
            ..marker
        };
        assert!(marker_compatibility(&malformed).is_err());
    }

    #[test]
    fn constraint_and_index_contracts_require_their_live_semantics() {
        let constraint = ConstraintInfo {
            table_name: "dovecote_events".to_owned(),
            name: "dovecote_events_pkey".to_owned(),
            kind: "p".to_owned(),
            columns: vec!["row_id".to_owned()],
            referenced_table: None,
            referenced_columns: Vec::new(),
            delete_action: None,
            validated: true,
            deferrable: false,
            deferred: false,
            definition: "PRIMARY KEY (row_id)".to_owned(),
        };
        let expected = ConstraintContract::primary_key(
            "dovecote_events_pkey",
            "dovecote_events",
            &["row_id"],
            &["PRIMARY KEY (row_id)"],
        );
        assert!(constraint.matches(&expected));
        let wrong_relation = ConstraintInfo {
            table_name: "other".to_owned(),
            ..constraint
        };
        assert!(!wrong_relation.matches(&expected));

        let check = ConstraintInfo {
            table_name: "dovecote_schema".to_owned(),
            name: "dovecote_schema_version_supported".to_owned(),
            kind: "c".to_owned(),
            columns: vec!["schema_version".to_owned()],
            referenced_table: None,
            referenced_columns: Vec::new(),
            delete_action: None,
            validated: true,
            deferrable: false,
            deferred: false,
            definition: "CHECK ((schema_version = 1))".to_owned(),
        };
        let expected_check = ConstraintContract::check(
            "dovecote_schema_version_supported",
            "dovecote_schema",
            &["CHECK ((schema_version = 1))"],
        );
        assert!(check.matches(&expected_check));

        let index = IndexInfo {
            table_name: "dovecote_events".to_owned(),
            name: "dovecote_events_source_event_id".to_owned(),
            access_method: "btree".to_owned(),
            is_unique: true,
            is_valid: true,
            is_ready: true,
            has_predicate: false,
            key_columns: 2,
            total_columns: 2,
            options: vec![0, 0],
            columns: vec!["source".to_owned(), "event_id".to_owned()],
            collations: vec!["C".to_owned(), "C".to_owned()],
        };
        let expected_index = IndexContract::new(
            "dovecote_events_source_event_id",
            "dovecote_events",
            true,
            &["source", "event_id"],
            Some(&["C", "C"]),
        );
        assert!(index.matches(&expected_index));
        let wrong_order = IndexInfo {
            options: vec![1, 0],
            ..index
        };
        assert!(!wrong_order.matches(&expected_index));
    }

    #[test]
    fn pg17_constraint_renderings_match_the_shipped_contracts() {
        let fixtures = [(
            ConstraintInfo {
                table_name: "dovecote_events".to_owned(),
                name: "dovecote_events_identity_size".to_owned(),
                kind: "c".to_owned(),
                columns: vec!["source".to_owned(), "event_id".to_owned()],
                referenced_table: None,
                referenced_columns: Vec::new(),
                delete_action: None,
                validated: true,
                deferrable: false,
                deferred: false,
                definition: "CHECK (((octet_length((source)::text) + octet_length((event_id)::text)) <= 2048))".to_owned(),
            },
            ConstraintContract::check(
                "dovecote_events_identity_size",
                "dovecote_events",
                &["CHECK (((octet_length((source)) + octet_length((event_id))) <= 2048))"],
            ),
        )];

        for (actual, expected) in fixtures {
            assert!(actual.matches(&expected));
        }
    }

    #[test]
    fn sql_normalization_preserves_boolean_grouping() {
        assert_ne!(
            normalize_sql("CHECK ((left IS NULL OR right IS NULL))"),
            normalize_sql("CHECK (((left IS NULL OR right IS NULL)))")
        );
        assert_ne!(
            normalize_sql("CHECK (((left IS NULL) OR (right IS NULL)))"),
            normalize_sql("CHECK (((left IS NULL) AND (right IS NULL)))")
        );
    }
}