use std::collections::BTreeMap;
use std::path::PathBuf;
use crate::context::DjogiContext;
use crate::error::DjogiError;
use super::ledger::{LEDGER_SELECT_COLS, LedgerRow, LedgerStatus};
use super::projection::BucketKey;
use super::schema::{
AppliedSchema, ColumnSchema, ForeignKeySchema, IndexColumnSchema, IndexKindSchema,
IndexNullsOrderSchema, IndexOrderSchema, IndexSchema, IndexTargetSchema, IndexTypeSchema,
OnDeleteSchema, PrimaryKeySchema, TableSchema,
};
pub(crate) const HEERANJID_ARTIFACT_TABLES: &[&str] = &[
"heer_config",
"heer_node_state",
"heer_nodes",
"heer_ranj_node_state",
];
pub(crate) fn is_heeranjid_artifact_table(name: &str) -> bool {
HEERANJID_ARTIFACT_TABLES.binary_search(&name).is_ok()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum VerifySeverity {
Info,
Warning,
Error,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerifyDiagnostic {
pub code: String,
pub severity: VerifySeverity,
pub message: String,
pub location: Option<String>,
}
impl VerifyDiagnostic {
fn sort_key(&self) -> (String, String) {
(self.code.clone(), self.location.clone().unwrap_or_default())
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerifyReport {
pub diagnostics: Vec<VerifyDiagnostic>,
pub latest_applied_version: Option<String>,
pub applied_count: usize,
pub unfinished_count: usize,
}
impl VerifyReport {
pub fn has_errors(&self) -> bool {
self.diagnostics
.iter()
.any(|d| d.severity == VerifySeverity::Error)
}
pub fn has_warnings(&self) -> bool {
self.diagnostics
.iter()
.any(|d| d.severity == VerifySeverity::Warning)
}
}
#[derive(Debug)]
pub enum VerifyRunError {
SnapshotLoadFailed {
path: PathBuf,
source: super::snapshot::SnapshotError,
},
CatalogQueryFailed {
query_label: &'static str,
source: DjogiError,
},
LedgerQueryFailed { source: DjogiError },
}
impl std::fmt::Display for VerifyRunError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
VerifyRunError::SnapshotLoadFailed { path, source } => {
write!(
f,
"verify could not load snapshot at {}: {source}",
path.display()
)
}
VerifyRunError::CatalogQueryFailed {
query_label,
source,
} => write!(f, "verify catalog query `{query_label}` failed: {source}"),
VerifyRunError::LedgerQueryFailed { source } => {
write!(f, "verify ledger read failed: {source}")
}
}
}
}
impl std::error::Error for VerifyRunError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
VerifyRunError::SnapshotLoadFailed { source, .. } => Some(source),
VerifyRunError::CatalogQueryFailed { source, .. } => Some(source),
VerifyRunError::LedgerQueryFailed { source } => Some(source),
}
}
}
pub async fn verify(
ctx: &mut DjogiContext,
snapshot: &AppliedSchema,
) -> Result<VerifyReport, VerifyRunError> {
verify_with_policy(ctx, snapshot, &crate::config::PolicyConfig::default()).await
}
pub async fn verify_with_policy(
ctx: &mut DjogiContext,
snapshot: &AppliedSchema,
policy: &crate::config::PolicyConfig,
) -> Result<VerifyReport, VerifyRunError> {
let mut diagnostics: Vec<VerifyDiagnostic> = Vec::new();
let ledger_present = ledger_table_exists(ctx).await?;
let (applied_count, unfinished_count, latest_applied_version) = if ledger_present {
let ledger_rows = read_applied_ledger(ctx).await?;
let applied_count = ledger_rows
.iter()
.filter(|r| r.status == LedgerStatus::Applied)
.count();
let unfinished_count = ledger_rows
.iter()
.filter(|r| matches!(r.status, LedgerStatus::Pending | LedgerStatus::Failed))
.count();
let latest_applied_version = ledger_rows
.iter()
.rev()
.find(|r| r.status == LedgerStatus::Applied)
.map(|r| r.version.clone());
let live_for_ledger_check = project_live_schema(ctx).await?;
if !ledger_rows.is_empty()
&& live_for_ledger_check.models.is_empty()
&& !snapshot.models.is_empty()
{
diagnostics.push(VerifyDiagnostic {
code: "D699".to_string(),
severity: VerifySeverity::Error,
message: format!(
"ledger reports {applied_count} applied migration(s) but the \
live database contains zero tables; the schema may have been \
dropped out-of-band",
),
location: None,
});
}
emit_out_of_order_diagnostics(&ledger_rows, policy, &mut diagnostics);
(applied_count, unfinished_count, latest_applied_version)
} else {
diagnostics.push(VerifyDiagnostic {
code: "D621".to_string(),
severity: VerifySeverity::Error,
message: "ledger table `djogi_schema_migrations` not found — run \
`djogi migrations apply` or `djogi migrations baseline` \
first; verify is read-only and will not bootstrap the ledger"
.to_string(),
location: None,
});
(0, 0, None)
};
let live = project_live_schema(ctx).await?;
diff_tables(snapshot, &live, &mut diagnostics);
diff_indexes(snapshot, &live, &mut diagnostics);
diff_advisory_fields(snapshot, &mut diagnostics);
diagnostics.sort_by_key(|d| d.sort_key());
Ok(VerifyReport {
diagnostics,
latest_applied_version,
applied_count,
unfinished_count,
})
}
pub async fn verify_bucket(
ctx: &mut DjogiContext,
bucket: &BucketKey,
snapshot: &AppliedSchema,
policy: &crate::config::PolicyConfig,
emit_ledger_diagnostics: bool,
database_has_models: bool,
) -> Result<VerifyReport, VerifyRunError> {
let mut diagnostics: Vec<VerifyDiagnostic> = Vec::new();
let ledger_present = ledger_table_exists(ctx).await?;
let (applied_count, unfinished_count, latest_applied_version) = if ledger_present {
let ledger_rows = read_applied_ledger(ctx).await?;
let applied_count = ledger_rows
.iter()
.filter(|r| r.status == LedgerStatus::Applied)
.count();
let unfinished_count = ledger_rows
.iter()
.filter(|r| matches!(r.status, LedgerStatus::Pending | LedgerStatus::Failed))
.count();
let latest_applied_version = ledger_rows
.iter()
.rev()
.find(|r| r.status == LedgerStatus::Applied)
.map(|r| r.version.clone());
if emit_ledger_diagnostics {
let live_for_ledger_check = project_live_schema(ctx).await?;
if !ledger_rows.is_empty()
&& live_for_ledger_check.models.is_empty()
&& database_has_models
{
diagnostics.push(VerifyDiagnostic {
code: "D699".to_string(),
severity: VerifySeverity::Error,
message: format!(
"ledger reports {applied_count} applied migration(s) but the \
live database contains zero tables; the schema may have been \
dropped out-of-band",
),
location: None,
});
}
emit_out_of_order_diagnostics(&ledger_rows, policy, &mut diagnostics);
}
(applied_count, unfinished_count, latest_applied_version)
} else {
if emit_ledger_diagnostics {
diagnostics.push(VerifyDiagnostic {
code: "D621".to_string(),
severity: VerifySeverity::Error,
message: "ledger table `djogi_schema_migrations` not found — run \
`djogi migrations apply` or `djogi migrations baseline` \
first; verify is read-only and will not bootstrap the ledger"
.to_string(),
location: None,
});
}
(0, 0, None)
};
let live = live_schema_for_repair(ctx, bucket).await?;
diff_tables(snapshot, &live, &mut diagnostics);
diff_indexes(snapshot, &live, &mut diagnostics);
diff_advisory_fields(snapshot, &mut diagnostics);
diagnostics.sort_by_key(|d| d.sort_key());
Ok(VerifyReport {
diagnostics,
latest_applied_version,
applied_count,
unfinished_count,
})
}
fn emit_out_of_order_diagnostics(
ledger_rows: &[LedgerRow],
policy: &crate::config::PolicyConfig,
diagnostics: &mut Vec<VerifyDiagnostic>,
) {
let severity = if policy.strict_out_of_order {
VerifySeverity::Error
} else {
VerifySeverity::Warning
};
for row in ledger_rows {
if !row.out_of_order_flag {
continue;
}
let conflicting_peer: Option<&LedgerRow> = ledger_rows
.iter()
.filter(|r| {
r.app_label == row.app_label
&& r.version > row.version
&& matches!(
r.status,
LedgerStatus::Applied | LedgerStatus::Faked | LedgerStatus::Baseline
)
})
.max_by(|a, b| a.version.cmp(&b.version));
let location = format!("version:{}", row.version);
let message = match conflicting_peer {
Some(peer) => format!(
"out-of-order migration `{version}` (app={app}) was applied \
below peer `{peer}` (which carries the same or higher \
version)",
version = row.version,
app = if row.app_label.is_empty() {
"_global_"
} else {
row.app_label.as_str()
},
peer = peer.version,
),
None => format!(
"out-of-order migration `{version}` (app={app}) was \
recorded with out_of_order_flag = TRUE but no \
higher-version peer is currently in the ledger; the \
peer may have been rolled back",
version = row.version,
app = if row.app_label.is_empty() {
"_global_"
} else {
row.app_label.as_str()
},
),
};
diagnostics.push(VerifyDiagnostic {
code: "D622".to_string(),
severity,
message,
location: Some(location),
});
}
}
async fn ledger_table_exists(ctx: &mut DjogiContext) -> Result<bool, VerifyRunError> {
let row = ctx
.query_one(
"SELECT EXISTS ( \
SELECT 1 FROM pg_class c \
JOIN pg_namespace n ON n.oid = c.relnamespace \
WHERE n.nspname = 'public' \
AND c.relname = 'djogi_schema_migrations' \
AND c.relkind = 'r' \
)",
&[],
)
.await
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "ledger_present_probe",
source: e,
})?;
let exists: bool = row
.try_get(0)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "ledger_present_probe.bool",
source: DjogiError::from(e),
})?;
Ok(exists)
}
async fn project_live_schema(ctx: &mut DjogiContext) -> Result<AppliedSchema, VerifyRunError> {
let foreign_keys = read_foreign_keys(ctx).await?;
let tables = read_tables(ctx, &foreign_keys).await?;
let mut models: BTreeMap<String, TableSchema> = BTreeMap::new();
for t in tables {
models.insert(t.table.clone(), t);
}
let indexes = read_indexes(ctx).await?;
Ok(AppliedSchema {
djogi_version: String::new(),
enums: BTreeMap::new(),
format_version: super::schema::SNAPSHOT_FORMAT_VERSION.to_string(),
generated_at: String::new(),
indexes,
models,
registered_apps: Vec::new(),
})
}
async fn read_tables(
ctx: &mut DjogiContext,
foreign_keys: &[LiveForeignKey],
) -> Result<Vec<TableSchema>, VerifyRunError> {
let table_rows = ctx
.query_all(
"SELECT c.relname::text \
FROM pg_class c \
JOIN pg_namespace n ON n.oid = c.relnamespace \
WHERE c.relkind = 'r' \
AND n.nspname = 'public' \
AND c.relname <> 'djogi_schema_migrations' \
ORDER BY c.relname",
&[],
)
.await
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "tables",
source: e,
})?;
let mut columns_by_table = read_all_columns(ctx).await?;
let mut pk_by_table = read_all_primary_key_columns(ctx).await?;
let mut out: Vec<TableSchema> = Vec::with_capacity(table_rows.len());
for row in &table_rows {
let table_name: String =
row.try_get(0)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "tables.relname",
source: DjogiError::from(e),
})?;
if is_heeranjid_artifact_table(&table_name) {
continue;
}
let mut columns = columns_by_table.remove(&table_name).unwrap_or_default();
attach_foreign_keys(&table_name, &mut columns, foreign_keys);
let primary_key_columns = pk_by_table.remove(&table_name).unwrap_or_default();
out.push(TableSchema {
app: None,
columns,
fts: None,
is_through: false,
moved_from_app: None,
partition: None,
primary_key: super::schema::PrimaryKeySchema {
columns: primary_key_columns,
kind: super::schema::PkKindSchema::HeerId,
},
rationale: None,
renamed_from: None,
rls_enabled: false,
table: table_name,
tenant_key: None,
exclusion_constraints: Vec::new(),
table_comment: None,
storage_params: None,
tablespace: None,
});
}
Ok(out)
}
async fn read_all_columns(
ctx: &mut DjogiContext,
) -> Result<BTreeMap<String, Vec<ColumnSchema>>, VerifyRunError> {
let rows = ctx
.query_all(
"SELECT c.relname::text, \
a.attname::text, \
format_type(a.atttypid, a.atttypmod)::text, \
a.attnotnull, \
pg_get_expr(d.adbin, d.adrelid) \
FROM pg_attribute a \
JOIN pg_class c ON c.oid = a.attrelid \
JOIN pg_namespace n ON n.oid = c.relnamespace \
LEFT JOIN pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum \
WHERE n.nspname = 'public' \
AND c.relkind = 'r' \
AND a.attnum > 0 \
AND NOT a.attisdropped \
ORDER BY c.relname, a.attnum",
&[],
)
.await
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "columns",
source: e,
})?;
let mut out: BTreeMap<String, Vec<ColumnSchema>> = BTreeMap::new();
for row in &rows {
let table_name: String =
row.try_get(0)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "columns.relname",
source: DjogiError::from(e),
})?;
let name: String = row
.try_get(1)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "columns.attname",
source: DjogiError::from(e),
})?;
let sql_type: String = row
.try_get(2)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "columns.type",
source: DjogiError::from(e),
})?;
let attnotnull: bool = row
.try_get(3)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "columns.attnotnull",
source: DjogiError::from(e),
})?;
let default_sql: Option<String> =
row.try_get(4)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "columns.default",
source: DjogiError::from(e),
})?;
out.entry(table_name).or_default().push(ColumnSchema {
check: None,
comment: None,
default_sql,
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name,
nullable: !attnotnull,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: render_type_for_compare(&sql_type),
unique: false,
type_change_using: None,
});
}
Ok(out)
}
fn attach_foreign_keys(
table_name: &str,
columns: &mut [ColumnSchema],
foreign_keys: &[LiveForeignKey],
) {
for column in columns {
if let Some(fk) = foreign_keys
.iter()
.find(|fk| fk.table == table_name && fk.column == column.name)
{
column.foreign_key = Some(ForeignKeySchema {
deferrable: fk.deferrable,
initially_deferred: fk.initially_deferred,
on_delete: fk.on_delete,
ref_column: fk.ref_column.clone(),
ref_table: fk.ref_table.clone(),
});
column.on_delete = Some(fk.on_delete);
}
}
}
async fn read_all_primary_key_columns(
ctx: &mut DjogiContext,
) -> Result<BTreeMap<String, Vec<String>>, VerifyRunError> {
let rows = ctx
.query_all(
"SELECT c.relname::text, \
a.attname::text \
FROM pg_constraint con \
JOIN pg_class c ON c.oid = con.conrelid \
JOIN pg_namespace n ON n.oid = c.relnamespace \
JOIN pg_attribute a ON a.attrelid = con.conrelid \
AND a.attnum = ANY(con.conkey) \
WHERE n.nspname = 'public' \
AND con.contype = 'p' \
ORDER BY c.relname, array_position(con.conkey, a.attnum)",
&[],
)
.await
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "primary_key",
source: e,
})?;
let mut out: BTreeMap<String, Vec<String>> = BTreeMap::new();
for row in &rows {
let table_name: String =
row.try_get(0)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "primary_key.relname",
source: DjogiError::from(e),
})?;
let col: String = row
.try_get(1)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "primary_key.attname",
source: DjogiError::from(e),
})?;
out.entry(table_name).or_default().push(col);
}
Ok(out)
}
async fn read_indexes(ctx: &mut DjogiContext) -> Result<Vec<IndexSchema>, VerifyRunError> {
let rows = ctx
.query_all(
"SELECT i.relname::text, \
t.relname::text, \
ix.indisunique, \
am.amname::text \
FROM pg_index ix \
JOIN pg_class i ON i.oid = ix.indexrelid \
JOIN pg_class t ON t.oid = ix.indrelid \
JOIN pg_namespace n ON n.oid = t.relnamespace \
JOIN pg_am am ON am.oid = i.relam \
WHERE n.nspname = 'public' \
AND ix.indisprimary = false \
AND t.relname <> 'djogi_schema_migrations' \
ORDER BY i.relname",
&[],
)
.await
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "indexes",
source: e,
})?;
let mut columns_by_index = read_all_index_columns(ctx).await?;
let mut out: Vec<IndexSchema> = Vec::with_capacity(rows.len());
for row in &rows {
let name: String = row
.try_get(0)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "indexes.relname",
source: DjogiError::from(e),
})?;
let table: String = row
.try_get(1)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "indexes.table",
source: DjogiError::from(e),
})?;
if is_heeranjid_artifact_table(&table) {
continue;
}
let is_unique: bool = row
.try_get(2)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "indexes.indisunique",
source: DjogiError::from(e),
})?;
let amname: String = row
.try_get(3)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "indexes.amname",
source: DjogiError::from(e),
})?;
let index_columns = columns_by_index.remove(&name).unwrap_or_default();
out.push(IndexSchema {
extension_dependency: None,
include: Vec::new(),
index_type: pg_amname_to_index_type(&amname),
kind: if is_unique {
IndexKindSchema::UniqueIndex
} else {
IndexKindSchema::NonUnique
},
name,
nulls_not_distinct: false,
predicate: None,
requires_out_of_transaction: false,
table,
target: IndexTargetSchema::Columns(index_columns),
});
}
Ok(out)
}
async fn read_all_index_columns(
ctx: &mut DjogiContext,
) -> Result<BTreeMap<String, Vec<IndexColumnSchema>>, VerifyRunError> {
let rows = ctx
.query_all(
"SELECT i.relname::text, \
a.attname::text \
FROM pg_index ix \
JOIN pg_class i ON i.oid = ix.indexrelid \
JOIN pg_namespace n ON n.oid = i.relnamespace \
JOIN unnest(ix.indkey) WITH ORDINALITY AS k(attnum, ord) ON TRUE \
JOIN pg_attribute a ON a.attrelid = ix.indrelid AND a.attnum = k.attnum \
WHERE n.nspname = 'public' \
AND a.attnum > 0 \
ORDER BY i.relname, k.ord",
&[],
)
.await
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "index_columns",
source: e,
})?;
let mut out: BTreeMap<String, Vec<IndexColumnSchema>> = BTreeMap::new();
for row in &rows {
let index_name: String =
row.try_get(0)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "index_columns.relname",
source: DjogiError::from(e),
})?;
let col_name: String = row
.try_get(1)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "index_columns.attname",
source: DjogiError::from(e),
})?;
out.entry(index_name).or_default().push(IndexColumnSchema {
name: col_name,
nulls: IndexNullsOrderSchema::Default,
opclass: None,
order: IndexOrderSchema::Asc,
});
}
Ok(out)
}
fn pg_amname_to_index_type(amname: &str) -> IndexTypeSchema {
match amname {
"btree" => IndexTypeSchema::BTree,
"hash" => IndexTypeSchema::Hash,
"gin" => IndexTypeSchema::Gin,
"gist" => IndexTypeSchema::Gist,
"spgist" => IndexTypeSchema::Spgist,
"brin" => IndexTypeSchema::Brin,
_ => IndexTypeSchema::BTree,
}
}
async fn read_applied_ledger(ctx: &mut DjogiContext) -> Result<Vec<LedgerRow>, VerifyRunError> {
let sql = format!("{LEDGER_SELECT_COLS} ORDER BY applied_at, version");
let rows = ctx
.query_all(&sql, &[])
.await
.map_err(|e| VerifyRunError::LedgerQueryFailed { source: e })?;
let mut out: Vec<LedgerRow> = Vec::with_capacity(rows.len());
for row in &rows {
let ledger_row =
LedgerRow::try_from(row).map_err(|e| VerifyRunError::LedgerQueryFailed {
source: DjogiError::from(e),
})?;
out.push(ledger_row);
}
Ok(out)
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct LiveForeignKey {
table: String,
column: String,
ref_table: String,
ref_column: String,
on_delete: OnDeleteSchema,
deferrable: bool,
initially_deferred: bool,
}
async fn read_foreign_keys(ctx: &mut DjogiContext) -> Result<Vec<LiveForeignKey>, VerifyRunError> {
let rows = ctx
.query_all(
"SELECT c.relname::text, \
a.attname::text, \
rc.relname::text, \
ra.attname::text, \
con.confdeltype::text, \
con.condeferrable, \
con.condeferred \
FROM pg_constraint con \
JOIN pg_class c ON c.oid = con.conrelid \
JOIN pg_namespace n ON n.oid = c.relnamespace \
JOIN pg_attribute a ON a.attrelid = con.conrelid \
AND a.attnum = ANY(con.conkey) \
JOIN pg_class rc ON rc.oid = con.confrelid \
JOIN pg_attribute ra ON ra.attrelid = con.confrelid \
AND ra.attnum = ANY(con.confkey) \
WHERE n.nspname = 'public' \
AND con.contype = 'f' \
AND array_position(con.conkey, a.attnum) \
= array_position(con.confkey, ra.attnum) \
ORDER BY c.relname, a.attname",
&[],
)
.await
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "foreign_keys",
source: e,
})?;
let mut out: Vec<LiveForeignKey> = Vec::with_capacity(rows.len());
for row in &rows {
let table: String = row
.try_get(0)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "foreign_keys.table",
source: DjogiError::from(e),
})?;
let column: String = row
.try_get(1)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "foreign_keys.column",
source: DjogiError::from(e),
})?;
let ref_table: String = row
.try_get(2)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "foreign_keys.ref_table",
source: DjogiError::from(e),
})?;
let ref_column: String =
row.try_get(3)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "foreign_keys.ref_column",
source: DjogiError::from(e),
})?;
let on_delete_code: String =
row.try_get(4)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "foreign_keys.confdeltype",
source: DjogiError::from(e),
})?;
let deferrable: bool = row
.try_get(5)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "foreign_keys.condeferrable",
source: DjogiError::from(e),
})?;
let initially_deferred: bool =
row.try_get(6)
.map_err(|e| VerifyRunError::CatalogQueryFailed {
query_label: "foreign_keys.condeferred",
source: DjogiError::from(e),
})?;
if is_heeranjid_artifact_table(&table) || is_heeranjid_artifact_table(&ref_table) {
continue;
}
let on_delete = match on_delete_code.as_bytes().first().copied() {
Some(b'r') => OnDeleteSchema::Restrict,
Some(b'c') => OnDeleteSchema::Cascade,
Some(b'n') => OnDeleteSchema::SetNull,
Some(b'd') => OnDeleteSchema::SetDefault,
Some(b'a') => OnDeleteSchema::NoAction,
_ => {
return Err(VerifyRunError::CatalogQueryFailed {
query_label: "foreign_keys.confdeltype",
source: DjogiError::Db(crate::error::DbError::other(format!(
"unexpected pg_constraint.confdeltype value {on_delete_code:?}"
))),
});
}
};
out.push(LiveForeignKey {
table,
column,
ref_table,
ref_column,
on_delete,
deferrable,
initially_deferred,
});
}
Ok(out)
}
fn diff_tables(
snapshot: &AppliedSchema,
live: &AppliedSchema,
diagnostics: &mut Vec<VerifyDiagnostic>,
) {
for name in snapshot.models.keys() {
if !live.models.contains_key(name) {
diagnostics.push(VerifyDiagnostic {
code: "D601".to_string(),
severity: VerifySeverity::Error,
message: format!(
"table `{name}` exists in snapshot but is missing from \
the live database; the schema may have been dropped \
out-of-band or the migration that creates it was \
never applied",
),
location: Some(name.clone()),
});
}
}
for name in live.models.keys() {
if !snapshot.models.contains_key(name) {
diagnostics.push(VerifyDiagnostic {
code: "D602".to_string(),
severity: VerifySeverity::Error,
message: format!(
"table `{name}` exists in the live database but is not \
in the snapshot; either an out-of-band migration ran \
or the snapshot is stale",
),
location: Some(name.clone()),
});
}
}
for (name, snap_table) in &snapshot.models {
let Some(live_table) = live.models.get(name) else {
continue;
};
let snap_cols: BTreeMap<&str, &ColumnSchema> = snap_table
.columns
.iter()
.map(|c| (c.name.as_str(), c))
.collect();
let live_cols: BTreeMap<&str, &ColumnSchema> = live_table
.columns
.iter()
.map(|c| (c.name.as_str(), c))
.collect();
for col_name in snap_cols.keys() {
if !live_cols.contains_key(col_name) {
diagnostics.push(VerifyDiagnostic {
code: "D603".to_string(),
severity: VerifySeverity::Error,
message: format!(
"column `{name}.{col_name}` exists in snapshot but \
is missing from the live database",
),
location: Some(format!("{name}.{col_name}")),
});
}
}
for col_name in live_cols.keys() {
if !snap_cols.contains_key(col_name) {
diagnostics.push(VerifyDiagnostic {
code: "D604".to_string(),
severity: VerifySeverity::Error,
message: format!(
"column `{name}.{col_name}` exists in the live \
database but not in the snapshot",
),
location: Some(format!("{name}.{col_name}")),
});
}
}
for (col_name, snap_col) in &snap_cols {
let Some(live_col) = live_cols.get(col_name) else {
continue;
};
if snap_col.nullable != live_col.nullable {
diagnostics.push(VerifyDiagnostic {
code: "D605".to_string(),
severity: VerifySeverity::Error,
message: format!(
"column `{name}.{col_name}` nullability differs: \
snapshot {snap_n}, live {live_n}",
snap_n = snap_col.nullable,
live_n = live_col.nullable,
),
location: Some(format!("{name}.{col_name}")),
});
}
let snap_canon = render_type_for_compare(&snap_col.sql_type);
let live_canon = render_type_for_compare(&live_col.sql_type);
if snap_canon != live_canon {
diagnostics.push(VerifyDiagnostic {
code: "D606".to_string(),
severity: VerifySeverity::Warning,
message: format!(
"column `{name}.{col_name}` type differs (advisory): \
snapshot `{s}`, live `{l}`",
s = snap_col.sql_type,
l = live_col.sql_type,
),
location: Some(format!("{name}.{col_name}")),
});
}
let snap_default = normalize_default_expr(snap_col.default_sql.as_deref());
let live_default = normalize_default_expr(live_col.default_sql.as_deref());
if snap_default != live_default {
diagnostics.push(VerifyDiagnostic {
code: "D607".to_string(),
severity: VerifySeverity::Error,
message: format!(
"column `{name}.{col_name}` DEFAULT differs: \
snapshot {s}, live {l}",
s = render_default_for_message(&snap_default),
l = render_default_for_message(&live_default),
),
location: Some(format!("{name}.{col_name}")),
});
}
if snap_col.foreign_key != live_col.foreign_key {
diagnostics.push(VerifyDiagnostic {
code: "D609".to_string(),
severity: VerifySeverity::Error,
message: format!(
"column `{name}.{col_name}` foreign key differs: \
snapshot {s:?}, live {l:?}",
s = snap_col.foreign_key,
l = live_col.foreign_key,
),
location: Some(format!("{name}.{col_name}")),
});
}
}
diff_primary_key(
name,
&snap_table.primary_key,
&live_table.primary_key,
diagnostics,
);
}
}
fn diff_primary_key(
table_name: &str,
snap_pk: &PrimaryKeySchema,
live_pk: &PrimaryKeySchema,
diagnostics: &mut Vec<VerifyDiagnostic>,
) {
let snap_empty = snap_pk.columns.is_empty();
let live_empty = live_pk.columns.is_empty();
if snap_empty && live_empty {
return;
}
if snap_empty != live_empty || snap_pk.columns != live_pk.columns {
diagnostics.push(VerifyDiagnostic {
code: "D608".to_string(),
severity: VerifySeverity::Error,
message: format!(
"table `{table_name}` primary key differs: \
snapshot {snap:?}, live {live:?}",
snap = snap_pk.columns,
live = live_pk.columns,
),
location: Some(format!("{table_name}.<pk>")),
});
}
}
fn diff_indexes(
snapshot: &AppliedSchema,
live: &AppliedSchema,
diagnostics: &mut Vec<VerifyDiagnostic>,
) {
let snap_by_name: BTreeMap<&str, &IndexSchema> = snapshot
.indexes
.iter()
.map(|i| (i.name.as_str(), i))
.collect();
let live_by_name: BTreeMap<&str, &IndexSchema> =
live.indexes.iter().map(|i| (i.name.as_str(), i)).collect();
for name in snap_by_name.keys() {
if !live_by_name.contains_key(name) {
diagnostics.push(VerifyDiagnostic {
code: "D610".to_string(),
severity: VerifySeverity::Error,
message: format!(
"index `{name}` exists in snapshot but is missing from \
the live database",
),
location: Some(format!("index:{name}")),
});
}
}
for name in live_by_name.keys() {
if !snap_by_name.contains_key(name) {
diagnostics.push(VerifyDiagnostic {
code: "D611".to_string(),
severity: VerifySeverity::Warning,
message: format!(
"index `{name}` exists in the live database but is not \
in the snapshot's index list (may be a constraint-backed \
auto-index)",
),
location: Some(format!("index:{name}")),
});
}
}
for (name, snap_idx) in &snap_by_name {
let Some(live_idx) = live_by_name.get(name) else {
continue;
};
if snap_idx.table != live_idx.table {
diagnostics.push(VerifyDiagnostic {
code: "D615".to_string(),
severity: VerifySeverity::Error,
message: format!(
"index `{name}` is on table `{l}` in the live database \
but the snapshot declares it on `{s}`",
s = snap_idx.table,
l = live_idx.table,
),
location: Some(format!("index:{name}")),
});
}
let snap_cols = index_target_column_names(&snap_idx.target);
let live_cols = index_target_column_names(&live_idx.target);
if snap_cols != live_cols {
diagnostics.push(VerifyDiagnostic {
code: "D612".to_string(),
severity: VerifySeverity::Error,
message: format!(
"index `{name}` columns differ: snapshot {s:?}, live {l:?}",
s = snap_cols,
l = live_cols,
),
location: Some(format!("index:{name}")),
});
}
let snap_unique = matches!(
snap_idx.kind,
IndexKindSchema::UniqueIndex | IndexKindSchema::UniqueConstraint
);
let live_unique = matches!(
live_idx.kind,
IndexKindSchema::UniqueIndex | IndexKindSchema::UniqueConstraint
);
if snap_unique != live_unique {
diagnostics.push(VerifyDiagnostic {
code: "D613".to_string(),
severity: VerifySeverity::Error,
message: format!(
"index `{name}` uniqueness differs: snapshot {s}, live {l}",
s = snap_unique,
l = live_unique,
),
location: Some(format!("index:{name}")),
});
}
if snap_idx.index_type != live_idx.index_type {
diagnostics.push(VerifyDiagnostic {
code: "D614".to_string(),
severity: VerifySeverity::Warning,
message: format!(
"index `{name}` method differs: snapshot {s:?}, live {l:?}",
s = snap_idx.index_type,
l = live_idx.index_type,
),
location: Some(format!("index:{name}")),
});
}
if !snap_idx.include.is_empty() || snap_idx.predicate.is_some() {
diagnostics.push(VerifyDiagnostic {
code: "D693".to_string(),
severity: VerifySeverity::Info,
message: format!(
"index `{name}` declares INCLUDE / partial-predicate; T5 \
verify does not yet project these from the live catalog \
(deferred to T8)",
),
location: Some(format!("index:{name}")),
});
}
}
}
fn index_target_column_names(target: &IndexTargetSchema) -> Vec<&str> {
match target {
IndexTargetSchema::Columns(cols) => cols.iter().map(|c| c.name.as_str()).collect(),
IndexTargetSchema::Expression(_) => Vec::new(),
}
}
fn normalize_default_expr(expr: Option<&str>) -> Option<String> {
let raw = expr?.trim();
if raw.is_empty() {
return None;
}
let mut current = raw.to_string();
loop {
let bytes = current.as_bytes();
let mut in_string = false;
let mut last_double_colon: Option<usize> = None;
let mut i = 0usize;
while i + 1 < bytes.len() {
let b = bytes[i];
if b == b'\'' {
in_string = !in_string;
} else if !in_string && b == b':' && bytes[i + 1] == b':' {
last_double_colon = Some(i);
i += 2;
continue;
}
i += 1;
}
match last_double_colon {
Some(idx) => {
let trimmed = current[..idx].trim_end();
if trimmed.is_empty() {
return None;
}
let next = trimmed.to_string();
if next == current {
break;
}
current = next;
}
None => break,
}
}
if current.is_empty() {
None
} else {
Some(current)
}
}
fn render_default_for_message(d: &Option<String>) -> String {
match d {
Some(s) => format!("`{s}`"),
None => "<no default>".to_string(),
}
}
fn diff_advisory_fields(snapshot: &AppliedSchema, diagnostics: &mut Vec<VerifyDiagnostic>) {
let fts_tables: Vec<&str> = snapshot
.models
.iter()
.filter_map(|(n, t)| t.fts.as_ref().map(|_| n.as_str()))
.collect();
if !fts_tables.is_empty() {
let location = fts_tables.first().copied().map(|s| s.to_string());
diagnostics.push(VerifyDiagnostic {
code: "D690".to_string(),
severity: VerifySeverity::Info,
message: format!(
"{n} table(s) declare FTS configuration; T5 verify does not \
yet check FTS triggers / generated columns against the live \
catalog (deferred to T8)",
n = fts_tables.len(),
),
location,
});
}
let partitioned: Vec<&str> = snapshot
.models
.iter()
.filter_map(|(n, t)| t.partition.as_ref().map(|_| n.as_str()))
.collect();
if !partitioned.is_empty() {
let location = partitioned.first().copied().map(|s| s.to_string());
diagnostics.push(VerifyDiagnostic {
code: "D691".to_string(),
severity: VerifySeverity::Info,
message: format!(
"{n} table(s) declare a partition strategy; T5 verify does \
not yet check partition method / column against the live \
catalog (deferred to T8)",
n = partitioned.len(),
),
location,
});
}
if !snapshot.enums.is_empty() {
diagnostics.push(VerifyDiagnostic {
code: "D692".to_string(),
severity: VerifySeverity::Info,
message: format!(
"{n} enum type(s) declared; T5 verify does not yet check \
enum variants against the live `pg_enum` catalog (deferred \
to T8)",
n = snapshot.enums.len(),
),
location: None,
});
}
}
fn render_type_for_compare(s: &str) -> String {
let lower: String = s
.as_bytes()
.iter()
.map(|b| b.to_ascii_lowercase() as char)
.collect();
let aliases: &[(&str, &str)] = &[
("character varying", "varchar"),
("timestamp with time zone", "timestamptz"),
("timestamp without time zone", "timestamp"),
("integer", "int4"),
("bigint", "int8"),
("smallint", "int2"),
("double precision", "float8"),
("real", "float4"),
("boolean", "bool"),
];
let mut out = lower;
for (from, to) in aliases {
out = out.replace(from, to);
}
out
}
pub(super) async fn live_schema_for_repair(
ctx: &mut DjogiContext,
bucket: &BucketKey,
) -> Result<AppliedSchema, VerifyRunError> {
let mut full = project_live_schema(ctx).await?;
use crate::AppDescriptor;
use crate::descriptor::ModelDescriptor;
let mut this_bucket_tables: std::collections::BTreeSet<&str> =
std::collections::BTreeSet::new();
let mut all_app_tables: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new();
for m in inventory::iter::<ModelDescriptor> {
let label = m.app.unwrap_or(AppDescriptor::GLOBAL_LABEL);
if label == bucket.app.as_str() {
this_bucket_tables.insert(m.table_name);
}
if label != AppDescriptor::GLOBAL_LABEL {
all_app_tables.insert(m.table_name);
}
}
let is_global_bucket = bucket.app.as_str() == AppDescriptor::GLOBAL_LABEL;
full.models.retain(|table_name, _| {
if is_global_bucket {
!all_app_tables.contains(table_name.as_str())
} else {
this_bucket_tables.contains(table_name.as_str())
}
});
let kept_tables: std::collections::BTreeSet<String> = full.models.keys().cloned().collect();
full.indexes.retain(|i| kept_tables.contains(&i.table));
full.registered_apps = vec![bucket.app.clone()];
Ok(full)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::migrate::schema::{
ColumnSchema, ForeignKeySchema, IndexKindSchema, IndexSchema, IndexTargetSchema,
IndexTypeSchema, OnDeleteSchema, PkKindSchema, PrimaryKeySchema, RelationKindSchema,
TableSchema,
};
fn empty_snapshot() -> AppliedSchema {
AppliedSchema {
djogi_version: "0.1.0".to_string(),
enums: BTreeMap::new(),
format_version: super::super::schema::SNAPSHOT_FORMAT_VERSION.to_string(),
generated_at: "2026-04-25T00:00:00Z".to_string(),
indexes: Vec::new(),
models: BTreeMap::new(),
registered_apps: vec!["".to_string()],
}
}
fn col(name: &str, sql_type: &str, nullable: bool) -> ColumnSchema {
ColumnSchema {
check: None,
comment: None,
default_sql: None,
foreign_key: None,
generated: None,
identity: None,
index_type: None,
indexed: false,
max_length: None,
name: name.to_string(),
nullable,
on_delete: None,
outbox_exclude: false,
rationale: None,
relation_kind: None,
renamed_from: None,
sequence_within: None,
sql_type: sql_type.to_string(),
unique: false,
type_change_using: None,
}
}
fn fk_column(
name: &str,
ref_table: &str,
ref_column: &str,
on_delete: OnDeleteSchema,
deferrable: bool,
initially_deferred: bool,
) -> ColumnSchema {
ColumnSchema {
foreign_key: Some(ForeignKeySchema {
deferrable,
initially_deferred,
on_delete,
ref_column: ref_column.to_string(),
ref_table: ref_table.to_string(),
}),
on_delete: Some(on_delete),
relation_kind: Some(RelationKindSchema::ForeignKey),
..col(name, "BIGINT", false)
}
}
fn table(name: &str, columns: Vec<ColumnSchema>) -> TableSchema {
TableSchema {
app: None,
columns,
exclusion_constraints: Vec::new(),
fts: None,
is_through: false,
moved_from_app: None,
partition: None,
primary_key: PrimaryKeySchema {
columns: vec!["id".to_string()],
kind: PkKindSchema::HeerId,
},
rationale: None,
renamed_from: None,
rls_enabled: false,
table: name.to_string(),
table_comment: None,
storage_params: None,
tablespace: None,
tenant_key: None,
}
}
fn idx(name: &str, table: &str, unique: bool) -> IndexSchema {
IndexSchema {
extension_dependency: None,
include: Vec::new(),
index_type: IndexTypeSchema::BTree,
kind: if unique {
IndexKindSchema::UniqueIndex
} else {
IndexKindSchema::NonUnique
},
name: name.to_string(),
nulls_not_distinct: false,
predicate: None,
requires_out_of_transaction: false,
table: table.to_string(),
target: IndexTargetSchema::Columns(Vec::new()),
}
}
#[test]
fn diagnostics_sort_by_code_then_location() {
let mut diagnostics = [
VerifyDiagnostic {
code: "D602".to_string(),
severity: VerifySeverity::Error,
message: "msg".to_string(),
location: Some("zebra".to_string()),
},
VerifyDiagnostic {
code: "D601".to_string(),
severity: VerifySeverity::Error,
message: "msg".to_string(),
location: Some("alpha".to_string()),
},
VerifyDiagnostic {
code: "D601".to_string(),
severity: VerifySeverity::Error,
message: "msg".to_string(),
location: Some("beta".to_string()),
},
];
diagnostics.sort_by_key(|d| d.sort_key());
let codes: Vec<(&str, &str)> = diagnostics
.iter()
.map(|d| (d.code.as_str(), d.location.as_deref().unwrap_or("")))
.collect();
assert_eq!(
codes,
vec![("D601", "alpha"), ("D601", "beta"), ("D602", "zebra")]
);
}
#[test]
fn diff_tables_clean_match_emits_no_diagnostics() {
let mut snap = empty_snapshot();
snap.models.insert(
"users".to_string(),
table("users", vec![col("id", "BIGINT", false)]),
);
let mut live = empty_snapshot();
live.models.insert(
"users".to_string(),
table("users", vec![col("id", "bigint", false)]),
);
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
assert!(
diagnostics.is_empty(),
"clean match must emit no diagnostics; got {diagnostics:?}"
);
}
#[test]
fn diff_tables_surfaces_missing_table_as_d601() {
let mut snap = empty_snapshot();
snap.models.insert(
"users".to_string(),
table("users", vec![col("id", "BIGINT", false)]),
);
let live = empty_snapshot();
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
assert_eq!(diagnostics.len(), 1);
assert_eq!(diagnostics[0].code, "D601");
assert_eq!(diagnostics[0].severity, VerifySeverity::Error);
assert_eq!(diagnostics[0].location.as_deref(), Some("users"));
}
#[test]
fn diff_tables_surfaces_extra_live_table_as_d602() {
let snap = empty_snapshot();
let mut live = empty_snapshot();
live.models.insert(
"widgets".to_string(),
table("widgets", vec![col("id", "bigint", false)]),
);
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
assert_eq!(diagnostics.len(), 1);
assert_eq!(diagnostics[0].code, "D602");
assert_eq!(diagnostics[0].severity, VerifySeverity::Error);
}
#[test]
fn diff_tables_surfaces_missing_column_as_d603() {
let mut snap = empty_snapshot();
snap.models.insert(
"users".to_string(),
table(
"users",
vec![col("id", "BIGINT", false), col("email", "TEXT", false)],
),
);
let mut live = empty_snapshot();
live.models.insert(
"users".to_string(),
table("users", vec![col("id", "bigint", false)]),
);
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
let by_code: BTreeMap<_, _> = diagnostics.iter().map(|d| (d.code.as_str(), d)).collect();
assert!(by_code.contains_key("D603"));
assert_eq!(by_code["D603"].location.as_deref(), Some("users.email"));
}
#[test]
fn diff_tables_surfaces_extra_live_column_as_d604() {
let mut snap = empty_snapshot();
snap.models.insert(
"users".to_string(),
table("users", vec![col("id", "BIGINT", false)]),
);
let mut live = empty_snapshot();
live.models.insert(
"users".to_string(),
table(
"users",
vec![col("id", "bigint", false), col("rogue", "text", true)],
),
);
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
let by_code: BTreeMap<_, _> = diagnostics.iter().map(|d| (d.code.as_str(), d)).collect();
assert!(by_code.contains_key("D604"));
assert_eq!(by_code["D604"].location.as_deref(), Some("users.rogue"));
}
#[test]
fn diff_tables_surfaces_nullability_drift_as_d605() {
let mut snap = empty_snapshot();
snap.models.insert(
"users".to_string(),
table("users", vec![col("email", "TEXT", false)]),
);
let mut live = empty_snapshot();
live.models.insert(
"users".to_string(),
table("users", vec![col("email", "text", true)]),
);
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
assert!(diagnostics.iter().any(|d| d.code == "D605"));
}
#[test]
fn diff_tables_surfaces_type_drift_as_advisory_d606() {
let mut snap = empty_snapshot();
snap.models.insert(
"users".to_string(),
table("users", vec![col("age", "INTEGER", true)]),
);
let mut live = empty_snapshot();
live.models.insert(
"users".to_string(),
table("users", vec![col("age", "BIGINT", true)]),
);
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
let d606 = diagnostics
.iter()
.find(|d| d.code == "D606")
.expect("D606 expected");
assert_eq!(d606.severity, VerifySeverity::Warning);
}
#[test]
fn diff_tables_surfaces_foreign_key_drift_as_d609() {
let mut snap = empty_snapshot();
snap.models.insert(
"posts".to_string(),
table(
"posts",
vec![
col("id", "BIGINT", false),
fk_column(
"author_id",
"users",
"id",
OnDeleteSchema::Restrict,
true,
true,
),
],
),
);
let mut live = empty_snapshot();
live.models.insert(
"posts".to_string(),
table(
"posts",
vec![
col("id", "bigint", false),
fk_column(
"author_id",
"users",
"id",
OnDeleteSchema::Restrict,
true,
false,
),
],
),
);
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
let d609 = diagnostics
.iter()
.find(|d| d.code == "D609")
.expect("D609 expected");
assert_eq!(d609.severity, VerifySeverity::Error);
assert_eq!(d609.location.as_deref(), Some("posts.author_id"));
}
#[test]
fn diff_indexes_missing_in_live_is_error() {
let mut snap = empty_snapshot();
snap.indexes.push(idx("users_email_idx", "users", false));
let live = empty_snapshot();
let mut diagnostics = Vec::new();
diff_indexes(&snap, &live, &mut diagnostics);
assert!(diagnostics.iter().any(|d| d.code == "D610"));
}
#[test]
fn diff_indexes_extra_in_live_is_warning() {
let snap = empty_snapshot();
let mut live = empty_snapshot();
live.indexes.push(idx("users_email_idx", "users", true));
let mut diagnostics = Vec::new();
diff_indexes(&snap, &live, &mut diagnostics);
let d611 = diagnostics
.iter()
.find(|d| d.code == "D611")
.expect("D611 expected");
assert_eq!(d611.severity, VerifySeverity::Warning);
}
#[test]
fn render_type_aliases_match_pg_renderings() {
assert_eq!(render_type_for_compare("BIGINT"), "int8");
assert_eq!(render_type_for_compare("bigint"), "int8");
assert_eq!(
render_type_for_compare("character varying(255)"),
"varchar(255)"
);
assert_eq!(
render_type_for_compare("timestamp with time zone"),
"timestamptz"
);
assert_eq!(render_type_for_compare("BOOLEAN"), "bool");
assert_eq!(render_type_for_compare("INTEGER"), "int4");
assert_eq!(render_type_for_compare("SMALLINT"), "int2");
assert_eq!(render_type_for_compare("DOUBLE PRECISION"), "float8");
assert_eq!(render_type_for_compare("REAL"), "float4");
}
#[test]
fn render_type_preserves_unknown_names_lowercase() {
assert_eq!(render_type_for_compare("CITEXT"), "citext");
}
#[test]
fn verify_report_has_errors_detects_error_diagnostics() {
let r = VerifyReport {
diagnostics: vec![VerifyDiagnostic {
code: "D601".to_string(),
severity: VerifySeverity::Error,
message: "x".to_string(),
location: None,
}],
latest_applied_version: None,
applied_count: 0,
unfinished_count: 0,
};
assert!(r.has_errors());
assert!(!r.has_warnings());
}
#[test]
fn verify_report_has_warnings_detects_warning_diagnostics() {
let r = VerifyReport {
diagnostics: vec![VerifyDiagnostic {
code: "D606".to_string(),
severity: VerifySeverity::Warning,
message: "x".to_string(),
location: None,
}],
latest_applied_version: None,
applied_count: 0,
unfinished_count: 0,
};
assert!(!r.has_errors());
assert!(r.has_warnings());
}
#[test]
fn verify_report_clean_run_reports_neither() {
let r = VerifyReport {
diagnostics: Vec::new(),
latest_applied_version: None,
applied_count: 0,
unfinished_count: 0,
};
assert!(!r.has_errors());
assert!(!r.has_warnings());
}
#[test]
fn advisory_emits_d692_when_enums_present() {
let mut snap = empty_snapshot();
snap.enums.insert(
"status".to_string(),
super::super::schema::EnumSchema {
name: "status".to_string(),
variants: vec!["active".to_string()],
},
);
let mut diagnostics = Vec::new();
diff_advisory_fields(&snap, &mut diagnostics);
assert!(diagnostics.iter().any(|d| d.code == "D692"));
}
#[test]
fn normalize_default_strips_trailing_text_cast() {
assert_eq!(
normalize_default_expr(Some("'foo'::text")),
Some("'foo'".to_string())
);
}
#[test]
fn normalize_default_strips_trailing_timestamptz_cast() {
assert_eq!(
normalize_default_expr(Some("now()::timestamp with time zone")),
Some("now()".to_string())
);
}
#[test]
fn normalize_default_preserves_unsuffixed_expr() {
assert_eq!(
normalize_default_expr(Some("now()")),
Some("now()".to_string())
);
assert_eq!(normalize_default_expr(Some("42")), Some("42".to_string()));
}
#[test]
fn normalize_default_treats_none_and_empty_as_no_default() {
assert_eq!(normalize_default_expr(None), None);
assert_eq!(normalize_default_expr(Some("")), None);
assert_eq!(normalize_default_expr(Some(" ")), None);
}
#[test]
fn normalize_default_preserves_double_colons_inside_string() {
assert_eq!(
normalize_default_expr(Some("'a::b'::text")),
Some("'a::b'".to_string())
);
}
#[test]
fn normalize_default_strips_nested_casts() {
assert_eq!(
normalize_default_expr(Some("'foo'::text::varchar")),
Some("'foo'".to_string())
);
assert_eq!(
normalize_default_expr(Some("123::int::bigint::numeric")),
Some("123".to_string())
);
assert_eq!(
normalize_default_expr(Some("'x'::text :: varchar")),
Some("'x'".to_string())
);
}
#[test]
fn diff_pk_match_emits_no_diagnostic() {
let mut diagnostics = Vec::new();
let snap = PrimaryKeySchema {
columns: vec!["id".to_string()],
kind: PkKindSchema::HeerId,
};
let live = PrimaryKeySchema {
columns: vec!["id".to_string()],
kind: PkKindSchema::HeerId,
};
diff_primary_key("users", &snap, &live, &mut diagnostics);
assert!(diagnostics.is_empty());
}
#[test]
fn diff_pk_snapshot_has_pk_live_does_not_emits_d608() {
let mut diagnostics = Vec::new();
let snap = PrimaryKeySchema {
columns: vec!["id".to_string()],
kind: PkKindSchema::HeerId,
};
let live = PrimaryKeySchema {
columns: Vec::new(),
kind: PkKindSchema::None,
};
diff_primary_key("users", &snap, &live, &mut diagnostics);
assert_eq!(diagnostics.len(), 1);
assert_eq!(diagnostics[0].code, "D608");
assert_eq!(diagnostics[0].severity, VerifySeverity::Error);
}
#[test]
fn diff_pk_live_has_pk_snapshot_does_not_emits_d608() {
let mut diagnostics = Vec::new();
let snap = PrimaryKeySchema {
columns: Vec::new(),
kind: PkKindSchema::None,
};
let live = PrimaryKeySchema {
columns: vec!["id".to_string()],
kind: PkKindSchema::HeerId,
};
diff_primary_key("users", &snap, &live, &mut diagnostics);
assert_eq!(diagnostics.len(), 1);
assert_eq!(diagnostics[0].code, "D608");
}
#[test]
fn diff_pk_column_list_mismatch_emits_d608() {
let mut diagnostics = Vec::new();
let snap = PrimaryKeySchema {
columns: vec!["a".to_string(), "b".to_string()],
kind: PkKindSchema::Composite,
};
let live = PrimaryKeySchema {
columns: vec!["b".to_string(), "a".to_string()],
kind: PkKindSchema::Composite,
};
diff_primary_key("t", &snap, &live, &mut diagnostics);
assert_eq!(diagnostics.len(), 1);
assert_eq!(diagnostics[0].code, "D608");
}
fn idx_with_columns(name: &str, table: &str, cols: &[&str]) -> IndexSchema {
IndexSchema {
extension_dependency: None,
include: Vec::new(),
index_type: IndexTypeSchema::BTree,
kind: IndexKindSchema::NonUnique,
name: name.to_string(),
nulls_not_distinct: false,
predicate: None,
requires_out_of_transaction: false,
table: table.to_string(),
target: IndexTargetSchema::Columns(
cols.iter()
.map(|c| super::super::schema::IndexColumnSchema {
name: c.to_string(),
nulls: super::super::schema::IndexNullsOrderSchema::Default,
opclass: None,
order: super::super::schema::IndexOrderSchema::Asc,
})
.collect(),
),
}
}
#[test]
fn diff_indexes_wrong_table_is_d615_error() {
let mut snap = empty_snapshot();
let mut live = empty_snapshot();
snap.indexes
.push(idx_with_columns("idx_x", "users", &["email"]));
live.indexes
.push(idx_with_columns("idx_x", "orders", &["email"]));
let mut diagnostics = Vec::new();
diff_indexes(&snap, &live, &mut diagnostics);
assert!(
diagnostics
.iter()
.any(|d| d.code == "D615" && d.severity == VerifySeverity::Error)
);
}
#[test]
fn diff_indexes_wrong_columns_is_d612_error() {
let mut snap = empty_snapshot();
let mut live = empty_snapshot();
snap.indexes
.push(idx_with_columns("idx_x", "users", &["email"]));
live.indexes
.push(idx_with_columns("idx_x", "users", &["name"]));
let mut diagnostics = Vec::new();
diff_indexes(&snap, &live, &mut diagnostics);
assert!(
diagnostics
.iter()
.any(|d| d.code == "D612" && d.severity == VerifySeverity::Error)
);
}
#[test]
fn diff_indexes_wrong_uniqueness_is_d613_error() {
let mut snap = empty_snapshot();
let mut live = empty_snapshot();
let mut s = idx_with_columns("idx_x", "users", &["email"]);
s.kind = IndexKindSchema::UniqueIndex;
snap.indexes.push(s);
live.indexes
.push(idx_with_columns("idx_x", "users", &["email"]));
let mut diagnostics = Vec::new();
diff_indexes(&snap, &live, &mut diagnostics);
assert!(
diagnostics
.iter()
.any(|d| d.code == "D613" && d.severity == VerifySeverity::Error)
);
}
#[test]
fn diff_indexes_method_drift_is_d614_warning() {
let mut snap = empty_snapshot();
let mut live = empty_snapshot();
let mut s = idx_with_columns("idx_x", "users", &["email"]);
s.index_type = IndexTypeSchema::Gin;
snap.indexes.push(s);
live.indexes
.push(idx_with_columns("idx_x", "users", &["email"]));
let mut diagnostics = Vec::new();
diff_indexes(&snap, &live, &mut diagnostics);
assert!(
diagnostics
.iter()
.any(|d| d.code == "D614" && d.severity == VerifySeverity::Warning)
);
}
#[test]
fn diff_indexes_clean_match_emits_no_shape_diagnostic() {
let mut snap = empty_snapshot();
let mut live = empty_snapshot();
snap.indexes
.push(idx_with_columns("idx_x", "users", &["email"]));
live.indexes
.push(idx_with_columns("idx_x", "users", &["email"]));
let mut diagnostics = Vec::new();
diff_indexes(&snap, &live, &mut diagnostics);
for code in ["D612", "D613", "D614", "D615"] {
assert!(
!diagnostics.iter().any(|d| d.code == code),
"unexpected {code} on clean match: {diagnostics:?}"
);
}
}
#[test]
fn diff_tables_default_drift_emits_d607() {
let mut snap = empty_snapshot();
let mut snap_col = col("created_at", "TIMESTAMPTZ", false);
snap_col.default_sql = Some("now()".to_string());
snap.models
.insert("users".to_string(), table("users", vec![snap_col]));
let mut live = empty_snapshot();
let live_col = col("created_at", "timestamptz", false);
live.models
.insert("users".to_string(), table("users", vec![live_col]));
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
assert!(
diagnostics
.iter()
.any(|d| d.code == "D607" && d.severity == VerifySeverity::Error),
"expected D607 on default drift; got {diagnostics:?}"
);
}
#[test]
fn diff_tables_default_canonicalisation_treats_text_cast_as_match() {
let mut snap = empty_snapshot();
let mut snap_col = col("status", "TEXT", false);
snap_col.default_sql = Some("'active'".to_string());
snap.models
.insert("users".to_string(), table("users", vec![snap_col]));
let mut live = empty_snapshot();
let mut live_col = col("status", "text", false);
live_col.default_sql = Some("'active'::text".to_string());
live.models
.insert("users".to_string(), table("users", vec![live_col]));
let mut diagnostics = Vec::new();
diff_tables(&snap, &live, &mut diagnostics);
assert!(
!diagnostics.iter().any(|d| d.code == "D607"),
"no D607 expected on canonicalised default; got {diagnostics:?}"
);
}
#[test]
fn heeranjid_allowlist_is_sorted() {
let mut sorted = HEERANJID_ARTIFACT_TABLES.to_vec();
sorted.sort();
assert_eq!(sorted.as_slice(), HEERANJID_ARTIFACT_TABLES);
}
#[test]
fn heeranjid_allowlist_recognises_known_substrate_tables() {
for name in HEERANJID_ARTIFACT_TABLES {
assert!(
is_heeranjid_artifact_table(name),
"{name} should be in allowlist"
);
}
}
#[test]
fn heeranjid_allowlist_does_not_match_adopter_heer_prefix_tables() {
assert!(!is_heeranjid_artifact_table("heer_user"));
assert!(!is_heeranjid_artifact_table("heer_orders"));
assert!(!is_heeranjid_artifact_table("heer"));
assert!(!is_heeranjid_artifact_table(""));
}
fn ledger_row_at(version: &str, app: &str, ooo: bool, status: LedgerStatus) -> LedgerRow {
LedgerRow {
version: version.to_string(),
description: String::new(),
checksum_up: "V1:0000000000000000000000000000000000000000000000000000000000000000"
.to_string(),
checksum_down: None,
execution_mode: super::super::ledger::ExecutionMode::Transactional,
status,
execution_time_ms: 0,
out_of_order_flag: ooo,
applied_steps_count: 0,
total_steps: None,
partial_apply_note: None,
run_id: 0,
snapshot_version: super::super::schema::SNAPSHOT_FORMAT_VERSION.to_string(),
app_label: app.to_string(),
leaf_identity: None,
}
}
#[test]
fn d622_warning_for_ooo_row_default_policy() {
let rows = vec![
ledger_row_at(
"V20260201000000__early",
"billing",
true,
LedgerStatus::Applied,
),
ledger_row_at(
"V20260301000000__later",
"billing",
false,
LedgerStatus::Applied,
),
];
let policy = crate::config::PolicyConfig::default();
let mut diagnostics = Vec::new();
emit_out_of_order_diagnostics(&rows, &policy, &mut diagnostics);
let d622: Vec<&VerifyDiagnostic> =
diagnostics.iter().filter(|d| d.code == "D622").collect();
assert_eq!(d622.len(), 1, "exactly one D622 expected");
assert_eq!(d622[0].severity, VerifySeverity::Warning);
assert!(
d622[0].message.contains("V20260201000000__early"),
"must name the offending row: {}",
d622[0].message
);
assert!(
d622[0].message.contains("V20260301000000__later"),
"must name the conflicting peer: {}",
d622[0].message
);
}
#[test]
fn d622_error_for_ooo_row_strict_policy() {
let rows = vec![
ledger_row_at(
"V20260201000000__early",
"billing",
true,
LedgerStatus::Applied,
),
ledger_row_at(
"V20260301000000__later",
"billing",
false,
LedgerStatus::Applied,
),
];
let policy = crate::config::PolicyConfig {
strict_out_of_order: true,
};
let mut diagnostics = Vec::new();
emit_out_of_order_diagnostics(&rows, &policy, &mut diagnostics);
let d622: Vec<&VerifyDiagnostic> =
diagnostics.iter().filter(|d| d.code == "D622").collect();
assert_eq!(d622.len(), 1);
assert_eq!(d622[0].severity, VerifySeverity::Error);
}
#[test]
fn d622_silent_when_no_rows_have_ooo_flag() {
let rows = vec![
ledger_row_at(
"V20260101000000__a",
"billing",
false,
LedgerStatus::Applied,
),
ledger_row_at(
"V20260201000000__b",
"billing",
false,
LedgerStatus::Applied,
),
];
let policy = crate::config::PolicyConfig::default();
let mut diagnostics = Vec::new();
emit_out_of_order_diagnostics(&rows, &policy, &mut diagnostics);
assert!(
!diagnostics.iter().any(|d| d.code == "D622"),
"no D622 expected when no rows are out-of-order"
);
}
#[test]
fn d622_one_per_ooo_row() {
let rows = vec![
ledger_row_at(
"V20260101000000__early1",
"billing",
true,
LedgerStatus::Applied,
),
ledger_row_at(
"V20260201000000__early2",
"billing",
true,
LedgerStatus::Applied,
),
ledger_row_at(
"V20260301000000__later",
"billing",
false,
LedgerStatus::Applied,
),
];
let policy = crate::config::PolicyConfig::default();
let mut diagnostics = Vec::new();
emit_out_of_order_diagnostics(&rows, &policy, &mut diagnostics);
let d622: Vec<&VerifyDiagnostic> =
diagnostics.iter().filter(|d| d.code == "D622").collect();
assert_eq!(d622.len(), 2, "one D622 per ooo row");
}
#[test]
fn d622_ignores_rolled_back_peer() {
let rows = vec![
ledger_row_at(
"V20260201000000__early",
"billing",
true,
LedgerStatus::Applied,
),
ledger_row_at(
"V20260301000000__rolled",
"billing",
false,
LedgerStatus::RolledBack,
),
];
let policy = crate::config::PolicyConfig::default();
let mut diagnostics = Vec::new();
emit_out_of_order_diagnostics(&rows, &policy, &mut diagnostics);
let d622: Vec<&VerifyDiagnostic> =
diagnostics.iter().filter(|d| d.code == "D622").collect();
assert_eq!(d622.len(), 1);
assert!(
d622[0].message.contains("no higher-version peer"),
"expected a `no higher-version peer` note: {}",
d622[0].message
);
}
#[test]
fn d622_scopes_peer_lookup_to_same_app_label() {
let rows = vec![
ledger_row_at(
"V20260201000000__early",
"billing",
true,
LedgerStatus::Applied,
),
ledger_row_at(
"V20260301000000__other_app_peer",
"users",
false,
LedgerStatus::Applied,
),
];
let policy = crate::config::PolicyConfig::default();
let mut diagnostics = Vec::new();
emit_out_of_order_diagnostics(&rows, &policy, &mut diagnostics);
let d622: Vec<&VerifyDiagnostic> =
diagnostics.iter().filter(|d| d.code == "D622").collect();
assert_eq!(d622.len(), 1);
assert!(
d622[0].message.contains("no higher-version peer"),
"no peer should match across app labels: {}",
d622[0].message
);
}
const D6XX_CODE_REGISTRY: &[(&str, &str)] = &[
("D601", "snapshot table missing in live"),
("D602", "live table not in snapshot"),
("D603", "snapshot column missing in live"),
("D604", "live column not in snapshot"),
("D605", "nullability drift"),
("D606", "type-string drift (advisory)"),
("D607", "default value drift"),
("D608", "primary key column list drift"),
("D609", "foreign-key shape drift"),
("D610", "snapshot index missing in live"),
("D611", "live index not in snapshot (advisory)"),
("D612", "index columns differ"),
("D613", "index uniqueness differs"),
("D614", "index method drift (advisory)"),
("D615", "index lives on wrong table"),
("D621", "ledger table not found"),
("D622", "out-of-order migration applied"),
("D623", "repair refused: leaf identity mismatch"),
("D624", "rollback refused: leaf identity mismatch"),
("D690", "FTS not yet checked (info)"),
("D691", "partition not yet checked (info)"),
("D692", "enums not yet checked (info)"),
("D693", "INCLUDE / partial-predicate not yet checked (info)"),
("D699", "ledger reports applied but DB has no tables"),
];
#[test]
fn d6xx_codes_have_unique_meanings() {
let mut seen: BTreeMap<&str, &str> = BTreeMap::new();
for (code, meaning) in D6XX_CODE_REGISTRY {
assert!(
seen.insert(code, meaning).is_none(),
"duplicate D6xx code {code}",
);
}
}
#[test]
fn d6xx_emit_sites_all_covered_by_registry() {
let source = include_str!("verify.rs");
let bytes = source.as_bytes();
let mut prefix_buf = Vec::with_capacity(8);
prefix_buf.extend_from_slice(b"code: ");
prefix_buf.push(b'"');
prefix_buf.push(b'D');
let prefix: &[u8] = &prefix_buf;
let mut scan = 0usize;
let mut emitted: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
while scan + prefix.len() + 3 <= bytes.len() {
if &bytes[scan..scan + prefix.len()] == prefix {
let code_start = scan + prefix.len() - 1; let code_end = code_start + 4;
if code_end < bytes.len() && bytes[code_end] == b'"' {
let body_ok = bytes[code_start + 1..code_end]
.iter()
.all(|b| b.is_ascii_alphanumeric());
if body_ok {
let s = String::from_utf8_lossy(&bytes[code_start..code_end]).into_owned();
emitted.insert(s);
}
}
scan += prefix.len();
continue;
}
scan += 1;
}
assert!(
!emitted.is_empty(),
"scanner found zero D6xx emit sites in verify.rs — \
the byte pattern probably drifted; investigate before \
trusting this test",
);
let registry: std::collections::BTreeSet<&str> =
D6XX_CODE_REGISTRY.iter().map(|(c, _)| *c).collect();
for code in &emitted {
assert!(
registry.contains(code.as_str()),
"emit site for {code} found in verify.rs but {code} is \
missing from D6XX_CODE_REGISTRY — add it (with a \
unique meaning) to keep the central registry honest",
);
}
for (code, _) in D6XX_CODE_REGISTRY {
if !emitted.contains(*code) {
eprintln!(
"note: D6xx code {code} is in the registry but has \
no emit site in verify.rs — either remove it from \
the registry or add the corresponding emit site"
);
}
}
}
}