Fix chatflow's type null or blank (#5065)

* fix(entities/ChatFlow.ts): make type column non-nullable with default value

* fix(postgres/ModifyChatflowType): set default type and make column non-nullable

* fix(sqlite/ModifyChatflowType): set default type and make column non-nullable

* fix(mysql/ModifyChatflowType): set default type and make column non-nullable

* chore(sqlite/ModifyChatflowType): standardize type column to VARCHAR(20)

* chore(postgres/ModifyChatflowType): standardize type column to VARCHAR(20)

* fix(mariadb/ModifyChatflowType): set default type and make column non-nullable

* chore: rename ChatflowType to EnumChatflowType and update references

* feat(chatflows): add chatflow type validation

* fix(chatflows): empty string bypassing type validation on update

packages/server/src/database/entities/ChatFlow.ts

@@ -2,6 +2,13 @@
 import { Entity, Column, CreateDateColumn, UpdateDateColumn, PrimaryGeneratedColumn } from 'typeorm'
 import { ChatflowType, IChatFlow } from '../../Interface'
 
+export enum EnumChatflowType {
+    CHATFLOW = 'CHATFLOW',
+    AGENTFLOW = 'AGENTFLOW',
+    MULTIAGENT = 'MULTIAGENT',
+    ASSISTANT = 'ASSISTANT'
+}
+
 @Entity()
 export class ChatFlow implements IChatFlow {
     @PrimaryGeneratedColumn('uuid')
@@ -40,7 +47,7 @@ export class ChatFlow implements IChatFlow {
     @Column({ nullable: true, type: 'text' })
     category?: string
 
-    @Column({ nullable: true, type: 'text' })
+    @Column({ type: 'varchar', length: 20, default: EnumChatflowType.CHATFLOW })
     type?: ChatflowType
 
     @Column({ type: 'timestamp' })

packages/server/src/database/migrations/mariadb/1755066758601-ModifyChatflowType.ts

@@ -0,0 +1,15 @@
+import { MigrationInterface, QueryRunner } from 'typeorm'
+import { EnumChatflowType } from '../../entities/ChatFlow'
+
+export class ModifyChatflowType1755066758601 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`
+            UPDATE \`chat_flow\` SET \`type\` = '${EnumChatflowType.CHATFLOW}' WHERE \`type\` IS NULL OR \`type\` = '';
+        `)
+        await queryRunner.query(`
+            ALTER TABLE \`chat_flow\` MODIFY COLUMN \`type\` VARCHAR(20) NOT NULL DEFAULT '${EnumChatflowType.CHATFLOW}';
+        `)
+    }
+
+    public async down(): Promise<void> {}
+}

packages/server/src/database/migrations/mariadb/index.ts

@@ -36,6 +36,7 @@ import { AddExecutionEntity1738090872625 } from './1738090872625-AddExecutionEnt
 import { FixOpenSourceAssistantTable1743758056188 } from './1743758056188-FixOpenSourceAssistantTable'
 import { AddErrorToEvaluationRun1744964560174 } from './1744964560174-AddErrorToEvaluationRun'
 import { ModifyExecutionDataColumnType1747902489801 } from './1747902489801-ModifyExecutionDataColumnType'
+import { ModifyChatflowType1755066758601 } from './1755066758601-ModifyChatflowType'
 
 import { AddAuthTables1720230151482 } from '../../../enterprise/database/migrations/mariadb/1720230151482-AddAuthTables'
 import { AddWorkspace1725437498242 } from '../../../enterprise/database/migrations/mariadb/1725437498242-AddWorkspace'
@@ -98,5 +99,6 @@ export const mariadbMigrations = [
     FixOpenSourceAssistantTable1743758056188,
     AddErrorToEvaluationRun1744964560174,
     ExecutionLinkWorkspaceId1746862866554,
-    ModifyExecutionDataColumnType1747902489801
+    ModifyExecutionDataColumnType1747902489801,
+    ModifyChatflowType1755066758601
 ]

packages/server/src/database/migrations/mysql/1755066758601-ModifyChatflowType.ts

