From 7495aadf8f709dfb68c9c026ea1cd2311d35b38e Mon Sep 17 00:00:00 2001 From: Bradley Axen Date: Thu, 8 Oct 2026 21:43:07 -0700 Subject: [PATCH] feat(relay): accept typing indicators over HTTP POST /events App-hosted agents sign events but hold no WebSocket, so they could not send typing indicators: POST /events rejected kind 20002 as an unknown kind. The bridge now hands 20002 to the same channel broadcast the WebSocket path uses, without storing it. Before broadcasting it checks the signature, that the signer is the authenticated pubkey, bans and timeouts, the community write fence, an h tag naming the channel, and channel membership. Other ephemeral kinds are still rejected over HTTP. Signed-off-by: Bradley Axen Co-authored-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 --- ARCHITECTURE.md | 2 + crates/buzz-relay/src/api/bridge.rs | 20 +- .../src/api/typing_postgres_tests.rs | 204 ++++++++++++++++++ crates/buzz-relay/src/handlers/event.rs | 111 +++++++--- crates/buzz-relay/src/handlers/ingest.rs | 6 +- 5 files changed, 309 insertions(+), 34 deletions(-) create mode 100644 crates/buzz-relay/src/api/typing_postgres_tests.rs diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 82d7cac7bc..531b1ceee7 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -314,6 +314,8 @@ Presence events skip membership checks and use local-only fan-out. Multi-node pr 5. LOCAL FAN-OUT — sub_registry.fan_out → conn_manager.send_to ``` +Typing indicators (kind 20002) are also accepted through HTTP `POST /events`, for clients that sign events but hold no WebSocket, such as app-hosted agents. The HTTP route requires an `h` tag, checks bans/timeouts and the community write fence, then runs steps 1–5 above. + Ephemeral events are never stored in Postgres and never appear in REQ historical queries. ### Handler Semaphore diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 5cb515a5b2..db12fe9429 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1158,7 +1158,21 @@ async fn submit_event_authed( auth_method: crate::handlers::ingest::HttpAuthMethod::Nip98, }; - match crate::handlers::ingest::ingest_event(state, tenant, event, auth).await { + // Typing indicators are ephemeral: broadcast to the channel, never stored. + let result = if kind_u32 == buzz_core::kind::KIND_TYPING_INDICATOR { + let event_id = event.id.to_hex(); + crate::handlers::event::publish_http_typing(state, tenant, event, pubkey) + .await + .map(|()| crate::handlers::ingest::IngestResult { + event_id, + accepted: true, + message: String::new(), + }) + } else { + crate::handlers::ingest::ingest_event(state, tenant, event, auth).await + }; + + match result { Ok(result) => { let response = Json(serde_json::json!({ "event_id": result.event_id, @@ -2917,6 +2931,10 @@ fn ban_json(b: &buzz_db::moderation::BanRecord) -> Value { #[path = "artifact_postgres_tests.rs"] mod artifact_postgres_tests; +#[cfg(test)] +#[path = "typing_postgres_tests.rs"] +mod typing_postgres_tests; + #[cfg(test)] pub(crate) mod postgres_tests { use super::*; diff --git a/crates/buzz-relay/src/api/typing_postgres_tests.rs b/crates/buzz-relay/src/api/typing_postgres_tests.rs new file mode 100644 index 0000000000..3883a92e19 --- /dev/null +++ b/crates/buzz-relay/src/api/typing_postgres_tests.rs @@ -0,0 +1,204 @@ +// Typing indicators (kind:20002) submitted through HTTP POST /events. +use super::postgres_tests::bridge_handler_test_state; +use super::*; +use nostr::{Event, EventBuilder, Keys, Kind, Tag}; +use serde_json::{json, Value}; +use uuid::Uuid; + +struct Fixture { + state: Arc, + pool: sqlx::PgPool, + host: String, + community: buzz_core::CommunityId, + channel: Uuid, + member: Keys, +} + +impl Fixture { + async fn new() -> Self { + let state = bridge_handler_test_state() + .await + .expect("Postgres and Redis"); + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let host = format!("typing-{}.local", Uuid::new_v4()); + state.db.ensure_configured_community(&host).await.unwrap(); + let community = crate::tenant::bind_community(&state.db, &host) + .await + .unwrap() + .community(); + let channel = Uuid::new_v4(); + let member = Keys::generate(); + sqlx::query("INSERT INTO channels(community_id,id,name,visibility,channel_type,created_by) VALUES($1,$2,$2::text,'private'::channel_visibility,'stream',$3)") + .bind(community.as_uuid()).bind(channel).bind(member.public_key().as_bytes().as_slice()).execute(&pool).await.unwrap(); + sqlx::query("INSERT INTO channel_members(community_id,channel_id,pubkey,role) VALUES($1,$2,$3,'owner')") + .bind(community.as_uuid()).bind(channel).bind(member.public_key().as_bytes().as_slice()).execute(&pool).await.unwrap(); + Self { + state, + pool, + host, + community, + channel, + member, + } + } + + fn typing(&self, key: &Keys, channel: Option) -> Event { + let tags = channel.map(|ch| Tag::parse(["h".to_string(), ch.to_string()]).unwrap()); + EventBuilder::new(Kind::Custom(20002), "") + .tags(tags) + .sign_with_keys(key) + .unwrap() + } + + async fn post(&self, auth: &Keys, event: &Event) -> (axum::http::StatusCode, Value) { + use tower::ServiceExt; + let response = crate::router::build_router(self.state.clone()) + .oneshot( + axum::http::Request::builder() + .method("POST") + .uri("/events") + .header("host", &self.host) + .header("content-type", "application/json") + .header("x-pubkey", auth.public_key().to_hex()) + .body(axum::body::Body::from(json!(event).to_string())) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .unwrap(); + (status, serde_json::from_slice(&bytes).unwrap()) + } + + async fn stored_typing_rows(&self) -> i64 { + sqlx::query_scalar( + "SELECT COUNT(*) FROM events WHERE community_id = $1 AND channel_id = $2 AND kind = 20002", + ) + .bind(self.community.as_uuid()) + .bind(self.channel) + .fetch_one(&self.pool) + .await + .unwrap() + } +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn member_typing_is_broadcast_and_not_stored() { + use futures::StreamExt; + let f = Fixture::new().await; + let topic = buzz_pubsub::EventTopicKey { + community_id: f.community, + topic: buzz_pubsub::EventTopic::Channel(f.channel), + } + .redis_channel(); + let client = redis::Client::open(f.state.config.redis_url.as_str()).unwrap(); + let mut pubsub = client.get_async_pubsub().await.unwrap(); + // Acknowledged by Redis before returning, so the publish below is not missed. + pubsub.subscribe(&topic).await.unwrap(); + + let event = f.typing(&f.member, Some(f.channel)); + let (status, body) = f.post(&f.member, &event).await; + assert_eq!(status, axum::http::StatusCode::OK, "{body}"); + assert_eq!(body["accepted"], true, "{body}"); + assert_eq!(body["event_id"], event.id.to_hex(), "{body}"); + + let mut messages = pubsub.into_on_message(); + let msg = tokio::time::timeout(std::time::Duration::from_secs(5), messages.next()) + .await + .expect("typing indicator published to the channel topic") + .unwrap(); + let published: Event = serde_json::from_str(&msg.get_payload::().unwrap()).unwrap(); + assert_eq!(published.id, event.id); + assert_eq!(f.stored_typing_rows().await, 0); +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn non_member_typing_is_rejected() { + let f = Fixture::new().await; + let outsider = Keys::generate(); + let (status, body) = f + .post(&outsider, &f.typing(&outsider, Some(f.channel))) + .await; + assert_eq!(status, axum::http::StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(body["error"], "restricted: not a channel member", "{body}"); +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn timed_out_member_typing_is_rejected() { + let f = Fixture::new().await; + f.state + .db + .timeout_community_member( + f.community, + f.member.public_key().as_bytes(), + Keys::generate().public_key().as_bytes(), + chrono::Utc::now() + chrono::Duration::hours(1), + None, + ) + .await + .unwrap(); + let (status, body) = f + .post(&f.member, &f.typing(&f.member, Some(f.channel))) + .await; + assert_eq!(status, axum::http::StatusCode::FORBIDDEN, "{body}"); +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn typing_signed_by_someone_else_is_rejected() { + let f = Fixture::new().await; + let (status, body) = f + .post(&Keys::generate(), &f.typing(&f.member, Some(f.channel))) + .await; + assert_eq!(status, axum::http::StatusCode::FORBIDDEN, "{body}"); + assert!( + body["error"] + .as_str() + .unwrap_or_default() + .contains("does not match authenticated identity"), + "{body}" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn tampered_typing_is_rejected() { + let f = Fixture::new().await; + let mut raw = serde_json::to_value(f.typing(&f.member, Some(f.channel))).unwrap(); + raw["content"] = json!("tampered"); + let event: Event = serde_json::from_value(raw).unwrap(); + let (status, body) = f.post(&f.member, &event).await; + assert_eq!(status, axum::http::StatusCode::BAD_REQUEST, "{body}"); + assert_eq!(f.stored_typing_rows().await, 0); +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn other_ephemeral_kinds_are_still_rejected() { + let f = Fixture::new().await; + let event = EventBuilder::new(Kind::Custom(20001), "") + .tags([Tag::parse(["h".to_string(), f.channel.to_string()]).unwrap()]) + .sign_with_keys(&f.member) + .unwrap(); + let (status, body) = f.post(&f.member, &event).await; + assert_eq!(status, axum::http::StatusCode::BAD_REQUEST, "{body}"); +} + +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn typing_without_channel_is_rejected() { + let f = Fixture::new().await; + let (status, body) = f.post(&f.member, &f.typing(&f.member, None)).await; + assert_eq!(status, axum::http::StatusCode::BAD_REQUEST, "{body}"); + assert!( + body["error"].as_str().unwrap_or_default().contains("h tag"), + "{body}" + ); +} diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 5b7b257c19..8dd05e5245 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -8,7 +8,7 @@ use tracing::{debug, error, info, warn}; use buzz_core::event::StoredEvent; use buzz_core::kind::{ event_kind_u32, is_ephemeral, is_unshared_gated_event, AUTHOR_ONLY_KINDS, - KIND_AGENT_OBSERVER_FRAME, KIND_GIFT_WRAP, KIND_PRESENCE_UPDATE, + KIND_AGENT_OBSERVER_FRAME, KIND_GIFT_WRAP, KIND_PRESENCE_UPDATE, KIND_TYPING_INDICATOR, }; use buzz_core::observer::{ content_looks_like_nip44, OBSERVER_AGENT_TAG, OBSERVER_FRAME_CONTROL, OBSERVER_FRAME_TAG, @@ -889,7 +889,9 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc, + tenant: &TenantContext, + event: Event, + ch_id: uuid::Uuid, + pubkey_bytes: &[u8], +) -> Result<(), IngestError> { + // Membership refusals are client-input rejections, and the shared + // gate's message text is surfaced verbatim. + super::ingest::check_channel_membership(tenant, state, ch_id, pubkey_bytes, None) + .await + .map_err(IngestError::Rejected)?; + + // Mark as local before Redis publish to prevent double-delivery when + // the event comes back through the Redis subscriber loop. + state.mark_local_event(tenant.community(), &event.id); + + if let Err(e) = state + .pubsub + .publish_event(tenant, EventTopic::Channel(ch_id), &event) + .await + { + state + .local_event_ids + .invalidate(&(tenant.community(), event.id.to_bytes())); + warn!(event_id = %event.id.to_hex(), "Ephemeral publish failed: {e}"); + } + + // Direct fan-out to local WS subscribers, through the guarded send path + // so a stale subscription on a removed/non-member connection cannot + // receive this private-channel ephemeral event. + // Pass the channel_id so fan_out() uses the channel-kind index. + let stored_event = StoredEvent::new(event, Some(ch_id)); + fan_out_event_to_local_subscribers(state, tenant.community(), &stored_event).await; + Ok(()) +} + +/// Publishes a typing indicator (kind:20002) submitted through HTTP +/// `POST /events`, for clients that sign but hold no WebSocket, such as +/// app-hosted agents. Applies the gates the WebSocket path applies to +/// ephemeral events, then the same channel publish. The caller has already +/// authenticated `auth_pubkey` and enforced relay membership. +pub(crate) async fn publish_http_typing( + state: &Arc, + tenant: &TenantContext, + event: Event, + auth_pubkey: PublicKey, +) -> Result<(), IngestError> { + if event.pubkey != auth_pubkey { + return Err(IngestError::AuthFailed( + "invalid: event pubkey does not match authenticated identity".into(), + )); + } + let ch_id = super::ingest::extract_channel_id(&event).ok_or_else(|| { + IngestError::Rejected("invalid: typing indicator needs a channel UUID h tag".into()) + })?; + let event_clone = event.clone(); + match tokio::task::spawn_blocking(move || verify_event(&event_clone)).await { + Ok(Ok(())) => {} + Ok(Err(e)) => return Err(IngestError::Rejected(format!("invalid: {e}"))), + Err(_) => return Err(IngestError::Internal("error: internal error".into())), + } + super::ingest::enforce_write_restriction(state, tenant, KIND_TYPING_INDICATOR, &auth_pubkey) + .await?; + super::ingest::map_serving_fence_state( + buzz_deletion::store(&state.db) + .is_serving_active(tenant.community()) + .await, + )?; + publish_channel_ephemeral(state, tenant, event, ch_id, auth_pubkey.as_bytes()).await +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum AgentObserverDirection { Telemetry, diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index b46911b4cb..25490ace3e 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -459,7 +459,9 @@ pub enum IngestError { /// `restricted:` wire text the ephemeral path uses. A lookup outage is a /// server fault and fails closed as `error:`/500 — a Postgres blip can /// neither admit a write past the fence nor read as a client mistake. -fn map_serving_fence_state(active: Result) -> Result<(), IngestError> { +pub(crate) fn map_serving_fence_state( + active: Result, +) -> Result<(), IngestError> { match active { Ok(true) => Ok(()), Ok(false) => Err(IngestError::Rejected( @@ -3559,7 +3561,7 @@ mod postgres_tests { #[ignore = "requires Postgres"] async fn check_channel_write_denies_when_channel_lookup_fails() { let state = crate::state::tests::test_state_with_database_url( - "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", + "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", // sadscan:disable np.postgres.1 -- local test fixture ) .await; let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::nil());