diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 45b1cc94f7..e456f46dfb 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -10,7 +10,7 @@ use uuid::Uuid; use crate::error::{DbError, Result}; use crate::Db; -use buzz_core::CommunityId; +use buzz_core::{CommunityId, StoredEvent}; use buzz_datastore_tracing::datastore_span; // Re-export the canonical enum definitions from buzz-core. @@ -200,10 +200,133 @@ pub async fn create_channel_with_id( ttl_seconds: Option, labels: &[String], ) -> Result<(ChannelRecord, bool)> { - create_channel_with_members( + let mut tx = crate::begin_community_write_transaction( pool, community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + + let was_created = insert_channel_rows( + &mut tx, channel_id, + NewChannel { + name, + channel_type, + visibility, + description, + created_by, + ttl_seconds, + labels, + members: &[], + }, + ) + .await?; + + let row = sqlx::query( + r#" + SELECT id, name, channel_type::text AS channel_type, visibility::text AS visibility, + description, canvas, + created_by, created_at, updated_at, archived_at, deleted_at, + nip29_group_id, topic_required, max_members, + topic, topic_set_by, topic_set_at, + purpose, purpose_set_by, purpose_set_at, + ttl_seconds, ttl_deadline, labels + FROM channels WHERE community_id = $1 AND id = $2 + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_one(tx.conn()) + .await?; + + let record = row_to_channel_record(row)?; + tx.commit().await?; + Ok((record, was_created)) +} + +/// A channel to create: the fields a kind:9007 sets. +pub struct NewChannel<'a> { + /// Display name, canonicalized before storage. + pub name: &'a str, + /// Stream, forum, DM, … + pub channel_type: ChannelType, + /// Open or private. + pub visibility: ChannelVisibility, + /// The `about` text. + pub description: Option<&'a str>, + /// The creator's 32-byte pubkey; becomes the owner. + pub created_by: &'a [u8], + /// Ephemeral channel lifetime. + pub ttl_seconds: Option, + /// Labels validated by `buzz_core::channel::parse_channel_labels`. + pub labels: &'a [String], + /// 32-byte pubkeys added as ordinary members invited by the creator. The + /// caller has authorized each addition. + pub members: &'a [Vec], +} + +/// What [`Db::create_channel_with_event`] did. +#[derive(Debug)] +pub enum ChannelCreate { + /// The channel, its members and the create event committed together. + Created(Box), + /// The UUID was already taken and nothing was written. `resent` is true + /// when the stored create that took it is this same event. + Exists { + /// This exact event is the channel's stored create. + resent: bool, + }, +} + +/// See [`Db::create_channel_with_event`]. +pub(crate) async fn create_channel_with_event( + pool: &PgPool, + push_enqueue: Option, + community_id: CommunityId, + channel_id: Uuid, + channel: NewChannel<'_>, + event: &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); + + if !insert_channel_rows(&mut tx, channel_id, channel).await? { + let resent = sqlx::query( + "SELECT 1 FROM events WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL LIMIT 1", + ) + .bind(community_id.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .fetch_optional(tx.conn()) + .await? + .is_some(); + return Ok(ChannelCreate::Exists { resent }); + } + + // 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 { + crate::insert_mentions_in_transaction(&mut tx, event, Some(channel_id)).await?; + } + tx.commit().await?; + Ok(ChannelCreate::Created(Box::new(stored))) +} + +/// Inserts the channel row, its owner and its initial members on `tx`. +/// Returns `false`, having written nothing, when `channel_id` already exists. +async fn insert_channel_rows( + tx: &mut crate::AdmittedTx, + channel_id: Uuid, + channel: NewChannel<'_>, +) -> Result { + let NewChannel { name, channel_type, visibility, @@ -211,29 +334,9 @@ pub async fn create_channel_with_id( created_by, ttl_seconds, labels, - &[], - ) - .await -} - -/// [`create_channel_with_id`], also adding `members` (32-byte pubkeys) as -/// ordinary members invited by the creator, in the transaction that creates the -/// channel. The caller has authorized each addition. Nothing is added when the -/// channel already exists. -#[allow(clippy::too_many_arguments)] -pub async fn create_channel_with_members( - pool: &PgPool, - community_id: CommunityId, - channel_id: Uuid, - name: &str, - channel_type: ChannelType, - visibility: ChannelVisibility, - description: Option<&str>, - created_by: &[u8], - ttl_seconds: Option, - labels: &[String], - members: &[Vec], -) -> Result<(ChannelRecord, bool)> { + members, + } = channel; + let community_id = tx.community(); if created_by.len() != 32 { return Err(DbError::InvalidData(format!( "pubkey must be 32 bytes, got {}", @@ -252,13 +355,6 @@ pub async fn create_channel_with_members( return Err(DbError::InvalidData("channel name is required".into())); } - let mut tx = crate::begin_community_write_transaction( - pool, - community_id, - crate::observability::WriterOperation::EventWrite, - ) - .await?; - let rows_affected = sqlx::query( r#" INSERT INTO channels (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline, labels) @@ -327,26 +423,7 @@ pub async fn create_channel_with_members( } } - let row = sqlx::query( - r#" - SELECT id, name, channel_type::text AS channel_type, visibility::text AS visibility, - description, canvas, - created_by, created_at, updated_at, archived_at, deleted_at, - nip29_group_id, topic_required, max_members, - topic, topic_set_by, topic_set_at, - purpose, purpose_set_by, purpose_set_at, - ttl_seconds, ttl_deadline, labels - FROM channels WHERE community_id = $1 AND id = $2 - "#, - ) - .bind(community_id.as_uuid()) - .bind(channel_id) - .fetch_one(tx.conn()) - .await?; - - let record = row_to_channel_record(row)?; - tx.commit().await?; - Ok((record, was_created)) + Ok(was_created) } /// Fetches a channel record by `(community_id, id)`. Returns `ChannelNotFound` if missing or deleted. @@ -1014,38 +1091,28 @@ impl Db { .await } - /// Creates a channel with a client-supplied UUID and an initial label set. + /// Creates `channel_id` from its kind:9007 `event`: the channel, its + /// owner, its initial members and the event itself commit in one + /// transaction, so a channel row always has its create event beside it. /// - /// The same as [`Db::create_channel_with_id`], plus `labels`, which the - /// caller has validated with `buzz_core::channel::parse_channel_labels`, - /// and initial `members` (see [`create_channel_with_members`]). - #[allow(clippy::too_many_arguments)] - #[datastore_span(name = "create_labeled_channel_with_id", system = "postgresql")] - pub async fn create_labeled_channel_with_id( + /// A concurrent copy of the same create waits on the first one's channel + /// insert, then sees either the committed event (`Exists { resent: true }`) + /// or, if the first rolled back, an untaken UUID. + #[datastore_span(name = "create_channel_with_event", system = "postgresql")] + pub async fn create_channel_with_event( &self, community_id: CommunityId, channel_id: Uuid, - name: &str, - channel_type: ChannelType, - visibility: ChannelVisibility, - description: Option<&str>, - created_by: &[u8], - ttl_seconds: Option, - labels: &[String], - members: &[Vec], - ) -> Result<(ChannelRecord, bool)> { - create_channel_with_members( + channel: NewChannel<'_>, + event: &nostr::Event, + ) -> Result { + create_channel_with_event( &self.pool, + self.push_enqueue.clone(), community_id, channel_id, - name, - channel_type, - visibility, - description, - created_by, - ttl_seconds, - labels, - members, + channel, + event, ) .await } @@ -1488,4 +1555,210 @@ mod postgres_tests { "reaper should carry the archived row's community id and host" ); } + + fn create_event(creator: &Keys, channel_id: Uuid) -> nostr::Event { + nostr::EventBuilder::new(nostr::Kind::Custom(9007), "") + .tags([ + nostr::Tag::parse(["h", &channel_id.to_string()]).expect("h tag"), + nostr::Tag::parse(["name", "start"]).expect("name tag"), + ]) + .sign_with_keys(creator) + .expect("sign 9007") + } + + fn new_channel<'a>(created_by: &'a [u8], members: &'a [Vec]) -> NewChannel<'a> { + NewChannel { + name: "start", + channel_type: ChannelType::Stream, + visibility: ChannelVisibility::Private, + description: None, + created_by, + ttl_seconds: None, + labels: &[], + members, + } + } + + /// Holds a create open on its own transaction (channel row and event both + /// written, not committed) while the same create runs again beside it. + async fn resend_beside_open_create(commit_first: bool) -> ChannelCreate { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(make_test_community(&pool).await); + let creator = Keys::generate(); + let creator_bytes = creator.public_key().to_bytes().to_vec(); + let members = vec![random_pubkey()]; + let channel_id = Uuid::new_v4(); + let event = create_event(&creator, channel_id); + + let mut first = crate::begin_community_write_transaction( + &pool, + community, + crate::observability::WriterOperation::EventWrite, + ) + .await + .expect("begin first create"); + assert!(insert_channel_rows( + &mut first, + channel_id, + new_channel(&creator_bytes, &members) + ) + .await + .expect("first create's rows")); + crate::event::insert_event_with_thread_metadata_tx( + &mut first, + &event, + Some(channel_id), + None, + ) + .await + .expect("first create's event"); + + let resend = tokio::spawn({ + let (pool, event, creator_bytes, members) = ( + pool.clone(), + event.clone(), + creator_bytes.clone(), + members.clone(), + ); + async move { + create_channel_with_event( + &pool, + None, + community, + channel_id, + new_channel(&creator_bytes, &members), + &event, + ) + .await + } + }); + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + assert!( + !resend.is_finished(), + "the resend waits for the open create instead of answering early" + ); + + if commit_first { + first.commit().await.expect("commit first create"); + } else { + drop(first); + } + let outcome = resend.await.expect("resend task").expect("resend outcome"); + let roster = sqlx::query_scalar::<_, Vec>( + "SELECT pubkey FROM channel_members WHERE community_id = $1 AND channel_id = $2 \ + AND removed_at IS NULL ORDER BY pubkey", + ) + .bind(community.as_uuid()) + .bind(channel_id) + .fetch_all(&pool) + .await + .expect("roster"); + let mut expected = vec![creator_bytes, members[0].clone()]; + expected.sort(); + assert_eq!( + roster, expected, + "the channel has its owner and member once" + ); + outcome + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn a_resend_beside_an_open_create_confirms_it_once_committed() { + let outcome = resend_beside_open_create(true).await; + assert!( + matches!(outcome, ChannelCreate::Exists { resent: true }), + "{outcome:?}" + ); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn a_resend_beside_a_failed_create_creates_the_channel() { + let outcome = resend_beside_open_create(false).await; + assert!(matches!(outcome, ChannelCreate::Created(_)), "{outcome:?}"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn a_create_whose_event_fails_leaves_no_channel() { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(make_test_community(&pool).await); + let creator = Keys::generate(); + let creator_bytes = creator.public_key().to_bytes().to_vec(); + let channel_id = Uuid::new_v4(); + // An ephemeral kind is refused by the event insert, after the rows. + let unstorable = nostr::EventBuilder::new(nostr::Kind::Custom(20_000), "") + .sign_with_keys(&creator) + .expect("sign"); + let members = [random_pubkey()]; + create_channel_with_event( + &pool, + None, + community, + channel_id, + new_channel(&creator_bytes, &members), + &unstorable, + ) + .await + .expect_err("the event insert fails"); + assert!( + get_channel(&pool, community, channel_id).await.is_err(), + "the channel rolls back with its event" + ); + let retry = create_channel_with_event( + &pool, + None, + community, + channel_id, + new_channel(&creator_bytes, &members), + &create_event(&creator, channel_id), + ) + .await + .expect("retry"); + assert!(matches!(retry, ChannelCreate::Created(_)), "{retry:?}"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn another_create_for_a_taken_channel_writes_nothing() { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(make_test_community(&pool).await); + let creator = Keys::generate(); + let creator_bytes = creator.public_key().to_bytes().to_vec(); + let channel_id = Uuid::new_v4(); + let first = create_channel_with_event( + &pool, + None, + community, + channel_id, + new_channel(&creator_bytes, &[]), + &create_event(&creator, channel_id), + ) + .await + .expect("first create"); + assert!(matches!(first, ChannelCreate::Created(_)), "{first:?}"); + + let other = Keys::generate(); + let other_bytes = other.public_key().to_bytes().to_vec(); + let intruder = random_pubkey(); + let second = create_channel_with_event( + &pool, + None, + community, + channel_id, + new_channel(&other_bytes, std::slice::from_ref(&intruder)), + &create_event(&other, channel_id), + ) + .await + .expect("second create"); + assert!( + matches!(second, ChannelCreate::Exists { resent: false }), + "{second:?}" + ); + let members = crate::channel_members::get_members(&pool, community, channel_id) + .await + .expect("members"); + assert_eq!(members.len(), 1, "only the first creator is a member"); + } } diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index a10aa4deb6..0b2e8d8f09 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -2381,6 +2381,18 @@ fn validate_canvas_future_timestamp(event_ts: i64, now: i64) -> Result<(), &'sta Ok(()) } +/// A validated kind:9007 whose channel is created when the event is stored. +struct ChannelCreateRequest { + channel_id: Uuid, + name: String, + channel_type: buzz_db::channel::ChannelType, + visibility: buzz_db::channel::ChannelVisibility, + description: Option, + ttl_seconds: Option, + labels: Vec, + members: Vec>, +} + async fn ingest_event_inner( state: &Arc, tracer: &Arc, @@ -3050,8 +3062,9 @@ async fn ingest_event_inner( .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; } - // Track pre-created channel UUID for compensation on insert failure. - let mut pre_created_channel: Option = None; + // A kind:9007 with an `h` tag creates its channel when it is stored, in the + // same transaction (see the storage dispatch below). + let mut channel_create: Option = None; if kind_u32 == KIND_NIP29_CREATE_GROUP { // Validate name tag is present and non-empty before any DB work. @@ -3112,66 +3125,24 @@ async fn ingest_event_inner( .unwrap_or_default(); if let Some(client_uuid) = channel_id { - let name = create_name.unwrap_or_default(); - let name = buzz_core::channel::canonical_channel_name(&name); - - let description = event.tags.iter().find_map(|t| { - if t.kind().to_string() == "about" { - t.content().map(|s| s.to_string()) - } else { - None - } - }); - - let ttl_seconds = super::resolve_ttl(&event, state.config.ephemeral_ttl_override); - - let actor_bytes = event.pubkey.to_bytes().to_vec(); let members = super::side_effects::initial_members(&event) .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; - let (_, was_created) = state - .db - .create_labeled_channel_with_id( - tenant.community(), - client_uuid, - name, - channel_type, - visibility, - description.as_deref(), - &actor_bytes, - ttl_seconds, - &labels, - &members, - ) - .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))?; - - if !was_created { - // Resending a create whose OK was lost confirms it. Anything - // else for an existing UUID is refused before storage, so it - // never runs the create's side effects again. - let resent = state - .db - .get_event_by_id_for_event_write(tenant.community(), event.id.as_bytes()) - .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))? - .is_some(); - return Ok(IngestResult { - event_id: event_id_hex, - accepted: resent, - message: if resent { - "duplicate:".into() + channel_create = Some(ChannelCreateRequest { + channel_id: client_uuid, + name: create_name.unwrap_or_default(), + channel_type, + visibility, + description: event.tags.iter().find_map(|t| { + if t.kind().to_string() == "about" { + t.content().map(|s| s.to_string()) } else { - "duplicate: channel already exists".into() - }, - }); - } - pre_created_channel = Some(client_uuid); - metrics::counter!( - "buzz_channels_created_total", - "community" => tenant.host().to_owned(), - "type" => channel_type.to_string() - ) - .increment(1); + None + } + }), + ttl_seconds: super::resolve_ttl(&event, state.config.ephemeral_ttl_override), + labels, + members, + }); } } @@ -3422,7 +3393,49 @@ async fn ingest_event_inner( }; let workflow_deletion = crate::handlers::side_effects::is_workflow_deletion(&event); - let (stored_event, was_inserted) = if workflow_deletion { + let (stored_event, was_inserted) = if let Some(create) = channel_create.as_ref() { + let creator = event.pubkey.to_bytes(); + let channel = buzz_db::channel::NewChannel { + name: &create.name, + channel_type: create.channel_type, + visibility: create.visibility, + description: create.description.as_deref(), + created_by: &creator, + ttl_seconds: create.ttl_seconds, + labels: &create.labels, + members: &create.members, + }; + match state + .db + .create_channel_with_event(tenant.community(), create.channel_id, channel, &event) + .await + .map_err(|e| IngestError::Internal(format!("error: {e}")))? + { + buzz_db::channel::ChannelCreate::Created(stored) => { + metrics::counter!( + "buzz_channels_created_total", + "community" => tenant.host().to_owned(), + "type" => create.channel_type.to_string() + ) + .increment(1); + (*stored, true) + } + // Resending a create whose OK was lost confirms it. Anything else + // for a taken UUID is refused. Neither writes anything, so the + // create's side effects never run again. + buzz_db::channel::ChannelCreate::Exists { resent } => { + return Ok(IngestResult { + event_id: event_id_hex, + accepted: resent, + message: if resent { + "duplicate:".into() + } else { + "duplicate: channel already exists".into() + }, + }); + } + } + } else if workflow_deletion { // A single commit owns public acceptance, domain mutation, and dispatch. // Failure rolls everything back; identical concurrent requests cannot // divide insertion and repair ownership between two relay workers. @@ -3502,18 +3515,6 @@ async fn ingest_event_inner( { Ok(result) => result, Err(e) => { - // Compensate: if we pre-created a channel for kind:9007, - // soft-delete it so no orphaned channel row remains. - if let Some(ch_id) = pre_created_channel { - if let Err(re) = state - .db - .soft_delete_channel(tenant.community(), ch_id) - .await - { - warn!(event_id = %event_id_hex, "channel compensation failed: {re}"); - } - state.invalidate_channel_deleted(tenant); - } return Err(match e { buzz_db::DbError::AuthEventRejected => { IngestError::Rejected("invalid: AUTH events cannot be stored".into()) diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 880369cb88..38d035910b 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use nostr::{Event, EventBuilder, Kind, Tag}; -use tracing::{info, warn}; +use tracing::{error, info, warn}; use uuid::Uuid; use buzz_core::kind::{ @@ -2174,8 +2174,8 @@ async fn handle_create_group( // Creator becomes owner — evict any stale negative membership lookup. state.invalidate_membership(tenant, channel.id, &actor_bytes); - // Open channels appear in everyone's accessible set; private channels only - // affect the creator (the sole initial member). + // Open channels appear in everyone's accessible set; a private channel only + // affects its initial members, whose entries the invalidations here drop. if visibility == buzz_db::channel::ChannelVisibility::Open { state.invalidate_all_accessible_channels(tenant); } @@ -2210,7 +2210,8 @@ async fn handle_create_group( } // Ingest added these in the create's transaction; announce them as a - // kind:9000 would. + // kind:9000 would. A failed announcement for one member must not cost the + // rest theirs: a resend writes nothing, so nothing would retry them. let members = if members_added { initial_members(event)? } else { @@ -2218,7 +2219,7 @@ async fn handle_create_group( }; for member in members { state.invalidate_membership(tenant, channel.id, &member); - emit_system_message( + if let Err(e) = emit_system_message( tenant, state, channel.id, @@ -2229,7 +2230,10 @@ async fn handle_create_group( }), chrono::Utc::now(), ) - .await?; + .await + { + error!(channel = %channel.id, error = %e, "member_joined system message failed"); + } if let Err(e) = emit_membership_notification( tenant, state,