feat(workspace): implement Yjs CRDT real-time collaborative editing (resolves FLAW-01)

internal/ws/hub.go

@@ -1,9 +1,11 @@
 package ws
 
 import (
+	"encoding/base64"
 	"encoding/json"
 	"log"
 	"net/http"
+	"strings"
 	"sync"
 	"time"
 
@@ -12,19 +14,20 @@ import (
 )
 
 var upgrader = websocket.Upgrader{
-	ReadBufferSize:  2048,
-	WriteBufferSize: 2048,
+	ReadBufferSize:  4096,
+	WriteBufferSize: 4096,
 	CheckOrigin: func(r *http.Request) bool {
 		return true // Self-hosted reverse proxy compatible
 	},
 }
 
 type Client struct {
-	hub     *Hub
-	conn    *websocket.Conn
-	roomID  string
-	userID  string
-	send    chan []byte
+	hub             *Hub
+	conn            *websocket.Conn
+	roomID          string
+	userID          string
+	send            chan []byte
+	subscribedRooms map[string]bool
 }
 
 type docLockItem struct {
@@ -35,12 +38,19 @@ type docLockItem struct {
 	expiresAt time.Time
 }
 
+type CRDTStore interface {
+	GetDocumentCRDTState(docID string) ([]byte, [][]byte, error)
+	AppendDocumentCRDTUpdate(docID string, update []byte, userID string) error
+	SaveDocumentCRDTState(docID string, stateBlob []byte, userID string) error
+}
+
 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
 }
 
 func NewHub() *Hub {
@@ -54,6 +64,12 @@ func NewHub() *Hub {
 	return h
 }
 
+func (h *Hub) SetCRDTStore(store CRDTStore) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.crdtStore = store
+}
+
 func (h *Hub) Register(c *Client) {
 	h.mu.Lock()
 	h.allClients[c] = true
@@ -90,6 +106,22 @@ func (h *Hub) Unregister(c *Client) {
 			}
 		}
 	}
