From 27c69845d075fec8882a1b3909360bebcaab9c0a Mon Sep 17 00:00:00 2001 From: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co> Date: Fri, 26 Jun 2026 14:24:40 -0400 Subject: [PATCH] test(db): lint tenant-scoped migration constraints Co-authored-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co> Signed-off-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co> --- crates/buzz-db/src/migration.rs | 543 ++++++++++++++++++++++++++++++++ 1 file changed, 543 insertions(+) diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index f4f3c9ab3..8bef9a717 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -212,6 +212,549 @@ mod tests { .replace(REMINDER_INDEX, "") } + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum ConstraintKind { + ForeignKey, + PrimaryKey, + Unique, + } + + #[derive(Debug, Clone, PartialEq, Eq)] + struct ConstraintLint { + table: String, + kind: ConstraintKind, + description: String, + columns: Vec, + } + + fn all_migration_sql() -> String { + MIGRATOR + .iter() + .map(|migration| migration.sql.as_str()) + .collect::>() + .join("\n") + } + + fn strip_sql_comments(sql: &str) -> String { + sql.lines() + .map(|line| line.split_once("--").map_or(line, |(before, _)| before)) + .collect::>() + .join("\n") + } + + fn normalize_sql(sql: &str) -> String { + strip_sql_comments(sql) + .split_whitespace() + .collect::>() + .join(" ") + .to_ascii_lowercase() + } + + fn split_sql_statements(sql: &str) -> Vec { + strip_sql_comments(sql) + .split(';') + .map(str::trim) + .filter(|statement| !statement.is_empty()) + .map(ToOwned::to_owned) + .collect() + } + + fn find_matching_paren(sql: &str, open: usize) -> Option { + let mut depth = 0usize; + for (offset, byte) in sql.as_bytes()[open..].iter().enumerate() { + match byte { + b'(' => depth += 1, + b')' => { + depth = depth.checked_sub(1)?; + if depth == 0 { + return Some(open + offset); + } + } + _ => {} + } + } + None + } + + fn split_top_level_csv(input: &str) -> Vec { + let mut parts = Vec::new(); + let mut start = 0usize; + let mut depth = 0usize; + for (idx, byte) in input.bytes().enumerate() { + match byte { + b'(' => depth += 1, + b')' => depth = depth.saturating_sub(1), + b',' if depth == 0 => { + parts.push(input[start..idx].trim().to_owned()); + start = idx + 1; + } + _ => {} + } + } + let tail = input[start..].trim(); + if !tail.is_empty() { + parts.push(tail.to_owned()); + } + parts + } + + fn identifier_after_keyword(statement: &str, keyword: &str) -> Option { + let lower = statement.to_ascii_lowercase(); + let keyword_pos = lower.find(keyword)?; + let mut remainder = statement[keyword_pos + keyword.len()..].trim_start(); + for prefix in ["if not exists", "if exists", "only"] { + if remainder.to_ascii_lowercase().starts_with(prefix) { + remainder = remainder[prefix.len()..].trim_start(); + } + } + + let identifier = remainder + .split(|ch: char| ch.is_whitespace() || ch == '(') + .next()? + .trim_matches('"') + .rsplit('.') + .next()? + .trim_matches('"') + .to_ascii_lowercase(); + (!identifier.is_empty()).then_some(identifier) + } + + fn first_parenthesized_columns(input: &str) -> Vec { + let Some(open) = input.find('(') else { + return Vec::new(); + }; + let Some(close) = find_matching_paren(input, open) else { + return Vec::new(); + }; + + split_top_level_csv(&input[open + 1..close]) + .into_iter() + .filter_map(|column| { + let name = column + .trim() + .trim_matches('"') + .split_whitespace() + .next()? + .trim_matches('"') + .to_ascii_lowercase(); + (!name.is_empty()).then_some(name) + }) + .collect() + } + + fn column_definition_name(definition: &str) -> Option { + let trimmed = definition.trim(); + let lower = trimmed.to_ascii_lowercase(); + if lower.starts_with("constraint ") + || lower.starts_with("primary key") + || lower.starts_with("foreign key") + || lower.starts_with("unique") + || lower.starts_with("check ") + || lower.starts_with("exclude ") + { + return None; + } + + let name = trimmed + .split_whitespace() + .next()? + .trim_matches('"') + .to_ascii_lowercase(); + (!name.is_empty()).then_some(name) + } + + fn create_table_body(statement: &str) -> Option<(String, Vec)> { + let table = identifier_after_keyword(statement, "create table")?; + let open = statement.find('(')?; + let close = find_matching_paren(statement, open)?; + Some((table, split_top_level_csv(&statement[open + 1..close]))) + } + + fn create_table_definitions(sql: &str) -> Vec<(String, Vec)> { + split_sql_statements(sql) + .into_iter() + .filter_map(|statement| { + statement + .trim_start() + .to_ascii_lowercase() + .starts_with("create table") + .then(|| create_table_body(&statement))? + }) + .collect() + } + + fn scoped_tables(sql: &str) -> Vec { + let mut tables = create_table_definitions(sql) + .into_iter() + .filter_map(|(table, definitions)| { + definitions + .iter() + .any(|definition| { + column_definition_name(definition).as_deref() == Some("community_id") + }) + .then_some(table) + }) + .collect::>(); + + for statement in split_sql_statements(sql) { + let normalized = normalize_sql(&statement); + if !normalized.starts_with("alter table") + || !(normalized.contains("add column community_id") + || normalized.contains("add column if not exists community_id")) + { + continue; + } + if let Some(table) = identifier_after_keyword(&statement, "alter table") { + tables.push(table); + } + } + + tables.sort(); + tables.dedup(); + tables + } + + fn constraint_lint_for_definition(table: &str, definition: &str) -> Option { + let normalized = normalize_sql(definition); + let definition_without_name = if normalized.starts_with("constraint ") { + let after_constraint = definition + .trim_start() + .splitn(3, char::is_whitespace) + .nth(2) + .unwrap_or(""); + normalize_sql(after_constraint) + } else { + normalized.clone() + }; + + if definition_without_name.starts_with("primary key") { + Some(ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::PrimaryKey, + description: definition.to_owned(), + columns: first_parenthesized_columns(&definition_without_name), + }) + } else if definition_without_name.starts_with("unique") { + Some(ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::Unique, + description: definition.to_owned(), + columns: first_parenthesized_columns(&definition_without_name), + }) + } else if definition_without_name.starts_with("foreign key") { + Some(ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::ForeignKey, + description: definition.to_owned(), + columns: first_parenthesized_columns(&definition_without_name), + }) + } else if normalized.contains(" primary key") { + column_definition_name(definition).map(|column| ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::PrimaryKey, + description: definition.to_owned(), + columns: vec![column], + }) + } else if normalized.contains(" references ") { + column_definition_name(definition).map(|column| ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::ForeignKey, + description: definition.to_owned(), + columns: vec![column], + }) + } else if normalized.contains(" unique") { + column_definition_name(definition).map(|column| ConstraintLint { + table: table.to_owned(), + kind: ConstraintKind::Unique, + description: definition.to_owned(), + columns: vec![column], + }) + } else { + None + } + } + + fn table_constraints(sql: &str, scoped_tables: &[String]) -> Vec { + create_table_definitions(sql) + .into_iter() + .filter(|(table, _)| scoped_tables.contains(table)) + .flat_map(|(table, definitions)| { + definitions.into_iter().filter_map(move |definition| { + constraint_lint_for_definition(&table, &definition) + }) + }) + .collect() + } + + fn alter_table_constraints(sql: &str, scoped_tables: &[String]) -> Vec { + split_sql_statements(sql) + .into_iter() + .filter_map(|statement| { + let normalized = normalize_sql(&statement); + if !normalized.starts_with("alter table") { + return None; + } + + let table = identifier_after_keyword(&statement, "alter table")?; + if !scoped_tables.contains(&table) { + return None; + } + + let add_pos = normalized.find(" add ")?; + let definition = normalized[add_pos + " add ".len()..].trim(); + constraint_lint_for_definition(&table, definition) + }) + .collect() + } + + fn unique_indexes(sql: &str, scoped_tables: &[String]) -> Vec { + split_sql_statements(sql) + .into_iter() + .filter_map(|statement| { + let normalized = normalize_sql(&statement); + if !normalized.starts_with("create unique index") { + return None; + } + + let lower_statement = statement.to_ascii_lowercase(); + let on_pos = lower_statement.find(" on ")?; + let table = statement[on_pos + " on ".len()..] + .trim_start() + .split(|ch: char| ch.is_whitespace() || ch == '(') + .next()? + .trim_matches('"') + .rsplit('.') + .next()? + .trim_matches('"') + .to_ascii_lowercase(); + + if !scoped_tables.contains(&table) { + return None; + } + + let columns = first_parenthesized_columns(&statement[on_pos + " on ".len()..]); + Some(ConstraintLint { + table, + kind: ConstraintKind::Unique, + description: statement, + columns, + }) + }) + .collect() + } + + fn scoped_constraint_lints(sql: &str, scoped_tables: &[String]) -> Vec { + let mut constraints = table_constraints(sql, scoped_tables); + constraints.extend(alter_table_constraints(sql, scoped_tables)); + constraints.extend(unique_indexes(sql, scoped_tables)); + constraints + } + + fn channels_has_community_id(sql: &str) -> bool { + scoped_tables(sql).iter().any(|table| table == "channels") + } + + fn has_channels_community_id_immutability_guard(sql: &str) -> bool { + let normalized = normalize_sql(sql); + normalized.contains("create trigger") + && normalized.contains("before update") + && normalized.contains(" on channels") + && normalized.contains("community_id") + && normalized.contains("old.community_id") + && normalized.contains("new.community_id") + && normalized.contains("raise exception") + } + + fn forbidden_channels_community_id_mutations(sql: &str) -> Vec { + split_sql_statements(sql) + .into_iter() + .filter(|statement| { + let normalized = normalize_sql(statement); + let updates_channels = + identifier_after_keyword(statement, "update").as_deref() == Some("channels"); + let mutates_with_update = updates_channels + && normalized.contains(" set ") + && normalized.contains("community_id"); + let alters_channels = identifier_after_keyword(statement, "alter table").as_deref() + == Some("channels"); + let drops_channels = identifier_after_keyword(statement, "drop table").as_deref() + == Some("channels"); + let drops_or_rewrites_column = alters_channels + && (normalized.contains("drop column community_id") + || normalized.contains("alter column community_id") + || normalized.contains("rename column community_id") + || normalized.contains("rename community_id") + || normalized.contains("drop trigger") + || normalized.contains("disable trigger")); + let drops_table = drops_channels; + + mutates_with_update || drops_or_rewrites_column || drops_table + }) + .collect() + } + + fn scoped_constraint_violations(sql: &str) -> Vec { + let scoped_tables = scoped_tables(sql); + scoped_constraint_lints(sql, &scoped_tables) + .into_iter() + .filter(|constraint| { + !constraint + .columns + .iter() + .any(|column| column == "community_id") + }) + .collect() + } + + #[test] + fn migration_lint_detects_scoped_key_constraints_missing_community_id() { + let sql = r#" + CREATE TABLE widgets ( + community_id UUID NOT NULL, + id UUID PRIMARY KEY, + channel_id UUID REFERENCES channels(id), + slug TEXT, + CONSTRAINT widgets_name_unique UNIQUE (slug), + CONSTRAINT widgets_parent_fk FOREIGN KEY (channel_id) REFERENCES channels(id) + ); + CREATE UNIQUE INDEX idx_widgets_slug ON widgets (slug); + ALTER TABLE widgets ADD CONSTRAINT widgets_alter_slug_unique UNIQUE (slug); + ALTER TABLE widgets ADD CONSTRAINT widgets_alter_parent_fk FOREIGN KEY (channel_id) REFERENCES channels(id); + "#; + + let violations = scoped_constraint_violations(sql); + + assert!(violations + .iter() + .any(|violation| violation.kind == ConstraintKind::PrimaryKey)); + assert_eq!( + violations + .iter() + .filter(|violation| violation.kind == ConstraintKind::ForeignKey) + .count(), + 3 + ); + assert_eq!( + violations + .iter() + .filter(|violation| violation.kind == ConstraintKind::Unique) + .count(), + 3 + ); + } + + #[test] + fn migration_lint_accepts_scoped_key_constraints_with_community_id() { + let sql = r#" + CREATE TABLE widgets ( + community_id UUID NOT NULL, + id UUID NOT NULL, + channel_id UUID NOT NULL, + slug TEXT NOT NULL, + PRIMARY KEY (community_id, id), + UNIQUE (community_id, slug), + FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id) + ); + CREATE UNIQUE INDEX idx_widgets_slug ON widgets (community_id, slug); + ALTER TABLE widgets ADD CONSTRAINT widgets_alter_slug_unique UNIQUE (community_id, slug); + ALTER TABLE widgets ADD CONSTRAINT widgets_alter_parent_fk FOREIGN KEY (community_id, channel_id) REFERENCES channels(community_id, id); + "#; + + assert!(scoped_constraint_violations(sql).is_empty()); + } + + #[test] + fn migration_lint_requires_channels_community_id_update_guard() { + let guarded = r#" + CREATE TABLE channels ( + community_id UUID NOT NULL, + id UUID NOT NULL, + PRIMARY KEY (community_id, id) + ); + CREATE FUNCTION prevent_channels_community_id_update() RETURNS trigger AS $$ + BEGIN + IF OLD.community_id <> NEW.community_id THEN + RAISE EXCEPTION 'channels.community_id is immutable'; + END IF; + RETURN NEW; + END; + $$ LANGUAGE plpgsql; + CREATE TRIGGER channels_community_id_immutable + BEFORE UPDATE ON channels + FOR EACH ROW EXECUTE FUNCTION prevent_channels_community_id_update(); + "#; + let unguarded = r#" + CREATE TABLE channels ( + community_id UUID NOT NULL, + id UUID NOT NULL, + PRIMARY KEY (community_id, id) + ); + "#; + let forbidden = r#" + UPDATE channels SET community_id = gen_random_uuid(); + ALTER TABLE channels DROP COLUMN community_id; + DROP TABLE IF EXISTS channels; + "#; + + assert!(channels_has_community_id(guarded)); + assert!(has_channels_community_id_immutability_guard(guarded)); + assert!(channels_has_community_id(unguarded)); + assert!(!has_channels_community_id_immutability_guard(unguarded)); + assert_eq!( + forbidden_channels_community_id_mutations(forbidden).len(), + 3 + ); + } + + #[test] + fn scoped_table_primary_key_unique_and_foreign_key_constraints_include_community_id() { + // docs/multi-tenant-conformance.md §Migration gates and + // docs/multi-tenant-relay.md C2.1 require every tenant-scoped key that + // can otherwise become a cross-community existence oracle to carry + // `community_id`. The current main schema is still single-tenant, so + // scope is detected structurally: a table becomes tenant-scoped in this + // lint as soon as a migration gives it a `community_id` column. + let sql = all_migration_sql(); + let violations = scoped_constraint_violations(&sql) + .into_iter() + .map(|constraint| { + format!( + "{}. {:?} constraint missing community_id: {}", + constraint.table, constraint.kind, constraint.description + ) + }) + .collect::>(); + + assert!( + violations.is_empty(), + "tenant-scoped tables are detected by a community_id column; every primary key, unique/FK constraint, or unique index on those tables must include community_id:\n{}", + violations.join("\n") + ); + } + + #[test] + fn channels_community_id_is_immutable_after_insert() { + // docs/multi-tenant-relay.md P-RESOLVE/S2 makes channel tenancy a + // load-bearing constant after resolution. The lint is intentionally + // green before `channels.community_id` exists, then requires a trigger + // guard and rejects re-tenanting migration statements once it does. + let sql = all_migration_sql(); + let forbidden_mutations = forbidden_channels_community_id_mutations(&sql); + + assert!( + forbidden_mutations.is_empty(), + "channels.community_id must not be re-tenanted after insert; forbidden migration statements:\n{}", + forbidden_mutations.join("\n---\n") + ); + + if channels_has_community_id(&sql) { + assert!( + has_channels_community_id_immutability_guard(&sql), + "migrations define channels.community_id but no BEFORE UPDATE trigger/function guard that rejects OLD.community_id <> NEW.community_id was found" + ); + } + } + #[tokio::test] #[ignore = "requires Postgres"] async fn run_migrations_applies_embedded_versions_on_fresh_database() {