buzz-db: purge retention-free deletions in the app and index mentions in the event transaction (#8128)

## Summary

Part 1a of removing database objects that exist only in migrations. This
PR moves into the app the work two migration-only triggers do today. It
doesn't drop anything.

On migration-built databases (production), `purge_soft_deleted_nip_rs`
(0009/0011) and `purge_soft_deleted_buzz_mesh_status` (0019) turn a
NIP-09 soft delete of a NIP-RS read-state event (kind 30078) or a
mesh-status heartbeat (kind 30003) into a hard delete, and delete its
mentions. `schema/schema.sql` has neither trigger, so on desired-state
databases (CI) those rows remain as soft-deleted. The two schemas behave
differently today.

Changes:
- `soft_delete_by_coordinate` and
`soft_delete_event_and_update_thread_in_tx` now hard-delete matching
NIP-RS and mesh-status rows and their `event_mentions` rows in the same
transaction, before the generic soft delete. The classification
predicate is the same one the 0011 and 0019 triggers use.
- `insert_event_with_serving_write_guard` now indexes mentions inside
the event transaction, under a savepoint, instead of after commit. A
mention failure still only logs a warning. A concurrent deletion can no
longer leave a mention row behind, and `guard_event_mention_live` will
no longer be needed. The other post-commit `insert_mentions` callers
write kinds 48101 and 13534 and a test helper, never 30078.
- Both soft-delete `UPDATE`s exclude the same predicate (`AND NOT
COALESCE(..., false)`). A retention-free row is either purged or left
live, never soft-deleted, even when a replacement races the deletion
under READ COMMITTED. The purge runs only for kinds 30078 and 30003.
- While the 0011 hard-delete fence is still installed, the purge sets
the transaction-local `buzz.nip_rs_hard_delete` opt-in. 1b removes it,
along with the fence.

**Rollout:** the triggers stay in place on purpose. Pods still running
N-1 during a deploy depend on them. A follow-up (1b) will add a
migration that drops the 5 functions and 5 triggers, remove the
`buzz.nip_rs_hard_delete` opt-in code, and add the catalog-parity test.
1b should merge only after this PR is deployed everywhere.

**What got simpler:** once 1b lands, NIP-RS and mesh deletion will
follow one rule owned by the app, the same on both schemas, and the
hidden trigger work will be gone. This PR adds about 130 lines to
`event.rs`. A known gap for 1b: the replace path's Rust mesh classifier
is stricter than the SQL predicate, and 1b has to make one of them
authoritative before the triggers are dropped. 1b will then delete 5
triggers and 5 functions, plus the opt-in plumbing in `replaceable.rs`
and `deletion.rs`.

### Related issue

BUZZ-175, BUZZ-177 (Linear).

### Testing

Blox, head `a2dac90646ca64fb3f961ffda292e3da976deffc`:
- `cargo fmt --check`, `clippy -D warnings` (`buzz-db`, `buzz-relay`),
and test discovery: all pass.
- New tests: 4 of 4 pass.
- `nip09_deletion_purges_retention_free_rows_and_mentions`,
desired-state and `migration_schema_` variants. On the migration schema,
a 13-shape table checks that the app and the triggers classify each
shape the same way. A separate case covers a mesh row with a NULL
`d_tag`.
-
`coordinate_deletion_racing_nip_rs_replacement_never_soft_deletes_read_state`,
a deterministic lock interleaving.
- `serving_write_guard_insert_indexes_mentions_in_event_transaction`:
mentions are indexed in the event's own transaction (checked via
`xmin`), and a mention failure leaves the event live.
- Falsification: the race test fails against the pre-fix `event.rs`. The
mention test fails with the merge-base post-commit `runtime/mod.rs`.
Both NULL-`d_tag` variants fail with `COALESCE` removed.
- `buzz-db` PostgreSQL lane (`scripts/postgres-test-run.sh -p buzz-db
--lib --tests`): **409 of 409 passed** in an unscoped run. One box hangs
on `writer_pool_rejects_non_read_committed_database_default` (an
environment issue; it also hangs there at merge-base), and there 408 of
408 others passed.
- E2E against a real relay on both schemas: 18 of 18 signed deletion
cases, the serving-write mention path (success, pre-commit invisibility,
isolated failure, recovery), and browser mention, send, reload and
delete.
- CI at `a2dac906`: all checks pass. The `PostgreSQL Tests` lane needed
one rerun: the first attempt hit a known flaky test,
`replica_fence::sample_writer_waits_for_shared_floor_lock_before_sampling_time`,
which fails the same way on `main` at `001290a34`.

