feat(nip-fi): wire deny-map check into WS connection admission

TESTING.md

@@ -358,7 +358,7 @@ CLI-side, only two matter for testing:
 | Symptom | Cause | Fix |
 |---------|-------|-----|
 | `relay error 500` or `400: restricted: not a channel member` after a code change | Stale binary | Rebuild and re-export `PATH`; or `cargo run` directly |
-| `Address already in use` on relay start (os error 48 on macOS, 98 on Linux) | Another relay (or stale process) holding `:3000` / `:8080` / `:9102` (or your override ports) | The panic line names the failing port — read it first. Then `lsof -iTCP:3000,8080,9102 -sTCP:LISTEN` (or your override equivalents). Kill the offender (`pkill -f buzz-relay`) or use the port-override block in step 3. If you already overrode and *still* collide, a prior reviewer left a relay running on the same alt ports — kill it or pick fresh ports |
+| `Address already in use` on relay start (os error 48 on macOS, 98 on Linux) | Another relay (or stale process) holding `:3000` / `:8080` / `:9102` (or your override ports) | Metrics-listener failures emit a `metrics_bind` lifecycle terminal with reason `bind`. Check the configured ports with `lsof -iTCP:3000,8080,9102 -sTCP:LISTEN` (or your override equivalents). Kill the offender (`pkill -f buzz-relay`) or use the port-override block in step 3. If you already overrode and *still* collide, a prior reviewer left a relay running on the same alt ports — kill it or pick fresh ports |
 | `auth_error: BUZZ_PRIVATE_KEY is required` | Env not exported into the CLI's shell | `export BUZZ_PRIVATE_KEY=...` (or pass `--private-key`) |
 | `auth_error: BUZZ_AUTH_TAG verification failed … signature verification failed` | A stale `BUZZ_AUTH_TAG` inherited from a parent shell. The local dev relay rejects it. | `unset BUZZ_AUTH_TAG` (see the scrub block in step 1) |
 | `auth-required: verification failed` on a closed relay | NIP-OA attestation needed | Set `BUZZ_AUTH_TAG` to the owner-issued JSON, or relax `BUZZ_REQUIRE_RELAY_MEMBERSHIP` |

crates/buzz-auth/src/nip_fi/assertion.rs

@@ -207,6 +207,48 @@ impl fmt::Debug for VerifiedAssertion {
     }
 }
 
