335 workflow implement workflow cron triggers backend (#9988)

[Backend side] Add cron triggers to workflow
Closes https://github.com/twentyhq/core-team-issues/issues/335

package.json

@@ -73,7 +73,7 @@
     "bcrypt": "^5.1.1",
     "better-sqlite3": "^9.2.2",
     "body-parser": "^1.20.2",
-    "bullmq": "^4.15.0",
+    "bullmq": "^5.40.0",
     "bytes": "^3.1.2",
     "class-transformer": "^0.5.1",
     "clsx": "^2.1.1",

packages/twenty-server/src/database/commands/data-seed-demo-workspace/crons/start-data-seed-demo-workspace.cron.command.ts

@@ -19,14 +19,14 @@ export class StartDataSeedDemoWorkspaceCronCommand extends CommandRunner {
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.addCron<undefined>(
+    await this.messageQueueService.addCron<undefined>({
-      DataSeedDemoWorkspaceJob.name,
+      jobName: DataSeedDemoWorkspaceJob.name,
-      undefined,
+      data: undefined,
-      {
+      options: {
         repeat: {
           pattern: dataSeedDemoWorkspaceCronPattern,
         },
       },
-    );
+    });
   }
 }

packages/twenty-server/src/database/commands/data-seed-demo-workspace/crons/stop-data-seed-demo-workspace.cron.command.ts

@@ -1,6 +1,5 @@
 import { Command, CommandRunner } from 'nest-commander';
 
-import { dataSeedDemoWorkspaceCronPattern } from 'src/database/commands/data-seed-demo-workspace/crons/data-seed-demo-workspace-cron-pattern';
 import { DataSeedDemoWorkspaceJob } from 'src/database/commands/data-seed-demo-workspace/jobs/data-seed-demo-workspace.job';
 import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
@@ -19,9 +18,8 @@ export class StopDataSeedDemoWorkspaceCronCommand extends CommandRunner {
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.removeCron(
+    await this.messageQueueService.removeCron({
-      DataSeedDemoWorkspaceJob.name,
+      jobName: DataSeedDemoWorkspaceJob.name,
-      dataSeedDemoWorkspaceCronPattern,
+    });
-    );
   }
 }

packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts

@@ -1,8 +1,8 @@
 import { OnModuleDestroy } from '@nestjs/common';
 
 import { JobsOptions, Queue, QueueOptions, Worker } from 'bullmq';
-import omitBy from 'lodash.omitby';
 import { v4 } from 'uuid';
