feat(relay): deliver pubkey mentions to relay companions (#7793)

We are introducing a companion service to the relay that needs pubkey
event mentions for pubkeys it manages. This PR introduces an outbox for
pubkey mentions that will be sent to these "operator listener" services.

## Summary

Deliver mentions of registered pubkeys to deployment-global operator
listeners.

- Configure listeners with `BUZZ_OPERATOR_LISTENERS`; listeners manage
target pubkeys through NIP-98-authenticated `POST`/`DELETE
/operator/listener/pubkeys` requests.
- During supported event ingestion, match `p` tags to registrations and
insert one outbox row per listener in the event transaction.
- A background worker claims up to 10 rows, sends NIP-98-signed
notifications, and retries transient failures with exponential backoff
for up to 9 attempts (idle polling backs off from 250 ms to 2 s).
- Outbox rows are ineligible for delivery after 15 minutes and are
reaped every 5 minutes. Registered pubkeys live for 30 days;
re-registering renews them, and a daily reaper removes expired
registrations.
- Keep persistence app-managed: no database triggers, functions, foreign
keys, or separate match queue.

## Changed behavior

Listener outbox rows are enqueued inside the shared event-insert and
thread-metadata transactions; enqueue failure rolls back the event.
`event_mentions` behavior is unchanged: ordinary event/reaction mention
indexing remains post-commit best effort with warning-only failures.

## Testing

- `cargo fmt --all`
- `cargo test -p buzz-db --lib`
- `cargo test -p buzz-relay --lib
config::tests::operator_listener_routes`
- `cargo check -p buzz-db -p buzz-relay`

.env.example

@@ -120,6 +120,12 @@ BUZZ_PUSH_GATEWAY_DELIVERY_URL=https://push.buzz.xyz/v1/deliveries/apns
 # and provisioning requests fail closed until it is set.
 # RELAY_OPERATOR_API_ORIGIN=http://127.0.0.1:3000
 
+# Deployment-global operator-listener routes. Each configured Nostr identity
+# may register target pubkeys and receives mention notifications at its URL.
+# Format: pubkey:deliveryUrl;pubkey:deliveryUrl
+# BUZZ_OPERATOR_LISTENER_TIMEOUT_MS=5000
+# BUZZ_OPERATOR_LISTENERS=<64-char hex pubkey>:https://listener.example/mentions
+
 # Optional relay-owned KLIPY key. When set, NIP-11 advertises GIF search and
 # authenticated desktop clients use this relay as the metadata/search proxy.
 # Keep the real value in your deployment's secret manager; never commit it.

crates/buzz-db/src/lib.rs

@@ -63,9 +63,10 @@ pub(crate) use runtime::{
 };
 pub use store::{
     admin_moderation, allowlist, api_token, archived_identities, channel, channel_members,
-    community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, push,
-    reaction, read_state, relay_admin_actions, relay_invite, relay_members, relay_operators,
-    reminder, replaceable, storage_accounting, thread, thread_window, usage, user, workflow,
+    community, deletion, dm, event, feed, git_repo, moderation, operator_listener, partition,
+    product_feedback, push, reaction, read_state, relay_admin_actions, relay_invite, relay_members,
+    relay_operators, reminder, replaceable, storage_accounting, thread, thread_window, usage, user,
+    workflow,
 };
 
 pub use allowlist::AllowlistEntry;

crates/buzz-db/src/runtime/migration.rs

@@ -490,6 +490,8 @@ mod postgres_tests {
             "relay_admin_outbox",
             "relay_operator_audit",
             "storage_accounting_snapshots",
+            "operator_listener_pubkeys",
+            "operator_listener_outbox",
         ] {
             if normalized[insert_pos..].contains(&format!("'{value}'")) {
                 globals.insert(value.to_owned());
@@ -703,7 +705,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 49);
+        assert_eq!(migrations.len(), 50);
         assert_eq!(migrations[48].version, 49);
         assert!(migrations[48]
             .sql
@@ -1293,10 +1295,63 @@ mod postgres_tests {
             .sql
             .as_str()
             .contains("CREATE TABLE storage_accounting_snapshots"));
-        // schema.sql exclusion list must match the restored (pre-0041) body.
+        assert_eq!(migrations[47].version, 48);
+        assert!(migrations[47]
+            .sql
+            .as_str()
+            .contains("ADD COLUMN hash_version SMALLINT"));
+        assert_eq!(migrations[49].version, 50);
+        assert!(migrations[49]
+            .sql
+            .as_str()
+            .contains("CREATE TABLE operator_listener_pubkeys"));
+        assert!(migrations[49]
+            .sql
+            .as_str()
+            .contains("CREATE TABLE operator_listener_outbox"));
+        assert!(migrations[49]
+            .sql
+            .as_str()
+            .contains("CREATE OR REPLACE FUNCTION community_write_fence_excluded_table"));
+        assert!(migrations[49]
+            .sql
+            .as_str()
+            .contains("'rate_limit_violations', 'operator_listener_outbox'"));
+        for index in [
+            "operator_listener_pubkeys_target",
+            "operator_listener_pubkeys_created_at",
+            "operator_listener_outbox_due",
+            "operator_listener_outbox_recovery",
+            "operator_listener_outbox_created_at",
+        ] {
+            assert!(
+                migrations[49].sql.as_str().contains(index),
+                "migration 0050 must declare {index}"
+            );
+            assert!(
+                desired_schema.contains(index),
+                "schema.sql must declare {index}"
+            );
+        }
+        for index_shape in [
+            "ON operator_listener_outbox (next_attempt_at, created_at, id)",
+            "ON operator_listener_outbox (lease_until, created_at, id)",
+        ] {
+            assert!(
+                migrations[49].sql.as_str().contains(index_shape),
+                "migration 0050 must declare {index_shape}"
+            );
+            assert!(
+                desired_schema.contains(index_shape),
+                "schema.sql must declare {index_shape}"
+            );
+        }
+        // schema.sql keeps the restored (pre-0041) deletion exclusions and also
+        // leaves the deployment-global listener outbox outside tenant fencing.
         assert!(
-            desired_schema.contains("'rate_limit_violations'\n    ]::TEXT[])"),
-            "schema.sql exclusion list must match the pre-0041 body after ledger removal"
+            desired_schema
+                .contains("'rate_limit_violations', 'operator_listener_outbox'\n    ]::TEXT[])"),
+            "schema.sql must exclude the deployment-global listener outbox from tenant fencing"
         );
     }
 

crates/buzz-db/src/runtime/tests.rs

@@ -638,6 +638,26 @@ async fn migration_schema_database_guard_covers_legacy_writer_and_nip09_deletion
     assert_eq!(watermark.1, c.id.as_bytes().as_slice());
 }
 
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn migration_schema_operator_listener_outbox_is_not_tenant_scoped() {
+    let db = setup_db().await;
+    let community = CommunityId::from_uuid(make_community(&db.pool).await);
+    let inventory = db
+        .deletion_store()
+        .inventory_schema(community)
+        .await
+        .expect("migrated deletion catalog should validate");
+
+    assert!(
+        !inventory
+            .scoped_tables
+            .iter()
+            .any(|table| table == "operator_listener_outbox"),
+        "operator-listener outbox is deployment-global, not tenant-scoped"
+    );
+}
+
 // ---- Read-replica routing ------------------------------------------------
 //
 // These tests pin the routing contract of `Db::read()` and the two routed

crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs

@@ -311,7 +311,7 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer()
         production_result.is_ok(),
         "production migrator must preserve ingestion progress: {production_result:?}"
     );
-    assert_eq!(version, 49);
+    assert_eq!(version, 50);
     assert_eq!(final_oid, oid, "prebuild must not be replaced");
     assert_eq!(count, 4, "all writer witnesses must persist");
 }

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

@@ -355,12 +355,15 @@ pub async fn insert_event(
     event: &Event,
     channel_id: Option<Uuid>,
 ) -> Result<(StoredEvent, bool)> {
-    let mut connection = crate::observability::acquire_writer(
+    let connection = crate::observability::acquire_writer(
         pool,
         crate::observability::WriterOperation::EventWrite,
     )
     .await?;
-    insert_event_on(&mut connection, community_id, event, channel_id).await
+    let mut tx = Transaction::begin(connection, None).await?;
+    let result = insert_event_in_transaction(&mut tx, community_id, event, channel_id).await?;
+    tx.commit().await?;
+    Ok(result)
 }
 
 /// Insert a Nostr event in a caller-owned PostgreSQL transaction.
@@ -374,7 +377,11 @@ pub async fn insert_event_in_transaction(
     event: &Event,
     channel_id: Option<Uuid>,
 ) -> Result<(StoredEvent, bool)> {
-    insert_event_on(tx.as_mut(), community_id, event, channel_id).await
+    let result = insert_event_on(tx.as_mut(), community_id, event, channel_id).await?;
+    if result.1 {
+        crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?;
+    }
+    Ok(result)
 }
 
 async fn insert_event_on(
@@ -1528,6 +1535,8 @@ pub(crate) async fn insert_event_with_thread_metadata_tx(
                 }
             }
         }
