Flowise

1
import { StatusCodes } from 'http-status-codes'
2
import { InternalFlowiseError } from '../../errors/internalFlowiseError'
3
import { getRunningExpressApp } from '../../utils/getRunningExpressApp'
4
import { IChatFlow } from '../../Interface'
5
import { ChatFlow } from '../../database/entities/ChatFlow'
6
import { getAppVersion, getTelemetryFlowObj, isFlowValidForStream, constructGraphs, getEndingNodes } from '../../utils'
7
import logger from '../../utils/logger'
8
import { removeFolderFromStorage } from 'flowise-components'
9
import { IReactFlowObject } from '../../Interface'
10
import { utilGetUploadsConfig } from '../../utils/getUploadsConfig'
11
import { ChatMessage } from '../../database/entities/ChatMessage'
12
import { ChatMessageFeedback } from '../../database/entities/ChatMessageFeedback'
13
import { UpsertHistory } from '../../database/entities/UpsertHistory'
14
import { containsBase64File, updateFlowDataWithFilePaths } from '../../utils/fileRepository'
15
import { getErrorMessage } from '../../errors/utils'
16
import documentStoreService from '../../services/documentstore'
17

18
// Check if chatflow valid for streaming
19
const checkIfChatflowIsValidForStreaming = async (chatflowId: string): Promise<any> => {
20
    try {
21
        const appServer = getRunningExpressApp()
22
        //**
23
        const chatflow = await appServer.AppDataSource.getRepository(ChatFlow).findOneBy({
24
            id: chatflowId
25
        })
26
        if (!chatflow) {
27
            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Chatflow ${chatflowId} not found`)
28
        }
29

30
        /*** Get Ending Node with Directed Graph  ***/
31
        const flowData = chatflow.flowData
32
        const parsedFlowData: IReactFlowObject = JSON.parse(flowData)
33
        const nodes = parsedFlowData.nodes
34
        const edges = parsedFlowData.edges
35
        const { graph, nodeDependencies } = constructGraphs(nodes, edges)
36

37
        const endingNodes = getEndingNodes(nodeDependencies, graph, nodes)
38

39
        let isStreaming = false
40
        for (const endingNode of endingNodes) {
41
            const endingNodeData = endingNode.data
42
            const isEndingNode = endingNodeData?.outputs?.output === 'EndingNode'
43
            // Once custom function ending node exists, flow is always unavailable to stream
44
            if (isEndingNode) {
45
                return { isStreaming: false }
46
            }
47
            isStreaming = isFlowValidForStream(nodes, endingNodeData)
48
        }
49

50
        const dbResponse = { isStreaming: isStreaming }
51
        return dbResponse
52
    } catch (error) {
53
        throw new InternalFlowiseError(
54
            StatusCodes.INTERNAL_SERVER_ERROR,
55
            `Error: chatflowsService.checkIfChatflowIsValidForStreaming - ${getErrorMessage(error)}`
56
        )
57
    }
58
}
59

60
// Check if chatflow valid for uploads
61
const checkIfChatflowIsValidForUploads = async (chatflowId: string): Promise<any> => {
62
    try {
63
        const dbResponse = await utilGetUploadsConfig(chatflowId)
64
        return dbResponse
65
    } catch (error) {
66
        throw new InternalFlowiseError(
67
            StatusCodes.INTERNAL_SERVER_ERROR,
68
            `Error: chatflowsService.checkIfChatflowIsValidForUploads - ${getErrorMessage(error)}`
69
        )
70
    }
71
}
72

73
const deleteChatflow = async (chatflowId: string): Promise<any> => {
74
    try {
75
        const appServer = getRunningExpressApp()
76
        const dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).delete({ id: chatflowId })
77
        try {
78
            // Delete all uploads corresponding to this chatflow
79
            await removeFolderFromStorage(chatflowId)
80
            await documentStoreService.updateDocumentStoreUsage(chatflowId, undefined)
81

82
            // Delete all chat messages
83
            await appServer.AppDataSource.getRepository(ChatMessage).delete({ chatflowid: chatflowId })
84

85
            // Delete all chat feedback
86
            await appServer.AppDataSource.getRepository(ChatMessageFeedback).delete({ chatflowid: chatflowId })
87

88
            // Delete all upsert history
89
            await appServer.AppDataSource.getRepository(UpsertHistory).delete({ chatflowid: chatflowId })
90
        } catch (e) {
91
            logger.error(`[server]: Error deleting file storage for chatflow ${chatflowId}: ${e}`)
92
        }
93
        return dbResponse
94
    } catch (error) {
95
        throw new InternalFlowiseError(
96
            StatusCodes.INTERNAL_SERVER_ERROR,
97
            `Error: chatflowsService.deleteChatflow - ${getErrorMessage(error)}`
98
        )
99
    }
100
}
101

102
const getAllChatflows = async (): Promise<IChatFlow[]> => {
103
    try {
104
        const appServer = getRunningExpressApp()
105
        const dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).find()
106
        return dbResponse
107
    } catch (error) {
108
        throw new InternalFlowiseError(
109
            StatusCodes.INTERNAL_SERVER_ERROR,
110
            `Error: chatflowsService.getAllChatflows - ${getErrorMessage(error)}`
111
        )
112
    }
113
}
114

115
const getChatflowByApiKey = async (apiKeyId: string): Promise<any> => {
116
    try {
117
        const appServer = getRunningExpressApp()
118
        const dbResponse = await appServer.AppDataSource.getRepository(ChatFlow)
119
            .createQueryBuilder('cf')
120
            .where('cf.apikeyid = :apikeyid', { apikeyid: apiKeyId })
121
            .orWhere('cf.apikeyid IS NULL')
122
            .orWhere('cf.apikeyid = ""')
123
            .orderBy('cf.name', 'ASC')
124
            .getMany()
125
        if (dbResponse.length < 1) {
126
            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Chatflow not found in the database!`)
127
        }
128
        return dbResponse
129
    } catch (error) {
130
        throw new InternalFlowiseError(
131
            StatusCodes.INTERNAL_SERVER_ERROR,
132
            `Error: chatflowsService.getChatflowByApiKey - ${getErrorMessage(error)}`
133
        )
134
    }
