use tokio::runtime::Handle;
use tokio_postgres::Client;
use pgevolve_core::catalog::{
CatalogError, CatalogFilter, CatalogQuerier, CatalogQuery, DriftReport, Row, Value,
};
use pgevolve_core::lint::{LINT_AT_PLAN_RULES, Severity};
use pgevolve_core::plan::Plan;
use super::error::ApplyError;
use crate::target_identity::compute_target_identity;
#[allow(clippy::struct_excessive_bools)]
#[derive(Debug, Clone, Copy, Default)]
pub struct PreflightOverrides {
pub allow_different_target: bool,
pub allow_drift: bool,
pub allow_unwaived_lint: bool,
pub allow_unapproved_intents: bool,
}
pub fn check_env_vars_resolvable(plan: &Plan) -> Result<(), ApplyError> {
use crate::executor::env_interp;
for group in &plan.groups {
for step in &group.steps {
let refs = env_interp::references(&step.sql);
for var in refs {
if std::env::var(&var).is_err() {
return Err(ApplyError::MissingEnvVar(var, step.step_no));
}
}
}
}
Ok(())
}
pub async fn run_preflight(
client: &Client,
plan: &Plan,
filter: &CatalogFilter,
overrides: PreflightOverrides,
) -> Result<(), ApplyError> {
check_env_vars_resolvable(plan)?;
let live = compute_target_identity(client).await?;
if live != plan.metadata.target_identity && !overrides.allow_different_target {
return Err(ApplyError::TargetIdentityMismatch {
plan: plan.metadata.target_identity.clone(),
live,
});
}
if !overrides.allow_drift {
let (live_catalog, live_drift) = read_live_catalog(client, filter)?;
let drift =
pgevolve_core::diff::diff(&plan.metadata.target_snapshot, &live_catalog, &live_drift);
if !drift.is_empty() {
return Err(ApplyError::DriftDetected(drift.len()));
}
}
if !overrides.allow_unwaived_lint {
check_lint_waivers(plan)?;
}
if !overrides.allow_unapproved_intents {
check_intent_approval(plan)?;
}
Ok(())
}
fn check_lint_waivers(plan: &Plan) -> Result<(), ApplyError> {
let malformed: Vec<_> = plan
.lint_waivers
.iter()
.filter(|w| w.rule.is_empty() || w.target.is_empty())
.map(|w| (w.rule.clone(), w.target.clone()))
.collect();
if !malformed.is_empty() {
return Err(ApplyError::LintWaiverMissing {
count: malformed.len(),
details: malformed,
});
}
let unknown: Vec<_> = plan
.lint_waivers
.iter()
.filter(|w| !LINT_AT_PLAN_RULES.contains(&w.rule.as_str()))
.map(|w| (w.rule.clone(), w.target.clone()))
.collect();
if !unknown.is_empty() {
for (rule, target) in &unknown {
eprintln!(
"pgevolve apply: warning: lint_waiver rule `{rule}` for `{target}` is not a \
known LintAtPlan rule; the waiver has no effect"
);
}
}
let unmatched: Vec<_> = plan
.metadata
.lint_at_plan_findings
.iter()
.filter(|f| {
!plan
.lint_waivers
.iter()
.any(|w| w.rule == f.rule && f.message.contains(&w.target))
})
.map(|f| (f.rule.clone(), f.target.clone()))
.collect();
if !unmatched.is_empty() {
return Err(ApplyError::LintWaiverMissing {
count: unmatched.len(),
details: unmatched,
});
}
if !plan.lint_waivers.is_empty() {
let _ = Severity::LintAtPlan; eprintln!(
"pgevolve apply: {} lint waiver(s) active:",
plan.lint_waivers.len()
);
for w in &plan.lint_waivers {
eprintln!(" - [{}] {} — {}", w.rule, w.target, w.reason);
}
}
Ok(())
}
fn check_intent_approval(plan: &Plan) -> Result<(), ApplyError> {
let unapproved: Vec<_> = plan
.intents
.iter()
.filter(|i| !i.approved)
.map(|i| (i.id, i.target.clone(), i.reason.clone()))
.collect();
if !unapproved.is_empty() {
return Err(ApplyError::UnapprovedIntents {
count: unapproved.len(),
details: unapproved,
});
}
Ok(())
}
fn read_live_catalog(
client: &Client,
filter: &CatalogFilter,
) -> Result<(pgevolve_core::ir::catalog::Catalog, DriftReport), CatalogError> {
struct BorrowedQuerier<'a> {
client: &'a Client,
runtime: Handle,
version: std::cell::Cell<Option<pgevolve_core::catalog::PgVersion>>,
}
impl CatalogQuerier for BorrowedQuerier<'_> {
fn fetch(
&self,
query: CatalogQuery,
text_array_param: &[&str],
) -> Result<Vec<Row>, CatalogError> {
use pgevolve_core::catalog::queries::query_for;
let version = if matches!(query, CatalogQuery::PgVersion) {
pgevolve_core::catalog::PgVersion::Pg16
} else if let Some(v) = self.version.get() {
v
} else {
let v = pgevolve_core::catalog::PgVersion::detect(self)?;
self.version.set(Some(v));
v
};
let sql = query_for(version, query);
let client = self.client;
let owned: Vec<String> = text_array_param.iter().map(ToString::to_string).collect();
let pg_rows: Vec<tokio_postgres::Row> = tokio::task::block_in_place(|| {
self.runtime.block_on(async move {
if query.takes_text_array_param() {
client.query(sql, &[&owned]).await
} else {
client.query(sql, &[]).await
}
})
})
.map_err(|e| CatalogError::QueryFailed {
query,
message: e.to_string(),
})?;
pg_rows
.iter()
.map(|r| pg_row_to_catalog_row(r, query))
.collect()
}
}
let runtime = Handle::try_current().map_err(|e| CatalogError::QueryFailed {
query: CatalogQuery::PgVersion,
message: format!("not inside a Tokio runtime: {e}"),
})?;
let querier = BorrowedQuerier {
client,
runtime,
version: std::cell::Cell::new(None),
};
pgevolve_core::catalog::read_catalog(&querier, filter)
}
fn pg_row_to_catalog_row(
row: &tokio_postgres::Row,
query: CatalogQuery,
) -> Result<Row, CatalogError> {
use tokio_postgres::types::Type;
let bad = |col: &str, msg: String| CatalogError::BadColumnType {
query,
column: col.to_string(),
message: msg,
};
macro_rules! get_opt {
($row:expr, $idx:expr, $col:expr, $t:ty, $ty:expr) => {
$row.try_get::<_, Option<$t>>($idx).map_err(|e| {
bad(
$col,
format!("decode {} as {}: {e}", $ty.name(), stringify!($t)),
)
})?
};
}
let mut out = Row::new();
for (i, col) in row.columns().iter().enumerate() {
let name = col.name();
let ty = col.type_();
let value = match *ty {
Type::BOOL => get_opt!(row, i, name, bool, ty).map_or(Value::Null, Value::Bool),
Type::INT2 => get_opt!(row, i, name, i16, ty).map_or(Value::Null, Value::SmallInt),
Type::INT4 => get_opt!(row, i, name, i32, ty)
.map_or(Value::Null, |v| Value::Integer(i64::from(v))),
Type::INT8 => get_opt!(row, i, name, i64, ty).map_or(Value::Null, Value::Integer),
Type::OID => get_opt!(row, i, name, u32, ty)
.map_or(Value::Null, |v| Value::Integer(i64::from(v))),
Type::TEXT | Type::VARCHAR | Type::NAME | Type::BPCHAR => {
get_opt!(row, i, name, String, ty).map_or(Value::Null, Value::Text)
}
Type::CHAR => {
let v = get_opt!(row, i, name, i8, ty);
#[allow(clippy::cast_sign_loss)]
v.map_or(Value::Null, |b| Value::Char(b as u8 as char))
}
Type::INT2_ARRAY => get_opt!(row, i, name, Vec<i16>, ty).map_or(Value::Null, |v| {
Value::IntegerArray(v.into_iter().map(i64::from).collect())
}),
Type::INT4_ARRAY => get_opt!(row, i, name, Vec<i32>, ty).map_or(Value::Null, |v| {
Value::IntegerArray(v.into_iter().map(i64::from).collect())
}),
Type::INT8_ARRAY => {
get_opt!(row, i, name, Vec<i64>, ty).map_or(Value::Null, Value::IntegerArray)
}
Type::TEXT_ARRAY | Type::NAME_ARRAY | Type::VARCHAR_ARRAY => {
get_opt!(row, i, name, Vec<String>, ty).map_or(Value::Null, Value::TextArray)
}
Type::BYTEA => get_opt!(row, i, name, Vec<u8>, ty).map_or(Value::Null, Value::Bytes),
_ => row
.try_get::<_, Option<String>>(i)
.map(|v| v.map_or(Value::Null, Value::Text))
.map_err(|e| bad(name, format!("unsupported type {} ({e})", ty.name())))?,
};
out.insert(name, value);
}
Ok(out)
}
#[cfg(test)]
mod tests {
use pgevolve_core::ir::catalog::Catalog;
use pgevolve_core::plan::{
DestructiveIntent, LintWaiver, Plan, PlanId, PlanMetadata, RecordedFinding,
};
use time::OffsetDateTime;
fn empty_plan_with_waivers(waivers: Vec<LintWaiver>) -> Plan {
empty_plan_with_waivers_and_findings(waivers, vec![])
}
fn empty_plan_with_waivers_and_findings(
waivers: Vec<LintWaiver>,
findings: Vec<RecordedFinding>,
) -> Plan {
let catalog = Catalog::empty();
Plan {
id: PlanId([0u8; 32]),
groups: vec![],
intents: vec![],
lint_waivers: waivers,
step_overrides: vec![],
metadata: PlanMetadata {
pgevolve_version: "0.0.0-test".into(),
planner_ruleset_version: 1,
source_rev: None,
target_identity: "test".into(),
target_snapshot: catalog,
created_at: OffsetDateTime::now_utc(),
lint_at_plan_findings: findings,
},
advisory_findings: vec![],
}
}
fn plan_with_intents(intents: Vec<DestructiveIntent>) -> Plan {
let catalog = Catalog::empty();
Plan {
id: PlanId([0u8; 32]),
groups: vec![],
intents,
lint_waivers: vec![],
step_overrides: vec![],
metadata: PlanMetadata {
pgevolve_version: "0.0.0-test".into(),
planner_ruleset_version: 1,
source_rev: None,
target_identity: "test".into(),
target_snapshot: catalog,
created_at: OffsetDateTime::now_utc(),
lint_at_plan_findings: vec![],
},
advisory_findings: vec![],
}
}
#[test]
fn empty_waiver_rule_fails_preflight() {
let plan = empty_plan_with_waivers(vec![LintWaiver {
rule: String::new(), target: "app.users".into(),
reason: "test".into(),
}]);
let result = super::check_lint_waivers(&plan);
assert!(
result.is_err(),
"expected Err for empty-rule waiver, got Ok"
);
}
#[test]
fn empty_waiver_target_fails_preflight() {
let plan = empty_plan_with_waivers(vec![LintWaiver {
rule: "column-position-drift".into(),
target: String::new(), reason: "test".into(),
}]);
let result = super::check_lint_waivers(&plan);
assert!(
result.is_err(),
"expected Err for empty-target waiver, got Ok"
);
}
#[test]
fn well_formed_waiver_passes_preflight() {
let plan = empty_plan_with_waivers(vec![LintWaiver {
rule: "column-position-drift".into(),
target: "app.users".into(),
reason: "acknowledged".into(),
}]);
assert!(
super::check_lint_waivers(&plan).is_ok(),
"expected Ok for well-formed waiver"
);
}
#[test]
fn no_waivers_passes_preflight() {
let plan = empty_plan_with_waivers(vec![]);
assert!(
super::check_lint_waivers(&plan).is_ok(),
"expected Ok for empty waiver list"
);
}
#[test]
fn recorded_finding_with_matching_waiver_passes() {
let plan = empty_plan_with_waivers_and_findings(
vec![LintWaiver {
rule: "column-position-drift".into(),
target: "app.users".into(),
reason: "acknowledged".into(),
}],
vec![RecordedFinding {
rule: "column-position-drift".into(),
target: "app.users".into(),
message: "app.users: column position drift. source order [id, created_at, email] \
vs catalog order [id, email, created_at]"
.into(),
}],
);
assert!(
super::check_lint_waivers(&plan).is_ok(),
"expected Ok when recorded finding has a matching waiver"
);
}
#[test]
fn recorded_finding_with_no_waiver_fails() {
let plan = empty_plan_with_waivers_and_findings(
vec![],
vec![RecordedFinding {
rule: "column-position-drift".into(),
target: "app.users".into(),
message: "app.users: column position drift.".into(),
}],
);
let result = super::check_lint_waivers(&plan);
assert!(
result.is_err(),
"expected Err when recorded finding has no waiver"
);
}
#[test]
fn removed_waiver_fails_preflight() {
let plan = empty_plan_with_waivers_and_findings(
vec![LintWaiver {
rule: "column-position-drift".into(),
target: "app.orders".into(), reason: "for a different table".into(),
}],
vec![RecordedFinding {
rule: "column-position-drift".into(),
target: "app.users".into(),
message: "app.users: column position drift.".into(),
}],
);
let result = super::check_lint_waivers(&plan);
assert!(
result.is_err(),
"expected Err when the correct waiver was removed (replaced with a non-matching one)"
);
}
#[test]
fn unapproved_intent_fails_preflight() {
let plan = plan_with_intents(vec![DestructiveIntent {
id: 1,
step: 1,
kind: "drop_column".into(),
target: "app.users.legacy_email".into(),
reason: "removing deprecated column".into(),
approved: false,
}]);
let result = super::check_intent_approval(&plan);
assert!(
result.is_err(),
"expected Err for unapproved destructive intent"
);
}
#[test]
fn approved_intent_passes_preflight() {
let plan = plan_with_intents(vec![DestructiveIntent {
id: 1,
step: 1,
kind: "drop_column".into(),
target: "app.users.legacy_email".into(),
reason: "removing deprecated column".into(),
approved: true,
}]);
assert!(
super::check_intent_approval(&plan).is_ok(),
"expected Ok when intent is approved"
);
}
#[test]
fn no_intents_passes_preflight() {
let plan = plan_with_intents(vec![]);
assert!(
super::check_intent_approval(&plan).is_ok(),
"expected Ok when there are no intents"
);
}
#[test]
fn allow_unapproved_intents_override_bypasses_check() {
let plan = plan_with_intents(vec![DestructiveIntent {
id: 1,
step: 1,
kind: "drop_table".into(),
target: "app.legacy".into(),
reason: "old table".into(),
approved: false,
}]);
assert!(
super::check_intent_approval(&plan).is_err(),
"check_intent_approval must fail for an unapproved intent (override tested at run_preflight level)"
);
}
}