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;
21use systemprompt_identifiers::ExtensionId;
22
23use crate::services::DatabaseProvider;
24
25#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
26pub struct UndeclaredTable {
27    pub schema: String,
28    pub table: String,
29    pub live_rows: i64,
30}
31
32#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
33pub struct OrphanMigrationLedger {
34    pub extension_id: ExtensionId,
35    pub rows: i64,
36}
37
38#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
39pub struct SchemaResidue {
40    pub undeclared_tables: Vec<UndeclaredTable>,
41    pub orphan_migration_ledgers: Vec<OrphanMigrationLedger>,
42}
43
44impl SchemaResidue {
45    #[must_use]
46    pub const fn is_empty(&self) -> bool {
47        self.undeclared_tables.is_empty() && self.orphan_migration_ledgers.is_empty()
48    }
49}
50
51pub async fn audit_schema_residue(
52    db: &dyn DatabaseProvider,
53    owned: &[String],
54    extension_ids: &[ExtensionId],
55) -> Result<SchemaResidue, LoaderError> {
56    let declared: BTreeSet<String> = owned.iter().map(|t| qualify(t)).collect();
57    let namespaces: BTreeSet<String> = declared
58        .iter()
59        .filter_map(|t| t.split_once('.').map(|(schema, _)| schema.to_owned()))
60        .collect();
61    let mut residue = SchemaResidue::default();
62    for (schema, table, live_rows) in live_tables(db).await? {
63        let qualified = format!("{schema}.{table}");
64        if namespaces.contains(&schema) && !declared.contains(&qualified) {
65            residue.undeclared_tables.push(UndeclaredTable {
66                schema,
67                table,
68                live_rows,
69            });
70        }
71    }
72    let registered: BTreeSet<&ExtensionId> = extension_ids.iter().collect();
73    for (extension_id, rows) in migration_ledgers(db).await? {
74        if !registered.contains(&extension_id) {
75            residue
76                .orphan_migration_ledgers
77                .push(OrphanMigrationLedger { extension_id, rows });
78        }
79    }
80    Ok(residue)
81}
82
83// Why: a bigint reaches the JSON row as a number or, past 2^53, as a
84// string; both spellings are a count.
85// JSON: `DatabaseProvider` runtime row cell — catalog queries return JSON rows.
86fn as_count(value: &serde_json::Value) -> i64 {
87    value
88        .as_i64()
89        .or_else(|| value.as_str().and_then(|s| s.parse().ok()))
90        .unwrap_or(0)
91}
92
93fn qualify(table: &str) -> String {
94    if table.contains('.') {
95        table.to_owned()
96    } else {
97        format!("public.{table}")
98    }
99}
100
101// Why: `extension_migrations` itself is core's ledger and `_sqlx_migrations`
102// is sqlx's; neither is declared by an extension's CREATE TABLE.
103const LEDGER_TABLES: [&str; 2] = ["extension_migrations", "_sqlx_migrations"];
104
105async fn live_tables(db: &dyn DatabaseProvider) -> Result<Vec<(String, String, i64)>, LoaderError> {
106    let result = db
107        .query_raw_with(
108            &"SELECT n.nspname AS schema, c.relname AS table, \
109                     COALESCE(s.n_live_tup, 0)::bigint AS live_rows \
110              FROM pg_class c \
111              JOIN pg_namespace n ON n.oid = c.relnamespace \
112              LEFT JOIN pg_stat_user_tables s ON s.relid = c.oid \
113              WHERE c.relkind = 'r' \
114                AND n.nspname NOT IN ('pg_catalog', 'information_schema') \
115                AND n.nspname NOT LIKE 'pg_toast%' \
116              ORDER BY 1, 2",
117            &[],
118        )
119        .await
120        .map_err(|e| LoaderError::SchemaInstallationStepFailed {
121            extension: ExtensionId::new("schema-residue"),
122            context: "could not list live tables".to_owned(),
123            source: Box::new(e),
124        })?;
125    Ok(result
126        .rows
127        .iter()
128        .filter_map(|row| {
129            let schema = row.get("schema")?.as_str()?.to_owned();
130            let table = row.get("table")?.as_str()?.to_owned();
131            let live_rows = row.get("live_rows").map_or(0, as_count);
132            (!LEDGER_TABLES.contains(&table.as_str())).then_some((schema, table, live_rows))
133        })
134        .collect())
135}
136
137async fn migration_ledgers(
138    db: &dyn DatabaseProvider,
139) -> Result<Vec<(ExtensionId, i64)>, LoaderError> {
140    let result = db
141        .query_raw_with(
142            &"SELECT extension_id, COUNT(*)::bigint AS rows \
143              FROM extension_migrations GROUP BY extension_id ORDER BY extension_id",
144            &[],
145        )
146        .await
147        .map_err(|e| LoaderError::SchemaInstallationStepFailed {
148            extension: ExtensionId::new("schema-residue"),
149            context: "could not read extension_migrations".to_owned(),
150            source: Box::new(e),
151        })?;
152    Ok(result
153        .rows
154        .iter()
155        .filter_map(|row| {
156            let id = ExtensionId::new(row.get("extension_id")?.as_str()?);
157            Some((id, row.get("rows").map_or(0, as_count)))
158        })
159        .collect())
160}
systemprompt_database/lifecycle/installation/undeclared.rs

systemprompt_database/lifecycle/installation/
undeclared.rs
