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::AlterTableStmt(stmt) => {
101 for cmd_node in &stmt.cmds {
102 if let Some(pg_query::protobuf::node::Node::AlterTableCmd(cmd)) = &cmd_node.node
103 {
104 inspect_alter_table_cmd(cmd, ddl, &mut issues, &mut overall_risk);
105 }
106 }
107 }
108 NodeRef::AlterTableCmd(cmd) => {
109 inspect_alter_table_cmd(cmd, ddl, &mut issues, &mut overall_risk);
110 }
111 _ => {}
112 }
113 }
114
115 let issues_json: Vec<Value> = issues
116 .into_iter()
117 .map(|i| {
118 json!({
119 "risk_level": i.risk_level.as_str(),
120 "lock_type": i.lock_type,
121 "issue": i.issue,
122 "recommendation": i.recommendation,
123 "safe_alternative_sql": i.safe_alternative_sql,
124 })
125 })
126 .collect();
127
128 json!({
129 "overall_risk": overall_risk.as_str(),
130 "statement_count": statement_count,
131 "issue_count": issues_json.len(),
132 "issues": issues_json,
133 "is_safe": overall_risk == RiskLevel::Safe,
134 })
135}
136
137fn inspect_alter_table_cmd(
138 cmd: &pg_query::protobuf::AlterTableCmd,
139 ddl: &str,
140 issues: &mut Vec<DdlSafetyIssue>,
141 overall_risk: &mut RiskLevel,
142) {
143 use pg_query::protobuf::AlterTableType;
144 if let Ok(subtype) = AlterTableType::try_from(cmd.subtype) {
145 match subtype {
146 AlterTableType::AtDropColumn => {
147 let issue = DdlSafetyIssue {
148 risk_level: RiskLevel::High,
149 lock_type: "AccessExclusiveLock",
150 issue: "Dropping a column acquires an AccessExclusiveLock, blocking all reads and writes.".into(),
151 recommendation: "Ensure application code has stopped referencing the column before dropping.".into(),
152 safe_alternative_sql: None,
153 };
154 update_overall_risk(overall_risk, RiskLevel::High);
155 issues.push(issue);
156 }
157 AlterTableType::AtAlterColumnType => {
158 let issue = DdlSafetyIssue {
159 risk_level: RiskLevel::Critical,
160 lock_type: "AccessExclusiveLock (Full Table Rewrite)",
161 issue: "Altering a column type forces a full table rewrite while holding an AccessExclusiveLock.".into(),
162 recommendation: "Add a new column, backfill data asynchronously, dual-write in app logic, then drop the old column.".into(),
163 safe_alternative_sql: None,
164 };
165 update_overall_risk(overall_risk, RiskLevel::Critical);
166 issues.push(issue);
167 }
168 AlterTableType::AtAddConstraint => {
169 if let Some(def_node) = &cmd.def
170 && let Some(pg_query::protobuf::node::Node::Constraint(c)) = &def_node.node
171 {
172 use pg_query::protobuf::ConstrType;
173 if let Ok(ConstrType::ConstrForeign) = ConstrType::try_from(c.contype)
174 && !c.skip_validation
175 {
176 let safe_sql = format!("{} NOT VALID;", ddl.trim_end_matches(';'));
177 let issue = DdlSafetyIssue {
178 risk_level: RiskLevel::High,
179 lock_type: "AccessExclusiveLock",
180 issue: "Adding a foreign key constraint scans the entire table under AccessExclusiveLock.".into(),
181 recommendation: "Add the constraint with NOT VALID first, then run ALTER TABLE ... VALIDATE CONSTRAINT separately.".into(),
182 safe_alternative_sql: Some(safe_sql),
183 };
184 update_overall_risk(overall_risk, RiskLevel::High);
185 issues.push(issue);
186 }
187 }
188 }
189 _ => {}
190 }
191 }
192}
193
194fn update_overall_risk(current: &mut RiskLevel, new_risk: RiskLevel) {
195 match (*current, new_risk) {
196 (RiskLevel::Critical, _) => {}
197 (_, RiskLevel::Critical) => *current = RiskLevel::Critical,
198 (RiskLevel::High, _) => {}
199 (_, RiskLevel::High) => *current = RiskLevel::High,
200 (RiskLevel::Medium, _) => {}
201 (_, RiskLevel::Medium) => *current = RiskLevel::Medium,
202 (RiskLevel::Safe, RiskLevel::Safe) => {}
203 }
204}
205
206#[cfg(test)]
207mod tests {
208 use super::*;
209
210 #[test]
211 fn test_non_concurrent_index_flagged() {
212 let sql = "CREATE INDEX idx_users_email ON users(email);";
213 let res = analyze_ddl_safety(sql);
214 assert_eq!(res["overall_risk"], "CRITICAL");
215 assert_eq!(res["issue_count"], 1);
216 assert!(
217 res["issues"][0]["safe_alternative_sql"]
218 .as_str()
219 .unwrap()
220 .contains("CONCURRENTLY")
221 );
222 }
223
224 #[test]
225 fn test_concurrent_index_is_safe() {
226 let sql = "CREATE INDEX CONCURRENTLY idx_users_email ON users(email);";
227 let res = analyze_ddl_safety(sql);
228 assert_eq!(res["overall_risk"], "SAFE");
229 assert_eq!(res["issue_count"], 0);
230 }
231
232 #[test]
233 fn test_keyword_in_string_literal_is_safe() {
234 let sql = "SELECT 'CREATE INDEX idx_test ON test(col);' AS query;";
235 let res = analyze_ddl_safety(sql);
236 assert_eq!(res["overall_risk"], "SAFE");
237 assert_eq!(res["issue_count"], 0);
238 }
239
240 #[test]
241 fn test_drop_table_flagged() {
242 let sql = "DROP TABLE users;";
243 let res = analyze_ddl_safety(sql);
244 assert_eq!(res["overall_risk"], "CRITICAL");
245 assert_eq!(res["issue_count"], 1);
246 }
247
248 #[test]
249 fn test_alter_drop_column_flagged() {
250 let sql = "ALTER TABLE users DROP COLUMN email;";
251 let res = analyze_ddl_safety(sql);
252 assert_eq!(res["overall_risk"], "HIGH");
253 assert_eq!(res["issue_count"], 1);
254 }
255
256 #[test]
257 fn test_alter_column_type_flagged() {
258 let sql = "ALTER TABLE users ALTER COLUMN email TYPE text;";
259 let res = analyze_ddl_safety(sql);
260 assert_eq!(res["overall_risk"], "CRITICAL");
261 assert_eq!(res["issue_count"], 1);
262 }
263}