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

feat(relay): add channel metadata to operator-listener notifications

Browse Source 
Operator listeners (kgoose) read the mention's kind:39000 channel metadata
before they can start a run, to tell a DM from a channel and, for Buzz
sessions, to read the session markers in `about`. The delivery access check
already reads the channel row, so return it and send it with the
notification instead of making the listener ask for it.

v2 notifications gain an optional `channel` object: `id`, `name`,
`channel_type`, `visibility` and `about` (omitted when empty, like the
39000 tag). The whole object is omitted for a channel-less event, or when
`about` exceeds 16 KiB so the body stays within the listener's size cap;
the listener then falls back to reading the channel. The access predicate
is unchanged, moved from an EXISTS subquery to a LEFT JOIN so its columns
can be selected.

Co-authored-by: claude00 <fb8b76a01257932b8ca35060862c29bd09b2137638573731c2d5074fc02c5385@buzz.block.builderlab.xyz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Bradley Axen <baxen@squareup.com>
This commit is contained in:
Bradley Axen
2026-10-14 20:27:50 -07:00
parent b77ba86262
commit d23795b05f
2 changed files with 207 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
crates/buzz-db/src/store/operator_listener.rs
View File

@@ -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<String>,
}
/// 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<DeliveryChannel>,
}
/// 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<Option<StoredEvent>> {
) -> Result<Option<DeliverableEvent>> {
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::<Option<Uuid>, _>("channel_id")?,
row.try_get::<Option<String>, _>("channel_name")?,
row.try_get::<Option<String>, _>("channel_type")?,
row.try_get::<Option<String>, _>("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<Option<StoredEvent>> {
) -> Result<Option<DeliverableEvent>> {
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());

135
crates/buzz-relay/src/operator_listener.rs
View File

@@ -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<NotificationChannel<'a>>,
}
/// 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<Self> {
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<Option<buzz_core::StoredEvent>, String>;
) -> Result<Option<buzz_db::operator_listener::DeliverableEvent>, String>;
async fn release(&self, id: Uuid, claim_id: Uuid, next: DateTime<Utc>) -> Result<bool, String>;
async fn complete(&self, id: Uuid, claim_id: Uuid) -> Result<bool, String>;
async fn retry(&self, id: Uuid, claim_id: Uuid, next: DateTime<Utc>) -> Result<bool, String>;
@@ -128,7 +164,7 @@ impl DeliveryStore for DbDeliveryStore<'_> {
async fn load(
&self,
delivery: &buzz_db::operator_listener::ClaimedDelivery,
) -> Result<Option<buzz_core::StoredEvent>, String> {
) -> Result<Option<buzz_db::operator_listener::DeliverableEvent>, String> {
self.db
.load_operator_listener_delivery_event(delivery, self.admission)
.await
@@ -258,9 +294,11 @@ async fn deliver_one<T: DeliveryTransport, S: DeliveryStore>(
.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<T: DeliveryTransport, S: DeliveryStore>(
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<Vec<StoreCall>>,
loaded: Result<Option<buzz_core::StoredEvent>, String>,
loaded: Result<Option<buzz_db::operator_listener::DeliverableEvent>, String>,
}
impl Default for MockStore {
@@ -422,7 +461,9 @@ mod tests {
}
impl MockStore {
fn loading(loaded: Result<Option<buzz_core::StoredEvent>, String>) -> Self {
fn loading(
loaded: Result<Option<buzz_db::operator_listener::DeliverableEvent>, 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<Option<buzz_core::StoredEvent>, String> {
) -> Result<Option<buzz_db::operator_listener::DeliverableEvent>, 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<buzz_db::operator_listener::DeliveryChannel>,
) -> 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<String>| 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.