buzz-review-completed (agent review and agent E2E complete; human test
waived by Luke Tornquist)

---------

Signed-off-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>

crates/buzz-db/src/runtime/mod.rs

@@ -13,7 +13,7 @@ use crate::{deletion, event, DbError, EventQuery, Result};
 use buzz_datastore_tracing::datastore_span;
 use chrono::{DateTime, Utc};
 use sqlx::postgres::PgPoolOptions;
-use sqlx::{PgPool, QueryBuilder};
+use sqlx::{Acquire, PgPool, QueryBuilder};
 use std::time::Duration;
 use uuid::Uuid;
 
@@ -1348,12 +1348,23 @@ impl Db {
             None,
         )
         .await?;
-        tx.commit().await?;
         if result.1 {
-            if let Err(e) = insert_mentions(&self.pool, community_id, event, channel_id).await {
-                tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
+            // Index mentions in the event's own transaction so a concurrent
+            // deletion cannot strand an orphan row. A savepoint keeps the
+            // existing best-effort contract: an indexing failure never rejects
+            // an otherwise valid event.
+            let mut savepoint = tx.begin().await?;
+            match insert_mentions_in_transaction(&mut savepoint, community_id, event, channel_id)
+                .await
+            {
+                Ok(()) => savepoint.commit().await?,
+                Err(e) => {
+                    savepoint.rollback().await?;
+                    tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
+                }
             }
         }
+        tx.commit().await?;
         Ok(result)
     }
 

crates/buzz-db/src/runtime/tests.rs

@@ -3811,3 +3811,159 @@ async fn e_tag_any_runs_with_real_binds_on_p_join_and_count() {
 
 #[path = "tests/thread_window_postgres_tests.rs"]
 mod thread_window_postgres_tests;
+
+/// `insert_event_with_serving_write_guard` indexes mentions in the event's own
+/// transaction, and an indexing failure stays best-effort: the savepoint rolls
+/// back only the mention rows, never the event.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn serving_write_guard_insert_indexes_mentions_in_event_transaction() {
+    use nostr::{EventBuilder, Keys, Kind, Tag};
+
+    let db = setup_db().await;
+    let community = db
+        .ensure_configured_community(&format!(
+            "serving-mentions-{}.example",
+            Uuid::new_v4().simple()
+        ))
+        .await
+        .expect("create serving-write mention community")
+        .id;
+    let store = db.deletion_store();
+    let lease = store
+        .acquire_serving_write_lease(
+            community,
+            "mention_index_test",
+            "mention-index-test",
+            std::time::Duration::from_secs(30),
+        )
+        .await
+        .expect("acquire serving-write lease");
+    let mentioned = Keys::generate().public_key().to_hex();
+    let build = |content: &str| {
+        EventBuilder::new(Kind::TextNote, content)
+            .tags([Tag::parse(["p", mentioned.as_str()]).expect("p tag")])
+            .sign_with_keys(&Keys::generate())
+            .expect("sign event")
+    };
+    let counts = |event_id: Vec<u8>| {
+        let pool = db.pool.clone();
+        async move {
+            let live: i64 = sqlx::query_scalar(
+                "SELECT count(*) FROM events \
+                 WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
+            )
+            .bind(community.as_uuid())
+            .bind(&event_id)
+            .fetch_one(&pool)
+            .await
+            .expect("count live events");
+            let mentions: i64 = sqlx::query_scalar(
+                "SELECT count(*) FROM event_mentions WHERE community_id = $1 AND event_id = $2",
+            )
+            .bind(community.as_uuid())
+            .bind(&event_id)
+            .fetch_one(&pool)
+            .await
+            .expect("count mentions");
+            (live, mentions)
+        }
+    };
+
+    // Pin in-transaction indexing: the mention insert must run in the same
+    // top-level transaction that inserted the event row. A post-commit index
+    // runs in a later transaction, raises here, is logged as a warning, and
+    // leaves zero mentions.
+    sqlx::query(sqlx::AssertSqlSafe(format!(
+        "CREATE FUNCTION test_require_event_tx_mentions() RETURNS trigger AS $$ \
+         BEGIN \
+             IF NEW.community_id = '{}'::uuid AND NOT EXISTS ( \
+                 SELECT 1 FROM events \
+                 WHERE community_id = NEW.community_id AND id = NEW.event_id \
+                   AND xmin = pg_current_xact_id()::xid) THEN \
+                 RAISE EXCEPTION 'test: post-commit mention'; \
+             END IF; \
+             RETURN NEW; \
+         END; $$ LANGUAGE plpgsql",
+        community.as_uuid()
+    )))
+    .execute(&db.pool)
+    .await
+    .expect("create same-transaction mention function");
+    sqlx::query(
+        "CREATE TRIGGER trg_test_require_event_tx_mentions BEFORE INSERT ON event_mentions \
+         FOR EACH ROW EXECUTE FUNCTION test_require_event_tx_mentions()",
+    )
+    .execute(&db.pool)
+    .await
+    .expect("install same-transaction mention trigger");
+
+    let indexed = build("indexed");
+    let result = db
+        .insert_event_with_serving_write_guard(&lease, &indexed, None)
+        .await;
+
+    sqlx::query("DROP TRIGGER trg_test_require_event_tx_mentions ON event_mentions")
+        .execute(&db.pool)
+        .await
+        .expect("drop same-transaction mention trigger");
+    sqlx::query("DROP FUNCTION test_require_event_tx_mentions()")
+        .execute(&db.pool)
+        .await
+        .expect("drop same-transaction mention function");
+
+    let (_, inserted) = result.expect("guarded insert");
+    assert!(inserted);
+    assert_eq!(
+        counts(indexed.id.as_bytes().to_vec()).await,
+        (1, 1),
+        "mentions must be indexed in the event's own transaction"
+    );
+
+    // Make mention indexing fail for this community only. Without the
+    // savepoint the aborted statement poisons the event transaction and the
+    // commit fails.
+    sqlx::query(sqlx::AssertSqlSafe(format!(
+        "CREATE FUNCTION test_reject_mentions() RETURNS trigger AS $$ \
+         BEGIN \
+             IF NEW.community_id = '{}'::uuid THEN \
+                 RAISE EXCEPTION 'test: mention indexing rejected'; \
+             END IF; \
+             RETURN NEW; \
+         END; $$ LANGUAGE plpgsql",
+        community.as_uuid()
+    )))
+    .execute(&db.pool)
+    .await
+    .expect("create mention rejection function");
+    sqlx::query(
+        "CREATE TRIGGER trg_test_reject_mentions BEFORE INSERT ON event_mentions \
+         FOR EACH ROW EXECUTE FUNCTION test_reject_mentions()",
+    )
+    .execute(&db.pool)
+    .await
+    .expect("install mention rejection trigger");
+
+    let rejected = build("mention indexing fails");
+    let result = db
+        .insert_event_with_serving_write_guard(&lease, &rejected, None)
+        .await;
+
+    sqlx::query("DROP TRIGGER trg_test_reject_mentions ON event_mentions")
+        .execute(&db.pool)
+        .await
+        .expect("drop mention rejection trigger");
+    sqlx::query("DROP FUNCTION test_reject_mentions()")
+        .execute(&db.pool)
+        .await
+        .expect("drop mention rejection function");
+
+    let (_, inserted) = result.expect("mention failure must not reject the event");
+    assert!(inserted);
+    assert_eq!(counts(rejected.id.as_bytes().to_vec()).await, (1, 0));
+
+    assert!(store
+        .release_serving_write_lease(&lease)
+        .await
+        .expect("release serving-write lease"));
+}

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

