diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index a14cb9aa02..6b548da4f9 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,198 @@ 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 mut locked = lock_channel_for_settings_write(&mut tx, channel_id, &updates).await?; + update_channel_on(&mut 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(()) +} + +/// The only way to write a channel's settings. Only +/// [`lock_channel_for_settings_write`] makes one, after it takes the locks +/// on the transaction it borrows, and every settings write is a method on +/// it. The borrow binds the locks to that transaction: a writer cannot use +/// them with another transaction or after its commit, skip a lock, or take +/// the locks in a different order. +pub(crate) struct ChannelSettingsWriter<'tx> { + tx: &'tx mut crate::AdmittedTx, + channel_id: Uuid, + admission: bool, + ttl: TtlLock, + previous_visibility: String, + archived: bool, +} + +impl ChannelSettingsWriter<'_> { + /// The channel's visibility when the lock was taken (`"open"` or `"private"`). + pub(crate) fn previous_visibility(&self) -> &str { + &self.previous_visibility + } + + /// Whether the channel was archived when the lock was taken. The row + /// lock keeps this current until the transaction ends. + pub(crate) fn archived(&self) -> bool { + self.archived + } + + /// The locked transaction, for writes that are not channel settings + /// (for example, storing the event that requested the change). + pub(crate) fn tx(&mut self) -> &mut crate::AdmittedTx { + self.tx + } + + /// 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, updates: &ChannelUpdate) -> Result<()> { + let covered = (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 NO KEY 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>( + tx: &'tx mut crate::AdmittedTx, + channel_id: Uuid, + updates: &ChannelUpdate, +) -> Result> { + let ttl = if updates.ttl_seconds.is_some() { + TtlLock::Exclusive + } else { + TtlLock::Shared + }; + lock_channel_row(tx, channel_id, updates.changes_admission(), ttl).await +} + +/// Takes the locks that deleting `channel_id` needs, in the same order as +/// [`lock_channel_for_settings_write`]. A delete ends admission, so it takes +/// the membership lock: a membership write or a huddle join either commits +/// first or, under the lock, finds the channel deleted. A delete does not +/// change the TTL, so the TTL lock is SHARED. +async fn lock_channel_for_delete( + tx: &mut crate::AdmittedTx, + channel_id: Uuid, +) -> Result> { + lock_channel_row(tx, channel_id, true, TtlLock::Shared).await +} + +async fn lock_channel_row( + tx: &mut crate::AdmittedTx, + channel_id: Uuid, + admission: bool, + ttl: TtlLock, +) -> Result> { + let community_id = tx.community(); + if admission { + crate::channel_members::acquire_channel_membership_lock( + tx.conn(), + community_id, + channel_id, + ) + .await?; + } + lock_channel_ttl(tx, channel_id, ttl).await?; + // `FOR NO KEY UPDATE`, like a huddle join: it orders this write against + // every other settings write and a huddle join, but not against the + // `KEY SHARE` lock that a membership INSERT takes for its foreign key. + let (previous_visibility, archived): (String, bool) = sqlx::query_as( + "SELECT visibility::text, archived_at IS NOT NULL FROM channels \ + WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL FOR NO KEY UPDATE", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await? + .ok_or(DbError::ChannelNotFound(channel_id))?; + Ok(ChannelSettingsWriter { + tx, + channel_id, + admission, + ttl, + previous_visibility, + archived, + }) +} + +/// Writes the set fields of `updates` to the locked channel row on `tx`. +/// The caller owns the commit. +async fn update_channel_on( + w: &mut ChannelSettingsWriter<'_>, + mut updates: ChannelUpdate, +) -> Result<()> { + w.check_covers(&updates)?; + let channel_id = w.channel_id; + let tx = &mut *w.tx; + 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 +1057,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 +1078,29 @@ pub async fn set_topic( crate::observability::WriterOperation::EventWrite, ) .await?; + let mut locked = + lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?; + set_topic_on(&mut locked, topic, set_by).await?; + tx.commit().await?; + Ok(()) +} + +async fn set_topic_on(w: &mut ChannelSettingsWriter<'_>, topic: &str, set_by: &[u8]) -> Result<()> { + let channel_id = w.channel_id; + let tx = &mut *w.tx; 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 +1118,33 @@ pub async fn set_purpose( crate::observability::WriterOperation::EventWrite, ) .await?; + let mut locked = + lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?; + set_purpose_on(&mut locked, purpose, set_by).await?; + tx.commit().await?; + Ok(()) +} + +async fn set_purpose_on( + w: &mut ChannelSettingsWriter<'_>, + purpose: &str, + set_by: &[u8], +) -> Result<()> { + let channel_id = w.channel_id; + let tx = &mut *w.tx; 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 +1163,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 mut locked = + lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?; + set_archived_on(&mut locked, true).await?; tx.commit().await?; Ok(()) } @@ -1040,42 +1185,174 @@ pub async fn unarchive_channel( crate::observability::WriterOperation::EventWrite, ) .await?; - // First check: does the channel exist and what is its state? + let mut locked = + lock_channel_for_settings_write(&mut tx, channel_id, &ChannelUpdate::default()).await?; + set_archived_on(&mut 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(w: &mut ChannelSettingsWriter<'_>, archive: bool) -> Result<()> { + let channel_id = w.channel_id; + let tx = &mut *w.tx; + 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", - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .execute(tx.conn()) - .await?; - - tx.commit().await?; + 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, + ) + .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 mut 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( + locked.tx(), + event, + Some(channel_id), + None, + ) + .await?; + if !inserted { + tx.rollback().await?; + return Ok(ChannelMetadataWrite { + stored, + inserted, + previous_visibility, + }); + } + // Decide the archive state here, after the duplicate check and under + // the row lock, not during validation. A resend that races the first + // commit then sees that commit as a duplicate (above), never as a + // channel already in the requested state. + match (locked.archived(), edit.archived) { + (true, Some(false)) | (false, None | Some(true)) => {} + (true, _) => return Err(DbError::AccessDenied("channel is archived".to_string())), + (false, Some(false)) => { + return Err(DbError::AccessDenied("channel is not archived".to_string())); + } + } + crate::insert_mentions_in_transaction(locked.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 locked, columns).await?; + } + if let Some(topic) = topic { + set_topic_on(&mut locked, &topic, &actor).await?; + } + if let Some(purpose) = purpose { + set_purpose_on(&mut locked, &purpose, &actor).await?; + } + if let Some(archive) = archived { + set_archived_on(&mut locked, archive).await?; + } + // The same writes as `Db::insert_event`. + for message in system_messages { + let (_, message_inserted) = + crate::event::insert_event_in_transaction(locked.tx(), message, Some(channel_id)) + .await?; + if message_inserted { + crate::insert_mentions_in_transaction(locked.tx(), message, Some(channel_id)).await?; + } + } + tx.commit().await?; + Ok(ChannelMetadataWrite { + stored, + inserted, + previous_visibility, + }) +} + /// Soft-delete a channel by setting `deleted_at = NOW()`. /// /// Returns `Ok(true)` if the channel was deleted, `Ok(false)` if already @@ -1091,16 +1368,30 @@ pub async fn soft_delete_channel( crate::observability::WriterOperation::EventWrite, ) .await?; - let result = sqlx::query( - "UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .execute(tx.conn()) - .await?; - + let mut writer = match lock_channel_for_delete(&mut tx, channel_id).await { + Ok(writer) => writer, + Err(DbError::ChannelNotFound(_)) => return Ok(false), + Err(error) => return Err(error), + }; + soft_delete_on(&mut writer).await?; tx.commit().await?; - Ok(result.rows_affected() > 0) + Ok(true) +} + +async fn soft_delete_on(w: &mut ChannelSettingsWriter<'_>) -> Result<()> { + if !w.admission { + return Err(DbError::InvalidData( + "channel delete is not covered by its lock".into(), + )); + } + let channel_id = w.channel_id; + let community_id = w.tx.community(); + sqlx::query("UPDATE channels SET deleted_at = NOW() WHERE community_id = $1 AND id = $2") + .bind(community_id.as_uuid()) + .bind(channel_id) + .execute(w.tx.conn()) + .await?; + Ok(()) } /// Archive ephemeral channels whose TTL deadline has passed. @@ -1133,6 +1424,16 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result 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 +1747,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()) @@ -2077,4 +2404,52 @@ mod postgres_tests { .expect("members"); assert_eq!(members.len(), 1, "only the first creator is a member"); } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn soft_delete_waits_for_membership_lock() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); + let owner = random_pubkey(); + let channel = create_test_channel( + &pool, + community_id, + "delete-vs-roster", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create channel"); + + // A roster publication holds the membership lock. A delete ends + // admission, so it must wait, like any membership write. + let snapshot = + crate::channel_members::lock_member_snapshot(&pool, community, channel.id, &[9u8; 32]) + .await + .expect("capture locked roster"); + let delete_pool = crate::test_support::named_pool("delete-vs-roster").await; + let channel_id = channel.id; + let delete = + tokio::spawn( + async move { soft_delete_channel(&delete_pool, community, channel_id).await }, + ); + crate::test_support::wait_for_lock_wait(&pool, "delete-vs-roster").await; + assert!( + !delete.is_finished(), + "delete must wait for the roster fence" + ); + snapshot.release().await.expect("release roster fence"); + assert!(delete.await.expect("delete task").expect("delete")); + + assert!( + !soft_delete_channel(&pool, community, channel.id) + .await + .expect("second delete"), + "a second delete finds no live channel" + ); + } } diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 28e64ff61c..6013ca869a 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -199,13 +199,13 @@ pub(crate) async fn acquire_channel_membership_lock( /// Check the same p-tag shape and canonical active membership as migration 0032. /// The caller holds this channel's membership advisory lock on the admitted -/// transaction before calling. The membership writers in this module take that -/// key before reading roles or changing rows. Not every writer does yet: -/// `execute_kick_with_marker` and `deploy_kick_member` (relay_admin_actions.rs) -/// set `removed_at` without it, so a kick can commit after this check passes -/// and before the roster commits. The retained trigger has the same gap. Both -/// kick paths must take the key, with a concurrent kick-vs-roster-publication -/// test, before anything relies on complete serialization (phase 2 included). +/// transaction before calling. Every writer that changes the membership of an +/// existing channel takes that key before reading roles or changing rows +/// (channel creation inserts rows that no other transaction can see yet), +/// including both kick paths +/// (`execute_kick_with_marker` and `deploy_kick_member` in +/// relay_admin_actions.rs), so no removal can commit between this check and +/// the roster commit. /// Legacy replacement takes the replacement key, locks /// old-head rows in its UPDATE, then takes the membership key at trigger INSERT. /// The app path takes replacement key -> membership key -> old-head row locks, @@ -287,17 +287,89 @@ pub(crate) async fn guard_channel_roster_event( // ── Transaction-level membership helpers (for commit_participant_join) ──────── -/// Acquire the per-channel membership advisory lock on a caller-owned transaction. +/// Proof that a caller-owned transaction holds the membership locks of a set +/// of channels. Only [`lock_channel_memberships_in_transaction`] makes one, +/// after it takes the locks on the transaction it borrows. The borrow binds +/// the locks to that transaction: the proof cannot be used with another +/// transaction or after its commit. Membership writes that compose with other +/// work on the caller's transaction (a huddle join) are methods on it, so +/// they cannot run without the lock. +pub struct ChannelMembershipLocks<'tx> { + tx: &'tx mut AdmittedTx, + channels: Vec, +} + +impl ChannelMembershipLocks<'_> { + /// The locked transaction, for the other reads and writes of the caller's + /// operation. + pub fn tx(&mut self) -> &mut AdmittedTx { + self.tx + } + + fn check_covers(&self, channel_id: Uuid) -> Result<()> { + if self.channels.contains(&channel_id) { + Ok(()) + } else { + Err(DbError::InvalidData( + "channel membership write is not covered by its lock".into(), + )) + } + } + + /// Auto-add a member (for ephemeral-channel admission). + /// + /// Inserts or reactivates the membership row at `Member` role with the + /// given `invited_by` (channel creator for huddle auto-add). Refuses a + /// channel whose membership lock this proof does not hold. + /// + /// Used by `commit_participant_join` to atomically add membership and the + /// `48101` event in a single transaction under a session effect permit. + pub async fn insert_auto_membership( + &mut self, + channel_id: Uuid, + pubkey: &[u8], + invited_by: &[u8], + ) -> Result<()> { + self.check_covers(channel_id)?; + let community_id = self.tx.community(); + sqlx::query( + r#" + INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) + VALUES ($1, $2, $3, 'member'::member_role, $4) + ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET + removed_at = NULL, + removed_by = NULL, + role = EXCLUDED.role + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(pubkey) + .bind(invited_by) + .execute(self.tx.conn()) + .await?; + Ok(()) + } +} + +/// Take the membership lock of each channel in `channels`, in that order, on +/// a caller-owned transaction, and return the proof that holds them. /// -/// Equivalent to the internal `acquire_channel_membership_lock`, but exposed -/// for callers that need to compose multiple operations in one transaction -/// (e.g., `commit_participant_join` in `audio/handler.rs`). -pub async fn acquire_channel_membership_lock_in_transaction( - tx: &mut AdmittedTx, - channel_id: Uuid, -) -> Result<()> { +/// These locks come first in the channel lock order (membership, then TTL, +/// then the channel row; see `channel::lock_channel_for_settings_write`). +/// Take them before any other channel lock on `tx`. +pub async fn lock_channel_memberships_in_transaction<'tx>( + tx: &'tx mut AdmittedTx, + channels: &[Uuid], +) -> Result> { let community_id = tx.community(); - acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await + for channel_id in channels { + acquire_channel_membership_lock(tx.conn(), community_id, *channel_id).await?; + } + Ok(ChannelMembershipLocks { + tx, + channels: channels.to_vec(), + }) } /// Lock the live channel row on a caller-owned transaction and return its @@ -405,41 +477,6 @@ pub async fn active_role_in_transaction( get_active_role_tx(tx.conn(), community_id, channel_id, pubkey).await } -/// Auto-add a member on a caller-owned transaction (for ephemeral-channel admission). -/// -/// Inserts or reactivates the membership row at `Member` role with the given -/// `invited_by` (channel creator for huddle auto-add). Does NOT acquire the -/// advisory lock — callers must have already called -/// `acquire_channel_membership_lock_in_transaction` before calling this. -/// -/// Used by `commit_participant_join` to atomically add membership and the -/// `48101` event in a single transaction under a session effect permit. -pub async fn insert_auto_membership_in_transaction( - tx: &mut AdmittedTx, - channel_id: Uuid, - pubkey: &[u8], - invited_by: &[u8], -) -> Result<()> { - let community_id = tx.community(); - sqlx::query( - r#" - INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) - VALUES ($1, $2, $3, 'member'::member_role, $4) - ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET - removed_at = NULL, - removed_by = NULL, - role = EXCLUDED.role - "#, - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .bind(pubkey) - .bind(invited_by) - .execute(tx.conn()) - .await?; - Ok(()) -} - // ── End transaction-level helpers ───────────────────────────────────────────── /// The immutable facts that every relay-signed group-state event @@ -3386,6 +3423,25 @@ mod postgres_tests { #[tokio::test] #[ignore = "requires Postgres"] async fn join_role_change_waits_for_a_roster_publication_on_an_ephemeral_channel() { + join_role_change_waits_for_roster_publication(SettingsWriter::UpdateChannel).await; + } + + /// The kind:9002 path stores the event and applies its changes in its own + /// transaction. It must take the same locks in the same order as + /// `update_channel`. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn kind_9002_join_role_change_waits_for_a_roster_publication_on_an_ephemeral_channel() { + join_role_change_waits_for_roster_publication(SettingsWriter::Kind9002).await; + } + + #[derive(Clone, Copy)] + enum SettingsWriter { + UpdateChannel, + Kind9002, + } + + async fn join_role_change_waits_for_roster_publication(writer: SettingsWriter) { let pool = setup_pool().await; let community_id = make_test_community(&pool).await; let community = CommunityId::from_uuid(community_id); @@ -3424,19 +3480,45 @@ mod postgres_tests { let pool = pool.clone(); let channel_id = channel.id; async move { - crate::channel::update_channel( - &pool, - community, - channel_id, - crate::channel::ChannelUpdate { - join_role: Some("guest".into()), - ..Default::default() - }, - ) - .await + let updates = crate::channel::ChannelUpdate { + join_role: Some("guest".into()), + ..Default::default() + }; + match writer { + SettingsWriter::UpdateChannel => { + crate::channel::update_channel(&pool, community, channel_id, updates).await + } + SettingsWriter::Kind9002 => { + let edit_event = nostr::EventBuilder::new(nostr::Kind::Custom(9002), "") + .tags(vec![ + nostr::Tag::parse(["h", &channel_id.to_string()]).expect("h tag"), + nostr::Tag::parse(["join_role", "guest"]).expect("join_role tag"), + ]) + .sign_with_keys(&Keys::generate()) + .expect("sign kind:9002"); + let write = crate::channel::apply_channel_metadata_event( + &pool, + None, + community, + channel_id, + &edit_event, + crate::channel::ChannelMetadataEdit { + columns: updates, + ..Default::default() + }, + &[], + ) + .await?; + assert!(write.inserted, "the kind:9002 event must be stored"); + crate::channel::get_channel(&pool, community, channel_id).await + } + } } }); - tokio::time::sleep(std::time::Duration::from_millis(300)).await; + // The kind:9002 writer also stores an event before it would reach + // the row, so give an unblocked writer time to finish. A blocked + // writer waits until the roster commits, so this cannot pass early. + tokio::time::sleep(std::time::Duration::from_secs(2)).await; assert!( !update.is_finished(), "the join_role change must wait for the roster publication" diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index 305ab18de2..6c819d3d24 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -703,6 +703,12 @@ pub async fn execute_kick_with_marker( .bind(action_id) .fetch_optional(tx.conn()) .await?; + // Like every membership writer, remove under the channel's membership + // lock, so a roster publication either sees the member removed or + // commits before the removal. No channel lock is held yet, so the lock + // order holds. + crate::channel_members::acquire_channel_membership_lock(tx.conn(), community_id, channel_id) + .await?; let kick = sqlx::query( r#" @@ -1689,6 +1695,11 @@ pub async fn deploy_kick_member( crate::observability::WriterOperation::Authorization, ) .await?; + // Like every membership writer, remove under the channel's membership + // lock, so a roster publication either sees the member removed or + // commits before the removal. + crate::channel_members::acquire_channel_membership_lock(tx.conn(), community_id, channel_id) + .await?; // Use a direct UPDATE to avoid the tenant ownership check in channel::remove_member. // This is the deployment-authority primitive: no actor role check. let result = sqlx::query( @@ -3840,14 +3851,27 @@ mod postgres_tests { }; // First kick: member is present → Removed + step_marker committed. - let r1 = execute_kick_with_marker( + // It runs while a roster publication holds the member snapshot, and + // must wait for it. + let (kicked, kick_actor) = (target.clone(), actork.clone()); + let r1 = assert_kick_waits_for_roster( &pool, - action_id, - lease_token1, - cid, + community_id, channel_id, &target, - &actork, + "marker-kick-vs-roster", + move |kick_pool| async move { + execute_kick_with_marker( + &kick_pool, + action_id, + lease_token1, + cid, + channel_id, + &kicked, + &kick_actor, + ) + .await + }, ) .await .expect("first kick"); @@ -4054,6 +4078,106 @@ mod postgres_tests { ); } + /// Runs `kick` on its own session while a roster publication holds the + /// channel's member snapshot. The kick must wait for the snapshot's + /// membership lock, so the published roster cannot list a member whose + /// removal committed before it. + async fn assert_kick_waits_for_roster( + pool: &PgPool, + community_id: Uuid, + channel_id: Uuid, + target: &[u8], + application_name: &str, + kick: impl FnOnce(PgPool) -> F, + ) -> T + where + F: std::future::Future + Send + 'static, + T: Send + 'static, + { + let snapshot = crate::channel_members::lock_member_snapshot( + pool, + CommunityId::from_uuid(community_id), + channel_id, + &[9u8; 32], + ) + .await + .expect("capture locked roster"); + assert!( + snapshot.members.iter().any(|m| m.pubkey == target), + "the captured roster lists the member before the kick" + ); + let kick_pool = crate::test_support::named_pool(application_name).await; + let kick = tokio::spawn(kick(kick_pool)); + crate::test_support::wait_for_lock_wait(pool, application_name).await; + assert!(!kick.is_finished(), "kick must wait for the roster fence"); + let active: bool = sqlx::query_scalar( + "SELECT removed_at IS NULL FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", + ) + .bind(community_id) + .bind(channel_id) + .bind(target) + .fetch_one(pool) + .await + .expect("read membership"); + assert!(active, "the member stays active while the roster is held"); + snapshot.release().await.expect("release roster fence"); + kick.await.expect("kick task") + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn deploy_kick_member_waits_for_roster_publication() { + let pool = setup_pool().await; + let community_id = make_community(&pool).await; + let actor = actor(); + let target = vec![4u8; 32]; + let channel_id = Uuid::new_v4(); + sqlx::query( + r#" + INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by) + VALUES ($1, $2, 'kick-vs-roster', 'stream', 'open', $3) + "#, + ) + .bind(channel_id) + .bind(community_id) + .bind(&actor) + .execute(&pool) + .await + .expect("create channel"); + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role) VALUES ($1, $2, $3, 'member')", + ) + .bind(community_id) + .bind(channel_id) + .bind(&target) + .execute(&pool) + .await + .expect("add member"); + + let kicked = target.clone(); + let result = assert_kick_waits_for_roster( + &pool, + community_id, + channel_id, + &target, + "deploy-kick-vs-roster", + move |kick_pool| async move { + deploy_kick_member( + &kick_pool, + CommunityId::from_uuid(community_id), + channel_id, + &kicked, + &actor, + ) + .await + }, + ) + .await + .expect("kick after roster publication"); + assert_eq!(result, KickResult::Removed); + } + // ── Reopen: terminal → open CAS + durable audit row ─────────────────────── async fn set_report_status(pool: &PgPool, community_id: Uuid, report_id: Uuid, status: &str) { diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index 18c8580817..7a09f97505 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -3282,17 +3282,21 @@ async fn commit_participant_join( #[cfg(test)] crate::nip_fi_test_hooks::before_membership_lock(tenant.community()).await; - buzz_db::channel_members::acquire_channel_membership_lock_in_transaction(&mut tx, channel_id) - .await?; - if parent_channel_id != channel_id { - buzz_db::channel_members::acquire_channel_membership_lock_in_transaction( - &mut tx, - parent_channel_id, - ) - .await?; - } - buzz_db::channel_members::lock_channel_ttl_shared_in_transaction(&mut tx, parent_channel_id) - .await?; + let locked_channels = if parent_channel_id == channel_id { + vec![channel_id] + } else { + vec![channel_id, parent_channel_id] + }; + let mut membership = buzz_db::channel_members::lock_channel_memberships_in_transaction( + &mut tx, + &locked_channels, + ) + .await?; + buzz_db::channel_members::lock_channel_ttl_shared_in_transaction( + membership.tx(), + parent_channel_id, + ) + .await?; // Archive re-check (ALL paths): re-read archived_at inside the // transaction before any write, taking a row-level write lock @@ -3306,9 +3310,11 @@ async fn commit_participant_join( // is held only for the duration of the join transaction (typically // sub-millisecond). See `lock_channel_archived_at_in_transaction` for why // the lock is `FOR NO KEY UPDATE` rather than `FOR UPDATE`. - let channel_archived = - buzz_db::channel_members::lock_channel_archived_at_in_transaction(&mut tx, channel_id) - .await?; + let channel_archived = buzz_db::channel_members::lock_channel_archived_at_in_transaction( + membership.tx(), + channel_id, + ) + .await?; // Test hook: fires after the FOR UPDATE lock is acquired but before the // archived check / any write. A test can attempt a concurrent archive here @@ -3338,9 +3344,12 @@ async fn commit_participant_join( // IMPORTANT 4b: Re-read parent membership under the lock. A parent // membership revocation in the same window would make the auto-add // unjustified; reject rather than grant access from stale authority. - let parent_still_member = - buzz_db::channel_members::is_member_in_transaction(&mut tx, *parent_id, pubkey_bytes) - .await?; + let parent_still_member = buzz_db::channel_members::is_member_in_transaction( + membership.tx(), + *parent_id, + pubkey_bytes, + ) + .await?; if !parent_still_member { retire_shadow(); @@ -3355,7 +3364,7 @@ async fn commit_participant_join( // but before this transaction acquires the lock; committing a join into // an unlinked channel violates the "creator authority" invariant. let link_still_exists = buzz_db::event::huddle_started_link_exists_in_transaction( - &mut tx, + membership.tx(), *parent_id, channel_id, channel_created_by.as_slice(), @@ -3370,18 +3379,17 @@ async fn commit_participant_join( // Re-read child membership — a concurrent legitimate add may have // already provided access; do not overwrite role/provenance. - let still_absent = - !buzz_db::channel_members::is_member_in_transaction(&mut tx, channel_id, pubkey_bytes) - .await?; + let still_absent = !buzz_db::channel_members::is_member_in_transaction( + membership.tx(), + channel_id, + pubkey_bytes, + ) + .await?; if still_absent { - buzz_db::channel_members::insert_auto_membership_in_transaction( - &mut tx, - channel_id, - pubkey_bytes, - channel_created_by.as_slice(), - ) - .await?; + membership + .insert_auto_membership(channel_id, pubkey_bytes, channel_created_by.as_slice()) + .await?; } // If not still_absent: concurrent add observed — membership preserved. } @@ -3392,22 +3400,22 @@ async fn commit_participant_join( // an open channel whose `join_role` is `member`. A non-member of a // private scope has lost the membership the pre-join check saw (the // auto-add above already made a private huddle's joiner a member). - let scopes = if parent_channel_id == channel_id { - vec![channel_id] - } else { - vec![channel_id, parent_channel_id] - }; - for scope in scopes { - let role = - buzz_db::channel_members::active_role_in_transaction(&mut tx, scope, pubkey_bytes) - .await?; + for scope in locked_channels.iter().copied() { + let role = buzz_db::channel_members::active_role_in_transaction( + membership.tx(), + scope, + pubkey_bytes, + ) + .await?; let refusal = match role.as_deref() { Some(role) => (!crate::handlers::ingest::role_may_post(Some(role))) .then_some(JoinCommitError::WriteRoleLost), None => { - let settings = - buzz_db::channel_members::channel_write_settings_in_transaction(&mut tx, scope) - .await?; + let settings = buzz_db::channel_members::channel_write_settings_in_transaction( + membership.tx(), + scope, + ) + .await?; match settings { Some((visibility, _)) if visibility != "open" => { Some(JoinCommitError::MembershipLost) @@ -9524,7 +9532,7 @@ mod tests { // Exercises the concurrent-external-add path in the AutoAddRequired branch // of `commit_participant_join`. An external transaction inserts the // membership row while our transaction is paused at `before_membership_lock` - // — just before `acquire_channel_membership_lock_in_transaction`. When our + // — just before `lock_channel_memberships_in_transaction`. When our // transaction resumes: // 1. It acquires the channel membership lock. // 2. Re-reads membership — the external insert is committed and visible. 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 0000000000..806e740b9a --- /dev/null +++ b/crates/buzz-relay/src/handlers/edit_metadata_postgres_tests.rs @@ -0,0 +1,507 @@ +//! 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 inside the write stores nothing and changes nothing, +/// and the same event succeeds once the failure is gone. A test trigger makes +/// the topic UPDATE fail after the event row and the `about` change were +/// written in the same transaction. 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; + + // Each test runs in its own database, so this trigger affects only it. + for statement in [ + "CREATE FUNCTION edit_atomic_fail() RETURNS trigger LANGUAGE plpgsql AS \ + $$ BEGIN RAISE EXCEPTION 'injected topic failure'; END $$", + "CREATE TRIGGER edit_atomic_fail BEFORE UPDATE ON channels FOR EACH ROW \ + WHEN (NEW.topic = 'fail-topic') EXECUTE FUNCTION edit_atomic_fail()", + ] { + sqlx::query(statement) + .execute(state.db.pool()) + .await + .expect("install failure trigger"); + } + + let event = EventBuilder::new(Kind::Custom(9002), "") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["about", "changed"]).unwrap(), + Tag::parse(["topic", "fail-topic"]).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.description, before.description, + "{attempt}: about is unchanged" + ); + assert_eq!(after.topic, before.topic, "{attempt}: topic is unchanged"); + assert_eq!( + count_kind(&state, &tenant, channel_id, 9002).await, + 0, + "{attempt}" + ); + assert_eq!( + count_kind(&state, &tenant, channel_id, 40099).await, + notices_before, + "{attempt}: no system message announces a change that did not happen" + ); + } + + // With the failure gone, the same signed event applies in full. + sqlx::query("DROP TRIGGER edit_atomic_fail ON channels") + .execute(state.db.pool()) + .await + .expect("drop failure trigger"); + let retry = ingest_event(&state, &tenant, event, http_auth(&owner)) + .await + .expect("retry"); + assert!(retry.accepted, "{:?}", retry.message); + assert!(retry.message.is_empty(), "{:?}", retry.message); + let after = state + .db + .get_channel(tenant.community(), channel_id) + .await + .expect("channel"); + assert_eq!(after.description.as_deref(), Some("changed")); + assert_eq!(after.topic.as_deref(), Some("fail-topic")); + assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 1); + assert_eq!( + count_kind(&state, &tenant, channel_id, 40099).await, + notices_before + 1 + ); +} + +/// 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 + ); +} + +/// Resending a committed archive or unarchive is a duplicate, not a refusal. +/// After the first send commits, the channel is already in the requested +/// state, so an archive-state check before the duplicate check would refuse +/// the identical resend (a lost response would look like a failure). +/// Mutation: check the archive state before the duplicate check in the +/// atomic write → RED ("channel is not archived"). +#[tokio::test] +#[ignore = "requires Postgres"] +async fn resent_archive_and_unarchive_are_duplicates() { + let state = state().await; + let tenant = community(&state, "edit-atomic-archive-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; + + for (value, archived) in [("true", true), ("false", false)] { + let event = EventBuilder::new(Kind::Custom(9002), "") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["archived", value]).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, "archived={value}: {:?}", first.message); + assert!( + first.message.is_empty(), + "archived={value}: {:?}", + first.message + ); + + let resend = ingest_event(&state, &tenant, event, http_auth(&owner)) + .await + .expect("resend"); + assert!(resend.accepted, "archived={value}: {:?}", resend.message); + assert_eq!(resend.message, "duplicate:", "archived={value}"); + + let channel = state + .db + .get_channel(tenant.community(), channel_id) + .await + .expect("channel"); + assert_eq!(channel.archived_at.is_some(), archived, "archived={value}"); + } + assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 2); + assert_eq!( + count_kind(&state, &tenant, channel_id, 40099).await, + notices_before + 2 + ); +} + +/// A kind:9002 `join_role` change goes through the same transaction as the +/// other settings: it is stored, and it is announced by one +/// `join_role_changed` system message. Mutation: drop the `join_role` arm +/// from the edit → RED (the event changes nothing). +#[tokio::test] +#[ignore = "requires Postgres"] +async fn join_role_edit_is_applied_and_announced() { + let state = state().await; + let tenant = community(&state, "edit-atomic-join-role").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(["join_role", "guest"]).unwrap(), + ]) + .sign_with_keys(&owner) + .expect("sign 9002"); + let result = ingest_event(&state, &tenant, event, http_auth(&owner)) + .await + .expect("ingest"); + assert!(result.accepted, "{:?}", result.message); + + let after = state + .db + .get_channel(tenant.community(), channel_id) + .await + .expect("channel"); + assert_eq!(after.join_role, "guest"); + let notices = state + .db + .query_events(&EventQuery { + kinds: Some(vec![40099]), + channel_id: Some(channel_id), + ..EventQuery::for_community(tenant.community()) + }) + .await + .expect("query events"); + assert_eq!(notices.len(), notices_before + 1); + let announced = notices + .iter() + .filter_map(|stored| serde_json::from_str::(&stored.event.content).ok()) + .any(|content| content["type"] == "join_role_changed" && content["join_role"] == "guest"); + assert!(announced, "one join_role_changed system message"); +} + +/// Sign a kind:9002 that sets `archived` to `value` on `channel_id`. +fn archive_event(owner: &Keys, channel_id: Uuid, value: &str) -> nostr::Event { + EventBuilder::new(Kind::Custom(9002), "") + .tags([ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["archived", value]).unwrap(), + ]) + .sign_with_keys(owner) + .expect("sign 9002") +} + +/// A resend checked after its first copy committed is a duplicate. +/// Validation runs on the new archive state and must not refuse the resend; +/// the write must then report it as a duplicate, with no change and no +/// notice. This is what a resend sees when it is validated after a racing +/// first copy commits. Before, validation read the archive state and refused +/// the resend ("channel is not archived"). Mutation: refuse a restore of an +/// active channel during validation → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn resend_validated_after_the_first_commit_is_a_duplicate() { + let state = state().await; + let tenant = community(&state, "edit-atomic-resend-window").await; + let owner = Keys::generate(); + let channel_id = create_channel(&state, &tenant, &owner, &[]).await; + + for value in ["true", "false"] { + let event = archive_event(&owner, channel_id, value); + let first = ingest_event(&state, &tenant, event.clone(), http_auth(&owner)) + .await + .expect("first send"); + assert!( + first.accepted && first.message.is_empty(), + "{:?}", + first.message + ); + let notices = count_kind(&state, &tenant, channel_id, 40099).await; + + crate::handlers::side_effects::validate_admin_event(&tenant, 9002, &event, &state) + .await + .unwrap_or_else(|e| panic!("archived={value}: resend refused in validation: {e}")); + let (_, inserted) = + crate::handlers::side_effects::persist_edit_metadata(&tenant, &event, &state) + .await + .expect("resend write"); + assert!( + !inserted, + "archived={value}: the resend changed the channel" + ); + assert_eq!( + count_kind(&state, &tenant, channel_id, 40099).await, + notices, + "archived={value}: the resend added a notice" + ); + } + assert_eq!(count_kind(&state, &tenant, channel_id, 9002).await, 2); +} + +/// Two copies of one restore, sent at the same time, change the channel +/// once. Both are held at the channel's TTL lock until both wait there, so +/// both pass validation before either commits. One applies the restore; the +/// other is reported as a duplicate, with no second notice. The case +/// where a copy validates after the other commits is covered by +/// `resend_validated_after_the_first_commit_is_a_duplicate`. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn concurrent_copies_of_a_restore_apply_once() { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect test pool"); + let state = crate::state::tests::test_state_with_database_pool(pool.clone()).await; + let tenant = community(&state, "edit-atomic-restore-race").await; + let owner = Keys::generate(); + let channel_id = create_channel(&state, &tenant, &owner, &[]).await; + edit_channel( + &state, + &tenant, + &owner, + channel_id, + &[&["archived", "true"]], + ) + .await + .expect("archive"); + let notices = count_kind(&state, &tenant, channel_id, 40099).await; + + let lock_key = format!( + "buzz_channel_ttl:{}:{channel_id}", + tenant.community().as_uuid() + ); + let mut holder = pool.begin().await.expect("begin lock holder"); + sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))") + .bind(&lock_key) + .execute(&mut *holder) + .await + .expect("hold the TTL lock"); + + let event = archive_event(&owner, channel_id, "false"); + let tasks: Vec<_> = (0..2) + .map(|_| { + let state = std::sync::Arc::clone(&state); + let tenant = tenant.clone(); + let event = event.clone(); + let auth = http_auth(&owner); + tokio::spawn(async move { ingest_event(&state, &tenant, event, auth).await }) + }) + .collect(); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + let waiting: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_locks WHERE locktype = 'advisory' AND NOT granted \ + AND ((classid::bigint << 32) | objid::bigint) = hashtextextended($1, 0)", + ) + .bind(&lock_key) + .fetch_one(&pool) + .await + .expect("read lock waiters"); + if waiting == 2 { + break; + } + assert!( + tasks.iter().all(|task| !task.is_finished()), + "a copy finished without waiting for the lock" + ); + assert!( + std::time::Instant::now() < deadline, + "both copies never waited for the TTL lock" + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + holder.commit().await.expect("release the TTL lock"); + + let mut messages = Vec::new(); + for task in tasks { + let result = tokio::time::timeout(std::time::Duration::from_secs(10), task) + .await + .expect("a copy returns after the lock is released") + .expect("task must not panic") + .expect("both copies are accepted"); + assert!(result.accepted, "{:?}", result.message); + messages.push(result.message); + } + messages.sort(); + assert_eq!(messages, ["", "duplicate:"]); + 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, 40099).await, + notices + 1 + ); +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index ce0c10a7bf..1b162cc329 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -3031,17 +3031,15 @@ async fn ingest_event_inner( } if channel_id.is_some() { - // Allow kind:9002 with archived=false (unarchive operation) - let is_unarchive = kind_u32 == KIND_NIP29_EDIT_METADATA - && event.tags.iter().any(|t| { - let parts = t.as_slice(); - parts.len() >= 2 && parts[0] == "archived" && parts[1] == "false" - }); + // kind:9002 is not checked here: its atomic write decides the + // archive state under the channel row lock, after its duplicate + // check, so a resend that races the first commit is a duplicate. + let is_edit_metadata = kind_u32 == KIND_NIP29_EDIT_METADATA; // A member may still leave an archived channel by themselves. let is_self_departure = crate::handlers::side_effects::is_self_departure(&event); - if !is_unarchive && !is_self_departure { + if !is_edit_metadata && !is_self_departure { if let Some(channel) = &channel_row { if channel.archived_at.is_some() { return Err(match target_denial { @@ -3523,6 +3521,19 @@ 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. + // A refusal decided under the lock (for example, restoring a channel + // that is not archived) is the client's error; anything else is ours. + crate::handlers::side_effects::persist_edit_metadata(tenant, &event, state) + .await + .map_err(|e| match e.downcast_ref::() { + Some(buzz_db::DbError::AccessDenied(reason)) => { + IngestError::Rejected(format!("invalid: {reason}")) + } + _ => 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 +3625,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 dd4db5e986..222512aa0c 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 2d29c8c124..c581405591 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, @@ -578,20 +579,17 @@ pub async fn validate_admin_event( let actor_bytes = event.pubkey.to_bytes().to_vec(); - // Reject mutations on archived channels — except kind:9002 with archived=false - // (unarchive), which must be allowed through so the channel can be restored, - // and a member leaving by themselves (still subject to the last-owner rule). + // Reject mutations on archived channels, except a member leaving by + // themselves (still subject to the last-owner rule). kind:9002 is not + // checked here: its atomic write decides the archive state under the + // channel row lock, after its duplicate check, so a resend that races + // the first commit is a duplicate and not a refusal. let channel = state .db .get_channel_for_event_write(tenant.community(), channel_id) .await .map_err(|_| anyhow::anyhow!("channel not found"))?; - let is_unarchive_request = kind == 9002 - && event.tags.iter().any(|t| { - let parts = t.as_slice(); - parts.len() >= 2 && parts[0] == "archived" && parts[1] == "false" - }); - if channel.archived_at.is_some() && !is_unarchive_request && !is_self_departure(event) { + if kind != 9002 && channel.archived_at.is_some() && !is_self_departure(event) { return Err(anyhow::anyhow!("channel is archived")); } @@ -974,14 +972,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 +981,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 +1707,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?; + 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. `["t", ""]` alone clears it. + if let Some(labels) = channel_labels_from_tags(event) { + edit.columns.labels = Some(labels.map_err(|e| anyhow::anyhow!("{e}"))?); + } - // 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 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"); } } } - - // All `t` tags together form the new label set, so they are applied - // once after the loop. `["t", ""]` alone clears the set. - 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?; + 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",