feat(relay): deliver full signed events to operator listeners (#8309)

## Summary

Operator-listener mention notifications now carry the full signed event,
so a listener such as kgoose can wake an agent without first fetching
the event (one signed round trip less on time-to-first-reply).

Because the notification now carries message content, two mitigations
come with it, as agreed in #buzz-security:

1. **Each target signs its own registration.** `POST
/operator/listener/pubkeys` takes `{"registrations": [proof, …]}`. Each
proof is a kind `24245` event signed by the target pubkey, with exactly
one `p` tag (the listener pubkey) and one `u` tag
(`{RELAY_OPERATOR_API_ORIGIN}/operator/listener/pubkeys`), and a
`created_at` within ±60s. The old unsigned `{"pubkeys": […]}` POST body
is rejected with 400. `DELETE` is unchanged.
2. **Only notify a target that can read the event.** Before each
delivery the relay re-checks, on the writer, that:
   - the registration still exists;
   - the event isn't deleted;
   - neither the target nor its recorded NIP-OA owner is banned;
- the target passes the relay-membership gate (with the owner path when
`allow_nip_oa_auth` is on);
   - the channel is open, or the target is an active member.

A delivery that fails the check retries with backoff instead of being
dropped, because a mention can commit just before the target is added to
the channel. It fails permanently, with no HTTP request, once attempts
are exhausted.

The payload is now `v: 2`: the v1 fields plus `event`, the full signed
NIP-01 event.

There's no migration. Existing registrations stay in place and age out
under the 30-day retention; only a configured listener could have
created them. Listeners must sign proofs to register new targets, so
receivers have to accept `v: 2`.

Companion kgoose change: squareup/cash-server#136794 signs registrations
with the agent key. The kgoose receiver in squareup/cash-server#136004
currently accepts only `v: 1` and bodies up to 4 KB, so it needs to
accept v2 and a larger body.

## Testing

- Unit tests:
- proof verification (forged signer, wrong kind, stale or future, other
or ambiguous listener, other or missing endpoint);
  - signed-only request bodies;
  - delivery payload carries a verifiable event;
- unreadable or failed loads retry, then fail, without an HTTP request.
- Postgres tests:
- every access predicate (open, member and non-member private,
channel-less, closed community with target or owner membership, owner
ban, removed member, deleted channel or event, removed registration);
- the route test (signed registration accepted; a proof for another
listener and the unsigned body both rejected).
- Mutation runs confirmed that the membership, owner, ban and
reader-visibility guards each fail a test when removed.
- Local relay exercise with a fake listener:
- signed registration returned 200; unsigned, stale and wrong-listener
proofs returned 400;
- open-channel mentions delivered v2 with an event whose id and
signature verify independently;
- a private-channel mention of a non-member retried without delivering,
then delivered once the target was added.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Bradley Axen <baxen@squareup.com>
Co-authored-by: murderbot <3754f8729004d95654c46dbab3129e4ab9ef05cc2534e2a3fbfc155983bd637b@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

.env.example

@@ -151,7 +151,8 @@ BUZZ_PUSH_GATEWAY_DELIVERY_URL=https://push.buzz.xyz/v1/deliveries/apns
 # Deployment-global operator-listener routes. Each configured Nostr identity
 # may register target pubkeys and receives mention notifications at its URL.
 # URLs may use HTTP or HTTPS. Use HTTP only for operator-controlled internal
-# connections: NIP-98 signs requests.
+# connections: NIP-98 signs requests but does not encrypt them, and
+# notifications carry full message content.
 # Format: pubkey:deliveryUrl;pubkey:deliveryUrl
 # BUZZ_OPERATOR_LISTENER_TIMEOUT_MS=5000
 # BUZZ_OPERATOR_LISTENERS=<64-char hex pubkey>:https://listener.example/mentions

crates/buzz-core/src/kind.rs

@@ -79,6 +79,9 @@ pub const KIND_AUTH: u32 = 22242;
 pub const KIND_BLOSSOM_AUTH: u32 = 24242;
 /// Buzz custom one-time identity binding proof (ephemeral, not stored).
 pub const KIND_NOSTR_IDENTITY_BINDING: u32 = 24243;
+/// Operator-listener registration proof: a target pubkey consents to mention
+/// delivery to the listener in its `p` tag (HTTP body only, not stored).
+pub const KIND_OPERATOR_LISTENER_REGISTRATION: u32 = 24245;
 /// NIP-98: HTTP auth event (used in nip98.rs, not stored).
 pub const KIND_HTTP_AUTH: u32 = 27235;
 
@@ -738,6 +741,7 @@ pub const ALL_KINDS: &[u32] = &[
     KIND_AGENT_PRESENCE,
     KIND_HUDDLE_REACTION,
     KIND_BLOSSOM_AUTH,
+    KIND_OPERATOR_LISTENER_REGISTRATION,
     KIND_PAIRING,
     KIND_AGENT_OBSERVER_FRAME,
     KIND_HTTP_AUTH,

crates/buzz-db/src/store/operator_listener.rs

@@ -11,6 +11,7 @@ use uuid::Uuid;
 use crate::error::Result;
 use crate::AdmittedTx;
 use crate::{observability, CommunityId, Db};
+use buzz_core::StoredEvent;
 
 /// Event kinds whose `p` tags represent user-visible message mentions.
 pub const OPERATOR_LISTENER_MENTION_KINDS: [u32; 4] = [9, 40002, 45001, 45003];
@@ -254,6 +255,84 @@ pub async fn claim_deliveries(
         .collect()
 }
 
+/// Community admission policy a delivery target must satisfy, mirroring the
+/// relay's connection gate.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DeliveryAdmission {
+    /// The target (or its owner, below) must be a community member.
+    pub require_membership: bool,
+    /// An agent whose recorded owner is a member counts as a member (NIP-OA).
+    pub allow_owner_membership: bool,
+    /// The target must be in the community's pubkey allowlist.
+    pub require_allowlist: bool,
+}
+
+/// Load the event for a claimed delivery 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
+/// banned, the target fails `admission`,
+/// or the target cannot read the event's channel: the channel was deleted, or
+/// it is private (including DMs) and the target is not an active member. Runs
+/// on the writer, which committed the outbox row with the event.
+pub async fn load_deliverable_event(
+    pool: &PgPool,
+    delivery: &ClaimedDelivery,
+    admission: DeliveryAdmission,
+) -> Result<Option<StoredEvent>> {
+    let mut connection = acquire_maintenance_writer(pool).await?;
+    let row = sqlx::query(
+        "WITH principals AS ( \
+             SELECT $2::bytea AS pubkey \
+             UNION ALL \
+             SELECT agent_owner_pubkey FROM users \
+             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 \
+         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 \
+         WHERE r.listener_pubkey = $1 AND r.target_pubkey = $2 \
+           AND e.deleted_at IS NULL \
+           AND (NOT $8 OR EXISTS ( \
+               SELECT 1 FROM pubkey_allowlist a \
+               WHERE a.community_id = $3 AND a.pubkey = $2)) \
+           AND NOT EXISTS ( \
+               SELECT 1 FROM community_bans b \
+               WHERE b.community_id = $3 AND b.pubkey IN (SELECT pubkey FROM principals) \
+                 AND b.banned AND (b.ban_expires_at IS NULL OR b.ban_expires_at > now())) \
+           AND (NOT $6 OR EXISTS ( \
+               SELECT 1 FROM relay_members rm \
+               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))))",
+    )
+    .bind(&delivery.listener_pubkey)
+    .bind(&delivery.target_pubkey)
+    .bind(delivery.community.as_uuid())
+    .bind(delivery.event_created_at)
+    .bind(&delivery.event_id)
+    .bind(admission.require_membership)
+    .bind(admission.allow_owner_membership)
+    .bind(admission.require_allowlist)
+    .fetch_optional(&mut *connection)
+    .await?;
+    match row {
+        Some(row) => crate::event::row_to_stored_event(row),
+        None => Ok(None),
+    }
+}
+
 /// Release a delivery claim that this pod cannot route.
 pub async fn release_unroutable_delivery(
     pool: &PgPool,
@@ -368,6 +447,16 @@ impl Db {
         claim_deliveries(&self.pool, limit, lease_until).await
     }
 
+    /// Load a claimed delivery's event 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>> {
+        load_deliverable_event(&self.pool, delivery, admission).await
+    }
+
     /// Mark one operator-listener delivery successful.
     #[datastore_span(name = "complete_operator_listener_delivery", system = "postgresql")]
     pub async fn complete_operator_listener_delivery(
@@ -438,6 +527,8 @@ mod tests {
 
 #[cfg(test)]
 mod postgres_tests {
+    use std::collections::HashMap;
+
     use super::*;
     use nostr::{EventBuilder, Keys, Kind, Tag};
     use sqlx::PgPool;
@@ -1035,4 +1126,289 @@ mod postgres_tests {
             .await
             .expect("delete test registrations");
     }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn deliverable_event_requires_registration_live_event_and_channel_access() {
+        let pool = setup_pool().await;
+        let community = CommunityId::from_uuid(Uuid::new_v4());
+        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+            .bind(community.as_uuid())
+            .bind(format!(
+                "operator-listener-access-{}.example",
+                community.as_uuid().simple()
+            ))
+            .execute(&pool)
+            .await
+            .expect("insert test community");
+        let listener = Keys::generate();
+        let target = Keys::generate();
+        let author = Keys::generate();
+        let listener_pubkey = listener.public_key().to_bytes().to_vec();
+        let target_pubkey = target.public_key().to_bytes().to_vec();
+
+        let open_channel = Uuid::new_v4();
+        let member_channel = Uuid::new_v4();
+        let other_channel = Uuid::new_v4();
+        for (channel, visibility) in [
+            (open_channel, "open"),
+            (member_channel, "private"),
+            (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)",
+            )
+            .bind(community.as_uuid())
+            .bind(channel)
+            .bind(channel.to_string())
+            .bind(visibility)
+            .bind(author.public_key().to_bytes().as_slice())
+            .execute(&pool)
+            .await
+            .expect("insert test channel");
+        }
+        sqlx::query(
+            "INSERT INTO channel_members (community_id, channel_id, pubkey, role) \
+             VALUES ($1, $2, $3, 'member')",
+        )
+        .bind(community.as_uuid())
+        .bind(member_channel)
+        .bind(&target_pubkey)
+        .execute(&pool)
+        .await
+        .expect("insert target membership");
+        register_pubkeys(
+            &pool,
+            &listener_pubkey,
+            std::slice::from_ref(&target_pubkey),
+        )
+        .await
+        .expect("register target");
+
+        let mut deliveries = HashMap::new();
+        for channel in [
+            Some(open_channel),
+            Some(member_channel),
+            Some(other_channel),
+            None,
+        ] {
+            let event = EventBuilder::new(Kind::Custom(9), format!("hello {channel:?}"))
+                .tag(Tag::public_key(target.public_key()))
+                .sign_with_keys(&author)
+                .expect("sign test event");
+            crate::event::insert_event(&pool, community, &event, channel)
+                .await
+                .expect("insert event");
+            deliveries.insert(
+                channel,
+                ClaimedDelivery {
+                    id: Uuid::new_v4(),
+                    claim_id: Uuid::new_v4(),
+                    listener_pubkey: listener_pubkey.clone(),
+                    target_pubkey: target_pubkey.clone(),
+                    community,
+                    community_host: String::new(),
+                    event_id: event.id.as_bytes().to_vec(),
+                    event_kind: 9,
+                    event_created_at: DateTime::from_timestamp(
+                        event.created_at.as_secs() as i64,
+                        0,
+                    )
+                    .expect("event timestamp"),
+                    attempt: 1,
+                },
+            );
+        }
+        const OPEN_RELAY: DeliveryAdmission = DeliveryAdmission {
+            require_membership: false,
+            allow_owner_membership: false,
+            require_allowlist: false,
+        };
+        let load_with = |delivery: ClaimedDelivery, admission: DeliveryAdmission| {
+            let pool = pool.clone();
+            async move {
+                load_deliverable_event(&pool, &delivery, admission)
+                    .await
+                    .expect("load deliverable event")
+            }
+        };
+        let load = |channel: Uuid| load_with(deliveries[&Some(channel)].clone(), OPEN_RELAY);
+
+        let open = load(open_channel).await.expect("open channel is readable");
+        assert_eq!(
+            open.event.id.as_bytes().as_slice(),
+            deliveries[&Some(open_channel)].event_id.as_slice()
+        );
+        open.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"
+        );
+        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());
+        let mut other_timestamp = deliveries[&Some(open_channel)].clone();
+        other_timestamp.event_created_at += TimeDelta::seconds(1);
+        assert!(load_with(other_timestamp, OPEN_RELAY).await.is_none());
+
+        // Closed community: direct membership, or a member owner when NIP-OA is on.
+        let closed = DeliveryAdmission {
+            require_membership: true,
+            ..OPEN_RELAY
+        };
+        let closed_with_owner = DeliveryAdmission {
+            allow_owner_membership: true,
+            ..closed
+        };
+        let open_delivery = deliveries[&Some(open_channel)].clone();
+        assert!(load_with(open_delivery.clone(), closed).await.is_none());
+        let owner = Keys::generate();
+        sqlx::query(
+            "INSERT INTO users (community_id, pubkey, agent_owner_pubkey) \
+             VALUES ($1, $3, NULL), ($1, $2, $3)",
+        )
+        .bind(community.as_uuid())
+        .bind(&target_pubkey)
+        .bind(owner.public_key().to_bytes().as_slice())
+        .execute(&pool)
+        .await
+        .expect("record target owner");
+        sqlx::query(
+            "INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'member')",
+        )
+        .bind(community.as_uuid())
+        .bind(owner.public_key().to_hex())
+        .execute(&pool)
+        .await
+        .expect("add owner membership");
+        assert!(load_with(open_delivery.clone(), closed).await.is_none());
+        assert!(load_with(open_delivery.clone(), closed_with_owner)
+            .await
+            .is_some());
+        sqlx::query(
+            "INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'member')",
+        )
+        .bind(community.as_uuid())
+        .bind(target.public_key().to_hex())
+        .execute(&pool)
+        .await
+        .expect("add target membership");
+        assert!(load_with(open_delivery.clone(), closed).await.is_some());
+
+        // A ban on the target's owner cascades to the target.
+        sqlx::query(
+            "INSERT INTO community_bans (community_id, pubkey, banned, actor_pubkey) \
+             VALUES ($1, $2, true, $2)",
+        )
+        .bind(community.as_uuid())
+        .bind(owner.public_key().to_bytes().as_slice())
+        .execute(&pool)
+        .await
+        .expect("ban target owner");
+        assert!(load_with(open_delivery.clone(), OPEN_RELAY).await.is_none());
+        sqlx::query("DELETE FROM community_bans WHERE community_id = $1")
+            .bind(community.as_uuid())
+            .execute(&pool)
+            .await
+            .expect("lift ban");
+        assert!(load_with(open_delivery.clone(), OPEN_RELAY).await.is_some());
+
+        sqlx::query(
+            "UPDATE channel_members SET removed_at = now() \
+             WHERE community_id = $1 AND channel_id = $2 AND pubkey = $3",
+        )
+        .bind(community.as_uuid())
+        .bind(member_channel)
+        .bind(&target_pubkey)
+        .execute(&pool)
+        .await
+        .expect("remove target membership");
+        assert!(load(member_channel).await.is_none(), "removed member");
+
+        sqlx::query("UPDATE channels SET deleted_at = now() WHERE community_id = $1 AND id = $2")
+            .bind(community.as_uuid())
+            .bind(open_channel)
+            .execute(&pool)
+            .await
+            .expect("delete open channel");
+        assert!(load(open_channel).await.is_none(), "deleted channel");
+        sqlx::query("UPDATE channels SET deleted_at = NULL WHERE community_id = $1 AND id = $2")
+            .bind(community.as_uuid())
+            .bind(open_channel)
+            .execute(&pool)
+            .await
+            .expect("restore open channel");
+
+        sqlx::query("UPDATE events SET deleted_at = now() WHERE community_id = $1 AND id = $2")
+            .bind(community.as_uuid())
+            .bind(&deliveries[&Some(open_channel)].event_id)
+            .execute(&pool)
+            .await
+            .expect("delete open event");
+        assert!(load(open_channel).await.is_none(), "deleted event");
+
+        sqlx::query("UPDATE events SET deleted_at = NULL WHERE community_id = $1 AND id = $2")
+            .bind(community.as_uuid())
+            .bind(&deliveries[&Some(open_channel)].event_id)
+            .execute(&pool)
+            .await
+            .expect("restore open event");
+
+        // The allowlist gate requires the target itself to be allowlisted.
+        let allowlisted = DeliveryAdmission {
+            require_allowlist: true,
+            ..OPEN_RELAY
+        };
+        assert!(load_with(open_delivery.clone(), allowlisted)
+            .await
+            .is_none());
+        sqlx::query("INSERT INTO pubkey_allowlist (community_id, pubkey) VALUES ($1, $2)")
+            .bind(community.as_uuid())
+            .bind(&target_pubkey)
+            .execute(&pool)
+            .await
+            .expect("allowlist target");
+        assert!(load_with(open_delivery.clone(), allowlisted)
+            .await
+            .is_some());
+
+        // An archived community delivers nothing. Deletion is one-way, so
+        // only the archive predicate is exercised here.
+        for (statement, expected) in [
+            (
+                "UPDATE communities SET archived_at = now() WHERE id = $1",
+                false,
+            ),
+            (
+                "UPDATE communities SET archived_at = NULL WHERE id = $1",
+                true,
+            ),
+        ] {
+            sqlx::query(statement)
+                .bind(community.as_uuid())
+                .execute(&pool)
+                .await
+                .expect("update community archive state");
+            assert_eq!(
+                load_with(open_delivery.clone(), OPEN_RELAY).await.is_some(),
+                expected,
+                "{statement}"
+            );
+        }
+        remove_pubkeys(
+            &pool,
+            &listener_pubkey,
+            std::slice::from_ref(&target_pubkey),
+        )
+        .await
+        .expect("remove registration");
+        assert!(load(open_channel).await.is_none(), "removed registration");
+    }
 }