135
}
136

137
const getChatflowById = async (chatflowId: string): Promise<any> => {
138
    try {
139
        const appServer = getRunningExpressApp()
140
        const dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).findOneBy({
141
            id: chatflowId
142
        })
143
        if (!dbResponse) {
144
            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Chatflow ${chatflowId} not found in the database!`)
145
        }
146
        return dbResponse
147
    } catch (error) {
148
        throw new InternalFlowiseError(
149
            StatusCodes.INTERNAL_SERVER_ERROR,
150
            `Error: chatflowsService.getChatflowById - ${getErrorMessage(error)}`
151
        )
152
    }
153
}
154

155
const saveChatflow = async (newChatFlow: ChatFlow): Promise<any> => {
156
    try {
157
        const appServer = getRunningExpressApp()
158
        let dbResponse: ChatFlow
159
        if (containsBase64File(newChatFlow)) {
160
            // we need a 2-step process, as we need to save the chatflow first and then update the file paths
161
            // this is because we need the chatflow id to create the file paths
162

163
            // step 1 - save with empty flowData
164
            const incomingFlowData = newChatFlow.flowData
165
            newChatFlow.flowData = JSON.stringify({})
166
            const chatflow = appServer.AppDataSource.getRepository(ChatFlow).create(newChatFlow)
167
            const step1Results = await appServer.AppDataSource.getRepository(ChatFlow).save(chatflow)
168

169
            // step 2 - convert base64 to file paths and update the chatflow
170
            step1Results.flowData = await updateFlowDataWithFilePaths(step1Results.id, incomingFlowData)
171
            await _checkAndUpdateDocumentStoreUsage(step1Results)
172
            dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(step1Results)
173
        } else {
174
            const chatflow = appServer.AppDataSource.getRepository(ChatFlow).create(newChatFlow)
175
            dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(chatflow)
176
        }
177
        await appServer.telemetry.sendTelemetry('chatflow_created', {
178
            version: await getAppVersion(),
179
            chatflowId: dbResponse.id,
180
            flowGraph: getTelemetryFlowObj(JSON.parse(dbResponse.flowData)?.nodes, JSON.parse(dbResponse.flowData)?.edges)
181
        })
182
        return dbResponse
183
    } catch (error) {
184
        throw new InternalFlowiseError(
185
            StatusCodes.INTERNAL_SERVER_ERROR,
186
            `Error: chatflowsService.saveChatflow - ${getErrorMessage(error)}`
187
        )
188
    }
189
}
190

191
const updateChatflow = async (chatflow: ChatFlow, updateChatFlow: ChatFlow): Promise<any> => {
192
    try {
193
        const appServer = getRunningExpressApp()
194
        if (updateChatFlow.flowData && containsBase64File(updateChatFlow)) {
195
            updateChatFlow.flowData = await updateFlowDataWithFilePaths(chatflow.id, updateChatFlow.flowData)
196
        }
197
        const newDbChatflow = appServer.AppDataSource.getRepository(ChatFlow).merge(chatflow, updateChatFlow)
198
        await _checkAndUpdateDocumentStoreUsage(newDbChatflow)
199
        const dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(newDbChatflow)
200

201
        // chatFlowPool is initialized only when a flow is opened
202
        // if the user attempts to rename/update category without opening any flow, chatFlowPool will be undefined
203
        if (appServer.chatflowPool) {
204
            // Update chatflowpool inSync to false, to build flow from scratch again because data has been changed
205
            appServer.chatflowPool.updateInSync(chatflow.id, false)
206
        }
207
        return dbResponse
208
    } catch (error) {
209
        throw new InternalFlowiseError(
210
            StatusCodes.INTERNAL_SERVER_ERROR,
211
            `Error: chatflowsService.updateChatflow - ${getErrorMessage(error)}`
212
        )
213
    }
214
}
215

216
// Get specific chatflow via id (PUBLIC endpoint, used when sharing chatbot link)
217
const getSinglePublicChatflow = async (chatflowId: string): Promise<any> => {
218
    try {
219
        const appServer = getRunningExpressApp()
220
        const dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).findOneBy({
221
            id: chatflowId
222
        })
223
        if (dbResponse && dbResponse.isPublic) {
224
            return dbResponse
225
        } else if (dbResponse && !dbResponse.isPublic) {
226
            throw new InternalFlowiseError(StatusCodes.UNAUTHORIZED, `Unauthorized`)
227
        }
228
        throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Chatflow ${chatflowId} not found`)
229
    } catch (error) {
230
        throw new InternalFlowiseError(
231
            StatusCodes.INTERNAL_SERVER_ERROR,
232
            `Error: chatflowsService.getSinglePublicChatflow - ${getErrorMessage(error)}`
233
        )
234
    }
