feat(files): live presence avatars + live file tree [3/N]

apps/realtime/src/handlers/index.ts

@@ -4,6 +4,7 @@ import { setupPresenceHandlers } from '@/handlers/presence'
 import { setupSubblocksHandlers } from '@/handlers/subblocks'
 import { setupVariablesHandlers } from '@/handlers/variables'
 import { setupWorkflowHandlers } from '@/handlers/workflow'
+import { setupWorkspaceFilesHandlers } from '@/handlers/workspace-files'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import type { IRoomManager } from '@/rooms'
 
@@ -13,5 +14,6 @@ export function setupAllHandlers(socket: AuthenticatedSocket, roomManager: IRoom
  setupSubblocksHandlers(socket, roomManager)
  setupVariablesHandlers(socket, roomManager)
  setupPresenceHandlers(socket, roomManager)
+ setupWorkspaceFilesHandlers(socket, roomManager)
  setupConnectionHandlers(socket, roomManager)
 }

apps/realtime/src/handlers/workspace-files.test.ts

@@ -0,0 +1,169 @@
+/**
+ * @vitest-environment node
+ */
+import { ROOM_TYPES } from '@sim/realtime-protocol/rooms'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import type { IRoomManager } from '@/rooms'
+
+const { mockAuthorizeRoom } = vi.hoisted(() => ({
+ mockAuthorizeRoom: vi.fn(),
+}))
+
+vi.mock('@sim/db', () => ({
+ db: { select: vi.fn() },
+ user: { image: 'image' },
+}))
+
+vi.mock('@sim/platform-authz/rooms', () => ({
+ authorizeRoom: mockAuthorizeRoom,
+}))
+
+import { setupWorkspaceFilesHandlers } from '@/handlers/workspace-files'
+
+interface JoinPayload {
+ workspaceId: string
+ folderId?: string | null
+ tabSessionId?: string
+}
+
+function createSocket(overrides?: Record<string, unknown>) {
+ const handlers: Record<string, (payload: JoinPayload) => Promise<void> | void> = {}
+ const socket = {
+ id: 'socket-1',
+ userId: 'user-1',
+ userName: 'Test User',
+ userImage: 'avatar.png',
+ on: vi.fn((event: string, handler: (payload: JoinPayload) => Promise<void> | void) => {
+ handlers[event] = handler
+ }),
+ emit: vi.fn(),
+ join: vi.fn(),
+ leave: vi.fn(),
+ to: vi.fn().mockReturnValue({ emit: vi.fn() }),
+ ...overrides,
+ }
+ return { handlers, socket }
+}
+
+function createRoomManager(overrides?: Partial<IRoomManager>): IRoomManager {
+ return {
+ isReady: vi.fn().mockReturnValue(true),
+ getRoomForSocket: vi.fn().mockResolvedValue(null),
+ getRoomsForSocket: vi.fn().mockResolvedValue([]),
+ removeUserFromRoom: vi.fn().mockResolvedValue(false),
+ removeSocketFromAllRooms: vi.fn().mockResolvedValue([]),
+ broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined),
+ getRoomUsers: vi.fn().mockResolvedValue([]),
+ hasRoom: vi.fn().mockResolvedValue(false),
+ addUserToRoom: vi.fn().mockResolvedValue(undefined),
+ getUserSession: vi.fn().mockResolvedValue(null),
+ updateUserActivity: vi.fn().mockResolvedValue(undefined),
+ updateRoomLastModified: vi.fn().mockResolvedValue(undefined),
+ emitToRoom: vi.fn(),
+ getUniqueUserCount: vi.fn().mockResolvedValue(1),
+ getTotalActiveConnections: vi.fn().mockResolvedValue(0),
+ shutdown: vi.fn().mockResolvedValue(undefined),
+ initialize: vi.fn().mockResolvedValue(undefined),
+ io: {
+ in: vi.fn().mockReturnValue({ socketsLeave: vi.fn().mockResolvedValue(undefined) }),
+ },
+ ...overrides,
+ } as unknown as IRoomManager
+}
+
+describe('setupWorkspaceFilesHandlers', () => {
+ beforeEach(() => {
+ vi.clearAllMocks()
+ mockAuthorizeRoom.mockResolvedValue({
+ allowed: true,
+ status: 200,
+ workspaceId: 'ws-1',
+ workspacePermission: 'admin',
+ })
+ })
+
+ it('rejects join when the socket is not authenticated', async () => {
+ const { socket, handlers } = createSocket({ userId: undefined, userName: undefined })
+ setupWorkspaceFilesHandlers(
+ socket as unknown as Parameters<typeof setupWorkspaceFilesHandlers>[0],
+ createRoomManager()
+ )
+
+ await handlers['join-workspace-files']({ workspaceId: 'ws-1' })
+
+ expect(socket.emit).toHaveBeenCalledWith('join-workspace-files-error', {
+ workspaceId: 'ws-1',
+ error: 'Authentication required',
+ code: 'AUTHENTICATION_REQUIRED',
+ retryable: false,
+ })
+ })
+
+ it('rejects join with a retryable error when realtime is unavailable', async () => {
+ const { socket, handlers } = createSocket()
+ setupWorkspaceFilesHandlers(
+ socket as unknown as Parameters<typeof setupWorkspaceFilesHandlers>[0],
+ createRoomManager({ isReady: vi.fn().mockReturnValue(false) })
+ )
+
+ await handlers['join-workspace-files']({ workspaceId: 'ws-1' })
+
+ expect(socket.emit).toHaveBeenCalledWith(
+ 'join-workspace-files-error',
+ expect.objectContaining({ code: 'ROOM_MANAGER_UNAVAILABLE', retryable: true })
+ )
+ })
+
+ it('rejects join when workspace access is denied', async () => {
+ mockAuthorizeRoom.mockResolvedValue({
+ allowed: false,
+ status: 403,
+ workspaceId: 'ws-1',
+ workspacePermission: null,
+ })
+ const { socket, handlers } = createSocket()
+ setupWorkspaceFilesHandlers(
+ socket as unknown as Parameters<typeof setupWorkspaceFilesHandlers>[0],
+ createRoomManager()
+ )
+
+ await handlers['join-workspace-files']({ workspaceId: 'ws-1' })
+
+ expect(socket.emit).toHaveBeenCalledWith(
+ 'join-workspace-files-error',
+ expect.objectContaining({ code: 'ACCESS_DENIED', retryable: false })
+ )
+ })
+
+ it('joins the workspace files room and broadcasts presence on success', async () => {
+ const { socket, handlers } = createSocket()
+ const roomManager = createRoomManager({
+ getRoomUsers: vi.fn().mockResolvedValue([]),
+ })
+ setupWorkspaceFilesHandlers(
+ socket as unknown as Parameters<typeof setupWorkspaceFilesHandlers>[0],
+ roomManager
+ )
+
+ await handlers['join-workspace-files']({
+ workspaceId: 'ws-1',
+ folderId: 'folder-1',
+ tabSessionId: 'tab-1',
+ })
+
+ expect(socket.join).toHaveBeenCalledWith('workspace-files:ws-1')
+ expect(roomManager.addUserToRoom).toHaveBeenCalledWith(
+ { type: ROOM_TYPES.WORKSPACE_FILES, id: 'ws-1' },
+ 'socket-1',
+ expect.objectContaining({ userId: 'user-1', folderId: 'folder-1', role: 'admin' })
+ )
+ expect(socket.emit).toHaveBeenCalledWith(
+ 'join-workspace-files-success',
+ expect.objectContaining({ workspaceId: 'ws-1', socketId: 'socket-1' })
+ )
+ expect(roomManager.broadcastPresenceUpdate).toHaveBeenCalledWith({
+ type: ROOM_TYPES.WORKSPACE_FILES,
+ id: 'ws-1',
+ })
+ })
+})

