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

Three bugs preventing any SSE events from reaching clients:

1. SubscriptionRegistry::clone() deep-cloned the DashMaps instead of sharing
   them. The sse_handler spawned a task with a disconnected clone, so
   subscriptions.update() wrote into a throwaway copy — the fanout always saw
   an empty registry and found zero connections to push to. Fixed by wrapping
   both DashMaps in Arc so clone() is a cheap pointer copy sharing live state.

2. The SSE event data payload used .join("\ndata: ") to pre-embed the `data:`
   prefix into the string, then handed it to axum's Event::data() which does
   its own \n-splitting to add `data:` prefixes. This produced double-prefixed
   lines (`data: data: elements ...`) that DataStar couldn't parse. Fixed by
   using .join("\n") and letting axum format the data lines correctly.

3. The fanout skipped connections with None signals (new connections that had
   not yet received a client action). This broke clock-style server-push events
   on fresh connections. Fixed by falling back to "{}" instead of continue-ing.

https://claude.ai/code/session_0168cLRd1wr6LK9FjBsjA37W

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: Arc<DashMap<CacheKey, HashSet<Uuid>>>,
-    key_to_conns: DashMap<CacheKey, HashSet<Uuid>>,
+    conn_to_keys: Arc<DashMap<Uuid, HashSet<CacheKey>>>,
-    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(),
-        }
-    }
 }
 
 impl SubscriptionRegistry {
     pub fn new() -> Self {
         Self {
-            key_to_conns: DashMap::new(),
+            key_to_conns: Arc::new(DashMap::new()),
-            conn_to_keys: 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 {
+                let signals_json = state.connections.get_signals(&conn_id)
-                    continue;
+                    .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.
+                            // Build a datastar-patch-elements SSE event.
-                            // DataStar expects the event name set once, then one
+                            // Each HTML line gets an `elements ` prefix; axum's
-                            // `data: elements <line>` entry per line of HTML.
+                            // Event::data() splits on \n and writes a `data:`
-                            // axum's Event::data() sets a single data field, so
+                            // line for each, which is correct SSE multiline format.
-                            // we pre-format the multi-line data payload manually
-                            // and send it as one data value.
                             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);