diff --git a/crates/buzz-db/src/store/channel.rs b/crates/buzz-db/src/store/channel.rs index 97f82d40c5..21cf59e46f 100644 --- a/crates/buzz-db/src/store/channel.rs +++ b/crates/buzz-db/src/store/channel.rs @@ -1655,6 +1655,31 @@ pub async fn soft_delete_channel( community_id: CommunityId, channel_id: Uuid, ) -> Result { + Ok( + soft_delete_channel_capturing_members(pool, community_id, channel_id) + .await? + .is_some(), + ) +} + +/// A deleted channel's type and the members active when it was deleted. +#[derive(Debug, Clone)] +pub struct DeletedChannel { + /// The channel's type, e.g. `stream`. + pub channel_type: String, + /// The pubkeys of its active members, oldest first. + pub members: Vec>, +} + +/// [`soft_delete_channel`], also returning who was a member at the deletion's +/// serialization point: read under the same membership lock, so a join either +/// commits first and is included, or waits and is then refused. `None` when the +/// channel was already deleted or never existed. +pub async fn soft_delete_channel_capturing_members( + pool: &PgPool, + community_id: CommunityId, + channel_id: Uuid, +) -> Result> { let mut tx = crate::begin_community_write_transaction( pool, community_id, @@ -1663,12 +1688,31 @@ pub async fn soft_delete_channel( .await?; let mut writer = match lock_channel_for_delete(&mut tx, channel_id).await { Ok(writer) => writer, - Err(DbError::ChannelNotFound(_)) => return Ok(false), + Err(DbError::ChannelNotFound(_)) => return Ok(None), Err(error) => return Err(error), }; + let channel_type: String = sqlx::query_scalar( + "SELECT channel_type::text FROM channels WHERE community_id = $1 AND id = $2", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_one(writer.tx.conn()) + .await?; + let members: Vec> = sqlx::query_scalar( + "SELECT pubkey FROM channel_members \ + WHERE community_id = $1 AND channel_id = $2 AND removed_at IS NULL \ + ORDER BY joined_at ASC", + ) + .bind(community_id.as_uuid()) + .bind(channel_id) + .fetch_all(writer.tx.conn()) + .await?; soft_delete_on(&mut writer).await?; tx.commit().await?; - Ok(true) + Ok(Some(DeletedChannel { + channel_type, + members, + })) } async fn soft_delete_on(w: &mut ChannelSettingsWriter<'_>) -> Result<()> { @@ -2061,6 +2105,16 @@ impl Db { soft_delete_channel(&self.pool, community_id, channel_id).await } + /// See [`soft_delete_channel_capturing_members`]. + #[datastore_span(name = "soft_delete_channel_capturing_members", system = "postgresql")] + pub async fn soft_delete_channel_capturing_members( + &self, + community_id: CommunityId, + channel_id: Uuid, + ) -> Result> { + soft_delete_channel_capturing_members(&self.pool, community_id, channel_id).await + } + /// Archive ephemeral channels whose TTL deadline has passed. #[datastore_span(name = "reap_expired_ephemeral_channels", system = "postgresql")] pub async fn reap_expired_ephemeral_channels(&self) -> Result> { @@ -2848,6 +2902,84 @@ mod postgres_tests { assert_eq!(members.len(), 1, "only the first creator is a member"); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn soft_delete_captures_members_at_its_serialization_point() { + let pool = setup_pool().await; + let community_id = make_test_community(&pool).await; + let community = CommunityId::from_uuid(community_id); + let owner = random_pubkey(); + let early = random_pubkey(); + let late = random_pubkey(); + let channel = create_test_channel( + &pool, + community_id, + "delete-captures-members", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner, + None, + ) + .await + .expect("create channel"); + let channel_id = channel.id; + + // Queue a join, then the delete, behind a held membership lock: the + // join is serialized before the delete, so the delete must see it. + let snapshot = + crate::channel_members::lock_member_snapshot(&pool, community, channel_id, &[9u8; 32]) + .await + .expect("hold membership lock"); + let join_pool = crate::test_support::named_pool("join-before-delete").await; + let early_key = early.clone(); + let join = tokio::spawn(async move { + crate::channel_members::add_member( + &join_pool, + community, + channel_id, + &early_key, + MemberRole::Member, + None, + ) + .await + }); + crate::test_support::wait_for_lock_wait(&pool, "join-before-delete").await; + let delete_pool = crate::test_support::named_pool("delete-after-join").await; + let delete = tokio::spawn(async move { + soft_delete_channel_capturing_members(&delete_pool, community, channel_id).await + }); + crate::test_support::wait_for_lock_wait(&pool, "delete-after-join").await; + snapshot.release().await.expect("release membership lock"); + + join.await.expect("join task").expect("join before delete"); + let deleted = delete + .await + .expect("delete task") + .expect("delete") + .expect("channel was live"); + assert_eq!(deleted.channel_type, "stream"); + assert_eq!(deleted.members, vec![owner, early]); + + // A join after the delete is refused, so nobody is left un-notified. + assert!(crate::channel_members::add_member( + &pool, + community, + channel_id, + &late, + MemberRole::Member, + None, + ) + .await + .is_err()); + assert!( + soft_delete_channel_capturing_members(&pool, community, channel_id) + .await + .expect("second delete") + .is_none() + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn soft_delete_waits_for_membership_lock() { diff --git a/crates/buzz-relay/src/api/admin/mod.rs b/crates/buzz-relay/src/api/admin/mod.rs index f6ee03af3c..296eba8fec 100644 --- a/crates/buzz-relay/src/api/admin/mod.rs +++ b/crates/buzz-relay/src/api/admin/mod.rs @@ -9467,6 +9467,190 @@ mod postgres_tests { // ── Kick live side effects: crash-recovery and re-add race ──────────────── + /// A channel deleted between a committed kick and its recovery leaves no + /// roster to refresh. Recovery must treat that as convergence and finalize, + /// without running the removal again. + #[tokio::test] + #[ignore = "requires Postgres — a deleted channel does not strand a committed kick"] + async fn crash_recovery_finalizes_kick_after_channel_deleted() { + let pool = e2e_pool().await; + let (community_id, host) = e2e_community(&pool, "kick-recovery-deleted").await; + let cid = buzz_core::CommunityId::from_uuid(community_id); + let target = vec![0xAAu8; 32]; + let actor = vec![0xBBu8; 32]; + + // Seed channel + member. + let channel_id = uuid::Uuid::new_v4(); + sqlx::query( + r#"INSERT INTO channels (id, community_id, name, channel_type, visibility, created_by) + VALUES ($1, $2, 'recovery-deleted-ch', 'stream', 'open', $3)"#, + ) + .bind(channel_id) + .bind(community_id) + .bind(&actor) + .execute(&pool) + .await + .expect("create channel"); + sqlx::query( + "INSERT INTO channel_members (community_id, channel_id, pubkey, role) VALUES ($1, $2, $3, 'member')", + ) + .bind(community_id) + .bind(channel_id) + .bind(&target) + .execute(&pool) + .await + .expect("add member"); + + let report_id = e2e_report_pubkey(&pool, community_id, &target).await; + + // Claim → enforcing → lease → kick + commit marker (simulates process + // that committed the kick but crashed before live side effects ran). + let action_id = match buzz_db::relay_admin_actions::claim_report( + &pool, + cid, + report_id, + uuid::Uuid::new_v4(), + &actor, + "operator", + "kick", + None, + None, + "resolve:kick", + "relay_operator", + Some(&target), + None, + Some(channel_id), + ) + .await + .expect("claim") + { + buzz_db::relay_admin_actions::ClaimResult::Claimed(a) => a.id, + other => panic!("expected Claimed, got {other:?}"), + }; + let _ = buzz_db::relay_admin_actions::begin_enforcing(&pool, action_id) + .await + .expect("begin_enforcing"); + let lease_until = chrono::Utc::now() + chrono::Duration::seconds(60); + let lease_token = + match buzz_db::relay_admin_actions::acquire_action_lease(&pool, action_id, lease_until) + .await + .expect("acquire lease") + { + buzz_db::relay_admin_actions::LeaseResult::Acquired(t) => t, + other => panic!("expected Acquired, got {other:?}"), + }; + let kick_result = buzz_db::relay_admin_actions::execute_kick_with_marker( + &pool, + action_id, + lease_token, + cid, + channel_id, + &target, + &actor, + ) + .await + .expect("execute_kick_with_marker"); + assert!( + matches!( + kick_result, + buzz_db::relay_admin_actions::KickWithMarkerResult::Removed + ), + "kick must commit before the simulated crash" + ); + + // Expire the lease so the recovery worker can re-claim. + sqlx::query( + "UPDATE relay_admin_actions SET action_lease_expires_at = $2, action_lease_token = NULL WHERE id = $1", + ) + .bind(action_id) + .bind(chrono::Utc::now() - chrono::Duration::seconds(300)) + .execute(&pool) + .await + .expect("expire lease"); + + let removed_at = || async { + sqlx::query_scalar::<_, Option>>( + "SELECT removed_at FROM channel_members WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3", + ) + .bind(community_id) + .bind(channel_id) + .bind(&target) + .fetch_one(&pool) + .await + .expect("member row") + }; + let removed_before = removed_at().await.expect("kick removed the member"); + + // Discovery published before the delete, which is stale once the + // channel is gone; recovery must clean it up before the notice. + let state = state_from_pool(pool.clone()).await; + let tenant = e2e_tenant(community_id, &host); + crate::handlers::side_effects::emit_group_discovery_events(&tenant, &state, channel_id) + .await + .expect("seed discovery"); + let live_discovery = || async { + sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM events WHERE community_id = $1 AND channel_id = $2 \ + AND kind IN (39000, 39001, 39002) AND deleted_at IS NULL", + ) + .bind(community_id) + .bind(channel_id) + .fetch_one(&pool) + .await + .expect("count discovery") + }; + assert!(live_discovery().await > 0, "stale discovery seeded"); + assert!( + buzz_db::channel::soft_delete_channel(&pool, cid, channel_id) + .await + .expect("delete channel") + ); + let mut barrier = crate::handlers::side_effects::notice_barrier::install(&target); + let batch = buzz_db::relay_admin_actions::claim_stranded_action_batch( + &pool, + "e2e-kick-recovery-deleted", + chrono::Utc::now() + chrono::Duration::seconds(120), + 1000, + ) + .await + .expect("claim_stranded_action_batch"); + let claim = batch + .into_iter() + .find(|c| c.record.id == action_id) + .expect("stranded action must appear in batch"); + let recovery = tokio::spawn({ + let state = state.clone(); + async move { crate::handlers::admin_action_worker::recover_one(&state, claim).await } + }); + let (noticed, release) = + tokio::time::timeout(std::time::Duration::from_secs(10), barrier.next()) + .await + .expect("recovery sent no removal notice") + .expect("barrier open"); + assert_eq!(noticed, channel_id); + assert_eq!( + live_discovery().await, + 0, + "removal notice sent before discovery cleanup" + ); + release.send(()).expect("recovery waiting"); + recovery.await.expect("recovery task"); + + let final_rec = buzz_db::relay_admin_actions::get_action(&pool, action_id) + .await + .expect("get_action") + .expect("record must still exist"); + assert_eq!( + final_rec.state, "succeeded", + "recovery must finalize the kick" + ); + assert_eq!( + removed_at().await, + Some(removed_before), + "recovery must not run the removal again" + ); + } + /// Recovery seam: commit kick marker, simulate crash, re-drive via the real /// recovery worker entry point and assert all three live side effects fire: /// membership cache cleared, channel subscription evicted, workflow disabled. diff --git a/crates/buzz-relay/src/handlers/membership_notice_postgres_tests.rs b/crates/buzz-relay/src/handlers/membership_notice_postgres_tests.rs new file mode 100644 index 0000000000..ded5b5e1a5 --- /dev/null +++ b/crates/buzz-relay/src/handlers/membership_notice_postgres_tests.rs @@ -0,0 +1,436 @@ +//! Postgres-backed tests for the membership notices sent by a channel delete, +//! a moderation kick and a moderation DM: each notice reaches the member's +//! live subscription, and a channel-list read made on receipt already shows +//! the change. "On receipt" is enforced by `notice_barrier`, which parks the +//! producer at the notice while the test reads. + +use std::sync::Arc; + +use buzz_core::kind::{ + KIND_DM_HIDE, KIND_DM_VISIBILITY, KIND_MEMBER_ADDED_NOTIFICATION, + KIND_MEMBER_REMOVED_NOTIFICATION, +}; +use buzz_core::TenantContext; +use nostr::{EventBuilder, Keys, Kind, Tag}; +use serde_json::{json, Value}; +use tokio::sync::mpsc::Receiver; +use uuid::Uuid; + +use super::group_state_postgres_tests::{create_channel, http_auth, test_conn}; +use super::side_effects::notice_barrier; +use crate::handlers::ingest::ingest_event; +use crate::state::AppState; + +async fn fixture() -> (Arc, TenantContext) { + let state = crate::api::bridge::postgres_tests::bridge_handler_test_state() + .await + .expect("local Postgres and Redis"); + let host = format!("membership-notices-{}.local", Uuid::new_v4().simple()); + let community = state + .db + .ensure_configured_community(&host) + .await + .expect("ensure community") + .id; + (state, TenantContext::resolved(community, &host)) +} + +async fn publish( + state: &Arc, + tenant: &TenantContext, + keys: &Keys, + kind: u16, + tags: &[&[&str]], +) { + let tags: Vec = tags + .iter() + .map(|tag| Tag::parse(tag.iter().copied()).unwrap()) + .collect(); + let event = EventBuilder::new(Kind::Custom(kind), "") + .tags(tags) + .sign_with_keys(keys) + .expect("sign"); + let result = ingest_event(state, tenant, event, http_auth(keys)) + .await + .unwrap_or_else(|error| panic!("ingest kind:{kind}: {error:?}")); + assert!(result.accepted, "kind:{kind} refused: {}", result.message); +} + +/// Register `reader` as a connected client holding the global notice +/// subscription clients send, and return its outbound frames. +fn subscribe_notices( + state: &Arc, + tenant: &TenantContext, + reader: &Keys, +) -> Receiver { + let conn_id = Uuid::new_v4(); + let (tx, rx) = tokio::sync::mpsc::channel(32); + state.conn_manager.register( + conn_id, + tx, + tokio::sync::mpsc::channel(4).0, + tokio::sync::mpsc::channel(1).0, + None, + tokio_util::sync::CancellationToken::new(), + tenant.community(), + Arc::new(std::sync::atomic::AtomicU8::new(0)), + Arc::new(tokio::sync::Mutex::new(Default::default())), + 3, + crate::state::CommunityConnectionControl::new(tokio_util::sync::CancellationToken::new()), + ); + state + .conn_manager + .set_authenticated_pubkey(conn_id, reader.public_key().to_bytes().to_vec()); + let filter = json!({ + "kinds": [KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION], + "#p": [reader.public_key().to_hex()], + }); + state.sub_registry.register_scoped( + tenant.community(), + conn_id, + "notices".into(), + vec![serde_json::from_value(filter).expect("filter")], + None, + ); + rx +} + +/// The `(kind, h, id)` of every live notice delivered so far. +fn live_notices(rx: &mut Receiver) -> Vec<(u64, String, String)> { + let mut notices = Vec::new(); + while let Ok(frame) = rx.try_recv() { + let axum::extract::ws::Message::Text(text) = frame else { + continue; + }; + let message: Value = serde_json::from_str(&text).expect("relay frame"); + if message[0] == "EVENT" { + let event = &message[2]; + notices.push(( + event["kind"].as_u64().unwrap(), + tag_values(event, "h").concat(), + event["id"].as_str().unwrap().to_owned(), + )); + } + } + notices +} + +fn notices(rx: &mut Receiver) -> Vec<(u64, String)> { + live_notices(rx) + .into_iter() + .map(|(kind, h, _)| (kind, h)) + .collect() +} + +/// The notices delivered once at least one arrives, for notices sent off +/// the request path. +async fn next_notices(rx: &mut Receiver) -> Vec<(u64, String)> { + tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + let delivered = notices(rx); + if !delivered.is_empty() { + return delivered; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + }) + .await + .expect("no notice delivered") +} + +fn tag_values(event: &Value, name: &str) -> Vec { + event["tags"] + .as_array() + .unwrap() + .iter() + .filter(|tag| tag[0] == name) + .map(|tag| tag[1].as_str().unwrap().to_owned()) + .collect() +} + +/// The `name` tag values of the events a one-shot read as `reader` returns. +async fn read_tags( + state: &Arc, + tenant: &TenantContext, + reader: &Keys, + filter: Value, + name: &str, +) -> Vec { + let (conn, mut rx) = test_conn(tenant, reader); + let filters: Vec = serde_json::from_value(json!([filter])).unwrap(); + crate::handlers::req::handle_req("read".into(), filters, vec![None], conn, state.clone()).await; + let mut values = Vec::new(); + while let Ok(frame) = rx.try_recv() { + let axum::extract::ws::Message::Text(text) = frame else { + continue; + }; + let message: Value = serde_json::from_str(&text).expect("relay frame"); + match message[0].as_str() { + Some("EVENT") => values.extend(tag_values(&message[2], name)), + Some("EOSE") | Some("CLOSED") => return values, + _ => {} + } + } + panic!("no EOSE"); +} + +/// A channel-list read as `reader`: the channels its 39002 discovery names, +/// as a client builds its sidebar. +async fn listed_channels( + state: &Arc, + tenant: &TenantContext, + reader: &Keys, +) -> Vec { + let filter = json!({"kinds": [39002], "#p": [reader.public_key().to_hex()]}); + read_tags(state, tenant, reader, filter, "d").await +} + +/// The DMs `reader`'s current NIP-DV snapshot hides from its sidebar, read +/// as a client does after its channel list refreshes. +async fn hidden_dms(state: &Arc, tenant: &TenantContext, reader: &Keys) -> Vec { + let filter = json!({"kinds": [KIND_DM_VISIBILITY], "#p": [reader.public_key().to_hex()]}); + read_tags(state, tenant, reader, filter, "h").await +} + +/// Wait for the producer to reach its next notice to the parked reader, run +/// `on_receipt` while it is held there, then let it continue. Returns the +/// notice's channel and what `on_receipt` saw. +pub(crate) async fn at_receipt( + barrier: &mut notice_barrier::Barrier, + on_receipt: impl AsyncFnOnce(String) -> T, +) -> (String, T) { + let (channel, release) = + tokio::time::timeout(std::time::Duration::from_secs(10), barrier.next()) + .await + .expect("producer reached no notice") + .expect("barrier open"); + let seen = on_receipt(channel.to_string()).await; + release.send(()).expect("producer waiting"); + (channel.to_string(), seen) +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn channel_delete_notifies_every_member_after_cleanup() { + let (state, tenant) = fixture().await; + let owner = Keys::generate(); + let member = Keys::generate(); + let channel = create_channel(&state, &tenant, &owner, &[]).await; + let h = channel.to_string(); + publish( + &state, + &tenant, + &owner, + 9000, + &[&["h", &h], &["p", &member.public_key().to_hex()]], + ) + .await; + assert!(listed_channels(&state, &tenant, &member).await.contains(&h)); + let mut owner_rx = subscribe_notices(&state, &tenant, &owner); + let mut member_rx = subscribe_notices(&state, &tenant, &member); + let mut barrier = notice_barrier::install(&member.public_key().to_bytes()); + + // The 9008 is accepted while the member's notice is still held, so the + // notices are off the request path. + tokio::time::timeout( + std::time::Duration::from_secs(5), + publish(&state, &tenant, &owner, 9008, &[&["h", &h]]), + ) + .await + .expect("9008 OK waited on the held notice"); + assert!( + notices(&mut member_rx).is_empty(), + "notice delivered before the OK" + ); + let (noticed, listed) = at_receipt(&mut barrier, async |_| { + listed_channels(&state, &tenant, &member).await + }) + .await; + + assert_eq!(noticed, h); + assert!(!listed.contains(&h), "notice sent before discovery cleanup"); + for rx in [&mut owner_rx, &mut member_rx] { + assert_eq!( + next_notices(rx).await, + vec![(KIND_MEMBER_REMOVED_NOTIFICATION as u64, h.clone())] + ); + } +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn moderation_kick_notifies_the_kicked_member_after_cleanup() { + let (state, tenant) = fixture().await; + let owner = Keys::generate(); + let member = Keys::generate(); + let channel = create_channel(&state, &tenant, &owner, &[]).await; + let h = channel.to_string(); + let member_bytes = member.public_key().to_bytes(); + publish( + &state, + &tenant, + &owner, + 9000, + &[&["h", &h], &["p", &member.public_key().to_hex()]], + ) + .await; + assert!(listed_channels(&state, &tenant, &member).await.contains(&h)); + let mut rx = subscribe_notices(&state, &tenant, &member); + let mut barrier = notice_barrier::install(&member_bytes); + + // The kick mutation commits the removal; live effects follow. + state + .db + .remove_member( + tenant.community(), + channel, + &member_bytes, + &owner.public_key().to_bytes(), + ) + .await + .expect("remove member"); + let effects = tokio::spawn({ + let (state, tenant) = (state.clone(), tenant.clone()); + async move { + crate::handlers::side_effects::apply_kick_live_side_effects( + &tenant, + &state, + channel, + &member_bytes, + ) + .await + } + }); + let (noticed, listed) = at_receipt(&mut barrier, async |_| { + listed_channels(&state, &tenant, &member).await + }) + .await; + effects.await.expect("task").expect("kick live effects"); + + assert_eq!(noticed, h); + assert!( + !listed.contains(&h), + "notice sent before the roster refresh" + ); + assert_eq!( + notices(&mut rx), + vec![(KIND_MEMBER_REMOVED_NOTIFICATION as u64, h.clone())] + ); +} + +/// A recipient whose channel set is cached when its moderation DM is +/// created still finds the DM in the read its notice prompts. +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn moderation_dm_notice_follows_a_cached_channel_list() { + let (state, tenant) = fixture().await; + let recipient = Keys::generate(); + let recipient_bytes = recipient.public_key().to_bytes(); + // Warm the recipient's cached channel set, as a recent REQ does. + assert_eq!( + listed_channels(&state, &tenant, &recipient).await, + Vec::::new() + ); + let mut barrier = notice_barrier::install(&recipient_bytes); + + let deliver = tokio::spawn({ + let (state, tenant) = (state.clone(), tenant.clone()); + async move { + crate::handlers::moderation_notices::send_moderation_notice( + &tenant, + &state, + &recipient_bytes, + crate::handlers::moderation_notices::ModerationNotice::ContentActioned { + action_id: Uuid::new_v4(), + public_reason: "spam".into(), + }, + chrono::Utc::now(), + ) + .await + } + }); + let (dm, listed) = at_receipt(&mut barrier, async |_| { + listed_channels(&state, &tenant, &recipient).await + }) + .await; + deliver.await.expect("task").expect("notice"); + + assert!(listed.contains(&dm), "notice sent before the cache drop"); +} + +/// A moderation DM resurfaces the DM, and on receipt of its notice the +/// channel list shows it and the NIP-DV snapshot no longer hides it. The +/// second delivery follows a real hide in the same (pinned) second and still +/// sends its own notice, so a client that only hears notices sees it return. +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn moderation_dm_notifies_the_recipient_once_visible() { + let (state, tenant) = fixture().await; + let recipient = Keys::generate(); + let recipient_bytes = recipient.public_key().to_bytes(); + let mut rx = subscribe_notices(&state, &tenant, &recipient); + let mut barrier = notice_barrier::install(&recipient_bytes); + barrier.pin_created_at(nostr::Timestamp::now().as_secs()); + let deliver = || { + let (state, tenant) = (state.clone(), tenant.clone()); + tokio::spawn(async move { + crate::handlers::moderation_notices::send_moderation_notice( + &tenant, + &state, + &recipient_bytes, + crate::handlers::moderation_notices::ModerationNotice::ContentActioned { + action_id: Uuid::new_v4(), + public_reason: "spam".into(), + }, + chrono::Utc::now(), + ) + .await + }) + }; + let on_receipt = async |dm: String| { + ( + listed_channels(&state, &tenant, &recipient) + .await + .contains(&dm), + hidden_dms(&state, &tenant, &recipient).await.contains(&dm), + ) + }; + + let first = deliver(); + let (dm, seen) = at_receipt(&mut barrier, on_receipt).await; + first.await.expect("task").expect("first notice"); + assert_eq!(seen, (true, false), "listed and visible on receipt"); + let first_notice = live_notices(&mut rx); + assert_eq!(first_notice.len(), 1, "{first_notice:?}"); + assert_eq!( + (first_notice[0].0, &first_notice[0].1), + (KIND_MEMBER_ADDED_NOTIFICATION as u64, &dm) + ); + + publish( + &state, + &tenant, + &recipient, + KIND_DM_HIDE as u16, + &[&["h", &dm]], + ) + .await; + assert!(hidden_dms(&state, &tenant, &recipient).await.contains(&dm)); + + let second = deliver(); + let (again, seen) = at_receipt(&mut barrier, on_receipt).await; + second.await.expect("task").expect("second notice"); + assert_eq!(again, dm); + assert_eq!(seen, (true, false), "listed and visible on receipt"); + let second_notice = live_notices(&mut rx); + assert_eq!( + second_notice.len(), + 1, + "the second delivery's notice coalesced" + ); + assert_eq!( + (second_notice[0].0, &second_notice[0].1), + (KIND_MEMBER_ADDED_NOTIFICATION as u64, &dm) + ); + assert_ne!(second_notice[0].2, first_notice[0].2); +} diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index cb18b8525c..43e13e973a 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -32,6 +32,8 @@ pub mod identity_archive; pub mod imeta; /// Transport-neutral event ingestion pipeline. pub mod ingest; +#[cfg(test)] +mod membership_notice_postgres_tests; /// Community moderation authorization seam (capability helper). pub mod moderation_authz; /// Community moderation command handler (kinds 9040–9044). diff --git a/crates/buzz-relay/src/handlers/moderation_notices.rs b/crates/buzz-relay/src/handlers/moderation_notices.rs index e5e4b9c3e3..d3914704c9 100644 --- a/crates/buzz-relay/src/handlers/moderation_notices.rs +++ b/crates/buzz-relay/src/handlers/moderation_notices.rs @@ -25,11 +25,14 @@ use nostr::{EventBuilder, Kind, Tag}; use tracing::warn; use uuid::Uuid; -use buzz_core::kind::{event_kind_u32, KIND_STREAM_MESSAGE}; +use buzz_core::kind::{event_kind_u32, KIND_MEMBER_ADDED_NOTIFICATION, 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_group_discovery_events, emit_membership_notification_tagged, + publish_dm_visibility_snapshot, +}; use crate::state::AppState; /// Tag naming the moderation source row (report/action) a notice was derived @@ -115,6 +118,13 @@ pub async fn send_moderation_notice( ) .await?; let dm_channel_id = dm_channel.id; + // Drop both participants' cached channel sets on every pod, as opening a + // DM does (`command_executor::handle_dm_open`), so the read the notice + // below prompts includes this DM. Done on every delivery, not just on + // creation, so a retry after a failed first delivery still drops them. + for participant in [recipient_pubkey, relay_pubkey_bytes.as_slice()] { + state.invalidate_membership(tenant, dm_channel_id, participant); + } // Count new DM creation; side-effect gates below intentionally do not // gate on was_created (see comment at step 2). @@ -135,6 +145,9 @@ pub async fn send_moderation_notice( .db .unhide_dm(tenant.community(), dm_channel_id, recipient_pubkey) .await?; + // Clients filter the sidebar on the NIP-DV snapshot, not `hidden_at`, so + // republish it before the notice that makes them re-read. + publish_dm_visibility_snapshot(tenant, state, recipient_pubkey).await?; // 2. Ensure the relay's "{host} Moderation" kind:0 profile exists, and 3. // the DM's kind:39000 discovery (with `hidden` / `t=dm` / `p`). Both are @@ -149,6 +162,25 @@ pub async fn send_moderation_notice( } emit_group_discovery_events(tenant, state, dm_channel_id).await?; + // The DM is now readable in the recipient's channel list: tell them, as + // opening a DM does. Sent on every delivery so a new, existing or hidden + // DM resurfaces live; a repeated notice is a harmless hint. The source tag + // keeps two deliveries in the same second from sharing one notice. + let source_id = notice.source_id(); + if let Err(e) = emit_membership_notification_tagged( + tenant, + state, + dm_channel_id, + recipient_pubkey, + relay_pubkey_bytes.as_slice(), + KIND_MEMBER_ADDED_NOTIFICATION, + &[Tag::parse([MODERATION_SOURCE_TAG, &source_id.to_string()])?], + ) + .await + { + warn!(channel = %dm_channel_id, error = %e, "membership notification emission failed"); + } + // 4. Insert the relay-signed kind:9 notice with `h=` and a // `moderation_source` tag naming the source row id (idempotency + // client linking). @@ -159,7 +191,6 @@ pub async fn send_moderation_notice( // (same pubkey + created_at + kind + tags + content = same SHA256 event ID). // `insert_event`'s ON CONFLICT DO NOTHING ensures exactly one row is // durably persisted regardless of how many workers reach this point. - let source_id = notice.source_id(); let tags = vec![ Tag::parse(["h", &dm_channel_id.to_string()])?, Tag::parse([MODERATION_SOURCE_TAG, &source_id.to_string()])?, diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 61228f4e13..d9238b6027 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -159,6 +159,43 @@ pub(crate) async fn apply_kick_live_side_effects( return Err(e.into()); } } + + // The removal is committed: refresh the channel's discovery roster, then + // tell the kicked member, as a kind:9001 removal does. The notice tells a + // client to re-read that roster, so a failed refresh returns before it and + // the moderation driver retries the action. A channel deleted since the + // kick has no roster to refresh: that is convergence, once its discovery + // is gone. A recovery re-drive repeats both; the refresh is replaceable + // and a repeated notice is a harmless hint. A failed notice is logged and + // does not undo the kick. + if let Err(e) = emit_group_discovery_events(tenant, state, channel_id).await { + if !matches!( + e.downcast_ref::(), + Some(buzz_db::DbError::ChannelNotFound(_)) + ) { + return Err(e); + } + state + .db + .soft_delete_discovery_events( + tenant.community(), + channel_id, + state.relay_keypair.public_key().as_bytes(), + ) + .await?; + } + if let Err(e) = emit_membership_notification( + tenant, + state, + channel_id, + target_pubkey, + state.relay_keypair.public_key().as_bytes(), + KIND_MEMBER_REMOVED_NOTIFICATION, + ) + .await + { + warn!(channel = %channel_id, error = %e, "membership notification emission failed"); + } Ok(()) } @@ -1162,6 +1199,87 @@ async fn publish_live_thread_summary( .await; } +/// Test-only barrier at the point a membership notice is sent: a test parks +/// the producer there, reads what a client would read on receipt, then lets +/// it continue. It can also pin the notices' clock. Keyed by recipient so +/// parallel tests do not interfere. +#[cfg(test)] +pub(crate) mod notice_barrier { + use std::collections::HashMap; + use std::sync::{Mutex, OnceLock}; + use tokio::sync::{mpsc, oneshot}; + + use uuid::Uuid; + + /// A parked notice: its channel, and the handle that lets it continue. + pub(crate) type Parked = (Uuid, oneshot::Sender<()>); + type Waiters = Mutex, mpsc::UnboundedSender>>; + + fn waiters() -> &'static Waiters { + static WAITERS: OnceLock = OnceLock::new(); + WAITERS.get_or_init(Default::default) + } + + /// Parks every notice to one recipient while held; dropping it removes + /// the recipient's waiter and clock pin. + pub(crate) struct Barrier { + recipient: Vec, + parked: mpsc::UnboundedReceiver, + } + + impl Barrier { + /// The next notice to reach the barrier, with its release handle. + pub(crate) async fn next(&mut self) -> Option { + self.parked.recv().await + } + + /// Sign every later notice to the recipient at `created_at`, as if all + /// were sent in the same second. + pub(crate) fn pin_created_at(&self, created_at: u64) { + pins() + .lock() + .unwrap() + .insert(self.recipient.clone(), created_at); + } + } + + impl Drop for Barrier { + fn drop(&mut self) { + waiters().lock().unwrap().remove(&self.recipient); + pins().lock().unwrap().remove(&self.recipient); + } + } + + /// Park every notice to `recipient` until the returned barrier drops. + pub(crate) fn install(recipient: &[u8]) -> Barrier { + let (tx, parked) = mpsc::unbounded_channel(); + waiters().lock().unwrap().insert(recipient.to_vec(), tx); + Barrier { + recipient: recipient.to_vec(), + parked, + } + } + + fn pins() -> &'static Mutex, u64>> { + static PINS: OnceLock, u64>>> = OnceLock::new(); + PINS.get_or_init(Default::default) + } + + pub(super) fn pinned_created_at(recipient: &[u8]) -> Option { + pins().lock().unwrap().get(recipient).copied() + } + + pub(super) async fn reached(channel_id: Uuid, recipient: &[u8]) { + let waiter = waiters().lock().unwrap().get(recipient).cloned(); + if let Some(waiter) = waiter { + let (release, released) = oneshot::channel(); + if waiter.send((channel_id, release)).is_ok() { + let _ = released.await; + } + } + } +} + /// Emit a relay-signed membership notification event stored globally (channel_id = None). /// /// kind:44100 = member added, kind:44101 = member removed. Every notice @@ -1179,6 +1297,30 @@ pub async fn emit_membership_notification( target_pubkey: &[u8], actor_pubkey: &[u8], notification_kind: u32, +) -> anyhow::Result<()> { + emit_membership_notification_tagged( + tenant, + state, + channel_id, + target_pubkey, + actor_pubkey, + notification_kind, + &[], + ) + .await +} + +/// [`emit_membership_notification`] with `extra_tags` after the standard +/// ones. A tag that differs per cause, such as a moderation source, keeps two +/// notices in the same second from sharing one event id. +pub(crate) async fn emit_membership_notification_tagged( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + target_pubkey: &[u8], + actor_pubkey: &[u8], + notification_kind: u32, + extra_tags: &[Tag], ) -> anyhow::Result<()> { // No channel row means no type to tag. Send nothing rather than an // untagged notice, which every client would treat as a legacy type. @@ -1198,12 +1340,14 @@ pub async fn emit_membership_notification( target_pubkey, actor_pubkey, notification_kind, + extra_tags, ) .await } /// [`emit_membership_notification`] for a caller that already read the /// channel's type, such as a kind:9000 notifying several people. +#[allow(clippy::too_many_arguments)] async fn emit_typed_membership_notification( tenant: &TenantContext, state: &Arc, @@ -1212,6 +1356,7 @@ async fn emit_typed_membership_notification( target_pubkey: &[u8], actor_pubkey: &[u8], notification_kind: u32, + extra_tags: &[Tag], ) -> anyhow::Result<()> { let target_hex = hex::encode(target_pubkey); let actor_hex = hex::encode(actor_pubkey); @@ -1241,8 +1386,18 @@ async fn emit_typed_membership_notification( let type_tag = Tag::parse([buzz_core::filter::NOTICE_CHANNEL_TYPE_TAG, channel_type]) .map_err(|e| anyhow::anyhow!("failed to build channel type tag: {e}"))?; - let event = EventBuilder::new(Kind::Custom(notification_kind as u16), content) + #[cfg(test)] + notice_barrier::reached(channel_id, target_pubkey).await; + + let builder = EventBuilder::new(Kind::Custom(notification_kind as u16), content) .tags([p_tag, h_tag, type_tag]) + .tags(extra_tags.iter().cloned()); + #[cfg(test)] + let builder = match notice_barrier::pinned_created_at(target_pubkey) { + Some(at) => builder.custom_created_at(nostr::Timestamp::from(at)), + None => builder, + }; + let event = builder .sign_with_keys(&state.relay_keypair) .map_err(|e| anyhow::anyhow!("failed to sign membership notification: {e}"))?; @@ -2048,6 +2203,7 @@ pub(crate) async fn persist_put_user( &target.pubkey, &actor_bytes, KIND_MEMBER_ADDED_NOTIFICATION, + &[], ) .await { @@ -2704,6 +2860,9 @@ async fn handle_create_group( Ok(()) } +/// Concurrent 44101 notices a channel delete sends to its members. +const DELETE_NOTICE_CONCURRENCY: usize = 8; + async fn handle_delete_group( tenant: &TenantContext, event: &Event, @@ -2713,14 +2872,15 @@ async fn handle_delete_group( extract_h_tag_channel(event).ok_or_else(|| anyhow::anyhow!("missing h tag"))?; let actor_bytes = event.pubkey.to_bytes().to_vec(); - // Soft-delete the channel. + // Soft-delete the channel, capturing its members under the same lock so a + // concurrent join is either included or refused. let deleted = state .db - .soft_delete_channel(tenant.community(), channel_id) + .soft_delete_channel_capturing_members(tenant.community(), channel_id) .await .map_err(|e| anyhow::anyhow!("soft_delete_channel failed: {e}"))?; - if !deleted { + if deleted.is_none() { warn!(channel = %channel_id, "channel already deleted or not found"); } @@ -2741,6 +2901,36 @@ async fn handle_delete_group( // Stale is_member=true entries would bypass the DB's deleted_at guard. state.invalidate_channel_deleted(tenant); + // Notify every member only after the delete and discovery cleanup, so a + // client that re-reads on the notice no longer sees the channel. The + // notices run off the request path, bounded like `handle_req`'s filter + // reads, so a large channel neither holds the OK nor depends on the + // system message below. + if let Some(deleted) = deleted { + let (tenant, state, actor_bytes) = (tenant.clone(), state.clone(), actor_bytes.clone()); + tokio::spawn(async move { + use futures_util::stream::{self, StreamExt}; + stream::iter(&deleted.members) + .for_each_concurrent(DELETE_NOTICE_CONCURRENCY, |member| async { + if let Err(e) = emit_typed_membership_notification( + &tenant, + &state, + channel_id, + &deleted.channel_type, + member, + &actor_bytes, + KIND_MEMBER_REMOVED_NOTIFICATION, + &[], + ) + .await + { + warn!(channel = %channel_id, error = %e, "membership notification emission failed"); + } + }) + .await; + }); + } + let actor_hex = hex::encode(&actor_bytes); emit_system_message( tenant,