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());