use mysql_async::Conn;
use mysql_async::prelude::Queryable;
use thiserror::Error;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProtectionModel {
TransactionalDmlProtection,
NonTransactionalDdlSnapshot,
UnprotectedDenied,
}
impl ProtectionModel {
pub fn warnings(&self) -> &'static [&'static str] {
match self {
ProtectionModel::TransactionalDmlProtection => &[],
ProtectionModel::NonTransactionalDdlSnapshot => &[
"pre-operation snapshot only",
"implicit commit may occur",
"snapshot and DDL are not one atomic transaction",
"automatic rollback is not guaranteed",
],
ProtectionModel::UnprotectedDenied => &["no backup protection can be guaranteed"],
}
}
pub fn is_transactional(&self) -> bool {
matches!(self, ProtectionModel::TransactionalDmlProtection)
}
}
pub fn protection_model_for(
category: crate::policy::model::SqlCategory,
ast_type: &str,
) -> ProtectionModel {
use crate::policy::model::SqlCategory;
match category {
SqlCategory::Ddl => ProtectionModel::NonTransactionalDdlSnapshot,
SqlCategory::Write => {
if crate::backup::extractor::is_backup_required(ast_type) {
ProtectionModel::TransactionalDmlProtection
} else {
ProtectionModel::TransactionalDmlProtection
}
}
_ => ProtectionModel::TransactionalDmlProtection,
}
}
#[derive(Debug, Clone, Error, PartialEq)]
pub enum DdlPreflightError {
#[error("table {schema}.{table} does not exist: {hint}")]
NotFound {
schema: String,
table: String,
hint: &'static str,
},
#[error("table {schema}.{table} already exists: {hint}")]
Conflict {
schema: String,
table: String,
hint: &'static str,
},
#[error("preflight query failed: {0}")]
Query(String),
}
#[derive(Debug, Clone, PartialEq)]
pub enum DdlPreflight {
Present,
MissingNoOp(Vec<(String, String)>),
Mixed {
existing: Vec<(String, String)>,
missing: Vec<(String, String)>,
},
}
pub fn rewrite_drop_subset(object_type: &str, existing: &[(String, String)]) -> Option<String> {
let keyword = match object_type {
"table" => "DROP TABLE IF EXISTS",
"view" => "DROP VIEW IF EXISTS",
_ => return None,
};
if existing.is_empty() {
return None;
}
let targets = existing
.iter()
.map(|(schema, table)| {
format!(
"`{}`.`{}`",
schema.replace('`', "``"),
table.replace('`', "``")
)
})
.collect::<Vec<_>>()
.join(", ");
Some(format!("{keyword} {targets}"))
}
pub async fn preflight_ddl(
conn: &mut Conn,
classified: &crate::policy::classifier::ClassifiedStatement,
fallback_db: Option<&str>,
) -> Result<DdlPreflight, DdlPreflightError> {
use crate::policy::model::SqlCategory;
if classified.category != SqlCategory::Ddl || classified.mutated_tables.is_empty() {
return Ok(DdlPreflight::Present);
}
let resolve_schema = |target: &crate::policy::classifier::TableRef| -> Option<String> {
target
.database
.clone()
.or_else(|| fallback_db.map(str::to_string))
};
async fn table_exists(
conn: &mut Conn,
schema: &str,
table: &str,
) -> Result<bool, DdlPreflightError> {
let found: Option<i8> = conn
.exec_first(
"SELECT 1 FROM information_schema.tables
WHERE table_schema = ? AND table_name = ?",
(schema, table),
)
.await
.map_err(|e| DdlPreflightError::Query(e.to_string()))?;
Ok(found.is_some())
}
match classified.ast_type {
"create" => Ok(DdlPreflight::Present),
"rename" => {
let pairs: Vec<(
&crate::policy::classifier::TableRef,
&crate::policy::classifier::TableRef,
)> = classified
.mutated_tables
.chunks(2)
.map(|c| (&c[0], &c[1]))
.collect();
let mut known: Vec<((String, String), bool)> = Vec::new();
for (src, dst) in &pairs {
let src_schema = resolve_schema(src).ok_or(DdlPreflightError::NotFound {
schema: "?".into(),
table: src.table.clone(),
hint: "no database in scope for the RENAME source",
})?;
let dst_schema = resolve_schema(dst).ok_or(DdlPreflightError::NotFound {
schema: "?".into(),
table: dst.table.clone(),
hint: "no database in scope for the RENAME destination",
})?;
let src_id = (src_schema.clone(), src.table.clone());
let dst_id = (dst_schema.clone(), dst.table.clone());
let src_present =
if let Some(p) = known.iter().find(|(id, _p)| *id == src_id).map(|(_, p)| *p) {
p
} else {
table_exists(conn, &src_schema, &src.table).await?
};
if !src_present {
return Err(DdlPreflightError::NotFound {
schema: src_schema,
table: src.table.clone(),
hint: "RENAME source does not exist",
});
}
let dst_present =
if let Some(p) = known.iter().find(|(id, _p)| *id == dst_id).map(|(_, p)| *p) {
p
} else {
table_exists(conn, &dst_schema, &dst.table).await?
};
if dst_present {
return Err(DdlPreflightError::Conflict {
schema: dst_schema,
table: dst.table.clone(),
hint: "RENAME destination already exists",
});
}
known.retain(|(id, _)| *id != src_id);
known.push((src_id, false));
known.push((dst_id, true));
}
Ok(DdlPreflight::Present)
}
_ => {
let mut existing: Vec<(String, String)> = Vec::new();
let mut missing: Vec<(String, String)> = Vec::new();
for target in &classified.mutated_tables {
let schema = resolve_schema(target).ok_or(DdlPreflightError::NotFound {
schema: "?".into(),
table: target.table.clone(),
hint: "no database in scope for the DDL target",
})?;
if table_exists(conn, &schema, &target.table).await? {
existing.push((schema, target.table.clone()));
} else {
missing.push((schema, target.table.clone()));
}
}
if missing.is_empty() {
return Ok(DdlPreflight::Present);
}
if !classified.if_exists {
let (schema, table) = missing[0].clone();
return Err(DdlPreflightError::NotFound {
schema,
table,
hint: "DDL target does not exist (no IF EXISTS)",
});
}
if existing.is_empty() {
Ok(DdlPreflight::MissingNoOp(missing))
} else {
Ok(DdlPreflight::Mixed { existing, missing })
}
}
}
}