diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index cf8d15094e..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,6 +200,143 @@ pub async fn create_channel_with_id( ttl_seconds: Option, labels: &[String], ) -> Result<(ChannelRecord, bool)> { + 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, + description, + created_by, + ttl_seconds, + labels, + members, + } = channel; + let community_id = tx.community(); if created_by.len() != 32 { return Err(DbError::InvalidData(format!( "pubkey must be 32 bytes, got {}", @@ -218,13 +355,6 @@ pub async fn create_channel_with_id( 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) @@ -266,28 +396,34 @@ pub async fn create_channel_with_id( .bind(created_by) .execute(tx.conn()) .await?; + + for member in members + .iter() + .filter(|member| member.as_slice() != created_by) + { + if member.len() != 32 { + return Err(DbError::InvalidData(format!( + "pubkey must be 32 bytes, got {}", + member.len() + ))); + } + sqlx::query( + r#" + INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by) + VALUES ($1, $2, $3, 'member', $4) + ON CONFLICT (community_id, channel_id, pubkey) DO NOTHING + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(member) + .bind(created_by) + .execute(tx.conn()) + .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)) + Ok(was_created) } /// Fetches a channel record by `(community_id, id)`. Returns `ChannelNotFound` if missing or deleted. @@ -955,35 +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`. - #[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], - ) -> Result<(ChannelRecord, bool)> { - create_channel_with_id( + 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, + channel, + event, ) .await } @@ -1426,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/create_with_members_postgres_tests.rs b/crates/buzz-relay/src/handlers/create_with_members_postgres_tests.rs new file mode 100644 index 0000000000..de8327ddb4 --- /dev/null +++ b/crates/buzz-relay/src/handlers/create_with_members_postgres_tests.rs @@ -0,0 +1,251 @@ +//! Postgres-backed tests for starting a channel with its members: a kind:9007 +//! that adds its `p` tags as members, and a creator's retry of it. + +use std::sync::Arc; + +use buzz_core::TenantContext; +use buzz_db::event::EventQuery; +use nostr::{Event, EventBuilder, Keys, Kind, Tag}; +use uuid::Uuid; + +use crate::handlers::group_state_postgres_tests::{community, group_state_tags, http_auth, state}; +use crate::handlers::ingest::{ingest_event, IngestError, IngestResult}; +use crate::state::AppState; + +fn create_event(creator: &Keys, channel_id: Uuid, members: &[&Keys]) -> Event { + let mut tags = vec![ + Tag::parse(["h", &channel_id.to_string()]).unwrap(), + Tag::parse(["name", "start"]).unwrap(), + Tag::parse(["visibility", "private"]).unwrap(), + ]; + for member in members { + tags.push(Tag::parse(["p", &member.public_key().to_hex()]).unwrap()); + } + EventBuilder::new(Kind::Custom(9007), "") + .tags(tags) + .sign_with_keys(creator) + .expect("sign 9007") +} + +fn message_event(author: &Keys, channel_id: Uuid) -> Event { + EventBuilder::new(Kind::Custom(9), "hello") + .tags([Tag::parse(["h", &channel_id.to_string()]).unwrap()]) + .sign_with_keys(author) + .expect("sign message") +} + +async fn ingest( + state: &Arc, + tenant: &TenantContext, + event: Event, + keys: &Keys, +) -> Result { + ingest_event(state, tenant, event, http_auth(keys)).await +} + +async fn roles( + state: &AppState, + tenant: &TenantContext, + channel_id: Uuid, +) -> Vec<(Vec, String)> { + let mut roles: Vec<_> = state + .db + .get_members(tenant.community(), channel_id) + .await + .expect("members") + .into_iter() + .map(|member| (member.pubkey, member.role)) + .collect(); + roles.sort(); + roles +} + +/// The `p` tags join in the create's transaction, the roster lists them, and +/// each gets the "joined" notice a kind:9000 would post. +/// Mutation: drop `members` from the create → no agent membership → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn create_adds_its_p_tags_as_members() { + let state = state().await; + let tenant = community(&state, "create-members").await; + let (creator, agent) = (Keys::generate(), Keys::generate()); + let channel_id = Uuid::new_v4(); + + let result = ingest( + &state, + &tenant, + create_event(&creator, channel_id, &[&agent, &agent]), + &creator, + ) + .await + .expect("create"); + assert!(result.accepted, "{}", result.message); + + let mut expected = vec![ + ( + creator.public_key().to_bytes().to_vec(), + "owner".to_string(), + ), + (agent.public_key().to_bytes().to_vec(), "member".to_string()), + ]; + expected.sort(); + assert_eq!(roles(&state, &tenant, channel_id).await, expected); + + let roster = group_state_tags(&state, &tenant, channel_id, 39002).await; + assert!( + roster + .iter() + .any(|tag| tag[0] == "p" && tag[1] == agent.public_key().to_hex()), + "{roster:?}" + ); + + let notices = state + .db + .query_events(&EventQuery { + kinds: Some(vec![buzz_core::kind::KIND_SYSTEM_MESSAGE as i32]), + channel_id: Some(channel_id), + ..EventQuery::for_community(tenant.community()) + }) + .await + .expect("system messages"); + let joined = notices + .iter() + .filter(|stored| { + let content: serde_json::Value = + serde_json::from_str(&stored.event.content).unwrap_or_default(); + content["type"] == "member_joined" && content["target"] == agent.public_key().to_hex() + }) + .count(); + assert_eq!(joined, 1, "one joined notice for the agent"); + + // The agent can answer in the chat with no separate add. + let reply = ingest(&state, &tenant, message_event(&agent, channel_id), &agent) + .await + .expect("agent reply"); + assert!(reply.accepted, "{}", reply.message); +} + +/// An agent whose policy refuses third-party adds keeps the whole create out: +/// no channel is made. Mutation: skip the policy check → channel exists → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn create_respects_the_members_add_policy() { + let state = state().await; + let tenant = community(&state, "create-members-policy").await; + let (creator, agent) = (Keys::generate(), Keys::generate()); + let agent_bytes = agent.public_key().to_bytes().to_vec(); + state + .db + .ensure_user(tenant.community(), &agent_bytes) + .await + .expect("agent user"); + state + .db + .set_channel_add_policy(tenant.community(), &agent_bytes, "nobody") + .await + .expect("policy"); + let channel_id = Uuid::new_v4(); + + let result = ingest( + &state, + &tenant, + create_event(&creator, channel_id, &[&agent]), + &creator, + ) + .await; + match result { + Err(IngestError::Rejected(message)) => { + assert!(message.starts_with("invalid:"), "{message}") + } + Err(error) => panic!("expected a rejection, got {error:?}"), + Ok(result) => panic!("expected a rejection, got OK {}", result.message), + } + assert!(state + .db + .get_channel(tenant.community(), channel_id) + .await + .is_err()); +} + +/// Resending a create whose OK was lost is accepted; a different create for +/// the same UUID, even from its creator, is refused before its side effects, +/// so it can neither add nor announce anyone. +/// Mutations: refuse every existing UUID → resend not accepted → RED; check the +/// members' policies before recognizing a stored resend → RED; let the +/// creator's changed create through → stranger announced → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn resending_a_create_confirms_it() { + let state = state().await; + let tenant = community(&state, "create-retry").await; + let (creator, agent, stranger, other) = ( + Keys::generate(), + Keys::generate(), + Keys::generate(), + Keys::generate(), + ); + let channel_id = Uuid::new_v4(); + let create = create_event(&creator, channel_id, &[&agent]); + + let first = ingest(&state, &tenant, create.clone(), &creator) + .await + .expect("create"); + assert!(first.accepted, "{}", first.message); + let before = roles(&state, &tenant, channel_id).await; + let resent = ingest(&state, &tenant, create.clone(), &creator) + .await + .expect("resend"); + assert!(resent.accepted, "{}", resent.message); + + for (author, members) in [(&creator, vec![&agent, &stranger]), (&other, vec![])] { + let changed = ingest( + &state, + &tenant, + create_event(author, channel_id, &members), + author, + ) + .await + .expect("changed create"); + assert!(!changed.accepted); + assert_eq!(changed.message, "duplicate: channel already exists"); + } + + // The agent closing its adds after the create doesn't refuse the resend. + let agent_bytes = agent.public_key().to_bytes().to_vec(); + state + .db + .ensure_user(tenant.community(), &agent_bytes) + .await + .expect("agent user"); + state + .db + .set_channel_add_policy(tenant.community(), &agent_bytes, "nobody") + .await + .expect("policy"); + let resent_after_policy = ingest(&state, &tenant, create, &creator) + .await + .expect("resend after policy change"); + assert!( + resent_after_policy.accepted, + "{}", + resent_after_policy.message + ); + + assert_eq!(roles(&state, &tenant, channel_id).await, before); + let notices = state + .db + .query_events(&EventQuery { + kinds: Some(vec![buzz_core::kind::KIND_SYSTEM_MESSAGE as i32]), + channel_id: Some(channel_id), + ..EventQuery::for_community(tenant.community()) + }) + .await + .expect("system messages"); + assert!( + !notices.iter().any(|stored| stored + .event + .content + .contains(&stranger.public_key().to_hex())), + "the refused create announced nobody" + ); +} diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index de4a19217d..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,50 +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 members = super::side_effects::initial_members(&event) + .map_err(|e| IngestError::Rejected(format!("invalid: {e}")))?; + 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 { + None + } + }), + ttl_seconds: super::resolve_ttl(&event, state.config.ephemeral_ttl_override), + labels, + members, }); - - let ttl_seconds = super::resolve_ttl(&event, state.config.ephemeral_ttl_override); - - let actor_bytes = event.pubkey.to_bytes().to_vec(); - 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, - ) - .await - .map_err(|e| IngestError::Internal(format!("error: {e}")))?; - - if !was_created { - return Ok(IngestResult { - event_id: event_id_hex, - accepted: false, - message: "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); } } @@ -3406,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. @@ -3486,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/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 24c368a564..dd4db5e986 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -12,6 +12,8 @@ pub mod command_executor; pub mod community_provisioning; /// NIP-45 COUNT handler. pub mod count; +#[cfg(test)] +mod create_with_members_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 ca4a2304dd..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::{ @@ -540,8 +540,31 @@ pub async fn validate_admin_event( event: &Event, state: &Arc, ) -> anyhow::Result<()> { - // CREATE_GROUP doesn't need an existing channel — skip h-tag extraction + // CREATE_GROUP doesn't need an existing channel — skip h-tag extraction. + // Its initial members get the third-party add check a kind:9000 would. if kind == 9007 { + let members = initial_members(event)?; + // Ingest confirms a resend of a stored create without writing anything, + // so a policy changed since then must not refuse it. + if members.is_empty() + || state + .db + .get_event_by_id_for_event_write(tenant.community(), event.id.as_bytes()) + .await? + .is_some() + { + return Ok(()); + } + let actor_bytes = event.pubkey.to_bytes().to_vec(); + for member in members { + if let Some((policy, owner)) = state + .db + .get_agent_channel_policy(tenant.community(), &member) + .await? + { + channel_authz::decide_channel_add_policy(&policy, owner.as_deref(), &actor_bytes)?; + } + } return Ok(()); } @@ -2093,13 +2116,14 @@ async fn handle_create_group( // inaccessible, so the counter correctly records a new creation. For the // no-h-tag path, ingest never creates the channel, so this is the sole // increment. - let channel = if let Some(client_uuid) = extract_h_tag_channel(event) { + // `members_added`: ingest's create also added the event's `p` tags. + let (channel, members_added) = if let Some(client_uuid) = extract_h_tag_channel(event) { match state .db .get_channel_for_event_write(tenant.community(), client_uuid) .await { - Ok(ch) => ch, + Ok(ch) => (ch, true), Err(_) => { // Channel not found — shouldn't happen (ingest_event pre-created it), // but fall back to creation to stay resilient. @@ -2122,7 +2146,7 @@ async fn handle_create_group( "type" => channel_type.to_string() ) .increment(1); - ch + (ch, false) } } } else { @@ -2145,13 +2169,13 @@ async fn handle_create_group( "type" => channel_type.to_string() ) .increment(1); - ch + (ch, false) }; // 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); } @@ -2185,6 +2209,45 @@ async fn handle_create_group( warn!(channel = %channel.id, error = %e, "membership notification emission failed"); } + // Ingest added these in the create's transaction; announce them as a + // 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 { + Vec::new() + }; + for member in members { + state.invalidate_membership(tenant, channel.id, &member); + if let Err(e) = emit_system_message( + tenant, + state, + channel.id, + serde_json::json!({ + "type": "member_joined", + "actor": actor_hex, + "target": hex::encode(&member), + }), + chrono::Utc::now(), + ) + .await + { + error!(channel = %channel.id, error = %e, "member_joined system message failed"); + } + if let Err(e) = emit_membership_notification( + tenant, + state, + channel.id, + &member, + &actor_bytes, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + warn!(channel = %channel.id, error = %e, "membership notification emission failed"); + } + } + info!(channel_id = %channel.id, name = %name, "NIP-29 CREATE_GROUP processed"); Ok(()) } @@ -2653,6 +2716,38 @@ async fn handle_standard_deletion_event( Ok(()) } +/// Most members a kind:9007 may add besides its creator. +pub(crate) const MAX_INITIAL_MEMBERS: usize = 32; + +/// The members a kind:9007 adds besides its creator: its `p` tags, deduplicated. +/// They need the client-chosen channel UUID (`h`), so a retry of the same +/// create names the same channel. +pub(crate) fn initial_members(event: &Event) -> anyhow::Result>> { + let creator = event.pubkey.to_bytes(); + let mut members: Vec> = Vec::new(); + for tag in event + .tags + .iter() + .filter(|tag| tag.kind().to_string() == "p") + { + let member = tag + .content() + .and_then(|value| hex::decode(value).ok()) + .filter(|bytes| bytes.len() == 32) + .ok_or_else(|| anyhow::anyhow!("p tag must be a 32-byte hex pubkey"))?; + if member != creator && !members.contains(&member) { + members.push(member); + } + } + if members.len() > MAX_INITIAL_MEMBERS { + anyhow::bail!("a channel may start with at most {MAX_INITIAL_MEMBERS} members"); + } + if !members.is_empty() && extract_h_tag_channel(event).is_none() { + anyhow::bail!("initial members need an h tag"); + } + Ok(members) +} + /// Extract channel UUID from `h` tag (NIP-29 group ID). fn extract_h_tag_channel(event: &Event) -> Option { for tag in event.tags.iter() { @@ -3996,6 +4091,57 @@ pub async fn publish_nipia_unarchived( mod tests { use super::*; + fn create_with(creator: &nostr::Keys, tags: Vec>) -> Event { + nostr::EventBuilder::new(nostr::Kind::Custom(9007), "") + .tags( + tags.into_iter() + .map(|tag| nostr::Tag::parse(tag).expect("tag")), + ) + .sign_with_keys(creator) + .expect("sign") + } + + fn tag(name: &str, value: &str) -> Vec { + vec![name.to_string(), value.to_string()] + } + + #[test] + fn initial_members_dedupes_and_leaves_out_the_creator() { + let (creator, agent) = (nostr::Keys::generate(), nostr::Keys::generate()); + let h = tag("h", &Uuid::new_v4().to_string()); + let event = create_with( + &creator, + vec![ + h, + tag("p", &agent.public_key().to_hex()), + tag("p", &agent.public_key().to_hex()), + tag("p", &creator.public_key().to_hex()), + ], + ); + assert_eq!( + initial_members(&event).expect("members"), + vec![agent.public_key().to_bytes().to_vec()] + ); + } + + #[test] + fn initial_members_rejects_bad_pubkeys_too_many_members_and_no_h_tag() { + let creator = nostr::Keys::generate(); + let h = tag("h", &Uuid::new_v4().to_string()); + let bad = create_with(&creator, vec![h.clone(), tag("p", "not-hex")]); + assert!(initial_members(&bad).is_err()); + + let mut crowd = vec![h]; + for _ in 0..=MAX_INITIAL_MEMBERS { + crowd.push(tag("p", &nostr::Keys::generate().public_key().to_hex())); + } + assert!(initial_members(&create_with(&creator, crowd)).is_err()); + + let member = nostr::Keys::generate().public_key().to_hex(); + let headless = create_with(&creator, vec![tag("p", &member)]); + assert!(initial_members(&headless).is_err()); + } + #[test] fn nip43_progress_logs_every_interval_inside_the_startup_sweep_only() { use Nip43ReconciliationPurpose::{Bootstrap, Maintenance};