Feature/Add bullmq redis for message queue processing (#3568)

* add bullmq redis for message queue processing * Update pnpm-lock.yaml * update queue manager * remove singleton patterns, add redis to cache pool * add bull board ui * update rate limit handler * update redis configuration * Merge add rate limit redis prefix * update rate limit queue events * update preview loader to queue * refractor namings to constants * update env variable for queue * update worker shutdown gracefully
2026-06-28 13:00:56 +03:00 · 2025-01-23 14:08:02 +00:00
parent 14adb936f2
commit a2a475ba7a
59 changed files with 38958 additions and 36985 deletions
@@ -1,4 +1,3 @@
-import express from 'express'
 import { Response } from 'express'
 import { IServerSideEventStreamer } from 'flowise-components'

@@ -13,11 +12,6 @@ type Client = {

 export class SSEStreamer implements IServerSideEventStreamer {
    clients: { [id: string]: Client } = {}
-    app: express.Application
-
-    constructor(app: express.Application) {
-        this.app = app
-    }

    addExternalClient(chatId: string, res: Response) {
        this.clients[chatId] = { clientType: 'EXTERNAL', response: res, started: false }
@@ -40,18 +34,6 @@ export class SSEStreamer implements IServerSideEventStreamer {
        }
    }

-    // Send SSE message to a specific client
-    streamEvent(chatId: string, data: string) {
-        const client = this.clients[chatId]
-        if (client) {
-            const clientResponse = {
-                event: 'start',
-                data: data
-            }
-            client.response.write('message:\ndata:' + JSON.stringify(clientResponse) + '\n\n')
-        }
-    }
-
    streamCustomEvent(chatId: string, eventType: string, data: any) {
        const client = this.clients[chatId]
        if (client) {
@@ -1,3 +1,4 @@
+import { DataSource } from 'typeorm'
 import { ChatMessage } from '../database/entities/ChatMessage'
 import { IChatMessage } from '../Interface'
 import { getRunningExpressApp } from '../utils/getRunningExpressApp'
@@ -6,14 +7,14 @@ import { getRunningExpressApp } from '../utils/getRunningExpressApp'
 * Method that add chat messages.
 * @param {Partial<IChatMessage>} chatMessage
 */
-export const utilAddChatMessage = async (chatMessage: Partial<IChatMessage>): Promise<ChatMessage> => {
-    const appServer = getRunningExpressApp()
+export const utilAddChatMessage = async (chatMessage: Partial<IChatMessage>, appDataSource?: DataSource): Promise<ChatMessage> => {
+    const dataSource = appDataSource ?? getRunningExpressApp().AppDataSource
    const newChatMessage = new ChatMessage()
    Object.assign(newChatMessage, chatMessage)
    if (!newChatMessage.createdDate) {
        newChatMessage.createdDate = new Date()
    }
-    const chatmessage = await appServer.AppDataSource.getRepository(ChatMessage).create(newChatMessage)
-    const dbResponse = await appServer.AppDataSource.getRepository(ChatMessage).save(chatmessage)
+    const chatmessage = await dataSource.getRepository(ChatMessage).create(newChatMessage)
+    const dbResponse = await dataSource.getRepository(ChatMessage).save(chatmessage)
    return dbResponse
 }
@@ -19,144 +19,77 @@ import { StatusCodes } from 'http-status-codes'
 import { v4 as uuidv4 } from 'uuid'
 import { StructuredTool } from '@langchain/core/tools'
 import { BaseMessage, HumanMessage, AIMessage, AIMessageChunk, ToolMessage } from '@langchain/core/messages'
-import {
-    IChatFlow,
-    IComponentNodes,
-    IDepthQueue,
-    IReactFlowNode,
-    IReactFlowObject,
-    IReactFlowEdge,
-    IMessage,
-    IncomingInput
-} from '../Interface'
-import {
-    buildFlow,
-    getStartingNodes,
-    getEndingNodes,
-    constructGraphs,
-    databaseEntities,
-    getSessionChatHistory,
-    getMemorySessionId,
-    clearSessionMemory,
-    getAPIOverrideConfig
-} from '../utils'
-import { getRunningExpressApp } from './getRunningExpressApp'
+import { IChatFlow, IComponentNodes, IDepthQueue, IReactFlowNode, IReactFlowEdge, IMessage, IncomingInput, IFlowConfig } from '../Interface'
+import { databaseEntities, clearSessionMemory, getAPIOverrideConfig } from '../utils'
 import { replaceInputsWithConfig, resolveVariables } from '.'
 import { InternalFlowiseError } from '../errors/internalFlowiseError'
 import { getErrorMessage } from '../errors/utils'
 import logger from './logger'
 import { Variable } from '../database/entities/Variable'
+import { DataSource } from 'typeorm'
+import { CachePool } from '../CachePool'

 /**
 * Build Agent Graph
- * @param {IChatFlow} chatflow
- * @param {string} chatId
- * @param {string} sessionId
- * @param {ICommonObject} incomingInput
- * @param {boolean} isInternal
- * @param {string} baseURL
 */
-export const buildAgentGraph = async (
-    chatflow: IChatFlow,
-    chatId: string,
-    apiMessageId: string,
-    sessionId: string,
-    incomingInput: IncomingInput,
-    isInternal: boolean,
-    baseURL?: string,
-    sseStreamer?: IServerSideEventStreamer,
-    shouldStreamResponse?: boolean,
-    uploadedFilesContent?: string
-): Promise<any> => {
+export const buildAgentGraph = async ({
+    agentflow,
+    flowConfig,
+    incomingInput,
+    nodes,
+    edges,
+    initializedNodes,
+    endingNodeIds,
+    startingNodeIds,
+    depthQueue,
+    chatHistory,
+    uploadedFilesContent,
+    appDataSource,
+    componentNodes,
+    sseStreamer,
+    shouldStreamResponse,
+    cachePool,
+    baseURL,
+    signal
+}: {
+    agentflow: IChatFlow
+    flowConfig: IFlowConfig
+    incomingInput: IncomingInput
+    nodes: IReactFlowNode[]
+    edges: IReactFlowEdge[]
+    initializedNodes: IReactFlowNode[]
+    endingNodeIds: string[]
+    startingNodeIds: string[]
+    depthQueue: IDepthQueue
+    chatHistory: IMessage[]
+    uploadedFilesContent: string
+    appDataSource: DataSource
+    componentNodes: IComponentNodes
+    sseStreamer: IServerSideEventStreamer
+    shouldStreamResponse: boolean
+    cachePool: CachePool
+    baseURL: string
+    signal?: AbortController
+}): Promise<any> => {
    try {
-        const appServer = getRunningExpressApp()
-        const chatflowid = chatflow.id
-
-        /*** Get chatflows and prepare data  ***/
-        const flowData = chatflow.flowData
-        const parsedFlowData: IReactFlowObject = JSON.parse(flowData)
-        const nodes = parsedFlowData.nodes
-        const edges = parsedFlowData.edges
-
-        /*** Get Ending Node with Directed Graph  ***/
-        const { graph, nodeDependencies } = constructGraphs(nodes, edges)
-        const directedGraph = graph
-
-        const endingNodes = getEndingNodes(nodeDependencies, directedGraph, nodes)
-
-        /*** Get Starting Nodes with Reversed Graph ***/
-        const constructedObj = constructGraphs(nodes, edges, { isReversed: true })
-        const nonDirectedGraph = constructedObj.graph
-        let startingNodeIds: string[] = []
-        let depthQueue: IDepthQueue = {}
-        const endingNodeIds = endingNodes.map((n) => n.id)
-        for (const endingNodeId of endingNodeIds) {
-            const resx = getStartingNodes(nonDirectedGraph, endingNodeId)
-            startingNodeIds.push(...resx.startingNodeIds)
-            depthQueue = Object.assign(depthQueue, resx.depthQueue)
-        }
-        startingNodeIds = [...new Set(startingNodeIds)]
-
-        /*** Get Memory Node for Chat History ***/
-        let chatHistory: IMessage[] = []
-        const agentMemoryList = ['agentMemory', 'sqliteAgentMemory', 'postgresAgentMemory', 'mySQLAgentMemory']
-        const memoryNode = nodes.find((node) => agentMemoryList.includes(node.data.name))
-        if (memoryNode) {
-            chatHistory = await getSessionChatHistory(
-                chatflowid,
-                getMemorySessionId(memoryNode, incomingInput, chatId, isInternal),
-                memoryNode,
-                appServer.nodesPool.componentNodes,
-                appServer.AppDataSource,
-                databaseEntities,
-                logger,
-                incomingInput.history
-            )
-        }
-
-        /*** Get API Config ***/
-        const availableVariables = await appServer.AppDataSource.getRepository(Variable).find()
-        const { nodeOverrides, variableOverrides, apiOverrideStatus } = getAPIOverrideConfig(chatflow)
-
-        // Initialize nodes like ChatModels, Tools, etc.
-        const reactFlowNodes: IReactFlowNode[] = await buildFlow({
-            startingNodeIds,
-            reactFlowNodes: nodes,
-            reactFlowEdges: edges,
-            apiMessageId,
-            graph,
-            depthQueue,
-            componentNodes: appServer.nodesPool.componentNodes,
-            question: incomingInput.question,
-            uploadedFilesContent,
-            chatHistory,
-            chatId,
-            sessionId,
-            chatflowid,
-            appDataSource: appServer.AppDataSource,
-            overrideConfig: incomingInput?.overrideConfig,
-            apiOverrideStatus,
-            nodeOverrides,
-            availableVariables,
-            variableOverrides,
-            cachePool: appServer.cachePool,
-            isUpsert: false,
-            uploads: incomingInput.uploads,
-            baseURL
-        })
+        const chatflowid = flowConfig.chatflowid
+        const chatId = flowConfig.chatId
+        const sessionId = flowConfig.sessionId
+        const analytic = agentflow.analytic
+        const uploads = incomingInput.uploads

        const options = {
            chatId,
            sessionId,
            chatflowid,
            logger,
-            analytic: chatflow.analytic,
-            appDataSource: appServer.AppDataSource,
-            databaseEntities: databaseEntities,
-            cachePool: appServer.cachePool,
-            uploads: incomingInput.uploads,
+            analytic,
+            appDataSource,
+            databaseEntities,
+            cachePool,
+            uploads,
            baseURL,
-            signal: new AbortController()
+            signal: signal ?? new AbortController()
        }

        let streamResults
@@ -171,9 +104,9 @@ export const buildAgentGraph = async (
        let totalUsedTools: IUsedTool[] = []
        let totalArtifacts: ICommonObject[] = []

-        const workerNodes = reactFlowNodes.filter((node) => node.data.name === 'worker')
-        const supervisorNodes = reactFlowNodes.filter((node) => node.data.name === 'supervisor')
-        const seqAgentNodes = reactFlowNodes.filter((node) => node.data.category === 'Sequential Agents')
+        const workerNodes = initializedNodes.filter((node) => node.data.name === 'worker')
+        const supervisorNodes = initializedNodes.filter((node) => node.data.name === 'supervisor')
+        const seqAgentNodes = initializedNodes.filter((node) => node.data.category === 'Sequential Agents')

        const mapNameToLabel: Record<string, { label: string; nodeName: string }> = {}

@@ -189,11 +122,12 @@ export const buildAgentGraph = async (
        try {
            if (!seqAgentNodes.length) {
                streamResults = await compileMultiAgentsGraph({
-                    chatflow,
+                    agentflow,
+                    appDataSource,
                    mapNameToLabel,
-                    reactFlowNodes,
+                    reactFlowNodes: initializedNodes,
                    workerNodeIds: endingNodeIds,
-                    componentNodes: appServer.nodesPool.componentNodes,
+                    componentNodes,
                    options,
                    startingNodeIds,
                    question: incomingInput.question,
@@ -208,10 +142,11 @@ export const buildAgentGraph = async (
                isSequential = true
                streamResults = await compileSeqAgentsGraph({
                    depthQueue,
-                    chatflow,
-                    reactFlowNodes,
+                    agentflow,
+                    appDataSource,
+                    reactFlowNodes: initializedNodes,
                    reactFlowEdges: edges,
-                    componentNodes: appServer.nodesPool.componentNodes,
+                    componentNodes,
                    options,
                    question: incomingInput.question,
                    prependHistoryMessages: incomingInput.history,
@@ -275,7 +210,7 @@ export const buildAgentGraph = async (
                                )

                                inputEdges.forEach((edge) => {
-                                    const parentNode = reactFlowNodes.find((nd) => nd.id === edge.source)
+                                    const parentNode = initializedNodes.find((nd) => nd.id === edge.source)
                                    if (parentNode) {
                                        if (parentNode.data.name.includes('seqCondition')) {
                                            const newMessages = messages.slice(0, -1)
@@ -366,7 +301,7 @@ export const buildAgentGraph = async (
                    // If last message is an AI Message with tool calls, that means the last node was interrupted
                    if (lastMessageRaw.tool_calls && lastMessageRaw.tool_calls.length > 0) {
                        // The last node that got interrupted
-                        const node = reactFlowNodes.find((node) => node.id === lastMessageRaw.additional_kwargs.nodeId)
+                        const node = initializedNodes.find((node) => node.id === lastMessageRaw.additional_kwargs.nodeId)

                        // Find the next tool node that is connected to the interrupted node, to get the approve/reject button text
                        const tooNodeId = edges.find(
@@ -374,7 +309,7 @@ export const buildAgentGraph = async (
                                edge.target.includes('seqToolNode') &&
                                edge.source === (lastMessageRaw.additional_kwargs && lastMessageRaw.additional_kwargs.nodeId)
                        )?.target
-                        const connectedToolNode = reactFlowNodes.find((node) => node.id === tooNodeId)
+                        const connectedToolNode = initializedNodes.find((node) => node.id === tooNodeId)

                        // Map raw tool calls to used tools, to be shown on interrupted message
                        const mappedToolCalls = lastMessageRaw.tool_calls.map((toolCall) => {
@@ -449,7 +384,7 @@ export const buildAgentGraph = async (
            }
        } catch (e) {
            // clear agent memory because checkpoints were saved during runtime
-            await clearSessionMemory(nodes, appServer.nodesPool.componentNodes, chatId, appServer.AppDataSource, sessionId)
+            await clearSessionMemory(nodes, componentNodes, chatId, appDataSource, sessionId)
            if (getErrorMessage(e).includes('Aborted')) {
                if (shouldStreamResponse && sseStreamer) {
                    sseStreamer.streamAbortEvent(chatId)
@@ -466,7 +401,8 @@ export const buildAgentGraph = async (
 }

 type MultiAgentsGraphParams = {
-    chatflow: IChatFlow
+    agentflow: IChatFlow
+    appDataSource: DataSource
    mapNameToLabel: Record<string, { label: string; nodeName: string }>
    reactFlowNodes: IReactFlowNode[]
    workerNodeIds: string[]
@@ -484,13 +420,13 @@ type MultiAgentsGraphParams = {

 const compileMultiAgentsGraph = async (params: MultiAgentsGraphParams) => {
    const {
-        chatflow,
+        agentflow,
+        appDataSource,
        mapNameToLabel,
        reactFlowNodes,
        workerNodeIds,
        componentNodes,
        options,
-        startingNodeIds,
        prependHistoryMessages = [],
        chatHistory = [],
        overrideConfig = {},
@@ -501,7 +437,6 @@ const compileMultiAgentsGraph = async (params: MultiAgentsGraphParams) => {

    let question = params.question

-    const appServer = getRunningExpressApp()
    const channels: ITeamState = {
        messages: {
            value: (x: BaseMessage[], y: BaseMessage[]) => x.concat(y),
@@ -522,8 +457,8 @@ const compileMultiAgentsGraph = async (params: MultiAgentsGraphParams) => {
    const workerNodes = reactFlowNodes.filter((node) => workerNodeIds.includes(node.data.id))

    /*** Get API Config ***/
-    const availableVariables = await appServer.AppDataSource.getRepository(Variable).find()
-    const { nodeOverrides, variableOverrides, apiOverrideStatus } = getAPIOverrideConfig(chatflow)
+    const availableVariables = await appDataSource.getRepository(Variable).find()
+    const { nodeOverrides, variableOverrides, apiOverrideStatus } = getAPIOverrideConfig(agentflow)

    let supervisorWorkers: { [key: string]: IMultiAgentNode[] } = {}

@@ -537,7 +472,6 @@ const compileMultiAgentsGraph = async (params: MultiAgentsGraphParams) => {
        if (overrideConfig && apiOverrideStatus)
            flowNodeData = replaceInputsWithConfig(flowNodeData, overrideConfig, nodeOverrides, variableOverrides)
        flowNodeData = await resolveVariables(
-            appServer.AppDataSource,
            flowNodeData,
            reactFlowNodes,
            question,
@@ -579,7 +513,6 @@ const compileMultiAgentsGraph = async (params: MultiAgentsGraphParams) => {
        if (overrideConfig && apiOverrideStatus)
            flowNodeData = replaceInputsWithConfig(flowNodeData, overrideConfig, nodeOverrides, variableOverrides)
        flowNodeData = await resolveVariables(
-            appServer.AppDataSource,
            flowNodeData,
            reactFlowNodes,
            question,
@@ -626,15 +559,7 @@ const compileMultiAgentsGraph = async (params: MultiAgentsGraphParams) => {

            //@ts-ignore
            workflowGraph.addEdge(START, supervisorResult.name)
-
-            // Add agentflow to pool
            ;(workflowGraph as any).signal = options.signal
-            appServer.chatflowPool.add(
-                `${chatflow.id}_${options.chatId}`,
-                workflowGraph as any,
-                reactFlowNodes.filter((node) => startingNodeIds.includes(node.id)),
-                overrideConfig
-            )

            // Get memory
            let memory = supervisorResult?.checkpointMemory
@@ -685,7 +610,8 @@ const compileMultiAgentsGraph = async (params: MultiAgentsGraphParams) => {

 type SeqAgentsGraphParams = {
    depthQueue: IDepthQueue
-    chatflow: IChatFlow
+    agentflow: IChatFlow
+    appDataSource: DataSource
    reactFlowNodes: IReactFlowNode[]
    reactFlowEdges: IReactFlowEdge[]
    componentNodes: IComponentNodes
@@ -702,7 +628,8 @@ type SeqAgentsGraphParams = {
 const compileSeqAgentsGraph = async (params: SeqAgentsGraphParams) => {
    const {
        depthQueue,
-        chatflow,
+        agentflow,
+        appDataSource,
        reactFlowNodes,
        reactFlowEdges,
        componentNodes,
@@ -717,8 +644,6 @@ const compileSeqAgentsGraph = async (params: SeqAgentsGraphParams) => {

    let question = params.question

-    const appServer = getRunningExpressApp()
-
    let channels: ISeqAgentsState = {
        messages: {
            value: (x: BaseMessage[], y: BaseMessage[]) => x.concat(y),
@@ -761,8 +686,8 @@ const compileSeqAgentsGraph = async (params: SeqAgentsGraphParams) => {
    let interruptToolNodeNames = []

    /*** Get API Config ***/
-    const availableVariables = await appServer.AppDataSource.getRepository(Variable).find()
-    const { nodeOverrides, variableOverrides, apiOverrideStatus } = getAPIOverrideConfig(chatflow)
+    const availableVariables = await appDataSource.getRepository(Variable).find()
+    const { nodeOverrides, variableOverrides, apiOverrideStatus } = getAPIOverrideConfig(agentflow)

    const initiateNode = async (node: IReactFlowNode) => {
        const nodeInstanceFilePath = componentNodes[node.data.name].filePath as string
@@ -773,7 +698,6 @@ const compileSeqAgentsGraph = async (params: SeqAgentsGraphParams) => {
        if (overrideConfig && apiOverrideStatus)
            flowNodeData = replaceInputsWithConfig(flowNodeData, overrideConfig, nodeOverrides, variableOverrides)
        flowNodeData = await resolveVariables(
-            appServer.AppDataSource,
            flowNodeData,
            reactFlowNodes,
            question,
@@ -1059,14 +983,8 @@ const compileSeqAgentsGraph = async (params: SeqAgentsGraphParams) => {
            routeMessage
        )
    }
-    /*** Add agentflow to pool ***/
+
    ;(seqGraph as any).signal = options.signal
-    appServer.chatflowPool.add(
-        `${chatflow.id}_${options.chatId}`,
-        seqGraph as any,
-        reactFlowNodes.filter((node) => startAgentNodes.map((nd) => nd.id).includes(node.id)),
-        overrideConfig
-    )

    /*** Get memory ***/
    const startNode = reactFlowNodes.find((node: IReactFlowNode) => node.data.name === 'seqStart')
@@ -20,6 +20,8 @@ export const WHITELIST_URLS = [
    '/api/v1/metrics'
 ]

+export const OMIT_QUEUE_JOB_DATA = ['componentNodes', 'appDataSource', 'sseStreamer', 'telemetry', 'cachePool']
+
 export const INPUT_PARAMS_TYPE = [
    'asyncOptions',
    'options',
@@ -560,7 +560,6 @@ export const buildFlow = async ({
            if (isUpsert) upsertHistory['flowData'] = saveUpsertFlowData(flowNodeData, upsertHistory)

            const reactFlowNodeData: INodeData = await resolveVariables(
-                appDataSource,
                flowNodeData,
                flowNodes,
                question,
@@ -762,10 +761,9 @@ export const clearSessionMemory = async (
 }

 const getGlobalVariable = async (
-    appDataSource: DataSource,
    overrideConfig?: ICommonObject,
    availableVariables: IVariable[] = [],
-    variableOverrides?: ICommonObject[]
+    variableOverrides: ICommonObject[] = []
 ) => {
    // override variables defined in overrideConfig
    // nodeData.inputs.vars is an Object, check each property and override the variable
@@ -826,13 +824,12 @@ const getGlobalVariable = async (
 * @returns {string}
 */
 export const getVariableValue = async (
-    appDataSource: DataSource,
    paramValue: string | object,
    reactFlowNodes: IReactFlowNode[],
    question: string,
    chatHistory: IMessage[],
    isAcceptVariable = false,
-    flowData?: ICommonObject,
+    flowConfig?: ICommonObject,
    uploadedFilesContent?: string,
    availableVariables: IVariable[] = [],
    variableOverrides: ICommonObject[] = []
@@ -877,7 +874,7 @@ export const getVariableValue = async (
            }

            if (variableFullPath.startsWith('$vars.')) {
-                const vars = await getGlobalVariable(appDataSource, flowData, availableVariables, variableOverrides)
+                const vars = await getGlobalVariable(flowConfig, availableVariables, variableOverrides)
                const variableValue = get(vars, variableFullPath.replace('$vars.', ''))
                if (variableValue != null) {
                    variableDict[`{{${variableFullPath}}}`] = variableValue
@@ -885,8 +882,8 @@ export const getVariableValue = async (
                }
            }

-            if (variableFullPath.startsWith('$flow.') && flowData) {
-                const variableValue = get(flowData, variableFullPath.replace('$flow.', ''))
+            if (variableFullPath.startsWith('$flow.') && flowConfig) {
+                const variableValue = get(flowConfig, variableFullPath.replace('$flow.', ''))
                if (variableValue != null) {
                    variableDict[`{{${variableFullPath}}}`] = variableValue
                    returnVal = returnVal.split(`{{${variableFullPath}}}`).join(variableValue)
@@ -980,12 +977,11 @@ export const getVariableValue = async (
 * @returns {INodeData}
 */
 export const resolveVariables = async (
-    appDataSource: DataSource,
    reactFlowNodeData: INodeData,
    reactFlowNodes: IReactFlowNode[],
    question: string,
    chatHistory: IMessage[],
-    flowData?: ICommonObject,
+    flowConfig?: ICommonObject,
    uploadedFilesContent?: string,
    availableVariables: IVariable[] = [],
    variableOverrides: ICommonObject[] = []
@@ -1000,13 +996,12 @@ export const resolveVariables = async (
                const resolvedInstances = []
                for (const param of paramValue) {
                    const resolvedInstance = await getVariableValue(
-                        appDataSource,
                        param,
                        reactFlowNodes,
                        question,
                        chatHistory,
                        undefined,
-                        flowData,
+                        flowConfig,
                        uploadedFilesContent,
                        availableVariables,
                        variableOverrides
@@ -1017,13 +1012,12 @@ export const resolveVariables = async (
            } else {
                const isAcceptVariable = reactFlowNodeData.inputParams.find((param) => param.name === key)?.acceptVariable ?? false
                const resolvedInstance = await getVariableValue(
-                    appDataSource,
                    paramValue,
                    reactFlowNodes,
                    question,
                    chatHistory,
                    isAcceptVariable,
-                    flowData,
+                    flowConfig,
                    uploadedFilesContent,
                    availableVariables,
                    variableOverrides
@@ -1,55 +1,174 @@
 import { NextFunction, Request, Response } from 'express'
 import { rateLimit, RateLimitRequestHandler } from 'express-rate-limit'
-import { IChatFlow } from '../Interface'
+import { IChatFlow, MODE } from '../Interface'
 import { Mutex } from 'async-mutex'
+import { RedisStore } from 'rate-limit-redis'
+import Redis from 'ioredis'
+import { QueueEvents, QueueEventsListener, QueueEventsProducer } from 'bullmq'

-let rateLimiters: Record<string, RateLimitRequestHandler> = {}
-const rateLimiterMutex = new Mutex()
+interface CustomListener extends QueueEventsListener {
+    updateRateLimiter: (args: { limitDuration: number; limitMax: number; limitMsg: string; id: string }) => void
+}

-async function addRateLimiter(id: string, duration: number, limit: number, message: string) {
-    const release = await rateLimiterMutex.acquire()
-    try {
-        rateLimiters[id] = rateLimit({
-            windowMs: duration * 1000,
-            max: limit,
-            handler: (_, res) => {
-                res.status(429).send(message)
+const QUEUE_NAME = 'ratelimit'
+const QUEUE_EVENT_NAME = 'updateRateLimiter'
+
+export class RateLimiterManager {
+    private rateLimiters: Record<string, RateLimitRequestHandler> = {}
+    private rateLimiterMutex: Mutex = new Mutex()
+    private redisClient: Redis
+    private static instance: RateLimiterManager
+    private queueEventsProducer: QueueEventsProducer
+    private queueEvents: QueueEvents
+
+    constructor() {
+        if (process.env.MODE === MODE.QUEUE) {
+            if (process.env.REDIS_URL) {
+                this.redisClient = new Redis(process.env.REDIS_URL)
+            } else {
+                this.redisClient = new Redis({
+                    host: process.env.REDIS_HOST || 'localhost',
+                    port: parseInt(process.env.REDIS_PORT || '6379'),
+                    username: process.env.REDIS_USERNAME || undefined,
+                    password: process.env.REDIS_PASSWORD || undefined,
+                    tls:
+                        process.env.REDIS_TLS === 'true'
+                            ? {
+                                  cert: process.env.REDIS_CERT ? Buffer.from(process.env.REDIS_CERT, 'base64') : undefined,
+                                  key: process.env.REDIS_KEY ? Buffer.from(process.env.REDIS_KEY, 'base64') : undefined,
+                                  ca: process.env.REDIS_CA ? Buffer.from(process.env.REDIS_CA, 'base64') : undefined
+                              }
+                            : undefined
+                })
            }
-        })
-    } finally {
-        release()
+            this.queueEventsProducer = new QueueEventsProducer(QUEUE_NAME, { connection: this.getConnection() })
+            this.queueEvents = new QueueEvents(QUEUE_NAME, { connection: this.getConnection() })
+        }
+    }
+
+    getConnection() {
+        let tlsOpts = undefined
+        if (process.env.REDIS_URL && process.env.REDIS_URL.startsWith('rediss://')) {
+            tlsOpts = {
+                rejectUnauthorized: false
+            }
+        } else if (process.env.REDIS_TLS === 'true') {
+            tlsOpts = {
+                cert: process.env.REDIS_CERT ? Buffer.from(process.env.REDIS_CERT, 'base64') : undefined,
+                key: process.env.REDIS_KEY ? Buffer.from(process.env.REDIS_KEY, 'base64') : undefined,
+                ca: process.env.REDIS_CA ? Buffer.from(process.env.REDIS_CA, 'base64') : undefined
+            }
+        }
+        return {
+            url: process.env.REDIS_URL || undefined,
+            host: process.env.REDIS_HOST || 'localhost',
+            port: parseInt(process.env.REDIS_PORT || '6379'),
+            username: process.env.REDIS_USERNAME || undefined,
+            password: process.env.REDIS_PASSWORD || undefined,
+            tls: tlsOpts
+        }
+    }
+
+    public static getInstance(): RateLimiterManager {
+        if (!RateLimiterManager.instance) {
+            RateLimiterManager.instance = new RateLimiterManager()
+        }
+        return RateLimiterManager.instance
+    }
+
+    public async addRateLimiter(id: string, duration: number, limit: number, message: string): Promise<void> {
+        const release = await this.rateLimiterMutex.acquire()
+        try {
+            if (process.env.MODE === MODE.QUEUE) {
+                this.rateLimiters[id] = rateLimit({
+                    windowMs: duration * 1000,
+                    max: limit,
+                    standardHeaders: true,
+                    legacyHeaders: false,
+                    message,
+                    store: new RedisStore({
+                        prefix: `rl:${id}`,
+                        // @ts-expect-error - Known issue: the `call` function is not present in @types/ioredis
+                        sendCommand: (...args: string[]) => this.redisClient.call(...args)
+                    })
+                })
+            } else {
+                this.rateLimiters[id] = rateLimit({
+                    windowMs: duration * 1000,
+                    max: limit,
+                    message
+                })
+            }
+        } finally {
+            release()
+        }
+    }
+
+    public removeRateLimiter(id: string): void {
+        if (this.rateLimiters[id]) {
+            delete this.rateLimiters[id]
+        }
+    }
+
+    public getRateLimiter(): (req: Request, res: Response, next: NextFunction) => void {
+        return (req: Request, res: Response, next: NextFunction) => {
+            const id = req.params.id
+            if (!this.rateLimiters[id]) return next()
+            const idRateLimiter = this.rateLimiters[id]
+            return idRateLimiter(req, res, next)
+        }
+    }
+
+    public async updateRateLimiter(chatFlow: IChatFlow, isInitialized?: boolean): Promise<void> {
+        if (!chatFlow.apiConfig) return
+        const apiConfig = JSON.parse(chatFlow.apiConfig)
+
+        const rateLimit: { limitDuration: number; limitMax: number; limitMsg: string; status?: boolean } = apiConfig.rateLimit
+        if (!rateLimit) return
+
+        const { limitDuration, limitMax, limitMsg, status } = rateLimit
+
+        if (!isInitialized && process.env.MODE === MODE.QUEUE && this.queueEventsProducer) {
+            await this.queueEventsProducer.publishEvent({
+                eventName: QUEUE_EVENT_NAME,
+                limitDuration,
+                limitMax,
+                limitMsg,
+                id: chatFlow.id
+            })
+        } else {
+            if (status === false) {
+                this.removeRateLimiter(chatFlow.id)
+            } else if (limitMax && limitDuration && limitMsg) {
+                await this.addRateLimiter(chatFlow.id, limitDuration, limitMax, limitMsg)
+            }
+        }
+    }
+
+    public async initializeRateLimiters(chatflows: IChatFlow[]): Promise<void> {
+        await Promise.all(
+            chatflows.map(async (chatFlow) => {
+                await this.updateRateLimiter(chatFlow, true)
+            })
+        )
+
+        if (process.env.MODE === MODE.QUEUE && this.queueEvents) {
+            this.queueEvents.on<CustomListener>(
+                QUEUE_EVENT_NAME,
+                async ({
+                    limitDuration,
+                    limitMax,
+                    limitMsg,
+                    id
+                }: {
+                    limitDuration: number
+                    limitMax: number
+                    limitMsg: string
+                    id: string
+                }) => {
+                    await this.addRateLimiter(id, limitDuration, limitMax, limitMsg)
+                }
+            )
+        }
    }
 }
-
-function removeRateLimit(id: string) {
-    if (rateLimiters[id]) {
-        delete rateLimiters[id]
-    }
-}
-
-export function getRateLimiter(req: Request, res: Response, next: NextFunction) {
-    const id = req.params.id
-    if (!rateLimiters[id]) return next()
-    const idRateLimiter = rateLimiters[id]
-    return idRateLimiter(req, res, next)
-}
-
-export async function updateRateLimiter(chatFlow: IChatFlow) {
-    if (!chatFlow.apiConfig) return
-    const apiConfig = JSON.parse(chatFlow.apiConfig)
-
-    const rateLimit: { limitDuration: number; limitMax: number; limitMsg: string; status?: boolean } = apiConfig.rateLimit
-    if (!rateLimit) return
-
-    const { limitDuration, limitMax, limitMsg, status } = rateLimit
-    if (status === false) removeRateLimit(chatFlow.id)
-    else if (limitMax && limitDuration && limitMsg) await addRateLimiter(chatFlow.id, limitDuration, limitMax, limitMsg)
-}
-
-export async function initializeRateLimiter(chatFlowPool: IChatFlow[]) {
-    await Promise.all(
-        chatFlowPool.map(async (chatFlow) => {
-            await updateRateLimiter(chatFlow)
-        })
-    )
-}
@@ -23,7 +23,7 @@ import {
    getAPIOverrideConfig
 } from '../utils'
 import { validateChatflowAPIKey } from './validateKey'
-import { IncomingInput, INodeDirectedGraph, IReactFlowObject, ChatType } from '../Interface'
+import { IncomingInput, INodeDirectedGraph, IReactFlowObject, ChatType, IExecuteFlowParams, MODE } from '../Interface'
 import { ChatFlow } from '../database/entities/ChatFlow'
 import { getRunningExpressApp } from '../utils/getRunningExpressApp'
 import { UpsertHistory } from '../database/entities/UpsertHistory'
@@ -33,17 +33,182 @@ import { getErrorMessage } from '../errors/utils'
 import { v4 as uuidv4 } from 'uuid'
 import { FLOWISE_COUNTER_STATUS, FLOWISE_METRIC_COUNTERS } from '../Interface.Metrics'
 import { Variable } from '../database/entities/Variable'
+import { OMIT_QUEUE_JOB_DATA } from './constants'
+
+export const executeUpsert = async ({
+    componentNodes,
+    incomingInput,
+    chatflow,
+    chatId,
+    appDataSource,
+    telemetry,
+    cachePool,
+    isInternal,
+    files
+}: IExecuteFlowParams) => {
+    const question = incomingInput.question
+    const overrideConfig = incomingInput.overrideConfig ?? {}
+    let stopNodeId = incomingInput?.stopNodeId ?? ''
+    const chatHistory: IMessage[] = []
+    const isUpsert = true
+    const chatflowid = chatflow.id
+    const apiMessageId = uuidv4()
+
+    if (files?.length) {
+        const overrideConfig: ICommonObject = { ...incomingInput }
+        for (const file of files) {
+            const fileNames: string[] = []
+            const fileBuffer = await getFileFromUpload(file.path ?? file.key)
+            // Address file name with special characters: https://github.com/expressjs/multer/issues/1104
+            file.originalname = Buffer.from(file.originalname, 'latin1').toString('utf8')
+            const storagePath = await addArrayFilesToStorage(file.mimetype, fileBuffer, file.originalname, fileNames, chatflowid)
+
+            const fileInputFieldFromMimeType = mapMimeTypeToInputField(file.mimetype)
+
+            const fileExtension = path.extname(file.originalname)
+
+            const fileInputFieldFromExt = mapExtToInputField(fileExtension)
+
+            let fileInputField = 'txtFile'
+
+            if (fileInputFieldFromExt !== 'txtFile') {
+                fileInputField = fileInputFieldFromExt
+            } else if (fileInputFieldFromMimeType !== 'txtFile') {
+                fileInputField = fileInputFieldFromExt
+            }
+
+            if (overrideConfig[fileInputField]) {
+                const existingFileInputField = overrideConfig[fileInputField].replace('FILE-STORAGE::', '')
+                const existingFileInputFieldArray = JSON.parse(existingFileInputField)
+
+                const newFileInputField = storagePath.replace('FILE-STORAGE::', '')
+                const newFileInputFieldArray = JSON.parse(newFileInputField)
+
+                const updatedFieldArray = existingFileInputFieldArray.concat(newFileInputFieldArray)
+
+                overrideConfig[fileInputField] = `FILE-STORAGE::${JSON.stringify(updatedFieldArray)}`
+            } else {
+                overrideConfig[fileInputField] = storagePath
+            }
+
+            await removeSpecificFileFromUpload(file.path ?? file.key)
+        }
+        if (overrideConfig.vars && typeof overrideConfig.vars === 'string') {
+            overrideConfig.vars = JSON.parse(overrideConfig.vars)
+        }
+        incomingInput = {
+            ...incomingInput,
+            question: '',
+            overrideConfig,
+            stopNodeId,
+            chatId
+        }
+    }
+
+    /*** Get chatflows and prepare data  ***/
+    const flowData = chatflow.flowData
+    const parsedFlowData: IReactFlowObject = JSON.parse(flowData)
+    const nodes = parsedFlowData.nodes
+    const edges = parsedFlowData.edges
+
+    /*** Get session ID ***/
+    const memoryNode = findMemoryNode(nodes, edges)
+    let sessionId = getMemorySessionId(memoryNode, incomingInput, chatId, isInternal)
+
+    /*** Find the 1 final vector store will be upserted  ***/
+    const vsNodes = nodes.filter((node) => node.data.category === 'Vector Stores')
+    const vsNodesWithFileUpload = vsNodes.filter((node) => node.data.inputs?.fileUpload)
+    if (vsNodesWithFileUpload.length > 1) {
+        throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, 'Multiple vector store nodes with fileUpload enabled')
+    } else if (vsNodesWithFileUpload.length === 1 && !stopNodeId) {
+        stopNodeId = vsNodesWithFileUpload[0].data.id
+    }
+
+    /*** Check if multiple vector store nodes exist, and if stopNodeId is specified ***/
+    if (vsNodes.length > 1 && !stopNodeId) {
+        throw new InternalFlowiseError(
+            StatusCodes.INTERNAL_SERVER_ERROR,
+            'There are multiple vector nodes, please provide stopNodeId in body request'
+        )
+    } else if (vsNodes.length === 1 && !stopNodeId) {
+        stopNodeId = vsNodes[0].data.id
+    } else if (!vsNodes.length && !stopNodeId) {
+        throw new InternalFlowiseError(StatusCodes.NOT_FOUND, 'No vector node found')
+    }
+
+    /*** Get Starting Nodes with Reversed Graph ***/
+    const { graph } = constructGraphs(nodes, edges, { isReversed: true })
+    const nodeIds = getAllConnectedNodes(graph, stopNodeId)
+    const filteredGraph: INodeDirectedGraph = {}
+    for (const key of nodeIds) {
+        if (Object.prototype.hasOwnProperty.call(graph, key)) {
+            filteredGraph[key] = graph[key]
+        }
+    }
+    const { startingNodeIds, depthQueue } = getStartingNodes(filteredGraph, stopNodeId)
+
+    /*** Get API Config ***/
+    const availableVariables = await appDataSource.getRepository(Variable).find()
+    const { nodeOverrides, variableOverrides, apiOverrideStatus } = getAPIOverrideConfig(chatflow)
+
+    const upsertedResult = await buildFlow({
+        startingNodeIds,
+        reactFlowNodes: nodes,
+        reactFlowEdges: edges,
+        apiMessageId,
+        graph: filteredGraph,
+        depthQueue,
+        componentNodes,
+        question,
+        chatHistory,
+        chatId,
+        sessionId,
+        chatflowid,
+        appDataSource,
+        overrideConfig,
+        apiOverrideStatus,
+        nodeOverrides,
+        availableVariables,
+        variableOverrides,
+        cachePool,
+        isUpsert,
+        stopNodeId
+    })
+
+    // Save to DB
+    if (upsertedResult['flowData'] && upsertedResult['result']) {
+        const result = cloneDeep(upsertedResult)
+        result['flowData'] = JSON.stringify(result['flowData'])
+        result['result'] = JSON.stringify(omit(result['result'], ['totalKeys', 'addedDocs']))
+        result.chatflowid = chatflowid
+        const newUpsertHistory = new UpsertHistory()
+        Object.assign(newUpsertHistory, result)
+        const upsertHistory = appDataSource.getRepository(UpsertHistory).create(newUpsertHistory)
+        await appDataSource.getRepository(UpsertHistory).save(upsertHistory)
+    }
+
+    await telemetry.sendTelemetry('vector_upserted', {
+        version: await getAppVersion(),
+        chatlowId: chatflowid,
+        type: isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
+        flowGraph: getTelemetryFlowObj(nodes, edges),
+        stopNodeId
+    })
+
+    return upsertedResult['result'] ?? { result: 'Successfully Upserted' }
+}
+
 /**
 * Upsert documents
 * @param {Request} req
 * @param {boolean} isInternal
 */
 export const upsertVector = async (req: Request, isInternal: boolean = false) => {
+    const appServer = getRunningExpressApp()
    try {
-        const appServer = getRunningExpressApp()
        const chatflowid = req.params.id
-        let incomingInput: IncomingInput = req.body

+        // Check if chatflow exists
        const chatflow = await appServer.AppDataSource.getRepository(ChatFlow).findOneBy({
            id: chatflowid
        })
@@ -51,6 +216,12 @@ export const upsertVector = async (req: Request, isInternal: boolean = false) =>
            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Chatflow ${chatflowid} not found`)
        }

+        const httpProtocol = req.get('x-forwarded-proto') || req.protocol
+        const baseURL = `${httpProtocol}://${req.get('host')}`
+        const incomingInput: IncomingInput = req.body
+        const chatId = incomingInput.chatId ?? incomingInput.overrideConfig?.sessionId ?? uuidv4()
+        const files = (req.files as Express.Multer.File[]) || []
+
        if (!isInternal) {
            const isKeyValidated = await validateChatflowAPIKey(req, chatflow)
            if (!isKeyValidated) {
@@ -58,168 +229,50 @@ export const upsertVector = async (req: Request, isInternal: boolean = false) =>
            }
        }

-        const files = (req.files as Express.Multer.File[]) || []
-
-        if (files.length) {
-            const overrideConfig: ICommonObject = { ...req.body }
-            for (const file of files) {
-                const fileNames: string[] = []
-                const fileBuffer = await getFileFromUpload(file.path ?? file.key)
-                // Address file name with special characters: https://github.com/expressjs/multer/issues/1104
-                file.originalname = Buffer.from(file.originalname, 'latin1').toString('utf8')
-                const storagePath = await addArrayFilesToStorage(file.mimetype, fileBuffer, file.originalname, fileNames, chatflowid)
-
-                const fileInputFieldFromMimeType = mapMimeTypeToInputField(file.mimetype)
-
-                const fileExtension = path.extname(file.originalname)
-
-                const fileInputFieldFromExt = mapExtToInputField(fileExtension)
-
-                let fileInputField = 'txtFile'
-
-                if (fileInputFieldFromExt !== 'txtFile') {
-                    fileInputField = fileInputFieldFromExt
-                } else if (fileInputFieldFromMimeType !== 'txtFile') {
-                    fileInputField = fileInputFieldFromExt
-                }
-
-                if (overrideConfig[fileInputField]) {
-                    const existingFileInputField = overrideConfig[fileInputField].replace('FILE-STORAGE::', '')
-                    const existingFileInputFieldArray = JSON.parse(existingFileInputField)
-
-                    const newFileInputField = storagePath.replace('FILE-STORAGE::', '')
-                    const newFileInputFieldArray = JSON.parse(newFileInputField)
-
-                    const updatedFieldArray = existingFileInputFieldArray.concat(newFileInputFieldArray)
-
-                    overrideConfig[fileInputField] = `FILE-STORAGE::${JSON.stringify(updatedFieldArray)}`
-                } else {
-                    overrideConfig[fileInputField] = storagePath
-                }
-
-                await removeSpecificFileFromUpload(file.path ?? file.key)
-            }
-            if (overrideConfig.vars && typeof overrideConfig.vars === 'string') {
-                overrideConfig.vars = JSON.parse(overrideConfig.vars)
-            }
-            incomingInput = {
-                question: req.body.question ?? 'hello',
-                overrideConfig,
-                stopNodeId: req.body.stopNodeId
-            }
-            if (req.body.chatId) {
-                incomingInput.chatId = req.body.chatId
-            }
-        }
-
-        /*** Get chatflows and prepare data  ***/
-        const flowData = chatflow.flowData
-        const parsedFlowData: IReactFlowObject = JSON.parse(flowData)
-        const nodes = parsedFlowData.nodes
-        const edges = parsedFlowData.edges
-
-        const apiMessageId = req.body.apiMessageId ?? uuidv4()
-
-        let stopNodeId = incomingInput?.stopNodeId ?? ''
-        let chatHistory: IMessage[] = []
-        let chatId = incomingInput.chatId ?? ''
-        let isUpsert = true
-
-        // Get session ID
-        const memoryNode = findMemoryNode(nodes, edges)
-        let sessionId = getMemorySessionId(memoryNode, incomingInput, chatId, isInternal)
-
-        const vsNodes = nodes.filter((node) => node.data.category === 'Vector Stores')
-
-        // Get StopNodeId for vector store which has fielUpload
-        const vsNodesWithFileUpload = vsNodes.filter((node) => node.data.inputs?.fileUpload)
-        if (vsNodesWithFileUpload.length > 1) {
-            throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, 'Multiple vector store nodes with fileUpload enabled')
-        } else if (vsNodesWithFileUpload.length === 1 && !stopNodeId) {
-            stopNodeId = vsNodesWithFileUpload[0].data.id
-        }
-
-        // Check if multiple vector store nodes exist, and if stopNodeId is specified
-        if (vsNodes.length > 1 && !stopNodeId) {
-            throw new InternalFlowiseError(
-                StatusCodes.INTERNAL_SERVER_ERROR,
-                'There are multiple vector nodes, please provide stopNodeId in body request'
-            )
-        } else if (vsNodes.length === 1 && !stopNodeId) {
-            stopNodeId = vsNodes[0].data.id
-        } else if (!vsNodes.length && !stopNodeId) {
-            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, 'No vector node found')
-        }
-
-        const { graph } = constructGraphs(nodes, edges, { isReversed: true })
-
-        const nodeIds = getAllConnectedNodes(graph, stopNodeId)
-
-        const filteredGraph: INodeDirectedGraph = {}
-        for (const key of nodeIds) {
-            if (Object.prototype.hasOwnProperty.call(graph, key)) {
-                filteredGraph[key] = graph[key]
-            }
-        }
-
-        const { startingNodeIds, depthQueue } = getStartingNodes(filteredGraph, stopNodeId)
-
-        /*** Get API Config ***/
-        const availableVariables = await appServer.AppDataSource.getRepository(Variable).find()
-        const { nodeOverrides, variableOverrides, apiOverrideStatus } = getAPIOverrideConfig(chatflow)
-
-        const upsertedResult = await buildFlow({
-            startingNodeIds,
-            reactFlowNodes: nodes,
-            reactFlowEdges: edges,
-            apiMessageId,
-            graph: filteredGraph,
-            depthQueue,
+        const executeData: IExecuteFlowParams = {
            componentNodes: appServer.nodesPool.componentNodes,
-            question: incomingInput.question,
-            chatHistory,
+            incomingInput,
+            chatflow,
            chatId,
-            sessionId: sessionId ?? '',
-            chatflowid,
            appDataSource: appServer.AppDataSource,
-            overrideConfig: incomingInput?.overrideConfig,
-            apiOverrideStatus,
-            nodeOverrides,
-            availableVariables,
-            variableOverrides,
+            telemetry: appServer.telemetry,
            cachePool: appServer.cachePool,
-            isUpsert,
-            stopNodeId
-        })
-
-        const startingNodes = nodes.filter((nd) => startingNodeIds.includes(nd.data.id))
-
-        await appServer.chatflowPool.add(chatflowid, undefined, startingNodes, incomingInput?.overrideConfig, chatId)
-
-        // Save to DB
-        if (upsertedResult['flowData'] && upsertedResult['result']) {
-            const result = cloneDeep(upsertedResult)
-            result['flowData'] = JSON.stringify(result['flowData'])
-            result['result'] = JSON.stringify(omit(result['result'], ['totalKeys', 'addedDocs']))
-            result.chatflowid = chatflowid
-            const newUpsertHistory = new UpsertHistory()
-            Object.assign(newUpsertHistory, result)
-            const upsertHistory = appServer.AppDataSource.getRepository(UpsertHistory).create(newUpsertHistory)
-            await appServer.AppDataSource.getRepository(UpsertHistory).save(upsertHistory)
+            sseStreamer: appServer.sseStreamer,
+            baseURL,
+            isInternal,
+            files,
+            isUpsert: true
        }

-        await appServer.telemetry.sendTelemetry('vector_upserted', {
-            version: await getAppVersion(),
-            chatlowId: chatflowid,
-            type: isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
-            flowGraph: getTelemetryFlowObj(nodes, edges),
-            stopNodeId
-        })
-        appServer.metricsProvider?.incrementCounter(FLOWISE_METRIC_COUNTERS.VECTORSTORE_UPSERT, { status: FLOWISE_COUNTER_STATUS.SUCCESS })
+        if (process.env.MODE === MODE.QUEUE) {
+            const upsertQueue = appServer.queueManager.getQueue('upsert')

-        return upsertedResult['result'] ?? { result: 'Successfully Upserted' }
+            const job = await upsertQueue.addJob(omit(executeData, OMIT_QUEUE_JOB_DATA))
+            logger.debug(`[server]: Job added to queue: ${job.id}`)
+
+            const queueEvents = upsertQueue.getQueueEvents()
+            const result = await job.waitUntilFinished(queueEvents)
+
+            if (!result) {
+                throw new Error('Job execution failed')
+            }
+
+            appServer.metricsProvider?.incrementCounter(FLOWISE_METRIC_COUNTERS.VECTORSTORE_UPSERT, {
+                status: FLOWISE_COUNTER_STATUS.SUCCESS
+            })
+            return result
+        } else {
+            const result = await executeUpsert(executeData)
+
+            appServer.metricsProvider?.incrementCounter(FLOWISE_METRIC_COUNTERS.VECTORSTORE_UPSERT, {
+                status: FLOWISE_COUNTER_STATUS.SUCCESS
+            })
+            return result
+        }
    } catch (e) {
        logger.error('[server]: Error:', e)
+        appServer.metricsProvider?.incrementCounter(FLOWISE_METRIC_COUNTERS.VECTORSTORE_UPSERT, { status: FLOWISE_COUNTER_STATUS.FAILURE })
+
        if (e instanceof InternalFlowiseError && e.statusCode === StatusCodes.UNAUTHORIZED) {
            throw e
        } else {