jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(relay): leave system channels out of reads that do not name them (#8181)

Browse Source 
🤖
## Summary
After part 1, a system channel is still returned by every query that
returns ordinary channels, so existing clients would show it as a chat.
This PR makes the relay leave system channels out of every read that
does not name them. Clients do not need to filter, so clients that know
nothing about system channels stay correct.

| Query | Before | After |
|---|---|---|
| A query that does not name a channel (no `#h`), COUNT, full-text
search, the HTTP `/query` and `/count` routes, artifact queries
(`{"artifact": ...}`), the sidebar | Includes system channels | Leaves
them out |
| "Channels I belong to" (kind:39002 `#p:[me]`) | Includes them | Leaves
them out |
| A filter that names the channel with `#h`, or with `#d` on kinds
39000–39003 | Ordinary read rules | Unchanged: members can read it, and
anyone can read it if it is open |

Push notifications and "you were added to a channel" notifications
(kinds 44100/44101) skip system channels, because a client that received
one would show the channel.

A client that is given a system channel's ID, for example in a pasted
link, can still open it. Naming the ID is the opt-in. This keeps system
channels ordinary NIP-29 groups.

Part 2 of 3, stacked on part 1.

### Related issue
None found.

### Details
- The rule lives in one place: `get_accessible_channel_ids`, the set of
channels a query may read when it does not name one. It is the union of
the reader's memberships and every open channel; both halves now exclude
`channel_type = 'system'`. Because the filter is in SQL before `LIMIT`,
pages stay full.
- A filter that names a system channel adds it to that filter's scope
only (`handlers/system_channels.rs`). A sibling filter in the same
request keeps the implicit set, so naming one system channel cannot
widen another filter.
- Thread windows and the workflows API name their channel, so they
follow the same named-channel rule.
- Artifact queries have their own read check in SQL. It reads a system
channel only when the query's `#h` names it.
- When a request re-checks membership for a named channel that its first
lookup missed, a confirmed system channel joins only the filters that
name it. Before, a reader who joined a system channel during the request
could have it added to a sibling filter's scope.
- The system-channel lookup runs only when a filter names a channel
outside the implicit set, so ordinary queries do not pay for it.
- **Spec.** NIP-CL lists the reads that leave system channels out, and
says when a filter names one. `AGENTS.md` notes this for `get_channels`.

### Testing
- End-to-end relay test (Postgres and Redis) with a private system
channel, an open system channel and an ordinary channel, as a member and
as a non-member: REQ without `#h`, NIP-50 search, COUNT, HTTP `/query`,
HTTP search, HTTP `/count`, the channel list, and 39002 `#p:[me]` return
only the ordinary channel. Named reads by `#h` and `#d` follow
member-or-open on REQ, COUNT, search and HTTP.
- Relay test: adding a member to a system channel sends no kind:44100;
an ordinary channel still does.
- Both tests fail when the SQL exclusion or the named-channel scope is
removed.
- Database test: an artifact in an open system channel is missing from
`{"artifact":"current"}` and present with `#h:[<system channel>]`, for a
member and a non-member. It fails without the channel-type condition.
- Relay test: the membership re-check confirms a system channel that the
system-channel lookup missed. The channel goes to the filter that names
it, and a sibling filter's scope does not include it. It fails without
the fix.
- Unit tests for the per-filter scope. They now run in `just test-unit`
and `scripts/run-tests.sh`.

### Stack
These are two stacks. Merge them in this order. Client support
(creating, labeling and finding these channels) will be in
block/buzz-app, not the buzz SDK and CLI.

**Stack 1: channel labels and creator lookup (#8174–#8176)**

1. #8174 channel type and creator tags, and the channel creation time
2. #8175 channel labels
3. #8176 database-side label and creator match, and the rule for
choosing one channel

**Stack 2: system channels (#8180–#8182)**, stacked on stack 1 because
it uses labels and the creator tag

4. #8180 the system channel type
5. #8181 leave system channels out of reads that do not name them ← this
PR
6. #8182 channel-state filters opt in to system channels

---------

Signed-off-by: Logan Johnson <loganj@squareup.com>
Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
This commit is contained in:
Logan Johnson
2026-10-15 08:40:15 -04:00
committed by GitHub
parent b77ba86262
commit cdf03f618a
19 changed files with 1220 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
crates/buzz-db/src/store/artifact_postgres_tests.rs
View File

@@ -255,6 +255,48 @@ async fn lifecycle_cas_queries_move_redaction_and_retention() {
f.db.validate_deletion_serving_catalog().await.unwrap();
}
/// An artifact in a system channel is read only by a query whose `#h` names
/// that channel (NIP-CL), even when the channel is open.
/// Mutation: drop the `channel_type` condition from `query_artifacts` → RED.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn system_channel_artifacts_are_read_only_by_name() {
let f = Fixture::new().await;
let system = Uuid::new_v4();
sqlx::query("INSERT INTO channels(community_id,id,name,visibility,channel_type,created_by) VALUES($1,$2,$3,'open','system',$4)")
.bind(f.community.as_uuid())
.bind(system)
.bind(system.to_string())
.bind(f.owner.public_key().to_bytes().as_slice())
.execute(&f.db.pool)
.await
.unwrap();
for home in [f.a, system] {
let create = f.revision(Uuid::new_v4(), "create", home, None, &f.owner, vec![]);
assert!(matches!(
f.accept(&create, None).await,
ArtifactOutcome::Accepted(_)
));
}
for reader in [&f.owner, &f.peer] {
let (events, _) = f
.query(reader, serde_json::json!({"artifact":"current"}), false)
.await;
let homes: Vec<_> = events.iter().map(|e| e.channel_id).collect();
assert_eq!(homes, [Some(f.a)], "no #h");
assert_eq!(
f.count(reader, serde_json::json!({"artifact":"current"}))
.await,
1
);
let named = serde_json::json!({"artifact":"current","#h":[system.to_string()]});
let (events, _) = f.query(reader, named.clone(), false).await;
let homes: Vec<_> = events.iter().map(|e| e.channel_id).collect();
assert_eq!(homes, [Some(system)], "#h names the system channel");
assert_eq!(f.count(reader, named).await, 1);
}
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn repeated_moves_get_distinct_source_safe_markers() {

12
crates/buzz-db/src/store/artifact_query.rs
View File

@@ -36,9 +36,19 @@ impl Db {
if query.view == ArtifactView::Current {
qb.push("JOIN artifact_heads a ON a.community_id=e.community_id AND a.event_id=e.id AND NOT a.deleted ");
}
// A system channel is read only by a query whose `#h` names it
// (NIP-CL); malformed `#h` values name no channel.
let named_channels: Vec<uuid::Uuid> = query
.tags
.iter()
.filter(|(name, _)| name == "h")
.flat_map(|(_, values)| values.iter().filter_map(|value| value.parse().ok()))
.collect();
// The explicit kind lets the community/kind index exclude chat before
// the join, even for an unfiltered current-state page.
qb.push("WHERE e.community_id=").push_bind(community.as_uuid()).push(" AND e.kind=45010 AND e.deleted_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.deleted_at IS NULL AND (c.visibility='open' OR EXISTS (SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id AND m.channel_id=c.id AND m.pubkey=")
qb.push("WHERE e.community_id=").push_bind(community.as_uuid()).push(" AND e.kind=45010 AND e.deleted_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.community_id=e.community_id AND c.id=e.channel_id AND c.deleted_at IS NULL AND (c.channel_type<>'system' OR c.id=ANY(")
.push_bind(named_channels)
.push(")) AND (c.visibility='open' OR EXISTS (SELECT 1 FROM channel_members m WHERE m.community_id=c.community_id AND m.channel_id=c.id AND m.pubkey=")
.push_bind(reader.to_vec()).push(" AND m.removed_at IS NULL)))");
for (name, values) in &query.tags {
// GIN-usable necessary prefilter; retain the positional recheck

93
crates/buzz-db/src/store/channel_members.rs
View File

@@ -1117,6 +1117,10 @@ pub async fn get_members_bulk(
///
/// Includes channels where the pubkey is an active member AND all open channels.
/// Open channels must be included in REQ filter resolution.
///
/// System channels are left out of both halves: this is the implicit scope of
/// queries that do not name a channel. A query that names a system channel,
/// or opts in to them, adds them from [`get_readable_system_channel_ids`].
pub async fn get_accessible_channel_ids(
pool: &PgPool,
community_id: CommunityId,
@@ -1133,10 +1137,12 @@ pub async fn get_accessible_channel_ids(
FROM channel_members cm
JOIN channels c ON cm.community_id = c.community_id AND cm.channel_id = c.id AND c.deleted_at IS NULL
WHERE cm.community_id = $1 AND cm.pubkey = $2 AND cm.removed_at IS NULL
AND c.channel_type <> 'system'
UNION
SELECT id AS channel_id
FROM channels
WHERE community_id = $1 AND visibility = 'open' AND deleted_at IS NULL
AND channel_type <> 'system'
"#,
)
.bind(community_id.as_uuid())
@@ -1152,6 +1158,72 @@ pub async fn get_accessible_channel_ids(
.collect()
}
/// Get the system channels a pubkey can read: those where it is an active
/// member, and every open system channel. Uncached, so a caller can trust it
/// for a channel the request names.
pub async fn get_readable_system_channel_ids(
pool: &PgPool,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Vec<Uuid>> {
let mut connection = crate::observability::acquire_writer(
pool,
crate::observability::WriterOperation::Authorization,
)
.await?;
let rows = sqlx::query(
r#"
SELECT c.id AS channel_id
FROM channels c
WHERE c.community_id = $1 AND c.deleted_at IS NULL AND c.channel_type = 'system'
AND (
c.visibility = 'open'
OR EXISTS (
SELECT 1 FROM channel_members cm
WHERE cm.community_id = c.community_id AND cm.channel_id = c.id
AND cm.pubkey = $2 AND cm.removed_at IS NULL
)
)
"#,
)
.bind(community_id.as_uuid())
.bind(pubkey)
.fetch_all(&mut *connection)
.await?;
rows.into_iter()
.map(|r| {
let id: Uuid = r.try_get("channel_id")?;
Ok(id)
})
.collect()
}
/// The IDs in `channel_ids` that name system channels, deleted or not.
/// Push matching and membership notifications use it to skip them.
pub async fn system_channel_ids_among(
pool: &PgPool,
community_id: CommunityId,
channel_ids: &[Uuid],
) -> Result<Vec<Uuid>> {
if channel_ids.is_empty() {
return Ok(Vec::new());
}
let mut connection = crate::observability::acquire_writer(
pool,
crate::observability::WriterOperation::Authorization,
)
.await?;
Ok(sqlx::query_scalar(
"SELECT id FROM channels \
WHERE community_id = $1 AND id = ANY($2) AND channel_type = 'system'",
)
.bind(community_id.as_uuid())
.bind(channel_ids)
.fetch_all(&mut *connection)
.await?)
}
/// A large channel whose canonical active-member count may need its legacy
/// discovery snapshot repaired.
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -1377,6 +1449,7 @@ pub async fn get_accessible_channels(
WHERE c.community_id = $1 AND c.deleted_at IS NULL
{membership_clause}
AND (c.channel_type != 'dm' OR cm.hidden_at IS NULL)
AND c.channel_type <> 'system'
"#
);
@@ -1782,6 +1855,26 @@ impl Db {
get_accessible_channel_ids(&self.pool, community_id, pubkey).await
}
/// Get the system channels a pubkey can read (member, or open).
#[datastore_span(name = "get_readable_system_channel_ids", system = "postgresql")]
pub async fn get_readable_system_channel_ids(
&self,
community_id: CommunityId,
pubkey: &[u8],
) -> Result<Vec<Uuid>> {
get_readable_system_channel_ids(&self.pool, community_id, pubkey).await
}
/// The IDs in `channel_ids` that name system channels, deleted or not.
#[datastore_span(name = "system_channel_ids_among", system = "postgresql")]
pub async fn system_channel_ids_among(
&self,
community_id: CommunityId,
channel_ids: &[Uuid],
) -> Result<Vec<Uuid>> {
system_channel_ids_among(&self.pool, community_id, channel_ids).await
}
/// Returns large active-channel rosters whose relay-authored snapshots differ.
#[datastore_span(
name = "list_large_channel_rosters_needing_reconciliation",

2
crates/buzz-db/src/store/personal_read/projection.rs
View File

@@ -99,6 +99,8 @@ impl Db {
WHERE cm.community_id=$1 AND cm.pubkey=$2 AND cm.removed_at IS NULL
AND c.deleted_at IS NULL AND ($3::uuid IS NULL OR c.id>$3)
AND ($9::uuid[] IS NULL OR c.id=ANY($9))
-- System channels only when the request names them.
AND ($9::uuid[] IS NOT NULL OR c.channel_type<>'system')
ORDER BY c.id LIMIT $4
)
SELECT r.*, COALESCE(e.latest_id, latest.latest_id) AS latest_id,

111
crates/buzz-relay/src/api/bridge.rs
View File

@@ -1438,12 +1438,26 @@ async fn query_events_authed(
.get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes)
.await
.map_err(|e| internal_error(&format!("channel access lookup: {e}")))?;
// System channels are outside `accessible_channels`; a filter that names
// one reads it in its own scope only (see `handlers::system_channels`).
let mut readable_system =
crate::handlers::system_channels::readable_system_channels_for_filters(
state,
tenant.community(),
&pubkey_bytes,
&filters,
&accessible_channels,
None,
)
.await
.map_err(|e| internal_error(&format!("system channel access lookup: {e}")))?;
repair_requested_channel_access(
state,
tenant,
&filters,
&pubkey_bytes,
&mut accessible_channels,
&mut readable_system,
)
.await?;
@@ -1459,6 +1473,7 @@ async fn query_events_authed(
&raw_filters,
&filters,
&accessible_channels,
&readable_system,
tenant,
&authed_pubkey_hex,
&pubkey_bytes,
@@ -1490,7 +1505,12 @@ async fn query_events_authed(
// Resolve reply owners from retained thread metadata, including tombstones.
for (idx, (raw, filter)) in raw_filters.iter().zip(filters.iter()).enumerate() {
if extension_flag(raw, "resolve_thread_roots") {
events.extend(thread_roots::query(state, tenant, filter, &accessible_channels).await?);
let scope = crate::handlers::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
);
events.extend(thread_roots::query(state, tenant, filter, &scope).await?);
handled.insert(idx);
}
}
@@ -1506,7 +1526,11 @@ async fn query_events_authed(
tenant,
raw,
filter,
&accessible_channels,
&crate::handlers::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
),
&pubkey_bytes,
&mut events,
)
@@ -1628,6 +1652,11 @@ async fn query_events_authed(
Ok(b) if b.len() == 32 => b,
_ => continue,
};
let accessible_channels = crate::handlers::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
);
if let Some(ch_id) = extract_channel_from_filter(filter) {
if !accessible_channels.contains(&ch_id) {
@@ -1717,6 +1746,11 @@ async fn query_events_authed(
if handled.contains(&idx) {
continue;
}
let accessible_channels = crate::handlers::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
);
if let Some(ch_id) = extract_channel_from_filter(filter) {
if !accessible_channels.contains(&ch_id) {
@@ -1818,6 +1852,11 @@ async fn query_events_authed(
// Phase 3 — post-processing, strictly in filter order.
while let Some((idx, filter_events)) = catchall_results.next().await {
let filter = &filters[idx];
let accessible_channels = crate::handlers::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
);
match filter_events {
Ok(stored_events) => {
for se in stored_events {
@@ -1849,13 +1888,15 @@ async fn query_events_authed(
Ok(Json(Value::Array(events)))
}
async fn repair_requested_channel_access(
pub(crate) async fn repair_requested_channel_access(
state: &AppState,
tenant: &TenantContext,
filters: &[nostr::Filter],
pubkey_bytes: &[u8],
accessible_channels: &mut Vec<uuid::Uuid>,
readable_system: &mut Vec<uuid::Uuid>,
) -> Result<(), (StatusCode, Json<Value>)> {
let repaired_from = accessible_channels.len();
for filter in filters {
let Some(requested) =
crate::handlers::req::extract_channel_ids_from_filters(std::slice::from_ref(filter))
@@ -1863,7 +1904,8 @@ async fn repair_requested_channel_access(
continue;
};
for channel_id in requested {
if accessible_channels.contains(&channel_id) {
// A readable system channel stays out of the request-wide set.
if accessible_channels.contains(&channel_id) || readable_system.contains(&channel_id) {
continue;
}
let is_member = state
@@ -1879,7 +1921,15 @@ async fn repair_requested_channel_access(
);
}
}
Ok(())
crate::handlers::system_channels::keep_repaired_system_channels_named(
state,
tenant.community(),
accessible_channels,
repaired_from,
readable_system,
)
.await
.map_err(|e| internal_error(&format!("system channel access lookup: {e}")))
}
/// Count events via HTTP bridge (NIP-98 auth). Returns `{"count": N}`.
@@ -2035,17 +2085,36 @@ async fn count_events_authed(
.get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes)
.await
.map_err(|e| internal_error(&format!("channel access lookup: {e}")))?;
// System channels are outside `accessible_channels`; a filter that names
// one reads it in its own scope only (see `handlers::system_channels`).
let mut readable_system =
crate::handlers::system_channels::readable_system_channels_for_filters(
state,
tenant.community(),
&pubkey_bytes,
&filters,
&accessible_channels,
None,
)
.await
.map_err(|e| internal_error(&format!("system channel access lookup: {e}")))?;
repair_requested_channel_access(
state,
tenant,
&filters,
&pubkey_bytes,
&mut accessible_channels,
&mut readable_system,
)
.await?;
let mut total: u64 = 0;
for (raw, filter) in raw.iter().zip(&filters) {
let accessible_channels = crate::handlers::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
);
let needs_author_only_filtering =
crate::handlers::req::filter_can_match_author_only_kinds(filter);
// Same result-gated guard as the WS COUNT handler: force the per-event
@@ -2270,26 +2339,17 @@ fn search_hit_accepted(
/// Handle search filters by routing to Postgres FTS, then fetching full events
/// from DB. Supports a bridge-only `page` extension over the FTS result set.
#[allow(clippy::too_many_arguments)]
async fn handle_bridge_search(
state: &AppState,
raw_filters: &[Value],
filters: &[nostr::Filter],
accessible_channels: &[uuid::Uuid],
implicit_channels: &[uuid::Uuid],
readable_system: &[uuid::Uuid],
tenant: &buzz_core::tenant::TenantContext,
reader_pubkey_hex: &str,
pubkey_bytes: &[u8],
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
// Bridge always includes global (channel-less) events — same as WS with
// full scopes. `None` means no accessible channels and no global access →
// empty result set (the caller short-circuits exactly as the WS door EOSEs).
let channel_scope = match crate::handlers::req::build_search_channel_scope_filter(
accessible_channels,
true, // include_global
) {
Some(scope) => scope,
None => return Ok(Json(Value::Array(Vec::new()))),
};
let mut events: Vec<Value> = Vec::new();
let mut seen_ids: std::collections::HashSet<[u8; 32]> = std::collections::HashSet::new();
@@ -2305,6 +2365,13 @@ async fn handle_bridge_search(
if limit == 0 {
continue;
}
// Implicit channels plus the readable system channels this filter
// names; never another filter's.
let accessible_channels = &*crate::handlers::system_channels::filter_channel_scope(
filter,
implicit_channels,
readable_system,
);
// Scope by channel — push the #h tag (intersected with accessible
// channels) if present, else the community-wide scope.
@@ -2321,7 +2388,15 @@ async fn handle_bridge_search(
}
buzz_search::ChannelScope::Channels(valid)
} else {
channel_scope.clone()
// Bridge always includes global (channel-less) events — same
// as WS with full scopes.
match crate::handlers::req::build_search_channel_scope_filter(
accessible_channels,
true, // include_global
) {
Some(scope) => scope,
None => continue,
}
};
let kinds = filter.kinds.as_ref().and_then(|ks| {

66
crates/buzz-relay/src/api/bridge/thread_window.rs
View File

@@ -84,6 +84,40 @@ fn append(events: &mut Vec<Value>, budget: &mut Budget, event: &nostr::Event) ->
Ok(())
}
/// The reader's implicit channel set and, when a request names a channel
/// outside it, the system channels it can read. A thread window names its
/// channel, so a readable system channel joins that request's scope only.
async fn access_sets(
state: &AppState,
tenant: &TenantContext,
reader: &nostr::PublicKey,
requests: &[&Request],
context: &str,
) -> Result<(Vec<uuid::Uuid>, Vec<uuid::Uuid>), Error> {
let implicit = state
.db
.get_accessible_channel_ids(tenant.community(), &reader.to_bytes())
.await
.map_err(|e| database_error(context, e))?;
let system = if requests
.iter()
.any(|request| !implicit.contains(&request.channel))
{
state
.db
.get_readable_system_channel_ids(tenant.community(), &reader.to_bytes())
.await
.map_err(|e| database_error(context, e))?
} else {
Vec::new()
};
Ok((implicit, system))
}
fn same_set(a: &[uuid::Uuid], b: &[uuid::Uuid]) -> bool {
a.iter().collect::<HashSet<_>>() == b.iter().collect::<HashSet<_>>()
}
/// Authorize the entire batch against one writer access set, then refresh it
/// once before releasing any output. A later window must never suppress only
/// its own rows while releasing an earlier window built before revocation.
@@ -93,26 +127,28 @@ pub(super) async fn query_batch<'a>(
reader: &nostr::PublicKey,
requests: impl IntoIterator<Item = &'a Request>,
) -> Result<Vec<Value>, Error> {
let accessible = state
.db
.get_accessible_channel_ids(tenant.community(), &reader.to_bytes())
.await
.map_err(|e| database_error("access", e))?;
let requests: Vec<&Request> = requests.into_iter().collect();
let (implicit, system) = access_sets(state, tenant, reader, &requests, "access").await?;
let mut budget = Budget::default();
let mut events = Vec::new();
for request in requests {
if accessible.contains(&request.channel) {
events.extend(query(state, tenant, reader, request, &accessible, &mut budget).await?);
}
for request in &requests {
let accessible: std::borrow::Cow<'_, [uuid::Uuid]> = if implicit.contains(&request.channel)
{
std::borrow::Cow::Borrowed(&implicit)
} else if system.contains(&request.channel) {
let mut scope = implicit.clone();
scope.push(request.channel);
std::borrow::Cow::Owned(scope)
} else {
continue;
};
events.extend(query(state, tenant, reader, request, &accessible, &mut budget).await?);
}
let current = state
.db
.get_accessible_channel_ids(tenant.community(), &reader.to_bytes())
.await
.map_err(|e| database_error("final access", e))?;
let (current_implicit, current_system) =
access_sets(state, tenant, reader, &requests, "final access").await?;
// Grants can expose auxiliary events omitted from the original closure;
// revocations can invalidate earlier windows or their cross-channel aux.
if accessible.iter().collect::<HashSet<_>>() != current.iter().collect::<HashSet<_>>() {
if !same_set(&implicit, &current_implicit) || !same_set(&system, &current_system) {
return Err(unavailable("thread authorization changed; retry query"));
}
Ok(events)

13
crates/buzz-relay/src/api/workflows.rs
View File

@@ -121,7 +121,18 @@ async fn authorize_workflow_read(
.map_err(|error| {
internal_error(&format!("workflow channel access lookup: {error}")).into_response()
})?;
if !accessible.contains(&channel_id) {
// A workflow names its channel, so a readable system channel counts.
let readable = accessible.contains(&channel_id)
|| state
.db
.get_readable_system_channel_ids(tenant.community(), &pubkey_bytes)
.await
.map_err(|error| {
internal_error(&format!("workflow system channel access lookup: {error}"))
.into_response()
})?
.contains(&channel_id);
if !readable {
return Err(not_found());
}

58
crates/buzz-relay/src/handlers/count.rs
View File

@@ -99,6 +99,25 @@ pub async fn handle_count(
if let Some(allowed) = token_channel_ids.as_deref() {
accessible_channels.retain(|channel_id| allowed.contains(channel_id));
}
// System channels are outside `accessible_channels`; a filter that names
// one counts it in its own scope only, as in the WS REQ handler.
let mut readable_system = match super::system_channels::readable_system_channels_for_filters(
&state,
conn.tenant.community(),
&pubkey_bytes,
&filters,
&accessible_channels,
token_channel_ids.as_deref(),
)
.await
{
Ok(ids) => ids,
Err(e) => {
warn!(sub_id = %sub_id, "Failed to get readable system channels: {e}");
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
return;
}
};
// B2: acquire effect permit immediately before the first DB count query.
// The permit is held through all count queries and the COUNT response.
@@ -144,8 +163,9 @@ pub async fn handle_count(
&& !result_gated_count_safe_for_pushdown(filter, &authed_pubkey_hex);
if let Some(requested_channels) = requested_channels {
let repaired_from = accessible_channels.len();
for &ch_id in &requested_channels {
if accessible_channels.contains(&ch_id) {
if accessible_channels.contains(&ch_id) || readable_system.contains(&ch_id) {
continue;
}
let token_allows = token_channel_ids
@@ -174,10 +194,28 @@ pub async fn handle_count(
db_is_member,
);
}
if let Err(e) = super::system_channels::keep_repaired_system_channels_named(
&state,
conn.tenant.community(),
&mut accessible_channels,
repaired_from,
&mut readable_system,
)
.await
{
warn!(sub_id = %sub_id, "Failed to classify repaired channels: {e}");
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
return;
}
let filter_scope = super::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
);
let authorized_requested: Vec<_> = requested_channels
.iter()
.copied()
.filter(|channel_id| accessible_channels.contains(channel_id))
.filter(|channel_id| filter_scope.contains(channel_id))
.collect();
if authorized_requested.is_empty() {
continue;
@@ -195,12 +233,7 @@ pub async fn handle_count(
conn.tenant.community(),
)
.await;
super::req::apply_channel_scope_to_query(
&mut query,
filter,
ch_id,
&accessible_channels,
);
super::req::apply_channel_scope_to_query(&mut query, filter, ch_id, &filter_scope);
// Shared-gated visibility pushdown: pre-filter the fallback
// query_events candidate page before ORDER/LIMIT.
if needs_shared_gate_filtering {
@@ -279,7 +312,14 @@ pub async fn handle_count(
conn.tenant.community(),
)
.await;
query.channel_ids = Some(accessible_channels.to_vec());
query.channel_ids = Some(
super::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
)
.into_owned(),
);
// Shared-gated visibility pushdown for the fallback query_events path.
if needs_shared_gate_filtering {
query.shared_gated_reader = Some(pubkey_bytes.clone());

35
crates/buzz-relay/src/handlers/group_state_postgres_tests.rs
View File

@@ -240,7 +240,7 @@ async fn creator_tag_survives_ownership_transfer() {
}
/// Ingest a kind:9002 that `keys` signs for `channel_id` with `tags`.
async fn edit_channel(
pub(crate) async fn edit_channel(
state: &Arc<AppState>,
tenant: &TenantContext,
keys: &Keys,
@@ -607,15 +607,16 @@ async fn racing_creates_resolve_to_the_same_oldest_channel() {
assert_eq!(pick(vec![&events[1], &events[0]]), Some(expected));
}
/// Run one historical REQ as `reader` and return the IDs of the events sent
/// before EOSE.
async fn req_event_ids(
state: &Arc<AppState>,
/// An authenticated WebSocket connection for `reader`, and the receiver of
/// the frames the relay sends on it.
pub(crate) fn test_conn(
tenant: &TenantContext,
reader: &Keys,
filter: serde_json::Value,
) -> Vec<String> {
let (send_tx, mut send_rx) = tokio::sync::mpsc::channel(256);
) -> (
Arc<crate::connection::ConnectionState>,
tokio::sync::mpsc::Receiver<axum::extract::ws::Message>,
) {
let (send_tx, send_rx) = tokio::sync::mpsc::channel(256);
let (ctrl_tx, _ctrl_rx) = tokio::sync::mpsc::channel(4);
let cancel = tokio_util::sync::CancellationToken::new();
let conn = Arc::new(crate::connection::ConnectionState {
@@ -643,6 +644,18 @@ async fn req_event_ids(
nip_fi_gate: crate::nip_fi_gate::SessionAdmissionGate::off_mode(cancel.clone()),
community_control: crate::state::CommunityConnectionControl::new(cancel.clone()),
});
(conn, send_rx)
}
/// Run one historical REQ as `reader` and return the IDs of the events sent
/// before EOSE.
pub(crate) async fn req_event_ids(
state: &Arc<AppState>,
tenant: &TenantContext,
reader: &Keys,
filter: serde_json::Value,
) -> Vec<String> {
let (conn, mut send_rx) = test_conn(tenant, reader);
let filter: nostr::Filter = serde_json::from_value(filter).expect("filter");
crate::handlers::req::handle_req("g".into(), vec![filter], vec![None], conn, state.clone())
.await;
@@ -662,7 +675,11 @@ async fn req_event_ids(
}
/// The current kind:39000 event ID for `channel_id`.
async fn metadata_event_id(state: &AppState, tenant: &TenantContext, channel_id: Uuid) -> String {
pub(crate) async fn metadata_event_id(
state: &AppState,
tenant: &TenantContext,
channel_id: Uuid,
) -> String {
state
.db
.query_events(&EventQuery {

4
crates/buzz-relay/src/handlers/mod.rs
View File

@@ -42,6 +42,10 @@ pub mod report_resolution;
pub mod req;
/// NIP-29 and NIP-25 side-effect handlers.
pub mod side_effects;
#[cfg(test)]
mod system_channel_postgres_tests;
/// Per-filter query scope for system channels.
pub(crate) mod system_channels;
/// Extract an optional TTL (in seconds) from a Nostr event's `ttl` tag,
/// applying the server-side override when configured.

108
crates/buzz-relay/src/handlers/req.rs
View File

@@ -126,6 +126,25 @@ pub async fn handle_req(
if let Some(allowed) = token_channel_ids.as_deref() {
accessible_channels.retain(|channel_id| allowed.contains(channel_id));
}
// System channels are outside `accessible_channels`. A filter that names
// one adds it to its own scope only (see `system_channels`).
let mut readable_system = match super::system_channels::readable_system_channels_for_filters(
&state,
conn.tenant.community(),
&pubkey_bytes,
&filters,
&accessible_channels,
token_channel_ids.as_deref(),
)
.await
{
Ok(ids) => ids,
Err(e) => {
warn!(conn_id = %conn_id, "Failed to get readable system channels: {e}");
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
return;
}
};
// Build the conformance `AbstractState` once at request entry. The
// `Option` only goes `None` on malformed pubkey bytes (already a
@@ -146,11 +165,17 @@ pub async fn handle_req(
// is what fixes the search false-miss: a `#h=<just-added>` search would
// otherwise be scoped against the stale vector and return empty.
if let Some(requested) = requested_channel_ids.as_ref() {
let repaired_from = accessible_channels.len();
for &ch_id in requested {
let token_allows = token_channel_ids
.as_deref()
.is_none_or(|allowed| allowed.contains(&ch_id));
let db_is_member = if !token_allows || accessible_channels.contains(&ch_id) {
// A readable system channel is authorized for the filters that
// name it; it must not enter the request-wide vector.
let db_is_member = if !token_allows
|| accessible_channels.contains(&ch_id)
|| readable_system.contains(&ch_id)
{
None
} else {
match state
@@ -185,13 +210,28 @@ pub async fn handle_req(
db_is_member,
);
}
if let Err(e) = super::system_channels::keep_repaired_system_channels_named(
&state,
conn.tenant.community(),
&mut accessible_channels,
repaired_from,
&mut readable_system,
)
.await
{
warn!(conn_id = %conn_id, "Failed to classify repaired channels: {e}");
conn.send(RelayMessage::closed(&sub_id, "error: database error"));
return;
}
}
let authorized_requested_channels = requested_channel_ids.as_ref().map(|requested| {
requested
.iter()
.copied()
.filter(|channel_id| accessible_channels.contains(channel_id))
.filter(|channel_id| {
accessible_channels.contains(channel_id) || readable_system.contains(channel_id)
})
.collect::<Vec<_>>()
});
// Partial authorization preserves NIP-01 OR semantics by omitting only
@@ -318,6 +358,7 @@ pub async fn handle_req(
owner,
&filters,
&accessible_channels,
&readable_system,
token_channel_ids.is_none(),
&conn.tenant,
&pubkey_bytes,
@@ -416,7 +457,11 @@ pub async fn handle_req(
&mut params,
filter,
per_filter_channel,
&accessible_channels,
&super::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
),
);
// Shared-gated visibility pushdown: set reader bytes so query_events
// appends the SQL visibility clause before ORDER/LIMIT, preventing
@@ -448,6 +493,11 @@ pub async fn handle_req(
// Phase 3 — post-processing, strictly in filter order.
while let Some((idx, per_filter_channel, filter_events)) = results.next().await {
let filter = &filters[idx];
let filter_scope = super::system_channels::filter_channel_scope(
filter,
&accessible_channels,
&readable_system,
);
let events = match filter_events {
Ok(evs) => evs,
Err(e) => {
@@ -510,7 +560,7 @@ pub async fn handle_req(
}
if let Some(ch_id) = stored.channel_id {
if !accessible_channels.contains(&ch_id) {
if !filter_scope.contains(&ch_id) {
continue;
}
}
@@ -684,7 +734,8 @@ async fn handle_search_req(
sub_id: &str,
owner: u64,
filters: &[Filter],
accessible_channels: &[uuid::Uuid],
implicit_channels: &[uuid::Uuid],
readable_system: &[uuid::Uuid],
include_global: bool,
tenant: &TenantContext,
reader_pubkey_bytes: &[u8],
@@ -692,17 +743,15 @@ async fn handle_search_req(
state: &AppState,
trace_state: Option<&crate::conformance::AbstractState>,
) {
// The community-wide channel scope (no #h tag on the filter). `None` means
// "no accessible channels and no global access" → EOSE, exactly as the
// legacy string-filter helper short-circuited.
let all_channels_scope =
match build_search_channel_scope_filter(accessible_channels, include_global) {
Some(scope) => scope,
None => {
conn.send(RelayMessage::eose(sub_id));
return;
}
};
// `None` means "no accessible channels and no global access" → EOSE,
// exactly as the legacy string-filter helper short-circuited. A filter
// that names a readable system channel can still match it.
if build_search_channel_scope_filter(implicit_channels, include_global).is_none()
&& readable_system.is_empty()
{
conn.send(RelayMessage::eose(sub_id));
return;
}
let mut seen_ids: HashSet<nostr::EventId> = HashSet::new();
@@ -711,6 +760,13 @@ async fn handle_search_req(
Some(s) if !s.is_empty() => s.clone(),
_ => continue,
};
// Implicit channels plus the readable system channels this filter
// names; never another filter's.
let accessible_channels = &*super::system_channels::filter_channel_scope(
filter,
implicit_channels,
readable_system,
);
let limit = filter
.limit
@@ -742,7 +798,11 @@ async fn handle_search_req(
}
buzz_search::ChannelScope::Channels(valid)
} else {
all_channels_scope.clone()
// The community-wide scope (no #h tag on the filter).
match build_search_channel_scope_filter(accessible_channels, include_global) {
Some(scope) => scope,
None => continue,
}
};
let kinds = filter.kinds.as_ref().and_then(|ks| {
@@ -1164,7 +1224,7 @@ fn filter_to_query_params(
.filter(|values| !values.is_empty())
.map(|values| values.iter().map(|v| v.to_string()).collect());
let tag_values = group_state_tag_values(filter, kinds.as_deref());
let tag_values = group_state_tag_values(filter);
EventQuery {
channel_id,
@@ -1195,16 +1255,8 @@ fn filter_to_query_params(
/// The SQL match is exact NIP-01 tag matching (see
/// [`EventQuery::tag_values`]), so no false candidate can take a place in
/// the page. The caller's NIP-01 post-filter still runs on each event.
fn group_state_tag_values(filter: &Filter, kinds: Option<&[i32]>) -> Vec<(String, Vec<String>)> {
let group_state_only = kinds.is_some_and(|ks| {
!ks.is_empty()
&& ks.iter().all(|&k| {
(buzz_core::kind::KIND_NIP29_GROUP_METADATA as i32
..=buzz_core::kind::KIND_NIP29_GROUP_ROLES as i32)
.contains(&k)
})
});
if !group_state_only {
fn group_state_tag_values(filter: &Filter) -> Vec<(String, Vec<String>)> {
if !super::system_channels::filter_is_group_state_only(filter) {
return Vec::new();
}
[

13
crates/buzz-relay/src/handlers/side_effects.rs
View File

@@ -1036,7 +1036,8 @@ pub fn emit_live_thread_summary(
/// Emit a relay-signed membership notification event stored globally (channel_id = None).
///
/// kind:44100 = member added, kind:44101 = member removed.
/// kind:44100 = member added, kind:44101 = member removed. System channels
/// emit none.
/// The p tag addresses the target pubkey; the h tag carries the channel UUID as metadata.
/// Stored with channel_id = None so global subscribers receive it via slow-path fan-out.
pub async fn emit_membership_notification(
@@ -1047,6 +1048,16 @@ pub async fn emit_membership_notification(
actor_pubkey: &[u8],
notification_kind: u32,
) -> anyhow::Result<()> {
// A system channel is left out of clients' channel lists, so it sends no
// membership notifications: a client that receives one would show it.
if !state
.db
.system_channel_ids_among(tenant.community(), &[channel_id])
.await?
.is_empty()
{
return Ok(());
}
let target_hex = hex::encode(target_pubkey);
let actor_hex = hex::encode(actor_pubkey);
let channel_id_str = channel_id.to_string();

445
crates/buzz-relay/src/handlers/system_channel_postgres_tests.rs Normal file
View File

@@ -0,0 +1,445 @@
//! Postgres-backed end-to-end tests for system channels: every read path
//! that does not name a system channel leaves it out, and naming it reads it
//! under the ordinary rules (member, or open channel).
use std::sync::Arc;
use buzz_core::TenantContext;
use buzz_db::event::EventQuery;
use nostr::{EventBuilder, Keys, Kind, Tag};
use serde_json::{json, Value};
use uuid::Uuid;
use super::group_state_postgres_tests::{create_channel, edit_channel, http_auth, test_conn};
use crate::handlers::ingest::ingest_event;
use crate::state::AppState;
struct Fixture {
state: Arc<AppState>,
tenant: TenantContext,
host: String,
}
async fn fixture() -> Fixture {
let state = crate::api::bridge::postgres_tests::bridge_handler_test_state()
.await
.expect("local Postgres and Redis");
let host = format!("system-channels-{}.local", Uuid::new_v4().simple());
let community = state
.db
.ensure_configured_community(&host)
.await
.expect("ensure community")
.id;
let tenant = TenantContext::resolved(community, &host);
Fixture {
state,
tenant,
host,
}
}
/// Ingest an event that `keys` signs and return its ID.
async fn publish(f: &Fixture, keys: &Keys, kind: u16, content: &str, tags: &[&[&str]]) -> String {
let tags: Vec<Tag> = tags
.iter()
.map(|tag| Tag::parse(tag.iter().copied()).unwrap())
.collect();
let event = EventBuilder::new(Kind::Custom(kind), content)
.tags(tags)
.sign_with_keys(keys)
.expect("sign");
let id = event.id.to_hex();
let result = ingest_event(&f.state, &f.tenant, event, http_auth(keys))
.await
.unwrap_or_else(|error| panic!("ingest kind:{kind}: {error:?}"));
assert!(result.accepted, "kind:{kind} refused: {}", result.message);
id
}
/// The current relay-signed group-state event ID of `kind` for `channel`.
async fn group_state_id(f: &Fixture, channel: Uuid, kind: i32) -> String {
f.state
.db
.query_events(&EventQuery {
kinds: Some(vec![kind]),
d_tag: Some(channel.to_string()),
limit: Some(1),
..EventQuery::for_community(f.tenant.community())
})
.await
.expect("query group state")
.first()
.unwrap_or_else(|| panic!("no kind:{kind} for {channel}"))
.event
.id
.to_hex()
}
/// Run one historical REQ and return the event IDs sent before EOSE. A
/// CLOSED reply counts as no events.
async fn req(f: &Fixture, reader: &Keys, filter: Value) -> Vec<String> {
let (conn, mut rx) = test_conn(&f.tenant, reader);
let filter: nostr::Filter = serde_json::from_value(filter).expect("filter");
crate::handlers::req::handle_req("s".into(), vec![filter], vec![None], conn, f.state.clone())
.await;
let mut ids = Vec::new();
while let Ok(frame) = rx.try_recv() {
let axum::extract::ws::Message::Text(text) = frame else {
continue;
};
let message: Value = serde_json::from_str(&text).expect("relay frame");
match message[0].as_str() {
Some("EVENT") => ids.push(message[2]["id"].as_str().expect("id").to_owned()),
Some("EOSE") | Some("CLOSED") => return ids,
other => panic!("unexpected frame {other:?}: {text}"),
}
}
panic!("no EOSE; events so far: {ids:?}");
}
/// Run one WebSocket COUNT and return the count. A CLOSED reply counts as 0.
async fn count(f: &Fixture, reader: &Keys, filter: Value) -> u64 {
let (conn, mut rx) = test_conn(&f.tenant, reader);
let filter: nostr::Filter = serde_json::from_value(filter).expect("filter");
crate::handlers::count::handle_count("c".into(), vec![filter], conn, f.state.clone()).await;
while let Ok(frame) = rx.try_recv() {
let axum::extract::ws::Message::Text(text) = frame else {
continue;
};
let message: Value = serde_json::from_str(&text).expect("relay frame");
match message[0].as_str() {
Some("COUNT") => return message[2]["count"].as_u64().expect("count"),
Some("CLOSED") => return 0,
other => panic!("unexpected frame {other:?}: {text}"),
}
}
panic!("no COUNT reply");
}
/// POST `filters` to an HTTP bridge route as `reader`. Returns the JSON body
/// of a 200, or `None` for a refusal.
async fn bridge(f: &Fixture, reader: &Keys, path: &str, filters: Value) -> Option<Value> {
use axum::body::Body;
use axum::http::{header, Request};
use tower::ServiceExt;
let response = crate::router::build_router(f.state.clone())
.oneshot(
Request::builder()
.method("POST")
.uri(path)
.header(header::HOST, &f.host)
.header("x-pubkey", reader.public_key().to_hex())
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&filters).unwrap()))
.expect("request"),
)
.await
.expect("router");
let status = response.status();
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.expect("body");
if status.is_success() {
Some(serde_json::from_slice(&body).expect("json body"))
} else {
assert!(
status.is_client_error(),
"{path}: {status} {}",
String::from_utf8_lossy(&body)
);
None
}
}
fn bridge_ids(body: Option<Value>) -> Vec<String> {
let Some(body) = body else {
return Vec::new();
};
let events = body
.as_array()
.or_else(|| body["events"].as_array())
.unwrap_or_else(|| panic!("bridge query body: {body}"));
events
.iter()
.map(|event| event["id"].as_str().expect("id").to_owned())
.collect()
}
fn bridge_count(body: Option<Value>) -> u64 {
body.map_or(0, |body| {
body["count"]
.as_u64()
.unwrap_or_else(|| panic!("bridge count body: {body}"))
})
}
fn sorted(mut ids: Vec<String>) -> Vec<String> {
ids.sort();
ids
}
/// The relay leaves system channels out of every read that does not name
/// them, and a read that names one follows the ordinary rules.
/// Mutation: drop `channel_type <> 'system'` from either half of
/// `get_accessible_channel_ids` → RED (implicit reads list them). Drop the
/// named-channel scope → RED (members cannot read their own system channel).
#[tokio::test]
#[ignore = "requires Postgres and Redis"]
async fn system_channels_are_read_only_by_name() {
let f = fixture().await;
let owner = Keys::generate();
let outsider = Keys::generate();
let private = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await;
let open = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await;
edit_channel(
&f.state,
&f.tenant,
&owner,
open,
&[&["visibility", "open"]],
)
.await
.expect("open the system channel");
let ordinary = create_channel(&f.state, &f.tenant, &owner, &[]).await;
edit_channel(
&f.state,
&f.tenant,
&owner,
ordinary,
&[&["visibility", "open"]],
)
.await
.expect("open the ordinary channel");
let word = format!("needle{}", Uuid::new_v4().simple());
let mut message = std::collections::HashMap::new();
for channel in [private, open, ordinary] {
let id = publish(&f, &owner, 9, &word, &[&["h", &channel.to_string()]]).await;
message.insert(channel, id);
}
let ordinary_message = vec![message[&ordinary].clone()];
for reader in [&owner, &outsider] {
let who = if std::ptr::eq(reader, &owner) {
"member"
} else {
"outsider"
};
// Implicit scope: no channel named.
assert_eq!(
req(&f, reader, json!({"kinds": [9]})).await,
ordinary_message,
"{who}: REQ without #h"
);
assert_eq!(
req(&f, reader, json!({"kinds": [9], "search": word})).await,
ordinary_message,
"{who}: NIP-50 search"
);
assert_eq!(
count(&f, reader, json!({"kinds": [9]})).await,
1,
"{who}: COUNT"
);
assert_eq!(
bridge_ids(bridge(&f, reader, "/query", json!([{"kinds": [9]}])).await),
ordinary_message,
"{who}: bridge /query"
);
assert_eq!(
bridge_ids(
bridge(
&f,
reader,
"/query",
json!([{"kinds": [9], "search": word}])
)
.await
),
ordinary_message,
"{who}: bridge search"
);
assert_eq!(
bridge_count(bridge(&f, reader, "/count", json!([{"kinds": [9]}])).await),
1,
"{who}: bridge /count"
);
assert_eq!(
req(&f, reader, json!({"kinds": [39000]})).await,
vec![group_state_id(&f, ordinary, 39000).await],
"{who}: channel list"
);
// Named scope: member, or the channel is open.
for (channel, readable) in [(private, std::ptr::eq(reader, &owner)), (open, true)] {
let h = channel.to_string();
let expected_message = if readable {
vec![message[&channel].clone()]
} else {
Vec::new()
};
let expected_count = u64::from(readable);
assert_eq!(
req(&f, reader, json!({"kinds": [9], "#h": [h]})).await,
expected_message,
"{who}: REQ #h {channel}"
);
assert_eq!(
count(&f, reader, json!({"kinds": [9], "#h": [h]})).await,
expected_count,
"{who}: COUNT #h {channel}"
);
assert_eq!(
bridge_ids(bridge(&f, reader, "/query", json!([{"kinds": [9], "#h": [h]}])).await),
expected_message,
"{who}: bridge /query #h {channel}"
);
assert_eq!(
bridge_count(
bridge(&f, reader, "/count", json!([{"kinds": [9], "#h": [h]}])).await
),
expected_count,
"{who}: bridge /count #h {channel}"
);
assert_eq!(
req(&f, reader, json!({"kinds": [9], "#h": [h], "search": word})).await,
expected_message,
"{who}: search #h {channel}"
);
let expected_metadata = if readable {
vec![group_state_id(&f, channel, 39000).await]
} else {
Vec::new()
};
assert_eq!(
req(&f, reader, json!({"kinds": [39000], "#d": [h]})).await,
expected_metadata,
"{who}: metadata #d {channel}"
);
}
}
// The member's own channel list (kind:39002 #p:[me]) leaves them out.
assert_eq!(
req(
&f,
&owner,
json!({"kinds": [39002], "#p": [owner.public_key().to_hex()]})
)
.await,
vec![group_state_id(&f, ordinary, 39002).await],
"member list by #p"
);
}
/// A reader can join a private system channel after the request loads its
/// readable system channels and before it re-checks a named channel. The
/// re-check then confirms membership. The channel must stay in the scope of
/// the filter that names it, not the request-wide implicit set, so a sibling
/// filter that never names it cannot read it.
/// Mutation: drop `keep_repaired_system_channels_named` from
/// `repair_requested_channel_access` → RED (the sibling filter's scope gets
/// the channel).
#[tokio::test]
#[ignore = "requires Postgres and Redis"]
async fn a_system_channel_joined_mid_request_stays_in_its_named_filter() {
let f = fixture().await;
let owner = Keys::generate();
let system = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await;
let ordinary = create_channel(&f.state, &f.tenant, &owner, &[]).await;
let filters: Vec<nostr::Filter> = serde_json::from_value(json!([
{"kinds": [9], "#h": [system.to_string()]},
{"kinds": [39000]},
]))
.expect("filters");
let mut implicit = f
.state
.db
.get_accessible_channel_ids(f.tenant.community(), &owner.public_key().to_bytes())
.await
.expect("implicit channels");
assert_eq!(implicit, [ordinary]);
// The request's system-channel lookup ran before the join.
let mut readable_system = Vec::new();
crate::api::bridge::repair_requested_channel_access(
&f.state,
&f.tenant,
&filters,
&owner.public_key().to_bytes(),
&mut implicit,
&mut readable_system,
)
.await
.expect("repair");
assert_eq!(implicit, [ordinary], "implicit set stays ordinary");
assert_eq!(readable_system, [system]);
let scope = |filter| {
crate::handlers::system_channels::filter_channel_scope(filter, &implicit, &readable_system)
.into_owned()
};
assert_eq!(scope(&filters[0]), [ordinary, system], "naming filter");
assert_eq!(scope(&filters[1]), [ordinary], "sibling filter");
}
/// Adding a member to a system channel sends no kind:44100 notification,
/// which an existing client would show as a new channel. An ordinary
/// channel still sends one, so the check is not vacuous.
/// Mutation: drop the system check in `emit_membership_notification` → RED.
#[tokio::test]
#[ignore = "requires Postgres and Redis"]
async fn system_channels_send_no_membership_notifications() {
let f = fixture().await;
let owner = Keys::generate();
let member = Keys::generate();
let member_hex = member.public_key().to_hex();
let system = create_channel(&f.state, &f.tenant, &owner, &[&["channel_type", "system"]]).await;
let ordinary = create_channel(&f.state, &f.tenant, &owner, &[]).await;
for channel in [system, ordinary] {
publish(
&f,
&owner,
9000,
"",
&[&["h", &channel.to_string()], &["p", &member_hex]],
)
.await;
}
// Notifications are written after the reply; give the task a moment.
let mut notified = Vec::new();
for _ in 0..50 {
notified = f
.state
.db
.query_events(&EventQuery {
kinds: Some(vec![buzz_core::kind::KIND_MEMBER_ADDED_NOTIFICATION as i32]),
p_tag_hex: Some(member_hex.clone()),
..EventQuery::for_community(f.tenant.community())
})
.await
.expect("query notifications")
.into_iter()
.map(|stored| {
stored
.event
.tags
.iter()
.find(|tag| tag.as_slice().first().map(String::as_str) == Some("h"))
.and_then(|tag| tag.as_slice().get(1).cloned())
.expect("h tag")
})
.collect::<Vec<_>>();
if !notified.is_empty() {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
assert_eq!(sorted(notified), vec![ordinary.to_string()]);
}

262
crates/buzz-relay/src/handlers/system_channels.rs Normal file
View File

@@ -0,0 +1,262 @@
//! Query scope for system channels.
//!
//! A system channel (`channel_type = 'system'`) exists only for access
//! control. The implicit channel set of a query, from
//! `get_accessible_channel_ids`, leaves system channels out. A filter adds
//! the system channels that the reader can read (member, or open) only when
//! it names them:
//!
//! - `#h` lists the channel, on any kinds;
//! - `#d` lists the channel, on a filter whose kinds are all 39000–39003.
//!
//! The addition is per filter. Another filter in the same request keeps the
//! implicit set, so naming one system channel cannot widen a sibling filter.
use std::borrow::Cow;
use nostr::{Alphabet, Filter, SingleLetterTag};
use uuid::Uuid;
use buzz_core::CommunityId;
use crate::state::AppState;
/// Whether every kind in `filter` is a NIP-29 group-state kind (39000–39003).
pub(crate) fn filter_is_group_state_only(filter: &Filter) -> bool {
filter.kinds.as_ref().is_some_and(|kinds| {
!kinds.is_empty()
&& kinds.iter().all(|kind| {
(buzz_core::kind::KIND_NIP29_GROUP_METADATA
..=buzz_core::kind::KIND_NIP29_GROUP_ROLES)
.contains(&u32::from(kind.as_u16()))
})
})
}
/// Channel IDs that `filter` names: its `#h` values, and its `#d` values when
/// the filter asks only for group-state kinds. Malformed values are skipped.
fn named_channels(filter: &Filter) -> impl Iterator<Item = Uuid> + '_ {
let h = filter
.generic_tags
.get(&SingleLetterTag::lowercase(Alphabet::H));
let d = filter_is_group_state_only(filter)
.then(|| {
filter
.generic_tags
.get(&SingleLetterTag::lowercase(Alphabet::D))
})
.flatten();
h.into_iter()
.chain(d)
.flatten()
.filter_map(|value| value.parse::<Uuid>().ok())
}
/// Whether `filter` could need a system channel added to its scope.
fn filter_needs_system_channels(filter: &Filter, implicit: &[Uuid]) -> bool {
named_channels(filter).any(|id| !implicit.contains(&id))
}
/// The channels `filter` may read: the implicit set, plus the readable
/// system channels the filter names.
pub(crate) fn filter_channel_scope<'a>(
filter: &Filter,
implicit: &'a [Uuid],
readable_system: &[Uuid],
) -> Cow<'a, [Uuid]> {
if readable_system.is_empty() {
return Cow::Borrowed(implicit);
}
let mut extra: Vec<Uuid> = Vec::new();
for id in named_channels(filter) {
if readable_system.contains(&id) && !implicit.contains(&id) && !extra.contains(&id) {
extra.push(id);
}
}
if extra.is_empty() {
return Cow::Borrowed(implicit);
}
let mut scope = implicit.to_vec();
scope.extend(extra);
Cow::Owned(scope)
}
/// Load the system channels the reader can read, narrowed to the token's
/// channel scope, when at least one filter could use them. Returns an empty
/// set, without a database read, when no filter names a channel outside the
/// implicit set.
pub(crate) async fn readable_system_channels_for_filters(
state: &AppState,
community: CommunityId,
pubkey: &[u8],
filters: &[Filter],
implicit: &[Uuid],
token_channel_ids: Option<&[Uuid]>,
) -> Result<Vec<Uuid>, buzz_db::DbError> {
if !filters
.iter()
.any(|filter| filter_needs_system_channels(filter, implicit))
{
return Ok(Vec::new());
}
let mut readable = state
.db
.get_readable_system_channel_ids(community, pubkey)
.await?;
if let Some(allowed) = token_channel_ids {
readable.retain(|id| allowed.contains(id));
}
Ok(readable)
}
/// Move system channels out of the repaired tail of the implicit set.
///
/// A request re-checks membership for named channels missing from its first
/// lookups, and [`super::req::resolve_request_local_access`] appends each
/// confirmed one to the implicit set. A system channel lands there when the
/// reader joins it between the lookups and the re-check. This moves it to
/// `readable_system`, so only the filters that name it can read it.
///
/// `repaired_from` is the length of `implicit` before the re-check. The tail
/// after it holds the repaired channels; nothing before it is a system
/// channel.
pub(crate) async fn keep_repaired_system_channels_named(
state: &AppState,
community: CommunityId,
implicit: &mut Vec<Uuid>,
repaired_from: usize,
readable_system: &mut Vec<Uuid>,
) -> Result<(), buzz_db::DbError> {
let Some(repaired) = implicit
.get(repaired_from..)
.filter(|tail| !tail.is_empty())
else {
return Ok(());
};
let system = state
.db
.system_channel_ids_among(community, repaired)
.await?;
move_system_channels(implicit, repaired_from, &system, readable_system);
Ok(())
}
fn move_system_channels(
implicit: &mut Vec<Uuid>,
repaired_from: usize,
system: &[Uuid],
readable_system: &mut Vec<Uuid>,
) {
if system.is_empty() {
return;
}
let repaired = implicit.split_off(repaired_from);
for id in repaired {
if !system.contains(&id) {
implicit.push(id);
} else if !readable_system.contains(&id) {
readable_system.push(id);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn filter(json: serde_json::Value) -> Filter {
serde_json::from_value(json).expect("filter")
}
#[test]
fn named_system_channel_widens_only_its_own_filter() {
let open = Uuid::new_v4();
let system = Uuid::new_v4();
let implicit = vec![open];
let readable = vec![system];
let by_h = filter(serde_json::json!({"#h": [system.to_string()]}));
assert_eq!(
filter_channel_scope(&by_h, &implicit, &readable).as_ref(),
&[open, system]
);
let by_d = filter(serde_json::json!({"kinds": [39000], "#d": [system.to_string()]}));
assert_eq!(
filter_channel_scope(&by_d, &implicit, &readable).as_ref(),
&[open, system]
);
// #d names a channel only on group-state kinds.
let d_other_kind =
filter(serde_json::json!({"kinds": [39000, 1], "#d": [system.to_string()]}));
assert_eq!(
filter_channel_scope(&d_other_kind, &implicit, &readable).as_ref(),
&[open]
);
// A sibling filter that names nothing keeps the implicit set.
let sibling = filter(serde_json::json!({"kinds": [9]}));
assert_eq!(
filter_channel_scope(&sibling, &implicit, &readable).as_ref(),
&[open]
);
// An unreadable system channel is never added.
let unreadable = Uuid::new_v4();
let other = filter(serde_json::json!({"#h": [unreadable.to_string()]}));
assert_eq!(
filter_channel_scope(&other, &implicit, &readable).as_ref(),
&[open]
);
}
#[test]
fn a_repaired_system_channel_stays_in_the_filter_that_names_it() {
let open = Uuid::new_v4();
let repaired_open = Uuid::new_v4();
let joined_system = Uuid::new_v4();
let mut implicit = vec![open, repaired_open, joined_system];
let mut readable = Vec::new();
move_system_channels(&mut implicit, 1, &[joined_system], &mut readable);
assert_eq!(implicit, [open, repaired_open]);
assert_eq!(readable, [joined_system]);
let names_it = filter(serde_json::json!({"#h": [joined_system.to_string()]}));
assert_eq!(
filter_channel_scope(&names_it, &implicit, &readable).as_ref(),
&[open, repaired_open, joined_system]
);
let sibling = filter(serde_json::json!({"kinds": [39000]}));
assert_eq!(
filter_channel_scope(&sibling, &implicit, &readable).as_ref(),
&[open, repaired_open]
);
}
#[test]
fn lookup_runs_only_when_a_filter_names_a_channel_outside_the_implicit_set() {
let open = Uuid::new_v4();
let implicit = vec![open];
assert!(!filter_needs_system_channels(
&filter(serde_json::json!({"kinds": [9]})),
&implicit
));
assert!(!filter_needs_system_channels(
&filter(serde_json::json!({"#h": [open.to_string()]})),
&implicit
));
assert!(filter_needs_system_channels(
&filter(serde_json::json!({"#h": [Uuid::new_v4().to_string()]})),
&implicit
));
assert!(filter_needs_system_channels(
&filter(serde_json::json!({"kinds": [39002], "#d": [Uuid::new_v4().to_string()]})),
&implicit
));
assert!(!filter_needs_system_channels(
&filter(serde_json::json!({"kinds": [1], "#d": [Uuid::new_v4().to_string()]})),
&implicit
));
}
}

7
crates/buzz-relay/src/push_runtime.rs
View File

@@ -138,11 +138,18 @@ async fn load_match_context(
let mut authors: Vec<Vec<u8>> = leases.iter().map(|lease| lease.author.clone()).collect();
authors.sort_unstable();
authors.dedup();
// System channels never push: drop their membership pairs, so a channel
// event there matches no lease (channel events require membership).
let system = state
.db
.system_channel_ids_among(batch.community, &channels)
.await?;
let memberships = state
.db
.membership_pairs(batch.community, &channels, &authors)
.await?
.into_iter()
.filter(|(channel, _)| !system.contains(channel))
.collect();
Ok(MatchContext {
leases,