mirror of
https://github.com/block/buzz.git
synced 2026-11-06 01:11:33 +00:00
fix(relay): rename duplicate channel names through the relay
Replace the direct-SQL repair, which needed every relay pod stopped, with renames that go through the relay's normal kind:9002 path: - A relay operator may send a name-only kind:9002 for any channel. Every other privileged change still needs a channel owner or admin. A moderator gets no new right. - scripts/maintenance/plan_duplicate_listed_channel_renames.sql lists the renames (read-only). rename_duplicate_listed_channel_names.sh sends each one with `buzz channels update`, signed by an operator key, then reads the plan again and fails if any rename remains. The relay enforces the name rule and publishes kind:39000 as for any rename, so relays keep serving while it runs. - Remove `buzz-admin reconcile-channel-groups --metadata` and the SQL repair script; nothing needs them now. - Refuse a kind:9007 with no h tag at ingest, before it is stored. Such a create used to run in side effects, so a refusal there (a taken name) still answered accepted. Every first-party client sends h. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
@@ -137,18 +137,12 @@ 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 unless
|
||||
/// `--metadata` is given.
|
||||
/// replaced; canonical metadata and admin events remain untouched.
|
||||
ReconcileChannels {
|
||||
/// Optional channel UUID to force-republish.
|
||||
#[arg(long)]
|
||||
channel: Option<String>,
|
||||
|
||||
/// 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).
|
||||
@@ -217,12 +211,8 @@ async fn run(cli: Cli) -> Result<i32> {
|
||||
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,
|
||||
metadata,
|
||||
relay_key,
|
||||
} => {
|
||||
reconcile_channels(channel, metadata, relay_key).await?;
|
||||
Command::ReconcileChannels { channel, relay_key } => {
|
||||
reconcile_channels(channel, relay_key).await?;
|
||||
Ok(0)
|
||||
}
|
||||
}
|
||||
@@ -731,7 +721,6 @@ async fn resolve_admin_tenant(db: &Db) -> Result<TenantContext> {
|
||||
|
||||
async fn reconcile_channels(
|
||||
channel_arg: Option<String>,
|
||||
republish_metadata: bool,
|
||||
relay_key_arg: Option<String>,
|
||||
) -> Result<()> {
|
||||
use buzz_core::channel::group_state_identity_tags;
|
||||
@@ -811,16 +800,11 @@ async fn reconcile_channels(
|
||||
|
||||
let members = db.get_members(tenant.community(), channel.id).await?;
|
||||
|
||||
// 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
|
||||
// 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
|
||||
// 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
|
||||
{
|
||||
@@ -890,74 +874,6 @@ 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<String>]| {
|
||||
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::<Result<Vec<_>, _>>()?,
|
||||
)
|
||||
.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};
|
||||
|
||||
@@ -1,111 +0,0 @@
|
||||
//! `buzz-admin reconcile-channels --channel <id> --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::<Result<Vec<_>, _>>()
|
||||
.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);
|
||||
}
|
||||
@@ -1,29 +1,32 @@
|
||||
//! 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.
|
||||
//! scripts/maintenance/plan_duplicate_listed_channel_renames.sql lists the
|
||||
//! renames that leave no two listed channels in a community with one name,
|
||||
//! ignoring case. It keeps the name on the oldest unarchived channel and picks
|
||||
//! a new name that no other listed channel has.
|
||||
|
||||
use super::*;
|
||||
|
||||
const REPAIR_SCRIPT: &str =
|
||||
include_str!("../../../../../scripts/maintenance/rename_duplicate_listed_channel_names.sql");
|
||||
const PLAN_SQL: &str =
|
||||
include_str!("../../../../../scripts/maintenance/plan_duplicate_listed_channel_renames.sql");
|
||||
|
||||
struct Seed {
|
||||
community: Uuid,
|
||||
host: String,
|
||||
created_by: Vec<u8>,
|
||||
}
|
||||
|
||||
impl Seed {
|
||||
async fn new(pool: &PgPool) -> Self {
|
||||
let community = Uuid::new_v4();
|
||||
let host = format!("listed-name-{}.example", community.simple());
|
||||
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
|
||||
.bind(community)
|
||||
.bind(format!("listed-name-{}.example", community.simple()))
|
||||
.bind(&host)
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("insert community");
|
||||
Self {
|
||||
community,
|
||||
host,
|
||||
created_by: nostr::Keys::generate().public_key().to_bytes().to_vec(),
|
||||
}
|
||||
}
|
||||
@@ -59,39 +62,43 @@ 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(';')
|
||||
/// The plan for `host`: (channel, current name, new name), as the rename
|
||||
/// script reads it.
|
||||
pub(super) async fn rename_plan(pool: &PgPool, host: &str) -> Vec<(Uuid, String, String)> {
|
||||
let (_, query) = PLAN_SQL.split_once("\nWITH").expect("plan query");
|
||||
sqlx::query_as(sqlx::AssertSqlSafe(format!(
|
||||
"WITH{}",
|
||||
query
|
||||
.trim_end()
|
||||
.trim_end_matches(';')
|
||||
.replace(":'host'", "$1")
|
||||
)))
|
||||
.fetch_one(pool)
|
||||
.bind(host)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.expect("run validation")
|
||||
.expect("run the rename plan")
|
||||
}
|
||||
|
||||
/// Applies the plan the way the relay applies each rename the script sends,
|
||||
/// and returns how many channels it renamed.
|
||||
pub(super) async fn rename_duplicates(pool: &PgPool, host: &str) -> usize {
|
||||
let plan = rename_plan(pool, host).await;
|
||||
for (channel, _, new_name) in &plan {
|
||||
sqlx::query("UPDATE channels SET name = $2 WHERE id = $1")
|
||||
.bind(channel)
|
||||
.bind(new_name)
|
||||
.execute(pool)
|
||||
.await
|
||||
.expect("rename channel");
|
||||
}
|
||||
plan.len()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn repair_keeps_each_listed_name_on_its_oldest_unarchived_channel() {
|
||||
async fn plan_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 (pool, name) = create_scratch_db(&admin, "listed_name_plan").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;
|
||||
@@ -113,37 +120,58 @@ async fn repair_keeps_each_listed_name_on_its_oldest_unarchived_channel() {
|
||||
.execute(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
// A community with no duplicates is not touched.
|
||||
// Another community's channel with the same name is not a duplicate.
|
||||
let other = Seed::new(&pool).await;
|
||||
other.open(&pool, Uuid::new_v4(), "square-ios", 0).await;
|
||||
let other_before = other.names(&pool).await;
|
||||
// A community that is not active is left alone.
|
||||
let inactive = Seed::new(&pool).await;
|
||||
inactive.open(&pool, Uuid::new_v4(), "payments", 20).await;
|
||||
inactive.open(&pool, Uuid::new_v4(), "payments", 10).await;
|
||||
crate::test_support::quiesce_community_for_tests(
|
||||
&pool,
|
||||
buzz_core::CommunityId::from_uuid(inactive.community),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert_eq!(run_repair(&pool).await, 0, "no duplicates remain");
|
||||
let mut expected = vec![
|
||||
(
|
||||
archived,
|
||||
"Square-iOS".to_owned(),
|
||||
renamed(archived, "Square-iOS").1,
|
||||
),
|
||||
(
|
||||
newer,
|
||||
"SQUARE-IOS".to_owned(),
|
||||
renamed(newer, "SQUARE-IOS").1,
|
||||
),
|
||||
];
|
||||
expected.sort();
|
||||
assert_eq!(rename_plan(&pool, &seed.host).await, expected);
|
||||
assert!(rename_plan(&pool, &other.host).await.is_empty());
|
||||
assert!(rename_plan(&pool, &inactive.host).await.is_empty());
|
||||
|
||||
assert_eq!(rename_duplicates(&pool, &seed.host).await, 2);
|
||||
assert!(
|
||||
rename_plan(&pool, &seed.host).await.is_empty(),
|
||||
"nothing is left to rename"
|
||||
);
|
||||
let mut names = 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"
|
||||
);
|
||||
names.sort();
|
||||
assert_eq!(seed.names(&pool).await, names);
|
||||
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() {
|
||||
async fn plan_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 (pool, name) = create_scratch_db(&admin, "listed_name_plan_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);
|
||||
@@ -155,72 +183,89 @@ async fn repair_picks_a_free_name_when_the_short_one_is_taken() {
|
||||
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));
|
||||
let new_name = format!("{}-{duplicate}", "a".repeat(218));
|
||||
assert_eq!(
|
||||
rename_plan(&pool, &seed.host).await,
|
||||
vec![(duplicate, long.clone(), new_name.clone())]
|
||||
);
|
||||
assert!(new_name.chars().count() <= 255);
|
||||
drop_scratch_db(&admin, pool, &name).await;
|
||||
}
|
||||
|
||||
/// The real script, with a stand-in CLI that logs each call and applies the
|
||||
/// rename. A name may hold any character but NUL, so a name holding control
|
||||
/// characters must still yield exactly one rename, of its own channel.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn repair_leaves_a_row_that_stopped_being_a_duplicate_while_it_waited() {
|
||||
async fn script_renames_each_planned_channel_once_whatever_its_name() {
|
||||
let admin = PgPool::connect(&admin_url().await).await.unwrap();
|
||||
let (pool, name) = create_scratch_db(&admin, "listed_name_repair_wait").await;
|
||||
let (pool, name) = create_scratch_db(&admin, "listed_name_script").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;
|
||||
let [kept, duplicate, other] = [(); 3].map(|_| Uuid::new_v4());
|
||||
let tricky = format!("ops\x1e{other}\x1fowned\x1f\n\t\"$(x)");
|
||||
seed.open(&pool, kept, &tricky, 30).await;
|
||||
seed.open(&pool, duplicate, &tricky, 10).await;
|
||||
seed.open(&pool, other, "other", 20).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"),
|
||||
let dir = std::env::temp_dir().join(format!("listed-name-script-{}", Uuid::new_v4().simple()));
|
||||
std::fs::create_dir(&dir).unwrap();
|
||||
let (cli, log) = (dir.join("buzz"), dir.join("calls"));
|
||||
std::fs::write(
|
||||
&cli,
|
||||
"#!/usr/bin/env bash\nset -euo pipefail\nprintf '%s\\0' \"$@\" >>\"$STUB_LOG\"\n\
|
||||
psql \"$DATABASE_URL\" -X -q -v ON_ERROR_STOP=1 -v id=\"$4\" -v name=\"$6\" <<'SQL'\n\
|
||||
UPDATE channels SET name = :'name' WHERE id = :'id';\nSQL\n",
|
||||
)
|
||||
.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;
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
std::fs::set_permissions(&cli, std::fs::Permissions::from_mode(0o755)).unwrap();
|
||||
|
||||
// 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"
|
||||
let base = admin_url().await;
|
||||
let database_url = format!("{}/{}", &base[..base.rfind('/').unwrap()], name);
|
||||
let mut path = std::env::var("PATH").unwrap_or_default();
|
||||
if let Ok(bin) = std::env::var("PG_BIN_DIR") {
|
||||
path = format!("{bin}:{path}");
|
||||
}
|
||||
let output = std::process::Command::new(concat!(
|
||||
env!("CARGO_MANIFEST_DIR"),
|
||||
"/../../scripts/maintenance/rename_duplicate_listed_channel_names.sh"
|
||||
))
|
||||
.arg(&seed.host)
|
||||
.env("PATH", path)
|
||||
.env("DATABASE_URL", &database_url)
|
||||
.env("BUZZ_PRIVATE_KEY", "unused")
|
||||
.env("BUZZ_CLI", &cli)
|
||||
.env("STUB_LOG", &log)
|
||||
.output()
|
||||
.expect("run the rename script");
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"script failed: {}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
repair_pool.close().await;
|
||||
|
||||
let new_name = renamed(duplicate, &tricky).1;
|
||||
let calls = std::fs::read(&log).unwrap();
|
||||
let args: Vec<&[u8]> = calls.split(|b| *b == 0).collect();
|
||||
let channel = duplicate.to_string();
|
||||
let expected: Vec<&[u8]> = vec![
|
||||
b"channels",
|
||||
b"update",
|
||||
b"--channel",
|
||||
channel.as_bytes(),
|
||||
b"--name",
|
||||
new_name.as_bytes(),
|
||||
b"",
|
||||
];
|
||||
assert_eq!(args, expected, "one call, for the duplicate only");
|
||||
let mut names = vec![
|
||||
(kept, tricky.clone()),
|
||||
(duplicate, new_name.clone()),
|
||||
(other, "other".to_owned()),
|
||||
];
|
||||
names.sort();
|
||||
assert_eq!(seed.names(&pool).await, names);
|
||||
std::fs::remove_dir_all(&dir).unwrap();
|
||||
drop_scratch_db(&admin, pool, &name).await;
|
||||
}
|
||||
|
||||
@@ -311,6 +311,18 @@ pub async fn resolve_admin_principal(
|
||||
.ok_or_else(ApiError::forbidden)
|
||||
}
|
||||
|
||||
/// True when `pubkey` holds the operator role (not moderator), with the same
|
||||
/// precedence as [`lookup_admin_principal`]. A roster lookup failure is an
|
||||
/// `Err`, never "not an operator".
|
||||
pub(crate) async fn is_relay_operator(
|
||||
state: &AppState,
|
||||
pubkey: [u8; 32],
|
||||
) -> Result<bool, ApiError> {
|
||||
Ok(lookup_admin_principal(state, pubkey)
|
||||
.await?
|
||||
.is_some_and(|p| p.role == AdminRole::Operator))
|
||||
}
|
||||
|
||||
/// Effective staff grant for `pubkey`, same precedence as
|
||||
/// [`resolve_admin_principal`]. `Ok(None)` means "not staff"; a roster lookup
|
||||
/// failure is an `Err`, so callers fail closed instead of treating it as
|
||||
|
||||
@@ -40,6 +40,7 @@ pub(crate) fn is_admin_host(state: &crate::state::AppState, headers: &HeaderMap)
|
||||
/// [`auth::admin_api_origin`]). Re-exported so the NIP-11 builder can derive
|
||||
/// the advertised origin without reaching into the private `auth` module.
|
||||
pub(crate) use auth::admin_api_origin;
|
||||
pub(crate) use auth::is_relay_operator;
|
||||
|
||||
/// Build the deployment-admin routes.
|
||||
///
|
||||
|
||||
@@ -8,7 +8,7 @@ use buzz_db::event::EventQuery;
|
||||
use nostr::{EventBuilder, Keys, Kind, Tag};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::handlers::ingest::{ingest_event, HttpAuthMethod, IngestAuth};
|
||||
use crate::handlers::ingest::{ingest_event, HttpAuthMethod, IngestAuth, IngestError};
|
||||
use crate::state::AppState;
|
||||
use buzz_auth::Scope;
|
||||
|
||||
@@ -885,79 +885,49 @@ async fn system_channel_defaults_private_and_hidden() {
|
||||
assert!(!has_marker(&tags, "hidden"), "{tags:?}");
|
||||
}
|
||||
|
||||
/// A kind:9007 without an `h` tag is created by the side-effect handler, not
|
||||
/// by the ingest pre-create. That path must give a system channel the same
|
||||
/// private default, so strangers can neither read nor write it.
|
||||
/// Mutation: default visibility to open in `handle_create_group` → RED.
|
||||
/// A kind:9007 without an `h` tag is refused before it is stored. Its channel
|
||||
/// would be created only after the event committed, so a refused create (a
|
||||
/// taken name, for example) would still answer accepted.
|
||||
/// Mutation: drop the `h`-tag refusal in `ingest_event` → RED.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn system_channel_without_h_tag_defaults_private() {
|
||||
async fn create_without_h_tag_is_refused_before_it_is_stored() {
|
||||
let state = state().await;
|
||||
let tenant = community(&state, "group-state-system-no-h").await;
|
||||
let tenant = community(&state, "group-state-create-no-h").await;
|
||||
let creator = Keys::generate();
|
||||
let stranger = Keys::generate();
|
||||
|
||||
let name = format!("system-no-h-{}", Uuid::new_v4().simple());
|
||||
let name = format!("no-h-{}", Uuid::new_v4().simple());
|
||||
let create = EventBuilder::new(Kind::Custom(9007), "")
|
||||
.tags([
|
||||
Tag::parse(["name", &name]).unwrap(),
|
||||
Tag::parse(["channel_type", "system"]).unwrap(),
|
||||
])
|
||||
.tags([Tag::parse(["name", &name]).unwrap()])
|
||||
.sign_with_keys(&creator)
|
||||
.expect("sign 9007");
|
||||
let result = ingest_event(&state, &tenant, create, http_auth(&creator))
|
||||
.await
|
||||
.unwrap_or_else(|error| panic!("create channel: {error:?}"));
|
||||
assert!(result.accepted, "{}", result.message);
|
||||
|
||||
let (channel_id, visibility): (Uuid, String) = sqlx::query_as(
|
||||
"SELECT id, visibility::text FROM channels WHERE community_id = $1 AND name = $2",
|
||||
)
|
||||
.bind(*tenant.community().as_uuid())
|
||||
.bind(&name)
|
||||
.fetch_one(state.db.pool())
|
||||
.await
|
||||
.expect("created channel");
|
||||
assert_eq!(visibility, "private");
|
||||
|
||||
let tags = group_state_tags(&state, &tenant, channel_id, 39000).await;
|
||||
assert!(has_marker(&tags, "private"), "{tags:?}");
|
||||
assert!(has_marker(&tags, "hidden"), "{tags:?}");
|
||||
assert!(!has_marker(&tags, "public"), "{tags:?}");
|
||||
|
||||
let stranger_bytes = stranger.public_key().to_bytes().to_vec();
|
||||
assert!(
|
||||
crate::handlers::ingest::check_channel_membership(
|
||||
tenant.community(),
|
||||
&state,
|
||||
channel_id,
|
||||
&stranger_bytes,
|
||||
None,
|
||||
crate::handlers::ingest::ChannelWrite::Membership,
|
||||
)
|
||||
.await
|
||||
.is_err(),
|
||||
"a stranger must not pass the channel access check"
|
||||
);
|
||||
let readable = state
|
||||
.db
|
||||
.get_accessible_channel_ids(tenant.community(), &stranger_bytes)
|
||||
.await
|
||||
.expect("accessible channels");
|
||||
assert!(!readable.contains(&channel_id), "stranger can read it");
|
||||
|
||||
let message = EventBuilder::new(Kind::Custom(9), "hello")
|
||||
.tags([Tag::parse(["h", &channel_id.to_string()]).unwrap()])
|
||||
.sign_with_keys(&stranger)
|
||||
.unwrap();
|
||||
match ingest_event(&state, &tenant, message, http_auth(&stranger)).await {
|
||||
Ok(result) => assert!(
|
||||
!result.accepted,
|
||||
"stranger wrote to a private system channel: {}",
|
||||
result.message
|
||||
let event_id = create.id;
|
||||
match ingest_event(&state, &tenant, create, http_auth(&creator)).await {
|
||||
Err(IngestError::Rejected(message)) => assert_eq!(
|
||||
message,
|
||||
"invalid: create-group must include an h tag with the new channel id"
|
||||
),
|
||||
Err(error) => assert!(format!("{error:?}").contains("restricted"), "{error:?}"),
|
||||
Err(other) => panic!("expected a refusal, got {other:?}"),
|
||||
Ok(result) => panic!("expected a refusal, got {:?}", result.message),
|
||||
}
|
||||
|
||||
let channels: i64 =
|
||||
sqlx::query_scalar("SELECT count(*) FROM channels WHERE community_id = $1 AND name = $2")
|
||||
.bind(*tenant.community().as_uuid())
|
||||
.bind(&name)
|
||||
.fetch_one(state.db.pool())
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(channels, 0, "no channel is created");
|
||||
assert!(
|
||||
state
|
||||
.db
|
||||
.get_event_by_id(tenant.community(), event_id.as_bytes())
|
||||
.await
|
||||
.unwrap()
|
||||
.is_none(),
|
||||
"the refused create is not stored"
|
||||
);
|
||||
}
|
||||
|
||||
/// A removed creator's kind:9021 that passes the open-channel check while the
|
||||
@@ -1082,8 +1052,6 @@ async fn removed_creator_join_racing_a_private_switch_is_refused() {
|
||||
#[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();
|
||||
@@ -1168,3 +1136,86 @@ async fn a_taken_open_name_is_refused_as_a_client_error() {
|
||||
.expect("look up the 9002");
|
||||
assert!(!stored, "the refused 9002 is not stored");
|
||||
}
|
||||
|
||||
/// A relay operator who is not a channel member may rename the channel, and
|
||||
/// the relay publishes the new name in kind:39000. The operator may change
|
||||
/// nothing else, and a relay moderator may not rename it.
|
||||
/// Mutation: drop the operator branch in `validate_admin_event`, or drop its
|
||||
/// `is_rename_only` or operator-role condition → RED.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires Postgres"]
|
||||
async fn a_relay_operator_may_only_rename_a_channel() {
|
||||
let state = state().await;
|
||||
let tenant = community(&state, "group-state-operator-rename").await;
|
||||
let owner = Keys::generate();
|
||||
let channel = create_channel(&state, &tenant, &owner, &[]).await;
|
||||
let operator = Keys::generate();
|
||||
let moderator = Keys::generate();
|
||||
for (keys, role) in [(&operator, "operator"), (&moderator, "moderator")] {
|
||||
state
|
||||
.db
|
||||
.upsert_relay_operator(
|
||||
&keys.public_key().to_bytes(),
|
||||
role,
|
||||
&owner.public_key().to_bytes(),
|
||||
true,
|
||||
)
|
||||
.await
|
||||
.expect("grant relay role");
|
||||
}
|
||||
let edit = |keys: &Keys, tags: &[&[&str]]| {
|
||||
let mut all = vec![Tag::parse(["h", &channel.to_string()]).unwrap()];
|
||||
all.extend(
|
||||
tags.iter()
|
||||
.map(|tag| Tag::parse(tag.iter().copied()).unwrap()),
|
||||
);
|
||||
EventBuilder::new(Kind::Custom(9002), "")
|
||||
.tags(all)
|
||||
.sign_with_keys(keys)
|
||||
.expect("sign 9002")
|
||||
};
|
||||
let community_id = tenant.community();
|
||||
let name = |state: Arc<AppState>| async move {
|
||||
state
|
||||
.db
|
||||
.get_channel(community_id, channel)
|
||||
.await
|
||||
.expect("channel")
|
||||
.name
|
||||
};
|
||||
let before = name(state.clone()).await;
|
||||
|
||||
for (keys, tags) in [
|
||||
(&moderator, &[&["name", "by-moderator"][..]][..]),
|
||||
(
|
||||
&operator,
|
||||
&[&["name", "by-operator"][..], &["about", "changed"][..]][..],
|
||||
),
|
||||
] {
|
||||
match ingest_event(&state, &tenant, edit(keys, tags), http_auth(keys)).await {
|
||||
Err(IngestError::Rejected(message)) => {
|
||||
assert!(message.contains("not authorized"), "{message}")
|
||||
}
|
||||
Err(other) => panic!("expected a refusal, got {other:?}"),
|
||||
Ok(result) => panic!("expected a refusal, got {:?}", result.message),
|
||||
}
|
||||
assert_eq!(name(state.clone()).await, before);
|
||||
}
|
||||
|
||||
let renamed = ingest_event(
|
||||
&state,
|
||||
&tenant,
|
||||
edit(&operator, &[&["name", "renamed-by-operator"]]),
|
||||
http_auth(&operator),
|
||||
)
|
||||
.await
|
||||
.unwrap_or_else(|error| panic!("operator rename: {error:?}"));
|
||||
assert!(renamed.accepted, "{}", renamed.message);
|
||||
assert_eq!(name(state.clone()).await, "renamed-by-operator");
|
||||
let tags = group_state_tags(&state, &tenant, channel, 39000).await;
|
||||
assert!(
|
||||
tags.iter()
|
||||
.any(|tag| tag.len() >= 2 && tag[0] == "name" && tag[1] == "renamed-by-operator"),
|
||||
"{tags:?}"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -3212,8 +3212,9 @@ async fn ingest_event_inner(
|
||||
.map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?;
|
||||
}
|
||||
|
||||
// A kind:9007 with an `h` tag creates its channel when it is stored, in the
|
||||
// same transaction (see the storage dispatch below).
|
||||
// A kind:9007 must carry the new channel's id in its `h` tag. It creates
|
||||
// its channel when it is stored, in the same transaction (see the storage
|
||||
// dispatch below).
|
||||
let mut channel_create: Option<ChannelCreateRequest> = None;
|
||||
|
||||
if kind_u32 == KIND_NIP29_CREATE_GROUP {
|
||||
@@ -3290,6 +3291,12 @@ async fn ingest_event_inner(
|
||||
unlisted,
|
||||
members,
|
||||
});
|
||||
} else {
|
||||
// The channel is created in the same commit as its event, so a
|
||||
// refused create (a taken name, for example) reaches the client.
|
||||
return Err(IngestError::Rejected(
|
||||
"invalid: create-group must include an h tag with the new channel id".into(),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -803,6 +803,16 @@ pub async fn validate_admin_event(
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
// A relay operator may rename any channel, so that
|
||||
// duplicate names can be repaired through this path
|
||||
// (scripts/maintenance/rename_duplicate_listed_channel_names.sh).
|
||||
// Every other privileged change still needs a channel
|
||||
// owner or admin.
|
||||
if is_rename_only(event)
|
||||
&& actor_is_relay_operator(state, event.pubkey.to_bytes()).await?
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
Err(anyhow::anyhow!(
|
||||
"actor not authorized for name/about/archived/visibility/ttl/label/join_role/unlisted changes"
|
||||
))
|
||||
@@ -1233,6 +1243,39 @@ async fn emit_typed_membership_notification(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// True when the only metadata a kind:9002 changes is the channel name.
|
||||
fn is_rename_only(event: &Event) -> bool {
|
||||
const METADATA_TAGS: &[&str] = &[
|
||||
"name",
|
||||
"about",
|
||||
"archived",
|
||||
"topic",
|
||||
"purpose",
|
||||
"visibility",
|
||||
"ttl",
|
||||
"t",
|
||||
"join_role",
|
||||
"unlisted",
|
||||
];
|
||||
let mut changes = event
|
||||
.tags
|
||||
.iter()
|
||||
.map(|t| t.kind().to_string())
|
||||
.filter(|kind| METADATA_TAGS.contains(&kind.as_str()));
|
||||
changes.all(|kind| kind == "name") && event.tags.iter().any(|t| t.kind().to_string() == "name")
|
||||
}
|
||||
|
||||
/// True when `pubkey` holds the relay operator role. A roster lookup failure
|
||||
/// is an error, never "not an operator".
|
||||
async fn actor_is_relay_operator(
|
||||
state: &crate::state::AppState,
|
||||
pubkey: [u8; 32],
|
||||
) -> anyhow::Result<bool> {
|
||||
crate::api::admin::is_relay_operator(state, pubkey)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("relay staff lookup failed: {}", e.message))
|
||||
}
|
||||
|
||||
/// Sign, store (replacing previous), and fan-out a single addressable discovery event.
|
||||
async fn emit_addressable_discovery_event(
|
||||
tenant: &TenantContext,
|
||||
@@ -2430,17 +2473,14 @@ async fn handle_create_group(
|
||||
.unwrap_or_default();
|
||||
let unlisted = create_unlisted_from_tags(event)?;
|
||||
|
||||
// If the event has an h-tag UUID, ingest_event() already created the channel
|
||||
// via create_channel_with_id(). Fetch it rather than creating a duplicate.
|
||||
// If no h-tag, fall back to the original auto-UUID creation path.
|
||||
// Ingest already created the channel named by the h tag, in the same
|
||||
// commit as this event. Fetch it rather than creating a duplicate.
|
||||
//
|
||||
// Double-count analysis (C5): the counter increments below do NOT
|
||||
// double-count vs. ingest.rs. For the h-tag path, ingest increments on
|
||||
// was_created=true and this handler only reaches create_channel() on a DB
|
||||
// lookup Err — an error recovery path where ingest's channel is
|
||||
// inaccessible, so the counter correctly records a new creation. For the
|
||||
// no-h-tag path, ingest never creates the channel, so this is the sole
|
||||
// increment.
|
||||
// Double-count analysis (C5): the counter increment below does NOT
|
||||
// double-count vs. ingest.rs. Ingest increments on was_created=true and
|
||||
// this handler only reaches create_channel() on a DB lookup Err — an
|
||||
// error recovery path where ingest's channel is inaccessible, so the
|
||||
// counter correctly records a new creation.
|
||||
// `members_added`: ingest's create also added the event's `p` tags.
|
||||
let (channel, members_added) = if let Some(client_uuid) = extract_h_tag_channel(event) {
|
||||
match state
|
||||
@@ -2476,27 +2516,8 @@ async fn handle_create_group(
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let ch = state
|
||||
.db
|
||||
.create_labeled_channel(
|
||||
tenant.community(),
|
||||
&name,
|
||||
channel_type,
|
||||
visibility,
|
||||
description.as_deref(),
|
||||
&actor_bytes,
|
||||
ttl_seconds,
|
||||
&labels,
|
||||
unlisted,
|
||||
)
|
||||
.await?;
|
||||
metrics::counter!(
|
||||
"buzz_channels_created_total",
|
||||
"community" => tenant.host().to_owned(),
|
||||
"type" => channel_type.to_string()
|
||||
)
|
||||
.increment(1);
|
||||
(ch, false)
|
||||
// Ingest refuses a kind:9007 without an `h` tag before storing it.
|
||||
anyhow::bail!("missing or invalid h tag");
|
||||
};
|
||||
|
||||
// Creator becomes owner — evict any stale negative membership lookup.
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
-- OUT-OF-BAND MAINTENANCE, read-only. rename_duplicate_listed_channel_names.sh
|
||||
-- runs this query and sends each rename it lists to the relay.
|
||||
--
|
||||
-- Lists the renames that give every listed channel in one community a name
|
||||
-- that no other listed channel there 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).
|
||||
--
|
||||
-- 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 gets "<name>-<first 8 characters of its
|
||||
-- id>", or "<name>-<id>" when another listed channel already has that name.
|
||||
-- It lists nothing for a community that is not active.
|
||||
--
|
||||
-- psql -X -v ON_ERROR_STOP=1 -v host=<community host> -f scripts/maintenance/plan_duplicate_listed_channel_renames.sql
|
||||
|
||||
WITH listed AS (
|
||||
SELECT ch.id, ch.name, ch.archived_at, ch.created_at
|
||||
FROM channels ch
|
||||
JOIN communities co ON co.id = ch.community_id
|
||||
WHERE co.host = :'host'
|
||||
AND co.deletion_state = 'active' AND co.deleted_at IS NULL
|
||||
AND ch.visibility = 'open' AND NOT ch.unlisted
|
||||
AND ch.channel_type IN ('stream', 'forum') AND ch.deleted_at IS NULL
|
||||
),
|
||||
duplicates AS (
|
||||
SELECT id, name,
|
||||
left(name, 246) || '-' || left(id::text, 8) AS short_name,
|
||||
left(name, 218) || '-' || id::text AS long_name
|
||||
FROM (
|
||||
SELECT id, name, row_number() OVER (
|
||||
PARTITION BY lower(name)
|
||||
ORDER BY archived_at IS NOT NULL, created_at, id
|
||||
) AS place
|
||||
FROM listed
|
||||
) ranked
|
||||
WHERE place > 1
|
||||
)
|
||||
SELECT d.id AS channel_id, d.name,
|
||||
CASE WHEN EXISTS (
|
||||
SELECT 1 FROM listed other
|
||||
WHERE other.id <> d.id AND lower(other.name) = lower(d.short_name)
|
||||
) THEN d.long_name ELSE d.short_name END AS new_name
|
||||
FROM duplicates d
|
||||
ORDER BY d.id;
|
||||
87
scripts/maintenance/rename_duplicate_listed_channel_names.sh
Executable file
87
scripts/maintenance/rename_duplicate_listed_channel_names.sh
Executable file
@@ -0,0 +1,87 @@
|
||||
#!/usr/bin/env bash
|
||||
# OUT-OF-BAND MAINTENANCE: renames listed channels that share a name in one
|
||||
# community, so the contract migration's unique index can build.
|
||||
#
|
||||
# It reads the renames from plan_duplicate_listed_channel_renames.sql and sends
|
||||
# each one to the relay as a kind:9002 rename, signed by a relay operator. The
|
||||
# relay applies a rename the way it applies any other: it refuses a name that
|
||||
# another listed channel has, and publishes the new kind:39000, so clients
|
||||
# show the new name. Relays keep serving while it runs.
|
||||
#
|
||||
# Run it only after every serving relay enforces the name rule and accepts an
|
||||
# operator rename. An older relay refuses the rename and can create new
|
||||
# duplicates.
|
||||
#
|
||||
# A rename that fails is reported and skipped: the relay refused it (for
|
||||
# example, because a channel took the new name meanwhile), or the CLI could
|
||||
# not reach the relay. At the end the script reads the plan again and fails
|
||||
# if any rename remains; run it again.
|
||||
#
|
||||
# Usage:
|
||||
# DATABASE_URL=postgres://... BUZZ_PRIVATE_KEY=<operator key> \
|
||||
# scripts/maintenance/rename_duplicate_listed_channel_names.sh [--dry-run] <community host>
|
||||
#
|
||||
# BUZZ_PRIVATE_KEY must belong to a relay operator (RELAY_OPERATOR_PUBKEYS, or
|
||||
# an `operator` row in relay_operators), not a moderator. BUZZ_RELAY_URL
|
||||
# defaults to wss://<community host>. BUZZ_CLI defaults to `buzz`.
|
||||
set -euo pipefail
|
||||
|
||||
usage() {
|
||||
echo "usage: $0 [--dry-run] <community host>" >&2
|
||||
exit 2
|
||||
}
|
||||
|
||||
dry_run=false
|
||||
if [[ ${1:-} == --dry-run ]]; then
|
||||
dry_run=true
|
||||
shift
|
||||
fi
|
||||
[[ $# -eq 1 ]] || usage
|
||||
host=$1
|
||||
: "${DATABASE_URL:?set DATABASE_URL}"
|
||||
if ! $dry_run; then
|
||||
: "${BUZZ_PRIVATE_KEY:?set BUZZ_PRIVATE_KEY to a relay operator key}"
|
||||
fi
|
||||
relay_url=${BUZZ_RELAY_URL:-wss://$host}
|
||||
buzz=${BUZZ_CLI:-buzz}
|
||||
plan_sql="$(cd "$(dirname "$0")" && pwd)/plan_duplicate_listed_channel_renames.sql"
|
||||
|
||||
plan=$(mktemp)
|
||||
trap 'rm -f "$plan"' EXIT
|
||||
|
||||
# Every field ends with a NUL byte. Postgres text cannot hold one, so any
|
||||
# name, with any control character in it, stays one field.
|
||||
read_plan() {
|
||||
psql "$DATABASE_URL" -X -q -v ON_ERROR_STOP=1 -v host="$host" \
|
||||
-A -t -z -0 -f "$plan_sql" >"$plan"
|
||||
}
|
||||
|
||||
read_plan
|
||||
# The plan is the loop's stdin, so the CLI reads /dev/null instead.
|
||||
renamed=0
|
||||
failed=0
|
||||
while IFS= read -r -d '' channel && IFS= read -r -d '' name &&
|
||||
IFS= read -r -d '' new_name; do
|
||||
if $dry_run; then
|
||||
printf 'would rename %s from %q to %q\n' "$channel" "$name" "$new_name"
|
||||
elif BUZZ_RELAY_URL=$relay_url "$buzz" channels update \
|
||||
--channel "$channel" --name "$new_name" </dev/null >/dev/null; then
|
||||
printf 'renamed %s from %q to %q\n' "$channel" "$name" "$new_name"
|
||||
renamed=$((renamed + 1))
|
||||
else
|
||||
printf 'could not rename %s from %q to %q\n' \
|
||||
"$channel" "$name" "$new_name" >&2
|
||||
failed=$((failed + 1))
|
||||
fi
|
||||
done <"$plan"
|
||||
|
||||
$dry_run && exit 0
|
||||
|
||||
read_plan
|
||||
# Three fields, so three NUL bytes, for each rename.
|
||||
remaining=$(($(tr -cd '\000' <"$plan" | wc -c) / 3))
|
||||
echo "renamed $renamed, failed $failed, remaining $remaining"
|
||||
if [[ $remaining -ne 0 ]]; then
|
||||
echo "duplicate listed channel names remain in $host; run the script again" >&2
|
||||
exit 1
|
||||
fi
|
||||
@@ -1,161 +0,0 @@
|
||||
-- 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 "<name>-<first 8 characters of its
|
||||
-- id>", or "<name>-<id>" 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 <id> --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;
|
||||
Reference in New Issue
Block a user