Merge branch 'main' into fix/cohere-rerank-custom-model

Author: VedantMadane

packages/server/src/controllers/internal-predictions/index.ts

@@ -33,6 +33,7 @@ const createInternalPrediction = async (req: Request, res: Response, next: NextF
 const createAndStreamInternalPrediction = async (req: Request, res: Response, next: NextFunction) => {
     const chatId = req.body.chatId
     const sseStreamer = getRunningExpressApp().sseStreamer
+    const isQueueMode = process.env.MODE === MODE.QUEUE
 
     try {
         sseStreamer.addClient(chatId, res)
@@ -42,8 +43,8 @@ const createAndStreamInternalPrediction = async (req: Request, res: Response, ne
         res.setHeader('X-Accel-Buffering', 'no') //nginx config: https://serverfault.com/a/801629
         res.flushHeaders()
 
-        if (process.env.MODE === MODE.QUEUE) {
-            getRunningExpressApp().redisSubscriber.subscribe(chatId)
+        if (isQueueMode) {
+            await getRunningExpressApp().redisSubscriber.subscribe(chatId)
         }
 
         const apiResponse = await utilBuildChatflow(req, true)
@@ -54,6 +55,9 @@ const createAndStreamInternalPrediction = async (req: Request, res: Response, ne
         }
         next(error)
     } finally {
+        if (isQueueMode && chatId) {
+            await getRunningExpressApp().redisSubscriber.unsubscribe(chatId)
+        }
         sseStreamer.removeClient(chatId)
     }
 }

packages/server/src/controllers/predictions/index.ts

@@ -64,6 +64,7 @@ const createPrediction = async (req: Request, res: Response, next: NextFunction)
                     chatId = req.body.chatId ?? req.body.overrideConfig?.sessionId ?? uuidv4()
                     req.body.chatId = chatId
                 }
+                const isQueueMode = process.env.MODE === MODE.QUEUE
                 try {
                     sseStreamer.addExternalClient(chatId, res)
                     res.setHeader('Content-Type', 'text/event-stream')
@@ -72,8 +73,8 @@ const createPrediction = async (req: Request, res: Response, next: NextFunction)
                     res.setHeader('X-Accel-Buffering', 'no') //nginx config: https://serverfault.com/a/801629
                     res.flushHeaders()
 
-                    if (process.env.MODE === MODE.QUEUE) {
-                        getRunningExpressApp().redisSubscriber.subscribe(chatId)
+                    if (isQueueMode) {
+                        await getRunningExpressApp().redisSubscriber.subscribe(chatId)
                     }
 
                     const apiResponse = await predictionsServices.buildChatflow(req)
@@ -84,6 +85,9 @@ const createPrediction = async (req: Request, res: Response, next: NextFunction)
                     }
                     next(error)
                 } finally {
+                    if (isQueueMode && chatId) {
+                        await getRunningExpressApp().redisSubscriber.unsubscribe(chatId)
+                    }
                     sseStreamer.removeClient(chatId)
                 }
             } else {

packages/server/src/controllers/variables/index.ts

@@ -22,9 +22,12 @@ const createVariable = async (req: Request, res: Response, next: NextFunction) =
             throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Error: toolsController.createTool - workspace ${workspaceId} not found!`)
         }
         const body = req.body
-        body.workspaceId = workspaceId
+        // Explicit allowlist — id/workspaceId/timestamps must not be overrideable by client
         const newVariable = new Variable()
-        Object.assign(newVariable, body)
+        if (body.name !== undefined) newVariable.name = body.name
+        if (body.value !== undefined) newVariable.value = body.value
+        if (body.type !== undefined) newVariable.type = body.type
+        newVariable.workspaceId = workspaceId
         const apiResponse = await variablesService.createVariable(newVariable, orgId)
         return res.json(apiResponse)
     } catch (error) {
@@ -91,8 +94,11 @@ const updateVariable = async (req: Request, res: Response, next: NextFunction) =
             return res.status(404).send('Variable not found in the database')
         }
         const body = req.body
+        // Explicit allowlist — id/workspaceId/timestamps must not be overrideable by client
         const updatedVariable = new Variable()
-        Object.assign(updatedVariable, body)
+        if (body.name !== undefined) updatedVariable.name = body.name
+        if (body.value !== undefined) updatedVariable.value = body.value
+        if (body.type !== undefined) updatedVariable.type = body.type
         const apiResponse = await variablesService.updateVariable(variable, updatedVariable)
         return res.json(apiResponse)
     } catch (error) {

packages/server/src/enterprise/controllers/user.controller.ts

@@ -56,11 +56,11 @@ export class UserController {
             if (!currentUser) {
                 throw new InternalFlowiseError(StatusCodes.UNAUTHORIZED, UserErrorMessage.USER_NOT_FOUND)
             }
-            const { id } = req.body
+            const { id, name, oldPassword, newPassword, confirmPassword } = req.body
             if (currentUser.id !== id) {
                 throw new InternalFlowiseError(StatusCodes.FORBIDDEN, UserErrorMessage.USER_NOT_FOUND)
             }
-            const user = await userService.updateUser(req.body)
+            const user = await userService.updateUser({ id, name, updatedBy: currentUser.id, oldPassword, newPassword, confirmPassword })
             return res.status(StatusCodes.OK).json(user)
         } catch (error) {
             next(error)

packages/server/src/enterprise/services/user.service.ts

@@ -150,16 +150,10 @@ export class UserService {
                 if (!updateUserData) throw new InternalFlowiseError(StatusCodes.NOT_FOUND, UserErrorMessage.USER_NOT_FOUND)
             }
 
-            newUserData.createdBy = oldUserData.createdBy
-
             if (newUserData.name) {
                 this.validateUserName(newUserData.name)
             }
 
-            if (newUserData.status) {
-                this.validateUserStatus(newUserData.status)
-            }
-
             if (newUserData.oldPassword && newUserData.newPassword && newUserData.confirmPassword) {
                 if (!oldUserData.credential) {
                     throw new InternalFlowiseError(StatusCodes.BAD_REQUEST, UserErrorMessage.INVALID_USER_CREDENTIAL)
@@ -176,7 +170,23 @@ export class UserService {
                 newUserData.tokenExpiry = undefined
             }
 
-            updatedUser = queryRunner.manager.merge(User, oldUserData, newUserData)
+            const safePatch: Partial<User> = {
+                createdBy: oldUserData.createdBy // always preserve from DB
+            }
+
+            if (newUserData.name) {
+                safePatch.name = newUserData.name
+            }
+
+            safePatch.updatedBy = newUserData.updatedBy // always set (controller forces req.user.id)
+            if (newUserData.oldPassword && newUserData.newPassword && newUserData.confirmPassword) {
+                // credential/tempToken/tokenExpiry were set by the validated workflow above
+                safePatch.credential = newUserData.credential
+                safePatch.tempToken = newUserData.tempToken
+                safePatch.tokenExpiry = newUserData.tokenExpiry
+            }
+
+            updatedUser = queryRunner.manager.merge(User, oldUserData, safePatch)
             await queryRunner.startTransaction()
             await this.saveUser(updatedUser, queryRunner)
             await queryRunner.commitTransaction()

packages/server/src/index.ts

@@ -128,6 +128,7 @@ export class App {
 
             // Initialize SSE Streamer
             this.sseStreamer = new SSEStreamer()
+            this.sseStreamer.startHeartbeat()
             logger.info('🌊 [server]: SSE Streamer initialized successfully')
 
             // Init Queues
@@ -148,6 +149,7 @@ export class App {
 
                 this.redisSubscriber = new RedisEventSubscriber(this.sseStreamer)
                 await this.redisSubscriber.connect()
+                this.redisSubscriber.startPeriodicCleanup()
                 logger.info('🔗 [server]: Redis event subscriber connected successfully')
             }
 
@@ -356,6 +358,7 @@ export class App {
 
     async stopApp() {
         try {
+            this.sseStreamer.stopHeartbeat()
             const removePromises: any[] = []
             removePromises.push(this.telemetry.flush())
             if (this.queueManager) {

packages/server/src/queue/PredictionQueue.ts

@@ -63,6 +63,9 @@ export class PredictionQueue extends BaseQueue {
     }
 
     async processJob(data: IExecuteFlowParams | IGenerateAgentflowv2Params) {
+        if (this.redisPublisher) {
+            await this.redisPublisher.connect()
+        }
         if (this.appDataSource) data.appDataSource = this.appDataSource
         if (this.telemetry) data.telemetry = this.telemetry
         if (this.cachePool) data.cachePool = this.cachePool