1use crate::{
4 error::SchemaError,
5 migration::{SchemaMarker, current_migration, marker_matches_migration},
6};
7use sqlx::{FromRow, PgConnection, PgPool, query_as};
8
9pub async fn check_schema(pool: &PgPool) -> Result<(), SchemaError> {
11 let mut connection = pool
12 .acquire()
13 .await
14 .map_err(|source| SchemaError::sql("acquire schema-check connection", source))?;
15 check_schema_connection(&mut connection).await
16}
17
18pub(crate) async fn check_schema_connection(
20 connection: &mut PgConnection,
21) -> Result<(), SchemaError> {
22 let namespace = resolve_namespace(connection).await?;
23
24 let marker_columns = [
25 ColumnSpec::required("schema_version", "integer", None),
26 ColumnSpec::required("minimum_crate_major", "smallint", None),
27 ColumnSpec::required("minimum_crate_minor", "smallint", None),
28 ColumnSpec::required("minimum_crate_patch", "smallint", None),
29 ColumnSpec::required("rolling_compatible", "boolean", None),
30 ];
31 check_columns(
32 connection,
33 &namespace.name,
34 "dovecote_schema",
35 &marker_columns,
36 )
37 .await?;
38 let marker = query_as::<_, SchemaMarker>(
39 r#"
40 SELECT schema_version, minimum_crate_major, minimum_crate_minor,
41 minimum_crate_patch, rolling_compatible
42 FROM dovecote_schema
43 ORDER BY schema_version DESC
44 LIMIT 1
45 "#,
46 )
47 .fetch_optional(&mut *connection)
48 .await
49 .map_err(|source| SchemaError::sql("check schema marker", source))?
50 .ok_or_else(|| SchemaError::MigrationMismatch {
51 detail: "schema marker is missing".to_owned(),
52 })?;
53 let migration =
54 current_migration().map_err(|detail| SchemaError::MigrationMismatch { detail })?;
55 if let Err(detail) = marker_matches_migration(&marker, migration) {
56 return Err(SchemaError::MigrationMismatch { detail });
57 }
58
59 let event_columns = [
60 ColumnSpec::required_identity("row_id", "bigint", None),
61 ColumnSpec::required("stream", "character varying", Some(255)),
62 ColumnSpec::required("specversion", "character varying", Some(8)),
63 ColumnSpec::required("event_id", "character varying", Some(1024)),
64 ColumnSpec::required("source", "character varying", Some(2048)),
65 ColumnSpec::required("event_type", "character varying", Some(1024)),
66 ColumnSpec::optional("subject", "character varying", Some(2048)),
67 ColumnSpec::optional("occurred_at", "timestamp with time zone", None),
68 ColumnSpec::optional("datacontenttype", "character varying", Some(255)),
69 ColumnSpec::optional("dataschema", "character varying", Some(2048)),
70 ColumnSpec::optional("partitionkey", "character varying", Some(255)),
71 ColumnSpec::required_with_default("extensions", "text", None, "'{}'::text"),
72 ColumnSpec::optional("data_kind", "character varying", Some(6)),
73 ColumnSpec::optional("data", "bytea", None),
74 ColumnSpec::required_with_default(
75 "enqueued_at",
76 "timestamp with time zone",
77 None,
78 "current_timestamp",
79 ),
80 ];
81 let delivery_columns = [
82 ColumnSpec::required("event_row_id", "bigint", None),
83 ColumnSpec::required("state", "character varying", Some(12)),
84 ColumnSpec::required_with_default(
85 "available_at",
86 "timestamp with time zone",
87 None,
88 "current_timestamp",
89 ),
90 ColumnSpec::required_with_default("attempts", "bigint", None, "0"),
91 ColumnSpec::optional("claim_token", "bytea", None),
92 ColumnSpec::optional("claimed_by", "character varying", Some(255)),
93 ColumnSpec::optional("claim_expires_at", "timestamp with time zone", None),
94 ColumnSpec::optional("last_failure_code", "character varying", Some(128)),
95 ColumnSpec::optional("last_failure_detail", "character varying", Some(2048)),
96 ColumnSpec::optional("delivered_at", "timestamp with time zone", None),
97 ColumnSpec::optional("quarantined_at", "timestamp with time zone", None),
98 ColumnSpec::optional("quarantine_reason", "character varying", Some(2048)),
99 ];
100 check_columns(
101 connection,
102 &namespace.name,
103 "dovecote_events",
104 &event_columns,
105 )
106 .await?;
107 check_columns(
108 connection,
109 &namespace.name,
110 "dovecote_deliveries",
111 &delivery_columns,
112 )
113 .await?;
114
115 let expected_constraints = [
116 ConstraintContract::check(
117 "dovecote_schema_version_supported",
118 "dovecote_schema",
119 &["CHECK ((schema_version = 1))"],
120 ),
121 ConstraintContract::check(
122 "dovecote_schema_minimum_nonnegative",
123 "dovecote_schema",
124 &[
125 "CHECK (((minimum_crate_major >= 0) AND (minimum_crate_minor >= 0) AND (minimum_crate_patch >= 0)))",
126 ],
127 ),
128 ConstraintContract::primary_key(
129 "dovecote_schema_pkey",
130 "dovecote_schema",
131 &["schema_version"],
132 &["PRIMARY KEY (schema_version)"],
133 ),
134 ConstraintContract::check(
135 "dovecote_events_row_id_positive",
136 "dovecote_events",
137 &["CHECK ((row_id > 0))"],
138 ),
139 ConstraintContract::primary_key(
140 "dovecote_events_pkey",
141 "dovecote_events",
142 &["row_id"],
143 &["PRIMARY KEY (row_id)"],
144 ),
145 ConstraintContract::check(
146 "dovecote_events_specversion",
147 "dovecote_events",
148 &["CHECK (((specversion) = '1.0'))"],
149 ),
150 ConstraintContract::check(
151 "dovecote_events_stream_size",
152 "dovecote_events",
153 &["CHECK ((octet_length((stream)) <= 255))"],
154 ),
155 ConstraintContract::check(
156 "dovecote_events_event_id_size",
157 "dovecote_events",
158 &["CHECK ((octet_length((event_id)) <= 1024))"],
159 ),
160 ConstraintContract::check(
161 "dovecote_events_source_size",
162 "dovecote_events",
163 &["CHECK ((octet_length((source)) <= 2048))"],
164 ),
165 ConstraintContract::check(
166 "dovecote_events_event_type_size",
167 "dovecote_events",
168 &["CHECK ((octet_length((event_type)) <= 1024))"],
169 ),
170 ConstraintContract::check(
171 "dovecote_events_subject_size",
172 "dovecote_events",
173 &["CHECK (((subject IS NULL) OR (octet_length((subject)) <= 2048)))"],
174 ),
175 ConstraintContract::check(
176 "dovecote_events_content_type_size",
177 "dovecote_events",
178 &["CHECK (((datacontenttype IS NULL) OR (octet_length((datacontenttype)) <= 255)))"],
179 ),
180 ConstraintContract::check(
181 "dovecote_events_schema_size",
182 "dovecote_events",
183 &["CHECK (((dataschema IS NULL) OR (octet_length((dataschema)) <= 2048)))"],
184 ),
185 ConstraintContract::check(
186 "dovecote_events_partition_size",
187 "dovecote_events",
188 &["CHECK (((partitionkey IS NULL) OR (octet_length((partitionkey)) <= 255)))"],
189 ),
190 ConstraintContract::check(
191 "dovecote_events_identity_size",
192 "dovecote_events",
193 &["CHECK (((octet_length((source)) + octet_length((event_id))) <= 2048))"],
194 ),
195 ConstraintContract::check(
196 "dovecote_events_data_kind",
197 "dovecote_events",
198 &["CHECK (((data_kind IS NULL) OR ((data_kind) = ANY ((ARRAY['json', 'binary'])))))"],
199 ),
200 ConstraintContract::check(
201 "dovecote_events_data_pair",
202 "dovecote_events",
203 &["CHECK (((data_kind IS NULL) = (data IS NULL)))"],
204 ),
205 ConstraintContract::check(
206 "dovecote_events_content_type",
207 "dovecote_events",
208 &[
209 "CHECK (((data IS NULL) OR (octet_length(data) = 0) OR (datacontenttype IS NOT NULL)))",
210 ],
211 ),
212 ConstraintContract::check(
213 "dovecote_deliveries_state",
214 "dovecote_deliveries",
215 &[
216 "CHECK (((state) = ANY ((ARRAY['pending', 'claimed', 'delivered', 'quarantined']))))",
217 ],
218 ),
219 ConstraintContract::primary_key(
220 "dovecote_deliveries_pkey",
221 "dovecote_deliveries",
222 &["event_row_id"],
223 &["PRIMARY KEY (event_row_id)"],
224 ),
225 ConstraintContract::check(
226 "dovecote_deliveries_attempts",
227 "dovecote_deliveries",
228 &["CHECK ((attempts >= 0))"],
229 ),
230 ConstraintContract::check(
231 "dovecote_deliveries_token_size",
232 "dovecote_deliveries",
233 &["CHECK (((claim_token IS NULL) OR (octet_length(claim_token) = 16)))"],
234 ),
235 ConstraintContract::check(
236 "dovecote_deliveries_worker_size",
237 "dovecote_deliveries",
238 &["CHECK (((claimed_by IS NULL) OR (octet_length((claimed_by)) <= 255)))"],
239 ),
240 ConstraintContract::check(
241 "dovecote_deliveries_failure_code_size",
242 "dovecote_deliveries",
243 &[
244 "CHECK (((last_failure_code IS NULL) OR (octet_length((last_failure_code)) <= 128)))",
245 ],
246 ),
247 ConstraintContract::check(
248 "dovecote_deliveries_failure_detail_size",
249 "dovecote_deliveries",
250 &[
251 "CHECK (((last_failure_detail IS NULL) OR (octet_length((last_failure_detail)) <= 2048)))",
252 ],
253 ),
254 ConstraintContract::check(
255 "dovecote_deliveries_quarantine_size",
256 "dovecote_deliveries",
257 &[
258 "CHECK (((quarantine_reason IS NULL) OR (octet_length((quarantine_reason)) <= 2048)))",
259 ],
260 ),
261 ConstraintContract::check(
262 "dovecote_deliveries_failure_pair",
263 "dovecote_deliveries",
264 &["CHECK (((last_failure_code IS NULL) = (last_failure_detail IS NULL)))"],
265 ),
266 ConstraintContract::check(
267 "dovecote_deliveries_state_shape",
268 "dovecote_deliveries",
269 &[
270 "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))))",
271 ],
272 ),
273 ConstraintContract::foreign_key(
274 "dovecote_deliveries_event_row_id_fkey",
275 "dovecote_deliveries",
276 &["event_row_id"],
277 "dovecote_events",
278 &["row_id"],
279 "r",
280 &["FOREIGN KEY (event_row_id) REFERENCES dovecote_events (row_id) ON DELETE RESTRICT"],
281 ),
282 ];
283
284 let constraints = query_as::<_, ConstraintInfo>(
285 r#"
286 SELECT table_class.relname AS table_name,
287 constraint_class.conname AS name,
288 constraint_class.contype::text AS kind,
289 ARRAY(
290 SELECT attribute.attname::text
291 FROM unnest(constraint_class.conkey) WITH ORDINALITY AS key(attnum, ordinality)
292 JOIN pg_attribute attribute
293 ON attribute.attrelid = constraint_class.conrelid
294 AND attribute.attnum = key.attnum
295 ORDER BY key.ordinality
296 ) AS columns,
297 parent_class.relname AS referenced_table,
298 ARRAY(
299 SELECT attribute.attname::text
300 FROM unnest(constraint_class.confkey) WITH ORDINALITY AS key(attnum, ordinality)
301 JOIN pg_attribute attribute
302 ON attribute.attrelid = constraint_class.confrelid
303 AND attribute.attnum = key.attnum
304 ORDER BY key.ordinality
305 ) AS referenced_columns,
306 CASE WHEN constraint_class.contype = 'f'
307 THEN constraint_class.confdeltype::text END AS delete_action,
308 constraint_class.convalidated AS validated,
309 constraint_class.condeferrable AS deferrable,
310 constraint_class.condeferred AS deferred,
311 pg_get_constraintdef(constraint_class.oid) AS definition
312 FROM pg_constraint constraint_class
313 JOIN pg_class table_class ON table_class.oid = constraint_class.conrelid
314 LEFT JOIN pg_class parent_class ON parent_class.oid = constraint_class.confrelid
315 WHERE table_class.relnamespace::bigint = $1
316 AND (constraint_class.confrelid = 0 OR parent_class.relnamespace::bigint = $1)
317 AND constraint_class.conname = ANY($2)
318 "#,
319 )
320 .bind(namespace.oid)
321 .bind(
322 expected_constraints
323 .iter()
324 .map(|constraint| constraint.name.to_owned())
325 .collect::<Vec<_>>(),
326 )
327 .fetch_all(&mut *connection)
328 .await
329 .map_err(|source| SchemaError::sql("check constraints", source))?;
330 for expected in &expected_constraints {
331 let Some(actual) = constraints
332 .iter()
333 .find(|constraint| constraint.name == expected.name)
334 else {
335 return Err(SchemaError::MigrationMismatch {
336 detail: format!("required constraint {} is missing", expected.name),
337 });
338 };
339
340 if !actual.matches(expected) {
341 return Err(SchemaError::MigrationMismatch {
342 detail: format!("required constraint {} is incompatible", expected.name),
343 });
344 }
345 }
346
347 let expected_indexes = [
348 IndexContract::new(
349 "dovecote_events_source_event_id",
350 "dovecote_events",
351 true,
352 &["source", "event_id"],
353 Some(&["C", "C"]),
354 ),
355 IndexContract::new(
356 "dovecote_deliveries_claimable",
357 "dovecote_deliveries",
358 false,
359 &["state", "available_at", "event_row_id"],
360 None,
361 ),
362 IndexContract::new(
363 "dovecote_deliveries_expired_claims",
364 "dovecote_deliveries",
365 false,
366 &["state", "claim_expires_at", "event_row_id"],
367 None,
368 ),
369 ];
370 let indexes = query_as::<_, IndexInfo>(
371 r#"
372 SELECT table_class.relname AS table_name,
373 index_class.relname AS name,
374 access_method.amname AS access_method,
375 i.indisunique AS is_unique,
376 i.indisvalid AS is_valid,
377 i.indisready AS is_ready,
378 i.indpred IS NOT NULL AS has_predicate,
379 i.indnkeyatts AS key_columns,
380 i.indnatts AS total_columns,
381 COALESCE(
382 ARRAY_AGG(i.indoption[keys.ordinality::integer - 1] ORDER BY keys.ordinality)
383 FILTER (WHERE keys.ordinality <= i.indnkeyatts),
384 ARRAY[]::smallint[]
385 ) AS options,
386 COALESCE(
387 ARRAY_AGG(a.attname::text ORDER BY keys.ordinality)
388 FILTER (WHERE keys.ordinality <= i.indnkeyatts),
389 ARRAY[]::text[]
390 ) AS columns,
391 COALESCE(
392 ARRAY_AGG(COALESCE(coll.collname::text, 'default') ORDER BY keys.ordinality)
393 FILTER (WHERE keys.ordinality <= i.indnkeyatts),
394 ARRAY[]::text[]
395 ) AS collations
396 FROM pg_class table_class
397 JOIN pg_namespace namespace ON namespace.oid = table_class.relnamespace
398 JOIN pg_index i ON i.indrelid = table_class.oid
399 JOIN pg_class index_class ON index_class.oid = i.indexrelid
400 JOIN pg_am access_method ON access_method.oid = index_class.relam
401 CROSS JOIN LATERAL unnest(i.indkey) WITH ORDINALITY AS keys(attnum, ordinality)
402 JOIN pg_attribute a ON a.attrelid = table_class.oid AND a.attnum = keys.attnum
403 LEFT JOIN pg_collation coll ON coll.oid = i.indcollation[keys.ordinality::integer - 1]
404 WHERE table_class.relnamespace::bigint = $1
405 AND index_class.relname = ANY($2)
406 GROUP BY table_class.relname, index_class.relname, access_method.amname,
407 i.indisunique, i.indisvalid, i.indisready, i.indpred IS NOT NULL,
408 i.indnkeyatts, i.indnatts
409 "#,
410 )
411 .bind(namespace.oid)
412 .bind(
413 expected_indexes
414 .iter()
415 .map(|index| index.name.to_owned())
416 .collect::<Vec<_>>(),
417 )
418 .fetch_all(&mut *connection)
419 .await
420 .map_err(|source| SchemaError::sql("check indexes", source))?;
421 for expected in &expected_indexes {
422 let Some(actual) = indexes.iter().find(|index| index.name == expected.name) else {
423 return Err(SchemaError::MigrationMismatch {
424 detail: format!("required index {} is missing", expected.name),
425 });
426 };
427
428 if !actual.matches(expected) {
429 return Err(SchemaError::MigrationMismatch {
430 detail: format!("required index {} is incompatible", expected.name),
431 });
432 }
433 }
434
435 Ok(())
436}
437
438#[derive(Debug, FromRow)]
439struct NamespaceInfo {
440 oid: i64,
441 name: String,
442}
443
444async fn resolve_namespace(connection: &mut PgConnection) -> Result<NamespaceInfo, SchemaError> {
445 query_as::<_, NamespaceInfo>(
446 r#"
447 SELECT oid::bigint AS oid, nspname AS name
448 FROM pg_namespace
449 WHERE nspname = current_schema()
450 "#,
451 )
452 .fetch_optional(&mut *connection)
453 .await
454 .map_err(|source| SchemaError::sql("resolve current schema", source))?
455 .ok_or_else(|| SchemaError::MigrationMismatch {
456 detail: "the transaction has no resolvable current schema".to_owned(),
457 })
458}
459
460#[derive(Clone, Copy)]
461struct ColumnSpec {
462 name: &'static str,
463 data_type: &'static str,
464 maximum_length: Option<i32>,
465 nullable: bool,
466 identity: bool,
467 default_fragment: Option<&'static str>,
468}
469
470impl ColumnSpec {
471 const fn required(
472 name: &'static str,
473 data_type: &'static str,
474 maximum_length: Option<i32>,
475 ) -> Self {
476 Self {
477 name,
478 data_type,
479 maximum_length,
480 nullable: false,
481 identity: false,
482 default_fragment: None,
483 }
484 }
485
486 const fn required_identity(
487 name: &'static str,
488 data_type: &'static str,
489 maximum_length: Option<i32>,
490 ) -> Self {
491 Self {
492 identity: true,
493 ..Self::required(name, data_type, maximum_length)
494 }
495 }
496
497 const fn optional(
498 name: &'static str,
499 data_type: &'static str,
500 maximum_length: Option<i32>,
501 ) -> Self {
502 Self {
503 nullable: true,
504 ..Self::required(name, data_type, maximum_length)
505 }
506 }
507
508 const fn required_with_default(
509 name: &'static str,
510 data_type: &'static str,
511 maximum_length: Option<i32>,
512 default_fragment: &'static str,
513 ) -> Self {
514 Self {
515 default_fragment: Some(default_fragment),
516 ..Self::required(name, data_type, maximum_length)
517 }
518 }
519}
520
521#[derive(Debug, FromRow)]
522struct ColumnInfo {
523 column_name: String,
524 data_type: String,
525 character_maximum_length: Option<i32>,
526 is_nullable: String,
527 column_default: Option<String>,
528 is_identity: String,
529 identity_generation: Option<String>,
530}
531
532async fn check_columns(
533 connection: &mut PgConnection,
534 schema_name: &str,
535 table: &str,
536 expected: &[ColumnSpec],
537) -> Result<(), SchemaError> {
538 let columns = query_as::<_, ColumnInfo>(
539 r#"
540 SELECT column_name, data_type, character_maximum_length,
541 is_nullable, column_default, is_identity, identity_generation
542 FROM information_schema.columns
543 WHERE table_schema = $1 AND table_name = $2
544 "#,
545 )
546 .bind(schema_name)
547 .bind(table)
548 .fetch_all(&mut *connection)
549 .await
550 .map_err(|source| SchemaError::sql("check columns", source))?;
551 if let Some(column) = columns.iter().find(|column| {
552 !expected
553 .iter()
554 .any(|specification| specification.name == column.column_name)
555 }) {
556 return Err(SchemaError::MigrationMismatch {
557 detail: format!("unexpected column {}.{}", table, column.column_name),
558 });
559 }
560
561 for specification in expected {
562 let Some(column) = columns
563 .iter()
564 .find(|column| column.column_name == specification.name)
565 else {
566 return Err(SchemaError::MigrationMismatch {
567 detail: format!(
568 "required column {}.{} is missing",
569 table, specification.name
570 ),
571 });
572 };
573 let default_matches = specification.default_fragment.is_none_or(|fragment| {
574 column
575 .column_default
576 .as_deref()
577 .is_some_and(|default| normalize_sql(default) == normalize_sql(fragment))
578 });
579 let identity_matches = if specification.identity {
580 column.is_identity == "YES" && column.identity_generation.as_deref() == Some("ALWAYS")
581 } else {
582 column.is_identity == "NO"
583 };
584 if column.data_type != specification.data_type
585 || column.character_maximum_length != specification.maximum_length
586 || (column.is_nullable == "YES") != specification.nullable
587 || !default_matches
588 || !identity_matches
589 {
590 return Err(SchemaError::MigrationMismatch {
591 detail: format!("column {}.{} is incompatible", table, specification.name),
592 });
593 }
594 }
595 Ok(())
596}
597
598#[derive(Debug, FromRow)]
599struct ConstraintInfo {
600 table_name: String,
601 name: String,
602 kind: String,
603 columns: Vec<String>,
604 referenced_table: Option<String>,
605 referenced_columns: Vec<String>,
606 delete_action: Option<String>,
607 validated: bool,
608 deferrable: bool,
609 deferred: bool,
610 definition: String,
611}
612
613struct ConstraintContract {
614 name: &'static str,
615 table_name: &'static str,
616 kind: &'static str,
617 columns: &'static [&'static str],
618 referenced_table: Option<&'static str>,
619 referenced_columns: &'static [&'static str],
620 delete_action: Option<&'static str>,
621 definition_variants: &'static [&'static str],
622}
623
624impl ConstraintContract {
625 fn check(
626 name: &'static str,
627 table_name: &'static str,
628 definition_variants: &'static [&'static str],
629 ) -> Self {
630 Self {
631 name,
632 table_name,
633 kind: "c",
634 columns: &[],
635 referenced_table: None,
636 referenced_columns: &[],
637 delete_action: None,
638 definition_variants,
639 }
640 }
641
642 fn primary_key(
643 name: &'static str,
644 table_name: &'static str,
645 columns: &'static [&'static str],
646 definition_variants: &'static [&'static str],
647 ) -> Self {
648 Self {
649 name,
650 table_name,
651 kind: "p",
652 columns,
653 referenced_table: None,
654 referenced_columns: &[],
655 delete_action: None,
656 definition_variants,
657 }
658 }
659
660 fn foreign_key(
661 name: &'static str,
662 table_name: &'static str,
663 columns: &'static [&'static str],
664 referenced_table: &'static str,
665 referenced_columns: &'static [&'static str],
666 delete_action: &'static str,
667 definition_variants: &'static [&'static str],
668 ) -> Self {
669 Self {
670 name,
671 table_name,
672 kind: "f",
673 columns,
674 referenced_table: Some(referenced_table),
675 referenced_columns,
676 delete_action: Some(delete_action),
677 definition_variants,
678 }
679 }
680}
681
682impl ConstraintInfo {
683 fn matches(&self, expected: &ConstraintContract) -> bool {
684 let definition = normalize_sql(&self.definition);
685 let columns_match = self.kind == "c"
686 || self.columns
687 == expected
688 .columns
689 .iter()
690 .map(|value| (*value).to_owned())
691 .collect::<Vec<_>>();
692 self.table_name == expected.table_name
693 && self.kind == expected.kind
694 && columns_match
695 && self.referenced_table.as_deref() == expected.referenced_table
696 && self.referenced_columns
697 == expected
698 .referenced_columns
699 .iter()
700 .map(|value| (*value).to_owned())
701 .collect::<Vec<_>>()
702 && self.delete_action.as_deref() == expected.delete_action
703 && self.validated
704 && !self.deferrable
705 && !self.deferred
706 && expected
707 .definition_variants
708 .iter()
709 .any(|variant| definition == normalize_sql(variant))
710 }
711}
712
713#[derive(Debug, FromRow)]
714struct IndexInfo {
715 table_name: String,
716 name: String,
717 access_method: String,
718 is_unique: bool,
719 is_valid: bool,
720 is_ready: bool,
721 has_predicate: bool,
722 key_columns: i16,
723 total_columns: i16,
724 options: Vec<i16>,
725 columns: Vec<String>,
726 collations: Vec<String>,
727}
728
729struct IndexContract {
730 name: &'static str,
731 table_name: &'static str,
732 is_unique: bool,
733 columns: &'static [&'static str],
734 collations: Option<&'static [&'static str]>,
735}
736
737impl IndexContract {
738 fn new(
739 name: &'static str,
740 table_name: &'static str,
741 is_unique: bool,
742 columns: &'static [&'static str],
743 collations: Option<&'static [&'static str]>,
744 ) -> Self {
745 Self {
746 name,
747 table_name,
748 is_unique,
749 columns,
750 collations,
751 }
752 }
753}
754
755impl IndexInfo {
756 fn matches(&self, expected: &IndexContract) -> bool {
757 self.table_name == expected.table_name
758 && self.access_method == "btree"
759 && self.is_unique == expected.is_unique
760 && self.is_valid
761 && self.is_ready
762 && !self.has_predicate
763 && self.key_columns == i16::try_from(expected.columns.len()).unwrap_or(i16::MAX)
764 && self.total_columns == self.key_columns
765 && self.options == vec![0; expected.columns.len()]
766 && self.columns
767 == expected
768 .columns
769 .iter()
770 .map(|value| (*value).to_owned())
771 .collect::<Vec<_>>()
772 && expected.collations.is_none_or(|collations| {
773 self.collations
774 == collations
775 .iter()
776 .map(|value| (*value).to_owned())
777 .collect::<Vec<_>>()
778 })
779 }
780}
781
782fn normalize_sql(value: &str) -> String {
783 let mut value = value.to_ascii_lowercase();
784 for cast in [
785 "::character varying[]",
786 "::character varying",
787 "::timestamp with time zone",
788 "::timestamp without time zone",
789 "::double precision",
790 "::numeric",
791 "::bigint",
792 "::integer",
793 "::smallint",
794 "::boolean",
795 "::text[]",
796 "::text",
797 ] {
798 value = value.replace(cast, "");
799 }
800 value
801 .chars()
802 .filter(|character| !character.is_ascii_whitespace())
803 .collect()
804}
805
806#[cfg(test)]
807mod tests {
808 use super::*;
809 use crate::migration::{
810 MIGRATIONS, SCHEMA_VERSION, current_crate_version, marker_compatibility,
811 };
812
813 #[test]
814 fn schema_marker_uses_the_shipped_compatibility_range() {
815 let migration = MIGRATIONS
816 .iter()
817 .find(|migration| migration.version() == SCHEMA_VERSION)
818 .expect("the v1 migration is shipped");
819 let marker = SchemaMarker {
820 schema_version: 1,
821 minimum_crate_major: 0,
822 minimum_crate_minor: 1,
823 minimum_crate_patch: 0,
824 rolling_compatible: false,
825 };
826 assert_eq!(marker_compatibility(&marker), Ok(migration.compatibility()));
827 assert!(marker_matches_migration(&marker, *migration).is_ok());
828 assert!(migration.compatibility().contains(current_crate_version()));
829
830 let wrong_version = SchemaMarker {
831 schema_version: 2,
832 ..marker
833 };
834 assert!(marker_matches_migration(&wrong_version, *migration).is_err());
835
836 let too_new = SchemaMarker {
837 minimum_crate_major: 9,
838 ..marker
839 };
840 assert!(marker_matches_migration(&too_new, *migration).is_err());
841
842 let malformed = SchemaMarker {
843 minimum_crate_minor: -1,
844 ..marker
845 };
846 assert!(marker_compatibility(&malformed).is_err());
847 }
848
849 #[test]
850 fn constraint_and_index_contracts_require_their_live_semantics() {
851 let constraint = ConstraintInfo {
852 table_name: "dovecote_events".to_owned(),
853 name: "dovecote_events_pkey".to_owned(),
854 kind: "p".to_owned(),
855 columns: vec!["row_id".to_owned()],
856 referenced_table: None,
857 referenced_columns: Vec::new(),
858 delete_action: None,
859 validated: true,
860 deferrable: false,
861 deferred: false,
862 definition: "PRIMARY KEY (row_id)".to_owned(),
863 };
864 let expected = ConstraintContract::primary_key(
865 "dovecote_events_pkey",
866 "dovecote_events",
867 &["row_id"],
868 &["PRIMARY KEY (row_id)"],
869 );
870 assert!(constraint.matches(&expected));
871 let wrong_relation = ConstraintInfo {
872 table_name: "other".to_owned(),
873 ..constraint
874 };
875 assert!(!wrong_relation.matches(&expected));
876
877 let check = ConstraintInfo {
878 table_name: "dovecote_schema".to_owned(),
879 name: "dovecote_schema_version_supported".to_owned(),
880 kind: "c".to_owned(),
881 columns: vec!["schema_version".to_owned()],
882 referenced_table: None,
883 referenced_columns: Vec::new(),
884 delete_action: None,
885 validated: true,
886 deferrable: false,
887 deferred: false,
888 definition: "CHECK ((schema_version = 1))".to_owned(),
889 };
890 let expected_check = ConstraintContract::check(
891 "dovecote_schema_version_supported",
892 "dovecote_schema",
893 &["CHECK ((schema_version = 1))"],
894 );
895 assert!(check.matches(&expected_check));
896
897 let index = IndexInfo {
898 table_name: "dovecote_events".to_owned(),
899 name: "dovecote_events_source_event_id".to_owned(),
900 access_method: "btree".to_owned(),
901 is_unique: true,
902 is_valid: true,
903 is_ready: true,
904 has_predicate: false,
905 key_columns: 2,
906 total_columns: 2,
907 options: vec![0, 0],
908 columns: vec!["source".to_owned(), "event_id".to_owned()],
909 collations: vec!["C".to_owned(), "C".to_owned()],
910 };
911 let expected_index = IndexContract::new(
912 "dovecote_events_source_event_id",
913 "dovecote_events",
914 true,
915 &["source", "event_id"],
916 Some(&["C", "C"]),
917 );
918 assert!(index.matches(&expected_index));
919 let wrong_order = IndexInfo {
920 options: vec![1, 0],
921 ..index
922 };
923 assert!(!wrong_order.matches(&expected_index));
924 }
925
926 #[test]
927 fn pg17_constraint_renderings_match_the_shipped_contracts() {
928 let fixtures = [(
929 ConstraintInfo {
930 table_name: "dovecote_events".to_owned(),
931 name: "dovecote_events_identity_size".to_owned(),
932 kind: "c".to_owned(),
933 columns: vec!["source".to_owned(), "event_id".to_owned()],
934 referenced_table: None,
935 referenced_columns: Vec::new(),
936 delete_action: None,
937 validated: true,
938 deferrable: false,
939 deferred: false,
940 definition: "CHECK (((octet_length((source)::text) + octet_length((event_id)::text)) <= 2048))".to_owned(),
941 },
942 ConstraintContract::check(
943 "dovecote_events_identity_size",
944 "dovecote_events",
945 &["CHECK (((octet_length((source)) + octet_length((event_id))) <= 2048))"],
946 ),
947 )];
948
949 for (actual, expected) in fixtures {
950 assert!(actual.matches(&expected));
951 }
952 }
953
954 #[test]
955 fn sql_normalization_preserves_boolean_grouping() {
956 assert_ne!(
957 normalize_sql("CHECK ((left IS NULL OR right IS NULL))"),
958 normalize_sql("CHECK (((left IS NULL OR right IS NULL)))")
959 );
960 assert_ne!(
961 normalize_sql("CHECK (((left IS NULL) OR (right IS NULL)))"),
962 normalize_sql("CHECK (((left IS NULL) AND (right IS NULL)))")
963 );
964 }
965}