dovecote_sqlx_sqlite/

schema.rs

//! `SQLite` schema verification.

use crate::{
    error::SchemaError,
    migration::{current_migration, migration_is_usable},
};
use sqlx::{FromRow, Row, SqliteConnection, SqlitePool, query, query_as, query_scalar};

#[derive(Debug, FromRow)]
struct SchemaMarker {
    schema_version: i64,
    minimum_crate_major: i64,
    minimum_crate_minor: i64,
    minimum_crate_patch: i64,
    rolling_compatible: i64,
}

/// Verifies the exact v2 table shape, constraints, indexes, and foreign key of
/// the installed schema. It never applies a migration.
///
/// # Errors
/// Returns an error for an unsupported backend, missing or incompatible
/// migration markers, tables, constraints or indexes, or failed catalog reads.
pub async fn check_schema(pool: &SqlitePool) -> 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 SqliteConnection,
) -> Result<(), SchemaError> {
    let enabled: i64 = query_scalar("PRAGMA foreign_keys")
        .fetch_one(&mut *connection)
        .await
        .map_err(|source| SchemaError::sql("check foreign-key enforcement", source))?;
    if enabled != 1 {
        return Err(mismatch("foreign-key enforcement is disabled"));
    }

    let migration = current_migration().map_err(mismatch)?;
    migration_is_usable(migration).map_err(mismatch)?;
    check_schema_marker(connection, migration).await?;

    check_columns(
        connection,
        "dovecote_events",
        &[
            ColumnSpec::required("row_id", "INTEGER", true),
            ColumnSpec::required("tenant_id", "TEXT", false),
            ColumnSpec::required("stream", "TEXT", false),
            ColumnSpec::required("specversion", "TEXT", false),
            ColumnSpec::required("event_id", "TEXT", false),
            ColumnSpec::required("source", "TEXT", false),
            ColumnSpec::required("event_type", "TEXT", false),
            ColumnSpec::optional("subject", "TEXT", false),
            ColumnSpec::optional("occurred_at", "TEXT", false),
            ColumnSpec::optional("datacontenttype", "TEXT", false),
            ColumnSpec::optional("dataschema", "TEXT", false),
            ColumnSpec::optional("partitionkey", "TEXT", false),
            ColumnSpec::required("extensions", "TEXT", false),
            ColumnSpec::optional("data_kind", "TEXT", false),
            ColumnSpec::optional("data", "BLOB", false),
            ColumnSpec::required("enqueued_at", "TEXT", false),
        ],
    )
    .await?;
    check_columns(
        connection,
        "dovecote_deliveries",
        &[
            ColumnSpec::required("event_row_id", "INTEGER", true),
            ColumnSpec::required("tenant_id", "TEXT", false),
            ColumnSpec::required("state", "TEXT", false),
            ColumnSpec::required("available_at", "TEXT", false),
            ColumnSpec::required("attempts", "INTEGER", false),
            ColumnSpec::optional("claim_token", "BLOB", false),
            ColumnSpec::optional("claimed_by", "TEXT", false),
            ColumnSpec::optional("claim_expires_at", "TEXT", false),
            ColumnSpec::optional("last_failure_code", "TEXT", false),
            ColumnSpec::optional("last_failure_detail", "TEXT", false),
            ColumnSpec::optional("delivered_at", "TEXT", false),
            ColumnSpec::optional("quarantined_at", "TEXT", false),
            ColumnSpec::optional("quarantine_reason", "TEXT", false),
        ],
    )
    .await?;

    let sources = query_as::<_, TableSource>(
        "SELECT name, sql FROM sqlite_master WHERE type = 'table' AND name IN ('dovecote_schema', 'dovecote_events', 'dovecote_deliveries')",
    ).fetch_all(&mut *connection).await
        .map_err(|source| SchemaError::sql("read schema definitions", source))?;
    for name in ["dovecote_schema", "dovecote_events", "dovecote_deliveries"] {
        if !sources.iter().any(|source| source.name == name) {
            return Err(mismatch(format!("required table {name} is missing")));
        }
    }

    for name in ["dovecote_schema", "dovecote_events", "dovecote_deliveries"] {
        let source = sources
            .iter()
            .find(|source| source.name == name)
            .expect("checked above");
        let expected = expected_table_source(migration.sql(), name).map_err(mismatch)?;
        if normalize_sql(&source.sql) != normalize_sql(&expected) {
            return Err(mismatch(format!(
                "table {name} definition is incompatible with schema version {}",
                migration.version()
            )));
        }
    }

    // TEMP triggers and indexes can target a main-schema table, so inspect
    // both catalogs.  Leaving sqlite_temp_master out would allow a caller to
    // add an unreviewed trigger that changes durable invariants for this
    // connection while the main schema still appears exact.
    let extra_objects: Vec<SchemaObject> = query_as(
        "SELECT type, name, COALESCE(tbl_name, '') AS tbl_name FROM sqlite_master WHERE (name LIKE 'dovecote_%' OR tbl_name IN ('dovecote_events', 'dovecote_deliveries')) AND NOT (type = 'table' AND name IN ('dovecote_schema', 'dovecote_events', 'dovecote_deliveries')) AND NOT (type = 'index' AND name IN ('dovecote_events_tenant_source_event_id', 'dovecote_events_tenant_row', 'dovecote_deliveries_claimable', 'dovecote_deliveries_expired_claims', 'sqlite_autoindex_dovecote_events_1')) UNION ALL SELECT type, name, COALESCE(tbl_name, '') AS tbl_name FROM sqlite_temp_master WHERE name LIKE 'dovecote_%' OR tbl_name IN ('dovecote_events', 'dovecote_deliveries')",
    )
    .fetch_all(&mut *connection)
    .await
    .map_err(|source| SchemaError::sql("check schema object isolation", source))?;
    if let Some(object) = extra_objects.first() {
        return Err(mismatch(format!(
            "unsupported SQLite schema object {} {} on {}",
            object.object_type, object.name, object.table_name
        )));
    }

    check_index(
        connection,
        "dovecote_events",
        "dovecote_events_tenant_source_event_id",
        true,
        &["tenant_id", "source", "event_id"],
        migration.sql(),
    )
    .await?;
    check_index(
        connection,
        "dovecote_events",
        "dovecote_events_tenant_row",
        false,
        &["tenant_id", "row_id"],
        migration.sql(),
    )
    .await?;
    check_index(
        connection,
        "dovecote_deliveries",
        "dovecote_deliveries_claimable",
        false,
        &["tenant_id", "state", "available_at", "event_row_id"],
        migration.sql(),
    )
    .await?;
    check_index(
        connection,
        "dovecote_deliveries",
        "dovecote_deliveries_expired_claims",
        false,
        &["tenant_id", "state", "claim_expires_at", "event_row_id"],
        migration.sql(),
    )
    .await?;
    check_foreign_key(connection).await?;
    let violations = query("PRAGMA foreign_key_check")
        .fetch_all(&mut *connection)
        .await
        .map_err(|source| SchemaError::sql("check foreign-key integrity", source))?;
    if !violations.is_empty() {
        return Err(mismatch("installed schema contains foreign-key violations"));
    }
    Ok(())
}

