From 055ee2dd58d5706b8a69297f21d23111b8b388b9 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 21 Oct 2026 09:38:45 -0400 Subject: [PATCH 1/7] fix(relay): store and apply kind:9002 channel edits in one transaction The relay stored a kind:9002 event, then applied each changed setting with its own database write. If a write failed, the relay logged it and still told the client the change succeeded. Settings after the failed one were skipped, and a resend of the same event returned duplicate: without trying again. Now one transaction stores the event, applies every change and stores the kind:40099 system messages. If any write fails, nothing is stored and the client gets an error, so it can resend the same event. Cache invalidation, subscription eviction, fan-out, the unarchive resubscribe notices and the kind:39000 republish run only after the commit. Restoring a channel that is not archived is now refused before storage, instead of storing the event and then failing the write. On top of #8143, every channel settings write now takes its locks through one function, lock_channel_for_settings_write: the membership lock for a visibility or join_role change, then the TTL lock (exclusive for a TTL change), then the channel row. The row-writing helpers take the lock it returns, so a new settings writer cannot skip a lock or change the order. kind:9002 join_role moves into the same transaction, with its join_role_changed system message. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Signed-off-by: Logan Johnson --- crates/buzz-db/src/store/channel.rs | 458 +++++++++++++---- .../handlers/edit_metadata_postgres_tests.rs | 209 ++++++++ crates/buzz-relay/src/handlers/ingest.rs | 11 +- crates/buzz-relay/src/handlers/mod.rs | 2 + .../buzz-relay/src/handlers/side_effects.rs | 486 ++++++++---------- 5 files changed, 794 insertions(+), 372 deletions(-) create mode 100644 crates/buzz-relay/src/handlers/edit_metadata_postgres_tests.rs diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index a14cb9aa026..5be704252d3 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -768,6 +768,24 @@ pub struct ChannelUpdate { pub join_role: Option, } +impl ChannelUpdate { + /// True when no field is set. + pub fn is_empty(&self) -> bool { + self.name.is_none() + && self.description.is_none() + && self.visibility.is_none() + && self.ttl_seconds.is_none() + && self.labels.is_none() + && self.join_role.is_none() + } + + /// True when the update changes a setting that decides admission: + /// visibility or `join_role`. + pub fn changes_admission(&self) -> bool { + self.visibility.is_some() || self.join_role.is_some() + } +} + /// Updates channel metadata dynamically. /// /// At least one field must be provided; returns `InvalidData` otherwise. @@ -776,20 +794,158 @@ pub async fn update_channel( pool: &PgPool, community_id: CommunityId, channel_id: Uuid, - mut updates: ChannelUpdate, + updates: ChannelUpdate, ) -> Result { - if updates.name.is_none() - && updates.description.is_none() - && updates.visibility.is_none() - && updates.ttl_seconds.is_none() - && updates.labels.is_none() - && updates.join_role.is_none() - { + if updates.is_empty() { return Err(DbError::InvalidData( "at least one field must be provided for update".to_string(), )); } + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + let locked = lock_channel_for_settings_write(&mut tx, channel_id, &updates).await?; + update_channel_on(&mut tx, &locked, updates).await?; + tx.commit().await?; + + get_channel_with_operation( + pool, + community_id, + channel_id, + crate::observability::WriterOperation::EventWrite, + ) + .await +} + +#[derive(Clone, Copy)] +enum TtlLock { + Shared, + Exclusive, +} + +/// Takes the per-channel TTL advisory lock (see [`channel_ttl_lock_key`]) on `tx`. +async fn lock_channel_ttl( + tx: &mut crate::AdmittedTx, + channel_id: Uuid, + mode: TtlLock, +) -> Result<()> { + let sql = match mode { + TtlLock::Shared => "SELECT pg_advisory_xact_lock_shared(hashtextextended($1, 0))", + TtlLock::Exclusive => "SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", + }; + sqlx::query(sql) + .bind(channel_ttl_lock_key(tx.community(), channel_id)) + .execute(tx.conn()) + .await?; + Ok(()) +} + +/// Proof that a transaction holds the locks a channel settings write needs. +/// Only [`lock_channel_for_settings_write`] makes one, and every function +/// that writes channel settings takes one. So a new settings writer cannot +/// skip a lock or take the locks in a different order. +pub(crate) struct ChannelSettingsLock { + channel_id: Uuid, + admission: bool, + ttl: TtlLock, + previous_visibility: String, +} + +impl ChannelSettingsLock { + /// The channel's visibility when the lock was taken (`"open"` or `"private"`). + pub(crate) fn previous_visibility(&self) -> &str { + &self.previous_visibility + } + + /// Refuse `updates` if this lock does not cover them. This catches a + /// caller that took the lock for one set of changes and then writes + /// another set. + fn check_covers(&self, channel_id: Uuid, updates: &ChannelUpdate) -> Result<()> { + let covered = self.channel_id == channel_id + && (self.admission || !updates.changes_admission()) + && (matches!(self.ttl, TtlLock::Exclusive) || updates.ttl_seconds.is_none()); + if covered { + Ok(()) + } else { + Err(DbError::InvalidData( + "channel settings write is not covered by its lock".into(), + )) + } + } +} + +/// Takes, on `tx`, every lock a write to `channel_id`'s settings needs, in +/// the one order that all channel writers use: +/// +/// 1. the channel's membership lock, when `updates` changes visibility or +/// `join_role`. Those two settings decide admission, and `put_member` and +/// a huddle join decide admission under this lock. So an admission either +/// sees the old setting and commits first, or sees the new one. +/// 2. the channel's TTL lock: EXCLUSIVE for a TTL change, SHARED otherwise. +/// A TTL change can switch the event-commit fast path +/// (`event_follow_up::refresh_channel_ttls`, which reads `ttl_seconds` +/// under the SHARED lock). With the EXCLUSIVE lock, a concurrent event +/// either sees the new TTL or commits before this change. +/// 3. the channel row, `FOR UPDATE`. +/// +/// Roster publication holds the membership lock and then, at commit, takes +/// the TTL lock and updates the row. A huddle join takes the membership +/// lock, then the TTL lock, then the row. A writer that took these locks in +/// any other order could wait for one of them while it waits for this one. +/// +/// Returns `ChannelNotFound` when the channel is missing or deleted. +pub(crate) async fn lock_channel_for_settings_write( + tx: &mut crate::AdmittedTx, + channel_id: Uuid, + updates: &ChannelUpdate, +) -> Result { + let community_id = tx.community(); + let admission = updates.changes_admission(); + if admission { + crate::channel_members::acquire_channel_membership_lock( + tx.conn(), + community_id, + channel_id, + ) + .await?; + } + let ttl = if updates.ttl_seconds.is_some() { + TtlLock::Exclusive + } else { + TtlLock::Shared + }; + lock_channel_ttl(tx, channel_id, ttl).await?; + let previous_visibility: String = sqlx::query_scalar( + "SELECT visibility::text FROM channels \ + WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL FOR UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await? + .ok_or(DbError::ChannelNotFound(channel_id))?; + Ok(ChannelSettingsLock { + channel_id, + admission, + ttl, + previous_visibility, + }) +} + +/// Writes the set fields of `updates` to the locked channel row on `tx`. +/// The caller owns the commit. +async fn update_channel_on( + tx: &mut crate::AdmittedTx, + locked: &ChannelSettingsLock, + mut updates: ChannelUpdate, +) -> Result<()> { + let channel_id = locked.channel_id; + locked.check_covers(channel_id, &updates)?; + let community_id = tx.community(); if let Some(name) = updates.name.as_mut() { *name = buzz_core::channel::canonical_channel_name(name).to_owned(); if name.is_empty() { @@ -861,57 +1017,11 @@ pub async fn update_channel( q = q.bind(community_id.as_uuid()); q = q.bind(channel_id); - // Every writer that takes more than one of a channel's locks takes them - // in this order: membership lock, then TTL lock, then the channel row. - // Roster publication holds the membership lock and then refreshes the - // TTL (TTL lock, row update) at commit; a huddle join takes the - // membership lock before it locks the row. Taking the row first here - // would let either of them wait on this transaction while it waits on - // them. - let mut tx = crate::begin_community_write_transaction( - pool, - community_id, - crate::observability::WriterOperation::EventWrite, - ) - .await?; - // Visibility and `join_role` decide admission, which `put_member` and a - // huddle join decide under the channel's membership lock. Take that lock - // too, so an admission either sees the old setting and commits first, or - // sees the new one. - if updates.visibility.is_some() || updates.join_role.is_some() { - crate::channel_members::acquire_channel_membership_lock( - tx.conn(), - community_id, - channel_id, - ) - .await?; - } - // T1a repair: a TTL change can flip this channel's event-commit fast - // path (`event_follow_up::refresh_channel_ttls` reads ttl_seconds under - // a SHARED per-channel advisory lock). Take the same key EXCLUSIVE - // before the UPDATE so a concurrent event either sees the committed TTL - // or strictly precedes this transition — whose own deadline reset is - // then the latest word. Non-TTL updates don't touch the fast path and - // skip the lock. - if updates.ttl_seconds.is_some() { - sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") - .bind(channel_ttl_lock_key(community_id, channel_id)) - .execute(tx.conn()) - .await?; - } let result = q.execute(tx.conn()).await?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } - tx.commit().await?; - - get_channel_with_operation( - pool, - community_id, - channel_id, - crate::observability::WriterOperation::EventWrite, - ) - .await + Ok(()) } /// Sets the topic for a channel, recording who set it and when. @@ -928,20 +1038,33 @@ pub async fn set_topic( crate::observability::WriterOperation::EventWrite, ) .await?; + let locked = + lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?; + set_topic_on(&mut tx, &locked, topic, set_by).await?; + tx.commit().await?; + Ok(()) +} + +async fn set_topic_on( + tx: &mut crate::AdmittedTx, + locked: &ChannelSettingsLock, + topic: &str, + set_by: &[u8], +) -> Result<()> { + let channel_id = locked.channel_id; let result = sqlx::query( "UPDATE channels SET topic = $1, topic_set_by = $2, topic_set_at = NOW() \ WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL", ) .bind(topic) .bind(set_by) - .bind(community_id.as_uuid()) + .bind(tx.community().as_uuid()) .bind(channel_id) .execute(tx.conn()) .await?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } - tx.commit().await?; Ok(()) } @@ -959,20 +1082,33 @@ pub async fn set_purpose( crate::observability::WriterOperation::EventWrite, ) .await?; + let locked = + lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?; + set_purpose_on(&mut tx, &locked, purpose, set_by).await?; + tx.commit().await?; + Ok(()) +} + +async fn set_purpose_on( + tx: &mut crate::AdmittedTx, + locked: &ChannelSettingsLock, + purpose: &str, + set_by: &[u8], +) -> Result<()> { + let channel_id = locked.channel_id; let result = sqlx::query( "UPDATE channels SET purpose = $1, purpose_set_by = $2, purpose_set_at = NOW() \ WHERE community_id = $3 AND id = $4 AND deleted_at IS NULL", ) .bind(purpose) .bind(set_by) - .bind(community_id.as_uuid()) + .bind(tx.community().as_uuid()) .bind(channel_id) .execute(tx.conn()) .await?; if result.rows_affected() == 0 { return Err(DbError::ChannelNotFound(channel_id)); } - tx.commit().await?; Ok(()) } @@ -991,36 +1127,9 @@ pub async fn archive_channel( crate::observability::WriterOperation::EventWrite, ) .await?; - // First check: does the channel exist and what is its state? - let row = sqlx::query( - "SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .fetch_optional(tx.conn()) - .await?; - - match row { - None => return Err(DbError::ChannelNotFound(channel_id)), - Some(r) => { - let archived_at: Option> = r.try_get("archived_at")?; - if archived_at.is_some() { - return Err(DbError::AccessDenied( - "channel is already archived".to_string(), - )); - } - } - } - - sqlx::query( - "UPDATE channels SET archived_at = NOW() \ - WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NULL", - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .execute(tx.conn()) - .await?; - + let locked = + lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?; + set_archived_on(&mut tx, &locked, true).await?; tx.commit().await?; Ok(()) } @@ -1040,40 +1149,159 @@ pub async fn unarchive_channel( crate::observability::WriterOperation::EventWrite, ) .await?; - // First check: does the channel exist and what is its state? + let locked = + lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?; + set_archived_on(&mut tx, &locked, false).await?; + tx.commit().await?; + Ok(()) +} + +/// Archives (`archive = true`) or restores the channel on `tx`. Refuses with +/// `AccessDenied` when the channel is already in that state. Restoring an +/// ephemeral channel restarts its TTL deadline. +async fn set_archived_on( + tx: &mut crate::AdmittedTx, + locked: &ChannelSettingsLock, + archive: bool, +) -> Result<()> { + let channel_id = locked.channel_id; + let community_id = tx.community(); let row = sqlx::query( "SELECT archived_at FROM channels WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .fetch_optional(tx.conn()) - .await?; + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await?; match row { None => return Err(DbError::ChannelNotFound(channel_id)), Some(r) => { let archived_at: Option> = r.try_get("archived_at")?; - if archived_at.is_none() { + if archive && archived_at.is_some() { + return Err(DbError::AccessDenied( + "channel is already archived".to_string(), + )); + } + if !archive && archived_at.is_none() { return Err(DbError::AccessDenied("channel is not archived".to_string())); } } } - sqlx::query( + let sql = if archive { + "UPDATE channels SET archived_at = NOW() \ + WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NULL" + } else { "UPDATE channels SET archived_at = NULL, \ ttl_deadline = CASE \ WHEN ttl_seconds IS NOT NULL THEN NOW() + (ttl_seconds || ' seconds')::interval \ ELSE ttl_deadline \ END \ - WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NOT NULL", + WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND archived_at IS NOT NULL" + }; + sqlx::query(sql) + .bind(community_id.as_uuid()) + .bind(channel_id) + .execute(tx.conn()) + .await?; + Ok(()) +} + +/// The channel changes named by one kind:9002 event. `None` leaves a value +/// unchanged. +#[derive(Default)] +pub struct ChannelMetadataEdit { + /// Name, description, visibility, TTL and labels. + pub columns: ChannelUpdate, + /// New topic, recorded as set by the event's author. + pub topic: Option, + /// New purpose, recorded as set by the event's author. + pub purpose: Option, + /// `Some(true)` archives the channel; `Some(false)` restores it. + pub archived: Option, +} + +/// The result of [`Db::apply_channel_metadata_event`]. +pub struct ChannelMetadataWrite { + /// The stored kind:9002 event. + pub stored: StoredEvent, + /// `false` when the event was already stored. Nothing was changed then: + /// its changes committed with it the first time. + pub inserted: bool, + /// The channel's visibility before this event (`"open"` or `"private"`). + pub previous_visibility: String, +} + +pub(crate) async fn apply_channel_metadata_event( + pool: &PgPool, + push_enqueue: Option, + community_id: CommunityId, + channel_id: Uuid, + event: &nostr::Event, + edit: ChannelMetadataEdit, + system_messages: &[nostr::Event], +) -> Result { + let mut tx = crate::begin_community_event_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .execute(tx.conn()) .await?; + tx.set_push_enqueue(push_enqueue); + // The row lock also orders two copies of the same event: the second + // waits, then sees the first stored. + let locked = lock_channel_for_settings_write(&mut tx, channel_id, &edit.columns).await?; + let previous_visibility = locked.previous_visibility().to_owned(); + + // The same writes as `Db::insert_event_with_thread_metadata`. + let (stored, inserted) = + crate::event::insert_event_with_thread_metadata_tx(&mut tx, event, Some(channel_id), None) + .await?; + if !inserted { + tx.rollback().await?; + return Ok(ChannelMetadataWrite { + stored, + inserted, + previous_visibility, + }); + } + crate::insert_mentions_in_transaction(&mut tx, event, Some(channel_id)).await?; + + let ChannelMetadataEdit { + columns, + topic, + purpose, + archived, + } = edit; + let actor = event.pubkey.to_bytes(); + if !columns.is_empty() { + update_channel_on(&mut tx, &locked, columns).await?; + } + if let Some(topic) = topic { + set_topic_on(&mut tx, &locked, &topic, &actor).await?; + } + if let Some(purpose) = purpose { + set_purpose_on(&mut tx, &locked, &purpose, &actor).await?; + } + if let Some(archive) = archived { + set_archived_on(&mut tx, &locked, archive).await?; + } + // The same writes as `Db::insert_event`. + for message in system_messages { + let (_, message_inserted) = + crate::event::insert_event_in_transaction(&mut tx, message, Some(channel_id)).await?; + if message_inserted { + crate::insert_mentions_in_transaction(&mut tx, message, Some(channel_id)).await?; + } + } tx.commit().await?; - Ok(()) + Ok(ChannelMetadataWrite { + stored, + inserted, + previous_visibility, + }) } /// Soft-delete a channel by setting `deleted_at = NOW()`. @@ -1248,6 +1476,32 @@ impl Db { .await } + /// Stores the kind:9002 `event` for `channel_id`, applies `edit`, and + /// stores the relay-signed `system_messages` that announce it, all in one + /// transaction. If any write fails, nothing is stored, so a client can + /// resend the same event. A resend of a stored event changes nothing and + /// returns `inserted: false`. + #[datastore_span(name = "apply_channel_metadata_event", system = "postgresql")] + pub async fn apply_channel_metadata_event( + &self, + community_id: CommunityId, + channel_id: Uuid, + event: &nostr::Event, + edit: ChannelMetadataEdit, + system_messages: &[nostr::Event], + ) -> Result { + apply_channel_metadata_event( + &self.pool, + self.push_enqueue.clone(), + community_id, + channel_id, + event, + edit, + system_messages, + ) + .await + } + /// Creates a channel with a relay-chosen UUID and an initial label set. #[allow(clippy::too_many_arguments)] #[datastore_span(name = "create_labeled_channel", system = "postgresql")] @@ -1420,7 +1674,7 @@ impl Db { /// The per-channel TTL advisory lock key. Event commits take it SHARED /// (`event_follow_up::refresh_channel_ttls`); TTL transitions in -/// [`update_channel`] take it EXCLUSIVE. Both sides build it here, so the keys +/// [`update_channel`] and [`apply_channel_metadata_event`] take it EXCLUSIVE. Both sides build it here, so the keys /// cannot drift. pub(crate) fn channel_ttl_lock_key(community: CommunityId, channel: Uuid) -> String { format!("buzz_channel_ttl:{}:{channel}", community.as_uuid()) diff --git a/crates/buzz-relay/src/handlers/edit_metadata_postgres_tests.rs b/crates/buzz-relay/src/handlers/edit_metadata_postgres_tests.rs new file mode 100644 index 00000000000..1286237d9bf --- /dev/null +++ b/crates/buzz-relay/src/handlers/edit_metadata_postgres_tests.rs @@ -0,0 +1,209 @@ +//! Postgres-backed tests for kind:9002: the event and the channel changes it +//! names commit together, or not at all. + +use buzz_db::event::EventQuery; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use uuid::Uuid; + +use crate::handlers::group_state_postgres_tests::{ + community, create_channel, edit_channel, http_auth, state, +}; +use crate::handlers::ingest::ingest_event; + +/// Count stored events of `kind` in `channel_id`. +async fn count_kind( + state: &crate::state::AppState, + tenant: &buzz_core::TenantContext, + channel_id: Uuid, + kind: i32, +) -> usize { + state + .db + .query_events(&EventQuery { + kinds: Some(vec![kind]), + channel_id: Some(channel_id), + ..EventQuery::for_community(tenant.community()) + }) + .await + .expect("query events") + .len() +} + +/// A change that fails after authorization stores nothing and changes +/// nothing. A name longer than the 255-character column passes the relay's +/// checks but fails the UPDATE. Before, the relay stored the event, applied +/// `about`, skipped `topic`, and reported success; a resend then returned +/// `duplicate:`. Mutation: store the event in its own transaction → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn failed_edit_stores_nothing() { + let state = state().await; + let tenant = community(&state, "edit-atomic-fail").await; + let owner = Keys::generate(); + let channel_id = create_channel(&state, &tenant, &owner, &[]).await; + let before = state + .db + .get_channel(tenant.community(), channel_id) + .await + .expect("channel"); + let notices_before = count_kind(&state, &tenant, channel_id, 40099).await; + + let long_name = "n".repeat(256); + let event = EventBuilder::new(Kind::Custom(9002), "") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["about", "changed"]).unwrap(), + Tag::parse(["name", &long_name]).unwrap(), + Tag::parse(["topic", "changed"]).unwrap(), + ]) + .sign_with_keys(&owner) + .expect("sign 9002"); + + for attempt in ["first send", "resend"] { + match ingest_event(&state, &tenant, event.clone(), http_auth(&owner)).await { + Ok(result) => assert!( + !result.accepted, + "{attempt}: a failed edit was reported as accepted: {:?}", + result.message + ), + Err(error) => assert!( + format!("{error:?}").contains("channel update failed"), + "{attempt}: {error:?}" + ), + } + } + + let after = state + .db + .get_channel(tenant.community(), channel_id) + .await + .expect("channel"); + assert_eq!(after.name, before.name); + assert_eq!(after.description, before.description, "about is unchanged"); + assert_eq!(after.topic, before.topic, "topic is unchanged"); + assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 0); + assert_eq!( + count_kind(&state, &tenant, channel_id, 40099).await, + notices_before, + "no system message announces a change that did not happen" + ); +} + +/// A resend of a stored edit is a duplicate and changes nothing: its changes +/// and system messages committed with it the first time. +/// Mutation: apply the edit again on a duplicate → RED (second notice). +#[tokio::test] +#[ignore = "requires Postgres"] +async fn resent_edit_is_a_duplicate() { + let state = state().await; + let tenant = community(&state, "edit-atomic-resend").await; + let owner = Keys::generate(); + let channel_id = create_channel(&state, &tenant, &owner, &[]).await; + let notices_before = count_kind(&state, &tenant, channel_id, 40099).await; + + let event = EventBuilder::new(Kind::Custom(9002), "") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["about", "about-1"]).unwrap(), + Tag::parse(["topic", "topic-1"]).unwrap(), + Tag::parse(["purpose", "purpose-1"]).unwrap(), + Tag::parse(["ttl", "3600"]).unwrap(), + Tag::parse(["t", "label-1"]).unwrap(), + ]) + .sign_with_keys(&owner) + .expect("sign 9002"); + + let first = ingest_event(&state, &tenant, event.clone(), http_auth(&owner)) + .await + .expect("first send"); + assert!(first.accepted, "{:?}", first.message); + assert!(first.message.is_empty(), "{:?}", first.message); + + let channel = state + .db + .get_channel(tenant.community(), channel_id) + .await + .expect("channel"); + assert_eq!(channel.description.as_deref(), Some("about-1")); + assert_eq!(channel.topic.as_deref(), Some("topic-1")); + assert_eq!( + channel.topic_set_by.as_deref(), + Some(owner.public_key().to_bytes().as_slice()) + ); + assert_eq!(channel.purpose.as_deref(), Some("purpose-1")); + assert_eq!(channel.ttl_seconds, Some(3600)); + assert!(channel.ttl_deadline.is_some()); + assert_eq!(channel.labels, vec!["label-1"]); + // topic, purpose and ttl each announce their change. + assert_eq!( + count_kind(&state, &tenant, channel_id, 40099).await, + notices_before + 3 + ); + + let resend = ingest_event(&state, &tenant, event, http_auth(&owner)) + .await + .expect("resend"); + assert!(resend.accepted, "{:?}", resend.message); + assert_eq!(resend.message, "duplicate:"); + assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 1); + assert_eq!( + count_kind(&state, &tenant, channel_id, 40099).await, + notices_before + 3 + ); +} + +/// Restoring a channel that is not archived is refused before storage. +/// Before, the event was stored and reported accepted while the restore +/// failed. Mutation: drop the pre-storage check → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn unarchive_of_an_active_channel_is_refused() { + let state = state().await; + let tenant = community(&state, "edit-atomic-unarchive").await; + let owner = Keys::generate(); + let channel_id = create_channel(&state, &tenant, &owner, &[]).await; + let notices_before = count_kind(&state, &tenant, channel_id, 40099).await; + + let error = edit_channel( + &state, + &tenant, + &owner, + channel_id, + &[&["archived", "false"]], + ) + .await + .expect_err("unarchive of an active channel"); + assert!(error.contains("channel is not archived"), "{error}"); + assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 0); + + // Archive then restore still works, and each step is announced. + edit_channel( + &state, + &tenant, + &owner, + channel_id, + &[&["archived", "true"]], + ) + .await + .expect("archive"); + edit_channel( + &state, + &tenant, + &owner, + channel_id, + &[&["archived", "false"]], + ) + .await + .expect("unarchive"); + let channel = state + .db + .get_channel(tenant.community(), channel_id) + .await + .expect("channel"); + assert!(channel.archived_at.is_none()); + assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 2); + assert_eq!( + count_kind(&state, &tenant, channel_id, 40099).await, + notices_before + 2 + ); +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index ce0c10a7bf5..ce121ab6b93 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -3523,6 +3523,12 @@ async fn ingest_event_inner( crate::handlers::side_effects::persist_workflow_deletion(tenant, &event, state) .await .map_err(|e| IngestError::Internal(format!("error: workflow deletion failed: {e}")))? + } else if kind_u32 == KIND_NIP29_EDIT_METADATA { + // One commit stores the event and applies its channel changes. A + // failure stores nothing, so the client is told and can resend it. + crate::handlers::side_effects::persist_edit_metadata(tenant, &event, state) + .await + .map_err(|e| IngestError::Internal(format!("error: channel update failed: {e}")))? } else if buzz_core::kind::is_replaceable(kind_u32) { // NIP-16 replaceable event — atomic replace with stale-write protection. // channel_id is None for global kinds (0, 1, 3) due to step 5b above. @@ -3614,7 +3620,10 @@ async fn ingest_event_inner( }); } - if !workflow_deletion && crate::handlers::side_effects::is_side_effect_kind(kind_u32) { + // Workflow deletion and kind:9002 applied their changes in the insert's + // own transaction above. + let persisted_atomically = workflow_deletion || kind_u32 == KIND_NIP29_EDIT_METADATA; + if !persisted_atomically && crate::handlers::side_effects::is_side_effect_kind(kind_u32) { if let Err(e) = crate::handlers::side_effects::handle_side_effects(tenant, kind_u32, &event, state) .await diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index dd4db5e9865..222512aa0c5 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -14,6 +14,8 @@ pub mod community_provisioning; pub mod count; #[cfg(test)] mod create_with_members_postgres_tests; +#[cfg(test)] +mod edit_metadata_postgres_tests; /// EVENT handler — WS dispatcher → ingest pipeline → fan-out. pub mod event; #[cfg(test)] diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 2d29c8c124d..6ed29e788c1 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -337,7 +337,8 @@ pub async fn handle_side_effects( 5 => handle_standard_deletion_event(tenant, event, state).await, 9000 => handle_put_user(tenant, event, state).await, 9001 => handle_remove_user(tenant, event, state).await, - 9002 => handle_edit_metadata(tenant, event, state).await, + // kind:9002 belongs to the atomic persistence path in ingest. + 9002 => Err(anyhow::anyhow!("kind:9002 requires atomic persistence")), 9005 => handle_delete_event_side_effect(tenant, event, state).await, 9007 => handle_create_group(tenant, event, state).await, 9008 => handle_delete_group(tenant, event, state).await, @@ -594,6 +595,11 @@ pub async fn validate_admin_event( if channel.archived_at.is_some() && !is_unarchive_request && !is_self_departure(event) { return Err(anyhow::anyhow!("channel is archived")); } + // Restoring an active channel is refused before storage, so that the + // atomic kind:9002 write cannot fail on it after authorization. + if channel.archived_at.is_none() && is_unarchive_request { + return Err(anyhow::anyhow!("channel is not archived")); + } match kind { 9000 => { @@ -974,14 +980,7 @@ pub async fn emit_system_message( content: serde_json::Value, idempotency_ts: chrono::DateTime, ) -> anyhow::Result<()> { - let channel_tag = Tag::parse(["h", &channel_id.to_string()])?; - - let ts = nostr::Timestamp::from(idempotency_ts.timestamp() as u64); - let event = EventBuilder::new(Kind::Custom(40099), content.to_string()) - .tags([channel_tag]) - .custom_created_at(ts) - .sign_with_keys(&state.relay_keypair) - .map_err(|e| anyhow::anyhow!("failed to sign system message: {e}"))?; + let event = sign_system_message(state, channel_id, content, idempotency_ts)?; // Durable insert is the completion boundary — propagate failure. state @@ -990,16 +989,41 @@ pub async fn emit_system_message( .await .map_err(|e| anyhow::anyhow!("system message insert failed: {e}"))?; - // Fan out to subscribers: best-effort, clients can retrieve the persisted event. + fan_out_system_message(tenant, state, channel_id, &event).await; + Ok(()) +} + +/// Sign a kind:40099 system message for `channel_id` with the relay keypair. +fn sign_system_message( + state: &Arc, + channel_id: Uuid, + content: serde_json::Value, + created_at: chrono::DateTime, +) -> anyhow::Result { + let channel_tag = Tag::parse(["h", &channel_id.to_string()])?; + let ts = nostr::Timestamp::from(created_at.timestamp() as u64); + EventBuilder::new(Kind::Custom(40099), content.to_string()) + .tags([channel_tag]) + .custom_created_at(ts) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| anyhow::anyhow!("failed to sign system message: {e}")) +} + +/// Fan out a stored system message. Best-effort: clients can retrieve the +/// persisted event. +async fn fan_out_system_message( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + event: &Event, +) { if let Err(e) = state .pubsub - .publish_event(tenant, EventTopic::Channel(channel_id), &event) + .publish_event(tenant, EventTopic::Channel(channel_id), event) .await { warn!("System message fan-out failed: {e}"); } - - Ok(()) } /// Sign and fan out a fresh relay-signed `kind:39005` thread-summary overlay @@ -1691,281 +1715,205 @@ async fn handle_remove_user( Ok(()) } -async fn handle_edit_metadata( +/// Persist an authorized kind:9002 event and the channel changes it names. +/// +/// One transaction stores the event, applies every change and stores the +/// system messages that announce them. If it fails, nothing is stored and the +/// client is told so, so it can resend the same event. Live effects (cache +/// invalidation, subscription eviction, fan-out, discovery events) run only +/// after the commit. +pub(crate) async fn persist_edit_metadata( tenant: &TenantContext, event: &Event, state: &Arc, -) -> anyhow::Result<()> { +) -> anyhow::Result<(buzz_core::StoredEvent, bool)> { let channel_id = extract_h_tag_channel(event).ok_or_else(|| anyhow::anyhow!("missing h tag"))?; let actor_bytes = event.pubkey.to_bytes().to_vec(); let actor_hex = hex::encode(&actor_bytes); + // kind:9002 is a partial update: a missing tag means "no change". If a + // tag repeats, its last value wins. + let mut edit = buzz_db::channel::ChannelMetadataEdit::default(); for tag in event.tags.iter() { - let key = tag.kind().to_string(); - if let Some(val) = tag.content() { - match key.as_str() { - "name" => { - state - .db - .update_channel( - tenant.community(), - channel_id, - buzz_db::channel::ChannelUpdate { - name: Some(val.to_string()), - ..Default::default() - }, - ) - .await?; - } - "about" => { - state - .db - .update_channel( - tenant.community(), - channel_id, - buzz_db::channel::ChannelUpdate { - description: Some(val.to_string()), - ..Default::default() - }, - ) - .await?; - } - "topic" => { - state - .db - .set_topic(tenant.community(), channel_id, val, &actor_bytes) - .await?; - emit_system_message( - tenant, - state, - channel_id, - serde_json::json!({ - "type": "topic_changed", "actor": actor_hex, "topic": val - }), - chrono::Utc::now(), - ) - .await?; - } - "purpose" => { - state - .db - .set_purpose(tenant.community(), channel_id, val, &actor_bytes) - .await?; - emit_system_message( - tenant, - state, - channel_id, - serde_json::json!({ - "type": "purpose_changed", "actor": actor_hex, "purpose": val - }), - chrono::Utc::now(), - ) - .await?; - } - "visibility" => { - let was_open = state - .db - .get_channel_for_event_write(tenant.community(), channel_id) - .await - .map(|c| c.visibility == "open") - .unwrap_or(false); - state - .db - .update_channel( - tenant.community(), - channel_id, - buzz_db::channel::ChannelUpdate { - visibility: Some(val.to_string()), - ..Default::default() - }, - ) - .await?; - // A visibility flip changes who can see the channel, so the - // accessible-channels and visibility caches must be cleared - // before any later event for this channel fans out. - state.invalidate_all_accessible_channels(tenant); - state.invalidate_channel_visibility(tenant, channel_id); - // On open -> private, eagerly close non-members' live subs - // for an immediate CLOSED on this node. The fan-out access - // filter is the cluster-wide correctness backstop. - if was_open && val == "private" { - evict_non_member_channel_subscriptions(tenant, state, channel_id).await?; - } - emit_system_message( - tenant, - state, - channel_id, - serde_json::json!({ - "type": "visibility_changed", "actor": actor_hex, "visibility": val - }), - chrono::Utc::now(), - ) - .await?; - } - "join_role" => { - // Validated during authorization; parse again so a bad - // value can never reach the column. - let join_role: buzz_db::channel::ChannelJoinRole = - val.parse().map_err(|e: String| anyhow::anyhow!(e))?; - state - .db - .update_channel( - tenant.community(), - channel_id, - buzz_db::channel::ChannelUpdate { - join_role: Some(join_role.as_str().to_string()), - ..Default::default() - }, - ) - .await?; - emit_system_message( - tenant, - state, - channel_id, - serde_json::json!({ - "type": "join_role_changed", "actor": actor_hex, "join_role": join_role.as_str() - }), - chrono::Utc::now(), - ) - .await?; - } - "ttl" => { - // Empty string clears the TTL (permanent); otherwise it is a - // positive integer of seconds, validated during authorization. - // Fail closed: a parse failure must reject, never silently - // clear the TTL to permanent. - let ttl_change: Option = if val.is_empty() { - None - } else { - Some(val.parse::().map_err(|_| { - anyhow::anyhow!("invalid ttl value: {val} (must be a positive integer)") - })?) - }; - state - .db - .update_channel( - tenant.community(), - channel_id, - buzz_db::channel::ChannelUpdate { - ttl_seconds: Some(ttl_change), - ..Default::default() - }, - ) - .await?; - emit_system_message( - tenant, - state, - channel_id, - serde_json::json!({ - "type": "ttl_changed", "actor": actor_hex, "ttl_seconds": ttl_change - }), - chrono::Utc::now(), - ) - .await?; - } - "archived" => { - match val { - "true" => { - state - .db - .archive_channel(tenant.community(), channel_id) - .await?; - emit_system_message( - tenant, - state, - channel_id, - serde_json::json!({ - "type": "channel_archived", "actor": actor_hex - }), - chrono::Utc::now(), - ) - .await?; - } - "false" => { - state - .db - .unarchive_channel(tenant.community(), channel_id) - .await?; - emit_system_message( - tenant, - state, - channel_id, - serde_json::json!({ - "type": "channel_unarchived", "actor": actor_hex - }), - chrono::Utc::now(), - ) - .await?; - - // Resubscribe connected agents after restore: archiving evicts their - // live subscriptions (CLOSED "channel access revoked") and unarchive - // otherwise emits no signal that makes a connected agent resubscribe. - // We reuse the member_added notification (44100) purely as a resubscribe - // trigger — no membership actually changed here — because it flows on the - // agent's always-live global membership subscription, the same path - // remove/re-add uses to recover. Humans self-heal via the re-emitted - // kind:39000 discovery, so this is intentionally agent-scoped. - // - // Known limitation: emit_membership_notification builds a created_at=now - // event with no nonce, and insert_event skips fan-out on a duplicate id. - // Four sub-second toggles (archive->unarchive->archive->unarchive) on the - // same channel by the same actor could collide ids and skip a fan-out. - // Not reachable in practice — unarchive has a single human-driven caller; - // the reaper only auto-archives — so we don't engineer around it. - for member in state - .db - .get_members_for_event_write(tenant.community(), channel_id) - .await? - { - if let Err(e) = emit_membership_notification( - tenant, - state, - channel_id, - &member.pubkey, - &actor_bytes, - KIND_MEMBER_ADDED_NOTIFICATION, - ) - .await - { - warn!( - channel = %channel_id, - error = %e, - "post-unarchive resubscribe notification failed" - ); - } - } - } - _ => {} // ignore invalid values - } - } - _ => {} + let Some(val) = tag.content() else { continue }; + match tag.kind().to_string().as_str() { + "name" => edit.columns.name = Some(val.to_string()), + "about" => edit.columns.description = Some(val.to_string()), + "topic" => edit.topic = Some(val.to_string()), + "purpose" => edit.purpose = Some(val.to_string()), + "visibility" => edit.columns.visibility = Some(val.to_string()), + "join_role" => { + // Validated during authorization; parse again so a bad + // value can never reach the column. + let join_role: buzz_db::channel::ChannelJoinRole = + val.parse().map_err(|e: String| anyhow::anyhow!(e))?; + edit.columns.join_role = Some(join_role.as_str().to_string()); } + "ttl" => { + // Empty string clears the TTL (permanent); otherwise it is a + // positive integer of seconds, validated during authorization. + // Fail closed: a parse failure must reject, never silently + // clear the TTL to permanent. + edit.columns.ttl_seconds = Some(if val.is_empty() { + None + } else { + Some(val.parse::().map_err(|_| { + anyhow::anyhow!("invalid ttl value: {val} (must be a positive integer)") + })?) + }); + } + "archived" => match val { + "true" => edit.archived = Some(true), + "false" => edit.archived = Some(false), + _ => {} // refused during authorization + }, + _ => {} } } - - // All `t` tags together form the new label set, so they are applied - // once after the loop. `["t", ""]` alone clears the set. + // All `t` tags together form the new label set. `["t", ""]` alone clears it. if let Some(labels) = channel_labels_from_tags(event) { - let labels = labels.map_err(|e| anyhow::anyhow!("{e}"))?; - state - .db - .update_channel( - tenant.community(), - channel_id, - buzz_db::channel::ChannelUpdate { - labels: Some(labels), - ..Default::default() - }, - ) - .await?; + edit.columns.labels = Some(labels.map_err(|e| anyhow::anyhow!("{e}"))?); } + let now = chrono::Utc::now(); + let mut notices = Vec::new(); + if let Some(topic) = &edit.topic { + notices.push(serde_json::json!({ + "type": "topic_changed", "actor": actor_hex, "topic": topic + })); + } + if let Some(purpose) = &edit.purpose { + notices.push(serde_json::json!({ + "type": "purpose_changed", "actor": actor_hex, "purpose": purpose + })); + } + if let Some(visibility) = &edit.columns.visibility { + notices.push(serde_json::json!({ + "type": "visibility_changed", "actor": actor_hex, "visibility": visibility + })); + } + if let Some(join_role) = &edit.columns.join_role { + notices.push(serde_json::json!({ + "type": "join_role_changed", "actor": actor_hex, "join_role": join_role + })); + } + if let Some(ttl_seconds) = edit.columns.ttl_seconds { + notices.push(serde_json::json!({ + "type": "ttl_changed", "actor": actor_hex, "ttl_seconds": ttl_seconds + })); + } + match edit.archived { + Some(true) => notices.push(serde_json::json!({ + "type": "channel_archived", "actor": actor_hex + })), + Some(false) => notices.push(serde_json::json!({ + "type": "channel_unarchived", "actor": actor_hex + })), + None => {} + } + let system_messages = notices + .into_iter() + .map(|content| sign_system_message(state, channel_id, content, now)) + .collect::>>()?; + + let new_visibility = edit.columns.visibility.clone(); + let unarchived = edit.archived == Some(false); + let write = state + .db + .apply_channel_metadata_event( + tenant.community(), + channel_id, + event, + edit, + &system_messages, + ) + .await?; + if !write.inserted { + return Ok((write.stored, false)); + } + + // The changes are committed. What follows only updates live state, so a + // failure is logged; the next read sees the committed channel. + if let Some(visibility) = new_visibility.as_deref() { + // A visibility flip changes who can see the channel, so the + // accessible-channels and visibility caches must be cleared + // before any later event for this channel fans out. + state.invalidate_all_accessible_channels(tenant); + state.invalidate_channel_visibility(tenant, channel_id); + // On open -> private, eagerly close non-members' live subs + // for an immediate CLOSED on this node. The fan-out access + // filter is the cluster-wide correctness backstop. + if write.previous_visibility == "open" && visibility == "private" { + if let Err(e) = evict_non_member_channel_subscriptions(tenant, state, channel_id).await + { + warn!(channel = %channel_id, error = %e, "non-member subscription eviction failed"); + } + } + } + for message in &system_messages { + fan_out_system_message(tenant, state, channel_id, message).await; + } + if unarchived { + resubscribe_members_after_unarchive(tenant, state, channel_id, &actor_bytes).await; + } if let Err(e) = emit_group_discovery_events(tenant, state, channel_id).await { warn!(channel = %channel_id, error = %e, "NIP-29 group discovery emission failed"); } - Ok(()) + Ok((write.stored, true)) +} + +/// Resubscribe connected agents after restore: archiving evicts their +/// live subscriptions (CLOSED "channel access revoked") and unarchive +/// otherwise emits no signal that makes a connected agent resubscribe. +/// We reuse the member_added notification (44100) purely as a resubscribe +/// trigger — no membership actually changed here — because it flows on the +/// agent's always-live global membership subscription, the same path +/// remove/re-add uses to recover. Humans self-heal via the re-emitted +/// kind:39000 discovery, so this is intentionally agent-scoped. +/// +/// Known limitation: emit_membership_notification builds a created_at=now +/// event with no nonce, and insert_event skips fan-out on a duplicate id. +/// Four sub-second toggles (archive->unarchive->archive->unarchive) on the +/// same channel by the same actor could collide ids and skip a fan-out. +/// Not reachable in practice — unarchive has a single human-driven caller; +/// the reaper only auto-archives — so we don't engineer around it. +async fn resubscribe_members_after_unarchive( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + actor_bytes: &[u8], +) { + let members = match state + .db + .get_members_for_event_write(tenant.community(), channel_id) + .await + { + Ok(members) => members, + Err(e) => { + warn!(channel = %channel_id, error = %e, "post-unarchive member read failed"); + return; + } + }; + for member in members { + if let Err(e) = emit_membership_notification( + tenant, + state, + channel_id, + &member.pubkey, + actor_bytes, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + warn!( + channel = %channel_id, + error = %e, + "post-unarchive resubscribe notification failed" + ); + } + } } /// The channel label set named by an event's `["t",