fix(file-editor): recover collaboration after reconnect

Author: waleedlatif1

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

@@ -545,6 +545,10 @@ describe('setupWorkspaceFileDocHandlers', () => {
 
     expect(socket.join).toHaveBeenCalledWith(ROOM_NAME)
     expect(joinSuccessFileId(socket)).toBe('file-1')
+    expect(socket.emit).toHaveBeenCalledWith(
+      FILE_DOC_EVENTS.JOIN_SUCCESS,
+      expect.objectContaining({ fileId: 'file-1', clientId: 1 })
+    )
 
     // A binary sync-step-1 message (type tag 0) is sent to kick off the handshake.
     const syncMessage = socket.emit.mock.calls.find(
@@ -843,7 +847,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(mockFetchFileDocMerge).toHaveBeenCalledTimes(2)
   })
 
-  it('relays a document update to the rest of the room, excluding the sender', async () => {
+  it('relays a document update to every provider, including siblings on the sender socket', async () => {
     const { io, sent } = createIo()
     const a = setup('socket-a', io)
     const b = setup('socket-b', io)
@@ -860,7 +864,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
 
     const relayed = sent.find((m) => m.event === FILE_DOC_EVENTS.MESSAGE)
     expect(relayed?.target).toBe(ROOM_NAME)
-    expect(relayed?.except).toBe('socket-a')
+    expect(relayed?.except).toBeUndefined()
     expect((relayed?.payload as Uint8Array)[0]).toBe(FILE_DOC_MESSAGE_TYPE.SYNC)
   })
 
@@ -959,6 +963,35 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(relayedFor(999)).toBeUndefined()
   })
 
+  it('accepts concurrent joins from co-mounted providers for the same file', async () => {
+    let resolveFirstAuth: (value: unknown) => void = () => {}
+    mockAuthorizeRoom.mockReturnValueOnce(
+      new Promise((resolve) => {
+        resolveFirstAuth = resolve
+      })
+    )
+    const { io } = createIo()
+    const { socket, handlers } = setup('socket-a', io)
+
+    const first = handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 500 })
+    await Promise.resolve()
+    const second = handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 501 })
+    await second
+
+    resolveFirstAuth({
+      allowed: true,
+      status: 200,
+      workspaceId: 'ws-1',
+      workspacePermission: 'write',
+    })
+    await first
+
+    const acceptedClientIds = socket.emit.mock.calls
+      .filter(([event]) => event === FILE_DOC_EVENTS.JOIN_SUCCESS)
+      .map(([, payload]) => (payload as { clientId: number }).clientId)
+    expect(acceptedClientIds).toEqual(expect.arrayContaining([500, 501]))
+  })
+
   it('preserves the existing caret when a rebind to a foreign client id is rejected', async () => {
     const { io, sent } = createIo()
     const { frame: awFrame } = awarenessFrame(10, 'A')

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

@@ -783,7 +783,7 @@ async function mergeMarkdownIntoRoom(
 /**
  * Get (or lazily create) the authoritative document for a room, wiring the two
  * relay handlers exactly once: document updates and awareness changes are
- * broadcast to the room, excluding the origin socket (it already applied them).
+ * broadcast to the room.
  */
 function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
   const name = roomName(ref)
@@ -821,18 +821,12 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
     const encoder = encoding.createEncoder()
     encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC)
     syncProtocol.writeUpdate(encoder, update)
-    // Fan out to THIS task's clients only (excluding the origin socket if local — a user edit OR an
-    // agent-streamed frame). Cross-task delivery rides the shared stream — every task's tailer applies +
-    // runs its own local fan-out.
-    // A client edit excludes its own sender socket (echo suppression). An agent frame broadcasts to the
-    // WHOLE room — no socket excluded — so a same-socket sibling provider (chat preview + Files editor)
-    // stays live mid-stream; the emitting provider no-ops on its own echo.
-    broadcastLocal(
-      io,
-      name,
-      encoding.toUint8Array(encoder),
-      origin === AGENT_SYNC_ORIGIN ? null : originSocketId(origin)
-    )
+    // Fan out to every client on THIS task, including the origin socket. One shared Socket.IO connection
+    // can host multiple providers for this file; excluding the whole socket would strand the sibling
+    // provider's distinct Y.Doc. Yjs updates are idempotent, and the originating provider applies its
+    // echo with the provider as transaction origin, so it does not send the update again. Cross-task
+    // delivery rides the shared stream, where every task's tailer runs its own local fan-out.
+    broadcastLocal(io, name, encoding.toUint8Array(encoder), null)
     // Share every locally-originated update to the stream so peers converge. Skip updates that already
     // came FROM the stream (REDIS_ORIGIN / REDIS_SNAPSHOT_ORIGIN / REDIS_AGENT_ORIGIN) and SEED_ORIGIN —
     // the seed is published EXPLICITLY and AWAITED under the seed lock (so it lands before the lock
@@ -898,12 +892,18 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
 function emitJoinError(
   socket: AuthenticatedSocket,
   fileId: unknown,
+  clientId: unknown,
   error: string,
   code: string,
   retryable: boolean
 ) {
+  const normalizedClientId =
+    typeof clientId === 'number' && Number.isInteger(clientId) && clientId >= 0
+      ? clientId
+      : undefined
   socket.emit(FILE_DOC_EVENTS.JOIN_ERROR, {
     fileId: typeof fileId === 'string' ? fileId : '',
+    clientId: normalizedClientId,
     error,
     code,
     retryable,
@@ -1113,11 +1113,25 @@ export function setupWorkspaceFileDocHandlers(
       const userName = socket.userName
 
       if (!userId || !userName) {
-        emitJoinError(socket, fileId, 'Authentication required', 'AUTHENTICATION_REQUIRED', false)
+        emitJoinError(
+          socket,
+          fileId,
+          clientId,
+          'Authentication required',
+          'AUTHENTICATION_REQUIRED',
+          false
+        )
         return
       }
       if (!roomManager.isReady()) {
-        emitJoinError(socket, fileId, 'Realtime unavailable', 'ROOM_MANAGER_UNAVAILABLE', true)
+        emitJoinError(
+          socket,
+          fileId,
+          clientId,
+          'Realtime unavailable',
+          'ROOM_MANAGER_UNAVAILABLE',
+          true
+        )
         return
       }
       if (
@@ -1128,15 +1142,20 @@ export function setupWorkspaceFileDocHandlers(
         !Number.isInteger(clientId) ||
         clientId < 0
       ) {
-        emitJoinError(socket, fileId, 'Invalid join payload', 'INVALID_PAYLOAD', false)
+        emitJoinError(socket, fileId, clientId, 'Invalid join payload', 'INVALID_PAYLOAD', false)
         return
       }
 
-      // Claim this JOIN's generation before the async authorize below, and record the file the
-      // socket now intends to edit so a leave for it can cancel this join if it's still in-flight.
-      generation = (joinGeneration.get(socket.id) ?? 0) + 1
-      joinGeneration.set(socket.id, generation)
-      currentFileId = fileId
+      // A generation represents the socket's intended FILE, not an individual provider. Co-mounted
+      // providers for the same file must be allowed to join concurrently; switching files advances the
+      // generation so every in-flight join for the old file is cancelled together.
+      if (currentFileId !== fileId) {
+        generation = (joinGeneration.get(socket.id) ?? 0) + 1
+        joinGeneration.set(socket.id, generation)
+        currentFileId = fileId
+      } else {
+        generation = joinGeneration.get(socket.id) ?? 0
+      }
 
       const room = fileDocRoom(fileId)
       const name = roomName(room)
@@ -1153,7 +1172,7 @@ export function setupWorkspaceFileDocHandlers(
           accessDenied: 'Access denied to file',
         },
         emitError: ({ error, code, retryable }) =>
-          emitJoinError(socket, fileId, error, code, retryable),
+          emitJoinError(socket, fileId, clientId, error, code, retryable),
       })
       if (!authorized) return
 
@@ -1190,7 +1209,7 @@ export function setupWorkspaceFileDocHandlers(
           logger.warn(
             `User ${userId} lost write access to file ${fileId} before the join completed`
           )
-          emitJoinError(socket, fileId, 'Access denied to file', 'ACCESS_DENIED', false)
+          emitJoinError(socket, fileId, clientId, 'Access denied to file', 'ACCESS_DENIED', false)
           return
         }
 
@@ -1219,7 +1238,14 @@ export function setupWorkspaceFileDocHandlers(
           const owner = clientMap.get(clientId)
           if (owner === undefined) continue
           if (owner.userId !== userId) {
-            emitJoinError(socket, fileId, 'Client id already in use', 'CLIENT_ID_IN_USE', false)
+            emitJoinError(
+              socket,
+              fileId,
+              clientId,
+              'Client id already in use',
+              'CLIENT_ID_IN_USE',
+              false
+            )
             return
           }
           // Same user reclaiming its client id on a stale prior socket: evict just THAT clientID's
@@ -1268,7 +1294,11 @@ export function setupWorkspaceFileDocHandlers(
         // Name the document this room holds, so a client that still carries a DIFFERENT one (its room
         // outlived by a document rebuilt in its place) can refuse to merge instead of unioning two
         // documents into the file twice over. Read after readiness — before it, the room has no doc yet.
-        socket.emit(FILE_DOC_EVENTS.JOIN_SUCCESS, { fileId, docId: docIdOf(entry.doc) })
+        socket.emit(FILE_DOC_EVENTS.JOIN_SUCCESS, {
+          fileId,
+          clientId,
+          docId: docIdOf(entry.doc),
+        })
         // Server-authenticated roster → everyone in the room, including this joiner.
         broadcastFileDocPresence(io, name, entry)
 
@@ -1321,7 +1351,7 @@ export function setupWorkspaceFileDocHandlers(
         (generation !== undefined && joinGeneration.get(socket.id) !== generation)
       )
         return
-      emitJoinError(socket, fileId, 'Failed to join file document', 'JOIN_FAILED', true)
+      emitJoinError(socket, fileId, clientId, 'Failed to join file document', 'JOIN_FAILED', true)
     }
   })