1use serde_json::{Value, json};
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub enum RiskLevel {
10 Critical,
11 High,
12 Medium,
13 Safe,
14}
15
16impl RiskLevel {
17 pub fn as_str(self) -> &'static str {
18 match self {
19 Self::Critical => "CRITICAL",
20 Self::High => "HIGH",
21 Self::Medium => "MEDIUM",
22 Self::Safe => "SAFE",
23 }
24 }
25}
26
27#[derive(Debug, Clone)]
28pub struct DdlSafetyIssue {
29 pub risk_level: RiskLevel,
30 pub lock_type: &'static str,
31 pub issue: String,
32 pub recommendation: String,
33 pub safe_alternative_sql: Option<String>,
34}
35
36use pg_query::{NodeRef, parse};
37
38pub fn analyze_ddl_safety(ddl: &str) -> Value {
39 let parse_result = match parse(ddl) {
40 Ok(res) => res,
41 Err(e) => {
42 return json!({
43 "overall_risk": "CRITICAL",
44 "statement_count": 0,
45 "issue_count": 1,
46 "issues": [{
47 "risk_level": "CRITICAL",
48 "lock_type": "Parse Error",
49 "issue": format!("Failed to parse SQL statement(s): {e}"),
50 "recommendation": "Check SQL syntax.",
51 "safe_alternative_sql": Value::Null,
52 }],
53 "is_safe": false,
54 });
55 }
56 };
57
58 let mut issues = Vec::new();
59 let mut overall_risk = RiskLevel::Safe;
60 let statement_count = parse_result.protobuf.stmts.len();
61
62 for (node_ref, _depth, _context, _loc) in parse_result.protobuf.nodes() {
63 match node_ref {
64 NodeRef::IndexStmt(idx) => {
65 if !idx.concurrent {
66 let safe_sql = ddl.replacen("CREATE INDEX", "CREATE INDEX CONCURRENTLY", 1);
67 let issue = DdlSafetyIssue {
68 risk_level: RiskLevel::Critical,
69 lock_type: "ShareLock / AccessExclusiveLock",
70 issue: "Building index without CONCURRENTLY blocks concurrent write operations on the table.".into(),
71 recommendation: "Use CREATE INDEX CONCURRENTLY to build indexes without blocking writes.".into(),
72 safe_alternative_sql: Some(safe_sql),
73 };
74 update_overall_risk(&mut overall_risk, RiskLevel::Critical);
75 issues.push(issue);
76 }
77 }
78 NodeRef::DropStmt(_drop) => {
79 let issue = DdlSafetyIssue {
80 risk_level: RiskLevel::Critical,
81 lock_type: "AccessExclusiveLock (Irreversible Data Loss)",
82 issue: "Dropping objects permanently removes data/schema and acquires AccessExclusiveLock.".into(),
83 recommendation: "Verify backup and confirm object is no longer in active use.".into(),
84 safe_alternative_sql: None,
85 };
86 update_overall_risk(&mut overall_risk, RiskLevel::Critical);
87 issues.push(issue);
88 }
89 NodeRef::TruncateStmt(_) => {
90 let issue = DdlSafetyIssue {
91 risk_level: RiskLevel::Critical,
92 lock_type: "AccessExclusiveLock (Irreversible Data Loss)",
93 issue: "Truncating tables permanently removes data and acquires AccessExclusiveLock.".into(),
94 recommendation: "Verify backup and confirm table is no longer in active use.".into(),
95 safe_alternative_sql: None,
96 };
97 update_overall_risk(&mut overall_risk, RiskLevel::Critical);
98 issues.push(issue);
99 }
100 NodeRef::AlterTableCmd(cmd) => {
101 use pg_query::protobuf::AlterTableType;
102 if let Ok(subtype) = AlterTableType::try_from(cmd.subtype) {
103 match subtype {
104 AlterTableType::AtDropColumn => {
105 let issue = DdlSafetyIssue {
106 risk_level: RiskLevel::High,
107 lock_type: "AccessExclusiveLock",
108 issue: "Dropping a column acquires an AccessExclusiveLock, blocking all reads and writes.".into(),
109 recommendation: "Ensure application code has stopped referencing the column before dropping.".into(),
110 safe_alternative_sql: None,
111 };
112 update_overall_risk(&mut overall_risk, RiskLevel::High);
113 issues.push(issue);
114 }
115 AlterTableType::AtAlterColumnType => {
116 let issue = DdlSafetyIssue {
117 risk_level: RiskLevel::Critical,
118 lock_type: "AccessExclusiveLock (Full Table Rewrite)",
119 issue: "Altering a column type forces a full table rewrite while holding an AccessExclusiveLock.".into(),
120 recommendation: "Add a new column, backfill data asynchronously, dual-write in app logic, then drop the old column.".into(),
121 safe_alternative_sql: None,
122 };
123 update_overall_risk(&mut overall_risk, RiskLevel::Critical);
124 issues.push(issue);
125 }
126 AlterTableType::AtAddConstraint => {
127 if let Some(def_node) = &cmd.def {
128 if let Some(pg_query::protobuf::node::Node::Constraint(c)) =
129 &def_node.node
130 {
131 use pg_query::protobuf::ConstrType;
132 if let Ok(ConstrType::ConstrForeign) =
133 ConstrType::try_from(c.contype)
134 {
135 if !c.skip_validation {
136 let safe_sql =
137 format!("{} NOT VALID;", ddl.trim_end_matches(';'));
138 let issue = DdlSafetyIssue {
139 risk_level: RiskLevel::High,
140 lock_type: "AccessExclusiveLock",
141 issue: "Adding a foreign key constraint scans the entire table under AccessExclusiveLock.".into(),
142 recommendation: "Add the constraint with NOT VALID first, then run ALTER TABLE ... VALIDATE CONSTRAINT separately.".into(),
143 safe_alternative_sql: Some(safe_sql),
144 };
145 update_overall_risk(&mut overall_risk, RiskLevel::High);
146 issues.push(issue);
147 }
148 }
149 }
150 }
151 }
152 _ => {}
153 }
154 }
155 }
156 _ => {}
157 }
158 }
159
160 let issues_json: Vec<Value> = issues
161 .into_iter()
162 .map(|i| {
163 json!({
164 "risk_level": i.risk_level.as_str(),
165 "lock_type": i.lock_type,
166 "issue": i.issue,
167 "recommendation": i.recommendation,
168 "safe_alternative_sql": i.safe_alternative_sql,
169 })
170 })
171 .collect();
172
173 json!({
174 "overall_risk": overall_risk.as_str(),
175 "statement_count": statement_count,
176 "issue_count": issues_json.len(),
177 "issues": issues_json,
178 "is_safe": overall_risk == RiskLevel::Safe,
179 })
180}
181
182fn update_overall_risk(current: &mut RiskLevel, new_risk: RiskLevel) {
183 match (*current, new_risk) {
184 (RiskLevel::Critical, _) => {}
185 (_, RiskLevel::Critical) => *current = RiskLevel::Critical,
186 (RiskLevel::High, _) => {}
187 (_, RiskLevel::High) => *current = RiskLevel::High,
188 (RiskLevel::Medium, _) => {}
189 (_, RiskLevel::Medium) => *current = RiskLevel::Medium,
190 (RiskLevel::Safe, RiskLevel::Safe) => {}
191 }
192}
193
194#[cfg(test)]
195mod tests {
196 use super::*;
197
198 #[test]
199 fn test_non_concurrent_index_flagged() {
200 let sql = "CREATE INDEX idx_users_email ON users(email);";
201 let res = analyze_ddl_safety(sql);
202 assert_eq!(res["overall_risk"], "CRITICAL");
203 assert_eq!(res["issue_count"], 1);
204 assert!(
205 res["issues"][0]["safe_alternative_sql"]
206 .as_str()
207 .unwrap()
208 .contains("CONCURRENTLY")
209 );
210 }
211
212 #[test]
213 fn test_concurrent_index_is_safe() {
214 let sql = "CREATE INDEX CONCURRENTLY idx_users_email ON users(email);";
215 let res = analyze_ddl_safety(sql);
216 assert_eq!(res["overall_risk"], "SAFE");
217 assert_eq!(res["issue_count"], 0);
218 }
219
220 #[test]
221 fn test_keyword_in_string_literal_is_safe() {
222 let sql = "SELECT 'CREATE INDEX idx_test ON test(col);' AS query;";
223 let res = analyze_ddl_safety(sql);
224 assert_eq!(res["overall_risk"], "SAFE");
225 assert_eq!(res["issue_count"], 0);
226 }
227
228 #[test]
229 fn test_drop_table_flagged() {
230 let sql = "DROP TABLE users;";
231 let res = analyze_ddl_safety(sql);
232 assert_eq!(res["overall_risk"], "CRITICAL");
233 assert_eq!(res["issue_count"], 1);
234 }
235}