Skip to content

Commit 1fbe302

Browse files
icecrasher321icecrasher321
committed
fix bugbot comments
1 parent e5d9b98 commit 1fbe302

File tree

4 files changed

+95
-51
lines changed

4 files changed

+95
-51
lines changed

‎apps/sim/app/workspace/providers/socket-provider.tsx‎

Lines changed: 29 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
127127
const [presenceUsers, setPresenceUsers] = useState<PresenceUser[]>([])
128128
const [authFailed, setAuthFailed] = useState(false)
129129
const initializedRef = useRef(false)
130+
const socketRef = useRef<Socket | null>(null)
130131

131132
const params = useParams()
132133
const urlWorkflowId = params?.workflowId as string | undefined
@@ -146,6 +147,7 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
146147
}>({})
147148

148149
const positionUpdateTimeouts = useRef<Map<string, number>>(new Map())
150+
const isRejoiningRef = useRef<boolean>(false)
149151
const pendingPositionUpdates = useRef<Map<string, any>>(new Map())
150152

151153
const generateSocketToken = async (): Promise<string> => {
@@ -315,13 +317,19 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
315317

316318
// Handle join workflow success - confirms room membership with presence list
317319
socketInstance.on('join-workflow-success', ({ workflowId, presenceUsers }) => {
320+
isRejoiningRef.current = false
318321
setCurrentWorkflowId(workflowId)
319322
setPresenceUsers(presenceUsers || [])
320323
logger.info(`Successfully joined workflow room: ${workflowId}`, {
321324
presenceCount: presenceUsers?.length || 0,
322325
})
323326
})
324327

328+
socketInstance.on('join-workflow-error', ({ error }) => {
329+
isRejoiningRef.current = false
330+
logger.error('Failed to join workflow:', error)
331+
})
332+
325333
socketInstance.on('workflow-operation', (data) => {
326334
eventHandlers.current.workflowOperation?.(data)
327335
})
@@ -485,6 +493,19 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
485493

486494
socketInstance.on('operation-forbidden', (error) => {
487495
logger.warn('Operation forbidden:', error)
496+
497+
if (error?.type === 'SESSION_ERROR') {
498+
const workflowId = urlWorkflowIdRef.current
499+
500+
if (workflowId && !isRejoiningRef.current) {
501+
isRejoiningRef.current = true
502+
logger.info(`Session expired, rejoining workflow: ${workflowId}`)
503+
socketInstance.emit('join-workflow', {
504+
workflowId,
505+
tabSessionId: getTabSessionId(),
506+
})
507+
}
508+
}
488509
})
489510

490511
socketInstance.on('workflow-state', async (workflowData) => {
@@ -495,11 +516,8 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
495516
}
496517
})
497518

519+
socketRef.current = socketInstance
498520
setSocket(socketInstance)
499-
500-
return () => {
501-
socketInstance.close()
502-
}
503521
} catch (error) {
504522
logger.error('Failed to initialize socket with token:', error)
505523
setIsConnecting(false)
@@ -514,6 +532,13 @@ export function SocketProvider({ children, user }: SocketProviderProps) {
514532
})
515533
positionUpdateTimeouts.current.clear()
516534
pendingPositionUpdates.current.clear()
535+
536+
// Close socket on unmount
537+
if (socketRef.current) {
538+
logger.info('Closing socket connection on unmount')
539+
socketRef.current.close()
540+
socketRef.current = null
541+
}
517542
}
518543
}, [user?.id, authFailed])
519544

‎apps/sim/socket/handlers/subblocks.ts‎

