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

dovecote_sqlx_postgres/schema/
mod.rs