@@ -1089,6 +1089,37 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer
     Ok(cnt)
 }
 
+/// SQL predicate over an `events` row for coordinates that carry no
+/// historical value: NIP-RS read state and Buzz mesh heartbeats. Their
+/// deletion is physical, including the mention index, so soft-deleted payloads
+/// never accumulate. This is exactly the classification used by the migration
+/// 0011 / 0019 purge triggers this path replaces.
+macro_rules! retention_free_event_predicate {
+    () => {
+        "((kind = 30078 \
+           AND d_tag ~ '^read-state:[0-9a-f]{32}$' \
+           AND (SELECT count(*) \
+                FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \
+                WHERE jsonb_typeof(tag) = 'array' AND tag->0 = '\"d\"'::jsonb) = 1 \
+           AND EXISTS (SELECT 1 \
+                FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \
+                WHERE jsonb_typeof(tag) = 'array' AND jsonb_array_length(tag) >= 2 \
+                  AND jsonb_typeof(tag->1) = 'string' AND tag->>0 = 'd' AND tag->>1 = d_tag) \
+           AND (SELECT count(*) \
+                FROM jsonb_array_elements(CASE WHEN jsonb_typeof(tags) = 'array' THEN tags ELSE '[]'::jsonb END) tag \
+                WHERE tag = '[\"t\", \"read-state\"]'::jsonb) = 1) \
+          OR (kind = 30003 \
+              AND d_tag LIKE 'buzz-mesh-member-status:%' \
+              AND tags @> '[[\"k\", \"buzz-mesh-status\"]]'::jsonb))"
+    };
+}
+
+/// Kinds that can hold a retention-free coordinate. Cheap Rust gate so
+/// ordinary deletions skip the purge statement entirely.
+fn may_be_retention_free(kind: i32) -> bool {
+    kind == 30078 || kind == 30003
+}
+
 /// Soft-delete the live row for an addressable coordinate
 /// `(kind, pubkey, d_tag)` — the NIP-33 replacement key — provided it is not
 /// newer than the deletion request.