+import { isDefined } from 'twenty-shared';
 
 import {
   QueueCronJobOptions,
@@ -13,6 +13,7 @@ import { MessageQueueJob } from 'src/engine/core-modules/message-queue/interface
 import { MessageQueueWorkerOptions } from 'src/engine/core-modules/message-queue/interfaces/message-queue-worker-options.interface';
 
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { getJobKey } from 'src/engine/core-modules/message-queue/utils/get-job-key.util';
 
 export type BullMQDriverOptions = QueueOptions;
 
@@ -49,54 +50,72 @@ export class BullMQDriver implements MessageQueueDriver, OnModuleDestroy {
     handler: (job: MessageQueueJob<T>) => Promise<void>,
     options?: MessageQueueWorkerOptions,
   ) {
-    const worker = new Worker(
+    const workerOptions = isDefined(options?.concurrency)
+      ? {
+          ...this.options,
+          concurrency: options.concurrency,
+        }
+      : this.options;
+
+    this.workerMap[queueName] = new Worker(
       queueName,
       async (job) => {
         // TODO: Correctly support for job.id
         await handler({ data: job.data, id: job.id ?? '', name: job.name });
       },
-      omitBy(
+      workerOptions,
-        {
-          ...this.options,
-          concurrency: options?.concurrency,
-        },
-        (value) => value === undefined,
-      ),
     );
-
-    this.workerMap[queueName] = worker;
   }
 
-  async addCron<T>(
+  async addCron<T>({
-    queueName: MessageQueue,
+    queueName,
-    jobName: string,
+    jobName,
-    data: T,
+    data,
-    options?: QueueCronJobOptions,
+    options,
-  ): Promise<void> {
+    jobId,
+  }: {
+    queueName: MessageQueue;
+    jobName: string;
+    data: T;
+    options: QueueCronJobOptions;
+    jobId?: string;
+  }): Promise<void> {
     if (!this.queueMap[queueName]) {
       throw new Error(
         `Queue ${queueName} is not registered, make sure you have added it as a queue provider`,
       );
     }
+
     const queueOptions: JobsOptions = {
-      jobId: options?.id,
       priority: options?.priority,
       repeat: options?.repeat,
       removeOnComplete: true,
       removeOnFail: 100,
     };
 
-    await this.queueMap[queueName].add(jobName, data, queueOptions);
+    await this.queueMap[queueName].upsertJobScheduler(
+      getJobKey({ jobName, jobId }),
+      options?.repeat,
+      {
+        name: jobName,
+        data,
+        opts: queueOptions,
+      },
+    );
   }
 
-  async removeCron(
+  async removeCron({
-    queueName: MessageQueue,
+    queueName,
-    jobName: string,
+    jobName,
-    pattern: string,
+    jobId,
-  ): Promise<void> {
+  }: {
-    await this.queueMap[queueName].removeRepeatable(jobName, {
+    queueName: MessageQueue;
-      pattern,
+    jobName: string;
-    });
+    jobId?: string;
+  }): Promise<void> {
+    await this.queueMap[queueName].removeJobScheduler(
+      getJobKey({ jobName, jobId }),
+    );
   }
 
   async add<T>(

packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/job-options.interface.ts

@@ -5,7 +5,7 @@ export interface QueueJobOptions {
 }
 
 export interface QueueCronJobOptions extends QueueJobOptions {
-  repeat?: {
+  repeat: {
     every?: number;
     pattern?: string;
     limit?: number;

packages/twenty-server/src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface.ts

@@ -19,12 +19,27 @@ export interface MessageQueueDriver {
     handler: ({ data, id }: { data: T; id: string }) => Promise<void> | void,
     options?: MessageQueueWorkerOptions,
   );
-  addCron<T extends MessageQueueJobData | undefined>(
+  addCron<T extends MessageQueueJobData | undefined>({
-    queueName: MessageQueue,
+    queueName,
-    jobName: string,
+    jobName,
-    data: T,
+    data,
-    options?: QueueCronJobOptions,
+    options,
-  );
+    jobId,
-  removeCron(queueName: MessageQueue, jobName: string, pattern?: string);
+  }: {
+    queueName: MessageQueue;
+    jobName: string;
+    data: T;
+    options: QueueCronJobOptions;
+    jobId?: string;
+  });
+  removeCron({
+    queueName,
+    jobName,
+    jobId,
+  }: {
+    queueName: MessageQueue;
+    jobName: string;
+    jobId?: string;
+  });
   register?(queueName: MessageQueue): void;
 }

packages/twenty-server/src/engine/core-modules/message-queue/drivers/pg-boss.driver.ts

@@ -11,6 +11,7 @@ import { MessageQueueWorkerOptions } from 'src/engine/core-modules/message-queue
 import { MessageQueueDriver } from 'src/engine/core-modules/message-queue/drivers/interfaces/message-queue-driver.interface';
 
 import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { getJobKey } from 'src/engine/core-modules/message-queue/utils/get-job-key.util';
 
 export type PgBossDriverOptions = PgBoss.ConstructorOptions;
 
@@ -62,27 +63,40 @@ export class PgBossDriver
     );
   }
 
-  async addCron<T>(
+  async addCron<T>({
-    queueName: MessageQueue,
+    queueName,
-    jobName: string,
+    jobName,
-    data: T,
+    data,
-    options?: QueueCronJobOptions,
+    options,
-  ): Promise<void> {
+    jobId,
+  }: {
+    queueName: MessageQueue;
+    jobName: string;
+    data: T;
+    options: QueueCronJobOptions;
+    jobId?: string;
+  }): Promise<void> {
+    const name = `${queueName}.${getJobKey({ jobName, jobId })}`;
+
     await this.pgBoss.schedule(
-      `${queueName}.${jobName}`,
+      name,
-      options?.repeat?.pattern ??
+      options.repeat.pattern ?? DEFAULT_PG_BOSS_CRON_PATTERN_WHEN_NOT_PROVIDED,
-        DEFAULT_PG_BOSS_CRON_PATTERN_WHEN_NOT_PROVIDED,
       data as object,
-      options
-        ? {
-            singletonKey: options?.id,
-          }
-        : {},
     );
   }
 
-  async removeCron(queueName: MessageQueue, jobName: string): Promise<void> {
+  async removeCron({
-    await this.pgBoss.unschedule(`${queueName}.${jobName}`);
+    queueName,
+    jobName,
+    jobId,
+  }: {
+    queueName: MessageQueue;
+    jobName: string;
+    jobId?: string;
+  }): Promise<void> {
+    const name = `${queueName}.${getJobKey({ jobName, jobId })}`;
+
+    await this.pgBoss.unschedule(name);
   }
 
   async add<T>(

packages/twenty-server/src/engine/core-modules/message-queue/drivers/sync.driver.ts

@@ -24,11 +24,15 @@ export class SyncDriver implements MessageQueueDriver {
     await this.processJob(queueName, { id: '', name: jobName, data });
   }
 
-  async addCron<T extends MessageQueueJobData | undefined>(
+  async addCron<T extends MessageQueueJobData | undefined>({
-    queueName: MessageQueue,
+    queueName,
-    jobName: string,
+    jobName,
-    data: T,
+    data,
-  ): Promise<void> {
+  }: {
+    queueName: MessageQueue;
+    jobName: string;
+    data: T;
+  }): Promise<void> {
     this.logger.log(`Running cron job with SyncDriver`);
     await this.processJob(queueName, {
       id: '',
@@ -38,7 +42,7 @@ export class SyncDriver implements MessageQueueDriver {
     });
   }
 
-  async removeCron(queueName: MessageQueue) {
+  async removeCron({ queueName }: { queueName: MessageQueue }) {
     this.logger.log(`Removing '${queueName}' cron job with SyncDriver`);
   }
 

packages/twenty-server/src/engine/core-modules/message-queue/services/message-queue.service.ts

@@ -35,16 +35,38 @@ export class MessageQueueService {
     return this.driver.add(this.queueName, jobName, data, options);
   }
 
-  addCron<T extends MessageQueueJobData | undefined>(
+  addCron<T extends MessageQueueJobData | undefined>({
-    jobName: string,
+    jobName,
-    data: T,
+    data,
-    options?: QueueCronJobOptions,
+    options,
-  ): Promise<void> {
+    jobId,
-    return this.driver.addCron(this.queueName, jobName, data, options);
+  }: {
+    jobName: string;
+    data: T;
+    options: QueueCronJobOptions;
+    jobId?: string;
+  }): Promise<void> {
+    return this.driver.addCron({
+      queueName: this.queueName,
+      jobName,
+      data,
+      options,
+      jobId,
+    });
   }
 
-  removeCron(jobName: string, pattern: string): Promise<void> {
+  removeCron({
-    return this.driver.removeCron(this.queueName, jobName, pattern);
+    jobName,
+    jobId,
+  }: {
+    jobName: string;
+    jobId?: string;
+  }): Promise<void> {
+    return this.driver.removeCron({
+      queueName: this.queueName,
+      jobName,
+      jobId,
+    });
   }
 
   work<T extends MessageQueueJobData>(

packages/twenty-server/src/engine/core-modules/message-queue/utils/get-job-key.util.ts

@@ -0,0 +1,9 @@
+export const getJobKey = ({
+  jobName,
+  jobId,
+}: {
+  jobName: string;
+  jobId?: string;
+}) => {
+  return `${jobName}${jobId ? `.${jobId}` : ''}`;
+};

packages/twenty-server/src/engine/workspace-manager/workspace-cleaner/commands/clean-suspended-workspaces.cron.command.ts

@@ -19,12 +19,12 @@ export class CleanSuspendedWorkspacesCronCommand extends CommandRunner {
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.addCron<undefined>(
+    await this.messageQueueService.addCron<undefined>({
-      CleanSuspendedWorkspacesJob.name,
+      jobName: CleanSuspendedWorkspacesJob.name,
-      undefined,
+      data: undefined,
-      {
+      options: {
         repeat: { pattern: cleanSuspendedWorkspaceCronPattern },
       },
-    );
+    });
   }
 }

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-event-list-fetch.cron.command.ts

@@ -20,14 +20,14 @@ export class CalendarEventListFetchCronCommand extends CommandRunner {
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.addCron<undefined>(
+    await this.messageQueueService.addCron<undefined>({
-      CalendarEventListFetchCronJob.name,
+      jobName: CalendarEventListFetchCronJob.name,
-      undefined,
+      data: undefined,
-      {
+      options: {
         repeat: {
           pattern: CALENDAR_EVENTS_LIST_CRON_PATTERN,
         },
       },
-    );
+    });
   }
 }

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-import.cron.command.ts

@@ -21,12 +21,12 @@ export class CalendarEventsImportCronCommand extends CommandRunner {
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.addCron<undefined>(
+    await this.messageQueueService.addCron<undefined>({
-      CalendarEventsImportCronJob.name,
+      jobName: CalendarEventsImportCronJob.name,
-      undefined,
+      data: undefined,
-      {
+      options: {
         repeat: { pattern: CALENDAR_EVENTS_IMPORT_CRON_PATTERN },
       },
-    );
+    });
   }
 }

packages/twenty-server/src/modules/calendar/calendar-event-import-manager/crons/commands/calendar-ongoing-stale.cron.command.ts

@@ -22,12 +22,12 @@ export class CalendarOngoingStaleCronCommand extends CommandRunner {
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.addCron<undefined>(
+    await this.messageQueueService.addCron<undefined>({
-      CalendarOngoingStaleCronJob.name,
+      jobName: CalendarOngoingStaleCronJob.name,
-      undefined,
+      data: undefined,
-      {
+      options: {
         repeat: { pattern: CALENDAR_ONGOING_STALE_CRON_PATTERN },
       },
-    );
+    });
   }
 }

packages/twenty-server/src/modules/messaging/message-import-manager/crons/commands/messaging-message-list-fetch.cron.command.ts

@@ -22,12 +22,12 @@ export class MessagingMessageListFetchCronCommand extends CommandRunner {
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.addCron<undefined>(
+    await this.messageQueueService.addCron<undefined>({
-      MessagingMessageListFetchCronJob.name,
+      jobName: MessagingMessageListFetchCronJob.name,
-      undefined,
+      data: undefined,
-      {
+      options: {
         repeat: { pattern: MESSAGING_MESSAGE_LIST_FETCH_CRON_PATTERN },
       },
-    );
+    });
   }
 }

packages/twenty-server/src/modules/messaging/message-import-manager/crons/commands/messaging-messages-import.cron.command.ts

@@ -21,14 +21,14 @@ export class MessagingMessagesImportCronCommand extends CommandRunner {
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.addCron<undefined>(
+    await this.messageQueueService.addCron<undefined>({
-      MessagingMessagesImportCronJob.name,
+      jobName: MessagingMessagesImportCronJob.name,
-      undefined,
+      data: undefined,
-      {
+      options: {
         repeat: {
           pattern: MESSAGING_MESSAGES_IMPORT_CRON_PATTERN,
         },
       },
-    );
+    });
   }
 }

packages/twenty-server/src/modules/messaging/message-import-manager/crons/commands/messaging-ongoing-stale.cron.command.ts

@@ -22,12 +22,12 @@ export class MessagingOngoingStaleCronCommand extends CommandRunner {
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.addCron<undefined>(
+    await this.messageQueueService.addCron<undefined>({
-      MessagingOngoingStaleCronJob.name,
+      jobName: MessagingOngoingStaleCronJob.name,
-      undefined,
+      data: undefined,
-      {
+      options: {
         repeat: { pattern: MESSAGING_ONGOING_STALE_CRON_PATTERN },
       },
-    );
+    });
   }
 }

packages/twenty-server/src/modules/messaging/monitoring/crons/commands/messaging-message-channel-sync-status-monitoring.cron.command.ts

@@ -22,15 +22,15 @@ export class MessagingMessageChannelSyncStatusMonitoringCronCommand extends Comm
   }
 
   async run(): Promise<void> {
-    await this.messageQueueService.addCron<undefined>(
+    await this.messageQueueService.addCron<undefined>({
-      MessagingMessageChannelSyncStatusMonitoringCronJob.name,
+      jobName: MessagingMessageChannelSyncStatusMonitoringCronJob.name,
-      undefined,
+      data: undefined,
-      {
+      options: {
         repeat: {
           pattern:
             MESSAGING_MESSAGE_CHANNEL_SYNC_STATUS_MONITORING_CRON_PATTERN,
         },
       },
-    );
+    });
   }
 }

packages/twenty-server/src/modules/workflow/workflow-trigger/database-event-trigger/listeners/database-event-trigger.listener.ts

@@ -13,9 +13,9 @@ import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.
 import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
 import { WorkflowEventListenerWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-event-listener.workspace-entity';
 import {
-  WorkflowEventTriggerJob,
+  WorkflowTriggerJob,
-  WorkflowEventTriggerJobData,
+  WorkflowTriggerJobData,
-} from 'src/modules/workflow/workflow-trigger/jobs/workflow-event-trigger.job';
+} from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
 import { OnDatabaseBatchEvent } from 'src/engine/api/graphql/graphql-query-runner/decorators/on-database-batch-event.decorator';
 import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
 
@@ -103,8 +103,8 @@ export class DatabaseEventTriggerListener {
 
     for (const eventListener of eventListeners) {
       for (const eventPayload of payload.events) {
-        this.messageQueueService.add<WorkflowEventTriggerJobData>(
+        this.messageQueueService.add<WorkflowTriggerJobData>(
-          WorkflowEventTriggerJob.name,
+          WorkflowTriggerJob.name,
           {
             workspaceId,
             workflowId: eventListener.workflowId,

packages/twenty-server/src/modules/workflow/workflow-trigger/jobs/workflow-event-trigger.job.ts

@@ -1,76 +0,0 @@
-import { Scope } from '@nestjs/common';
-
-import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
-import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
-import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
-import { FieldActorSource } from 'src/engine/metadata-modules/field-metadata/composite-types/actor.composite-type';
-import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
-import {
-  WorkflowVersionStatus,
-  WorkflowVersionWorkspaceEntity,
-} from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
-import { WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity';
-import { WorkflowRunnerWorkspaceService } from 'src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service';
-import {
-  WorkflowTriggerException,
-  WorkflowTriggerExceptionCode,
-} from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception';
-
-export type WorkflowEventTriggerJobData = {
-  workspaceId: string;
-  workflowId: string;
-  payload: object;
-};
-
-@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST })
-export class WorkflowEventTriggerJob {
-  constructor(
-    private readonly twentyORMManager: TwentyORMManager,
-    private readonly workflowRunnerWorkspaceService: WorkflowRunnerWorkspaceService,
-  ) {}
-
-  @Process(WorkflowEventTriggerJob.name)
-  async handle(data: WorkflowEventTriggerJobData): Promise<void> {
-    const workflowRepository =
-      await this.twentyORMManager.getRepository<WorkflowWorkspaceEntity>(
-        'workflow',
-      );
-
-    const workflow = await workflowRepository.findOneByOrFail({
-      id: data.workflowId,
-    });
-
-    if (!workflow.lastPublishedVersionId) {
-      throw new WorkflowTriggerException(
-        'Workflow has no published version',
-        WorkflowTriggerExceptionCode.INTERNAL_ERROR,
-      );
-    }
-
-    const workflowVersionRepository =
-      await this.twentyORMManager.getRepository<WorkflowVersionWorkspaceEntity>(
-        'workflowVersion',
-      );
-
-    const workflowVersion = await workflowVersionRepository.findOneByOrFail({
-      id: workflow.lastPublishedVersionId,
-    });
-
-    if (workflowVersion.status !== WorkflowVersionStatus.ACTIVE) {
-      throw new WorkflowTriggerException(
-        'Workflow version is not active',
-        WorkflowTriggerExceptionCode.INTERNAL_ERROR,
-      );
-    }
-
-    await this.workflowRunnerWorkspaceService.run(
-      data.workspaceId,
-      workflow.lastPublishedVersionId,
-      data.payload,
-      {
-        source: FieldActorSource.WORKFLOW,
-        name: workflow.name,
-      },
-    );
-  }
-}

packages/twenty-server/src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job.ts

@@ -0,0 +1,89 @@
+import { Scope } from '@nestjs/common';
+
+import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
+import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { FieldActorSource } from 'src/engine/metadata-modules/field-metadata/composite-types/actor.composite-type';
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import {
+  WorkflowVersionStatus,
+  WorkflowVersionWorkspaceEntity,
+} from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
+import { WorkflowWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow.workspace-entity';
+import { WorkflowRunnerWorkspaceService } from 'src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service';
+import {
+  WorkflowTriggerException,
+  WorkflowTriggerExceptionCode,
+} from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+
+export type WorkflowTriggerJobData = {
+  workspaceId: string;
+  workflowId: string;
+  payload: object;
+};
+
+@Processor({ queueName: MessageQueue.workflowQueue, scope: Scope.REQUEST })
+export class WorkflowTriggerJob {
+  constructor(
+    private readonly twentyORMManager: TwentyORMManager,
+    private readonly workflowRunnerWorkspaceService: WorkflowRunnerWorkspaceService,
+    @InjectMessageQueue(MessageQueue.workflowQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {}
+
+  @Process(WorkflowTriggerJob.name)
+  async handle(data: WorkflowTriggerJobData): Promise<void> {
+    try {
+      const workflowRepository =
+        await this.twentyORMManager.getRepository<WorkflowWorkspaceEntity>(
+          'workflow',
+        );
+
+      const workflow = await workflowRepository.findOneByOrFail({
+        id: data.workflowId,
+      });
+
+      if (!workflow.lastPublishedVersionId) {
+        throw new WorkflowTriggerException(
+          'Workflow has no published version',
+          WorkflowTriggerExceptionCode.INTERNAL_ERROR,
+        );
+      }
+
+      const workflowVersionRepository =
+        await this.twentyORMManager.getRepository<WorkflowVersionWorkspaceEntity>(
+          'workflowVersion',
+        );
+
+      const workflowVersion = await workflowVersionRepository.findOneByOrFail({
+        id: workflow.lastPublishedVersionId,
+      });
+
+      if (workflowVersion.status !== WorkflowVersionStatus.ACTIVE) {
+        throw new WorkflowTriggerException(
+          'Workflow version is not active',
+          WorkflowTriggerExceptionCode.INTERNAL_ERROR,
+        );
+      }
+
+      await this.workflowRunnerWorkspaceService.run(
+        data.workspaceId,
+        workflow.lastPublishedVersionId,
+        data.payload,
+        {
+          source: FieldActorSource.WORKFLOW,
+          name: workflow.name,
+        },
+      );
+    } catch (e) {
+      // We remove cron if it exists when no valid workflowVersion exists
+      await this.messageQueueService.removeCron({
+        jobName: WorkflowTriggerJob.name,
+        jobId: data.workflowId,
+      });
+      throw e;
+    }
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-trigger/types/workflow-trigger.type.ts

@@ -3,6 +3,7 @@ import { OutputSchema } from 'src/modules/workflow/workflow-builder/types/output
 export enum WorkflowTriggerType {
   DATABASE_EVENT = 'DATABASE_EVENT',
   MANUAL = 'MANUAL',
+  CRON = 'CRON',
 }
 
 type BaseWorkflowTriggerSettings = {
@@ -35,8 +36,16 @@ export type WorkflowManualTrigger = BaseTrigger & {
   };
 };
 
+export type WorkflowCronTrigger = BaseTrigger & {
+  type: WorkflowTriggerType.CRON;
+  settings: {
+    pattern: string;
+  };
+};
+
 export type WorkflowManualTriggerSettings = WorkflowManualTrigger['settings'];
 
 export type WorkflowTrigger =
   | WorkflowDatabaseEventTrigger
-  | WorkflowManualTrigger;
+  | WorkflowManualTrigger
+  | WorkflowCronTrigger;

packages/twenty-server/src/modules/workflow/workflow-trigger/workflow-trigger.module.ts

@@ -6,7 +6,7 @@ import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/s
 import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-common.module';
 import { WorkflowRunnerModule } from 'src/modules/workflow/workflow-runner/workflow-runner.module';
 import { DatabaseEventTriggerModule } from 'src/modules/workflow/workflow-trigger/database-event-trigger/database-event-trigger.module';
-import { WorkflowEventTriggerJob } from 'src/modules/workflow/workflow-trigger/jobs/workflow-event-trigger.job';
+import { WorkflowTriggerJob } from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
 import { WorkflowTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service';
 import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
 
@@ -20,7 +20,7 @@ import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadat
   providers: [
     WorkflowTriggerWorkspaceService,
     ScopedWorkspaceContextFactory,
-    WorkflowEventTriggerJob,
+    WorkflowTriggerJob,
   ],
   exports: [WorkflowTriggerWorkspaceService],
 })

packages/twenty-server/src/modules/workflow/workflow-trigger/workspace-services/workflow-trigger.workspace-service.ts

@@ -29,6 +29,13 @@ import {
 import { WorkflowTriggerType } from 'src/modules/workflow/workflow-trigger/types/workflow-trigger.type';
 import { assertVersionCanBeActivated } from 'src/modules/workflow/workflow-trigger/utils/assert-version-can-be-activated.util';
 import { assertNever } from 'src/utils/assert';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import {
+  WorkflowTriggerJob,
+  WorkflowTriggerJobData,
+} from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
 
 @Injectable()
 export class WorkflowTriggerWorkspaceService {
@@ -41,6 +48,8 @@ export class WorkflowTriggerWorkspaceService {
     private readonly workspaceEventEmitter: WorkspaceEventEmitter,
     @InjectRepository(ObjectMetadataEntity, 'metadata')
     private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
+    @InjectMessageQueue(MessageQueue.workflowQueue)
+    private readonly messageQueueService: MessageQueueService,
   ) {}
 
   private getWorkspaceId() {
@@ -329,6 +338,23 @@ export class WorkflowTriggerWorkspaceService {
 
         return;
       case WorkflowTriggerType.MANUAL:
+        return;
+      case WorkflowTriggerType.CRON:
+        await this.messageQueueService.addCron<WorkflowTriggerJobData>({
+          jobName: WorkflowTriggerJob.name,
+          jobId: workflowVersion.workflowId,
+          data: {
+            workspaceId: this.getWorkspaceId(),
+            workflowId: workflowVersion.workflowId,
+            payload: {},
+          },
+          options: {
+            repeat: {
+              pattern: workflowVersion.trigger.settings.pattern,
+            },
+          },
+        });
+
         return;
       default: {
         assertNever(workflowVersion.trigger);
@@ -351,6 +377,13 @@ export class WorkflowTriggerWorkspaceService {
 
         return;
       case WorkflowTriggerType.MANUAL:
+        return;
+      case WorkflowTriggerType.CRON:
+        await this.messageQueueService.removeCron({
+          jobName: WorkflowTriggerJob.name,
+          jobId: workflowVersion.workflowId,
+        });
+
         return;
       default:
         assertNever(workflowVersion.trigger);

packages/twenty-shared/src/constants/index.ts

@@ -0,0 +1,5 @@
+export * from './FieldForTotalCountAggregateOperation';
+export * from './Locales';
+export * from './TwentyCompaniesBaseUrl';
+export * from './TwentyIconsBaseUrl';
+export * from './SettingsFeatures';

packages/twenty-shared/src/index.ts

@@ -1,13 +1,4 @@
-export * from './constants/FieldForTotalCountAggregateOperation';
+export * from './constants';
-export * from './constants/Locales';
+export * from './types';
-export * from './constants/SettingsFeatures';
+export * from './utils';
-export * from './constants/TwentyCompaniesBaseUrl';
-export * from './constants/TwentyIconsBaseUrl';
-export * from './types/ConnectedAccountProvider';
-export * from './types/FieldMetadataType';
-export * from './utils/fieldMetadata/isFieldMetadataDateKind';
-export * from './utils/image/getImageAbsoluteURI';
-export * from './utils/isDefined';
-export * from './utils/isValidLocale';
-export * from './utils/strings';
 export * from './workspace';

packages/twenty-shared/src/types/index.ts

@@ -0,0 +1,2 @@
+export * from './ConnectedAccountProvider';
+export * from './FieldMetadataType';

packages/twenty-shared/src/utils/fieldMetadata/index.ts

@@ -0,0 +1 @@
+export * from './isFieldMetadataDateKind';

packages/twenty-shared/src/utils/image/index.ts

@@ -0,0 +1 @@
+export * from './getImageAbsoluteURI';

packages/twenty-shared/src/utils/index.ts

@@ -0,0 +1,5 @@
+export * from './fieldMetadata';
+export * from './image';
+export * from './strings';
+export * from './validation';
+export * from './validation';

packages/twenty-shared/src/utils/strings/__tests__/capitalize.test.ts

@@ -1,4 +1,4 @@
-import { capitalize } from '../capitalize.util';
+import { capitalize } from '../capitalize';
 describe('capitalize', () => {
   it('should capitalize a string', () => {
     expect(capitalize('test')).toBe('Test');

packages/twenty-shared/src/utils/strings/__tests__/isValidUuid.test.ts

@@ -1,11 +0,0 @@
-import { isValidUuid } from '../isValidUuid.util';
-
-describe('isValidUuid', () => {
-  it('should return true for a valid UUID', () => {
-    expect(isValidUuid('123e4567-e89b-12d3-a456-426614174000')).toBe(true);
-  });
-
-  it('should return false for an invalid UUID', () => {
-    expect(isValidUuid('123e4567-e89b-12d3-a456-426614174000')).toBe(false);
-  });
-});

packages/twenty-shared/src/utils/strings/index.ts

@@ -1,2 +1 @@
-export * from './capitalize.util';
+export * from './capitalize';
-export * from './isValidUuid.util';

packages/twenty-shared/src/utils/strings/isValidUuid.util.ts

@@ -1,5 +0,0 @@
-export const isValidUuid = (value: string) => {
-  return /^[0-9A-F]{8}-[0-9A-F]{4}-[4][0-9A-F]{3}-[89AB][0-9A-F]{3}-[0-9A-F]{12}$/i.test(
-    value,
-  );
-};

packages/twenty-shared/src/utils/validation/__tests__/isValidUuid.test.ts

@@ -0,0 +1,19 @@
+import { isValidUuid } from '../isValidUuid';
+
+describe('isValidUuid', () => {
+  it('should return true for a valid UUID', () => {
+    expect(isValidUuid('123e4567-e89b-12d3-a456-426614174000')).toBe(true);
+    expect(isValidUuid('550e8400-e29b-41d4-a716-446655440000')).toBe(true);
+  });
+
+  it('should return false for an invalid UUID', () => {
+    expect(isValidUuid('invalid-uuid')).toBe(false);
+    expect(isValidUuid('12345')).toBe(false);
+    expect(isValidUuid('550e8400e29b41d4a716446655440000')).toBe(false);
+    expect(isValidUuid('')).toBe(false);
+    expect(isValidUuid('123e4567-e89b-12d3-a456-42661417400-')).toBe(false);
+    expect(isValidUuid('123e4567-e89b-12d3-a456-42661417400')).toBe(false);
+    expect(isValidUuid('123e4567-e89b-12d3-a456-42661417400)')).toBe(false);
+    expect(isValidUuid('123e4567-e89b-12d3-a456-4266141740001')).toBe(false);
+  });
+});

packages/twenty-shared/src/utils/validation/__tests__/isValideLocale.test.ts

@@ -0,0 +1,15 @@
+import { isValidLocale } from '../isValidLocale';
+import { APP_LOCALES } from 'src/constants/Locales';
+
+describe('isValidLocale', () => {
+  it('should return true for valid locales', () => {
+    Object.keys(APP_LOCALES).forEach((locale) => {
+      expect(isValidLocale(locale)).toBe(true);
+    });
+  });
+
+  it('should return false for invalid locales', () => {
+    expect(isValidLocale('invalidLocale')).toBe(false);
+    expect(isValidLocale(null)).toBe(false);
+  });
+});

packages/twenty-shared/src/utils/validation/index.ts

@@ -0,0 +1,3 @@
+export * from './isValidUuid';
+export * from './isDefined';
+export * from './isValidLocale';

packages/twenty-shared/src/utils/validation/isValidUuid.ts

@@ -0,0 +1,5 @@
+export const isValidUuid = (value: string): boolean => {
+  const uuidRegex =
+    /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
+  return uuidRegex.test(value);
+};

packages/twenty-shared/src/workspace/index.ts

@@ -1,2 +1,2 @@
-export * from './types/WorkspaceActivationStatus';
+export * from './types';
-export * from './utils/isWorkspaceActiveOrSuspended';
+export * from './utils';

packages/twenty-shared/src/workspace/types/index.ts

@@ -0,0 +1 @@
+export * from './WorkspaceActivationStatus';

packages/twenty-shared/src/workspace/utils/index.ts

@@ -0,0 +1 @@
+export * from './isWorkspaceActiveOrSuspended';

yarn.lock

@@ -22018,20 +22018,18 @@ __metadata:
   languageName: node
   linkType: hard
 
-"bullmq@npm:^4.15.0":
+"bullmq@npm:^5.40.0":
-  version: 4.18.2
+  version: 5.40.0
-  resolution: "bullmq@npm:4.18.2"
+  resolution: "bullmq@npm:5.40.0"
   dependencies:
-    cron-parser: "npm:^4.6.0"
+    cron-parser: "npm:^4.9.0"
-    glob: "npm:^8.0.3"
+    ioredis: "npm:^5.4.1"
-    ioredis: "npm:^5.3.2"
+    msgpackr: "npm:^1.11.2"
-    lodash: "npm:^4.17.21"
-    msgpackr: "npm:^1.6.2"
     node-abort-controller: "npm:^3.1.1"
     semver: "npm:^7.5.4"
     tslib: "npm:^2.0.0"
     uuid: "npm:^9.0.0"
-  checksum: 10c0/09371a6d53377e556a37e3e046576bb20056a14ac26d29528bd0a4054c33f458f1764ae696ae6a39fda688e8215ef34f1ba94fe56ba89db5b9ad2c9aa0082f2f
+  checksum: 10c0/00ca72939af44815cfd33e366d2bf650232ac14bc2e3afe32c3aa4bbd74e6b23bec6622f1ecf4a0da684c548ac5553738457a9780bba9bfe23bb831efb766eb8
   languageName: node
   linkType: hard
 
@@ -24025,7 +24023,7 @@ __metadata:
   languageName: node
   linkType: hard
 
-"cron-parser@npm:^4.0.0, cron-parser@npm:^4.6.0":
+"cron-parser@npm:^4.0.0, cron-parser@npm:^4.9.0":
   version: 4.9.0
   resolution: "cron-parser@npm:4.9.0"
   dependencies:
@@ -29005,7 +29003,7 @@ __metadata:
   languageName: node
   linkType: hard
 
-"glob@npm:^8.0.1, glob@npm:^8.0.3, glob@npm:^8.1.0":
+"glob@npm:^8.0.1, glob@npm:^8.1.0":
   version: 8.1.0
   resolution: "glob@npm:8.1.0"
   dependencies:
@@ -31081,9 +31079,9 @@ __metadata:
   languageName: node
   linkType: hard
 
-"ioredis@npm:^5.3.2":
+"ioredis@npm:^5.4.1":
-  version: 5.4.1
+  version: 5.4.2
-  resolution: "ioredis@npm:5.4.1"
+  resolution: "ioredis@npm:5.4.2"
   dependencies:
     "@ioredis/commands": "npm:^1.1.1"
     cluster-key-slot: "npm:^1.1.0"
@@ -31094,7 +31092,7 @@ __metadata:
     redis-errors: "npm:^1.2.0"
     redis-parser: "npm:^3.0.0"
     standard-as-callback: "npm:^2.1.0"
-  checksum: 10c0/5d28b7c89a3cab5b76d75923d7d4ce79172b3a1ca9be690133f6e8e393a7a4b4ffd55513e618bbb5504fed80d9e1395c9d9531a7c5c5c84aa4c4e765cca75456
+  checksum: 10c0/e59d2cceb43ed74b487d7b50fa91b93246e734e5d4835c7e62f64e44da072f12ab43b044248012e6f8b76c61a7c091a2388caad50e8ad69a8ce5515a730b23b8
   languageName: node
   linkType: hard
 
@@ -36774,15 +36772,15 @@ __metadata:
   languageName: node
   linkType: hard
 
-"msgpackr@npm:^1.6.2":
+"msgpackr@npm:^1.11.2":
-  version: 1.11.0
+  version: 1.11.2
-  resolution: "msgpackr@npm:1.11.0"
+  resolution: "msgpackr@npm:1.11.2"
   dependencies:
     msgpackr-extract: "npm:^3.0.2"
   dependenciesMeta:
     msgpackr-extract:
       optional: true
-  checksum: 10c0/a7edc36754ec9f8469bc14c896f0f36e0e3de595c0bb5ac7b2ab8c2a72a2e188c12f1345d71a127f8537d9bbc880407a7073ac1d29c27822178bc0b81ae7370e
+  checksum: 10c0/7d2e81ca82c397b2352d470d6bc8f4a967fe4fe14f8fc1fc9906b23009fdfb543999b1ad29c700b8861581e0b6bf903d6f0fefb69a09375cbca6d4d802e6c906
   languageName: node
   linkType: hard
 
@@ -46188,7 +46186,7 @@ __metadata:
     bcrypt: "npm:^5.1.1"
     better-sqlite3: "npm:^9.2.2"
     body-parser: "npm:^1.20.2"
-    bullmq: "npm:^4.15.0"
+    bullmq: "npm:^5.40.0"
     bytes: "npm:^3.1.2"
     chromatic: "npm:^6.18.0"
     class-transformer: "npm:^0.5.1"