Lines changed: 27 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -36,32 +36,36 @@ export function cleanupPendingSubblocksForSocket(socketId: string): void {
3636

3737
export function setupSubblocksHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
3838
socket.on('subblock-update', async (data) => {
39-
const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
40-
const session = await roomManager.getUserSession(socket.id)
41-
42-
if (!workflowId || !session) {
43-
logger.debug(`Ignoring subblock update: socket not connected to any workflow room`, {
44-
socketId: socket.id,
45-
hasWorkflowId: !!workflowId,
46-
hasSession: !!session,
47-
})
48-
return
49-
}
50-
5139
const { blockId, subblockId, value, timestamp, operationId } = data
5240

53-
const hasRoom = await roomManager.hasWorkflowRoom(workflowId)
54-
if (!hasRoom) {
55-
logger.debug(`Ignoring subblock update: workflow room not found`, {
56-
socketId: socket.id,
57-
workflowId,
58-
blockId,
59-
subblockId,
60-
})
61-
return
62-
}
63-
6441
try {
42+
const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
43+
const session = await roomManager.getUserSession(socket.id)
44+
45+
if (!workflowId || !session) {
46+
logger.debug(`Ignoring subblock update: socket not connected to any workflow room`, {
47+
socketId: socket.id,
48+
hasWorkflowId: !!workflowId,
49+
hasSession: !!session,
50+
})
51+
socket.emit('operation-forbidden', {
52+
type: 'SESSION_ERROR',
53+
message: 'Session expired, please rejoin workflow',
54+
})
55+
return
56+
}
57+
58+
const hasRoom = await roomManager.hasWorkflowRoom(workflowId)
59+
if (!hasRoom) {
60+
logger.debug(`Ignoring subblock update: workflow room not found`, {
61+
socketId: socket.id,
62+
workflowId,
63+
blockId,
64+
subblockId,
65+
})
66+
return
67+
}
68+
6569
// Update user activity
6670
await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() })
6771

‎apps/sim/socket/handlers/variables.ts‎

Lines changed: 27 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -32,32 +32,36 @@ export function cleanupPendingVariablesForSocket(socketId: string): void {
3232

3333
export function setupVariablesHandlers(socket: AuthenticatedSocket, roomManager: IRoomManager) {
3434
socket.on('variable-update', async (data) => {
35-
const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
36-
const session = await roomManager.getUserSession(socket.id)
37-
38-
if (!workflowId || !session) {
39-
logger.debug(`Ignoring variable update: socket not connected to any workflow room`, {
40-
socketId: socket.id,
41-
hasWorkflowId: !!workflowId,
42-
hasSession: !!session,
43-
})
44-
return
45-
}
46-
4735
const { variableId, field, value, timestamp, operationId } = data
4836

49-
const hasRoom = await roomManager.hasWorkflowRoom(workflowId)
50-
if (!hasRoom) {
51-
logger.debug(`Ignoring variable update: workflow room not found`, {
52-
socketId: socket.id,
53-
workflowId,
54-
variableId,
55-
field,
56-
})
57-
return
58-
}
59-
6037
try {
38+
const workflowId = await roomManager.getWorkflowIdForSocket(socket.id)
39+
const session = await roomManager.getUserSession(socket.id)
40+
41+
if (!workflowId || !session) {
42+
logger.debug(`Ignoring variable update: socket not connected to any workflow room`, {
43+
socketId: socket.id,
44+
hasWorkflowId: !!workflowId,
45+
hasSession: !!session,
46+
})
47+
socket.emit('operation-forbidden', {
48+
type: 'SESSION_ERROR',
49+
message: 'Session expired, please rejoin workflow',
50+
})
51+
return
52+
}
53+
54+
const hasRoom = await roomManager.hasWorkflowRoom(workflowId)
55+
if (!hasRoom) {
56+
logger.debug(`Ignoring variable update: workflow room not found`, {
57+
socketId: socket.id,
58+
workflowId,
59+
variableId,
60+
field,
61+
})
62+
return
63+
}
64+
6165
// Update user activity
6266
await roomManager.updateUserActivity(workflowId, socket.id, { lastActivity: Date.now() })
6367

‎apps/sim/socket/rooms/redis-manager.ts‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,13 +48,17 @@ return workflowId
4848
/**
4949
* Lua script for atomic user activity update.
5050
* Performs read-modify-write atomically to prevent lost updates.
51+
* Also refreshes TTL on socket keys to prevent expiry during long sessions.
5152
*/
5253
const UPDATE_ACTIVITY_SCRIPT = `
5354
local workflowUsersKey = KEYS[1]
55+
local socketWorkflowKey = KEYS[2]
56+
local socketSessionKey = KEYS[3]
5457
local socketId = ARGV[1]
5558
local cursorJson = ARGV[2]
5659
local selectionJson = ARGV[3]
5760
local lastActivity = ARGV[4]
61+
local ttl = tonumber(ARGV[5])
5862
5963
local existingJson = redis.call('HGET', workflowUsersKey, socketId)
6064
if not existingJson then
@@ -72,6 +76,8 @@ end
7276
existing.lastActivity = tonumber(lastActivity)
7377
7478
redis.call('HSET', workflowUsersKey, socketId, cjson.encode(existing))
79+
redis.call('EXPIRE', socketWorkflowKey, ttl)
80+
redis.call('EXPIRE', socketSessionKey, ttl)
7581
return 1
7682
`
7783

@@ -269,12 +275,17 @@ export class RedisRoomManager implements IRoomManager {
269275

270276
try {
271277
await this.redis.evalSha(this.updateActivityScriptSha, {
272-
keys: [KEYS.workflowUsers(workflowId)],
278+
keys: [
279+
KEYS.workflowUsers(workflowId),
280+
KEYS.socketWorkflow(socketId),
281+
KEYS.socketSession(socketId),
282+
],
273283
arguments: [
274284
socketId,
275285
updates.cursor ? JSON.stringify(updates.cursor) : '',
276286
updates.selection ? JSON.stringify(updates.selection) : '',
277287
(updates.lastActivity ?? Date.now()).toString(),
288+
SOCKET_KEY_TTL.toString(),
278289
],
279290
})
280291
} catch (error) {

0 commit comments

Comments
 (0)