From 0ab0f79749e895dab7cb56d1f3926d3917e206c6 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sun, 8 Mar 2026 19:24:22 +0000
Subject: [PATCH] Fix SSE pipe: subscriptions never registered, malformed
 events, fanout skipping
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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 | 23 +++++++++--------------
 src/server.rs     | 19 ++++++++-----------
 2 files changed, 17 insertions(+), 25 deletions(-)

diff --git a/src/connection.rs b/src/connection.rs
index 5c48f88..d74b333 100644
--- a/src/connection.rs
+++ b/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()),
         }
     }
 
diff --git a/src/server.rs b/src/server.rs
index b6bddfd..9fd7b8b 100644
--- a/src/server.rs
+++ b/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);