feat: implement scheduling functionality for AgentflowV2

- Added ScheduleQueue class for managing scheduled jobs using BullMQ.
- Introduced scheduleService for creating, updating, and disabling schedules.
- Integrated schedule validation and cron expression handling.
- Enhanced chatflow service to manage schedules for agentflows.
- Added UI components for time and day selection (TimePicker, WeekDaysPicker, MonthDaysPicker).
- Updated NodeInputHandler to support new input types for scheduling.
- Improved validation for schedule configurations in agentflows.

Author: prd-hoang-doan

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

@@ -18,7 +18,7 @@ class Start_Agentflow implements INode {
     constructor() {
         this.label = 'Start'
         this.name = 'startAgentflow'
-        this.version = 1.1
+        this.version = 1.2
         this.type = 'Start'
         this.category = 'Agent Flows'
         this.description = 'Starting point of the agentflow'
@@ -40,6 +40,11 @@ class Start_Agentflow implements INode {
                         label: 'Form Input',
                         name: 'formInput',
                         description: 'Start the workflow with form inputs'
+                    },
+                    {
+                        label: 'Schedule Input',
+                        name: 'scheduleInput',
+                        description: 'Start the workflow on a recurring schedule (cron)'
                     }
                 ],
                 default: 'chatInput'
@@ -125,6 +130,135 @@ class Start_Agentflow implements INode {
                     }
                 ]
             },
