diff --git a/crates/buzz-db/src/store/operator_listener.rs b/crates/buzz-db/src/store/operator_listener.rs index 18d9c1cca0..e461356930 100644 --- a/crates/buzz-db/src/store/operator_listener.rs +++ b/crates/buzz-db/src/store/operator_listener.rs @@ -267,7 +267,32 @@ pub struct DeliveryAdmission { pub require_allowlist: bool, } -/// Load the event for a claimed delivery if the target may still receive it. +/// The channel a deliverable event belongs to, read with its access check. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DeliveryChannel { + /// Channel id, the event's `h` tag. + pub id: Uuid, + /// Channel name as stored; DMs may have an empty name. + pub name: String, + /// Channel type: `stream`, `forum`, `dm`, `workflow` or `system`. + pub channel_type: String, + /// Channel visibility: `open` or `private`. + pub visibility: String, + /// Channel description, published as the kind:39000 `about` tag. + pub about: Option, +} + +/// An event a claimed delivery may send, with its channel when it has one. +#[derive(Debug, Clone)] +pub struct DeliverableEvent { + /// The stored mentioning event. + pub stored: StoredEvent, + /// The event's channel, or `None` for a channel-less event. + pub channel: Option, +} + +/// Load the event for a claimed delivery, with its channel, if the target +/// may still receive it. /// /// Returns `None` when the registration was removed, the community is no /// longer active, the event was deleted, the target or its recorded owner is @@ -279,7 +304,7 @@ pub async fn load_deliverable_event( pool: &PgPool, delivery: &ClaimedDelivery, admission: DeliveryAdmission, -) -> Result> { +) -> Result> { let mut connection = acquire_maintenance_writer(pool).await?; let row = sqlx::query( "WITH principals AS ( \ @@ -289,11 +314,14 @@ pub async fn load_deliverable_event( WHERE community_id = $3 AND pubkey = $2 AND agent_owner_pubkey IS NOT NULL \ ) \ SELECT e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, e.sig, \ - e.received_at, e.channel_id \ + e.received_at, e.channel_id, c.name AS channel_name, \ + c.channel_type::text AS channel_type, c.visibility::text AS channel_visibility, \ + c.description AS channel_about \ FROM operator_listener_pubkeys r \ JOIN communities co ON co.id = $3 AND co.archived_at IS NULL \ AND co.deleted_at IS NULL AND co.deletion_state = 'active' \ JOIN events e ON e.community_id = $3 AND e.created_at = $4 AND e.id = $5 \ + LEFT JOIN channels c ON c.community_id = e.community_id AND c.id = e.channel_id \ WHERE r.listener_pubkey = $1 AND r.target_pubkey = $2 \ AND e.deleted_at IS NULL \ AND (NOT $8 OR EXISTS ( \ @@ -308,14 +336,12 @@ pub async fn load_deliverable_event( WHERE rm.community_id = $3 AND rm.pubkey IN ( \ SELECT encode(p.pubkey, 'hex') FROM principals p \ WHERE $7 OR p.pubkey = $2))) \ - AND (e.channel_id IS NULL OR 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 = r.target_pubkey AND m.removed_at IS NULL))))", + AND (e.channel_id IS NULL OR ( \ + c.id IS NOT NULL 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 = r.target_pubkey AND m.removed_at IS NULL))))", ) .bind(&delivery.listener_pubkey) .bind(&delivery.target_pubkey) @@ -327,10 +353,25 @@ pub async fn load_deliverable_event( .bind(admission.require_allowlist) .fetch_optional(&mut *connection) .await?; - match row { - Some(row) => crate::event::row_to_stored_event(row), - None => Ok(None), - } + let Some(row) = row else { + return Ok(None); + }; + let channel = match ( + row.try_get::, _>("channel_id")?, + row.try_get::, _>("channel_name")?, + row.try_get::, _>("channel_type")?, + row.try_get::, _>("channel_visibility")?, + ) { + (Some(id), Some(name), Some(channel_type), Some(visibility)) => Some(DeliveryChannel { + id, + name, + channel_type, + visibility, + about: row.try_get("channel_about")?, + }), + _ => None, + }; + Ok(crate::event::row_to_stored_event(row)?.map(|stored| DeliverableEvent { stored, channel })) } /// Release a delivery claim that this pod cannot route. @@ -447,13 +488,13 @@ impl Db { claim_deliveries(&self.pool, limit, lease_until).await } - /// Load a claimed delivery's event if its target may still receive it. + /// Load a claimed delivery's event and channel if its target may still receive it. #[datastore_span(name = "load_operator_listener_delivery_event", system = "postgresql")] pub async fn load_operator_listener_delivery_event( &self, delivery: &ClaimedDelivery, admission: DeliveryAdmission, - ) -> Result> { + ) -> Result> { load_deliverable_event(&self.pool, delivery, admission).await } @@ -1156,14 +1197,16 @@ mod postgres_tests { (other_channel, "private"), ] { sqlx::query( - "INSERT INTO channels (community_id, id, name, visibility, channel_type, created_by) \ - VALUES ($1, $2, $3, $4::channel_visibility, 'stream', $5)", + "INSERT INTO channels \ + (community_id, id, name, visibility, channel_type, created_by, description) \ + VALUES ($1, $2, $3, $4::channel_visibility, 'stream', $5, $6)", ) .bind(community.as_uuid()) .bind(channel) .bind(channel.to_string()) .bind(visibility) .bind(author.public_key().to_bytes().as_slice()) + .bind(format!("about {visibility}")) .execute(&pool) .await .expect("insert test channel"); @@ -1237,20 +1280,35 @@ mod postgres_tests { let open = load(open_channel).await.expect("open channel is readable"); assert_eq!( - open.event.id.as_bytes().as_slice(), + open.stored.event.id.as_bytes().as_slice(), deliveries[&Some(open_channel)].event_id.as_slice() ); - open.event + open.stored + .event .verify() .expect("stored event round-trips with a valid signature"); - assert!(load(member_channel).await.is_some(), "member may read"); - assert!(load(other_channel).await.is_none(), "non-member may not"); - assert!( - load_with(deliveries[&None].clone(), OPEN_RELAY) - .await - .is_some(), - "channel-less event" + assert_eq!( + open.channel, + Some(DeliveryChannel { + id: open_channel, + name: open_channel.to_string(), + channel_type: "stream".to_owned(), + visibility: "open".to_owned(), + about: Some("about open".to_owned()), + }) ); + let member = load(member_channel).await.expect("member may read"); + assert_eq!( + member + .channel + .map(|channel| (channel.id, channel.visibility)), + Some((member_channel, "private".to_owned())) + ); + assert!(load(other_channel).await.is_none(), "non-member may not"); + let channel_less = load_with(deliveries[&None].clone(), OPEN_RELAY) + .await + .expect("channel-less event"); + assert_eq!(channel_less.channel, None); let mut other_community = deliveries[&Some(open_channel)].clone(); other_community.community = CommunityId::from_uuid(Uuid::new_v4()); assert!(load_with(other_community, OPEN_RELAY).await.is_none()); diff --git a/crates/buzz-relay/src/operator_listener.rs b/crates/buzz-relay/src/operator_listener.rs index 8adeb8df57..a0b3db582e 100644 --- a/crates/buzz-relay/src/operator_listener.rs +++ b/crates/buzz-relay/src/operator_listener.rs @@ -18,10 +18,16 @@ const IDLE_POLL_FLOOR: Duration = Duration::from_millis(250); const IDLE_POLL_CEILING: Duration = Duration::from_secs(2); const OUTBOX_REAP_INTERVAL: Duration = Duration::from_secs(5 * 60); const REGISTRATION_REAP_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); +/// Largest channel `about` a notification carries. A larger one omits the +/// whole `channel` object, so the listener reads the channel itself and the +/// body stays within the listener's size cap. +const MAX_NOTIFICATION_ABOUT_BYTES: usize = 16 * 1024; /// Notification sent to the configured operator-listener endpoint. /// -/// Version 2 adds the full signed `event` to version 1's metadata. +/// Version 2 adds the full signed `event` to version 1's metadata. `channel` +/// is an optional version 2 field, omitted for a channel-less event or an +/// oversized `about`. #[derive(Debug, Serialize)] struct MentionNotification<'a> { v: u8, @@ -31,6 +37,36 @@ struct MentionNotification<'a> { event_kind: i32, event_created_at: i64, event: &'a nostr::Event, + #[serde(skip_serializing_if = "Option::is_none")] + channel: Option>, +} + +/// The mentioning event's channel, from the row the delivery access check read. +#[derive(Debug, Serialize)] +struct NotificationChannel<'a> { + id: String, + name: &'a str, + channel_type: &'a str, + visibility: &'a str, + /// Omitted when empty, like the kind:39000 `about` tag. + #[serde(skip_serializing_if = "Option::is_none")] + about: Option<&'a str>, +} + +impl<'a> NotificationChannel<'a> { + fn new(channel: &'a buzz_db::operator_listener::DeliveryChannel) -> Option { + let about = channel.about.as_deref().filter(|about| !about.is_empty()); + if about.is_some_and(|about| about.len() > MAX_NOTIFICATION_ABOUT_BYTES) { + return None; + } + Some(Self { + id: channel.id.to_string(), + name: &channel.name, + channel_type: &channel.channel_type, + visibility: &channel.visibility, + about, + }) + } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -111,7 +147,7 @@ trait DeliveryStore: Sync { async fn load( &self, delivery: &buzz_db::operator_listener::ClaimedDelivery, - ) -> Result, String>; + ) -> Result, String>; async fn release(&self, id: Uuid, claim_id: Uuid, next: DateTime) -> Result; async fn complete(&self, id: Uuid, claim_id: Uuid) -> Result; async fn retry(&self, id: Uuid, claim_id: Uuid, next: DateTime) -> Result; @@ -128,7 +164,7 @@ impl DeliveryStore for DbDeliveryStore<'_> { async fn load( &self, delivery: &buzz_db::operator_listener::ClaimedDelivery, - ) -> Result, String> { + ) -> Result, String> { self.db .load_operator_listener_delivery_event(delivery, self.admission) .await @@ -258,9 +294,11 @@ async fn deliver_one( .increment(1); return; }; - let stored = match store.load(&delivery).await { - Ok(Some(stored)) if event_visible_to_reader(&stored.event, &delivery.target_pubkey) => { - stored + let loaded = match store.load(&delivery).await { + Ok(Some(loaded)) + if event_visible_to_reader(&loaded.stored.event, &delivery.target_pubkey) => + { + loaded } // Retry rather than drop: a mention can commit just before the // target is added to the channel. @@ -280,7 +318,8 @@ async fn deliver_one( event_id: hex::encode(&delivery.event_id), event_kind: delivery.event_kind, event_created_at: delivery.event_created_at.timestamp(), - event: &stored.event, + event: &loaded.stored.event, + channel: loaded.channel.as_ref().and_then(NotificationChannel::new), }) { Ok(body) => body, Err(error) => { @@ -412,7 +451,7 @@ mod tests { struct MockStore { calls: Mutex>, - loaded: Result, String>, + loaded: Result, String>, } impl Default for MockStore { @@ -422,7 +461,9 @@ mod tests { } impl MockStore { - fn loading(loaded: Result, String>) -> Self { + fn loading( + loaded: Result, String>, + ) -> Self { Self { calls: Mutex::default(), loaded, @@ -434,11 +475,14 @@ mod tests { } } - fn stored_event(kind: Kind) -> buzz_core::StoredEvent { + fn stored_event(kind: Kind) -> buzz_db::operator_listener::DeliverableEvent { let event = EventBuilder::new(kind, "hello agent") .sign_with_keys(&Keys::generate()) .expect("sign test event"); - buzz_core::StoredEvent::new(event, None) + buzz_db::operator_listener::DeliverableEvent { + stored: buzz_core::StoredEvent::new(event, None), + channel: None, + } } #[async_trait::async_trait] @@ -446,7 +490,7 @@ mod tests { async fn load( &self, _delivery: &buzz_db::operator_listener::ClaimedDelivery, - ) -> Result, String> { + ) -> Result, String> { self.loaded.clone() } @@ -614,8 +658,16 @@ mod tests { #[tokio::test] async fn successful_delivery_posts_payload_and_completes_claim() { let (routes, relay_keypair, delivery) = delivery_fixture(1); - let loaded = stored_event(Kind::Custom(9)); - let expected = loaded.event.clone(); + let mut loaded = stored_event(Kind::Custom(9)); + let channel = buzz_db::operator_listener::DeliveryChannel { + id: Uuid::new_v4(), + name: "general".to_owned(), + channel_type: "stream".to_owned(), + visibility: "private".to_owned(), + about: Some("Buzz session (buzz.sessions/v1)\n[personal]".to_owned()), + }; + loaded.channel = Some(channel.clone()); + let expected = loaded.stored.event.clone(); let store = MockStore::loading(Ok(Some(loaded))); let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT)); @@ -647,12 +699,67 @@ mod tests { event .verify() .expect("delivered event keeps a valid signature"); + assert_eq!( + body["channel"], + serde_json::json!({ + "id": channel.id.to_string(), + "name": "general", + "channel_type": "stream", + "visibility": "private", + "about": "Buzz session (buzz.sessions/v1)\n[personal]", + }) + ); assert_eq!( store.calls(), [StoreCall::Complete(delivery.id, delivery.claim_id)] ); } + async fn delivered_channel( + channel: Option, + ) -> serde_json::Value { + let (routes, relay_keypair, delivery) = delivery_fixture(1); + let mut loaded = stored_event(Kind::Custom(9)); + loaded.channel = channel; + let store = MockStore::loading(Ok(Some(loaded))); + let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT)); + + deliver_one(&routes, &relay_keypair, &store, &transport, delivery).await; + + let request = transport.take_request().expect("delivery request"); + let body: serde_json::Value = serde_json::from_slice(&request.body).expect("JSON body"); + assert!(body.get("event").is_some()); + body.get("channel") + .cloned() + .unwrap_or(serde_json::Value::Null) + } + + #[tokio::test] + async fn channel_is_omitted_without_a_channel_or_with_an_oversized_about() { + let channel = |about: Option| buzz_db::operator_listener::DeliveryChannel { + id: Uuid::new_v4(), + name: String::new(), + channel_type: "dm".to_owned(), + visibility: "private".to_owned(), + about, + }; + assert_eq!(delivered_channel(None).await, serde_json::Value::Null); + for about in [None, Some(String::new())] { + let delivered = delivered_channel(Some(channel(about))).await; + assert_eq!(delivered["channel_type"], "dm"); + assert!(delivered.get("about").is_none(), "{delivered}"); + } + let largest = "a".repeat(MAX_NOTIFICATION_ABOUT_BYTES); + assert_eq!( + delivered_channel(Some(channel(Some(largest.clone())))).await["about"], + largest + ); + assert_eq!( + delivered_channel(Some(channel(Some(format!("{largest}a"))))).await, + serde_json::Value::Null + ); + } + #[tokio::test] async fn unreadable_event_retries_then_fails_without_http_request() { // Author-only kinds pass the database check but not the reader check.