refactor(collab): remove live document collaboration and enforce strictly read-only sharing
MindSift CI & Docker Release / Build, Test in Container & Push (push) Successful in 1m2s
MindSift CI & Docker Release / Build, Test in Container & Push (push) Successful in 1m2s
This commit is contained in:
sandrodach committed
1 parent
aea05e868a
commit
04fa774d10
16 files changed
+179
-3556
No files matched your search
@@ -1,736 +0,0 @@
|
||||
package ws
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
"github.com/sandrodach/mindsift/internal/domain"
|
||||
)
|
||||
|
||||
// CursorPresence represents an active user's ephemeral cursor position in a document room.
|
||||
type CursorPresence struct {
|
||||
ClientID string `json:"clientId"`
|
||||
UserID string `json:"userId"`
|
||||
UserName string `json:"userName"`
|
||||
UserColor string `json:"userColor"`
|
||||
BlockID string `json:"blockId,omitempty"`
|
||||
SelectionStart *int `json:"selectionStart,omitempty"`
|
||||
SelectionEnd *int `json:"selectionEnd,omitempty"`
|
||||
UpdatedAt int64 `json:"updatedAt"`
|
||||
}
|
||||
|
||||
// DocWSClient represents an active WebSocket connection to a specific document room.
|
||||
type DocWSClient struct {
|
||||
hub *Hub
|
||||
room *DocRoom
|
||||
conn *websocket.Conn
|
||||
userID string
|
||||
clientID string
|
||||
canWrite bool
|
||||
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"`
|
||||
UserName string `json:"userName,omitempty"`
|
||||
UserColor string `json:"userColor,omitempty"`
|
||||
SelectionStart *int `json:"selectionStart,omitempty"`
|
||||
SelectionEnd *int `json:"selectionEnd,omitempty"`
|
||||
ActiveCursors []CursorPresence `json:"activeCursors,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 block state and 1.5s debounced SQLite persistence for one document.
|
||||
type DocRoom struct {
|
||||
mu sync.Mutex
|
||||
docID string
|
||||
hub *Hub
|
||||
title string
|
||||
blocks []domain.DocumentBlock
|
||||
loaded bool
|
||||
wsClients map[*DocWSClient]bool
|
||||
activeCursors map[string]*CursorPresence
|
||||
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()
|
||||
if h.docRooms == nil {
|
||||
h.docRooms = make(map[string]*DocRoom)
|
||||
}
|
||||
room, exists := h.docRooms[docID]
|
||||
if !exists {
|
||||
room = &DocRoom{
|
||||
docID: docID,
|
||||
hub: h,
|
||||
wsClients: make(map[*DocWSClient]bool),
|
||||
activeCursors: make(map[string]*CursorPresence),
|
||||
}
|
||||
h.docRooms[docID] = room
|
||||
}
|
||||
store := h.docStore
|
||||
h.mu.Unlock()
|
||||
|
||||
room.mu.Lock()
|
||||
if !room.loaded && store != nil {
|
||||
title, blocks, err := store.LoadDocumentState(docID)
|
||||
if err == nil {
|
||||
room.title = title
|
||||
room.blocks = normalizeRoomBlocks(docID, blocks)
|
||||
room.loaded = true
|
||||
}
|
||||
}
|
||||
room.mu.Unlock()
|
||||
|
||||
return room
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
room.mu.Lock()
|
||||
if title != "" {
|
||||
room.title = title
|
||||
}
|
||||
if blocks != nil {
|
||||
room.blocks = normalizeRoomBlocks(docID, blocks)
|
||||
}
|
||||
room.loaded = true
|
||||
room.mu.Unlock()
|
||||
}
|
||||
|
||||
// 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()
|
||||
|
||||
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()
|
||||
}
|
||||
}
|
||||
|
||||
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,
|
||||
},
|
||||
}
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
evt.UserID = sender.userID
|
||||
}
|
||||
|
||||
mutated := false
|
||||
|
||||
switch evt.Type {
|
||||
case "cursor_moved":
|
||||
if evt.ClientID != "" {
|
||||
cp := &CursorPresence{
|
||||
ClientID: evt.ClientID,
|
||||
UserID: evt.UserID,
|
||||
UserName: evt.UserName,
|
||||
UserColor: evt.UserColor,
|
||||
BlockID: evt.BlockID,
|
||||
SelectionStart: evt.SelectionStart,
|
||||
SelectionEnd: evt.SelectionEnd,
|
||||
UpdatedAt: time.Now().UnixMilli(),
|
||||
}
|
||||
if r.activeCursors == nil {
|
||||
r.activeCursors = make(map[string]*CursorPresence)
|
||||
}
|
||||
r.activeCursors[evt.ClientID] = cp
|
||||
|
||||
msgBytes, err := json.Marshal(evt)
|
||||
if err == nil {
|
||||
for client := range r.wsClients {
|
||||
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.mu.Unlock()
|
||||
return
|
||||
|
||||
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 != "" {
|
||||
found := false
|
||||
for i := range r.blocks {
|
||||
if r.blocks[i].ID == targetID {
|
||||
found = true
|
||||
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
|
||||
}
|
||||
}
|
||||
if !found && evt.Block != nil {
|
||||
newBlock := *evt.Block
|
||||
newBlock.DocumentID = r.docID
|
||||
if newBlock.Type == "" {
|
||||
newBlock.Type = "text"
|
||||
}
|
||||
if newBlock.Level < 0 {
|
||||
newBlock.Level = 0
|
||||
}
|
||||
r.blocks = append(r.blocks, newBlock)
|
||||
r.blocks = normalizeRoomBlocks(r.docID, r.blocks)
|
||||
bCopy := newBlock
|
||||
evt.Block = &bCopy
|
||||
evt.BlockID = targetID
|
||||
mutated = true
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
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:
|
||||
slog.Warn("Doc room WS client send buffer full, dropping message",
|
||||
"doc_id", r.docID,
|
||||
"client_id", client.clientID,
|
||||
"user_id", client.userID,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
r.scheduleDebouncedFlushLocked()
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
func (r *DocRoom) scheduleDebouncedFlushLocked() {
|
||||
if r.debounceTimer != nil {
|
||||
r.debounceTimer.Stop()
|
||||
}
|
||||
r.debounceTimer = time.AfterFunc(1500*time.Millisecond, func() {
|
||||
r.Flush()
|
||||
})
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
docID := r.docID
|
||||
title := r.title
|
||||
blocksCopy := make([]domain.DocumentBlock, len(r.blocks))
|
||||
copy(blocksCopy, r.blocks)
|
||||
r.dirty = false
|
||||
r.mu.Unlock()
|
||||
|
||||
tagsCount := 0
|
||||
for _, b := range blocksCopy {
|
||||
tagsCount += len(domain.ExtractTags(b.Content))
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
r.hub.mu.RLock()
|
||||
store := r.hub.docStore
|
||||
r.hub.mu.RUnlock()
|
||||
|
||||
if store != nil {
|
||||
if err := store.FlushDocumentState(docID, title, blocksCopy); err != nil {
|
||||
slog.Error("SQLite document flush failed",
|
||||
"doc_id", docID,
|
||||
"error", err,
|
||||
)
|
||||
} else {
|
||||
durationMs := float64(time.Since(start).Microseconds()) / 1000.0
|
||||
slog.Debug("SQLite document flushed",
|
||||
"doc_id", docID,
|
||||
"blocks_count", len(blocksCopy),
|
||||
"tags_count", tagsCount,
|
||||
"duration_ms", durationMs,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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, canWrite bool) {
|
||||
conn, err := upgrader.Upgrade(w, req, nil)
|
||||
if err != nil {
|
||||
slog.Error("WebSocket upgrade error", "doc_id", r.docID, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
if clientID == "" {
|
||||
clientID = req.URL.Query().Get("clientId")
|
||||
}
|
||||
if clientID == "" {
|
||||
clientID = domain.NewID("cli")
|
||||
}
|
||||
|
||||
userName := req.URL.Query().Get("userName")
|
||||
userColor := req.URL.Query().Get("userColor")
|
||||
|
||||
client := &DocWSClient{
|
||||
hub: r.hub,
|
||||
room: r,
|
||||
conn: conn,
|
||||
userID: userID,
|
||||
clientID: clientID,
|
||||
canWrite: canWrite,
|
||||
send: make(chan []byte, 256),
|
||||
}
|
||||
|
||||
r.mu.Lock()
|
||||
r.wsClients[client] = true
|
||||
activeClients := len(r.wsClients)
|
||||
if r.activeCursors == nil {
|
||||
r.activeCursors = make(map[string]*CursorPresence)
|
||||
}
|
||||
if userName != "" {
|
||||
r.activeCursors[clientID] = &CursorPresence{
|
||||
ClientID: clientID,
|
||||
UserID: userID,
|
||||
UserName: userName,
|
||||
UserColor: userColor,
|
||||
UpdatedAt: time.Now().UnixMilli(),
|
||||
}
|
||||
}
|
||||
title := r.title
|
||||
blocksCopy := make([]domain.DocumentBlock, len(r.blocks))
|
||||
copy(blocksCopy, r.blocks)
|
||||
|
||||
activeList := make([]CursorPresence, 0, len(r.activeCursors))
|
||||
for _, cp := range r.activeCursors {
|
||||
if cp != nil {
|
||||
activeList = append(activeList, *cp)
|
||||
}
|
||||
}
|
||||
r.mu.Unlock()
|
||||
|
||||
slog.Debug("WS client joined doc room",
|
||||
"doc_id", r.docID,
|
||||
"client_id", clientID,
|
||||
"user_id", userID,
|
||||
"active_clients", activeClients,
|
||||
)
|
||||
|
||||
// Immediately send live in-memory state and current cursors to the newly connected client
|
||||
initEvt := DocSyncEvent{
|
||||
Type: "doc_state",
|
||||
DocID: r.docID,
|
||||
ClientID: clientID,
|
||||
Title: title,
|
||||
Blocks: blocksCopy,
|
||||
ActiveCursors: activeList,
|
||||
}
|
||||
if initBytes, err := json.Marshal(initEvt); err == nil {
|
||||
client.send <- initBytes
|
||||
}
|
||||
|
||||
// If userName was provided, broadcast initial presence to existing clients in room
|
||||
if userName != "" {
|
||||
joinEvt := DocSyncEvent{
|
||||
Type: "cursor_moved",
|
||||
DocID: r.docID,
|
||||
ClientID: clientID,
|
||||
UserID: userID,
|
||||
UserName: userName,
|
||||
UserColor: userColor,
|
||||
}
|
||||
if joinBytes, err := json.Marshal(joinEvt); err == nil {
|
||||
r.mu.Lock()
|
||||
for c := range r.wsClients {
|
||||
if c != client && c.clientID != clientID {
|
||||
select {
|
||||
case c.send <- joinBytes:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
r.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
go client.writePump()
|
||||
go client.readPump()
|
||||
}
|
||||
|
||||
func (c *DocWSClient) readPump() {
|
||||
defer func() {
|
||||
c.room.mu.Lock()
|
||||
delete(c.room.wsClients, c)
|
||||
delete(c.room.activeCursors, c.clientID)
|
||||
remaining := len(c.room.wsClients)
|
||||
|
||||
// Broadcast client_left to all remaining clients in room
|
||||
leftEvt := DocSyncEvent{
|
||||
Type: "client_left",
|
||||
DocID: c.room.docID,
|
||||
ClientID: c.clientID,
|
||||
UserID: c.userID,
|
||||
}
|
||||
if leftBytes, err := json.Marshal(leftEvt); err == nil {
|
||||
for client := range c.room.wsClients {
|
||||
if client != c && client.clientID != c.clientID {
|
||||
select {
|
||||
case client.send <- leftBytes:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
c.room.mu.Unlock()
|
||||
|
||||
slog.Debug("WS client left doc room",
|
||||
"doc_id", c.room.docID,
|
||||
"client_id", c.clientID,
|
||||
"user_id", c.userID,
|
||||
"active_clients", remaining,
|
||||
)
|
||||
|
||||
_ = c.conn.Close()
|
||||
|
||||
// Flush immediately when the last client leaves the document room
|
||||
if remaining == 0 {
|
||||
c.room.Flush()
|
||||
}
|
||||
}()
|
||||
|
||||
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))
|
||||
return nil
|
||||
})
|
||||
|
||||
for {
|
||||
_, message, err := c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
|
||||
var evt DocSyncEvent
|
||||
if err := json.Unmarshal(message, &evt); err == nil && evt.Type != "" {
|
||||
targetBlockID := evt.BlockID
|
||||
if targetBlockID == "" && evt.Block != nil {
|
||||
targetBlockID = evt.Block.ID
|
||||
}
|
||||
|
||||
slog.Debug("WS doc event received",
|
||||
"type", evt.Type,
|
||||
"doc_id", c.room.docID,
|
||||
"block_id", targetBlockID,
|
||||
"sender_client_id", c.clientID,
|
||||
)
|
||||
|
||||
// Read-only clients can still move their cursor and send presence updates
|
||||
if !c.canWrite && evt.Type != "cursor_moved" {
|
||||
continue
|
||||
}
|
||||
c.room.HandleEvent(evt, c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *DocWSClient) writePump() {
|
||||
ticker := time.NewTicker(25 * 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
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,85 +0,0 @@
|
||||
package ws
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestDocRoomCursorPresenceAndEchoProtection(t *testing.T) {
|
||||
hub := NewHub()
|
||||
room := hub.GetOrCreateDocRoom("doc_test_presence")
|
||||
|
||||
c1 := &DocWSClient{
|
||||
hub: hub,
|
||||
room: room,
|
||||
userID: "user_1",
|
||||
clientID: "cli_1",
|
||||
canWrite: true,
|
||||
send: make(chan []byte, 10),
|
||||
}
|
||||
c2 := &DocWSClient{
|
||||
hub: hub,
|
||||
room: room,
|
||||
userID: "user_2",
|
||||
clientID: "cli_2",
|
||||
canWrite: false, // read-only viewer
|
||||
send: make(chan []byte, 10),
|
||||
}
|
||||
|
||||
room.mu.Lock()
|
||||
room.wsClients[c1] = true
|
||||
room.wsClients[c2] = true
|
||||
room.mu.Unlock()
|
||||
|
||||
// 1. Send cursor_moved from c1
|
||||
selStart := 5
|
||||
selEnd := 5
|
||||
room.HandleEvent(DocSyncEvent{
|
||||
Type: "cursor_moved",
|
||||
DocID: "doc_test_presence",
|
||||
ClientID: "cli_1",
|
||||
UserID: "user_1",
|
||||
UserName: "Alice",
|
||||
UserColor: "#3b82f6",
|
||||
BlockID: "blk_123",
|
||||
SelectionStart: &selStart,
|
||||
SelectionEnd: &selEnd,
|
||||
}, c1)
|
||||
|
||||
// c1 should NOT have received its own echo
|
||||
select {
|
||||
case <-c1.send:
|
||||
t.Fatal("c1 received echo of its own cursor_moved event!")
|
||||
default:
|
||||
}
|
||||
|
||||
// c2 SHOULD receive the event
|
||||
select {
|
||||
case msg := <-c2.send:
|
||||
if len(msg) == 0 {
|
||||
t.Fatal("empty message received by c2")
|
||||
}
|
||||
case <-time.After(500 * time.Millisecond):
|
||||
t.Fatal("timed out waiting for cursor_moved event on c2")
|
||||
}
|
||||
|
||||
// Room should have registered c1's presence
|
||||
room.mu.Lock()
|
||||
cp, exists := room.activeCursors["cli_1"]
|
||||
room.mu.Unlock()
|
||||
|
||||
if !exists || cp == nil {
|
||||
t.Fatal("expected active cursor for cli_1 in room")
|
||||
}
|
||||
if cp.UserName != "Alice" || cp.BlockID != "blk_123" {
|
||||
t.Fatalf("unexpected cursor data: %+v", cp)
|
||||
}
|
||||
|
||||
// Ephemeral cursor events must NOT mark the room dirty for disk flush
|
||||
room.mu.Lock()
|
||||
dirty := room.dirty
|
||||
room.mu.Unlock()
|
||||
if dirty {
|
||||
t.Fatal("cursor_moved should not mark room dirty")
|
||||
}
|
||||
}
|
||||
+2
-507
@@ -84,37 +84,6 @@ func (c *SSEClient) trySend(data []byte) {
|
||||
}
|
||||
}
|
||||
|
||||
type docLockItem struct {
|
||||
docID string
|
||||
userID string
|
||||
userName string
|
||||
lockedAt time.Time
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
type blockLockItem struct {
|
||||
docID string
|
||||
blockID string
|
||||
userID string
|
||||
userName string
|
||||
userColor string
|
||||
lockedAt time.Time
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
type liveSessionItem struct {
|
||||
docID string
|
||||
initiatorID string
|
||||
initiatorName string
|
||||
startedAt time.Time
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
type DocStore interface {
|
||||
LoadDocumentState(docID string) (string, []domain.DocumentBlock, error)
|
||||
FlushDocumentState(docID string, title string, blocks []domain.DocumentBlock) error
|
||||
}
|
||||
|
||||
type Hub struct {
|
||||
mu sync.RWMutex
|
||||
rooms map[string]map[*Client]bool
|
||||
@@ -123,37 +92,18 @@ type Hub struct {
|
||||
sseClients map[*SSEClient]bool
|
||||
sseUserRooms map[string]map[*SSEClient]bool
|
||||
sseRooms map[string]map[*SSEClient]bool
|
||||
locks map[string]*docLockItem
|
||||
blockLocks map[string]*blockLockItem // key: docID + ":" + blockID
|
||||
liveSessions map[string]*liveSessionItem // key: docID
|
||||
docRooms map[string]*DocRoom
|
||||
docStore DocStore
|
||||
stopCleanup chan struct{}
|
||||
closed bool
|
||||
}
|
||||
|
||||
func NewHub() *Hub {
|
||||
h := &Hub{
|
||||
return &Hub{
|
||||
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),
|
||||
blockLocks: make(map[string]*blockLockItem),
|
||||
liveSessions: make(map[string]*liveSessionItem),
|
||||
docRooms: make(map[string]*DocRoom),
|
||||
stopCleanup: make(chan struct{}),
|
||||
}
|
||||
go h.cleanupLocksLoop()
|
||||
return h
|
||||
}
|
||||
|
||||
func (h *Hub) SetDocStore(store DocStore) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
h.docStore = store
|
||||
}
|
||||
|
||||
func (h *Hub) Register(c *Client) {
|
||||
@@ -193,7 +143,6 @@ 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 {
|
||||
@@ -202,52 +151,19 @@ func (h *Hub) Unregister(c *Client) {
|
||||
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 {
|
||||
delete(uRoom, c)
|
||||
userRemaining += len(uRoom)
|
||||
if len(uRoom) == 0 {
|
||||
delete(h.userRooms, c.userID)
|
||||
}
|
||||
}
|
||||
if sseRoom, ok := h.sseUserRooms[c.userID]; ok {
|
||||
userRemaining += len(sseRoom)
|
||||
}
|
||||
}
|
||||
c.closeSend()
|
||||
|
||||
// If this user has no other active client connections, release their document locks
|
||||
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()
|
||||
|
||||
// Broadcast unlock events for released locks
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Hub) RegisterSSE(c *SSEClient) {
|
||||
@@ -294,7 +210,6 @@ func (h *Hub) UnregisterSSE(c *SSEClient) {
|
||||
}
|
||||
}
|
||||
}
|
||||
var docRoomsLeft []string
|
||||
if c.subscribedRooms != nil {
|
||||
for rID := range c.subscribedRooms {
|
||||
if room, ok := h.sseRooms[rID]; ok {
|
||||
@@ -303,48 +218,18 @@ func (h *Hub) UnregisterSSE(c *SSEClient) {
|
||||
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)
|
||||
}
|
||||
}
|
||||
c.closeSend()
|
||||
|
||||
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 {
|
||||
@@ -386,354 +271,9 @@ func (h *Hub) LeaveSSERoomByID(sseID, roomID string) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// Document Locking Methods
|
||||
|
||||
func (h *Hub) AcquireDocLock(docID, userID, userName string, ttl time.Duration) (*domain.DocLockInfo, bool, *domain.DocLockInfo) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
existing, exists := h.locks[docID]
|
||||
if exists && existing.expiresAt.After(now) {
|
||||
if existing.userID != userID {
|
||||
// Locked by another user
|
||||
info := &domain.DocLockInfo{
|
||||
DocID: existing.docID,
|
||||
UserID: existing.userID,
|
||||
UserName: existing.userName,
|
||||
ExpiresAt: existing.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
return nil, false, info
|
||||
}
|
||||
// Same user extending lock
|
||||
existing.expiresAt = now.Add(ttl)
|
||||
if userName != "" {
|
||||
existing.userName = userName
|
||||
}
|
||||
info := &domain.DocLockInfo{
|
||||
DocID: existing.docID,
|
||||
UserID: existing.userID,
|
||||
UserName: existing.userName,
|
||||
ExpiresAt: existing.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
return info, true, nil
|
||||
}
|
||||
|
||||
// Create fresh lock
|
||||
item := &docLockItem{
|
||||
docID: docID,
|
||||
userID: userID,
|
||||
userName: userName,
|
||||
lockedAt: now,
|
||||
expiresAt: now.Add(ttl),
|
||||
}
|
||||
h.locks[docID] = item
|
||||
info := &domain.DocLockInfo{
|
||||
DocID: item.docID,
|
||||
UserID: item.userID,
|
||||
UserName: item.userName,
|
||||
ExpiresAt: item.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
return info, true, nil
|
||||
}
|
||||
|
||||
func (h *Hub) HeartbeatDocLock(docID, userID string, ttl time.Duration) bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
item, ok := h.locks[docID]
|
||||
if !ok || item.userID != userID || item.expiresAt.Before(now) {
|
||||
return false
|
||||
}
|
||||
item.expiresAt = now.Add(ttl)
|
||||
return true
|
||||
}
|
||||
|
||||
func (h *Hub) ReleaseDocLock(docID, userID string, isAdmin bool) bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
item, ok := h.locks[docID]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
if item.userID == userID || isAdmin || userID == "admin" {
|
||||
delete(h.locks, docID)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (h *Hub) GetDocLock(docID string) *domain.DocLockInfo {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
|
||||
item, ok := h.locks[docID]
|
||||
if !ok || item.expiresAt.Before(time.Now()) {
|
||||
return nil
|
||||
}
|
||||
return &domain.DocLockInfo{
|
||||
DocID: item.docID,
|
||||
UserID: item.userID,
|
||||
UserName: item.userName,
|
||||
ExpiresAt: item.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
}
|
||||
|
||||
// Block Locking Methods (Option B)
|
||||
|
||||
func (h *Hub) AcquireBlockLock(docID, blockID, userID, userName, userColor string, ttl time.Duration) (*domain.BlockLockInfo, bool, *domain.BlockLockInfo) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
key := docID + ":" + blockID
|
||||
existing, exists := h.blockLocks[key]
|
||||
if exists && existing.expiresAt.After(now) {
|
||||
if existing.userID != userID {
|
||||
info := &domain.BlockLockInfo{
|
||||
DocID: existing.docID,
|
||||
BlockID: existing.blockID,
|
||||
UserID: existing.userID,
|
||||
UserName: existing.userName,
|
||||
UserColor: existing.userColor,
|
||||
ExpiresAt: existing.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
return nil, false, info
|
||||
}
|
||||
// Refresh lease
|
||||
existing.expiresAt = now.Add(ttl)
|
||||
if userName != "" {
|
||||
existing.userName = userName
|
||||
}
|
||||
if userColor != "" {
|
||||
existing.userColor = userColor
|
||||
}
|
||||
info := &domain.BlockLockInfo{
|
||||
DocID: existing.docID,
|
||||
BlockID: existing.blockID,
|
||||
UserID: existing.userID,
|
||||
UserName: existing.userName,
|
||||
UserColor: existing.userColor,
|
||||
ExpiresAt: existing.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
return info, true, nil
|
||||
}
|
||||
|
||||
item := &blockLockItem{
|
||||
docID: docID,
|
||||
blockID: blockID,
|
||||
userID: userID,
|
||||
userName: userName,
|
||||
userColor: userColor,
|
||||
lockedAt: now,
|
||||
expiresAt: now.Add(ttl),
|
||||
}
|
||||
h.blockLocks[key] = item
|
||||
info := &domain.BlockLockInfo{
|
||||
DocID: item.docID,
|
||||
BlockID: item.blockID,
|
||||
UserID: item.userID,
|
||||
UserName: item.userName,
|
||||
UserColor: item.userColor,
|
||||
ExpiresAt: item.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
return info, true, nil
|
||||
}
|
||||
|
||||
func (h *Hub) ReleaseBlockLock(docID, blockID, userID string) bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
key := docID + ":" + blockID
|
||||
existing, exists := h.blockLocks[key]
|
||||
if !exists {
|
||||
return false
|
||||
}
|
||||
if userID != "" && existing.userID != userID && userID != "admin" {
|
||||
return false
|
||||
}
|
||||
delete(h.blockLocks, key)
|
||||
return true
|
||||
}
|
||||
|
||||
func (h *Hub) GetDocBlockLocks(docID string) []domain.BlockLockInfo {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
|
||||
now := time.Now()
|
||||
var result []domain.BlockLockInfo
|
||||
prefix := docID + ":"
|
||||
for k, l := range h.blockLocks {
|
||||
if strings.HasPrefix(k, prefix) && l.expiresAt.After(now) {
|
||||
result = append(result, domain.BlockLockInfo{
|
||||
DocID: l.docID,
|
||||
BlockID: l.blockID,
|
||||
UserID: l.userID,
|
||||
UserName: l.userName,
|
||||
UserColor: l.userColor,
|
||||
ExpiresAt: l.expiresAt.Format(time.RFC3339),
|
||||
})
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// Live Session Methods (Opt-In Live Collaboration)
|
||||
|
||||
func (h *Hub) StartLiveSession(docID, userID, userName string, ttl time.Duration) (*domain.LiveSessionInfo, bool, *domain.LiveSessionInfo) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
now := time.Now()
|
||||
existing, exists := h.liveSessions[docID]
|
||||
if exists && existing.expiresAt.After(now) {
|
||||
if existing.initiatorID != userID {
|
||||
info := &domain.LiveSessionInfo{
|
||||
DocID: existing.docID,
|
||||
InitiatorID: existing.initiatorID,
|
||||
InitiatorName: existing.initiatorName,
|
||||
StartedAt: existing.startedAt.Format(time.RFC3339),
|
||||
ExpiresAt: existing.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
return nil, false, info
|
||||
}
|
||||
// Refresh lease
|
||||
existing.expiresAt = now.Add(ttl)
|
||||
info := &domain.LiveSessionInfo{
|
||||
DocID: existing.docID,
|
||||
InitiatorID: existing.initiatorID,
|
||||
InitiatorName: existing.initiatorName,
|
||||
StartedAt: existing.startedAt.Format(time.RFC3339),
|
||||
ExpiresAt: existing.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
return info, true, nil
|
||||
}
|
||||
|
||||
item := &liveSessionItem{
|
||||
docID: docID,
|
||||
initiatorID: userID,
|
||||
initiatorName: userName,
|
||||
startedAt: now,
|
||||
expiresAt: now.Add(ttl),
|
||||
}
|
||||
h.liveSessions[docID] = item
|
||||
info := &domain.LiveSessionInfo{
|
||||
DocID: item.docID,
|
||||
InitiatorID: item.initiatorID,
|
||||
InitiatorName: item.initiatorName,
|
||||
StartedAt: item.startedAt.Format(time.RFC3339),
|
||||
ExpiresAt: item.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
return info, true, nil
|
||||
}
|
||||
|
||||
func (h *Hub) EndLiveSession(docID, userID string) bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
existing, exists := h.liveSessions[docID]
|
||||
if !exists {
|
||||
return false
|
||||
}
|
||||
if userID != "" && existing.initiatorID != userID && userID != "admin" {
|
||||
return false
|
||||
}
|
||||
delete(h.liveSessions, docID)
|
||||
return true
|
||||
}
|
||||
|
||||
func (h *Hub) ForceEndLiveSession(docID string) bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
if _, exists := h.liveSessions[docID]; !exists {
|
||||
return false
|
||||
}
|
||||
delete(h.liveSessions, docID)
|
||||
return true
|
||||
}
|
||||
|
||||
func (h *Hub) GetLiveSession(docID string) *domain.LiveSessionInfo {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
|
||||
now := time.Now()
|
||||
existing, exists := h.liveSessions[docID]
|
||||
if !exists || existing.expiresAt.Before(now) {
|
||||
return nil
|
||||
}
|
||||
return &domain.LiveSessionInfo{
|
||||
DocID: existing.docID,
|
||||
InitiatorID: existing.initiatorID,
|
||||
InitiatorName: existing.initiatorName,
|
||||
StartedAt: existing.startedAt.Format(time.RFC3339),
|
||||
ExpiresAt: existing.expiresAt.Format(time.RFC3339),
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Hub) cleanupLocksLoop() {
|
||||
ticker := time.NewTicker(2 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-h.stopCleanup:
|
||||
return
|
||||
case <-ticker.C:
|
||||
h.mu.Lock()
|
||||
now := time.Now()
|
||||
|
||||
var expiredDocIDs []string
|
||||
for docID, l := range h.locks {
|
||||
if l.expiresAt.Before(now) {
|
||||
expiredDocIDs = append(expiredDocIDs, docID)
|
||||
delete(h.locks, docID)
|
||||
}
|
||||
}
|
||||
|
||||
type expBlock struct {
|
||||
docID string
|
||||
blockID string
|
||||
}
|
||||
var expiredBlocks []expBlock
|
||||
for k, bl := range h.blockLocks {
|
||||
if bl.expiresAt.Before(now) {
|
||||
expiredBlocks = append(expiredBlocks, expBlock{docID: bl.docID, blockID: bl.blockID})
|
||||
delete(h.blockLocks, k)
|
||||
}
|
||||
}
|
||||
|
||||
var expiredLiveSessions []string
|
||||
for docID, ls := range h.liveSessions {
|
||||
if ls.expiresAt.Before(now) {
|
||||
expiredLiveSessions = append(expiredLiveSessions, docID)
|
||||
delete(h.liveSessions, docID)
|
||||
}
|
||||
}
|
||||
h.mu.Unlock()
|
||||
|
||||
for _, docID := range expiredDocIDs {
|
||||
h.BroadcastToAll("DOC_UNLOCKED", map[string]string{"docId": docID, "doc_id": docID})
|
||||
}
|
||||
for _, b := range expiredBlocks {
|
||||
h.BroadcastRoom("doc:"+b.docID, "DOC_BLOCK_UNLOCKED", map[string]string{
|
||||
"docId": b.docID,
|
||||
"blockId": b.blockID,
|
||||
}, nil)
|
||||
}
|
||||
for _, docID := range expiredLiveSessions {
|
||||
h.BroadcastRoom("doc:"+docID, "DOC_LIVE_ENDED", map[string]string{
|
||||
"docId": docID,
|
||||
}, nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Close gracefully shuts down the Hub, stops background tickers, flushes dirty document rooms to DB,
|
||||
// and immediately disconnects all WebSocket and SSE connections so HTTP shutdown completes in milliseconds.
|
||||
// Close gracefully disconnects all WebSocket and SSE connections so HTTP shutdown completes immediately.
|
||||
func (h *Hub) Close() {
|
||||
h.mu.Lock()
|
||||
if h.closed {
|
||||
@@ -742,19 +282,6 @@ func (h *Hub) Close() {
|
||||
}
|
||||
h.closed = true
|
||||
|
||||
if h.stopCleanup != nil {
|
||||
select {
|
||||
case <-h.stopCleanup:
|
||||
default:
|
||||
close(h.stopCleanup)
|
||||
}
|
||||
}
|
||||
|
||||
docRooms := make([]*DocRoom, 0, len(h.docRooms))
|
||||
for _, r := range h.docRooms {
|
||||
docRooms = append(docRooms, r)
|
||||
}
|
||||
|
||||
allWsClients := make([]*Client, 0, len(h.allClients))
|
||||
for c := range h.allClients {
|
||||
allWsClients = append(allWsClients, c)
|
||||
@@ -766,28 +293,11 @@ func (h *Hub) Close() {
|
||||
}
|
||||
h.mu.Unlock()
|
||||
|
||||
// 1. Flush any active dirty document rooms and close doc WS connections
|
||||
for _, r := range docRooms {
|
||||
r.Flush()
|
||||
r.mu.Lock()
|
||||
if r.debounceTimer != nil {
|
||||
r.debounceTimer.Stop()
|
||||
r.debounceTimer = nil
|
||||
}
|
||||
for c := range r.wsClients {
|
||||
_ = c.conn.Close()
|
||||
}
|
||||
r.wsClients = make(map[*DocWSClient]bool)
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
// 2. Close all WebSocket connections
|
||||
for _, c := range allWsClients {
|
||||
_ = c.conn.Close()
|
||||
c.closeSend()
|
||||
}
|
||||
|
||||
// 3. Close all SSE client channels so ServeSSE select loop exits immediately
|
||||
for _, c := range allSSEClients {
|
||||
c.closeSend()
|
||||
}
|
||||
@@ -1116,21 +626,6 @@ func (c *Client) readPump() {
|
||||
c.hub.LeaveRoom(c, room)
|
||||
}
|
||||
}
|
||||
|
||||
if docID != "" {
|
||||
switch incoming.Event {
|
||||
case "LOCK_DOC":
|
||||
if info, ok, _ := c.hub.AcquireDocLock(docID, c.userID, "", 30*time.Second); ok && info != nil {
|
||||
c.hub.BroadcastDocEvent(docID, "DOC_LOCKED", info)
|
||||
}
|
||||
case "HEARTBEAT_DOC":
|
||||
c.hub.HeartbeatDocLock(docID, c.userID, 30*time.Second)
|
||||
case "UNLOCK_DOC":
|
||||
if c.hub.ReleaseDocLock(docID, c.userID, false) {
|
||||
c.hub.BroadcastDocEvent(docID, "DOC_UNLOCKED", map[string]string{"docId": docID, "doc_id": docID})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user