Skip to content

Commit 0a36d64

Browse files
authored
feat(realtime): accurate in-file presence + collaborative-caret polish (#5965)
Per-session file-doc presence (avatars count other sessions like the canvas), StrictMode-safe stable Y.Doc (fixes blank-doc on join), flush caret cap + restored hover hit-slop, and three join-lifecycle race fixes unifying file-doc + workspace-files on one intent-tracked monotonic generation model. All findings root-caused with regression tests.
1 parent 790d93a commit 0a36d64

22 files changed

Lines changed: 1112 additions & 385 deletions

File tree

apps/realtime/src/handlers/connection.ts

Lines changed: 17 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { createLogger } from '@sim/logger'
2-
import { parseRoomName, type RoomRef, roomName } from '@sim/realtime-protocol/rooms'
2+
import { parseRoomName, ROOM_TYPES, type RoomRef, roomName } from '@sim/realtime-protocol/rooms'
33
import { cleanupFileDocForSocket } from '@/handlers/file-doc'
44
import { cleanupPendingSubblocksForSocket } from '@/handlers/subblocks'
55
import { cleanupPendingVariablesForSocket } from '@/handlers/variables'
@@ -8,6 +8,14 @@ import type { IRoomManager } from '@/rooms'
88

99
const logger = createLogger('ConnectionHandlers')
1010

11+
/**
12+
* Room types whose presence lives in the room manager (Redis-backed), so a disconnect must
13+
* remove the socket + broadcast a correction. The workspace-files and file-doc rooms are
14+
* NOT here: workspace-files carries no presence (native Socket.IO membership only), and
15+
* file-doc broadcasts its own server-authenticated roster via `cleanupFileDocForSocket`.
16+
*/
17+
const PRESENCE_BEARING_TYPES = new Set<RoomRef['type']>([ROOM_TYPES.WORKFLOW, ROOM_TYPES.TABLE])
18+
1119
export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
1220
socket.on('error', (error) => {
1321
logger.error(`Socket ${socket.id} error:`, error)
@@ -33,8 +41,9 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager
3341
cleanupPendingSubblocksForSocket(socket.id)
3442
cleanupPendingVariablesForSocket(socket.id)
3543
// Clear the socket's collaborative-document awareness (removes its caret for
36-
// everyone else) and drop the room if it was the last editor.
37-
cleanupFileDocForSocket(socket.id, roomManager.io)
44+
// everyone else) and drop the room if it was the last editor. `endOfLife` drops the
45+
// socket's join-generation entry — safe only here, on true disconnect (see cleanup).
46+
cleanupFileDocForSocket(socket.id, roomManager.io, true)
3847

3948
// A socket may occupy multiple rooms (one per type). Remove it from every
4049
// room the manager knows about.
@@ -47,12 +56,13 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager
4756
const wasInRooms = new Map<string, RoomRef>()
4857
for (const room of removedRooms) wasInRooms.set(roomName(room), room)
4958
for (const name of liveRoomNames) {
50-
// `wasInRooms.has(name)` already excludes every room the manager removed
51-
// (same room-name key via the roomName/parseRoomName bijection), so any
52-
// room reaching here was NOT in `removedRooms` and needs a removal attempt.
59+
// `wasInRooms.has(name)` already excludes every room the manager removed (same
60+
// room-name key via the roomName/parseRoomName bijection). Skip room types with no
61+
// manager-tracked presence (workspace-files, file-doc): removing there is a no-op and
62+
// broadcasting a correction would emit a dead presence-update no client listens to.
5363
if (name === socket.id || wasInRooms.has(name)) continue
5464
const ref = parseRoomName(name)
55-
if (!ref) continue
65+
if (!ref || !PRESENCE_BEARING_TYPES.has(ref.type)) continue
5666
wasInRooms.set(name, ref)
5767
await roomManager.removeUserFromRoom(ref, socket.id)
5868
}

apps/realtime/src/handlers/file-doc.test.ts

Lines changed: 85 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,8 @@ function createSocket(id: string, overrides?: Record<string, unknown>) {
5454
id,
5555
userId: 'user-1',
5656
userName: 'Test User',
57+
// Set so the server's roster resolves the avatar from the socket (never the DB).
58+
userImage: 'avatar.png',
5759
disconnected: false,
5860
on: vi.fn((event: string, handler: Handler) => {
5961
handlers[event] = handler
@@ -133,7 +135,9 @@ describe('setupWorkspaceFileDocHandlers', () => {
133135
afterEach(() => {
134136
// The room store is module-global; drop every room the test's sockets opened.
135137
const { io } = createIo()
136-
for (const id of createdSocketIds) cleanupFileDocForSocket(id, io)
138+
// Simulate a full disconnect between tests (`endOfLife`) so the module-global join-generation
139+
// map is cleared and never bleeds a counter into the next test.
140+
for (const id of createdSocketIds) cleanupFileDocForSocket(id, io, true)
137141
createdSocketIds.clear()
138142
vi.clearAllTimers()
139143
vi.useRealTimers()
@@ -435,7 +439,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
435439

436440
const pending = s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
437441
s.socket.disconnected = true
438-
cleanupFileDocForSocket('socket-a', io) // disconnect cleanup — no-op, nothing registered yet
442+
cleanupFileDocForSocket('socket-a', io, true) // disconnect cleanup — no-op, nothing registered yet
439443
resolveAuth({ allowed: true, status: 200, workspacePermission: 'write' })
440444
await pending
441445

@@ -462,6 +466,49 @@ describe('setupWorkspaceFileDocHandlers', () => {
462466
expect(s.socket.join).toHaveBeenCalledWith('workspace-file-doc:file-2')
463467
})
464468

469+
it('does not reset the join generation on a leave, so an in-flight join still binds', async () => {
470+
const { io } = createIo()
471+
const s = setup('socket-a', io)
472+
473+
// file-1 join completes; the socket is registered in file-1.
474+
await s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
475+
476+
// file-2 join goes in-flight (authorize deferred).
477+
let resolveAuth: (v: unknown) => void = () => {}
478+
mockAuthorizeRoom.mockReturnValueOnce(new Promise((resolve) => (resolveAuth = resolve)))
479+
const pending = s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-2', clientId: 1 })
480+
481+
// A deferred leave for the prior file-1 lands while file-2's join awaits authorization. Its
482+
// cleanup must NOT reset the monotonic join generation, or file-2's guard would see an emptied
483+
// map (`undefined !== generation`) and abort the join the client actually wants.
484+
s.handlers[FILE_DOC_EVENTS.LEAVE]({ fileId: 'file-1' })
485+
486+
resolveAuth({ allowed: true, status: 200, workspacePermission: 'write' })
487+
await pending
488+
489+
expect(joinSuccessFileId(s.socket)).toBe('file-2')
490+
expect(s.socket.join).toHaveBeenCalledWith('workspace-file-doc:file-2')
491+
})
492+
493+
it('cancels an in-flight join when the client leaves that same file (no ghost owner)', async () => {
494+
const { io, sent } = createIo()
495+
let resolveAuth: (v: unknown) => void = () => {}
496+
mockAuthorizeRoom.mockReturnValueOnce(new Promise((resolve) => (resolveAuth = resolve)))
497+
const s = setup('socket-a', io)
498+
499+
// Join file-1 is awaiting authorization when the client leaves file-1 (fast open→close).
500+
const pending = s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
501+
s.handlers[FILE_DOC_EVENTS.LEAVE]({ fileId: 'file-1' })
502+
resolveAuth({ allowed: true, status: 200, workspacePermission: 'write' })
503+
await pending
504+
505+
// The stale join must not register: no success, no room join, and no presence broadcast that
506+
// would leave a ghost collaborator until disconnect.
507+
expect(s.socket.join).not.toHaveBeenCalled()
508+
expect(joinSuccessFileId(s.socket)).toBeUndefined()
509+
expect(sent.some((m) => m.event === FILE_DOC_EVENTS.PRESENCE)).toBe(false)
510+
})
511+
465512
it('scopes LEAVE to the named file (a leave for a different file is a no-op)', async () => {
466513
const { io } = createIo()
467514
const a = setup('socket-a', io)
@@ -574,4 +621,40 @@ describe('setupWorkspaceFileDocHandlers', () => {
574621
// No re-election: the successful seed cancelled the deadline.
575622
expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)).toBeUndefined()
576623
})
624+
625+
it('broadcasts a server-authenticated presence roster on join, one entry per session', async () => {
626+
const { io, sent } = createIo()
627+
const a = setup('socket-a', io, { userId: 'user-a', userName: 'Ada', userImage: 'ada.png' })
628+
const b = setup('socket-b', io, { userId: 'user-b', userName: 'Bob', userImage: 'bob.png' })
629+
630+
await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
631+
await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 })
632+
633+
const roster = sent.filter((m) => m.event === FILE_DOC_EVENTS.PRESENCE).at(-1)?.payload as {
634+
fileId: string
635+
users: Array<{ socketId: string; userId: string; userName: string; avatarUrl: string | null }>
636+
}
637+
expect(roster.fileId).toBe('file-1')
638+
// Identity is each socket's authenticated session — not any client-supplied value.
639+
expect([...roster.users].sort((x, y) => x.userId.localeCompare(y.userId))).toEqual([
640+
{ socketId: 'socket-a', userId: 'user-a', userName: 'Ada', avatarUrl: 'ada.png' },
641+
{ socketId: 'socket-b', userId: 'user-b', userName: 'Bob', avatarUrl: 'bob.png' },
642+
])
643+
})
644+
645+
it('keeps a per-session entry for two sockets of the SAME user (no server-side user dedup)', async () => {
646+
const { io, sent } = createIo()
647+
// Two tabs of one account: the client self-excludes its own socket, so the roster must carry
648+
// BOTH sessions or a client could never see the other tab as present.
649+
const a = setup('socket-a', io, { userId: 'user-a', userName: 'Ada', userImage: 'ada.png' })
650+
const b = setup('socket-b', io, { userId: 'user-a', userName: 'Ada', userImage: 'ada.png' })
651+
652+
await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
653+
await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 })
654+
655+
const roster = sent.filter((m) => m.event === FILE_DOC_EVENTS.PRESENCE).at(-1)?.payload as {
656+
users: Array<{ socketId: string; userId: string }>
657+
}
658+
expect([...roster.users].map((u) => u.socketId).sort()).toEqual(['socket-a', 'socket-b'])
659+
})
577660
})

apps/realtime/src/handlers/file-doc.ts

Lines changed: 73 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import {
44
FILE_DOC_EVENTS,
55
FILE_DOC_MESSAGE_TYPE,
66
FILE_DOC_SEED,
7+
type FileDocPresenceUser,
78
type JoinFileDocPayload,
89
type LeaveFileDocPayload,
910
toFileDocBytes,
@@ -15,6 +16,7 @@ import type { Server } from 'socket.io'
1516
import * as awarenessProtocol from 'y-protocols/awareness'
1617
import * as syncProtocol from 'y-protocols/sync'
1718
import * as Y from 'yjs'
19+
import { resolveAvatarUrl } from '@/handlers/avatar'
1820
import type { AuthenticatedSocket } from '@/middleware/auth'
1921
import type { IRoomManager } from '@/rooms'
2022

@@ -57,6 +59,10 @@ interface FileDocOwner {
5759
/** The owning user — used to tell a reconnect (same user reusing its Yjs client
5860
* id) from a spoof (a different user binding a peer's id). */
5961
userId: string
62+
/** Server-authenticated display identity for the presence roster (from the socket's
63+
* session, never the client-set awareness — so a peer cannot spoof it). */
64+
userName: string
65+
avatarUrl: string | null
6066
}
6167

6268
interface FileDocRoom {
@@ -116,6 +122,28 @@ function broadcast(io: Server, name: string, payload: Uint8Array, exceptSocketId
116122
channel.emit(FILE_DOC_EVENTS.MESSAGE, payload)
117123
}
118124

125+
/**
126+
* Broadcast the room's collaborator roster to everyone in it, for the avatar stack. One entry
127+
* PER SESSION (socket) — the client excludes its own socket and dedupes the remainder per user
128+
* for display, so a second tab of the same account still registers as present (mirroring the
129+
* canvas presence model). Deduping here instead would drop the current user's other sessions
130+
* asymmetrically (only one socket survives), so each client could never reliably self-exclude.
131+
* Identity comes from each owner's server-authenticated session — never the client-set awareness
132+
* — so a peer cannot spoof or suppress an entry.
133+
*/
134+
function broadcastFileDocPresence(io: Server, name: string, room: FileDocRoom) {
135+
const users: FileDocPresenceUser[] = []
136+
for (const [socketId, owner] of room.owners) {
137+
users.push({
138+
socketId,
139+
userId: owner.userId,
140+
userName: owner.userName,
141+
avatarUrl: owner.avatarUrl,
142+
})
143+
}
144+
io.to(name).emit(FILE_DOC_EVENTS.PRESENCE, { fileId: room.fileId, users })
145+
}
146+
119147
/** Whether the client has recorded that it seeded the document's initial content. */
120148
function isDocSeeded(doc: Y.Doc): boolean {
121149
return doc.getMap(FILE_DOC_SEED.configMap).get(FILE_DOC_SEED.flag) === true
@@ -316,10 +344,15 @@ function handleMessage(socket: AuthenticatedSocket, data: unknown) {
316344
* seeding, and drop the room's document when the last collaborator leaves.
317345
* Exported for the disconnect handler; safe to call for a socket in no room.
318346
*/
319-
export function cleanupFileDocForSocket(socketId: string, io: Server): void {
320-
// Drop the join generation so an in-flight JOIN for this socket aborts after
321-
// its authorize resolves, and the map never leaks across the socket's life.
322-
joinGeneration.delete(socketId)
347+
export function cleanupFileDocForSocket(socketId: string, io: Server, endOfLife = false): void {
348+
// The join-generation counter is monotonic for the socket's WHOLE life and must survive a room
349+
// switch/leave: resetting it here would let the next join reuse a low number that a still
350+
// in-flight earlier join also holds, so that stale join passes the generation guard and rebinds
351+
// the socket to the wrong document. Drop it ONLY when the socket is truly gone (disconnect),
352+
// which is also the only place the map would otherwise leak. An in-flight join is already
353+
// aborted on disconnect by the `socket.disconnected` check, and on a switch by a newer join
354+
// bumping the generation — neither needs this delete.
355+
if (endOfLife) joinGeneration.delete(socketId)
323356

324357
const name = socketToRoomName.get(socketId)
325358
if (!name) return
@@ -334,6 +367,8 @@ export function cleanupFileDocForSocket(socketId: string, io: Server): void {
334367
// Fires the awareness `update` handler with a non-socket origin → the removal
335368
// is broadcast to every remaining client, so the departed caret vanishes.
336369
awarenessProtocol.removeAwarenessStates(room.awareness, [owner.clientId], null)
370+
// Refresh the roster for whoever remains (server-authenticated identity).
371+
broadcastFileDocPresence(io, name, room)
337372
}
338373

339374
// Hand off the seeder role: if the elected seeder left before it seeded, elect
@@ -351,12 +386,22 @@ export function cleanupFileDocForSocket(socketId: string, io: Server): void {
351386
* file id; joining requires workspace `write` (editing a document). Mirrors the
352387
* workspace-files join shape (auth → readiness → validate → authorize → join),
353388
* then runs the Yjs sync/awareness handshake.
389+
*
390+
* The avatar roster is derived from this room's own `owners` map and broadcast as
391+
* `FILE_DOC_EVENTS.PRESENCE` — NOT the Redis-backed room-manager presence the workflow /
392+
* table rooms use — because the file-doc room already owns an authoritative in-memory Y.Doc
393+
* pinned to a single replica, so the session identity is right here with no extra store.
354394
*/
355395
export function setupWorkspaceFileDocHandlers(
356396
socket: AuthenticatedSocket,
357397
roomManager: IRoomManager
358398
) {
359399
const io = roomManager.io
400+
// The file this socket currently intends to edit (set when a join starts). A leave targeting it
401+
// — or an unscoped leave — advances the join generation to cancel an in-flight join, so a join
402+
// awaiting authorization can't complete after the client left and register a ghost owner. A
403+
// leave for a DIFFERENT file must NOT cancel it (a document switch), mirroring workspace-files.
404+
let currentFileId: string | null = null
360405

361406
socket.on(FILE_DOC_EVENTS.JOIN, async ({ fileId, clientId }: JoinFileDocPayload) => {
362407
try {
@@ -376,9 +421,11 @@ export function setupWorkspaceFileDocHandlers(
376421
return
377422
}
378423

379-
// Claim this JOIN's generation before the async authorize below.
424+
// Claim this JOIN's generation before the async authorize below, and record the file the
425+
// socket now intends to edit so a leave for it can cancel this join if it's still in-flight.
380426
const generation = (joinGeneration.get(socket.id) ?? 0) + 1
381427
joinGeneration.set(socket.id, generation)
428+
currentFileId = fileId
382429

383430
const room = fileDocRoom(fileId)
384431
const name = roomName(room)
@@ -408,9 +455,13 @@ export function setupWorkspaceFileDocHandlers(
408455
return
409456
}
410457

411-
// Abort a JOIN superseded during authorization: the socket disconnected, or
412-
// a newer JOIN (a document switch) bumped the generation. Registering here
413-
// would leak a dead socket's room or bind the socket to the wrong document.
458+
// Server-authenticated identity for the presence roster (never trusts the client-set
459+
// awareness). Resolved here so the generation guard below also covers this await.
460+
const avatarUrl = await resolveAvatarUrl(socket, userId)
461+
462+
// Abort a JOIN superseded during authorization/identity resolution: the socket
463+
// disconnected, or a newer JOIN (a document switch) bumped the generation. Registering
464+
// here would leak a dead socket's room or bind the socket to the wrong document.
414465
if (socket.disconnected || joinGeneration.get(socket.id) !== generation) return
415466

416467
// Switched documents on the same socket — leave the previous one first (a
@@ -449,11 +500,13 @@ export function setupWorkspaceFileDocHandlers(
449500
awarenessProtocol.removeAwarenessStates(entry.awareness, [previous.clientId], null)
450501
}
451502

452-
entry.owners.set(socket.id, { clientId, userId })
503+
entry.owners.set(socket.id, { clientId, userId, userName, avatarUrl })
453504
socketToRoomName.set(socket.id, name)
454505
socket.join(name)
455506

456507
socket.emit(FILE_DOC_EVENTS.JOIN_SUCCESS, { fileId })
508+
// Server-authenticated roster → everyone in the room, including this joiner.
509+
broadcastFileDocPresence(io, name, entry)
457510

458511
// Begin the sync handshake: send the server's state (sync step 1). The
459512
// client replies with its updates and requests the server's in return.
@@ -496,10 +549,17 @@ export function setupWorkspaceFileDocHandlers(
496549

497550
socket.on(FILE_DOC_EVENTS.LEAVE, (payload?: LeaveFileDocPayload) => {
498551
try {
499-
// Only affect a REGISTERED room; never touch the join generation here. A
500-
// leave that raced ahead of an in-flight join (no room registered yet) is a
501-
// no-op — bumping the generation would silently abort an unrelated join for
502-
// a different file (a document switch), leaving the socket bound to nothing.
552+
// Cancel an in-flight join whose file the client is now leaving (or an unscoped leave): a
553+
// join still awaiting authorization would otherwise complete after the client left, register
554+
// as an owner, and broadcast a ghost collaborator until disconnect. Guard on the current
555+
// file intent so a stale/deferred leave for a DIFFERENT file can't abort the join the client
556+
// has since switched to (bumping the generation blindly caused that regression in #5941).
557+
if (!payload?.fileId || payload.fileId === currentFileId) {
558+
joinGeneration.set(socket.id, (joinGeneration.get(socket.id) ?? 0) + 1)
559+
currentFileId = null
560+
}
561+
// Tear down membership only for a REGISTERED room; a leave that raced ahead of an in-flight
562+
// join (nothing registered yet) has already cancelled it above.
503563
const name = socketToRoomName.get(socket.id)
504564
if (!name) return
505565
// Scope the leave to the named file when provided: a deferred leave from a

0 commit comments

Comments
 (0)