@@ -0,0 +1,15 @@
+import { MigrationInterface, QueryRunner } from 'typeorm'
+import { EnumChatflowType } from '../../entities/ChatFlow'
+
+export class ModifyChatflowType1755066758601 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`
+            UPDATE \`chat_flow\` SET \`type\` = '${EnumChatflowType.CHATFLOW}' WHERE \`type\` IS NULL OR \`type\` = '';
+        `)
+        await queryRunner.query(`
+            ALTER TABLE \`chat_flow\` MODIFY COLUMN \`type\` VARCHAR(20) NOT NULL DEFAULT '${EnumChatflowType.CHATFLOW}';
+        `)
+    }
+
+    public async down(): Promise<void> {}
+}

packages/server/src/database/migrations/mysql/index.ts

@@ -37,6 +37,7 @@ import { FixOpenSourceAssistantTable1743758056188 } from './1743758056188-FixOpe
 import { AddErrorToEvaluationRun1744964560174 } from './1744964560174-AddErrorToEvaluationRun'
 import { FixErrorsColumnInEvaluationRun1746437114935 } from './1746437114935-FixErrorsColumnInEvaluationRun'
 import { ModifyExecutionDataColumnType1747902489801 } from './1747902489801-ModifyExecutionDataColumnType'
+import { ModifyChatflowType1755066758601 } from './1755066758601-ModifyChatflowType'
 
 import { AddAuthTables1720230151482 } from '../../../enterprise/database/migrations/mysql/1720230151482-AddAuthTables'
 import { AddWorkspace1720230151484 } from '../../../enterprise/database/migrations/mysql/1720230151484-AddWorkspace'
@@ -100,5 +101,6 @@ export const mysqlMigrations = [
     AddErrorToEvaluationRun1744964560174,
     FixErrorsColumnInEvaluationRun1746437114935,
     ExecutionLinkWorkspaceId1746862866554,
-    ModifyExecutionDataColumnType1747902489801
+    ModifyExecutionDataColumnType1747902489801,
+    ModifyChatflowType1755066758601
 ]

packages/server/src/database/migrations/postgres/1755066758601-ModifyChatflowType.ts

@@ -0,0 +1,21 @@
+import { MigrationInterface, QueryRunner } from 'typeorm'
+import { EnumChatflowType } from '../../entities/ChatFlow'
+
+export class ModifyChatflowType1755066758601 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`
+            UPDATE "chat_flow" SET "type" = '${EnumChatflowType.CHATFLOW}' WHERE "type" IS NULL OR "type" = '';
+        `)
+        await queryRunner.query(`
+            ALTER TABLE "chat_flow" ALTER COLUMN "type" SET DEFAULT '${EnumChatflowType.CHATFLOW}';
+        `)
+        await queryRunner.query(`
+            ALTER TABLE "chat_flow" ALTER COLUMN "type" TYPE VARCHAR(20);
+        `)
+        await queryRunner.query(`
+            ALTER TABLE "chat_flow" ALTER COLUMN "type" SET NOT NULL;
+        `)
+    }
+
+    public async down(): Promise<void> {}
+}

packages/server/src/database/migrations/postgres/index.ts

@@ -36,6 +36,7 @@ import { AddExecutionEntity1738090872625 } from './1738090872625-AddExecutionEnt
 import { FixOpenSourceAssistantTable1743758056188 } from './1743758056188-FixOpenSourceAssistantTable'
 import { AddErrorToEvaluationRun1744964560174 } from './1744964560174-AddErrorToEvaluationRun'
 import { ModifyExecutionSessionIdFieldType1748450230238 } from './1748450230238-ModifyExecutionSessionIdFieldType'
+import { ModifyChatflowType1755066758601 } from './1755066758601-ModifyChatflowType'
 
 import { AddAuthTables1720230151482 } from '../../../enterprise/database/migrations/postgres/1720230151482-AddAuthTables'
 import { AddWorkspace1720230151484 } from '../../../enterprise/database/migrations/postgres/1720230151484-AddWorkspace'