async fn check_schema_marker(
    connection: &mut SqliteConnection,
    migration: crate::migration::Migration,
) -> Result<(), SchemaError> {
    let markers = query_as::<_, SchemaMarker>(
        "SELECT schema_version, minimum_crate_major, minimum_crate_minor, minimum_crate_patch, rolling_compatible FROM dovecote_schema",
    )
    .fetch_all(&mut *connection)
    .await
    .map_err(|source| SchemaError::sql("check schema marker", source))?;
    if markers.len() != 1 {
        return Err(mismatch(format!(
            "expected exactly one schema marker row, found {}",
            markers.len()
        )));
    }

    let marker = &markers[0];
    let minimum = migration.compatibility().minimum();
    if marker.schema_version != i64::from(migration.version())
        || marker.minimum_crate_major != i64::from(minimum.major())
        || marker.minimum_crate_minor != i64::from(minimum.minor())
        || marker.minimum_crate_patch != i64::from(minimum.patch())
        || marker.rolling_compatible != i64::from(migration.rolling_compatible())
    {
        return Err(mismatch("schema marker is incompatible with this adapter"));
    }
    Ok(())
}

fn mismatch(detail: impl Into<String>) -> SchemaError {
    SchemaError::MigrationMismatch {
        detail: detail.into(),
    }
}

