866 refactor cron trigger only one cron each minutes triggers all cron triggers (#11809)

<img width="1123" alt="image"
src="https://github.com/user-attachments/assets/75447922-81dd-4cfc-805d-f511f73cc778"
/>

packages/twenty-server/src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity.ts

@@ -0,0 +1,84 @@
+import { msg } from '@lingui/core/macro';
+import { Relation } from 'typeorm';
+import { FieldMetadataType } from 'twenty-shared/types';
+
+import { RelationType } from 'src/engine/metadata-modules/field-metadata/interfaces/relation-type.interface';
+
+import { BaseWorkspaceEntity } from 'src/engine/twenty-orm/base.workspace-entity';
+import { WorkspaceEntity } from 'src/engine/twenty-orm/decorators/workspace-entity.decorator';
+import { WorkspaceIsSystem } from 'src/engine/twenty-orm/decorators/workspace-is-system.decorator';
+import { STANDARD_OBJECT_ICONS } from 'src/engine/workspace-manager/workspace-sync-metadata/constants/standard-object-icons';
+import { STANDARD_OBJECT_IDS } from 'src/engine/workspace-manager/workspace-sync-metadata/constants/standard-object-ids';
+import { WORKFLOW_AUTOMATED_TRIGGER_STANDARD_FIELD_IDS } from 'src/engine/workspace-manager/workspace-sync-metadata/constants/standard-field-ids';
+import { WorkspaceRelation } from 'src/engine/twenty-orm/decorators/workspace-relation.decorator';
+import { RelationOnDeleteAction } from 'src/engine/metadata-modules/relation-metadata/relation-metadata.entity';
+import { WorkspaceJoinColumn } from 'src/engine/twenty-orm/decorators/workspace-join-column.decorator';
+import { WorkspaceField } from 'src/engine/twenty-orm/decorators/workspace-field.decorator';
+
+import { WorkflowWorkspaceEntity } from './workflow.workspace-entity';
+
+export enum AutomatedTriggerType {
+  DATABASE_EVENT = 'DATABASE_EVENT',
+  CRON = 'CRON',
+}
+
+export type AutomatedTriggerSettings = {
+  pattern?: string;
+  eventName?: string;
+};
+
+@WorkspaceEntity({
+  standardId: STANDARD_OBJECT_IDS.workflowAutomatedTrigger,
+  namePlural: 'workflowAutomatedTriggers',
+  labelSingular: msg`WorkflowAutomatedTrigger`,
+  labelPlural: msg`WorkflowAutomatedTriggers`,
+  description: msg`A workflow automated trigger`,
+  icon: STANDARD_OBJECT_ICONS.workflowAutomatedTrigger,
+})
+@WorkspaceIsSystem()
+export class WorkflowAutomatedTriggerWorkspaceEntity extends BaseWorkspaceEntity {
+  @WorkspaceField({
+    standardId: WORKFLOW_AUTOMATED_TRIGGER_STANDARD_FIELD_IDS.type,
+    type: FieldMetadataType.SELECT,
+    label: msg`Automated Trigger Type`,
+    description: msg`The workflow automated trigger type`,
+    options: [
+      {
+        value: AutomatedTriggerType.DATABASE_EVENT,
+        label: 'Database Event',
+        position: 0,
+        color: 'green',
+      },
+      {
+        value: AutomatedTriggerType.CRON,
+        label: 'Cron',
+        position: 1,
+        color: 'blue',
+      },
+    ],
+  })
+  type: AutomatedTriggerType;
+
+  @WorkspaceField({
+    standardId: WORKFLOW_AUTOMATED_TRIGGER_STANDARD_FIELD_IDS.settings,
+    type: FieldMetadataType.RAW_JSON,
+    label: msg`Settings`,
+    description: msg`The workflow automated trigger settings`,
+  })
+  settings: AutomatedTriggerSettings;
+
+  @WorkspaceRelation({
+    standardId: WORKFLOW_AUTOMATED_TRIGGER_STANDARD_FIELD_IDS.workflow,
+    type: RelationType.MANY_TO_ONE,
+    label: msg`Workflow`,
+    description: msg`WorkflowAutomatedTrigger workflow`,
+    icon: 'IconSettingsAutomation',
+    inverseSideTarget: () => WorkflowWorkspaceEntity,
+    inverseSideFieldKey: 'automatedTriggers',
+    onDelete: RelationOnDeleteAction.CASCADE,
+  })
+  workflow: Relation<WorkflowWorkspaceEntity>;
+
+  @WorkspaceJoinColumn('workflow')
+  workflowId: string;
+}

packages/twenty-server/src/modules/workflow/common/standard-objects/workflow.workspace-entity.ts

@@ -21,6 +21,7 @@ import { TimelineActivityWorkspaceEntity } from 'src/modules/timeline/standard-o
 import { WorkflowEventListenerWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-event-listener.workspace-entity';
 import { WorkflowRunWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
 import { WorkflowVersionWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-version.workspace-entity';
+import { WorkflowAutomatedTriggerWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity';
 
 export enum WorkflowStatus {
   DRAFT = 'DRAFT',
@@ -135,6 +136,17 @@ export class WorkflowWorkspaceEntity extends BaseWorkspaceEntity {
   @WorkspaceIsSystem()
   eventListeners: Relation<WorkflowEventListenerWorkspaceEntity[]>;
 
+  @WorkspaceRelation({
+    standardId: WORKFLOW_STANDARD_FIELD_IDS.automatedTriggers,
+    type: RelationType.ONE_TO_MANY,
+    label: msg`Automated Triggers`,
+    description: msg`Workflow automated triggers linked to the workflow.`,
+    inverseSideTarget: () => WorkflowAutomatedTriggerWorkspaceEntity,
+    onDelete: RelationOnDeleteAction.CASCADE,
+  })
+  @WorkspaceIsSystem()
+  automatedTriggers: Relation<WorkflowAutomatedTriggerWorkspaceEntity[]>;
+
   @WorkspaceRelation({
     standardId: WORKFLOW_STANDARD_FIELD_IDS.favorites,
     type: RelationType.ONE_TO_MANY,

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

@@ -19,6 +19,7 @@ import {
   WorkflowTriggerException,
   WorkflowTriggerExceptionCode,
 } from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception';
+import { WorkflowAutomatedTriggerWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity';
 
 @Injectable()
 export class WorkflowCommonWorkspaceService {
@@ -141,11 +142,20 @@ export class WorkflowCommonWorkspaceService {
         'workflowEventListener',
       );
 
+    const workflowAutomatedTriggerRepository =
+      await this.twentyORMManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
+        'workflowAutomatedTrigger',
+      );
+
     workflowIds.forEach((workflowId) => {
       workflowEventListenerRepository.softDelete({
         workflowId,
       });
 
+      workflowAutomatedTriggerRepository.softDelete({
+        workflowId,
+      });
+
       workflowRunRepository.softDelete({
         workflowId,
       });

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

@@ -0,0 +1,21 @@
+import { Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+
+import { AutomatedTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.workspace-service';
+import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
+import { DatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener';
+import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
+import { CronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command';
+import { CronTriggerCronJob } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/cron-trigger.cron.job';
+
+@Module({
+  imports: [FeatureFlagModule, TypeOrmModule.forFeature([Workspace], 'core')],
+  providers: [
+    AutomatedTriggerWorkspaceService,
+    DatabaseEventTriggerListener,
+    CronTriggerCronJob,
+    CronTriggerCronCommand,
+  ],
+  exports: [AutomatedTriggerWorkspaceService],
+})
+export class AutomatedTriggerModule {}

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

@@ -0,0 +1,94 @@
+import { Injectable } from '@nestjs/common';
+
+import { EntityManager } from 'typeorm';
+
+import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
+import { WorkflowEventListenerWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-event-listener.workspace-entity';
+import {
+  AutomatedTriggerType,
+  AutomatedTriggerSettings,
+  WorkflowAutomatedTriggerWorkspaceEntity,
+} from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity';
+
+@Injectable()
+export class AutomatedTriggerWorkspaceService {
+  constructor(private readonly twentyORMManager: TwentyORMManager) {}
+
+  async addAutomatedTrigger({
+    workflowId,
+    manager,
+    type,
+    settings,
+  }: {
+    workflowId: string;
+    manager: EntityManager;
+    type: AutomatedTriggerType;
+    settings: AutomatedTriggerSettings;
+  }) {
+    if (type === AutomatedTriggerType.DATABASE_EVENT) {
+      // Todo: remove workflowEventListenerRepository updates when data are migrated to workflowAutomatedTrigger
+      const workflowEventListenerRepository =
+        await this.twentyORMManager.getRepository<WorkflowEventListenerWorkspaceEntity>(
+          'workflowEventListener',
+        );
+
+      const workflowEventListener = workflowEventListenerRepository.create({
+        workflowId,
+        eventName: settings.eventName,
+      });
+
+      await workflowEventListenerRepository.save(
+        workflowEventListener,
+        {},
+        manager,
+      );
+      // end-Todo
+    }
+
+    const workflowAutomatedTriggerRepository =
+      await this.twentyORMManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
+        'workflowAutomatedTrigger',
+      );
+
+    const workflowAutomatedTrigger = workflowAutomatedTriggerRepository.create({
+      type,
+      settings,
+      workflowId,
+    });
+
+    await workflowAutomatedTriggerRepository.save(
+      workflowAutomatedTrigger,
+      {},
+      manager,
+    );
+  }
+
+  async deleteAutomatedTrigger({
+    workflowId,
+    manager,
+  }: {
+    workflowId: string;
+    manager: EntityManager;
+  }) {
+    // Todo: remove workflowEventListenerRepository updates when data are migrated to workflowAutomatedTrigger
+    const workflowEventListenerRepository =
+      await this.twentyORMManager.getRepository<WorkflowEventListenerWorkspaceEntity>(
+        'workflowEventListener',
+      );
+
+    await workflowEventListenerRepository.delete(
+      {
+        workflowId,
+      },
+      manager,
+    );
+    // end-Todo
+
+    const workflowAutomatedTriggerRepository =
+      await this.twentyORMManager.getRepository<WorkflowAutomatedTriggerWorkspaceEntity>(
+        'workflowAutomatedTrigger',
+      );
+
+    await workflowAutomatedTriggerRepository.delete({ workflowId }, manager);
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/cron-trigger.cron.command.ts

@@ -0,0 +1,34 @@
+import { Command, CommandRunner } from 'nest-commander';
+
+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 {
+  CRON_TRIGGER_CRON_PATTERN,
+  CronTriggerCronJob,
+} from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/cron-trigger.cron.job';
+
+@Command({
+  name: 'cron:workflow:automated-cron-trigger',
+  description: 'Starts a cron job to trigger cron triggered workflows',
+})
+export class CronTriggerCronCommand extends CommandRunner {
+  constructor(
+    @InjectMessageQueue(MessageQueue.cronQueue)
+    private readonly messageQueueService: MessageQueueService,
+  ) {
+    super();
+  }
+
+  async run(): Promise<void> {
+    await this.messageQueueService.addCron<undefined>({
+      jobName: CronTriggerCronJob.name,
+      data: undefined,
+      options: {
+        repeat: {
+          pattern: CRON_TRIGGER_CRON_PATTERN,
+        },
+      },
+    });
+  }
+}

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

@@ -0,0 +1,81 @@
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+import { Repository } from 'typeorm';
+import { isDefined } from 'twenty-shared/utils';
+
+import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
+import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator';
+import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
+import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
+import {
+  AutomatedTriggerType,
+  WorkflowAutomatedTriggerWorkspaceEntity,
+} from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity';
+import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
+import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+import {
+  WorkflowTriggerJob,
+  WorkflowTriggerJobData,
+} from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
+import { shouldRunNow } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/utils/should-run-now.utils';
+
+export const CRON_TRIGGER_CRON_PATTERN = '* * * * *';
+
+@Processor(MessageQueue.cronQueue)
+export class CronTriggerCronJob {
+  constructor(
+    @InjectRepository(Workspace, 'core')
+    private readonly workspaceRepository: Repository<Workspace>,
+    @InjectMessageQueue(MessageQueue.workflowQueue)
+    private readonly messageQueueService: MessageQueueService,
+    private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
+  ) {}
+
+  @Process(CronTriggerCronJob.name)
+  @SentryCronMonitor(CronTriggerCronJob.name, CRON_TRIGGER_CRON_PATTERN)
+  async handle() {
+    const activeWorkspaces = await this.workspaceRepository.find({
+      where: {
+        activationStatus: WorkspaceActivationStatus.ACTIVE,
+      },
+    });
+
+    const now = new Date();
+
+    for (const activeWorkspace of activeWorkspaces) {
+      const workflowAutomatedTriggerRepository =
+        await this.twentyORMGlobalManager.getRepositoryForWorkspace<WorkflowAutomatedTriggerWorkspaceEntity>(
+          activeWorkspace.id,
+          'workflowAutomatedTrigger',
+        );
+
+      const workflowAutomatedCronTriggers =
+        await workflowAutomatedTriggerRepository.find({
+          where: { type: AutomatedTriggerType.CRON },
+        });
+
+      for (const workflowAutomatedCronTrigger of workflowAutomatedCronTriggers) {
+        if (!isDefined(workflowAutomatedCronTrigger.settings.pattern)) {
+          continue;
+        }
+
+        if (!shouldRunNow(workflowAutomatedCronTrigger.settings.pattern, now)) {
+          continue;
+        }
+
+        await this.messageQueueService.add<WorkflowTriggerJobData>(
+          WorkflowTriggerJob.name,
+          {
+            workspaceId: activeWorkspace.id,
+            workflowId: workflowAutomatedCronTrigger.workflowId,
+            payload: {},
+          },
+          { retryLimit: 3 },
+        );
+      }
+    }
+  }
+}

packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/utils/should-run-now.utils.ts

@@ -0,0 +1,20 @@
+import { CronExpressionParser } from 'cron-parser';
+
+export const shouldRunNow = (
+  pattern: string,
+  now: Date,
+  rootCronIntervalMs = 60_000,
+) => {
+  try {
+    const interval = CronExpressionParser.parse(pattern, {
+      currentDate: now,
+    });
+
+    const prevTriggerDate = interval.prev();
+    const diff = Math.abs(prevTriggerDate.getTime() - now.getTime());
+
+    return diff < rootCronIntervalMs;
+  } catch {
+    return false;
+  }
+};

packages/twenty-server/src/modules/workflow/workflow-trigger/automated-trigger/crons/utils/tests/should-run-now.utils.spec.ts

@@ -0,0 +1,47 @@
+import { shouldRunNow } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/utils/should-run-now.utils';
+
+const getNowDate = (hour: string) => {
+  return new Date(`2025-01-01T${hour}.100Z`);
+};
+
+describe('shouldRunNow', () => {
+  it('returns true when now matches cron pattern */1 * * * *', () => {
+    const cron = '*/1 * * * *';
+
+    expect(shouldRunNow(cron, getNowDate('10:00:00'))).toBe(true);
+  });
+
+  it('returns true with a 50s root cron delay', () => {
+    const cron = '*/1 * * * *';
+
+    expect(shouldRunNow(cron, getNowDate('10:00:50'))).toBe(true);
+  });
+
+  it('returns true 5 times in a row for a */5 pattern', () => {
+    const cron = '*/5 * * * *'; // every 5 minutes
+
+    expect(shouldRunNow(cron, getNowDate('09:59:00'))).toBe(false);
+    expect(shouldRunNow(cron, getNowDate('10:00:00'))).toBe(true);
+    expect(shouldRunNow(cron, getNowDate('10:01:00'))).toBe(false);
+    expect(shouldRunNow(cron, getNowDate('10:02:00'))).toBe(false);
+    expect(shouldRunNow(cron, getNowDate('10:03:00'))).toBe(false);
+    expect(shouldRunNow(cron, getNowDate('10:04:00'))).toBe(false);
+    expect(shouldRunNow(cron, getNowDate('10:05:00'))).toBe(true);
+    expect(shouldRunNow(cron, getNowDate('10:06:00'))).toBe(false);
+  });
+
+  it('returns false for invalid cron pattern', () => {
+    const cron = 'invalid-cron';
+
+    expect(shouldRunNow(cron, getNowDate('10:00:00'))).toBe(false);
+  });
+
+  it('returns false if the next run is outside the interval window (2 minutes)', () => {
+    const cron = '*/10 * * * *'; // every 10 minutes
+    const interval2min = 2 * 60_000;
+
+    expect(shouldRunNow(cron, getNowDate('10:06:00'), interval2min)).toBe(
+      false,
+    );
+  });
+});

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

@@ -13,11 +13,11 @@ import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queu
 import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
 import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
 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 {
   WorkflowTriggerJob,
   WorkflowTriggerJobData,
 } from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
+import { WorkflowEventListenerWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-event-listener.workspace-entity';
 
 @Injectable()
 export class DatabaseEventTriggerListener {
@@ -67,9 +67,9 @@ export class DatabaseEventTriggerListener {
     >,
   ) {
     const workspaceId = payload.workspaceId;
-    const eventName = payload.name;
+    const databaseEventName = payload.name;
 
-    if (!workspaceId || !eventName) {
+    if (!workspaceId || !databaseEventName) {
       this.logger.error(
         `Missing workspaceId or eventName in payload ${JSON.stringify(
           payload,
@@ -89,19 +89,43 @@ export class DatabaseEventTriggerListener {
       return;
     }
 
+    // Todo: uncomment that when data are migrated to workflowAutomatedTrigger
+    /*
+    const workflowAutomatedTriggerRepository =
+      await this.twentyORMGlobalManager.getRepositoryForWorkspace<WorkflowAutomatedTriggerWorkspaceEntity>(
+        workspaceId,
+        'workflowAutomatedTrigger',
+      );
+
+    const eventListeners = await workflowAutomatedTriggerRepository.find({
+      where: {
+        type: AutomatedTriggerType.DATABASE_EVENT,
+        settings: { eventName: databaseEventName },
+      },
+    });
+    */
+    // end Todo
+
+    // Todo: remove that when data are migrated to workflowAutomatedTrigger
     const workflowEventListenerRepository =
       await this.twentyORMGlobalManager.getRepositoryForWorkspace<WorkflowEventListenerWorkspaceEntity>(
         workspaceId,
         'workflowEventListener',
       );
 
-    const eventListeners = await workflowEventListenerRepository.find({
-      where: {
-        eventName,
-      },
+    const oldEventListeners = await workflowEventListenerRepository.find({
+      where: { eventName: databaseEventName },
     });
 
-    for (const eventListener of eventListeners) {
+    // end Todo
+
+    // Todo: uncomment that when data are migrated to workflowAutomatedTrigger
+    //for (const eventListener of eventListeners) {
+    // end Todo
+
+    // Todo: remove that when data are migrated to workflowAutomatedTrigger
+    for (const eventListener of oldEventListeners) {
+      // end Todo
       for (const eventPayload of payload.events) {
         this.messageQueueService.add<WorkflowTriggerJobData>(
           WorkflowTriggerJob.name,

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

@@ -1,12 +0,0 @@
-import { Module } from '@nestjs/common';
-
-import { FeatureFlagModule } from 'src/engine/core-modules/feature-flag/feature-flag.module';
-import { DatabaseEventTriggerService } from 'src/modules/workflow/workflow-trigger/database-event-trigger/database-event-trigger.service';
-import { DatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trigger/database-event-trigger/listeners/database-event-trigger.listener';
-
-@Module({
-  imports: [FeatureFlagModule],
-  providers: [DatabaseEventTriggerService, DatabaseEventTriggerListener],
-  exports: [DatabaseEventTriggerService],
-})
-export class DatabaseEventTriggerModule {}

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

@@ -1,50 +0,0 @@
-import { Injectable } from '@nestjs/common';
-
-import { EntityManager } from 'typeorm';
-
-import { TwentyORMManager } from 'src/engine/twenty-orm/twenty-orm.manager';
-import { WorkflowEventListenerWorkspaceEntity } from 'src/modules/workflow/common/standard-objects/workflow-event-listener.workspace-entity';
-import { WorkflowDatabaseEventTrigger } from 'src/modules/workflow/workflow-trigger/types/workflow-trigger.type';
-
-@Injectable()
-export class DatabaseEventTriggerService {
-  constructor(private readonly twentyORMManager: TwentyORMManager) {}
-
-  async createEventListener(
-    workflowId: string,
-    trigger: WorkflowDatabaseEventTrigger,
-    manager: EntityManager,
-  ) {
-    const eventName = trigger.settings.eventName;
-
-    const workflowEventListenerRepository =
-      await this.twentyORMManager.getRepository<WorkflowEventListenerWorkspaceEntity>(
-        'workflowEventListener',
-      );
-
-    const workflowEventListener = await workflowEventListenerRepository.create({
-      workflowId,
-      eventName,
-    });
-
-    await workflowEventListenerRepository.save(
-      workflowEventListener,
-      {},
-      manager,
-    );
-  }
-
-  async deleteEventListener(workflowId: string, manager: EntityManager) {
-    const workflowEventListenerRepository =
-      await this.twentyORMManager.getRepository<WorkflowEventListenerWorkspaceEntity>(
-        'workflowEventListener',
-      );
-
-    await workflowEventListenerRepository.delete(
-      {
-        workflowId,
-      },
-      manager,
-    );
-  }
-}

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

@@ -5,16 +5,16 @@ import { NestjsQueryTypeOrmModule } from '@ptc-org/nestjs-query-typeorm';
 import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/scoped-workspace-context.factory';
 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 { 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';
+import { AutomatedTriggerModule } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.module';
 
 @Module({
   imports: [
     WorkflowCommonModule,
     WorkflowRunnerModule,
-    DatabaseEventTriggerModule,
+    AutomatedTriggerModule,
     NestjsQueryTypeOrmModule.forFeature([ObjectMetadataEntity], 'metadata'),
   ],
   providers: [

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

@@ -4,9 +4,6 @@ import { InjectRepository } from '@nestjs/typeorm';
 import { EntityManager, Repository } from 'typeorm';
 
 import { DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
-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 { ActorMetadata } from 'src/engine/metadata-modules/field-metadata/composite-types/actor.composite-type';
 import { ObjectMetadataEntity } from 'src/engine/metadata-modules/object-metadata/object-metadata.entity';
 import { ScopedWorkspaceContextFactory } from 'src/engine/twenty-orm/factories/scoped-workspace-context.factory';
@@ -23,19 +20,16 @@ import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/work
 import { WorkflowRunnerWorkspaceService } from 'src/modules/workflow/workflow-runner/workspace-services/workflow-runner.workspace-service';
 import { WORKFLOW_VERSION_STATUS_UPDATED } from 'src/modules/workflow/workflow-status/constants/workflow-version-status-updated.constants';
 import { WorkflowVersionStatusUpdate } from 'src/modules/workflow/workflow-status/jobs/workflow-statuses-update.job';
-import { DatabaseEventTriggerService } from 'src/modules/workflow/workflow-trigger/database-event-trigger/database-event-trigger.service';
 import {
   WorkflowTriggerException,
   WorkflowTriggerExceptionCode,
 } from 'src/modules/workflow/workflow-trigger/exceptions/workflow-trigger.exception';
-import {
-  WorkflowTriggerJob,
-  WorkflowTriggerJobData,
-} from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
 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 { computeCronPatternFromSchedule } from 'src/modules/workflow/workflow-trigger/utils/compute-cron-pattern-from-schedule';
 import { assertNever } from 'src/utils/assert';
+import { AutomatedTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.workspace-service';
+import { AutomatedTriggerType } from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity';
 
 @Injectable()
 export class WorkflowTriggerWorkspaceService {
@@ -44,12 +38,10 @@ export class WorkflowTriggerWorkspaceService {
     private readonly workflowCommonWorkspaceService: WorkflowCommonWorkspaceService,
     private readonly scopedWorkspaceContextFactory: ScopedWorkspaceContextFactory,
     private readonly workflowRunnerWorkspaceService: WorkflowRunnerWorkspaceService,
-    private readonly databaseEventTriggerService: DatabaseEventTriggerService,
+    private readonly automatedTriggerWorkspaceService: AutomatedTriggerWorkspaceService,
     private readonly workspaceEventEmitter: WorkspaceEventEmitter,
     @InjectRepository(ObjectMetadataEntity, 'metadata')
     private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
-    @InjectMessageQueue(MessageQueue.workflowQueue)
-    private readonly messageQueueService: MessageQueueService,
   ) {}
 
   private getWorkspaceId() {
@@ -332,33 +324,29 @@ export class WorkflowTriggerWorkspaceService {
     assertWorkflowVersionTriggerIsDefined(workflowVersion);
 
     switch (workflowVersion.trigger.type) {
-      case WorkflowTriggerType.DATABASE_EVENT:
-        await this.databaseEventTriggerService.createEventListener(
-          workflowVersion.workflowId,
-          workflowVersion.trigger,
-          manager,
-        );
-
-        return;
       case WorkflowTriggerType.MANUAL:
       case WorkflowTriggerType.WEBHOOK:
         return;
+      case WorkflowTriggerType.DATABASE_EVENT: {
+        const eventName = workflowVersion.trigger.settings.eventName;
+
+        await this.automatedTriggerWorkspaceService.addAutomatedTrigger({
+          workflowId: workflowVersion.workflowId,
+          type: AutomatedTriggerType.DATABASE_EVENT,
+          settings: { eventName },
+          manager,
+        });
+
+        return;
+      }
       case WorkflowTriggerType.CRON: {
         const pattern = computeCronPatternFromSchedule(workflowVersion.trigger);
 
-        await this.messageQueueService.addCron<WorkflowTriggerJobData>({
-          jobName: WorkflowTriggerJob.name,
-          jobId: workflowVersion.workflowId,
-          data: {
-            workspaceId: this.getWorkspaceId(),
-            workflowId: workflowVersion.workflowId,
-            payload: {},
-          },
-          options: {
-            repeat: {
-              pattern,
-            },
-          },
+        await this.automatedTriggerWorkspaceService.addAutomatedTrigger({
+          workflowId: workflowVersion.workflowId,
+          type: AutomatedTriggerType.CRON,
+          settings: { pattern },
+          manager,
         });
 
         return;
@@ -377,21 +365,15 @@ export class WorkflowTriggerWorkspaceService {
 
     switch (workflowVersion.trigger.type) {
       case WorkflowTriggerType.DATABASE_EVENT:
-        await this.databaseEventTriggerService.deleteEventListener(
-          workflowVersion.workflowId,
+      case WorkflowTriggerType.CRON:
+        await this.automatedTriggerWorkspaceService.deleteAutomatedTrigger({
+          workflowId: workflowVersion.workflowId,
           manager,
-        );
+        });
 
         return;
       case WorkflowTriggerType.MANUAL:
       case WorkflowTriggerType.WEBHOOK:
-        return;
-      case WorkflowTriggerType.CRON:
-        await this.messageQueueService.removeCron({
-          jobName: WorkflowTriggerJob.name,
-          jobId: workflowVersion.workflowId,
-        });
-
         return;
       default:
         assertNever(workflowVersion.trigger);