@@ -98,5 +99,6 @@ export const postgresMigrations = [
     FixOpenSourceAssistantTable1743758056188,
     AddErrorToEvaluationRun1744964560174,
     ExecutionLinkWorkspaceId1746862866554,
-    ModifyExecutionSessionIdFieldType1748450230238
+    ModifyExecutionSessionIdFieldType1748450230238,
+    ModifyChatflowType1755066758601
 ]

packages/server/src/database/migrations/sqlite/1755066758601-ModifyChatflowType.ts

@@ -0,0 +1,40 @@
+import { MigrationInterface, QueryRunner } from 'typeorm'
+import { EnumChatflowType } from '../../entities/ChatFlow'
+
+export class ModifyChatflowType1755066758601 implements MigrationInterface {
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`
+            CREATE TABLE "temp_chat_flow" (
+                "id" varchar PRIMARY KEY NOT NULL, 
+                "name" varchar NOT NULL, 
+                "flowData" text NOT NULL, 
+                "deployed" boolean, 
+                "isPublic" boolean, 
+                "apikeyid" varchar, 
+                "chatbotConfig" text, 
+                "createdDate" datetime NOT NULL DEFAULT (datetime('now')), 
+                "updatedDate" datetime NOT NULL DEFAULT (datetime('now')), 
+                "apiConfig" TEXT, 
+                "analytic" TEXT, 
+                "category" TEXT, 
+                "speechToText" TEXT, 
+                "type" VARCHAR(20) NOT NULL DEFAULT '${EnumChatflowType.CHATFLOW}', 
+                "workspaceId" TEXT, 
+                "followUpPrompts" TEXT,
+                FOREIGN KEY ("workspaceId") REFERENCES "workspace"("id")
+            );
+        `)
+
+        await queryRunner.query(`
+            INSERT INTO "temp_chat_flow" ("id", "name", "flowData", "deployed", "isPublic", "apikeyid", "chatbotConfig", "createdDate", "updatedDate", "apiConfig", "analytic", "category", "speechToText", "type", "workspaceId", "followUpPrompts")
+            SELECT "id", "name", "flowData", "deployed", "isPublic", "apikeyid", "chatbotConfig", "createdDate", "updatedDate", "apiConfig", "analytic", "category", "speechToText",
+            CASE WHEN "type" IS NULL OR "type" = '' THEN '${EnumChatflowType.CHATFLOW}' ELSE "type" END, "workspaceId", "followUpPrompts" FROM "chat_flow";
+        `)
+
+        await queryRunner.query(`DROP TABLE "chat_flow";`)
+
+        await queryRunner.query(`ALTER TABLE "temp_chat_flow" RENAME TO "chat_flow";`)
+    }
+
+    public async down(): Promise<void> {}
+}

packages/server/src/database/migrations/sqlite/index.ts

@@ -34,6 +34,7 @@ import { AddSeqNoToDatasetRow1733752119696 } from './1733752119696-AddSeqNoToDat
 import { AddExecutionEntity1738090872625 } from './1738090872625-AddExecutionEntity'
 import { FixOpenSourceAssistantTable1743758056188 } from './1743758056188-FixOpenSourceAssistantTable'
 import { AddErrorToEvaluationRun1744964560174 } from './1744964560174-AddErrorToEvaluationRun'
+import { ModifyChatflowType1755066758601 } from './1755066758601-ModifyChatflowType'
 
 import { AddAuthTables1720230151482 } from '../../../enterprise/database/migrations/sqlite/1720230151482-AddAuthTables'
 import { AddWorkspace1720230151484 } from '../../../enterprise/database/migrations/sqlite/1720230151484-AddWorkspace'
@@ -94,5 +95,6 @@ export const sqliteMigrations = [
     AddExecutionEntity1738090872625,
     FixOpenSourceAssistantTable1743758056188,
     AddErrorToEvaluationRun1744964560174,
-    ExecutionLinkWorkspaceId1746862866554
+    ExecutionLinkWorkspaceId1746862866554,
+    ModifyChatflowType1755066758601
 ]

packages/server/src/services/chatflows/index.ts

