feat(relay): project compact viewer memberships (BUZZ-615) (#9118)

## Summary

Cold login downloads other people's full rosters just to find the
viewer's channels. This adds a compact projection from canonical
memberships through the existing Nostr `/query` bridge. One primary SQL
statement reads the viewer's rows and channel state, then the relay
signs that exact result.

Merged spec PR #9117 owns both relay-only kind registrations. Self mode
(`membership_snapshot`, kind 39008) requires the viewer's own NIP-98
key. Public mode (`membership_public`, kind 39009) lets an admitted
requester look up any subject's listed, open forum/stream/workflow
memberships. It excludes private/unlisted channels, DMs, system channels
and deleted or removed memberships, including shared private channels.
Both modes require payload-bound NIP-98 and return signed request-bound
envelopes; overflow or backend errors fail the whole request. Legacy
kind-39002 reads and signatures stay intact. No schema migration or
persisted per-user list is added.

### Related issue

[BUZZ-615](https://linear.app/squareup/issue/BUZZ-615). Implements the
compact discovery contract merged in #9117. Related:
[#8898](https://github.com/block/buzz/pull/8898),
[#9060](https://github.com/block/buzz/pull/9060). Client lazy-loading
follows separately; this does not include
[buzz-app#2627](https://github.com/block/buzz-app/pull/2627).

### Testing

Current head `5c0c35400a` rebases the five implementation patches
unchanged onto main `c39ee507`, then repairs the removed Redis test API.
McKeon independently verified the rebase and reviewed that repair. Full
core/database/relay package tests, PostgreSQL membership tests (9/9),
Clippy and every mandatory publication hook pass at this head. Old-head
CI run 42132655155 failed six jobs on that same removed API; four
summary checks propagated those failures. New-head hosted CI is pending.
Main advanced to `e3a0f3d59d` during publication; a local merge-tree
check found no conflicts. Historical runtime evidence below remains
attributed to its tested heads.

Prior self-mode validation at `6d2ffceeee` (historical evidence, not
validation of the public-mode change):

- `just ci`, pre-push hooks and all 2,221 core/DB/relay package tests
passed.
- All six new Postgres protocol tests passed. All five guard mutations
failed assertions: mandatory auth, flag gate, single-statement cut,
overflow probe and tenant predicate. A/B are older runs against
production files unchanged from `9784df51`; C/D/E were rerun on final
SQL/fixtures. The earlier surviving tenant mutation is superseded by the
final duplicate-ID witness.
- The full isolated Postgres run finished 1,288/1,294 green; six
existing push/media/FI/startup cases failed. All six passed when rerun
serially after the concurrent local gates stopped. The failed full run
is retained and is not represented as a clean full-suite pass. The
preceding correction `1c47dfc9` had passed all 1,294 tests.
- `just test` passed its unit steps but integration setup was blocked by
Docker Desktop organization sign-in enforcement. The waiting wrapper was
stopped with SIGTERM; this is not a complete pass.
- A real local relay rebuilt at this head returned 67,896 bytes for 468
memberships in 33 ms unthrottled. Signatures, request/viewer/tenant
binding, content identity, wrong-viewer refusal, removal and role
changes verified.

The scale fixture adds five large rosters and 20,000 unrelated channels.
Compact bytes stay unchanged as other members grow; EXPLAIN checks 468
indexed viewer rows and bounded channel-ID lookups. A same-UUID
cross-tenant fixture verifies fields and duplicate rejection
independently of row order.

Durable local evidence: `WORK_LOGS/BUZZ_615_20261105/MANIFEST.md` and
`MANIFEST.json` map commands/results to commits or verified mutant-file
hashes and index the exported logs. Concrete human-test steps and the
signed-request helper are included. Human testing remains pending; ready
state follows the author’s explicit request. Native/browser 140/205 KB/s
acceptance belongs to the client PRs.

Public-mode validation at `c6e269170e`: McKeon independently ran all
nine isolated PostgreSQL membership tests and five mutations. Every
mutation failed its expected assertion: public selection predicate,
subject tag, public routing, identity domain and self serialization
order. The full core/DB/relay package run, all nine isolated PostgreSQL
membership tests and the relay build passed at c6. A real signed HTTP
workflow verified signatures, bindings, identity hashes, all visibility
exclusions, archived guest roles, requester-independent content identity
and malformed-request refusals. After a channel became private, the
public list shrank from three entries to two; a role change appeared as
admin. The local test relay was stopped afterward.

The restacked child `8c56f97d6409` has the same code, tests and
configuration as c6; only the spec differs. A new full core/DB/relay
package run passed at that exact child head. The parent `17a481a89716`
separately passed the full core package check. The first child attempt
used the wrong local database and failed five existing tests on a
missing table; its log is retained separately from the successful
configured run. These checks do not attest human testing.

### Rollout and rollback

This PR does not deploy or enable the feature. Both
`BUZZ_MEMBERSHIP_SNAPSHOT_ENABLED` and `BUZZ_MEMBERSHIP_PUBLIC_ENABLED`
default to false and independently gate advertisement and execution. An
authorized operator sets them in the relay Helm environment managed by
`squareup/block-coder-tf-stacks`. First deploy the image to every
staging pod with both flags off; then enable each capability separately
for staging validation. Enabling either mode does not require or enable
the other. Repeat that dark-then-enable order for bb-block after staging
passes. Opt-in client rollout follows the relay contract and the
throttled desktop/browser workflow.

Watch `buzz_membership_snapshot_query_seconds`,
`buzz_membership_snapshot_encode_sign_seconds`,
`buzz_membership_snapshot_response_bytes`, split by
`mode=membership_snapshot` or `mode=membership_public`, plus `/query`
413/503 responses and primary-pool wait time for the Authorization
operation. Confirm zero unexpected full-roster discovery downloads on
supported clients and correct roles/membership against the canonical
cut. Abort enablement for any binding/signature mismatch, access
leakage, incorrect complete result, stale-response regrant, unexpected
overflow for the 468-channel account, sustained 503s or a material
regression in primary-pool wait/foreground reads. Turn the affected flag
off to roll back. Self-mode clients recheck support and start a new
legacy authority generation; public-mode clients recheck support and
back off without treating public absence as membership loss. Keep cached
display/user data.

The first named usable channel target of five seconds at 140 KB/s
remains a client acceptance target, not a result established by this
relay PR. The 119-event approximately 5 MB read-state contention and
native/browser throttled tests are required before broad client rollout.

Generated with Codex

---------

Signed-off-by: Brad Seiler <seiler@squareup.com>
Co-authored-by: Honor <e0bf479eeebe47dde1e35705212a4df70616a7bc834d907ddfb18cf0405f8b1b@seiler-test.communities.public.builderlabstaging.xyz>
This commit is contained in:
Brad Seiler
2026-11-05 20:14:08 -05:00
committed by GitHub
parent fd0ec10505
commit bb3c5f7175
11 changed files with 1624 additions and 6 deletions

View File

@@ -473,3 +473,8 @@ RUST_LOG=buzz_relay=debug,buzz_datastore=info,buzz_db=debug,buzz_auth=debug,buzz
# New irreversible community FI acceptances stay disabled until the fleet's
# enforcement and activation baseline is verified. Disabling preserves receipts.
BUZZ_NIP_FI_UPGRADES_ENABLED=false
# NIP-CL compact membership discovery. Deploy fleet-wide dark before enabling.
BUZZ_MEMBERSHIP_SNAPSHOT_ENABLED=false
# Independent listed-open public membership lookup; also deploy disabled.
BUZZ_MEMBERSHIP_PUBLIC_ENABLED=false

View File

@@ -90,6 +90,10 @@ pub enum DbError {
#[error("read-state snapshot exceeds event or byte limit")]
ReadStateSnapshotTooLarge,
/// A complete membership snapshot exceeds its bounded resource budget.
#[error("membership snapshot exceeds channel or byte limit")]
MembershipSnapshotTooLarge,
/// A complete thread window exceeds its request-wide work allowance.
#[error("thread window exceeds {0} budget")]
ThreadWindowBudgetExceeded(&'static str),

View File

@@ -89,10 +89,10 @@ pub(crate) use runtime::{
pub use store::{
admin_moderation, allowlist, api_token, archived_identities, artifact, artifact_sync, channel,
channel_members, community, community_auth, deletion, dm, entity_deletion, event, feed,
git_repo, leader_lock, moderation, nip_fi_binding, operator_listener, partition, personal_read,
product_feedback, push, reaction, read_state, relay_admin_actions, relay_invite, relay_members,
relay_operators, reminder, replaceable, storage_accounting, thread, thread_window, user,
user_deletion, user_deletion_fences, workflow,
git_repo, leader_lock, membership_snapshot, moderation, nip_fi_binding, operator_listener,
partition, personal_read, product_feedback, push, reaction, read_state, relay_admin_actions,
relay_invite, relay_members, relay_operators, reminder, replaceable, storage_accounting,
thread, thread_window, user, user_deletion, user_deletion_fences, workflow,
};
pub use allowlist::AllowlistEntry;

View File

@@ -0,0 +1,182 @@
//! Bounded, primary-only projections of the viewer's canonical memberships.
use buzz_core::CommunityId;
use serde::Serialize;
use sha2::{Digest, Sha256};
use sqlx::Row;
use uuid::Uuid;
use crate::{observability, Db, DbError, Result};
/// Maximum memberships in a complete NIP-CL version-1 response.
pub const MAX_CHANNELS: usize = 4096;
/// Maximum encoded response bytes, including the signed event and envelope.
pub const MAX_RESPONSE_BYTES: usize = 2 * 1024 * 1024;
/// Types understood by version 1, in canonical wire order.
pub const CHANNEL_TYPES: &[&str] = &["dm", "forum", "stream", "system", "workflow"];
/// Exactly the fields needed to establish the viewer's membership.
/// Field declaration order is part of the snapshot content-identity encoding.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ViewerMembership {
/// Canonical channel UUID.
pub id: Uuid,
/// This viewer's role; never a list of other members.
pub role: String,
/// Channel type, within the explicitly requested scope.
pub channel_type: String,
/// Current channel visibility, independently of membership.
pub visibility: String,
/// Archived channels remain memberships, with read-only behavior.
pub archived: bool,
}
/// A complete projection at one primary statement's MVCC cut.
#[derive(Debug)]
pub struct MembershipSnapshot {
/// Content identity, never an ordered revision or a live-delivery cursor.
pub snapshot_id: String,
/// Every active membership in the requested types, sorted by channel UUID.
pub channels: Vec<ViewerMembership>,
}
// No events, rosters, profiles, other-member aggregates or replica routing.
// Probe one additional row; never silently certify a capped set as complete.
// Keep the lookup dependent on the viewer row to avoid a community-wide scan.
// Probe two rows so an accidental loss of the tenant/key predicate cannot hide
// a duplicate behind LIMIT 1: strict ID validation below then fails closed.
// The outer probe counts AFTER type/deletion filters.
const SNAPSHOT_SQL: &str = "SELECT cm.channel_id AS id, cm.role::text AS role,
c.channel_type, c.visibility, c.archived
FROM channel_members cm
CROSS JOIN LATERAL (
SELECT c.channel_type::text AS channel_type, c.visibility::text AS visibility,
c.archived_at IS NOT NULL AS archived
FROM channels c
WHERE c.community_id = cm.community_id AND c.id = cm.channel_id
AND (NOT $5 OR (c.visibility = 'open' AND NOT c.unlisted))
AND c.deleted_at IS NULL AND c.channel_type = ANY($3::text[]::channel_type[])
LIMIT 2
) c
WHERE cm.community_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL
ORDER BY cm.channel_id LIMIT $4";
/// Validate the explicit type scope without silently dropping unknown types.
pub fn valid_types(types: &[String]) -> bool {
!types.is_empty()
&& types.len() <= CHANNEL_TYPES.len()
&& types.iter().all(|t| CHANNEL_TYPES.contains(&t.as_str()))
&& types.windows(2).all(|pair| pair[0] < pair[1])
}
/// Public mode accepts only listed channel types, never DMs or system channels.
pub fn valid_public_types(types: &[String]) -> bool {
valid_types(types)
&& types
.iter()
.all(|t| ["forum", "stream", "workflow"].contains(&t.as_str()))
}
impl Db {
/// Project only the viewer's active memberships from one primary MVCC cut.
///
/// Archive/visibility/role/deletion changes share the same statement cut.
/// Hidden DMs are deliberately included; NIP-DV owns hiding separately.
/// No full-roster or per-user projection cache is consulted or maintained.
pub async fn membership_snapshot(
&self,
community: CommunityId,
viewer: &nostr::PublicKey,
types: &[String],
) -> Result<MembershipSnapshot> {
self.project_memberships(community, viewer, types, false)
.await
}
/// Project a subject's listed open memberships at one primary statement cut.
pub async fn public_memberships(
&self,
community: CommunityId,
subject: &nostr::PublicKey,
types: &[String],
) -> Result<MembershipSnapshot> {
self.project_memberships(community, subject, types, true)
.await
}
async fn project_memberships(
&self,
community: CommunityId,
viewer: &nostr::PublicKey,
types: &[String],
public: bool,
) -> Result<MembershipSnapshot> {
if !valid_types(types) || (public && !valid_public_types(types)) {
return Err(DbError::InvalidData("invalid membership type scope".into()));
}
let started = std::time::Instant::now();
let mut conn = observability::acquire_writer(
&self.pool,
observability::WriterOperation::Authorization,
)
.await?;
let rows = sqlx::query(SNAPSHOT_SQL)
.bind(community.as_uuid())
.bind(viewer.to_bytes().to_vec())
.bind(types)
.bind((MAX_CHANNELS + 1) as i64)
.bind(public)
.fetch_all(&mut *conn)
.await?;
metrics::histogram!("buzz_membership_snapshot_query_seconds", "mode" => if public {"membership_public"} else {"membership_snapshot"})
.record(started.elapsed().as_secs_f64());
if rows.len() > MAX_CHANNELS {
return Err(DbError::MembershipSnapshotTooLarge);
}
let channels = rows
.into_iter()
.map(|row| {
let entry = ViewerMembership {
id: row.try_get("id")?,
role: row.try_get("role")?,
channel_type: row.try_get("channel_type")?,
visibility: row.try_get("visibility")?,
archived: row.try_get("archived")?,
};
if !["owner", "admin", "member", "bot", "guest"].contains(&entry.role.as_str())
|| !types.contains(&entry.channel_type)
|| !["open", "private"].contains(&entry.visibility.as_str())
|| (public && entry.visibility != "open")
{
return Err(DbError::InvalidData(
"invalid membership projection row".into(),
));
}
Ok(entry)
})
.collect::<Result<Vec<_>>>()?;
if channels.windows(2).any(|pair| pair[0].id >= pair[1].id) {
return Err(DbError::InvalidData(
"membership projection IDs are not unique and ordered".into(),
));
}
let identity = serde_json::to_vec(&(
if public {
"nip-cl-public-membership-v1"
} else {
"nip-cl-membership-state-v1"
},
community.as_uuid(),
viewer.to_hex(),
types,
&channels,
))?;
Ok(MembershipSnapshot {
snapshot_id: hex::encode(Sha256::digest(identity)),
channels,
})
}
}
#[cfg(test)]
mod postgres_tests;

View File

@@ -0,0 +1,520 @@
use super::*;
use nostr::Keys;
use sqlx::{postgres::PgPoolOptions, PgPool};
use std::time::Duration;
async fn fixture(count: usize) -> (Db, PgPool, CommunityId, Keys, Vec<Uuid>) {
let pool = PgPool::connect(&crate::test_support::database_url())
.await
.unwrap();
let db = Db::from_pool(pool.clone());
let community = db
.ensure_configured_community(&format!("membership-{}.local", Uuid::new_v4().simple()))
.await
.unwrap()
.id;
let viewer = Keys::generate();
let ids: Vec<Uuid> = (0..count).map(|_| Uuid::new_v4()).collect();
let mut tx = crate::begin_community_write_transaction(
&pool,
community,
observability::WriterOperation::EventWrite,
)
.await
.unwrap();
sqlx::query(
"INSERT INTO channels (community_id,id,name,created_by)
SELECT $1,id,id::text,$3 FROM unnest($2::uuid[]) id",
)
.bind(community.as_uuid())
.bind(&ids)
.bind(viewer.public_key().to_bytes().to_vec())
.execute(tx.conn())
.await
.unwrap();
sqlx::query(
"INSERT INTO channel_members (community_id,channel_id,pubkey,role)
SELECT $1,id,$3,'owner' FROM unnest($2::uuid[]) id",
)
.bind(community.as_uuid())
.bind(&ids)
.bind(viewer.public_key().to_bytes().to_vec())
.execute(tx.conn())
.await
.unwrap();
tx.commit().await.unwrap();
(db, pool, community, viewer, ids)
}
fn all_types() -> Vec<String> {
CHANNEL_TYPES.iter().map(|s| s.to_string()).collect()
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn membership_snapshot_scope_roles_types_archive_hidden_and_overflow() {
let (db, pool, community, viewer, ids) = fixture(8).await;
for (i, (kind, role)) in CHANNEL_TYPES
.iter()
.zip(["owner", "admin", "member", "bot", "guest"])
.enumerate()
{
sqlx::query(
"UPDATE channels SET channel_type=$3::channel_type, visibility='private',
archived_at=CASE WHEN $4 THEN now() ELSE NULL END WHERE community_id=$1 AND id=$2",
)
.bind(community.as_uuid())
.bind(ids[i])
.bind(kind)
.bind(i == 1)
.execute(&pool)
.await
.unwrap();
sqlx::query(
"UPDATE channel_members SET role=$3::member_role, hidden_at=now()
WHERE community_id=$1 AND channel_id=$2",
)
.bind(community.as_uuid())
.bind(ids[i])
.bind(role)
.execute(&pool)
.await
.unwrap();
}
sqlx::query("UPDATE channels SET deleted_at=now() WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid())
.bind(ids[5])
.execute(&pool)
.await
.unwrap();
sqlx::query(
"UPDATE channel_members SET removed_at=now() WHERE community_id=$1 AND channel_id=$2",
)
.bind(community.as_uuid())
.bind(ids[6])
.execute(&pool)
.await
.unwrap();
// Open but not a member is excluded as well.
sqlx::query("DELETE FROM channel_members WHERE community_id=$1 AND channel_id=$2")
.bind(community.as_uuid())
.bind(ids[7])
.execute(&pool)
.await
.unwrap();
let dead = PgPoolOptions::new()
.connect_lazy("postgres://unused@127.0.0.1:1/absent")
.unwrap();
dead.close().await;
let routed = Db::from_pools(pool.clone(), dead);
let cut = routed
.membership_snapshot(community, &viewer.public_key(), &all_types())
.await
.unwrap();
assert_eq!(cut.channels.len(), 5);
assert!(cut.channels.windows(2).all(|p| p[0].id < p[1].id));
for (i, role) in ["owner", "admin", "member", "bot", "guest"]
.iter()
.enumerate()
{
let row = cut.channels.iter().find(|c| c.id == ids[i]).unwrap();
assert_eq!(&row.role, role);
assert_eq!(row.channel_type, CHANNEL_TYPES[i]);
assert_eq!(row.visibility, "private");
assert_eq!(row.archived, i == 1);
}
let systems = db
.membership_snapshot(community, &viewer.public_key(), &["system".into()])
.await
.unwrap();
assert_eq!(systems.channels.len(), 1);
assert_ne!(cut.snapshot_id, systems.snapshot_id);
let other = db
.membership_snapshot(community, &Keys::generate().public_key(), &all_types())
.await
.unwrap();
assert!(other.channels.is_empty());
let other_tenant = db
.ensure_configured_community(&format!("other-{}.local", Uuid::new_v4().simple()))
.await
.unwrap()
.id;
assert!(db
.membership_snapshot(other_tenant, &viewer.public_key(), &all_types())
.await
.unwrap()
.channels
.is_empty());
// Same UUID in another tenant is legal. A lost tenant predicate must
// produce a duplicate that fails closed, independently of physical order.
let shared_id = Uuid::new_v4();
sqlx::query(
"INSERT INTO channels (community_id,id,name,created_by,channel_type,visibility,archived_at)
VALUES ($1,$2,'collision-other',$3,'system','private',now())",
)
.bind(other_tenant.as_uuid())
.bind(shared_id)
.bind(Keys::generate().public_key().to_bytes().to_vec())
.execute(&pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO channels (community_id,id,name,created_by,channel_type,visibility)
VALUES ($1,$2,'collision-own',$3,'stream','open')",
)
.bind(community.as_uuid())
.bind(shared_id)
.bind(viewer.public_key().to_bytes().to_vec())
.execute(&pool)
.await
.unwrap();
sqlx::query("INSERT INTO channel_members (community_id,channel_id,pubkey,role) VALUES ($1,$2,$3,'guest')")
.bind(community.as_uuid()).bind(shared_id).bind(viewer.public_key().to_bytes().to_vec())
.execute(&pool).await.unwrap();
let collision_cut = db
.membership_snapshot(community, &viewer.public_key(), &all_types())
.await
.unwrap();
let entries: Vec<_> = collision_cut
.channels
.iter()
.filter(|c| c.id == shared_id)
.collect();
assert_eq!(entries.len(), 1);
assert_eq!(entries[0].channel_type, "stream");
assert_eq!(entries[0].visibility, "open");
assert_eq!(entries[0].role, "guest");
assert!(!entries[0].archived);
assert!(db
.membership_snapshot(community, &viewer.public_key(), &["unknown".into()])
.await
.is_err());
let (db, _, community, viewer, _) = fixture(MAX_CHANNELS + 1).await;
assert!(matches!(
db.membership_snapshot(community, &viewer.public_key(), &all_types())
.await,
Err(DbError::MembershipSnapshotTooLarge)
));
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn membership_snapshot_468_channels_size_independent_of_other_members_and_index_plan() {
let (db, pool, community, viewer, ids) = fixture(468).await;
let before = db
.membership_snapshot(community, &viewer.public_key(), &all_types())
.await
.unwrap();
let bytes = serde_json::to_vec(&before.channels).unwrap().len();
for (i, id) in ids.iter().enumerate() {
let members = [5898, 5847, 5307, 3314, 2007].get(i).copied().unwrap_or(5);
sqlx::query(
"INSERT INTO channel_members (community_id,channel_id,pubkey)
SELECT $1,$2,decode(md5(n::text)||md5(n::text),'hex') FROM generate_series(1,$3) n",
)
.bind(community.as_uuid())
.bind(id)
.bind(members - 1)
.execute(&pool)
.await
.unwrap();
}
// The community also has DMs and open channels the viewer cannot discover.
// This prevents a tiny all-member community from rewarding a directory scan.
let other = Keys::generate().public_key().to_bytes().to_vec();
sqlx::query(
"INSERT INTO channels (community_id,id,name,created_by,channel_type,visibility)
SELECT $1, md5('nonmember-'||n::text)::uuid, 'nonmember-'||n::text, $2,
CASE WHEN n%2=0 THEN 'dm' ELSE 'stream' END::channel_type,
CASE WHEN n%2=0 THEN 'private' ELSE 'open' END::channel_visibility
FROM generate_series(1,20000) n",
)
.bind(community.as_uuid())
.bind(&other)
.execute(&pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO channel_members (community_id,channel_id,pubkey)
SELECT $1,md5('nonmember-'||n::text)::uuid,$2 FROM generate_series(1,20000) n",
)
.bind(community.as_uuid())
.bind(other)
.execute(&pool)
.await
.unwrap();
sqlx::raw_sql("ANALYZE channel_members; ANALYZE channels")
.execute(&pool)
.await
.unwrap();
let start = std::time::Instant::now();
let after = db
.membership_snapshot(community, &viewer.public_key(), &all_types())
.await
.unwrap();
assert_eq!(before.channels, after.channels);
assert_eq!(before.snapshot_id, after.snapshot_id);
assert_eq!(bytes, serde_json::to_vec(&after.channels).unwrap().len());
let row = sqlx::query(sqlx::AssertSqlSafe(format!(
"EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) {SNAPSHOT_SQL}"
)))
.bind(community.as_uuid())
.bind(viewer.public_key().to_bytes().to_vec())
.bind(all_types())
.bind((MAX_CHANNELS + 1) as i64)
.bind(false)
.fetch_one(&pool)
.await
.unwrap();
let plan: serde_json::Value = row.get(0);
let mut nodes = vec![&plan[0]["Plan"]];
let mut scanned_members = false;
let mut viewer_index = false;
let mut channel_key_lookup = false;
while let Some(node) = nodes.pop() {
if let Some(children) = node["Plans"].as_array() {
nodes.extend(children);
}
if node["Index Name"] == "idx_channel_members_pubkey" {
viewer_index = true;
}
if node["Relation Name"] == "channel_members" {
assert_eq!(node["Actual Rows"], 468, "{plan}");
assert_eq!(node["Actual Loops"], 1, "{plan}");
scanned_members = true;
}
if node["Relation Name"] == "channels" {
assert!(
matches!(
node["Index Name"].as_str(),
Some("channels_pkey" | "idx_channels_id_live")
),
"{plan}"
);
assert_eq!(node["Actual Loops"], 468, "{plan}");
let rows = node["Actual Rows"].as_u64().unwrap();
let filtered = node["Rows Removed by Filter"].as_u64().unwrap_or(0);
assert!(rows <= 1 && filtered <= 1, "{plan}");
assert!(
rows * node["Actual Loops"].as_u64().unwrap() <= 468,
"{plan}"
);
channel_key_lookup = true;
}
}
assert!(
scanned_members && viewer_index && channel_key_lookup,
"{plan}"
);
println!(
"468-channel projection with 20000 unrelated channels: bytes={bytes}, query_ms={}\n{}",
start.elapsed().as_secs_f64() * 1000.0,
serde_json::to_string_pretty(&plan).unwrap()
);
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn cluster_global_membership_snapshot_role_and_removal_commit_during_statement_keep_one_cut()
{
let (db, pool, community, viewer, ids) = fixture(4).await;
let before = db
.membership_snapshot(community, &viewer.public_key(), &all_types())
.await
.unwrap();
// A private view pauses the REAL production query without changing its SQL.
// The gate runs under its statement snapshot while public canonical rows change.
let schema = format!("membership_gate_{}", Uuid::new_v4().simple());
let lock_key = i64::from_be_bytes(Uuid::new_v4().as_bytes()[..8].try_into().unwrap());
sqlx::raw_sql(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {schema};
CREATE FUNCTION {schema}.gate() RETURNS boolean LANGUAGE plpgsql VOLATILE AS
$$ BEGIN PERFORM pg_advisory_xact_lock({lock_key}); RETURN true; END $$;
CREATE VIEW {schema}.channel_members AS SELECT cm.* FROM public.channel_members cm WHERE {schema}.gate()")))
.execute(&pool).await.unwrap();
let application = schema.clone();
let path = schema.clone();
let isolated = PgPoolOptions::new()
.max_connections(1)
.after_connect(move |c, _| {
let setup = format!(
"SET search_path TO {path},pg_catalog,public; SET application_name='{path}'"
);
Box::pin(async move {
sqlx::raw_sql(sqlx::AssertSqlSafe(setup)).execute(c).await?;
Ok(())
})
})
.connect(&crate::test_support::database_url())
.await
.unwrap();
let snapshot_db = Db::from_pool(isolated);
let mut blocker = pool.acquire().await.unwrap();
sqlx::query("SELECT pg_advisory_lock($1)")
.bind(lock_key)
.execute(&mut *blocker)
.await
.unwrap();
let reader = snapshot_db.clone();
let pubkey = viewer.public_key();
let pending = tokio::spawn(async move {
reader
.membership_snapshot(community, &pubkey, &all_types())
.await
});
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let waiting: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM pg_stat_activity
WHERE application_name=$1 AND wait_event='advisory')",
)
.bind(&application)
.fetch_one(&pool)
.await
.unwrap();
if waiting {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("production projection must reach statement gate");
let mut tx = pool.begin().await.unwrap();
sqlx::query("UPDATE channel_members SET role='guest' WHERE community_id=$1 AND channel_id=$2")
.bind(community.as_uuid())
.bind(ids[0])
.execute(&mut *tx)
.await
.unwrap();
sqlx::query(
"UPDATE channel_members SET removed_at=now() WHERE community_id=$1 AND channel_id=$2",
)
.bind(community.as_uuid())
.bind(ids[1])
.execute(&mut *tx)
.await
.unwrap();
sqlx::query("UPDATE channels SET archived_at=now() WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid())
.bind(ids[2])
.execute(&mut *tx)
.await
.unwrap();
sqlx::query("UPDATE channels SET deleted_at=now() WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid())
.bind(ids[3])
.execute(&mut *tx)
.await
.unwrap();
tx.commit().await.unwrap();
sqlx::query("SELECT pg_advisory_unlock($1)")
.bind(lock_key)
.execute(&mut *blocker)
.await
.unwrap();
let cut = pending.await.unwrap().unwrap();
assert_eq!(cut.channels, before.channels);
assert_eq!(cut.snapshot_id, before.snapshot_id);
let next = snapshot_db
.membership_snapshot(community, &viewer.public_key(), &all_types())
.await
.unwrap();
assert_eq!(next.channels.len(), 2);
assert_eq!(
next.channels.iter().find(|c| c.id == ids[0]).unwrap().role,
"guest"
);
assert!(
next.channels
.iter()
.find(|c| c.id == ids[2])
.unwrap()
.archived
);
assert!(!next
.channels
.iter()
.any(|c| c.id == ids[1] || c.id == ids[3]));
assert_ne!(next.snapshot_id, cut.snapshot_id);
sqlx::query(sqlx::AssertSqlSafe(format!("DROP SCHEMA {schema} CASCADE")))
.execute(&pool)
.await
.unwrap();
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn public_memberships_exclude_private_unlisted_removed_deleted_and_other_types() {
let (db, pool, community, subject, ids) = fixture(9).await;
for (index, kind) in [
(0, "stream"),
(1, "stream"),
(2, "stream"),
(3, "dm"),
(4, "system"),
(5, "stream"),
(6, "stream"),
(7, "forum"),
(8, "workflow"),
] {
sqlx::query("UPDATE channels SET channel_type=$3::channel_type, visibility=$4::channel_visibility, unlisted=$5, deleted_at=CASE WHEN $6 THEN now() ELSE NULL END, archived_at=CASE WHEN $7 THEN now() ELSE NULL END WHERE community_id=$1 AND id=$2")
.bind(community.as_uuid()).bind(ids[index]).bind(kind)
.bind(if index == 1 {"private"} else {"open"}).bind(index == 2)
.bind(index == 5).bind(index == 7).execute(&pool).await.unwrap();
}
sqlx::query(
"UPDATE channel_members SET removed_at=now() WHERE community_id=$1 AND channel_id=$2",
)
.bind(community.as_uuid())
.bind(ids[6])
.execute(&pool)
.await
.unwrap();
sqlx::query("UPDATE channel_members SET role='guest' WHERE community_id=$1 AND channel_id=$2")
.bind(community.as_uuid())
.bind(ids[7])
.execute(&pool)
.await
.unwrap();
let types = vec!["forum".into(), "stream".into(), "workflow".into()];
let snapshot = db
.public_memberships(community, &subject.public_key(), &types)
.await
.unwrap();
let mut expected = vec![ids[0], ids[7], ids[8]];
expected.sort();
assert_eq!(
snapshot.channels.iter().map(|c| c.id).collect::<Vec<_>>(),
expected
);
assert!(snapshot.channels.iter().all(|c| c.visibility == "open"));
let archived = snapshot.channels.iter().find(|c| c.id == ids[7]).unwrap();
assert_eq!(archived.role, "guest");
assert!(archived.archived);
let identity = serde_json::to_vec(&(
"nip-cl-public-membership-v1",
community.as_uuid(),
subject.public_key().to_hex(),
&types,
&snapshot.channels,
))
.unwrap();
assert_eq!(snapshot.snapshot_id, hex::encode(Sha256::digest(identity)));
assert_ne!(
snapshot.snapshot_id,
db.membership_snapshot(community, &subject.public_key(), &types)
.await
.unwrap()
.snapshot_id
);
assert!(db
.public_memberships(community, &subject.public_key(), &all_types())
.await
.is_err());
assert!(db
.public_memberships(community, &Keys::generate().public_key(), &types)
.await
.unwrap()
.channels
.is_empty());
}

View File

@@ -47,6 +47,8 @@ pub mod feed;
pub mod git_repo;
/// Deployment-wide leader election over session advisory locks.
pub mod leader_lock;
/// Compact viewer membership projections.
pub mod membership_snapshot;
/// Community moderation: reports, bans/timeouts, audit actions.
pub mod moderation;
/// Deployment-global operator-listener mention registrations and delivery queues.

View File

@@ -3,6 +3,7 @@
//! These endpoints provide HTTP access to the relay's Nostr protocol,
//! authenticated via NIP-98 signed events.
mod membership_snapshot;
mod read_state_snapshot;
pub(crate) mod search_slots;
@@ -1619,6 +1620,11 @@ pub async fn query_events(
.await
.enforces();
// This authority surface always needs payload-bound NIP-98, including dev.
// Detect before admission, and before Filter can drop the extension key.
let compact_membership = serde_json::from_slice::<Vec<Value>>(&body)
.is_ok_and(|filters| membership_snapshot::requested(&filters));
// NIP-FI admission. [FI-TRACE-AUTHORITY-UNIFORM]
let proof = crate::nip_fi_shadow::EnforcedProof::new(&state, &headers, "bridge");
crate::nip_fi_shadow::observe_strict_proof(&state, &headers, "bridge", || {
@@ -1633,8 +1639,8 @@ pub async fn query_events(
"POST",
&url,
Some(&body),
state.config.require_auth_token || nip_fi_active,
nip_fi_enforce,
state.config.require_auth_token || nip_fi_active || compact_membership,
nip_fi_enforce || compact_membership,
))
.map(|auth| auth.proof((auth.event_id_bytes, auth.signed_created_at)))
.map_err(|e| e.into_response())
@@ -1713,6 +1719,17 @@ async fn query_member_events(
// depth_limit, feed_types) that nostr::Filter silently drops.
let raw_filters: Vec<Value> = serde_json::from_slice(body)
.map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid filters: {e}")))?;
if membership_snapshot::requested(&raw_filters) {
caller.read_artifact(&raw_filters);
return membership_snapshot::query(
state,
tenant,
&pubkey,
caller.event_id_bytes,
&raw_filters,
)
.await;
}
let routed = || caller.read_artifact(&raw_filters);
if let Some(result) =
super::artifact::query(state, tenant, &pubkey, &raw_filters, false, routed).await

View File

@@ -0,0 +1,249 @@
//! Opt-in NIP-CL compact discovery; ordinary arrays never assert completeness.
use axum::{http::StatusCode, Json};
use buzz_core::{
kind::{KIND_MEMBERSHIP_PUBLIC, KIND_MEMBERSHIP_SNAPSHOT},
TenantContext,
};
use buzz_db::membership_snapshot::{valid_public_types, valid_types, MAX_RESPONSE_BYTES};
use serde::Serialize;
use serde_json::{json, Value};
use crate::{api::api_error, state::AppState};
pub(super) fn requested(filters: &[Value]) -> bool {
filters
.iter()
.any(|f| f.get("membership_snapshot").is_some() || f.get("membership_public").is_some())
}
fn types(
filters: &[Value],
viewer: &nostr::PublicKey,
public: bool,
) -> Result<Vec<String>, &'static str> {
let types: Vec<String> = filters
.first()
.and_then(|f| f.get("#t"))
.and_then(|v| serde_json::from_value(v.clone()).ok())
.filter(|types: &Vec<String>| valid_types(types) && (!public || valid_public_types(types)))
.ok_or("compact membership requires a sorted unique nonempty supported type set")?;
let expected = json!({
"kinds": [if public { KIND_MEMBERSHIP_PUBLIC } else { KIND_MEMBERSHIP_SNAPSHOT }],
"#p": [viewer.to_hex()],
"#t": types,
(if public { "membership_public" } else { "membership_snapshot" }): 1,
});
if filters != [expected] {
return Err(
"compact membership requires one exact version-1 filter for its mode and subject",
);
}
Ok(types)
}
#[derive(Serialize)]
struct Content<'a> {
version: u8,
community_id: uuid::Uuid,
host: &'a str,
viewer: String,
request_id: String,
types: &'a [String],
complete: bool,
snapshot_id: &'a str,
channels: &'a [buzz_db::membership_snapshot::ViewerMembership],
}
pub(super) async fn query(
state: &AppState,
tenant: &TenantContext,
viewer: &nostr::PublicKey,
request_id: [u8; 32],
filters: &[Value],
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
let public = filters.iter().any(|f| f.get("membership_public").is_some());
if public
&& filters
.iter()
.any(|f| f.get("membership_snapshot").is_some())
{
return Err(api_error(
StatusCode::BAD_REQUEST,
"membership modes cannot be mixed",
));
}
for filter in filters {
for (key, kind) in [
("membership_public", KIND_MEMBERSHIP_PUBLIC),
("membership_snapshot", KIND_MEMBERSHIP_SNAPSHOT),
] {
if filter.get(key).is_some() && filter.get("kinds") != Some(&json!([kind])) {
return Err(api_error(
StatusCode::BAD_REQUEST,
"membership mode does not match response kind",
));
}
}
}
let mode = if public {
"membership_public"
} else {
"membership_snapshot"
};
if !(if public {
state.config.membership_public_enabled
} else {
state.config.membership_snapshot_enabled
}) {
return Err(api_error(
StatusCode::NOT_FOUND,
&format!("{mode} unsupported"),
));
}
// Even dev mode must provide real NIP-98 for this new authority surface.
if request_id == [0; 32] {
return Err(api_error(
StatusCode::UNAUTHORIZED,
&format!("{mode} requires NIP-98"),
));
}
let subject = if public {
let subject = filters
.first()
.and_then(|f| f.get("#p"))
.and_then(Value::as_array)
.filter(|p| p.len() == 1)
.and_then(|p| p[0].as_str())
.filter(|p| {
p.len() == 64
&& p.bytes()
.all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
})
.and_then(|p| nostr::PublicKey::from_hex(p).ok())
.ok_or_else(|| {
api_error(
StatusCode::BAD_REQUEST,
"membership_public requires one lowercase subject key",
)
})?;
subject
} else {
*viewer
};
let types = types(filters, &subject, public)
.map_err(|reason| api_error(StatusCode::BAD_REQUEST, reason))?;
let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), async {
if public {
state
.db
.public_memberships(tenant.community(), &subject, &types)
.await
} else {
state
.db
.membership_snapshot(tenant.community(), viewer, &types)
.await
}
})
.await
.map_err(|_| {
api_error(
StatusCode::SERVICE_UNAVAILABLE,
&format!("{mode} timed out"),
)
})?
.map_err(|error| match error {
buzz_db::DbError::MembershipSnapshotTooLarge => api_error(
StatusCode::PAYLOAD_TOO_LARGE,
&format!("{mode} exceeds limit"),
),
_ => {
tracing::error!(mode, error_kind = %error.kind(), "membership projection failed");
api_error(
StatusCode::SERVICE_UNAVAILABLE,
&format!("{mode} unavailable"),
)
}
})?;
let started = std::time::Instant::now();
let content = Content {
version: 1,
community_id: *tenant.community().as_uuid(),
host: tenant.host(),
viewer: viewer.to_hex(),
request_id: hex::encode(request_id),
types: &types,
complete: true,
snapshot_id: &snapshot.snapshot_id,
channels: &snapshot.channels,
};
let serialized = if public {
serde_json::to_string(&json!({
"version": 1, "scope": "public", "community_id": content.community_id,
"host": content.host, "requester": viewer.to_hex(), "subject": subject.to_hex(),
"request_id": content.request_id, "types": content.types, "complete": true,
"snapshot_id": content.snapshot_id, "channels": content.channels,
}))
} else {
serde_json::to_string(&content)
}
.map_err(|_| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("{mode} encoding failed"),
)
})?;
let tags = [
["d".to_string(), content.request_id],
["p".to_string(), subject.to_hex()],
]
.into_iter()
.map(nostr::Tag::parse)
.collect::<Result<Vec<_>, _>>()
.map_err(|_| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("{mode} tags failed"),
)
})?;
let event = nostr::EventBuilder::new(
nostr::Kind::Custom(if public {
KIND_MEMBERSHIP_PUBLIC
} else {
KIND_MEMBERSHIP_SNAPSHOT
} as u16),
serialized,
)
.tags(tags)
.sign_with_keys(&state.relay_keypair)
.map_err(|_| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("{mode} signing failed"),
)
})?;
let envelope = json!({(mode):1, "event":event});
let bytes = serde_json::to_vec(&envelope)
.map_err(|_| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("{mode} encoding failed"),
)
})?
.len();
if bytes > MAX_RESPONSE_BYTES {
return Err(api_error(
StatusCode::PAYLOAD_TOO_LARGE,
&format!("{mode} exceeds limit"),
));
}
metrics::histogram!("buzz_membership_snapshot_encode_sign_seconds", "mode" => mode)
.record(started.elapsed().as_secs_f64());
metrics::histogram!("buzz_membership_snapshot_response_bytes", "mode" => mode)
.record(bytes as f64);
Ok(Json(envelope))
}
#[cfg(test)]
mod postgres_tests;

View File

@@ -0,0 +1,603 @@
use super::*;
use axum::{
body::{to_bytes, Body},
http::Request,
};
use base64::Engine;
use nostr::{EventBuilder, Keys, Kind, Tag};
use sha2::{Digest, Sha256};
use std::sync::Arc;
use tower::ServiceExt;
fn filter(viewer: &Keys) -> Value {
json!([{"kinds":[39008],"#p":[viewer.public_key().to_hex()],
"#t":["dm","forum","stream","system","workflow"],"membership_snapshot":1}])
}
fn proof(key: &Keys, host: &str, body: &Value, payload: bool) -> (String, nostr::Event) {
proof_at(key, host, "/query", body, payload)
}
fn proof_at(
key: &Keys,
host: &str,
path: &str,
body: &Value,
payload: bool,
) -> (String, nostr::Event) {
let mut tags = vec![
Tag::parse(["u", &format!("https://{host}{path}")]).unwrap(),
Tag::parse(["method", "POST"]).unwrap(),
Tag::parse(["nonce", &uuid::Uuid::new_v4().to_string()]).unwrap(),
];
if payload {
tags.push(
Tag::parse([
"payload",
&hex::encode(Sha256::digest(serde_json::to_vec(body).unwrap())),
])
.unwrap(),
);
}
let event = EventBuilder::new(Kind::Custom(27235), "")
.tags(tags)
.sign_with_keys(key)
.unwrap();
(
format!(
"Nostr {}",
base64::engine::general_purpose::STANDARD.encode(serde_json::to_vec(&event).unwrap())
),
event,
)
}
async fn post(
state: Arc<AppState>,
host: &str,
body: Value,
auth: Option<String>,
owner_tag: Option<String>,
) -> (StatusCode, Value) {
let mut request = Request::builder()
.method("POST")
.uri("/query")
.header("host", host);
if let Some(auth) = auth {
request = request.header("authorization", auth);
}
if let Some(tag) = owner_tag {
request = request.header("x-auth-tag", tag);
}
let response = crate::router::build_router(state)
.oneshot(
request
.body(Body::from(serde_json::to_vec(&body).unwrap()))
.unwrap(),
)
.await
.unwrap();
let status = response.status();
let bytes = to_bytes(response.into_body(), 3 * 1024 * 1024)
.await
.unwrap();
(status, serde_json::from_slice(&bytes).unwrap())
}
async fn setup() -> (Arc<AppState>, String, buzz_core::CommunityId, Keys) {
let fixture = super::super::postgres_tests::bridge_handler_test_state()
.await
.expect("isolated Postgres/Redis fixture");
let host = format!("membership-{}.local", uuid::Uuid::new_v4().simple());
let community = fixture
.db
.ensure_configured_community(&host)
.await
.unwrap()
.id;
let mut state = (*fixture).clone();
Arc::make_mut(&mut state.config).membership_snapshot_enabled = true;
state.nip98_replay = Arc::new(buzz_pubsub::RedisNip98ReplayGuard::new(
state.redis_pool.clone(),
));
(Arc::new(state), host, community, Keys::generate())
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn membership_snapshot_router_signature_scope_strict_auth_replay_and_owner_substitution() {
let (state, host, community, viewer) = setup().await;
let channel = state
.db
.create_channel(
community,
"private",
buzz_db::channel::ChannelType::Stream,
buzz_db::channel::ChannelVisibility::Private,
None,
&viewer.public_key().to_bytes(),
None,
)
.await
.unwrap();
let body = filter(&viewer);
assert_eq!(
post(state.clone(), &host, body.clone(), None, None).await.0,
StatusCode::UNAUTHORIZED
);
for (signed_host, payload, submitted) in [
("wrong.local", true, body.clone()),
(host.as_str(), false, body.clone()),
(host.as_str(), true, json!([])),
] {
let (auth, _) = proof(&viewer, signed_host, &body, payload);
assert_eq!(
post(state.clone(), &host, submitted, Some(auth), None)
.await
.0,
StatusCode::UNAUTHORIZED
);
}
let (auth, event) = proof(&viewer, &host, &body, true);
let (status, response) =
post(state.clone(), &host, body.clone(), Some(auth.clone()), None).await;
assert_eq!(status, StatusCode::OK, "{response}");
assert_eq!(response["membership_snapshot"], 1);
let signed: nostr::Event = serde_json::from_value(response["event"].clone()).unwrap();
signed.verify().unwrap();
assert_eq!(signed.pubkey, state.relay_keypair.public_key());
assert_eq!(signed.kind, Kind::Custom(39008));
assert!(signed
.content
.starts_with("{\"version\":1,\"community_id\":"));
let content: Value = serde_json::from_str(&signed.content).unwrap();
assert_eq!(content["request_id"], event.id.to_hex());
assert!(signed
.tags
.iter()
.any(|t| t.as_slice() == ["d", event.id.to_hex().as_str()]));
assert_eq!(content["viewer"], viewer.public_key().to_hex());
assert_eq!(content["community_id"], community.as_uuid().to_string());
assert_eq!(content["host"], host);
assert_eq!(content["types"], body[0]["#t"]);
assert_eq!(content["complete"], true);
assert_eq!(
content["channels"],
json!([{"id":channel.id,"role":"owner","channel_type":"stream","visibility":"private","archived":false}])
);
assert!(serde_json::to_vec(&response).unwrap().len() < 2000);
assert_eq!(
post(state.clone(), &host, body.clone(), Some(auth), None)
.await
.0,
StatusCode::UNAUTHORIZED
);
let (auth, _) = proof(&viewer, &host, &body, true);
assert_eq!(
post(state.clone(), &host, body.clone(), Some(auth), None)
.await
.0,
StatusCode::OK
);
// Real NIP-OA owner admission never changes the projection's viewer.
let agent = Keys::generate();
let mut enforced = (*state).clone();
Arc::make_mut(&mut enforced.config).require_relay_membership = true;
Arc::make_mut(&mut enforced.config).allow_nip_oa_auth = true;
let _ = enforced
.db
.add_relay_member(
community,
&viewer.public_key().to_hex(),
"owner",
None,
&enforced.relay_keypair,
)
.await
.unwrap();
let tag = buzz_sdk::nip_oa::compute_auth_tag(&viewer, &agent.public_key(), "").unwrap();
let enforced = Arc::new(enforced);
let (auth, _) = proof(&agent, &host, &body, true);
assert_eq!(
post(enforced.clone(), &host, body, Some(auth), Some(tag.clone()))
.await
.0,
StatusCode::BAD_REQUEST
);
let own = filter(&agent);
let (auth, _) = proof(&agent, &host, &own, true);
let (status, own_response) = post(enforced, &host, own, Some(auth), Some(tag)).await;
assert_eq!(status, StatusCode::OK, "{own_response}");
let own_content: Value =
serde_json::from_str(own_response["event"]["content"].as_str().unwrap()).unwrap();
assert_eq!(own_content["channels"], json!([]));
assert_eq!(own_content["viewer"], agent.public_key().to_hex());
let other_host = format!("other-{}.local", uuid::Uuid::new_v4().simple());
state
.db
.ensure_configured_community(&other_host)
.await
.unwrap();
let own = filter(&viewer);
let (auth, _) = proof(&viewer, &other_host, &own, true);
let (status, other) = post(state.clone(), &other_host, own, Some(auth), None).await;
assert_eq!(status, StatusCode::OK, "{other}");
let other: Value = serde_json::from_str(other["event"]["content"].as_str().unwrap()).unwrap();
assert_ne!(other["community_id"], content["community_id"]);
assert_eq!(other["channels"], json!([]));
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn membership_snapshot_router_disabled_malformed_and_legacy_roster_compatibility() {
let (state, host, community, viewer) = setup().await;
let capability = crate::nip11::nip11_document(&state, &host)
.await
.unwrap()
.membership_snapshot
.unwrap();
assert_eq!(capability["version"], 1);
assert_eq!(capability["community_id"], community.as_uuid().to_string());
assert_eq!(
capability["max_channels"],
buzz_db::membership_snapshot::MAX_CHANNELS
);
assert_eq!(capability["max_response_bytes"], MAX_RESPONSE_BYTES);
let body = filter(&viewer);
let mut disabled = (*state).clone();
Arc::make_mut(&mut disabled.config).membership_snapshot_enabled = false;
assert!(crate::nip11::nip11_document(&disabled, &host)
.await
.unwrap()
.membership_snapshot
.is_none());
let (auth, _) = proof(&viewer, &host, &body, true);
let (status, response) = post(Arc::new(disabled), &host, body.clone(), Some(auth), None).await;
assert_eq!(status, StatusCode::NOT_FOUND);
assert_eq!(response, json!({"error":"membership_snapshot unsupported"}));
let mut malformed = vec![];
for (key, value) in [
("membership_snapshot", json!(2)),
("#p", json!([Keys::generate().public_key().to_hex()])),
("#t", json!(["system", "stream"])),
("#t", json!(["unknown"])),
("#t", json!([])),
("#t", json!(["dm", "dm"])),
("limit", json!(1)),
("artifact", json!("incremental")),
("read_state_snapshot", json!(1)),
] {
let mut changed = body.clone();
changed[0][key] = value;
malformed.push(changed);
}
malformed.push(json!([body[0], body[0]]));
for bad in malformed {
let (auth, _) = proof(&viewer, &host, &bad, true);
let (status, error) = post(state.clone(), &host, bad, Some(auth), None).await;
assert_eq!(status, StatusCode::BAD_REQUEST, "{error}");
assert!(error.get("event").is_none());
}
let channel = state
.db
.create_channel(
community,
"legacy",
buzz_db::channel::ChannelType::Stream,
buzz_db::channel::ChannelVisibility::Private,
None,
&viewer.public_key().to_bytes(),
None,
)
.await
.unwrap();
// Ordinary legacy reads and full signatures are unchanged on the new relay.
let roster = EventBuilder::new(Kind::Custom(39002), "")
.tags([
Tag::identifier(channel.id.to_string()),
Tag::parse(["t", "stream"]).unwrap(),
Tag::parse(["p", &viewer.public_key().to_hex(), "", "owner"]).unwrap(),
])
.sign_with_keys(&state.relay_keypair)
.unwrap();
state
.db
.replace_addressable_event(community, &roster, Some(channel.id))
.await
.unwrap();
let legacy = json!([{"kinds":[39002],"#p":[viewer.public_key().to_hex()]}]);
let (auth, _) = proof(&viewer, &host, &legacy, true);
let (status, events) = post(state.clone(), &host, legacy, Some(auth), None).await;
assert_eq!(status, StatusCode::OK, "{events}");
assert_eq!(events, json!([roster]));
// The relay-only gate is exercised through the real ingest boundary.
for kind in [39008, 39009] {
let forged = EventBuilder::new(Kind::Custom(kind), "{}")
.tags([Tag::identifier(viewer.public_key().to_hex())])
.sign_with_keys(&viewer)
.unwrap();
let forged = serde_json::to_value(forged).unwrap();
let (auth, _) = proof_at(&viewer, &host, "/events", &forged, true);
let response = crate::router::build_router(state.clone())
.oneshot(
Request::builder()
.method("POST")
.uri("/events")
.header("host", &host)
.header("authorization", auth)
.body(Body::from(serde_json::to_vec(&forged).unwrap()))
.unwrap(),
)
.await
.unwrap();
assert!(!response.status().is_success());
// Ordinary reads never synthesize a compact snapshot (old-relay wire shape).
let ordinary = json!([{"kinds":[kind],"#p":[viewer.public_key().to_hex()]}]);
let (auth, _) = proof(&viewer, &host, &ordinary, true);
let (status, events) = post(state.clone(), &host, ordinary, Some(auth), None).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(events, json!([]));
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn membership_snapshot_router_scale_overflow_and_backend_failure_never_complete_partial() {
for public in [false, true] {
let (state, host, community, viewer) = setup().await;
let pool = sqlx::PgPool::connect(&crate::test_support::database_url())
.await
.unwrap();
let ids: Vec<uuid::Uuid> = (0..4097).map(|_| uuid::Uuid::new_v4()).collect();
// 468 memberships first: the query and signing path must handle an ordinary
// large account without pagination or full roster demand.
sqlx::query(
"INSERT INTO channels (community_id,id,name,created_by)
SELECT $1,id,id::text,$3 FROM unnest($2::uuid[]) id",
)
.bind(community.as_uuid())
.bind(&ids)
.bind(viewer.public_key().to_bytes().to_vec())
.execute(&pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO channel_members (community_id,channel_id,pubkey)
SELECT $1,id,$3 FROM unnest($2::uuid[]) id",
)
.bind(community.as_uuid())
.bind(&ids[..468])
.bind(viewer.public_key().to_bytes().to_vec())
.execute(&pool)
.await
.unwrap();
let mut enabled = (*state).clone();
Arc::make_mut(&mut enabled.config).membership_public_enabled = public;
let state = Arc::new(enabled);
let body = if public {
json!([{"kinds":[39009],"#p":[viewer.public_key().to_hex()],"#t":["stream"],"membership_public":1}])
} else {
filter(&viewer)
};
let started = std::time::Instant::now();
let (auth, _) = proof(&viewer, &host, &body, true);
let (status, response) = post(state.clone(), &host, body.clone(), Some(auth), None).await;
assert_eq!(status, StatusCode::OK, "{response}");
let signed: nostr::Event = serde_json::from_value(response["event"].clone()).unwrap();
signed.verify().unwrap();
let content: Value = serde_json::from_str(&signed.content).unwrap();
assert_eq!(content["channels"].as_array().unwrap().len(), 468);
let bytes = serde_json::to_vec(&response).unwrap().len();
println!(
"468-channel signed HTTP envelope bytes={bytes}, router_ms={}",
started.elapsed().as_secs_f64() * 1000.0
);
assert!(bytes < 100_000);
sqlx::query(
"INSERT INTO channel_members (community_id,channel_id,pubkey)
SELECT $1,id,$3 FROM unnest($2::uuid[]) id",
)
.bind(community.as_uuid())
.bind(&ids[468..])
.bind(viewer.public_key().to_bytes().to_vec())
.execute(&pool)
.await
.unwrap();
let (auth, _) = proof(&viewer, &host, &body, true);
let (status, error) = post(state.clone(), &host, body.clone(), Some(auth), None).await;
assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE, "{error}");
assert!(error.get("event").is_none());
sqlx::query("ALTER TABLE channel_members RENAME TO membership_snapshot_unavailable")
.execute(&pool)
.await
.unwrap();
let (auth, _) = proof(&viewer, &host, &body, true);
let (status, error) = post(state, &host, body, Some(auth), None).await;
sqlx::query("ALTER TABLE membership_snapshot_unavailable RENAME TO channel_members")
.execute(&pool)
.await
.unwrap();
assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{error}");
assert!(error.get("event").is_none());
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn public_memberships_router_bindings_scope_flags_and_strict_filters() {
let (state, host, community, requester) = setup().await;
let subject = Keys::generate();
let channel = state
.db
.create_channel(
community,
"public",
buzz_db::channel::ChannelType::Stream,
buzz_db::channel::ChannelVisibility::Open,
None,
&subject.public_key().to_bytes(),
None,
)
.await
.unwrap();
let body = json!([{"kinds":[39009],"#p":[subject.public_key().to_hex()],"#t":["forum","stream","workflow"],"membership_public":1}]);
assert!(crate::nip11::nip11_document(&state, &host)
.await
.unwrap()
.membership_public
.is_none());
let (auth, _) = proof(&requester, &host, &body, true);
let (status, error) = post(state.clone(), &host, body.clone(), Some(auth), None).await;
assert_eq!(status, StatusCode::NOT_FOUND);
assert_eq!(error, json!({"error":"membership_public unsupported"}));
let mut enabled = (*state).clone();
Arc::make_mut(&mut enabled.config).membership_public_enabled = true;
Arc::make_mut(&mut enabled.config).membership_snapshot_enabled = false;
let state = Arc::new(enabled);
let info = crate::nip11::nip11_document(&state, &host).await.unwrap();
assert!(info.membership_snapshot.is_none());
assert_eq!(info.membership_public.unwrap()["kind"], 39009);
assert!(crate::nip11::nip11_document(&state, "unknown.local")
.await
.is_err());
assert_eq!(
post(state.clone(), &host, body.clone(), None, None).await.0,
StatusCode::UNAUTHORIZED
);
for (signed_host, payload, submitted) in [
("wrong.local", true, body.clone()),
(host.as_str(), false, body.clone()),
(host.as_str(), true, json!([])),
] {
let (auth, _) = proof(&requester, signed_host, &body, payload);
assert_eq!(
post(state.clone(), &host, submitted, Some(auth), None)
.await
.0,
StatusCode::UNAUTHORIZED
);
}
let (auth, event) = proof(&requester, &host, &body, true);
let (status, response) =
post(state.clone(), &host, body.clone(), Some(auth.clone()), None).await;
assert_eq!(status, StatusCode::OK, "{response}");
assert_eq!(response["membership_public"], 1);
assert!(response.get("membership_snapshot").is_none());
let signed: nostr::Event = serde_json::from_value(response["event"].clone()).unwrap();
signed.verify().unwrap();
assert_eq!(signed.kind, Kind::Custom(39009));
assert_eq!(signed.pubkey, state.relay_keypair.public_key());
assert_eq!(
signed
.tags
.iter()
.map(|t| t.as_slice().to_vec())
.collect::<Vec<_>>(),
vec![
vec!["d".to_string(), event.id.to_hex()],
vec!["p".to_string(), subject.public_key().to_hex()]
]
);
let content: Value = serde_json::from_str(&signed.content).unwrap();
assert_eq!(content["scope"], "public");
assert_eq!(content["requester"], requester.public_key().to_hex());
assert_eq!(content["subject"], subject.public_key().to_hex());
assert_eq!(content["request_id"], event.id.to_hex());
assert_eq!(content["community_id"], community.as_uuid().to_string());
assert_eq!(content["host"], host);
assert_eq!(content["types"], body[0]["#t"]);
assert_eq!(content["complete"], true);
assert_eq!(
content["channels"],
json!([{"id":channel.id,"role":"owner","channel_type":"stream","visibility":"open","archived":false}])
);
assert_eq!(
post(state.clone(), &host, body.clone(), Some(auth), None)
.await
.0,
StatusCode::UNAUTHORIZED
);
for (key, value) in [
("kinds", json!([39008])),
("membership_public", json!(2)),
("membership_snapshot", json!(1)),
("#t", json!(["dm", "stream"])),
("#t", json!(["system"])),
("#t", json!(["stream", "forum"])),
("#t", json!(["stream", "stream"])),
("#t", json!([])),
(
"#p",
json!([
requester.public_key().to_hex(),
subject.public_key().to_hex()
]),
),
("#p", json!([subject.public_key().to_hex().to_uppercase()])),
("limit", json!(1)),
] {
let mut bad = body.clone();
bad[0][key] = value;
let (auth, _) = proof(&requester, &host, &bad, true);
assert_eq!(
post(state.clone(), &host, bad.clone(), Some(auth), None)
.await
.0,
StatusCode::BAD_REQUEST,
"{bad}"
);
}
let ordinary = json!([{"kinds":[39009],"#p":[subject.public_key().to_hex()]}]);
let (auth, _) = proof(&requester, &host, &ordinary, true);
assert_eq!(
post(state, &host, ordinary, Some(auth), None).await.1,
json!([])
);
}
#[tokio::test]
#[ignore = "requires Postgres and Redis"]
async fn public_memberships_share_requester_quota_across_subjects_and_fail_closed() {
let (base, host, _, requester) = setup().await;
let mut quota = (*base).clone();
Arc::make_mut(&mut quota.config).membership_public_enabled = true;
let mut config = quota.config.auth.clone();
config.rate_limits.human_api_calls_per_min = 1;
quota.auth = Arc::new(buzz_auth::AuthService::new(config));
let state = Arc::new(quota);
let body = |subject: &Keys| json!([{"kinds":[39009],"#p":[subject.public_key().to_hex()],"#t":["stream"],"membership_public":1}]);
for (subject, expected) in [
(Keys::generate(), StatusCode::OK),
(Keys::generate(), StatusCode::TOO_MANY_REQUESTS),
] {
let body = body(&subject);
let (auth, _) = proof(&requester, &host, &body, true);
let (status, response) = post(state.clone(), &host, body, Some(auth), None).await;
assert_eq!(status, expected, "{response}");
if status == StatusCode::TOO_MANY_REQUESTS {
assert!(response["error"]
.as_str()
.unwrap()
.starts_with("rate-limited: quota exceeded; retry in "));
assert!(response.get("event").is_none());
}
}
let other = Keys::generate();
let body = body(&requester);
let (auth, _) = proof(&other, &host, &body, true);
assert_eq!(
post(state.clone(), &host, body.clone(), Some(auth), None)
.await
.0,
StatusCode::OK
);
let mut unavailable = (*state).clone();
let dead = buzz_pubsub::redis_connection("redis://127.0.0.1:1").unwrap();
unavailable.admission_rate_limiter =
Arc::new(buzz_pubsub::rate_limiter::RedisRateLimiter::new(dead));
let (auth, _) = proof(&Keys::generate(), &host, &body, true);
let (status, error) = post(Arc::new(unavailable), &host, body, Some(auth), None).await;
assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
assert!(error.get("event").is_none());
}

View File

@@ -180,6 +180,10 @@ pub struct Config {
pub require_auth_token: bool,
/// Opt-in private accessory API; disabled until explicitly deployed.
pub buzz_v1_enabled: bool,
/// Opt-in NIP-CL compact membership projection; dark by default.
pub membership_snapshot_enabled: bool,
/// Opt-in NIP-CL public memberships, independent of self discovery.
pub membership_public_enabled: bool,
/// Author-time unread tracking duration, independent of NIP-RS retention.
pub buzz_v1_retention_seconds: u32,
/// Comma-separated list of allowed CORS origins.
@@ -1453,6 +1457,10 @@ impl Config {
auth,
require_auth_token,
buzz_v1_enabled,
membership_snapshot_enabled: std::env::var("BUZZ_MEMBERSHIP_SNAPSHOT_ENABLED")
.is_ok_and(|v| v == "true"),
membership_public_enabled: std::env::var("BUZZ_MEMBERSHIP_PUBLIC_ENABLED")
.is_ok_and(|v| v == "true"),
buzz_v1_retention_seconds,
cors_origins,
relay_private_key,

View File

@@ -37,6 +37,12 @@ pub struct RelayInfo {
/// Host-bound atomic read-state snapshot capability; absent on unresolved hosts.
#[serde(skip_serializing_if = "Option::is_none")]
pub read_state_snapshot: Option<serde_json::Value>,
/// Opt-in host-bound signed compact membership discovery.
#[serde(skip_serializing_if = "Option::is_none")]
pub membership_snapshot: Option<serde_json::Value>,
/// Opt-in host-bound public membership discovery.
#[serde(skip_serializing_if = "Option::is_none")]
pub membership_public: Option<serde_json::Value>,
/// Opt-in, host-bound private accessory API; separate from legacy NIP-RS.
#[serde(skip_serializing_if = "Option::is_none")]
pub buzz_v1: Option<BuzzV1Descriptor>,
@@ -285,6 +291,8 @@ impl RelayInfo {
description: "Buzz — private team communication relay".to_string(),
icon: icon.filter(|s| !s.is_empty()).map(|s| s.to_string()),
read_state_snapshot: None,
membership_snapshot: None,
membership_public: None,
buzz_v1: None,
artifacts: serde_json::json!({
"version": 1, "revision_kind": 45010, "removal_kind": 45011,
@@ -482,6 +490,26 @@ pub(crate) async fn nip11_document(
eligible_kinds: ELIGIBLE_KINDS,
});
}
if state.config.membership_snapshot_enabled {
info.membership_snapshot = Some(serde_json::json!({
"version": 1,
"community_id": tenant.community().as_uuid(),
"pubkey": state.relay_keypair.public_key().to_hex(),
"kind": buzz_core::kind::KIND_MEMBERSHIP_SNAPSHOT,
"max_channels": buzz_db::membership_snapshot::MAX_CHANNELS,
"max_response_bytes": buzz_db::membership_snapshot::MAX_RESPONSE_BYTES,
}));
}
if state.config.membership_public_enabled {
info.membership_public = Some(serde_json::json!({
"version": 1,
"community_id": tenant.community().as_uuid(),
"pubkey": state.relay_keypair.public_key().to_hex(),
"kind": buzz_core::kind::KIND_MEMBERSHIP_PUBLIC,
"max_channels": buzz_db::membership_snapshot::MAX_CHANNELS,
"max_response_bytes": buzz_db::membership_snapshot::MAX_RESPONSE_BYTES,
}));
}
info.read_state_snapshot = Some(serde_json::json!({
"version": 1,
"community_id": tenant.community().as_uuid(),