use crate::error::Result;
use serde::{Deserialize, Serialize};
use sqlx::{PgPool, Row};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ForeignKeyInfo {
pub constraint_name: String,
pub table_name: String,
pub column_name: String,
pub referenced_table: String,
pub referenced_column: String,
pub has_index: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OrphanedRecordInfo {
pub table_name: String,
pub column_name: String,
pub referenced_table: String,
pub orphaned_count: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CheckConstraintInfo {
pub constraint_name: String,
pub table_name: String,
pub check_definition: String,
pub is_validated: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ValidationSeverity {
Info,
Warning,
Error,
Critical,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidationIssue {
pub severity: ValidationSeverity,
pub category: String,
pub description: String,
pub affected_object: String,
pub recommendation: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaValidationReport {
pub issues: Vec<ValidationIssue>,
pub foreign_keys: Vec<ForeignKeyInfo>,
pub orphaned_records: Vec<OrphanedRecordInfo>,
pub check_constraints: Vec<CheckConstraintInfo>,
pub critical_count: usize,
pub error_count: usize,
pub warning_count: usize,
pub info_count: usize,
pub validated_at: chrono::DateTime<chrono::Utc>,
}
pub async fn get_foreign_keys(pool: &PgPool) -> Result<Vec<ForeignKeyInfo>> {
let rows = sqlx::query(
r#"
SELECT
tc.constraint_name,
tc.table_name,
kcu.column_name,
ccu.table_name AS referenced_table,
ccu.column_name AS referenced_column,
EXISTS(
SELECT 1 FROM pg_indexes
WHERE tablename = tc.table_name
AND indexdef LIKE '%' || kcu.column_name || '%'
) AS has_index
FROM information_schema.table_constraints AS tc
JOIN information_schema.key_column_usage AS kcu
ON tc.constraint_name = kcu.constraint_name
AND tc.table_schema = kcu.table_schema
JOIN information_schema.constraint_column_usage AS ccu
ON ccu.constraint_name = tc.constraint_name
AND ccu.table_schema = tc.table_schema
WHERE tc.constraint_type = 'FOREIGN KEY'
AND tc.table_schema = 'public'
ORDER BY tc.table_name, kcu.column_name
"#,
)
.fetch_all(pool)
.await?;
let mut foreign_keys = Vec::new();
for row in rows {
foreign_keys.push(ForeignKeyInfo {
constraint_name: row.try_get("constraint_name")?,
table_name: row.try_get("table_name")?,
column_name: row.try_get("column_name")?,
referenced_table: row.try_get("referenced_table").unwrap_or_default(),
referenced_column: row.try_get("referenced_column").unwrap_or_default(),
has_index: row.try_get("has_index").unwrap_or(false),
});
}
Ok(foreign_keys)
}
pub async fn detect_orphaned_records(
pool: &PgPool,
table_name: &str,
column_name: &str,
referenced_table: &str,
referenced_column: &str,
) -> Result<i64> {
let query = format!(
r#"
SELECT COUNT(*) as count
FROM "{}" AS t
WHERE t."{}" IS NOT NULL
AND NOT EXISTS (
SELECT 1 FROM "{}" AS r
WHERE r."{}" = t."{}"
)
"#,
table_name, column_name, referenced_table, referenced_column, column_name
);
let row: (i64,) = sqlx::query_as(&query).fetch_one(pool).await?;
Ok(row.0)
}
pub async fn get_all_orphaned_records(pool: &PgPool) -> Result<Vec<OrphanedRecordInfo>> {
let foreign_keys = get_foreign_keys(pool).await?;
let mut orphaned_records = Vec::new();
for fk in foreign_keys {
let count = detect_orphaned_records(
pool,
&fk.table_name,
&fk.column_name,
&fk.referenced_table,
&fk.referenced_column,
)
.await?;
if count > 0 {
orphaned_records.push(OrphanedRecordInfo {
table_name: fk.table_name,
column_name: fk.column_name,
referenced_table: fk.referenced_table,
orphaned_count: count,
});
}
}
Ok(orphaned_records)
}
pub async fn get_check_constraints(pool: &PgPool) -> Result<Vec<CheckConstraintInfo>> {
let rows = sqlx::query(
r#"
SELECT
tc.constraint_name,
tc.table_name,
pg_get_constraintdef(pgc.oid) AS check_definition,
pgc.convalidated AS is_validated
FROM information_schema.table_constraints AS tc
JOIN pg_constraint AS pgc
ON pgc.conname = tc.constraint_name
WHERE tc.constraint_type = 'CHECK'
AND tc.table_schema = 'public'
ORDER BY tc.table_name, tc.constraint_name
"#,
)
.fetch_all(pool)
.await?;
let mut constraints = Vec::new();
for row in rows {
constraints.push(CheckConstraintInfo {
constraint_name: row.try_get("constraint_name")?,
table_name: row.try_get("table_name")?,
check_definition: row.try_get("check_definition").unwrap_or_default(),
is_validated: row.try_get("is_validated").unwrap_or(false),
});
}
Ok(constraints)
}
pub async fn validate_schema(pool: &PgPool) -> Result<SchemaValidationReport> {
let mut issues = Vec::new();
let foreign_keys = get_foreign_keys(pool).await?;
for fk in &foreign_keys {
if !fk.has_index {
issues.push(ValidationIssue {
severity: ValidationSeverity::Warning,
category: "Missing Index".to_string(),
description: format!(
"Foreign key column '{}' in table '{}' lacks an index",
fk.column_name, fk.table_name
),
affected_object: format!("{}.{}", fk.table_name, fk.column_name),
recommendation: format!(
"CREATE INDEX idx_{}_{} ON \"{}\" (\"{}\");",
fk.table_name, fk.column_name, fk.table_name, fk.column_name
),
});
}
}
let orphaned_records = get_all_orphaned_records(pool).await?;
for orphaned in &orphaned_records {
let severity = if orphaned.orphaned_count > 100 {
ValidationSeverity::Critical
} else if orphaned.orphaned_count > 10 {
ValidationSeverity::Error
} else {
ValidationSeverity::Warning
};
issues.push(ValidationIssue {
severity,
category: "Orphaned Records".to_string(),
description: format!(
"Table '{}' has {} orphaned records in column '{}'",
orphaned.table_name, orphaned.orphaned_count, orphaned.column_name
),
affected_object: format!("{}.{}", orphaned.table_name, orphaned.column_name),
recommendation: format!(
"Clean up orphaned records or fix references to '{}'",
orphaned.referenced_table
),
});
}
let check_constraints = get_check_constraints(pool).await?;
for constraint in &check_constraints {
if !constraint.is_validated {
issues.push(ValidationIssue {
severity: ValidationSeverity::Warning,
category: "Unvalidated Constraint".to_string(),
description: format!(
"Check constraint '{}' on table '{}' is not validated",
constraint.constraint_name, constraint.table_name
),
affected_object: format!(
"{}.{}",
constraint.table_name, constraint.constraint_name
),
recommendation: format!(
"ALTER TABLE \"{}\" VALIDATE CONSTRAINT \"{}\";",
constraint.table_name, constraint.constraint_name
),
});
}
}
let critical_count = issues
.iter()
.filter(|i| i.severity == ValidationSeverity::Critical)
.count();
let error_count = issues
.iter()
.filter(|i| i.severity == ValidationSeverity::Error)
.count();
let warning_count = issues
.iter()
.filter(|i| i.severity == ValidationSeverity::Warning)
.count();
let info_count = issues
.iter()
.filter(|i| i.severity == ValidationSeverity::Info)
.count();
Ok(SchemaValidationReport {
issues,
foreign_keys,
orphaned_records,
check_constraints,
critical_count,
error_count,
warning_count,
info_count,
validated_at: chrono::Utc::now(),
})
}
pub async fn get_tables_without_primary_keys(pool: &PgPool) -> Result<Vec<String>> {
let rows = sqlx::query(
r#"
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_type = 'BASE TABLE'
AND table_name NOT IN (
SELECT table_name
FROM information_schema.table_constraints
WHERE constraint_type = 'PRIMARY KEY'
AND table_schema = 'public'
)
ORDER BY table_name
"#,
)
.fetch_all(pool)
.await?;
let mut tables = Vec::new();
for row in rows {
tables.push(row.try_get("table_name")?);
}
Ok(tables)
}
pub async fn get_nullable_foreign_key_columns(pool: &PgPool) -> Result<Vec<String>> {
let rows = sqlx::query(
r#"
SELECT DISTINCT
kcu.table_name || '.' || kcu.column_name AS column_path
FROM information_schema.key_column_usage AS kcu
JOIN information_schema.table_constraints AS tc
ON kcu.constraint_name = tc.constraint_name
JOIN information_schema.columns AS c
ON c.table_name = kcu.table_name
AND c.column_name = kcu.column_name
WHERE tc.constraint_type = 'FOREIGN KEY'
AND c.is_nullable = 'YES'
AND kcu.table_schema = 'public'
ORDER BY column_path
"#,
)
.fetch_all(pool)
.await?;
let mut columns = Vec::new();
for row in rows {
columns.push(row.try_get("column_path")?);
}
Ok(columns)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_foreign_key_info_creation() {
let fk = ForeignKeyInfo {
constraint_name: "fk_user_id".to_string(),
table_name: "orders".to_string(),
column_name: "user_id".to_string(),
referenced_table: "users".to_string(),
referenced_column: "id".to_string(),
has_index: true,
};
assert_eq!(fk.table_name, "orders");
assert_eq!(fk.referenced_table, "users");
assert!(fk.has_index);
}
#[test]
fn test_orphaned_record_info_creation() {
let orphaned = OrphanedRecordInfo {
table_name: "orders".to_string(),
column_name: "user_id".to_string(),
referenced_table: "users".to_string(),
orphaned_count: 42,
};
assert_eq!(orphaned.orphaned_count, 42);
}
#[test]
fn test_validation_issue_severity() {
let issue = ValidationIssue {
severity: ValidationSeverity::Critical,
category: "Data Integrity".to_string(),
description: "Test issue".to_string(),
affected_object: "test_table".to_string(),
recommendation: "Fix it".to_string(),
};
assert_eq!(issue.severity, ValidationSeverity::Critical);
}
#[test]
fn test_validation_severity_ordering() {
assert!(ValidationSeverity::Critical != ValidationSeverity::Error);
assert!(ValidationSeverity::Warning != ValidationSeverity::Info);
}
#[test]
fn test_check_constraint_info() {
let constraint = CheckConstraintInfo {
constraint_name: "check_positive".to_string(),
table_name: "balances".to_string(),
check_definition: "CHECK (amount >= 0)".to_string(),
is_validated: true,
};
assert_eq!(constraint.table_name, "balances");
assert!(constraint.is_validated);
}
#[test]
fn test_schema_validation_report_counts() {
let report = SchemaValidationReport {
issues: vec![
ValidationIssue {
severity: ValidationSeverity::Critical,
category: "Test".to_string(),
description: "Critical issue".to_string(),
affected_object: "obj1".to_string(),
recommendation: "Fix".to_string(),
},
ValidationIssue {
severity: ValidationSeverity::Warning,
category: "Test".to_string(),
description: "Warning issue".to_string(),
affected_object: "obj2".to_string(),
recommendation: "Fix".to_string(),
},
],
foreign_keys: vec![],
orphaned_records: vec![],
check_constraints: vec![],
critical_count: 1,
error_count: 0,
warning_count: 1,
info_count: 0,
validated_at: chrono::Utc::now(),
};
assert_eq!(report.critical_count, 1);
assert_eq!(report.warning_count, 1);
assert_eq!(report.issues.len(), 2);
}
#[test]
fn test_validation_issue_serialization() {
let issue = ValidationIssue {
severity: ValidationSeverity::Error,
category: "Integrity".to_string(),
description: "Test".to_string(),
affected_object: "table.column".to_string(),
recommendation: "Fix immediately".to_string(),
};
let json = serde_json::to_string(&issue).unwrap();
let deserialized: ValidationIssue = serde_json::from_str(&json).unwrap();
assert_eq!(issue.severity, deserialized.severity);
assert_eq!(issue.category, deserialized.category);
}
#[test]
fn test_foreign_key_serialization() {
let fk = ForeignKeyInfo {
constraint_name: "fk_test".to_string(),
table_name: "test".to_string(),
column_name: "id".to_string(),
referenced_table: "ref".to_string(),
referenced_column: "id".to_string(),
has_index: false,
};
let json = serde_json::to_string(&fk).unwrap();
let deserialized: ForeignKeyInfo = serde_json::from_str(&json).unwrap();
assert_eq!(fk.table_name, deserialized.table_name);
assert_eq!(fk.has_index, deserialized.has_index);
}
#[test]
fn test_orphaned_record_serialization() {
let orphaned = OrphanedRecordInfo {
table_name: "orders".to_string(),
column_name: "user_id".to_string(),
referenced_table: "users".to_string(),
orphaned_count: 10,
};
let json = serde_json::to_string(&orphaned).unwrap();
let deserialized: OrphanedRecordInfo = serde_json::from_str(&json).unwrap();
assert_eq!(orphaned.orphaned_count, deserialized.orphaned_count);
}
#[test]
fn test_schema_validation_report_serialization() {
let report = SchemaValidationReport {
issues: vec![],
foreign_keys: vec![],
orphaned_records: vec![],
check_constraints: vec![],
critical_count: 0,
error_count: 0,
warning_count: 0,
info_count: 0,
validated_at: chrono::Utc::now(),
};
let json = serde_json::to_string(&report).unwrap();
let deserialized: SchemaValidationReport = serde_json::from_str(&json).unwrap();
assert_eq!(report.critical_count, deserialized.critical_count);
}
}