feat(workspace): tabula-rasa outliner architecture, SQLite blocks/block_tags schema & echo-protected live room sync

internal/ws/docroom.go

@@ -1,9 +1,7 @@
 package ws
 
 import (
-	"encoding/base64"
 	"encoding/json"
-	"fmt"
 	"log"
 	"net/http"
 	"sync"
@@ -15,386 +13,488 @@ import (
 
 // DocWSClient represents an active WebSocket connection to a specific document room.
 type DocWSClient struct {
-	hub    *Hub
-	room   *DocRoom
-	conn   *websocket.Conn
-	userID string
-	send   chan []byte
+	hub      *Hub
+	room     *DocRoom
+	conn     *websocket.Conn
+	userID   string
+	clientID string
+	send     chan []byte
 }
 
-// DocSSEClient represents an active SSE connection to a specific document room.
-type DocSSEClient struct {
-	hub    *Hub
-	room   *DocRoom
-	id     string
-	userID string
-	send   chan []byte
+// DocSyncEvent is the lightweight JSON payload exchanged over /api/v1/ws/docs/{id}.
+type DocSyncEvent struct {
+	Type         string                 `json:"type"`
+	DocID        string                 `json:"docId,omitempty"`
+	ClientID     string                 `json:"clientId,omitempty"`
+	UserID       string                 `json:"userId,omitempty"`
+	Title        string                 `json:"title,omitempty"`
+	Block        *domain.DocumentBlock  `json:"block,omitempty"`
+	BlockID      string                 `json:"blockId,omitempty"`
+	AfterBlockID string                 `json:"afterBlockId,omitempty"`
+	NewIndex     *int                   `json:"newIndex,omitempty"`
+	Content      *string                `json:"content,omitempty"`
+	Level        *int                   `json:"level,omitempty"`
+	Checked      *bool                  `json:"checked,omitempty"`
+	Collapsed    *bool                  `json:"collapsed,omitempty"`
+	BlockType    *string                `json:"blockType,omitempty"`
+	Blocks       []domain.DocumentBlock `json:"blocks,omitempty"`
 }
 
-// DocRoom coordinates in-memory real-time updates and debounced persistence for one document.
+// DocRoom coordinates in-memory real-time block state and 1.5s debounced SQLite persistence for one document.
 type DocRoom struct {
-	mu                     sync.Mutex
-	docID                  string
-	hub                    *Hub
-	stateBlob              []byte
-	pendingUpdates         [][]byte
-	persistedUpdatesCount  int
-	wsClients              map[*DocWSClient]bool
-	sseClients             map[*DocSSEClient]bool
-	debounceTimer          *time.Timer
-	dirty                  bool
-	lastUserID             string
+	mu            sync.Mutex
+	docID         string
+	hub           *Hub
+	title         string
+	blocks        []domain.DocumentBlock
+	loaded        bool
+	wsClients     map[*DocWSClient]bool
+	debounceTimer *time.Timer
+	dirty         bool
 }
 
+// GetOrCreateDocRoom returns the active in-memory room for docID or initializes one from SQLite.
 func (h *Hub) GetOrCreateDocRoom(docID string) *DocRoom {
 	h.mu.Lock()
-	defer h.mu.Unlock()
-
 	if h.docRooms == nil {
 		h.docRooms = make(map[string]*DocRoom)
 	}
-
-	if room, exists := h.docRooms[docID]; exists {
-		return room
+	room, exists := h.docRooms[docID]
+	if !exists {
+		room = &DocRoom{
+			docID:     docID,
+			hub:       h,
+			wsClients: make(map[*DocWSClient]bool),
+		}
+		h.docRooms[docID] = room
 	}
+	store := h.docStore
+	h.mu.Unlock()
 
-	room := &DocRoom{
-		docID:      docID,
-		hub:        h,
-		wsClients:  make(map[*DocWSClient]bool),
-		sseClients: make(map[*DocSSEClient]bool),
-	}
-
-	// Pre-load current state from CRDT store if available
-	if h.crdtStore != nil {
-		stateBlob, updates, err := h.crdtStore.GetDocumentCRDTState(docID)
+	room.mu.Lock()
+	if !room.loaded && store != nil {
+		title, blocks, err := store.LoadDocumentState(docID)
 		if err == nil {
-			room.stateBlob = stateBlob
-			room.pendingUpdates = updates
-			room.persistedUpdatesCount = len(updates)
+			room.title = title
+			room.blocks = normalizeRoomBlocks(docID, blocks)
+			room.loaded = true
 		}
 	}
+	room.mu.Unlock()
 
-	h.docRooms[docID] = room
 	return room
 }
 
-// ApplyUpdate processes an incremental Yjs binary update in-memory, broadcasts it immediately
-// to all other connected clients (WS & SSE), and schedules an asynchronous debounced flush to SQLite.
-func (r *DocRoom) ApplyUpdate(update []byte, fromUserID string, senderWS *DocWSClient, senderSSE *DocSSEClient) {
-	if len(update) == 0 {
+// GetActiveDocState returns the live in-memory title and blocks if an active DocRoom exists in RAM.
+func (h *Hub) GetActiveDocState(docID string) (string, []domain.DocumentBlock, bool) {
+	h.mu.RLock()
+	room, exists := h.docRooms[docID]
+	h.mu.RUnlock()
+	if !exists || room == nil {
+		return "", nil, false
+	}
+
+	room.mu.Lock()
+	defer room.mu.Unlock()
+	if !room.loaded {
+		return "", nil, false
+	}
+	blocksCopy := make([]domain.DocumentBlock, len(room.blocks))
+	copy(blocksCopy, room.blocks)
+	return room.title, blocksCopy, true
+}
+
+// SyncActiveDocRoom updates an in-memory room if one is currently loaded (e.g. after REST PUT/PATCH).
+func (h *Hub) SyncActiveDocRoom(docID string, title string, blocks []domain.DocumentBlock) {
+	h.mu.RLock()
+	room, exists := h.docRooms[docID]
+	h.mu.RUnlock()
+	if !exists || room == nil {
 		return
 	}
 
-	r.mu.Lock()
-	r.pendingUpdates = append(r.pendingUpdates, update)
-	r.dirty = true
-	if fromUserID != "" {
-		r.lastUserID = fromUserID
+	room.mu.Lock()
+	if title != "" {
+		room.title = title
 	}
-
-	b64Update := base64.StdEncoding.EncodeToString(update)
-	msg := domain.WSMessage{
-		Event: "CRDT_UPDATE",
-		Payload: map[string]interface{}{
-			"docId":      r.docID,
-			"update":     b64Update,
-			"fromUserId": fromUserID,
-		},
-		Timestamp: domain.NowISO(),
+	if blocks != nil {
+		room.blocks = normalizeRoomBlocks(docID, blocks)
 	}
-	msgBytes, _ := json.Marshal(msg)
-
-	// Broadcast to all WS clients except sender
-	for client := range r.wsClients {
-		if senderWS != nil && client == senderWS {
-			continue
-		}
-		select {
-		case client.send <- msgBytes:
-		default:
-		}
-	}
-
-	// Broadcast to all SSE clients except sender
-	for client := range r.sseClients {
-		if senderSSE != nil && client == senderSSE {
-			continue
-		}
-		select {
-		case client.send <- msgBytes:
-		default:
-		}
-	}
-
-	// Schedule 2.5s debounced persistence
-	r.scheduleDebouncedFlushLocked()
-	r.mu.Unlock()
-
-	// Also inform legacy room subscribers if any
-	r.hub.BroadcastRoom("doc:"+r.docID, "CRDT_UPDATE", map[string]interface{}{
-		"docId":      r.docID,
-		"update":     b64Update,
-		"fromUserId": fromUserID,
-	}, nil)
+	room.loaded = true
+	room.mu.Unlock()
 }
 
-// CompactState replaces the incremental pending updates with a full compacted state blob.
-func (r *DocRoom) CompactState(stateBlob []byte, fromUserID string) {
-	if len(stateBlob) == 0 {
-		return
+// CloseDocRoom removes a document room from memory when a document is deleted.
+func (h *Hub) CloseDocRoom(docID string) {
+	h.mu.Lock()
+	room, exists := h.docRooms[docID]
+	if exists {
+		delete(h.docRooms, docID)
 	}
+	h.mu.Unlock()
 
-	r.mu.Lock()
-	r.stateBlob = stateBlob
-	r.pendingUpdates = nil
-	r.persistedUpdatesCount = 0
-	r.dirty = true
-	if fromUserID != "" {
-		r.lastUserID = fromUserID
+	if exists && room != nil {
+		room.mu.Lock()
+		if room.debounceTimer != nil {
+			room.debounceTimer.Stop()
+		}
+		room.dirty = false
+		for c := range room.wsClients {
+			_ = c.conn.Close()
+		}
+		room.wsClients = make(map[*DocWSClient]bool)
+		room.mu.Unlock()
 	}
-	r.scheduleDebouncedFlushLocked()
-	r.mu.Unlock()
 }
 
-// BroadcastAwareness forwards user awareness state to all other participants in the room.
-func (r *DocRoom) BroadcastAwareness(payload json.RawMessage, fromUserID string, senderWS *DocWSClient, senderSSE *DocSSEClient) {
-	msg := domain.WSMessage{
-		Event:     "CRDT_AWARENESS",
-		Payload:   payload,
-		Timestamp: domain.NowISO(),
+func normalizeRoomBlocks(docID string, blocks []domain.DocumentBlock) []domain.DocumentBlock {
+	if len(blocks) == 0 {
+		return []domain.DocumentBlock{
+			{
+				ID:         docID + "-b0",
+				DocumentID: docID,
+				Type:       "text",
+				Content:    "",
+				Level:      0,
+				Position:   1000,
+			},
+		}
 	}
-	msgBytes, _ := json.Marshal(msg)
+	out := make([]domain.DocumentBlock, len(blocks))
+	for i, b := range blocks {
+		if b.ID == "" {
+			b.ID = domain.NewID("blk")
+		}
+		b.DocumentID = docID
+		if b.Type == "" {
+			b.Type = "text"
+		}
+		if b.Level < 0 {
+			b.Level = 0
+		}
+		b.Position = float64((i + 1) * 1000)
+		out[i] = b
+	}
+	return out
+}
 
+// GetStateSnapshot returns a copy of the room's current live in-memory title and blocks.
+func (r *DocRoom) GetStateSnapshot() (string, []domain.DocumentBlock) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
+	blocksCopy := make([]domain.DocumentBlock, len(r.blocks))
+	copy(blocksCopy, r.blocks)
+	return r.title, blocksCopy
+}
 
-	for client := range r.wsClients {
-		if senderWS != nil && client == senderWS {
-			continue
+// HandleEvent applies a granular block or title mutation to the in-memory room,
+// broadcasts it to all OTHER clients in the room (excluding senderClientID for echo protection),
+// and resets the 1.5-second inactivity debounce timer for SQLite persistence.
+func (r *DocRoom) HandleEvent(evt DocSyncEvent, sender *DocWSClient) {
+	r.mu.Lock()
+
+	evt.DocID = r.docID
+	if sender != nil {
+		if evt.ClientID == "" {
+			evt.ClientID = sender.clientID
 		}
-		select {
-		case client.send <- msgBytes:
-		default:
+		evt.UserID = sender.userID
+	}
+
+	mutated := false
+
+	switch evt.Type {
+	case "title_updated":
+		if evt.Title != "" || r.title != evt.Title {
+			r.title = evt.Title
+			mutated = true
+		}
+
+	case "block_created":
+		if evt.Block != nil {
+			newBlock := *evt.Block
+			if newBlock.ID == "" {
+				newBlock.ID = domain.NewID("blk")
+			}
+			newBlock.DocumentID = r.docID
+			if newBlock.Type == "" {
+				newBlock.Type = "text"
+			}
+			if newBlock.Level < 0 {
+				newBlock.Level = 0
+			}
+
+			// Avoid duplicate ID if already created
+			existsIdx := -1
+			for i, b := range r.blocks {
+				if b.ID == newBlock.ID {
+					existsIdx = i
+					break
+				}
+			}
+
+			if existsIdx >= 0 {
+				r.blocks[existsIdx] = newBlock
+			} else {
+				insertIdx := len(r.blocks)
+				if evt.AfterBlockID != "" {
+					for i, b := range r.blocks {
+						if b.ID == evt.AfterBlockID {
+							insertIdx = i + 1
+							break
+						}
+					}
+				} else if evt.NewIndex != nil && *evt.NewIndex >= 0 && *evt.NewIndex <= len(r.blocks) {
+					insertIdx = *evt.NewIndex
+				}
+
+				updated := make([]domain.DocumentBlock, 0, len(r.blocks)+1)
+				updated = append(updated, r.blocks[:insertIdx]...)
+				updated = append(updated, newBlock)
+				updated = append(updated, r.blocks[insertIdx:]...)
+				r.blocks = normalizeRoomBlocks(r.docID, updated)
+			}
+			evt.Block = &newBlock
+			mutated = true
+		}
+
+	case "block_updated":
+		targetID := evt.BlockID
+		if targetID == "" && evt.Block != nil {
+			targetID = evt.Block.ID
+		}
+		if targetID != "" {
+			for i := range r.blocks {
+				if r.blocks[i].ID == targetID {
+					if evt.Block != nil {
+						r.blocks[i].Content = evt.Block.Content
+						r.blocks[i].Level = evt.Block.Level
+						if evt.Block.Type != "" {
+							r.blocks[i].Type = evt.Block.Type
+						}
+						r.blocks[i].Checked = evt.Block.Checked
+						r.blocks[i].Collapsed = evt.Block.Collapsed
+						r.blocks[i].Lang = evt.Block.Lang
+					} else {
+						if evt.Content != nil {
+							r.blocks[i].Content = *evt.Content
+						}
+						if evt.Level != nil {
+							lvl := *evt.Level
+							if lvl < 0 {
+								lvl = 0
+							}
+							r.blocks[i].Level = lvl
+						}
+						if evt.Checked != nil {
+							r.blocks[i].Checked = *evt.Checked
+						}
+						if evt.Collapsed != nil {
+							r.blocks[i].Collapsed = *evt.Collapsed
+						}
+						if evt.BlockType != nil && *evt.BlockType != "" {
+							r.blocks[i].Type = *evt.BlockType
+						}
+					}
+					bCopy := r.blocks[i]
+					evt.Block = &bCopy
+					evt.BlockID = targetID
+					mutated = true
+					break
+				}
+			}
+		}
+
+	case "block_deleted":
+		if evt.BlockID != "" {
+			filtered := make([]domain.DocumentBlock, 0, len(r.blocks))
+			for _, b := range r.blocks {
+				if b.ID != evt.BlockID {
+					filtered = append(filtered, b)
+				}
+			}
+			r.blocks = normalizeRoomBlocks(r.docID, filtered)
+			mutated = true
+		}
+
+	case "block_moved":
+		if evt.BlockID != "" && evt.NewIndex != nil {
+			fromIdx := -1
+			for i, b := range r.blocks {
+				if b.ID == evt.BlockID {
+					fromIdx = i
+					break
+				}
+			}
+			if fromIdx >= 0 {
+				moving := r.blocks[fromIdx]
+				if evt.Level != nil {
+					lvl := *evt.Level
+					if lvl < 0 {
+						lvl = 0
+					}
+					moving.Level = lvl
+				}
+				without := make([]domain.DocumentBlock, 0, len(r.blocks)-1)
+				without = append(without, r.blocks[:fromIdx]...)
+				without = append(without, r.blocks[fromIdx+1:]...)
+
+				toIdx := *evt.NewIndex
+				if toIdx < 0 {
+					toIdx = 0
+				}
+				if toIdx > len(without) {
+					toIdx = len(without)
+				}
+
+				reordered := make([]domain.DocumentBlock, 0, len(r.blocks))
+				reordered = append(reordered, without[:toIdx]...)
+				reordered = append(reordered, moving)
+				reordered = append(reordered, without[toIdx:]...)
+				r.blocks = normalizeRoomBlocks(r.docID, reordered)
+				mutated = true
+			}
+		}
+
+	case "blocks_replaced":
+		if evt.Blocks != nil {
+			r.blocks = normalizeRoomBlocks(r.docID, evt.Blocks)
+			evt.Blocks = r.blocks
+			mutated = true
 		}
 	}
 
-	for client := range r.sseClients {
-		if senderSSE != nil && client == senderSSE {
-			continue
-		}
-		select {
-		case client.send <- msgBytes:
-		default:
+	if !mutated {
+		r.mu.Unlock()
+		return
+	}
+
+	r.dirty = true
+	r.loaded = true
+
+	msgBytes, err := json.Marshal(evt)
+	if err == nil {
+		for client := range r.wsClients {
+			// Strict Echo Protection: never send back to the originating connection or clientId
+			if sender != nil && (client == sender || (sender.clientID != "" && client.clientID == sender.clientID)) {
+				continue
+			}
+			if evt.ClientID != "" && client.clientID == evt.ClientID {
+				continue
+			}
+			select {
+			case client.send <- msgBytes:
+			default:
+			}
 		}
 	}
+
+	r.scheduleDebouncedFlushLocked()
+	r.mu.Unlock()
 }
 
 func (r *DocRoom) scheduleDebouncedFlushLocked() {
 	if r.debounceTimer != nil {
 		r.debounceTimer.Stop()
 	}
-	r.debounceTimer = time.AfterFunc(2500*time.Millisecond, func() {
+	r.debounceTimer = time.AfterFunc(1500*time.Millisecond, func() {
 		r.Flush()
 	})
 }
 
-// Flush persists in-memory pending updates to SQLite asynchronously in a background goroutine.
+// Flush persists the current in-memory document state (blocks, #tags in block_tags, and markdown_cache) to SQLite.
 func (r *DocRoom) Flush() {
 	r.mu.Lock()
 	if !r.dirty {
 		r.mu.Unlock()
 		return
 	}
+	if r.debounceTimer != nil {
+		r.debounceTimer.Stop()
+		r.debounceTimer = nil
+	}
 
-	// Prepare data to flush
 	docID := r.docID
-	userID := r.lastUserID
-	stateBlob := r.stateBlob
-	var updatesToFlush [][]byte
-
-	if len(stateBlob) > 0 && len(r.pendingUpdates) == 0 {
-		// Compacted state snapshot
-		r.dirty = false
-		r.mu.Unlock()
-
-		go func() {
-			if r.hub.crdtStore != nil {
-				_ = r.hub.crdtStore.SaveDocumentCRDTState(docID, stateBlob, userID)
-			}
-		}()
-		return
-	}
-
-	if r.persistedUpdatesCount < len(r.pendingUpdates) {
-		updatesToFlush = make([][]byte, len(r.pendingUpdates)-r.persistedUpdatesCount)
-		copy(updatesToFlush, r.pendingUpdates[r.persistedUpdatesCount:])
-		r.persistedUpdatesCount = len(r.pendingUpdates)
-	}
+	title := r.title
+	blocksCopy := make([]domain.DocumentBlock, len(r.blocks))
+	copy(blocksCopy, r.blocks)
 	r.dirty = false
 	r.mu.Unlock()
 
-	if len(updatesToFlush) > 0 {
-		go func() {
-			if r.hub.crdtStore != nil {
-				for _, u := range updatesToFlush {
-					_ = r.hub.crdtStore.AppendDocumentCRDTUpdate(docID, u, userID)
-				}
-			}
-		}()
-	}
-}
+	r.hub.mu.RLock()
+	store := r.hub.docStore
+	r.hub.mu.RUnlock()
 
-// GetInitialSnapshot returns the base64-encoded state blob and array of incremental updates.
-func (r *DocRoom) GetInitialSnapshot() (string, []string) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	b64State := ""
-	if len(r.stateBlob) > 0 {
-		b64State = base64.StdEncoding.EncodeToString(r.stateBlob)
-	}
-
-	var b64Updates []string
-	for _, u := range r.pendingUpdates {
-		if len(u) > 0 {
-			b64Updates = append(b64Updates, base64.StdEncoding.EncodeToString(u))
+	if store != nil {
+		if err := store.FlushDocumentState(docID, title, blocksCopy); err != nil {
+			log.Printf("[doc-room] flush error for doc %s: %v", docID, err)
 		}
 	}
-	return b64State, b64Updates
 }
 
-// ServeDocWS handles WebSocket connection for GET /api/v1/ws/docs/{id}
-func (r *DocRoom) ServeDocWS(w http.ResponseWriter, req *http.Request, userID string) {
+// ServeDocWS handles WebSocket connections for GET /api/v1/ws/docs/{id}
+func (r *DocRoom) ServeDocWS(w http.ResponseWriter, req *http.Request, userID string, clientID string) {
 	conn, err := upgrader.Upgrade(w, req, nil)
 	if err != nil {
 		log.Printf("[doc-ws] upgrade error for doc %s: %v", r.docID, err)
 		return
 	}
 
+	if clientID == "" {
+		clientID = req.URL.Query().Get("clientId")
+	}
+	if clientID == "" {
+		clientID = domain.NewID("cli")
+	}
+
 	client := &DocWSClient{
-		hub:    r.hub,
-		room:   r,
-		conn:   conn,
-		userID: userID,
-		send:   make(chan []byte, 256),
+		hub:      r.hub,
+		room:     r,
+		conn:     conn,
+		userID:   userID,
+		clientID: clientID,
+		send:     make(chan []byte, 256),
 	}
 
 	r.mu.Lock()
 	r.wsClients[client] = true
+	title := r.title
+	blocksCopy := make([]domain.DocumentBlock, len(r.blocks))
+	copy(blocksCopy, r.blocks)
 	r.mu.Unlock()
 
-	// Send initial snapshot
-	state, updates := r.GetInitialSnapshot()
-	initMsg := domain.WSMessage{
-		Event: "CRDT_INIT",
-		Payload: map[string]interface{}{
-			"docId":   r.docID,
-			"state":   state,
-			"updates": updates,
-		},
-		Timestamp: domain.NowISO(),
+	// Immediately send live in-memory state to the newly connected client
+	initEvt := DocSyncEvent{
+		Type:     "doc_state",
+		DocID:    r.docID,
+		ClientID: clientID,
+		Title:    title,
+		Blocks:   blocksCopy,
+	}
+	if initBytes, err := json.Marshal(initEvt); err == nil {
+		client.send <- initBytes
 	}
-	initBytes, _ := json.Marshal(initMsg)
-	client.send <- initBytes
 
 	go client.writePump()
 	go client.readPump()
 }
 
-// ServeDocSSE handles SSE stream fallback for GET /api/v1/sync/docs/{id}/stream
-func (r *DocRoom) ServeDocSSE(w http.ResponseWriter, req *http.Request, userID 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 := &DocSSEClient{
-		hub:    r.hub,
-		room:   r,
-		id:     domain.NewID("sse-doc"),
-		userID: userID,
-		send:   make(chan []byte, 256),
-	}
-
-	r.mu.Lock()
-	r.sseClients[client] = true
-	r.mu.Unlock()
-
-	defer func() {
-		r.mu.Lock()
-		delete(r.sseClients, client)
-		clientCount := len(r.wsClients) + len(r.sseClients)
-		r.mu.Unlock()
-
-		if clientCount == 0 {
-			r.Flush()
-		}
-	}()
-
-	// Send initial connected comment and CRDT_INIT
-	fmt.Fprintf(w, ": connected (transport: sse, docId: %s)\n\n", r.docID)
-	state, updates := r.GetInitialSnapshot()
-	initMsg, _ := json.Marshal(domain.WSMessage{
-		Event: "CRDT_INIT",
-		Payload: map[string]interface{}{
-			"docId":   r.docID,
-			"state":   state,
-			"updates": updates,
-		},
-		Timestamp: domain.NowISO(),
-	})
-	fmt.Fprintf(w, "data: %s\n\n", initMsg)
-	flusher.Flush()
-
-	ticker := time.NewTicker(15 * time.Second)
-	defer ticker.Stop()
-
-	notify := req.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:
-			// Ping to keep proxy connection alive
-			fmt.Fprintf(w, ": ping\n\n")
-			flusher.Flush()
-		}
-	}
-}
-
 func (c *DocWSClient) readPump() {
 	defer func() {
 		c.room.mu.Lock()
 		delete(c.room.wsClients, c)
-		clientCount := len(c.room.wsClients) + len(c.room.sseClients)
+		remaining := len(c.room.wsClients)
 		c.room.mu.Unlock()
 
 		_ = c.conn.Close()
 
-		// Broadcast awareness leave
-		c.room.BroadcastAwareness([]byte(fmt.Sprintf(`{"userId":"%s","activeBlockId":null}`, c.userID)), c.userID, nil, nil)
-
-		if clientCount == 0 {
+		// Flush immediately when the last client leaves the document room
+		if remaining == 0 {
 			c.room.Flush()
 		}
 	}()
 
-	c.conn.SetReadLimit(8 * 1024 * 1024) // 8MB limit for binary/text updates
+	c.conn.SetReadLimit(4 * 1024 * 1024)
 	_ = c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
 	c.conn.SetPongHandler(func(string) error {
 		_ = c.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
@@ -402,51 +502,20 @@ func (c *DocWSClient) readPump() {
 	})
 
 	for {
-		msgType, message, err := c.conn.ReadMessage()
+		_, message, err := c.conn.ReadMessage()
 		if err != nil {
 			break
 		}
 
-		if msgType == websocket.BinaryMessage {
-			// Direct binary Yjs update
-			c.room.ApplyUpdate(message, c.userID, c, nil)
-			continue
-		}
-
-		// Text JSON message
-		var incoming struct {
-			Event   string          `json:"event"`
-			Payload json.RawMessage `json:"payload"`
-		}
-		if json.Unmarshal(message, &incoming) == nil {
-			switch incoming.Event {
-			case "CRDT_UPDATE":
-				var p struct {
-					Update string `json:"update"`
-				}
-				if json.Unmarshal(incoming.Payload, &p) == nil && p.Update != "" {
-					if raw, err := base64.StdEncoding.DecodeString(p.Update); err == nil && len(raw) > 0 {
-						c.room.ApplyUpdate(raw, c.userID, c, nil)
-					}
-				}
-			case "CRDT_COMPACT":
-				var p struct {
-					State string `json:"state"`
-				}
-				if json.Unmarshal(incoming.Payload, &p) == nil && p.State != "" {
-					if raw, err := base64.StdEncoding.DecodeString(p.State); err == nil && len(raw) > 0 {
-						c.room.CompactState(raw, c.userID)
-					}
-				}
-			case "CRDT_AWARENESS":
-				c.room.BroadcastAwareness(incoming.Payload, c.userID, c, nil)
-			}
+		var evt DocSyncEvent
+		if err := json.Unmarshal(message, &evt); err == nil && evt.Type != "" {
+			c.room.HandleEvent(evt, c)
 		}
 	}
 }
 
 func (c *DocWSClient) writePump() {
-	ticker := time.NewTicker(30 * time.Second)
+	ticker := time.NewTicker(25 * time.Second)
 	defer func() {
 		ticker.Stop()
 		_ = c.conn.Close()