fix(db): drop a superseded head's mention rows with it (#8948)

## Summary

Replacing a replaceable or addressable event soft-deletes the previous
head, but its `event_mentions` rows stayed behind. Only the hard-delete
kinds (read-state and mesh status) removed theirs, in a separate
statement. No mention-index read (`#p` filters and feeds) returns a
soft-deleted event, so those rows are pure cost: every `#p` read walks
them and throws them away.

A kind:39002 roster republish leaves one dead row per channel member, so
a large channel piles them up with every join. Relay-emitted replaceable
lists with p-tags do the same.

The statement that retires the old head now deletes that head's mention
rows as well, through a data-modifying CTE over `RETURNING id`, on all
three supersede paths:
- the relay's kind:39002 roster writer
(`LockedMemberSnapshot::replace_member_event`)
- `replace_addressable_event` (kind 0/3/10000–19999, including the
relay-signed kind:13535 list)
- the NIP-33 parameterized path

Because the CTE covers every row the retire statement touches, a
coordinate that somehow had more than one live row also gets cleaned up.
The shared `superseded_with_mentions_sql!` macro sits at the crate root
so all three mention-indexing replacement paths use the same statement.
The kind:13534 membership-list and push-lease writers also retire heads,
with a bare `UPDATE`, but neither indexes p-tags, so they leave no
mention rows.

## What got simpler

The hard-delete special case (`if hard_delete_superseded { DELETE FROM
event_mentions … }`) is gone. Both retire statements go through one
macro, so a head and its mention rows leave together on all three paths.

## Not in this PR

- Mention rows already left behind by earlier supersedes.
`docs/database.md` rules out data backfills in migrations, so these get
cleaned up out of band, in batches.
- Other soft-delete paths (NIP-09 deletion, moderation, channel deletion
retiring its 39000–39002 events) also keep mention rows. They are
low-volume, and are noted here as identified debt.

## Tests

`superseded_heads_drop_their_mention_rows` (pg) has one table row per
supersede path: the member-snapshot roster writer (lock, replace,
release), the addressable path, the NIP-33 path, and the two NIP-33
kinds that are hard-deleted (NIP-RS read state and mesh status). In each
case the old head keeps zero mention rows and the new head is indexed.
Soft-delete rows check that the old head stays soft-deleted, and
hard-delete rows check that it is gone. Reverting the macro, any single
call site, or the hard-delete statement alone fails this test.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Luke Tornquist <tornquist@squareup.com>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

crates/buzz-db/src/lib.rs

@@ -21,6 +21,28 @@
 //! Existing crate-root modules, records, and [`Db`] methods remain the public
 //! API. The internal `runtime` and `store` namespaces are not public APIs.
 
+/// Wrap `$retire`, an `UPDATE`/`DELETE` on `events` that retires a
+/// coordinate's previous head, so the same statement drops those rows' mention
+/// index entries. Mention-index reads (`#p` filters and feeds) never return a
+/// retired row, so its mention rows are dead weight they must walk and
+/// discard: one per member for every kind:39002 roster republish. `$1` must
+/// bind the community.
+///
+/// The three replacement paths that index mentions (NIP-33 parameterized,
+/// addressable, and the relay's member snapshot) go through this, so a head
+/// and its mention rows leave together on those paths.
+macro_rules! superseded_with_mentions_sql {
+    ($retire:literal) => {
+        concat!(
+            "WITH superseded AS (",
+            $retire,
+            " RETURNING id) \
+             DELETE FROM event_mentions mention USING superseded \
+             WHERE mention.community_id = $1 AND mention.event_id = superseded.id"
+        )
+    };
+}
+
 mod runtime;
 mod store;
 

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

@@ -500,7 +500,9 @@ impl LockedMemberSnapshot {
         }
         let tags = serde_json::to_value(&event.tags)?;
         validate_channel_roster_snapshot(&mut self.tx, channel_id, &tags).await?;
-        sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id = $4 AND deleted_at IS NULL")
+        sqlx::query(superseded_with_mentions_sql!(
+            "UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND channel_id = $4 AND deleted_at IS NULL"
+        ))
             .bind(community_id.as_uuid()).bind(kind).bind(pubkey.as_slice()).bind(channel_id)
             .execute(self.tx.conn()).await?;
         let received_at = Utc::now();

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

