fix(relay): send membership notices on channel delete, moderation kick and moderation DM (#8977)

Related: [buzz-app#2215](https://github.com/block/buzz-app/pull/2215)

## Problem

A connected client learns about membership changes from kind:44100/44101
notices and then re-reads its channel list. Three relay paths changed a
member's channel list without sending a notice, so clients only caught
up on their periodic full read:

- **Channel delete (9008):** no member was told.
- **Moderation kick:** the kicked member wasn't told, and the channel's
39000/39002 discovery wasn't refreshed, so it still listed them.
- **Moderation DM:** a new, existing or hidden moderation DM didn't
surface live.

## Change

- **Channel delete:** `soft_delete_channel_capturing_members` (buzz-db)
reads the channel type and active members under deletion's existing
membership lock, in the same transaction. A join either commits first
and is included, or waits and is then refused. After the commit,
discovery cleanup and cache invalidation, the relay sends a type-tagged
44101 to each captured member in a spawned task, eight at a time,
outside any lock. The 9008 OK doesn't wait for them, and the task is
started before the `channel_deleted` system message is emitted, so a
system-message failure can't drop them. An already-deleted channel sends
nothing. `soft_delete_channel` keeps its signature and delegates.
- **Moderation kick:** `apply_kick_live_side_effects` refreshes
39000/39002 after the removal fence commits, then sends 44101 to the
kicked member. If the refresh fails, it returns the error before the
notice, so the moderation driver's existing mutation-marker recovery
retries. If the channel was deleted since the kick, there's no roster to
refresh: it removes the channel's discovery events and treats that as
converged, so the action still finalizes. The removal/re-add fence and
the re-added early return are unchanged.
- **Moderation DM:** `send_moderation_notice` drops both participants'
cached channel sets on every pod (`invalidate_membership`), as opening a
DM does in `command_executor::handle_dm_open`, so the read the notice
prompts includes the DM. After clearing `hidden_at`, it republishes the
recipient's NIP-DV kind:30622 visibility snapshot (clients filter the
sidebar on it), then emits discovery, then sends 44100. It sends the
notice on every delivery, tagged `moderation_source` with the delivery's
source id, so two deliveries in the same second each get their own
44100. The content JSON is unchanged. A failed snapshot or discovery
returns an error, so the outbox retries.
- **Ordering guarantee:** each notice is sent after its change is
committed and readable on the relay instance that sends it. Ordering
across relay instances, and other instances' cached channel sets, are
best-effort, the same as for existing membership notices.
- A failed notice delivery is logged and never undoes the change, like
the existing 9001 path. There's no new retry system.

## Tests

- `membership_notice_postgres_tests.rs`: one journey per path. A
`#[cfg(test)]` barrier at the point a notice is signed parks the
producer, keyed by recipient. While it's parked, the test reads the
channel list (and, for the DM, the 30622 snapshot) as a client would on
receipt. The barrier can also pin a notice's `created_at`.
- The DM journey pins both deliveries to the same second, hides the DM
through the real kind:41012 path between them, and checks what a client
sees: a live 44100 for each delivery, with distinct ids, then a
channel-list read that lists the DM and a visibility read that no longer
hides it.
- `moderation_dm_notice_follows_a_cached_channel_list` warms the
recipient's channel set with a read before the DM exists, then checks
the read on receipt of the notice lists the DM.
- `soft_delete_captures_members_at_its_serialization_point` (buzz-db): a
join and then a delete queue behind a held membership lock. The delete
must return the joiner, and a later join is refused.
- `crash_recovery_finalizes_kick_after_channel_deleted` (admin): commits
a kick, deletes the channel, then runs the recovery worker. The action
finalizes, and the member's `removed_at` is unchanged, so the removal
doesn't run again.

Each ordering test fails when its prerequisite is moved or removed: the
kick notice ahead of the roster refresh, the delete notices ahead of
discovery cleanup, the DM without the snapshot republish or the cache
drop, the DM notice without its source tag (the second notice
coalesces), and the member capture outside the lock. The recovery test
fails if a deleted channel's `ChannelNotFound` propagates.

No spec files change. This reads a channel delete as removing every
member under the existing `NOSTR.md` notice promise.

---------

Signed-off-by: Will Pfleger <wpfleger@block.xyz>
Co-authored-by: Duncan <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@buzz.block.builderlab.xyz>

crates/buzz-db/src/store/channel.rs

@@ -1655,6 +1655,31 @@ pub async fn soft_delete_channel(
     community_id: CommunityId,
     channel_id: Uuid,
 ) -> Result<bool> {
+    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<Vec<u8>>,
+}
+
+/// [`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<Option<DeletedChannel>> {
     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<Vec<u8>> = 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<Option<DeletedChannel>> {
+        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<Vec<ReapedEphemeralChannel>> {
@@ -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() {

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<chrono::DateTime<chrono::Utc>>>(
+                "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.

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<AppState>, 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<AppState>,
+    tenant: &TenantContext,
+    keys: &Keys,
+    kind: u16,
+    tags: &[&[&str]],
+) {
+    let tags: Vec<Tag> = 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<AppState>,
+    tenant: &TenantContext,
+    reader: &Keys,
+) -> Receiver<axum::extract::ws::Message> {
+    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<axum::extract::ws::Message>) -> 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<axum::extract::ws::Message>) -> 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<axum::extract::ws::Message>) -> 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<String> {
+    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<AppState>,
+    tenant: &TenantContext,
+    reader: &Keys,
+    filter: Value,
+    name: &str,
+) -> Vec<String> {
+    let (conn, mut rx) = test_conn(tenant, reader);
+    let filters: Vec<nostr::Filter> = 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<AppState>,
+    tenant: &TenantContext,
+    reader: &Keys,
+) -> Vec<String> {
+    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<AppState>, tenant: &TenantContext, reader: &Keys) -> Vec<String> {
+    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<T>(
+    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::<String>::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);
+}

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).

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=<dm_channel_id>` 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()])?,

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::<buzz_db::DbError>(),
+            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<HashMap<Vec<u8>, mpsc::UnboundedSender<Parked>>>;
+
+    fn waiters() -> &'static Waiters {
+        static WAITERS: OnceLock<Waiters> = 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<u8>,
+        parked: mpsc::UnboundedReceiver<Parked>,
+    }
+
+    impl Barrier {
+        /// The next notice to reach the barrier, with its release handle.
+        pub(crate) async fn next(&mut self) -> Option<Parked> {
+            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<HashMap<Vec<u8>, u64>> {
+        static PINS: OnceLock<Mutex<HashMap<Vec<u8>, u64>>> = OnceLock::new();
+        PINS.get_or_init(Default::default)
+    }
+
+    pub(super) fn pinned_created_at(recipient: &[u8]) -> Option<u64> {
+        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<AppState>,
+    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<AppState>,
@@ -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,