From db45b39771621d76818700d6cffce8b1405c6827 Mon Sep 17 00:00:00 2001 From: morgmart <98432065+morgmart@users.noreply.github.com> Date: Thu, 5 Nov 2026 10:55:52 -0800 Subject: [PATCH 1/8] feat(relay): grow DMs in place, identified by their people A DM is now identified by its people: agents never change which DM it is and do not count toward the 9-person limit. kind:41011 gains ["history", "include"]: the added people join this DM with its history, and its identity moves to the grown set of people. If a DM of those people already exists, that DM is returned instead. Without the tag an added person starts a new DM of everyone, as before. Agents always join in place, subject to their channel_add_policy. kind:41010 finds the DM of the named people and adds any named agents it lacks. DMs made before this change are still found by all members. Only people grow DMs: an agent cannot send the history tag, its adds start a new DM, and a DM it opens is identified by all its members. Growing never reveals a DM the actor is not in. Signed-off-by: morgmart <98432065+morgmart@users.noreply.github.com> Co-authored-by: Carl Co-Authored-By: Claude Opus 5.5 --- crates/buzz-cli/src/commands/dms.rs | 10 +- crates/buzz-cli/src/lib.rs | 4 + crates/buzz-db/src/store/channel_members.rs | 5 +- crates/buzz-db/src/store/dm.rs | 327 ++++++++++- .../src/handlers/command_executor.rs | 436 ++++++++++---- .../src/handlers/dm_growth_postgres_tests.rs | 537 ++++++++++++++++++ crates/buzz-relay/src/handlers/mod.rs | 2 + crates/buzz-sdk/src/builders.rs | 26 +- 8 files changed, 1199 insertions(+), 148 deletions(-) create mode 100644 crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs diff --git a/crates/buzz-cli/src/commands/dms.rs b/crates/buzz-cli/src/commands/dms.rs index 092e4a3cb3f..c3797635dad 100644 --- a/crates/buzz-cli/src/commands/dms.rs +++ b/crates/buzz-cli/src/commands/dms.rs @@ -137,11 +137,13 @@ pub async fn cmd_add_dm_member( client: &BuzzClient, channel_id: &str, pubkey: &str, + include_history: bool, ) -> Result<(), CliError> { let channel_uuid = parse_uuid(channel_id)?; validate_hex64(pubkey)?; - let builder = buzz_sdk::build_dm_add_member(channel_uuid, pubkey).map_err(sdk_err)?; + let builder = + buzz_sdk::build_dm_add_member(channel_uuid, pubkey, include_history).map_err(sdk_err)?; let event = client.sign_event(builder).await?; let resp = client.submit_event(event).await?; @@ -154,7 +156,11 @@ pub async fn dispatch(cmd: crate::DmsCmd, client: &BuzzClient) -> Result<(), Cli match cmd { DmsCmd::List { limit } => cmd_list_dms(client, limit).await, DmsCmd::Open { pubkeys } => cmd_open_dm(client, &pubkeys).await, - DmsCmd::AddMember { channel, pubkey } => cmd_add_dm_member(client, &channel, &pubkey).await, + DmsCmd::AddMember { + channel, + pubkey, + include_history, + } => cmd_add_dm_member(client, &channel, &pubkey, include_history).await, DmsCmd::Hide { channel } => cmd_hide_dm(client, &channel).await, } } diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 973fd0d2a57..3477112be62 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1326,6 +1326,10 @@ pub enum DmsCmd { /// User pubkey to add (64-char hex) #[arg(long)] pubkey: String, + /// Add a person to this DM with its history, instead of starting a new + /// DM of everyone. Agents always join this DM. + #[arg(long)] + include_history: bool, }, /// Hide a DM conversation from your DM list Hide { diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 3be879aaf00..048ea9dbda7 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -894,7 +894,10 @@ pub(crate) async fn put_members_with_event( /// tag, the TTL reaper's single UPDATE), and none of them takes the /// membership lock. So an archive either commits before [`put_members_on`] /// reads `archived_at`, or waits until this write commits. -async fn lock_channel_for_member_put(tx: &mut AdmittedTx, channel_id: Uuid) -> Result<()> { +pub(crate) async fn lock_channel_for_member_put( + tx: &mut AdmittedTx, + channel_id: Uuid, +) -> Result<()> { let community_id = tx.community(); acquire_channel_membership_lock(tx.conn(), community_id, channel_id).await?; lock_channel_ttl_shared_in_transaction(tx, channel_id).await?; diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index ebdbc45d271..c695c0434d2 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -1,7 +1,13 @@ //! Direct message channel persistence. //! //! DMs are channels with channel_type='dm' and visibility='private'. -//! Participant sets are immutable -- adding a member creates a NEW DM. +//! +//! A DM is identified by its people: at most one DM exists for a set of +//! people, and "message these people" finds it. Agents are add-ons: they join +//! a DM without changing which DM it is (see [`dm_identity_hash`]). A DM can +//! grow in place ([`grow_dm`]); its identity then moves to the grown set. + +use std::collections::HashSet; use chrono::{DateTime, Utc}; use sha2::{Digest, Sha256}; @@ -59,6 +65,63 @@ pub fn compute_participant_hash(pubkeys: &[&[u8]]) -> [u8; 32] { hasher.finalize().into() } +/// The most people a DM can hold. Agents do not count. +pub const MAX_DM_PEOPLE: usize = 9; + +/// The fingerprint that identifies a DM with these `members`, of whom +/// `agents` are agents. +/// +/// A DM is identified by its people, so adding or removing an agent never +/// changes which DM it is. A DM with at most one person (someone and their +/// agents) is identified by all its members instead, so each such DM keeps its +/// own identity, as every DM did before agents were told apart. +pub fn dm_identity_hash(members: &[&[u8]], agents: &HashSet>) -> [u8; 32] { + let people: Vec<&[u8]> = members + .iter() + .copied() + .filter(|pk| !agents.contains(*pk)) + .collect(); + if people.len() >= 2 { + compute_participant_hash(&people) + } else { + compute_participant_hash(members) + } +} + +/// The fingerprint a DM of `members` opened by `creator` gets. An agent +/// opening a DM gets one identified by all its members, as before agents were +/// told apart: only people decide which DM their people share. +fn opened_identity_hash(members: &[&[u8]], agents: &HashSet>, creator: &[u8]) -> [u8; 32] { + if agents.contains(creator) { + compute_participant_hash(members) + } else { + dm_identity_hash(members, agents) + } +} + +/// How many of `members` are people rather than agents. +fn count_people(members: &[&[u8]], agents: &HashSet>) -> usize { + members.iter().filter(|pk| !agents.contains(**pk)).count() +} + +/// The pubkeys among `candidates` that belong to agents: users with an owner. +async fn agents_among<'e>( + executor: impl sqlx::PgExecutor<'e>, + community_id: CommunityId, + candidates: &[&[u8]], +) -> Result>> { + let candidates: Vec> = candidates.iter().map(|pk| pk.to_vec()).collect(); + let agents: Vec> = sqlx::query_scalar( + "SELECT pubkey FROM users \ + WHERE community_id = $1 AND pubkey = ANY($2) AND agent_owner_pubkey IS NOT NULL", + ) + .bind(community_id.as_uuid()) + .bind(&candidates) + .fetch_all(executor) + .await?; + Ok(agents.into_iter().collect()) +} + // -- DB functions ------------------------------------------------------------- /// Find an existing DM by its participant hash. @@ -97,9 +160,11 @@ pub async fn find_dm_by_participants( /// existing one if a DM with the same participant set already exists. /// /// Rules: -/// - `participants` must contain 2-9 entries (enforced here). +/// - `participants` must contain at least 2 entries, and at most +/// [`MAX_DM_PEOPLE`] people; agents do not count (enforced here). /// - `created_by` must be one of the participants. -/// - The operation is idempotent: same participant set -> same channel returned. +/// - The operation is idempotent: the same people (see [`dm_identity_hash`]) +/// -> the same channel returned. pub async fn create_dm( pool: &PgPool, community_id: CommunityId, @@ -111,11 +176,6 @@ pub async fn create_dm( "DM requires at least 2 participants".to_string(), )); } - if participants.len() > 9 { - return Err(DbError::InvalidData( - "DM supports at most 9 participants".to_string(), - )); - } for pk in participants { if pk.len() != 32 { return Err(DbError::InvalidData(format!( @@ -124,8 +184,14 @@ pub async fn create_dm( ))); } } + let agents = agents_among(pool, community_id, participants).await?; + if count_people(participants, &agents) > MAX_DM_PEOPLE { + return Err(DbError::InvalidData(format!( + "DM supports at most {MAX_DM_PEOPLE} people" + ))); + } - let hash = compute_participant_hash(participants); + let hash = opened_identity_hash(participants, &agents, created_by); let mut tx = crate::begin_community_write_transaction( pool, @@ -358,6 +424,13 @@ pub async fn list_dms_for_user( /// `created_by` is automatically added to `pubkeys` if not already present, /// ensuring the caller is always a participant in their own DM. /// +/// The DM is found by its people ([`dm_identity_hash`]), so it may hold agents +/// that `pubkeys` does not name, and lack agents it does. Adding those is +/// [`grow_dm`]'s job. A DM made before agents were told apart is still found by +/// all its members. An agent opening a DM finds or makes one identified by all +/// its members. A found DM is returned only to an active member of it: +/// an agent naming its people cannot use this to see or join their DM. +/// /// Returns `(channel, was_created)`: /// - `was_created = true` -- a new DM was created. /// - `was_created = false` -- an existing DM was returned. @@ -367,23 +440,35 @@ pub async fn open_dm( pubkeys: &[&[u8]], created_by: &[u8], ) -> Result<(ChannelRecord, bool)> { - // Merge created_by into the participant set (dedup handled by compute_participant_hash). - let mut all: Vec<&[u8]> = pubkeys.to_vec(); - if !all.contains(&created_by) { - all.push(created_by); + let mut all: Vec<&[u8]> = vec![created_by]; + for pk in pubkeys { + if !all.contains(pk) { + all.push(pk); + } } - // Enforce max before hitting the DB. - if all.len() > 9 { - return Err(DbError::InvalidData( - "DM supports at most 9 participants".to_string(), - )); + let agents = agents_among(pool, community_id, &all).await?; + if count_people(&all, &agents) > MAX_DM_PEOPLE { + return Err(DbError::InvalidData(format!( + "DM supports at most {MAX_DM_PEOPLE} people" + ))); } - let hash = compute_participant_hash(&all); + let hash = opened_identity_hash(&all, &agents, created_by); + // DMs made before agents were told apart are identified by all members. + let legacy = compute_participant_hash(&all); // Check for existing DM first (fast path, no transaction). - if let Some(existing) = find_dm_by_participants(pool, community_id, &hash).await? { + let mut existing = find_dm_by_participants(pool, community_id, &hash).await?; + if existing.is_none() && legacy != hash { + existing = find_dm_by_participants(pool, community_id, &legacy).await?; + } + if let Some(existing) = existing { + if !crate::channel_members::is_member(pool, community_id, existing.id, created_by).await? { + return Err(DbError::AccessDenied( + "a DM with these people already exists".to_string(), + )); + } // Clear hidden_at for the caller so the DM reappears in their sidebar. unhide_dm(pool, community_id, existing.id, created_by).await?; return Ok((existing, false)); @@ -395,6 +480,196 @@ pub async fn open_dm( Ok((channel, true)) } +/// The outcome of [`grow_dm`]. +#[derive(Debug, Clone)] +pub enum DmGrowth { + /// The DM now holds everyone asked for. `added` lists those who were not + /// already active members, in the order asked. + Grown { + /// Pubkeys that became members. + added: Vec>, + }, + /// Another DM, with this ID, already has the grown set of people, so this + /// DM cannot take its identity. Nothing was written. + Existing(Uuid), +} + +/// Add `additions` to the DM `channel_id` in place, keeping its history. +/// +/// Agents join without changing which DM this is. When people join, the DM's +/// identity moves to the grown set of people ([`dm_identity_hash`]), so +/// opening a DM with the original people afterwards starts a new one. If a DM +/// with the grown set already exists, nothing is written and that DM is +/// returned as [`DmGrowth::Existing`], provided `actor` is in it. +/// +/// `actor` must be an active member. The DM must be live and not archived, and +/// may hold at most [`MAX_DM_PEOPLE`] people afterwards. Re-adding members is +/// a no-op, so a retry after a lost commit acknowledgement is safe. +/// +/// Runs under the channel's membership locks ([`put_members`]'s), so it is +/// serialized against every other membership write and archive of the DM. +/// +/// [`put_members`]: crate::channel_members::put_members +pub async fn grow_dm( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, + actor: &[u8], + additions: &[&[u8]], +) -> Result { + for pk in additions { + if pk.len() != 32 { + return Err(DbError::InvalidData(format!( + "pubkey must be 32 bytes, got {}", + pk.len() + ))); + } + } + + let mut tx = crate::begin_community_write_transaction( + pool, + community_id, + crate::observability::WriterOperation::EventWrite, + ) + .await?; + crate::channel_members::lock_channel_for_member_put(&mut tx, channel_id).await?; + + let channel = sqlx::query( + "SELECT channel_type::text AS channel_type, archived_at FROM channels \ + WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await? + .ok_or(DbError::ChannelNotFound(channel_id))?; + let channel_type: String = channel.try_get("channel_type")?; + if channel_type != "dm" { + return Err(DbError::InvalidData("channel is not a DM".to_string())); + } + if channel + .try_get::>, _>("archived_at")? + .is_some() + { + return Err(DbError::AccessDenied("DM is archived".to_string())); + } + + let members: Vec> = sqlx::query_scalar( + "SELECT pubkey FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 AND removed_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_all(tx.conn()) + .await?; + if !members.iter().any(|m| m == actor) { + return Err(DbError::AccessDenied("not a member of this DM".to_string())); + } + + let mut added: Vec<&[u8]> = Vec::new(); + for pk in additions { + if !members.iter().any(|m| m == pk) && !added.contains(pk) { + added.push(pk); + } + } + if added.is_empty() { + tx.commit().await?; + return Ok(DmGrowth::Grown { added: Vec::new() }); + } + + let mut grown: Vec<&[u8]> = members.iter().map(Vec::as_slice).collect(); + grown.extend(&added); + let agents = agents_among(tx.conn(), community_id, &grown).await?; + if count_people(&grown, &agents) > MAX_DM_PEOPLE { + return Err(DbError::InvalidData(format!( + "DM supports at most {MAX_DM_PEOPLE} people" + ))); + } + + if count_people(&added, &agents) > 0 { + let hash = dm_identity_hash(&grown, &agents); + let existing = sqlx::query_scalar::<_, Uuid>( + "SELECT id FROM channels \ + WHERE community_id = $1 AND participant_hash = $2 AND id <> $3 \ + AND deleted_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(hash.as_slice()) + .bind(channel_id) + .fetch_optional(tx.conn()) + .await?; + if let Some(existing) = existing { + tx.rollback().await?; + return existing_for(pool, community_id, existing, actor).await; + } + let moved = sqlx::query( + "UPDATE channels SET participant_hash = $3, updated_at = NOW() \ + WHERE community_id = $1 AND id = $2", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(hash.as_slice()) + .execute(tx.conn()) + .await; + // A DM created or grown into the same people since the check above. + if let Err(sqlx::Error::Database(e)) = &moved { + if e.code().as_deref() == Some("23505") { + tx.rollback().await?; + return match find_dm_by_participants(pool, community_id, &hash).await? { + Some(existing) => existing_for(pool, community_id, existing.id, actor).await, + None => Err(DbError::InvalidData( + "a DM with these people already exists".to_string(), + )), + }; + } + } + moved?; + } + + for pk in &added { + 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 UPDATE SET + sync_modified_at = clock_timestamp(), + removed_at = NULL, + removed_by = NULL, + hidden_at = NULL, + role = EXCLUDED.role + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(*pk) + .bind(actor) + .execute(tx.conn()) + .await?; + } + + tx.commit().await?; + Ok(DmGrowth::Grown { + added: added.into_iter().map(<[u8]>::to_vec).collect(), + }) +} + +/// [`DmGrowth::Existing`] for `actor`, who must be an active member of that +/// DM: growing a DM must not reveal a DM the actor is not in. +async fn existing_for( + pool: &PgPool, + community_id: CommunityId, + existing: Uuid, + actor: &[u8], +) -> Result { + if crate::channel_members::is_member(pool, community_id, existing, actor).await? { + Ok(DmGrowth::Existing(existing)) + } else { + Err(DbError::AccessDenied( + "a DM with these people already exists".to_string(), + )) + } +} + // -- Hide / unhide ------------------------------------------------------------ /// Hide a DM for a specific user by setting `hidden_at = NOW()`. @@ -587,6 +862,18 @@ impl Db { crate::dm::open_dm(&self.pool, community_id, pubkeys, created_by).await } + /// Add members to a DM in place, keeping its history. + #[datastore_span(name = "grow_dm", system = "postgresql")] + pub async fn grow_dm( + &self, + community_id: CommunityId, + channel_id: Uuid, + actor: &[u8], + additions: &[&[u8]], + ) -> Result { + crate::dm::grow_dm(&self.pool, community_id, channel_id, actor, additions).await + } + /// Hide a DM channel for a specific user. /// /// The DM is not deleted — it can be restored by opening a new DM with diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 73d42985178..33eb0b5c662 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -9,6 +9,7 @@ //! 3. Pubkey/auth identity match //! 4. Per-kind scope authorization +use std::collections::HashMap; use std::sync::Arc; use chrono::Utc; @@ -20,6 +21,7 @@ use uuid::Uuid; use buzz_core::kind::*; use buzz_core::tenant::{CommunityId, TenantContext}; use buzz_datastore_tracing::datastore_span; +use buzz_db::dm::{DmGrowth, MAX_DM_PEOPLE}; use buzz_db::workflow::{ApprovalStatus, RunStatus}; use buzz_db::DbError; use buzz_workflow::executor::TriggerContext; @@ -291,6 +293,158 @@ fn compute_definition_hash(json_str: &str) -> Vec { Sha256::digest(json_str.as_bytes()).to_vec() } +/// Map a DM write's refusal to the client's error. +fn dm_write_error(e: DbError) -> IngestError { + match e { + DbError::AccessDenied(m) => IngestError::Rejected(format!("forbidden: {m}")), + DbError::InvalidData(m) => IngestError::Rejected(format!("invalid: {m}")), + DbError::ChannelNotFound(_) => IngestError::Rejected("invalid: DM not found".into()), + e => IngestError::Internal(format!("error: db: {e}")), + } +} + +/// The side effects of members joining an existing DM (post-commit, +/// best-effort): membership caches, a `member_joined` system message each, +/// kind:39000/39002 and kind:44100. +async fn announce_dm_additions( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + actor: &[u8], + added: &[Vec], +) { + if added.is_empty() { + return; + } + for pk in added { + state.invalidate_membership(tenant, channel_id, pk); + } + let actor_hex = hex::encode(actor); + for pk in added { + if let Err(e) = emit_system_message( + tenant, + state, + channel_id, + serde_json::json!({ + "type": "member_joined", + "actor": actor_hex, + "target": hex::encode(pk), + }), + chrono::Utc::now(), + ) + .await + { + warn!("DM add: system message failed: {e}"); + } + } + if let Err(e) = emit_group_discovery_events(tenant, state, channel_id).await { + warn!(channel = %channel_id, "DM add: discovery emission failed: {e}"); + } + for pk in added { + if let Err(e) = emit_membership_notification( + tenant, + state, + channel_id, + pk, + actor, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + warn!("DM add: membership notification failed: {e}"); + } + } +} + +/// The side effects of a newly created DM (post-commit, best-effort). +async fn announce_dm_created( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + actor: &[u8], + participants: &[Vec], +) { + metrics::counter!( + "buzz_channels_created_total", + "community" => tenant.host().to_owned(), + "type" => "dm" + ) + .increment(1); + + for pk in participants { + state.invalidate_membership(tenant, channel_id, pk); + } + + let participant_hexes: Vec = participants.iter().map(hex::encode).collect(); + if let Err(e) = emit_system_message( + tenant, + state, + channel_id, + serde_json::json!({ + "type": "dm_created", + "actor": hex::encode(actor), + "participants": participant_hexes, + }), + chrono::Utc::now(), + ) + .await + { + warn!("DM open: system message failed: {e}"); + } + + if let Err(e) = emit_group_discovery_events(tenant, state, channel_id).await { + warn!(channel = %channel_id, "DM open: discovery emission failed: {e}"); + } + + for participant in participants { + if let Err(e) = emit_membership_notification( + tenant, + state, + channel_id, + participant, + actor, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + warn!("DM open: membership notification failed: {e}"); + } + } +} + +/// The agents among `pubkeys` (users with an owner): each one's +/// `channel_add_policy` and owner. +async fn dm_agents( + tenant: &TenantContext, + state: &Arc, + pubkeys: &[Vec], +) -> Result, (String, Vec)>, IngestError> { + let policies = state + .db + .get_agent_channel_policies(tenant.community(), pubkeys) + .await + .map_err(|e| IngestError::Internal(format!("error: agent lookup: {e}")))?; + Ok(policies + .into_iter() + .filter_map(|(pubkey, (policy, owner))| owner.map(|owner| (pubkey, (policy, owner)))) + .collect()) +} + +/// Refuse unless `actor` may put each of `agents` in a DM with other people, +/// by its `channel_add_policy`, as a kind:9000 add would decide. +fn check_dm_agent_policies( + agents: &HashMap, (String, Vec)>, + actor: &[u8], +) -> Result<(), IngestError> { + for (agent, (policy, owner)) in agents { + if agent.as_slice() != actor { + super::channel_authz::decide_channel_add_policy(policy, Some(owner), actor) + .map_err(|e| IngestError::Rejected(format!("forbidden: {e}")))?; + } + } + Ok(()) +} + async fn handle_dm_open( tenant: &TenantContext, state: &Arc, @@ -298,36 +452,38 @@ async fn handle_dm_open( auth: &IngestAuth, ) -> Result { let self_bytes = auth.pubkey().to_bytes().to_vec(); - let self_hex = hex::encode(&self_bytes); - // 1. Extract participant pubkeys from `p` tags - let p_tags = extract_p_tags(event); - - // 2. Validate: at least 1 other participant, max 8 others (9 total) - if p_tags.is_empty() { - return Err(IngestError::Rejected( - "invalid: pubkeys must contain at least 1 other participant".into(), - )); + // 1. Decode the other participants from `p` tags (deduplicated). + let mut others: Vec> = Vec::new(); + for hex_str in &extract_p_tags(event) { + let pk = decode_pubkey(hex_str)?; + if pk != self_bytes && !others.contains(&pk) { + others.push(pk); + } } - if p_tags.len() > 8 { + if others.is_empty() { return Err(IngestError::Rejected( - "invalid: pubkeys may contain at most 8 other participants (9 total)".into(), + "invalid: pubkeys must contain at least 1 other participant".into(), )); } - // Decode all provided pubkeys - let mut other_bytes: Vec> = Vec::with_capacity(p_tags.len()); - for hex_str in &p_tags { - other_bytes.push(decode_pubkey(hex_str)?); + // 2. At most 8 other people (9 total); agents do not count. Putting an + // agent in a DM with other people needs its channel_add_policy; a DM + // of just the caller and agents does not. + let agents = dm_agents(tenant, state, &others).await?; + let other_people = others.iter().filter(|pk| !agents.contains_key(*pk)).count(); + if other_people > 0 { + check_dm_agent_policies(&agents, &self_bytes)?; + } + if other_people > MAX_DM_PEOPLE - 1 { + return Err(IngestError::Rejected(format!( + "invalid: pubkeys may contain at most {} other people ({MAX_DM_PEOPLE} total); agents do not count", + MAX_DM_PEOPLE - 1 + ))); } - // 3. Build full participant set (self + others, deduplicated) let mut all_bytes: Vec> = vec![self_bytes.clone()]; - for ob in &other_bytes { - if !all_bytes.iter().any(|b| b == ob) { - all_bytes.push(ob.clone()); - } - } + all_bytes.extend(others); // Persist the command event (idempotency) — returns open transaction let tx = match persist_command_event(&state.db, tenant, event, None).await? { @@ -341,77 +497,38 @@ async fn handle_dm_open( PersistResult::Inserted(tx) => tx, }; - // 4. Execute: open_dm + // 3. Execute: open the DM of these people, then add any named agents it + // does not hold yet. let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect(); let (channel, was_created) = state .db .open_dm(tenant.community(), &all_refs, &self_bytes) .await - .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + .map_err(dm_write_error)?; + let added = if was_created { + Vec::new() + } else { + grow_with_agents(tenant, state, channel.id, &self_bytes, &agents).await? + }; // Finalize the idempotency record after the separate mutation succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; - // 5. Side effects if newly created (post-commit, best-effort) + // 4. Side effects (post-commit, best-effort) if was_created { - metrics::counter!( - "buzz_channels_created_total", - "community" => tenant.host().to_owned(), - "type" => "dm" - ) - .increment(1); - - // Invalidate caches for all participants - for pk in &all_bytes { - state.invalidate_membership(tenant, channel.id, pk); - } - - let participant_hexes: Vec = all_bytes.iter().map(hex::encode).collect(); - if let Err(e) = emit_system_message( - tenant, - state, - channel.id, - serde_json::json!({ - "type": "dm_created", - "actor": self_hex, - "participants": participant_hexes, - }), - chrono::Utc::now(), - ) - .await - { - warn!("DM open: system message failed: {e}"); - } - - if let Err(e) = emit_group_discovery_events(tenant, state, channel.id).await { - warn!(channel = %channel.id, "DM open: discovery emission failed: {e}"); - } - - for participant in &all_bytes { - if let Err(e) = emit_membership_notification( - tenant, - state, - channel.id, - participant, - &self_bytes, - KIND_MEMBER_ADDED_NOTIFICATION, - ) - .await - { - warn!("DM open: membership notification failed: {e}"); - } - } + announce_dm_created(tenant, state, channel.id, &self_bytes, &all_bytes).await; } else { // Re-open of an existing DM cleared the caller's hidden_at; refresh // their NIP-DV snapshot so the DM reappears in the sidebar. if let Err(e) = publish_dm_visibility_snapshot(tenant, state, &self_bytes).await { warn!("DM re-open: visibility snapshot failed: {e}"); } + announce_dm_additions(tenant, state, channel.id, &self_bytes, &added).await; } - // 6. Return response + // 5. Return response Ok(IngestResult { event_id: event.id.to_hex(), accepted: true, @@ -425,6 +542,40 @@ async fn handle_dm_open( }) } +/// Add `agents` to the existing DM `channel_id` in place. Agents never change +/// which DM it is, so the DM cannot turn out to be another one. +async fn grow_with_agents( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + actor: &[u8], + agents: &HashMap, (String, Vec)>, +) -> Result>, IngestError> { + if agents.is_empty() { + return Ok(Vec::new()); + } + let refs: Vec<&[u8]> = agents.keys().map(Vec::as_slice).collect(); + match state + .db + .grow_dm(tenant.community(), channel_id, actor, &refs) + .await + .map_err(dm_write_error)? + { + DmGrowth::Grown { added } => Ok(added), + DmGrowth::Existing(_) => Err(IngestError::Internal( + "error: adding agents changed the DM's people".into(), + )), + } +} + +/// kind:41011: add people and agents to a DM. +/// +/// Agents always join in place, with the DM's history. People join in place +/// too when the event carries `["history", "include"]`; the DM's identity then +/// moves to the grown set of people, or, if a DM of those people already +/// exists, that DM is returned instead and nothing changes here. Without the +/// tag, people start a new DM of everyone (no history), as before. An agent +/// cannot send the tag, and its adds always start a new DM of everyone. async fn handle_dm_add_member( tenant: &TenantContext, state: &Arc, @@ -445,6 +596,15 @@ async fn handle_dm_add_member( "invalid: must specify at least 1 new participant in p tags".into(), )); } + let include_history = match extract_tag(event, "history").as_deref() { + None => false, + Some("include") => true, + Some(other) => { + return Err(IngestError::Rejected(format!( + "invalid: history must be \"include\", got \"{other}\"" + ))); + } + }; // 2. Validate caller is member of existing DM let is_member = state @@ -467,29 +627,46 @@ async fn handle_dm_add_member( return Err(IngestError::Rejected("invalid: channel is not a DM".into())); } - // 4. Get existing members, merge with new - let existing_members = state + // 4. Read the current members. The relay's moderation DM stays two-party. + let members: Vec> = state .db .get_members_for_event_write(tenant.community(), channel_id) .await - .map_err(|e| IngestError::Internal(format!("error: get members: {e}")))?; - - let mut all_bytes: Vec> = existing_members.into_iter().map(|m| m.pubkey).collect(); + .map_err(|e| IngestError::Internal(format!("error: get members: {e}")))? + .into_iter() + .map(|m| m.pubkey) + .collect(); + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); + if members + .iter() + .any(|m| m.as_slice() == relay_pubkey.as_slice()) + { + return Err(IngestError::Rejected( + "invalid: this DM cannot be changed".into(), + )); + } - // Decode and merge new pubkeys + let mut additions: Vec> = Vec::new(); for hex_str in &p_tags { - let bytes = decode_pubkey(hex_str)?; - if !all_bytes.iter().any(|b| b == &bytes) { - all_bytes.push(bytes); + let pk = decode_pubkey(hex_str)?; + if !members.contains(&pk) && !additions.contains(&pk) { + additions.push(pk); } } - - // 5. Enforce max 9 participants - if all_bytes.len() > 9 { + let agents = dm_agents(tenant, state, &additions).await?; + check_dm_agent_policies(&agents, &self_bytes)?; + let adds_people = additions.iter().any(|pk| !agents.contains_key(pk)); + // Only people grow a DM or share its history. An agent's add starts a new + // DM of everyone, as every add did before. + let actor_is_agent = !dm_agents(tenant, state, std::slice::from_ref(&self_bytes)) + .await? + .is_empty(); + if actor_is_agent && include_history { return Err(IngestError::Rejected( - "invalid: DM supports at most 9 participants".into(), + "forbidden: only people can share a DM's history".into(), )); } + let in_place = !actor_is_agent && (include_history || !adds_people); // Persist the command event — returns open transaction let tx = match persist_command_event(&state.db, tenant, event, None).await? { @@ -503,60 +680,77 @@ async fn handle_dm_add_member( PersistResult::Inserted(tx) => tx, }; - // 6. Execute: open_dm with expanded set (creates NEW DM — DM sets are immutable) - let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect(); - let (new_channel, was_created) = state - .db - .open_dm(tenant.community(), &all_refs, &self_bytes) - .await - .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + // 5. Execute. + let addition_refs: Vec<&[u8]> = additions.iter().map(Vec::as_slice).collect(); + let (target, was_created, added, moved_elsewhere) = if in_place { + // Grow this DM in place. + match state + .db + .grow_dm(tenant.community(), channel_id, &self_bytes, &addition_refs) + .await + .map_err(dm_write_error)? + { + DmGrowth::Grown { added } => (channel_id, false, added, false), + // These people already have a DM: the agents join them there. + DmGrowth::Existing(other) => { + let added = grow_with_agents(tenant, state, other, &self_bytes, &agents).await?; + (other, false, added, true) + } + } + } else { + // Start a new DM of everyone (no history); this DM is untouched. + let mut all_bytes = members.clone(); + all_bytes.extend(additions.iter().cloned()); + let all_refs: Vec<&[u8]> = all_bytes.iter().map(Vec::as_slice).collect(); + let (channel, created) = state + .db + .open_dm(tenant.community(), &all_refs, &self_bytes) + .await + .map_err(dm_write_error)?; + let added = if created { + Vec::new() + } else { + grow_with_agents(tenant, state, channel.id, &self_bytes, &agents).await? + }; + (channel.id, created, added, !created) + }; // Finalize the idempotency record after the separate mutation succeeds. tx.commit() .await .map_err(|e| IngestError::Internal(format!("error: commit transaction: {e}")))?; - // 7. Cache invalidation + notifications for new DM (post-commit, best-effort) + // 6. Side effects (post-commit, best-effort) if was_created { - metrics::counter!( - "buzz_channels_created_total", - "community" => tenant.host().to_owned(), - "type" => "dm" - ) - .increment(1); - - for pk in &all_bytes { - state.invalidate_membership(tenant, new_channel.id, pk); - } - - if let Err(e) = emit_group_discovery_events(tenant, state, new_channel.id).await { - warn!(channel = %new_channel.id, "DM add_member: discovery emission failed: {e}"); - } - - for participant_bytes in &all_bytes { - if let Err(e) = emit_membership_notification( - tenant, - state, - new_channel.id, - participant_bytes, - &self_bytes, - KIND_MEMBER_ADDED_NOTIFICATION, - ) - .await + let mut participants = members; + participants.extend(additions); + announce_dm_created(tenant, state, target, &self_bytes, &participants).await; + } else { + if moved_elsewhere { + // The caller lands in an existing DM; make sure they can see it. + if let Err(e) = state + .db + .unhide_dm(tenant.community(), target, &self_bytes) + .await + { + warn!("DM add: unhide failed: {e}"); + } else if let Err(e) = publish_dm_visibility_snapshot(tenant, state, &self_bytes).await { - warn!("DM add_member: membership notification failed: {e}"); + warn!("DM add: visibility snapshot failed: {e}"); } } + announce_dm_additions(tenant, state, target, &self_bytes, &added).await; } - // 8. Return response + // 7. Return response Ok(IngestResult { event_id: event.id.to_hex(), accepted: true, message: format!( "response:{}", serde_json::json!({ - "channel_id": new_channel.id.to_string(), + "channel_id": target.to_string(), + "created": was_created, }) ), }) diff --git a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs new file mode 100644 index 00000000000..769e87d19df --- /dev/null +++ b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs @@ -0,0 +1,537 @@ +//! Postgres-backed tests for growing a DM: kind:41011 adding people in place +//! (`["history", "include"]`) or into a new DM, agents joining without +//! changing which DM it is, and kind:41010 finding a DM by its people. + +use std::sync::Arc; + +use buzz_core::TenantContext; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use uuid::Uuid; + +use super::group_state_postgres_tests::{community, http_auth, state}; +use crate::handlers::ingest::{ingest_event, IngestError, IngestResult}; +use crate::state::AppState; + +fn hex(keys: &Keys) -> String { + keys.public_key().to_hex() +} + +async fn submit( + state: &Arc, + tenant: &TenantContext, + actor: &Keys, + kind: u16, + tags: Vec>, +) -> Result { + let tags: Vec = tags + .into_iter() + .map(|tag| Tag::parse(tag).expect("tag")) + // A unique tag so identical commands are distinct events. + .chain([Tag::parse(["client", &Uuid::new_v4().to_string()]).expect("tag")]) + .collect(); + let event = EventBuilder::new(Kind::Custom(kind), "") + .tags(tags) + .sign_with_keys(actor) + .expect("sign"); + ingest_event(state, tenant, event, http_auth(actor)).await +} + +fn response_channel(result: &IngestResult) -> (Uuid, bool) { + let json: serde_json::Value = serde_json::from_str( + result + .message + .strip_prefix("response:") + .unwrap_or_else(|| panic!("no response in {:?}", result.message)), + ) + .expect("response json"); + ( + json["channel_id"] + .as_str() + .expect("channel_id") + .parse() + .expect("uuid"), + json["created"].as_bool().expect("created"), + ) +} + +async fn open( + state: &Arc, + tenant: &TenantContext, + actor: &Keys, + others: &[&Keys], +) -> Result { + let tags = others.iter().map(|k| vec!["p".into(), hex(k)]).collect(); + submit(state, tenant, actor, 41010, tags) + .await + .map(|r| response_channel(&r).0) +} + +async fn add( + state: &Arc, + tenant: &TenantContext, + actor: &Keys, + dm: Uuid, + others: &[&Keys], + include_history: bool, +) -> Result<(Uuid, bool), IngestError> { + let mut tags: Vec> = vec![vec!["h".into(), dm.to_string()]]; + tags.extend(others.iter().map(|k| vec!["p".into(), hex(k)])); + if include_history { + tags.push(vec!["history".into(), "include".into()]); + } + submit(state, tenant, actor, 41011, tags) + .await + .map(|r| response_channel(&r)) +} + +async fn members(state: &AppState, tenant: &TenantContext, dm: Uuid) -> Vec { + let mut members: Vec = state + .db + .get_members(tenant.community(), dm) + .await + .expect("members") + .into_iter() + .map(|m| ::hex::encode(m.pubkey)) + .collect(); + members.sort(); + members +} + +fn set_of(keys: &[&Keys]) -> Vec { + let mut set: Vec = keys.iter().map(|k| hex(k)).collect(); + set.sort(); + set +} + +/// Make `agent` an agent owned by `owner`, with `policy` for third-party adds. +async fn make_agent( + state: &AppState, + tenant: &TenantContext, + agent: &Keys, + owner: &Keys, + policy: &str, +) { + let agent_bytes = agent.public_key().to_bytes(); + let owner_bytes = owner.public_key().to_bytes(); + for pk in [&agent_bytes, &owner_bytes] { + state + .db + .ensure_user(tenant.community(), pk) + .await + .expect("user"); + } + state + .db + .set_agent_owner(tenant.community(), &agent_bytes, &owner_bytes) + .await + .expect("owner"); + state + .db + .set_channel_add_policy(tenant.community(), &agent_bytes, policy) + .await + .expect("policy"); +} + +fn rejected(result: Result, needle: &str) { + match result { + Err(IngestError::Rejected(message)) => { + assert!(message.contains(needle), "{message:?} lacks {needle:?}") + } + other => panic!("expected a refusal containing {needle:?}, got {other:?}"), + } +} + +/// Adding a person with history keeps the DM and moves its identity to the +/// grown set of people: the original people alone now get a new DM, and the +/// grown set finds this one. +/// Mutations: fork instead of growing → a different channel → RED; leave the +/// fingerprint behind → opening {A, B} finds the grown DM → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn adding_a_person_with_history_grows_the_dm_in_place() { + let state = state().await; + let tenant = community(&state, "dm-grow-in-place").await; + let (a, b, c) = (Keys::generate(), Keys::generate(), Keys::generate()); + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + + let (target, created) = add(&state, &tenant, &a, dm, &[&c], true) + .await + .expect("add"); + + assert_eq!((target, created), (dm, false)); + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b, &c])); + assert_eq!( + open(&state, &tenant, &b, &[&a, &c]).await.expect("reopen"), + dm + ); + let fresh = open(&state, &tenant, &a, &[&b]) + .await + .expect("open original people"); + assert_ne!(fresh, dm, "the original people start a new DM"); + assert_eq!(members(&state, &tenant, fresh).await, set_of(&[&a, &b])); +} + +/// Without the history tag, adding a person starts a new DM of everyone and +/// leaves the original untouched, as before. +/// Mutation: treat a missing tag as include → the original grows → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn adding_a_person_without_history_starts_a_new_dm() { + let state = state().await; + let tenant = community(&state, "dm-grow-fork").await; + let (a, b, c) = (Keys::generate(), Keys::generate(), Keys::generate()); + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + + let (target, created) = add(&state, &tenant, &a, dm, &[&c], false) + .await + .expect("add"); + + assert!(created); + assert_ne!(target, dm); + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); + assert_eq!( + members(&state, &tenant, target).await, + set_of(&[&a, &b, &c]) + ); + assert_eq!(open(&state, &tenant, &a, &[&b]).await.expect("reopen"), dm); +} + +/// When a DM of the grown people already exists, growing returns it and +/// changes nothing. +/// Mutation: drop the collision check → the unique index refuses the move, or +/// the DM grows anyway → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn growing_into_an_existing_group_returns_that_dm() { + let state = state().await; + let tenant = community(&state, "dm-grow-collision").await; + let (a, b, c) = (Keys::generate(), Keys::generate(), Keys::generate()); + let pair = open(&state, &tenant, &a, &[&b]).await.expect("open pair"); + let trio = open(&state, &tenant, &a, &[&b, &c]) + .await + .expect("open trio"); + + let (target, created) = add(&state, &tenant, &a, pair, &[&c], true) + .await + .expect("add"); + + assert_eq!((target, created), (trio, false)); + assert_eq!(members(&state, &tenant, pair).await, set_of(&[&a, &b])); + assert_eq!( + open(&state, &tenant, &a, &[&b]).await.expect("reopen"), + pair + ); +} + +/// An agent joins in place, with or without the history tag, and never +/// changes which DM it is: the same people still find it. +/// Mutation: fork on agents without the tag → a different channel → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_agent_joins_in_place_without_changing_the_dm() { + let state = state().await; + let tenant = community(&state, "dm-grow-agent").await; + let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &agent, &a, "owner_only").await; + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + + let (target, created) = add(&state, &tenant, &a, dm, &[&agent], false) + .await + .expect("add"); + + assert_eq!((target, created), (dm, false)); + assert_eq!( + members(&state, &tenant, dm).await, + set_of(&[&a, &b, &agent]) + ); + assert_eq!(open(&state, &tenant, &b, &[&a]).await.expect("reopen"), dm); +} + +/// Opening a DM with people and an agent finds the people's DM and adds the +/// agent to it. +/// Mutation: skip adding named agents to a found DM → agent missing → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn opening_with_an_agent_finds_the_peoples_dm_and_adds_it() { + let state = state().await; + let tenant = community(&state, "dm-open-agent").await; + let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &agent, &a, "anyone").await; + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + + assert_eq!( + open(&state, &tenant, &a, &[&b, &agent]) + .await + .expect("open"), + dm + ); + assert_eq!( + members(&state, &tenant, dm).await, + set_of(&[&a, &b, &agent]) + ); +} + +/// A DM of someone and their agents is still identified by all its members, +/// so each agent keeps its own DM with its owner. +/// Mutation: identify one-person DMs by people only → both agents share one +/// DM → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn each_agent_keeps_its_own_dm_with_its_owner() { + let state = state().await; + let tenant = community(&state, "dm-own-agents").await; + let (a, first, second) = (Keys::generate(), Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &first, &a, "owner_only").await; + make_agent(&state, &tenant, &second, &a, "owner_only").await; + + let dm_first = open(&state, &tenant, &a, &[&first]).await.expect("first"); + let dm_second = open(&state, &tenant, &a, &[&second]).await.expect("second"); + + assert_ne!(dm_first, dm_second); + assert_eq!( + open(&state, &tenant, &a, &[&first]).await.expect("again"), + dm_first + ); +} + +/// The nine-person limit counts people only. +/// Mutations: count agents toward the limit → the agent add is refused → RED; +/// drop the limit → a tenth person joins → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn the_limit_counts_people_not_agents() { + let state = state().await; + let tenant = community(&state, "dm-limit").await; + let a = Keys::generate(); + let others: Vec = (0..8).map(|_| Keys::generate()).collect(); + let other_refs: Vec<&Keys> = others.iter().collect(); + let agent = Keys::generate(); + make_agent(&state, &tenant, &agent, &a, "anyone").await; + let dm = open(&state, &tenant, &a, &other_refs) + .await + .expect("nine people"); + + add(&state, &tenant, &a, dm, &[&agent], false) + .await + .expect("agent beyond nine"); + let tenth = Keys::generate(); + rejected( + add(&state, &tenant, &a, dm, &[&tenth], true).await, + "at most 9 people", + ); + rejected( + add(&state, &tenant, &a, dm, &[&tenth], false).await, + "at most 9 people", + ); +} + +/// Adding someone else's agent follows its `channel_add_policy`. +/// Mutation: skip the policy check → the agent joins → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn adding_an_agent_follows_its_add_policy() { + let state = state().await; + let tenant = community(&state, "dm-agent-policy").await; + let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &agent, &b, "owner_only").await; + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + + rejected( + add(&state, &tenant, &a, dm, &[&agent], false).await, + "policy:owner_only", + ); + rejected( + open(&state, &tenant, &a, &[&b, &agent]).await, + "policy:owner_only", + ); + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); + + let (target, _) = add(&state, &tenant, &b, dm, &[&agent], false) + .await + .expect("owner adds"); + assert_eq!(target, dm); +} + +/// An agent naming its owner's people cannot open or join their DM: it gets +/// its own DM with them, identified by all its members. +/// Mutation: return a found DM to a non-member, or identify an agent's DM by +/// its people → the agent gets the people's DM → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_agent_cannot_open_its_owners_dm_with_others() { + let state = state().await; + let tenant = community(&state, "dm-agent-actor").await; + let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &agent, &a, "anyone").await; + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + + let own = open(&state, &tenant, &agent, &[&a, &b]) + .await + .expect("agent open"); + + assert_ne!(own, dm); + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); +} + +/// A DM an agent starts with people does not become those people's DM, and +/// they still find it when they name the agent too. +/// Mutation: identify an agent's DM by its people → B finds it without the +/// agent → RED; drop the all-members lookup → B naming the agent forks → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_agents_group_dm_keeps_its_own_identity() { + let state = state().await; + let tenant = community(&state, "dm-agent-group").await; + let (owner, b, c, agent) = ( + Keys::generate(), + Keys::generate(), + Keys::generate(), + Keys::generate(), + ); + make_agent(&state, &tenant, &agent, &owner, "anyone").await; + let agents_dm = open(&state, &tenant, &agent, &[&b, &c]) + .await + .expect("agent open"); + + assert_eq!( + open(&state, &tenant, &b, &[&c, &agent]) + .await + .expect("open with agent"), + agents_dm + ); + let peoples = open(&state, &tenant, &b, &[&c]).await.expect("people open"); + assert_ne!(peoples, agents_dm); +} + +/// Only people share a DM's history: an agent's add with the tag is refused, +/// and without it starts a new DM, as before. +/// Mutation: let agents grow DMs → C joins the original DM → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_agent_cannot_share_a_dms_history() { + let state = state().await; + let tenant = community(&state, "dm-agent-history").await; + let (a, b, c, agent) = ( + Keys::generate(), + Keys::generate(), + Keys::generate(), + Keys::generate(), + ); + make_agent(&state, &tenant, &agent, &a, "anyone").await; + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + add(&state, &tenant, &a, dm, &[&agent], false) + .await + .expect("add agent"); + + rejected( + add(&state, &tenant, &agent, dm, &[&c], true).await, + "only people", + ); + let (target, created) = add(&state, &tenant, &agent, dm, &[&c], false) + .await + .expect("agent add"); + + assert!(created); + assert_ne!(target, dm); + assert_eq!( + members(&state, &tenant, dm).await, + set_of(&[&a, &b, &agent]) + ); +} + +/// Growing a DM never reveals another DM the actor is not in. +/// Mutation: return `Existing` without checking membership → the agent +/// learns the people's DM → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn growing_does_not_reveal_a_dm_the_actor_is_not_in() { + let state = state().await; + let tenant = community(&state, "dm-grow-reveal").await; + let (a, b, c, agent) = ( + Keys::generate(), + Keys::generate(), + Keys::generate(), + Keys::generate(), + ); + make_agent(&state, &tenant, &agent, &a, "anyone").await; + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + add(&state, &tenant, &a, dm, &[&agent], false) + .await + .expect("add agent"); + open(&state, &tenant, &a, &[&b, &c]).await.expect("trio"); + + let c_bytes = c.public_key().to_bytes(); + let result = state + .db + .grow_dm( + tenant.community(), + dm, + &agent.public_key().to_bytes(), + &[c_bytes.as_slice()], + ) + .await; + + assert!( + matches!(result, Err(buzz_db::DbError::AccessDenied(_))), + "{result:?}" + ); +} + +/// The relay's moderation DM stays two-party. +/// Mutation: drop the relay-key guard → a person joins → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn the_moderation_dm_cannot_grow() { + let state = state().await; + let tenant = community(&state, "dm-moderation").await; + let (a, b) = (Keys::generate(), Keys::generate()); + let relay = state.relay_keypair.public_key().to_bytes(); + let (dm, _) = state + .db + .open_dm(tenant.community(), &[&a.public_key().to_bytes()], &relay) + .await + .expect("moderation dm"); + + rejected( + add(&state, &tenant, &a, dm.id, &[&b], true).await, + "cannot be changed", + ); + rejected( + add(&state, &tenant, &a, dm.id, &[&b], false).await, + "cannot be changed", + ); +} + +/// Only `include` is a valid history value. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_unknown_history_value_is_refused() { + let state = state().await; + let tenant = community(&state, "dm-history-value").await; + let (a, b, c) = (Keys::generate(), Keys::generate(), Keys::generate()); + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + let tags = vec![ + vec!["h".into(), dm.to_string()], + vec!["p".into(), hex(&c)], + vec!["history".into(), "exclude".into()], + ]; + rejected( + submit(&state, &tenant, &a, 41011, tags).await.map(|_| ()), + "history", + ); +} + +/// A DM of just the caller and someone else's agent needs no add policy, as +/// before: anyone may message an agent. +/// Mutation: check the policy on every open → the DM is refused → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn messaging_an_agent_alone_ignores_its_add_policy() { + let state = state().await; + let tenant = community(&state, "dm-agent-alone").await; + let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &agent, &b, "nobody").await; + + let dm = open(&state, &tenant, &a, &[&agent]).await.expect("open"); + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &agent])); +} diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 3a0216ecdd3..0877964b569 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -23,6 +23,8 @@ mod create_with_members_postgres_tests; #[cfg(test)] mod deletion_postgres_tests; #[cfg(test)] +mod dm_growth_postgres_tests; +#[cfg(test)] mod edit_metadata_postgres_tests; /// EVENT handler — WS dispatcher → ingest pipeline → fan-out. pub mod event; diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index 9970e2f0002..c7f013097ca 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -1970,9 +1970,20 @@ pub fn build_dm_open(pubkeys: &[&str]) -> Result { } /// Build a DM add-member event (kind 41011). -pub fn build_dm_add_member(channel_id: Uuid, pubkey: &str) -> Result { +/// +/// With `include_history`, a person joins this DM and sees its history. +/// Without it, adding a person starts a new DM of everyone. Agents always join +/// this DM in place. +pub fn build_dm_add_member( + channel_id: Uuid, + pubkey: &str, + include_history: bool, +) -> Result { let pk = check_pubkey_hex(pubkey, "pubkey")?; - let tags = vec![tag(&["h", &channel_id.to_string()])?, tag(&["p", &pk])?]; + let mut tags = vec![tag(&["h", &channel_id.to_string()])?, tag(&["p", &pk])?]; + if include_history { + tags.push(tag(&["history", "include"])?); + } Ok(EventBuilder::new(Kind::Custom(KIND_DM_ADD_MEMBER as u16), "").tags(tags)) } @@ -5073,15 +5084,22 @@ mod tests { fn dm_add_member_happy_path() { let cid = uuid(); let pk = "b".repeat(64); - let ev = sign(build_dm_add_member(cid, &pk).unwrap()); + let ev = sign(build_dm_add_member(cid, &pk, false).unwrap()); assert_eq!(ev.kind.as_u16(), 41011); assert!(has_tag(&ev, "h", &cid.to_string())); assert!(has_tag(&ev, "p", &pk)); + assert!(!ev.tags.iter().any(|t| t.as_slice()[0] == "history")); + } + + #[test] + fn dm_add_member_with_history() { + let ev = sign(build_dm_add_member(uuid(), &"b".repeat(64), true).unwrap()); + assert!(has_tag(&ev, "history", "include")); } #[test] fn dm_add_member_rejects_bad_pubkey() { - let err = build_dm_add_member(uuid(), "short").unwrap_err(); + let err = build_dm_add_member(uuid(), "short", false).unwrap_err(); assert!(matches!(err, SdkError::InvalidInput(_))); } From cd42f94789e2631f7dedc0b138b43b3c03640d5a Mon Sep 17 00:00:00 2001 From: morgmart <98432065+morgmart@users.noreply.github.com> Date: Thu, 5 Nov 2026 12:00:41 -0800 Subject: [PATCH 2/8] fix(dms): keep DM identity in step with removals Generic membership kinds 9000, 9001 and 9022 are refused on DMs, so a DM's members change only through kind:41010/41011, which keep its identity current. Moderation and admin kicks remove members directly in the database and still work on DMs; any removal from a DM now releases its identity under the membership lock, so the DM keeps its history and opening that group again starts a new DM instead of a stale match. Signed-off-by: morgmart <98432065+morgmart@users.noreply.github.com> Co-authored-by: Carl Co-Authored-By: Claude Opus 5.5 --- crates/buzz-db/src/store/channel_members.rs | 1 + crates/buzz-db/src/store/dm.rs | 23 ++++++ .../buzz-db/src/store/relay_admin_actions.rs | 4 + .../src/handlers/dm_growth_postgres_tests.rs | 79 +++++++++++++++++++ .../buzz-relay/src/handlers/side_effects.rs | 11 +++ 5 files changed, 118 insertions(+) diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 048ea9dbda7..692ce3425e9 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -1198,6 +1198,7 @@ pub async fn remove_member( if result.rows_affected() == 0 { return Err(DbError::MemberNotFound(channel_id)); } + crate::dm::release_dm_identity(tx.conn(), community_id, channel_id).await?; tx.commit().await?; Ok(()) diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index c695c0434d2..8ffc414fdd2 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -653,6 +653,29 @@ pub async fn grow_dm( }) } +/// Stop `channel_id`, if it is a DM, from being any set of people's DM, after +/// a member was removed from it. Its fingerprint would otherwise name someone +/// no longer in it, so neither they nor the rest could reach that group again. +/// The DM keeps its history for those still in it; opening that group again +/// starts a new DM. The caller holds the channel's membership lock, as every +/// membership writer does, so this cannot race [`grow_dm`]. +pub(crate) async fn release_dm_identity<'e>( + executor: impl sqlx::PgExecutor<'e>, + community_id: CommunityId, + channel_id: Uuid, +) -> Result<()> { + sqlx::query( + "UPDATE channels SET participant_hash = NULL \ + WHERE community_id = $1 AND id = $2 AND channel_type = 'dm' \ + AND participant_hash IS NOT NULL", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .execute(executor) + .await?; + Ok(()) +} + /// [`DmGrowth::Existing`] for `actor`, who must be an active member of that /// DM: growing a DM must not reveal a DM the actor is not in. async fn existing_for( diff --git a/crates/buzz-db/src/store/relay_admin_actions.rs b/crates/buzz-db/src/store/relay_admin_actions.rs index fc2fb32c987..5dc4a43c2b8 100644 --- a/crates/buzz-db/src/store/relay_admin_actions.rs +++ b/crates/buzz-db/src/store/relay_admin_actions.rs @@ -736,6 +736,7 @@ pub async fn execute_kick_with_marker( tx.rollback().await?; return Ok(KickWithMarkerResult::AlreadyGone); } + crate::dm::release_dm_identity(tx.conn(), community_id, channel_id).await?; let marker = sqlx::query( r#" @@ -1742,6 +1743,9 @@ pub async fn deploy_kick_member( .bind(target_pubkey) .execute(tx.conn()) .await?; + if result.rows_affected() > 0 { + crate::dm::release_dm_identity(tx.conn(), community_id, channel_id).await?; + } tx.commit().await?; if result.rows_affected() > 0 { diff --git a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs index 769e87d19df..54b273025ea 100644 --- a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs @@ -535,3 +535,82 @@ async fn messaging_an_agent_alone_ignores_its_add_policy() { let dm = open(&state, &tenant, &a, &[&agent]).await.expect("open"); assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &agent])); } + +/// A DM's members change only through kind:41011: a plain add, a removal and +/// a leave are refused, so its identity never names someone who is not in it +/// and the people can always reopen it. +/// Mutation: drop either DM guard → the add or the leave lands → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn generic_membership_kinds_cannot_change_a_dm() { + let state = state().await; + let tenant = community(&state, "dm-generic-membership").await; + let (a, b, c) = (Keys::generate(), Keys::generate(), Keys::generate()); + let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + let h = vec!["h".to_string(), dm.to_string()]; + + rejected( + submit( + &state, + &tenant, + &a, + 9000, + vec![h.clone(), vec!["p".into(), hex(&c)]], + ) + .await + .map(|r| (r.accepted, r.message)), + "kind:41011", + ); + rejected( + submit( + &state, + &tenant, + &b, + 9001, + vec![h.clone(), vec!["p".into(), hex(&b)]], + ) + .await + .map(|r| (r.accepted, r.message)), + "kind:41011", + ); + rejected( + submit(&state, &tenant, &b, 9022, vec![h]) + .await + .map(|r| (r.accepted, r.message)), + "kind:41011", + ); + + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); + assert_eq!(open(&state, &tenant, &b, &[&a]).await.expect("reopen"), dm); +} + +/// A moderation kick from a DM still removes the member, and the DM stops +/// being any group's DM: the person removed, like the rest, can message that +/// group again, in a new DM. +/// Mutation: keep the DM's identity after the kick → the removed person is +/// refused ("already exists") → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_kick_from_a_dm_releases_its_identity() { + let state = state().await; + let tenant = community(&state, "dm-kick").await; + let (a, b, c) = (Keys::generate(), Keys::generate(), Keys::generate()); + let dm = open(&state, &tenant, &a, &[&b, &c]).await.expect("open"); + + state + .db + .deploy_kick_member( + tenant.community(), + dm, + &c.public_key().to_bytes(), + &a.public_key().to_bytes(), + ) + .await + .expect("kick"); + + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); + let again = open(&state, &tenant, &c, &[&a, &b]) + .await + .expect("message the group again"); + assert_ne!(again, dm); +} diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 0bcc267a3cb..2de24d72687 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -621,6 +621,9 @@ async fn actor_owns_any_owner_agent( Ok(false) } +/// Why kinds 9000, 9001 and 9022 are refused on a DM. +const DM_MEMBERSHIP_REFUSAL: &str = "DM members change only through kind:41011"; + /// Validate an admin kind event BEFORE storage. pub async fn validate_admin_event( tenant: &TenantContext, @@ -675,6 +678,11 @@ pub async fn validate_admin_event( if kind != 9002 && channel.archived_at.is_some() && !is_self_departure(event) { return Err(anyhow::anyhow!("channel is archived")); } + // A DM is identified by its people, so its members change only through + // kind:41011, which keeps that identity in step. + if channel.channel_type == "dm" && matches!(kind, 9001 | 9022) { + return Err(anyhow::anyhow!(DM_MEMBERSHIP_REFUSAL)); + } match kind { // Ingest validates kind:9000 with `validate_put_user`, which keeps @@ -1930,6 +1938,9 @@ pub(crate) async fn validate_put_user( if channel.archived_at.is_some() { anyhow::bail!("channel is archived"); } + if channel.channel_type == "dm" { + anyhow::bail!(DM_MEMBERSHIP_REFUSAL); + } let targets = put_user_targets(event)?; let actor_bytes = event.pubkey.to_bytes().to_vec(); From 4b619d5be5bb2e7d531c1702a9fd9817fee6f460 Mon Sep 17 00:00:00 2001 From: morgmart <98432065+morgmart@users.noreply.github.com> Date: Thu, 5 Nov 2026 12:15:49 -0800 Subject: [PATCH 3/8] fix(dms): keep a group DM's identity when an agent is removed A removal releases a DM's identity only when the remaining members no longer have it. Agents are not part of a DM of people's identity, so kicking one leaves the DM as that group's DM; a DM of someone and their agents, identified by all its members, is still released. Signed-off-by: morgmart <98432065+morgmart@users.noreply.github.com> Co-authored-by: Carl Co-Authored-By: Claude Opus 5.5 --- crates/buzz-db/src/store/dm.rs | 47 ++++++++++++++----- .../src/handlers/dm_growth_postgres_tests.rs | 43 +++++++++++++++++ 2 files changed, 79 insertions(+), 11 deletions(-) diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index 8ffc414fdd2..5641aabf7ee 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -653,25 +653,50 @@ pub async fn grow_dm( }) } -/// Stop `channel_id`, if it is a DM, from being any set of people's DM, after -/// a member was removed from it. Its fingerprint would otherwise name someone -/// no longer in it, so neither they nor the rest could reach that group again. -/// The DM keeps its history for those still in it; opening that group again -/// starts a new DM. The caller holds the channel's membership lock, as every -/// membership writer does, so this cannot race [`grow_dm`]. -pub(crate) async fn release_dm_identity<'e>( - executor: impl sqlx::PgExecutor<'e>, +/// Keep `channel_id`'s fingerprint, if it is a DM, true to who is still in it +/// after a member was removed. When the remaining members no longer have that +/// fingerprint (a person left, or the DM was identified by all its members), +/// the DM stops being any group's DM: it keeps its history for those still in +/// it, and opening that group again starts a new DM. Removing an agent from a +/// DM of people changes nothing. The caller holds the channel's membership +/// lock, as every membership writer does, so this cannot race [`grow_dm`]. +pub(crate) async fn release_dm_identity( + conn: &mut sqlx::PgConnection, community_id: CommunityId, channel_id: Uuid, ) -> Result<()> { + let stored: Option> = sqlx::query_scalar( + "SELECT participant_hash FROM channels \ + WHERE community_id = $1 AND id = $2 AND channel_type = 'dm'", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_optional(&mut *conn) + .await? + .flatten(); + let Some(stored) = stored else { + return Ok(()); + }; + let members: Vec> = sqlx::query_scalar( + "SELECT pubkey FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 AND removed_at IS NULL", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_all(&mut *conn) + .await?; + let members: Vec<&[u8]> = members.iter().map(Vec::as_slice).collect(); + let agents = agents_among(&mut *conn, community_id, &members).await?; + if dm_identity_hash(&members, &agents).as_slice() == stored.as_slice() { + return Ok(()); + } sqlx::query( "UPDATE channels SET participant_hash = NULL \ - WHERE community_id = $1 AND id = $2 AND channel_type = 'dm' \ - AND participant_hash IS NOT NULL", + WHERE community_id = $1 AND id = $2", ) .bind(community_id.as_uuid()) .bind(channel_id) - .execute(executor) + .execute(&mut *conn) .await?; Ok(()) } diff --git a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs index 54b273025ea..5d3c6d06e80 100644 --- a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs @@ -614,3 +614,46 @@ async fn a_kick_from_a_dm_releases_its_identity() { .expect("message the group again"); assert_ne!(again, dm); } + +/// Agents are not part of a DM of people's identity, so kicking one leaves +/// the DM as that group's DM; kicking an agent from a DM of someone and +/// their agent releases it, since all its members identify it. +/// Mutation: release on every removal → the group gets a new DM → RED; +/// never release when an agent is kicked → the owner's reopen finds the +/// emptied agent DM → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn kicking_an_agent_keeps_a_group_dm_its_identity() { + let state = state().await; + let tenant = community(&state, "dm-kick-agent").await; + let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &agent, &a, "anyone").await; + let group = open(&state, &tenant, &a, &[&b, &agent]) + .await + .expect("group"); + let own = open(&state, &tenant, &a, &[&agent]).await.expect("own"); + + for dm in [group, own] { + state + .db + .deploy_kick_member( + tenant.community(), + dm, + &agent.public_key().to_bytes(), + &a.public_key().to_bytes(), + ) + .await + .expect("kick"); + } + + assert_eq!( + open(&state, &tenant, &b, &[&a]).await.expect("reopen"), + group + ); + assert_ne!( + open(&state, &tenant, &a, &[&agent]) + .await + .expect("own again"), + own + ); +} From db192bf372b66372f7ef64c8dd509fd1a4869972 Mon Sep 17 00:00:00 2001 From: morgmart <98432065+morgmart@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:26:48 -0800 Subject: [PATCH 4/8] fix(dms): let people leave DMs and owners remove their agents Refusing kinds 9001 and 9022 on DMs took away leaving a DM, which clients offer today, and broke agent deletion, which removes the agent from each of its channels with kind:9001. Every removal already releases the DM's identity when its people change, so a DM again accepts kind:9022, and kind:9001 for yourself or your own agent. People still join only through kind:41011. Signed-off-by: morgmart <98432065+morgmart@users.noreply.github.com> Co-authored-by: Carl Co-Authored-By: Claude Opus 5.5 --- .../src/handlers/dm_growth_postgres_tests.rs | 70 +++++++++++++++++-- .../buzz-relay/src/handlers/side_effects.rs | 21 ++++-- 2 files changed, 79 insertions(+), 12 deletions(-) diff --git a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs index 5d3c6d06e80..86af6c04558 100644 --- a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs @@ -536,10 +536,9 @@ async fn messaging_an_agent_alone_ignores_its_add_policy() { assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &agent])); } -/// A DM's members change only through kind:41011: a plain add, a removal and -/// a leave are refused, so its identity never names someone who is not in it -/// and the people can always reopen it. -/// Mutation: drop either DM guard → the add or the leave lands → RED. +/// People join a DM only through kind:41011, so a plain add, and removing +/// someone other than yourself or your own agent, are refused. +/// Mutation: drop either DM guard → the add or the removal lands → RED. #[tokio::test] #[ignore = "requires Postgres"] async fn generic_membership_kinds_cannot_change_a_dm() { @@ -565,20 +564,77 @@ async fn generic_membership_kinds_cannot_change_a_dm() { submit( &state, &tenant, - &b, + &a, 9001, - vec![h.clone(), vec!["p".into(), hex(&b)]], + vec![h, vec!["p".into(), hex(&b)]], ) .await .map(|r| (r.accepted, r.message)), "kind:41011", ); + + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); + assert_eq!(open(&state, &tenant, &b, &[&a]).await.expect("reopen"), dm); +} + +/// Anyone can leave a DM. The DM then stops being that group's DM, so the +/// person who left can message the group again, in a new DM. +/// Mutation: refuse kind:9022 on DMs → the leave is refused → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_person_can_leave_a_dm() { + let state = state().await; + let tenant = community(&state, "dm-leave").await; + let (a, b, c) = (Keys::generate(), Keys::generate(), Keys::generate()); + let dm = open(&state, &tenant, &a, &[&b, &c]).await.expect("open"); + + let left = submit( + &state, + &tenant, + &c, + 9022, + vec![vec!["h".into(), dm.to_string()]], + ) + .await + .expect("leave"); + assert!(left.accepted, "{}", left.message); + + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); + let again = open(&state, &tenant, &c, &[&a, &b]) + .await + .expect("message the group again"); + assert_ne!(again, dm); +} + +/// An owner can remove their own agent from a DM, as deleting an agent does; +/// the DM stays its people's DM. Someone else's agent cannot be removed. +/// Mutation: refuse every kind:9001 for another member on DMs → the owner's +/// removal is refused → RED; allow any → B removes A's agent → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_owner_can_remove_their_agent_from_a_dm() { + let state = state().await; + let tenant = community(&state, "dm-remove-agent").await; + let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &agent, &a, "anyone").await; + let dm = open(&state, &tenant, &a, &[&b, &agent]) + .await + .expect("open"); + let remove = vec![ + vec!["h".to_string(), dm.to_string()], + vec!["p".into(), hex(&agent)], + ]; + rejected( - submit(&state, &tenant, &b, 9022, vec![h]) + submit(&state, &tenant, &b, 9001, remove.clone()) .await .map(|r| (r.accepted, r.message)), "kind:41011", ); + let removed = submit(&state, &tenant, &a, 9001, remove) + .await + .expect("remove"); + assert!(removed.accepted, "{}", removed.message); assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); assert_eq!(open(&state, &tenant, &b, &[&a]).await.expect("reopen"), dm); diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 2de24d72687..ad6eb8754f0 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -621,7 +621,8 @@ async fn actor_owns_any_owner_agent( Ok(false) } -/// Why kinds 9000, 9001 and 9022 are refused on a DM. +/// Why kind:9000, and kind:9001 for anyone but yourself or your own agent, +/// are refused on a DM. const DM_MEMBERSHIP_REFUSAL: &str = "DM members change only through kind:41011"; /// Validate an admin kind event BEFORE storage. @@ -678,10 +679,20 @@ pub async fn validate_admin_event( if kind != 9002 && channel.archived_at.is_some() && !is_self_departure(event) { return Err(anyhow::anyhow!("channel is archived")); } - // A DM is identified by its people, so its members change only through - // kind:41011, which keeps that identity in step. - if channel.channel_type == "dm" && matches!(kind, 9001 | 9022) { - return Err(anyhow::anyhow!(DM_MEMBERSHIP_REFUSAL)); + // A DM is identified by its people, so people join it only through + // kind:41011, which keeps that identity in step. Anyone may leave, and + // an owner may remove their own agent: every removal releases the DM's + // identity when its people change (`release_dm_identity`). + if channel.channel_type == "dm" && kind == 9001 { + let target = extract_p_tag(event).ok_or_else(|| anyhow::anyhow!("missing p tag"))?; + if target != actor_bytes + && !state + .db + .is_agent_owner(tenant.community(), &target, &actor_bytes) + .await? + { + return Err(anyhow::anyhow!(DM_MEMBERSHIP_REFUSAL)); + } } match kind { From 88358944c553ad802c5425f3115309df969a6f60 Mon Sep 17 00:00:00 2001 From: morgmart <98432065+morgmart@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:52:37 -0800 Subject: [PATCH 5/8] fix(dms): close the moderation, agent-policy and one-person identity gaps - Opening a DM can no longer put anyone else in the relay's moderation DM. - Opening a DM checks each agent's add policy unless it is just the caller and that one agent. - An agent joining a DM of at most one person moves its identity to all its members, so reopening the original pair starts a new DM. Signed-off-by: morgmart <98432065+morgmart@users.noreply.github.com> Co-authored-by: Carl Co-Authored-By: Claude Opus 5.5 --- crates/buzz-db/src/store/dm.rs | 13 ++-- .../src/handlers/command_executor.rs | 24 +++++-- .../src/handlers/dm_growth_postgres_tests.rs | 69 ++++++++++++++++++- 3 files changed, 92 insertions(+), 14 deletions(-) diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index 5641aabf7ee..a8d5619e14c 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -496,9 +496,9 @@ pub enum DmGrowth { /// Add `additions` to the DM `channel_id` in place, keeping its history. /// -/// Agents join without changing which DM this is. When people join, the DM's -/// identity moves to the grown set of people ([`dm_identity_hash`]), so -/// opening a DM with the original people afterwards starts a new one. If a DM +/// Agents join a DM of two or more people without changing which DM it is. +/// Otherwise the DM's identity moves to the grown set ([`dm_identity_hash`]), so +/// opening a DM with the original members afterwards starts a new one. If a DM /// with the grown set already exists, nothing is written and that DM is /// returned as [`DmGrowth::Existing`], provided `actor` is in it. /// @@ -586,8 +586,11 @@ pub async fn grow_dm( ))); } - if count_people(&added, &agents) > 0 { - let hash = dm_identity_hash(&grown, &agents); + // People move the DM's identity to the grown set of people. So do agents + // joining a DM of at most one person, which is identified by all members. + let hash = dm_identity_hash(&grown, &agents); + let current: Vec<&[u8]> = members.iter().map(Vec::as_slice).collect(); + if hash != dm_identity_hash(¤t, &agents) { let existing = sqlx::query_scalar::<_, Uuid>( "SELECT id FROM channels \ WHERE community_id = $1 AND participant_hash = $2 AND id <> $3 \ diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 33eb0b5c662..4b367e0bb83 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -412,6 +412,9 @@ async fn announce_dm_created( } } +/// Why the relay's moderation DM cannot take anyone else. +const MODERATION_DM_REFUSAL: &str = "invalid: this DM cannot be changed"; + /// The agents among `pubkeys` (users with an owner): each one's /// `channel_add_policy` and owner. async fn dm_agents( @@ -467,12 +470,21 @@ async fn handle_dm_open( )); } - // 2. At most 8 other people (9 total); agents do not count. Putting an - // agent in a DM with other people needs its channel_add_policy; a DM - // of just the caller and agents does not. + // 2. The relay's moderation DM stays two-party. At most 8 other people + // (9 total); agents do not count. Putting an agent in a DM with anyone + // besides the caller needs its channel_add_policy; messaging one alone + // does not. + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); + if others.len() > 1 + && others + .iter() + .any(|pk| pk.as_slice() == relay_pubkey.as_slice()) + { + return Err(IngestError::Rejected(MODERATION_DM_REFUSAL.into())); + } let agents = dm_agents(tenant, state, &others).await?; let other_people = others.iter().filter(|pk| !agents.contains_key(*pk)).count(); - if other_people > 0 { + if others.len() > 1 { check_dm_agent_policies(&agents, &self_bytes)?; } if other_people > MAX_DM_PEOPLE - 1 { @@ -641,9 +653,7 @@ async fn handle_dm_add_member( .iter() .any(|m| m.as_slice() == relay_pubkey.as_slice()) { - return Err(IngestError::Rejected( - "invalid: this DM cannot be changed".into(), - )); + return Err(IngestError::Rejected(MODERATION_DM_REFUSAL.into())); } let mut additions: Vec> = Vec::new(); diff --git a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs index 86af6c04558..3827cb405e4 100644 --- a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs @@ -477,8 +477,9 @@ async fn growing_does_not_reveal_a_dm_the_actor_is_not_in() { ); } -/// The relay's moderation DM stays two-party. -/// Mutation: drop the relay-key guard → a person joins → RED. +/// The relay's moderation DM stays two-party, whether added to or opened with +/// someone else. +/// Mutation: drop either relay-key guard → a member joins → RED. #[tokio::test] #[ignore = "requires Postgres"] async fn the_moderation_dm_cannot_grow() { @@ -500,6 +501,16 @@ async fn the_moderation_dm_cannot_grow() { add(&state, &tenant, &a, dm.id, &[&b], false).await, "cannot be changed", ); + let agent = Keys::generate(); + make_agent(&state, &tenant, &agent, &a, "anyone").await; + rejected( + open(&state, &tenant, &a, &[&state.relay_keypair, &agent]).await, + "cannot be changed", + ); + assert_eq!( + members(&state, &tenant, dm.id).await, + set_of(&[&a, &state.relay_keypair]) + ); } /// Only `include` is a valid history value. @@ -713,3 +724,57 @@ async fn kicking_an_agent_keeps_a_group_dm_its_identity() { own ); } + +/// Opening a DM that puts someone else's agent with anyone besides the caller +/// follows its add policy, even when no other person is in it. +/// Mutation: check policies only when people are named → the DM opens → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn opening_with_several_agents_follows_their_add_policy() { + let state = state().await; + let tenant = community(&state, "dm-agents-policy").await; + let (a, b) = (Keys::generate(), Keys::generate()); + let (theirs, mine) = (Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &theirs, &b, "nobody").await; + make_agent(&state, &tenant, &mine, &a, "owner_only").await; + + rejected( + open(&state, &tenant, &a, &[&theirs, &mine]).await, + "policy:nobody", + ); +} + +/// Adding an agent to a DM of one person and their agent moves its identity +/// to all its members: the original pair gets a new DM. +/// Mutation: keep the old identity → reopening the pair finds the grown DM → +/// RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn an_agent_joining_a_one_person_dm_moves_its_identity() { + let state = state().await; + let tenant = community(&state, "dm-one-person-grow").await; + let (a, b) = (Keys::generate(), Keys::generate()); + let (mine, theirs) = (Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &mine, &a, "anyone").await; + make_agent(&state, &tenant, &theirs, &b, "anyone").await; + let dm = open(&state, &tenant, &a, &[&mine]).await.expect("open"); + + let (target, _) = add(&state, &tenant, &a, dm, &[&theirs], false) + .await + .expect("add agent"); + assert_eq!(target, dm); + assert_eq!( + members(&state, &tenant, dm).await, + set_of(&[&a, &mine, &theirs]) + ); + assert_ne!( + open(&state, &tenant, &a, &[&mine]).await.expect("reopen"), + dm + ); + assert_eq!( + open(&state, &tenant, &a, &[&mine, &theirs]) + .await + .expect("grown"), + dm + ); +} From 1d1b80862bca11855c3f523457557ccc1d4c28e6 Mon Sep 17 00:00:00 2001 From: morgmart <98432065+morgmart@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:08:03 -0800 Subject: [PATCH 6/8] fix(dms): join a DM in place only with the history tag Without ["history","include"], kind:41010 and kind:41011 behave as before members could join in place: opening finds or makes the DM of exactly the named members, agents included, and any add starts a new DM of everyone. With the tag, opening finds the people's DM and adds the named agents, and adds join in place. An older client can no longer put an agent into an existing DM, and show it the history, without the user seeing a warning. Signed-off-by: morgmart <98432065+morgmart@users.noreply.github.com> Co-authored-by: Carl Co-Authored-By: Claude Opus 5.5 --- crates/buzz-cli/src/lib.rs | 4 +- .../src/store/admission_postgres_tests.rs | 14 ++- crates/buzz-db/src/store/dm.rs | 56 ++++++++---- .../src/handlers/command_executor.rs | 58 ++++++------ .../src/handlers/dm_growth_postgres_tests.rs | 90 +++++++++++++++---- .../handlers/group_state_postgres_tests.rs | 1 + .../src/handlers/moderation_notices.rs | 1 + crates/buzz-sdk/src/builders.rs | 5 +- 8 files changed, 152 insertions(+), 77 deletions(-) diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 3477112be62..bf1a827fb4d 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1326,8 +1326,8 @@ pub enum DmsCmd { /// User pubkey to add (64-char hex) #[arg(long)] pubkey: String, - /// Add a person to this DM with its history, instead of starting a new - /// DM of everyone. Agents always join this DM. + /// Add them to this DM with its history, instead of starting a new DM + /// of everyone. #[arg(long)] include_history: bool, }, diff --git a/crates/buzz-db/src/store/admission_postgres_tests.rs b/crates/buzz-db/src/store/admission_postgres_tests.rs index c5b23f41e6a..61e688e853d 100644 --- a/crates/buzz-db/src/store/admission_postgres_tests.rs +++ b/crates/buzz-db/src/store/admission_postgres_tests.rs @@ -491,7 +491,7 @@ async fn dm_writers_require_application_admission() { let participants: [&[u8]; 2] = [&[1; 32], &[2; 32]]; for community in &communities[1..] { assert_fenced( - super::dm::create_dm(&db.pool, *community, &participants, participants[0]) + super::dm::create_dm(&db.pool, *community, &participants, participants[0], false) .await .expect_err("DM create must reject"), ); @@ -501,9 +501,15 @@ async fn dm_writers_require_application_admission() { .expect_err("DM hide must reject"), ); } - super::dm::create_dm(&db.pool, communities[0], &participants, participants[0]) - .await - .expect("active DM create succeeds"); + super::dm::create_dm( + &db.pool, + communities[0], + &participants, + participants[0], + false, + ) + .await + .expect("active DM create succeeds"); } #[tokio::test] diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index a8d5619e14c..58d5fc0f336 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -88,14 +88,20 @@ pub fn dm_identity_hash(members: &[&[u8]], agents: &HashSet>) -> [u8; 32 } } -/// The fingerprint a DM of `members` opened by `creator` gets. An agent -/// opening a DM gets one identified by all its members, as before agents were -/// told apart: only people decide which DM their people share. -fn opened_identity_hash(members: &[&[u8]], agents: &HashSet>, creator: &[u8]) -> [u8; 32] { - if agents.contains(creator) { - compute_participant_hash(members) - } else { +/// The fingerprint a DM of `members` opened by `creator` gets. With +/// `by_people`, the DM of its people ([`dm_identity_hash`]). Otherwise, and +/// always for an agent, the DM of exactly these members, as before agents were +/// told apart: only people, choosing to, decide which DM their people share. +fn opened_identity_hash( + members: &[&[u8]], + agents: &HashSet>, + creator: &[u8], + by_people: bool, +) -> [u8; 32] { + if by_people && !agents.contains(creator) { dm_identity_hash(members, agents) + } else { + compute_participant_hash(members) } } @@ -163,13 +169,14 @@ pub async fn find_dm_by_participants( /// - `participants` must contain at least 2 entries, and at most /// [`MAX_DM_PEOPLE`] people; agents do not count (enforced here). /// - `created_by` must be one of the participants. -/// - The operation is idempotent: the same people (see [`dm_identity_hash`]) -/// -> the same channel returned. +/// - The operation is idempotent: the same identity (see +/// [`opened_identity_hash`] for `by_people`) -> the same channel returned. pub async fn create_dm( pool: &PgPool, community_id: CommunityId, participants: &[&[u8]], created_by: &[u8], + by_people: bool, ) -> Result { if participants.len() < 2 { return Err(DbError::InvalidData( @@ -191,7 +198,7 @@ pub async fn create_dm( ))); } - let hash = opened_identity_hash(participants, &agents, created_by); + let hash = opened_identity_hash(participants, &agents, created_by, by_people); let mut tx = crate::begin_community_write_transaction( pool, @@ -424,11 +431,12 @@ pub async fn list_dms_for_user( /// `created_by` is automatically added to `pubkeys` if not already present, /// ensuring the caller is always a participant in their own DM. /// -/// The DM is found by its people ([`dm_identity_hash`]), so it may hold agents -/// that `pubkeys` does not name, and lack agents it does. Adding those is -/// [`grow_dm`]'s job. A DM made before agents were told apart is still found by -/// all its members. An agent opening a DM finds or makes one identified by all -/// its members. A found DM is returned only to an active member of it: +/// With `by_people`, the DM is found by its people ([`dm_identity_hash`]), so +/// it may hold agents that `pubkeys` does not name, and lack agents it does. +/// Adding those is [`grow_dm`]'s job. A DM made before agents were told apart +/// is still found by all its members. Without `by_people`, and always for an +/// agent opening a DM, only a DM of exactly these members, agents included, is +/// found or made. A found DM is returned only to an active member of it: /// an agent naming its people cannot use this to see or join their DM. /// /// Returns `(channel, was_created)`: @@ -439,6 +447,7 @@ pub async fn open_dm( community_id: CommunityId, pubkeys: &[&[u8]], created_by: &[u8], + by_people: bool, ) -> Result<(ChannelRecord, bool)> { let mut all: Vec<&[u8]> = vec![created_by]; for pk in pubkeys { @@ -454,7 +463,7 @@ pub async fn open_dm( ))); } - let hash = opened_identity_hash(&all, &agents, created_by); + let hash = opened_identity_hash(&all, &agents, created_by, by_people); // DMs made before agents were told apart are identified by all members. let legacy = compute_participant_hash(&all); @@ -475,7 +484,7 @@ pub async fn open_dm( } // Create new DM. - let channel = create_dm(pool, community_id, &all, created_by).await?; + let channel = create_dm(pool, community_id, &all, created_by, by_people).await?; Ok((channel, true)) } @@ -886,8 +895,16 @@ impl Db { community_id: CommunityId, participants: &[&[u8]], created_by: &[u8], + by_people: bool, ) -> Result { - crate::dm::create_dm(&self.pool, community_id, participants, created_by).await + crate::dm::create_dm( + &self.pool, + community_id, + participants, + created_by, + by_people, + ) + .await } /// List all DMs for a user. @@ -909,8 +926,9 @@ impl Db { community_id: CommunityId, pubkeys: &[&[u8]], created_by: &[u8], + by_people: bool, ) -> Result<(ChannelRecord, bool)> { - crate::dm::open_dm(&self.pool, community_id, pubkeys, created_by).await + crate::dm::open_dm(&self.pool, community_id, pubkeys, created_by, by_people).await } /// Add members to a DM in place, keeping its history. diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 4b367e0bb83..194fc2db248 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -412,6 +412,19 @@ async fn announce_dm_created( } } +/// Whether `event` carries `["history", "include"]`: the caller has agreed to +/// share an existing DM's history with whoever joins it. Without it, DM +/// commands behave as before members could join in place. +fn history_opt_in(event: &Event) -> Result { + match extract_tag(event, "history").as_deref() { + None => Ok(false), + Some("include") => Ok(true), + Some(other) => Err(IngestError::Rejected(format!( + "invalid: history must be \"include\", got \"{other}\"" + ))), + } +} + /// Why the relay's moderation DM cannot take anyone else. const MODERATION_DM_REFUSAL: &str = "invalid: this DM cannot be changed"; @@ -455,6 +468,7 @@ async fn handle_dm_open( auth: &IngestAuth, ) -> Result { let self_bytes = auth.pubkey().to_bytes().to_vec(); + let include_history = history_opt_in(event)?; // 1. Decode the other participants from `p` tags (deduplicated). let mut others: Vec> = Vec::new(); @@ -509,15 +523,15 @@ async fn handle_dm_open( PersistResult::Inserted(tx) => tx, }; - // 3. Execute: open the DM of these people, then add any named agents it - // does not hold yet. + // 3. Execute. With history, open the DM of these people and add any named + // agents it does not hold yet; without, the DM of exactly these members. let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect(); let (channel, was_created) = state .db - .open_dm(tenant.community(), &all_refs, &self_bytes) + .open_dm(tenant.community(), &all_refs, &self_bytes, include_history) .await .map_err(dm_write_error)?; - let added = if was_created { + let added = if was_created || !include_history { Vec::new() } else { grow_with_agents(tenant, state, channel.id, &self_bytes, &agents).await? @@ -582,12 +596,11 @@ async fn grow_with_agents( /// kind:41011: add people and agents to a DM. /// -/// Agents always join in place, with the DM's history. People join in place -/// too when the event carries `["history", "include"]`; the DM's identity then -/// moves to the grown set of people, or, if a DM of those people already -/// exists, that DM is returned instead and nothing changes here. Without the -/// tag, people start a new DM of everyone (no history), as before. An agent -/// cannot send the tag, and its adds always start a new DM of everyone. +/// With `["history", "include"]`, people and agents join in place, with the +/// DM's history. People move the DM's identity to the grown set of people, or, +/// if a DM of those people already exists, that DM is returned instead and +/// nothing changes here. Without the tag, everyone starts a new DM of exactly +/// its members (no history), as before. An agent cannot send the tag. async fn handle_dm_add_member( tenant: &TenantContext, state: &Arc, @@ -608,15 +621,7 @@ async fn handle_dm_add_member( "invalid: must specify at least 1 new participant in p tags".into(), )); } - let include_history = match extract_tag(event, "history").as_deref() { - None => false, - Some("include") => true, - Some(other) => { - return Err(IngestError::Rejected(format!( - "invalid: history must be \"include\", got \"{other}\"" - ))); - } - }; + let include_history = history_opt_in(event)?; // 2. Validate caller is member of existing DM let is_member = state @@ -665,8 +670,7 @@ async fn handle_dm_add_member( } let agents = dm_agents(tenant, state, &additions).await?; check_dm_agent_policies(&agents, &self_bytes)?; - let adds_people = additions.iter().any(|pk| !agents.contains_key(pk)); - // Only people grow a DM or share its history. An agent's add starts a new + // Only people grow a DM or share its history. Any other add starts a new // DM of everyone, as every add did before. let actor_is_agent = !dm_agents(tenant, state, std::slice::from_ref(&self_bytes)) .await? @@ -676,7 +680,6 @@ async fn handle_dm_add_member( "forbidden: only people can share a DM's history".into(), )); } - let in_place = !actor_is_agent && (include_history || !adds_people); // Persist the command event — returns open transaction let tx = match persist_command_event(&state.db, tenant, event, None).await? { @@ -692,7 +695,7 @@ async fn handle_dm_add_member( // 5. Execute. let addition_refs: Vec<&[u8]> = additions.iter().map(Vec::as_slice).collect(); - let (target, was_created, added, moved_elsewhere) = if in_place { + let (target, was_created, added, moved_elsewhere) = if include_history { // Grow this DM in place. match state .db @@ -714,15 +717,10 @@ async fn handle_dm_add_member( let all_refs: Vec<&[u8]> = all_bytes.iter().map(Vec::as_slice).collect(); let (channel, created) = state .db - .open_dm(tenant.community(), &all_refs, &self_bytes) + .open_dm(tenant.community(), &all_refs, &self_bytes, false) .await .map_err(dm_write_error)?; - let added = if created { - Vec::new() - } else { - grow_with_agents(tenant, state, channel.id, &self_bytes, &agents).await? - }; - (channel.id, created, added, !created) + (channel.id, created, Vec::new(), !created) }; // Finalize the idempotency record after the separate mutation succeeds. diff --git a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs index 3827cb405e4..a746d57ad46 100644 --- a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs @@ -60,7 +60,30 @@ async fn open( actor: &Keys, others: &[&Keys], ) -> Result { - let tags = others.iter().map(|k| vec!["p".into(), hex(k)]).collect(); + open_tagged(state, tenant, actor, others, false).await +} + +/// Open a DM with `others`, agreeing to share an existing DM's history. +async fn open_with_history( + state: &Arc, + tenant: &TenantContext, + actor: &Keys, + others: &[&Keys], +) -> Result { + open_tagged(state, tenant, actor, others, true).await +} + +async fn open_tagged( + state: &Arc, + tenant: &TenantContext, + actor: &Keys, + others: &[&Keys], + history: bool, +) -> Result { + let mut tags: Vec> = others.iter().map(|k| vec!["p".into(), hex(k)]).collect(); + if history { + tags.push(vec!["history".into(), "include".into()]); + } submit(state, tenant, actor, 41010, tags) .await .map(|r| response_channel(&r).0) @@ -223,9 +246,12 @@ async fn growing_into_an_existing_group_returns_that_dm() { ); } -/// An agent joins in place, with or without the history tag, and never -/// changes which DM it is: the same people still find it. -/// Mutation: fork on agents without the tag → a different channel → RED. +/// With the history tag, an agent joins in place and never changes which DM it +/// is: the same people still find it. Without it, the agent starts a new DM of +/// everyone, as before, and this DM is untouched: an older client cannot share +/// a DM's history unseen. +/// Mutations: fork on agents with the tag → a different channel → RED; join +/// in place without it → the same channel → RED. #[tokio::test] #[ignore = "requires Postgres"] async fn an_agent_joins_in_place_without_changing_the_dm() { @@ -235,9 +261,16 @@ async fn an_agent_joins_in_place_without_changing_the_dm() { make_agent(&state, &tenant, &agent, &a, "owner_only").await; let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); - let (target, created) = add(&state, &tenant, &a, dm, &[&agent], false) + let (forked, created) = add(&state, &tenant, &a, dm, &[&agent], false) .await - .expect("add"); + .expect("add without history"); + assert!(created); + assert_ne!(forked, dm); + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); + + let (target, created) = add(&state, &tenant, &a, dm, &[&agent], true) + .await + .expect("add with history"); assert_eq!((target, created), (dm, false)); assert_eq!( @@ -247,9 +280,11 @@ async fn an_agent_joins_in_place_without_changing_the_dm() { assert_eq!(open(&state, &tenant, &b, &[&a]).await.expect("reopen"), dm); } -/// Opening a DM with people and an agent finds the people's DM and adds the -/// agent to it. -/// Mutation: skip adding named agents to a found DM → agent missing → RED. +/// Opening a DM with people and an agent, with the history tag, finds the +/// people's DM and adds the agent to it. Without the tag it finds or makes the +/// DM of exactly those members, as before, and the people's DM is untouched. +/// Mutations: skip adding named agents to a found DM → agent missing → RED; +/// find the people's DM without the tag → the same channel → RED. #[tokio::test] #[ignore = "requires Postgres"] async fn opening_with_an_agent_finds_the_peoples_dm_and_adds_it() { @@ -259,10 +294,22 @@ async fn opening_with_an_agent_finds_the_peoples_dm_and_adds_it() { make_agent(&state, &tenant, &agent, &a, "anyone").await; let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); + let separate = open(&state, &tenant, &a, &[&b, &agent]) + .await + .expect("open without history"); + assert_ne!(separate, dm); + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); assert_eq!( open(&state, &tenant, &a, &[&b, &agent]) .await - .expect("open"), + .expect("again without history"), + separate + ); + + assert_eq!( + open_with_history(&state, &tenant, &a, &[&b, &agent]) + .await + .expect("open with history"), dm ); assert_eq!( @@ -311,7 +358,7 @@ async fn the_limit_counts_people_not_agents() { .await .expect("nine people"); - add(&state, &tenant, &a, dm, &[&agent], false) + add(&state, &tenant, &a, dm, &[&agent], true) .await .expect("agent beyond nine"); let tenth = Keys::generate(); @@ -337,7 +384,7 @@ async fn adding_an_agent_follows_its_add_policy() { let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); rejected( - add(&state, &tenant, &a, dm, &[&agent], false).await, + add(&state, &tenant, &a, dm, &[&agent], true).await, "policy:owner_only", ); rejected( @@ -346,7 +393,7 @@ async fn adding_an_agent_follows_its_add_policy() { ); assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &b])); - let (target, _) = add(&state, &tenant, &b, dm, &[&agent], false) + let (target, _) = add(&state, &tenant, &b, dm, &[&agent], true) .await .expect("owner adds"); assert_eq!(target, dm); @@ -419,7 +466,7 @@ async fn an_agent_cannot_share_a_dms_history() { ); make_agent(&state, &tenant, &agent, &a, "anyone").await; let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); - add(&state, &tenant, &a, dm, &[&agent], false) + add(&state, &tenant, &a, dm, &[&agent], true) .await .expect("add agent"); @@ -455,7 +502,7 @@ async fn growing_does_not_reveal_a_dm_the_actor_is_not_in() { ); make_agent(&state, &tenant, &agent, &a, "anyone").await; let dm = open(&state, &tenant, &a, &[&b]).await.expect("open"); - add(&state, &tenant, &a, dm, &[&agent], false) + add(&state, &tenant, &a, dm, &[&agent], true) .await .expect("add agent"); open(&state, &tenant, &a, &[&b, &c]).await.expect("trio"); @@ -489,7 +536,12 @@ async fn the_moderation_dm_cannot_grow() { let relay = state.relay_keypair.public_key().to_bytes(); let (dm, _) = state .db - .open_dm(tenant.community(), &[&a.public_key().to_bytes()], &relay) + .open_dm( + tenant.community(), + &[&a.public_key().to_bytes()], + &relay, + false, + ) .await .expect("moderation dm"); @@ -628,7 +680,7 @@ async fn an_owner_can_remove_their_agent_from_a_dm() { let tenant = community(&state, "dm-remove-agent").await; let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); make_agent(&state, &tenant, &agent, &a, "anyone").await; - let dm = open(&state, &tenant, &a, &[&b, &agent]) + let dm = open_with_history(&state, &tenant, &a, &[&b, &agent]) .await .expect("open"); let remove = vec![ @@ -695,7 +747,7 @@ async fn kicking_an_agent_keeps_a_group_dm_its_identity() { let tenant = community(&state, "dm-kick-agent").await; let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); make_agent(&state, &tenant, &agent, &a, "anyone").await; - let group = open(&state, &tenant, &a, &[&b, &agent]) + let group = open_with_history(&state, &tenant, &a, &[&b, &agent]) .await .expect("group"); let own = open(&state, &tenant, &a, &[&agent]).await.expect("own"); @@ -759,7 +811,7 @@ async fn an_agent_joining_a_one_person_dm_moves_its_identity() { make_agent(&state, &tenant, &theirs, &b, "anyone").await; let dm = open(&state, &tenant, &a, &[&mine]).await.expect("open"); - let (target, _) = add(&state, &tenant, &a, dm, &[&theirs], false) + let (target, _) = add(&state, &tenant, &a, dm, &[&theirs], true) .await .expect("add agent"); assert_eq!(target, dm); diff --git a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs index 010240aeb45..eb86dfa5895 100644 --- a/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/group_state_postgres_tests.rs @@ -1620,6 +1620,7 @@ async fn a_relay_operator_may_not_rename_an_unlisted_channel() { &other.public_key().to_bytes(), ], &owner.public_key().to_bytes(), + false, ) .await .expect("open dm"); diff --git a/crates/buzz-relay/src/handlers/moderation_notices.rs b/crates/buzz-relay/src/handlers/moderation_notices.rs index d5d6bc2c5c5..e9e873394aa 100644 --- a/crates/buzz-relay/src/handlers/moderation_notices.rs +++ b/crates/buzz-relay/src/handlers/moderation_notices.rs @@ -115,6 +115,7 @@ pub async fn send_moderation_notice( tenant.community(), &[recipient_pubkey], relay_pubkey_bytes.as_slice(), + false, ) .await?; let dm_channel_id = dm_channel.id; diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index c7f013097ca..9241056f4e4 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -1971,9 +1971,8 @@ pub fn build_dm_open(pubkeys: &[&str]) -> Result { /// Build a DM add-member event (kind 41011). /// -/// With `include_history`, a person joins this DM and sees its history. -/// Without it, adding a person starts a new DM of everyone. Agents always join -/// this DM in place. +/// With `include_history`, the member joins this DM and sees its history. +/// Without it, adding anyone starts a new DM of everyone. pub fn build_dm_add_member( channel_id: Uuid, pubkey: &str, From fdad7b86ed56ea13e5bbf0bee1be44e131467403 Mon Sep 17 00:00:00 2001 From: morgmart <98432065+morgmart@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:01:55 -0800 Subject: [PATCH 7/8] fix(relay): keep agent-only DMs from gaining people in place A DM of at most one person and their agents now never grows into a DM of two or more people, even with ["history","include"]. grow_dm reports PeopleStartNewDm under the membership lock, and 41011 falls back to the fork path: it opens or creates the DM of everyone and leaves the original untouched. This matches main, where adding a person to an agent chat always started a new DM. Agents still join in place, and DMs of two or more people are unchanged. Signed-off-by: morgmart <98432065+morgmart@users.noreply.github.com> Co-authored-by: Carl Co-Authored-By: Claude Opus 5.5 --- crates/buzz-cli/src/lib.rs | 3 +- crates/buzz-db/src/store/dm.rs | 16 ++++- .../src/handlers/command_executor.rs | 59 +++++++++++-------- .../src/handlers/dm_growth_postgres_tests.rs | 36 +++++++++++ crates/buzz-sdk/src/builders.rs | 5 +- 5 files changed, 88 insertions(+), 31 deletions(-) diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index bf1a827fb4d..6c4e159115d 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1327,7 +1327,8 @@ pub enum DmsCmd { #[arg(long)] pubkey: String, /// Add them to this DM with its history, instead of starting a new DM - /// of everyone. + /// of everyone. A person added to a DM of one person and their agents + /// always starts a new DM. #[arg(long)] include_history: bool, }, diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index 58d5fc0f336..079822575da 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -501,6 +501,10 @@ pub enum DmGrowth { /// Another DM, with this ID, already has the grown set of people, so this /// DM cannot take its identity. Nothing was written. Existing(Uuid), + /// This DM has at most one person, and the additions would bring it to + /// two or more. Such a DM never gains people in place; the caller starts + /// a DM of everyone instead. Nothing was written. + PeopleStartNewDm, } /// Add `additions` to the DM `channel_id` in place, keeping its history. @@ -509,7 +513,8 @@ pub enum DmGrowth { /// Otherwise the DM's identity moves to the grown set ([`dm_identity_hash`]), so /// opening a DM with the original members afterwards starts a new one. If a DM /// with the grown set already exists, nothing is written and that DM is -/// returned as [`DmGrowth::Existing`], provided `actor` is in it. +/// returned as [`DmGrowth::Existing`], provided `actor` is in it. People never +/// join a DM of at most one person: that is [`DmGrowth::PeopleStartNewDm`]. /// /// `actor` must be an active member. The DM must be live and not archived, and /// may hold at most [`MAX_DM_PEOPLE`] people afterwards. Re-adding members is @@ -595,10 +600,17 @@ pub async fn grow_dm( ))); } + // A DM of at most one person and their agents stays theirs: people + // joining it start a DM of everyone instead. + let current: Vec<&[u8]> = members.iter().map(Vec::as_slice).collect(); + if count_people(¤t, &agents) < 2 && count_people(&grown, &agents) >= 2 { + tx.rollback().await?; + return Ok(DmGrowth::PeopleStartNewDm); + } + // People move the DM's identity to the grown set of people. So do agents // joining a DM of at most one person, which is identified by all members. let hash = dm_identity_hash(&grown, &agents); - let current: Vec<&[u8]> = members.iter().map(Vec::as_slice).collect(); if hash != dm_identity_hash(¤t, &agents) { let existing = sqlx::query_scalar::<_, Uuid>( "SELECT id FROM channels \ diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 194fc2db248..8c87e739674 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -588,7 +588,7 @@ async fn grow_with_agents( .map_err(dm_write_error)? { DmGrowth::Grown { added } => Ok(added), - DmGrowth::Existing(_) => Err(IngestError::Internal( + DmGrowth::Existing(_) | DmGrowth::PeopleStartNewDm => Err(IngestError::Internal( "error: adding agents changed the DM's people".into(), )), } @@ -600,7 +600,8 @@ async fn grow_with_agents( /// DM's history. People move the DM's identity to the grown set of people, or, /// if a DM of those people already exists, that DM is returned instead and /// nothing changes here. Without the tag, everyone starts a new DM of exactly -/// its members (no history), as before. An agent cannot send the tag. +/// its members (no history), as before. So do people added to a DM of at most +/// one person, even with the tag. An agent cannot send the tag. async fn handle_dm_add_member( tenant: &TenantContext, state: &Arc, @@ -695,32 +696,38 @@ async fn handle_dm_add_member( // 5. Execute. let addition_refs: Vec<&[u8]> = additions.iter().map(Vec::as_slice).collect(); - let (target, was_created, added, moved_elsewhere) = if include_history { + let growth = if include_history { // Grow this DM in place. - match state - .db - .grow_dm(tenant.community(), channel_id, &self_bytes, &addition_refs) - .await - .map_err(dm_write_error)? - { - DmGrowth::Grown { added } => (channel_id, false, added, false), - // These people already have a DM: the agents join them there. - DmGrowth::Existing(other) => { - let added = grow_with_agents(tenant, state, other, &self_bytes, &agents).await?; - (other, false, added, true) - } - } + Some( + state + .db + .grow_dm(tenant.community(), channel_id, &self_bytes, &addition_refs) + .await + .map_err(dm_write_error)?, + ) } else { - // Start a new DM of everyone (no history); this DM is untouched. - let mut all_bytes = members.clone(); - all_bytes.extend(additions.iter().cloned()); - let all_refs: Vec<&[u8]> = all_bytes.iter().map(Vec::as_slice).collect(); - let (channel, created) = state - .db - .open_dm(tenant.community(), &all_refs, &self_bytes, false) - .await - .map_err(dm_write_error)?; - (channel.id, created, Vec::new(), !created) + None + }; + let (target, was_created, added, moved_elsewhere) = match growth { + Some(DmGrowth::Grown { added }) => (channel_id, false, added, false), + // These people already have a DM: the agents join them there. + Some(DmGrowth::Existing(other)) => { + let added = grow_with_agents(tenant, state, other, &self_bytes, &agents).await?; + (other, false, added, true) + } + // Start a new DM of everyone (no history); this DM is untouched. A DM + // of at most one person never gains people, even with the tag. + Some(DmGrowth::PeopleStartNewDm) | None => { + let mut all_bytes = members.clone(); + all_bytes.extend(additions.iter().cloned()); + let all_refs: Vec<&[u8]> = all_bytes.iter().map(Vec::as_slice).collect(); + let (channel, created) = state + .db + .open_dm(tenant.community(), &all_refs, &self_bytes, false) + .await + .map_err(dm_write_error)?; + (channel.id, created, Vec::new(), !created) + } }; // Finalize the idempotency record after the separate mutation succeeds. diff --git a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs index a746d57ad46..f73111090de 100644 --- a/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs @@ -830,3 +830,39 @@ async fn an_agent_joining_a_one_person_dm_moves_its_identity() { dm ); } + +/// A DM of one person and their agents never gains a person in place, even +/// with the history tag: adding one starts a DM of everyone, and adding them +/// again opens that DM. The original keeps its members and its identity. +/// Mutation: let the tag grow it in place → the original gains B → RED. +#[tokio::test] +#[ignore = "requires Postgres"] +async fn a_person_added_to_an_agent_only_dm_starts_a_new_one() { + let state = state().await; + let tenant = community(&state, "dm-agent-only-person").await; + let (a, b, agent) = (Keys::generate(), Keys::generate(), Keys::generate()); + make_agent(&state, &tenant, &agent, &a, "anyone").await; + let dm = open(&state, &tenant, &a, &[&agent]).await.expect("open"); + + let (target, created) = add(&state, &tenant, &a, dm, &[&b], true) + .await + .expect("add person"); + + assert!(created); + assert_ne!(target, dm); + assert_eq!(members(&state, &tenant, dm).await, set_of(&[&a, &agent])); + assert_eq!( + members(&state, &tenant, target).await, + set_of(&[&a, &b, &agent]) + ); + assert_eq!( + add(&state, &tenant, &a, dm, &[&b], true) + .await + .expect("add again"), + (target, false) + ); + assert_eq!( + open(&state, &tenant, &a, &[&agent]).await.expect("reopen"), + dm + ); +} diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index 9241056f4e4..3190694177d 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -1971,8 +1971,9 @@ pub fn build_dm_open(pubkeys: &[&str]) -> Result { /// Build a DM add-member event (kind 41011). /// -/// With `include_history`, the member joins this DM and sees its history. -/// Without it, adding anyone starts a new DM of everyone. +/// With `include_history`, the member joins this DM and sees its history, +/// unless a person is joining a DM of at most one person and their agents. +/// Otherwise adding anyone starts a new DM of everyone. pub fn build_dm_add_member( channel_id: Uuid, pubkey: &str, From 0279a892e1fc9a796b91d876b454f8bccb7a12aa Mon Sep 17 00:00:00 2001 From: morgmart <98432065+morgmart@users.noreply.github.com> Date: Thu, 5 Nov 2026 18:51:35 -0800 Subject: [PATCH 8/8] docs(sdk): document the DM history tag and the created response kind:41010 and kind:41011 have no NIP yet, so the SDK builder carries the contract: the opt-in history tag and the channel_id/created response. Signed-off-by: morgmart <98432065+morgmart@users.noreply.github.com> Co-authored-by: Carl Co-Authored-By: Claude Opus 5.5 --- crates/buzz-sdk/src/builders.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index 3190694177d..008ee61b476 100644 --- a/crates/buzz-sdk/src/builders.rs +++ b/crates/buzz-sdk/src/builders.rs @@ -1955,6 +1955,13 @@ pub fn build_workflow_approval( /// Build a DM open event (kind 41010). /// /// `pubkeys` must be 1–8 hex-encoded pubkeys to include in the DM conversation. +/// The relay finds or makes the DM of exactly these members. With a +/// `["history", "include"]` tag, it instead finds the DM of the named people +/// and adds any named agents it lacks. +/// +/// The relay answers kind 41010 and kind 41011 with +/// `response:{"channel_id":"","created":}`, where `created` is +/// true only when a new DM was made. pub fn build_dm_open(pubkeys: &[&str]) -> Result { if pubkeys.is_empty() || pubkeys.len() > 8 { return Err(SdkError::InvalidInput(