From ed4eab3875e400515ebe8bbacc25c59ebb391cc3 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 11:16:27 -0700 Subject: [PATCH 1/5] refactor(realtime): generalize presence server to multi-room (RoomRef) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Generalizes the Socket.IO presence layer from single-workflow-room-per-socket to a domain-neutral, multi-room-per-socket model keyed by RoomRef, so a second domain (workspace files, next PR) can reuse the same membership + presence engine. Behavior-preserving for workflow collaboration. IRoomManager is now domain-neutral (addUserToRoom/removeUserFromRoom/ getRoomForSocket/getRoomUsers/updateUserActivity/... all take a RoomRef). The workflow lifecycle broadcasts (deletion/revert/update/deploy) move out of the manager into WorkflowRoomService, composed over the generic manager. Backward-compat by design (no workflow migration, no regression): - Workflow Socket.IO room name stays the bare workflowId (roomName() maps workflow -> bare id), so the ~40 io.to(workflowId) callsites are untouched. - Workflow Redis presence keys stay workflow:{id}:users/:meta (the type prefix IS "workflow"). Multi-room correctness (from adversarial audit): - socket:{id}:workflow single-value key -> socket:{id}:rooms HASH (type->id). - The SHARED socket:{id}:session key is deleted only when the socket leaves its LAST room (refcount via HLEN) — a leave from one room no longer breaks the other room's handlers. - disconnect enumerates the socket's stored rooms and rebroadcasts presence per room, instead of picking an arbitrary socket.rooms entry. - presence broadcasts use a per-room-type event name (workflow keeps the bare presence-update; others are namespaced). Workflow handlers wrap manager calls with a shared workflowRoom(id) helper; UserPresence.workflowId -> room (the client never reads that field). Tests: existing 112 realtime tests pass unchanged (behavior gate) + 7 new multi-room tests (refcounted session, presence isolation, multi-room disconnect, per-type event names). tsc clean, boundaries + prune (14/25) green. --- apps/realtime/src/handlers/connection.ts | 17 +- apps/realtime/src/handlers/operations.ts | 41 +- apps/realtime/src/handlers/presence.ts | 21 +- apps/realtime/src/handlers/subblocks.ts | 12 +- apps/realtime/src/handlers/variables.ts | 12 +- apps/realtime/src/handlers/workflow.test.ts | 19 +- apps/realtime/src/handlers/workflow.ts | 45 ++- apps/realtime/src/index.test.ts | 34 +- apps/realtime/src/rooms/index.ts | 3 +- .../realtime/src/rooms/memory-manager.test.ts | 130 ++++++ apps/realtime/src/rooms/memory-manager.ts | 277 +++++-------- apps/realtime/src/rooms/redis-manager.ts | 375 +++++++----------- apps/realtime/src/rooms/types.ts | 136 +++---- .../src/rooms/workflow-room-service.ts | 110 +++++ apps/realtime/src/routes/http.ts | 12 +- packages/realtime-protocol/src/rooms.ts | 10 + 16 files changed, 668 insertions(+), 586 deletions(-) create mode 100644 apps/realtime/src/rooms/memory-manager.test.ts create mode 100644 apps/realtime/src/rooms/workflow-room-service.ts diff --git a/apps/realtime/src/handlers/connection.ts b/apps/realtime/src/handlers/connection.ts index 90eddb82464..c59f3d74ee7 100644 --- a/apps/realtime/src/handlers/connection.ts +++ b/apps/realtime/src/handlers/connection.ts @@ -21,14 +21,17 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager cleanupPendingSubblocksForSocket(socket.id) cleanupPendingVariablesForSocket(socket.id) - const workflowIdHint = [...socket.rooms].find((roomId) => roomId !== socket.id) - const workflowId = await roomManager.removeUserFromRoom(socket.id, workflowIdHint) + // A socket may occupy multiple rooms (one per type). Remove it from all of + // them and rebroadcast presence per room so no room leaks a stale entry. + const removedRooms = await roomManager.removeSocketFromAllRooms(socket.id) - if (workflowId) { - await roomManager.broadcastPresenceUpdate(workflowId) - logger.info( - `Socket ${socket.id} disconnected from workflow ${workflowId} (reason: ${reason})` - ) + for (const room of removedRooms) { + await roomManager.broadcastPresenceUpdate(room) + } + + if (removedRooms.length > 0) { + const rooms = removedRooms.map((room) => `${room.type}:${room.id}`).join(', ') + logger.info(`Socket ${socket.id} disconnected from [${rooms}] (reason: ${reason})`) } } catch (error) { logger.error(`Error handling disconnect for socket ${socket.id}:`, error) diff --git a/apps/realtime/src/handlers/operations.ts b/apps/realtime/src/handlers/operations.ts index eef51847718..d3d603887db 100644 --- a/apps/realtime/src/handlers/operations.ts +++ b/apps/realtime/src/handlers/operations.ts @@ -9,6 +9,7 @@ import { type VariableOperation, WORKFLOW_OPERATIONS, } from '@sim/realtime-protocol/constants' +import { ROOM_TYPES } from '@sim/realtime-protocol/rooms' import { WorkflowOperationSchema } from '@sim/realtime-protocol/schemas' import { getErrorMessage } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' @@ -16,7 +17,7 @@ import { ZodError } from 'zod' import { persistWorkflowOperation } from '@/database/operations' import type { AuthenticatedSocket } from '@/middleware/auth' import { checkWorkflowOperationPermission } from '@/middleware/permissions' -import type { IRoomManager, UserSession } from '@/rooms' +import { type IRoomManager, type UserSession, workflowRoom as wf } from '@/rooms' const logger = createLogger('OperationsHandlers') @@ -44,7 +45,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager let session: UserSession | null = null try { - workflowId = await roomManager.getWorkflowIdForSocket(socket.id) + workflowId = (await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKFLOW))?.id ?? null session = await roomManager.getUserSession(socket.id) } catch (error) { logger.error('Error loading session for workflow operation:', error) @@ -65,7 +66,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager let hasRoom = false try { - hasRoom = await roomManager.hasWorkflowRoom(workflowId) + hasRoom = await roomManager.hasRoom(wf(workflowId)) } catch (error) { logger.error('Error checking workflow room:', error) emitOperationError( @@ -98,14 +99,16 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager const operationTimestamp = isPositionUpdate ? timestamp : Date.now() // Get user presence for permission checking - const users = await roomManager.getWorkflowUsers(workflowId) + const users = await roomManager.getRoomUsers(wf(workflowId)) const userPresence = users.find((u) => u.socketId === socket.id) // Skip permission checks for non-committed position updates (broadcasts only, no persistence) if (isPositionUpdate && !commitPositionUpdate) { // Update last activity if (userPresence) { - await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() }) + await roomManager.updateUserActivity(wf(workflowId), socket.id, { + lastActivity: Date.now(), + }) } } else { // Check permissions from cached role for all other operations @@ -123,7 +126,9 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager return } - await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() }) + await roomManager.updateUserActivity(wf(workflowId), socket.id, { + lastActivity: Date.now(), + }) // Re-validate the workspace role against the DB (cached per pod for a short // window) so revoked or downgraded collaborators lose write access live. @@ -198,7 +203,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager timestamp: operationTimestamp, userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) if (operationId) { socket.emit('operation-confirmed', { @@ -244,7 +249,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager timestamp: operationTimestamp, userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) if (operationId) { socket.emit('operation-confirmed', { operationId, serverTimestamp: Date.now() }) @@ -277,7 +282,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) const broadcastData = { operation, @@ -317,7 +322,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) const broadcastData = { operation, @@ -354,7 +359,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) socket.to(workflowId).emit('workflow-operation', { operation, @@ -386,7 +391,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) socket.to(workflowId).emit('workflow-operation', { operation, @@ -415,7 +420,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) socket.to(workflowId).emit('workflow-operation', { operation, @@ -447,7 +452,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) socket.to(workflowId).emit('workflow-operation', { operation, @@ -479,7 +484,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) socket.to(workflowId).emit('workflow-operation', { operation, @@ -511,7 +516,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) socket.to(workflowId).emit('workflow-operation', { operation, @@ -540,7 +545,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) socket.to(workflowId).emit('workflow-operation', { operation, @@ -569,7 +574,7 @@ export function setupOperationsHandlers(socket: AuthenticatedSocket, roomManager userId: session.userId, }) - await roomManager.updateRoomLastModified(workflowId) + await roomManager.updateRoomLastModified(wf(workflowId)) const broadcastData = { operation, diff --git a/apps/realtime/src/handlers/presence.ts b/apps/realtime/src/handlers/presence.ts index 13aadc22f34..78b53176e2f 100644 --- a/apps/realtime/src/handlers/presence.ts +++ b/apps/realtime/src/handlers/presence.ts @@ -1,4 +1,5 @@ import { createLogger } from '@sim/logger' +import { ROOM_TYPES } from '@sim/realtime-protocol/rooms' import type { AuthenticatedSocket } from '@/middleware/auth' import type { IRoomManager } from '@/rooms' @@ -7,16 +8,16 @@ const logger = createLogger('PresenceHandlers') export function setupPresenceHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) { socket.on('cursor-update', async ({ cursor }) => { try { - const workflowId = await roomManager.getWorkflowIdForSocket(socket.id) + const room = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKFLOW) const session = await roomManager.getUserSession(socket.id) - if (!workflowId || !session) return + if (!room || !session) return // Update cursor in room state - await roomManager.updateUserActivity(workflowId, socket.id, { cursor }) + await roomManager.updateUserActivity(room, socket.id, { cursor }) - // Broadcast to other users in the room - socket.to(workflowId).emit('cursor-update', { + // Broadcast to other users in the room (workflow room name is the bare id) + socket.to(room.id).emit('cursor-update', { socketId: socket.id, userId: session.userId, userName: session.userName, @@ -30,16 +31,16 @@ export function setupPresenceHandlers(socket: AuthenticatedSocket, roomManager: socket.on('selection-update', async ({ selection }) => { try { - const workflowId = await roomManager.getWorkflowIdForSocket(socket.id) + const room = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKFLOW) const session = await roomManager.getUserSession(socket.id) - if (!workflowId || !session) return + if (!room || !session) return // Update selection in room state - await roomManager.updateUserActivity(workflowId, socket.id, { selection }) + await roomManager.updateUserActivity(room, socket.id, { selection }) - // Broadcast to other users in the room - socket.to(workflowId).emit('selection-update', { + // Broadcast to other users in the room (workflow room name is the bare id) + socket.to(room.id).emit('selection-update', { socketId: socket.id, userId: session.userId, userName: session.userName, diff --git a/apps/realtime/src/handlers/subblocks.ts b/apps/realtime/src/handlers/subblocks.ts index b2f94b6fb98..93e7411d148 100644 --- a/apps/realtime/src/handlers/subblocks.ts +++ b/apps/realtime/src/handlers/subblocks.ts @@ -3,12 +3,13 @@ import { workflow, workflowBlocks } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { assertWorkflowMutable, WorkflowLockedError } from '@sim/platform-authz/workflow' import { SUBBLOCK_OPERATIONS } from '@sim/realtime-protocol/constants' +import { ROOM_TYPES } from '@sim/realtime-protocol/rooms' import { getErrorMessage } from '@sim/utils/errors' import { isWorkflowBlockProtected } from '@sim/workflow-types/workflow' import { and, eq } from 'drizzle-orm' import type { AuthenticatedSocket } from '@/middleware/auth' import { checkWorkflowOperationPermission } from '@/middleware/permissions' -import type { IRoomManager } from '@/rooms' +import { type IRoomManager, workflowRoom as wf } from '@/rooms' const logger = createLogger('SubblocksHandlers') @@ -69,7 +70,8 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager: } try { - const sessionWorkflowId = await roomManager.getWorkflowIdForSocket(socket.id) + const sessionWorkflowId = + (await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKFLOW))?.id ?? null const session = await roomManager.getUserSession(socket.id) if (!sessionWorkflowId || !session) { @@ -106,7 +108,7 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager: return } - const hasRoom = await roomManager.hasWorkflowRoom(workflowId) + const hasRoom = await roomManager.hasRoom(wf(workflowId)) if (!hasRoom) { logger.debug(`Ignoring subblock update: workflow room not found`, { socketId: socket.id, @@ -117,7 +119,7 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager: return } - const users = await roomManager.getWorkflowUsers(workflowId) + const users = await roomManager.getRoomUsers(wf(workflowId)) const userPresence = users.find((user) => user.socketId === socket.id) if (!userPresence) { socket.emit('operation-forbidden', { @@ -182,7 +184,7 @@ export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager: } // Update user activity - await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() }) + await roomManager.updateUserActivity(wf(workflowId), socket.id, { lastActivity: Date.now() }) // Server-side debounce/coalesce by workflowId+blockId+subblockId const debouncedKey = `${workflowId}:${blockId}:${subblockId}` diff --git a/apps/realtime/src/handlers/variables.ts b/apps/realtime/src/handlers/variables.ts index 7a4303e70b3..0d6eaac2e3a 100644 --- a/apps/realtime/src/handlers/variables.ts +++ b/apps/realtime/src/handlers/variables.ts @@ -4,11 +4,12 @@ import { workflow } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { assertWorkflowMutable, WorkflowLockedError } from '@sim/platform-authz/workflow' import { VARIABLE_OPERATIONS } from '@sim/realtime-protocol/constants' +import { ROOM_TYPES } from '@sim/realtime-protocol/rooms' import { getErrorMessage } from '@sim/utils/errors' import { eq } from 'drizzle-orm' import type { AuthenticatedSocket } from '@/middleware/auth' import { checkWorkflowOperationPermission } from '@/middleware/permissions' -import type { IRoomManager } from '@/rooms' +import { type IRoomManager, workflowRoom as wf } from '@/rooms' const logger = createLogger('VariablesHandlers') @@ -61,7 +62,8 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager: } try { - const sessionWorkflowId = await roomManager.getWorkflowIdForSocket(socket.id) + const sessionWorkflowId = + (await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKFLOW))?.id ?? null const session = await roomManager.getUserSession(socket.id) if (!sessionWorkflowId || !session) { @@ -98,7 +100,7 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager: return } - const hasRoom = await roomManager.hasWorkflowRoom(workflowId) + const hasRoom = await roomManager.hasRoom(wf(workflowId)) if (!hasRoom) { logger.debug(`Ignoring variable update: workflow room not found`, { socketId: socket.id, @@ -109,7 +111,7 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager: return } - const users = await roomManager.getWorkflowUsers(workflowId) + const users = await roomManager.getRoomUsers(wf(workflowId)) const userPresence = users.find((user) => user.socketId === socket.id) if (!userPresence) { socket.emit('operation-forbidden', { @@ -174,7 +176,7 @@ export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager: } // Update user activity - await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() }) + await roomManager.updateUserActivity(wf(workflowId), socket.id, { lastActivity: Date.now() }) const debouncedKey = `${workflowId}:${variableId}:${field}` const existing = pendingVariableUpdates.get(debouncedKey) diff --git a/apps/realtime/src/handlers/workflow.test.ts b/apps/realtime/src/handlers/workflow.test.ts index 9dd82db87c4..89ae5521e2f 100644 --- a/apps/realtime/src/handlers/workflow.test.ts +++ b/apps/realtime/src/handlers/workflow.test.ts @@ -54,21 +54,20 @@ function createSocket(overrides?: Partial>) { function createRoomManager(overrides?: Partial): IRoomManager { return { isReady: vi.fn().mockReturnValue(true), - getWorkflowIdForSocket: vi.fn().mockResolvedValue(null), - removeUserFromRoom: vi.fn().mockResolvedValue(null), + getRoomForSocket: vi.fn().mockResolvedValue(null), + getRoomsForSocket: vi.fn().mockResolvedValue([]), + removeUserFromRoom: vi.fn().mockResolvedValue(false), + removeSocketFromAllRooms: vi.fn().mockResolvedValue([]), broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined), - getWorkflowUsers: vi.fn().mockResolvedValue([]), - hasWorkflowRoom: vi.fn().mockResolvedValue(false), + 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), - emitToWorkflow: vi.fn(), + emitToRoom: vi.fn(), getUniqueUserCount: vi.fn().mockResolvedValue(1), getTotalActiveConnections: vi.fn().mockResolvedValue(0), - handleWorkflowDeletion: vi.fn().mockResolvedValue(undefined), - handleWorkflowRevert: vi.fn().mockResolvedValue(undefined), - handleWorkflowUpdate: vi.fn().mockResolvedValue(undefined), shutdown: vi.fn().mockResolvedValue(undefined), initialize: vi.fn().mockResolvedValue(undefined), io: { @@ -173,8 +172,8 @@ describe('setupWorkflowHandlers', () => { it('includes workflowId when an unexpected join failure occurs', async () => { const { socket, handlers } = createSocket() const roomManager = createRoomManager({ - getWorkflowIdForSocket: vi.fn().mockRejectedValue(new Error('boom')), - removeUserFromRoom: vi.fn().mockResolvedValue(null), + getRoomForSocket: vi.fn().mockRejectedValue(new Error('boom')), + removeUserFromRoom: vi.fn().mockResolvedValue(false), }) setupWorkflowHandlers( diff --git a/apps/realtime/src/handlers/workflow.ts b/apps/realtime/src/handlers/workflow.ts index da977fcdb5e..553a92c7b78 100644 --- a/apps/realtime/src/handlers/workflow.ts +++ b/apps/realtime/src/handlers/workflow.ts @@ -1,10 +1,11 @@ import { db, user } from '@sim/db' import { createLogger } from '@sim/logger' +import { ROOM_TYPES } from '@sim/realtime-protocol/rooms' import { eq } from 'drizzle-orm' import { getWorkflowState } from '@/database/operations' import type { AuthenticatedSocket } from '@/middleware/auth' import { verifyWorkflowAccess } from '@/middleware/permissions' -import type { IRoomManager, UserPresence } from '@/rooms' +import { type IRoomManager, type UserPresence, workflowRoom as wf } from '@/rooms' const logger = createLogger('WorkflowHandlers') @@ -64,18 +65,18 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager: return } - // Leave current room if in one - const currentWorkflowId = await roomManager.getWorkflowIdForSocket(socket.id) - if (currentWorkflowId) { - socket.leave(currentWorkflowId) - await roomManager.removeUserFromRoom(socket.id, currentWorkflowId) - await roomManager.broadcastPresenceUpdate(currentWorkflowId) + // Leave current workflow room if in one + const currentRoom = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKFLOW) + if (currentRoom) { + socket.leave(currentRoom.id) + await roomManager.removeUserFromRoom(currentRoom, socket.id) + await roomManager.broadcastPresenceUpdate(currentRoom) } // Keep this above Redis socket key TTL (1h) so a normal idle user is not evicted too aggressively. const STALE_THRESHOLD_MS = 75 * 60 * 1000 const now = Date.now() - const existingUsers = await roomManager.getWorkflowUsers(workflowId) + const existingUsers = await roomManager.getRoomUsers(wf(workflowId)) let liveSocketIds = new Set() let canCheckLiveness = false @@ -106,7 +107,7 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager: logger.info( `Cleaning up socket ${existingUser.socketId} for user ${existingUser.userId} (same tab)` ) - await roomManager.removeUserFromRoom(existingUser.socketId, workflowId) + await roomManager.removeUserFromRoom(wf(workflowId), existingUser.socketId) await roomManager.io.in(existingUser.socketId).socketsLeave(workflowId) continue } @@ -124,7 +125,7 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager: logger.info( `Cleaning up socket ${existingUser.socketId} for user ${existingUser.userId} (stale activity)` ) - await roomManager.removeUserFromRoom(existingUser.socketId, workflowId) + await roomManager.removeUserFromRoom(wf(workflowId), existingUser.socketId) await roomManager.io.in(existingUser.socketId).socketsLeave(workflowId) } catch (error) { logger.warn(`Best-effort cleanup failed for socket ${existingUser.socketId}`, error) @@ -153,7 +154,7 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager: // Create presence entry const userPresence: UserPresence = { userId, - workflowId, + room: wf(workflowId), userName, socketId: socket.id, tabSessionId, @@ -164,10 +165,10 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager: } // Add user to room - await roomManager.addUserToRoom(workflowId, socket.id, userPresence) + await roomManager.addUserToRoom(wf(workflowId), socket.id, userPresence) // Get current presence list for the join acknowledgment - const presenceUsers = await roomManager.getWorkflowUsers(workflowId) + const presenceUsers = await roomManager.getRoomUsers(wf(workflowId)) // Get workflow state const workflowState = await getWorkflowState(workflowId) @@ -183,9 +184,9 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager: socket.emit('workflow-state', workflowState) // Broadcast presence update to all users in the room - await roomManager.broadcastPresenceUpdate(workflowId) + await roomManager.broadcastPresenceUpdate(wf(workflowId)) - const uniqueUserCount = await roomManager.getUniqueUserCount(workflowId) + const uniqueUserCount = await roomManager.getUniqueUserCount(wf(workflowId)) logger.info( `User ${userId} (${userName}) joined workflow ${workflowId}. Room now has ${uniqueUserCount} unique users.` ) @@ -193,7 +194,7 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager: logger.error('Error joining workflow:', error) // Undo socket.join and room manager entry if any operation failed socket.leave(workflowId) - await roomManager.removeUserFromRoom(socket.id, workflowId) + await roomManager.removeUserFromRoom(wf(workflowId), socket.id) const isReady = roomManager.isReady() socket.emit('join-workflow-error', { workflowId, @@ -210,15 +211,15 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager: return } - const workflowId = await roomManager.getWorkflowIdForSocket(socket.id) + const room = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKFLOW) const session = await roomManager.getUserSession(socket.id) - if (workflowId && session) { - socket.leave(workflowId) - await roomManager.removeUserFromRoom(socket.id, workflowId) - await roomManager.broadcastPresenceUpdate(workflowId) + if (room && session) { + socket.leave(room.id) + await roomManager.removeUserFromRoom(room, socket.id) + await roomManager.broadcastPresenceUpdate(room) - logger.info(`User ${session.userId} (${session.userName}) left workflow ${workflowId}`) + logger.info(`User ${session.userId} (${session.userName}) left workflow ${room.id}`) } } catch (error) { logger.error('Error leaving workflow:', error) diff --git a/apps/realtime/src/index.test.ts b/apps/realtime/src/index.test.ts index 92ddc101c69..5015b58d9bf 100644 --- a/apps/realtime/src/index.test.ts +++ b/apps/realtime/src/index.test.ts @@ -7,8 +7,9 @@ import { createServer, request as httpRequest } from 'http' import { createMockLogger } from '@sim/testing' import { randomInt } from '@sim/utils/random' import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' +import { ROOM_TYPES } from '@sim/realtime-protocol/rooms' import { createSocketIOServer } from '@/config/socket' -import { MemoryRoomManager } from '@/rooms' +import { MemoryRoomManager, workflowRoom } from '@/rooms' import { createHttpHandler } from '@/routes/http' vi.mock('@/auth', () => ({ @@ -230,9 +231,10 @@ describe('Socket Server Index Integration', () => { const workflowId = 'test-workflow-123' const socketId = 'test-socket-123' + const room = workflowRoom(workflowId) const presence = { userId: 'user-123', - workflowId, + room, userName: 'Test User', socketId, joinedAt: Date.now(), @@ -240,10 +242,10 @@ describe('Socket Server Index Integration', () => { role: 'admin', } - await roomManager.addUserToRoom(workflowId, socketId, presence) + await roomManager.addUserToRoom(room, socketId, presence) - expect(await roomManager.hasWorkflowRoom(workflowId)).toBe(true) - const users = await roomManager.getWorkflowUsers(workflowId) + expect(await roomManager.hasRoom(room)).toBe(true) + const users = await roomManager.getRoomUsers(room) expect(users).toHaveLength(1) expect(users[0].socketId).toBe(socketId) }) @@ -252,9 +254,10 @@ describe('Socket Server Index Integration', () => { const socketId = 'test-socket-123' const workflowId = 'test-workflow-456' + const room = workflowRoom(workflowId) const presence = { userId: 'user-123', - workflowId, + room, userName: 'Test User', socketId, joinedAt: Date.now(), @@ -262,9 +265,9 @@ describe('Socket Server Index Integration', () => { role: 'admin', } - await roomManager.addUserToRoom(workflowId, socketId, presence) + await roomManager.addUserToRoom(room, socketId, presence) - expect(await roomManager.getWorkflowIdForSocket(socketId)).toBe(workflowId) + expect(await roomManager.getRoomForSocket(socketId, ROOM_TYPES.WORKFLOW)).toEqual(room) const session = await roomManager.getUserSession(socketId) expect(session).toBeDefined() expect(session?.userId).toBe('user-123') @@ -274,9 +277,10 @@ describe('Socket Server Index Integration', () => { const workflowId = 'test-workflow-789' const socketId = 'test-socket-789' + const room = workflowRoom(workflowId) const presence = { userId: 'user-789', - workflowId, + room, userName: 'Test User', socketId, joinedAt: Date.now(), @@ -284,16 +288,16 @@ describe('Socket Server Index Integration', () => { role: 'admin', } - await roomManager.addUserToRoom(workflowId, socketId, presence) + await roomManager.addUserToRoom(room, socketId, presence) - expect(await roomManager.hasWorkflowRoom(workflowId)).toBe(true) + expect(await roomManager.hasRoom(room)).toBe(true) // Remove user - await roomManager.removeUserFromRoom(socketId) + await roomManager.removeUserFromRoom(room, socketId) // Room should be cleaned up since it's now empty - expect(await roomManager.hasWorkflowRoom(workflowId)).toBe(false) - expect(await roomManager.getWorkflowIdForSocket(socketId)).toBeNull() + expect(await roomManager.hasRoom(room)).toBe(false) + expect(await roomManager.getRoomForSocket(socketId, ROOM_TYPES.WORKFLOW)).toBeNull() }) }) @@ -324,7 +328,7 @@ describe('Socket Server Index Integration', () => { expect(typeof roomManager.addUserToRoom).toBe('function') expect(typeof roomManager.removeUserFromRoom).toBe('function') - expect(typeof roomManager.handleWorkflowDeletion).toBe('function') + expect(typeof roomManager.removeSocketFromAllRooms).toBe('function') expect(typeof roomManager.broadcastPresenceUpdate).toBe('function') }) }) diff --git a/apps/realtime/src/rooms/index.ts b/apps/realtime/src/rooms/index.ts index 8067fc1b215..29157aa4581 100644 --- a/apps/realtime/src/rooms/index.ts +++ b/apps/realtime/src/rooms/index.ts @@ -1,3 +1,4 @@ export { MemoryRoomManager } from '@/rooms/memory-manager' export { RedisRoomManager } from '@/rooms/redis-manager' -export type { IRoomManager, UserPresence, UserSession, WorkflowRoom } from '@/rooms/types' +export type { IRoomManager, RoomState, UserPresence, UserSession } from '@/rooms/types' +export { WorkflowRoomService, workflowRoom } from '@/rooms/workflow-room-service' diff --git a/apps/realtime/src/rooms/memory-manager.test.ts b/apps/realtime/src/rooms/memory-manager.test.ts new file mode 100644 index 00000000000..fe57009170d --- /dev/null +++ b/apps/realtime/src/rooms/memory-manager.test.ts @@ -0,0 +1,130 @@ +/** + * Multi-room semantics for the room manager. Exercises the invariants the + * single-room → multi-room migration must preserve: a socket in two rooms, + * refcounted session cleanup, presence isolation, and full-disconnect cleanup. + * + * @vitest-environment node + */ +import { ROOM_TYPES, type RoomRef } from '@sim/realtime-protocol/rooms' +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { MemoryRoomManager } from '@/rooms/memory-manager' +import type { UserPresence } from '@/rooms/types' + +function fakeIo() { + const emit = vi.fn() + return { + emit, + io: { to: vi.fn().mockReturnValue({ emit }) } as never, + } +} + +function presence(room: RoomRef, socketId: string, userId: string): UserPresence { + return { + userId, + room, + userName: `user-${userId}`, + socketId, + joinedAt: Date.now(), + lastActivity: Date.now(), + role: 'admin', + } +} + +const WORKFLOW: RoomRef = { type: ROOM_TYPES.WORKFLOW, id: 'wf-1' } +const FILES: RoomRef = { type: ROOM_TYPES.WORKSPACE_FILES, id: 'ws-1' } + +describe('MemoryRoomManager multi-room', () => { + let manager: MemoryRoomManager + + beforeEach(async () => { + manager = new MemoryRoomManager(fakeIo().io) + await manager.initialize() + }) + + it('tracks a single socket in two rooms of different types', async () => { + await manager.addUserToRoom(WORKFLOW, 'socket-1', presence(WORKFLOW, 'socket-1', 'user-1')) + await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + + const rooms = await manager.getRoomsForSocket('socket-1') + expect(rooms).toHaveLength(2) + expect(rooms).toContainEqual(WORKFLOW) + expect(rooms).toContainEqual(FILES) + + expect(await manager.getRoomForSocket('socket-1', ROOM_TYPES.WORKFLOW)).toEqual(WORKFLOW) + expect(await manager.getRoomForSocket('socket-1', ROOM_TYPES.WORKSPACE_FILES)).toEqual(FILES) + }) + + it('keeps the shared session alive when leaving one of two rooms (refcount)', async () => { + await manager.addUserToRoom(WORKFLOW, 'socket-1', presence(WORKFLOW, 'socket-1', 'user-1')) + await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + + const removed = await manager.removeUserFromRoom(WORKFLOW, 'socket-1') + expect(removed).toBe(true) + + // The files room and the shared session must survive. + expect(await manager.hasRoom(WORKFLOW)).toBe(false) + expect(await manager.hasRoom(FILES)).toBe(true) + expect(await manager.getUserSession('socket-1')).not.toBeNull() + expect(await manager.getRoomForSocket('socket-1', ROOM_TYPES.WORKFLOW)).toBeNull() + expect(await manager.getRoomForSocket('socket-1', ROOM_TYPES.WORKSPACE_FILES)).toEqual(FILES) + }) + + it('drops the shared session only when the last room is left', async () => { + await manager.addUserToRoom(WORKFLOW, 'socket-1', presence(WORKFLOW, 'socket-1', 'user-1')) + await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + + await manager.removeUserFromRoom(WORKFLOW, 'socket-1') + expect(await manager.getUserSession('socket-1')).not.toBeNull() + + await manager.removeUserFromRoom(FILES, 'socket-1') + expect(await manager.getUserSession('socket-1')).toBeNull() + expect(await manager.getRoomsForSocket('socket-1')).toHaveLength(0) + }) + + it('isolates presence between rooms of different types', async () => { + await manager.addUserToRoom(WORKFLOW, 'socket-1', presence(WORKFLOW, 'socket-1', 'user-1')) + await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + await manager.addUserToRoom(FILES, 'socket-2', presence(FILES, 'socket-2', 'user-2')) + + expect(await manager.getRoomUsers(WORKFLOW)).toHaveLength(1) + expect(await manager.getRoomUsers(FILES)).toHaveLength(2) + }) + + it('removes a socket from every room on disconnect and reports them', async () => { + await manager.addUserToRoom(WORKFLOW, 'socket-1', presence(WORKFLOW, 'socket-1', 'user-1')) + await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + await manager.addUserToRoom(FILES, 'socket-2', presence(FILES, 'socket-2', 'user-2')) + + const removed = await manager.removeSocketFromAllRooms('socket-1') + expect(removed).toHaveLength(2) + expect(removed).toContainEqual(WORKFLOW) + expect(removed).toContainEqual(FILES) + + expect(await manager.hasRoom(WORKFLOW)).toBe(false) + // The files room still has socket-2. + expect(await manager.getRoomUsers(FILES)).toHaveLength(1) + expect(await manager.getUserSession('socket-1')).toBeNull() + }) + + it('does not clobber another type when two sockets share a room', async () => { + await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + await manager.addUserToRoom(FILES, 'socket-2', presence(FILES, 'socket-2', 'user-2')) + + await manager.removeUserFromRoom(FILES, 'socket-1') + expect(await manager.hasRoom(FILES)).toBe(true) + expect(await manager.getUserSession('socket-2')).not.toBeNull() + }) + + it('broadcasts presence on the room-type-specific event name', async () => { + const { emit, io } = fakeIo() + const m = new MemoryRoomManager(io) + await m.initialize() + await m.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + + await m.broadcastPresenceUpdate(FILES) + expect(emit).toHaveBeenCalledWith('workspace-files:presence-update', expect.any(Array)) + + await m.broadcastPresenceUpdate(WORKFLOW) + expect(emit).toHaveBeenCalledWith('presence-update', expect.any(Array)) + }) +}) diff --git a/apps/realtime/src/rooms/memory-manager.ts b/apps/realtime/src/rooms/memory-manager.ts index a032e785bb5..a7279842249 100644 --- a/apps/realtime/src/rooms/memory-manager.ts +++ b/apps/realtime/src/rooms/memory-manager.ts @@ -1,16 +1,29 @@ import { createLogger } from '@sim/logger' +import { + presenceEventName, + type RoomRef, + type RoomType, + roomName, +} from '@sim/realtime-protocol/rooms' import type { Server } from 'socket.io' -import type { IRoomManager, UserPresence, UserSession, WorkflowRoom } from '@/rooms/types' +import type { IRoomManager, RoomState, UserPresence, UserSession } from '@/rooms/types' const logger = createLogger('MemoryRoomManager') +/** Stable string key for a room in the local maps (distinct from the Socket.IO room name). */ +function roomKey(room: RoomRef): string { + return `${room.type}:${room.id}` +} + /** - * In-memory room manager for single-pod deployments - * Used as fallback when REDIS_URL is not configured + * In-memory room manager for single-pod deployments. Used when REDIS_URL is not + * configured. Domain-neutral: keyed by {@link RoomRef}, supports a socket in + * multiple rooms (one per {@link RoomType}). */ export class MemoryRoomManager implements IRoomManager { - private workflowRooms = new Map() - private socketToWorkflow = new Map() + private rooms = new Map() + /** socketId -> (roomType -> roomId) */ + private socketRooms = new Map>() private userSessions = new Map() private _io: Server @@ -31,228 +44,150 @@ export class MemoryRoomManager implements IRoomManager { } async shutdown(): Promise { - this.workflowRooms.clear() - this.socketToWorkflow.clear() + this.rooms.clear() + this.socketRooms.clear() this.userSessions.clear() logger.info('MemoryRoomManager shutdown complete') } - async addUserToRoom(workflowId: string, socketId: string, presence: UserPresence): Promise { - // Create room if it doesn't exist - if (!this.workflowRooms.has(workflowId)) { - this.workflowRooms.set(workflowId, { - workflowId, - users: new Map(), - lastModified: Date.now(), - activeConnections: 0, - }) + async addUserToRoom(room: RoomRef, socketId: string, presence: UserPresence): Promise { + const key = roomKey(room) + let state = this.rooms.get(key) + if (!state) { + state = { room, users: new Map(), lastModified: Date.now(), activeConnections: 0 } + this.rooms.set(key, state) } - const room = this.workflowRooms.get(workflowId)! - room.users.set(socketId, presence) - room.activeConnections++ - room.lastModified = Date.now() + state.users.set(socketId, presence) + state.activeConnections++ + state.lastModified = Date.now() - // Map socket to workflow - this.socketToWorkflow.set(socketId, workflowId) + let socketRoomMap = this.socketRooms.get(socketId) + if (!socketRoomMap) { + socketRoomMap = new Map() + this.socketRooms.set(socketId, socketRoomMap) + } + socketRoomMap.set(room.type, room.id) - // Store session this.userSessions.set(socketId, { userId: presence.userId, userName: presence.userName, avatarUrl: presence.avatarUrl, }) - logger.debug(`Added user ${presence.userId} to workflow ${workflowId} (socket: ${socketId})`) + logger.debug(`Added user ${presence.userId} to room ${key} (socket: ${socketId})`) } - async removeUserFromRoom(socketId: string, _workflowIdHint?: string): Promise { - const workflowId = this.socketToWorkflow.get(socketId) + async removeUserFromRoom(room: RoomRef, socketId: string): Promise { + const key = roomKey(room) + const state = this.rooms.get(key) + let existed = false - if (!workflowId) { - return null + if (state?.users.has(socketId)) { + existed = true + state.users.delete(socketId) + state.activeConnections = Math.max(0, state.activeConnections - 1) + if (state.users.size === 0) { + this.rooms.delete(key) + logger.info(`Cleaned up empty room: ${key}`) + } } - const room = this.workflowRooms.get(workflowId) - if (room) { - room.users.delete(socketId) - room.activeConnections = Math.max(0, room.activeConnections - 1) - - // Clean up empty rooms - if (room.activeConnections === 0) { - this.workflowRooms.delete(workflowId) - logger.info(`Cleaned up empty workflow room: ${workflowId}`) + const socketRoomMap = this.socketRooms.get(socketId) + if (socketRoomMap && socketRoomMap.get(room.type) === room.id) { + socketRoomMap.delete(room.type) + // Drop the shared session only when the socket has left its last room. + if (socketRoomMap.size === 0) { + this.socketRooms.delete(socketId) + this.userSessions.delete(socketId) } } - this.socketToWorkflow.delete(socketId) + return existed + } + + async removeSocketFromAllRooms(socketId: string): Promise { + const socketRoomMap = this.socketRooms.get(socketId) + if (!socketRoomMap || socketRoomMap.size === 0) { + this.userSessions.delete(socketId) + return [] + } + + const rooms: RoomRef[] = Array.from(socketRoomMap.entries()).map(([type, id]) => ({ type, id })) + for (const room of rooms) { + await this.removeUserFromRoom(room, socketId) + } + // Belt-and-suspenders: ensure session is gone even if the map drifted. + this.socketRooms.delete(socketId) this.userSessions.delete(socketId) + return rooms + } - logger.debug(`Removed socket ${socketId} from workflow ${workflowId}`) - return workflowId + async getRoomsForSocket(socketId: string): Promise { + const socketRoomMap = this.socketRooms.get(socketId) + if (!socketRoomMap) return [] + return Array.from(socketRoomMap.entries()).map(([type, id]) => ({ type, id })) } - async getWorkflowIdForSocket(socketId: string): Promise { - return this.socketToWorkflow.get(socketId) ?? null + async getRoomForSocket(socketId: string, type: RoomType): Promise { + const id = this.socketRooms.get(socketId)?.get(type) + return id ? { type, id } : null } async getUserSession(socketId: string): Promise { return this.userSessions.get(socketId) ?? null } - async getWorkflowUsers(workflowId: string): Promise { - const room = this.workflowRooms.get(workflowId) - if (!room) return [] - return Array.from(room.users.values()) + async getRoomUsers(room: RoomRef): Promise { + const state = this.rooms.get(roomKey(room)) + if (!state) return [] + return Array.from(state.users.values()) } - async hasWorkflowRoom(workflowId: string): Promise { - return this.workflowRooms.has(workflowId) + async hasRoom(room: RoomRef): Promise { + return this.rooms.has(roomKey(room)) } async updateUserActivity( - workflowId: string, + room: RoomRef, socketId: string, updates: Partial> ): Promise { - const room = this.workflowRooms.get(workflowId) - if (!room) return + const presence = this.rooms.get(roomKey(room))?.users.get(socketId) + if (!presence) return - const presence = room.users.get(socketId) - if (presence) { - if (updates.cursor !== undefined) presence.cursor = updates.cursor - if (updates.selection !== undefined) presence.selection = updates.selection - presence.lastActivity = updates.lastActivity ?? Date.now() - } + if (updates.cursor !== undefined) presence.cursor = updates.cursor + if (updates.selection !== undefined) presence.selection = updates.selection + presence.lastActivity = updates.lastActivity ?? Date.now() } - async updateRoomLastModified(workflowId: string): Promise { - const room = this.workflowRooms.get(workflowId) - if (room) { - room.lastModified = Date.now() - } + async updateRoomLastModified(room: RoomRef): Promise { + const state = this.rooms.get(roomKey(room)) + if (state) state.lastModified = Date.now() } - async broadcastPresenceUpdate(workflowId: string): Promise { - const users = await this.getWorkflowUsers(workflowId) - this._io.to(workflowId).emit('presence-update', users) + async broadcastPresenceUpdate(room: RoomRef): Promise { + const users = await this.getRoomUsers(room) + this._io.to(roomName(room)).emit(presenceEventName(room.type), users) } - emitToWorkflow(workflowId: string, event: string, payload: T): void { - this._io.to(workflowId).emit(event, payload) + emitToRoom(room: RoomRef, event: string, payload: T): void { + this._io.to(roomName(room)).emit(event, payload) } - async getUniqueUserCount(workflowId: string): Promise { - const room = this.workflowRooms.get(workflowId) - if (!room) return 0 - + async getUniqueUserCount(room: RoomRef): Promise { + const state = this.rooms.get(roomKey(room)) + if (!state) return 0 const uniqueUsers = new Set() - room.users.forEach((presence) => { - uniqueUsers.add(presence.userId) - }) - + state.users.forEach((presence) => uniqueUsers.add(presence.userId)) return uniqueUsers.size } async getTotalActiveConnections(): Promise { let total = 0 - for (const room of this.workflowRooms.values()) { - total += room.activeConnections + for (const state of this.rooms.values()) { + total += state.activeConnections } return total } - - async handleWorkflowDeletion(workflowId: string): Promise { - logger.info(`Handling workflow deletion notification for ${workflowId}`) - - const room = this.workflowRooms.get(workflowId) - if (!room) { - logger.debug(`No active room found for deleted workflow ${workflowId}`) - return - } - - this._io.to(workflowId).emit('workflow-deleted', { - workflowId, - message: 'This workflow has been deleted', - timestamp: Date.now(), - }) - - const socketsToDisconnect: string[] = [] - room.users.forEach((_presence, socketId) => { - socketsToDisconnect.push(socketId) - }) - - for (const socketId of socketsToDisconnect) { - const socket = this._io.sockets.sockets.get(socketId) - if (socket) { - socket.leave(workflowId) - logger.debug(`Disconnected socket ${socketId} from deleted workflow ${workflowId}`) - } - await this.removeUserFromRoom(socketId) - } - - this.workflowRooms.delete(workflowId) - logger.info( - `Cleaned up workflow room ${workflowId} after deletion (${socketsToDisconnect.length} users disconnected)` - ) - } - - async handleWorkflowRevert(workflowId: string, timestamp: number): Promise { - logger.info(`Handling workflow revert notification for ${workflowId}`) - - const room = this.workflowRooms.get(workflowId) - if (!room) { - logger.debug(`No active room found for reverted workflow ${workflowId}`) - return - } - - this._io.to(workflowId).emit('workflow-reverted', { - workflowId, - message: 'Workflow has been reverted to deployed state', - timestamp, - }) - - room.lastModified = timestamp - - logger.info(`Notified ${room.users.size} users about workflow revert: ${workflowId}`) - } - - async handleWorkflowUpdate(workflowId: string): Promise { - logger.info(`Handling workflow update notification for ${workflowId}`) - - const room = this.workflowRooms.get(workflowId) - if (!room) { - logger.debug(`No active room found for updated workflow ${workflowId}`) - return - } - - const timestamp = Date.now() - - this._io.to(workflowId).emit('workflow-updated', { - workflowId, - message: 'Workflow has been updated externally', - timestamp, - }) - - room.lastModified = timestamp - - logger.info(`Notified ${room.users.size} users about workflow update: ${workflowId}`) - } - - async handleWorkflowDeployed(workflowId: string): Promise { - logger.info(`Handling workflow deployed notification for ${workflowId}`) - - const room = this.workflowRooms.get(workflowId) - if (!room) { - logger.debug(`No active room found for deployed workflow ${workflowId}`) - return - } - - this._io.to(workflowId).emit('workflow-deployed', { - workflowId, - timestamp: Date.now(), - }) - - logger.info(`Notified ${room.users.size} users about workflow deployment change: ${workflowId}`) - } } diff --git a/apps/realtime/src/rooms/redis-manager.ts b/apps/realtime/src/rooms/redis-manager.ts index 0e6b3eadf2b..6dd6d1f3ceb 100644 --- a/apps/realtime/src/rooms/redis-manager.ts +++ b/apps/realtime/src/rooms/redis-manager.ts @@ -1,86 +1,95 @@ import { createLogger } from '@sim/logger' +import { + presenceEventName, + type RoomRef, + type RoomType, + roomName, +} from '@sim/realtime-protocol/rooms' import { createClient, type RedisClientType } from 'redis' import type { Server } from 'socket.io' import type { IRoomManager, UserPresence, UserSession } from '@/rooms/types' const logger = createLogger('RedisRoomManager') +/** + * Redis key scheme (all room-scoped keys are prefixed by room type): + * {type}:{id}:users HASH socketId -> UserPresence JSON (room membership) + * {type}:{id}:meta HASH room metadata (lastModified) + * socket:{sid}:rooms HASH roomType -> roomId (the socket's rooms, one per type) + * socket:{sid}:session HASH userId/userName/avatarUrl (shared across the socket's rooms) + * + * Workflow rooms keep their historical `workflow:{id}:users`/`:meta` keys (the + * type prefix IS `workflow`), so no presence-state migration is needed for them. + */ const KEYS = { - workflowUsers: (wfId: string) => `workflow:${wfId}:users`, - workflowMeta: (wfId: string) => `workflow:${wfId}:meta`, - socketWorkflow: (socketId: string) => `socket:${socketId}:workflow`, + roomUsers: (room: RoomRef) => `${room.type}:${room.id}:users`, + roomMeta: (room: RoomRef) => `${room.type}:${room.id}:meta`, + socketRooms: (socketId: string) => `socket:${socketId}:rooms`, socketSession: (socketId: string) => `socket:${socketId}:session`, - socketPresenceWorkflow: (socketId: string) => `socket:${socketId}:presence-workflow`, } as const -const SOCKET_KEY_TTL = 3600 -const SOCKET_PRESENCE_WORKFLOW_KEY_TTL = 24 * 60 * 60 +/** TTL for the socket's room-set. Long enough that an idle-but-connected socket is not evicted. */ +const SOCKET_ROOMS_TTL = 24 * 60 * 60 +/** TTL for the shared session key; refreshed on every activity update. */ +const SESSION_TTL = 60 * 60 /** - * Lua script for atomic user removal from room. - * Returns workflowId if user was removed, null otherwise. - * Handles room cleanup atomically to prevent race conditions. + * Atomic single-room removal. Removes a socket from one room's presence, drops + * the room from the socket's room-set, and — critically — deletes the SHARED + * session key only when the socket has left its LAST room (otherwise a leave from + * one room would break the socket's other rooms). Cleans up empty room state. + * + * KEYS: [socketRooms, socketSession, roomUsers, roomMeta] + * ARGV: [roomType, socketId] + * Returns 1 if the socket was a member of the room, else 0. */ -const REMOVE_USER_SCRIPT = ` -local socketWorkflowKey = KEYS[1] +const REMOVE_ROOM_SCRIPT = ` +local socketRoomsKey = KEYS[1] local socketSessionKey = KEYS[2] -local socketPresenceWorkflowKey = KEYS[3] -local workflowUsersPrefix = ARGV[1] -local workflowMetaPrefix = ARGV[2] -local socketId = ARGV[3] -local workflowIdHint = ARGV[4] - -local workflowId = redis.call('GET', socketWorkflowKey) -if not workflowId then - workflowId = redis.call('GET', socketPresenceWorkflowKey) -end +local roomUsersKey = KEYS[3] +local roomMetaKey = KEYS[4] +local roomType = ARGV[1] +local socketId = ARGV[2] -if not workflowId and workflowIdHint ~= '' then - workflowId = workflowIdHint -end +local removed = redis.call('HDEL', roomUsersKey, socketId) +redis.call('HDEL', socketRoomsKey, roomType) -if not workflowId then - return nil +if redis.call('HLEN', socketRoomsKey) == 0 then + redis.call('DEL', socketRoomsKey, socketSessionKey) end -local workflowUsersKey = workflowUsersPrefix .. workflowId .. ':users' -local workflowMetaKey = workflowMetaPrefix .. workflowId .. ':meta' - -redis.call('HDEL', workflowUsersKey, socketId) -redis.call('DEL', socketWorkflowKey, socketSessionKey, socketPresenceWorkflowKey) - -local remaining = redis.call('HLEN', workflowUsersKey) -if remaining == 0 then - redis.call('DEL', workflowUsersKey, workflowMetaKey) +if redis.call('HLEN', roomUsersKey) == 0 then + redis.call('DEL', roomUsersKey, roomMetaKey) end -return workflowId +return removed ` /** - * Lua script for atomic user activity update. - * Performs read-modify-write atomically to prevent lost updates. - * Also refreshes TTL on socket keys to prevent expiry during long sessions. + * Atomic presence-activity update (read-modify-write) that also refreshes the + * socket key TTLs to keep a long-lived session alive. + * + * KEYS: [roomUsers, socketRooms, socketSession] + * ARGV: [socketId, cursorJson, selectionJson, lastActivity, roomsTtl, sessionTtl] + * Returns 1 if the socket had presence in the room, else 0. */ const UPDATE_ACTIVITY_SCRIPT = ` -local workflowUsersKey = KEYS[1] -local socketWorkflowKey = KEYS[2] +local roomUsersKey = KEYS[1] +local socketRoomsKey = KEYS[2] local socketSessionKey = KEYS[3] -local socketPresenceWorkflowKey = KEYS[4] local socketId = ARGV[1] local cursorJson = ARGV[2] local selectionJson = ARGV[3] local lastActivity = ARGV[4] -local ttl = tonumber(ARGV[5]) -local presenceWorkflowTtl = tonumber(ARGV[6]) +local roomsTtl = tonumber(ARGV[5]) +local sessionTtl = tonumber(ARGV[6]) -local existingJson = redis.call('HGET', workflowUsersKey, socketId) +local existingJson = redis.call('HGET', roomUsersKey, socketId) if not existingJson then return 0 end local existing = cjson.decode(existingJson) - if cursorJson ~= '' then existing.cursor = cjson.decode(cursorJson) end @@ -89,44 +98,39 @@ if selectionJson ~= '' then end existing.lastActivity = tonumber(lastActivity) -redis.call('HSET', workflowUsersKey, socketId, cjson.encode(existing)) -redis.call('EXPIRE', socketWorkflowKey, ttl) -redis.call('EXPIRE', socketSessionKey, ttl) -redis.call('EXPIRE', socketPresenceWorkflowKey, presenceWorkflowTtl) +redis.call('HSET', roomUsersKey, socketId, cjson.encode(existing)) +redis.call('EXPIRE', socketRoomsKey, roomsTtl) +redis.call('EXPIRE', socketSessionKey, sessionTtl) return 1 ` /** - * Redis-backed room manager for multi-pod deployments. - * Uses Lua scripts for atomic operations to prevent race conditions. + * Redis-backed room manager for multi-pod deployments. Domain-neutral: keyed by + * {@link RoomRef}, supports a socket in multiple rooms (one per {@link RoomType}). + * Uses Lua scripts for atomic multi-key operations. */ export class RedisRoomManager implements IRoomManager { private redis: RedisClientType private _io: Server private isConnected = false - private removeUserScriptSha: string | null = null + private removeRoomScriptSha: string | null = null private updateActivityScriptSha: string | null = null constructor(io: Server, redisUrl: string) { this._io = io - this.redis = createClient({ - url: redisUrl, - }) + this.redis = createClient({ url: redisUrl }) this.redis.on('error', (err) => { logger.error('Redis client error:', err) }) - this.redis.on('reconnecting', () => { logger.warn('Redis client reconnecting...') this.isConnected = false }) - this.redis.on('ready', () => { logger.info('Redis client ready') this.isConnected = true }) - this.redis.on('end', () => { logger.warn('Redis client connection closed') this.isConnected = false @@ -148,8 +152,7 @@ export class RedisRoomManager implements IRoomManager { await this.redis.connect() this.isConnected = true - // Pre-load Lua scripts for better performance - this.removeUserScriptSha = await this.redis.scriptLoad(REMOVE_USER_SCRIPT) + this.removeRoomScriptSha = await this.redis.scriptLoad(REMOVE_ROOM_SCRIPT) this.updateActivityScriptSha = await this.redis.scriptLoad(UPDATE_ACTIVITY_SCRIPT) logger.info('RedisRoomManager connected to Redis and scripts loaded') @@ -161,7 +164,6 @@ export class RedisRoomManager implements IRoomManager { async shutdown(): Promise { if (!this.isConnected) return - try { await this.redis.quit() this.isConnected = false @@ -171,93 +173,102 @@ export class RedisRoomManager implements IRoomManager { } } - async addUserToRoom(workflowId: string, socketId: string, presence: UserPresence): Promise { + async addUserToRoom(room: RoomRef, socketId: string, presence: UserPresence): Promise { try { const pipeline = this.redis.multi() - pipeline.hSet(KEYS.workflowUsers(workflowId), socketId, JSON.stringify(presence)) - pipeline.hSet(KEYS.workflowMeta(workflowId), 'lastModified', Date.now().toString()) - pipeline.set(KEYS.socketWorkflow(socketId), workflowId) - pipeline.expire(KEYS.socketWorkflow(socketId), SOCKET_KEY_TTL) - pipeline.set(KEYS.socketPresenceWorkflow(socketId), workflowId) - pipeline.expire(KEYS.socketPresenceWorkflow(socketId), SOCKET_PRESENCE_WORKFLOW_KEY_TTL) + pipeline.hSet(KEYS.roomUsers(room), socketId, JSON.stringify(presence)) + pipeline.hSet(KEYS.roomMeta(room), 'lastModified', Date.now().toString()) + pipeline.hSet(KEYS.socketRooms(socketId), room.type, room.id) + pipeline.expire(KEYS.socketRooms(socketId), SOCKET_ROOMS_TTL) pipeline.hSet(KEYS.socketSession(socketId), { userId: presence.userId, userName: presence.userName, avatarUrl: presence.avatarUrl || '', }) - pipeline.expire(KEYS.socketSession(socketId), SOCKET_KEY_TTL) + pipeline.expire(KEYS.socketSession(socketId), SESSION_TTL) const results = await pipeline.exec() - // Check if any command failed const failed = results.some((result) => result instanceof Error) if (failed) { - logger.error(`Pipeline partially failed when adding user to room`, { workflowId, socketId }) + logger.error('Pipeline partially failed when adding user to room', { + room, + socketId, + }) throw new Error('Failed to store user session data in Redis') } - logger.debug(`Added user ${presence.userId} to workflow ${workflowId} (socket: ${socketId})`) + logger.debug(`Added user ${presence.userId} to room ${room.type}:${room.id} (${socketId})`) } catch (error) { - logger.error(`Failed to add user to room: ${socketId} -> ${workflowId}`, error) + logger.error(`Failed to add user to room: ${socketId} -> ${room.type}:${room.id}`, error) throw error } } - async removeUserFromRoom( - socketId: string, - workflowIdHint?: string, - retried = false - ): Promise { - if (!this.removeUserScriptSha) { + async removeUserFromRoom(room: RoomRef, socketId: string, retried = false): Promise { + if (!this.removeRoomScriptSha) { logger.error('removeUserFromRoom called before initialize()') - return null + return false } try { - const workflowId = await this.redis.evalSha(this.removeUserScriptSha, { + const removed = await this.redis.evalSha(this.removeRoomScriptSha, { keys: [ - KEYS.socketWorkflow(socketId), + KEYS.socketRooms(socketId), KEYS.socketSession(socketId), - KEYS.socketPresenceWorkflow(socketId), + KEYS.roomUsers(room), + KEYS.roomMeta(room), ], - arguments: ['workflow:', 'workflow:', socketId, workflowIdHint ?? ''], + arguments: [room.type, socketId], }) - - if (typeof workflowId === 'string' && workflowId.length > 0) { - logger.debug(`Removed socket ${socketId} from workflow ${workflowId}`) - return workflowId - } - - return null + return typeof removed === 'number' ? removed > 0 : Number(removed) > 0 } catch (error) { if ((error as Error).message?.includes('NOSCRIPT') && !retried) { logger.warn('Lua script not found, reloading...') - this.removeUserScriptSha = await this.redis.scriptLoad(REMOVE_USER_SCRIPT) - return this.removeUserFromRoom(socketId, workflowIdHint, true) + this.removeRoomScriptSha = await this.redis.scriptLoad(REMOVE_ROOM_SCRIPT) + return this.removeUserFromRoom(room, socketId, true) } - logger.error(`Failed to remove user from room: ${socketId}`, error) - return null + logger.error(`Failed to remove socket ${socketId} from room ${room.type}:${room.id}`, error) + return false } } - async getWorkflowIdForSocket(socketId: string): Promise { - const workflowId = await this.redis.get(KEYS.socketWorkflow(socketId)) - if (workflowId) { - return workflowId + async removeSocketFromAllRooms(socketId: string): Promise { + const rooms = await this.getRoomsForSocket(socketId) + if (rooms.length === 0) { + // Nothing tracked (already cleaned up or TTL-expired); ensure session is gone. + await this.redis.del(KEYS.socketSession(socketId)).catch(() => {}) + return [] } - return this.redis.get(KEYS.socketPresenceWorkflow(socketId)) + const removed: RoomRef[] = [] + for (const room of rooms) { + const wasMember = await this.removeUserFromRoom(room, socketId) + if (wasMember) removed.push(room) + } + return removed } - async getUserSession(socketId: string): Promise { + async getRoomsForSocket(socketId: string): Promise { try { - const session = await this.redis.hGetAll(KEYS.socketSession(socketId)) + const entries = await this.redis.hGetAll(KEYS.socketRooms(socketId)) + return Object.entries(entries).map(([type, id]) => ({ type: type as RoomType, id })) + } catch (error) { + logger.error(`Failed to get rooms for socket ${socketId}:`, error) + return [] + } + } - if (!session.userId) { - return null - } + async getRoomForSocket(socketId: string, type: RoomType): Promise { + const id = await this.redis.hGet(KEYS.socketRooms(socketId), type) + return id ? { type, id } : null + } + async getUserSession(socketId: string): Promise { + try { + const session = await this.redis.hGetAll(KEYS.socketSession(socketId)) + if (!session.userId) return null return { userId: session.userId, userName: session.userName, @@ -269,9 +280,9 @@ export class RedisRoomManager implements IRoomManager { } } - async getWorkflowUsers(workflowId: string): Promise { + async getRoomUsers(room: RoomRef): Promise { try { - const users = await this.redis.hGetAll(KEYS.workflowUsers(workflowId)) + const users = await this.redis.hGetAll(KEYS.roomUsers(room)) return Object.entries(users) .map(([socketId, json]) => { try { @@ -283,18 +294,18 @@ export class RedisRoomManager implements IRoomManager { }) .filter((u): u is UserPresence => u !== null) } catch (error) { - logger.error(`Failed to get workflow users for ${workflowId}:`, error) + logger.error(`Failed to get room users for ${room.type}:${room.id}:`, error) return [] } } - async hasWorkflowRoom(workflowId: string): Promise { - const exists = await this.redis.exists(KEYS.workflowUsers(workflowId)) + async hasRoom(room: RoomRef): Promise { + const exists = await this.redis.exists(KEYS.roomUsers(room)) return exists > 0 } async updateUserActivity( - workflowId: string, + room: RoomRef, socketId: string, updates: Partial>, retried = false @@ -306,155 +317,47 @@ export class RedisRoomManager implements IRoomManager { try { await this.redis.evalSha(this.updateActivityScriptSha, { - keys: [ - KEYS.workflowUsers(workflowId), - KEYS.socketWorkflow(socketId), - KEYS.socketSession(socketId), - KEYS.socketPresenceWorkflow(socketId), - ], + keys: [KEYS.roomUsers(room), KEYS.socketRooms(socketId), KEYS.socketSession(socketId)], arguments: [ socketId, updates.cursor !== undefined ? JSON.stringify(updates.cursor) : '', updates.selection !== undefined ? JSON.stringify(updates.selection) : '', (updates.lastActivity ?? Date.now()).toString(), - SOCKET_KEY_TTL.toString(), - SOCKET_PRESENCE_WORKFLOW_KEY_TTL.toString(), + SOCKET_ROOMS_TTL.toString(), + SESSION_TTL.toString(), ], }) } catch (error) { if ((error as Error).message?.includes('NOSCRIPT') && !retried) { logger.warn('Lua script not found, reloading...') this.updateActivityScriptSha = await this.redis.scriptLoad(UPDATE_ACTIVITY_SCRIPT) - return this.updateUserActivity(workflowId, socketId, updates, true) + return this.updateUserActivity(room, socketId, updates, true) } logger.error(`Failed to update user activity: ${socketId}`, error) } } - async updateRoomLastModified(workflowId: string): Promise { - await this.redis.hSet(KEYS.workflowMeta(workflowId), 'lastModified', Date.now().toString()) + async updateRoomLastModified(room: RoomRef): Promise { + await this.redis.hSet(KEYS.roomMeta(room), 'lastModified', Date.now().toString()) } - async broadcastPresenceUpdate(workflowId: string): Promise { - const users = await this.getWorkflowUsers(workflowId) - // io.to() with Redis adapter broadcasts to all pods - this._io.to(workflowId).emit('presence-update', users) + async broadcastPresenceUpdate(room: RoomRef): Promise { + const users = await this.getRoomUsers(room) + // io.to() with the Redis adapter broadcasts to all pods. + this._io.to(roomName(room)).emit(presenceEventName(room.type), users) } - emitToWorkflow(workflowId: string, event: string, payload: T): void { - this._io.to(workflowId).emit(event, payload) + emitToRoom(room: RoomRef, event: string, payload: T): void { + this._io.to(roomName(room)).emit(event, payload) } - async getUniqueUserCount(workflowId: string): Promise { - const users = await this.getWorkflowUsers(workflowId) - const uniqueUserIds = new Set(users.map((u) => u.userId)) - return uniqueUserIds.size + async getUniqueUserCount(room: RoomRef): Promise { + const users = await this.getRoomUsers(room) + return new Set(users.map((u) => u.userId)).size } async getTotalActiveConnections(): Promise { - // This is more complex with Redis - we'd need to scan all workflow:*:users keys - // For now, just count sockets in this server instance - // The true count would require aggregating across all pods + // Local instance only; the true cross-pod count would require aggregation. return this._io.sockets.sockets.size } - - async handleWorkflowDeletion(workflowId: string): Promise { - logger.info(`Handling workflow deletion notification for ${workflowId}`) - - try { - const users = await this.getWorkflowUsers(workflowId) - if (users.length === 0) { - logger.debug(`No active users found for deleted workflow ${workflowId}`) - return - } - - // Notify all clients across all pods via Redis adapter - this._io.to(workflowId).emit('workflow-deleted', { - workflowId, - message: 'This workflow has been deleted', - timestamp: Date.now(), - }) - - // Use Socket.IO's cross-pod socketsLeave() to remove all sockets from the room - // This works across all pods when using the Redis adapter - await this._io.in(workflowId).socketsLeave(workflowId) - logger.debug(`All sockets left workflow room ${workflowId} via socketsLeave()`) - - // Remove all users from Redis state - for (const user of users) { - await this.removeUserFromRoom(user.socketId, workflowId) - } - - // Clean up room data - await this.redis.del([KEYS.workflowUsers(workflowId), KEYS.workflowMeta(workflowId)]) - - logger.info( - `Cleaned up workflow room ${workflowId} after deletion (${users.length} users disconnected)` - ) - } catch (error) { - logger.error(`Failed to handle workflow deletion for ${workflowId}:`, error) - } - } - - async handleWorkflowRevert(workflowId: string, timestamp: number): Promise { - logger.info(`Handling workflow revert notification for ${workflowId}`) - - const hasRoom = await this.hasWorkflowRoom(workflowId) - if (!hasRoom) { - logger.debug(`No active room found for reverted workflow ${workflowId}`) - return - } - - this._io.to(workflowId).emit('workflow-reverted', { - workflowId, - message: 'Workflow has been reverted to deployed state', - timestamp, - }) - - await this.updateRoomLastModified(workflowId) - - const userCount = await this.getUniqueUserCount(workflowId) - logger.info(`Notified ${userCount} users about workflow revert: ${workflowId}`) - } - - async handleWorkflowUpdate(workflowId: string): Promise { - logger.info(`Handling workflow update notification for ${workflowId}`) - - const hasRoom = await this.hasWorkflowRoom(workflowId) - if (!hasRoom) { - logger.debug(`No active room found for updated workflow ${workflowId}`) - return - } - - const timestamp = Date.now() - - this._io.to(workflowId).emit('workflow-updated', { - workflowId, - message: 'Workflow has been updated externally', - timestamp, - }) - - await this.updateRoomLastModified(workflowId) - - const userCount = await this.getUniqueUserCount(workflowId) - logger.info(`Notified ${userCount} users about workflow update: ${workflowId}`) - } - - async handleWorkflowDeployed(workflowId: string): Promise { - logger.info(`Handling workflow deployed notification for ${workflowId}`) - - const hasRoom = await this.hasWorkflowRoom(workflowId) - if (!hasRoom) { - logger.debug(`No active room found for deployed workflow ${workflowId}`) - return - } - - this._io.to(workflowId).emit('workflow-deployed', { - workflowId, - timestamp: Date.now(), - }) - - const userCount = await this.getUniqueUserCount(workflowId) - logger.info(`Notified ${userCount} users about workflow deployment change: ${workflowId}`) - } } diff --git a/apps/realtime/src/rooms/types.ts b/apps/realtime/src/rooms/types.ts index 9553a427e1e..0a4024f7a11 100644 --- a/apps/realtime/src/rooms/types.ts +++ b/apps/realtime/src/rooms/types.ts @@ -1,11 +1,15 @@ +import type { RoomRef, RoomType } from '@sim/realtime-protocol/rooms' import type { Server } from 'socket.io' /** - * User presence data stored in room state + * User presence data stored in room state. + * + * `room` is the generic room address (see `@sim/realtime-protocol/rooms`). A + * socket may hold presence in more than one room, but only one room per type. */ export interface UserPresence { userId: string - workflowId: string + room: RoomRef userName: string socketId: string tabSessionId?: string @@ -18,7 +22,8 @@ export interface UserPresence { } /** - * User session data (minimal info for quick lookups) + * User session data (minimal info for quick lookups). Shared across all rooms a + * socket is in — keyed by socket, not by room. */ export interface UserSession { userId: string @@ -27,120 +32,89 @@ export interface UserSession { } /** - * Workflow room state + * Room presence state. */ -export interface WorkflowRoom { - workflowId: string +export interface RoomState { + room: RoomRef users: Map lastModified: number activeConnections: number } /** - * Common interface for room managers (in-memory and Redis) - * All methods that access state are async to support Redis operations + * Common interface for room managers (in-memory and Redis). + * + * The manager is domain-neutral: it tracks room membership and presence keyed by + * {@link RoomRef}, and knows nothing about workflows, files, or any specific + * domain. Domain lifecycle concerns (e.g. workflow deletion/deploy broadcasts) + * live in domain services that compose a manager — see `WorkflowRoomService`. + * + * A socket may occupy multiple rooms, at most one per {@link RoomType}. The + * shared session key is dropped only when a socket leaves its last room. + * + * All state-accessing methods are async to support the Redis implementation. */ export interface IRoomManager { readonly io: Server - /** - * Initialize the room manager (connect to Redis, etc.) - */ + /** Initialize the manager (connect to Redis, load scripts, etc.). */ initialize(): Promise - /** - * Whether the room manager is ready to serve requests - */ + /** Whether the manager is ready to serve requests. */ isReady(): boolean - /** - * Clean shutdown - */ + /** Clean shutdown. */ shutdown(): Promise - /** - * Add a user to a workflow room - */ - addUserToRoom(workflowId: string, socketId: string, presence: UserPresence): Promise + /** Add a socket's presence to a room. */ + addUserToRoom(room: RoomRef, socketId: string, presence: UserPresence): Promise /** - * Remove a user from their current room - * Optional workflowIdHint is used when socket mapping keys are missing/expired. - * Returns the workflowId they were in, or null if not in any room. + * Remove a socket from a single room. Returns `true` if it was a member. The + * shared session is dropped only if this was the socket's last room. */ - removeUserFromRoom(socketId: string, workflowIdHint?: string): Promise + removeUserFromRoom(room: RoomRef, socketId: string): Promise /** - * Get the workflow ID for a socket + * Remove a socket from every room it occupies (disconnect). Returns the rooms + * it was in, so the caller can rebroadcast presence per room. */ - getWorkflowIdForSocket(socketId: string): Promise + removeSocketFromAllRooms(socketId: string): Promise - /** - * Get user session data for a socket - */ + /** Every room the socket currently occupies. */ + getRoomsForSocket(socketId: string): Promise + + /** The socket's room of a given type (at most one per type), or `null`. */ + getRoomForSocket(socketId: string, type: RoomType): Promise + + /** Session data for a socket (shared across its rooms). */ getUserSession(socketId: string): Promise - /** - * Get all users in a workflow room - */ - getWorkflowUsers(workflowId: string): Promise + /** All users present in a room. */ + getRoomUsers(room: RoomRef): Promise - /** - * Check if a workflow room exists - */ - hasWorkflowRoom(workflowId: string): Promise + /** Whether a room currently has any presence. */ + hasRoom(room: RoomRef): Promise - /** - * Update user activity (cursor, selection, lastActivity) - */ + /** Update a socket's activity (cursor, selection, lastActivity) within a room. */ updateUserActivity( - workflowId: string, + room: RoomRef, socketId: string, updates: Partial> ): Promise - /** - * Update room's lastModified timestamp - */ - updateRoomLastModified(workflowId: string): Promise + /** Bump a room's lastModified timestamp. */ + updateRoomLastModified(room: RoomRef): Promise - /** - * Broadcast presence update to all clients in a workflow room - */ - broadcastPresenceUpdate(workflowId: string): Promise + /** Broadcast the room's presence list to all clients in the room. */ + broadcastPresenceUpdate(room: RoomRef): Promise - /** - * Emit an event to all clients in a workflow room - */ - emitToWorkflow(workflowId: string, event: string, payload: T): void + /** Emit an event to all clients in a room. */ + emitToRoom(room: RoomRef, event: string, payload: T): void - /** - * Get the number of unique users in a workflow room - */ - getUniqueUserCount(workflowId: string): Promise + /** Number of unique users in a room. */ + getUniqueUserCount(room: RoomRef): Promise - /** - * Get total active connections across all rooms - */ + /** Total active connections tracked by this instance. */ getTotalActiveConnections(): Promise - - /** - * Handle workflow deletion - notify users and clean up room - */ - handleWorkflowDeletion(workflowId: string): Promise - - /** - * Handle workflow revert - notify users - */ - handleWorkflowRevert(workflowId: string, timestamp: number): Promise - - /** - * Handle workflow update - notify users - */ - handleWorkflowUpdate(workflowId: string): Promise - - /** - * Handle workflow deployment change - notify users to refresh deployment state - */ - handleWorkflowDeployed(workflowId: string): Promise } diff --git a/apps/realtime/src/rooms/workflow-room-service.ts b/apps/realtime/src/rooms/workflow-room-service.ts new file mode 100644 index 00000000000..4623647a7da --- /dev/null +++ b/apps/realtime/src/rooms/workflow-room-service.ts @@ -0,0 +1,110 @@ +import { createLogger } from '@sim/logger' +import { ROOM_TYPES, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' +import type { IRoomManager } from '@/rooms/types' + +const logger = createLogger('WorkflowRoomService') + +/** The workflow room ref for a workflow id. Its Socket.IO room name is the bare id. */ +export function workflowRoom(workflowId: string): RoomRef { + return { type: ROOM_TYPES.WORKFLOW, id: workflowId } +} + +/** + * Workflow-domain lifecycle broadcasts, composed over a domain-neutral + * {@link IRoomManager}. Keeps workflow-specific concerns (deletion, revert, + * update, deploy notifications) out of the generic manager, mirroring how the + * workflow socket handlers own workflow semantics. + */ +export class WorkflowRoomService { + constructor(private readonly manager: IRoomManager) {} + + async handleWorkflowDeletion(workflowId: string): Promise { + logger.info(`Handling workflow deletion notification for ${workflowId}`) + const room = workflowRoom(workflowId) + + const users = await this.manager.getRoomUsers(room) + if (users.length === 0) { + logger.debug(`No active users found for deleted workflow ${workflowId}`) + return + } + + this.manager.emitToRoom(room, 'workflow-deleted', { + workflowId, + message: 'This workflow has been deleted', + timestamp: Date.now(), + }) + + // Remove every socket from the Socket.IO room (cross-pod via the Redis adapter). + const name = roomName(room) + await this.manager.io.in(name).socketsLeave(name) + + // Drop presence state for each socket; empty-room cleanup is handled by the manager. + for (const user of users) { + await this.manager.removeUserFromRoom(room, user.socketId) + } + + logger.info( + `Cleaned up workflow room ${workflowId} after deletion (${users.length} users disconnected)` + ) + } + + async handleWorkflowRevert(workflowId: string, timestamp: number): Promise { + logger.info(`Handling workflow revert notification for ${workflowId}`) + const room = workflowRoom(workflowId) + + if (!(await this.manager.hasRoom(room))) { + logger.debug(`No active room found for reverted workflow ${workflowId}`) + return + } + + this.manager.emitToRoom(room, 'workflow-reverted', { + workflowId, + message: 'Workflow has been reverted to deployed state', + timestamp, + }) + + await this.manager.updateRoomLastModified(room) + + const userCount = await this.manager.getUniqueUserCount(room) + logger.info(`Notified ${userCount} users about workflow revert: ${workflowId}`) + } + + async handleWorkflowUpdate(workflowId: string): Promise { + logger.info(`Handling workflow update notification for ${workflowId}`) + const room = workflowRoom(workflowId) + + if (!(await this.manager.hasRoom(room))) { + logger.debug(`No active room found for updated workflow ${workflowId}`) + return + } + + this.manager.emitToRoom(room, 'workflow-updated', { + workflowId, + message: 'Workflow has been updated externally', + timestamp: Date.now(), + }) + + await this.manager.updateRoomLastModified(room) + + const userCount = await this.manager.getUniqueUserCount(room) + logger.info(`Notified ${userCount} users about workflow update: ${workflowId}`) + } + + async handleWorkflowDeployed(workflowId: string): Promise { + logger.info(`Handling workflow deployed notification for ${workflowId}`) + const room = workflowRoom(workflowId) + + if (!(await this.manager.hasRoom(room))) { + logger.debug(`No active room found for deployed workflow ${workflowId}`) + return + } + + this.manager.emitToRoom(room, 'workflow-deployed', { + workflowId, + timestamp: Date.now(), + }) + + const userCount = await this.manager.getUniqueUserCount(room) + logger.info(`Notified ${userCount} users about workflow deployment change: ${workflowId}`) + } +} diff --git a/apps/realtime/src/routes/http.ts b/apps/realtime/src/routes/http.ts index 0f8ed73cc52..233bee4d174 100644 --- a/apps/realtime/src/routes/http.ts +++ b/apps/realtime/src/routes/http.ts @@ -1,7 +1,7 @@ import type { IncomingMessage, ServerResponse } from 'http' import { safeCompare } from '@sim/security/compare' import { env } from '@/env' -import type { IRoomManager } from '@/rooms' +import { type IRoomManager, WorkflowRoomService } from '@/rooms' interface Logger { info: (message: string, ...args: unknown[]) => void @@ -58,6 +58,8 @@ function sendError(res: ServerResponse, message: string, status = 500): void { * @returns HTTP request handler function */ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { + const workflowRoomService = new WorkflowRoomService(roomManager) + return async (req: IncomingMessage, res: ServerResponse) => { // Health check doesn't require auth if (req.method === 'GET' && req.url === '/health') { @@ -99,7 +101,7 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { try { const body = await readRequestBody(req) const { workflowId } = JSON.parse(body) - await roomManager.handleWorkflowDeletion(workflowId) + await workflowRoomService.handleWorkflowDeletion(workflowId) sendSuccess(res) } catch (error) { logger.error('Error handling workflow deletion notification:', error) @@ -113,7 +115,7 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { try { const body = await readRequestBody(req) const { workflowId } = JSON.parse(body) - await roomManager.handleWorkflowUpdate(workflowId) + await workflowRoomService.handleWorkflowUpdate(workflowId) sendSuccess(res) } catch (error) { logger.error('Error handling workflow update notification:', error) @@ -127,7 +129,7 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { try { const body = await readRequestBody(req) const { workflowId } = JSON.parse(body) - await roomManager.handleWorkflowDeployed(workflowId) + await workflowRoomService.handleWorkflowDeployed(workflowId) sendSuccess(res) } catch (error) { logger.error('Error handling workflow deployed notification:', error) @@ -141,7 +143,7 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { try { const body = await readRequestBody(req) const { workflowId, timestamp } = JSON.parse(body) - await roomManager.handleWorkflowRevert(workflowId, timestamp) + await workflowRoomService.handleWorkflowRevert(workflowId, timestamp) sendSuccess(res) } catch (error) { logger.error('Error handling workflow revert notification:', error) diff --git a/packages/realtime-protocol/src/rooms.ts b/packages/realtime-protocol/src/rooms.ts index 994bf6b3cac..c03ad621d45 100644 --- a/packages/realtime-protocol/src/rooms.ts +++ b/packages/realtime-protocol/src/rooms.ts @@ -87,3 +87,13 @@ export function parseRoomName(name: string): RoomRef | null { export function isSameRoom(a: RoomRef, b: RoomRef): boolean { return a.type === b.type && a.id === b.id } + +/** + * The `presence-update` broadcast event name for a room type. `WORKFLOW` keeps + * the historical bare `presence-update` name (client backward-compat); every + * other type is namespaced so a socket joined to more than one room can tell the + * presence streams apart on a single connection. + */ +export function presenceEventName(type: RoomType): string { + return type === ROOM_TYPES.WORKFLOW ? 'presence-update' : `${type}:presence-update` +} From 1fee9a74eb31a10e0a5696c7bb96a7d4bc5eea13 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 11:31:10 -0700 Subject: [PATCH 2/5] fix(realtime): harden multi-room disconnect + id-guard room removal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two fixes from an adversarial regression audit of the multi-room refactor: - Disconnect now handles `disconnecting` (where `socket.rooms` is still populated and authoritative) and falls back to the live Socket.IO room set for any room the manager's stored state no longer tracked. This restores reliable presence cleanup + departure broadcast even if the Redis `socket:{id}:rooms` key was evicted or TTL-expired — the one behavioral gap vs the pre-refactor disconnect. - REMOVE_ROOM_SCRIPT now only drops the socket's room mapping (and runs the last-room session cleanup) when the stored id matches the room being removed, matching the memory manager's existing id guard. Prevents a mismatched-room call from wiping a different room's mapping or the shared session. +1 test (id-guarded no-op removal). 120 realtime tests pass, tsc clean. --- apps/realtime/src/handlers/connection.ts | 29 +++++++++++++++---- .../realtime/src/rooms/memory-manager.test.ts | 12 ++++++++ apps/realtime/src/rooms/redis-manager.ts | 16 ++++++---- 3 files changed, 46 insertions(+), 11 deletions(-) diff --git a/apps/realtime/src/handlers/connection.ts b/apps/realtime/src/handlers/connection.ts index c59f3d74ee7..c413a6edaa3 100644 --- a/apps/realtime/src/handlers/connection.ts +++ b/apps/realtime/src/handlers/connection.ts @@ -1,4 +1,5 @@ import { createLogger } from '@sim/logger' +import { parseRoomName, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' import { cleanupPendingSubblocksForSocket } from '@/handlers/subblocks' import { cleanupPendingVariablesForSocket } from '@/handlers/variables' import type { AuthenticatedSocket } from '@/middleware/auth' @@ -15,22 +16,38 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager logger.error(`Socket ${socket.id} connection error:`, error) }) - socket.on('disconnect', async (reason) => { + // `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) => { try { // Clean up pending debounce entries for this socket to prevent memory leaks cleanupPendingSubblocksForSocket(socket.id) cleanupPendingVariablesForSocket(socket.id) - // A socket may occupy multiple rooms (one per type). Remove it from all of - // them and rebroadcast presence per room so no room leaks a stale entry. + // A socket may occupy multiple rooms (one per type). Remove it from every + // room the manager knows about. const removedRooms = await roomManager.removeSocketFromAllRooms(socket.id) + const handled = new Set(removedRooms.map((room) => roomName(room))) - for (const room of removedRooms) { + // Fallback: clean up any room the live Socket.IO membership still lists that + // the manager's (possibly-evicted) state no longer tracked. + const fallbackRooms: RoomRef[] = [] + for (const name of socket.rooms) { + if (name === socket.id || handled.has(name)) continue + const ref = parseRoomName(name) + if (!ref) continue + await roomManager.removeUserFromRoom(ref, socket.id) + fallbackRooms.push(ref) + } + + const allRooms = [...removedRooms, ...fallbackRooms] + for (const room of allRooms) { await roomManager.broadcastPresenceUpdate(room) } - if (removedRooms.length > 0) { - const rooms = removedRooms.map((room) => `${room.type}:${room.id}`).join(', ') + if (allRooms.length > 0) { + const rooms = allRooms.map((room) => `${room.type}:${room.id}`).join(', ') logger.info(`Socket ${socket.id} disconnected from [${rooms}] (reason: ${reason})`) } } catch (error) { diff --git a/apps/realtime/src/rooms/memory-manager.test.ts b/apps/realtime/src/rooms/memory-manager.test.ts index fe57009170d..c0d968df7b4 100644 --- a/apps/realtime/src/rooms/memory-manager.test.ts +++ b/apps/realtime/src/rooms/memory-manager.test.ts @@ -115,6 +115,18 @@ describe('MemoryRoomManager multi-room', () => { expect(await manager.getUserSession('socket-2')).not.toBeNull() }) + it('ignores removal of a room the socket is not in (id-guarded)', async () => { + await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + + // Removing a workflow room the socket never joined must be a no-op — it must + // not wipe the files mapping or the shared session. + const removed = await manager.removeUserFromRoom(WORKFLOW, 'socket-1') + expect(removed).toBe(false) + expect(await manager.hasRoom(FILES)).toBe(true) + expect(await manager.getUserSession('socket-1')).not.toBeNull() + expect(await manager.getRoomForSocket('socket-1', ROOM_TYPES.WORKSPACE_FILES)).toEqual(FILES) + }) + it('broadcasts presence on the room-type-specific event name', async () => { const { emit, io } = fakeIo() const m = new MemoryRoomManager(io) diff --git a/apps/realtime/src/rooms/redis-manager.ts b/apps/realtime/src/rooms/redis-manager.ts index 6dd6d1f3ceb..9df818398fe 100644 --- a/apps/realtime/src/rooms/redis-manager.ts +++ b/apps/realtime/src/rooms/redis-manager.ts @@ -40,7 +40,7 @@ const SESSION_TTL = 60 * 60 * one room would break the socket's other rooms). Cleans up empty room state. * * KEYS: [socketRooms, socketSession, roomUsers, roomMeta] - * ARGV: [roomType, socketId] + * ARGV: [roomType, socketId, roomId] * Returns 1 if the socket was a member of the room, else 0. */ const REMOVE_ROOM_SCRIPT = ` @@ -50,12 +50,18 @@ local roomUsersKey = KEYS[3] local roomMetaKey = KEYS[4] local roomType = ARGV[1] local socketId = ARGV[2] +local roomId = ARGV[3] local removed = redis.call('HDEL', roomUsersKey, socketId) -redis.call('HDEL', socketRoomsKey, roomType) -if redis.call('HLEN', socketRoomsKey) == 0 then - redis.call('DEL', socketRoomsKey, socketSessionKey) +-- Only drop the socket's mapping for this type if it points at THIS room, so +-- removing a room the socket isn't in can't wipe a different room's mapping or +-- spuriously trigger the last-room session cleanup (mirrors the memory manager). +if redis.call('HGET', socketRoomsKey, roomType) == roomId then + redis.call('HDEL', socketRoomsKey, roomType) + if redis.call('HLEN', socketRoomsKey) == 0 then + redis.call('DEL', socketRoomsKey, socketSessionKey) + end end if redis.call('HLEN', roomUsersKey) == 0 then @@ -220,7 +226,7 @@ export class RedisRoomManager implements IRoomManager { KEYS.roomUsers(room), KEYS.roomMeta(room), ], - arguments: [room.type, socketId], + arguments: [room.type, socketId, room.id], }) return typeof removed === 'number' ? removed > 0 : Number(removed) > 0 } catch (error) { From 7f70ab729cb03595fc6a75f7d4b454dc5232f466 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 11:47:53 -0700 Subject: [PATCH 3/5] fix(realtime): only rebroadcast disconnect-fallback rooms whose removal succeeded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Greptile 4/5 follow-up: the disconnecting-time fallback ignored removeUserFromRoom's boolean and rebroadcast presence even when the removal reported false. Now it only treats a room as removed (and rebroadcasts) when the manager confirms it — symmetric with removeSocketFromAllRooms, which already only returns rooms it actually removed. --- apps/realtime/src/handlers/connection.ts | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/apps/realtime/src/handlers/connection.ts b/apps/realtime/src/handlers/connection.ts index c413a6edaa3..185bd9c253f 100644 --- a/apps/realtime/src/handlers/connection.ts +++ b/apps/realtime/src/handlers/connection.ts @@ -31,14 +31,19 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager const handled = new Set(removedRooms.map((room) => roomName(room))) // Fallback: clean up any room the live Socket.IO membership still lists that - // the manager's (possibly-evicted) state no longer tracked. + // the manager's (possibly-evicted) state no longer tracked. Only treat a room + // as removed when the manager confirms it (returns true) — symmetric with + // `removeSocketFromAllRooms`, which only reports rooms it actually removed. A + // false result (already absent, or a transient Redis error) is not rebroadcast + // here: on a real error `broadcastPresenceUpdate` would emit a stale list, and + // any orphaned entry is reclaimed by the next join's stale-presence sweep. const fallbackRooms: RoomRef[] = [] for (const name of socket.rooms) { if (name === socket.id || handled.has(name)) continue const ref = parseRoomName(name) if (!ref) continue - await roomManager.removeUserFromRoom(ref, socket.id) - fallbackRooms.push(ref) + const removed = await roomManager.removeUserFromRoom(ref, socket.id) + if (removed) fallbackRooms.push(ref) } const allRooms = [...removedRooms, ...fallbackRooms] From 7f9610c52aa771004917c31482cb6d722e98839e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 11:59:44 -0700 Subject: [PATCH 4/5] fix(realtime): exclude the disconnecting socket from its farewell broadcast Greptile follow-up (transient-Redis-failure edge): if removeUserFromRoom fails on disconnect, the socket's presence entry can outlive it (room hashes have no TTL) and reappear as a ghost. Disconnect now broadcasts a correction to EVERY room the socket was in (union of the manager's removed rooms and the live Socket.IO membership) and passes the disconnecting socket id as excludeSocketId, so it is never shown as a collaborator regardless of whether the Redis delete succeeded. Any orphaned entry is still reclaimed by the next join's stale-presence sweep. broadcastPresenceUpdate gains an optional excludeSocketId; normal broadcasts are unchanged. +1 test. --- apps/realtime/src/handlers/connection.ts | 43 +++++++++++-------- .../realtime/src/rooms/memory-manager.test.ts | 14 ++++++ apps/realtime/src/rooms/memory-manager.ts | 5 ++- apps/realtime/src/rooms/redis-manager.ts | 5 ++- apps/realtime/src/rooms/types.ts | 9 +++- 5 files changed, 51 insertions(+), 25 deletions(-) diff --git a/apps/realtime/src/handlers/connection.ts b/apps/realtime/src/handlers/connection.ts index 185bd9c253f..ecb2665fe50 100644 --- a/apps/realtime/src/handlers/connection.ts +++ b/apps/realtime/src/handlers/connection.ts @@ -1,5 +1,5 @@ import { createLogger } from '@sim/logger' -import { parseRoomName, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' +import { isSameRoom, parseRoomName, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' import { cleanupPendingSubblocksForSocket } from '@/handlers/subblocks' import { cleanupPendingVariablesForSocket } from '@/handlers/variables' import type { AuthenticatedSocket } from '@/middleware/auth' @@ -28,31 +28,36 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager // A socket may occupy multiple rooms (one per type). Remove it from every // room the manager knows about. const removedRooms = await roomManager.removeSocketFromAllRooms(socket.id) - const handled = new Set(removedRooms.map((room) => roomName(room))) - - // Fallback: clean up any room the live Socket.IO membership still lists that - // the manager's (possibly-evicted) state no longer tracked. Only treat a room - // as removed when the manager confirms it (returns true) — symmetric with - // `removeSocketFromAllRooms`, which only reports rooms it actually removed. A - // false result (already absent, or a transient Redis error) is not rebroadcast - // here: on a real error `broadcastPresenceUpdate` would emit a stale list, and - // any orphaned entry is reclaimed by the next join's stale-presence sweep. - const fallbackRooms: RoomRef[] = [] + + // Union with the live Socket.IO membership (authoritative here, and it + // survives a Redis eviction/TTL lapse that would leave the manager's tracked + // rooms empty). Attempt removal for any room the manager didn't already + // remove — best-effort, since a transient Redis error can't be recovered here. + const wasInRooms = new Map() + for (const room of removedRooms) wasInRooms.set(roomName(room), room) for (const name of socket.rooms) { - if (name === socket.id || handled.has(name)) continue + if (name === socket.id || wasInRooms.has(name)) continue const ref = parseRoomName(name) if (!ref) continue - const removed = await roomManager.removeUserFromRoom(ref, socket.id) - if (removed) fallbackRooms.push(ref) + wasInRooms.set(name, ref) + if (!removedRooms.some((room) => isSameRoom(room, ref))) { + await roomManager.removeUserFromRoom(ref, socket.id) + } } - const allRooms = [...removedRooms, ...fallbackRooms] - for (const room of allRooms) { - await roomManager.broadcastPresenceUpdate(room) + // Broadcast a correction to every room this socket was in, EXCLUDING this + // socket — so it is never shown as a ghost collaborator even if its presence + // entry outlived a failed removal (transient Redis error; the hashes have no + // TTL). Any orphaned entry is additionally reclaimed by the next join's + // stale-presence sweep. + for (const room of wasInRooms.values()) { + await roomManager.broadcastPresenceUpdate(room, socket.id) } - if (allRooms.length > 0) { - const rooms = allRooms.map((room) => `${room.type}:${room.id}`).join(', ') + if (wasInRooms.size > 0) { + const rooms = Array.from(wasInRooms.values()) + .map((room) => `${room.type}:${room.id}`) + .join(', ') logger.info(`Socket ${socket.id} disconnected from [${rooms}] (reason: ${reason})`) } } catch (error) { diff --git a/apps/realtime/src/rooms/memory-manager.test.ts b/apps/realtime/src/rooms/memory-manager.test.ts index c0d968df7b4..687fd663242 100644 --- a/apps/realtime/src/rooms/memory-manager.test.ts +++ b/apps/realtime/src/rooms/memory-manager.test.ts @@ -139,4 +139,18 @@ describe('MemoryRoomManager multi-room', () => { await m.broadcastPresenceUpdate(WORKFLOW) expect(emit).toHaveBeenCalledWith('presence-update', expect.any(Array)) }) + + it('omits an excluded socket from the presence broadcast (disconnect ghost guard)', async () => { + const { emit, io } = fakeIo() + const m = new MemoryRoomManager(io) + await m.initialize() + await m.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + await m.addUserToRoom(FILES, 'socket-2', presence(FILES, 'socket-2', 'user-2')) + + // Broadcast as if socket-1 is disconnecting: even though its presence entry is + // still present, it must not appear in the emitted list. + await m.broadcastPresenceUpdate(FILES, 'socket-1') + const emitted = emit.mock.calls.at(-1)?.[1] as Array<{ socketId: string }> + expect(emitted.map((u) => u.socketId)).toEqual(['socket-2']) + }) }) diff --git a/apps/realtime/src/rooms/memory-manager.ts b/apps/realtime/src/rooms/memory-manager.ts index a7279842249..b1490acfe20 100644 --- a/apps/realtime/src/rooms/memory-manager.ts +++ b/apps/realtime/src/rooms/memory-manager.ts @@ -166,9 +166,10 @@ export class MemoryRoomManager implements IRoomManager { if (state) state.lastModified = Date.now() } - async broadcastPresenceUpdate(room: RoomRef): Promise { + async broadcastPresenceUpdate(room: RoomRef, excludeSocketId?: string): Promise { const users = await this.getRoomUsers(room) - this._io.to(roomName(room)).emit(presenceEventName(room.type), users) + const visible = excludeSocketId ? users.filter((u) => u.socketId !== excludeSocketId) : users + this._io.to(roomName(room)).emit(presenceEventName(room.type), visible) } emitToRoom(room: RoomRef, event: string, payload: T): void { diff --git a/apps/realtime/src/rooms/redis-manager.ts b/apps/realtime/src/rooms/redis-manager.ts index 9df818398fe..e02de256b77 100644 --- a/apps/realtime/src/rooms/redis-manager.ts +++ b/apps/realtime/src/rooms/redis-manager.ts @@ -347,10 +347,11 @@ export class RedisRoomManager implements IRoomManager { await this.redis.hSet(KEYS.roomMeta(room), 'lastModified', Date.now().toString()) } - async broadcastPresenceUpdate(room: RoomRef): Promise { + async broadcastPresenceUpdate(room: RoomRef, excludeSocketId?: string): Promise { const users = await this.getRoomUsers(room) + const visible = excludeSocketId ? users.filter((u) => u.socketId !== excludeSocketId) : users // io.to() with the Redis adapter broadcasts to all pods. - this._io.to(roomName(room)).emit(presenceEventName(room.type), users) + this._io.to(roomName(room)).emit(presenceEventName(room.type), visible) } emitToRoom(room: RoomRef, event: string, payload: T): void { diff --git a/apps/realtime/src/rooms/types.ts b/apps/realtime/src/rooms/types.ts index 0a4024f7a11..67cee450832 100644 --- a/apps/realtime/src/rooms/types.ts +++ b/apps/realtime/src/rooms/types.ts @@ -106,8 +106,13 @@ export interface IRoomManager { /** Bump a room's lastModified timestamp. */ updateRoomLastModified(room: RoomRef): Promise - /** Broadcast the room's presence list to all clients in the room. */ - broadcastPresenceUpdate(room: RoomRef): Promise + /** + * Broadcast the room's presence list to all clients in the room. Pass + * `excludeSocketId` (e.g. a disconnecting socket) to omit that socket from the + * broadcast even if its presence entry outlived a failed removal — so it is + * never shown as a ghost collaborator. + */ + broadcastPresenceUpdate(room: RoomRef, excludeSocketId?: string): Promise /** Emit an event to all clients in a room. */ emitToRoom(room: RoomRef, event: string, payload: T): void From 21298323214ec4922b861e91a438dcb593d7b4a5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 12:10:41 -0700 Subject: [PATCH 5/5] fix(realtime): make presence broadcasts liveness-aware (root-cause ghost fix) Presence broadcasts now reconcile the stored list against the live Socket.IO membership (io.in(room).fetchSockets()) before emitting, via a shared filterVisiblePresence helper. This closes the residual behind the earlier disconnect fixes: an entry orphaned by a failed removal (room hashes have no TTL) could reappear in a LATER join's presence snapshot until the 75-min stale sweep. Now such an entry is never emitted, because a non-live socket is filtered out of every broadcast. Combined with excludeSocketId (which handles the disconnecting socket, still momentarily live). Fail-safe: on a fetchSockets throw or an empty result while entries remain, emit the unfiltered list rather than hide live collaborators. Also drops a dead guard in the disconnect union loop (rooms already removed are skipped by the wasInRooms check) and the now-unused isSameRoom import. +1 ghost-guard test. 122 realtime tests pass. --- apps/realtime/src/handlers/connection.ts | 9 ++--- .../realtime/src/rooms/memory-manager.test.ts | 23 ++++++++++-- apps/realtime/src/rooms/memory-manager.ts | 3 +- .../realtime/src/rooms/presence-visibility.ts | 36 +++++++++++++++++++ apps/realtime/src/rooms/redis-manager.ts | 3 +- 5 files changed, 66 insertions(+), 8 deletions(-) create mode 100644 apps/realtime/src/rooms/presence-visibility.ts diff --git a/apps/realtime/src/handlers/connection.ts b/apps/realtime/src/handlers/connection.ts index ecb2665fe50..2f180e7fa53 100644 --- a/apps/realtime/src/handlers/connection.ts +++ b/apps/realtime/src/handlers/connection.ts @@ -1,5 +1,5 @@ import { createLogger } from '@sim/logger' -import { isSameRoom, parseRoomName, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' +import { parseRoomName, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' import { cleanupPendingSubblocksForSocket } from '@/handlers/subblocks' import { cleanupPendingVariablesForSocket } from '@/handlers/variables' import type { AuthenticatedSocket } from '@/middleware/auth' @@ -36,13 +36,14 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager const wasInRooms = new Map() for (const room of removedRooms) wasInRooms.set(roomName(room), room) for (const name of socket.rooms) { + // `wasInRooms.has(name)` already excludes every room the manager removed + // (same room-name key via the roomName/parseRoomName bijection), so any + // room reaching here was NOT in `removedRooms` and needs a removal attempt. if (name === socket.id || wasInRooms.has(name)) continue const ref = parseRoomName(name) if (!ref) continue wasInRooms.set(name, ref) - if (!removedRooms.some((room) => isSameRoom(room, ref))) { - await roomManager.removeUserFromRoom(ref, socket.id) - } + await roomManager.removeUserFromRoom(ref, socket.id) } // Broadcast a correction to every room this socket was in, EXCLUDING this diff --git a/apps/realtime/src/rooms/memory-manager.test.ts b/apps/realtime/src/rooms/memory-manager.test.ts index 687fd663242..d5d2512a2d8 100644 --- a/apps/realtime/src/rooms/memory-manager.test.ts +++ b/apps/realtime/src/rooms/memory-manager.test.ts @@ -10,11 +10,16 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' import { MemoryRoomManager } from '@/rooms/memory-manager' import type { UserPresence } from '@/rooms/types' -function fakeIo() { +function fakeIo(liveSocketIds: string[] = []) { const emit = vi.fn() return { emit, - io: { to: vi.fn().mockReturnValue({ emit }) } as never, + io: { + to: vi.fn().mockReturnValue({ emit }), + in: vi.fn().mockReturnValue({ + fetchSockets: vi.fn().mockResolvedValue(liveSocketIds.map((id) => ({ id }))), + }), + } as never, } } @@ -153,4 +158,18 @@ describe('MemoryRoomManager multi-room', () => { const emitted = emit.mock.calls.at(-1)?.[1] as Array<{ socketId: string }> expect(emitted.map((u) => u.socketId)).toEqual(['socket-2']) }) + + it('never emits a presence entry whose socket is no longer live (ghost guard)', async () => { + // Only socket-2 is a live Socket.IO member; socket-1 is an orphaned entry that + // outlived a failed removal. + const { emit, io } = fakeIo(['socket-2']) + const m = new MemoryRoomManager(io) + await m.initialize() + await m.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1')) + await m.addUserToRoom(FILES, 'socket-2', presence(FILES, 'socket-2', 'user-2')) + + await m.broadcastPresenceUpdate(FILES) + const emitted = emit.mock.calls.at(-1)?.[1] as Array<{ socketId: string }> + expect(emitted.map((u) => u.socketId)).toEqual(['socket-2']) + }) }) diff --git a/apps/realtime/src/rooms/memory-manager.ts b/apps/realtime/src/rooms/memory-manager.ts index b1490acfe20..0e25921a47e 100644 --- a/apps/realtime/src/rooms/memory-manager.ts +++ b/apps/realtime/src/rooms/memory-manager.ts @@ -6,6 +6,7 @@ import { roomName, } from '@sim/realtime-protocol/rooms' import type { Server } from 'socket.io' +import { filterVisiblePresence } from '@/rooms/presence-visibility' import type { IRoomManager, RoomState, UserPresence, UserSession } from '@/rooms/types' const logger = createLogger('MemoryRoomManager') @@ -168,7 +169,7 @@ export class MemoryRoomManager implements IRoomManager { async broadcastPresenceUpdate(room: RoomRef, excludeSocketId?: string): Promise { const users = await this.getRoomUsers(room) - const visible = excludeSocketId ? users.filter((u) => u.socketId !== excludeSocketId) : users + const visible = await filterVisiblePresence(this._io, room, users, excludeSocketId) this._io.to(roomName(room)).emit(presenceEventName(room.type), visible) } diff --git a/apps/realtime/src/rooms/presence-visibility.ts b/apps/realtime/src/rooms/presence-visibility.ts new file mode 100644 index 00000000000..4680a581964 --- /dev/null +++ b/apps/realtime/src/rooms/presence-visibility.ts @@ -0,0 +1,36 @@ +import { type RoomRef, roomName } from '@sim/realtime-protocol/rooms' +import type { Server } from 'socket.io' + +/** + * Filters a room's stored presence down to what should actually be broadcast: + * drops `excludeSocketId` (e.g. a socket mid-disconnect that is still connected), + * then reconciles against the live Socket.IO membership so an entry orphaned by a + * failed removal (the room hashes have no TTL) is never emitted as a ghost. + * + * Fail-safe: if the liveness lookup throws, or returns an empty set while we still + * hold presence entries (a cross-pod `fetchSockets` timeout, not a truly empty + * room), we emit the unfiltered list rather than hide everyone — a transient + * ghost self-corrects on the next broadcast, but hiding live collaborators would + * be a worse, visible failure. + */ +export async function filterVisiblePresence( + io: Server, + room: RoomRef, + users: T[], + excludeSocketId?: string +): Promise { + const candidates = excludeSocketId + ? users.filter((user) => user.socketId !== excludeSocketId) + : users + + try { + const liveSockets = await io.in(roomName(room)).fetchSockets() + if (liveSockets.length === 0) { + return candidates + } + const liveIds = new Set(liveSockets.map((socket) => socket.id)) + return candidates.filter((user) => liveIds.has(user.socketId)) + } catch { + return candidates + } +} diff --git a/apps/realtime/src/rooms/redis-manager.ts b/apps/realtime/src/rooms/redis-manager.ts index e02de256b77..2de65bc14c5 100644 --- a/apps/realtime/src/rooms/redis-manager.ts +++ b/apps/realtime/src/rooms/redis-manager.ts @@ -7,6 +7,7 @@ import { } from '@sim/realtime-protocol/rooms' import { createClient, type RedisClientType } from 'redis' import type { Server } from 'socket.io' +import { filterVisiblePresence } from '@/rooms/presence-visibility' import type { IRoomManager, UserPresence, UserSession } from '@/rooms/types' const logger = createLogger('RedisRoomManager') @@ -349,7 +350,7 @@ export class RedisRoomManager implements IRoomManager { async broadcastPresenceUpdate(room: RoomRef, excludeSocketId?: string): Promise { const users = await this.getRoomUsers(room) - const visible = excludeSocketId ? users.filter((u) => u.socketId !== excludeSocketId) : users + const visible = await filterVisiblePresence(this._io, room, users, excludeSocketId) // io.to() with the Redis adapter broadcasts to all pods. this._io.to(roomName(room)).emit(presenceEventName(room.type), visible) }