@@ -1134,11 +1165,34 @@ pub async fn soft_delete_by_coordinate(
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
-    let result = sqlx::query(
+    let purged = if may_be_retention_free(kind) {
+        purge_retention_free_events(
+            &mut tx,
+            community_id,
+            RetentionFreeTarget::Coordinate {
+                kind,
+                pubkey,
+                d_tag,
+                created_at_or_before: deletion_created_at,
+            },
+        )
+        .await?
+    } else {
+        0
+    };
+    // `AND NOT COALESCE(<predicate>, false)` holds the invariant in SQL rather than through
+    // statement order: under READ COMMITTED this UPDATE takes a fresh snapshot,
+    // so a retention-free head committed by a racing replacement after the
+    // purge ran is spared instead of soft-deleted (the "deletion arrived
+    // first" outcome documented above). COALESCE keeps a NULL predicate
+    // (e.g. NULL `d_tag`) on the ordinary soft-delete path, as the triggers do.
+    let result = sqlx::query(concat!(
         "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 \
-         AND created_at <= $5",
-    )
+         AND created_at <= $5 AND NOT COALESCE(",
+        retention_free_event_predicate!(),
+        ", false)"
+    ))
     .bind(community_id.as_uuid())
     .bind(kind)
     .bind(pubkey)
@@ -1149,7 +1203,86 @@ pub async fn soft_delete_by_coordinate(
 
     tx.commit().await?;
 
-    Ok(result.rows_affected() > 0)
+    Ok(purged > 0 || result.rows_affected() > 0)
+}
+
+/// Purge live retention-free rows matching `$target`, plus their mention rows,
+/// and return the number of events removed.
+macro_rules! purge_retention_free_sql {
+    ($target:literal) => {
+        concat!(
+            "WITH purged AS ( \
+                 DELETE FROM events \
+                 WHERE community_id = $1 AND ",
+            $target,
+            " AND deleted_at IS NULL AND ",
+            retention_free_event_predicate!(),
+            " RETURNING id \
+             ), purged_mentions AS ( \
+                 DELETE FROM event_mentions mention USING purged \
+                 WHERE mention.community_id = $1 AND mention.event_id = purged.id \
+             ) \
+             SELECT count(*) FROM purged"
+        )
+    };
+}
+
+/// Which live rows a deletion targets.
+enum RetentionFreeTarget<'a> {
+    Id(&'a [u8]),
+    Coordinate {
+        kind: i32,
+        pubkey: &'a [u8],
+        d_tag: &'a str,
+        created_at_or_before: DateTime<Utc>,
+    },
+}
+
+/// Physically delete live retention-free rows matched by `target`, plus their
+/// mention rows, in the caller's transaction. The generic soft delete that
+/// follows excludes the same predicate, so a retention-free row is either
+/// purged here or left live, never soft-deleted. Returns the number of events
+/// removed.
+async fn purge_retention_free_events(
+    tx: &mut PgConnection,
+    community_id: CommunityId,
+    target: RetentionFreeTarget<'_>,
+) -> Result<u64> {
+    // Migration 0011 fences NIP-RS hard deletes behind a transaction-local
+    // opt-in. The fence and this opt-in are removed together once the
+    // migration-only triggers are dropped.
+    sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
+        .execute(&mut *tx)
+        .await?;
+
+    let purged: i64 = match target {
+        RetentionFreeTarget::Id(event_id) => {
+            sqlx::query_scalar(purge_retention_free_sql!("id = $2"))
+                .bind(community_id.as_uuid())
+                .bind(event_id)
+                .fetch_one(&mut *tx)
+                .await?
+        }
+        RetentionFreeTarget::Coordinate {
+            kind,
+            pubkey,
+            d_tag,
+            created_at_or_before,
+        } => {
+            sqlx::query_scalar(purge_retention_free_sql!(
+                "kind = $2 AND pubkey = $3 AND d_tag = $4 AND created_at <= $5"
+            ))
+            .bind(community_id.as_uuid())
+            .bind(kind)
+            .bind(pubkey)
+            .bind(d_tag)
+            .bind(created_at_or_before)
+            .fetch_one(&mut *tx)
+            .await?
+        }
+    };
+
+    Ok(purged as u64)
 }
 
 /// Atomically soft-delete an event and decrement thread reply counters.
@@ -1236,15 +1369,24 @@ pub(crate) async fn soft_delete_event_and_update_thread_in_tx(
         }
     }
 
-    let result = sqlx::query(
-        "UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL",
-    )
+    let purged = if target.is_some_and(|(kind, _)| may_be_retention_free(kind)) {
+        purge_retention_free_events(&mut *tx, community_id, RetentionFreeTarget::Id(event_id))
+            .await?
+    } else {
+        0
+    };
+    let result = sqlx::query(concat!(
+        "UPDATE events SET deleted_at = NOW() \
+         WHERE community_id = $1 AND id = $2 AND deleted_at IS NULL AND NOT COALESCE(",
+        retention_free_event_predicate!(),
+        ", false)"
+    ))
     .bind(community_id.as_uuid())
     .bind(event_id)
     .execute(&mut *tx)
     .await?;
 
-    let deleted = result.rows_affected() > 0;
+    let deleted = purged > 0 || result.rows_affected() > 0;
 
     if deleted {
         if let Some(pid) = parent_event_id {

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

@@ -1871,4 +1871,484 @@ mod postgres_tests {
             assert_ne!(leaked.as_deref(), Some("on"));
         }
     }
+
+    /// NIP-09 deletion of a retention-free coordinate (NIP-RS read state, mesh
+    /// status) is physical in application code, so it holds on desired-state
+    /// databases that never installed the migration 0009/0019 purge triggers.
+    /// Ordinary addressable events, and look-alikes that fail classification,
+    /// keep the soft-delete contract.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn nip09_deletion_purges_retention_free_rows_and_mentions() {
+        assert_nip09_deletion_purges_retention_free_rows_and_mentions(false).await;
+    }
+
+    /// The same contract on a migration-built database, where the 0009/0011/
+    /// 0019 purge and hard-delete fence triggers are still installed. Each row
+    /// shape is also soft-deleted the way a pre-0009 writer would, so the
+    /// table proves the app classification agrees with the triggers.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn migration_schema_nip09_deletion_purges_retention_free_rows_and_mentions() {
+        assert_nip09_deletion_purges_retention_free_rows_and_mentions(true).await;
+    }
+
+    async fn assert_nip09_deletion_purges_retention_free_rows_and_mentions(migration: bool) {
+        use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
+
+        #[derive(Clone, Copy, Debug)]
+        enum Delete {
+            Coordinate,
+            Id,
+        }
+
+        let db = setup_db().await;
+        let triggers_installed: bool = sqlx::query_scalar(
+            "SELECT EXISTS(SELECT 1 FROM pg_trigger \
+             WHERE tgname = 'trg_events_purge_soft_deleted_nip_rs')",
+        )
+        .fetch_one(&db.pool)
+        .await
+        .expect("inspect purge trigger");
+        assert_eq!(
+            triggers_installed, migration,
+            "schema mode must match the test variant (run through the Postgres wrapper)"
+        );
+        let community = CommunityId::from_uuid(make_community(&db.pool).await);
+        let nip_rs = buzz_core::kind::KIND_READ_STATE as u16;
+        let mesh = buzz_core::kind::KIND_BOOKMARK_SET as u16;
+        let read_state_d = format!("read-state:{}", "b".repeat(32));
+        let mesh_d = "buzz-mesh-member-status:purge-test".to_string();
+        let tag = |parts: &[&str]| parts.iter().map(|p| p.to_string()).collect::<Vec<_>>();
+        let t_read_state = tag(&["t", "read-state"]);
+        let k_mesh = tag(&["k", "buzz-mesh-status"]);
+        // (label, kind, column d_tag, tags, physically purged)
+        type Case = (&'static str, u16, String, Vec<Vec<String>>, bool);
+        let cases: Vec<Case> = vec![
+            (
+                "nip-rs",
+                nip_rs,
+                read_state_d.clone(),
+                vec![tag(&["d", &read_state_d]), t_read_state.clone()],
+                true,
+            ),
+            (
+                "nip-rs without t tag",
+                nip_rs,
+                read_state_d.clone(),
+                vec![tag(&["d", &read_state_d])],
+                false,
+            ),
+            (
+                "nip-rs uppercase hex slot",
+                nip_rs,
+                format!("read-state:{}", "B".repeat(32)),
+                vec![
+                    tag(&["d", &format!("read-state:{}", "B".repeat(32))]),
+                    t_read_state.clone(),
+                ],
+                false,
+            ),
+            (
+                "nip-rs 31-char slot",
+                nip_rs,
+                format!("read-state:{}", "b".repeat(31)),
+                vec![
+                    tag(&["d", &format!("read-state:{}", "b".repeat(31))]),
+                    t_read_state.clone(),
+                ],
+                false,
+            ),
+            (
+                "nip-rs duplicate d tag",
+                nip_rs,
+                read_state_d.clone(),
+                vec![
+                    tag(&["d", &read_state_d]),
+                    tag(&["d", &read_state_d]),
+                    t_read_state.clone(),
+                ],
+                false,
+            ),
+            (
+                "nip-rs duplicate t tag",
+                nip_rs,
+                read_state_d.clone(),
+                vec![
+                    tag(&["d", &read_state_d]),
+                    t_read_state.clone(),
+                    t_read_state.clone(),
+                ],
+                false,
+            ),
+            (
+                "nip-rs d tag differs from column",
+                nip_rs,
+                read_state_d.clone(),
+                vec![
+                    tag(&["d", &format!("read-state:{}", "c".repeat(32))]),
+                    t_read_state.clone(),
+                ],
+                false,
+            ),
+            (
+                "mesh-status",
+                mesh,
+                mesh_d.clone(),
+                vec![tag(&["d", &mesh_d]), k_mesh.clone()],
+                true,
+            ),
+            (
+                // Containment semantics (`tags @>`), shared by trigger and app.
+                "mesh-status k tag with extra element",
+                mesh,
+                mesh_d.clone(),
+                vec![
+                    tag(&["d", &mesh_d]),
+                    tag(&["k", "buzz-mesh-status", "extra"]),
+                ],
+                true,
+            ),
+            (
+                "mesh-status wrong d prefix",
+                mesh,
+                "buzz-mesh-member:purge-test".into(),
+                vec![tag(&["d", "buzz-mesh-member:purge-test"]), k_mesh.clone()],
+                false,
+            ),
+            (
+                "mesh-status without k tag",
+                mesh,
+                mesh_d.clone(),
+                vec![tag(&["d", &mesh_d])],
+                false,
+            ),
+            (
+                "mesh-status wrong k value",
+                mesh,
+                mesh_d.clone(),
+                vec![tag(&["d", &mesh_d]), tag(&["k", "buzz-mesh-other"])],
+                false,
+            ),
+            (
+                "ordinary addressable",
+                buzz_core::kind::KIND_PROJECT as u16,
+                "retained-project".into(),
+                vec![tag(&["d", "retained-project"])],
+                false,
+            ),
+        ];
+
+        // Store one event of this shape with one mention; return
+        // (keys, id, created_at secs).
+        let store = |label: &'static str, kind: u16, d_tag: String, tags: Vec<Vec<String>>| {
+            let db = &db;
+            async move {
+                let keys = Keys::generate();
+                let created_at = Timestamp::now().as_secs();
+                let event = EventBuilder::new(Kind::Custom(kind), label)
+                    .tags(
+                        tags.iter()
+                            .map(|tag| Tag::parse(tag.clone()).expect("tag"))
+                            .collect::<Vec<_>>(),
+                    )
+                    .custom_created_at(Timestamp::from(created_at))
+                    .sign_with_keys(&keys)
+                    .expect("sign event");
+                assert!(
+                    db.replace_parameterized_event(community, &event, &d_tag, None)
+                        .await
+                        .expect("store event")
+                        .1,
+                    "{label}: event must be stored"
+                );
+                let id = event.id.as_bytes().to_vec();
+                sqlx::query(
+                    "INSERT INTO event_mentions \
+                     (community_id, pubkey_hex, event_id, event_created_at, channel_id, event_kind) \
+                     SELECT community_id, $3, id, created_at, NULL, kind \
+                     FROM events WHERE community_id = $1 AND id = $2",
+                )
+                .bind(community.as_uuid())
+                .bind(&id)
+                .bind("c".repeat(64))
+                .execute(&db.pool)
+                .await
+                .expect("index mention");
+                (keys, id, created_at)
+            }
+        };
+        // (event rows, live event rows, mention rows) for one id.
+        let state = |id: Vec<u8>| {
+            let pool = db.pool.clone();
+            async move {
+                let (rows, live): (i64, i64) = sqlx::query_as(
+                    "SELECT count(*), count(*) FILTER (WHERE deleted_at IS NULL) \
+                     FROM events WHERE community_id=$1 AND id=$2",
+                )
+                .bind(community.as_uuid())
+                .bind(&id)
+                .fetch_one(&pool)
+                .await
+                .expect("count event rows");
+                let mentions: i64 = sqlx::query_scalar(
+                    "SELECT count(*) FROM event_mentions WHERE community_id=$1 AND event_id=$2",
+                )
+                .bind(community.as_uuid())
+                .bind(&id)
+                .fetch_one(&pool)
+                .await
+                .expect("count mentions");
+                (rows, live, mentions)
+            }
+        };
+        let expected = |purged: bool| -> (i64, i64, i64) {
+            if purged {
+                (0, 0, 0)
+            } else {
+                (1, 0, 1)
+            }
+        };
+
+        for (label, kind, d_tag, tags, purged) in cases {
+            if migration {
+                // Trigger half of the parity table: a pre-0009 writer's plain
+                // soft delete, classified only by the 0009/0011/0019 triggers.
+                let (_, id, _) = store(label, kind, d_tag.clone(), tags.clone()).await;
+                sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id=$1 AND id=$2")
+                    .bind(community.as_uuid())
+                    .bind(&id)
+                    .execute(&db.pool)
+                    .await
+                    .expect("legacy soft delete");
+                assert_eq!(
+                    state(id).await,
+                    expected(purged),
+                    "{label}: trigger classification"
+                );
+            }
+
+            for mode in [Delete::Coordinate, Delete::Id] {
+                let (keys, id, created_at) = store(label, kind, d_tag.clone(), tags.clone()).await;
+                let delete = || async {
+                    match mode {
+                        Delete::Coordinate => event::soft_delete_by_coordinate(
+                            &db.pool,
+                            community,
+                            i32::from(kind),
+                            &keys.public_key().to_bytes(),
+                            &d_tag,
+                            (created_at + 1) as i64,
+                        )
+                        .await
+                        .expect("coordinate delete"),
+                        Delete::Id => event::soft_delete_event_and_update_thread(
+                            &db.pool, community, &id, None, None,
+                        )
+                        .await
+                        .expect("id delete"),
+                    }
+                };
+                assert!(
+                    delete().await,
+                    "{label}/{mode:?}: deletion must report success"
+                );
+                assert_eq!(
+                    state(id.clone()).await,
+                    expected(purged),
+                    "{label}/{mode:?}: app classification"
+                );
+                assert!(
+                    !delete().await,
+                    "{label}/{mode:?}: repeated deletion is a no-op"
+                );
+            }
+        }
+
+        // A mesh-status-shaped row whose `d_tag` column was never backfilled:
+        // `kind = 30003 AND NULL LIKE … AND tags @> …` makes the purge
+        // predicate NULL (a read-state row cannot: its tag/column EXISTS is
+        // false), which the triggers treat as false. The id path must still
+        // soft-delete it (the `COALESCE` in the generic soft delete), not
+        // silently skip it. Coordinate deletes key on `d_tag`, so only the id
+        // path can reach this row.
+        let null_d_tag_row = || async {
+            let (_, id, _) = store(
+                "mesh-status with NULL d_tag column",
+                mesh,
+                mesh_d.clone(),
+                vec![tag(&["d", &mesh_d]), k_mesh.clone()],
+            )
+            .await;
+            sqlx::query("UPDATE events SET d_tag = NULL WHERE community_id=$1 AND id=$2")
+                .bind(community.as_uuid())
+                .bind(&id)
+                .execute(&db.pool)
+                .await
+                .expect("clear d_tag column");
+            id
+        };
+        if migration {
+            let id = null_d_tag_row().await;
+            sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id=$1 AND id=$2")
+                .bind(community.as_uuid())
+                .bind(&id)
+                .execute(&db.pool)
+                .await
+                .expect("legacy soft delete");
+            assert_eq!(
+                state(id).await,
+                expected(false),
+                "NULL d_tag: trigger classification"
+            );
+        }
+        let id = null_d_tag_row().await;
+        assert!(
+            event::soft_delete_event_and_update_thread(&db.pool, community, &id, None, None)
+                .await
+                .expect("id delete"),
+            "NULL d_tag/Id: deletion must report success"
+        );
+        assert_eq!(
+            state(id).await,
+            expected(false),
+            "NULL d_tag/Id: app classification"
+        );
+    }
+
+    /// A coordinate deletion racing a NIP-RS replacement must never leave a
+    /// soft-deleted read-state row behind. The purge waits on the old head's
+    /// row lock; the replacement commits a newer head that the purge's snapshot
+    /// cannot see; the generic soft delete then runs on a fresh snapshot and
+    /// must spare that head rather than soft-delete it.
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn coordinate_deletion_racing_nip_rs_replacement_never_soft_deletes_read_state() {
+        use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
+        use sqlx::postgres::PgConnectOptions;
+        use std::str::FromStr;
+
+        let db = setup_db().await;
+        let community = CommunityId::from_uuid(make_community(&db.pool).await);
+        let keys = Keys::generate();
+        let d_tag = format!("read-state:{}", "d".repeat(32));
+        let created_at = Timestamp::now().as_secs();
+        let old_head = EventBuilder::new(
+            Kind::Custom(buzz_core::kind::KIND_READ_STATE as u16),
+            "old head",
+        )
+        .tags([
+            Tag::parse(["d", d_tag.as_str()]).expect("d tag"),
+            Tag::parse(["t", "read-state"]).expect("t tag"),
+        ])
+        .custom_created_at(Timestamp::from(created_at))
+        .sign_with_keys(&keys)
+        .expect("sign old head");
+        assert!(
+            db.replace_parameterized_event(community, &old_head, &d_tag, None)
+                .await
+                .expect("store old head")
+                .1
+        );
+        let old_id = old_head.id.as_bytes().to_vec();
+        let new_id = vec![0x5a_u8; 32];
+
+        // The replacement: insert a newer head (still at or before the
+        // tombstone), hard-delete the old head, and hold the transaction open.
+        let mut replacement = db.pool.begin().await.expect("begin replacement");
+        sqlx::query("SELECT set_config('buzz.nip_rs_hard_delete', 'on', true)")
+            .execute(&mut *replacement)
+            .await
+            .expect("opt in to hard delete");
+        sqlx::query(
+            "INSERT INTO events \
+             (community_id, id, pubkey, created_at, kind, tags, content, sig, channel_id, d_tag) \
+             SELECT community_id, $3, pubkey, created_at + interval '1 second', kind, tags, \
+                    'new head', sig, channel_id, d_tag \
+             FROM events WHERE community_id = $1 AND id = $2",
+        )
+        .bind(community.as_uuid())
+        .bind(&old_id)
+        .bind(&new_id)
+        .execute(&mut *replacement)
+        .await
+        .expect("insert new head");
+        sqlx::query("DELETE FROM events WHERE community_id = $1 AND id = $2")
+            .bind(community.as_uuid())
+            .bind(&old_id)
+            .execute(&mut *replacement)
+            .await
+            .expect("delete old head");
+
+        let database_url =
+            std::env::var("TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.into());
+        let application_name = format!("nip-rs-race-{}", Uuid::new_v4().simple());
+        let deleter_pool = PgPoolOptions::new()
+            .max_connections(1)
+            .connect_with(
+                PgConnectOptions::from_str(&database_url)
+                    .expect("parse test database URL")
+                    .application_name(&application_name),
+            )
+            .await
+            .expect("connect deleter pool");
+        let pubkey = keys.public_key().to_bytes();
+        let deletion = tokio::spawn({
+            let d_tag = d_tag.clone();
+            async move {
+                event::soft_delete_by_coordinate(
+                    &deleter_pool,
+                    community,
+                    buzz_core::kind::KIND_READ_STATE as i32,
+                    &pubkey,
+                    &d_tag,
+                    (created_at + 5) as i64,
+                )
+                .await
+            }
+        });
+
+        tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                let waiting: bool = sqlx::query_scalar(
+                    "SELECT EXISTS(SELECT 1 FROM pg_stat_activity \
+                     WHERE datname = current_database() AND application_name = $1 \
+                       AND wait_event_type = 'Lock')",
+                )
+                .bind(&application_name)
+                .fetch_one(&db.pool)
+                .await
+                .expect("inspect deleter lock");
+                if waiting {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("deletion reached the old head's row lock");
+
+        replacement.commit().await.expect("commit replacement");
+        deletion
+            .await
+            .expect("join deletion")
+            .expect("coordinate delete");
+
+        let rows: Vec<(Vec<u8>, bool)> = sqlx::query_as(
+            "SELECT id, deleted_at IS NOT NULL FROM events \
+             WHERE community_id = $1 AND kind = $2 AND pubkey = $3 AND d_tag = $4",
+        )
+        .bind(community.as_uuid())
+        .bind(buzz_core::kind::KIND_READ_STATE as i32)
+        .bind(keys.public_key().to_bytes().as_slice())
+        .bind(&d_tag)
+        .fetch_all(&db.pool)
+        .await
+        .expect("read coordinate rows");
+        assert_eq!(
+            rows,
+            vec![(new_id, false)],
+            "the racing head is spared and no soft-deleted read-state row remains"
+        );
+    }
 }