dovecote_sqlx_postgres/schema/
mod.rs

1//! `PostgreSQL` catalog verification for the installed Dovecote schema.
2
3mod catalog;
4mod contracts;
5mod normalization;
6#[cfg(test)]
7mod tests;
8
9use crate::{
10    error::SchemaError,
11    migration::{SchemaMarker, current_migration, marker_matches_migration},
12};
13use catalog::{ColumnSpec, check_columns, resolve_namespace};
14use contracts::{ConstraintInfo, IndexInfo};
15use sqlx::{PgConnection, PgPool, query_as};
16
17/// Verifies the tables and all columns required by schema version 2.
18///
19/// # Errors
20/// Returns an error for an unsupported backend, missing or incompatible
21/// migration markers, tables, constraints or indexes, or failed catalog reads.
22pub async fn check_schema(pool: &PgPool) -> Result<(), SchemaError> {
23    let mut connection = pool
24        .acquire()
25        .await
26        .map_err(|source| SchemaError::sql("acquire schema-check connection", source))?;
27    check_schema_connection(&mut connection).await
28}
29
30/// Performs the complete schema check on an already-owned connection.
31pub(crate) async fn check_schema_connection(
32    connection: &mut PgConnection,
33) -> Result<(), SchemaError> {
34    let namespace = resolve_namespace(connection).await?;
35
36    let marker_columns = [
37        ColumnSpec::required("schema_version", "integer", None),
38        ColumnSpec::required("minimum_crate_major", "smallint", None),
39        ColumnSpec::required("minimum_crate_minor", "smallint", None),
40        ColumnSpec::required("minimum_crate_patch", "smallint", None),
41        ColumnSpec::required("rolling_compatible", "boolean", None),
42    ];
43    check_columns(
44        connection,
45        &namespace.name,
46        "dovecote_schema",
47        &marker_columns,
48    )
49    .await?;
50    let markers = query_as::<_, SchemaMarker>(
51        r"
52        SELECT schema_version, minimum_crate_major, minimum_crate_minor,
53               minimum_crate_patch, rolling_compatible
54        FROM dovecote_schema
55        ORDER BY schema_version DESC
56        ",
57    )
58    .fetch_all(&mut *connection)
59    .await
60    .map_err(|source| SchemaError::sql("check schema marker", source))?;
61    if markers.len() != 1 {
62        return Err(SchemaError::MigrationMismatch {
63            detail: format!(
64                "expected exactly one schema marker row, found {}",
65                markers.len()
66            ),
67        });
68    }
69
70    let marker = &markers[0];
71    let migration =
72        current_migration().map_err(|detail| SchemaError::MigrationMismatch { detail })?;
73    if let Err(detail) = marker_matches_migration(marker, migration) {
74        return Err(SchemaError::MigrationMismatch { detail });
75    }
76
77    let event_columns = [
78        ColumnSpec::required_identity("row_id", "bigint", None),
79        ColumnSpec::required("tenant_id", "character varying", Some(255)),
80        ColumnSpec::required("stream", "character varying", Some(255)),
81        ColumnSpec::required("specversion", "character varying", Some(8)),
82        ColumnSpec::required("event_id", "character varying", Some(1024)),
83        ColumnSpec::required("source", "character varying", Some(2048)),
84        ColumnSpec::required("event_type", "character varying", Some(1024)),
85        ColumnSpec::optional("subject", "character varying", Some(2048)),
86        ColumnSpec::optional("occurred_at", "timestamp with time zone", None),
87        ColumnSpec::optional("datacontenttype", "character varying", Some(255)),
88        ColumnSpec::optional("dataschema", "character varying", Some(2048)),
89        ColumnSpec::optional("partitionkey", "character varying", Some(255)),
90        ColumnSpec::required_with_default("extensions", "text", None, "'{}'::text"),
91        ColumnSpec::optional("data_kind", "character varying", Some(6)),
92        ColumnSpec::optional("data", "bytea", None),
93        ColumnSpec::required_with_default(
94            "enqueued_at",
95            "timestamp with time zone",
96            None,
97            "current_timestamp",
98        ),
99    ];
100    let delivery_columns = [
101        ColumnSpec::required("event_row_id", "bigint", None),
102        ColumnSpec::required("tenant_id", "character varying", Some(255)),
103        ColumnSpec::required("state", "character varying", Some(12)),
104        ColumnSpec::required_with_default(
105            "available_at",
106            "timestamp with time zone",
107            None,
108            "current_timestamp",
109        ),
110        ColumnSpec::required_with_default("attempts", "bigint", None, "0"),
111        ColumnSpec::optional("claim_token", "bytea", None),
112        ColumnSpec::optional("claimed_by", "character varying", Some(255)),
113        ColumnSpec::optional("claim_expires_at", "timestamp with time zone", None),
114        ColumnSpec::optional("last_failure_code", "character varying", Some(128)),
115        ColumnSpec::optional("last_failure_detail", "character varying", Some(2048)),
116        ColumnSpec::optional("delivered_at", "timestamp with time zone", None),
117        ColumnSpec::optional("quarantined_at", "timestamp with time zone", None),
118        ColumnSpec::optional("quarantine_reason", "character varying", Some(2048)),
119    ];
120    check_columns(
121        connection,
122        &namespace.name,
123        "dovecote_events",
124        &event_columns,
125    )
126    .await?;
127    check_columns(
128        connection,
129        &namespace.name,
130        "dovecote_deliveries",
131        &delivery_columns,
132    )
133    .await?;
134
135    let expected_constraints = contracts::expected_constraints();
136
137    let constraints = query_as::<_, ConstraintInfo>(
138        r"
139        SELECT table_class.relname AS table_name,
140               constraint_class.conname AS name,
141               constraint_class.contype::text AS kind,
142               ARRAY(
143                   SELECT attribute.attname::text
144                   FROM unnest(constraint_class.conkey) WITH ORDINALITY AS key(attnum, ordinality)
145                   JOIN pg_attribute attribute
146                     ON attribute.attrelid = constraint_class.conrelid
147                    AND attribute.attnum = key.attnum
148                   ORDER BY key.ordinality
149               ) AS columns,
150               parent_class.relname AS referenced_table,
151               ARRAY(
152                   SELECT attribute.attname::text
153                   FROM unnest(constraint_class.confkey) WITH ORDINALITY AS key(attnum, ordinality)
154                   JOIN pg_attribute attribute
155                     ON attribute.attrelid = constraint_class.confrelid
156                    AND attribute.attnum = key.attnum
157                   ORDER BY key.ordinality
158               ) AS referenced_columns,
159               CASE WHEN constraint_class.contype = 'f'
160                    THEN constraint_class.confdeltype::text END AS delete_action,
161               constraint_class.convalidated AS validated,
162               constraint_class.condeferrable AS deferrable,
163               constraint_class.condeferred AS deferred,
164               pg_get_constraintdef(constraint_class.oid) AS definition
165        FROM pg_constraint constraint_class
166        JOIN pg_class table_class ON table_class.oid = constraint_class.conrelid
167        LEFT JOIN pg_class parent_class ON parent_class.oid = constraint_class.confrelid
168        WHERE table_class.relnamespace::bigint = $1
169          AND (constraint_class.confrelid = 0 OR parent_class.relnamespace::bigint = $1)
170          AND table_class.relname IN ('dovecote_schema', 'dovecote_events', 'dovecote_deliveries')
171        ",
172    )
173    .bind(namespace.oid)
174    .fetch_all(&mut *connection)
175    .await
176    .map_err(|source| SchemaError::sql("check constraints", source))?;
177    for expected in &expected_constraints {
178        let Some(actual) = constraints
179            .iter()
180            .find(|constraint| constraint.name == expected.name)
181        else {
182            return Err(SchemaError::MigrationMismatch {
183                detail: format!("required constraint {} is missing", expected.name),
184            });
185        };
186
187        if !actual.matches(expected) {
188            return Err(SchemaError::MigrationMismatch {
189                detail: format!("required constraint {} is incompatible", expected.name),
190            });
191        }
192    }
193
194    let expected_constraint_names = expected_constraints
195        .iter()
196        .map(|constraint| constraint.name)
197        .collect::<Vec<_>>();
198    if let Some(unexpected) = constraints
199        .iter()
200        .find(|actual| !is_expected_name(&actual.name, &expected_constraint_names))
201    {
202        return Err(SchemaError::MigrationMismatch {
203            detail: format!("unexpected constraint {}", unexpected.name),
204        });
205    }
206
207    let expected_indexes = contracts::expected_indexes();
208
209    let indexes = query_as::<_, IndexInfo>(
210        r"
211        SELECT table_class.relname AS table_name,
212               index_class.relname AS name,
213               access_method.amname AS access_method,
214               i.indisunique AS is_unique,
215               i.indisvalid AS is_valid,
216               i.indisready AS is_ready,
217               i.indpred IS NOT NULL AS has_predicate,
218               i.indnkeyatts AS key_columns,
219               i.indnatts AS total_columns,
220               COALESCE(
221                   ARRAY_AGG(i.indoption[keys.ordinality::integer - 1] ORDER BY keys.ordinality)
222                       FILTER (WHERE keys.ordinality <= i.indnkeyatts),
223                   ARRAY[]::smallint[]
224               ) AS options,
225               COALESCE(
226                   ARRAY_AGG(a.attname::text ORDER BY keys.ordinality)
227                       FILTER (WHERE keys.ordinality <= i.indnkeyatts),
228                   ARRAY[]::text[]
229               ) AS columns,
230               COALESCE(
231                   ARRAY_AGG(COALESCE(coll.collname::text, 'default') ORDER BY keys.ordinality)
232                       FILTER (WHERE keys.ordinality <= i.indnkeyatts),
233                   ARRAY[]::text[]
234               ) AS collations
235        FROM pg_class table_class
236        JOIN pg_namespace namespace ON namespace.oid = table_class.relnamespace
237        JOIN pg_index i ON i.indrelid = table_class.oid
238        JOIN pg_class index_class ON index_class.oid = i.indexrelid
239        JOIN pg_am access_method ON access_method.oid = index_class.relam
240        CROSS JOIN LATERAL unnest(i.indkey) WITH ORDINALITY AS keys(attnum, ordinality)
241        JOIN pg_attribute a ON a.attrelid = table_class.oid AND a.attnum = keys.attnum
242        LEFT JOIN pg_collation coll ON coll.oid = i.indcollation[keys.ordinality::integer - 1]
243        WHERE table_class.relnamespace::bigint = $1
244          AND table_class.relname IN ('dovecote_schema', 'dovecote_events', 'dovecote_deliveries')
245          AND NOT EXISTS (
246              SELECT 1
247              FROM pg_constraint constraint_class
248              WHERE constraint_class.conindid = index_class.oid
249          )
250        GROUP BY table_class.relname, index_class.relname, access_method.amname,
251                 i.indisunique, i.indisvalid, i.indisready, i.indpred IS NOT NULL,
252                 i.indnkeyatts, i.indnatts
253        ",
254    )
255    .bind(namespace.oid)
256    .fetch_all(&mut *connection)
257    .await
258    .map_err(|source| SchemaError::sql("check indexes", source))?;
259    for expected in &expected_indexes {
260        let Some(actual) = indexes.iter().find(|index| index.name == expected.name) else {
261            return Err(SchemaError::MigrationMismatch {
262                detail: format!("required index {} is missing", expected.name),
263            });
264        };
265
266        if !actual.matches(expected) {
267            return Err(SchemaError::MigrationMismatch {
268                detail: format!("required index {} is incompatible", expected.name),
269            });
270        }
271    }
272
273    let expected_index_names = expected_indexes
274        .iter()
275        .map(|index| index.name)
276        .collect::<Vec<_>>();
277    if let Some(unexpected) = indexes
278        .iter()
279        .find(|actual| !is_expected_name(&actual.name, &expected_index_names))
280    {
281        return Err(SchemaError::MigrationMismatch {
282            detail: format!("unexpected index {}", unexpected.name),
283        });
284    }
285
286    Ok(())
287}
288
289fn is_expected_name(name: &str, expected: &[&str]) -> bool {
290    expected.contains(&name)
291}
dovecote_sqlx_postgres/schema/mod.rs

dovecote_sqlx_postgres/schema/
mod.rs
