systemprompt_database/lifecycle/installation/
undeclared.rs

1//! What the live database holds that no registered extension declares.
2//!
3//! A crate that is deleted takes its schema files with it and leaves its
4//! tables, its `extension_migrations` ledger and its scheduled job behind
5//! on every installed database; a drop migration that was baseline-stamped
6//! rather than executed leaves the same residue with a ledger that says
7//! there is nothing left to run. This audit runs after every schema install and
8//! in `infra db doctor`: a live base table in a namespace the extensions
9//! declare into that no extension's `CREATE TABLE` names, and a ledger whose
10//! extension id no registered extension carries. It reads; the fix is a
11//! migration.
12//!
13//! Copyright (c) systemprompt.io — Business Source License 1.1.
14//! See <https://systemprompt.io> for licensing details.
15
16use std::collections::BTreeSet;
17
18use serde::Serialize;
19use systemprompt_extension::LoaderError;
20use systemprompt_identifiers::ExtensionId;
21
22use crate::services::DatabaseProvider;
23
24#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
25pub struct UndeclaredTable {
26    pub schema: String,
27    pub table: String,
28    pub live_rows: i64,
29}
30
31#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
32pub struct OrphanMigrationLedger {
33    pub extension_id: ExtensionId,
34    pub rows: i64,
35}
36
37#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
38pub struct SchemaResidue {
39    pub undeclared_tables: Vec<UndeclaredTable>,
40    pub orphan_migration_ledgers: Vec<OrphanMigrationLedger>,
41}
42
43impl SchemaResidue {
44    #[must_use]
45    pub const fn is_empty(&self) -> bool {
46        self.undeclared_tables.is_empty() && self.orphan_migration_ledgers.is_empty()
47    }
48}
49
50pub async fn audit_schema_residue(
51    db: &dyn DatabaseProvider,
52    owned: &[String],
53    extension_ids: &[ExtensionId],
54) -> Result<SchemaResidue, LoaderError> {
55    let declared: BTreeSet<String> = owned.iter().map(|t| qualify(t)).collect();
56    let namespaces: BTreeSet<String> = declared
57        .iter()
58        .filter_map(|t| t.split_once('.').map(|(schema, _)| schema.to_owned()))
59        .collect();
60    let mut residue = SchemaResidue::default();
61    for (schema, table, live_rows) in live_tables(db).await? {
62        let qualified = format!("{schema}.{table}");
63        if namespaces.contains(&schema) && !declared.contains(&qualified) {
64            residue.undeclared_tables.push(UndeclaredTable {
65                schema,
66                table,
67                live_rows,
68            });
69        }
70    }
71    let registered: BTreeSet<&ExtensionId> = extension_ids.iter().collect();
72    for (extension_id, rows) in migration_ledgers(db).await? {
73        if !registered.contains(&extension_id) {
74            residue
75                .orphan_migration_ledgers
76                .push(OrphanMigrationLedger { extension_id, rows });
77        }
78    }
79    Ok(residue)
80}
81
82// Why: a bigint reaches the JSON row as a number or, past 2^53, as a
83// string; both spellings are a count.
84// JSON: `DatabaseProvider` runtime row cell — catalog queries return JSON rows.
85fn as_count(value: &serde_json::Value) -> i64 {
86    value
87        .as_i64()
88        .or_else(|| value.as_str().and_then(|s| s.parse().ok()))
89        .unwrap_or(0)
90}
91
92fn qualify(table: &str) -> String {
93    if table.contains('.') {
94        table.to_owned()
95    } else {
96        format!("public.{table}")
97    }
98}
99
100// Why: `extension_migrations` itself is core's ledger and `_sqlx_migrations`
101// is sqlx's; neither is declared by an extension's CREATE TABLE.
102const LEDGER_TABLES: [&str; 2] = ["extension_migrations", "_sqlx_migrations"];
103
104async fn live_tables(db: &dyn DatabaseProvider) -> Result<Vec<(String, String, i64)>, LoaderError> {
105    let result = db
106        .query_raw_with(
107            &"SELECT n.nspname AS schema, c.relname AS table, \
108                     COALESCE(s.n_live_tup, 0)::bigint AS live_rows \
109              FROM pg_class c \
110              JOIN pg_namespace n ON n.oid = c.relnamespace \
111              LEFT JOIN pg_stat_user_tables s ON s.relid = c.oid \
112              WHERE c.relkind = 'r' \
113                AND n.nspname NOT IN ('pg_catalog', 'information_schema') \
114                AND n.nspname NOT LIKE 'pg_toast%' \
115              ORDER BY 1, 2",
116            &[],
117        )
118        .await
119        .map_err(|e| LoaderError::SchemaInstallationStepFailed {
120            extension: ExtensionId::new("schema-residue"),
121            context: "could not list live tables".to_owned(),
122            source: Box::new(e),
123        })?;
124    Ok(result
125        .rows
126        .iter()
127        .filter_map(|row| {
128            let schema = row.get("schema")?.as_str()?.to_owned();
129            let table = row.get("table")?.as_str()?.to_owned();
130            let live_rows = row.get("live_rows").map_or(0, as_count);
131            (!LEDGER_TABLES.contains(&table.as_str())).then_some((schema, table, live_rows))
132        })
133        .collect())
134}
135
136async fn migration_ledgers(
137    db: &dyn DatabaseProvider,
138) -> Result<Vec<(ExtensionId, i64)>, LoaderError> {
139    let result = db
140        .query_raw_with(
141            &"SELECT extension_id, COUNT(*)::bigint AS rows \
142              FROM extension_migrations GROUP BY extension_id ORDER BY extension_id",
143            &[],
144        )
145        .await
146        .map_err(|e| LoaderError::SchemaInstallationStepFailed {
147            extension: ExtensionId::new("schema-residue"),
148            context: "could not read extension_migrations".to_owned(),
149            source: Box::new(e),
150        })?;
151    Ok(result
152        .rows
153        .iter()
154        .filter_map(|row| {
155            let id = ExtensionId::new(row.get("extension_id")?.as_str()?);
156            Some((id, row.get("rows").map_or(0, as_count)))
157        })
158        .collect())
159}
systemprompt_database/lifecycle/installation/undeclared.rs

systemprompt_database/lifecycle/installation/
undeclared.rs
