diff --git a/apps/realtime/src/handlers/avatar.ts b/apps/realtime/src/handlers/avatar.ts new file mode 100644 index 00000000000..349b166e270 --- /dev/null +++ b/apps/realtime/src/handlers/avatar.ts @@ -0,0 +1,29 @@ +import { db, user } from '@sim/db' +import { createLogger } from '@sim/logger' +import { eq } from 'drizzle-orm' +import type { AuthenticatedSocket } from '@/middleware/auth' + +const logger = createLogger('PresenceAvatar') + +/** + * The avatar URL for a presence entry: the socket's authenticated image when + * present, otherwise a single lookup of the user's stored image. Never throws — + * presence must not fail on an avatar lookup, so a DB error resolves to `null`. + */ +export async function resolveAvatarUrl( + socket: AuthenticatedSocket, + userId: string +): Promise { + if (socket.userImage) return socket.userImage + try { + const [record] = await db + .select({ image: user.image }) + .from(user) + .where(eq(user.id, userId)) + .limit(1) + return record?.image ?? null + } catch (error) { + logger.warn('Failed to load user avatar for presence', { userId, error }) + return null + } +} diff --git a/apps/realtime/src/handlers/index.ts b/apps/realtime/src/handlers/index.ts index 9e034d77db1..485e2853170 100644 --- a/apps/realtime/src/handlers/index.ts +++ b/apps/realtime/src/handlers/index.ts @@ -3,6 +3,7 @@ import { setupWorkspaceFileDocHandlers } from '@/handlers/file-doc' import { setupOperationsHandlers } from '@/handlers/operations' import { setupPresenceHandlers } from '@/handlers/presence' import { setupSubblocksHandlers } from '@/handlers/subblocks' +import { setupTablesHandlers } from '@/handlers/tables' import { setupVariablesHandlers } from '@/handlers/variables' import { setupWorkflowHandlers } from '@/handlers/workflow' import { setupWorkspaceFilesHandlers } from '@/handlers/workspace-files' @@ -17,5 +18,6 @@ export function setupAllHandlers(socket: AuthenticatedSocket, roomManager: IRoom setupPresenceHandlers(socket, roomManager) setupWorkspaceFilesHandlers(socket, roomManager) setupWorkspaceFileDocHandlers(socket, roomManager) + setupTablesHandlers(socket, roomManager) setupConnectionHandlers(socket, roomManager) } diff --git a/apps/realtime/src/handlers/tables.test.ts b/apps/realtime/src/handlers/tables.test.ts new file mode 100644 index 00000000000..2e47099f09a --- /dev/null +++ b/apps/realtime/src/handlers/tables.test.ts @@ -0,0 +1,234 @@ +/** + * @vitest-environment node + */ +import { ROOM_TYPES } from '@sim/realtime-protocol/rooms' +import { TABLE_PRESENCE_EVENTS } from '@sim/realtime-protocol/table-presence' +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { IRoomManager } from '@/rooms' + +const { mockAuthorizeRoom } = vi.hoisted(() => ({ + mockAuthorizeRoom: vi.fn(), +})) + +vi.mock('@sim/db', () => ({ + db: { select: vi.fn() }, + user: { image: 'image' }, +})) + +vi.mock('@sim/platform-authz/rooms', () => ({ + authorizeRoom: mockAuthorizeRoom, +})) + +import { setupTablesHandlers } from '@/handlers/tables' + +const TABLE_ROOM = { type: ROOM_TYPES.TABLE, id: 'table-1' } + +function createSocket(overrides?: Record) { + const handlers: Record Promise | void> = {} + const toEmit = vi.fn() + const socket = { + id: 'socket-1', + userId: 'user-1', + userName: 'Test User', + userImage: 'avatar.png', + on: vi.fn((event: string, handler: (payload: unknown) => Promise | void) => { + handlers[event] = handler + }), + emit: vi.fn(), + join: vi.fn(), + leave: vi.fn(), + to: vi.fn().mockReturnValue({ emit: toEmit }), + ...overrides, + } + return { handlers, socket, toEmit } +} + +function createRoomManager(overrides?: Partial): IRoomManager { + return { + isReady: vi.fn().mockReturnValue(true), + getRoomForSocket: vi.fn().mockResolvedValue(null), + getRoomsForSocket: vi.fn().mockResolvedValue([]), + removeUserFromRoom: vi.fn().mockResolvedValue(false), + removeSocketFromAllRooms: vi.fn().mockResolvedValue([]), + broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined), + getRoomUsers: vi.fn().mockResolvedValue([]), + hasRoom: vi.fn().mockResolvedValue(false), + deleteRoom: vi.fn().mockResolvedValue(undefined), + addUserToRoom: vi.fn().mockResolvedValue(undefined), + getUserSession: vi.fn().mockResolvedValue(null), + updateUserActivity: vi.fn().mockResolvedValue(undefined), + updateRoomLastModified: vi.fn().mockResolvedValue(undefined), + emitToRoom: vi.fn(), + getUniqueUserCount: vi.fn().mockResolvedValue(1), + getTotalActiveConnections: vi.fn().mockResolvedValue(0), + shutdown: vi.fn().mockResolvedValue(undefined), + initialize: vi.fn().mockResolvedValue(undefined), + io: { + in: vi.fn().mockReturnValue({ socketsLeave: vi.fn().mockResolvedValue(undefined) }), + }, + ...overrides, + } as unknown as IRoomManager +} + +type SetupArg = Parameters[0] + +describe('setupTablesHandlers', () => { + beforeEach(() => { + vi.clearAllMocks() + mockAuthorizeRoom.mockResolvedValue({ + allowed: true, + status: 200, + workspaceId: 'ws-1', + workspacePermission: 'admin', + }) + }) + + it('rejects join when the socket is not authenticated', async () => { + const { socket, handlers } = createSocket({ userId: undefined, userName: undefined }) + setupTablesHandlers(socket as unknown as SetupArg, createRoomManager()) + + await handlers[TABLE_PRESENCE_EVENTS.JOIN]({ tableId: 'table-1' }) + + expect(socket.emit).toHaveBeenCalledWith(TABLE_PRESENCE_EVENTS.JOIN_ERROR, { + tableId: 'table-1', + error: 'Authentication required', + code: 'AUTHENTICATION_REQUIRED', + retryable: false, + }) + }) + + it('rejects join with a retryable error when realtime is unavailable', async () => { + const { socket, handlers } = createSocket() + setupTablesHandlers( + socket as unknown as SetupArg, + createRoomManager({ isReady: vi.fn().mockReturnValue(false) }) + ) + + await handlers[TABLE_PRESENCE_EVENTS.JOIN]({ tableId: 'table-1' }) + + expect(socket.emit).toHaveBeenCalledWith( + TABLE_PRESENCE_EVENTS.JOIN_ERROR, + expect.objectContaining({ code: 'ROOM_MANAGER_UNAVAILABLE', retryable: true }) + ) + }) + + it('rejects join when table access is denied', async () => { + mockAuthorizeRoom.mockResolvedValue({ + allowed: false, + status: 403, + workspaceId: 'ws-1', + workspacePermission: null, + }) + const { socket, handlers } = createSocket() + setupTablesHandlers(socket as unknown as SetupArg, createRoomManager()) + + await handlers[TABLE_PRESENCE_EVENTS.JOIN]({ tableId: 'table-1' }) + + expect(socket.emit).toHaveBeenCalledWith( + TABLE_PRESENCE_EVENTS.JOIN_ERROR, + expect.objectContaining({ code: 'ACCESS_DENIED', retryable: false }) + ) + }) + + it('joins the table room and broadcasts presence on success', async () => { + const { socket, handlers } = createSocket() + const roomManager = createRoomManager() + setupTablesHandlers(socket as unknown as SetupArg, roomManager) + + await handlers[TABLE_PRESENCE_EVENTS.JOIN]({ tableId: 'table-1', tabSessionId: 'tab-1' }) + + expect(socket.join).toHaveBeenCalledWith('table:table-1') + expect(roomManager.addUserToRoom).toHaveBeenCalledWith( + TABLE_ROOM, + 'socket-1', + expect.objectContaining({ userId: 'user-1', role: 'admin' }) + ) + expect(socket.emit).toHaveBeenCalledWith( + TABLE_PRESENCE_EVENTS.JOIN_SUCCESS, + expect.objectContaining({ tableId: 'table-1', socketId: 'socket-1' }) + ) + expect(roomManager.broadcastPresenceUpdate).toHaveBeenCalledWith(TABLE_ROOM) + }) + + it('persists and relays a cell selection to the namespaced room (id + cell only)', async () => { + const { socket, handlers, toEmit } = createSocket() + const roomManager = createRoomManager({ + getRoomForSocket: vi.fn().mockResolvedValue(TABLE_ROOM), + }) + setupTablesHandlers(socket as unknown as SetupArg, roomManager) + + const cell = { + anchor: { rowId: 'row-1', columnId: 'col-a' }, + focus: { rowId: 'row-1', columnId: 'col-a' }, + editing: true, + } + await handlers[TABLE_PRESENCE_EVENTS.CELL_SELECTION]({ cell }) + + expect(roomManager.updateUserActivity).toHaveBeenCalledWith(TABLE_ROOM, 'socket-1', { cell }) + // Namespaced room → broadcast targets roomName(room), not the bare id. + expect(socket.to).toHaveBeenCalledWith('table:table-1') + // The delta carries only the socket id + cell — identity comes from the roster. + expect(toEmit).toHaveBeenCalledWith(TABLE_PRESENCE_EVENTS.CELL_SELECTION, { + socketId: 'socket-1', + cell, + }) + }) + + it('drops a malformed cell selection without storing or relaying it', async () => { + const { socket, handlers, toEmit } = createSocket() + const roomManager = createRoomManager({ + getRoomForSocket: vi.fn().mockResolvedValue(TABLE_ROOM), + }) + setupTablesHandlers(socket as unknown as SetupArg, roomManager) + + await handlers[TABLE_PRESENCE_EVENTS.CELL_SELECTION]({ cell: { anchor: 'x"]' } }) + + expect(roomManager.updateUserActivity).not.toHaveBeenCalled() + expect(toEmit).not.toHaveBeenCalled() + }) + + it('aborts a stale join whose authorize resolves after a newer join', async () => { + const { socket, handlers } = createSocket() + const roomManager = createRoomManager() + // First join's authorize hangs until released; the second resolves immediately. + let releaseA: (value: unknown) => void = () => {} + const pendingA = new Promise((resolve) => { + releaseA = resolve + }) + mockAuthorizeRoom.mockReturnValueOnce(pendingA).mockResolvedValue({ + allowed: true, + status: 200, + workspaceId: 'ws-1', + workspacePermission: 'admin', + }) + setupTablesHandlers(socket as unknown as SetupArg, roomManager) + + const joinA = handlers[TABLE_PRESENCE_EVENTS.JOIN]({ tableId: 'table-A' }) + await handlers[TABLE_PRESENCE_EVENTS.JOIN]({ tableId: 'table-B' }) + releaseA({ allowed: true, status: 200, workspaceId: 'ws-1', workspacePermission: 'admin' }) + await joinA + + // The newer join (B) wins; the stale A join aborts before touching room state. + expect(socket.join).toHaveBeenCalledWith('table:table-B') + expect(socket.join).not.toHaveBeenCalledWith('table:table-A') + expect(roomManager.addUserToRoom).not.toHaveBeenCalledWith( + { type: ROOM_TYPES.TABLE, id: 'table-A' }, + expect.anything(), + expect.anything() + ) + }) + + it('leaves the table room on leave', async () => { + const { socket, handlers } = createSocket() + const roomManager = createRoomManager({ + getRoomForSocket: vi.fn().mockResolvedValue(TABLE_ROOM), + }) + setupTablesHandlers(socket as unknown as SetupArg, roomManager) + + await handlers[TABLE_PRESENCE_EVENTS.LEAVE]({ tableId: 'table-1' }) + + expect(socket.leave).toHaveBeenCalledWith('table:table-1') + expect(roomManager.removeUserFromRoom).toHaveBeenCalledWith(TABLE_ROOM, 'socket-1') + expect(roomManager.broadcastPresenceUpdate).toHaveBeenCalledWith(TABLE_ROOM, 'socket-1') + }) +}) diff --git a/apps/realtime/src/handlers/tables.ts b/apps/realtime/src/handlers/tables.ts new file mode 100644 index 00000000000..d1fcdd632c2 --- /dev/null +++ b/apps/realtime/src/handlers/tables.ts @@ -0,0 +1,255 @@ +import { createLogger } from '@sim/logger' +import { authorizeRoom } from '@sim/platform-authz/rooms' +import { ROOM_TYPES, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' +import { + type JoinTablePayload, + TABLE_PRESENCE_EVENTS, + type TableCellRef, + type TableCellSelection, +} from '@sim/realtime-protocol/table-presence' +import { resolveAvatarUrl } from '@/handlers/avatar' +import type { AuthenticatedSocket } from '@/middleware/auth' +import type { IRoomManager, UserPresence } from '@/rooms' +import { filterVisiblePresence, sweepStalePresence } from '@/rooms/presence-visibility' + +const logger = createLogger('TablePresenceHandlers') + +/** Longest accepted row/column id — real ids are UUIDs/short ids; this bounds a hostile payload. */ +const MAX_CELL_ID_LENGTH = 200 + +/** The table presence room ref for a table id. */ +const tableRoom = (tableId: string): RoomRef => ({ type: ROOM_TYPES.TABLE, id: tableId }) + +function isCellRef(value: unknown): value is TableCellRef { + if (typeof value !== 'object' || value === null) return false + const ref = value as { rowId?: unknown; columnId?: unknown } + return ( + typeof ref.rowId === 'string' && + ref.rowId.length <= MAX_CELL_ID_LENGTH && + typeof ref.columnId === 'string' && + ref.columnId.length <= MAX_CELL_ID_LENGTH + ) +} + +/** + * Validate + whitelist an untrusted peer's selection before it is stored and + * rebroadcast (it ultimately flows into a DOM query on every viewer). Returns the + * normalized selection — `null` for a legitimately cleared selection — or `undefined` + * for anything malformed, so the caller drops it. Only the known fields survive, so a + * hostile client can't amplify an oversized object through the room. + */ +function normalizeCellSelection(cell: unknown): TableCellSelection | undefined { + if (cell === null) return null + if (typeof cell !== 'object') return undefined + const candidate = cell as { anchor?: unknown; focus?: unknown; editing?: unknown } + if (!isCellRef(candidate.anchor) || !isCellRef(candidate.focus)) return undefined + return { + anchor: { rowId: candidate.anchor.rowId, columnId: candidate.anchor.columnId }, + focus: { rowId: candidate.focus.rowId, columnId: candidate.focus.columnId }, + ...(candidate.editing === true ? { editing: true } : {}), + } +} + +/** + * Live cell-selection presence for the table grid. Mirrors the workspace-files + * join flow but is table-scoped (room id = tableId) with a bidirectional + * cell-selection channel — the grid analog of the workflow cursor/selection + * relay. Table *data* still flows through the one-way durable event stream + * (`lib/table/events.ts`); this socket carries only ephemeral presence. + * + * Table rooms are namespaced (`table:${id}`), so every broadcast targets + * `roomName(room)`, never the bare `room.id` (which the workflow handler can use + * only because a workflow room's name equals its id). + */ +export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) { + // Monotonic per-socket join counter: each JOIN captures its number and, after the async + // authorize, aborts if a newer JOIN has started — a fast table switch A→B can otherwise + // let A's late handler leave B and strand the socket in room A while the client views B. + let joinGeneration = 0 + + socket.on(TABLE_PRESENCE_EVENTS.JOIN, async ({ tableId, tabSessionId }: JoinTablePayload) => { + const joinAttempt = (joinGeneration += 1) + try { + const userId = socket.userId + const userName = socket.userName + + if (!userId || !userName) { + socket.emit(TABLE_PRESENCE_EVENTS.JOIN_ERROR, { + tableId, + error: 'Authentication required', + code: 'AUTHENTICATION_REQUIRED', + retryable: false, + }) + return + } + + if (!roomManager.isReady()) { + socket.emit(TABLE_PRESENCE_EVENTS.JOIN_ERROR, { + tableId, + error: 'Realtime unavailable', + code: 'ROOM_MANAGER_UNAVAILABLE', + retryable: true, + }) + return + } + + // Validate the client-supplied id before it reaches the DB query. + if (typeof tableId !== 'string' || tableId.length === 0) { + socket.emit(TABLE_PRESENCE_EVENTS.JOIN_ERROR, { + tableId: typeof tableId === 'string' ? tableId : '', + error: 'Invalid table id', + code: 'INVALID_PAYLOAD', + retryable: false, + }) + return + } + + const room = tableRoom(tableId) + + let authorized: Awaited> + try { + authorized = await authorizeRoom({ userId, room, action: 'read' }) + } catch (error) { + logger.warn(`Error authorizing table room for ${userId}:`, error) + socket.emit(TABLE_PRESENCE_EVENTS.JOIN_ERROR, { + tableId, + error: 'Failed to verify table access', + code: 'VERIFY_ACCESS_FAILED', + retryable: true, + }) + return + } + + if (!authorized.allowed) { + socket.emit(TABLE_PRESENCE_EVENTS.JOIN_ERROR, { + tableId, + error: authorized.status === 404 ? 'Table not found' : 'Access denied to table', + code: authorized.status === 404 ? 'NOT_FOUND' : 'ACCESS_DENIED', + retryable: false, + }) + return + } + + // A newer JOIN started on this socket during authorize (or the socket dropped): + // abort so a stale join can't leave the room the client has since moved to. + if (joinGeneration !== joinAttempt || socket.disconnected) return + + // Leave a previously-joined table room if switching tables. + const currentRoom = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.TABLE) + if (currentRoom && currentRoom.id !== tableId) { + socket.leave(roomName(currentRoom)) + await roomManager.removeUserFromRoom(currentRoom, socket.id) + await roomManager.broadcastPresenceUpdate(currentRoom) + } + + // Clean up the same user's stale socket from the same tab (a reconnect that + // raced the old socket's disconnect), so presence shows one entry. + if (tabSessionId) { + const existingUsers = await roomManager.getRoomUsers(room) + for (const existing of existingUsers) { + if ( + existing.socketId !== socket.id && + existing.userId === userId && + existing.tabSessionId === tabSessionId + ) { + await roomManager.removeUserFromRoom(room, existing.socketId) + await roomManager.io.in(existing.socketId).socketsLeave(roomName(room)) + } + } + } + + // Reclaim presence orphaned by an ungraceful disconnect (no `disconnecting` + // event fires on a pod crash; the room hashes have no TTL). + await sweepStalePresence(roomManager, room) + + socket.join(roomName(room)) + + const presence: UserPresence = { + userId, + room, + userName, + socketId: socket.id, + tabSessionId, + joinedAt: Date.now(), + lastActivity: Date.now(), + role: authorized.workspacePermission ?? 'read', + avatarUrl: await resolveAvatarUrl(socket, userId), + } + + await roomManager.addUserToRoom(room, socket.id, presence) + + // Filter the join ack to live members so a new joiner never briefly sees a + // ghost from an entry the sweep hasn't reclaimed yet. + const presenceUsers = await filterVisiblePresence( + roomManager.io, + room, + await roomManager.getRoomUsers(room) + ) + socket.emit(TABLE_PRESENCE_EVENTS.JOIN_SUCCESS, { + tableId, + socketId: socket.id, + presenceUsers, + }) + + await roomManager.broadcastPresenceUpdate(room) + + logger.info(`User ${userId} (${userName}) joined table room ${tableId}`) + } catch (error) { + logger.error('Error joining table room:', error) + // Roll back any partial join so a failed attempt can't leave the socket in the + // Socket.IO room or a stale presence entry behind, before signalling a retry. + try { + const room = tableRoom(tableId) + socket.leave(roomName(room)) + await roomManager.removeUserFromRoom(room, socket.id) + } catch {} + socket.emit(TABLE_PRESENCE_EVENTS.JOIN_ERROR, { + tableId, + error: 'Failed to join table', + code: 'JOIN_FAILED', + retryable: true, + }) + } + }) + + socket.on(TABLE_PRESENCE_EVENTS.LEAVE, async (payload?: { tableId?: string }) => { + try { + if (!roomManager.isReady()) return + const room = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.TABLE) + if (!room) return + // Scope the leave to a specific table when the client provides one: a deferred + // leave from a prior view must not evict the socket from a room it has since + // switched into (table A→B leaves A's leave targeting B). + if (payload?.tableId && payload.tableId !== room.id) return + socket.leave(roomName(room)) + await roomManager.removeUserFromRoom(room, socket.id) + await roomManager.broadcastPresenceUpdate(room, socket.id) + } catch (error) { + logger.error('Error leaving table room:', error) + } + }) + + socket.on(TABLE_PRESENCE_EVENTS.CELL_SELECTION, async ({ cell }: { cell: unknown }) => { + try { + // Drop a malformed/oversized selection from an untrusted peer before it is stored + // or rebroadcast (`undefined` = invalid; `null` = a legitimately cleared selection). + const selection = normalizeCellSelection(cell) + if (selection === undefined) return + + const room = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.TABLE) + if (!room) return + + // Persist so a later joiner sees this viewer's current selection in the join ack. + await roomManager.updateUserActivity(room, socket.id, { cell: selection }) + + // Relay to peers (namespaced room → roomName, not room.id). Peers already know this + // socket's identity from the presence roster, so the delta carries only id + cell. + socket.to(roomName(room)).emit(TABLE_PRESENCE_EVENTS.CELL_SELECTION, { + socketId: socket.id, + cell: selection, + }) + } catch (error) { + logger.error(`Error handling table cell selection for socket ${socket.id}:`, error) + } + }) +} diff --git a/apps/realtime/src/handlers/workspace-files.ts b/apps/realtime/src/handlers/workspace-files.ts index 964857e0633..0c7d3ef8c9b 100644 --- a/apps/realtime/src/handlers/workspace-files.ts +++ b/apps/realtime/src/handlers/workspace-files.ts @@ -1,8 +1,7 @@ -import { db, user } from '@sim/db' import { createLogger } from '@sim/logger' import { authorizeRoom } from '@sim/platform-authz/rooms' import { ROOM_TYPES, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' -import { eq } from 'drizzle-orm' +import { resolveAvatarUrl } from '@/handlers/avatar' import type { AuthenticatedSocket } from '@/middleware/auth' import type { IRoomManager, UserPresence } from '@/rooms' import { filterVisiblePresence, sweepStalePresence } from '@/rooms/presence-visibility' @@ -21,24 +20,6 @@ interface JoinPayload { tabSessionId?: string } -async function resolveAvatarUrl( - socket: AuthenticatedSocket, - userId: string -): Promise { - if (socket.userImage) return socket.userImage - try { - const [record] = await db - .select({ image: user.image }) - .from(user) - .where(eq(user.id, userId)) - .limit(1) - return record?.image ?? null - } catch (error) { - logger.warn('Failed to load user avatar for files presence', { userId, error }) - return null - } -} - /** * Presence handlers for the workspace file browser. Mirrors the workflow join * flow but is workspace-scoped (room id = workspaceId) and read-only presence: diff --git a/apps/realtime/src/rooms/memory-manager.ts b/apps/realtime/src/rooms/memory-manager.ts index 90001f53b53..f7ebed862ac 100644 --- a/apps/realtime/src/rooms/memory-manager.ts +++ b/apps/realtime/src/rooms/memory-manager.ts @@ -156,13 +156,14 @@ export class MemoryRoomManager implements IRoomManager { async updateUserActivity( room: RoomRef, socketId: string, - updates: Partial> + updates: Partial> ): Promise { const presence = this.rooms.get(roomKey(room))?.users.get(socketId) if (!presence) return if (updates.cursor !== undefined) presence.cursor = updates.cursor if (updates.selection !== undefined) presence.selection = updates.selection + if (updates.cell !== undefined) presence.cell = updates.cell presence.lastActivity = updates.lastActivity ?? Date.now() } diff --git a/apps/realtime/src/rooms/redis-manager.ts b/apps/realtime/src/rooms/redis-manager.ts index 945d03d539c..cffb9a6d661 100644 --- a/apps/realtime/src/rooms/redis-manager.ts +++ b/apps/realtime/src/rooms/redis-manager.ts @@ -77,7 +77,7 @@ return removed * socket key TTLs to keep a long-lived session alive. * * KEYS: [roomUsers, socketRooms, socketSession] - * ARGV: [socketId, cursorJson, selectionJson, lastActivity, roomsTtl, sessionTtl] + * ARGV: [socketId, cursorJson, selectionJson, lastActivity, roomsTtl, sessionTtl, cellJson] * Returns 1 if the socket had presence in the room, else 0. */ const UPDATE_ACTIVITY_SCRIPT = ` @@ -90,6 +90,7 @@ local selectionJson = ARGV[3] local lastActivity = ARGV[4] local roomsTtl = tonumber(ARGV[5]) local sessionTtl = tonumber(ARGV[6]) +local cellJson = ARGV[7] local existingJson = redis.call('HGET', roomUsersKey, socketId) if not existingJson then @@ -103,6 +104,9 @@ end if selectionJson ~= '' then existing.selection = cjson.decode(selectionJson) end +if cellJson ~= '' then + existing.cell = cjson.decode(cellJson) +end existing.lastActivity = tonumber(lastActivity) redis.call('HSET', roomUsersKey, socketId, cjson.encode(existing)) @@ -325,7 +329,7 @@ export class RedisRoomManager implements IRoomManager { async updateUserActivity( room: RoomRef, socketId: string, - updates: Partial>, + updates: Partial>, retried = false ): Promise { if (!this.updateActivityScriptSha) { @@ -343,6 +347,9 @@ export class RedisRoomManager implements IRoomManager { (updates.lastActivity ?? Date.now()).toString(), SOCKET_ROOMS_TTL.toString(), SESSION_TTL.toString(), + // Trailing arg (ARGV[7]) so existing indices stay stable. `null` (cleared + // selection) serializes to 'null'; `undefined` (no cell change) to '' (skip). + updates.cell !== undefined ? JSON.stringify(updates.cell) : '', ], }) } catch (error) { diff --git a/apps/realtime/src/rooms/types.ts b/apps/realtime/src/rooms/types.ts index c9c8ee18704..f1a67d66abb 100644 --- a/apps/realtime/src/rooms/types.ts +++ b/apps/realtime/src/rooms/types.ts @@ -1,4 +1,5 @@ import type { RoomRef, RoomType } from '@sim/realtime-protocol/rooms' +import type { TableCellSelection } from '@sim/realtime-protocol/table-presence' import type { Server } from 'socket.io' /** @@ -18,6 +19,8 @@ export interface UserPresence { role: string cursor?: { x: number; y: number } selection?: { type: 'block' | 'edge' | 'none'; id?: string } + /** The viewer's current table cell selection, for table presence rooms. */ + cell?: TableCellSelection avatarUrl?: string | null /** * The subfolder the user is viewing, recorded at join for room types that track @@ -108,11 +111,11 @@ export interface IRoomManager { */ deleteRoom(room: RoomRef): Promise - /** Update a socket's activity (cursor, selection, lastActivity) within a room. */ + /** Update a socket's activity (cursor, selection, cell, lastActivity) within a room. */ updateUserActivity( room: RoomRef, socketId: string, - updates: Partial> + updates: Partial> ): Promise /** Bump a room's lastModified timestamp. */ diff --git a/apps/sim/app/api/table/[tableId]/columns/route.ts b/apps/sim/app/api/table/[tableId]/columns/route.ts index 7eecb5ee466..5b60573bd72 100644 --- a/apps/sim/app/api/table/[tableId]/columns/route.ts +++ b/apps/sim/app/api/table/[tableId]/columns/route.ts @@ -17,6 +17,7 @@ import { updateColumnConstraints, updateColumnType, } from '@/lib/table' +import { signalTableSchemaChanged } from '@/lib/table/events' import { accessError, checkAccess, normalizeColumn, rootErrorMessage } from '@/app/api/table/utils' const logger = createLogger('TableColumnsAPI') @@ -51,6 +52,7 @@ export const POST = withRouteHandler(async (request: NextRequest, context: Colum } const updatedTable = await addTableColumn(tableId, validated.column, requestId) + signalTableSchemaChanged(tableId) return NextResponse.json({ success: true, @@ -138,6 +140,7 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: Colu if (!updatedTable) { return NextResponse.json({ error: 'No updates specified' }, { status: 400 }) } + signalTableSchemaChanged(tableId) return NextResponse.json({ success: true, @@ -202,6 +205,7 @@ export const DELETE = withRouteHandler( { tableId, columnName: validated.columnName }, requestId ) + signalTableSchemaChanged(tableId) return NextResponse.json({ success: true, diff --git a/apps/sim/app/api/table/[tableId]/groups/route.ts b/apps/sim/app/api/table/[tableId]/groups/route.ts index 84aecb13c0c..23a2d5f0811 100644 --- a/apps/sim/app/api/table/[tableId]/groups/route.ts +++ b/apps/sim/app/api/table/[tableId]/groups/route.ts @@ -10,6 +10,7 @@ import { parseRequest } from '@/lib/api/server' import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid' import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { signalTableSchemaChanged } from '@/lib/table/events' import { addWorkflowGroup, deleteWorkflowGroup, @@ -99,6 +100,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro }, requestId ) + signalTableSchemaChanged(tableId) return NextResponse.json({ success: true, data: { @@ -161,6 +163,7 @@ export const PATCH = withRouteHandler(async (request: NextRequest, { params }: R }, requestId ) + signalTableSchemaChanged(tableId) return NextResponse.json({ success: true, data: { @@ -194,6 +197,7 @@ export const DELETE = withRouteHandler(async (request: NextRequest, { params }: { tableId, groupId: validated.groupId }, requestId ) + signalTableSchemaChanged(tableId) return NextResponse.json({ success: true, data: { diff --git a/apps/sim/app/api/table/[tableId]/import/route.ts b/apps/sim/app/api/table/[tableId]/import/route.ts index 1d6482390fd..eab163d583b 100644 --- a/apps/sim/app/api/table/[tableId]/import/route.ts +++ b/apps/sim/app/api/table/[tableId]/import/route.ts @@ -37,6 +37,7 @@ import { wouldExceedRowLimit, } from '@/lib/table' import { sniffCsvDelimiterFromStream } from '@/lib/table/csv-delimiter-stream' +import { signalTableSchemaChanged } from '@/lib/table/events' import { importAppendRows, importReplaceRows } from '@/lib/table/import-data' import { getUserSettings } from '@/lib/users/queries' import { @@ -322,6 +323,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro mappedColumns: validation.mappedHeaders.length, skippedHeaders: validation.skippedHeaders.length, }) + signalTableSchemaChanged(tableId) return NextResponse.json({ success: true, @@ -378,6 +380,7 @@ export const POST = withRouteHandler(async (request: NextRequest, { params }: Ro createdColumns: additions.length, mappedColumns: validation.mappedHeaders.length, }) + signalTableSchemaChanged(tableId) return NextResponse.json({ success: true, diff --git a/apps/sim/app/api/table/[tableId]/metadata/route.ts b/apps/sim/app/api/table/[tableId]/metadata/route.ts index f85df0af4bd..6883b56038c 100644 --- a/apps/sim/app/api/table/[tableId]/metadata/route.ts +++ b/apps/sim/app/api/table/[tableId]/metadata/route.ts @@ -7,6 +7,7 @@ import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import type { TableMetadata } from '@/lib/table' import { updateTableMetadata } from '@/lib/table' +import { signalTableMetadataChanged, signalTableSchemaChanged } from '@/lib/table/events' import { accessError, checkAccess } from '@/app/api/table/utils' const logger = createLogger('TableMetadataAPI') @@ -43,11 +44,20 @@ export const PUT = withRouteHandler(async (request: NextRequest, context: TableR return NextResponse.json({ error: 'Invalid workspace ID' }, { status: 400 }) } - const updated = await updateTableMetadata( + const { metadata: updated, schemaChanged } = await updateTableMetadata( tableId, validated.metadata, table.metadata as TableMetadata | null ) + // Signal collaborators to re-apply the new column layout (width/pin/order) live; the + // grid reconciles against its in-progress resize/drag so a peer's change never clobbers + // the local gesture. A reorder that scrubs a workflow-group's dependencies also mutated + // the schema — escalate to the schema signal so peers refresh run-state/enrichment too. + if (schemaChanged) { + signalTableSchemaChanged(tableId) + } else { + signalTableMetadataChanged(tableId) + } return NextResponse.json({ success: true, data: { metadata: updated } }) } catch (error) { diff --git a/apps/sim/app/api/table/[tableId]/route.ts b/apps/sim/app/api/table/[tableId]/route.ts index 5d0069fa570..2dd9a759828 100644 --- a/apps/sim/app/api/table/[tableId]/route.ts +++ b/apps/sim/app/api/table/[tableId]/route.ts @@ -9,6 +9,7 @@ import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { captureServerEvent } from '@/lib/posthog/server' import { deleteTable, renameTable, TableConflictError, type TableSchema } from '@/lib/table' import { getWorkspaceTableLimits } from '@/lib/table/billing' +import { signalTableSchemaChanged } from '@/lib/table/events' import { accessError, checkAccess, normalizeColumn } from '@/app/api/table/utils' const logger = createLogger('TableDetailAPI') @@ -126,6 +127,7 @@ export const PATCH = withRouteHandler( } const updated = await renameTable(tableId, validated.name, requestId, authResult.userId) + signalTableSchemaChanged(tableId) return NextResponse.json({ success: true, diff --git a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts index b1865223f83..4fc0da1871d 100644 --- a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts +++ b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts @@ -15,6 +15,7 @@ import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import type { RowData, TableSchema } from '@/lib/table' import { deleteRow, updateRow } from '@/lib/table' +import { signalTableRowsChanged } from '@/lib/table/events' import { rowWireTranslators } from '@/app/api/table/row-wire' import { accessError, @@ -146,6 +147,9 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR // Only `null` when a `cancellationGuard` is supplied and the SQL guard // rejects the write — this route doesn't pass one, so reaching null is a bug. if (!updatedRow) throw new Error('updateRow returned null without a cancellationGuard') + // An edit that also triggers a dispatch already emits dispatch/cell events; the + // debounced rows refetch on the peer coalesces the two. + signalTableRowsChanged(tableId) // Auto-dispatch for user edits is handled inside `updateRow` (mode: 'new'). // Firing a second mode: 'incomplete' dispatch here would race with the // `mode: 'new'` one AND bulk-clear sibling-group outputs (the incomplete @@ -212,6 +216,7 @@ export const DELETE = withRouteHandler(async (request: NextRequest, context: Row } await deleteRow(tableId, rowId, validated.workspaceId, requestId) + signalTableRowsChanged(tableId) return NextResponse.json({ success: true, diff --git a/apps/sim/app/api/table/[tableId]/rows/route.ts b/apps/sim/app/api/table/[tableId]/rows/route.ts index e200220ea11..49ba83bd322 100644 --- a/apps/sim/app/api/table/[tableId]/rows/route.ts +++ b/apps/sim/app/api/table/[tableId]/rows/route.ts @@ -25,6 +25,7 @@ import { validateRowData, validateRowSize, } from '@/lib/table' +import { signalTableRowsChanged } from '@/lib/table/events' import { queryRows } from '@/lib/table/rows/service' import { TableQueryValidationError } from '@/lib/table/sql' import { rowWireTranslators } from '@/app/api/table/row-wire' @@ -79,6 +80,7 @@ async function handleBatchInsert( table, requestId ) + signalTableRowsChanged(tableId) return NextResponse.json({ success: true, @@ -164,6 +166,7 @@ export const POST = withRouteHandler( table, requestId ) + signalTableRowsChanged(tableId) return NextResponse.json({ success: true, @@ -371,6 +374,7 @@ export const PUT = withRouteHandler( { status: 200 } ) } + signalTableRowsChanged(tableId) return NextResponse.json({ success: true, @@ -436,6 +440,7 @@ export const DELETE = withRouteHandler( { tableId, rowIds: validated.rowIds, workspaceId: validated.workspaceId }, requestId ) + signalTableRowsChanged(tableId) return NextResponse.json({ success: true, @@ -461,6 +466,7 @@ export const DELETE = withRouteHandler( }, requestId ) + signalTableRowsChanged(tableId) return NextResponse.json({ success: true, @@ -534,6 +540,7 @@ export const PATCH = withRouteHandler( table, requestId ) + signalTableRowsChanged(tableId) return NextResponse.json({ success: true, diff --git a/apps/sim/app/api/table/[tableId]/rows/upsert/route.ts b/apps/sim/app/api/table/[tableId]/rows/upsert/route.ts index bc97623ef9a..d6656c11ab4 100644 --- a/apps/sim/app/api/table/[tableId]/rows/upsert/route.ts +++ b/apps/sim/app/api/table/[tableId]/rows/upsert/route.ts @@ -8,6 +8,7 @@ import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import type { RowData, TableSchema } from '@/lib/table' import { upsertRow } from '@/lib/table' +import { signalTableRowsChanged } from '@/lib/table/events' import { rowWireTranslators } from '@/app/api/table/row-wire' import { accessError, checkAccess, rowWriteErrorResponse } from '@/app/api/table/utils' @@ -54,6 +55,7 @@ export const POST = withRouteHandler(async (request: NextRequest, context: Upser table, requestId ) + signalTableRowsChanged(tableId) return NextResponse.json({ success: true, diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/remote-selection-overlay.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/remote-selection-overlay.tsx new file mode 100644 index 00000000000..74ed697f76a --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/remote-selection-overlay.tsx @@ -0,0 +1,196 @@ +'use client' + +import { useCallback, useEffect, useLayoutEffect, useRef, useState } from 'react' +import { getUserColor, withAlpha } from '@/lib/workspaces/colors' +import type { RemoteTableSelection } from '@/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-room' + +/** A measured remote selection, positioned in the grid content wrapper's space. */ +interface SelectionBox { + socketId: string + userName: string + color: string + editing: boolean + top: number + left: number + width: number + height: number +} + +interface RemoteSelectionOverlayProps { + remoteSelections: RemoteTableSelection[] + /** Column id → its rendered column index (matches the cells' `data-col`). */ + columnIndexById: Map + /** The grid's scroll container (`data-table-scroll`), queried for cell rects. */ + scrollElement: HTMLElement | null +} + +/** The cell `` for a (rowId, columnIndex), or undefined when virtualized off-window. */ +function cellRect( + scrollEl: HTMLElement, + rowId: string, + columnIndex: number | undefined +): DOMRect | undefined { + if (columnIndex === undefined) return undefined + // `rowId` is a remote peer's value — escape it so a hostile id can't break the + // selector and throw (`columnIndex` is a local numeric index, already safe). + const cell = scrollEl.querySelector( + `[data-row-id="${CSS.escape(rowId)}"][data-col="${columnIndex}"]` + ) + return cell?.getBoundingClientRect() +} + +/** + * Renders remote collaborators' cell selections over the table grid — a colored + * border per user (Google-Sheets style), a darker fill while they are editing, and + * their name on hover. Mounted inside the grid's `relative` content wrapper, so + * content-space coordinates scroll with the grid automatically. + * + * Positions are measured from the live cell rects (the same `[data-row-id][data-col]` + * idiom the reveal effect uses), keyed by stable ids so each client renders under its + * own sort/scroll. A selection whose rows are virtualized off-window is simply not + * drawn. The layer is `pointer-events-none` so it never intercepts cell clicks; the + * name-on-hover is driven by hit-testing pointer moves against the measured boxes. + */ +export function RemoteSelectionOverlay({ + remoteSelections, + columnIndexById, + scrollElement, +}: RemoteSelectionOverlayProps) { + const rootRef = useRef(null) + const [boxes, setBoxes] = useState([]) + const [hoveredSocketId, setHoveredSocketId] = useState(null) + + // Latest data read by the subscribe-once effect + the pointer hit-test, so neither + // re-subscribes on every incoming selection delta. + const boxesRef = useRef([]) + boxesRef.current = boxes + const remoteSelectionsRef = useRef(remoteSelections) + remoteSelectionsRef.current = remoteSelections + const columnIndexByIdRef = useRef(columnIndexById) + columnIndexByIdRef.current = columnIndexById + // Cached content-wrapper origin, refreshed on each measure (scroll/resize/data change), + // so the pointer hit-test never forces a layout read per mouse move. + const originRef = useRef({ top: 0, left: 0 }) + + const measure = useCallback(() => { + const scrollEl = scrollElement + const root = rootRef.current + if (!scrollEl || !root) return + const origin = root.getBoundingClientRect() + originRef.current = { top: origin.top, left: origin.left } + const next: SelectionBox[] = [] + for (const selection of remoteSelectionsRef.current) { + const { anchor, focus, editing } = selection.cell + const rects = [ + cellRect(scrollEl, anchor.rowId, columnIndexByIdRef.current.get(anchor.columnId)), + cellRect(scrollEl, focus.rowId, columnIndexByIdRef.current.get(focus.columnId)), + ].filter((rect): rect is DOMRect => rect !== undefined) + if (rects.length === 0) continue + + const top = Math.min(...rects.map((r) => r.top)) - origin.top + const left = Math.min(...rects.map((r) => r.left)) - origin.left + const bottom = Math.max(...rects.map((r) => r.bottom)) - origin.top + const right = Math.max(...rects.map((r) => r.right)) - origin.left + next.push({ + socketId: selection.socketId, + userName: selection.userName, + color: getUserColor(selection.userId), + editing: editing === true, + top, + left, + width: right - left, + height: bottom - top, + }) + } + setBoxes(next) + }, [scrollElement]) + + // Subscribe once per scroll element: re-measure on scroll/resize, and hit-test pointer + // moves against the cached boxes/origin — no layout read per move, stays pointer-events-none. + useEffect(() => { + const scrollEl = scrollElement + if (!scrollEl) return + + let raf = 0 + const schedule = () => { + if (!raf) + raf = requestAnimationFrame(() => { + raf = 0 + measure() + }) + } + const handleMove = (event: PointerEvent) => { + const { top, left } = originRef.current + const x = event.clientX - left + const y = event.clientY - top + const hit = boxesRef.current.find( + (b) => x >= b.left && x <= b.left + b.width && y >= b.top && y <= b.top + b.height + ) + setHoveredSocketId((prev) => + prev === (hit?.socketId ?? null) ? prev : (hit?.socketId ?? null) + ) + } + const handleLeave = () => setHoveredSocketId(null) + + // No measure() here — the re-measure layout effect below runs on mount and whenever + // `measure` changes (it depends on `scrollElement`), so it already covers the initial + // and scroll-element-changed measures without a redundant pass. + scrollEl.addEventListener('scroll', schedule, { passive: true }) + scrollEl.addEventListener('pointermove', handleMove, { passive: true }) + scrollEl.addEventListener('pointerleave', handleLeave) + const resizeObserver = new ResizeObserver(schedule) + resizeObserver.observe(scrollEl) + // Also observe the content layer (this overlay fills it): a column resize or a + // row-count change grows/shrinks the content without resizing the scroll container, + // yet moves cell rects — so measure off the content, not just the viewport. + if (rootRef.current) resizeObserver.observe(rootRef.current) + // Re-measure when rows are added/removed/reordered/virtualized (a live refetch moves + // cells without a scroll/resize) — childList only, so a cell-content edit doesn't fire. + const tbody = scrollEl.querySelector('tbody') + const rowObserver = new MutationObserver(schedule) + if (tbody) rowObserver.observe(tbody, { childList: true }) + + return () => { + scrollEl.removeEventListener('scroll', schedule) + scrollEl.removeEventListener('pointermove', handleMove) + scrollEl.removeEventListener('pointerleave', handleLeave) + resizeObserver.disconnect() + rowObserver.disconnect() + if (raf) cancelAnimationFrame(raf) + } + }, [scrollElement, measure]) + + // Re-measure when the selections or column layout change (listeners stay subscribed). + // Layout effect so positions update before paint — no one-frame lag as a peer moves. + useLayoutEffect(() => { + measure() + }, [remoteSelections, columnIndexById, measure]) + + return ( +
+ {boxes.map((box) => ( +
+ {hoveredSocketId === box.socketId && ( + + {box.userName} + + )} +
+ ))} +
+ ) +} diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx index 330df1c2d2d..8ae1abd1df7 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/components/table-grid/table-grid.tsx @@ -5,6 +5,7 @@ import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } fr import { cn, toast, useToast } from '@sim/emcn' import { Loader, TableX } from '@sim/emcn/icons' import { createLogger } from '@sim/logger' +import type { TableCellSelection } from '@sim/realtime-protocol/table-presence' import { useVirtualizer } from '@tanstack/react-virtual' import { useParams } from 'next/navigation' import { usePostHog } from 'posthog-js/react' @@ -14,6 +15,7 @@ import type { ColumnDefinition, Filter, TableRow as TableRowType, WorkflowGroup import { getColumnId } from '@/lib/table/column-keys' import { TABLE_LIMITS } from '@/lib/table/constants' import { useUserPermissionsContext } from '@/app/workspace/[workspaceId]/providers/workspace-permissions-provider' +import type { RemoteTableSelection } from '@/app/workspace/[workspaceId]/tables/[tableId]/hooks/use-table-room' import { useTimezone } from '@/hooks/queries/general-settings' import { useAddTableColumn, @@ -47,6 +49,7 @@ import { ExpandedCellPopover } from './cells' import { ADD_COL_WIDTH, COL_WIDTH, SELECTION_TINT_BG } from './constants' import { DataRow } from './data-row' import { ColumnHeaderMenu, WorkflowGroupMetaCell } from './headers' +import { RemoteSelectionOverlay } from './remote-selection-overlay' import { TableFind } from './table-find' import { AddRowButton, SelectAllCheckbox, TableColGroup } from './table-primitives' import type { DisplayColumn } from './types' @@ -143,6 +146,10 @@ interface TableGridProps { workspaceId?: string tableId?: string embedded?: boolean + /** Remote collaborators' cell selections, rendered as presence overlays. */ + remoteSelections: RemoteTableSelection[] + /** Broadcast the local viewer's cell selection to the table presence room. */ + emitCellSelection: (cell: TableCellSelection) => void /** * Pixel width to reserve on the right of the table's scroll content for the * currently-open slideout panel (column config, workflow config, or log @@ -288,6 +295,8 @@ export function TableGrid({ workspaceId: propWorkspaceId, tableId: propTableId, embedded, + remoteSelections, + emitCellSelection, sidebarReservedPx, onOpenColumnConfig, onOpenWorkflowConfig, @@ -348,6 +357,8 @@ export function TableGrid({ const columnWidthsRef = useRef(columnWidths) columnWidthsRef.current = columnWidths const [resizingColumn, setResizingColumn] = useState(null) + const resizingColumnRef = useRef(resizingColumn) + resizingColumnRef.current = resizingColumn const [columnOrder, setColumnOrder] = useState(null) const columnOrderRef = useRef(columnOrder) columnOrderRef.current = columnOrder @@ -645,6 +656,13 @@ export function TableGrid({ return expandToDisplayColumns(ordered, tableWorkflowGroups) }, [columns, columnOrder, tableWorkflowGroups]) + /** Column id → its rendered index (matches the cells' `data-col`), for placing overlays. */ + const columnIndexById = useMemo(() => { + const map = new Map() + displayColumns.forEach((col, index) => map.set(col.key, index)) + return map + }, [displayColumns]) + const workflowGroupById = useMemo( () => new Map(tableWorkflowGroups.map((g) => [g.id, g])), [tableWorkflowGroups] @@ -819,6 +837,30 @@ export function TableGrid({ ? (rowsRef.current[selectionFocus.rowIndex]?.id ?? null) : null + // Broadcast the local viewer's cell selection to the presence room. Resolves the + // index-based selection to stable (rowId, columnId), re-running on `rows`/`displayColumns` + // too so a peer's row insert/delete/reorder re-broadcasts the shifted id under the same + // index (the emitter dedups an unchanged result). `editing` marks the active cell so + // peers darken it (the "someone is typing here" signal). + useEffect(() => { + const resolve = (coord: CellCoord | null) => { + if (!coord) return null + const rowId = rows[coord.rowIndex]?.id + const columnId = displayColumns[coord.colIndex]?.key + return rowId && columnId ? { rowId, columnId } : null + } + const anchor = resolve(selectionAnchor) + if (!anchor) { + emitCellSelection(null) + return + } + // A single-cell click leaves `selectionFocus` null; the grid treats that as a + // one-cell selection at the anchor (`focus ?? anchor`). Mirror that — otherwise the + // most common selection would never broadcast and would clear the prior outline. + const focus = resolve(selectionFocus) ?? anchor + emitCellSelection({ anchor, focus, editing: editingCell !== null }) + }, [selectionAnchor, selectionFocus, editingCell, rows, displayColumns, emitCellSelection]) + const { data: findData, isFetching: isFindFetching } = useFindTableRows({ workspaceId, tableId, @@ -1740,20 +1782,32 @@ export function TableGrid({ } return } - // After first load: only re-seed `columnOrder` when the *set of columns* - // changes (e.g. a workflow group adds/removes outputs server-side). Pure - // reorders are left alone so an in-flight optimistic drag isn't clobbered - // by a refetch returning the pre-drag order. + // After first load a collaborator (or our own committed edit) reshaped the layout. + // Re-apply it live, but never clobber the gesture the local user is mid-way through — + // their in-progress value leads the server's. Each field is guarded by reference: + // React Query structural sharing keeps an unchanged sub-object referentially stable, + // so an unrelated change (e.g. a peer's pin) doesn't re-apply widths/order. + // Width: keep the column being actively resized on its live local value. + const serverWidths = tableData.metadata.columnWidths + if (serverWidths && serverWidths !== columnWidthsRef.current) { + const resizing = resizingColumnRef.current + const localWidth = resizing ? columnWidthsRef.current[resizing] : undefined + setColumnWidths( + resizing && localWidth !== undefined + ? { ...serverWidths, [resizing]: localWidth } + : serverWidths + ) + } + // Pins toggle instantly (no in-progress gesture) — apply on change. + const serverPins = tableData.metadata.pinnedColumns + if (serverPins && serverPins !== pinnedColumnsRef.current) { + setPinnedColumns(serverPins) + } + // Order: apply unless a local column drag is in flight (an optimistic reorder would + // otherwise be reverted to the pre-drag order the refetch returns). const serverOrder = tableData.metadata.columnOrder - if (serverOrder) { - const localOrder = columnOrderRef.current - const serverSet = new Set(serverOrder) - const localSet = new Set(localOrder ?? []) - const setChanged = - !localOrder || serverSet.size !== localSet.size || serverOrder.some((n) => !localSet.has(n)) - if (setChanged) { - setColumnOrder(serverOrder) - } + if (serverOrder && serverOrder !== columnOrderRef.current && !dragColumnNameRef.current) { + setColumnOrder(serverOrder) } }, [tableData?.metadata]) @@ -3885,6 +3939,13 @@ export function TableGrid({ })()} + {remoteSelections.length > 0 && ( + + )} {resizingColumn && (
+} + +interface UseTableRoomResult { + /** Collaborators viewing this table, excluding the current socket (for avatars). */ + otherUsers: PresenceAvatarUser[] + /** Remote viewers that currently have a cell selected (for overlays). */ + remoteSelections: RemoteTableSelection[] + /** Broadcast the local viewer's current cell selection (`null` clears it). Throttled. */ + emitCellSelection: (cell: TableCellSelection) => void +} + +/** + * Joins the table presence room for live collaborator avatars + cell-selection + * highlights. Presence rides the shared socket (`useSocket`); table data is + * unchanged (it flows through the one-way durable event stream). The full roster + * arrives via the presence broadcast (join/leave); individual selection moves + * arrive as lower-latency {@link TABLE_PRESENCE_EVENTS.CELL_SELECTION} deltas that + * patch the matching roster entry. + */ +export function useTableRoom(tableId: string): UseTableRoomResult { + const { socket, currentSocketId } = useSocket() + + const [presenceUsers, setPresenceUsers] = useState([]) + + const tabSessionIdRef = useRef('') + if (!tabSessionIdRef.current) tabSessionIdRef.current = generateShortId() + + // The local viewer's current selection, re-broadcast on (re)join. Emits only fire on a + // selection change, and the server drops a CELL_SELECTION for a socket not yet in the + // room — so a selection made before the join completes (or held across a reconnect) + // would otherwise never reach peers until it next changes. + const currentCellRef = useRef(null) + + useEffect(() => { + if (!socket || !tableId) return + + let retries = 0 + let retryTimer: ReturnType | null = null + + const join = () => { + socket.emit(TABLE_PRESENCE_EVENTS.JOIN, { tableId, tabSessionId: tabSessionIdRef.current }) + } + + const handleJoinSuccess = (data: JoinTableSuccess) => { + if (data.tableId !== tableId) return + retries = 0 + if (retryTimer) { + clearTimeout(retryTimer) + retryTimer = null + } + setPresenceUsers(data.presenceUsers ?? []) + // Re-send our current selection now that the room is joined (earlier emits were + // dropped server-side), so peers see it without waiting for the next change. + if (currentCellRef.current) { + socket.emit(TABLE_PRESENCE_EVENTS.CELL_SELECTION, { cell: currentCellRef.current }) + } + } + const handleJoinError = (data: JoinTableError) => { + if (data.tableId !== tableId) return + logger.warn('Failed to join table room', { code: data.code, error: data.error }) + if (data.retryable && retries < MAX_JOIN_RETRIES) { + retries += 1 + retryTimer = setTimeout(join, JOIN_RETRY_BASE_MS * retries) + } + } + const handlePresence = (users: TablePresenceUser[]) => { + // Take membership from the roster snapshot but keep the `cell` we already hold for + // a known socket. Trade-off: a snapshot can lag the lower-latency CELL_SELECTION + // deltas, so a blind replace could revert a fresher selection (the common case). + // The cost is that a *dropped* delta is no longer healed by the next snapshot, only + // by the peer's next delta — fine, since deltas flow continuously during selection + // and a cleared selection also sends `null` via delta. + setPresenceUsers((prev) => { + const cellBySocket = new Map(prev.map((user) => [user.socketId, user.cell])) + return (users ?? []).map((user) => + cellBySocket.has(user.socketId) + ? { ...user, cell: cellBySocket.get(user.socketId) } + : user + ) + }) + } + const handleCellSelection = (data: TableCellSelectionBroadcast) => { + // Patch the matching roster entry's selection. The peer is always already in + // the roster: the server broadcasts their join (→ presence-update) before they + // can select, and Socket.IO preserves that order — so a delta for an unknown + // socket only means a dropped broadcast, which the next presence-update heals. + setPresenceUsers((prev) => + prev.map((user) => (user.socketId === data.socketId ? { ...user, cell: data.cell } : user)) + ) + } + + // Join now if the socket is already connected; `connect` covers (re)connects. + if (socket.connected) join() + socket.on('connect', join) + socket.on(TABLE_PRESENCE_EVENTS.JOIN_SUCCESS, handleJoinSuccess) + socket.on(TABLE_PRESENCE_EVENTS.JOIN_ERROR, handleJoinError) + socket.on(TABLE_PRESENCE_UPDATE_EVENT, handlePresence) + socket.on(TABLE_PRESENCE_EVENTS.CELL_SELECTION, handleCellSelection) + + return () => { + if (retryTimer) clearTimeout(retryTimer) + socket.off('connect', join) + socket.off(TABLE_PRESENCE_EVENTS.JOIN_SUCCESS, handleJoinSuccess) + socket.off(TABLE_PRESENCE_EVENTS.JOIN_ERROR, handleJoinError) + socket.off(TABLE_PRESENCE_UPDATE_EVENT, handlePresence) + socket.off(TABLE_PRESENCE_EVENTS.CELL_SELECTION, handleCellSelection) + setPresenceUsers([]) + // Leave scoped to THIS table so a table A→B switch (B joins first, auto-leaving + // A) can't have A's deferred leave evict the fresh B membership. + socket.emit(TABLE_PRESENCE_EVENTS.LEAVE, { tableId }) + } + }, [socket, tableId]) + + const socketRef = useRef(socket) + socketRef.current = socket + const lastEmitRef = useRef(0) + const trailingTimerRef = useRef | null>(null) + const pendingCellRef = useRef(null) + /** Key of the last selection sent, to skip re-emitting an unchanged selection. */ + const lastSentKeyRef = useRef(null) + + // Reset the throttle when the table changes (or on unmount): a pending selection for + // the table we're leaving must not flush into the next table's room after a switch. + useEffect( + () => () => { + if (trailingTimerRef.current) { + clearTimeout(trailingTimerRef.current) + trailingTimerRef.current = null + } + pendingCellRef.current = null + currentCellRef.current = null + lastSentKeyRef.current = null + lastEmitRef.current = 0 + }, + [tableId] + ) + + const emitCellSelection = useCallback((cell: TableCellSelection) => { + // Skip re-emitting an unchanged selection: the caller re-resolves on every data + // refetch (so a peer's row insert re-broadcasts the shifted rowId), but most refetches + // don't move the selection — dedup those, and the no-selection state on table open. + const key = cell === null ? 'null' : JSON.stringify(cell) + if (key === lastSentKeyRef.current) return + lastSentKeyRef.current = key + currentCellRef.current = cell + pendingCellRef.current = cell + const flush = () => { + lastEmitRef.current = Date.now() + trailingTimerRef.current = null + socketRef.current?.emit(TABLE_PRESENCE_EVENTS.CELL_SELECTION, { + cell: pendingCellRef.current, + }) + } + const elapsed = Date.now() - lastEmitRef.current + if (elapsed >= SELECTION_EMIT_THROTTLE_MS) { + flush() + } else if (!trailingTimerRef.current) { + trailingTimerRef.current = setTimeout(flush, SELECTION_EMIT_THROTTLE_MS - elapsed) + } + }, []) + + const otherUsers = useMemo( + () => presenceUsers.filter((user) => user.socketId !== currentSocketId), + [presenceUsers, currentSocketId] + ) + const remoteSelections = useMemo( + () => + otherUsers + .filter( + (user): user is TablePresenceUser & { cell: NonNullable } => + user.cell != null + ) + .map((user) => ({ + socketId: user.socketId, + userId: user.userId, + userName: user.userName, + cell: user.cell, + })), + [otherUsers] + ) + + return { otherUsers, remoteSelections, emitCellSelection } +} diff --git a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/table.tsx b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/table.tsx index 0f7485636bb..08c18c0ef73 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/table.tsx +++ b/apps/sim/app/workspace/[workspaceId]/tables/[tableId]/table.tsx @@ -24,6 +24,7 @@ import { Resource, type SortConfig, } from '@/app/workspace/[workspaceId]/components' +import { PresenceAvatars } from '@/app/workspace/[workspaceId]/components/presence/presence-avatars' import { LogDetails } from '@/app/workspace/[workspaceId]/logs/components' import { useUserPermissionsContext } from '@/app/workspace/[workspaceId]/providers/workspace-permissions-provider' import { ImportCsvDialog } from '@/app/workspace/[workspaceId]/tables/components/import-csv-dialog' @@ -59,7 +60,7 @@ import { } from './components' import { COLUMN_SIDEBAR_WIDTH } from './components/table-grid/constants' import { COLUMN_TYPE_ICONS } from './components/table-grid/headers' -import { useTable, useTableEventStream } from './hooks' +import { useTable, useTableEventStream, useTableRoom } from './hooks' import { DEFAULT_TABLE_DETAIL_SORT_DIRECTION, tableDetailParsers, @@ -153,6 +154,14 @@ export function Table({ } useTableEventStream({ tableId, workspaceId, onUsageLimitReached }) + // Live table presence (avatars + cell-selection highlights). Scoped to the + // dedicated tables page — the embedded surface passes no id, disabling it. + const { + otherUsers: presenceUsers, + remoteSelections, + emitCellSelection, + } = useTableRoom(embedded ? '' : tableId) + const [slideout, dispatch] = useReducer(slideoutReducer, { kind: 'none' }) const [showDeleteTableConfirm, setShowDeleteTableConfirm] = useState(false) const [isImportCsvOpen, setIsImportCsvOpen] = useState(false) @@ -654,6 +663,7 @@ export function Table({ breadcrumbs={breadcrumbs} aside={
+ {presenceUsers.length > 0 && } {selection.totalRunning > 0 || selection.hasActiveDispatch ? ( { +): Promise<{ metadata: TableMetadata; schemaChanged: boolean }> { const merged: TableMetadata = { ...(existingMetadata ?? {}), ...metadata } // When `columnOrder` is in the patch, scrub any workflow-group dependency @@ -635,7 +635,7 @@ export async function updateTableMetadata( .set(nextSchema ? { metadata: merged, schema: nextSchema } : { metadata: merged }) .where(eq(userTableDefinitions.id, tableId)) - return merged + return { metadata: merged, schemaChanged: nextSchema !== null } } /** diff --git a/packages/platform-authz/src/rooms.ts b/packages/platform-authz/src/rooms.ts index f8676cfde45..acb4d2779e5 100644 --- a/packages/platform-authz/src/rooms.ts +++ b/packages/platform-authz/src/rooms.ts @@ -1,4 +1,4 @@ -import { db, workspace, workspaceFiles } from '@sim/db' +import { db, userTableDefinitions, workspace, workspaceFiles } from '@sim/db' import { ROOM_TYPES, type RoomRef, type RoomType } from '@sim/realtime-protocol/rooms' import { and, eq, isNull } from 'drizzle-orm' import { getActiveWorkflowContext } from './workflow' @@ -55,6 +55,23 @@ async function resolveFileDocWorkspace(fileId: string): Promise { + const [table] = await db + .select({ workspaceId: userTableDefinitions.workspaceId }) + .from(userTableDefinitions) + .where(and(eq(userTableDefinitions.id, tableId), isNull(userTableDefinitions.archivedAt))) + .limit(1) + + if (!table?.workspaceId) return null + return resolveWorkspaceRoomWorkspace(table.workspaceId) +} + /** * Single source of truth mapping each room type to its resource→workspace * lookup. Every realtime room is workspace-scoped and authorizes through the @@ -74,6 +91,8 @@ const ROOM_WORKSPACE_RESOLVERS: Record = { [ROOM_TYPES.WORKSPACE_FILES]: resolveWorkspaceRoomWorkspace, // A file-doc room is addressed by file id; resolve it to its workspace. [ROOM_TYPES.WORKSPACE_FILE_DOC]: resolveFileDocWorkspace, + // A table room is addressed by table id; resolve it to its workspace. + [ROOM_TYPES.TABLE]: resolveTableWorkspace, } /** Resolves a room's owning workspace, or `null` if the room resource is gone. */ diff --git a/packages/realtime-protocol/package.json b/packages/realtime-protocol/package.json index 7023ec6a12a..e1995aae687 100644 --- a/packages/realtime-protocol/package.json +++ b/packages/realtime-protocol/package.json @@ -29,6 +29,10 @@ "./file-doc": { "types": "./src/file-doc.ts", "default": "./src/file-doc.ts" + }, + "./table-presence": { + "types": "./src/table-presence.ts", + "default": "./src/table-presence.ts" } }, "scripts": { diff --git a/packages/realtime-protocol/src/rooms.ts b/packages/realtime-protocol/src/rooms.ts index b81b65e72bb..de6a97dd82d 100644 --- a/packages/realtime-protocol/src/rooms.ts +++ b/packages/realtime-protocol/src/rooms.ts @@ -29,6 +29,12 @@ export const ROOM_TYPES = { * workspace-scoped {@link ROOM_TYPES.WORKSPACE_FILES} browser room. */ WORKSPACE_FILE_DOC: 'workspace-file-doc', + /** + * A single table's grid (one room per table). Carries live cell-selection + * presence — which cells each viewer has selected — so its id space is the + * table id. + */ + TABLE: 'table', } as const export type RoomType = (typeof ROOM_TYPES)[keyof typeof ROOM_TYPES] diff --git a/packages/realtime-protocol/src/table-presence.ts b/packages/realtime-protocol/src/table-presence.ts new file mode 100644 index 00000000000..45d06a94dc7 --- /dev/null +++ b/packages/realtime-protocol/src/table-presence.ts @@ -0,0 +1,103 @@ +/** + * Wire protocol for live table presence — which cell(s) each viewer has selected + * in a table grid. Carried over the shared, already-authenticated Socket.IO + * connection (the server relay is `apps/realtime/src/handlers/tables.ts`), + * separate from the one-way durable cell-status stream (`lib/table/events.ts`). + * + * Centralized here so the server emits and the client subscriptions cannot drift. + * This module is pure so both `apps/sim` and `apps/realtime` can import it. + */ + +/** Socket.IO event names for the table presence channel. */ +export const TABLE_PRESENCE_EVENTS = { + JOIN: 'join-table', + JOIN_SUCCESS: 'join-table-success', + JOIN_ERROR: 'join-table-error', + LEAVE: 'leave-table', + /** + * The sender's current cell selection. Sent client→server, then relayed + * server→peers with the sender's identity attached. + */ + CELL_SELECTION: 'table-cell-selection', +} as const + +/** + * A single cell address. Keyed by stable ids (never positional indices): row and + * column order differ per client under their own sort/filter, so an index would + * point at the wrong cell on the receiver. + */ +export interface TableCellRef { + rowId: string + columnId: string +} + +/** + * A viewer's grid selection: the `anchor` and `focus` corners of a rectangular + * range (a single-cell selection has `anchor` equal to `focus`). `null` clears + * the selection (the viewer has nothing selected). + */ +export type TableCellSelection = { + anchor: TableCellRef + focus: TableCellRef + /** + * True while the viewer is actively editing the `focus` cell (they double-clicked + * or started typing). Peers render the cell with a slightly darker fill — the + * Google-Sheets "someone is typing here" signal — on top of the color border. + */ + editing?: boolean +} | null + +/** Client→server join request for a table presence room. */ +export interface JoinTablePayload { + tableId: string + /** Stable per-tab id so a reconnecting tab replaces its own stale presence entry. */ + tabSessionId?: string +} + +/** Server→client rejection of a {@link TABLE_PRESENCE_EVENTS.JOIN}. */ +export interface JoinTableError { + tableId: string + error: string + code: + | 'AUTHENTICATION_REQUIRED' + | 'ROOM_MANAGER_UNAVAILABLE' + | 'INVALID_PAYLOAD' + | 'VERIFY_ACCESS_FAILED' + | 'NOT_FOUND' + | 'ACCESS_DENIED' + | 'JOIN_FAILED' + /** Whether re-attempting the join (e.g. after reconnect) could succeed. */ + retryable: boolean +} + +/** + * A remote viewer of a table, as carried in the join ack and every + * `table:presence-update` broadcast. `cell` is the viewer's current selection at + * broadcast time (absent until they select something). + */ +export interface TablePresenceUser { + socketId: string + userId: string + userName: string + avatarUrl?: string | null + cell?: TableCellSelection +} + +/** Server→client ack of a successful join, carrying the room's current viewers. */ +export interface JoinTableSuccess { + tableId: string + socketId: string + presenceUsers: TablePresenceUser[] +} + +/** + * A single remote viewer's cell-selection delta, relayed to peers on + * {@link TABLE_PRESENCE_EVENTS.CELL_SELECTION}. Lower-latency than a full presence + * broadcast for the frequent case of just moving the selection. Carries only the + * socket id + selection — peers already hold the viewer's identity (name, color) from + * the presence roster, so it is not repeated on this high-frequency channel. + */ +export interface TableCellSelectionBroadcast { + socketId: string + cell: TableCellSelection +}