Merge pull request #1 from Markk116/claude/fix-sse-pipe-events-y7Nxd

Fix SSE pipe: subscriptions never registered, malformed events, fanout skipping

src/connection.rs

@@ -2,6 +2,7 @@ use crate::query::CacheKey;
 use axum::response::sse::Event;
 use dashmap::DashMap;
 use std::collections::HashSet;
+use std::sync::Arc;
 use tokio::sync::mpsc;
 use uuid::Uuid;
 
@@ -106,26 +107,20 @@ impl ConnectionManager {
 /// This is the hot path of the reactive pipeline: the fanout task calls
 /// `get_connections_for_key` on every broadcast invalidation event to find
 /// which connections need rerendering.
+///
+/// Clone is cheap (Arc pointer copy) and all clones share the same underlying
+/// maps, so updates from any clone are visible to the fanout task.
+#[derive(Clone)]
 pub struct SubscriptionRegistry {
-    // DashMap is Arc-backed internally, so clone is cheap.
-    key_to_conns: DashMap<CacheKey, HashSet<Uuid>>,
-    conn_to_keys: DashMap<Uuid, HashSet<CacheKey>>,
-}
-
-impl Clone for SubscriptionRegistry {
-    fn clone(&self) -> Self {
-        Self {
-            key_to_conns: self.key_to_conns.clone(),
-            conn_to_keys: self.conn_to_keys.clone(),
-        }
-    }
+    key_to_conns: Arc<DashMap<CacheKey, HashSet<Uuid>>>,
+    conn_to_keys: Arc<DashMap<Uuid, HashSet<CacheKey>>>,
 }
 
 impl SubscriptionRegistry {
     pub fn new() -> Self {
         Self {
-            key_to_conns: DashMap::new(),
-            conn_to_keys: DashMap::new(),
+            key_to_conns: Arc::new(DashMap::new()),
+            conn_to_keys: Arc::new(DashMap::new()),
         }
     }
 

src/server.rs

@@ -109,9 +109,8 @@ pub fn spawn_fanout(state: SharedState) {
             let conn_ids = state.subscriptions.get_connections_for_key(&key);
 
             for conn_id in conn_ids {
-                let Some(signals_json) = state.connections.get_signals(&conn_id) else {
-                    continue;
-                };
+                let signals_json = state.connections.get_signals(&conn_id)
+                    .unwrap_or_else(|| "{}".to_string());
                 let Some(sse_tx) = state.connections.get_sender(&conn_id) else {
                     continue;
                 };
@@ -145,17 +144,15 @@ pub fn spawn_fanout(state: SharedState) {
                         Ok((html, keys)) => {
                             state_ref.subscriptions.update(conn_id, keys);
 
-                            // Build a single datastar-patch-elements SSE event.
-                            // DataStar expects the event name set once, then one
-                            // `data: elements <line>` entry per line of HTML.
-                            // axum's Event::data() sets a single data field, so
-                            // we pre-format the multi-line data payload manually
-                            // and send it as one data value.
+                            // Build a datastar-patch-elements SSE event.
+                            // Each HTML line gets an `elements ` prefix; axum's
+                            // Event::data() splits on \n and writes a `data:`
+                            // line for each, which is correct SSE multiline format.
                             let data_payload: String = html
                                 .lines()
                                 .map(|line| format!("elements {}", line))
                                 .collect::<Vec<_>>()
-                                .join("\ndata: ");
+                                .join("\n");
 
                             let event = Event::default()
                                 .event("datastar-patch-elements")
@@ -241,7 +238,7 @@ pub async fn sse_handler(
                     .lines()
                     .map(|line| format!("elements {}", line))
                     .collect::<Vec<_>>()
-                    .join("\ndata: ");
+                    .join("\n");
                 let event = Event::default()
                     .event("datastar-patch-elements")
                     .data(data_payload);