diff --git a/.env.example b/.env.example index c364f4d3df..9d821ab559 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 8cf397d1a7..56cf748c34 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -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), diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index ce8defd71b..d83c5f38c9 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -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; diff --git a/crates/buzz-db/src/store/membership_snapshot.rs b/crates/buzz-db/src/store/membership_snapshot.rs new file mode 100644 index 0000000000..d0fededffc --- /dev/null +++ b/crates/buzz-db/src/store/membership_snapshot.rs @@ -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, +} + +// 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 { + 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 { + self.project_memberships(community, subject, types, true) + .await + } + + async fn project_memberships( + &self, + community: CommunityId, + viewer: &nostr::PublicKey, + types: &[String], + public: bool, + ) -> Result { + 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::>>()?; + 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; diff --git a/crates/buzz-db/src/store/membership_snapshot/postgres_tests.rs b/crates/buzz-db/src/store/membership_snapshot/postgres_tests.rs new file mode 100644 index 0000000000..eb7dace21a --- /dev/null +++ b/crates/buzz-db/src/store/membership_snapshot/postgres_tests.rs @@ -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) { + 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 = (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 { + 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::>(), + 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()); +} diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 4a16ec05c6..017f1ba4da 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -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. diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index d843e05ecd..131ffef48c 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -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::>(&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 = 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 diff --git a/crates/buzz-relay/src/api/bridge/membership_snapshot.rs b/crates/buzz-relay/src/api/bridge/membership_snapshot.rs new file mode 100644 index 0000000000..f440680b74 --- /dev/null +++ b/crates/buzz-relay/src/api/bridge/membership_snapshot.rs @@ -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, &'static str> { + let types: Vec = filters + .first() + .and_then(|f| f.get("#t")) + .and_then(|v| serde_json::from_value(v.clone()).ok()) + .filter(|types: &Vec| 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, (StatusCode, Json)> { + 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::, _>>() + .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; diff --git a/crates/buzz-relay/src/api/bridge/membership_snapshot/postgres_tests.rs b/crates/buzz-relay/src/api/bridge/membership_snapshot/postgres_tests.rs new file mode 100644 index 0000000000..838581ebcd --- /dev/null +++ b/crates/buzz-relay/src/api/bridge/membership_snapshot/postgres_tests.rs @@ -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, + host: &str, + body: Value, + auth: Option, + owner_tag: Option, +) -> (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, 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 = (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!["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()); +} diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 6aadf72d19..215c1472fc 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -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, diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index c250c2949a..6c1eee3aba 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -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, + /// Opt-in host-bound signed compact membership discovery. + #[serde(skip_serializing_if = "Option::is_none")] + pub membership_snapshot: Option, + /// Opt-in host-bound public membership discovery. + #[serde(skip_serializing_if = "Option::is_none")] + pub membership_public: Option, /// Opt-in, host-bound private accessory API; separate from legacy NIP-RS. #[serde(skip_serializing_if = "Option::is_none")] pub buzz_v1: Option, @@ -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(),