@@ -4,7 +4,7 @@ import { In } from 'typeorm'
 import { ChatflowType, IReactFlowObject } from '../../Interface'
 import { FLOWISE_COUNTER_STATUS, FLOWISE_METRIC_COUNTERS } from '../../Interface.Metrics'
 import { UsageCacheManager } from '../../UsageCacheManager'
-import { ChatFlow } from '../../database/entities/ChatFlow'
+import { ChatFlow, EnumChatflowType } from '../../database/entities/ChatFlow'
 import { ChatMessage } from '../../database/entities/ChatMessage'
 import { ChatMessageFeedback } from '../../database/entities/ChatMessageFeedback'
 import { UpsertHistory } from '../../database/entities/UpsertHistory'
@@ -20,6 +20,15 @@ import { utilGetUploadsConfig } from '../../utils/getUploadsConfig'
 import logger from '../../utils/logger'
 import { updateStorageUsage } from '../../utils/quotaUsage'
 
+export const enum ChatflowErrorMessage {
+    INVALID_CHATFLOW_TYPE = 'Invalid Chatflow Type'
+}
+
+export function validateChatflowType(type: ChatflowType | undefined) {
+    if (!Object.values(EnumChatflowType).includes(type as EnumChatflowType))
+        throw new InternalFlowiseError(StatusCodes.BAD_REQUEST, ChatflowErrorMessage.INVALID_CHATFLOW_TYPE)
+}
+
 // Check if chatflow valid for streaming
 const checkIfChatflowIsValidForStreaming = async (chatflowId: string): Promise<any> => {
     try {
@@ -254,57 +263,51 @@ const saveChatflow = async (
     subscriptionId: string,
     usageCacheManager: UsageCacheManager
 ): Promise<any> => {
-    try {
-        const appServer = getRunningExpressApp()
+    validateChatflowType(newChatFlow.type)
+    const appServer = getRunningExpressApp()
 
-        let dbResponse: ChatFlow
-        if (containsBase64File(newChatFlow)) {
-            // we need a 2-step process, as we need to save the chatflow first and then update the file paths
-            // this is because we need the chatflow id to create the file paths
+    let dbResponse: ChatFlow
+    if (containsBase64File(newChatFlow)) {
+        // we need a 2-step process, as we need to save the chatflow first and then update the file paths
+        // this is because we need the chatflow id to create the file paths
 
-            // step 1 - save with empty flowData
-            const incomingFlowData = newChatFlow.flowData
-            newChatFlow.flowData = JSON.stringify({})
-            const chatflow = appServer.AppDataSource.getRepository(ChatFlow).create(newChatFlow)
-            const step1Results = await appServer.AppDataSource.getRepository(ChatFlow).save(chatflow)
+        // step 1 - save with empty flowData
+        const incomingFlowData = newChatFlow.flowData
+        newChatFlow.flowData = JSON.stringify({})
+        const chatflow = appServer.AppDataSource.getRepository(ChatFlow).create(newChatFlow)
+        const step1Results = await appServer.AppDataSource.getRepository(ChatFlow).save(chatflow)
 
-            // step 2 - convert base64 to file paths and update the chatflow
-            step1Results.flowData = await updateFlowDataWithFilePaths(
-                step1Results.id,
-                incomingFlowData,
-                orgId,
-                workspaceId,
-                subscriptionId,
-                usageCacheManager
-            )
-            await _checkAndUpdateDocumentStoreUsage(step1Results, newChatFlow.workspaceId)
-            dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(step1Results)
-        } else {
-            const chatflow = appServer.AppDataSource.getRepository(ChatFlow).create(newChatFlow)
-            dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(chatflow)
-        }
-        await appServer.telemetry.sendTelemetry(
-            'chatflow_created',
-            {
-                version: await getAppVersion(),
-                chatflowId: dbResponse.id,
-                flowGraph: getTelemetryFlowObj(JSON.parse(dbResponse.flowData)?.nodes, JSON.parse(dbResponse.flowData)?.edges)
-            },
-            orgId
-        )
-
-        appServer.metricsProvider?.incrementCounter(
-            dbResponse?.type === 'MULTIAGENT' ? FLOWISE_METRIC_COUNTERS.AGENTFLOW_CREATED : FLOWISE_METRIC_COUNTERS.CHATFLOW_CREATED,
-            { status: FLOWISE_COUNTER_STATUS.SUCCESS }
-        )
-
-        return dbResponse
-    } catch (error) {
-        throw new InternalFlowiseError(
-            StatusCodes.INTERNAL_SERVER_ERROR,
-            `Error: chatflowsService.saveChatflow - ${getErrorMessage(error)}`
+        // step 2 - convert base64 to file paths and update the chatflow
+        step1Results.flowData = await updateFlowDataWithFilePaths(
+            step1Results.id,
+            incomingFlowData,
+            orgId,
+            workspaceId,
+            subscriptionId,
+            usageCacheManager
         )
+        await _checkAndUpdateDocumentStoreUsage(step1Results, newChatFlow.workspaceId)
+        dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(step1Results)
+    } else {
+        const chatflow = appServer.AppDataSource.getRepository(ChatFlow).create(newChatFlow)
+        dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(chatflow)
     }
+    await appServer.telemetry.sendTelemetry(
+        'chatflow_created',
+        {
+            version: await getAppVersion(),
+            chatflowId: dbResponse.id,
+            flowGraph: getTelemetryFlowObj(JSON.parse(dbResponse.flowData)?.nodes, JSON.parse(dbResponse.flowData)?.edges)
+        },
+        orgId
+    )
+
+    appServer.metricsProvider?.incrementCounter(
+        dbResponse?.type === 'MULTIAGENT' ? FLOWISE_METRIC_COUNTERS.AGENTFLOW_CREATED : FLOWISE_METRIC_COUNTERS.CHATFLOW_CREATED,
+        { status: FLOWISE_COUNTER_STATUS.SUCCESS }
+    )
+
+    return dbResponse
 }
 
 const updateChatflow = async (
@@ -314,29 +317,27 @@ const updateChatflow = async (
     workspaceId: string,
     subscriptionId: string
 ): Promise<any> => {
-    try {
-        const appServer = getRunningExpressApp()
-        if (updateChatFlow.flowData && containsBase64File(updateChatFlow)) {
-            updateChatFlow.flowData = await updateFlowDataWithFilePaths(
-                chatflow.id,
-                updateChatFlow.flowData,
-                orgId,
-                workspaceId,
-                subscriptionId,
-                appServer.usageCacheManager
-            )
-        }
-        const newDbChatflow = appServer.AppDataSource.getRepository(ChatFlow).merge(chatflow, updateChatFlow)
-        await _checkAndUpdateDocumentStoreUsage(newDbChatflow, chatflow.workspaceId)
-        const dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(newDbChatflow)
-
-        return dbResponse
-    } catch (error) {
-        throw new InternalFlowiseError(
-            StatusCodes.INTERNAL_SERVER_ERROR,
-            `Error: chatflowsService.updateChatflow - ${getErrorMessage(error)}`
+    const appServer = getRunningExpressApp()
+    if (updateChatFlow.flowData && containsBase64File(updateChatFlow)) {
+        updateChatFlow.flowData = await updateFlowDataWithFilePaths(
+            chatflow.id,
+            updateChatFlow.flowData,
+            orgId,
+            workspaceId,
+            subscriptionId,
+            appServer.usageCacheManager
         )
     }
+    if (updateChatFlow.type || updateChatFlow.type === '') {
+        validateChatflowType(updateChatFlow.type)
+    } else {
+        updateChatFlow.type = chatflow.type
+    }
+    const newDbChatflow = appServer.AppDataSource.getRepository(ChatFlow).merge(chatflow, updateChatFlow)
+    await _checkAndUpdateDocumentStoreUsage(newDbChatflow, chatflow.workspaceId)
+    const dbResponse = await appServer.AppDataSource.getRepository(ChatFlow).save(newDbChatflow)
+
+    return dbResponse
 }
 
 // Get specific chatflow chatbotConfig via id (PUBLIC endpoint, used to retrieve config for embedded chat)