@@ -278,11 +278,15 @@ async fn replace_parameterized_event_in_transaction_impl(
                     .await?;
             }
             let statement = if hard_delete_superseded {
-                "DELETE FROM events \
-                 WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL"
+                superseded_with_mentions_sql!(
+                    "DELETE FROM events \
+                     WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL"
+                )
             } else {
-                "UPDATE events SET deleted_at = NOW() \
-                 WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL"
+                superseded_with_mentions_sql!(
+                    "UPDATE events SET deleted_at = NOW() \
+                     WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4 AND deleted_at IS NULL"
+                )
             };
             sqlx::query(statement)
                 .bind(community_id.as_uuid())
@@ -299,16 +303,6 @@ async fn replace_parameterized_event_in_transaction_impl(
                     .execute(tx.conn())
                     .await?;
             }
-
-            if hard_delete_superseded {
-                if let Some((_, existing_id)) = &existing {
-                    sqlx::query("DELETE FROM event_mentions WHERE community_id = $1 AND event_id = $2")
-                        .bind(community_id.as_uuid())
-                        .bind(existing_id)
-                        .execute(tx.conn())
-                        .await?;
-                }
-            }
         }
 
         let sig_bytes = event.sig.serialize();
@@ -494,12 +488,12 @@ impl Db {
                 .await?;
 
                 // Soft-delete the old event (if any). IS NOT DISTINCT FROM for NULL safety.
-                sqlx::query(
+                sqlx::query(superseded_with_mentions_sql!(
                     "UPDATE events SET deleted_at = NOW() \
                      WHERE community_id = $1 AND kind = $2 AND pubkey = $3 \
                      AND channel_id IS NOT DISTINCT FROM $4 \
-                     AND deleted_at IS NULL",
-                )
+                     AND deleted_at IS NULL"
+                ))
                 .bind(community_id.as_uuid())
                 .bind(kind_i32)
                 .bind(pubkey_bytes.as_slice())
@@ -784,6 +778,197 @@ mod postgres_tests {
         assert_eq!(new_rows, 0, "new roster must roll back with its index");
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn superseded_heads_drop_their_mention_rows() {
+        use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
+
+        let (pool, _scratch) = ScratchDb::migrated(&admin_url().await, "superseded_mentions").await;
+        let db = Db::from_pool(pool.clone());
+        let community_uuid = Uuid::new_v4();
+        let channel = Uuid::new_v4();
+        // The signer must differ from every p-tagged key: the builder drops
+        // self-references.
+        let keys = Keys::generate();
+        let owner_keys = Keys::generate();
+        seed_community_channel(&pool, community_uuid, channel, &owner_keys).await;
+        let community = CommunityId::from_uuid(community_uuid);
+        let owner = owner_keys.public_key().to_hex();
+        let mentioned: Vec<String> = (0..3)
+            .map(|_| Keys::generate().public_key().to_hex())
+            .collect();
+        let mention_rows = |event: &nostr::Event| {
+            let id = event.id.as_bytes().to_vec();
+            let pool = pool.clone();
+            async move {
+                sqlx::query_scalar::<_, i64>(
+                    "SELECT count(*) FROM event_mentions WHERE community_id = $1 AND event_id = $2",
+                )
+                .bind(community_uuid)
+                .bind(id)
+                .fetch_one(&pool)
+                .await
+                .expect("count mention rows")
+            }
+        };
+        let base = Timestamp::now().as_secs();
+
+        // Every supersede path retires the old head with a soft delete, except
+        // NIP-RS read state and mesh status, which the NIP-33 path hard-deletes.
+        // The relay roster writer is the production kind:39002 path; its p tags
+        // must match the channel's members. The relay-signed kind:13535 list
+        // goes through the addressable path.
+        enum Path {
+            MemberSnapshot,
+            Addressable,
+            Parameterized,
+        }
+        let roster_p = vec![vec![owner.as_str(), "", "owner"]];
+        let people_p =
+            || -> Vec<Vec<&str>> { mentioned.iter().map(|pk| vec![pk.as_str()]).collect() };
+        let relay_pubkey = keys.public_key().to_bytes();
+        for (case, path, kind, d_tag, extra_tags, p_tags, hard_delete) in [
+            (
+                "member snapshot",
+                Path::MemberSnapshot,
+                39002_u16,
+                Some(channel.to_string()),
+                vec![],
+                roster_p,
+                false,
+            ),
+            (
+                "addressable",
+                Path::Addressable,
+                13535_u16,
+                None,
+                vec![],
+                people_p(),
+                false,
+            ),
+            (
+                "parameterized",
+                Path::Parameterized,
+                30000_u16,
+                Some("people".to_owned()),
+                vec![],
+                people_p(),
+                false,
+            ),
+            (
+                "NIP-RS read state",
+                Path::Parameterized,
+                buzz_core::kind::KIND_READ_STATE as u16,
+                Some(format!("read-state:{}", "a".repeat(32))),
+                vec![["t", "read-state"]],
+                people_p(),
+                true,
+            ),
+            (
+                "mesh status",
+                Path::Parameterized,
+                buzz_core::kind::KIND_BOOKMARK_SET as u16,
+                Some("buzz-mesh-member-status:owner-test".to_owned()),
+                vec![["k", "buzz-mesh-status"]],
+                people_p(),
+                true,
+            ),
+        ] {
+            let indexed = p_tags.len() as i64;
+            let version = |offset: u64, content: &str| {
+                let mut tags: Vec<Tag> = d_tag
+                    .iter()
+                    .map(|d| Tag::parse(["d", d.as_str()]).expect("d tag"))
+                    .collect();
+                tags.extend(
+                    extra_tags
+                        .iter()
+                        .map(|t| Tag::parse(*t).expect("extra tag")),
+                );
+                tags.extend(p_tags.iter().map(|p| {
+                    let mut parts = vec!["p"];
+                    parts.extend(p.iter().copied());
+                    Tag::parse(parts).expect("p tag")
+                }));
+                EventBuilder::new(Kind::Custom(kind), content)
+                    .tags(tags)
+                    .custom_created_at(Timestamp::from(base + offset))
+                    .sign_with_keys(&keys)
+                    .expect("sign version")
+            };
+            let replace = |event: nostr::Event| {
+                let db = &db;
+                let d_tag = d_tag.clone();
+                let path = &path;
+                async move {
+                    let inserted = match path {
+                        Path::MemberSnapshot => {
+                            let mut snapshot = db
+                                .lock_member_snapshot(community, channel, &relay_pubkey)
+                                .await
+                                .expect("lock member snapshot");
+                            let inserted = snapshot
+                                .replace_member_event(&event)
+                                .await
+                                .expect("replace roster")
+                                .1;
+                            snapshot.release().await.expect("commit roster");
+                            inserted
+                        }
+                        Path::Addressable => {
+                            db.replace_addressable_event(community, &event, None)
+                                .await
+                                .expect("replace addressable")
+                                .1
+                        }
+                        Path::Parameterized => {
+                            let d_tag = d_tag.expect("parameterized d tag");
+                            db.replace_parameterized_event(community, &event, &d_tag, None)
+                                .await
+                                .expect("replace parameterized")
+                                .1
+                        }
+                    };
+                    assert!(inserted, "{case}: newer version must be inserted");
+                }
+            };
+
+            let old = version(0, "old");
+            replace(old.clone()).await;
+            assert_eq!(mention_rows(&old).await, indexed, "{case}: head is indexed");
+
+            let new = version(1, "new");
+            replace(new.clone()).await;
+            assert_eq!(
+                mention_rows(&old).await,
+                0,
+                "{case}: retired head keeps no mention rows"
+            );
+            assert_eq!(
+                mention_rows(&new).await,
+                indexed,
+                "{case}: new head is indexed"
+            );
+            let retired: Option<bool> = sqlx::query_scalar(
+                "SELECT deleted_at IS NOT NULL FROM events WHERE community_id = $1 AND id = $2",
+            )
+            .bind(community_uuid)
+            .bind(old.id.as_bytes().as_slice())
+            .fetch_optional(&pool)
+            .await
+            .expect("read retired head");
+            if hard_delete {
+                assert_eq!(retired, None, "{case}: the old head is removed");
+            } else {
+                assert_eq!(
+                    retired,
+                    Some(true),
+                    "{case}: the old head is soft-deleted, not removed"
+                );
+            }
+        }
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn stale_legacy_roster_cannot_replace_new_locked_snapshot() {