#[derive(Clone, Copy)]
struct ColumnSpec {
    name: &'static str,
    kind: &'static str,
    primary_key: bool,
    not_null: bool,
}
impl ColumnSpec {
    const fn required(name: &'static str, kind: &'static str, primary_key: bool) -> Self {
        Self {
            name,
            kind,
            primary_key,
            not_null: !primary_key,
        }
    }
    const fn optional(name: &'static str, kind: &'static str, primary_key: bool) -> Self {
        Self {
            name,
            kind,
            primary_key,
            not_null: false,
        }
    }
}

#[derive(Debug, FromRow)]
struct TableSource {
    name: String,
    sql: String,
}

#[derive(Debug, FromRow)]
struct SchemaObject {
    #[sqlx(rename = "type")]
    object_type: String,
    name: String,
    #[sqlx(rename = "tbl_name")]
    table_name: String,
}

async fn check_columns(
    connection: &mut SqliteConnection,
    table: &str,
    expected: &[ColumnSpec],
) -> Result<(), SchemaError> {
    let sql = sqlx::AssertSqlSafe(format!("PRAGMA table_info({table})"));
    let rows = query(sql)
        .fetch_all(&mut *connection)
        .await
        .map_err(|source| SchemaError::sql("check table columns", source))?;
    for spec in expected {
        let Some(row) = rows
            .iter()
            .find(|row| row.try_get::<String, _>("name").ok().as_deref() == Some(spec.name))
        else {
            return Err(mismatch(format!(
                "required column {table}.{} is missing",
                spec.name
            )));
        };

        let kind = row
            .try_get::<String, _>("type")
            .map_err(|_| mismatch(format!("column {table}.{} has no type", spec.name)))?;
        if !kind.eq_ignore_ascii_case(spec.kind) {
            return Err(mismatch(format!(
                "column {table}.{} has type {kind}, expected {}",
                spec.name, spec.kind
            )));
        }

        let pk = row.try_get::<i64, _>("pk").unwrap_or_default();
        if spec.primary_key && pk != 1 {
            return Err(mismatch(format!(
                "column {table}.{} is not the primary key",
                spec.name
            )));
        }

        let not_null = row.try_get::<i64, _>("notnull").unwrap_or_default() != 0;
        if spec.not_null && !not_null {
            return Err(mismatch(format!(
                "column {table}.{} must be NOT NULL",
                spec.name
            )));
        }
    }

    Ok(())
}

async fn check_index(
    connection: &mut SqliteConnection,
    table: &str,
    expected_name: &str,
    unique: bool,
    columns: &[&str],
    migration: &str,
) -> Result<(), SchemaError> {
    let sql = sqlx::AssertSqlSafe(format!("PRAGMA index_list({table})"));
    let indexes = query(sql)
        .fetch_all(&mut *connection)
        .await
        .map_err(|source| SchemaError::sql("check schema indexes", source))?;
    let Some(index) = indexes
        .iter()
        .find(|row| row.try_get::<String, _>("name").ok().as_deref() == Some(expected_name))
    else {
        return Err(mismatch(format!(
            "required index {expected_name} is missing"
        )));
    };

    let actual_unique = index.try_get::<i64, _>("unique").unwrap_or_default() != 0;
    if actual_unique != unique {
        return Err(mismatch(format!(
            "index {expected_name} uniqueness is incompatible"
        )));
    }

    let source: Option<String> =
        query_scalar("SELECT sql FROM sqlite_master WHERE type = 'index' AND name = ?")
            .bind(expected_name)
            .fetch_optional(&mut *connection)
            .await
            .map_err(|source| SchemaError::sql("read schema index definition", source))?;
    let Some(source) = source else {
        return Err(mismatch(format!("index {expected_name} has no definition")));
    };

    let expected = expected_index_source(migration, expected_name).map_err(mismatch)?;
    if normalize_sql(&source) != normalize_sql(&expected) {
        return Err(mismatch(format!(
            "index {expected_name} definition is incompatible"
        )));
    }

    let info_sql = sqlx::AssertSqlSafe(format!("PRAGMA index_info({expected_name})"));
    let info = query(info_sql)
        .fetch_all(&mut *connection)
        .await
        .map_err(|source| SchemaError::sql("read schema index columns", source))?;
    let actual = info
        .iter()
        .filter_map(|row| row.try_get::<String, _>("name").ok())
        .collect::<Vec<_>>();
    if actual
        != columns
            .iter()
            .map(|column| (*column).to_owned())
            .collect::<Vec<_>>()
    {
        return Err(mismatch(format!(
            "index {expected_name} columns are incompatible"
        )));
    }

    if expected_name == "dovecote_events_tenant_source_event_id" {
        let xinfo_sql = sqlx::AssertSqlSafe(format!("PRAGMA index_xinfo({expected_name})"));
        let xinfo = query(xinfo_sql)
            .fetch_all(&mut *connection)
            .await
            .map_err(|source| SchemaError::sql("read identity index collation", source))?;
        let collations = xinfo
            .iter()
            .filter_map(|row| row.try_get::<i64, _>("key").ok().filter(|key| *key != 0))
            .zip(
                xinfo
                    .iter()
                    .filter_map(|row| row.try_get::<String, _>("coll").ok()),
            )
            .map(|(_, collation)| collation)
            .collect::<Vec<_>>();
        if collations
            != [
                "BINARY".to_owned(),
                "BINARY".to_owned(),
                "BINARY".to_owned(),
            ]
        {
            return Err(mismatch("identity index collation is not BINARY"));
        }
    }

    Ok(())
}