+            {
+                label: 'Schedule Type',
+                name: 'scheduleType',
+                type: 'options',
+                options: [
+                    {
+                        label: 'Visual Picker',
+                        name: 'visualPicker',
+                        description: 'Use a visual picker to select schedule options'
+                    },
+                    {
+                        label: 'Cron Expression',
+                        name: 'cronExpression',
+                        description: 'Use a cron expression to define the schedule'
+                    }
+                ],
+                show: {
+                    startInputType: 'scheduleInput'
+                }
+            },
+            {
+                label: 'Cron Expression',
+                name: 'scheduleCronExpression',
+                type: 'string',
+                placeholder: '0 9 * * 1-5',
+                description:
+                    'Standard 5-field cron expression (minute hour day month weekday). Example: "0 9 * * 1-5" runs at 09:00 every weekday.',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'cronExpression'
+                }
+            },
+            {
+                label: 'Frequency',
+                name: 'scheduleFrequency',
+                type: 'options',
+                options: [
+                    {
+                        label: 'Hourly',
+                        name: 'hourly',
+                        description: 'Run every hour at the specified time'
+                    },
+                    {
+                        label: 'Daily',
+                        name: 'daily',
+                        description: 'Run every day at the specified time'
+                    },
+                    {
+                        label: 'Weekly',
+                        name: 'weekly',
+                        description: 'Run every week on the specified day and time'
+                    },
+                    {
+                        label: 'Monthly',
+                        name: 'monthly',
+                        description: 'Run every month on the specified date and time'
+                    }
+                ],
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker'
+                }
+            },
+            {
+                label: 'On Minute',
+                name: 'scheduleOnMinute',
+                type: 'number',
+                placeholder: '30',
+                description:
+                    'Minute of the hour when the schedule should run (0-59). For example, "30" means the schedule will run at the 30th minute of the hour.',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker',
+                    scheduleFrequency: 'hourly'
+                }
+            },
+            {
+                label: 'On Time',
+                name: 'scheduleOnTime',
+                type: 'timePicker',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker',
+                    scheduleFrequency: ['daily', 'weekly', 'monthly']
+                }
+            },
+            {
+                label: 'On Day of Week',
+                name: 'scheduleOnDayOfWeek',
+                type: 'weekDaysPicker',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker',
+                    scheduleFrequency: 'weekly'
+                }
+            },
+            {
+                label: 'On Day of Month',
+                name: 'scheduleOnDayOfMonth',
+                type: 'monthDaysPicker',
+                show: {
+                    startInputType: 'scheduleInput',
+                    scheduleType: 'visualPicker',
+                    scheduleFrequency: 'monthly'
+                }
+            },
+            {
+                label: 'Timezone',
+                name: 'scheduleTimezone',
+                type: 'string',
+                placeholder: 'UTC',
+                description: 'IANA timezone name, e.g. America/New_York. Defaults to UTC.',
+                optional: true,
+                show: {
+                    startInputType: 'scheduleInput'
+                }
+            },
+            {
+                label: 'Default Input',
+                name: 'scheduleDefaultInput',
+                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'
+                }
+            },
             {
                 label: 'Ephemeral Memory',
                 name: 'startEphemeralMemory',
@@ -213,6 +347,14 @@ class Start_Agentflow implements INode {
             outputData.form = form
         }
 
+        if (startInputType === 'scheduleInput') {
+            const defaultInput = nodeData.inputs?.scheduleDefaultInput as string
+            const effectiveInput = (typeof input === 'string' && input) || defaultInput || ''
+            inputData.question = effectiveInput
+            outputData.question = effectiveInput
+            outputData.scheduledAt = options.agentflowRuntime?.scheduledAt ?? new Date().toISOString()
+        }
+
         if (startEphemeralMemory) {
             outputData.ephemeralMemory = true
         }

packages/server/src/Interface.Schedule.ts

@@ -0,0 +1,23 @@
+import { DataSource } from 'typeorm'
+import { IComponentNodes } from './Interface'
+import { Telemetry } from './utils/telemetry'
+import { CachePool } from './CachePool'
+import { UsageCacheManager } from './UsageCacheManager'
+
+export interface IScheduleQueueAppServer {
+    appDataSource: DataSource
+    componentNodes: IComponentNodes
+    telemetry: Telemetry
+    cachePool: CachePool
+    usageCacheManager: UsageCacheManager
+}
+
+export interface IScheduleAgentflowJobData extends IScheduleQueueAppServer {
+    scheduleRecordId: string
+    targetId: string
+    cronExpression: string
+    timezone: string
+    defaultInput?: string
+    workspaceId: string
+    scheduledAt: string // ISO string
+}

packages/server/src/commands/worker.ts

@@ -16,6 +16,7 @@ interface CustomListener extends QueueEventsListener {
 export default class Worker extends BaseCommand {
     predictionWorkerId: string
     upsertionWorkerId: string
+    scheduleWorkerId: string
 
     async run(): Promise<void> {
         logger.info('Starting Flowise Worker...')
@@ -51,6 +52,12 @@ export default class Worker extends BaseCommand {
         this.upsertionWorkerId = upsertionWorker.id
         logger.info(`Upsertion Worker ${this.upsertionWorkerId} created`)
 
+        /** Schedule */
+        const scheduleQueue = queueManager.getQueue('schedule')
+        const scheduleWorker = scheduleQueue.createWorker()
+        this.scheduleWorkerId = scheduleWorker.id
+        logger.info(`Schedule Worker ${this.scheduleWorkerId} created`)
+
         // Keep the process running
         process.stdin.resume()
     }
@@ -98,6 +105,10 @@ export default class Worker extends BaseCommand {
             const upsertWorker = queueManager.getQueue('upsert').getWorker()
             logger.info(`Shutting down Flowise Upsertion Worker ${this.upsertionWorkerId}...`)
             await upsertWorker.close()
+
+            const scheduleWorker = queueManager.getQueue('schedule').getWorker()
+            logger.info(`Shutting down Flowise Schedule Worker...`)
+            await scheduleWorker.close()
         } catch (error) {
             logger.error('There was an error shutting down Flowise Worker...', error)
             await this.failExit()

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

@@ -0,0 +1,58 @@
+/* eslint-disable */
+import { Entity, Column, PrimaryGeneratedColumn, CreateDateColumn, UpdateDateColumn, Index } from 'typeorm'
+
+export enum ScheduleTriggerType {
+    AGENTFLOW = 'AGENTFLOW'
+}
+
+@Entity()
+export class ScheduleRecord {
+    @PrimaryGeneratedColumn('uuid')
+    id: string
+
+    /** Discriminator: which entity type is being scheduled */
+    @Column({ type: 'varchar', length: 32 })
+    triggerType: ScheduleTriggerType
+
+    /** FK to the target entity (ChatFlow.id for AGENTFLOW) */
+    @Index()
+    @Column({ type: 'varchar' })
+    targetId: string
+
+    /** Node ID within the flow (for traceability) */
+    @Column({ nullable: true, type: 'text' })
+    nodeId?: string
+
+    /** Standard 5 or 6 field cron expression */
+    @Column({ type: 'text' })
+    cronExpression: string
+
+    /** IANA timezone string, e.g. "UTC" or "America/New_York" */
+    @Column({ type: 'varchar', length: 64, default: 'UTC' })
+    timezone: string
+
+    /** Whether the schedule is active */
+    @Column({ type: 'boolean', default: true })
+    enabled: boolean
+
+    /** Optional static text sent as question when the flow fires */
+    @Column({ nullable: true, type: 'text' })
+    defaultInput?: string
+
+    @Column({ nullable: true, type: 'timestamp' })
+    lastRunAt?: Date
+
+    @Column({ nullable: true, type: 'timestamp' })
+    nextRunAt?: Date
+
+    @Column({ type: 'varchar' })
+    workspaceId: string
+
+    @Column({ type: 'timestamp' })
+    @CreateDateColumn()
+    createdDate: Date
+
+    @Column({ type: 'timestamp' })
+    @UpdateDateColumn()
+    updatedDate: Date
+}

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

@@ -0,0 +1,51 @@
+/* eslint-disable */
+import { Entity, Column, PrimaryGeneratedColumn, CreateDateColumn, Index } from 'typeorm'
+import { ScheduleTriggerType } from './ScheduleRecord'
+
+export enum ScheduleTriggerStatus {
+    QUEUED = 'QUEUED',
+    RUNNING = 'RUNNING',
+    SUCCEEDED = 'SUCCEEDED',
+    FAILED = 'FAILED',
+    SKIPPED = 'SKIPPED'
+}
+
+@Entity()
+export class ScheduleTriggerLog {
+    @PrimaryGeneratedColumn('uuid')
+    id: string
+
+    @Index()
+    @Column({ type: 'varchar' })
+    scheduleRecordId: string
+
+    @Column({ type: 'varchar', length: 32 })
+    triggerType: ScheduleTriggerType
+
+    @Index()
+    @Column({ type: 'varchar' })
+    targetId: string
+
+    /** Resulting execution/chatMessage ID (for agentflow triggers) */
+    @Column({ nullable: true, type: 'varchar' })
+    executionId?: string
+
+    @Column({ type: 'varchar', length: 32 })
+    status: ScheduleTriggerStatus
+
+    @Column({ nullable: true, type: 'text' })
+    error?: string
+
+    @Column({ nullable: true, type: 'integer' })
+    elapsedTimeMs?: number
+
+    @Column({ type: 'timestamp' })
+    scheduledAt: Date
+
+    @Column({ type: 'varchar' })
+    workspaceId: string
+
+    @Column({ type: 'timestamp' })
+    @CreateDateColumn()
+    createdDate: Date
+}

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

@@ -26,6 +26,8 @@ import { Workspace } from '../../enterprise/database/entities/workspace.entity'
 import { WorkspaceUser } from '../../enterprise/database/entities/workspace-user.entity'
 import { LoginMethod } from '../../enterprise/database/entities/login-method.entity'
 import { LoginSession } from '../../enterprise/database/entities/login-session.entity'
+import { ScheduleRecord } from './ScheduleRecord'
+import { ScheduleTriggerLog } from './ScheduleTriggerLog'
 
 export const entities = {
     ChatFlow,
@@ -57,5 +59,7 @@ export const entities = {
     Workspace,
     WorkspaceUser,
     LoginMethod,
-    LoginSession
+    LoginSession,
+    ScheduleRecord,
+    ScheduleTriggerLog
 }