From 116200b527acd587ffc75214833293ad12c133c5 Mon Sep 17 00:00:00 2001 From: ghostelle1 <292656566+ghostelle1@users.noreply.github.com> Date: Mon, 31 Aug 2026 10:39:02 +0900 Subject: [PATCH 1/2] fix(dm): restore missing participant memberships Signed-off-by: ghostelle1 <292656566+ghostelle1@users.noreply.github.com> --- crates/buzz-db/src/channel.rs | 2 +- crates/buzz-db/src/dm.rs | 355 +++++++++++++++++- crates/buzz-db/src/lib.rs | 2 +- .../src/handlers/command_executor.rs | 32 +- .../src/handlers/moderation_notices.rs | 15 +- .../buzz-relay/src/handlers/side_effects.rs | 86 ++++- .../tests/e2e_nostr_interop.rs | 158 ++++++++ 7 files changed, 627 insertions(+), 23 deletions(-) diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index 98790e3d623..87411192f52 100644 --- a/crates/buzz-db/src/channel.rs +++ b/crates/buzz-db/src/channel.rs @@ -473,7 +473,7 @@ pub async fn verify_channel_roster_fence_behavior(pool: &sqlx::PgPool) -> Result /// Take the per-channel membership lock. MUST be the first statement in the /// transaction that then reads roles/owner counts and writes membership, so the /// whole check-then-write sequence is atomic against a concurrent one. -async fn acquire_channel_membership_lock( +pub(crate) async fn acquire_channel_membership_lock( tx: &mut Transaction<'_, Postgres>, community_id: CommunityId, channel_id: Uuid, diff --git a/crates/buzz-db/src/dm.rs b/crates/buzz-db/src/dm.rs index 89e15c70260..6981de79b7b 100644 --- a/crates/buzz-db/src/dm.rs +++ b/crates/buzz-db/src/dm.rs @@ -8,7 +8,7 @@ use sha2::{Digest, Sha256}; use sqlx::{PgPool, Row}; use uuid::Uuid; -use crate::channel::ChannelRecord; +use crate::channel::{self, ChannelRecord}; use crate::error::{DbError, Result}; use buzz_core::CommunityId; @@ -38,6 +38,17 @@ pub struct DmParticipant { pub role: String, } +/// Outcome of opening a direct-message participant set. +#[derive(Debug, Clone)] +pub struct OpenDmResult { + /// The existing or newly created DM channel. + pub channel: ChannelRecord, + /// Whether this call created the channel. + pub was_created: bool, + /// Participants whose missing or soft-removed memberships were restored. + pub restored_participants: Vec>, +} + // -- Pure helpers ------------------------------------------------------------- /// Compute a stable SHA-256 fingerprint for a set of participant pubkeys. @@ -350,15 +361,15 @@ 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. /// -/// Returns `(channel, was_created)`: -/// - `was_created = true` -- a new DM was created. -/// - `was_created = false` -- an existing DM was returned. +/// An active participant reopening an existing immutable participant set also +/// restores any missing or soft-removed peer memberships. A removed caller +/// cannot use this path to restore themselves or anyone else. pub async fn open_dm( pool: &PgPool, community_id: CommunityId, pubkeys: &[&[u8]], created_by: &[u8], -) -> Result<(ChannelRecord, bool)> { +) -> Result { // 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) { @@ -376,15 +387,113 @@ pub async fn open_dm( // Check for existing DM first (fast path, no transaction). if let Some(existing) = find_dm_by_participants(pool, community_id, &hash).await? { - // 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)); + let restored_participants = + reopen_existing_dm(pool, community_id, existing.id, &all, created_by).await?; + return Ok(OpenDmResult { + channel: existing, + was_created: false, + restored_participants, + }); } // Create new DM. let channel = create_dm(pool, community_id, &all, created_by).await?; - Ok((channel, true)) + Ok(OpenDmResult { + channel, + was_created: true, + restored_participants: Vec::new(), + }) +} + +/// Reopen an existing immutable DM participant set under the same membership +/// serialization lock used by normal channel membership changes. +async fn reopen_existing_dm( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, + participants: &[&[u8]], + opened_by: &[u8], +) -> Result>> { + let mut tx = pool.begin().await?; + channel::acquire_channel_membership_lock(&mut tx, community_id, channel_id).await?; + + let opener_is_active: bool = sqlx::query_scalar( + r#" + SELECT EXISTS( + SELECT 1 + FROM channel_members + WHERE community_id = $1 + AND channel_id = $2 + AND pubkey = $3 + AND removed_at IS NULL + ) + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(opened_by) + .fetch_one(&mut *tx) + .await?; + + if !opener_is_active { + return Err(DbError::AccessDenied( + "only an active DM participant may restore the participant set".to_string(), + )); + } + + let participant_bytes: Vec> = participants.iter().map(|pk| pk.to_vec()).collect(); + let rows = sqlx::query( + r#" + WITH requested(pubkey) AS ( + SELECT DISTINCT unnest($3::bytea[]) + ) + INSERT INTO channel_members + (community_id, channel_id, pubkey, role, invited_by) + SELECT $1, $2, requested.pubkey, 'member', $4 + FROM requested + ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET + removed_at = NULL, + removed_by = NULL, + hidden_at = NULL, + role = EXCLUDED.role, + invited_by = EXCLUDED.invited_by + WHERE channel_members.removed_at IS NOT NULL + RETURNING pubkey + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(&participant_bytes) + .bind(opened_by) + .fetch_all(&mut *tx) + .await?; + + // Reopening always resurfaces the DM for its active caller. Active peers' + // independent hidden preferences remain untouched. + sqlx::query( + r#" + UPDATE channel_members + SET hidden_at = NULL + WHERE community_id = $1 + AND channel_id = $2 + AND pubkey = $3 + AND removed_at IS NULL + "#, + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .bind(opened_by) + .execute(&mut *tx) + .await?; + + let mut restored = rows + .into_iter() + .map(|row| row.try_get::, _>("pubkey").map_err(Into::into)) + .collect::>>()?; + restored.sort_unstable(); + tx.commit().await?; + Ok(restored) } // -- Hide / unhide ------------------------------------------------------------ @@ -520,6 +629,25 @@ fn row_to_channel_record(row: sqlx::postgres::PgRow) -> Result { mod tests { use super::*; + type MembershipRemovalState = ( + Option>, + Option>, + Option>, + ); + + async fn setup_pool() -> PgPool { + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_owned()); + let pool = PgPool::connect(&database_url) + .await + .expect("connect to test database"); + crate::migration::run_migrations(&pool) + .await + .expect("migrate test database"); + pool + } + #[test] fn participant_hash_is_order_independent() { let a = [1u8; 32]; @@ -554,4 +682,213 @@ mod tests { let h = compute_participant_hash(&[&a, &b]); assert_eq!(h.len(), 32); } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn active_participant_reopening_dm_restores_removed_peer() { + let pool = setup_pool().await; + let community_uuid = Uuid::new_v4(); + let community = CommunityId::from_uuid(community_uuid); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(format!("dm-recovery-{}.example", community_uuid.simple())) + .execute(&pool) + .await + .expect("insert test community"); + + let opener = [1u8; 32]; + let peer = [2u8; 32]; + let opened = open_dm(&pool, community, &[&peer], &opener) + .await + .expect("create dm"); + assert!(opened.was_created, "fixture must create a fresh DM"); + let channel = opened.channel; + + sqlx::query( + r#" + UPDATE channel_members + SET removed_at = NOW(), removed_by = $4, hidden_at = NOW() + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 + "#, + ) + .bind(community_uuid) + .bind(channel.id) + .bind(peer.as_slice()) + .bind(opener.as_slice()) + .execute(&pool) + .await + .expect("soft-remove peer fixture"); + + let reopened = open_dm(&pool, community, &[&peer], &opener) + .await + .expect("reopen existing dm"); + assert_eq!( + reopened.channel.id, channel.id, + "reopen must preserve DM history" + ); + assert!( + !reopened.was_created, + "reopen must not create a replacement DM" + ); + assert_eq!(reopened.restored_participants, vec![peer.to_vec()]); + + let restored: MembershipRemovalState = sqlx::query_as( + r#" + SELECT removed_at, removed_by, hidden_at + FROM channel_members + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 + "#, + ) + .bind(community_uuid) + .bind(channel.id) + .bind(peer.as_slice()) + .fetch_one(&pool) + .await + .expect("read restored peer membership"); + + assert_eq!(restored, (None, None, None)); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn removed_participant_cannot_reopen_dm_or_restore_peer() { + let pool = setup_pool().await; + let community_uuid = Uuid::new_v4(); + let community = CommunityId::from_uuid(community_uuid); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(format!( + "dm-recovery-denied-{}.example", + community_uuid.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + + let removed_caller = [3u8; 32]; + let removed_peer = [4u8; 32]; + let opened = open_dm(&pool, community, &[&removed_peer], &removed_caller) + .await + .expect("create dm"); + assert!(opened.was_created, "fixture must create a fresh DM"); + let channel = opened.channel; + + sqlx::query( + r#" + UPDATE channel_members + SET removed_at = NOW(), removed_by = $4 + WHERE community_id = $1 AND channel_id = $2 AND pubkey = ANY($3) + "#, + ) + .bind(community_uuid) + .bind(channel.id) + .bind(vec![removed_caller.to_vec(), removed_peer.to_vec()]) + .bind(removed_caller.as_slice()) + .execute(&pool) + .await + .expect("soft-remove caller and peer fixtures"); + + let error = open_dm(&pool, community, &[&removed_peer], &removed_caller) + .await + .expect_err("removed caller must not resurrect the DM participant set"); + assert!(matches!(error, DbError::AccessDenied(_))); + + let active_count: i64 = sqlx::query_scalar( + r#" + SELECT count(*) + FROM channel_members + WHERE community_id = $1 AND channel_id = $2 AND removed_at IS NULL + "#, + ) + .bind(community_uuid) + .bind(channel.id) + .fetch_one(&pool) + .await + .expect("count active memberships"); + assert_eq!(active_count, 0, "denied reopen must not restore any peer"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn reopening_dm_restores_missing_peer_without_unhiding_active_peer() { + let pool = setup_pool().await; + let community_uuid = Uuid::new_v4(); + let community = CommunityId::from_uuid(community_uuid); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(format!( + "dm-recovery-hidden-{}.example", + community_uuid.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + + let opener = [5u8; 32]; + let missing_peer = [6u8; 32]; + let hidden_active_peer = [7u8; 32]; + let opened = open_dm( + &pool, + community, + &[&missing_peer, &hidden_active_peer], + &opener, + ) + .await + .expect("create group dm"); + assert!(opened.was_created, "fixture must create a fresh DM"); + + sqlx::query( + r#" + DELETE FROM channel_members + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 + "#, + ) + .bind(community_uuid) + .bind(opened.channel.id) + .bind(missing_peer.as_slice()) + .execute(&pool) + .await + .expect("delete peer membership fixture"); + sqlx::query( + r#" + UPDATE channel_members + SET hidden_at = NOW() + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 + "#, + ) + .bind(community_uuid) + .bind(opened.channel.id) + .bind(hidden_active_peer.as_slice()) + .execute(&pool) + .await + .expect("hide active peer fixture"); + + let reopened = open_dm( + &pool, + community, + &[&missing_peer, &hidden_active_peer], + &opener, + ) + .await + .expect("reopen group dm"); + assert_eq!(reopened.restored_participants, vec![missing_peer.to_vec()]); + + let hidden_at: Option> = sqlx::query_scalar( + r#" + SELECT hidden_at + FROM channel_members + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 + "#, + ) + .bind(community_uuid) + .bind(opened.channel.id) + .bind(hidden_active_peer.as_slice()) + .fetch_one(&pool) + .await + .expect("read active peer visibility"); + assert!( + hidden_at.is_some(), + "reopening must preserve an active peer's hidden preference" + ); + } } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index b37dcedff8c..d02771537e1 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2341,7 +2341,7 @@ impl Db { community_id: CommunityId, pubkeys: &[&[u8]], created_by: &[u8], - ) -> Result<(channel::ChannelRecord, bool)> { + ) -> Result { dm::open_dm(&self.pool, community_id, pubkeys, created_by).await } diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 007db43ffd9..d0822acd636 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -29,8 +29,8 @@ use crate::webhook_secret; use super::ingest::{extract_channel_id, IngestAuth, IngestError, IngestResult}; use super::side_effects::{ - emit_group_discovery_events, emit_membership_notification, emit_system_message, - publish_dm_visibility_snapshot, + emit_dm_membership_recovery_side_effects, emit_group_discovery_events, + emit_membership_notification, emit_system_message, publish_dm_visibility_snapshot, }; /// Route a command-kind event to the appropriate handler. @@ -346,11 +346,14 @@ async fn handle_dm_open( // 4. Execute: open_dm let all_refs: Vec<&[u8]> = all_bytes.iter().map(|b| b.as_slice()).collect(); - let (channel, was_created) = state + let opened = state .db .open_dm(tenant.community(), &all_refs, &self_bytes) .await .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + let channel = opened.channel; + let was_created = opened.was_created; + let restored_participants = opened.restored_participants; // Finalize the idempotency record after the separate mutation succeeds. tx.commit() @@ -406,6 +409,15 @@ async fn handle_dm_open( } } } else { + emit_dm_membership_recovery_side_effects( + tenant, + state, + channel.id, + &restored_participants, + &self_bytes, + ) + .await; + // 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 { @@ -507,11 +519,14 @@ async fn handle_dm_add_member( // 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 + let opened = state .db .open_dm(tenant.community(), &all_refs, &self_bytes) .await .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + let new_channel = opened.channel; + let was_created = opened.was_created; + let restored_participants = opened.restored_participants; // Finalize the idempotency record after the separate mutation succeeds. tx.commit() @@ -549,6 +564,15 @@ async fn handle_dm_add_member( warn!("DM add_member: membership notification failed: {e}"); } } + } else { + emit_dm_membership_recovery_side_effects( + tenant, + state, + new_channel.id, + &restored_participants, + &self_bytes, + ) + .await; } // 8. Return response diff --git a/crates/buzz-relay/src/handlers/moderation_notices.rs b/crates/buzz-relay/src/handlers/moderation_notices.rs index 8f57eea71f8..5bcc9c06aca 100644 --- a/crates/buzz-relay/src/handlers/moderation_notices.rs +++ b/crates/buzz-relay/src/handlers/moderation_notices.rs @@ -29,7 +29,7 @@ use buzz_core::kind::{event_kind_u32, KIND_STREAM_MESSAGE}; use buzz_core::tenant::TenantContext; use super::event::dispatch_persistent_event; -use super::side_effects::emit_group_discovery_events; +use super::side_effects::{emit_dm_membership_recovery_side_effects, emit_group_discovery_events}; use crate::state::AppState; /// Tag naming the moderation source row (report/action) a notice was derived @@ -97,7 +97,7 @@ pub async fn send_moderation_notice( // 1. Create/reuse the two-party DM channel {relay mod key, recipient}. // `open_dm` is participant-hash idempotent, so re-delivery to the same // user reuses the one thread per (community, user). - let (dm_channel, was_created) = state + let opened = state .db .open_dm( tenant.community(), @@ -105,6 +105,9 @@ pub async fn send_moderation_notice( relay_pubkey_bytes.as_slice(), ) .await?; + let dm_channel = opened.channel; + let was_created = opened.was_created; + let restored_participants = opened.restored_participants; let dm_channel_id = dm_channel.id; // Count new DM creation; side-effect gates below intentionally do not @@ -126,6 +129,14 @@ pub async fn send_moderation_notice( .db .unhide_dm(tenant.community(), dm_channel_id, recipient_pubkey) .await?; + emit_dm_membership_recovery_side_effects( + tenant, + state, + dm_channel_id, + &restored_participants, + relay_pubkey_bytes.as_slice(), + ) + .await; // Idempotency: a notice for this source id already exists in this DM ⇒ no-op. // The source (report/action) row id is carried in a `moderation_source` tag diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index d37e7b375ee..24b93837cbc 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -934,8 +934,33 @@ pub async fn emit_membership_notification( }) .to_string(); + // A member can be removed and restored in the same second as the original + // add. Because these relay-signed notifications otherwise have identical + // kind, content, tags, and timestamp, Nostr gives them the same event id and + // `insert_event` suppresses the recovery fan-out as a duplicate. Advance + // past the latest notification for this target/channel pair so every real + // membership transition remains observable by live clients. + let now = nostr::Timestamp::now().as_secs(); + let previous = state + .db + .query_events(&buzz_db::event::EventQuery { + kinds: Some(vec![notification_kind as i32]), + pubkey: Some(state.relay_keypair.public_key().to_bytes().to_vec()), + p_tag_hex: Some(target_hex.clone()), + custom_tag: Some(("h".to_owned(), channel_id_str.clone())), + limit: Some(1), + global_only: true, + ..buzz_db::event::EventQuery::for_community(tenant.community()) + }) + .await?; + let created_at = previous + .first() + .map(|event| (event.event.created_at.as_secs() + 1).max(now)) + .unwrap_or(now); + let event = EventBuilder::new(Kind::Custom(notification_kind as u16), content) .tags([p_tag, h_tag]) + .custom_created_at(nostr::Timestamp::from(created_at)) .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign membership notification: {e}"))?; @@ -1669,12 +1694,6 @@ async fn handle_edit_metadata( // remove/re-add uses to recover. Humans self-heal via the re-emitted // kind:39000 discovery, so this is intentionally agent-scoped. // - // Known limitation: emit_membership_notification builds a created_at=now - // event with no nonce, and insert_event skips fan-out on a duplicate id. - // Four sub-second toggles (archive->unarchive->archive->unarchive) on the - // same channel by the same actor could collide ids and skip a fan-out. - // Not reachable in practice — unarchive has a single human-driven caller; - // the reaper only auto-archives — so we don't engineer around it. for member in state.db.get_members(tenant.community(), channel_id).await? { @@ -3444,6 +3463,61 @@ pub async fn publish_nipia_archival_list( ) } +/// Refresh relay and client state after an existing DM repairs inactive +/// participant memberships. +pub async fn emit_dm_membership_recovery_side_effects( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + restored_participants: &[Vec], + actor: &[u8], +) { + if restored_participants.is_empty() { + return; + } + + for participant in restored_participants { + state.invalidate_membership(tenant, channel_id, participant); + } + + if let Err(error) = emit_group_discovery_events(tenant, state, channel_id).await { + warn!( + channel = %channel_id, + error = %error, + "DM membership recovery discovery emission failed" + ); + } + + for participant in restored_participants { + if let Err(error) = emit_membership_notification( + tenant, + state, + channel_id, + participant, + actor, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + warn!( + channel = %channel_id, + participant = %hex::encode(participant), + error = %error, + "DM membership recovery notification failed" + ); + } + + if let Err(error) = publish_dm_visibility_snapshot(tenant, state, participant).await { + warn!( + channel = %channel_id, + participant = %hex::encode(participant), + error = %error, + "DM membership recovery visibility snapshot failed" + ); + } + } +} + /// NIP-DV: publish the relay-signed, per-viewer DM visibility snapshot for /// `viewer`. The event is parameterized-replaceable (`d` = viewer pubkey) and /// carries one `h` tag per DM the viewer currently has hidden. Called after any diff --git a/crates/buzz-test-client/tests/e2e_nostr_interop.rs b/crates/buzz-test-client/tests/e2e_nostr_interop.rs index fce78776764..9285a38b80e 100644 --- a/crates/buzz-test-client/tests/e2e_nostr_interop.rs +++ b/crates/buzz-test-client/tests/e2e_nostr_interop.rs @@ -24,6 +24,12 @@ use std::time::Duration; use buzz_test_client::{BuzzTestClient, RelayMessage, TestClientError}; use nostr::{Alphabet, EventBuilder, Filter, Keys, Kind, SingleLetterTag, Tag}; +type MembershipRemovalState = ( + Option>, + Option>, + Option>, +); + fn relay_url() -> String { std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()) } @@ -36,6 +42,16 @@ fn relay_http_url() -> String { .to_string() } +async fn e2e_db_pool() -> sqlx::Pool { + let database_url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); + sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&database_url) + .await + .expect("connect to e2e Postgres") +} + fn sub_id(name: &str) -> String { format!("e2e-{name}-{}", uuid::Uuid::new_v4()) } @@ -105,6 +121,26 @@ async fn send_rest_message(keys: &Keys, channel_id: &str, content: &str) -> Stri body["event_id"].as_str().expect("event_id").to_string() } +async fn post_rest_message(keys: &Keys, channel_id: &str, content: &str) -> serde_json::Value { + let client = reqwest::Client::new(); + let pubkey_hex = keys.public_key().to_hex(); + let event = EventBuilder::new(Kind::Custom(9), content) + .tags(vec![Tag::parse(["h", channel_id]).unwrap()]) + .sign_with_keys(keys) + .unwrap(); + client + .post(format!("{}/events", relay_http_url())) + .header("X-Pubkey", &pubkey_hex) + .header("Content-Type", "application/json") + .body(serde_json::to_string(&event).unwrap()) + .send() + .await + .expect("submit message event") + .json() + .await + .expect("parse message response") +} + /// Create a DM via a signed kind:41010 (DM open) command event and return the /// channel_id UUID string parsed from the relay's `response:{...}` message. async fn create_dm(requester_keys: &Keys, other_pubkey_hex: &str) -> String { @@ -1474,6 +1510,128 @@ async fn test_nipdv_two_viewers_independent_snapshots() { client_b.disconnect().await.expect("disconnect B"); } +/// Reopening an existing immutable DM must repair an inactive peer, evict a +/// cached negative membership decision, and notify the peer so a live agent can +/// subscribe to the restored channel without restarting. +#[tokio::test] +#[ignore] +async fn test_dm_reopen_restores_removed_peer_and_emits_notification() { + let url = relay_url(); + let keys_a = Keys::generate(); + let keys_b = Keys::generate(); + let b_pubkey = keys_b.public_key().to_bytes(); + let b_pubkey_hex = keys_b.public_key().to_hex(); + let channel_id = create_dm(&keys_a, &b_pubkey_hex).await; + let channel_uuid = uuid::Uuid::parse_str(&channel_id).expect("DM channel UUID"); + + let mut client_b = BuzzTestClient::connect(&url, &keys_b) + .await + .expect("client B connect"); + let sid_membership = sub_id("dm-recovery-44100"); + let membership_filter = Filter::new().kind(Kind::Custom(44100)).custom_tag( + SingleLetterTag::lowercase(Alphabet::P), + b_pubkey_hex.as_str(), + ); + client_b + .subscribe(&sid_membership, vec![membership_filter]) + .await + .expect("subscribe membership recovery"); + client_b + .collect_until_eose(&sid_membership, Duration::from_secs(10)) + .await + .expect("drain initial membership history"); + + let pool = e2e_db_pool().await; + let community_id: uuid::Uuid = + sqlx::query_scalar("SELECT community_id FROM channels WHERE id = $1") + .bind(channel_uuid) + .fetch_one(&pool) + .await + .expect("read DM community"); + let removed = sqlx::query( + r#" + UPDATE channel_members + SET removed_at = NOW(), removed_by = $4, hidden_at = NOW() + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 AND removed_at IS NULL + "#, + ) + .bind(community_id) + .bind(channel_uuid) + .bind(b_pubkey.as_slice()) + .bind(keys_a.public_key().to_bytes().as_slice()) + .execute(&pool) + .await + .expect("soft-remove B fixture"); + assert_eq!(removed.rows_affected(), 1, "fixture must remove B once"); + + // This rejection seeds the relay's negative membership cache for B. + let blocked = post_rest_message(&keys_b, &channel_id, "blocked-before-dm-recovery").await; + assert!( + !blocked["accepted"].as_bool().unwrap_or(false), + "removed B must be rejected before recovery: {blocked}" + ); + + post_signed_event( + &keys_a, + 41010, + vec![Tag::parse(["p", &b_pubkey_hex]).unwrap()], + ) + .await; + + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + let remaining = deadline + .checked_duration_since(tokio::time::Instant::now()) + .unwrap_or(Duration::ZERO); + assert!( + !remaining.is_zero(), + "B did not receive a live kind:44100 recovery notification" + ); + match client_b + .recv_event(remaining) + .await + .expect("receive membership recovery notification") + { + RelayMessage::Event { + subscription_id, + event, + } if subscription_id == sid_membership + && event.kind == Kind::Custom(44100) + && event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.len() >= 2 && values[0] == "h" && values[1] == channel_id + }) => + { + break; + } + _ => {} + } + } + + let membership: MembershipRemovalState = sqlx::query_as( + r#" + SELECT removed_at, removed_by, hidden_at + FROM channel_members + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 + "#, + ) + .bind(community_id) + .bind(channel_uuid) + .bind(b_pubkey.as_slice()) + .fetch_one(&pool) + .await + .expect("read recovered membership"); + assert_eq!(membership, (None, None, None)); + + let accepted = post_rest_message(&keys_b, &channel_id, "accepted-after-dm-recovery").await; + assert!( + accepted["accepted"].as_bool().unwrap_or(false), + "recovery must evict B's cached rejection: {accepted}" + ); + + client_b.disconnect().await.expect("disconnect B"); +} + /// NIP-DV privacy via WebSocket REQ: a third party subscribing to another /// viewer's snapshot (`kind:30622 #p=A` as B) must be rejected with CLOSED, not /// served A's hidden set. From 7c2af82a3fb5c4a3efd37c02d1374f89d0a1ce59 Mon Sep 17 00:00:00 2001 From: ghostelle1 <292656566+ghostelle1@users.noreply.github.com> Date: Mon, 31 Aug 2026 10:58:32 +0900 Subject: [PATCH 2/2] fix(dm): harden membership recovery delivery Signed-off-by: ghostelle1 <292656566+ghostelle1@users.noreply.github.com> --- crates/buzz-db/src/store/dm.rs | 139 ++++++++++++++++++ .../src/handlers/command_executor.rs | 21 ++- .../buzz-relay/src/handlers/side_effects.rs | 60 ++++---- 3 files changed, 193 insertions(+), 27 deletions(-) diff --git a/crates/buzz-db/src/store/dm.rs b/crates/buzz-db/src/store/dm.rs index 01a1d9b70b1..a9496f75d61 100644 --- a/crates/buzz-db/src/store/dm.rs +++ b/crates/buzz-db/src/store/dm.rs @@ -977,4 +977,143 @@ mod tests { "reopening must preserve an active peer's hidden preference" ); } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn reopening_group_dm_restores_multiple_missing_and_removed_peers() { + let pool = setup_pool().await; + let community_uuid = Uuid::new_v4(); + let community = CommunityId::from_uuid(community_uuid); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(format!( + "dm-recovery-multiple-{}.example", + community_uuid.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + + let opener = [8u8; 32]; + let missing_peer = [9u8; 32]; + let removed_peer = [10u8; 32]; + let opened = open_dm(&pool, community, &[&missing_peer, &removed_peer], &opener) + .await + .expect("create group dm"); + + sqlx::query( + r#" + DELETE FROM channel_members + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 + "#, + ) + .bind(community_uuid) + .bind(opened.channel.id) + .bind(missing_peer.as_slice()) + .execute(&pool) + .await + .expect("delete missing peer fixture"); + sqlx::query( + r#" + UPDATE channel_members + SET removed_at = NOW(), removed_by = $4, hidden_at = NOW() + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 + "#, + ) + .bind(community_uuid) + .bind(opened.channel.id) + .bind(removed_peer.as_slice()) + .bind(opener.as_slice()) + .execute(&pool) + .await + .expect("soft-remove peer fixture"); + + let reopened = open_dm(&pool, community, &[&missing_peer, &removed_peer], &opener) + .await + .expect("reopen group dm"); + + assert_eq!( + reopened.restored_participants, + vec![missing_peer.to_vec(), removed_peer.to_vec()] + ); + let active_count: i64 = sqlx::query_scalar( + r#" + SELECT count(*) + FROM channel_members + WHERE community_id = $1 AND channel_id = $2 AND removed_at IS NULL + "#, + ) + .bind(community_uuid) + .bind(opened.channel.id) + .fetch_one(&pool) + .await + .expect("count active memberships"); + assert_eq!(active_count, 3); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn concurrent_reopens_restore_peer_exactly_once() { + let pool = setup_pool().await; + let community_uuid = Uuid::new_v4(); + let community = CommunityId::from_uuid(community_uuid); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_uuid) + .bind(format!( + "dm-recovery-concurrent-{}.example", + community_uuid.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + + let opener = [11u8; 32]; + let peer = [12u8; 32]; + let opened = open_dm(&pool, community, &[&peer], &opener) + .await + .expect("create dm"); + sqlx::query( + r#" + UPDATE channel_members + SET removed_at = NOW(), removed_by = $4 + WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3 + "#, + ) + .bind(community_uuid) + .bind(opened.channel.id) + .bind(peer.as_slice()) + .bind(opener.as_slice()) + .execute(&pool) + .await + .expect("soft-remove peer fixture"); + + let first_participants = [&peer[..]]; + let second_participants = [&peer[..]]; + let (first, second) = tokio::join!( + open_dm(&pool, community, &first_participants, &opener), + open_dm(&pool, community, &second_participants, &opener) + ); + let first = first.expect("first concurrent reopen"); + let second = second.expect("second concurrent reopen"); + + let restoration_count = [first, second] + .into_iter() + .filter(|result| result.restored_participants == vec![peer.to_vec()]) + .count(); + assert_eq!(restoration_count, 1, "peer must be restored exactly once"); + + let active_count: i64 = sqlx::query_scalar( + r#" + SELECT count(*) + FROM channel_members + WHERE community_id = $1 AND channel_id = $2 AND removed_at IS NULL + "#, + ) + .bind(community_uuid) + .bind(opened.channel.id) + .fetch_one(&pool) + .await + .expect("count active memberships"); + assert_eq!(active_count, 2); + } } diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index f7e10f79001..9eee66b5f71 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -33,6 +33,13 @@ use super::side_effects::{ emit_membership_notification, emit_system_message, publish_dm_visibility_snapshot, }; +fn map_open_dm_error(error: DbError) -> IngestError { + match error { + DbError::AccessDenied(message) => IngestError::Rejected(format!("restricted: {message}")), + other => IngestError::Internal(format!("error: db open_dm: {other}")), + } +} + /// Route a command-kind event to the appropriate handler. pub async fn handle_command( tenant: &TenantContext, @@ -350,7 +357,7 @@ async fn handle_dm_open( .db .open_dm(tenant.community(), &all_refs, &self_bytes) .await - .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + .map_err(map_open_dm_error)?; let channel = opened.channel; let was_created = opened.was_created; let restored_participants = opened.restored_participants; @@ -524,7 +531,7 @@ async fn handle_dm_add_member( .db .open_dm(tenant.community(), &all_refs, &self_bytes) .await - .map_err(|e| IngestError::Internal(format!("error: db open_dm: {e}")))?; + .map_err(map_open_dm_error)?; let new_channel = opened.channel; let was_created = opened.was_created; let restored_participants = opened.restored_participants; @@ -1395,6 +1402,16 @@ mod tests { use super::*; use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + #[test] + fn open_dm_access_denied_maps_to_restricted_client_error() { + let mapped = map_open_dm_error(DbError::AccessDenied("removed participant".to_string())); + + assert!(matches!( + mapped, + IngestError::Rejected(message) if message == "restricted: removed participant" + )); + } + async fn persistence_test_context() -> (buzz_db::Db, TenantContext) { let url = std::env::var("BUZZ_TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index d182bd58b41..843348887d0 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -905,6 +905,11 @@ pub fn emit_live_thread_summary( }); } +fn membership_transition_tag(transition_id: Uuid) -> anyhow::Result { + Tag::parse(["transition", &transition_id.to_string()]) + .map_err(|e| anyhow::anyhow!("failed to build transition tag: {e}")) +} + /// Emit a relay-signed membership notification event stored globally (channel_id = None). /// /// kind:44100 = member added, kind:44101 = member removed. @@ -926,6 +931,7 @@ pub async fn emit_membership_notification( .map_err(|e| anyhow::anyhow!("failed to build p tag: {e}"))?; let h_tag = Tag::parse(["h", &channel_id_str]) .map_err(|e| anyhow::anyhow!("failed to build h tag: {e}"))?; + let transition_tag = membership_transition_tag(Uuid::new_v4())?; let event_type = match notification_kind { KIND_MEMBER_ADDED_NOTIFICATION => "member_added", @@ -945,32 +951,11 @@ pub async fn emit_membership_notification( .to_string(); // A member can be removed and restored in the same second as the original - // add. Because these relay-signed notifications otherwise have identical - // kind, content, tags, and timestamp, Nostr gives them the same event id and - // `insert_event` suppresses the recovery fan-out as a duplicate. Advance - // past the latest notification for this target/channel pair so every real - // membership transition remains observable by live clients. - let now = nostr::Timestamp::now().as_secs(); - let previous = state - .db - .query_events(&buzz_db::event::EventQuery { - kinds: Some(vec![notification_kind as i32]), - pubkey: Some(state.relay_keypair.public_key().to_bytes().to_vec()), - p_tag_hex: Some(target_hex.clone()), - custom_tag: Some(("h".to_owned(), channel_id_str.clone())), - limit: Some(1), - global_only: true, - ..buzz_db::event::EventQuery::for_community(tenant.community()) - }) - .await?; - let created_at = previous - .first() - .map(|event| (event.event.created_at.as_secs() + 1).max(now)) - .unwrap_or(now); - + // add. Give every actual transition a collision-resistant tag so concurrent + // emitters cannot sign the same Nostr event and suppress one fan-out as a + // duplicate. let event = EventBuilder::new(Kind::Custom(notification_kind as u16), content) - .tags([p_tag, h_tag]) - .custom_created_at(nostr::Timestamp::from(created_at)) + .tags([p_tag, h_tag, transition_tag]) .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign membership notification: {e}"))?; @@ -3758,6 +3743,31 @@ pub async fn publish_nipia_unarchived( mod tests { use super::*; + #[test] + fn membership_transition_tag_changes_same_second_event_identity() { + let keys = nostr::Keys::generate(); + let timestamp = nostr::Timestamp::from(1_700_000_000); + let base_tags = || { + vec![ + Tag::parse(["p", &"01".repeat(32)]).expect("p tag"), + Tag::parse(["h", &Uuid::nil().to_string()]).expect("h tag"), + ] + }; + let build = |transition_id| { + let mut tags = base_tags(); + tags.push(membership_transition_tag(transition_id).expect("transition tag")); + EventBuilder::new(Kind::Custom(KIND_MEMBER_ADDED_NOTIFICATION as u16), "same") + .tags(tags) + .custom_created_at(timestamp) + .sign_with_keys(&keys) + .expect("sign notification") + }; + + let first = build(Uuid::from_u128(1)); + let second = build(Uuid::from_u128(2)); + assert_ne!(first.id, second.id); + } + #[test] fn group_members_snapshot_keeps_members_past_one_thousand() { let channel_id = Uuid::new_v4();