feat(buzz-relay): NIP-FI stateless enforcement (S3) — upgrade gate, NIP-42 pairing, session lifetime, JWKS warm

Justfile

@@ -459,7 +459,13 @@ test-unit:
         # the ~30s sqlx acquire timeout, so they do not belong in the infra-free
         # unit job either.
         cargo nextest run -p buzz-relay --lib \
-            -E '(test(/^api::admin::/) - test(=api::admin::tests::disabled_mode_allows_unauthenticated_requests_on_the_admin_host) - test(=api::admin::tests::nip98_mode_unrostered_signer_does_not_consume_a_replay_slot)) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/)'
+            -E '(test(/^api::admin::/) - test(=api::admin::tests::disabled_mode_allows_unauthenticated_requests_on_the_admin_host) - test(=api::admin::tests::nip98_mode_unrostered_signer_does_not_consume_a_replay_slot)) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^storage_sweep::tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/)'
+        # Note on audio::join::tests scope: the full suite is infra-free (no DB,
+        # no Redis). The infra-free audio/FI regression witnesses — bootstrap
+        # ordering barrier, CommitConfirmed arm, pending-close invisibility,
+        # abnormal-stream-close fanout, and never-ready-sink writer witnesses —
+        # are all selected by audio::join::tests and audio::handler::tests.
+        # DB-backed audio join tests use #[ignore] and run in the postgres lane.
         # ACP author-gate and queue tests protect the trust boundary between
         # relay events and agent prompts. They are infra-free; ignored lifecycle
         # tests remain excluded and run in their dedicated integration lanes.

crates/buzz-auth/src/lib.rs

@@ -61,6 +61,8 @@ pub use access::MockAccessChecker;
 #[cfg(any(test, feature = "test-utils"))]
 pub use nip98_replay::AlwaysFreshReplayGuard;
 #[cfg(any(test, feature = "test-utils"))]
+pub use nip_fi::ToggleJwksFetcher;
+#[cfg(any(test, feature = "test-utils"))]
 pub use rate_limit::AlwaysAllowRateLimiter;
 
 /// How the connection was authenticated.

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-auth/src/nip_fi/jwks/mod.rs

@@ -734,5 +734,74 @@ impl<F> std::fmt::Debug for ProductionJwksSource<F> {
     }
 }
 
+/// A toggle-controlled [`JwksFetcher`] for use in downstream-crate integration
+/// tests. Returns a minimal but valid JWKS document when the toggle is `true`,
+/// and `NetworkError` when `false`.
+///
+/// This is the only path by which a crate outside `buzz-auth` can build a
+/// `ProductionJwksSource` with a controllable fetch outcome — `sealed::Sealed`
+/// is crate-private, so downstream crates cannot implement `JwksFetcher`
+/// directly. Because the returned JWKS is real (not a synthetic shortcut), the
+/// full `ProductionJwksSource` code path — parse, bound, cache, hard-deadline —
+/// exercises itself normally, and the resulting `AssertionKeySet` is valid for
+/// verifier lookups.
+///
+/// Only available with the `test-utils` feature enabled.
+#[cfg(any(test, feature = "test-utils"))]
+#[derive(Clone, Debug)]
+pub struct ToggleJwksFetcher {
+    /// When `true` the fetcher returns a minimal valid JWKS body; when `false`
+    /// it returns `JwksFetchError::NetworkError`.
+    pub available: std::sync::Arc<std::sync::atomic::AtomicBool>,
+    /// Fired (via `notify_one`) after every fetch attempt completes, whether
+    /// successful or not. Tests can await this to deterministically observe
+    /// that the fetch loop executed a given attempt before proceeding.
+    pub fetch_done: std::sync::Arc<tokio::sync::Notify>,
+}
+
+#[cfg(any(test, feature = "test-utils"))]
+impl ToggleJwksFetcher {
+    /// Construct a new `ToggleJwksFetcher`. Pass `initial` as the starting
+    /// availability state; `available` and `fetch_done` are externally
+    /// observable and can be driven from the test after construction.
+    pub fn new(initial: bool) -> Self {
+        Self {
+            available: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(initial)),
+            fetch_done: std::sync::Arc::new(tokio::sync::Notify::new()),
+        }
+    }
+}
+
+#[cfg(any(test, feature = "test-utils"))]
+impl super::verifier::sealed::Sealed for ToggleJwksFetcher {}
+
+#[cfg(any(test, feature = "test-utils"))]
+impl JwksFetcher for ToggleJwksFetcher {
+    fn fetch_jwks<'a>(
+        &'a self,
+        _uri: &'a str,
+    ) -> impl std::future::Future<Output = Result<String, JwksFetchError>> + Send + 'a {
+        // A minimal P-256 JWK.  The coordinates are the same values used in
+        // buzz-auth's own test suite (tests.rs `minimal_jwks_json`).
+        const TOGGLE_JWKS: &str = concat!(
+            r#"{"keys":[{"kty":"EC","crv":"P-256","#,
+            r#""x":"f83OJ3D2xF1Bg8vub9tLe1gHMzV76e8Tus9uPHvRVEU","#,
+            r#""y":"x_FEzRu9m36HLN_tue659LNpXW6pCyStikYjKIWI5a0","#,
+            r#""use":"sig","alg":"ES256","kid":"toggle-kid"}]}"#
+        );
+        let available = self.available.load(std::sync::atomic::Ordering::SeqCst);
+        let fetch_done = std::sync::Arc::clone(&self.fetch_done);
+        async move {
+            let result = if available {
+                Ok(TOGGLE_JWKS.to_string())
+            } else {
+                Err(JwksFetchError::NetworkError)
+            };
+            fetch_done.notify_one();
+            result
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests;

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

@@ -33,5 +33,8 @@ pub use jwks::{
     HttpJwksFetcher, IssuerJwksConfig, JwksFetchError, JwksFetcher, JwksSourceContract,
     ProductionJwksSource,
 };
+
+#[cfg(any(test, feature = "test-utils"))]
+pub use jwks::ToggleJwksFetcher;
 pub use startup::{validate_nip_fi_config, NipFiMode, NipFiStartupError};
 pub use verifier::{AssertionKeySet, FederatedAssertionVerifier, IssuerKeySource, VerifierError};

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

@@ -1169,6 +1169,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 {