diff --git a/apps/realtime/src/handlers/connection.ts b/apps/realtime/src/handlers/connection.ts index 90eddb82464..2f180e7fa53 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,20 +16,50 @@ 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) - 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 every + // room the manager knows about. + const removedRooms = await roomManager.removeSocketFromAllRooms(socket.id) - if (workflowId) { - await roomManager.broadcastPresenceUpdate(workflowId) - logger.info( - `Socket ${socket.id} disconnected from workflow ${workflowId} (reason: ${reason})` - ) + // 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) { + // `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) + await roomManager.removeUserFromRoom(ref, socket.id) + } + + // 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 (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) { 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..d5d2512a2d8 --- /dev/null +++ b/apps/realtime/src/rooms/memory-manager.test.ts @@ -0,0 +1,175 @@ +/** + * 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(liveSocketIds: string[] = []) { + const emit = vi.fn() + return { + emit, + io: { + to: vi.fn().mockReturnValue({ emit }), + in: vi.fn().mockReturnValue({ + fetchSockets: vi.fn().mockResolvedValue(liveSocketIds.map((id) => ({ id }))), + }), + } 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('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) + 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)) + }) + + 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']) + }) + + 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 a032e785bb5..0e25921a47e 100644 --- a/apps/realtime/src/rooms/memory-manager.ts +++ b/apps/realtime/src/rooms/memory-manager.ts @@ -1,16 +1,30 @@ 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 { filterVisiblePresence } from '@/rooms/presence-visibility' +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 +45,151 @@ 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, excludeSocketId?: string): Promise { + const users = await this.getRoomUsers(room) + const visible = await filterVisiblePresence(this._io, room, users, excludeSocketId) + this._io.to(roomName(room)).emit(presenceEventName(room.type), visible) } - 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/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 0e6b3eadf2b..2de65bc14c5 100644 --- a/apps/realtime/src/rooms/redis-manager.ts +++ b/apps/realtime/src/rooms/redis-manager.ts @@ -1,86 +1,102 @@ 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 { filterVisiblePresence } from '@/rooms/presence-visibility' 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, roomId] + * 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 - -if not workflowId and workflowIdHint ~= '' then - workflowId = workflowIdHint +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) + +-- 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 not workflowId then - return nil +if redis.call('HLEN', roomUsersKey) == 0 then + redis.call('DEL', roomUsersKey, roomMetaKey) 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) -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 +105,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 +159,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 +171,6 @@ export class RedisRoomManager implements IRoomManager { async shutdown(): Promise { if (!this.isConnected) return - try { await this.redis.quit() this.isConnected = false @@ -171,93 +180,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, room.id], }) - - 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 +287,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 +301,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 +324,48 @@ 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, excludeSocketId?: string): Promise { + const users = await this.getRoomUsers(room) + 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) } - 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..67cee450832 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,94 @@ 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 + * 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(workflowId: string): Promise + broadcastPresenceUpdate(room: RoomRef, excludeSocketId?: string): 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` +}