+
+	var docRoomsLeft []string
+	if c.subscribedRooms != nil {
+		for rID := range c.subscribedRooms {
+			if room, ok := h.rooms[rID]; ok {
+				delete(room, c)
+				if len(room) == 0 {
+					delete(h.rooms, 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.userRooms[c.userID]; ok {
@@ -119,6 +151,14 @@ func (h *Hub) Unregister(c *Client) {
 	for _, docID := range releasedDocIDs {
 		h.BroadcastToAll("DOC_UNLOCKED", map[string]string{"docId": docID, "doc_id": docID})
 	}
+
+	// Broadcast awareness leave for doc rooms
+	for _, docID := range docRoomsLeft {
+		h.BroadcastRoom("doc:"+docID, "CRDT_AWARENESS_LEAVE", map[string]interface{}{
+			"docId":  docID,
+			"userId": c.userID,
+		}, nil)
+	}
 }
 
 // Document Locking Methods
@@ -329,6 +369,75 @@ func (h *Hub) BroadcastListEvent(listID, event string, payload interface{}) {
 	h.Broadcast("", listID, event, payload)
 }
 
+func (h *Hub) JoinRoom(c *Client, roomID string) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if h.rooms[roomID] == nil {
+		h.rooms[roomID] = make(map[*Client]bool)
+	}
+	h.rooms[roomID][c] = true
+	if c.subscribedRooms == nil {
+		c.subscribedRooms = make(map[string]bool)
+	}
+	c.subscribedRooms[roomID] = true
+}
+
+func (h *Hub) LeaveRoom(c *Client, roomID string) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if room, ok := h.rooms[roomID]; ok {
+		delete(room, c)
+		if len(room) == 0 {
+			delete(h.rooms, roomID)
+		}
+	}
+	if c.subscribedRooms != nil {
+		delete(c.subscribedRooms, roomID)
+	}
+}
+
+func (h *Hub) BroadcastRoom(roomID, event string, payload interface{}, excludeClient *Client) {
+	msg := domain.WSMessage{
+		Event:     event,
+		Payload:   payload,
+		Timestamp: domain.NowISO(),
+	}
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return
+	}
+
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+
+	if room, ok := h.rooms[roomID]; ok {
+		for client := range room {
+			if excludeClient != nil && client == excludeClient {
+				continue
+			}
+			select {
+			case client.send <- data:
+			default:
+			}
+		}
+	}
+}
+
+func (c *Client) SendEvent(event string, payload interface{}) {
+	msg := domain.WSMessage{
+		Event:     event,
+		Payload:   payload,
+		Timestamp: domain.NowISO(),
+	}
+	data, err := json.Marshal(msg)
+	if err == nil {
+		select {
+		case c.send <- data:
+		default:
+		}
+	}
+}
+
 func (h *Hub) ServeWS(w http.ResponseWriter, r *http.Request, roomID, userID string) {
 	conn, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
@@ -355,7 +464,7 @@ func (c *Client) readPump() {
 		_ = c.conn.Close()
 	}()
 
-	c.conn.SetReadLimit(16384)
+	c.conn.SetReadLimit(4 * 1024 * 1024) // 4MB for large document CRDT updates
 	_ = c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
 	c.conn.SetPongHandler(func(string) error {
 		_ = c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
@@ -368,7 +477,7 @@ func (c *Client) readPump() {
 			break
 		}
 
-		// Handle client-initiated lock and unlock messages
+		// Handle client-initiated lock, CRDT, and presence messages
 		var incoming struct {
 			Event   string          `json:"event"`
 			DocID   string          `json:"docId"`
@@ -396,6 +505,61 @@ func (c *Client) readPump() {
 					if c.hub.ReleaseDocLock(docID, c.userID, false) {
 						c.hub.BroadcastDocEvent(docID, "DOC_UNLOCKED", map[string]string{"docId": docID, "doc_id": docID})
 					}
+				case "CRDT_JOIN":
+					c.hub.JoinRoom(c, "doc:"+docID)
+					if c.hub.crdtStore != nil {
+						stateBlob, updates, err := c.hub.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)
+							}
+							c.SendEvent("CRDT_INIT", map[string]interface{}{
+								"docId":   docID,
+								"state":   b64State,
+								"updates": b64Updates,
+							})
+						}
+					}
+				case "CRDT_UPDATE":
+					var p struct {
+						Update string `json:"update"`
+					}
+					_ = json.Unmarshal(incoming.Payload, &p)
+					if p.Update != "" {
+						c.hub.BroadcastRoom("doc:"+docID, "CRDT_UPDATE", map[string]interface{}{
+							"docId":      docID,
+							"update":     p.Update,
+							"fromUserId": c.userID,
+						}, c)
+						if c.hub.crdtStore != nil {
+							if raw, err := base64.StdEncoding.DecodeString(p.Update); err == nil && len(raw) > 0 {
+								_ = c.hub.crdtStore.AppendDocumentCRDTUpdate(docID, raw, c.userID)
+							}
+						}
+					}
+				case "CRDT_COMPACT":
+					var p struct {
+						State string `json:"state"`
+					}
+					_ = json.Unmarshal(incoming.Payload, &p)
+					if p.State != "" && c.hub.crdtStore != nil {
+						if raw, err := base64.StdEncoding.DecodeString(p.State); err == nil && len(raw) > 0 {
+							_ = c.hub.crdtStore.SaveDocumentCRDTState(docID, raw, c.userID)
+						}
+					}
+				case "CRDT_AWARENESS":
+					c.hub.BroadcastRoom("doc:"+docID, "CRDT_AWARENESS", incoming.Payload, c)
+				case "CRDT_LEAVE":
+					c.hub.LeaveRoom(c, "doc:"+docID)
+					c.hub.BroadcastRoom("doc:"+docID, "CRDT_AWARENESS_LEAVE", map[string]interface{}{
+						"docId":  docID,
+						"userId": c.userID,
+					}, c)
 				}
 			}
 		}