+#[cfg(any(test, feature = "test-utils"))]
+impl VerifiedAssertion {
+    /// Test-only factory for building `VerifiedAssertion` fixtures without
+    /// going through the full JWT/JWKS verification path. NOT available in
+    /// production builds.
+    ///
+    /// # Panics
+    ///
+    /// Panics when `authority_deadlines` is empty — an empty set violates the
+    /// non-empty invariant that `upstream_authority_deadline()` relies on.
+    pub fn for_test(
+        asserted_key: Option<PublicKey>,
+        authority_deadlines: Vec<DateTime<Utc>>,
+    ) -> Self {
+        assert!(
+            !authority_deadlines.is_empty(),
+            "VerifiedAssertion::for_test: authority_deadlines must be non-empty \
+             (upstream_authority_deadline() panics on empty)"
+        );
+        use super::config::{AssertionPolicyId, TransportContractId};
+        Self {
+            identity: FederatedIdentity {
+                issuer: "test-issuer".to_string(),
+                subject: "test-subject".to_string(),
+            },
+            asserted_key,
+            capabilities: CanonicalCapabilities::from_pairs(vec![]),
+            authority_deadlines,
+            assertion_policy_id: AssertionPolicyId::zero(),
+            transport_contract_id: TransportContractId::zero(),
+            revalidation_dependencies: RevalidationDependencies {
+                verification_key_id: "test-kid".to_string(),
+                key_snapshot_generation: 0,
+                key_snapshot_hard_deadline: DateTime::<Utc>::MAX_UTC,
+                confidential_assertion: ConfidentialAssertion {
+                    compact_jws: "test.test.test".to_string(),
+                },
+            },
+        }
+    }
+}
+
 impl RevalidationDependencies {
     pub(super) fn new(
         verification_key_id: String,

crates/buzz-auth/src/nip_fi/config.rs

@@ -104,6 +104,13 @@ impl AssertionPolicyId {
     pub const fn as_bytes(&self) -> &[u8; 32] {
         &self.0
     }
+
+    /// Zero value for tests. Available in production builds only with the
+    /// `test-utils` feature enabled.
+    #[cfg(any(test, feature = "test-utils"))]
+    pub const fn zero() -> Self {
+        Self([0u8; 32])
+    }
 }
 
 impl fmt::Debug for AssertionPolicyId {
@@ -144,6 +151,13 @@ impl TransportContractId {
     pub const fn as_bytes(&self) -> &[u8; 32] {
         &self.0
     }
+
+    /// Zero value for tests. Available in production builds only with the
+    /// `test-utils` feature enabled.
+    #[cfg(any(test, feature = "test-utils"))]
+    pub const fn zero() -> Self {
+        Self([0u8; 32])
+    }
 }
 
 impl fmt::Debug for TransportContractId {

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

@@ -1055,6 +1055,16 @@ impl Db {
         }
     }
 
+    /// Return a reference to the writer pool.
+    ///
+    /// Callers that need a pool handle for standalone free functions (e.g.,
+    /// `buzz_db::insert_mentions`) can use this. Prefer the `Db` method
+    /// equivalents when they exist; use `pool()` only for functions that have
+    /// no `Db` wrapper yet.
+    pub fn pool(&self) -> &PgPool {
+        &self.pool
+    }
+
     /// Refresh all expected operation-specific waiter gauges, including zero.
     ///
     /// The relay pool sampler calls this periodically so an exporter idle

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

@@ -197,6 +197,82 @@ async fn acquire_channel_membership_lock(
     Ok(())
 }
 
+// ── Transaction-level membership helpers (for commit_participant_join) ────────
+
+/// Acquire the per-channel membership advisory lock on a caller-owned transaction.
+///
+/// Equivalent to the internal `acquire_channel_membership_lock`, but exposed
+/// for callers that need to compose multiple operations in one transaction
+/// (e.g., `commit_participant_join` in `audio/handler.rs`).
+pub async fn acquire_channel_membership_lock_in_transaction(
+    tx: &mut Transaction<'_, Postgres>,
+    community_id: CommunityId,
+    channel_id: Uuid,
+) -> Result<()> {
+    acquire_channel_membership_lock(tx, community_id, channel_id).await
+}
+
+/// Check whether a pubkey is an active channel member on a caller-owned transaction.
+///
+/// Runs the same query as `is_member` but within the caller's transaction so
+/// the read is serialized with any concurrent membership writes on the same lock.
+pub async fn is_member_in_transaction(
+    tx: &mut Transaction<'_, Postgres>,
+    community_id: CommunityId,
+    channel_id: Uuid,
+    pubkey: &[u8],
+) -> Result<bool> {
+    let row = sqlx::query(
+        "SELECT COUNT(*) as cnt FROM channel_members cm \
+         JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL \
+         WHERE cm.community_id = $1 AND cm.channel_id = $2 AND cm.pubkey = $3 AND cm.removed_at IS NULL",
+    )
+    .bind(community_id.as_uuid())
+    .bind(channel_id)
+    .bind(pubkey)
+    .fetch_one(&mut **tx)
+    .await?;
+    let cnt: i64 = row.try_get("cnt")?;
+    Ok(cnt > 0)
+}
+
+/// Auto-add a member on a caller-owned transaction (for ephemeral-channel admission).
+///
+/// Inserts or reactivates the membership row at `Member` role with the given
+/// `invited_by` (channel creator for huddle auto-add). Does NOT acquire the
+/// advisory lock — callers must have already called
+/// `acquire_channel_membership_lock_in_transaction` before calling this.
+///
+/// Used by `commit_participant_join` to atomically add membership and the
+/// `48101` event in a single transaction under a session effect permit.
+pub async fn insert_auto_membership_in_transaction(
+    tx: &mut Transaction<'_, Postgres>,
+    community_id: CommunityId,
+    channel_id: Uuid,
+    pubkey: &[u8],
+    invited_by: &[u8],
+) -> Result<()> {
+    sqlx::query(
+        r#"
+        INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by)
+        VALUES ($1, $2, $3, 'member'::member_role, $4)
+        ON CONFLICT (community_id, channel_id, pubkey) DO UPDATE SET
+            removed_at = NULL,
+            removed_by = NULL,
+            role = EXCLUDED.role
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(channel_id)
+    .bind(pubkey)
+    .bind(invited_by)
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
+}
+
+// ── End transaction-level helpers ─────────────────────────────────────────────
+
 /// An active member roster captured while holding the channel's membership
 /// serialization lock on one writer connection.
 pub struct LockedMemberSnapshot {

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

@@ -341,6 +341,59 @@ async fn huddle_started_link_exists_with_operation(
         .any(|content| huddle_started_content_links(content, ephemeral_channel_id)))
 }
 
+/// Return whether a creator-signed huddle-start event links a parent channel
+/// to the requested ephemeral huddle channel — checked inside an open
+/// transaction with a shared row lock on matching rows.
+///
+/// Uses `SELECT ... FOR SHARE` so any concurrent `soft_delete_event()` that
+/// attempts `UPDATE events SET deleted_at = NOW() WHERE ...` on the same row
+/// must wait until this transaction commits or rolls back. This makes the
+/// re-read authoritative against concurrent deletion — "visibility" alone
+/// (i.e. a plain SELECT) is insufficient under READ COMMITTED because deletion
+/// can commit between the SELECT and the join commit in the same transaction.
+///
+/// Uses `tx.as_mut()` so the lock participates in the caller's transaction.
+/// A `false` return means the link was deleted or was never inserted, and the
+/// caller should abort the surrounding transaction.
+pub async fn huddle_started_link_exists_in_transaction(
+    tx: &mut Transaction<'_, Postgres>,
+    community_id: CommunityId,
+    parent_channel_id: Uuid,
+    ephemeral_channel_id: Uuid,
+    creator_pubkey: &[u8],
+) -> Result<bool> {
+    let uuid_needle = format!("%{}%", ephemeral_channel_id);
+    let candidates: Vec<String> = sqlx::query_scalar(
+        r#"
+        SELECT content
+        FROM events
+        WHERE deleted_at IS NULL
+          AND community_id = $1
+          AND channel_id = $2
+          AND kind = $3
+          AND pubkey = $4
+          AND octet_length(content) <= $5
+          AND content ILIKE $6
+        ORDER BY created_at DESC, id ASC
+        LIMIT $7
+        FOR SHARE
+        "#,
+    )
+    .bind(community_id.as_uuid())
+    .bind(parent_channel_id)
+    .bind(KIND_HUDDLE_STARTED as i32)
+    .bind(creator_pubkey)
+    .bind(HUDDLE_LINK_CONTENT_MAX_BYTES)
+    .bind(uuid_needle)
+    .bind(HUDDLE_LINK_CANDIDATE_LIMIT)
+    .fetch_all(tx.as_mut())
+    .await?;
+
+    Ok(candidates
+        .iter()
+        .any(|content| huddle_started_content_links(content, ephemeral_channel_id)))
+}
+
 /// Insert a Nostr event. Rejects AUTH and ephemeral kinds.
 ///
 /// Returns `(StoredEvent, was_inserted)` — `was_inserted` is `false` on duplicate.
@@ -2837,6 +2890,135 @@ mod postgres_tests {
         assert_eq!(links, vec![(session, parent, creator)]);
     }
 
+    // I4 deletion-race witness:
+    // `huddle_started_link_exists_in_transaction` acquires FOR SHARE on the
+    // matching row. A concurrent `soft_delete_event` (UPDATE events SET
+    // deleted_at = NOW() WHERE ...) must BLOCK until the join transaction
+    // commits or rolls back — it cannot race past the re-read and commit
+    // deletion before the join completes.
+    //
+    // Test protocol:
+    //   1. Insert a huddle_started event row.
+    //   2. Open a transaction and call `huddle_started_link_exists_in_transaction`
+    //      (acquires FOR SHARE).
+    //   3. Concurrently try `soft_delete_event` from a second connection —
+    //      the UPDATE blocks because FOR SHARE conflicts with UPDATE.
+    //   4. Commit the first transaction.
+    //   5. The concurrent delete now completes — confirm it succeeds.
+    //
+    // Mutation evidence:
+    //   Remove `FOR SHARE` from the SELECT in `huddle_started_link_exists_in_transaction` →
+    //   the concurrent delete completes before the join tx commits →
+    //   `link_gone_before_commit` becomes true before the tx commits →
+    //   assertion panics ("FOR SHARE must make delete block").
+    #[tokio::test]
+    #[ignore = "requires Postgres — link deletion contends with join transaction via FOR SHARE"]
+    async fn i4_huddle_link_deletion_blocked_by_join_transaction_for_share() {
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use std::sync::Arc;
+        use tokio::sync::Notify;
+
+        let pool = setup_pool().await;
+        let community = make_test_community(&pool).await;
+        let community_id = buzz_core::CommunityId::from_uuid(community);
+        let parent = make_test_channel(&pool, community, None).await;
+        let session = make_test_channel(&pool, community, None).await;
+        let creator = vec![0xAAu8; 32];
+        let event_id = vec![0xBBu8; 32];
+
+        // Insert the huddle_started event row.
+        let content = serde_json::json!({"ephemeral_channel_id": session.to_string()}).to_string();
+        sqlx::query(
+            "INSERT INTO events \
+             (community_id, id, pubkey, created_at, kind, tags, content, sig, channel_id) \
+             VALUES ($1, $2, $3, NOW(), $4, '[]', $5, $6, $7)",
+        )
+        .bind(community)
+        .bind(&event_id)
+        .bind(&creator)
+        .bind(KIND_HUDDLE_STARTED as i32)
+        .bind(&content)
+        .bind(vec![0u8; 64])
+        .bind(parent)
+        .execute(&pool)
+        .await
+        .expect("insert huddle_started event");
+
+        // Signal: join transaction has acquired FOR SHARE, delete may attempt.
+        let delete_may_start = Arc::new(Notify::new());
+        // Signal: delete completed (or timed out).
+        let delete_completed = Arc::new(AtomicBool::new(false));
+        let link_gone_before_commit = Arc::new(AtomicBool::new(false));
+
+        let delete_may_start2 = delete_may_start.clone();
+        let delete_completed2 = delete_completed.clone();
+        let link_gone2 = link_gone_before_commit.clone();
+        let pool2 = pool.clone();
+        let event_id2 = event_id.clone();
+        let community2 = community_id;
+
+        // Spawn the deleter: waits for the join tx to hold FOR SHARE, then tries
+        // to delete. It should block until the join tx commits.
+        let delete_handle = tokio::spawn(async move {
+            delete_may_start2.notified().await;
+            // Record whether the link row is still live at delete time.
+            // Under FOR SHARE this call will block until the join tx commits.
+            let result = soft_delete_event(&pool2, community2, &event_id2)
+                .await
+                .expect("soft_delete_event should not error");
+            // Mark whether the link was deleted (not already gone).
+            link_gone2.store(result, Ordering::Relaxed);
+            delete_completed2.store(true, Ordering::Relaxed);
+        });
+
+        // Open the join transaction and acquire FOR SHARE.
+        let mut tx = pool.begin().await.expect("begin join tx");
+        let exists = huddle_started_link_exists_in_transaction(
+            &mut tx,
+            community_id,
+            parent,
+            session,
+            &creator,
+        )
+        .await
+        .expect("huddle_started_link_exists_in_transaction");
+        assert!(exists, "I4: link must exist before commit");
+
+        // Signal the deleter to attempt its UPDATE now.
+        delete_may_start.notify_one();
+
+        // Give the deleter a brief window to attempt the DELETE. Under correct
+        // FOR SHARE locking, it blocks here and `delete_completed` stays false.
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+
+        assert!(
+            !delete_completed.load(Ordering::Relaxed),
+            "I4: FOR SHARE must make soft_delete_event block — \
+             delete completed before the join transaction committed, \
+             which proves deletion can race past the re-read. \
+             Remove FOR SHARE from the SELECT in \
+             huddle_started_link_exists_in_transaction to reproduce."
+        );
+
+        // Commit the join transaction — delete should unblock.
+        tx.commit().await.expect("commit join tx");
+
+        tokio::time::timeout(std::time::Duration::from_secs(5), delete_handle)
+            .await
+            .expect("I4: delete must complete within 5s after join tx commit")
+            .expect("delete_handle must not panic");
+
+        // After the join tx commits, the delete should have succeeded.
+        assert!(
+            link_gone_before_commit.load(Ordering::Relaxed),
+            "I4: soft_delete_event must succeed once the join tx releases FOR SHARE"
+        );
+        assert!(
+            delete_completed.load(Ordering::Relaxed),
+            "I4: delete must complete after join tx commit"
+        );
+    }
+
     #[test]
     fn huddle_started_content_requires_matching_ephemeral_field() {
         let channel_id = Uuid::new_v4();

crates/buzz-relay/Cargo.toml

@@ -39,6 +39,7 @@ tower-http = { workspace = true }
 nostr = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+jsonwebtoken = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 tracing-opentelemetry = { workspace = true }
@@ -95,7 +96,7 @@ mesh-llm-host-runtime = { git = "https://github.com/Mesh-LLM/mesh-llm.git", tag
 buzz-test-client = { path = "../buzz-test-client" }
 ed25519-dalek = "=3.0.0-rc.0"
 buzz-core = { workspace = true, features = ["test-utils"] }
-buzz-auth = { workspace = true, features = ["dev"] }
+buzz-auth = { workspace = true, features = ["dev", "test-utils"] }
 reqwest = { workspace = true }
 tokio-tungstenite = { workspace = true }
 futures = "0.3"