diff --git a/.env.example b/.env.example index 683bceb4cc0..7ec2e35be01 100644 --- a/.env.example +++ b/.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. diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index f749d1a6255..ce4a947d711 100644 --- a/crates/buzz-db/src/lib.rs +++ b/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; diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 6e5e14c2c62..8fde5d5d5a9 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/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" ); } diff --git a/crates/buzz-db/src/runtime/tests.rs b/crates/buzz-db/src/runtime/tests.rs index cce1927be69..2ae0791f213 100644 --- a/crates/buzz-db/src/runtime/tests.rs +++ b/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 diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index a6c3eae8f0a..f8968f5b84f 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -355,12 +355,15 @@ pub async fn insert_event( event: &Event, channel_id: Option, ) -> 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, ) -> 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(( diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 3aeb2257aae..99dc734e63c 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/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. diff --git a/crates/buzz-db/src/store/operator_listener.rs b/crates/buzz-db/src/store/operator_listener.rs new file mode 100644 index 00000000000..7b08a34572d --- /dev/null +++ b/crates/buzz-db/src/store/operator_listener.rs @@ -0,0 +1,1036 @@ +//! Deployment-global operator-listener mention registrations and delivery outbox. + +use std::collections::HashSet; + +use buzz_datastore_tracing::datastore_span; +use chrono::{DateTime, TimeDelta, Utc}; +use nostr::Event; +use sqlx::{PgPool, Postgres, QueryBuilder, Row as _, Transaction}; +use uuid::Uuid; + +use crate::error::Result; +use crate::{observability, CommunityId, Db}; + +/// Event kinds whose `p` tags represent user-visible message mentions. +pub const OPERATOR_LISTENER_MENTION_KINDS: [u32; 4] = [9, 40002, 45001, 45003]; +/// Maximum notification attempts before a delivery becomes terminally failed. +pub const MAX_DELIVERY_ATTEMPTS: i32 = 9; +/// Maximum time a queued delivery remains useful when no worker can route it. +pub const OUTBOX_RETENTION: TimeDelta = TimeDelta::minutes(15); +/// Registration lifetime before the daily cleanup removes it. +pub const PUBKEY_RETENTION: TimeDelta = TimeDelta::days(30); + +async fn acquire_maintenance_writer( + pool: &PgPool, +) -> sqlx::Result> { + observability::acquire_writer(pool, observability::WriterOperation::Maintenance).await +} + +/// A claimed operator-listener notification. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClaimedDelivery { + /// Durable delivery identifier and claim fence key. + pub id: Uuid, + /// Claim fencing token. + pub claim_id: Uuid, + /// Configured operator-listener identity. + pub listener_pubkey: Vec, + /// Registered target identity mentioned by the event. + pub target_pubkey: Vec, + /// Community containing the event. + pub community: CommunityId, + /// Community host used in the notification payload. + pub community_host: String, + /// Mentioning event id. + pub event_id: Vec, + /// Mentioning event kind. + pub event_kind: i32, + /// Author-controlled event timestamp. + pub event_created_at: DateTime, + /// Attempt number, starting at one. + pub attempt: i32, +} + +/// Return whether an event kind should generate operator-listener notifications. +#[must_use] +pub const fn is_listener_mention_kind(kind: u32) -> bool { + let mut index = 0; + while index < OPERATOR_LISTENER_MENTION_KINDS.len() { + if OPERATOR_LISTENER_MENTION_KINDS[index] == kind { + return true; + } + index += 1; + } + false +} + +fn event_targets(event: &Event) -> Vec> { + let mut targets = Vec::new(); + for tag in event.tags.iter() { + let parts = tag.as_slice(); + if parts.first().map(String::as_str) != Some("p") { + continue; + } + let Some(value) = parts.get(1) else { + continue; + }; + let Ok(target) = hex::decode(value) else { + continue; + }; + if target.len() == 32 && !targets.iter().any(|known| known == &target) { + targets.push(target); + } + } + targets +} + +/// Insert one outbox row for each registered listener matching the event's +/// `p` tags. The caller owns the transaction and must commit it with the event. +pub(crate) async fn enqueue_mentions_in_transaction( + tx: &mut Transaction<'_, Postgres>, + community_id: CommunityId, + event: &Event, +) -> Result { + let kind = u32::from(event.kind.as_u16()); + if !is_listener_mention_kind(kind) { + return Ok(0); + } + let targets = event_targets(event); + if targets.is_empty() { + return Ok(0); + } + let event_created_at = DateTime::from_timestamp(event.created_at.as_secs() as i64, 0).ok_or( + crate::error::DbError::InvalidTimestamp(event.created_at.as_secs() as i64), + )?; + + let mut query = QueryBuilder::::new( + "INSERT INTO operator_listener_outbox \ + (listener_pubkey, target_pubkey, community_id, event_id, event_kind, event_created_at) \ + SELECT r.listener_pubkey, r.target_pubkey, ", + ); + query + .push_bind(community_id.as_uuid()) + .push(", ") + .push_bind(event.id.as_bytes().as_slice()) + .push(", ") + .push_bind(kind as i32) + .push(", ") + .push_bind(event_created_at) + .push(" FROM operator_listener_pubkeys r WHERE r.target_pubkey IN ("); + let mut separated = query.separated(", "); + for target in &targets { + separated.push_bind(target.as_slice()); + } + separated.push_unseparated(") ON CONFLICT DO NOTHING"); + + Ok(query.build().execute(&mut **tx).await?.rows_affected()) +} + +/// Register target pubkeys for one deployment-global listener. +pub async fn register_pubkeys( + pool: &PgPool, + listener_pubkey: &[u8], + target_pubkeys: &[Vec], +) -> Result { + if target_pubkeys.is_empty() { + return Ok(0); + } + let mut seen = HashSet::with_capacity(target_pubkeys.len()); + let unique_target_pubkeys: Vec<&[u8]> = target_pubkeys + .iter() + .filter_map(|target| seen.insert(target.as_slice()).then_some(target.as_slice())) + .collect(); + let mut connection = acquire_maintenance_writer(pool).await?; + let mut query = QueryBuilder::::new( + "INSERT INTO operator_listener_pubkeys (listener_pubkey, target_pubkey) ", + ); + query.push_values(unique_target_pubkeys, |mut bind, target| { + bind.push_bind(listener_pubkey).push_bind(target); + }); + query.push( + " ON CONFLICT (listener_pubkey, target_pubkey) DO UPDATE SET created_at = EXCLUDED.created_at", + ); + Ok(query + .build() + .execute(&mut *connection) + .await? + .rows_affected()) +} + +/// Remove target pubkeys for one deployment-global listener. +pub async fn remove_pubkeys( + pool: &PgPool, + listener_pubkey: &[u8], + target_pubkeys: &[Vec], +) -> Result { + if target_pubkeys.is_empty() { + return Ok(0); + } + let mut connection = acquire_maintenance_writer(pool).await?; + let mut query = QueryBuilder::::new( + "DELETE FROM operator_listener_pubkeys WHERE listener_pubkey = ", + ); + query + .push_bind(listener_pubkey) + .push(" AND target_pubkey IN ("); + let mut separated = query.separated(", "); + for target in target_pubkeys { + separated.push_bind(target.as_slice()); + } + separated.push_unseparated(")"); + Ok(query + .build() + .execute(&mut *connection) + .await? + .rows_affected()) +} + +/// Delete registrations older than the operator-listener retention period. +pub async fn delete_expired_pubkeys(pool: &PgPool) -> Result { + let mut connection = acquire_maintenance_writer(pool).await?; + let cutoff = Utc::now() - PUBKEY_RETENTION; + Ok( + sqlx::query("DELETE FROM operator_listener_pubkeys WHERE created_at < $1") + .bind(cutoff) + .execute(&mut *connection) + .await? + .rows_affected(), + ) +} + +/// Claim due notification deliveries and recover claims whose visibility timeout expired. +pub async fn claim_deliveries( + pool: &PgPool, + limit: i64, + lease_until: DateTime, +) -> Result> { + let claim_id = Uuid::new_v4(); + let cutoff = Utc::now() - OUTBOX_RETENTION; + let mut connection = acquire_maintenance_writer(pool).await?; + let rows = sqlx::query( + "WITH candidates AS ( \ + SELECT o.id, o.community_id \ + FROM operator_listener_outbox o \ + WHERE o.attempts < $2 \ + AND o.next_attempt_at <= now() \ + AND o.created_at >= $4 \ + AND (o.state = 'pending' OR (o.state = 'sending' AND o.lease_until < now())) \ + ORDER BY o.next_attempt_at, o.created_at, o.id \ + FOR UPDATE OF o SKIP LOCKED \ + LIMIT $3 \ + ) \ + UPDATE operator_listener_outbox o \ + SET state = 'sending', claim_id = $1, lease_until = $5, attempts = o.attempts + 1 \ + FROM candidates c \ + JOIN communities community ON community.id = c.community_id \ + WHERE o.id = c.id \ + RETURNING o.id, o.claim_id, o.listener_pubkey, o.target_pubkey, o.community_id, \ + community.host, o.event_id, o.event_kind, o.event_created_at, o.attempts", + ) + .bind(claim_id) + .bind(MAX_DELIVERY_ATTEMPTS) + .bind(limit) + .bind(cutoff) + .bind(lease_until) + .fetch_all(&mut *connection) + .await?; + + rows.into_iter() + .map(|row| { + Ok(ClaimedDelivery { + id: row.try_get("id")?, + claim_id: row.try_get("claim_id")?, + listener_pubkey: row.try_get("listener_pubkey")?, + target_pubkey: row.try_get("target_pubkey")?, + community: CommunityId::from_uuid(row.try_get("community_id")?), + community_host: row.try_get("host")?, + event_id: row.try_get("event_id")?, + event_kind: row.try_get("event_kind")?, + event_created_at: row.try_get("event_created_at")?, + attempt: row.try_get("attempts")?, + }) + }) + .collect() +} + +/// Release a delivery claim that this pod cannot route. +pub async fn release_unroutable_delivery( + pool: &PgPool, + id: Uuid, + claim_id: Uuid, + next: DateTime, +) -> Result { + let mut connection = acquire_maintenance_writer(pool).await?; + Ok(sqlx::query( + "UPDATE operator_listener_outbox \ + SET state = 'pending', claim_id = NULL, lease_until = NULL, \ + next_attempt_at = $3, attempts = GREATEST(attempts - 1, 0) \ + WHERE id = $1 AND claim_id = $2 AND state = 'sending'", + ) + .bind(id) + .bind(claim_id) + .bind(next) + .execute(&mut *connection) + .await? + .rows_affected() + == 1) +} + +/// Mark one fenced delivery successful. +pub async fn complete_delivery(pool: &PgPool, id: Uuid, claim_id: Uuid) -> Result { + delete_claimed_delivery(pool, id, claim_id).await +} + +/// Retry one fenced delivery after a transient failure. +pub async fn retry_delivery( + pool: &PgPool, + id: Uuid, + claim_id: Uuid, + next: DateTime, +) -> Result { + let mut connection = acquire_maintenance_writer(pool).await?; + Ok(sqlx::query( + "UPDATE operator_listener_outbox \ + SET state = 'pending', claim_id = NULL, lease_until = NULL, next_attempt_at = $3 \ + WHERE id = $1 AND claim_id = $2 AND state = 'sending' AND attempts < $4", + ) + .bind(id) + .bind(claim_id) + .bind(next) + .bind(MAX_DELIVERY_ATTEMPTS) + .execute(&mut *connection) + .await? + .rows_affected() + == 1) +} + +/// Delete one fenced delivery after all delivery attempts fail. +pub async fn fail_delivery(pool: &PgPool, id: Uuid, claim_id: Uuid) -> Result { + delete_claimed_delivery(pool, id, claim_id).await +} + +async fn delete_claimed_delivery(pool: &PgPool, id: Uuid, claim_id: Uuid) -> Result { + let mut connection = acquire_maintenance_writer(pool).await?; + Ok(sqlx::query( + "DELETE FROM operator_listener_outbox \ + WHERE id = $1 AND claim_id = $2 AND state = 'sending'", + ) + .bind(id) + .bind(claim_id) + .execute(&mut *connection) + .await? + .rows_affected() + == 1) +} + +/// Delete delivery rows that have exceeded the outbox retention window. +pub async fn reap_deliveries(pool: &PgPool) -> Result { + let mut connection = acquire_maintenance_writer(pool).await?; + let cutoff = Utc::now() - OUTBOX_RETENTION; + Ok( + sqlx::query("DELETE FROM operator_listener_outbox WHERE created_at < $1") + .bind(cutoff) + .execute(&mut *connection) + .await? + .rows_affected(), + ) +} + +impl Db { + /// Register target pubkeys for one configured operator listener. + #[datastore_span(name = "register_operator_listener_pubkeys", system = "postgresql")] + pub async fn register_operator_listener_pubkeys( + &self, + listener_pubkey: &[u8], + target_pubkeys: &[Vec], + ) -> Result { + register_pubkeys(&self.pool, listener_pubkey, target_pubkeys).await + } + + /// Remove target pubkeys for one configured operator listener. + #[datastore_span(name = "remove_operator_listener_pubkeys", system = "postgresql")] + pub async fn remove_operator_listener_pubkeys( + &self, + listener_pubkey: &[u8], + target_pubkeys: &[Vec], + ) -> Result { + remove_pubkeys(&self.pool, listener_pubkey, target_pubkeys).await + } + + /// Claim due operator-listener deliveries. + #[datastore_span(name = "claim_operator_listener_deliveries", system = "postgresql")] + pub async fn claim_operator_listener_deliveries( + &self, + limit: i64, + lease_until: DateTime, + ) -> Result> { + claim_deliveries(&self.pool, limit, lease_until).await + } + + /// Mark one operator-listener delivery successful. + #[datastore_span(name = "complete_operator_listener_delivery", system = "postgresql")] + pub async fn complete_operator_listener_delivery( + &self, + id: Uuid, + claim_id: Uuid, + ) -> Result { + complete_delivery(&self.pool, id, claim_id).await + } + + /// Retry one operator-listener delivery. + #[datastore_span(name = "retry_operator_listener_delivery", system = "postgresql")] + pub async fn retry_operator_listener_delivery( + &self, + id: Uuid, + claim_id: Uuid, + next: DateTime, + ) -> Result { + retry_delivery(&self.pool, id, claim_id, next).await + } + + /// Release a delivery claim that this pod cannot route. + #[datastore_span(name = "release_operator_listener_delivery", system = "postgresql")] + pub async fn release_operator_listener_delivery( + &self, + id: Uuid, + claim_id: Uuid, + next: DateTime, + ) -> Result { + release_unroutable_delivery(&self.pool, id, claim_id, next).await + } + + /// Delete one operator-listener delivery after terminal failure. + #[datastore_span(name = "fail_operator_listener_delivery", system = "postgresql")] + pub async fn fail_operator_listener_delivery(&self, id: Uuid, claim_id: Uuid) -> Result { + fail_delivery(&self.pool, id, claim_id).await + } + + /// Reap stale deliveries. + #[datastore_span(name = "reap_operator_listener_deliveries", system = "postgresql")] + pub async fn reap_operator_listener_deliveries(&self) -> Result { + reap_deliveries(&self.pool).await + } + + /// Delete operator-listener registrations older than thirty days. + #[datastore_span( + name = "delete_expired_operator_listener_pubkeys", + system = "postgresql" + )] + pub async fn delete_expired_operator_listener_pubkeys(&self) -> Result { + delete_expired_pubkeys(&self.pool).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_message_kinds_are_listener_mentions() { + for kind in OPERATOR_LISTENER_MENTION_KINDS { + assert!(is_listener_mention_kind(kind)); + } + assert!(!is_listener_mention_kind(1)); + assert!(!is_listener_mention_kind(40003)); + } +} + +#[cfg(test)] +mod postgres_tests { + use super::*; + use nostr::{EventBuilder, Keys, Kind, Tag}; + use sqlx::PgPool; + + async fn setup_pool() -> PgPool { + PgPool::connect(&crate::test_support::database_url()) + .await + .expect("connect to test DB") + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn event_insert_enqueues_registered_listener_in_same_transaction() { + let pool = setup_pool().await; + let community_id = Uuid::new_v4(); + let listener = Keys::generate(); + let target = Keys::generate(); + let target_pubkey = target.public_key().to_bytes().to_vec(); + + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!( + "operator-listener-test-{}.example", + community_id.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + register_pubkeys( + &pool, + listener.public_key().as_bytes(), + std::slice::from_ref(&target_pubkey), + ) + .await + .expect("register listener target"); + + let event = EventBuilder::new(Kind::Custom(9), "mention") + .tag(Tag::public_key(target.public_key())) + .sign_with_keys(&Keys::generate()) + .expect("sign test event"); + let (_, inserted) = + crate::event::insert_event(&pool, CommunityId::from_uuid(community_id), &event, None) + .await + .expect("insert event"); + assert!(inserted); + + let outbox_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM operator_listener_outbox WHERE community_id = $1 AND event_id = $2", + ) + .bind(community_id) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("count listener deliveries"); + assert_eq!(outbox_count, 1); + + sqlx::query("DELETE FROM operator_listener_outbox WHERE community_id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test outbox rows"); + sqlx::query("DELETE FROM operator_listener_pubkeys WHERE listener_pubkey = $1") + .bind(listener.public_key().as_bytes().as_slice()) + .execute(&pool) + .await + .expect("delete test registration"); + sqlx::query("DELETE FROM event_mentions WHERE community_id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test mentions"); + sqlx::query("DELETE FROM events WHERE community_id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test event"); + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test community"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn thread_metadata_insert_fans_out_once_to_matching_listeners() { + let pool = setup_pool().await; + let community_id = Uuid::new_v4(); + let first_listener = Keys::generate(); + let second_listener = Keys::generate(); + let unrelated_listener = Keys::generate(); + let target = Keys::generate(); + let unrelated_target = Keys::generate(); + let target_pubkey = target.public_key().to_bytes().to_vec(); + let unrelated_pubkey = unrelated_target.public_key().to_bytes().to_vec(); + + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!( + "operator-listener-thread-test-{}.example", + community_id.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + for listener in [&first_listener, &second_listener] { + register_pubkeys( + &pool, + listener.public_key().as_bytes(), + std::slice::from_ref(&target_pubkey), + ) + .await + .expect("register matching listener target"); + } + register_pubkeys( + &pool, + unrelated_listener.public_key().as_bytes(), + std::slice::from_ref(&unrelated_pubkey), + ) + .await + .expect("register unrelated listener target"); + + let event = EventBuilder::new(Kind::Custom(9), "thread mention") + .tags([ + Tag::public_key(target.public_key()), + Tag::public_key(target.public_key()), + ]) + .sign_with_keys(&Keys::generate()) + .expect("sign test event"); + let community = CommunityId::from_uuid(community_id); + let (_, inserted) = + crate::event::insert_event_with_thread_metadata(&pool, community, &event, None, None) + .await + .expect("insert event through thread-metadata path"); + assert!(inserted); + + let delivered_listeners: Vec> = sqlx::query_scalar( + "SELECT listener_pubkey FROM operator_listener_outbox WHERE community_id = $1 AND event_id = $2 ORDER BY listener_pubkey", + ) + .bind(community_id) + .bind(event.id.as_bytes().as_slice()) + .fetch_all(&pool) + .await + .expect("read listener fanout"); + let mut expected_listeners = vec![ + first_listener.public_key().to_bytes().to_vec(), + second_listener.public_key().to_bytes().to_vec(), + ]; + expected_listeners.sort(); + assert_eq!(delivered_listeners, expected_listeners); + + let (_, duplicate_inserted) = + crate::event::insert_event_with_thread_metadata(&pool, community, &event, None, None) + .await + .expect("retry duplicate event through thread-metadata path"); + assert!(!duplicate_inserted); + let outbox_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM operator_listener_outbox WHERE community_id = $1 AND event_id = $2", + ) + .bind(community_id) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("count listener deliveries after duplicate event"); + assert_eq!(outbox_count, 2); + + sqlx::query("DELETE FROM operator_listener_outbox WHERE community_id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test outbox rows"); + sqlx::query("DELETE FROM operator_listener_pubkeys WHERE listener_pubkey IN ($1, $2, $3)") + .bind(first_listener.public_key().as_bytes().as_slice()) + .bind(second_listener.public_key().as_bytes().as_slice()) + .bind(unrelated_listener.public_key().as_bytes().as_slice()) + .execute(&pool) + .await + .expect("delete test registrations"); + sqlx::query("DELETE FROM events WHERE community_id = $1 AND id = $2") + .bind(community_id) + .bind(event.id.as_bytes().as_slice()) + .execute(&pool) + .await + .expect("delete test event"); + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test community"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn event_and_listener_outbox_roll_back_together() { + let pool = setup_pool().await; + let community_id = Uuid::new_v4(); + let listener = Keys::generate(); + let target = Keys::generate(); + let target_pubkey = target.public_key().to_bytes().to_vec(); + + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!( + "operator-listener-rollback-test-{}.example", + community_id.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + register_pubkeys( + &pool, + listener.public_key().as_bytes(), + std::slice::from_ref(&target_pubkey), + ) + .await + .expect("register listener target"); + let event = EventBuilder::new(Kind::Custom(9), "mention") + .tag(Tag::public_key(target.public_key())) + .sign_with_keys(&Keys::generate()) + .expect("sign test event"); + + let mut tx = pool.begin().await.expect("begin event transaction"); + let (_, inserted) = crate::event::insert_event_in_transaction( + &mut tx, + CommunityId::from_uuid(community_id), + &event, + None, + ) + .await + .expect("insert event and enqueue mention"); + assert!(inserted); + tx.rollback().await.expect("roll back event transaction"); + + let event_count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM events WHERE community_id = $1 AND id = $2") + .bind(community_id) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("count rolled-back event"); + let outbox_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM operator_listener_outbox WHERE community_id = $1 AND event_id = $2", + ) + .bind(community_id) + .bind(event.id.as_bytes().as_slice()) + .fetch_one(&pool) + .await + .expect("count rolled-back delivery"); + assert_eq!(event_count, 0); + assert_eq!(outbox_count, 0); + + sqlx::query("DELETE FROM operator_listener_pubkeys WHERE listener_pubkey = $1") + .bind(listener.public_key().as_bytes().as_slice()) + .execute(&pool) + .await + .expect("delete test registration"); + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test community"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn repeated_registration_renews_created_at() { + let pool = setup_pool().await; + let listener = Keys::generate(); + let target = Keys::generate(); + let listener_pubkey = listener.public_key().as_bytes().to_vec(); + let target_pubkey = target.public_key().as_bytes().to_vec(); + + register_pubkeys( + &pool, + &listener_pubkey, + std::slice::from_ref(&target_pubkey), + ) + .await + .expect("register listener target"); + + let expired_at = Utc::now() - PUBKEY_RETENTION - TimeDelta::hours(1); + sqlx::query( + "UPDATE operator_listener_pubkeys SET created_at = $3 \ + WHERE listener_pubkey = $1 AND target_pubkey = $2", + ) + .bind(&listener_pubkey) + .bind(&target_pubkey) + .bind(expired_at) + .execute(&pool) + .await + .expect("age test registration"); + + register_pubkeys( + &pool, + &listener_pubkey, + std::slice::from_ref(&target_pubkey), + ) + .await + .expect("renew listener target"); + + let renewed_at: DateTime = sqlx::query_scalar( + "SELECT created_at FROM operator_listener_pubkeys \ + WHERE listener_pubkey = $1 AND target_pubkey = $2", + ) + .bind(&listener_pubkey) + .bind(&target_pubkey) + .fetch_one(&pool) + .await + .expect("read renewed registration"); + assert!(renewed_at > expired_at); + + sqlx::query( + "DELETE FROM operator_listener_pubkeys \ + WHERE listener_pubkey = $1 AND target_pubkey = $2", + ) + .bind(&listener_pubkey) + .bind(&target_pubkey) + .execute(&pool) + .await + .expect("delete test registration"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn claim_deliveries_skips_rows_past_retention() { + let pool = setup_pool().await; + let community_id = Uuid::new_v4(); + let listener = Keys::generate(); + let target = Keys::generate(); + let stale_id = Uuid::new_v4(); + let fresh_id = Uuid::new_v4(); + let stale_at = Utc::now() - OUTBOX_RETENTION - TimeDelta::seconds(1); + let event_created_at = Utc::now(); + + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!( + "operator-listener-claim-test-{}.example", + community_id.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + for (id, event_marker, created_at) in [ + (stale_id, 1_u8, stale_at), + (fresh_id, 2_u8, event_created_at), + ] { + sqlx::query( + "INSERT INTO operator_listener_outbox \ + (id, listener_pubkey, target_pubkey, community_id, event_id, event_kind, \ + event_created_at, created_at) \ + VALUES ($1, $2, $3, $4, $5, 9, $6, $7)", + ) + .bind(id) + .bind(listener.public_key().as_bytes().as_slice()) + .bind(target.public_key().as_bytes().as_slice()) + .bind(community_id) + .bind(vec![event_marker; 32]) + .bind(event_created_at) + .bind(created_at) + .execute(&pool) + .await + .expect("insert test delivery"); + } + + let claimed = claim_deliveries(&pool, 1, Utc::now() + TimeDelta::seconds(30)) + .await + .expect("claim deliveries"); + assert_eq!(claimed.len(), 1); + assert_eq!(claimed[0].id, fresh_id); + + sqlx::query("DELETE FROM operator_listener_outbox WHERE community_id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test deliveries"); + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test community"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn expired_claim_is_recovered_and_old_claim_token_is_fenced() { + let pool = setup_pool().await; + let community_id = Uuid::new_v4(); + let listener = Keys::generate(); + let target = Keys::generate(); + let delivery_id = Uuid::new_v4(); + + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!( + "operator-listener-lease-test-{}.example", + community_id.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + sqlx::query( + "INSERT INTO operator_listener_outbox \ + (id, listener_pubkey, target_pubkey, community_id, event_id, event_kind, event_created_at) \ + VALUES ($1, $2, $3, $4, $5, 9, $6)", + ) + .bind(delivery_id) + .bind(listener.public_key().as_bytes().as_slice()) + .bind(target.public_key().as_bytes().as_slice()) + .bind(community_id) + .bind(vec![0x33_u8; 32]) + .bind(Utc::now()) + .execute(&pool) + .await + .expect("insert test delivery"); + + let first = claim_deliveries(&pool, 1, Utc::now() + TimeDelta::seconds(30)) + .await + .expect("claim delivery"); + assert_eq!(first.len(), 1); + assert_eq!(first[0].id, delivery_id); + assert!( + claim_deliveries(&pool, 1, Utc::now() + TimeDelta::seconds(30)) + .await + .expect("don't claim active lease") + .is_empty() + ); + + sqlx::query("UPDATE operator_listener_outbox SET lease_until = $2 WHERE id = $1") + .bind(delivery_id) + .bind(Utc::now() - TimeDelta::seconds(1)) + .execute(&pool) + .await + .expect("expire first lease"); + let recovered = claim_deliveries(&pool, 1, Utc::now() + TimeDelta::seconds(30)) + .await + .expect("recover expired claim"); + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].id, delivery_id); + assert_ne!(recovered[0].claim_id, first[0].claim_id); + assert_eq!(recovered[0].attempt, first[0].attempt + 1); + assert!(!complete_delivery(&pool, delivery_id, first[0].claim_id) + .await + .expect("old claim cannot complete delivery")); + assert!(!retry_delivery( + &pool, + delivery_id, + first[0].claim_id, + Utc::now() + TimeDelta::seconds(10), + ) + .await + .expect("old claim cannot retry delivery")); + assert!(!release_unroutable_delivery( + &pool, + delivery_id, + first[0].claim_id, + Utc::now() + TimeDelta::seconds(10), + ) + .await + .expect("old claim cannot release delivery")); + assert!(!fail_delivery(&pool, delivery_id, first[0].claim_id) + .await + .expect("old claim cannot delete delivery")); + assert!(complete_delivery(&pool, delivery_id, recovered[0].claim_id) + .await + .expect("current claim completes delivery")); + + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test community"); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn reapers_delete_expired_rows_and_preserve_recent_rows() { + let pool = setup_pool().await; + let listener = Keys::generate(); + let old_target = Keys::generate(); + let recent_target = Keys::generate(); + let listener_pubkey = listener.public_key().as_bytes().to_vec(); + let old_target_pubkey = old_target.public_key().as_bytes().to_vec(); + let recent_target_pubkey = recent_target.public_key().as_bytes().to_vec(); + + register_pubkeys( + &pool, + &listener_pubkey, + &[old_target_pubkey.clone(), recent_target_pubkey.clone()], + ) + .await + .expect("register test targets"); + let expired_at = Utc::now() - PUBKEY_RETENTION - TimeDelta::seconds(1); + sqlx::query( + "UPDATE operator_listener_pubkeys SET created_at = $3 \ + WHERE listener_pubkey = $1 AND target_pubkey = $2", + ) + .bind(&listener_pubkey) + .bind(&old_target_pubkey) + .bind(expired_at) + .execute(&pool) + .await + .expect("age test registration"); + + let community_id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community_id) + .bind(format!( + "operator-listener-reaper-test-{}.example", + community_id.simple() + )) + .execute(&pool) + .await + .expect("insert test community"); + let old_delivery = Uuid::new_v4(); + let recent_delivery = Uuid::new_v4(); + for (id, marker, created_at) in [ + ( + old_delivery, + 0x44_u8, + Utc::now() - OUTBOX_RETENTION - TimeDelta::seconds(1), + ), + (recent_delivery, 0x55_u8, Utc::now()), + ] { + sqlx::query( + "INSERT INTO operator_listener_outbox \ + (id, listener_pubkey, target_pubkey, community_id, event_id, event_kind, \ + event_created_at, created_at) \ + VALUES ($1, $2, $3, $4, $5, 9, $6, $7)", + ) + .bind(id) + .bind(&listener_pubkey) + .bind(&recent_target_pubkey) + .bind(community_id) + .bind(vec![marker; 32]) + .bind(Utc::now()) + .bind(created_at) + .execute(&pool) + .await + .expect("insert test delivery"); + } + + assert!( + delete_expired_pubkeys(&pool) + .await + .expect("reap expired registrations") + >= 1 + ); + assert!( + reap_deliveries(&pool) + .await + .expect("reap expired deliveries") + >= 1 + ); + + let registration_count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM operator_listener_pubkeys \ + WHERE listener_pubkey = $1 AND target_pubkey = $2", + ) + .bind(&listener_pubkey) + .bind(&recent_target_pubkey) + .fetch_one(&pool) + .await + .expect("count recent registration"); + let old_delivery_count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM operator_listener_outbox WHERE id = $1") + .bind(old_delivery) + .fetch_one(&pool) + .await + .expect("count expired delivery"); + let recent_delivery_count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM operator_listener_outbox WHERE id = $1") + .bind(recent_delivery) + .fetch_one(&pool) + .await + .expect("count recent delivery"); + assert_eq!(registration_count, 1); + assert_eq!(old_delivery_count, 0); + assert_eq!(recent_delivery_count, 1); + + sqlx::query("DELETE FROM operator_listener_outbox WHERE community_id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test deliveries"); + sqlx::query("DELETE FROM communities WHERE id = $1") + .bind(community_id) + .execute(&pool) + .await + .expect("delete test community"); + sqlx::query("DELETE FROM operator_listener_pubkeys WHERE listener_pubkey = $1") + .bind(&listener_pubkey) + .execute(&pool) + .await + .expect("delete test registrations"); + } +} diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index 2c49ca6a5c3..66a7420cfad 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/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, +} + +fn parse_listener_pubkeys(body: &[u8]) -> Result>, (StatusCode, Json)> { + 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. @@ -105,6 +143,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, + headers: &HeaderMap, + method: &str, + path: &str, + body: &[u8], +) -> Result)> { + 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_bridge_auth_with_options(headers, method, &url, Some(body), true, true)?; + 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>, + headers: HeaderMap, + body: axum::body::Bytes, +) -> Result, (StatusCode, Json)> { + 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>, + headers: HeaderMap, + body: axum::body::Bytes, +) -> Result, (StatusCode, Json)> { + 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], @@ -503,7 +621,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}, @@ -536,6 +657,26 @@ mod postgres_tests { Box::pin(async { Ok(true) }) } } + + struct SeenOnceReplayGuard(Mutex>); + + 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> + 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 { @@ -636,6 +777,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, + method: &str, + path: &str, + body: Option, + auth: String, + ) -> axum::response::Response { let mut request = Request::builder() .method(method) .uri(path) @@ -777,6 +928,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::>(); + 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> = 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() { diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 5d831b3f651..1d110055d8e 100644 --- a/crates/buzz-relay/src/config.rs +++ b/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, + /// Configured operator-listener identities and their HTTPS delivery URLs. + pub operator_listener_delivery_urls: HashMap, + /// 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, - /// 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. @@ -470,6 +475,58 @@ fn parse_push_gateway_delivery_url(raw: &str) -> Result { Ok(url) } +fn parse_operator_listener_delivery_urls( + raw: &str, +) -> Result, 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 { match std::env::var(name) { Err(std::env::VarError::NotPresent) => Ok(default), @@ -789,6 +846,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()?, }; @@ -1004,6 +1087,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::() + .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, ConfigError> { @@ -1240,6 +1350,7 @@ impl Config { relay_owner_pubkey, relay_operator_api_origin, relay_operator_pubkeys, + operator_listener_delivery_urls, allow_nip_oa_auth, klipy, media, @@ -1261,6 +1372,7 @@ impl Config { push_executor_key_id, push_gateway_delivery_url, push_gateway_timeout, + operator_listener_timeout, join_policy, admin, web_dir, @@ -1421,14 +1533,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, String) { + /// Capture the tracing output emitted during `Config::from_env()`. + fn capture_config_logs() -> (Result, String) { use std::sync::{Arc, Mutex}; #[derive(Clone)] @@ -1463,12 +1569,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, 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, 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) { @@ -2273,6 +2423,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(); diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index 18ea187fc7d..ccf4674256d 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -33,6 +33,9 @@ pub mod mesh_boot; pub mod metrics; /// NIP-11 relay information document. pub mod nip11; +mod nip98; +/// 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. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 532defbd8da..c2c44e8bb65 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -818,6 +818,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. diff --git a/crates/buzz-relay/src/nip98.rs b/crates/buzz-relay/src/nip98.rs new file mode 100644 index 00000000000..117131396dd --- /dev/null +++ b/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 { + 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)?) + )) +} diff --git a/crates/buzz-relay/src/operator_listener.rs b/crates/buzz-relay/src/operator_listener.rs new file mode 100644 index 00000000000..77fd1ab87c4 --- /dev/null +++ b/crates/buzz-relay/src/operator_listener.rs @@ -0,0 +1,601 @@ +//! 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) { + 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) { + let http = match reqwest::Client::builder() + .timeout(state.config.operator_listener_timeout) + .build() + { + 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, + } + } +} + +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, + ) -> Result; +} + +#[async_trait::async_trait] +trait DeliveryStore: Sync { + async fn release(&self, id: Uuid, claim_id: Uuid, next: DateTime) -> Result; + async fn complete(&self, id: Uuid, claim_id: Uuid) -> Result; + async fn retry(&self, id: Uuid, claim_id: Uuid, next: DateTime) -> Result; + async fn fail(&self, id: Uuid, claim_id: Uuid) -> Result; +} + +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) -> Result { + 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 { + 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) -> Result { + 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 { + 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, + ) -> Result { + 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( + 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( + routes: &HashMap, + 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) => retry_or_fail(store, &delivery, format!("HTTP {status}")).await, + Err(error) => retry_or_fail(store, &delivery, error.to_string()).await, + } +} + +async fn fail_permanently( + 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( + 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 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), + Complete(Uuid, Uuid), + Retry(Uuid, Uuid, DateTime), + Fail(Uuid, Uuid), + } + + #[derive(Default)] + struct MockStore { + calls: Mutex>, + } + + impl MockStore { + fn calls(&self) -> Vec { + 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, + ) -> Result { + 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 { + 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, + ) -> Result { + 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 { + self.calls + .lock() + .expect("store calls lock") + .push(StoreCall::Fail(id, claim_id)); + Ok(true) + } + } + + struct SentRequest { + url: url::Url, + authorization: String, + body: Vec, + } + + struct MockTransport { + response: Mutex>>, + request: Mutex>, + } + + impl MockTransport { + fn new(response: Result) -> Self { + Self { + response: Mutex::new(Some(response)), + request: Mutex::new(None), + } + } + + fn take_request(&self) -> Option { + self.request.lock().expect("transport request lock").take() + } + } + + #[async_trait::async_trait] + impl DeliveryTransport for MockTransport { + async fn post( + &self, + url: &url::Url, + authorization: &str, + body: Vec, + ) -> Result { + *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, + 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 transient_http_failure_retries_with_exponential_delay() { + let (routes, relay_keypair, delivery) = delivery_fixture(3); + let store = MockStore::default(); + let transport = MockTransport::new(Ok(StatusCode::SERVICE_UNAVAILABLE)); + 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!(transport.take_request().is_some()); + let calls = store.calls(); + assert_eq!(calls.len(), 1); + 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 retry, got {call:?}"), + } + } + + #[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)] + ); + } +} diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index 246997aac22..020b564dc66 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/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 { - 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 } @@ -690,6 +672,7 @@ fn class_rank(_: &str) -> u8 { mod tests { use super::*; use axum::{extract::State, routing::post, Json, Router}; + use nostr::{EventBuilder, Kind, Tag}; use serde_json::Value; use std::{future::IntoFuture, sync::Arc}; use tokio::sync::Mutex; diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 61aedf70be0..0035a693d57 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -88,6 +88,11 @@ pub fn build_router(state: Arc) -> 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), diff --git a/migrations/0049_operator_listener_mentions.sql b/migrations/0049_operator_listener_mentions.sql new file mode 100644 index 00000000000..10f81f50aa8 --- /dev/null +++ b/migrations/0049_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'); diff --git a/schema/schema.sql b/schema/schema.sql index 797a83f5d10..62b25b9db6b 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -1492,7 +1492,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[]) $$; @@ -1876,6 +1876,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;