feat(realtime): shared room spine + live Files/Tables collaboration + Yjs document editing

apps/realtime/package.json

@@ -33,9 +33,12 @@
     "@sim/workflow-types": "workspace:*",
     "@socket.io/redis-adapter": "8.3.0",
     "drizzle-orm": "^0.45.2",
+    "lib0": "0.2.117",
     "postgres": "^3.4.5",
     "redis": "5.10.0",
     "socket.io": "^4.8.1",
+    "y-protocols": "1.0.7",
+    "yjs": "13.6.31",
     "zod": "4.3.6"
   },
   "devDependencies": {

apps/realtime/src/access-revalidation.test.ts

@@ -50,16 +50,14 @@ function makeManager(sockets: FakeSocket[], presence: Partial<UserPresence>[] =
   const manager = {
     io: { sockets: { sockets: socketMap } },
     isReady: () => true,
-    getWorkflowUsers: vi.fn().mockResolvedValue(presence),
-    getWorkflowIdForSocket: vi.fn().mockResolvedValue(null),
-    removeUserFromRoom: vi
-      .fn()
-      .mockImplementation(async (_socketId: string, workflowId?: string) => workflowId ?? null),
+    getRoomUsers: vi.fn().mockResolvedValue(presence),
+    getRoomForSocket: vi.fn().mockResolvedValue(null),
+    removeUserFromRoom: vi.fn().mockResolvedValue(true),
     broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined),
   }
   return manager as unknown as IRoomManager & {
-    getWorkflowUsers: ReturnType<typeof vi.fn>
-    getWorkflowIdForSocket: ReturnType<typeof vi.fn>
+    getRoomUsers: ReturnType<typeof vi.fn>
+    getRoomForSocket: ReturnType<typeof vi.fn>
     removeUserFromRoom: ReturnType<typeof vi.fn>
     broadcastPresenceUpdate: ReturnType<typeof vi.fn>
   }
@@ -84,8 +82,11 @@ describe('access-revalidation sweep', () => {
       expect.objectContaining({ workflowId: 'wf-1' })
     )
     expect(socket.leave).toHaveBeenCalledWith('wf-1')
-    expect(manager.removeUserFromRoom).toHaveBeenCalledWith('sock-1', 'wf-1')
-    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith('wf-1')
+    expect(manager.removeUserFromRoom).toHaveBeenCalledWith(
+      { type: 'workflow', id: 'wf-1' },
+      'sock-1'
+    )
+    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith({ type: 'workflow', id: 'wf-1' })
   })
 
   it('keeps a socket whose access is still valid', async () => {
@@ -141,7 +142,30 @@ describe('access-revalidation sweep', () => {
 
     expect(mockResolveRole).toHaveBeenCalledWith('user-1', 'wf-1', 'read')
     // The security scan must stay Redis-free — presence is never consulted.
-    expect(manager.getWorkflowUsers).not.toHaveBeenCalled()
+    expect(manager.getRoomUsers).not.toHaveBeenCalled()
+  })
+
+  it('never evicts a socket joined only to a non-workflow room (files/tables/file-doc)', async () => {
+    // The sweep shares one io with the files/tables/file-doc handlers. Those rooms are
+    // namespaced (`workspace-files:ws-1`, `table:t-1`), so treating every socket.rooms
+    // entry as a workflow id would resolve a bogus permission → null → evict the socket
+    // from its files/table room every pass. Non-workflow rooms must be filtered out.
+    const filesSocket = makeSocket('sock-1', 'user-1', 'workspace-files:ws-1')
+    const tableSocket = makeSocket('sock-2', 'user-2', 'table:t-1')
+    const manager = makeManager([filesSocket, tableSocket])
+    // Even if the role resolver would say "no access", these must never be swept.
+    mockResolveRole.mockResolvedValue(null)
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(mockResolveRole).not.toHaveBeenCalled()
+    expect(filesSocket.leave).not.toHaveBeenCalled()
+    expect(filesSocket.emit).not.toHaveBeenCalled()
+    expect(tableSocket.leave).not.toHaveBeenCalled()
+    expect(tableSocket.emit).not.toHaveBeenCalled()
+    expect(manager.removeUserFromRoom).not.toHaveBeenCalled()
   })
 
   it('evicts only the revoked socket, not co-members of the room', async () => {
@@ -199,37 +223,37 @@ describe('access-revalidation sweep', () => {
     sweep.stop()
 
     expect(manager.removeUserFromRoom).toHaveBeenCalledTimes(2)
-    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith('wf-1')
+    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith({ type: 'workflow', id: 'wf-1' })
   })
 
-  it('defers cleanup when removal fails with expired socket mappings', async () => {
+  it('drops eviction cleanup when the socket is no longer mapped to the room (no infinite retry)', async () => {
     const socket = makeSocket('sock-1', 'user-1', 'wf-1')
     const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
-    // Mapping keys already expired (lookup resolves null) AND the removal fails
-    // (the Redis manager swallows the transport error into null) — the failed
-    // removal must still defer instead of reading as success.
-    manager.removeUserFromRoom.mockResolvedValueOnce(null)
+    // A healthy lookup shows the socket is no longer mapped to any workflow room (its presence
+    // is already gone), and removeUserFromRoom reports a no-op `false`. This is "already clean",
+    // not a deferrable failure — the cleanup must drop it, never re-enqueue a still-connected
+    // socket forever. (A genuine failure — still mapped + false — is covered by the next test.)
+    manager.getRoomForSocket.mockResolvedValue(null)
+    manager.removeUserFromRoom.mockResolvedValue(false)
     mockResolveRole.mockResolvedValue(null)
 
     const sweep = startAccessRevalidationSweep(manager)
     await sweep.runOnce()
-
-    expect(manager.broadcastPresenceUpdate).not.toHaveBeenCalled()
-
     await sweep.runOnce()
     sweep.stop()
 
-    expect(manager.removeUserFromRoom).toHaveBeenCalledTimes(2)
-    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith('wf-1')
+    // Attempted once, then dropped — not re-enqueued across passes, and no broadcast.
+    expect(manager.removeUserFromRoom).toHaveBeenCalledTimes(1)
+    expect(manager.broadcastPresenceUpdate).not.toHaveBeenCalled()
   })
 
   it('defers cleanup when the manager swallows a removal failure into null', async () => {
     const socket = makeSocket('sock-1', 'user-1', 'wf-1')
     const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
     // Live mapping but the removal reports nothing removed — the Redis manager
     // swallows transport errors into null, so this is the only failure signal.
-    manager.getWorkflowIdForSocket.mockResolvedValue('wf-1')
-    manager.removeUserFromRoom.mockResolvedValueOnce(null)
+    manager.getRoomForSocket.mockResolvedValue({ type: 'workflow', id: 'wf-1' })
+    manager.removeUserFromRoom.mockResolvedValueOnce(false)
     mockResolveRole.mockResolvedValue(null)
 
     const sweep = startAccessRevalidationSweep(manager)
@@ -243,15 +267,15 @@ describe('access-revalidation sweep', () => {
     sweep.stop()
 
     expect(manager.removeUserFromRoom).toHaveBeenCalledTimes(2)
-    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith('wf-1')
+    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith({ type: 'workflow', id: 'wf-1' })
   })
 
   it('skips removal when the socket has since moved to a different workflow', async () => {
     const socket = makeSocket('sock-1', 'user-1', 'wf-1')
     const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
     // Between the membership snapshot and cleanup, the socket switched to a
     // workflow it can still access — removal must not touch its new presence.
-    manager.getWorkflowIdForSocket.mockResolvedValue('wf-2')
+    manager.getRoomForSocket.mockResolvedValue({ type: 'workflow', id: 'wf-2' })
     mockResolveRole.mockResolvedValue(null)
 
     const sweep = startAccessRevalidationSweep(manager)
@@ -357,7 +381,7 @@ describe('access-revalidation sweep', () => {
       const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
       // A Redis outage where commands hang in the offline queue instead of
       // failing: the cleanup lane stalls, but scans must keep running.
-      manager.getWorkflowIdForSocket.mockReturnValue(new Promise(() => {}))
+      manager.getRoomForSocket.mockReturnValue(new Promise(() => {}))
       mockResolveRole.mockResolvedValue(null)
 
       const sweep = startAccessRevalidationSweep(manager)

apps/realtime/src/access-revalidation.ts

@@ -1,9 +1,10 @@
 import { createLogger } from '@sim/logger'
 import type { AccessRevokedBroadcast } from '@sim/realtime-protocol/events'
+import { parseRoomName, ROOM_TYPES } from '@sim/realtime-protocol/rooms'
 import { sleep } from '@sim/utils/helpers'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import { ROLE_REVALIDATION_TTL_MS, resolveCurrentWorkflowRole } from '@/middleware/permissions'
-import type { IRoomManager } from '@/rooms'
+import { type IRoomManager, workflowRoom as wf } from '@/rooms'
 
 const logger = createLogger('AccessRevalidation')
 
@@ -65,9 +66,14 @@ interface ScanTarget {
  * Collects this pod's authenticated sockets with the workflow room each has
  * joined, in stable socket order.
  *
- * The workflow room is derived from the socket's own `rooms` set (pod-local, no
- * Redis round-trips): a socket joins exactly one workflow room, so its rooms are
- * `{ ownSocketId, workflowId }`. Only local sockets are evaluated — sockets are
+ * Rooms are derived from the socket's own `rooms` set (pod-local, no Redis
+ * round-trips). A socket may occupy several rooms of different types at once
+ * (workflow canvas, workspace-files browser, table, file-doc), all on the same
+ * io — so each name is decoded with {@link parseRoomName} and only **workflow**
+ * rooms are swept here. Non-workflow room names are namespaced (`type:id`) and
+ * resolve to a non-workflow type; sweeping them as workflow ids would resolve a
+ * bogus permission, come back `null`, and spuriously evict the socket from its
+ * files/table room every pass. Only local sockets are evaluated — sockets are
  * sticky to a pod, so every socket is swept by exactly one pod using that pod's
  * warm role cache (mirroring the per-pod reasoning of the write-path cache).
  */
@@ -78,7 +84,9 @@ function collectScanTargets(io: IRoomManager['io']): ScanTarget[] {
     if (!authed.userId) continue
     for (const room of socket.rooms) {
       if (room === socket.id) continue
-      targets.push({ workflowId: room, socket: authed, userId: authed.userId })
+      const ref = parseRoomName(room)
+      if (ref?.type !== ROOM_TYPES.WORKFLOW) continue
+      targets.push({ workflowId: ref.id, socket: authed, userId: authed.userId })
     }
   }
   return targets
@@ -129,9 +137,20 @@ export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessR
   async function cleanupEvictedSocket(socketId: string, workflowId: string): Promise<void> {
     const key = `${socketId}:${workflowId}`
     try {
+      // A fully-disconnected socket already had its presence removed by the
+      // disconnect handler (removeSocketFromAllRooms), so there is nothing left to
+      // clean. Dropping here also keeps the boolean removeUserFromRoom below from
+      // reporting a false "not a member" for an already-gone entry and retrying it
+      // forever (the pre-generalization manager returned the target on a no-op).
+      if (!io.sockets.sockets.get(socketId)) {
+        pendingCleanups.delete(key)
+        return
+      }
+
       // Unlike removeUserFromRoom, this read does not swallow transport errors,
       // so a Redis outage lands in the catch below and defers the cleanup.
-      const currentWorkflowId = await roomManager.getWorkflowIdForSocket(socketId)
+      const currentRoom = await roomManager.getRoomForSocket(socketId, ROOM_TYPES.WORKFLOW)
+      const currentWorkflowId = currentRoom?.id ?? null
       if (currentWorkflowId !== null && currentWorkflowId !== workflowId) {
         // The socket has since moved to a different workflow it can still
         // access; that join's room switch already removed this room's presence
@@ -148,16 +167,26 @@ export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessR
         return
       }
 
-      const removed = await roomManager.removeUserFromRoom(socketId, workflowId)
-      if (removed === null) {
-        // The sweep always passes the target room, and both managers report a
-        // performed removal by returning it — the Redis manager swallows
-        // transport errors into null, so null means the removal did not happen
-        // (even when the socket's mapping keys have already expired).
-        throw new Error('room-state removal not confirmed')
+      // A null mapping here is the normal case (the socket's mapping key may have
+      // expired) and does NOT mean "skip" — the eviction still removes the presence
+      // entry from the known target room via the explicit ref below.
+      const removed = await roomManager.removeUserFromRoom(wf(workflowId), socketId)
+      if (!removed) {
+        // `false` conflates two outcomes: the entry was already gone (a no-op), or a
+        // transport error the manager swallowed. Only retry when the socket is still mapped
+        // to THIS room — then a false result is a genuine, deferrable failure. When a healthy
+        // getRoomForSocket above returned no workflow mapping (`currentWorkflowId === null`),
+        // the presence entry is already gone, so the cleanup is complete: dropping it avoids
+        // re-enqueuing a still-connected socket forever. (A real Redis outage throws at
+        // getRoomForSocket and is deferred by the outer catch, never reaching here.)
+        if (currentWorkflowId === workflowId) {
+          throw new Error('room-state removal not confirmed')
+        }
+        pendingCleanups.delete(key)
+        return
       }
 
-      await roomManager.broadcastPresenceUpdate(workflowId)
+      await roomManager.broadcastPresenceUpdate(wf(workflowId))
       pendingCleanups.delete(key)
     } catch (error) {
       pendingCleanups.set(key, { socketId, workflowId })

apps/realtime/src/handlers/avatar.ts

@@ -0,0 +1,29 @@
+import { db, user } from '@sim/db'
+import { createLogger } from '@sim/logger'
+import { eq } from 'drizzle-orm'
+import type { AuthenticatedSocket } from '@/middleware/auth'
+
+const logger = createLogger('PresenceAvatar')
+
+/**
+ * The avatar URL for a presence entry: the socket's authenticated image when
+ * present, otherwise a single lookup of the user's stored image. Never throws —
+ * presence must not fail on an avatar lookup, so a DB error resolves to `null`.
+ */
+export async function resolveAvatarUrl(
+  socket: AuthenticatedSocket,
+  userId: string
+): Promise<string | null> {
+  if (socket.userImage) return socket.userImage
+  try {
+    const [record] = await db
+      .select({ image: user.image })
+      .from(user)
+      .where(eq(user.id, userId))
+      .limit(1)
+    return record?.image ?? null
+  } catch (error) {
+    logger.warn('Failed to load user avatar for presence', { userId, error })
+    return null
+  }
+}