crates/buzz-relay/src/api/operator.rs

@@ -70,6 +70,17 @@ struct ListenerPubkeysRequest {
     pubkeys: Vec<String>,
 }
 
+/// Registration body: one proof signed by each target pubkey.
+#[derive(Debug, Deserialize)]
+struct ListenerRegistrationRequest {
+    registrations: Vec<nostr::Event>,
+}
+
+const MAX_LISTENER_PUBKEYS: usize = 1_000;
+const LISTENER_PUBKEYS_PATH: &str = "/operator/listener/pubkeys";
+/// Matches the NIP-98 request window: proofs are signed just before sending.
+const REGISTRATION_PROOF_MAX_SKEW_SECS: u64 = 60;
+
 fn parse_listener_pubkeys(body: &[u8]) -> Result<Vec<Vec<u8>>, (StatusCode, Json<Value>)> {
     let request: ListenerPubkeysRequest = serde_json::from_slice(body).map_err(|e| {
         api_error(
@@ -77,7 +88,7 @@ fn parse_listener_pubkeys(body: &[u8]) -> Result<Vec<Vec<u8>>, (StatusCode, Json
             &format!("invalid operator-listener pubkeys JSON: {e}"),
         )
     })?;
-    if request.pubkeys.is_empty() || request.pubkeys.len() > 1_000 {
+    if request.pubkeys.is_empty() || request.pubkeys.len() > MAX_LISTENER_PUBKEYS {
         return Err(api_error(
             StatusCode::BAD_REQUEST,
             "pubkeys must contain between 1 and 1000 entries",
@@ -103,6 +114,75 @@ fn parse_listener_pubkeys(body: &[u8]) -> Result<Vec<Vec<u8>>, (StatusCode, Json
         .collect()
 }
 
+/// Parse a registration body and return the target pubkeys whose proofs
+/// verify for `listener` at `url`.
+fn parse_listener_registrations(
+    body: &[u8],
+    listener: &nostr::PublicKey,
+    url: &str,
+    now: u64,
+) -> Result<Vec<Vec<u8>>, (StatusCode, Json<Value>)> {
+    let request: ListenerRegistrationRequest = serde_json::from_slice(body).map_err(|e| {
+        api_error(
+            StatusCode::BAD_REQUEST,
+            &format!("invalid operator-listener registration JSON: {e}"),
+        )
+    })?;
+    if request.registrations.is_empty() || request.registrations.len() > MAX_LISTENER_PUBKEYS {
+        return Err(api_error(
+            StatusCode::BAD_REQUEST,
+            "registrations must contain between 1 and 1000 entries",
+        ));
+    }
+    request
+        .registrations
+        .iter()
+        .map(|proof| verify_registration_proof(proof, listener, url, now))
+        .collect::<Result<_, _>>()
+        .map_err(|reason| api_error(StatusCode::BAD_REQUEST, reason))
+}
+
+/// Verify one target-signed registration proof and return the target pubkey.
+///
+/// The proof binds the target's consent to this listener (`p`) and this
+/// deployment's registration endpoint (`u`). The outer NIP-98 request is
+/// replay-checked; within the skew window a proof can only re-register the
+/// pair it was signed for.
+fn verify_registration_proof(
+    proof: &nostr::Event,
+    listener: &nostr::PublicKey,
+    url: &str,
+    now: u64,
+) -> Result<Vec<u8>, &'static str> {
+    if u32::from(proof.kind.as_u16()) != buzz_core::kind::KIND_OPERATOR_LISTENER_REGISTRATION {
+        return Err("registration proof has the wrong kind");
+    }
+    if proof.verify().is_err() {
+        return Err("registration proof signature is invalid");
+    }
+    if proof.created_at.as_secs().abs_diff(now) > REGISTRATION_PROOF_MAX_SKEW_SECS {
+        return Err("registration proof is outside the timestamp window");
+    }
+    let listener_hex = listener.to_hex();
+    let tag_values = |name: &str| {
+        proof
+            .tags
+            .iter()
+            .filter_map(|tag| match tag.as_slice() {
+                [tag_name, value, ..] if tag_name == name => Some(value.as_str()),
+                _ => None,
+            })
+            .collect::<Vec<_>>()
+    };
+    if tag_values("p") != [listener_hex.as_str()] {
+        return Err("registration proof must name exactly this listener in one p tag");
+    }
+    if tag_values("u") != [url] {
+        return Err("registration proof must name this endpoint in one u tag");
+    }
+    Ok(proof.pubkey.to_bytes().to_vec())
+}
+
 /// Shared deployment-global operator auth prelude. The canonical management
 /// origin and replay namespace are configuration, never tenant registry state
 /// or an inbound proxy `Host` header.
@@ -187,15 +267,20 @@ pub async fn register_listener_pubkeys(
     headers: HeaderMap,
     body: axum::body::Bytes,
 ) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
-    let listener = authorize_operator_listener_request(
-        &state,
-        &headers,
-        "POST",
-        "/operator/listener/pubkeys",
+    let listener =
+        authorize_operator_listener_request(&state, &headers, "POST", LISTENER_PUBKEYS_PATH, &body)
+            .await?;
+    let origin = state
+        .config
+        .relay_operator_api_origin
+        .as_deref()
+        .ok_or_else(|| internal_error("operator API origin is not configured"))?;
+    let target_pubkeys = parse_listener_registrations(
         &body,
-    )
-    .await?;
-    let target_pubkeys = parse_listener_pubkeys(&body)?;
+        &listener,
+        &format!("{origin}{LISTENER_PUBKEYS_PATH}"),
+        nostr::Timestamp::now().as_secs(),
+    )?;
     state
         .db
         .register_operator_listener_pubkeys(listener.as_bytes(), &target_pubkeys)
@@ -214,7 +299,7 @@ pub async fn remove_listener_pubkeys(
         &state,
         &headers,
         "DELETE",
-        "/operator/listener/pubkeys",
+        LISTENER_PUBKEYS_PATH,
         &body,
     )
     .await?;
@@ -816,6 +901,130 @@ pub async fn community_availability(
     })))
 }
 
+#[cfg(test)]
+mod tests {
+    use nostr::{EventBuilder, Keys, Kind, Tag, TagKind, Timestamp};
+
+    use super::*;
+
+    const URL: &str = "https://operator.example/operator/listener/pubkeys";
+    const NOW: u64 = 1_800_000_000;
+
+    fn proof(target: &Keys, kind: u16, created_at: u64, tags: Vec<Tag>) -> nostr::Event {
+        EventBuilder::new(Kind::Custom(kind), "")
+            .tags(tags)
+            .custom_created_at(Timestamp::from(created_at))
+            .sign_with_keys(target)
+            .expect("sign registration proof")
+    }
+
+    fn proof_tags(listener: &Keys, url: &str) -> Vec<Tag> {
+        vec![
+            Tag::public_key(listener.public_key()),
+            Tag::custom(TagKind::custom("u"), [url]),
+        ]
+    }
+
+    const PROOF_KIND: u16 = buzz_core::kind::KIND_OPERATOR_LISTENER_REGISTRATION as u16;
+
+    #[test]
+    fn registration_proof_returns_signing_target() {
+        let listener = Keys::generate();
+        let target = Keys::generate();
+        let valid = proof(&target, PROOF_KIND, NOW - 30, proof_tags(&listener, URL));
+
+        assert_eq!(
+            verify_registration_proof(&valid, &listener.public_key(), URL, NOW),
+            Ok(target.public_key().to_bytes().to_vec())
+        );
+    }
+
+    #[test]
+    fn registration_proof_rejects_unbound_or_forged_consent() {
+        let listener = Keys::generate();
+        let other_listener = Keys::generate();
+        let target = Keys::generate();
+        let mut forged = proof(&target, PROOF_KIND, NOW, proof_tags(&listener, URL));
+        forged.pubkey = Keys::generate().public_key();
+        let mut two_listeners = proof_tags(&listener, URL);
+        two_listeners.push(Tag::public_key(other_listener.public_key()));
+
+        for (case, event) in [
+            ("forged signer", forged),
+            (
+                "wrong kind",
+                proof(&target, 27235, NOW, proof_tags(&listener, URL)),
+            ),
+            (
+                "stale",
+                proof(&target, PROOF_KIND, NOW - 61, proof_tags(&listener, URL)),
+            ),
+            (
+                "future",
+                proof(&target, PROOF_KIND, NOW + 61, proof_tags(&listener, URL)),
+            ),
+            (
+                "other listener",
+                proof(&target, PROOF_KIND, NOW, proof_tags(&other_listener, URL)),
+            ),
+            (
+                "ambiguous listener",
+                proof(&target, PROOF_KIND, NOW, two_listeners),
+            ),
+            (
+                "other endpoint",
+                proof(
+                    &target,
+                    PROOF_KIND,
+                    NOW,
+                    proof_tags(&listener, "https://other.example/operator/listener/pubkeys"),
+                ),
+            ),
+            (
+                "missing endpoint",
+                proof(
+                    &target,
+                    PROOF_KIND,
+                    NOW,
+                    vec![Tag::public_key(listener.public_key())],
+                ),
+            ),
+        ] {
+            assert!(
+                verify_registration_proof(&event, &listener.public_key(), URL, NOW).is_err(),
+                "{case} proof was accepted"
+            );
+        }
+    }
+
+    #[test]
+    fn registration_body_requires_signed_proofs() {
+        let listener = Keys::generate();
+        let target = Keys::generate();
+        let valid = proof(&target, PROOF_KIND, NOW, proof_tags(&listener, URL));
+        let parse = |body: serde_json::Value| {
+            parse_listener_registrations(
+                body.to_string().as_bytes(),
+                &listener.public_key(),
+                URL,
+                NOW,
+            )
+        };
+
+        assert_eq!(
+            parse(serde_json::json!({"registrations": [valid]})).expect("signed registration"),
+            vec![target.public_key().to_bytes().to_vec()]
+        );
+        for body in [
+            serde_json::json!({}),
+            serde_json::json!({"registrations": []}),
+            serde_json::json!({"pubkeys": [target.public_key().to_hex()]}),
+        ] {
+            assert!(parse(body.clone()).is_err(), "accepted {body}");
+        }
+    }
+}
+
 #[cfg(test)]
 mod postgres_tests {
     use std::{
@@ -2200,7 +2409,28 @@ mod postgres_tests {
             .iter()
             .map(|target| target.public_key().to_hex())
             .collect::<Vec<_>>();
-        let register_body = serde_json::json!({"pubkeys": target_hex}).to_string();
+        let registration_proof = |target: &Keys, listener: &Keys| {
+            EventBuilder::new(
+                Kind::Custom(buzz_core::kind::KIND_OPERATOR_LISTENER_REGISTRATION as u16),
+                "",
+            )
+            .tags([
+                Tag::public_key(listener.public_key()),
+                Tag::custom(
+                    nostr::TagKind::custom("u"),
+                    [format!("http://{INGRESS_HOST}/operator/listener/pubkeys")],
+                ),
+            ])
+            .sign_with_keys(target)
+            .expect("sign registration proof")
+        };
+        let register_body = serde_json::json!({
+            "registrations": targets
+                .iter()
+                .map(|target| registration_proof(target, &listener))
+                .collect::<Vec<_>>()
+        })
+        .to_string();
         let response = signed_operator_request(
             Arc::clone(&state),
             &listener,
@@ -2228,6 +2458,26 @@ mod postgres_tests {
         assert_eq!(listener_count, 2);
         assert_eq!(other_listener_count, 0);
 
+        // A proof signed for another listener, and the old unsigned form.
+        for rejected_body in [
+            serde_json::json!({"registrations": [registration_proof(&targets[1], &other_listener)]}),
+            serde_json::json!({"pubkeys": [target_hex[1]]}),
+        ] {
+            let rejected = signed_operator_request(
+                Arc::clone(&state),
+                &listener,
+                "POST",
+                "/operator/listener/pubkeys",
+                Some(rejected_body.to_string()),
+            )
+            .await;
+            assert_eq!(
+                rejected.status(),
+                StatusCode::BAD_REQUEST,
+                "{rejected_body}"
+            );
+        }
+
         let body = serde_json::json!({"pubkeys": [target_hex[0]]}).to_string();
         let missing_payload_auth = nip98_auth_header_without_payload(
             &listener,
@@ -2259,7 +2509,7 @@ mod postgres_tests {
             &listener,
             "POST",
             "/operator/listener/pubkeys",
-            Some(r#"{"pubkeys":["not-a-pubkey"]}"#.to_string()),
+            Some(r#"{"registrations":["not-an-event"]}"#.to_string()),
         )
         .await;
         assert_eq!(malformed.status(), StatusCode::BAD_REQUEST);

crates/buzz-relay/src/operator_listener.rs

@@ -8,7 +8,7 @@ use serde::Serialize;
 use tracing::{error, warn};
 use uuid::Uuid;
 
-use crate::{nip98::nip98_header, state::AppState};
+use crate::{handlers::req::event_visible_to_reader, nip98::nip98_header, state::AppState};
 
 use reqwest::StatusCode;
 
@@ -20,14 +20,17 @@ const OUTBOX_REAP_INTERVAL: Duration = Duration::from_secs(5 * 60);
 const REGISTRATION_REAP_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);
 
 /// Notification sent to the configured operator-listener endpoint.
+///
+/// Version 2 adds the full signed `event` to version 1's metadata.
 #[derive(Debug, Serialize)]
-struct MentionNotification {
+struct MentionNotification<'a> {
     v: u8,
     pubkey: String,
     community_host: String,
     event_id: String,
     event_kind: i32,
     event_created_at: i64,
+    event: &'a nostr::Event,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -105,6 +108,10 @@ trait DeliveryTransport: Sync {
 
 #[async_trait::async_trait]
 trait DeliveryStore: Sync {
+    async fn load(
+        &self,
+        delivery: &buzz_db::operator_listener::ClaimedDelivery,
+    ) -> Result<Option<buzz_core::StoredEvent>, 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>;
@@ -113,10 +120,21 @@ trait DeliveryStore: Sync {
 
 struct DbDeliveryStore<'a> {
     db: &'a buzz_db::Db,
+    admission: buzz_db::operator_listener::DeliveryAdmission,
 }
 
 #[async_trait::async_trait]
 impl DeliveryStore for DbDeliveryStore<'_> {
+    async fn load(
+        &self,
+        delivery: &buzz_db::operator_listener::ClaimedDelivery,
+    ) -> Result<Option<buzz_core::StoredEvent>, String> {
+        self.db
+            .load_operator_listener_delivery_event(delivery, self.admission)
+            .await
+            .map_err(|error| error.to_string())
+    }
+
     async fn release(&self, id: Uuid, claim_id: Uuid, next: DateTime<Utc>) -> Result<bool, String> {
         self.db
             .release_operator_listener_delivery(id, claim_id, next)
@@ -191,7 +209,14 @@ async fn run_delivery_once_with_transport<T: DeliveryTransport>(
     if claimed.is_empty() {
         return WorkerIteration::Idle;
     }
-    let store = DbDeliveryStore { db: &state.db };
+    let store = DbDeliveryStore {
+        db: &state.db,
+        admission: buzz_db::operator_listener::DeliveryAdmission {
+            require_membership: state.config.require_relay_membership,
+            allow_owner_membership: state.config.allow_nip_oa_auth,
+            require_allowlist: state.config.pubkey_allowlist_enabled,
+        },
+    };
     join_all(claimed.into_iter().map(|delivery| {
         deliver_one(
             &state.config.operator_listener_delivery_urls,
@@ -233,13 +258,29 @@ 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
+        }
+        // Retry rather than drop: a mention can commit just before the
+        // target is added to the channel.
+        Ok(_) => {
+            retry_or_fail(store, &delivery, "target cannot read event".to_owned()).await;
+            return;
+        }
+        Err(error) => {
+            retry_or_fail(store, &delivery, format!("event load failed: {error}")).await;
+            return;
+        }
+    };
     let body = match serde_json::to_vec(&MentionNotification {
-        v: 1,
+        v: 2,
         pubkey: hex::encode(&delivery.target_pubkey),
         community_host: delivery.community_host.clone(),
         event_id: hex::encode(&delivery.event_id),
         event_kind: delivery.event_kind,
         event_created_at: delivery.event_created_at.timestamp(),
+        event: &stored.event,
     }) {
         Ok(body) => body,
         Err(error) => {
@@ -350,7 +391,7 @@ mod tests {
 
     use super::*;
     use buzz_core::tenant::CommunityId;
-    use nostr::Keys;
+    use nostr::{EventBuilder, Keys, Kind};
 
     #[test]
     fn reapers_use_their_expected_intervals() {
@@ -369,19 +410,46 @@ mod tests {
         Fail(Uuid, Uuid),
     }
 
-    #[derive(Default)]
     struct MockStore {
         calls: Mutex<Vec<StoreCall>>,
+        loaded: Result<Option<buzz_core::StoredEvent>, String>,
+    }
+
+    impl Default for MockStore {
+        fn default() -> Self {
+            Self::loading(Ok(Some(stored_event(Kind::Custom(9)))))
+        }
     }
 
     impl MockStore {
+        fn loading(loaded: Result<Option<buzz_core::StoredEvent>, String>) -> Self {
+            Self {
+                calls: Mutex::default(),
+                loaded,
+            }
+        }
+
         fn calls(&self) -> Vec<StoreCall> {
             self.calls.lock().expect("store calls lock").clone()
         }
     }
 
+    fn stored_event(kind: Kind) -> buzz_core::StoredEvent {
+        let event = EventBuilder::new(kind, "hello agent")
+            .sign_with_keys(&Keys::generate())
+            .expect("sign test event");
+        buzz_core::StoredEvent::new(event, None)
+    }
+
     #[async_trait::async_trait]
     impl DeliveryStore for MockStore {
+        async fn load(
+            &self,
+            _delivery: &buzz_db::operator_listener::ClaimedDelivery,
+        ) -> Result<Option<buzz_core::StoredEvent>, String> {
+            self.loaded.clone()
+        }
+
         async fn release(
             &self,
             id: Uuid,
@@ -546,7 +614,9 @@ mod tests {
     #[tokio::test]
     async fn successful_delivery_posts_payload_and_completes_claim() {
         let (routes, relay_keypair, delivery) = delivery_fixture(1);
-        let store = MockStore::default();
+        let loaded = stored_event(Kind::Custom(9));
+        let expected = loaded.event.clone();
+        let store = MockStore::loading(Ok(Some(loaded)));
         let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT));
 
         deliver_one(
@@ -562,7 +632,7 @@ mod tests {
         assert_eq!(request.url, routes[&hex::encode(&delivery.listener_pubkey)]);
         assert!(request.authorization.starts_with("Nostr "));
         let body: serde_json::Value = serde_json::from_slice(&request.body).expect("JSON body");
-        assert_eq!(body["v"], 1);
+        assert_eq!(body["v"], 2);
         assert_eq!(body["pubkey"], hex::encode(&delivery.target_pubkey));
         assert_eq!(body["community_host"], delivery.community_host);
         assert_eq!(body["event_id"], hex::encode(&delivery.event_id));
@@ -571,12 +641,75 @@ mod tests {
             body["event_created_at"],
             delivery.event_created_at.timestamp()
         );
+        let event: nostr::Event =
+            serde_json::from_value(body["event"].clone()).expect("signed event in body");
+        assert_eq!(event, expected);
+        event
+            .verify()
+            .expect("delivered event keeps a valid signature");
         assert_eq!(
             store.calls(),
             [StoreCall::Complete(delivery.id, delivery.claim_id)]
         );
     }
 
+    #[tokio::test]
+    async fn unreadable_event_retries_then_fails_without_http_request() {
+        // Author-only kinds pass the database check but not the reader check.
+        for loaded in [None, Some(stored_event(Kind::Custom(30300)))] {
+            for (attempt, expect_retry) in [
+                (1, true),
+                (buzz_db::operator_listener::MAX_DELIVERY_ATTEMPTS, false),
+            ] {
+                let (routes, relay_keypair, delivery) = delivery_fixture(attempt);
+                let store = MockStore::loading(Ok(loaded.clone()));
+                let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT));
+
+                deliver_one(
+                    &routes,
+                    &relay_keypair,
+                    &store,
+                    &transport,
+                    delivery.clone(),
+                )
+                .await;
+
+                assert_eq!(transport.request_count(), 0);
+                match store.calls().as_slice() {
+                    [StoreCall::Retry(id, claim_id, _)] if expect_retry => {
+                        assert_eq!((*id, *claim_id), (delivery.id, delivery.claim_id));
+                    }
+                    [StoreCall::Fail(id, claim_id)] if !expect_retry => {
+                        assert_eq!((*id, *claim_id), (delivery.id, delivery.claim_id));
+                    }
+                    calls => panic!("attempt {attempt}: unexpected store calls {calls:?}"),
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn event_load_error_retries_without_http_request() {
+        let (routes, relay_keypair, delivery) = delivery_fixture(1);
+        let store = MockStore::loading(Err("database unavailable".to_owned()));
+        let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT));
+
+        deliver_one(
+            &routes,
+            &relay_keypair,
+            &store,
+            &transport,
+            delivery.clone(),
+        )
+        .await;
+
+        assert_eq!(transport.request_count(), 0);
+        assert!(matches!(
+            store.calls().as_slice(),
+            [StoreCall::Retry(id, claim_id, _)] if *id == delivery.id && *claim_id == delivery.claim_id
+        ));
+    }
+
     #[tokio::test]
     async fn rate_limit_and_server_errors_retry_with_exponential_delay() {
         for status in [