235
}
236

237
// Get specific chatflow chatbotConfig via id (PUBLIC endpoint, used to retrieve config for embedded chat)
238
// Safe as public endpoint as chatbotConfig doesn't contain sensitive credential
239
const getSinglePublicChatbotConfig = async (chatflowId: string): Promise<any> => {
240
    try {
241
        const appServer = getRunningExpressApp()
242
        const dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).findOneBy({
243
            id: chatflowId
244
        })
245
        if (!dbResponse) {
246
            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Chatflow ${chatflowId} not found`)
247
        }
248
        const uploadsConfig = await utilGetUploadsConfig(chatflowId)
249
        // even if chatbotConfig is not set but uploads are enabled
250
        // send uploadsConfig to the chatbot
251
        if (dbResponse.chatbotConfig || uploadsConfig) {
252
            try {
253
                const parsedConfig = dbResponse.chatbotConfig ? JSON.parse(dbResponse.chatbotConfig) : {}
254
                return { ...parsedConfig, uploads: uploadsConfig }
255
            } catch (e) {
256
                throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, `Error parsing Chatbot Config for Chatflow ${chatflowId}`)
257
            }
258
        }
259
        return 'OK'
260
    } catch (error) {
261
        throw new InternalFlowiseError(
262
            StatusCodes.INTERNAL_SERVER_ERROR,
263
            `Error: chatflowsService.getSinglePublicChatbotConfig - ${getErrorMessage(error)}`
264
        )
265
    }
266
}
267

268
const _checkAndUpdateDocumentStoreUsage = async (chatflow: ChatFlow) => {
269
    const parsedFlowData: IReactFlowObject = JSON.parse(chatflow.flowData)
270
    const nodes = parsedFlowData.nodes
271
    // from the nodes array find if there is a node with name == documentStore)
272
    const node = nodes.length > 0 && nodes.find((node) => node.data.name === 'documentStore')
273
    if (!node || !node.data || !node.data.inputs || node.data.inputs['selectedStore'] === undefined) {
274
        await documentStoreService.updateDocumentStoreUsage(chatflow.id, undefined)
275
    } else {
276
        await documentStoreService.updateDocumentStoreUsage(chatflow.id, node.data.inputs['selectedStore'])
277
    }
278
}
279

280
export default {
281
    checkIfChatflowIsValidForStreaming,
282
    checkIfChatflowIsValidForUploads,
283
    deleteChatflow,
284
    getAllChatflows,
285
    getChatflowByApiKey,
286
    getChatflowById,
287
    saveChatflow,
288
    updateChatflow,
289
    getSinglePublicChatflow,
290
    getSinglePublicChatbotConfig
291
}
292