Enha: sse update

broker/sse.go

@@ -1,6 +1,7 @@
 package broker
 
 import (
+	"context"
 	"fmt"
 	"log/slog"
 	"net/http"
@@ -20,8 +21,10 @@ type (
 	NotifierChan chan NotificationEvent
 	Broker       struct {
 		// Events are pushed to this channel by the main events-gathering routine
-		Notifier NotifierChan
+		Notifier  NotifierChan
-		log      *slog.Logger
+		log       *slog.Logger
+		addClient chan NotifierChan
+		clients   map[NotifierChan]struct{}
 	}
 )
 
@@ -33,6 +36,8 @@ func NewBroker() (broker *Broker) {
 			Level:     slog.LevelDebug,
 			AddSource: true,
 		})),
+		addClient: make(chan NotifierChan, 10),
+		clients:   map[NotifierChan]struct{}{},
 	}
 }
 
@@ -56,8 +61,10 @@ func (broker *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Access-Control-Allow-Origin", origin)
 	w.Header().Set("Access-Control-Allow-Credentials", "true")
 	ctx := r.Context()
+	msgChan := make(NotifierChan, 10)
+	broker.addClient <- msgChan
 	// browser can close sse on its own; ping every 2s to prevent
-	heartbeat := time.NewTicker(2 * time.Second)
+	heartbeat := time.NewTicker(8 * time.Second)
 	defer heartbeat.Stop()
 	for {
 		select {
@@ -67,13 +74,20 @@ func (broker *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			return
 		case event := <-broker.Notifier:
 			broker.log.Debug("got event", "event", event)
-			_, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event.EventName, event.Payload)
+			for i := 0; i < 10; i++ { // Repeat 3 times
-			if err != nil {
+				_, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event.EventName, event.Payload)
-				broker.log.Error("failed to write event", "error", err)
+				if err != nil {
-				// Client disconnected
+					broker.log.Error("write failed", "error", err)
-				return
+					return
+				}
+				w.(http.Flusher).Flush()
+				// Short delay between sends (non-blocking)
+				select {
+				case <-time.After(20 * time.Millisecond): // Adjust delay as needed
+				case <-ctx.Done():
+					return
+				}
 			}
-			w.(http.Flusher).Flush()
 		case <-heartbeat.C:
 			// Send SSE heartbeat comment
 			if _, err := fmt.Fprint(w, ":\n\n"); err != nil {
@@ -84,3 +98,15 @@ func (broker *Broker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 }
+
+func (broker *Broker) Listen(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case clientChan := <-broker.addClient:
+			// mutex
+			broker.clients[clientChan] = struct{}{}
+		}
+	}
+}

components/room.html

@@ -53,7 +53,7 @@
   </div>
   <hr/>
   <div class="grid grid-cols-1 md:grid-cols-5 md:gap-4">
-    <div hx-get="/actionhistory" hx-trigger="sse:backlog_{{.Room.ID}}" class="md:col-span-1">
+    <div hx-get="/actionhistory" class="md:col-span-1">
       {{template "actionhistory" .Room.ActionHistory}}
     </div>
     <div id="cardtable" class="md:col-span-3">