fix(relay): reject presence updates when Redis storage fails

crates/buzz-relay/src/handlers/event.rs

@@ -829,16 +829,37 @@ async fn handle_ephemeral_event(
             raw
         };
 
+        // Presence mutation is the inclusion contract for the live fan-out
+        // below: a client that observes the fanned-out event treats a later
+        // snapshot as reflecting it (see `synthesize_presence` in
+        // `api/bridge.rs`, which reads Redis). If the mutation failed we
+        // published nothing — so we must also fan out nothing and reject the
+        // ACK, or a snapshot later "confirms" stale storage over a live event
+        // the sender believes was delivered.
         if status == "offline" {
-            let _ = state
+            if let Err(e) = state
                 .pubsub
                 .clear_presence(&conn.tenant, &auth_pubkey)
-                .await;
-        } else {
-            let _ = state
-                .pubsub
-                .set_presence(&conn.tenant, &auth_pubkey, &status)
-                .await;
+                .await
+            {
+                warn!(
+                    conn_id = %conn_id,
+                    event_id = %event_id,
+                    "Presence clear failed, refusing publish and fan-out: {e}"
+                );
+                return Err("error: presence storage unavailable".to_string());
+            }
+        } else if let Err(e) = state
+            .pubsub
+            .set_presence(&conn.tenant, &auth_pubkey, &status)
+            .await
+        {
+            warn!(
+                conn_id = %conn_id,
+                event_id = %event_id,
+                "Presence set failed, refusing publish and fan-out: {e}"
+            );
+            return Err("error: presence storage unavailable".to_string());
         }
 
         // Presence is a channel-less ephemeral event. After updating Redis
@@ -1435,6 +1456,147 @@ mod tests {
         );
     }
 
+    // PostgreSQL/Redis tests are discovered by the isolated postgres-ci lane.
+    mod presence_storage_postgres_tests {
+        use super::*;
+        use buzz_core::{CommunityId, TenantContext};
+        use nostr::Filter;
+
+        async fn exercise(storage_available: bool, statuses: &[&str]) {
+            let redis_url = std::env::var("REDIS_URL").expect("test Redis URL required");
+            // Pick an unused local endpoint for a real connection failure.
+            let dead_socket = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+            let dead_port = dead_socket.local_addr().unwrap().port();
+            drop(dead_socket);
+            let dead_url = format!("redis://127.0.0.1:{dead_port}");
+            let state = fanout_access::test_state_with_redis_url(if storage_available {
+                &redis_url
+            } else {
+                &dead_url
+            })
+            .await;
+            let pool = sqlx::PgPool::connect(&state.config.database_url)
+                .await
+                .unwrap();
+            let community_uuid = Uuid::new_v4();
+            let host = format!("presence-storage-{}.example", community_uuid.simple());
+            sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
+                .bind(community_uuid)
+                .bind(&host)
+                .execute(&pool)
+                .await
+                .expect("seed active community");
+            let tenant = TenantContext::resolved(CommunityId::from_uuid(community_uuid), host);
+            let keys = Keys::generate();
+            let (send_tx, mut send_rx) = mpsc::channel(10);
+            let (ctrl_tx, _ctrl_rx) = mpsc::channel(10);
+            let conn = Arc::new(crate::connection::ConnectionState {
+                conn_id: Uuid::new_v4(),
+                tenant: tenant.clone(),
+                remote_addr: "127.0.0.1:1234".parse().unwrap(),
+                auth_state: RwLock::new(crate::connection::AuthState::Authenticated(
+                    buzz_auth::AuthContext {
+                        pubkey: keys.public_key(),
+                        scopes: vec![],
+                        channel_ids: None,
+                        auth_method: buzz_auth::AuthMethod::Nip42,
+                        agent_owner_pubkey: None,
+                    },
+                )),
+                subscriptions: Arc::new(Mutex::new(HashMap::new())),
+                send_tx,
+                ctrl_tx,
+                cancel: CancellationToken::new(),
+                backpressure_count: Arc::new(AtomicU8::new(0)),
+                grace_limit: 3,
+            });
+            let watcher = Uuid::new_v4();
+            let (tx, mut rx) = mpsc::channel(10);
+            let (ctrl, _ctrl_rx) = mpsc::channel(10);
+            state.conn_manager.register(
+                watcher,
+                tx,
+                ctrl,
+                None,
+                CancellationToken::new(),
+                tenant.community(),
+                Arc::new(AtomicU8::new(0)),
+                Arc::new(Mutex::new(HashMap::new())),
+                3,
+            );
+            state.sub_registry.register_scoped(
+                tenant.community(),
+                watcher,
+                "presence".into(),
+                vec![Filter::new().kind(Kind::Custom(KIND_PRESENCE_UPDATE as u16))],
+                None,
+            );
+            // Online followed by offline also proves DEL removes an existing value.
+            for &status in statuses {
+                let event = EventBuilder::new(Kind::Custom(KIND_PRESENCE_UPDATE as u16), status)
+                    .sign_with_keys(&keys)
+                    .unwrap();
+                super::super::handle_event(event.clone(), conn.clone(), state.clone()).await;
+                let axum::extract::ws::Message::Text(text) = send_rx.try_recv().expect("ACK")
+                else {
+                    panic!("expected text ACK");
+                };
+                let ack: serde_json::Value = serde_json::from_str(&text).unwrap();
+                assert_eq!(ack[0], "OK");
+                assert_eq!(ack[1], event.id.to_hex());
+                assert_eq!(ack[2], storage_available);
+                if storage_available {
+                    assert_eq!(ack[3], "");
+                    let stored = state
+                        .pubsub
+                        .get_presence(&tenant, &keys.public_key())
+                        .await
+                        .unwrap();
+                    assert_eq!(stored.as_deref(), (status != "offline").then_some(status));
+                    let axum::extract::ws::Message::Text(text) =
+                        rx.try_recv().expect("live fanout")
+                    else {
+                        panic!("expected text event");
+                    };
+                    let frame: serde_json::Value = serde_json::from_str(&text).unwrap();
+                    assert_eq!(frame[0], "EVENT");
+                    assert_eq!(frame[2]["id"], event.id.to_hex());
+                } else {
+                    assert_eq!(ack[3], "error: presence storage unavailable");
+                    assert!(state
+                        .local_event_ids
+                        .get(&(tenant.community(), event.id.to_bytes()))
+                        .is_none());
+                }
+                assert!(rx.try_recv().is_err(), "no extra or rejected-event fanout");
+                assert!(send_rx.try_recv().is_err(), "exactly one ACK");
+            }
+            sqlx::query("DELETE FROM communities WHERE id = $1")
+                .bind(community_uuid)
+                .execute(&pool)
+                .await
+                .unwrap();
+        }
+
+        #[tokio::test]
+        #[ignore = "requires PostgreSQL and Redis"]
+        async fn rejects_online_when_presence_storage_fails() {
+            exercise(false, &["online"]).await;
+        }
+
+        #[tokio::test]
+        #[ignore = "requires PostgreSQL and Redis"]
+        async fn rejects_offline_when_presence_storage_fails() {
+            exercise(false, &["offline"]).await;
+        }
+
+        #[tokio::test]
+        #[ignore = "requires PostgreSQL and Redis"]
+        async fn accepts_stores_and_fans_out_online_and_offline() {
+            exercise(true, &["online", "offline"]).await;
+        }
+    }
+
     mod pubsub_fanout {
         use std::collections::HashMap;
         use std::sync::atomic::AtomicU8;