From aa5cd2c7315a5d1812392438485c3c5b1baea475 Mon Sep 17 00:00:00 2001 From: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Date: Mon, 3 Aug 2026 12:58:17 -0400 Subject: [PATCH] test(db): ratchet PostgreSQL SQLite schema parity Signed-off-by: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> --- .github/workflows/ci.yml | 8 ++ crates/buzz-db/src/migration.rs | 169 +++++++++++++++++++++++++++++++- 2 files changed, 176 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5e2e3fcbf2..997b4125cc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -708,6 +708,14 @@ jobs: --run-ignored ignored-only env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: PostgreSQL/SQLite schema parity + run: | + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E 'package(buzz-db) and test(/migration::tests::postgres_and_sqlite_schema_inventories_match_allowlist/)' \ + --run-ignored ignored-only + env: + DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: EventQuery backend parity run: | cargo nextest run \ diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 37f54d0fa2..d0fec40d69 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -98,7 +98,8 @@ async fn reject_legacy_nip_rs_cardinality_ambiguity(pool: &PgPool) -> Result<()> #[cfg(test)] mod tests { use super::*; - use std::collections::BTreeSet; + use sqlx::Row as _; + use std::collections::{BTreeMap, BTreeSet}; const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 @@ -1091,6 +1092,172 @@ mod tests { ); } + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] + enum SchemaDifference { + PostgresOnlyTable(&'static str), + SqliteOnlyTable(&'static str), + PostgresOnlyColumn(&'static str, &'static str), + SqliteOnlyColumn(&'static str, &'static str), + } + + // This is the sole inventory of intentional catalog divergence. Each entry needs a reason; + // differences not listed here and entries that no longer exist both fail the ratchet. + const ALLOWED_SCHEMA_DIFFERENCES: &[(SchemaDifference, &str)] = &[ + (SchemaDifference::PostgresOnlyTable("_operator_global_tables"), "Postgres migration-lint metadata has no local runtime role."), + (SchemaDifference::PostgresOnlyTable("audit_log"), "Central relay audit retention is outside the single-node profile."), + (SchemaDifference::PostgresOnlyTable("delivery_log"), "Central relay delivery accounting is outside the single-node profile."), + (SchemaDifference::PostgresOnlyTable("event_mentions"), "SQLite derives mentions from event JSON instead of a join table."), + (SchemaDifference::PostgresOnlyTable("parameterized_event_watermarks"), "Server-side parameterized-event retention is outside local mode."), + (SchemaDifference::PostgresOnlyTable("push_gateway_challenges"), "Push gateway authority is a hosted-service concern."), + (SchemaDifference::PostgresOnlyTable("push_gateway_delegations"), "Push gateway authority is a hosted-service concern."), + (SchemaDifference::PostgresOnlyTable("push_gateway_delivery_auth_replays"), "Push gateway replay defense is a hosted-service concern."), + (SchemaDifference::PostgresOnlyTable("push_gateway_delivery_request_replays"), "Push gateway replay defense is a hosted-service concern."), + (SchemaDifference::PostgresOnlyTable("push_gateway_endpoint_quotas"), "Push gateway quotas are a hosted-service concern."), + (SchemaDifference::PostgresOnlyTable("push_gateway_installations"), "Push gateway installations are a hosted-service concern."), + (SchemaDifference::PostgresOnlyTable("push_leases"), "Distributed push-worker leasing is outside local mode."), + (SchemaDifference::PostgresOnlyTable("push_match_queue"), "Distributed push matching is outside local mode."), + (SchemaDifference::PostgresOnlyTable("push_wake_outbox"), "Distributed push delivery is outside local mode."), + (SchemaDifference::PostgresOnlyTable("rate_limit_violations"), "Hosted relay abuse telemetry is outside local mode."), + (SchemaDifference::PostgresOnlyTable("replica_heartbeat"), "Local mode has no read replica to fence."), + (SchemaDifference::PostgresOnlyTable("subscriptions"), "Local subscriptions are connection-local rather than persisted."), + (SchemaDifference::SqliteOnlyTable("moderation_restrictions"), "Local mode uses a compact restriction table instead of the hosted moderation pipeline."), + (SchemaDifference::PostgresOnlyColumn("channel_members", "community_id"), "SQLite scopes membership through channels and omits the redundant tenant key."), + (SchemaDifference::PostgresOnlyColumn("channel_members", "removed_by"), "Removal actor provenance is not persisted locally."), + (SchemaDifference::PostgresOnlyColumn("communities", "signing_key"), "Desktop signing material is managed outside SQLite."), + (SchemaDifference::PostgresOnlyColumn("events", "d_tag"), "SQLite currently derives d tags from event JSON."), + (SchemaDifference::PostgresOnlyColumn("events", "deleted_at"), "SQLite physically deletes events instead of retaining server tombstones."), + (SchemaDifference::PostgresOnlyColumn("events", "search_tsv"), "SQLite uses FTS5 instead of a generated tsvector."), + (SchemaDifference::PostgresOnlyColumn("events", "tags"), "SQLite stores equivalent serialized tags in tags_json."), + (SchemaDifference::SqliteOnlyColumn("events", "event_json"), "SQLite retains serialized events for local reads."), + (SchemaDifference::SqliteOnlyColumn("events", "tags_json"), "SQLite stores tags as JSON text because it has no JSONB."), + (SchemaDifference::PostgresOnlyColumn("reactions", "created_at"), "SQLite uses the source event timestamp and does not duplicate creation time."), + (SchemaDifference::PostgresOnlyColumn("relay_invites", "created_at"), "Local invite lifecycle omits creation-time audit metadata."), + (SchemaDifference::PostgresOnlyColumn("relay_invites", "role"), "Local invites grant the profile's fixed member role."), + (SchemaDifference::PostgresOnlyColumn("users", "agent_type"), "SQLite uses the compact is_agent flag."), + (SchemaDifference::PostgresOnlyColumn("users", "capabilities"), "Hosted agent capability metadata is not used locally."), + (SchemaDifference::PostgresOnlyColumn("users", "deactivated_at"), "Account deactivation is unsupported locally."), + (SchemaDifference::PostgresOnlyColumn("users", "metadata_event_id"), "Metadata-event provenance is not persisted locally."), + (SchemaDifference::PostgresOnlyColumn("users", "okta_user_id"), "Enterprise identity linkage is hosted-only."), + (SchemaDifference::PostgresOnlyColumn("users", "updated_at"), "Local user profiles omit server audit timestamps."), + (SchemaDifference::SqliteOnlyColumn("users", "is_agent"), "SQLite uses a compact agent flag instead of agent_type."), + ]; + + async fn postgres_schema(pool: &PgPool) -> BTreeMap> { + let rows = sqlx::query("SELECT c.relname table_name, a.attname column_name FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid=c.relnamespace JOIN pg_catalog.pg_attribute a ON a.attrelid=c.oid WHERE n.nspname='public' AND c.relkind IN ('r','p') AND c.relname <> '_sqlx_migrations' AND a.attnum>0 AND NOT a.attisdropped AND NOT EXISTS (SELECT 1 FROM pg_catalog.pg_inherits i WHERE i.inhrelid=c.oid) ORDER BY c.relname,a.attnum") + .fetch_all(pool).await.expect("introspect Postgres catalog"); + let mut schema = BTreeMap::>::new(); + for row in rows { + schema + .entry(row.get("table_name")) + .or_default() + .insert(row.get("column_name")); + } + schema + } + + async fn sqlite_schema() -> BTreeMap> { + let pool = crate::sqlite::connect("sqlite::memory:") + .await + .expect("migrate in-memory SQLite"); + let tables = sqlx::query_scalar::<_, String>("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' AND name <> 'schema_version' AND name NOT GLOB 'events_fts*' ORDER BY name") + .fetch_all(&pool).await.expect("introspect SQLite tables"); + let mut schema = BTreeMap::new(); + for table in tables { + let columns = sqlx::query(sqlx::AssertSqlSafe(format!("PRAGMA table_info({table})"))) + .fetch_all(&pool) + .await + .expect("introspect SQLite columns") + .into_iter() + .map(|row| row.get::("name")) + .collect(); + schema.insert(table, columns); + } + schema + } + + fn schema_differences( + postgres: &BTreeMap>, + sqlite: &BTreeMap>, + ) -> BTreeSet { + let mut differences = BTreeSet::new(); + for table in postgres.keys() { + let table: &'static str = Box::leak(table.clone().into_boxed_str()); + if !sqlite.contains_key(table) { + differences.insert(SchemaDifference::PostgresOnlyTable(table)); + } + } + for table in sqlite.keys() { + let table: &'static str = Box::leak(table.clone().into_boxed_str()); + if !postgres.contains_key(table) { + differences.insert(SchemaDifference::SqliteOnlyTable(table)); + } + } + for table in postgres.keys().filter(|table| sqlite.contains_key(*table)) { + for column in postgres[table].difference(&sqlite[table]) { + differences.insert(SchemaDifference::PostgresOnlyColumn( + Box::leak(table.clone().into_boxed_str()), + Box::leak(column.clone().into_boxed_str()), + )); + } + for column in sqlite[table].difference(&postgres[table]) { + differences.insert(SchemaDifference::SqliteOnlyColumn( + Box::leak(table.clone().into_boxed_str()), + Box::leak(column.clone().into_boxed_str()), + )); + } + } + differences + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn postgres_and_sqlite_schema_inventories_match_allowlist() { + let admin = connect_test_pool().await; + let database_name = format!("schema_parity_{}", uuid::Uuid::new_v4().simple()); + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE DATABASE {database_name}" + ))) + .execute(&admin) + .await + .expect("create schema parity scratch database"); + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_owned()); + let slash = database_url.rfind('/').expect("database URL has path"); + let pool = PgPool::connect(&format!("{}/{database_name}", &database_url[..slash])) + .await + .expect("connect schema parity scratch database"); + run_migrations(&pool).await.expect("migrate Postgres"); + let actual = schema_differences(&postgres_schema(&pool).await, &sqlite_schema().await); + let allowed = ALLOWED_SCHEMA_DIFFERENCES + .iter() + .map(|(difference, reason)| { + assert!( + !reason.trim().is_empty(), + "schema allowlist reasons are required" + ); + difference.clone() + }) + .collect::>(); + let undeclared = actual.difference(&allowed).collect::>(); + let stale = allowed.difference(&actual).collect::>(); + let result = if undeclared.is_empty() && stale.is_empty() { + Ok(()) + } else { + Err(format!("Postgres/SQLite schema divergence must exactly match ALLOWED_SCHEMA_DIFFERENCES.\nUndeclared: {undeclared:#?}\nStale allowlist entries: {stale:#?}")) + }; + pool.close().await; + sqlx::query(sqlx::AssertSqlSafe(format!( + "DROP DATABASE IF EXISTS {database_name} WITH (FORCE)" + ))) + .execute(&admin) + .await + .expect("drop schema parity scratch database"); + if let Err(message) = result { + panic!("{message}"); + } + } + async fn connect_test_pool() -> PgPool { let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL"))