+
+        crate::operator_listener::enqueue_mentions_in_transaction(tx, community_id, event).await?;
     }
 
     Ok((

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

@@ -26,6 +26,8 @@ pub mod feed;
 pub mod git_repo;
 /// Community moderation: reports, bans/timeouts, audit actions.
 pub mod moderation;
+/// Deployment-global operator-listener mention registrations and delivery queues.
+pub mod operator_listener;
 /// Monthly table partition management.
 pub mod partition;
 /// Buzz product-feedback sidecar persistence.

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

@@ -54,6 +54,44 @@ struct TransferCommunityResponse {
 
 const OPERATOR_REPLAY_SCOPE: &str = "operator-management";
 
+#[derive(Debug, Deserialize)]
+struct ListenerPubkeysRequest {
+    pubkeys: Vec<String>,
+}
+
+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(
+            StatusCode::BAD_REQUEST,
+            &format!("invalid operator-listener pubkeys JSON: {e}"),
+        )
+    })?;
+    if request.pubkeys.is_empty() || request.pubkeys.len() > 1_000 {
+        return Err(api_error(
+            StatusCode::BAD_REQUEST,
+            "pubkeys must contain between 1 and 1000 entries",
+        ));
+    }
+    request
+        .pubkeys
+        .into_iter()
+        .map(|value| {
+            let normalized = validate_pubkey_hex(&value).ok_or_else(|| {
+                api_error(
+                    StatusCode::BAD_REQUEST,
+                    "pubkeys must contain 64-char hex public keys",
+                )
+            })?;
+            hex::decode(normalized).map_err(|_| {
+                api_error(
+                    StatusCode::BAD_REQUEST,
+                    "pubkeys must contain 64-char hex public keys",
+                )
+            })
+        })
+        .collect()
+}
+
 /// 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.
@@ -98,6 +136,86 @@ async fn authorize_operator_request(
     Ok(pubkey)
 }
 
