1#![forbid(unsafe_code)]
44
45mod drift;
46pub use drift::{UNSEEN, compare, drift, unseen};
47
48use cratefield_core::{Database, DbError, Row, Rows, Statement};
49use cratefield_tables::{FieldDef, FieldKind, ForeignKey, Schema, TableDef};
50
51fn harness_exclusion(whole_catalog: bool, column: &str) -> String {
77 if whole_catalog {
78 String::new()
79 } else {
80 format!("AND {column} NOT LIKE 'harness\\_%' ESCAPE '\\'")
81 }
82}
83
84pub async fn schema(db: &dyn Database) -> Result<Schema, DbError> {
103 read_schema(db, false).await
104}
105
106pub async fn full_schema(db: &dyn Database) -> Result<Schema, DbError> {
130 read_schema(db, true).await
131}
132
133async fn read_schema(db: &dyn Database, whole_catalog: bool) -> Result<Schema, DbError> {
136 match sqlite_tables(db, whole_catalog).await {
143 Ok(tables) => sqlite_schema(db, tables).await,
144 Err(sqlite_probe) => match postgres_schema(db, whole_catalog).await {
145 Ok(schema) => Ok(schema),
146 Err(postgres_probe) => Err(DbError::Query(format!(
147 "could not read a schema from either catalog: sqlite said \
148 [{sqlite_probe}]; postgres said [{postgres_probe}]"
149 ))),
150 },
151 }
152}
153
154pub async fn row_count(db: &dyn Database, table: &str) -> Result<u64, DbError> {
161 let rows = db
162 .query(&Statement::new(format!(
163 "SELECT COUNT(*) AS n FROM {}",
164 quote_ident(table)
165 )))
166 .await?;
167 let count = rows
168 .first()
169 .and_then(|row| row.get::<i64>("n"))
170 .unwrap_or(0);
171 Ok(u64::try_from(count).unwrap_or(0))
172}
173
174pub async fn rows(
183 db: &dyn Database,
184 table: &str,
185 order_by: &[&str],
186 limit: u64,
187 offset: u64,
188) -> Result<Rows, DbError> {
189 let order = order_by
190 .iter()
191 .map(|column| quote_ident(column))
192 .collect::<Vec<_>>()
193 .join(", ");
194 let sql = if order.is_empty() {
195 format!("SELECT * FROM {} LIMIT ? OFFSET ?", quote_ident(table))
196 } else {
197 format!(
198 "SELECT * FROM {} ORDER BY {order} LIMIT ? OFFSET ?",
199 quote_ident(table)
200 )
201 };
202 db.query(&Statement::with_values(
203 sql,
204 vec![
205 sea_query::Value::BigInt(Some(i64::try_from(limit).unwrap_or(i64::MAX))),
206 sea_query::Value::BigInt(Some(i64::try_from(offset).unwrap_or(0))),
207 ],
208 ))
209 .await
210}
211
212async fn sqlite_tables(db: &dyn Database, whole_catalog: bool) -> Result<Vec<String>, DbError> {
221 let rows = db
222 .query(&Statement::new(format!(
223 "SELECT name FROM sqlite_master \
231 WHERE type = 'table' \
232 AND name NOT LIKE 'sqlite\\_%' ESCAPE '\\' \
233 {harness} \
234 ORDER BY name ASC",
235 harness = harness_exclusion(whole_catalog, "name")
236 )))
237 .await?;
238 Ok(rows
239 .rows
240 .iter()
241 .filter_map(|row| row.get::<String>("name"))
242 .collect())
243}
244
245async fn sqlite_schema(db: &dyn Database, names: Vec<String>) -> Result<Schema, DbError> {
249 let mut tables = Vec::with_capacity(names.len());
250 for name in names {
251 let columns = db
252 .query(&Statement::with_values(
253 "SELECT name AS column_name, type AS declared, \"notnull\" AS not_null, \
254 pk AS pk_pos FROM pragma_table_info(?) ORDER BY cid",
255 vec![text(&name)],
256 ))
257 .await?;
258 let foreign_keys = db
259 .query(&Statement::with_values(
260 "SELECT \"table\" AS parent, \"from\" AS child_column \
261 FROM pragma_foreign_key_list(?) ORDER BY id, seq",
262 vec![text(&name)],
263 ))
264 .await?;
265 let indexes = db
266 .query(&Statement::with_values(
267 "SELECT name AS index_name, \"unique\" AS is_unique, origin AS origin \
268 FROM pragma_index_list(?) ORDER BY seq",
269 vec![text(&name)],
270 ))
271 .await?;
272
273 let mut unique: Vec<String> = Vec::new();
281 let mut indexed_columns: Vec<String> = Vec::new();
282 for index in &indexes.rows {
283 let Some(index_name) = index.get::<String>("index_name") else {
284 continue;
285 };
286 let is_unique = index.get::<i64>("is_unique").unwrap_or(0) == 1;
287 let origin = index.get::<String>("origin").unwrap_or_default();
288 let members = db
289 .query(&Statement::with_values(
290 "SELECT name AS column_name FROM pragma_index_info(?) ORDER BY seqno",
291 vec![text(&index_name)],
292 ))
293 .await?
294 .rows
295 .iter()
296 .filter_map(|row| row.get::<String>("column_name"))
297 .collect::<Vec<_>>();
298 if members.len() != 1 || origin == "pk" {
299 continue;
300 }
301 if is_unique {
302 unique.push(members[0].clone());
303 } else if origin == "c" {
304 indexed_columns.push(members[0].clone());
305 }
306 }
307
308 tables.push(sqlite_table(
309 &name,
310 &columns.rows,
311 &foreign_keys.rows,
312 &unique,
313 &indexed_columns,
314 ));
315 }
316 Ok(Schema::new(tables))
317}
318
319fn sqlite_table(
321 name: &str,
322 columns: &[Row],
323 foreign_keys: &[Row],
324 unique: &[String],
325 indexed: &[String],
326) -> TableDef {
327 let mut fields = Vec::with_capacity(columns.len());
328 let mut key: Vec<(String, i64)> = Vec::new();
331 for column in columns {
332 let column_name: String = column.get("column_name").unwrap_or_default();
333 let declared: String = column.get("declared").unwrap_or_default();
334 let not_null = column.get::<i64>("not_null").unwrap_or(0) == 1;
335 let pk_pos = column.get::<i64>("pk_pos").unwrap_or(0);
336 if pk_pos > 0 {
337 key.push((column_name.clone(), pk_pos));
338 }
339 fields.push(FieldDef {
340 kind: field_kind(&declared),
341 required: not_null || pk_pos > 0,
342 unique: unique.contains(&column_name),
343 indexed: indexed.contains(&column_name),
344 ..FieldDef::new(column_name, FieldKind::text())
345 });
346 }
347 key.sort_by_key(|(_, position)| *position);
350 TableDef {
351 name: name.to_owned(),
352 primary_key: key.into_iter().map(|(column, _)| column).collect(),
353 foreign_keys: foreign_keys
354 .iter()
355 .filter_map(|row| {
356 Some(ForeignKey::new(
362 row.get::<String>("child_column")?,
363 row.get::<String>("parent")?,
364 ))
365 })
366 .collect(),
367 fields,
368 }
369}
370
371async fn postgres_schema(db: &dyn Database, whole_catalog: bool) -> Result<Schema, DbError> {
380 let names = db
381 .query(&Statement::new(format!(
382 "SELECT table_name AS name FROM information_schema.tables \
383 WHERE table_schema = current_schema() AND table_type = 'BASE TABLE' \
384 {harness} \
385 ORDER BY table_name ASC",
386 harness = harness_exclusion(whole_catalog, "table_name")
387 )))
388 .await?
389 .rows
390 .iter()
391 .filter_map(|row| row.get::<String>("name"))
392 .collect::<Vec<_>>();
393
394 let mut tables = Vec::with_capacity(names.len());
395 for name in names {
396 let columns = db
397 .query(&Statement::with_values(
398 "SELECT column_name AS column_name, udt_name AS declared, \
399 is_nullable AS is_nullable FROM information_schema.columns \
400 WHERE table_schema = current_schema() AND table_name = ? \
401 ORDER BY ordinal_position",
402 vec![text(&name)],
403 ))
404 .await?;
405 let keys = db
408 .query(&Statement::with_values(
409 "SELECT tc.constraint_type AS constraint_type, kcu.column_name AS column_name, \
410 kcu.ordinal_position AS pos \
411 FROM information_schema.table_constraints tc \
412 JOIN information_schema.key_column_usage kcu \
413 ON kcu.constraint_name = tc.constraint_name \
414 AND kcu.constraint_schema = tc.constraint_schema \
415 WHERE tc.table_schema = current_schema() AND tc.table_name = ? \
416 AND tc.constraint_type IN ('PRIMARY KEY', 'UNIQUE') \
417 ORDER BY tc.constraint_name, kcu.ordinal_position",
418 vec![text(&name)],
419 ))
420 .await?;
421 let foreign_keys = db
422 .query(&Statement::with_values(
423 "SELECT kcu.column_name AS child_column, ccu.table_name AS parent \
433 FROM information_schema.table_constraints tc \
434 JOIN information_schema.key_column_usage kcu \
435 ON kcu.constraint_name = tc.constraint_name \
436 AND kcu.constraint_schema = tc.constraint_schema \
437 JOIN information_schema.constraint_column_usage ccu \
438 ON ccu.constraint_name = tc.constraint_name \
439 AND ccu.constraint_schema = tc.constraint_schema \
440 WHERE tc.table_schema = current_schema() AND tc.table_name = ? \
441 AND tc.constraint_type = 'FOREIGN KEY' \
442 ORDER BY kcu.column_name, ccu.table_name, ccu.column_name",
443 vec![text(&name)],
444 ))
445 .await?;
446 let indexes = db
447 .query(&Statement::with_values(
448 "SELECT indexdef AS indexdef FROM pg_indexes \
449 WHERE schemaname = current_schema() AND tablename = ? \
450 ORDER BY indexname",
451 vec![text(&name)],
452 ))
453 .await?;
454
455 tables.push(postgres_table(
456 &name,
457 &columns.rows,
458 &keys.rows,
459 &foreign_keys.rows,
460 &indexes.rows,
461 ));
462 }
463 Ok(Schema::new(tables))
464}
465
466fn postgres_table(
468 name: &str,
469 columns: &[Row],
470 keys: &[Row],
471 foreign_keys: &[Row],
472 indexes: &[Row],
473) -> TableDef {
474 let mut key: Vec<(String, i64)> = Vec::new();
475 let mut unique_constraints: Vec<String> = Vec::new();
476 for row in keys {
477 let column: String = row.get::<String>("column_name").unwrap_or_default();
478 if row.get::<String>("constraint_type").as_deref() == Some("PRIMARY KEY") {
479 key.push((column, row.get::<i64>("pos").unwrap_or(0)));
480 } else {
481 unique_constraints.push(column);
482 }
483 }
484 key.sort_by_key(|(_, position)| *position);
485 let primary_key: Vec<String> = key.into_iter().map(|(column, _)| column).collect();
486
487 let mut unique: Vec<String> = Vec::new();
497 let mut indexed_columns: Vec<String> = Vec::new();
498 let column_names: Vec<String> = columns
499 .iter()
500 .filter_map(|row| row.get::<String>("column_name"))
501 .collect();
502 for index in indexes {
503 let Some(def) = index.get::<String>("indexdef") else {
504 continue;
505 };
506 let Some(members) = pg_index_columns(&def) else {
507 continue;
508 };
509 if members == primary_key {
510 continue;
511 }
512 let is_unique = def.starts_with("CREATE UNIQUE INDEX");
513 if members.len() != 1 {
514 continue;
515 }
516 if !column_names.contains(&members[0]) {
517 continue;
518 }
519 if is_unique {
520 unique.push(members[0].clone());
521 } else {
522 indexed_columns.push(members[0].clone());
523 }
524 }
525 for column in unique_constraints {
526 if !unique.contains(&column) {
527 unique.push(column);
528 }
529 }
530
531 TableDef {
532 name: name.to_owned(),
533 primary_key,
534 foreign_keys: foreign_keys
535 .iter()
536 .filter_map(|row| {
537 Some(ForeignKey::new(
538 row.get::<String>("child_column")?,
539 row.get::<String>("parent")?,
540 ))
541 })
542 .collect(),
543 fields: columns
544 .iter()
545 .filter_map(|row| {
546 let column_name: String = row.get::<String>("column_name")?;
547 Some(FieldDef {
548 kind: field_kind(&row.get::<String>("declared").unwrap_or_default()),
549 required: row.get::<String>("is_nullable").as_deref() == Some("NO"),
550 unique: unique.contains(&column_name),
551 indexed: indexed_columns.contains(&column_name),
552 ..FieldDef::new(column_name, FieldKind::text())
553 })
554 })
555 .collect(),
556 }
557}
558
559fn pg_index_columns(def: &str) -> Option<Vec<String>> {
562 let body = def.split(" WHERE ").next()?;
565 let open = body.rfind('(')?;
566 let close = body.rfind(')')?;
567 if close < open {
568 return None;
569 }
570 Some(
571 body[open + 1..close]
572 .split(',')
573 .map(|entry| {
574 entry
575 .split_whitespace()
576 .next()
577 .unwrap_or_default()
578 .to_owned()
579 })
580 .collect(),
581 )
582}
583
584fn field_kind(declared: &str) -> FieldKind {
597 let lowered = declared.trim().to_ascii_lowercase();
598 let base = lowered
599 .split('(')
600 .next()
601 .unwrap_or("")
602 .split_whitespace()
603 .next()
604 .unwrap_or("");
605 match base {
606 "uuid" => FieldKind::Uuid,
607 "json" | "jsonb" => FieldKind::Json,
608 "bool" | "boolean" => FieldKind::Boolean,
609 "date" | "datetime" | "timestamp" | "timestamptz" | "time" => FieldKind::Timestamp,
610 "int" | "int2" | "int4" | "int8" | "integer" | "bigint" | "smallint" | "serial"
611 | "bigserial" => FieldKind::integer(),
612 "real" | "float" | "float4" | "float8" | "double" | "numeric" | "decimal" | "money" => {
613 FieldKind::real()
614 }
615 _ => FieldKind::text(),
616 }
617}
618
619fn quote_ident(name: &str) -> String {
624 format!("\"{}\"", name.replace('"', "\"\""))
625}
626
627fn text(value: &str) -> sea_query::Value {
628 sea_query::Value::String(Some(Box::new(value.to_owned())))
629}
630
631#[cfg(test)]
636mod tests {
637 use super::*;
638
639 const DDL: &[&str] = &[
644 "CREATE TABLE parent (\
645 id TEXT PRIMARY KEY, \
646 email TEXT NOT NULL UNIQUE, \
647 note TEXT)",
648 "CREATE TABLE child (\
649 id TEXT PRIMARY KEY, \
650 parent_id TEXT NOT NULL REFERENCES parent (id), \
651 qty INTEGER, \
652 created_at TEXT)",
653 "CREATE INDEX child_by_created ON child (created_at)",
654 "CREATE UNIQUE INDEX child_qty_unique ON child (qty)",
655 ];
656
657 async fn seed(db: &dyn Database) {
658 for statement in DDL {
659 db.execute(&Statement::new(*statement)).await.expect("ddl");
660 }
661 db.execute(&Statement::new(
665 "CREATE TABLE harness_secrets (\
666 name TEXT PRIMARY KEY, \
667 key_id TEXT NOT NULL)",
668 ))
669 .await
670 .expect("a harness-named table");
671 for id in ["p1", "p2"] {
672 db.execute(&Statement::with_values(
673 "INSERT INTO parent (id, email, note) VALUES (?, ?, ?)",
674 vec![text(id), text(&format!("{id}@example.com")), text("")],
675 ))
676 .await
677 .expect("parent row");
678 }
679 for (id, parent, qty) in [("c1", "p1", 1), ("c2", "p2", 2), ("c3", "p1", 3)] {
680 db.execute(&Statement::with_values(
681 "INSERT INTO child (id, parent_id, qty, created_at) VALUES (?, ?, ?, ?)",
682 vec![
683 text(id),
684 text(parent),
685 sea_query::Value::Int(Some(qty)),
686 text("2026-01-01T00:00:00Z"),
687 ],
688 ))
689 .await
690 .expect("child row");
691 }
692 }
693
694 async fn assert_catalog(db: &dyn Database) {
697 let schema = schema(db).await.expect("read the catalog");
698
699 let names: Vec<&str> = schema.tables.iter().map(|t| t.name.as_str()).collect();
704 assert_eq!(
705 names,
706 ["child", "parent"],
707 "sorted, the harness prefix excluded"
708 );
709 assert!(
710 schema.table("harness_secrets").is_none(),
711 "no harness_ table is listable"
712 );
713
714 let whole = full_schema(db).await.expect("read the whole catalog");
718 let whole_names: Vec<&str> = whole.tables.iter().map(|t| t.name.as_str()).collect();
719 assert!(
720 whole_names.contains(&"harness_secrets"),
721 "the whole-catalog read still sees the harness-named table: {whole_names:?}"
722 );
723 assert!(
724 whole_names.len() > names.len(),
725 "the browsable catalog hid something the database holds"
726 );
727
728 let parent = schema.table("parent").expect("parent");
729 assert_eq!(parent.primary_key, ["id"]);
730 let id = parent.field("id").expect("id column");
731 assert!(id.required, "a primary key column is NOT NULL");
732 assert!(!id.unique, "the pk marker is its own fact, not also unique");
733 assert_eq!(id.kind.as_str(), "text");
734 let email = parent.field("email").expect("email column");
735 assert!(email.required);
736 assert!(email.unique, "a UNIQUE column reads as unique");
737 assert!(!email.indexed, "unique adds no second index");
738 let note = parent.field("note").expect("note column");
739 assert!(!note.required, "note is the nullable column");
740 assert!(!note.unique);
741
742 let child = schema.table("child").expect("child");
743 assert_eq!(child.primary_key, ["id"]);
744 assert_eq!(
745 child
746 .foreign_keys
747 .iter()
748 .map(|key| (key.field.as_str(), key.references.as_str()))
749 .collect::<Vec<_>>(),
750 [("parent_id", "parent")],
751 "the foreign key comes back"
752 );
753 let parent_id = child.field("parent_id").expect("parent_id column");
754 assert!(parent_id.required);
755 assert_eq!(
756 child.field("qty").expect("qty column").kind.as_str(),
757 "integer",
758 "an INTEGER column maps onto the vocabulary"
759 );
760 assert!(
761 child.field("qty").expect("qty column").unique,
762 "the unique INDEX reads as unique"
763 );
764 assert!(
765 child
766 .field("created_at")
767 .expect("created_at column")
768 .indexed,
769 "the plain index reads as indexed"
770 );
771 assert!(!parent_id.indexed && !child.field("id").expect("id").indexed);
772
773 assert_eq!(row_count(db, "parent").await.expect("count"), 2);
775 assert_eq!(row_count(db, "child").await.expect("count"), 3);
776 let page = rows(db, "child", &["id"], 2, 1).await.expect("page");
777 let ids: Vec<String> = page
778 .rows
779 .iter()
780 .map(|row| row.get::<String>("id").unwrap_or_default())
781 .collect();
782 assert_eq!(ids, ["c2", "c3"], "ordered by primary key, offset applies");
783 }
784
785 #[pollster::test]
786 async fn the_sqlite_catalog_reads_back_every_declared_fact() {
787 let db = cratefield_adapter_sqlite::SqliteDatabase::in_memory().expect("sqlite");
788 db.apply_migrations("probe", &[]).expect("ledger");
790 seed(&db).await;
791 assert_catalog(&db).await;
792 }
793
794 #[pollster::test]
799 async fn no_harness_table_is_listable_and_the_whole_catalog_names_them() {
800 let db = cratefield_adapter_sqlite::SqliteDatabase::in_memory().expect("sqlite");
801 db.apply_migrations("probe", &[]).expect("ledger");
802 db.execute(&Statement::new(
803 "CREATE TABLE harness_secret_keys (key_id TEXT PRIMARY KEY, wrapped TEXT)".to_owned(),
804 ))
805 .await
806 .expect("harness-named table");
807 db.execute(&Statement::new(
808 "CREATE TABLE ordinary (id TEXT PRIMARY KEY)".to_owned(),
809 ))
810 .await
811 .expect("ordinary table");
812
813 let browsable = schema(&db).await.expect("catalog");
814 let names: Vec<&str> = browsable.tables.iter().map(|t| t.name.as_str()).collect();
815 assert_eq!(
816 names,
817 ["ordinary"],
818 "no harness_ table is listable, the ledger included: {names:?}"
819 );
820
821 let whole = full_schema(&db).await.expect("whole catalog");
822 let mut whole_names: Vec<&str> = whole.tables.iter().map(|t| t.name.as_str()).collect();
823 whole_names.sort_unstable();
824 assert_eq!(
825 whole_names,
826 ["harness_migrations", "harness_secret_keys", "ordinary"],
827 "the whole-catalog read is the whole catalog: {whole_names:?}"
828 );
829 }
830
831 #[tokio::test]
832 async fn the_postgres_catalog_reads_back_the_same_facts() {
833 let Some(base) = cratefield_adapter_postgres::testing::base_url() else {
834 eprintln!(
835 "SKIP: {}",
836 cratefield_adapter_postgres::testing::skip_reason()
837 );
838 return;
839 };
840 let temp = cratefield_adapter_postgres::testing::TempDb::create(&base, "introspect")
841 .await
842 .expect("throwaway database");
843 let db = cratefield_adapter_postgres::Postgres::connect(&temp.url)
844 .await
845 .expect("connect");
846 seed(&db).await;
847 assert_catalog(&db).await;
848 db.close().await.expect("close");
849 temp.finish().await;
850 }
851
852 #[pollster::test]
853 async fn an_unreadable_catalog_is_an_error_not_an_empty_schema() {
854 let err = schema(&EmptyCatalog).await.expect_err("no catalog answers");
858 match err {
859 DbError::Query(message) => {
860 assert!(message.contains("either catalog"), "{message}");
861 }
862 other => panic!("expected a query error, got {other:?}"),
863 }
864 }
865
866 struct EmptyCatalog;
869
870 #[async_trait::async_trait]
871 impl Database for EmptyCatalog {
872 async fn execute(&self, _stmt: &Statement) -> Result<u64, DbError> {
873 Err(DbError::Execute("not a database".to_owned()))
874 }
875 async fn query(&self, stmt: &Statement) -> Result<Rows, DbError> {
876 if stmt.sql.trim().starts_with("SELECT 1") {
877 Ok(Rows::new(Vec::new()))
878 } else {
879 Err(DbError::Query("no catalog here".to_owned()))
880 }
881 }
882 async fn batch_atomic(&self, _stmts: &[Statement]) -> Result<(), DbError> {
883 Err(DbError::Batch("no".to_owned()))
884 }
885 }
886
887 #[test]
888 fn identifiers_are_quoted_so_a_catalog_name_cannot_become_a_statement() {
889 assert_eq!(quote_ident("venture"), "\"venture\"");
890 assert_eq!(quote_ident("od\"d"), "\"od\"\"d\"");
891 }
892
893 #[test]
894 fn types_map_onto_the_vocabulary_with_an_explicit_text_fallback() {
895 assert_eq!(field_kind("TEXT").as_str(), "text");
896 assert_eq!(field_kind("varchar(254)").as_str(), "text");
897 assert_eq!(field_kind("INTEGER").as_str(), "integer");
898 assert_eq!(field_kind("bigint").as_str(), "integer");
899 assert_eq!(field_kind("int4").as_str(), "integer");
900 assert_eq!(field_kind("float8").as_str(), "real");
901 assert_eq!(field_kind("DOUBLE PRECISION").as_str(), "real");
902 assert_eq!(field_kind("BOOLEAN").as_str(), "boolean");
903 assert_eq!(field_kind("timestamptz").as_str(), "timestamp");
904 assert_eq!(field_kind("uuid").as_str(), "uuid");
905 assert_eq!(field_kind("jsonb").as_str(), "json");
906 assert_eq!(field_kind("BLOB").as_str(), "text");
909 }
910}