fix(files): preserve collaboration through reconnects and peer edits

Author: waleedlatif1

apps/docs/openapi-v2-files-audit.json

@@ -3167,7 +3167,7 @@
           "uploadedByEmail": {
             "type": "string",
             "format": "email",
-            "pattern": "^(?!\\.)(?!.*\\.\\.)([A-Za-z0-9_'+\\-\\.]*)[A-Za-z0-9_+-]@([A-Za-z0-9][A-Za-z0-9\\-]*\\.)+[A-Za-z]{2,}$",
+            "pattern": "^[a-zA-Z0-9.!#$%&'*+/=?^_`{|}~-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$",
             "description": "Current email address of the uploader.",
             "examples": ["jane@example.com"]
           },
@@ -4029,7 +4029,7 @@
           "uploadedByEmail": {
             "type": "string",
             "format": "email",
-            "pattern": "^(?!\\.)(?!.*\\.\\.)([A-Za-z0-9_'+\\-\\.]*)[A-Za-z0-9_+-]@([A-Za-z0-9][A-Za-z0-9\\-]*\\.)+[A-Za-z]{2,}$",
+            "pattern": "^[a-zA-Z0-9.!#$%&'*+/=?^_`{|}~-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$",
             "description": "Current email address of the uploader.",
             "examples": ["jane@example.com"]
           },

apps/realtime/src/handlers/connection.test.ts

@@ -0,0 +1,73 @@
+/**
+ * @vitest-environment node
+ */
+import { createServer, type Server as HttpServer } from 'node:http'
+import type { AddressInfo } from 'node:net'
+import { Server } from 'socket.io'
+import { io as connect, type Socket } from 'socket.io-client'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { setupConnectionHandlers, waitForConnectionCleanup } from '@/handlers/connection'
+import type { AuthenticatedSocket } from '@/middleware/auth'
+import { MemoryRoomManager } from '@/rooms'
+
+vi.mock('@/handlers/file-doc', () => ({ cleanupFileDocForSocket: vi.fn() }))
+vi.mock('@/handlers/subblocks', () => ({ cleanupPendingSubblocksForSocket: vi.fn() }))
+vi.mock('@/handlers/variables', () => ({ cleanupPendingVariablesForSocket: vi.fn() }))
+
+describe('server shutdown connection drain', () => {
+  let httpServer: HttpServer
+  let io: Server
+  let manager: MemoryRoomManager
+  let client: Socket
+
+  beforeEach(async () => {
+    httpServer = createServer()
+    io = new Server(httpServer, { transports: ['websocket'] })
+    manager = new MemoryRoomManager(io)
+    await manager.initialize()
+    io.on('connection', (socket) => setupConnectionHandlers(socket as AuthenticatedSocket, manager))
+    await new Promise<void>((resolve) => httpServer.listen(0, '127.0.0.1', resolve))
+    const port = (httpServer.address() as AddressInfo).port
+    client = connect(`http://127.0.0.1:${port}`, { transports: ['websocket'], autoConnect: false })
+    const connected = new Promise<void>((resolve) => client.once('connect', resolve))
+    client.connect()
+    await connected
+  })
+
+  afterEach(async () => {
+    client.disconnect()
+    await io.close()
+    await waitForConnectionCleanup()
+    await manager.shutdown()
+    vi.restoreAllMocks()
+  })
+
+  it('keeps automatic reconnection active after transport shutdown', async () => {
+    const disconnected = new Promise<string>((resolve) => client.once('disconnect', resolve))
+    await io.close()
+    expect(await disconnected).toBe('transport close')
+    expect(client.active).toBe(true)
+    await waitForConnectionCleanup()
+  })
+
+  it('waits for asynchronous presence cleanup before releasing its dependencies', async () => {
+    let finishRemoval: (() => void) | undefined
+    vi.spyOn(manager, 'removeSocketFromAllRooms').mockImplementation(
+      () =>
+        new Promise((resolve) => {
+          finishRemoval = () => resolve([])
+        })
+    )
+    await io.close()
+    let drained = false
+    const drain = waitForConnectionCleanup().then(() => {
+      drained = true
+    })
+    await Promise.resolve()
+    expect(drained).toBe(false)
+    expect(finishRemoval).toBeDefined()
+    finishRemoval?.()
+    await drain
+    expect(drained).toBe(true)
+  })
+})

apps/realtime/src/handlers/connection.ts

@@ -16,6 +16,13 @@ const logger = createLogger('ConnectionHandlers')
  */
 const PRESENCE_BEARING_TYPES = new Set<RoomRef['type']>([ROOM_TYPES.WORKFLOW, ROOM_TYPES.TABLE])
 
+const pendingDisconnects = new Set<Promise<void>>()
+
+/** Keep Redis available until disconnect listeners finish removing presence. */
+export async function waitForConnectionCleanup(): Promise<void> {
+  await Promise.all(pendingDisconnects)
+}
+
 export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
   socket.on('error', (error) => {
     logger.error(`Socket ${socket.id} error:`, error)
@@ -28,7 +35,7 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager
   // `disconnecting` (not `disconnect`): here `socket.rooms` is still populated and
   // authoritative, so presence is cleaned up even if the Redis room-set key was
   // evicted or TTL-expired (which would leave the manager's stored rooms empty).
-  socket.on('disconnecting', async (reason) => {
+  const handleDisconnect = async (reason: string) => {
     try {
       // Snapshot the live Socket.IO room membership SYNCHRONOUSLY, before any
       // await: Socket.IO clears `socket.rooms` via leaveAll() as soon as the
@@ -91,5 +98,11 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager
     } catch (error) {
       logger.error(`Error handling disconnect for socket ${socket.id}:`, error)
     }
+  }
+
+  socket.on('disconnecting', (reason) => {
+    const cleanup = handleDisconnect(reason)
+    pendingDisconnects.add(cleanup)
+    void cleanup.finally(() => pendingDisconnects.delete(cleanup))
   })
 }