+/// Authenticate a deployment-global operator listener using its configured
+/// identity and a NIP-98 request signature.
+async fn authorize_operator_listener_request(
+    state: &Arc<AppState>,
+    headers: &HeaderMap,
+    method: &str,
+    path: &str,
+    body: &[u8],
+) -> Result<nostr::PublicKey, (StatusCode, Json<Value>)> {
+    let origin = state
+        .config
+        .relay_operator_api_origin
+        .as_deref()
+        .ok_or_else(|| internal_error("operator API origin is not configured"))?;
+    let url = format!("{origin}{path}");
+    let bridge::VerifiedBridgeAuth {
+        pubkey,
+        event_id_bytes,
+        ..
+    } = bridge::verify_nip98_exempt_operator(headers, method, &url, Some(body))?;
+    check_operator_replay(state, event_id_bytes).await?;
+    if !state
+        .config
+        .operator_listener_delivery_urls
+        .contains_key(&pubkey.to_hex())
+    {
+        return Err(api_error(
+            StatusCode::FORBIDDEN,
+            "actor not authorized: not a configured operator listener",
+        ));
+    }
+    Ok(pubkey)
+}
+
+/// Register target pubkeys for the authenticated operator listener.
+pub async fn register_listener_pubkeys(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    body: axum::body::Bytes,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    let listener = authorize_operator_listener_request(
+        &state,
+        &headers,
+        "POST",
+        "/operator/listener/pubkeys",
+        &body,
+    )
+    .await?;
+    let target_pubkeys = parse_listener_pubkeys(&body)?;
+    state
+        .db
+        .register_operator_listener_pubkeys(listener.as_bytes(), &target_pubkeys)
+        .await
+        .map_err(|e| internal_error(&format!("register operator-listener pubkeys: {e}")))?;
+    Ok(Json(serde_json::json!({})))
+}
+
+/// Remove target pubkeys for the authenticated operator listener.
+pub async fn remove_listener_pubkeys(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    body: axum::body::Bytes,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    let listener = authorize_operator_listener_request(
+        &state,
+        &headers,
+        "DELETE",
+        "/operator/listener/pubkeys",
+        &body,
+    )
+    .await?;
+    let target_pubkeys = parse_listener_pubkeys(&body)?;
+    state
+        .db
+        .remove_operator_listener_pubkeys(listener.as_bytes(), &target_pubkeys)
+        .await
+        .map_err(|e| internal_error(&format!("remove operator-listener pubkeys: {e}")))?;
+    Ok(Json(serde_json::json!({})))
+}
+
 async fn check_operator_replay(
     state: &AppState,
     event_id_bytes: [u8; 32],
@@ -496,7 +614,10 @@ pub async fn community_availability(
 
 #[cfg(test)]
 mod postgres_tests {
-    use std::sync::Arc;
+    use std::{
+        collections::HashSet,
+        sync::{Arc, Mutex},
+    };
 
     use axum::{
         body::{to_bytes, Body},
@@ -529,6 +650,26 @@ mod postgres_tests {
             Box::pin(async { Ok(true) })
         }
     }
+
+    struct SeenOnceReplayGuard(Mutex<HashSet<[u8; 32]>>);
+
+    impl buzz_auth::Nip98ReplayGuard for SeenOnceReplayGuard {
+        fn try_mark_in_scope<'a>(
+            &'a self,
+            _scope: &'a str,
+            event_id: &'a nostr::EventId,
+            _ttl_secs: u64,
+        ) -> std::pin::Pin<
+            Box<dyn std::future::Future<Output = Result<bool, buzz_auth::AuthError>> + Send + 'a>,
+        > {
+            let inserted = self
+                .0
+                .lock()
+                .expect("replay set")
+                .insert(*event_id.as_bytes());
+            Box::pin(async move { Ok(inserted) })
+        }
+    }
     const INGRESS_HOST: &str = "operator-ingress.example";
 
     fn nip98_auth_header(keys: &Keys, url: &str, method: &str, body: Option<&[u8]>) -> String {
@@ -629,6 +770,16 @@ mod postgres_tests {
     ) -> axum::response::Response {
         let url = format!("http://{INGRESS_HOST}{path}");
         let auth = nip98_auth_header(keys, &url, method, body.as_deref().map(str::as_bytes));
+        operator_request_with_auth(state, method, path, body, auth).await
+    }
+
+    async fn operator_request_with_auth(
+        state: Arc<AppState>,
+        method: &str,
+        path: &str,
+        body: Option<String>,
+        auth: String,
+    ) -> axum::response::Response {
         let mut request = Request::builder()
             .method(method)
             .uri(path)
@@ -770,6 +921,162 @@ mod postgres_tests {
         );
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn listener_pubkey_routes_register_and_remove_only_for_authenticated_listener() {
+        let listener = Keys::generate();
+        let other_listener = Keys::generate();
+        let outsider = Keys::generate();
+        let Some(mut state) = operator_test_state(&[]).await else {
+            return;
+        };
+        Arc::get_mut(&mut state)
+            .expect("test state has a single owner")
+            .nip98_replay = Arc::new(SeenOnceReplayGuard(Mutex::new(HashSet::new())));
+        let config = Arc::make_mut(
+            &mut Arc::get_mut(&mut state)
+                .expect("test state has a single owner")
+                .config,
+        );
+        config.operator_listener_delivery_urls.insert(
+            listener.public_key().to_hex(),
+            url::Url::parse("http://listener.example/deliver").expect("listener URL"),
+        );
+        config.operator_listener_delivery_urls.insert(
+            other_listener.public_key().to_hex(),
+            url::Url::parse("http://other-listener.example/deliver").expect("listener URL"),
+        );
+
+        let targets = [Keys::generate(), Keys::generate()];
+        let pool = sqlx::PgPool::connect(&crate::test_support::database_url())
+            .await
+            .expect("connect to test DB");
+        let target_hex = targets
+            .iter()
+            .map(|target| target.public_key().to_hex())
+            .collect::<Vec<_>>();
+        let register_body = serde_json::json!({"pubkeys": target_hex}).to_string();
+        let response = signed_operator_request(
+            Arc::clone(&state),
+            &listener,
+            "POST",
+            "/operator/listener/pubkeys",
+            Some(register_body),
+        )
+        .await;
+        assert_eq!(response.status(), StatusCode::OK);
+
+        let listener_count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM operator_listener_pubkeys WHERE listener_pubkey = $1",
+        )
+        .bind(listener.public_key().as_bytes().as_slice())
+        .fetch_one(&pool)
+        .await
+        .expect("count registered listener targets");
+        let other_listener_count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM operator_listener_pubkeys WHERE listener_pubkey = $1",
+        )
+        .bind(other_listener.public_key().as_bytes().as_slice())
+        .fetch_one(&pool)
+        .await
+        .expect("count other listener targets");
+        assert_eq!(listener_count, 2);
+        assert_eq!(other_listener_count, 0);
+
+        let body = serde_json::json!({"pubkeys": [target_hex[0]]}).to_string();
+        let missing_payload_auth = nip98_auth_header_without_payload(
+            &listener,
+            &format!("http://{INGRESS_HOST}/operator/listener/pubkeys"),
+            "DELETE",
+        );
+        let missing_payload = operator_request_with_auth(
+            Arc::clone(&state),
+            "DELETE",
+            "/operator/listener/pubkeys",
+            Some(body),
+            missing_payload_auth,
+        )
+        .await;
+        assert_eq!(missing_payload.status(), StatusCode::UNAUTHORIZED);
+
+        let unauthorized = signed_operator_request(
+            Arc::clone(&state),
+            &outsider,
+            "POST",
+            "/operator/listener/pubkeys",
+            Some(serde_json::json!({"pubkeys": [target_hex[0]]}).to_string()),
+        )
+        .await;
+        assert_eq!(unauthorized.status(), StatusCode::FORBIDDEN);
+
+        let malformed = signed_operator_request(
+            Arc::clone(&state),
+            &listener,
+            "POST",
+            "/operator/listener/pubkeys",
+            Some(r#"{"pubkeys":["not-a-pubkey"]}"#.to_string()),
+        )
+        .await;
+        assert_eq!(malformed.status(), StatusCode::BAD_REQUEST);
+
+        let replay_body = serde_json::json!({"pubkeys": [target_hex[1]]}).to_string();
+        let replay_auth = nip98_auth_header(
+            &listener,
+            &format!("http://{INGRESS_HOST}/operator/listener/pubkeys"),
+            "DELETE",
+            Some(replay_body.as_bytes()),
+        );
+        let first_use = operator_request_with_auth(
+            Arc::clone(&state),
+            "DELETE",
+            "/operator/listener/pubkeys",
+            Some(replay_body.clone()),
+            replay_auth.clone(),
+        )
+        .await;
+        assert_eq!(first_use.status(), StatusCode::OK);
+        let replay = operator_request_with_auth(
+            Arc::clone(&state),
+            "DELETE",
+            "/operator/listener/pubkeys",
+            Some(replay_body),
+            replay_auth,
+        )
+        .await;
+        assert_eq!(replay.status(), StatusCode::UNAUTHORIZED);
+        let replay_error = read_json(replay).await;
+        assert!(replay_error["error"]
+            .as_str()
+            .unwrap_or_default()
+            .contains("replay"));
+
+        let remove_body = serde_json::json!({"pubkeys": [target_hex[0]]}).to_string();
+        let removed = signed_operator_request(
+            Arc::clone(&state),
+            &listener,
+            "DELETE",
+            "/operator/listener/pubkeys",
+            Some(remove_body),
+        )
+        .await;
+        assert_eq!(removed.status(), StatusCode::OK);
+        let remaining: Vec<Vec<u8>> = sqlx::query_scalar(
+            "SELECT target_pubkey FROM operator_listener_pubkeys WHERE listener_pubkey = $1",
+        )
+        .bind(listener.public_key().as_bytes().as_slice())
+        .fetch_all(&pool)
+        .await
+        .expect("read remaining listener targets");
+        assert!(remaining.is_empty());
+
+        sqlx::query("DELETE FROM operator_listener_pubkeys WHERE listener_pubkey = $1 OR listener_pubkey = $2")
+            .bind(listener.public_key().as_bytes().as_slice())
+            .bind(other_listener.public_key().as_bytes().as_slice())
+            .execute(&pool)
+            .await
+            .expect("clean up listener registrations");
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn unmapped_management_host_can_check_availability() {

crates/buzz-relay/src/config.rs

@@ -1,11 +1,11 @@
 //! Relay configuration from environment variables.
 
-use std::net::SocketAddr;
 use std::time::Duration;
+use std::{collections::HashMap, net::SocketAddr};
 
 use sha2::{Digest, Sha256};
 use thiserror::Error;
-use tracing::warn;
+use tracing::{error, warn};
 
 /// Default maximum inbound WebSocket frame size in bytes.
 ///
@@ -272,6 +272,9 @@ pub struct Config {
     /// skipped — a typo must not silently disable an operator.
     pub relay_operator_pubkeys: Vec<String>,
 
+    /// Configured operator-listener identities and their HTTPS delivery URLs.
+    pub operator_listener_delivery_urls: HashMap<String, url::Url>,
+
     /// Allow NIP-OA owner attestation for relay membership.
     ///
     /// When `true` and `require_relay_membership` is also `true`, agents
@@ -350,8 +353,10 @@ pub struct Config {
     /// Required while push is enabled. An explicitly empty setting is allowed
     /// only while push is disabled.
     pub push_gateway_delivery_url: Option<url::Url>,
-    /// Hard timeout for one gateway delivery request.
+    /// Hard timeout for one push gateway delivery request.
     pub push_gateway_timeout: Duration,
+    /// Hard timeout for one operator-listener delivery request.
+    pub operator_listener_timeout: Duration,
 
     /// Optional relay-hosted policy shown on join surfaces. Disabled when no
     /// documents or age attestation are configured.
@@ -478,6 +483,58 @@ fn parse_push_gateway_delivery_url(raw: &str) -> Result<url::Url, ConfigError> {
     Ok(url)
 }
 
+fn parse_operator_listener_delivery_urls(
+    raw: &str,
+) -> Result<HashMap<String, url::Url>, ConfigError> {
+    let mut endpoints = HashMap::new();
+    let entries: Vec<&str> = raw
+        .split(';')
+        .map(str::trim)
+        .filter(|entry| !entry.is_empty())
+        .collect();
+    if !raw.trim().is_empty() && entries.is_empty() {
+        return Err(ConfigError::InvalidValue(
+            "BUZZ_OPERATOR_LISTENERS must contain pubkey:delivery_url pairs".to_string(),
+        ));
+    }
+    for entry in entries {
+        let (pubkey, delivery_url) = entry.split_once(':').ok_or_else(|| {
+            ConfigError::InvalidValue(
+                "BUZZ_OPERATOR_LISTENERS entries must be pubkey:delivery_url pairs".to_string(),
+            )
+        })?;
+        let pubkey = pubkey.trim().to_ascii_lowercase();
+        if pubkey.len() != 64 || !pubkey.chars().all(|c| c.is_ascii_hexdigit()) {
+            return Err(ConfigError::InvalidValue(format!(
+                "BUZZ_OPERATOR_LISTENERS entry has an invalid pubkey: {pubkey:?}"
+            )));
+        }
+        let url = url::Url::parse(delivery_url.trim()).map_err(|e| {
+            ConfigError::InvalidValue(format!(
+                "BUZZ_OPERATOR_LISTENERS endpoint is not a valid URL: {e}"
+            ))
+        })?;
+        if url.scheme() != "https"
+            || url.host().is_none()
+            || !url.username().is_empty()
+            || url.password().is_some()
+            || url.query().is_some()
+            || url.fragment().is_some()
+        {
+            return Err(ConfigError::InvalidValue(
+                "BUZZ_OPERATOR_LISTENERS endpoints must be HTTPS URLs without credentials, query, or fragment"
+                    .to_string(),
+            ));
+        }
+        if endpoints.insert(pubkey.clone(), url).is_some() {
+            return Err(ConfigError::InvalidValue(format!(
+                "BUZZ_OPERATOR_LISTENERS contains duplicate pubkey: {pubkey}"
+            )));
+        }
+    }
+    Ok(endpoints)
+}
+
 fn parse_bool(name: &str, default: bool) -> Result<bool, ConfigError> {
     match std::env::var(name) {
         Err(std::env::VarError::NotPresent) => Ok(default),
@@ -797,6 +854,32 @@ impl Config {
             );
         }
 
+        let operator_listener_delivery_urls = match std::env::var("BUZZ_OPERATOR_LISTENERS") {
+            Ok(raw) => match parse_operator_listener_delivery_urls(&raw) {
+                Ok(urls) => urls,
+                Err(parse_error) => {
+                    error!(
+                        error = %parse_error,
+                        "invalid BUZZ_OPERATOR_LISTENERS; operator-listener mention delivery is disabled"
+                    );
+                    HashMap::new()
+                }
+            },
+            Err(std::env::VarError::NotPresent) => HashMap::new(),
+            Err(error) => {
+                error!(
+                    error = %error,
+                    "BUZZ_OPERATOR_LISTENERS must be valid UTF-8; operator-listener mention delivery is disabled"
+                );
+                HashMap::new()
+            }
+        };
+        if !operator_listener_delivery_urls.is_empty() && relay_operator_api_origin.is_none() {
+            error!(
+                "BUZZ_OPERATOR_LISTENERS is set but RELAY_OPERATOR_API_ORIGIN is not — operator-listener registration requests will reject every request until RELAY_OPERATOR_API_ORIGIN is set"
+            );
+        }
+
         let auth = buzz_auth::AuthConfig {
             rate_limits: rate_limit_config_from_env()?,
         };
@@ -1012,6 +1095,33 @@ impl Config {
             Err(_) => 2_000,
         };
         let push_gateway_timeout = Duration::from_millis(push_gateway_timeout_millis);
+        let operator_listener_timeout_millis = match std::env::var(
+            "BUZZ_OPERATOR_LISTENER_TIMEOUT_MS",
+        ) {
+            Ok(raw) => match raw
+                .parse::<u64>()
+                .ok()
+                .filter(|millis| (100..=10_000).contains(millis))
+            {
+                Some(millis) => millis,
+                None => {
+                    error!(
+                        value = %raw,
+                        "invalid BUZZ_OPERATOR_LISTENER_TIMEOUT_MS; using default 5000ms (expected integer in 100..=10000)"
+                    );
+                    5_000
+                }
+            },
+            Err(std::env::VarError::NotPresent) => 5_000,
+            Err(err) => {
+                error!(
+                    error = %err,
+                    "invalid BUZZ_OPERATOR_LISTENER_TIMEOUT_MS; using default 5000ms"
+                );
+                5_000
+            }
+        };
+        let operator_listener_timeout = Duration::from_millis(operator_listener_timeout_millis);
 
         const MAX_POLICY_MARKDOWN_BYTES: usize = 256 * 1024;
         let read_policy_markdown = |name: &str| -> Result<Option<String>, ConfigError> {
@@ -1248,6 +1358,7 @@ impl Config {
             relay_owner_pubkey,
             relay_operator_api_origin,
             relay_operator_pubkeys,
+            operator_listener_delivery_urls,
             allow_nip_oa_auth,
             klipy,
             media,
@@ -1269,6 +1380,7 @@ impl Config {
             push_executor_key_id,
             push_gateway_delivery_url,
             push_gateway_timeout,
+            operator_listener_timeout,
             join_policy,
             admin,
             web_dir,
@@ -1430,14 +1542,8 @@ mod tests {
         config
     }
 
-    /// Like `config_with_admin_env`, but also captures the tracing output
-    /// emitted during `Config::from_env()` so a test can assert the startup
-    /// warning fired. The `BUZZ_ADMIN_TOKEN` warning is the sole behavioral
-    /// value of retaining the guards (the variable is otherwise inert), so it
-    /// must be regression-protected: deleting a warn block has to fail a test.
-    fn config_with_admin_env_capturing_logs(
-        values: &[(&str, Option<&str>)],
-    ) -> (Result<Config, ConfigError>, String) {
+    /// Capture the tracing output emitted during `Config::from_env()`.
+    fn capture_config_logs() -> (Result<Config, ConfigError>, String) {
         use std::sync::{Arc, Mutex};
 
         #[derive(Clone)]
@@ -1472,12 +1578,56 @@ mod tests {
             })
             .with_ansi(false)
             .finish();
-        let config =
-            tracing::subscriber::with_default(subscriber, || config_with_admin_env(values));
+        let config = tracing::subscriber::with_default(subscriber, Config::from_env);
         let captured = String::from_utf8(buf.lock().unwrap().clone()).unwrap_or_default();
         (config, captured)
     }
 
+    /// Like `config_with_admin_env`, but also captures the tracing output
+    /// emitted during `Config::from_env()` so a test can assert the startup
+    /// warning fired. The `BUZZ_ADMIN_TOKEN` warning is the sole behavioral
+    /// value of retaining the guards (the variable is otherwise inert), so it
+    /// must be regression-protected: deleting a warn block has to fail a test.
+    fn config_with_admin_env_capturing_logs(
+        values: &[(&str, Option<&str>)],
+    ) -> (Result<Config, ConfigError>, String) {
+        const KEYS: [&str; 3] = ["BUZZ_ADMIN_HOST", "BUZZ_ADMIN_TOKEN", "BUZZ_ADMIN_AUTH"];
+        let previous: Vec<_> = KEYS
+            .iter()
+            .map(|key| (*key, std::env::var_os(key)))
+            .collect();
+        for key in KEYS {
+            std::env::remove_var(key);
+        }
+        for (key, value) in values {
+            match value {
+                Some(value) => std::env::set_var(key, value),
+                None => std::env::remove_var(key),
+            }
+        }
+        let result = capture_config_logs();
+        for (key, value) in previous {
+            match value {
+                Some(value) => std::env::set_var(key, value),
+                None => std::env::remove_var(key),
+            }
+        }
+        result
+    }
+
+    fn config_with_operator_listeners_capturing_logs(
+        raw: &str,
+    ) -> (Result<Config, ConfigError>, String) {
+        let previous = std::env::var_os("BUZZ_OPERATOR_LISTENERS");
+        std::env::set_var("BUZZ_OPERATOR_LISTENERS", raw);
+        let result = capture_config_logs();
+        match previous {
+            Some(value) => std::env::set_var("BUZZ_OPERATOR_LISTENERS", value),
+            None => std::env::remove_var("BUZZ_OPERATOR_LISTENERS"),
+        }
+        result
+    }
+
     /// Assert `captured` contains a WARN naming the removal of `BUZZ_ADMIN_TOKEN`
     /// so the migration breadcrumb Will's ruling preserved cannot silently regress.
     fn assert_admin_token_removal_warning(captured: &str) {
@@ -2293,6 +2443,52 @@ mod tests {
         }
     }
 
+    #[test]
+    fn operator_listener_routes_parse_and_normalize() {
+        let routes = parse_operator_listener_delivery_urls(
+            "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA:https://one.example/mentions;bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb:https://two.example/mentions",
+        )
+        .expect("valid operator-listener routes");
+        assert_eq!(routes.len(), 2);
+        assert!(
+            routes.contains_key("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
+        );
+        assert!(
+            routes.contains_key("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
+        );
+    }
+
+    #[test]
+    fn operator_listener_routes_reject_unsafe_or_duplicate_entries() {
+        let key = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+        for raw in [
+            format!("{key}:http://listener.example/mentions"),
+            format!("{key}:https://user:pass@listener.example/mentions"),
+            format!("{key}:https://listener.example/mentions?token=secret"),
+            format!("{key}:https://listener.example/mentions;{key}:https://other.example"),
+        ] {
+            assert!(
+                parse_operator_listener_delivery_urls(&raw).is_err(),
+                "unsafe or duplicate route accepted: {raw}"
+            );
+        }
+    }
+
+    #[test]
+    fn malformed_operator_listener_config_is_logged_and_disables_delivery() {
+        let _guard = ENV_MUTEX.lock().unwrap();
+        let (result, logs) = config_with_operator_listeners_capturing_logs(";");
+        let config = result.expect("malformed listener config must not stop relay startup");
+
+        assert!(config.operator_listener_delivery_urls.is_empty());
+        assert!(logs.contains("ERROR"), "expected an ERROR line: {logs:?}");
+        assert!(logs.contains("BUZZ_OPERATOR_LISTENERS"), "logs: {logs:?}");
+        assert!(
+            logs.contains("operator-listener mention delivery is disabled"),
+            "logs: {logs:?}"
+        );
+    }
+
     #[test]
     fn invalid_push_gateway_timeout_is_not_silently_defaulted() {
         let _guard = ENV_MUTEX.lock().unwrap();

crates/buzz-relay/src/lib.rs

@@ -33,11 +33,14 @@ pub mod mesh_boot;
 pub mod metrics;
 /// NIP-11 relay information document.
 pub mod nip11;
+mod nip98;
 /// NIP-FI relay configuration: mode, issuer registry, JWKS warm/refresh.
 pub mod nip_fi_config;
 /// NIP-FI HTTP ingress enforcement: assertion extraction, verification,
 /// key-pairing check, and deny-map gate for every protected HTTP surface.
 pub(crate) mod nip_fi_http;
+/// Deployment-global operator-listener mention delivery worker.
+pub mod operator_listener;
 /// NIP-01 client/relay message parsing.
 pub mod protocol;
 /// Durable NIP-PL matcher and delivery worker.

crates/buzz-relay/src/main.rs

@@ -857,6 +857,24 @@ async fn run_relay_main(boot: BootTracker) -> anyhow::Result<()> {
         info!("NIP-PL push disabled by BUZZ_PUSH_ENABLED");
     }
 
+    // Registration cleanup is independent of configured routes. Delivery is
+    // opt-in, while accepted events populate the outbox transactionally.
+    tokio::spawn(buzz_relay::operator_listener::run_reaper(Arc::clone(
+        &state,
+    )));
+    if !state.config.operator_listener_delivery_urls.is_empty() {
+        tokio::spawn(buzz_relay::operator_listener::run_delivery_worker(
+            Arc::clone(&state),
+        ));
+        info!(
+            listeners = state.config.operator_listener_delivery_urls.len(),
+            "operator-listener mention delivery worker started"
+        );
+    } else {
+        info!("operator-listener mention delivery disabled by BUZZ_OPERATOR_LISTENERS");
+    }
+    info!("operator-listener registration cleanup started");
+
     // Admin outbox delivery worker — drives `relay_admin_outbox` rows.
     // Uses DB-level leases (held_by / lease_expires_at) so multiple pods can
     // run the worker concurrently without double-delivery.

crates/buzz-relay/src/nip98.rs

@@ -0,0 +1,22 @@
+//! Shared NIP-98 signing helpers for relay-originated HTTP requests.
+
+use base64::Engine as _;
+use nostr::{EventBuilder, Kind, Tag};
+use sha2::{Digest as _, Sha256};
+
+/// Build a NIP-98 authorization header for a signed JSON `POST` request.
+pub(crate) fn nip98_header(keys: &nostr::Keys, url: &str, body: &[u8]) -> anyhow::Result<String> {
+    let hash = hex::encode(Sha256::digest(body));
+    let event = EventBuilder::new(Kind::HttpAuth, "")
+        .tags([
+            Tag::parse(["u", url])?,
+            Tag::parse(["method", "POST"])?,
+            Tag::parse(["payload", &hash])?,
+            Tag::parse(["nonce", &uuid::Uuid::new_v4().to_string()])?,
+        ])
+        .sign_with_keys(keys)?;
+    Ok(format!(
+        "Nostr {}",
+        base64::engine::general_purpose::STANDARD.encode(serde_json::to_vec(&event)?)
+    ))
+}

crates/buzz-relay/src/operator_listener.rs

@@ -0,0 +1,776 @@
+//! Deployment-global operator-listener mention matching and notification delivery.
+
+use std::{collections::HashMap, sync::Arc, time::Duration};
+
+use chrono::{DateTime, TimeDelta, Utc};
+use futures_util::future::join_all;
+use serde::Serialize;
+use tracing::{error, warn};
+use uuid::Uuid;
+
+use crate::{nip98::nip98_header, state::AppState};
+
+use reqwest::StatusCode;
+
+const CLAIM_SECS: i64 = 30;
+const DELIVERY_BATCH_LIMIT: i64 = 10;
+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);
+
+/// Notification sent to the configured operator-listener endpoint.
+#[derive(Debug, Serialize)]
+struct MentionNotification {
+    v: u8,
+    pubkey: String,
+    community_host: String,
+    event_id: String,
+    event_kind: i32,
+    event_created_at: i64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum WorkerIteration {
+    Worked,
+    Idle,
+    Failed,
+}
+
+/// Reap stale deliveries every five minutes and expired registrations daily.
+pub async fn run_reaper(state: Arc<AppState>) {
+    tokio::join!(
+        async {
+            loop {
+                if let Err(error) = state.db.delete_expired_operator_listener_pubkeys().await {
+                    warn!(%error, "operator-listener registration cleanup failed");
+                }
+                tokio::time::sleep(REGISTRATION_REAP_INTERVAL).await;
+            }
+        },
+        async {
+            loop {
+                if let Err(error) = state.db.reap_operator_listener_deliveries().await {
+                    warn!(%error, "operator-listener outbox reap failed");
+                }
+                tokio::time::sleep(OUTBOX_REAP_INTERVAL).await;
+            }
+        }
+    );
+}
+
+/// Continuously deliver operator-listener notification rows with bounded concurrency and retries.
+pub async fn run_delivery_worker(state: Arc<AppState>) {
+    let http = match build_delivery_http_client(state.config.operator_listener_timeout) {
+        Ok(http) => http,
+        Err(error) => {
+            error!(%error, "operator-listener HTTP client initialization failed");
+            return;
+        }
+    };
+    let mut idle_delay = IDLE_POLL_FLOOR;
+    loop {
+        match run_delivery_once(&state, &http).await {
+            WorkerIteration::Worked => idle_delay = IDLE_POLL_FLOOR,
+            WorkerIteration::Idle => {
+                tokio::time::sleep(idle_delay).await;
+                idle_delay = (idle_delay * 2).min(IDLE_POLL_CEILING);
+            }
+            WorkerIteration::Failed => tokio::time::sleep(Duration::from_secs(2)).await,
+        }
+    }
+}
+
+fn build_delivery_http_client(timeout: Duration) -> Result<reqwest::Client, reqwest::Error> {
+    reqwest::Client::builder()
+        .timeout(timeout)
+        .redirect(reqwest::redirect::Policy::none())
+        .build()
+}
+
+async fn run_delivery_once(state: &AppState, http: &reqwest::Client) -> WorkerIteration {
+    let transport = ReqwestDeliveryTransport { http };
+    run_delivery_once_with_transport(state, &transport).await
+}
+
+#[async_trait::async_trait]
+trait DeliveryTransport: Sync {
+    async fn post(
+        &self,
+        url: &url::Url,
+        authorization: &str,
+        body: Vec<u8>,
+    ) -> Result<StatusCode, String>;
+}
+
+#[async_trait::async_trait]
+trait DeliveryStore: Sync {
+    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>;
+    async fn fail(&self, id: Uuid, claim_id: Uuid) -> Result<bool, String>;
+}
+
+struct DbDeliveryStore<'a> {
+    db: &'a buzz_db::Db,
+}
+
+#[async_trait::async_trait]
+impl DeliveryStore for DbDeliveryStore<'_> {
+    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)
+            .await
+            .map_err(|error| error.to_string())
+    }
+
+    async fn complete(&self, id: Uuid, claim_id: Uuid) -> Result<bool, String> {
+        self.db
+            .complete_operator_listener_delivery(id, claim_id)
+            .await
+            .map_err(|error| error.to_string())
+    }
+
+    async fn retry(&self, id: Uuid, claim_id: Uuid, next: DateTime<Utc>) -> Result<bool, String> {
+        self.db
+            .retry_operator_listener_delivery(id, claim_id, next)
+            .await
+            .map_err(|error| error.to_string())
+    }
+
+    async fn fail(&self, id: Uuid, claim_id: Uuid) -> Result<bool, String> {
+        self.db
+            .fail_operator_listener_delivery(id, claim_id)
+            .await
+            .map_err(|error| error.to_string())
+    }
+}
+
+struct ReqwestDeliveryTransport<'a> {
+    http: &'a reqwest::Client,
+}
+
+#[async_trait::async_trait]
+impl DeliveryTransport for ReqwestDeliveryTransport<'_> {
+    async fn post(
+        &self,
+        url: &url::Url,
+        authorization: &str,
+        body: Vec<u8>,
+    ) -> Result<StatusCode, String> {
+        self.http
+            .post(url.clone())
+            .header("Authorization", authorization)
+            .header("Content-Type", "application/json")
+            .body(body)
+            .send()
+            .await
+            .map(|response| response.status())
+            .map_err(|error| error.to_string())
+    }
+}
+
+async fn run_delivery_once_with_transport<T: DeliveryTransport>(
+    state: &AppState,
+    transport: &T,
+) -> WorkerIteration {
+    let claimed = match state
+        .db
+        .claim_operator_listener_deliveries(
+            DELIVERY_BATCH_LIMIT,
+            Utc::now() + TimeDelta::seconds(CLAIM_SECS),
+        )
+        .await
+    {
+        Ok(claimed) => claimed,
+        Err(error) => {
+            error!(%error, "operator-listener delivery claim failed");
+            return WorkerIteration::Failed;
+        }
+    };
+    if claimed.is_empty() {
+        return WorkerIteration::Idle;
+    }
+    let store = DbDeliveryStore { db: &state.db };
+    join_all(claimed.into_iter().map(|delivery| {
+        deliver_one(
+            &state.config.operator_listener_delivery_urls,
+            &state.relay_keypair,
+            &store,
+            transport,
+            delivery,
+        )
+    }))
+    .await;
+    WorkerIteration::Worked
+}
+
+async fn deliver_one<T: DeliveryTransport, S: DeliveryStore>(
+    routes: &HashMap<String, url::Url>,
+    relay_keypair: &nostr::Keys,
+    store: &S,
+    transport: &T,
+    delivery: buzz_db::operator_listener::ClaimedDelivery,
+) {
+    let listener_hex = hex::encode(&delivery.listener_pubkey);
+    let Some(url) = routes.get(&listener_hex) else {
+        warn!(
+            delivery=%delivery.id,
+            listener=%listener_hex,
+            "operator-listener delivery has no route on this pod; releasing claim"
+        );
+        if let Err(error) = store
+            .release(
+                delivery.id,
+                delivery.claim_id,
+                Utc::now() + TimeDelta::seconds(CLAIM_SECS),
+            )
+            .await
+        {
+            error!(%error, delivery=%delivery.id, "failed to release unroutable operator-listener delivery");
+        }
+        metrics::counter!("buzz_operator_listener_deliveries_total", "outcome" => "unroutable")
+            .increment(1);
+        return;
+    };
+    let body = match serde_json::to_vec(&MentionNotification {
+        v: 1,
+        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(),
+    }) {
+        Ok(body) => body,
+        Err(error) => {
+            fail_permanently(
+                store,
+                &delivery,
+                &format!("notification encoding failed: {error}"),
+            )
+            .await;
+            return;
+        }
+    };
+    let auth = match nip98_header(relay_keypair, url.as_str(), &body) {
+        Ok(auth) => auth,
+        Err(error) => {
+            fail_permanently(
+                store,
+                &delivery,
+                &format!("notification auth failed: {error}"),
+            )
+            .await;
+            return;
+        }
+    };
+    let response = transport.post(url, &auth, body).await;
+    match response {
+        Ok(status) if status.is_success() => {
+            match store.complete(delivery.id, delivery.claim_id).await {
+                Ok(true) => {}
+                Ok(false) => warn!(
+                    delivery=%delivery.id,
+                    "operator-listener delivery completion lost its claim"
+                ),
+                Err(error) => error!(
+                    delivery=%delivery.id,
+                    %error,
+                    "failed to persist operator-listener delivery completion"
+                ),
+            }
+            metrics::counter!("buzz_operator_listener_deliveries_total", "outcome" => "accepted")
+                .increment(1);
+        }
+        Ok(status)
+            if status == StatusCode::REQUEST_TIMEOUT
+                || status == StatusCode::TOO_MANY_REQUESTS
+                || status.is_server_error() =>
+        {
+            retry_or_fail(store, &delivery, format!("HTTP {status}")).await
+        }
+        Ok(status) => {
+            fail_permanently(store, &delivery, &format!("HTTP {status}")).await;
+        }
+        Err(error) => retry_or_fail(store, &delivery, error.to_string()).await,
+    }
+}
+
+async fn fail_permanently<S: DeliveryStore>(
+    store: &S,
+    delivery: &buzz_db::operator_listener::ClaimedDelivery,
+    reason: &str,
+) {
+    error!(delivery=%delivery.id, %reason, "operator-listener delivery failed permanently");
+    if let Err(error) = store.fail(delivery.id, delivery.claim_id).await {
+        error!(delivery=%delivery.id, %error, "failed to delete terminal operator-listener delivery");
+    }
+    metrics::counter!("buzz_operator_listener_deliveries_total", "outcome" => "failed")
+        .increment(1);
+}
+
+async fn retry_or_fail<S: DeliveryStore>(
+    store: &S,
+    delivery: &buzz_db::operator_listener::ClaimedDelivery,
+    reason: String,
+) {
+    if delivery.attempt >= buzz_db::operator_listener::MAX_DELIVERY_ATTEMPTS {
+        fail_permanently(store, delivery, &format!("retries exhausted: {reason}")).await;
+        return;
+    }
+    let delay = 2_i64.pow((delivery.attempt - 1).clamp(0, 7) as u32);
+    warn!(
+        delivery=%delivery.id,
+        attempt=delivery.attempt,
+        retry_in_seconds=delay,
+        %reason,
+        "operator-listener delivery failed; retrying"
+    );
+    if let Err(error) = store
+        .retry(
+            delivery.id,
+            delivery.claim_id,
+            Utc::now() + TimeDelta::seconds(delay),
+        )
+        .await
+    {
+        error!(delivery=%delivery.id, %error, "failed to persist operator-listener retry");
+    }
+    metrics::counter!("buzz_operator_listener_deliveries_total", "outcome" => "retry").increment(1);
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Mutex;
+
+    use tokio::{
+        io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
+        net::{TcpListener, TcpStream},
+    };
+
+    use super::*;
+    use buzz_core::tenant::CommunityId;
+    use nostr::Keys;
+
+    #[test]
+    fn reapers_use_their_expected_intervals() {
+        assert_eq!(OUTBOX_REAP_INTERVAL, Duration::from_secs(5 * 60));
+        assert_eq!(
+            REGISTRATION_REAP_INTERVAL,
+            Duration::from_secs(24 * 60 * 60)
+        );
+    }
+
+    #[derive(Debug, Clone, PartialEq, Eq)]
+    enum StoreCall {
+        Release(Uuid, Uuid, DateTime<Utc>),
+        Complete(Uuid, Uuid),
+        Retry(Uuid, Uuid, DateTime<Utc>),
+        Fail(Uuid, Uuid),
+    }
+
+    #[derive(Default)]
+    struct MockStore {
+        calls: Mutex<Vec<StoreCall>>,
+    }
+
+    impl MockStore {
+        fn calls(&self) -> Vec<StoreCall> {
+            self.calls.lock().expect("store calls lock").clone()
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl DeliveryStore for MockStore {
+        async fn release(
+            &self,
+            id: Uuid,
+            claim_id: Uuid,
+            next: DateTime<Utc>,
+        ) -> Result<bool, String> {
+            self.calls
+                .lock()
+                .expect("store calls lock")
+                .push(StoreCall::Release(id, claim_id, next));
+            Ok(true)
+        }
+
+        async fn complete(&self, id: Uuid, claim_id: Uuid) -> Result<bool, String> {
+            self.calls
+                .lock()
+                .expect("store calls lock")
+                .push(StoreCall::Complete(id, claim_id));
+            Ok(true)
+        }
+
+        async fn retry(
+            &self,
+            id: Uuid,
+            claim_id: Uuid,
+            next: DateTime<Utc>,
+        ) -> Result<bool, String> {
+            self.calls
+                .lock()
+                .expect("store calls lock")
+                .push(StoreCall::Retry(id, claim_id, next));
+            Ok(true)
+        }
+
+        async fn fail(&self, id: Uuid, claim_id: Uuid) -> Result<bool, String> {
+            self.calls
+                .lock()
+                .expect("store calls lock")
+                .push(StoreCall::Fail(id, claim_id));
+            Ok(true)
+        }
+    }
+
+    struct SentRequest {
+        url: url::Url,
+        authorization: String,
+        body: Vec<u8>,
+    }
+
+    struct MockTransport {
+        response: Mutex<Option<Result<StatusCode, String>>>,
+        request: Mutex<Option<SentRequest>>,
+        request_count: Mutex<usize>,
+    }
+
+    impl MockTransport {
+        fn new(response: Result<StatusCode, String>) -> Self {
+            Self {
+                response: Mutex::new(Some(response)),
+                request: Mutex::new(None),
+                request_count: Mutex::new(0),
+            }
+        }
+
+        fn take_request(&self) -> Option<SentRequest> {
+            self.request.lock().expect("transport request lock").take()
+        }
+
+        fn request_count(&self) -> usize {
+            *self
+                .request_count
+                .lock()
+                .expect("transport request count lock")
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl DeliveryTransport for MockTransport {
+        async fn post(
+            &self,
+            url: &url::Url,
+            authorization: &str,
+            body: Vec<u8>,
+        ) -> Result<StatusCode, String> {
+            *self
+                .request_count
+                .lock()
+                .expect("transport request count lock") += 1;
+            *self.request.lock().expect("transport request lock") = Some(SentRequest {
+                url: url.clone(),
+                authorization: authorization.to_owned(),
+                body,
+            });
+            self.response
+                .lock()
+                .expect("transport response lock")
+                .take()
+                .expect("delivery should make one request")
+        }
+    }
+
+    fn delivery_fixture(
+        attempt: i32,
+    ) -> (
+        HashMap<String, url::Url>,
+        Keys,
+        buzz_db::operator_listener::ClaimedDelivery,
+    ) {
+        let listener = Keys::generate();
+        let listener_pubkey = listener.public_key().to_bytes().to_vec();
+        let route = url::Url::parse("https://listener.example/mentions").expect("test URL");
+        let routes = HashMap::from([(hex::encode(&listener_pubkey), route)]);
+        (
+            routes,
+            Keys::generate(),
+            buzz_db::operator_listener::ClaimedDelivery {
+                id: Uuid::new_v4(),
+                claim_id: Uuid::new_v4(),
+                listener_pubkey,
+                target_pubkey: vec![0x11; 32],
+                community: CommunityId::from_uuid(Uuid::new_v4()),
+                community_host: "community.example".to_owned(),
+                event_id: vec![0x22; 32],
+                event_kind: 9,
+                event_created_at: DateTime::from_timestamp(1_700_000_000, 0)
+                    .expect("test timestamp"),
+                attempt,
+            },
+        )
+    }
+
+    #[tokio::test]
+    async fn unroutable_delivery_releases_claim_without_http_request() {
+        let (_, relay_keypair, delivery) = delivery_fixture(1);
+        let store = MockStore::default();
+        let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT));
+        let before = Utc::now() + TimeDelta::seconds(CLAIM_SECS);
+
+        deliver_one(
+            &HashMap::new(),
+            &relay_keypair,
+            &store,
+            &transport,
+            delivery.clone(),
+        )
+        .await;
+
+        let after = Utc::now() + TimeDelta::seconds(CLAIM_SECS);
+        assert!(transport.take_request().is_none());
+        let calls = store.calls();
+        assert_eq!(calls.len(), 1);
+        match &calls[0] {
+            StoreCall::Release(id, claim_id, next) => {
+                assert_eq!(*id, delivery.id);
+                assert_eq!(*claim_id, delivery.claim_id);
+                assert!((before..=after).contains(next));
+            }
+            call => panic!("expected claim release, got {call:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn successful_delivery_posts_payload_and_completes_claim() {
+        let (routes, relay_keypair, delivery) = delivery_fixture(1);
+        let store = MockStore::default();
+        let transport = MockTransport::new(Ok(StatusCode::NO_CONTENT));
+
+        deliver_one(
+            &routes,
+            &relay_keypair,
+            &store,
+            &transport,
+            delivery.clone(),
+        )
+        .await;
+
+        let request = transport.take_request().expect("delivery request");
+        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["pubkey"], hex::encode(&delivery.target_pubkey));
+        assert_eq!(body["community_host"], delivery.community_host);
+        assert_eq!(body["event_id"], hex::encode(&delivery.event_id));
+        assert_eq!(body["event_kind"], delivery.event_kind);
+        assert_eq!(
+            body["event_created_at"],
+            delivery.event_created_at.timestamp()
+        );
+        assert_eq!(
+            store.calls(),
+            [StoreCall::Complete(delivery.id, delivery.claim_id)]
+        );
+    }
+
+    #[tokio::test]
+    async fn rate_limit_and_server_errors_retry_with_exponential_delay() {
+        for status in [
+            StatusCode::TOO_MANY_REQUESTS,
+            StatusCode::SERVICE_UNAVAILABLE,
+        ] {
+            let (routes, relay_keypair, delivery) = delivery_fixture(3);
+            let store = MockStore::default();
+            let transport = MockTransport::new(Ok(status));
+            let before = Utc::now() + TimeDelta::seconds(4);
+
+            deliver_one(
+                &routes,
+                &relay_keypair,
+                &store,
+                &transport,
+                delivery.clone(),
+            )
+            .await;
+
+            let after = Utc::now() + TimeDelta::seconds(4);
+            assert_eq!(transport.request_count(), 1, "HTTP {status}");
+            assert!(transport.take_request().is_some());
+            let calls = store.calls();
+            assert_eq!(calls.len(), 1, "HTTP {status}");
+            match &calls[0] {
+                StoreCall::Retry(id, claim_id, next) => {
+                    assert_eq!(*id, delivery.id);
+                    assert_eq!(*claim_id, delivery.claim_id);
+                    assert!((before..=after).contains(next));
+                }
+                call => panic!("expected HTTP {status} retry, got {call:?}"),
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn deterministic_client_error_fails_without_retrying() {
+        let (routes, relay_keypair, delivery) = delivery_fixture(1);
+        let store = MockStore::default();
+        let transport = MockTransport::new(Ok(StatusCode::BAD_REQUEST));
+
+        deliver_one(
+            &routes,
+            &relay_keypair,
+            &store,
+            &transport,
+            delivery.clone(),
+        )
+        .await;
+
+        assert_eq!(transport.request_count(), 1);
+        assert!(transport.take_request().is_some());
+        assert_eq!(
+            store.calls(),
+            [StoreCall::Fail(delivery.id, delivery.claim_id)]
+        );
+    }
+
+    async fn read_http_request(stream: &mut TcpStream) -> (String, Vec<u8>) {
+        let mut reader = BufReader::new(stream);
+        let mut request_line = String::new();
+        reader
+            .read_line(&mut request_line)
+            .await
+            .expect("read request line");
+
+        let mut content_length = 0;
+        loop {
+            let mut line = String::new();
+            reader.read_line(&mut line).await.expect("read header");
+            if line == "\r\n" || line.is_empty() {
+                break;
+            }
+            if let Some((name, value)) = line.split_once(':') {
+                if name.eq_ignore_ascii_case("content-length") {
+                    content_length = value.trim().parse().expect("valid content length");
+                }
+            }
+        }
+
+        let mut body = vec![0; content_length];
+        reader
+            .read_exact(&mut body)
+            .await
+            .expect("read request body");
+        (request_line, body)
+    }
+
+    #[tokio::test]
+    async fn redirect_response_fails_delivery_without_retrying() {
+        let source = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("source listener");
+        let source_addr = source.local_addr().expect("source address");
+        let target = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("target listener");
+        let target_addr = target.local_addr().expect("target address");
+        let target_url = format!("http://{target_addr}/redirected");
+        let target_task = tokio::spawn(async move {
+            match tokio::time::timeout(Duration::from_secs(1), target.accept()).await {
+                Ok(Ok((mut stream, _))) => {
+                    let request = read_http_request(&mut stream).await;
+                    stream
+                        .write_all(
+                            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
+                        )
+                        .await
+                        .expect("write target success response");
+                    Some(request)
+                }
+                Ok(Err(error)) => panic!("accept target request: {error}"),
+                Err(_) => None,
+            }
+        });
+        let (_, relay_keypair, delivery) = delivery_fixture(1);
+        let source_url =
+            url::Url::parse(&format!("http://{source_addr}/mentions")).expect("source URL");
+        let routes = HashMap::from([(hex::encode(&delivery.listener_pubkey), source_url)]);
+        let store = MockStore::default();
+        let http = build_delivery_http_client(Duration::from_secs(2)).expect("delivery client");
+        let transport = ReqwestDeliveryTransport { http: &http };
+
+        let source_task = tokio::spawn(async move {
+            let (mut stream, _) = source.accept().await.expect("accept source request");
+            let request = read_http_request(&mut stream).await;
+            let response = format!(
+                "HTTP/1.1 302 Found\r\nLocation: {target_url}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+            );
+            stream
+                .write_all(response.as_bytes())
+                .await
+                .expect("write redirect response");
+            let second_request = tokio::time::timeout(Duration::from_millis(250), source.accept())
+                .await
+                .is_ok();
+            (request, second_request)
+        });
+
+        deliver_one(
+            &routes,
+            &relay_keypair,
+            &store,
+            &transport,
+            delivery.clone(),
+        )
+        .await;
+
+        let (request, second_source_request) =
+            tokio::time::timeout(Duration::from_secs(2), source_task)
+                .await
+                .expect("source server completed")
+                .expect("source server task");
+        assert!(request.0.starts_with("POST /mentions HTTP/1.1"));
+        assert!(!request.1.is_empty());
+        assert!(
+            !second_source_request,
+            "source received more than one request"
+        );
+        let target_request = tokio::time::timeout(Duration::from_secs(2), target_task)
+            .await
+            .expect("target server completed")
+            .expect("target server task");
+        assert_eq!(
+            store.calls(),
+            [StoreCall::Fail(delivery.id, delivery.claim_id)]
+        );
+        assert!(
+            target_request.is_none(),
+            "redirect target unexpectedly received a request"
+        );
+    }
+
+    #[tokio::test]
+    async fn exhausted_delivery_attempts_fail_permanently() {
+        let (routes, relay_keypair, delivery) =
+            delivery_fixture(buzz_db::operator_listener::MAX_DELIVERY_ATTEMPTS);
+        let store = MockStore::default();
+        let transport = MockTransport::new(Err("connection reset".to_owned()));
+
+        deliver_one(
+            &routes,
+            &relay_keypair,
+            &store,
+            &transport,
+            delivery.clone(),
+        )
+        .await;
+
+        assert!(transport.take_request().is_some());
+        assert_eq!(
+            store.calls(),
+            [StoreCall::Fail(delivery.id, delivery.claim_id)]
+        );
+    }
+}

crates/buzz-relay/src/push_runtime.rs

@@ -5,15 +5,13 @@ use std::{
     time::{Duration, Instant},
 };
 
-use base64::Engine as _;
 use buzz_core::filter::{filters_match, reader_authorized_for_event};
 use chrono::{TimeDelta, Utc};
-use nostr::{EventBuilder, Filter, Kind, Tag};
+use nostr::Filter;
 use serde::{Deserialize, Serialize};
-use sha2::{Digest as _, Sha256};
 use tracing::{error, warn};
 
-use crate::{handlers::push_lease::Subscription, state::AppState};
+use crate::{handlers::push_lease::Subscription, nip98::nip98_header, state::AppState};
 
 const CLAIM_SECS: i64 = 30;
 const EVENT_USEFUL_SECS: i64 = 3600;
@@ -666,22 +664,6 @@ async fn retry_or_fail(
     }
 }
 
-fn nip98_header(keys: &nostr::Keys, url: &str, body: &[u8]) -> anyhow::Result<String> {
-    let hash = hex::encode(Sha256::digest(body));
-    let event = EventBuilder::new(Kind::HttpAuth, "")
-        .tags([
-            Tag::parse(["u", url])?,
-            Tag::parse(["method", "POST"])?,
-            Tag::parse(["payload", &hash])?,
-            Tag::parse(["nonce", &uuid::Uuid::new_v4().to_string()])?,
-        ])
-        .sign_with_keys(keys)?;
-    Ok(format!(
-        "Nostr {}",
-        base64::engine::general_purpose::STANDARD.encode(serde_json::to_vec(&event)?)
-    ))
-}
-
 fn class_rank(_: &str) -> u8 {
     1
 }
@@ -696,6 +678,7 @@ mod tests {
         routing::post,
         Json, Router,
     };
+    use nostr::{EventBuilder, Kind, Tag};
     use serde_json::Value;
     use std::{future::IntoFuture, sync::Arc};
     use tokio::sync::Mutex;

crates/buzz-relay/src/router.rs

@@ -317,6 +317,11 @@ pub fn build_router(state: Arc<AppState>) -> Router {
             "/operator/communities",
             get(api::operator::list_owned_communities).post(api::operator::provision_community),
         )
+        .route(
+            "/operator/listener/pubkeys",
+            post(api::operator::register_listener_pubkeys)
+                .delete(api::operator::remove_listener_pubkeys),
+        )
         .route(
             "/operator/communities/archive",
             post(api::operator::archive_community),

migrations/0050_operator_listener_mentions.sql

@@ -0,0 +1,57 @@
+-- Deployment-global operator-listener mention delivery.
+-- Listener registrations span communities; community_id on the outbox is event
+-- provenance, not a tenant boundary.
+
+-- Migration 0044 restored this function to the pre-0041 exclusion set. Extend
+-- it here so migrated databases keep the outbox outside tenant fencing and
+-- community-deletion catalog discovery, matching the desired-state schema.
+CREATE OR REPLACE FUNCTION community_write_fence_excluded_table(target NAME) RETURNS BOOLEAN
+LANGUAGE SQL IMMUTABLE STRICT PARALLEL SAFE AS $$
+    SELECT target::TEXT = ANY (ARRAY[
+        'community_deletion_requests', 'community_deletion_approvals',
+        'community_deletion_checkpoints', 'community_serving_write_leases',
+        'community_deletion_executor_heartbeats', 'product_feedback',
+        'rate_limit_violations', 'operator_listener_outbox'
+    ]::TEXT[])
+$$;
+
+CREATE TABLE operator_listener_pubkeys (
+    listener_pubkey BYTEA NOT NULL CHECK (length(listener_pubkey) = 32),
+    target_pubkey   BYTEA NOT NULL CHECK (length(target_pubkey) = 32),
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (listener_pubkey, target_pubkey)
+);
+CREATE INDEX operator_listener_pubkeys_target
+    ON operator_listener_pubkeys (target_pubkey, listener_pubkey);
+CREATE INDEX operator_listener_pubkeys_created_at
+    ON operator_listener_pubkeys (created_at);
+
+CREATE TABLE operator_listener_outbox (
+    id                UUID NOT NULL PRIMARY KEY DEFAULT gen_random_uuid(),
+    listener_pubkey   BYTEA NOT NULL CHECK (length(listener_pubkey) = 32),
+    target_pubkey     BYTEA NOT NULL CHECK (length(target_pubkey) = 32),
+    community_id      UUID NOT NULL,
+    event_id          BYTEA NOT NULL CHECK (length(event_id) = 32),
+    event_kind        INTEGER NOT NULL,
+    event_created_at  TIMESTAMPTZ NOT NULL,
+    state             TEXT NOT NULL DEFAULT 'pending'
+                      CHECK (state IN ('pending', 'sending')),
+    attempts          INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+    next_attempt_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    lease_until       TIMESTAMPTZ,
+    claim_id          UUID,
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (listener_pubkey, target_pubkey, community_id, event_id)
+);
+CREATE INDEX operator_listener_outbox_due
+    ON operator_listener_outbox (next_attempt_at, created_at, id)
+    WHERE state = 'pending';
+CREATE INDEX operator_listener_outbox_recovery
+    ON operator_listener_outbox (lease_until, created_at, id)
+    WHERE state = 'sending';
+CREATE INDEX operator_listener_outbox_created_at
+    ON operator_listener_outbox (created_at);
+
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('operator_listener_pubkeys', 'deployment-global target registrations for operator listeners'),
+    ('operator_listener_outbox', 'deployment-global mention delivery queue; community_id is event provenance');

schema/schema.sql

@@ -1494,7 +1494,7 @@ LANGUAGE SQL IMMUTABLE STRICT PARALLEL SAFE AS $$
         'community_deletion_requests', 'community_deletion_approvals',
         'community_deletion_checkpoints', 'community_serving_write_leases',
         'community_deletion_executor_heartbeats', 'product_feedback',
-        'rate_limit_violations'
+        'rate_limit_violations', 'operator_listener_outbox'
     ]::TEXT[])
 $$;
 
@@ -1878,6 +1878,51 @@ CREATE INDEX idx_relay_admin_outbox_pending
 INSERT INTO _operator_global_tables (table_name, reason) VALUES
     ('relay_admin_outbox', 'deployment-global enforcement artifact delivery queue');
 
+-- ── Operator-listener mention delivery ──────────────────────────────────────
+-- Listener registrations are deployment-global. The outbox records community
+-- provenance for the event, but is intentionally not tenant-owned.
+
+CREATE TABLE operator_listener_pubkeys (
+    listener_pubkey BYTEA NOT NULL CHECK (length(listener_pubkey) = 32),
+    target_pubkey   BYTEA NOT NULL CHECK (length(target_pubkey) = 32),
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (listener_pubkey, target_pubkey)
+);
+CREATE INDEX operator_listener_pubkeys_target
+    ON operator_listener_pubkeys (target_pubkey, listener_pubkey);
+CREATE INDEX operator_listener_pubkeys_created_at
+    ON operator_listener_pubkeys (created_at);
+
+CREATE TABLE operator_listener_outbox (
+    id                UUID NOT NULL PRIMARY KEY DEFAULT gen_random_uuid(),
+    listener_pubkey   BYTEA NOT NULL CHECK (length(listener_pubkey) = 32),
+    target_pubkey     BYTEA NOT NULL CHECK (length(target_pubkey) = 32),
+    community_id      UUID NOT NULL,
+    event_id          BYTEA NOT NULL CHECK (length(event_id) = 32),
+    event_kind        INTEGER NOT NULL,
+    event_created_at  TIMESTAMPTZ NOT NULL,
+    state             TEXT NOT NULL DEFAULT 'pending'
+                      CHECK (state IN ('pending', 'sending')),
+    attempts          INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
+    next_attempt_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    lease_until       TIMESTAMPTZ,
+    claim_id          UUID,
+    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (listener_pubkey, target_pubkey, community_id, event_id)
+);
+CREATE INDEX operator_listener_outbox_due
+    ON operator_listener_outbox (next_attempt_at, created_at, id)
+    WHERE state = 'pending';
+CREATE INDEX operator_listener_outbox_recovery
+    ON operator_listener_outbox (lease_until, created_at, id)
+    WHERE state = 'sending';
+CREATE INDEX operator_listener_outbox_created_at
+    ON operator_listener_outbox (created_at);
+
+INSERT INTO _operator_global_tables (table_name, reason) VALUES
+    ('operator_listener_pubkeys', 'deployment-global target registrations for operator listeners'),
+    ('operator_listener_outbox', 'deployment-global mention delivery queue; community_id is event provenance');
+
 -- ── Relay operator audit (append-only roster mutation trail) ─────────────────
 -- One row per PUT/DELETE /operators/{pubkey} mutation. The roster is the
 -- deployment-wide root of trust and its mutations overwrite/remove in place;