package ws import ( "encoding/json" "log" "net/http" "sync" "time" "github.com/gorilla/websocket" "github.com/sandrodach/mindsift/internal/domain" ) var upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, CheckOrigin: func(r *http.Request) bool { return true // Self-hosted reverse proxy compatible }, } type Client struct { hub *Hub conn *websocket.Conn spaceID string userID string send chan []byte } type Hub struct { mu sync.RWMutex rooms map[string]map[*Client]bool } func NewHub() *Hub { return &Hub{ rooms: make(map[string]map[*Client]bool), } } func (h *Hub) Register(c *Client) { h.mu.Lock() if h.rooms[c.spaceID] == nil { h.rooms[c.spaceID] = make(map[*Client]bool) } h.rooms[c.spaceID][c] = true onlineCount := len(h.rooms[c.spaceID]) h.mu.Unlock() h.Broadcast(c.spaceID, "", "USER_PRESENCE", map[string]interface{}{ "onlineCount": onlineCount, }) } func (h *Hub) Unregister(c *Client) { h.mu.Lock() var onlineCount int if room, ok := h.rooms[c.spaceID]; ok { if _, exists := room[c]; exists { delete(room, c) close(c.send) } onlineCount = len(room) if onlineCount == 0 { delete(h.rooms, c.spaceID) } } h.mu.Unlock() if onlineCount > 0 { h.Broadcast(c.spaceID, "", "USER_PRESENCE", map[string]interface{}{ "onlineCount": onlineCount, }) } } func (h *Hub) Broadcast(spaceID, listID, event string, payload interface{}) { msg := domain.WSMessage{ Event: event, SpaceID: spaceID, ListID: listID, Payload: payload, Timestamp: domain.NowISO(), } data, err := json.Marshal(msg) if err != nil { return } h.mu.RLock() defer h.mu.RUnlock() room, ok := h.rooms[spaceID] if !ok { return } for client := range room { select { case client.send <- data: default: // Slow client buffer full } } } func (h *Hub) ServeWS(w http.ResponseWriter, r *http.Request, spaceID, userID string) { conn, err := upgrader.Upgrade(w, r, nil) if err != nil { log.Printf("[ws] upgrade error: %v", err) return } client := &Client{ hub: h, conn: conn, spaceID: spaceID, userID: userID, send: make(chan []byte, 64), } h.Register(client) go client.writePump() go client.readPump() } func (c *Client) readPump() { defer func() { c.hub.Unregister(c) _ = c.conn.Close() }() c.conn.SetReadLimit(8192) _ = c.conn.SetReadDeadline(time.Now().Add(60 * time.Second)) c.conn.SetPongHandler(func(string) error { _ = c.conn.SetReadDeadline(time.Now().Add(60 * time.Second)) return nil }) for { _, _, err := c.conn.ReadMessage() if err != nil { break } } } func (c *Client) writePump() { ticker := time.NewTicker(30 * time.Second) defer func() { ticker.Stop() _ = c.conn.Close() }() for { select { case message, ok := <-c.send: _ = c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) if !ok { _ = c.conn.WriteMessage(websocket.CloseMessage, []byte{}) return } if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil { return } case <-ticker.C: _ = c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { return } } } }