From 3ffbfa693c6727b04b01ef727ce4848e4695ac40 Mon Sep 17 00:00:00 2001 From: Bradley Axen Date: Sun, 18 Oct 2026 11:32:09 -0700 Subject: [PATCH] feat(relay): start a channel with its members in one create (#8421) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What A kind:9007 can now name the channel's first members with `p` tags. Those members are added in the same database transaction that creates the channel. Resending a create whose OK was lost is now accepted instead of refused. ## Why Starting a new Solo chat today takes about 6–8 sequential relay round trips before the first message goes out: create, read, add each agent (kind:9000), read the roster, then send. With members in the create, a client can start a chat with its agents as two writes: the create, then the message as soon as the create is OK. The resend fix makes the create safe for an outbox to retry. Today an identical resend after a lost OK is refused with `duplicate: channel already exists`, so the client can't tell success from a real UUID collision. ## How - **buzz-db:** `Db::create_channel_with_event` writes the channel, its owner, its members (role `member`, `invited_by` = creator) and the 9007 event itself in one transaction. A channel row therefore never exists without its create event. `create_channel_with_id` is unchanged for its other callers. - **Validation** (`side_effects::initial_members` and the `validate_admin_event` 9007 branch): - Each `p` must be a 32-byte hex pubkey. - Duplicates and the creator are dropped. - At most 32 members. - An `h` tag is required when there are members. - Each member goes through the same `decide_channel_add_policy` check a kind:9000 uses, so an agent that refuses third-party adds keeps the whole create out. - **Side effects:** each initial member gets the same `member_joined` system message and member-added notification a kind:9000 would post. The fallback in `handle_create_group` creates a bare channel when the pre-created row is missing; it announces no members, because it added none. - **Resend:** when the UUID is already taken, the same transaction checks whether this exact event is the stored create. - If it is, the create is OK'd as `duplicate:` (it is the same signed create). - Anything else, including a changed create from the same author, is refused and writes nothing, so it can never re-run side effects or announce members. - A resend that arrives while the first create is still uncommitted waits on its channel insert. It is confirmed once the first commits, or creates the channel itself if the first rolled back. Before this, a resend in that window was refused, and the app then withdrew a chat that existed. - **No compensation:** ingest no longer soft-deletes a pre-created channel when the event insert fails. The rollback covers it. - **Announcements:** a failed `member_joined` notice for one initial member is logged at `error` and the loop continues, so the members after it still get theirs. A resend writes nothing, so nothing would retry them. ## Compatibility A 9007 with malformed `p` tags, or more than 32 of them, used to be accepted with the tags ignored. It is now rejected with `invalid:`. The relay's own SDK and CLI builders send no `p` tags on 9007. ## Out of scope - The kind:9000 OK still doesn't prove the membership change committed: side-effect failures are logged and OK'd, and a duplicate replay skips side effects. The new path doesn't depend on kind:9000. - Events in separate HTTP requests are still unordered. Clients send the first message after the create's OK. ## Tests `store::channel::postgres_tests` (buzz-db): - `a_resend_beside_an_open_create_confirms_it_once_committed`: holds a create open, checks that the resend waits, commits, and expects `Exists { resent: true }` with one owner and one member. - `a_resend_beside_a_failed_create_creates_the_channel`: the same, but the first create rolls back. - `a_create_whose_event_fails_leaves_no_channel`: an event the insert refuses leaves no channel row, and a retry creates the channel. - `another_create_for_a_taken_channel_writes_nothing`. Two mutations, each caught: committing the channel before its event fails the "leaves no channel" test, and looking up the event before the channel insert waits fails the "confirmed once committed" test. `handlers/create_with_members_postgres_tests.rs`, all through the real ingest path: - `create_adds_its_p_tags_as_members` checks: - the roles and the 39002 roster - exactly one `member_joined` notice - that the agent can post with no separate add - `create_respects_the_members_add_policy`: an agent with policy `nobody` gets an `invalid:` rejection, and no channel is created. - `resending_a_create_confirms_it`: - an identical resend is accepted - a changed create (from the creator or from someone else) is refused - the roster is unchanged and nobody new is announced - Unit tests for `initial_members`: dedupe, creator excluded, bad hex, the 32-member cap, and the missing `h` tag. Each new test was mutation-checked against the change it covers: no members, no policy check, refuse every resend, and let a changed creator create through. Each mutation turns its test red. Agent exercise on a local relay, over `POST /events` the way the app sends: - The 9007 with `p` = agent was accepted in 54–97ms (debug build). - The kind 9 was accepted 12–15ms after the create's OK. - The agent was in the 39002 roster, and the operator listener delivered the mention. - An identical resend came back `duplicate:` (accepted), and a changed create was refused. Local suite runs at 10adcf6dd (the transaction fix), on PG14: - `postgres-test-run.sh -p buzz-relay -p buzz-db`: 28 failures. That is the same 27 as below, plus `personal_read_frontier_is_monotonic…`, which hit a statement timeout under load and passes when run alone. - `cargo test -p buzz-db -p buzz-relay`: the same 6 `api::media` failures and the flaky `mesh_demo` test. Earlier local suite runs: - New tests pass and clippy is clean. - `postgres-test-run.sh -p buzz-relay -p buzz-db`: 27 failures (partition maintenance, a huddle links test, a redis boot test), and the same 27 fail on a96ace4e5. - `cargo test -p buzz-relay`: 6 `api::media` failures (they need a DB at the default URL) also fail on base. - One `api::mesh_demo` test is flaky: it passed 2 of 3 branch runs and failed once on a timeout. πŸ€– Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Bradley Axen Co-authored-by: mbot <3754f8729004d95654c46dbab3129e4ab9ef05cc2534e2a3fbfc155983bd637b@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 Co-authored-by: claude00 Co-authored-by: ART --- crates/buzz-db/src/store/channel.rs | 435 ++++++++++++++++-- .../create_with_members_postgres_tests.rs | 251 ++++++++++ crates/buzz-relay/src/handlers/ingest.rs | 133 +++--- crates/buzz-relay/src/handlers/mod.rs | 2 + .../buzz-relay/src/handlers/side_effects.rs | 162 ++++++- 5 files changed, 867 insertions(+), 116 deletions(-) create mode 100644 crates/buzz-relay/src/handlers/create_with_members_postgres_tests.rs 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};