diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 13e7f6df82..f138cb7827 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -137,12 +137,18 @@ enum Command { /// /// Without `--channel`, only channels missing discovery metadata are /// reconciled. With `--channel`, only that channel's member snapshot is - /// replaced; canonical metadata and admin events remain untouched. + /// replaced; canonical metadata and admin events remain untouched unless + /// `--metadata` is given. ReconcileChannels { /// Optional channel UUID to force-republish. #[arg(long)] channel: Option, + /// With `--channel`, also republish its kind:39000 metadata from the + /// channel row, for example after an operator rename. + #[arg(long, requires = "channel")] + metadata: bool, + /// Relay private key (hex) for signing events. Falls back to /// BUZZ_RELAY_PRIVATE_KEY env var. If neither is set, generates /// an ephemeral key (events will be unverifiable after restart). @@ -211,8 +217,12 @@ async fn run(cli: Cli) -> Result { Command::Deletions { command } => deletions::run(command).await, Command::UserDeletions { command } => user_deletions::run(command).await, Command::NipFi { command } => nip_fi::run(command).await, - Command::ReconcileChannels { channel, relay_key } => { - reconcile_channels(channel, relay_key).await?; + Command::ReconcileChannels { + channel, + metadata, + relay_key, + } => { + reconcile_channels(channel, metadata, relay_key).await?; Ok(0) } } @@ -721,6 +731,7 @@ async fn resolve_admin_tenant(db: &Db) -> Result { async fn reconcile_channels( channel_arg: Option, + republish_metadata: bool, relay_key_arg: Option, ) -> Result<()> { use buzz_core::channel::group_state_identity_tags; @@ -800,11 +811,16 @@ async fn reconcile_channels( let members = db.get_members(tenant.community(), channel.id).await?; - // A targeted repair is deliberately roster-only. kind:39000 metadata - // is richer than this legacy backfill builder, and kind:39001 is not - // part of the stale-roster incident; replacing either can destroy - // canonical state. Full backfill still creates all three event kinds + // A targeted repair is roster-only unless `--metadata` asks for + // kind:39000 too; kind:39001 is never replaced by a targeted repair. + // kind:39000 is built from the channel row by the same function the + // relay uses, so an operator rename (for example + // scripts/maintenance/rename_duplicate_listed_channel_names.sql) + // reaches clients. Full backfill still creates all three event kinds // for channels with no discovery metadata. + if republish_metadata { + republish_channel_metadata(&db, &tenant, channel.id, &relay_keys).await?; + } if target_channel.is_none() { // kind:39000 — channel metadata { @@ -874,6 +890,74 @@ async fn reconcile_channels( Ok(()) } +/// Makes the newest relay-signed kind:39000 of a channel match its row, for +/// example after an operator rename. +/// +/// Run it with no relay serving: the relay builds kind:39000 from a channel +/// read outside the replacement (`emit_group_discovery_events`), so a live +/// edit between this command's read and write could leave the older +/// snapshot. The command signs later than the stored event, as the relay +/// does, so it also replaces an event from the same second. It fails unless +/// a read after its write finds the stored event matching the row. +async fn republish_channel_metadata( + db: &Db, + tenant: &TenantContext, + channel_id: uuid::Uuid, + relay_keys: &Keys, +) -> Result<()> { + let latest = || async { + let stored = db + .query_events(&buzz_db::event::EventQuery { + kinds: Some(vec![39000]), + pubkey: Some(relay_keys.public_key().to_bytes().to_vec()), + d_tag: Some(channel_id.to_string()), + limit: Some(1), + ..buzz_db::event::EventQuery::for_community(tenant.community()) + }) + .await?; + Ok::<_, anyhow::Error>(stored.into_iter().next().map(|stored| stored.event)) + }; + let row_tags = || async { + let channel = db.get_channel(tenant.community(), channel_id).await?; + let members = db.get_members(tenant.community(), channel_id).await?; + Ok::<_, anyhow::Error>(buzz_db::channel::group_metadata_tags(&channel, &members)) + }; + let matches = |event: &nostr::Event, tags: &[Vec]| { + event + .tags + .iter() + .map(|tag| tag.as_slice()) + .eq(tags.iter().map(Vec::as_slice)) + }; + + let tags = row_tags().await?; + let stored = latest().await?; + if stored.as_ref().is_some_and(|event| matches(event, &tags)) { + return Ok(()); + } + let created_at = stored + .map_or(0, |event| event.created_at.as_secs() + 1) + .max(nostr::Timestamp::now().as_secs()); + let event = EventBuilder::new(Kind::Custom(39000), "") + .tags( + tags.into_iter() + .map(Tag::parse) + .collect::, _>>()?, + ) + .custom_created_at(nostr::Timestamp::from(created_at)) + .sign_with_keys(relay_keys) + .map_err(|e| anyhow::anyhow!("sign kind:39000: {e}"))?; + db.replace_addressable_event(tenant.community(), &event, Some(channel_id)) + .await?; + let tags = row_tags().await?; + if !latest().await?.is_some_and(|event| matches(&event, &tags)) { + return Err(anyhow::anyhow!( + "channel {channel_id} changed while its kind:39000 was republished; stop every relay and run the command again" + )); + } + Ok(()) +} + #[cfg(test)] mod storage_snapshot_tests { use std::sync::atomic::{AtomicUsize, Ordering}; diff --git a/crates/buzz-admin/tests/postgres_reconcile_metadata.rs b/crates/buzz-admin/tests/postgres_reconcile_metadata.rs new file mode 100644 index 0000000000..2a05101c8d --- /dev/null +++ b/crates/buzz-admin/tests/postgres_reconcile_metadata.rs @@ -0,0 +1,111 @@ +//! `buzz-admin reconcile-channels --channel --metadata` leaves the newest +//! relay-signed kind:39000 matching the channel row, also when an older +//! snapshot of the row holds the replacement with a newer timestamp. + +use buzz_core::CommunityId; +use buzz_db::event::EventQuery; +use buzz_db::{Db, DbConfig}; +use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; +use tokio::process::Command; +use uuid::Uuid; + +fn admin_binary() -> std::ffi::OsString { + // Nextest relocates binaries when extracting an archive. Cargo's build-time + // path remains the fallback for local cargo test runs. + std::env::var_os("NEXTEST_BIN_EXE_buzz_admin") + .unwrap_or_else(|| env!("CARGO_BIN_EXE_buzz-admin").into()) +} + +fn name_tag(event: &nostr::Event) -> Option<&str> { + event.tags.iter().find_map(|tag| match tag.as_slice() { + [key, value, ..] if key == "name" => Some(value.as_str()), + _ => None, + }) +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn metadata_republish_replaces_a_stale_snapshot_that_holds_the_replacement() { + let url = std::env::var("BUZZ_TEST_DATABASE_URL") + .expect("run through scripts/postgres-test-run.sh for an isolated database"); + let db = Db::new(&DbConfig { + database_url: url.clone(), + ..DbConfig::default() + }) + .await + .expect("connect to test DB"); + let community = Uuid::new_v4(); + let host = format!("reconcile-{}.example", community.simple()); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community) + .bind(&host) + .execute(db.pool()) + .await + .expect("insert community"); + let community = CommunityId::from_uuid(community); + let channel = Uuid::new_v4(); + sqlx::query( + "INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by) \ + VALUES ($1, $2, 'square-ios-79f8e9d4', 'stream', 'open', $3)", + ) + .bind(channel) + .bind(community.as_uuid()) + .bind(Keys::generate().public_key().to_bytes().to_vec()) + .execute(db.pool()) + .await + .expect("insert renamed channel"); + + // The row before the repair renamed it, dated ahead of this command's + // clock: a relay that read the old row published it, signing later than + // the event it replaced. The command must still replace it. + let relay = Keys::generate(); + let mut before_rename = db.get_channel(community, channel).await.unwrap(); + before_rename.name = "square-ios".to_owned(); + let stale = EventBuilder::new(Kind::Custom(39000), "") + .tags( + buzz_db::channel::group_metadata_tags(&before_rename, &[]) + .into_iter() + .map(Tag::parse) + .collect::, _>>() + .unwrap(), + ) + .custom_created_at(Timestamp::from(Timestamp::now().as_secs() + 60)) + .sign_with_keys(&relay) + .unwrap(); + let (_, inserted) = db + .replace_addressable_event(community, &stale, Some(channel)) + .await + .unwrap(); + assert!(inserted); + + let output = Command::new(admin_binary()) + .args(["reconcile-channels", "--channel", &channel.to_string()]) + .args([ + "--metadata", + "--relay-key", + &relay.secret_key().to_secret_hex(), + ]) + .env("DATABASE_URL", &url) + .env("RELAY_URL", format!("wss://{host}")) + .output() + .await + .expect("run buzz-admin"); + assert!( + output.status.success(), + "reconcile-channels failed:\n{}", + String::from_utf8_lossy(&output.stderr) + ); + + let newest = db + .query_events(&EventQuery { + kinds: Some(vec![39000]), + d_tag: Some(channel.to_string()), + limit: Some(1), + ..EventQuery::for_community(community) + }) + .await + .unwrap(); + let newest = &newest.first().expect("kind:39000 stored").event; + assert_eq!(name_tag(newest), Some("square-ios-79f8e9d4")); + assert!(newest.created_at > stale.created_at); +} diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 5b43d44364..39e03f5392 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -25,6 +25,11 @@ pub enum DbError { #[error("channel not found: {0}")] ChannelNotFound(uuid::Uuid), + /// Another listed open channel in the community already has this name + /// (see `channel::ensure_listed_channel_name_unique`). + #[error("an open channel named \"{0}\" already exists")] + ChannelNameTaken(String), + /// The requested member is not in the channel. #[error("member not found in channel {0}")] MemberNotFound(uuid::Uuid), diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index cbacf501e6..06d0779696 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/crates/buzz-db/src/runtime/tests.rs @@ -4277,6 +4277,9 @@ async fn e_tag_any_runs_with_real_binds_on_p_join_and_count() { #[path = "tests/thread_window_postgres_tests.rs"] mod thread_window_postgres_tests; +#[path = "tests/listed_channel_name_postgres_tests.rs"] +mod listed_channel_name_postgres_tests; + #[path = "tests/reader_pool_postgres_tests.rs"] mod reader_pool_postgres_tests; diff --git a/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs b/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs new file mode 100644 index 0000000000..b223312dde --- /dev/null +++ b/crates/buzz-db/src/runtime/tests/listed_channel_name_postgres_tests.rs @@ -0,0 +1,226 @@ +//! scripts/maintenance/rename_duplicate_listed_channel_names.sql leaves no two +//! listed channels in a community with one name, ignoring case. It keeps the +//! name on the oldest unarchived channel, picks a new name that is free, and +//! leaves alone a row that a relay changed while the script waited for it. + +use super::*; + +const REPAIR_SCRIPT: &str = + include_str!("../../../../../scripts/maintenance/rename_duplicate_listed_channel_names.sql"); + +struct Seed { + community: Uuid, + created_by: Vec, +} + +impl Seed { + async fn new(pool: &PgPool) -> Self { + let community = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community) + .bind(format!("listed-name-{}.example", community.simple())) + .execute(pool) + .await + .expect("insert community"); + Self { + community, + created_by: nostr::Keys::generate().public_key().to_bytes().to_vec(), + } + } + + /// Inserts an open stream `age` seconds old with plain SQL, as a relay + /// without the name rule did. + async fn open(&self, pool: &PgPool, id: Uuid, name: &str, age: i64) { + sqlx::query( + "INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by, \ + created_at) \ + VALUES ($1, $2, $3, 'stream', 'open', $4, now() - make_interval(secs => $5))", + ) + .bind(id) + .bind(self.community) + .bind(name) + .bind(&self.created_by) + .bind(age as f64) + .execute(pool) + .await + .expect("insert channel"); + } + + async fn names(&self, pool: &PgPool) -> Vec<(Uuid, String)> { + sqlx::query_as("SELECT id, name FROM channels WHERE community_id = $1 ORDER BY id") + .bind(self.community) + .fetch_all(pool) + .await + .expect("read channel names") + } +} + +fn renamed(id: Uuid, name: &str) -> (Uuid, String) { + (id, format!("{name}-{}", &id.to_string()[..8])) +} + +/// Runs the repair the way `psql -f` does: the `DO` block as its own +/// statement, so that its per-batch COMMITs run outside a transaction block. +/// A repair that stops making progress fails at the statement timeout. +/// Returns the number of rows the script's validation query reports. +pub(super) async fn run_repair(pool: &PgPool) -> i64 { + let (repair, validation) = REPAIR_SCRIPT + .split_once("\n-- Validation.") + .expect("script has a validation query"); + let mut conn = pool.acquire().await.expect("acquire repair connection"); + sqlx::query("SET statement_timeout = '30s'") + .execute(&mut *conn) + .await + .expect("bound the repair"); + sqlx::raw_sql(sqlx::AssertSqlSafe(repair)) + .execute(&mut *conn) + .await + .expect("run repair"); + drop(conn); + let (_, query) = validation.split_once("\nSELECT").expect("validation query"); + sqlx::query_scalar::<_, i64>(sqlx::AssertSqlSafe(format!( + "SELECT count(*) FROM (SELECT{}) AS remaining", + query.trim_end().trim_end_matches(';') + ))) + .fetch_one(pool) + .await + .expect("run validation") +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn repair_keeps_each_listed_name_on_its_oldest_unarchived_channel() { + let admin = PgPool::connect(&admin_url().await).await.unwrap(); + let (pool, name) = create_scratch_db(&admin, "listed_name_repair").await; + let seed = Seed::new(&pool).await; + let [kept, archived, newer, private, deleted] = [(); 5].map(|_| Uuid::new_v4()); + seed.open(&pool, kept, "square-ios", 30).await; + seed.open(&pool, archived, "Square-iOS", 60).await; + seed.open(&pool, newer, "SQUARE-IOS", 10).await; + seed.open(&pool, private, "square-ios", 5).await; + seed.open(&pool, deleted, "square-ios", 90).await; + sqlx::query( + "UPDATE channels SET \ + archived_at = CASE WHEN id = $2 THEN now() END, \ + visibility = CASE WHEN id = $3 THEN 'private'::channel_visibility ELSE visibility END, \ + deleted_at = CASE WHEN id = $4 THEN now() END \ + WHERE community_id = $1", + ) + .bind(seed.community) + .bind(archived) + .bind(private) + .bind(deleted) + .execute(&pool) + .await + .unwrap(); + // A community with no duplicates is not touched. + let other = Seed::new(&pool).await; + other.open(&pool, Uuid::new_v4(), "square-ios", 0).await; + let other_before = other.names(&pool).await; + + assert_eq!(run_repair(&pool).await, 0, "no duplicates remain"); + let mut expected = vec![ + (kept, "square-ios".to_owned()), + renamed(archived, "Square-iOS"), + renamed(newer, "SQUARE-IOS"), + (private, "square-ios".to_owned()), + (deleted, "square-ios".to_owned()), + ]; + expected.sort(); + assert_eq!(seed.names(&pool).await, expected); + assert_eq!(other.names(&pool).await, other_before); + + assert_eq!(run_repair(&pool).await, 0); + assert_eq!( + seed.names(&pool).await, + expected, + "a second run changes nothing" + ); + drop_scratch_db(&admin, pool, &name).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn repair_picks_a_free_name_when_the_short_one_is_taken() { + let admin = PgPool::connect(&admin_url().await).await.unwrap(); + let (pool, name) = create_scratch_db(&admin, "listed_name_repair_taken").await; + let seed = Seed::new(&pool).await; + // The longest name the column holds, so the new name must truncate it. + let long = "a".repeat(255); + let duplicate = Uuid::parse_str("20000000-0000-4000-8000-000000000000").unwrap(); + let (kept, squatter) = (Uuid::new_v4(), Uuid::new_v4()); + seed.open(&pool, kept, &long, 30).await; + seed.open(&pool, duplicate, &long, 10).await; + // Another listed channel already has the short new name, in another case. + let short = format!("{}-20000000", "A".repeat(246)); + seed.open(&pool, squatter, &short, 60).await; + + assert_eq!(run_repair(&pool).await, 0, "no duplicates remain"); + let mut expected = vec![ + (kept, long.clone()), + (duplicate, format!("{}-{duplicate}", "a".repeat(218))), + (squatter, short), + ]; + expected.sort(); + assert_eq!(seed.names(&pool).await, expected); + assert!(expected.iter().all(|(_, name)| name.chars().count() <= 255)); + drop_scratch_db(&admin, pool, &name).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn repair_leaves_a_row_that_stopped_being_a_duplicate_while_it_waited() { + let admin = PgPool::connect(&admin_url().await).await.unwrap(); + let (pool, name) = create_scratch_db(&admin, "listed_name_repair_wait").await; + let seed = Seed::new(&pool).await; + let (kept, duplicate) = (Uuid::new_v4(), Uuid::new_v4()); + seed.open(&pool, kept, "payments", 30).await; + seed.open(&pool, duplicate, "payments", 10).await; + + // A relay write holds the duplicate's row while the script starts. + let mut relay = pool.begin().await.unwrap(); + sqlx::query("SELECT 1 FROM channels WHERE community_id = $1 AND id = $2 FOR NO KEY UPDATE") + .bind(seed.community) + .bind(duplicate) + .execute(&mut *relay) + .await + .unwrap(); + let repair_pool = PgPool::connect_with( + (*pool.connect_options()) + .clone() + .application_name("listed-name-repair"), + ) + .await + .unwrap(); + let repair = tokio::spawn({ + let repair_pool = repair_pool.clone(); + async move { run_repair(&repair_pool).await } + }); + crate::test_support::wait_for_lock_wait(&pool, "listed-name-repair").await; + + // The relay makes the channel private and renames it, then commits. + sqlx::query( + "UPDATE channels SET name = 'payments-private', visibility = 'private' \ + WHERE community_id = $1 AND id = $2", + ) + .bind(seed.community) + .bind(duplicate) + .execute(&mut *relay) + .await + .unwrap(); + relay.commit().await.unwrap(); + + assert_eq!(repair.await.unwrap(), 0); + let mut expected = vec![ + (kept, "payments".to_owned()), + (duplicate, "payments-private".to_owned()), + ]; + expected.sort(); + assert_eq!( + seed.names(&pool).await, + expected, + "the private row is not renamed" + ); + repair_pool.close().await; + drop_scratch_db(&admin, pool, &name).await; +} diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index cbe9ce2053..4cde838cf1 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -155,7 +155,9 @@ pub async fn create_channel( .bind(labels) .bind(unlisted) .execute(tx.conn()) - .await?; + .await + .map_err(|e| listed_name_taken(e, Some(name)))?; + ensure_listed_channel_name_unique(&mut tx, id).await?; sqlx::query( r#" @@ -392,12 +394,14 @@ async fn insert_channel_rows( .bind(labels) .bind(unlisted) .execute(tx.conn()) - .await? + .await + .map_err(|e| listed_name_taken(e, Some(name)))? .rows_affected(); let was_created = rows_affected > 0; if was_created { + ensure_listed_channel_name_unique(tx, channel_id).await?; // Bootstrap the creator as owner. sqlx::query( r#" @@ -446,6 +450,81 @@ async fn insert_channel_rows( Ok(was_created) } +/// SQL that is true for a channel the channel directory lists: an open, +/// listed stream or forum that is not deleted. An archived channel is still +/// listed, so it keeps its name. Within a community, no two of these share a +/// name, ignoring case ([`ensure_listed_channel_name_unique`]). +pub(crate) const LISTED_CHANNEL_SQL: &str = + "visibility = 'open' AND NOT unlisted AND channel_type IN ('stream', 'forum') \ + AND deleted_at IS NULL"; + +/// Refuses, with [`DbError::ChannelNameTaken`], a write on `tx` that left +/// `channel_id` listed ([`LISTED_CHANNEL_SQL`]) under a name another listed +/// channel in the community already has, ignoring case. Private channels, +/// DMs and the other types may repeat names: nobody browses them side by +/// side, and refusing a name that a private channel holds would tell anyone +/// that the channel exists. +/// +/// Call it after the write that creates the channel or changes its name, +/// visibility or unlisted flag. It first takes a transaction advisory lock +/// on the name, so two writers that claim one name take turns and the second +/// sees the first's committed row. A writer takes this lock last, after the +/// locks of [`lock_channel_for_settings_write`], and holds at most one. +async fn ensure_listed_channel_name_unique( + tx: &mut crate::AdmittedTx, + channel_id: Uuid, +) -> Result<()> { + let community_id = tx.community(); + let name: Option = sqlx::query_scalar(sqlx::AssertSqlSafe(format!( + "SELECT name FROM channels WHERE community_id = $1 AND id = $2 AND {LISTED_CHANNEL_SQL}" + ))) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await?; + let Some(name) = name else { + return Ok(()); + }; + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1 || lower($2), 0))") + .bind(format!("buzz_channel_name:{}:", community_id.as_uuid())) + .bind(&name) + .execute(tx.conn()) + .await?; + let taken: bool = sqlx::query_scalar(sqlx::AssertSqlSafe(format!( + "SELECT EXISTS (SELECT 1 FROM channels WHERE community_id = $1 AND id <> $2 \ + AND lower(name) = lower($3) AND {LISTED_CHANNEL_SQL})" + ))) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(&name) + .fetch_one(tx.conn()) + .await?; + if taken { + return Err(DbError::ChannelNameTaken(name)); + } + Ok(()) +} + +/// The unique index on `(community_id, lower(name))` over +/// [`LISTED_CHANNEL_SQL`] channels that migration 0079 adds. It refuses a +/// duplicate at the write itself, before +/// [`ensure_listed_channel_name_unique`] runs, so this release maps the +/// refusal too: it can run beside that index during the rollout of 0079 and +/// after a rollback. +const LISTED_NAME_INDEX: &str = "idx_channels_listed_name"; + +/// Reports a channel write that [`LISTED_NAME_INDEX`] refused as +/// [`DbError::ChannelNameTaken`] for `name`, the name the channel would have +/// had. +fn listed_name_taken(error: sqlx::Error, name: Option<&str>) -> DbError { + match (&error, name) { + (sqlx::Error::Database(db), Some(name)) if db.constraint() == Some(LISTED_NAME_INDEX) => { + DbError::ChannelNameTaken(name.to_owned()) + } + _ => error.into(), + } +} + /// SQL that is true when `column` holds a legacy channel type /// ([`buzz_core::channel::LEGACY_CHANNEL_TYPES`]): ` IN ('stream', …)`. /// @@ -1090,21 +1169,30 @@ async fn update_channel_on( q = q.bind(community_id.as_uuid()); q = q.bind(channel_id); - let policy_changed = if updates.visibility.is_some() + // The current name is what a visibility or unlisted change that + // [`LISTED_NAME_INDEX`] refuses would have listed. + let (policy_changed, current_name) = if updates.visibility.is_some() || updates.ttl_seconds.is_some() || updates.unlisted.is_some() { - sqlx::query_scalar::<_,bool>("SELECT (($3::text IS NOT NULL AND visibility::text IS DISTINCT FROM $3) OR ($4 AND ttl_seconds IS DISTINCT FROM $5::int) OR ($6::bool IS NOT NULL AND unlisted IS DISTINCT FROM $6)) FROM channels WHERE community_id=$1 AND id=$2") + sqlx::query_as::<_,(bool, String)>("SELECT (($3::text IS NOT NULL AND visibility::text IS DISTINCT FROM $3) OR ($4 AND ttl_seconds IS DISTINCT FROM $5::int) OR ($6::bool IS NOT NULL AND unlisted IS DISTINCT FROM $6)), name FROM channels WHERE community_id=$1 AND id=$2") .bind(community_id.as_uuid()).bind(channel_id).bind(updates.visibility.as_deref()) .bind(updates.ttl_seconds.is_some()).bind(updates.ttl_seconds.flatten()).bind(updates.unlisted) - .fetch_one(tx.conn()).await? + .fetch_one(tx.conn()).await + .map(|(changed, name)| (changed, Some(name)))? } else { - false + (false, None) }; - let result = q.execute(tx.conn()).await?; + let result = q + .execute(tx.conn()) + .await + .map_err(|e| listed_name_taken(e, updates.name.as_deref().or(current_name.as_deref())))?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } + if updates.name.is_some() || updates.visibility.is_some() || updates.unlisted.is_some() { + ensure_listed_channel_name_unique(tx, channel_id).await?; + } if policy_changed { sqlx::query( "UPDATE communities SET artifact_policy_modified_at=clock_timestamp() WHERE id=$1", @@ -2401,6 +2489,86 @@ mod postgres_tests { assert!(matches!(outcome, ChannelCreate::Created(_)), "{outcome:?}"); } + /// Holds an open create of the listed channel `race` on its own + /// transaction while a second create claims the same name beside it. + async fn second_listed_create_beside_open_create(commit_first: bool) -> Result { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(make_test_community(&pool).await); + let creator = Keys::generate(); + let creator_bytes = creator.public_key().to_bytes().to_vec(); + let listed = |name: &'static str| NewChannel { + name, + visibility: ChannelVisibility::Open, + ..new_channel(&creator_bytes, &[]) + }; + + let mut first = crate::begin_community_write_transaction( + &pool, + community, + crate::observability::WriterOperation::EventWrite, + ) + .await + .expect("begin first create"); + assert!( + insert_channel_rows(&mut first, Uuid::new_v4(), listed("race")) + .await + .expect("first create's rows") + ); + + let second_id = Uuid::new_v4(); + let second = tokio::spawn({ + let (creator, creator_bytes) = (creator.clone(), creator_bytes.clone()); + async move { + let pool = crate::test_support::named_pool("listed-name-race").await; + create_channel_with_event( + &pool, + None, + community, + second_id, + NewChannel { + name: "RACE", + visibility: ChannelVisibility::Open, + ..new_channel(&creator_bytes, &[]) + }, + &create_event(&creator, second_id), + ) + .await + } + }); + crate::test_support::wait_for_lock_wait(&pool, "listed-name-race").await; + assert!( + !second.is_finished(), + "the second create waits for the open create's name" + ); + + if commit_first { + first.commit().await.expect("commit first create"); + } else { + drop(first); + } + second.await.expect("second create task") + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn a_listed_create_beside_an_open_create_of_its_name_is_refused_once_committed() { + let outcome = second_listed_create_beside_open_create(true).await; + assert!( + matches!(&outcome, Err(DbError::ChannelNameTaken(name)) if name == "RACE"), + "{outcome:?}" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn a_listed_create_beside_a_failed_create_of_its_name_succeeds() { + let outcome = second_listed_create_beside_open_create(false).await; + assert!( + matches!(outcome, Ok(ChannelCreate::Created(_))), + "{outcome:?}" + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn a_create_whose_event_fails_leaves_no_channel() { diff --git a/crates/buzz-db/src/store/channel_name_postgres_tests.rs b/crates/buzz-db/src/store/channel_name_postgres_tests.rs new file mode 100644 index 0000000000..bc387dc2a3 --- /dev/null +++ b/crates/buzz-db/src/store/channel_name_postgres_tests.rs @@ -0,0 +1,270 @@ +//! Two listed open channels in a community never share a name, ignoring +//! case. Every write that can make a channel claim a listed name goes through +//! the real store functions here: create, rename, private to open, and +//! unlisted to listed. An archived channel keeps its name; a deleted one +//! frees it. + +use crate::channel::ChannelUpdate; +use crate::error::DbError; +use crate::Db; +use buzz_core::channel::{ChannelType, ChannelVisibility}; +use buzz_core::CommunityId; +use nostr::Keys; +use sqlx::PgPool; +use uuid::Uuid; + +struct Fixture { + db: Db, + community: CommunityId, + owner: Keys, +} + +impl Fixture { + async fn new() -> Self { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB"); + if std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref() != Ok("desired") { + crate::migration::run_migrations(&pool) + .await + .expect("migrate test DB"); + } + Self { + community: insert_community(&pool).await, + db: Db::from_pool(pool), + owner: Keys::generate(), + } + } + + async fn create( + &self, + name: &str, + channel_type: ChannelType, + visibility: ChannelVisibility, + ) -> crate::error::Result { + self.create_in(self.community, name, channel_type, visibility) + .await + } + + async fn create_in( + &self, + community: CommunityId, + name: &str, + channel_type: ChannelType, + visibility: ChannelVisibility, + ) -> crate::error::Result { + self.db + .create_channel( + community, + name, + channel_type, + visibility, + None, + &self.owner.public_key().to_bytes(), + None, + ) + .await + .map(|channel| channel.id) + } + + async fn open(&self, name: &str) -> Uuid { + self.create(name, ChannelType::Stream, ChannelVisibility::Open) + .await + .expect("create open channel") + } + + async fn update(&self, channel: Uuid, update: ChannelUpdate) -> crate::error::Result<()> { + self.db + .update_channel(self.community, channel, update) + .await + .map(|_| ()) + } + + async fn listing_state(&self, channel: Uuid) -> (String, String, bool) { + let channel = self + .db + .get_channel(self.community, channel) + .await + .expect("reload channel"); + (channel.name, channel.visibility, channel.unlisted) + } +} + +async fn insert_community(pool: &PgPool) -> CommunityId { + let community = CommunityId::from_uuid(Uuid::new_v4()); + sqlx::query("INSERT INTO communities(id,host) VALUES($1,$2)") + .bind(community.as_uuid()) + .bind(format!("channel-name-{}.test", community.as_uuid())) + .execute(pool) + .await + .expect("insert community"); + community +} + +fn assert_taken(result: crate::error::Result, name: &str) { + match result { + Err(DbError::ChannelNameTaken(taken)) => assert_eq!(taken, name), + other => panic!("expected {name:?} to be taken, got {other:?}"), + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_listed_name_is_taken_only_for_listed_channels() { + let f = Fixture::new().await; + f.open("Square-iOS").await; + + // Another listed channel cannot take it, whatever its case or type, and + // the display prefix is not part of the name. + for (name, channel_type) in [ + ("square-ios", ChannelType::Stream), + ("SQUARE-IOS", ChannelType::Forum), + ("#square-ios ", ChannelType::Stream), + ] { + assert_taken( + f.create(name, channel_type, ChannelVisibility::Open).await, + buzz_core::channel::canonical_channel_name(name), + ); + } + + // Channels the directory does not list may repeat it. + for (channel_type, visibility) in [ + (ChannelType::Stream, ChannelVisibility::Private), + (ChannelType::Forum, ChannelVisibility::Private), + (ChannelType::Dm, ChannelVisibility::Private), + (ChannelType::Workflow, ChannelVisibility::Open), + (ChannelType::System, ChannelVisibility::Open), + ] { + f.create("square-ios", channel_type, visibility) + .await + .unwrap_or_else(|e| panic!("{channel_type:?}/{visibility:?}: {e}")); + } + + // Names are per community. + let other = insert_community(f.db.pool()).await; + f.create_in( + other, + "square-ios", + ChannelType::Stream, + ChannelVisibility::Open, + ) + .await + .expect("another community may use the name"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_change_that_would_list_a_taken_name_changes_nothing() { + let f = Fixture::new().await; + f.open("payments").await; + + let private = f + .create("payments", ChannelType::Stream, ChannelVisibility::Private) + .await + .expect("private channel"); + let unlisted = f.open("payments-unlisted").await; + f.update( + unlisted, + ChannelUpdate { + name: Some("payments".into()), + unlisted: Some(true), + ..Default::default() + }, + ) + .await + .expect("an unlisted channel may share the name"); + let other = f.open("payments-team").await; + + let cases = [ + ( + private, + ChannelUpdate { + visibility: Some("open".into()), + ..Default::default() + }, + ), + ( + unlisted, + ChannelUpdate { + unlisted: Some(false), + ..Default::default() + }, + ), + ( + other, + ChannelUpdate { + name: Some("PAYMENTS".into()), + description: Some("rolled back with the name".into()), + ..Default::default() + }, + ), + ]; + for (channel, update) in cases { + let before = f.listing_state(channel).await; + let description = + f.db.get_channel(f.community, channel) + .await + .expect("reload") + .description; + let attempted = update.name.clone().unwrap_or_else(|| before.0.clone()); + assert_taken(f.update(channel, update).await, &attempted); + assert_eq!(f.listing_state(channel).await, before); + assert_eq!( + f.db.get_channel(f.community, channel) + .await + .expect("reload") + .description, + description, + "the whole update rolls back" + ); + } + + // A rename to a free name, and the same name in another case, still work. + f.update( + other, + ChannelUpdate { + name: Some("payments-core".into()), + ..Default::default() + }, + ) + .await + .expect("rename to a free name"); + f.update( + other, + ChannelUpdate { + name: Some("Payments-Core".into()), + ..Default::default() + }, + ) + .await + .expect("a channel may change the case of its own name"); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_archived_channel_keeps_its_name_until_it_is_deleted() { + let f = Fixture::new().await; + let archived = f.open("square-ios").await; + f.db.archive_channel(f.community, archived) + .await + .expect("archive"); + + assert_taken( + f.create("square-ios", ChannelType::Stream, ChannelVisibility::Open) + .await, + "square-ios", + ); + f.db.unarchive_channel(f.community, archived) + .await + .expect("unarchive never collides"); + f.db.archive_channel(f.community, archived) + .await + .expect("archive again"); + + assert!(f + .db + .soft_delete_channel(f.community, archived) + .await + .expect("delete")); + f.open("square-ios").await; +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index d5bbde4347..5fe2927ae2 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -96,5 +96,8 @@ mod legacy_channel_types_postgres_tests; #[cfg(test)] mod unlisted_channels_postgres_tests; +#[cfg(test)] +mod channel_name_postgres_tests; + #[cfg(test)] mod admission_postgres_tests; diff --git a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs index 0e3859c05b..fd1996340e 100644 --- a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs @@ -1074,3 +1074,97 @@ async fn removed_creator_join_racing_a_private_switch_is_refused() { "a removed creator must not rejoin a channel that became private" ); } + +/// A create or rename that would list a second open channel under a taken +/// name is refused with its reason as a client error, and stores nothing. +/// Mutation: let `ChannelNameTaken` fall through to the internal-error arm +/// of either kind → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_taken_open_name_is_refused_as_a_client_error() { + use crate::handlers::ingest::IngestError; + + let state = state().await; + let tenant = community(&state, "group-state-name-taken").await; + let owner = Keys::generate(); + let create = |name: &str| { + let channel_id = Uuid::new_v4(); + let event = EventBuilder::new(Kind::Custom(9007), "") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["name", name]).unwrap(), + ]) + .sign_with_keys(&owner) + .expect("sign 9007"); + (channel_id, event) + }; + + let (_, first) = create("launch"); + assert!( + ingest_event(&state, &tenant, first, http_auth(&owner)) + .await + .expect("first create") + .accepted + ); + + let (second, event) = create("Launch"); + match ingest_event(&state, &tenant, event, http_auth(&owner)).await { + Err(IngestError::Rejected(message)) => assert_eq!( + message, + "invalid: an open channel named \"Launch\" already exists" + ), + Err(other) => panic!("expected a refusal, got {other:?}"), + Ok(result) => panic!("expected a refusal, got {:?}", result.message), + } + assert!( + state + .db + .get_channel(tenant.community(), second) + .await + .is_err(), + "the refused create stores no channel" + ); + + let other = create_channel(&state, &tenant, &owner, &[]).await; + let before = state + .db + .get_channel(tenant.community(), other) + .await + .expect("channel") + .name; + let rename = EventBuilder::new(Kind::Custom(9002), "") + .tags([ + Tag::parse(["h", &other.to_string()]).unwrap(), + Tag::parse(["name", "LAUNCH"]).unwrap(), + ]) + .sign_with_keys(&owner) + .expect("sign 9002"); + let rename_id = rename.id; + match ingest_event(&state, &tenant, rename, http_auth(&owner)).await { + Err(IngestError::Rejected(message)) => assert_eq!( + message, + "invalid: an open channel named \"LAUNCH\" already exists" + ), + Err(other) => panic!("expected a refusal, got {other:?}"), + Ok(result) => panic!("expected a refusal, got {:?}", result.message), + } + assert_eq!( + state + .db + .get_channel(tenant.community(), other) + .await + .expect("channel") + .name, + before, + "the refused rename keeps the name" + ); + let stored: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM events WHERE community_id = $1 AND id = $2)", + ) + .bind(tenant.community().as_uuid()) + .bind(rename_id.as_bytes().as_slice()) + .fetch_one(state.db.pool()) + .await + .expect("look up the 9002"); + assert!(!stored, "the refused 9002 is not stored"); +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 0e44eb8c1e..4cb74418b8 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -3558,8 +3558,12 @@ async fn ingest_event_inner( .db .create_channel_with_event(tenant.community(), create.channel_id, channel, &event) .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))? - { + .map_err(|e| match e { + buzz_db::DbError::ChannelNameTaken(_) => { + IngestError::Rejected(format!("invalid: {e}")) + } + _ => IngestError::Internal(format!("error: {e}")), + })? { buzz_db::channel::ChannelCreate::Created(stored) => { metrics::counter!( "buzz_channels_created_total", @@ -3624,6 +3628,9 @@ async fn ingest_event_inner( Some(buzz_db::DbError::AccessDenied(reason)) => { IngestError::Rejected(format!("invalid: {reason}")) } + Some(taken @ buzz_db::DbError::ChannelNameTaken(_)) => { + IngestError::Rejected(format!("invalid: {taken}")) + } _ => IngestError::Internal(format!("error: channel update failed: {e}")), })? } else if buzz_core::kind::is_replaceable(kind_u32) { diff --git a/scripts/maintenance/rename_duplicate_listed_channel_names.sql b/scripts/maintenance/rename_duplicate_listed_channel_names.sql new file mode 100644 index 0000000000..0264031bb0 --- /dev/null +++ b/scripts/maintenance/rename_duplicate_listed_channel_names.sql @@ -0,0 +1,161 @@ +-- OUT-OF-BAND MAINTENANCE: do not run from relay startup migrations. +-- +-- Gives every listed open channel a name that no other listed open channel in +-- its community has, ignoring case. A listed channel is an open, not unlisted +-- stream or forum that is not deleted; an archived one still counts. This is +-- the rule the relay enforces on create and rename (`LISTED_CHANNEL_SQL` in +-- crates/buzz-db/src/store/channel.rs). It repairs the duplicates created +-- before that rule, so the contract migration's unique index can build. +-- +-- Run it only after every serving relay enforces the rule. An older relay can +-- create a new duplicate while the script runs. +-- +-- In each group of channels that share a name, one channel keeps it: the +-- oldest unarchived one, or the oldest archived one when all are archived. +-- Every other channel in the group becomes "-", or "-" when another listed channel already has that name. +-- The script prints a NOTICE for each rename. When both names are taken, it +-- leaves the channel unchanged with a NOTICE, and the validation below +-- reports it. +-- +-- The rename writes only the channel row. Clients read the name from the +-- relay-signed kind:39000, so republish it for each renamed channel: +-- +-- buzz-admin reconcile-channels --channel --metadata +-- +-- Each batch renames up to 100 channels in one community and commits. It +-- takes the community deletion lock first and skips a community that is not +-- active. It locks every row of the batch before it takes the relay's name +-- lock for any new name, in the order the relay takes them. After it locks a +-- row, it checks again that the row is still a listed duplicate, because a +-- relay may have changed it meanwhile. The script is idempotent: a second run +-- finds nothing to rename. +-- +-- Run it outside a transaction, so that each batch can commit: +-- +-- psql -v ON_ERROR_STOP=1 -f scripts/maintenance/rename_duplicate_listed_channel_names.sql + +DO $$ +DECLARE + target UUID; + skipped UUID[] := '{}'; + batch UUID[]; + duplicate UUID; + live RECORD; + new_name TEXT; + renamed INT; +BEGIN + LOOP + SELECT ch.community_id INTO target + FROM channels ch + WHERE ch.visibility = 'open' AND NOT ch.unlisted + AND ch.channel_type IN ('stream', 'forum') AND ch.deleted_at IS NULL + AND ch.community_id <> ALL (skipped) + GROUP BY ch.community_id, lower(ch.name) + HAVING count(*) > 1 + LIMIT 1; + EXIT WHEN target IS NULL; + + PERFORM set_config('lock_timeout', '5s', true); + PERFORM pg_advisory_xact_lock_shared(community_deletion_lock_key(target)); + IF NOT EXISTS ( + SELECT 1 FROM communities + WHERE id = target AND deletion_state = 'active' AND deleted_at IS NULL + ) THEN + RAISE NOTICE 'skipped community %: it is not active', target; + skipped := skipped || target; + COMMIT; + CONTINUE; + END IF; + + -- Lock every row of the batch before any name lock, in the order + -- the relay takes them. + SELECT array_agg(locked.id ORDER BY locked.id) INTO batch + FROM ( + WITH ranked AS ( + SELECT id, row_number() OVER ( + PARTITION BY lower(name) + ORDER BY archived_at IS NOT NULL, created_at, id + ) AS place + FROM channels + WHERE community_id = target + AND visibility = 'open' AND NOT unlisted + AND channel_type IN ('stream', 'forum') AND deleted_at IS NULL + ) + SELECT ch.id + FROM channels ch + JOIN ranked ON ranked.id = ch.id + WHERE ch.community_id = target AND ranked.place > 1 + ORDER BY ch.id + LIMIT 100 + FOR NO KEY UPDATE OF ch + ) locked; + + renamed := 0; + FOREACH duplicate IN ARRAY coalesce(batch, '{}') + LOOP + -- The ranking used the statement's snapshot. Now that the row is + -- locked, read it again: a relay may have renamed it, made it + -- private, or renamed the channel that kept the name. + SELECT ch.name, ch.archived_at, ch.created_at INTO live + FROM channels ch + WHERE ch.community_id = target AND ch.id = duplicate + AND ch.visibility = 'open' AND NOT ch.unlisted + AND ch.channel_type IN ('stream', 'forum') AND ch.deleted_at IS NULL; + CONTINUE WHEN NOT FOUND OR NOT EXISTS ( + SELECT 1 FROM channels other + WHERE other.community_id = target AND other.id <> duplicate + AND lower(other.name) = lower(live.name) + AND other.visibility = 'open' AND NOT other.unlisted + AND other.channel_type IN ('stream', 'forum') AND other.deleted_at IS NULL + AND (other.archived_at IS NOT NULL, other.created_at, other.id) + < (live.archived_at IS NOT NULL, live.created_at, duplicate) + ); + + new_name := NULL; + FOREACH new_name IN ARRAY ARRAY[ + left(live.name, 246) || '-' || left(duplicate::text, 8), + left(live.name, 218) || '-' || duplicate::text + ] LOOP + PERFORM pg_advisory_xact_lock(hashtextextended( + 'buzz_channel_name:' || target::text || ':' || lower(new_name), 0)); + EXIT WHEN NOT EXISTS ( + SELECT 1 FROM channels other + WHERE other.community_id = target AND other.id <> duplicate + AND lower(other.name) = lower(new_name) + AND other.visibility = 'open' AND NOT other.unlisted + AND other.channel_type IN ('stream', 'forum') AND other.deleted_at IS NULL + ); + new_name := NULL; + END LOOP; + IF new_name IS NULL THEN + RAISE NOTICE 'left channel % in community % as "%": both new names are taken', + duplicate, target, live.name; + CONTINUE; + END IF; + + UPDATE channels SET name = new_name, updated_at = NOW() + WHERE community_id = target AND id = duplicate; + RAISE NOTICE 'renamed channel % in community % from "%" to "%"', + duplicate, target, live.name, new_name; + renamed := renamed + 1; + END LOOP; + + -- A batch that found nothing to rename would find the same community + -- again; skip it so the loop ends. The validation below reports it. + IF renamed = 0 THEN + skipped := skipped || target; + END IF; + COMMIT; + END LOOP; +END +$$; + +-- Validation. Expected result: zero rows. A row names a community the script +-- skipped (not active) or a duplicate created while it ran; rerun the script. +SELECT community_id, lower(name) AS name, count(*) AS channels +FROM channels +WHERE visibility = 'open' AND NOT unlisted + AND channel_type IN ('stream', 'forum') AND deleted_at IS NULL +GROUP BY community_id, lower(name) +HAVING count(*) > 1;