feat: enhance schedule job for agentflows

- Refactor ScheduleQueue to utilize executeScheduleJob for job execution.
- Add new endpoints for schedule status retrieval and enabling/disabling schedules in chatflows.
- Update chatflow service to handle schedule creation, updates, and deletions more effectively.
- Introduce DatePicker component for date input in UI.
- Enhance CanvasHeader to manage schedule state and toggle functionality.
- Integrate schedule validation logic to ensure proper configuration before enabling schedules.
- Update dependencies to include node-cron for improved scheduling capabilities.

Author: prd-hoang-doan

packages/components/nodes/agentflow/Start/Start.ts

@@ -236,6 +236,16 @@ class Start_Agentflow implements INode {
                     scheduleFrequency: 'monthly'
                 }
             },
+            {
+                label: 'End Date',
+                name: 'scheduleEndDate',
+                type: 'datePicker',
+                description: 'Optional date after which the schedule will stop firing.',
+                optional: true,
+                show: {
+                    startInputType: 'scheduleInput'
+                }
+            },
             {
                 label: 'Timezone',
                 name: 'scheduleTimezone',
@@ -253,7 +263,6 @@ class Start_Agentflow implements INode {
                 type: 'string',
                 placeholder: 'Run the daily report',
                 description: 'Default question/input passed to the flow when it is triggered by the scheduler.',
-                acceptVariable: true,
                 rows: 4,
                 show: {
                     startInputType: 'scheduleInput'
@@ -352,7 +361,6 @@ class Start_Agentflow implements INode {
             const effectiveInput = (typeof input === 'string' && input) || defaultInput || ''
             inputData.question = effectiveInput
             outputData.question = effectiveInput
-            outputData.scheduledAt = options.agentflowRuntime?.scheduledAt ?? new Date().toISOString()
         }
 
         if (startEphemeralMemory) {

packages/server/package.json

@@ -126,6 +126,7 @@
         "multer-s3": "^3.0.1",
         "mysql2": "^3.11.3",
         "nanoid": "3",
+        "node-cron": "^4.2.1",
         "nodemailer": "^7.0.7",
         "openai": "6.19.0",
         "passport": "^0.7.0",

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

@@ -13,6 +13,8 @@ import { WorkspaceUserErrorMessage, WorkspaceUserService } from '../../enterpris
 import { QueryRunner } from 'typeorm'
 import { GeneralErrorMessage } from '../../utils/constants'
 import { sanitizeFlowDataForPublicEndpoint } from '../../utils/sanitizeFlowData'
+import scheduleService from '../../services/schedule'
+import { ScheduleBeat } from '../../queue/ScheduleBeat'
 
 const checkIfChatflowIsValidForStreaming = async (req: Request, res: Response, next: NextFunction) => {
     try {
@@ -273,6 +275,54 @@ const checkIfChatflowHasChanged = async (req: Request, res: Response, next: Next
     }
 }
 
+const getScheduleStatus = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+        if (!req.params?.id) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                'Error: chatflowsController.getScheduleStatus - id not provided!'
+            )
+        }
+        const workspaceId = req.user?.activeWorkspaceId
+        if (!workspaceId) {
+            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, 'Error: chatflowsController.getScheduleStatus - workspace not found!')
+        }
+        const status = await scheduleService.getScheduleStatus(req.params.id, workspaceId)
+        return res.json({
+            enabled: status.record?.enabled ?? false,
+            canEnable: status.canEnable,
+            reason: status.reason,
+            record: status.record
+        })
+    } catch (error) {
+        next(error)
+    }
+}
+
+const toggleScheduleEnabled = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+        if (!req.params?.id) {
+            throw new InternalFlowiseError(
+                StatusCodes.PRECONDITION_FAILED,
+                'Error: chatflowsController.toggleScheduleEnabled - id not provided!'
+            )
+        }
+        const workspaceId = req.user?.activeWorkspaceId
+        if (!workspaceId) {
+            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, 'Error: chatflowsController.toggleScheduleEnabled - workspace not found!')
+        }
+        const { enabled } = req.body
+        if (typeof enabled !== 'boolean') {
+            throw new InternalFlowiseError(StatusCodes.BAD_REQUEST, '"enabled" must be a boolean')
+        }
+        const record = await scheduleService.toggleScheduleEnabled(req.params.id, workspaceId, enabled)
+        await ScheduleBeat.getInstance().onScheduleChanged(record.id, enabled ? 'upsert' : 'delete')
+        return res.json(record)
+    } catch (error) {
+        next(error)
+    }
+}
+
 export default {
     checkIfChatflowIsValidForStreaming,
     checkIfChatflowIsValidForUploads,
@@ -284,5 +334,7 @@ export default {
     updateChatflow,
     getSinglePublicChatflow,
     getSinglePublicChatbotConfig,
-    checkIfChatflowHasChanged
+    checkIfChatflowHasChanged,
+    getScheduleStatus,
+    toggleScheduleEnabled
 }

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

@@ -40,10 +40,14 @@ export class ScheduleRecord {
     defaultInput?: string
 
     @Column({ nullable: true, type: 'timestamp' })
-    lastRunAt?: Date
+    lastRunAt?: Date | null
 
     @Column({ nullable: true, type: 'timestamp' })
-    nextRunAt?: Date
+    nextRunAt?: Date | null
+
+    /** Optional date/time after which the schedule will no longer fire */
+    @Column({ nullable: true, type: 'timestamp' })
+    endDate?: Date | null
 
     @Column({ type: 'varchar' })
     workspaceId: string

packages/server/src/database/migrations/mariadb/1772000000000-AddScheduleEntities.ts

@@ -14,6 +14,7 @@ export class AddScheduleEntities1772000000000 implements MigrationInterface {
                 \`defaultInput\` text,
                 \`lastRunAt\` datetime(6),
                 \`nextRunAt\` datetime(6),
+                \`endDate\` datetime(6),
                 \`workspaceId\` varchar(255) NOT NULL,
                 \`createdDate\` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
                 \`updatedDate\` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),

packages/server/src/database/migrations/mysql/1772000000000-AddScheduleEntities.ts

@@ -14,6 +14,7 @@ export class AddScheduleEntities1772000000000 implements MigrationInterface {
                 \`defaultInput\` text,
                 \`lastRunAt\` datetime(6),
                 \`nextRunAt\` datetime(6),
+                \`endDate\` datetime(6),
                 \`workspaceId\` varchar(255) NOT NULL,
                 \`createdDate\` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
                 \`updatedDate\` datetime(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),

packages/server/src/database/migrations/postgres/1772000000000-AddScheduleEntities.ts

@@ -14,6 +14,7 @@ export class AddScheduleEntities1772000000000 implements MigrationInterface {
                 "defaultInput" text,
                 "lastRunAt" timestamp,
                 "nextRunAt" timestamp,
+                "endDate" timestamp,
                 "workspaceId" varchar NOT NULL,
                 "createdDate" timestamp NOT NULL DEFAULT now(),
                 "updatedDate" timestamp NOT NULL DEFAULT now(),

packages/server/src/database/migrations/sqlite/1772000000000-AddScheduleEntities.ts

@@ -14,6 +14,7 @@ export class AddScheduleEntities1772000000000 implements MigrationInterface {
                 "defaultInput" text,
                 "lastRunAt" datetime,
                 "nextRunAt" datetime,
+                "endDate" datetime,
                 "workspaceId" varchar NOT NULL,
                 "createdDate" datetime NOT NULL DEFAULT (datetime('now')),
                 "updatedDate" datetime NOT NULL DEFAULT (datetime('now'))