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. The 2026-09-22 production audit found 39
8//! such tables and a dead `evaluation` ledger, none reported anywhere. This
9//! audit runs after every schema install and in `infra db doctor`: a live
10//! base table in a namespace the extensions declare into that no
11//! extension's `CREATE TABLE` names, and a ledger whose extension id no
12//! registered extension carries. It reads; the fix is a migration.
13//!
14//! Copyright (c) systemprompt.io — Business Source License 1.1.
15//! See <https://systemprompt.io> for licensing details.
16
17use std::collections::BTreeSet;
18
19use serde::Serialize;
20use systemprompt_extension::LoaderError;
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: String,
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: &[String],
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<&str> = extension_ids.iter().map(String::as_str).collect();
72    for (extension_id, rows) in migration_ledgers(db).await? {
73        if !registered.contains(extension_id.as_str()) {
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.
84fn as_count(value: &serde_json::Value) -> i64 {
85    value
86        .as_i64()
87        .or_else(|| value.as_str().and_then(|s| s.parse().ok()))
88        .unwrap_or(0)
89}
90
91fn qualify(table: &str) -> String {
92    if table.contains('.') {
93        table.to_owned()
94    } else {
95        format!("public.{table}")
96    }
97}
98
99// Why: `extension_migrations` itself is core's ledger and `_sqlx_migrations`
100// is sqlx's; neither is declared by an extension's CREATE TABLE.
101const LEDGER_TABLES: [&str; 2] = ["extension_migrations", "_sqlx_migrations"];
102
103async fn live_tables(db: &dyn DatabaseProvider) -> Result<Vec<(String, String, i64)>, LoaderError> {
104    let result = db
105        .query_raw_with(
106            &"SELECT n.nspname AS schema, c.relname AS table, \
107                     COALESCE(s.n_live_tup, 0)::bigint AS live_rows \
108              FROM pg_class c \
109              JOIN pg_namespace n ON n.oid = c.relnamespace \
110              LEFT JOIN pg_stat_user_tables s ON s.relid = c.oid \
111              WHERE c.relkind = 'r' \
112                AND n.nspname NOT IN ('pg_catalog', 'information_schema') \
113                AND n.nspname NOT LIKE 'pg_toast%' \
114              ORDER BY 1, 2",
115            &[],
116        )
117        .await
118        .map_err(|e| LoaderError::SchemaInstallationFailed {
119            extension: "schema-residue".to_owned(),
120            message: format!("could not list live tables: {e}"),
121        })?;
122    Ok(result
123        .rows
124        .iter()
125        .filter_map(|row| {
126            let schema = row.get("schema")?.as_str()?.to_owned();
127            let table = row.get("table")?.as_str()?.to_owned();
128            let live_rows = row.get("live_rows").map_or(0, as_count);
129            (!LEDGER_TABLES.contains(&table.as_str())).then_some((schema, table, live_rows))
130        })
131        .collect())
132}
133
134async fn migration_ledgers(db: &dyn DatabaseProvider) -> Result<Vec<(String, i64)>, LoaderError> {
135    let result = db
136        .query_raw_with(
137            &"SELECT extension_id, COUNT(*)::bigint AS rows \
138              FROM extension_migrations GROUP BY extension_id ORDER BY extension_id",
139            &[],
140        )
141        .await
142        .map_err(|e| LoaderError::SchemaInstallationFailed {
143            extension: "schema-residue".to_owned(),
144            message: format!("could not read extension_migrations: {e}"),
145        })?;
146    Ok(result
147        .rows
148        .iter()
149        .filter_map(|row| {
150            let id = row.get("extension_id")?.as_str()?.to_owned();
151            Some((id, row.get("rows").map_or(0, as_count)))
152        })
153        .collect())
154}
systemprompt_database/lifecycle/installation/undeclared.rs

systemprompt_database/lifecycle/installation/
undeclared.rs