apps/realtime/src/handlers/workspace-files.ts

@@ -0,0 +1,184 @@
+import { db, user } from '@sim/db'
+import { createLogger } from '@sim/logger'
+import { authorizeRoom } from '@sim/platform-authz/rooms'
+import { ROOM_TYPES, type RoomRef, roomName } from '@sim/realtime-protocol/rooms'
+import { eq } from 'drizzle-orm'
+import type { AuthenticatedSocket } from '@/middleware/auth'
+import type { IRoomManager, UserPresence } from '@/rooms'
+
+const logger = createLogger('WorkspaceFilesHandlers')
+
+/** The workspace-files room ref for a workspace id. */
+const filesRoom = (workspaceId: string): RoomRef => ({
+ type: ROOM_TYPES.WORKSPACE_FILES,
+ id: workspaceId,
+})
+
+interface JoinPayload {
+ workspaceId: string
+ folderId?: string | null
+ tabSessionId?: string
+}
+
+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 files presence', { userId, error })
+ return null
+ }
+}
+
+/**
+ * Presence handlers for the workspace file browser. Mirrors the workflow join
+ * flow but is workspace-scoped (room id = workspaceId) and read-only presence:
+ * there are no persisted file operations over the socket — file mutations go
+ * through the HTTP API, which fans out a `workspace-files-changed` event
+ * separately. The viewer's `folderId` is recorded at join (a future hook for
+ * folder-scoped presence); there is no cursor channel here yet.
+ */
+export function setupWorkspaceFilesHandlers(
+ socket: AuthenticatedSocket,
+ roomManager: IRoomManager
+) {
+ socket.on(
+ 'join-workspace-files',
+ async ({ workspaceId, folderId, tabSessionId }: JoinPayload) => {
+ try {
+ const userId = socket.userId
+ const userName = socket.userName
+
+ if (!userId || !userName) {
+ socket.emit('join-workspace-files-error', {
+ workspaceId,
+ error: 'Authentication required',
+ code: 'AUTHENTICATION_REQUIRED',
+ retryable: false,
+ })
+ return
+ }
+
+ if (!roomManager.isReady()) {
+ socket.emit('join-workspace-files-error', {
+ workspaceId,
+ error: 'Realtime unavailable',
+ code: 'ROOM_MANAGER_UNAVAILABLE',
+ retryable: true,
+ })
+ return
+ }
+
+ const room = filesRoom(workspaceId)
+
+ let authorized: Awaited<ReturnType<typeof authorizeRoom>>
+ try {
+ authorized = await authorizeRoom({ userId, room, action: 'read' })
+ } catch (error) {
+ logger.warn(`Error authorizing files room for ${userId}:`, error)
+ socket.emit('join-workspace-files-error', {
+ workspaceId,
+ error: 'Failed to verify workspace access',
+ code: 'VERIFY_ACCESS_FAILED',
+ retryable: true,
+ })
+ return
+ }
+
+ if (!authorized.allowed) {
+ socket.emit('join-workspace-files-error', {
+ workspaceId,
+ error: authorized.status === 404 ? 'Workspace not found' : 'Access denied to workspace',
+ code: authorized.status === 404 ? 'NOT_FOUND' : 'ACCESS_DENIED',
+ retryable: false,
+ })
+ return
+ }
+
+ // Leave a previously-joined files room if switching workspaces.
+ const currentRoom = await roomManager.getRoomForSocket(
+ socket.id,
+ ROOM_TYPES.WORKSPACE_FILES
+ )
+ if (currentRoom && currentRoom.id !== workspaceId) {
+ socket.leave(roomName(currentRoom))
+ await roomManager.removeUserFromRoom(currentRoom, socket.id)
+ await roomManager.broadcastPresenceUpdate(currentRoom)
+ }
+
+ // Clean up the same user's stale socket from the same tab (e.g. a reconnect
+ // that raced the old socket's disconnect), so presence shows one entry.
+ if (tabSessionId) {
+ const existingUsers = await roomManager.getRoomUsers(room)
+ for (const existing of existingUsers) {
+ if (
+ existing.socketId !== socket.id &&
+ existing.userId === userId &&
+ existing.tabSessionId === tabSessionId
+ ) {
+ await roomManager.removeUserFromRoom(room, existing.socketId)
+ await roomManager.io.in(existing.socketId).socketsLeave(roomName(room))
+ }
+ }
+ }
+
+ socket.join(roomName(room))
+
+ const presence: UserPresence = {
+ userId,
+ room,
+ userName,
+ socketId: socket.id,
+ tabSessionId,
+ joinedAt: Date.now(),
+ lastActivity: Date.now(),
+ role: authorized.workspacePermission ?? 'read',
+ folderId: folderId ?? null,
+ avatarUrl: await resolveAvatarUrl(socket, userId),
+ }
+
+ await roomManager.addUserToRoom(room, socket.id, presence)
+
+ const presenceUsers = await roomManager.getRoomUsers(room)
+ socket.emit('join-workspace-files-success', {
+ workspaceId,
+ socketId: socket.id,
+ presenceUsers,
+ })
+
+ await roomManager.broadcastPresenceUpdate(room)
+
+ logger.info(`User ${userId} (${userName}) joined files room for workspace ${workspaceId}`)
+ } catch (error) {
+ logger.error('Error joining workspace files room:', error)
+ socket.emit('join-workspace-files-error', {
+ workspaceId,
+ error: 'Failed to join workspace files',
+ code: 'JOIN_FAILED',
+ retryable: true,
+ })
+ }
+ }
+ )
+
+ socket.on('leave-workspace-files', async () => {
+ try {
+ if (!roomManager.isReady()) return
+ const room = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKSPACE_FILES)
+ if (!room) return
+ socket.leave(roomName(room))
+ await roomManager.removeUserFromRoom(room, socket.id)
+ await roomManager.broadcastPresenceUpdate(room)
+ } catch (error) {
+ logger.error('Error leaving workspace files room:', error)
+ }
+ })
+}

apps/realtime/src/rooms/types.ts

@@ -19,6 +19,11 @@ export interface UserPresence {
  cursor?: { x: number; y: number }
  selection?: { type: 'block' | 'edge' | 'none'; id?: string }
  avatarUrl?: string | null
+ /**
+ * The subfolder the user is viewing, recorded at join for room types that track
+ * a per-viewer location (e.g. the workspace file browser). `null` is the root.
+ */
+ folderId?: string | null
 }
 
 /**