feat(relay): deliver pubkey mentions to relay companions

.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,9 @@ 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, 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, 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(), 48);
+        assert_eq!(migrations.len(), 49);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1288,10 +1290,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[48].version, 49);
+        assert!(migrations[48]
+            .sql
+            .as_str()
+            .contains("CREATE TABLE operator_listener_pubkeys"));
+        assert!(migrations[48]
+            .sql
+            .as_str()
+            .contains("CREATE TABLE operator_listener_outbox"));
+        assert!(migrations[48]
+            .sql
+            .as_str()
+            .contains("CREATE OR REPLACE FUNCTION community_write_fence_excluded_table"));
+        assert!(migrations[48]
+            .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[48].sql.as_str().contains(index),
+                "migration 0049 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[48].sql.as_str().contains(index_shape),
+                "migration 0049 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/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(
@@ -1497,6 +1504,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.