fix(files): authorize joins before subscribing to document frames

Author: waleedlatif1

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

@@ -38,6 +38,7 @@ vi.mock('@/handlers/file-doc-app', () => ({
 import {
   applyMarkdownToLiveFileDoc,
   cleanupFileDocForSocket,
+  fileDocAdmissionRoom,
   flushAllFileDocRooms,
   invalidateLiveFileDocument,
   setupWorkspaceFileDocHandlers,
@@ -1006,6 +1007,76 @@ describe('setupWorkspaceFileDocHandlers', () => {
     }
   )
 
+  it.each(['write', 'read', null] as const)(
+    'withholds document and presence broadcasts until final authorization resolves to %s',
+    async (permission) => {
+      mockFetchFileDocSeed.mockResolvedValue(seedResult('# Private', 'doc-private'))
+      const memberships = new Set<string>()
+      let pending: ReturnType<typeof setup>
+      const { io } = createIo(({ target, event, payload }) => {
+        if (memberships.has(target)) pending.socket.emit(event, payload)
+      })
+      pending = setup('socket-pending-authorization', io, {
+        join: vi.fn((name: string) => {
+          memberships.add(name)
+        }),
+        leave: vi.fn((name: string) => {
+          memberships.delete(name)
+        }),
+      })
+      let resolvePermission!: (value: 'write' | 'read' | null) => void
+      const authorization = new Promise<'write' | 'read' | null>((resolve) => {
+        resolvePermission = resolve
+      })
+      const guard = vi
+        .spyOn(permissions, 'resolveCurrentRoomPermission')
+        .mockResolvedValueOnce('write')
+        .mockImplementationOnce(() => authorization)
+      const joining = pending.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+      try {
+        await vi.waitFor(() => expect(guard).toHaveBeenCalledTimes(2))
+        const content = frame(FILE_DOC_MESSAGE_TYPE.SYNC, (encoder) =>
+          syncProtocol.writeUpdate(encoder, seedResult('# Private update').update)
+        )
+        io.local.to(ROOM_NAME).emit(FILE_DOC_EVENTS.MESSAGE, content)
+        io.to(ROOM_NAME).emit(
+          FILE_DOC_EVENTS.MESSAGE,
+          new Uint8Array([FILE_DOC_MESSAGE_TYPE.AWARENESS])
+        )
+        io.to(ROOM_NAME).emit(FILE_DOC_EVENTS.PRESENCE, [{ userId: 'private-peer' }])
+        expect(pending.socket.emit).not.toHaveBeenCalledWith(
+          FILE_DOC_EVENTS.MESSAGE,
+          expect.anything()
+        )
+        expect(pending.socket.emit).not.toHaveBeenCalledWith(
+          FILE_DOC_EVENTS.PRESENCE,
+          expect.anything()
+        )
+        expect(memberships.has(ROOM_NAME)).toBe(false)
+        resolvePermission(permission)
+        await joining
+        expect(memberships.has(ROOM_NAME)).toBe(permission === 'write')
+        if (permission === 'write') {
+          expect(joinSuccessFileId(pending.socket)).toBe('file-1')
+          expect(pending.socket.emit).toHaveBeenCalledWith(
+            FILE_DOC_EVENTS.MESSAGE,
+            expect.any(Uint8Array)
+          )
+        } else {
+          expect(joinSuccessFileId(pending.socket)).toBeUndefined()
+          expect(pending.socket.emit).toHaveBeenCalledWith(
+            FILE_DOC_EVENTS.JOIN_ERROR,
+            expect.objectContaining({ code: 'ACCESS_DENIED', retryable: false })
+          )
+        }
+      } finally {
+        resolvePermission(permission)
+        await joining
+        guard.mockRestore()
+      }
+    }
+  )
+
   it('receives invalidation while the subscribed generation check is pending', async () => {
     mockFetchFileDocSeed.mockResolvedValue(seedResult('# Old', 'doc-old'))
     const memberships = new Set<string>()
@@ -1031,9 +1102,10 @@ describe('setupWorkspaceFileDocHandlers', () => {
     try {
       const joining = pending.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
       await vi.waitFor(() => expect(guard).toHaveBeenCalledOnce())
-      expect(memberships.has(ROOM_NAME)).toBe(true)
+      expect(memberships.has(ROOM_NAME)).toBe(false)
+      expect(memberships.has(fileDocAdmissionRoom('file-1'))).toBe(true)
       await getFileDocStore().invalidateDocument(ROOM_NAME, 2)
-      io.to(ROOM_NAME).emit(FILE_DOC_EVENTS.INVALIDATED, { fileId: 'file-1' })
+      io.to(fileDocAdmissionRoom('file-1')).emit(FILE_DOC_EVENTS.INVALIDATED, { fileId: 'file-1' })
       expect(pending.socket.emit).toHaveBeenCalledWith(FILE_DOC_EVENTS.INVALIDATED, {
         fileId: 'file-1',
       })
@@ -1043,6 +1115,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
 
       expect(joinSuccessFileId(pending.socket)).toBeUndefined()
       expect(memberships.has(ROOM_NAME)).toBe(false)
+      expect(memberships.has(fileDocAdmissionRoom('file-1'))).toBe(false)
     } finally {
       resolveGeneration(true)
       guard.mockRestore()
@@ -1104,7 +1177,8 @@ describe('setupWorkspaceFileDocHandlers', () => {
         FILE_DOC_EVENTS.JOIN_SUCCESS,
         expect.objectContaining({ clientId: 2, docId: 'doc-shared' })
       )
-      expect(pending.socket.leave).not.toHaveBeenCalled()
+      expect(pending.socket.leave).not.toHaveBeenCalledWith(ROOM_NAME)
+      expect(pending.socket.leave).toHaveBeenCalledWith(fileDocAdmissionRoom('file-1'))
     } finally {
       for (const resolve of checks) resolve(true)
       guard.mockRestore()
@@ -1121,7 +1195,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
       .mockRejectedValueOnce(new Error('Temporary generation read failure'))
     try {
       await current.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 })
-      expect(current.socket.leave).not.toHaveBeenCalled()
+      expect(current.socket.leave).not.toHaveBeenCalledWith(ROOM_NAME)
       current.socket.emit.mockClear()
       current.handlers[FILE_DOC_EVENTS.MESSAGE](
         frame(FILE_DOC_MESSAGE_TYPE.SYNC, (encoder) =>

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

@@ -227,6 +227,11 @@ const fileDocRoom = (fileId: string): RoomRef => ({
   id: fileId,
 })
 
+/** Pending admissions receive invalidations here, never document or presence frames. */
+export function fileDocAdmissionRoom(fileId: string): string {
+  return `file-doc-admission:${fileId}`
+}
+
 /**
  * A `y-protocols` transaction/awareness origin is the emitting socket id (a
  * string) when it came from a client, and something else (`null` / `'local'` /
@@ -1344,7 +1349,7 @@ export function setupWorkspaceFileDocHandlers(
   // awaiting authorization can't complete after the client left and register a ghost owner. A
   // leave for a DIFFERENT file must NOT cancel it (a document switch), mirroring workspace-files.
   let currentFileId: string | null = null
-  /** Co-mounted providers share transport membership while their admissions are still pending. */
+  /** Co-mounted providers share invalidation membership until their last admission settles. */
   const pendingMemberships = new Map<string, number>()
 
   socket.on(FILE_DOC_EVENTS.JOIN, async (payload: JoinFileDocPayload) => {
@@ -1413,6 +1418,7 @@ export function setupWorkspaceFileDocHandlers(
 
       const room = fileDocRoom(fileId)
       const name = roomName(room)
+      const admissionName = fileDocAdmissionRoom(fileId)
 
       const authorized = await resolveRoomJoinAuth({
         userId,
@@ -1485,13 +1491,12 @@ export function setupWorkspaceFileDocHandlers(
         if (!isCurrentJoin()) return
 
         /**
-         * Subscribe after hydration, before checking the generation: an invalidation either fails
-         * this check or reaches the subscribed client, including one broadcast by another replica.
-         * Count concurrent same-socket admissions so one failed attempt cannot unsubscribe another.
+         * Watch invalidations before checking the generation, including broadcasts from another
+         * replica. Pending clients must not receive document or presence frames before authorization.
          */
         pendingMemberships.set(name, (pendingMemberships.get(name) ?? 0) + 1)
         subscribed = true
-        await socket.join(name)
+        await socket.join(admissionName)
         const joinedVersion =
           Math.max(entry.syncedVersion ?? 0, (await store.getSyncedVersion(name)) ?? 0) || undefined
         const currentDocument = await store.isDocumentGenerationCurrent(name, docIdOf(entry.doc))
@@ -1514,11 +1519,10 @@ export function setupWorkspaceFileDocHandlers(
           return
         }
 
-        // Abort a JOIN superseded while the room was being prepared: the socket disconnected, a newer
-        // JOIN (a document switch) bumped the generation, or the room was dropped and re-created.
-        // Registering here would leak a dead socket's room, bind the socket to the wrong document, or
-        // attach it to a doc no longer registered. Last await before the commit, so nothing can
-        // interleave between the access re-check above and the registration below.
+        /** Commit content membership only after the final authorization decision. */
+        if (!isCurrentJoin()) return
+        await socket.join(name)
+        /** An asynchronous adapter join can be superseded by a leave, switch, or disconnect. */
         if (!isCurrentJoin()) return
 
         // A client id must be owned by at most one user, or a peer could bind an active
@@ -1631,6 +1635,7 @@ export function setupWorkspaceFileDocHandlers(
           if (remaining > 0) pendingMemberships.set(name, remaining)
           else {
             pendingMemberships.delete(name)
+            await socket.leave(admissionName)
             if (socketToRoomName.get(socket.id) !== name) await socket.leave(name)
           }
         }

apps/realtime/src/routes/http.test.ts

@@ -6,6 +6,7 @@ const { mockInvalidateDocument } = vi.hoisted(() => ({ mockInvalidateDocument: v
 
 vi.mock('@/handlers/file-doc', () => ({
   applyMarkdownToLiveFileDoc: vi.fn(),
+  fileDocAdmissionRoom: (fileId: string) => `file-doc-admission:${fileId}`,
   invalidateLiveFileDocument: mockInvalidateDocument,
 }))
 
@@ -16,7 +17,10 @@ function createMocks(req: Partial<IncomingMessage>) {
   const writeHead = vi.fn()
   const end = vi.fn()
   const logger = { info: vi.fn(), error: vi.fn(), debug: vi.fn(), warn: vi.fn() }
+  const emit = vi.fn()
+  const to = vi.fn(() => ({ emit }))
   const roomManager = {
+    io: { to },
     getTotalActiveConnections: vi.fn().mockResolvedValue(0),
     isReady: vi.fn().mockReturnValue(true),
     emitToRoom: vi.fn(),
@@ -30,6 +34,8 @@ function createMocks(req: Partial<IncomingMessage>) {
     writeHead,
     end,
     roomManager,
+    to,
+    emit,
   }
 }
 
@@ -86,31 +92,32 @@ describe('createHttpHandler', () => {
 
   it('invalidates the shared generation before notifying every open editor', async () => {
     mockInvalidateDocument.mockResolvedValueOnce({ status: 'applied', docId: 'old-document' })
-    const { handler, req, res, writeHead, roomManager } = createMocks(
+    const { handler, req, res, writeHead, to, emit } = createMocks(
       requestWithBody('/api/file-doc/invalidate', { fileId: 'file-1', version: 100 })
     )
 
     await handler(req, res)
 
     expect(mockInvalidateDocument).toHaveBeenCalledWith('file-1', 100)
-    expect(roomManager.emitToRoom).toHaveBeenCalledWith(
-      { type: 'workspace-file-doc', id: 'file-1' },
+    expect(to).toHaveBeenCalledWith(['workspace-file-doc:file-1', 'file-doc-admission:file-1'])
+    expect(emit).toHaveBeenCalledWith(
       'file-doc-invalidated',
       expect.objectContaining({ fileId: 'file-1', version: 100, docId: 'old-document' })
     )
     expect(mockInvalidateDocument.mock.invocationCallOrder[0]).toBeLessThan(
-      vi.mocked(roomManager.emitToRoom).mock.invocationCallOrder[0]
+      emit.mock.invocationCallOrder[0]
     )
     expect(writeHead).toHaveBeenCalledWith(200, { 'Content-Type': 'application/json' })
   })
 
   it('does not evict editors for a superseded invalidation', async () => {
     mockInvalidateDocument.mockResolvedValueOnce({ status: 'stale' })
-    const { handler, req, res, end, roomManager } = createMocks(
+    const { handler, req, res, end, roomManager, to } = createMocks(
       requestWithBody('/api/file-doc/invalidate', { fileId: 'file-1', version: 100 })
     )
     await handler(req, res)
     expect(roomManager.emitToRoom).not.toHaveBeenCalled()
+    expect(to).not.toHaveBeenCalled()
     expect(end).toHaveBeenCalledWith(JSON.stringify({ status: 'stale' }))
   })
 

apps/realtime/src/routes/http.ts

@@ -1,9 +1,13 @@
 import type { IncomingMessage, ServerResponse } from 'http'
 import { FILE_DOC_EVENTS, type FileDocInvalidated } from '@sim/realtime-protocol/file-doc'
-import { ROOM_TYPES, WORKSPACE_LIST_ROOM_TYPES } from '@sim/realtime-protocol/rooms'
+import { ROOM_TYPES, roomName, WORKSPACE_LIST_ROOM_TYPES } from '@sim/realtime-protocol/rooms'
 import { safeCompare } from '@sim/security/compare'
 import { env } from '@/env'
-import { applyMarkdownToLiveFileDoc, invalidateLiveFileDocument } from '@/handlers/file-doc'
+import {
+  applyMarkdownToLiveFileDoc,
+  fileDocAdmissionRoom,
+  invalidateLiveFileDocument,
+} from '@/handlers/file-doc'
 import { type IRoomManager, WorkflowRoomService } from '@/rooms'
 
 interface Logger {
@@ -233,7 +237,9 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
           message: 'This file changed outside the editor. Reload to continue editing.',
         }
         if (result.status === 'applied')
-          roomManager.emitToRoom(room, FILE_DOC_EVENTS.INVALIDATED, payload)
+          roomManager.io
+            .to([roomName(room), fileDocAdmissionRoom(fileId)])
+            .emit(FILE_DOC_EVENTS.INVALIDATED, payload)
         res.writeHead(200, { 'Content-Type': 'application/json' })
         res.end(JSON.stringify({ status: result.status }))
       } catch (error) {