async fn check_foreign_key(connection: &mut SqliteConnection) -> Result<(), SchemaError> {
    let rows = query("PRAGMA foreign_key_list(dovecote_deliveries)")
        .fetch_all(&mut *connection)
        .await
        .map_err(|source| SchemaError::sql("check delivery foreign key", source))?;
    let matching = rows
        .iter()
        .filter(|row| row.try_get::<String, _>("table").ok().as_deref() == Some("dovecote_events"))
        .collect::<Vec<_>>();
    if matching.len() != 2 {
        return Err(mismatch("delivery foreign key is missing"));
    }

    let columns = matching
        .iter()
        .map(|row| {
            (
                row.try_get::<String, _>("from").unwrap_or_default(),
                row.try_get::<String, _>("to").unwrap_or_default(),
            )
        })
        .collect::<Vec<_>>();
    if !columns.contains(&("tenant_id".to_owned(), "tenant_id".to_owned()))
        || !columns.contains(&("event_row_id".to_owned(), "row_id".to_owned()))
    {
        return Err(mismatch("delivery foreign key is incompatible"));
    }
    Ok(())
}

fn expected_table_source(migration: &str, table: &str) -> Result<String, String> {
    let needle = format!("CREATE TABLE {table}");
    let start = migration
        .find(&needle)
        .ok_or_else(|| format!("migration does not define {table}"))?;
    let mut depth = 0_u32;
    let mut quoted = false;
    for (offset, character) in migration[start..].char_indices() {
        match character {
            '\'' => quoted = !quoted,
            '(' if !quoted => depth = depth.saturating_add(1),
            ')' if !quoted => depth = depth.saturating_sub(1),
            ';' if !quoted && depth == 0 => return Ok(migration[start..start + offset].to_owned()),
            _ => {}
        }
    }
    Err(format!("migration statement for {table} is unterminated"))
}

fn expected_index_source(migration: &str, index: &str) -> Result<String, String> {
    let needle = if migration.contains(&format!("CREATE UNIQUE INDEX {index}")) {
        format!("CREATE UNIQUE INDEX {index}")
    } else {
        format!("CREATE INDEX {index}")
    };
    let start = migration
        .find(&needle)
        .ok_or_else(|| format!("migration does not define {index}"))?;
    let end = migration[start..]
        .find(';')
        .map(|offset| start + offset)
        .ok_or_else(|| format!("migration statement for {index} is unterminated"))?;
    Ok(migration[start..end].to_owned())
}

fn normalize_sql(value: &str) -> String {
    let mut normalized = String::with_capacity(value.len());
    let mut in_string = false;
    for character in value.chars() {
        match (character, in_string) {
            ('\'', _) => {
                in_string = !in_string;
                normalized.push(character);
            }
            ('"', false) => {
                // SQLite quotes a rebuilt table name after ALTER TABLE RENAME;
                // identifier quoting does not change the table contract.
            }
            (character, true) if !character.is_ascii_whitespace() => {
                normalized.push(character);
            }
            (character, false) if !character.is_ascii_whitespace() => {
                normalized.extend(character.to_lowercase());
            }
            _ => {}
        }
    }
    normalized
}