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..6c4e159115d 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -1326,6 +1326,11 @@ pub enum DmsCmd { /// User pubkey to add (64-char hex) #[arg(long)] pubkey: String, + /// Add them to this DM with its history, instead of starting a new DM + /// of everyone. A person added to a DM of one person and their agents + /// always starts a new DM. + #[arg(long)] + include_history: bool, }, /// Hide a DM conversation from your DM list Hide { 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/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index 3be879aaf00..692ce3425e9 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?; @@ -1195,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 ebdbc45d271..079822575da 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,69 @@ 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. 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) + } +} + +/// 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,25 +166,23 @@ 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 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( "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 +191,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, by_people); let mut tx = crate::begin_community_write_transaction( pool, @@ -358,6 +431,14 @@ 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. /// +/// 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)`: /// - `was_created = true` -- a new DM was created. /// - `was_created = false` -- an existing DM was returned. @@ -366,35 +447,301 @@ pub async fn open_dm( community_id: CommunityId, pubkeys: &[&[u8]], created_by: &[u8], + by_people: bool, ) -> 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, by_people); + // 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)); } // 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)) } +/// 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), + /// 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. +/// +/// 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. 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 +/// 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" + ))); + } + + // 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); + 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 \ + 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(), + }) +} + +/// 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", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .execute(&mut *conn) + .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( + 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()`. @@ -560,8 +907,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. @@ -583,8 +938,21 @@ 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. + #[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. 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/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 73d42985178..8c87e739674 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,174 @@ 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}"); + } + } +} + +/// 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"; + +/// 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 +468,48 @@ 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(), - )); + let include_history = history_opt_in(event)?; + + // 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. 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 others.len() > 1 { + 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 +523,38 @@ async fn handle_dm_open( PersistResult::Inserted(tx) => tx, }; - // 4. Execute: open_dm + // 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(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + .map_err(dm_write_error)?; + let added = if was_created || !include_history { + 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 +568,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(_) | DmGrowth::PeopleStartNewDm => Err(IngestError::Internal( + "error: adding agents changed the DM's people".into(), + )), + } +} + +/// kind:41011: add people and agents to a DM. +/// +/// 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. 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, @@ -445,6 +622,7 @@ async fn handle_dm_add_member( "invalid: must specify at least 1 new participant in p tags".into(), )); } + let include_history = history_opt_in(event)?; // 2. Validate caller is member of existing DM let is_member = state @@ -467,27 +645,40 @@ 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(MODERATION_DM_REFUSAL.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)?; + // 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? + .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(), )); } @@ -503,60 +694,78 @@ 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 growth = if include_history { + // Grow this DM in place. + Some( + state + .db + .grow_dm(tenant.community(), channel_id, &self_bytes, &addition_refs) + .await + .map_err(dm_write_error)?, + ) + } else { + 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. 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..f73111090de --- /dev/null +++ b/crates/buzz-relay/src/handlers/dm_growth_postgres_tests.rs @@ -0,0 +1,868 @@ +//! 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 { + 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) +} + +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 + ); +} + +/// 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() { + 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 (forked, created) = add(&state, &tenant, &a, dm, &[&agent], false) + .await + .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!( + 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, 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() { + 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"); + + 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("again without history"), + separate + ); + + assert_eq!( + open_with_history(&state, &tenant, &a, &[&b, &agent]) + .await + .expect("open with history"), + 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], true) + .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], true).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], true) + .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], true) + .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], true) + .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, 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() { + 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, + false, + ) + .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", + ); + 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. +#[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])); +} + +/// 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() { + 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, + &a, + 9001, + 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_with_history(&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, 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); +} + +/// 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); +} + +/// 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_with_history(&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 + ); +} + +/// 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], true) + .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 + ); +} + +/// 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-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/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-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-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 0bcc267a3cb..ad6eb8754f0 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -621,6 +621,10 @@ async fn actor_owns_any_owner_agent( Ok(false) } +/// 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. pub async fn validate_admin_event( tenant: &TenantContext, @@ -675,6 +679,21 @@ 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 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 { // Ingest validates kind:9000 with `validate_put_user`, which keeps @@ -1930,6 +1949,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(); diff --git a/crates/buzz-sdk/src/builders.rs b/crates/buzz-sdk/src/builders.rs index 9970e2f0002..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( @@ -1970,9 +1977,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`, 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, + 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 +5091,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(_))); }