diff --git a/crates/buzz-db/src/store/channel_members.rs b/crates/buzz-db/src/store/channel_members.rs index b92cc4aaa4..95f97bd2eb 100644 --- a/crates/buzz-db/src/store/channel_members.rs +++ b/crates/buzz-db/src/store/channel_members.rs @@ -1321,31 +1321,6 @@ pub async fn channel_type_of( .await?) } -/// The IDs in `channel_ids` that name system channels, deleted or not. -/// Push matching uses it to skip them. -pub async fn system_channel_ids_among( - pool: &PgPool, - community_id: CommunityId, - channel_ids: &[Uuid], -) -> Result> { - 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?) -} - /// The channels in `channel_ids` whose type is not a legacy type /// ([`buzz_core::channel::LEGACY_CHANNEL_TYPES`]), deleted or not, each with /// its type: system channels and any type added later. A filter reads them @@ -2029,16 +2004,6 @@ impl Db { channel_type_of(&self.pool, community_id, channel_id).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> { - system_channel_ids_among(&self.pool, community_id, channel_ids).await - } - /// The channels in `channel_ids` whose type is not a legacy type, deleted /// or not, each with its type. #[datastore_span(name = "named_only_channels_among", system = "postgresql")] diff --git a/crates/buzz-relay/src/handlers/system_channels.rs b/crates/buzz-relay/src/handlers/system_channels.rs index 13ad914fd3..308b87ddce 100644 --- a/crates/buzz-relay/src/handlers/system_channels.rs +++ b/crates/buzz-relay/src/handlers/system_channels.rs @@ -67,6 +67,11 @@ impl NamedOnlyChannels { &self.ids } + /// Whether `id` is a system channel. + pub(crate) fn is_system(&self, id: &Uuid) -> bool { + self.system.contains(id) + } + pub(crate) fn is_empty(&self) -> bool { self.ids.is_empty() } diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index c89284f948..01dd7db06a 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -11,7 +11,14 @@ use nostr::Filter; use serde::{Deserialize, Serialize}; use tracing::{debug, error, warn}; -use crate::{handlers::push_lease::Subscription, nip98::nip98_header, state::AppState}; +use crate::{ + handlers::{ + push_lease::Subscription, + system_channels::{filter_channel_scope, NamedOnlyChannels}, + }, + nip98::nip98_header, + state::AppState, +}; // Dropping a SQLx transaction rolls it back; committed claims remain leased // until expiry. Dropping HTTP cannot recall requests already accepted by the @@ -116,11 +123,13 @@ async fn matcher_loop(state: &AppState) { } } -/// Per-batch state shared by every job: the community's active leases and -/// the exact (channel, lease author) membership pairs the jobs can consult. +/// Per-batch state shared by every job: the community's active leases, the +/// exact (channel, lease author) membership pairs the jobs can consult, and +/// the batch's named-only channels (system channels and any type added later). struct MatchContext { leases: Vec, memberships: std::collections::HashSet<(uuid::Uuid, Vec)>, + named_only: NamedOnlyChannels, } async fn load_match_context( @@ -138,22 +147,22 @@ async fn load_match_context( let mut authors: Vec> = 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(); + let named_only = NamedOnlyChannels::from_typed( + state + .db + .named_only_channels_among(batch.community, &channels) + .await?, + ); Ok(MatchContext { leases, memberships, + named_only, }) } @@ -301,6 +310,7 @@ fn match_job( let filter: Filter = serde_json::from_value(serde_json::Value::Object(sub.filter.clone()))?; if !push_filter_authorized_for_event(&filter, &job.event.event, &author_hex) + || !push_filter_reads_channel(&filter, job.event.channel_id, &context.named_only) || !filters_match(std::slice::from_ref(&filter), &job.event) { continue; @@ -346,6 +356,29 @@ fn match_job( Ok(wakes) } +/// Whether a lease filter may push an event in `channel`, by the same rule as +/// REQ: a legacy-type channel is in every filter's scope (membership is +/// checked separately), and a named-only channel is in scope only when the +/// filter names it (see [`crate::handlers::system_channels`]). Lease filters +/// cannot carry `#t` or `#P`, so for push "names it" means `#h`. The one +/// exception: NIP-CL forbids push for system channels, even when named. +fn push_filter_reads_channel( + filter: &Filter, + channel: Option, + named_only: &NamedOnlyChannels, +) -> bool { + let Some(channel) = channel else { + return true; + }; + if !named_only.ids().contains(&channel) { + return true; + } + if named_only.is_system(&channel) { + return false; + } + filter_channel_scope(filter, &[], named_only).contains(&channel) +} + /// Match-time counterpart of REQ's filter-level `#p` authorization gate. /// Kind 1059 is globally stored and leaks recipient activity through wake /// timing, so a lease may only match gift wraps addressed to its own author. @@ -764,6 +797,90 @@ mod tests { )); } + #[test] + fn named_only_channel_pushes_only_to_a_filter_that_names_it() { + let member = nostr::Keys::generate(); + let outsider = nostr::Keys::generate(); + let sender = nostr::Keys::generate(); + let legacy = uuid::Uuid::new_v4(); + let system = uuid::Uuid::new_v4(); + let future = uuid::Uuid::new_v4(); + let author = |keys: &nostr::Keys| keys.public_key().to_bytes().to_vec(); + let context = |keys: &nostr::Keys, filter: Value| MatchContext { + leases: vec![buzz_db::push::MatchLease { + author: author(keys), + installation_id: "phone".into(), + generation: 1, + subscriptions: serde_json::json!([{"filter": filter, "class": "default"}]), + expires_at: Utc::now().timestamp() + 3600, + }], + memberships: [legacy, system, future] + .into_iter() + .map(|channel| (channel, author(&member))) + .collect(), + named_only: NamedOnlyChannels::from_typed([ + (system, "system".to_owned()), + (future, "later-type".to_owned()), + ]), + }; + let job = |channel: Option| { + let mut event = EventBuilder::new(Kind::Custom(9), "hi"); + if let Some(channel) = channel { + event = event.tag(Tag::parse(["h", &channel.to_string()]).unwrap()); + } + buzz_db::push::BatchedMatch { + event: buzz_core::StoredEvent::new(event.sign_with_keys(&sender).unwrap(), channel), + attempt: 1, + } + }; + let wakes = |keys: &nostr::Keys, filter: Value, channel| { + match_job(&job(channel), &context(keys, filter)) + .unwrap() + .len() + }; + let names = + |channel: uuid::Uuid| serde_json::json!({"kinds": [9], "#h": [channel.to_string()]}); + let broad = serde_json::json!({"kinds": [9]}); + + // A filter that names no channel reads legacy-type channels and + // channel-less events, but no named-only channel. + assert_eq!(wakes(&member, broad.clone(), Some(legacy)), 1, "legacy"); + assert_eq!(wakes(&member, broad.clone(), None), 1, "no channel"); + assert_eq!( + wakes(&member, broad.clone(), Some(system)), + 0, + "unnamed system" + ); + assert_eq!(wakes(&member, broad, Some(future)), 0, "unnamed later type"); + + // Naming a later-type channel by #h reads it, as in REQ. + assert_eq!( + wakes(&member, names(future), Some(future)), + 1, + "named later type" + ); + // NIP-CL: a system channel never pushes, even when named. + assert_eq!( + wakes(&member, names(system), Some(system)), + 0, + "named system" + ); + // Naming one channel does not widen the filter to another. + let other = uuid::Uuid::new_v4(); + assert_eq!(wakes(&member, names(other), Some(future)), 0, "other #h"); + // Membership is still required. + assert_eq!( + wakes(&outsider, names(future), Some(future)), + 0, + "non-member" + ); + assert_eq!( + wakes(&outsider, names(legacy), Some(legacy)), + 0, + "non-member legacy" + ); + } + async fn capture( State(seen): State>>>, OriginalUri(uri): OriginalUri,