sandrodach/MindSift
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(realtime): implement SSE fallback, dual-transport hub & proxy resilience (resolves FLAW-03)
MindSift CI & Docker Release / Build, Test in Container & Push (push) Successful in 59s
Details

Browse Source
This commit is contained in:
sandrodach committed 2026-10-26 16:55:11 +01:00
1 parent 80bb74b5b2
commit 6343d62bab
10 files changed
+1402 -240
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/ws/hub.go
+297 -13
View File
@@ -3,6 +3,7 @@ package ws
import (
"encoding/base64"
"encoding/json"
"fmt"
"log"
"net/http"
"strings"
@@ -30,6 +31,14 @@ type Client struct {
subscribedRooms map[string]bool
}
type SSEClient struct {
id string
userID string
roomID string
send chan []byte
subscribedRooms map[string]bool
}
type docLockItem struct {
docID string
userID string
@@ -45,20 +54,26 @@ type CRDTStore interface {
}
type Hub struct {
mu sync.RWMutex
rooms map[string]map[*Client]bool
userRooms map[string]map[*Client]bool
allClients map[*Client]bool
locks map[string]*docLockItem
crdtStore CRDTStore
mu sync.RWMutex
rooms map[string]map[*Client]bool
userRooms map[string]map[*Client]bool
allClients map[*Client]bool
sseClients map[*SSEClient]bool
sseUserRooms map[string]map[*SSEClient]bool
sseRooms map[string]map[*SSEClient]bool
locks map[string]*docLockItem
crdtStore CRDTStore
}
func NewHub() *Hub {
h := &Hub{
rooms: make(map[string]map[*Client]bool),
userRooms: make(map[string]map[*Client]bool),
allClients: make(map[*Client]bool),
locks: make(map[string]*docLockItem),
rooms: make(map[string]map[*Client]bool),
userRooms: make(map[string]map[*Client]bool),
allClients: make(map[*Client]bool),
sseClients: make(map[*SSEClient]bool),
sseUserRooms: make(map[string]map[*SSEClient]bool),
sseRooms: make(map[string]map[*SSEClient]bool),
locks: make(map[string]*docLockItem),
}
go h.cleanupLocksLoop()
return h
@@ -85,7 +100,7 @@ func (h *Hub) Register(c *Client) {
}
h.userRooms[c.userID][c] = true
}
onlineCount := len(h.allClients)
onlineCount := len(h.allClients) + len(h.sseClients)
h.mu.Unlock()
if c.roomID != "" {
@@ -126,11 +141,14 @@ func (h *Hub) Unregister(c *Client) {
if c.userID != "" && c.userID != "anonymous" {
if uRoom, ok := h.userRooms[c.userID]; ok {
delete(uRoom, c)
userRemaining = len(uRoom)
if userRemaining == 0 {
userRemaining += len(uRoom)
if len(uRoom) == 0 {
delete(h.userRooms, c.userID)
}
}
if sseRoom, ok := h.sseUserRooms[c.userID]; ok {
userRemaining += len(sseRoom)
}
}
close(c.send)
@@ -161,6 +179,142 @@ func (h *Hub) Unregister(c *Client) {
}
}
func (h *Hub) RegisterSSE(c *SSEClient) {
h.mu.Lock()
h.sseClients[c] = true
if c.roomID != "" {
if h.sseRooms[c.roomID] == nil {
h.sseRooms[c.roomID] = make(map[*SSEClient]bool)
}
h.sseRooms[c.roomID][c] = true
}
if c.userID != "" && c.userID != "anonymous" {
if h.sseUserRooms[c.userID] == nil {
h.sseUserRooms[c.userID] = make(map[*SSEClient]bool)
}
h.sseUserRooms[c.userID][c] = true
}
if c.subscribedRooms != nil {
for rID := range c.subscribedRooms {
if h.sseRooms[rID] == nil {
h.sseRooms[rID] = make(map[*SSEClient]bool)
}
h.sseRooms[rID][c] = true
}
}
onlineCount := len(h.allClients) + len(h.sseClients)
h.mu.Unlock()
if c.roomID != "" {
h.Broadcast(c.roomID, "", "USER_PRESENCE", map[string]interface{}{
"onlineCount": onlineCount,
})
}
}
func (h *Hub) UnregisterSSE(c *SSEClient) {
h.mu.Lock()
delete(h.sseClients, c)
if c.roomID != "" {
if room, ok := h.sseRooms[c.roomID]; ok {
delete(room, c)
if len(room) == 0 {
delete(h.sseRooms, c.roomID)
}
}
}
var docRoomsLeft []string
if c.subscribedRooms != nil {
for rID := range c.subscribedRooms {
if room, ok := h.sseRooms[rID]; ok {
delete(room, c)
if len(room) == 0 {
delete(h.sseRooms, rID)
}
}
if strings.HasPrefix(rID, "doc:") {
docRoomsLeft = append(docRoomsLeft, strings.TrimPrefix(rID, "doc:"))
}
}
}
userRemaining := 0
if c.userID != "" && c.userID != "anonymous" {
if uRoom, ok := h.sseUserRooms[c.userID]; ok {
delete(uRoom, c)
userRemaining += len(uRoom)
if len(uRoom) == 0 {
delete(h.sseUserRooms, c.userID)
}
}
if wsRoom, ok := h.userRooms[c.userID]; ok {
userRemaining += len(wsRoom)
}
}
close(c.send)
var releasedDocIDs []string
if userRemaining == 0 && c.userID != "" {
now := time.Now()
for docID, l := range h.locks {
if l.userID == c.userID || l.expiresAt.Before(now) {
releasedDocIDs = append(releasedDocIDs, docID)
delete(h.locks, docID)
}
}
}
h.mu.Unlock()
for _, docID := range releasedDocIDs {
h.BroadcastToAll("DOC_UNLOCKED", map[string]string{"docId": docID, "doc_id": docID})
}
for _, docID := range docRoomsLeft {
h.BroadcastRoom("doc:"+docID, "CRDT_AWARENESS_LEAVE", map[string]interface{}{
"docId": docID,
"userId": c.userID,
}, nil)
}
}
func (h *Hub) JoinSSERoomByID(sseID, roomID string) bool {
h.mu.Lock()
defer h.mu.Unlock()
for client := range h.sseClients {
if client.id == sseID {
if h.sseRooms[roomID] == nil {
h.sseRooms[roomID] = make(map[*SSEClient]bool)
}
h.sseRooms[roomID][client] = true
if client.subscribedRooms == nil {
client.subscribedRooms = make(map[string]bool)
}
client.subscribedRooms[roomID] = true
return true
}
}
return false
}
func (h *Hub) LeaveSSERoomByID(sseID, roomID string) bool {
h.mu.Lock()
defer h.mu.Unlock()
for client := range h.sseClients {
if client.id == sseID {
if room, ok := h.sseRooms[roomID]; ok {
delete(room, client)
if len(room) == 0 {
delete(h.sseRooms, roomID)
}
}
if client.subscribedRooms != nil {
delete(client.subscribedRooms, roomID)
}
return true
}
}
return false
}
// Document Locking Methods
func (h *Hub) AcquireDocLock(docID, userID, userName string, ttl time.Duration) (*domain.DocLockInfo, bool, *domain.DocLockInfo) {
@@ -300,6 +454,12 @@ func (h *Hub) BroadcastToAll(event string, payload interface{}) {
default:
}
}
for sseClient := range h.sseClients {
select {
case sseClient.send <- data:
default:
}
}
}
func (h *Hub) BroadcastToUser(userID, event string, payload interface{}) {
@@ -324,6 +484,14 @@ func (h *Hub) BroadcastToUser(userID, event string, payload interface{}) {
}
}
}
if sseRoom, ok := h.sseUserRooms[userID]; ok {
for sseClient := range sseRoom {
select {
case sseClient.send <- data:
default:
}
}
}
}
func (h *Hub) BroadcastTaskEvent(userID, event string, payload interface{}) {
@@ -363,6 +531,14 @@ func (h *Hub) Broadcast(spaceID, listID, event string, payload interface{}) {
}
}
}
if sseRoom, ok := h.sseRooms[targetRoom]; ok {
for sseClient := range sseRoom {
select {
case sseClient.send <- data:
default:
}
}
}
}
func (h *Hub) BroadcastListEvent(listID, event string, payload interface{}) {
@@ -421,6 +597,14 @@ func (h *Hub) BroadcastRoom(roomID, event string, payload interface{}, excludeCl
}
}
}
if sseRoom, ok := h.sseRooms[roomID]; ok {
for sseClient := range sseRoom {
select {
case sseClient.send <- data:
default:
}
}
}
}
func (c *Client) SendEvent(event string, payload interface{}) {
@@ -458,6 +642,106 @@ func (h *Hub) ServeWS(w http.ResponseWriter, r *http.Request, roomID, userID str
go client.readPump()
}
func (h *Hub) ServeSSE(w http.ResponseWriter, r *http.Request, roomID, userID string, initialRooms []string) {
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "Streaming unsupported!", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache, no-transform")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("X-Accel-Buffering", "no")
w.Header().Set("Access-Control-Allow-Origin", "*")
client := &SSEClient{
id: domain.NewID("sse"),
userID: userID,
roomID: roomID,
send: make(chan []byte, 128),
subscribedRooms: make(map[string]bool),
}
for _, room := range initialRooms {
if strings.TrimSpace(room) != "" {
client.subscribedRooms[strings.TrimSpace(room)] = true
}
}
h.RegisterSSE(client)
defer h.UnregisterSSE(client)
// Send initial connected comment and event
fmt.Fprintf(w, ": connected (transport: sse, id: %s)\n\n", client.id)
connectedMsg, _ := json.Marshal(domain.WSMessage{
Event: "SSE_CONNECTED",
Payload: map[string]interface{}{
"sseId": client.id,
"transport": "sse",
"userId": userID,
},
Timestamp: domain.NowISO(),
})
fmt.Fprintf(w, "data: %s\n\n", connectedMsg)
flusher.Flush()
// If initial rooms had doc rooms, send CRDT_INIT
if h.crdtStore != nil {
for rID := range client.subscribedRooms {
if strings.HasPrefix(rID, "doc:") {
docID := strings.TrimPrefix(rID, "doc:")
stateBlob, updates, err := h.crdtStore.GetDocumentCRDTState(docID)
if err == nil && (len(stateBlob) > 0 || len(updates) > 0) {
var b64Updates []string
for _, u := range updates {
b64Updates = append(b64Updates, base64.StdEncoding.EncodeToString(u))
}
b64State := ""
if len(stateBlob) > 0 {
b64State = base64.StdEncoding.EncodeToString(stateBlob)
}
initMsg := domain.WSMessage{
Event: "CRDT_INIT",
Payload: map[string]interface{}{
"docId": docID,
"state": b64State,
"updates": b64Updates,
},
Timestamp: domain.NowISO(),
}
if initBytes, err := json.Marshal(initMsg); err == nil {
fmt.Fprintf(w, "data: %s\n\n", initBytes)
flusher.Flush()
}
}
}
}
}
ticker := time.NewTicker(15 * time.Second)
defer ticker.Stop()
notify := r.Context().Done()
for {
select {
case <-notify:
return
case msg, ok := <-client.send:
if !ok {
return
}
fmt.Fprintf(w, "data: %s\n\n", msg)
flusher.Flush()
case <-ticker.C:
// Periodic ping comment to keep HTTP connection alive and prevent proxy timeouts
fmt.Fprintf(w, ": ping\n\n")
flusher.Flush()
}
}
}
func (c *Client) readPump() {
defer func() {
c.hub.Unregister(c)