Add cron trigger table (#14110)

This Pr begins the extensibility journey
- adds a `core.cronTrigger` table
- add a oneToMany relation between core.serverlessFunction and
`core.cronTrigger` (one serverlessFunction can be triggered by multiple
cronTriggers)
- add a job to trigger a serverless function
- adds a cron to trigger serverlessFunction (via the trigger job) based
on the core.cronTrigger.setting.pattern
- adds a command to register the cron
- add the command in `cron-register-all.command.ts`

packages/twenty-server/src/engine/metadata-modules/serverless-function/jobs/serverless-function-trigger.job.ts

@@ -0,0 +1,32 @@
+import { Scope } from '@nestjs/common';
+
+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 { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator';
+import { ServerlessFunctionService } from 'src/engine/metadata-modules/serverless-function/serverless-function.service';
+
+export type ServerlessFunctionTriggerJobData = {
+  serverlessFunctionId: string;
+  workspaceId: string;
+  payload?: object;
+};
+
+@Processor({
+  queueName: MessageQueue.serverlessFunctionQueue,
+  scope: Scope.REQUEST,
+})
+export class ServerlessFunctionTriggerJob {
+  constructor(
+    private readonly serverlessFunctionService: ServerlessFunctionService,
+  ) {}
+
+  @Process(ServerlessFunctionTriggerJob.name)
+  async handle(data: ServerlessFunctionTriggerJobData) {
+    await this.serverlessFunctionService.executeOneServerlessFunction(
+      data.serverlessFunctionId,
+      data.workspaceId,
+      data.payload || {},
+      'draft',
+    );
+  }
+}

packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.entity.ts

@@ -7,9 +7,11 @@ import {
   Index,
   PrimaryGeneratedColumn,
   UpdateDateColumn,
+  OneToMany,
 } from 'typeorm';
 
 import { InputSchema } from 'src/modules/workflow/workflow-builder/workflow-schema/types/input-schema.type';
+import { CronTrigger } from 'src/engine/metadata-modules/trigger/entities/cron-trigger.entity';
 
 const DEFAULT_SERVERLESS_TIMEOUT_SECONDS = 300; // 5 minutes
 
@@ -52,6 +54,15 @@ export class ServerlessFunctionEntity {
   @Column({ nullable: false, type: 'uuid' })
   workspaceId: string;
 
+  @OneToMany(
+    () => CronTrigger,
+    (cronTrigger) => cronTrigger.serverlessFunction,
+    {
+      cascade: true,
+    },
+  )
+  cronTriggers: CronTrigger[];
+
   @CreateDateColumn({ type: 'timestamptz' })
   createdAt: Date;
 

packages/twenty-server/src/engine/metadata-modules/serverless-function/serverless-function.module.ts

@@ -12,6 +12,7 @@ import { ThrottlerModule } from 'src/engine/core-modules/throttler/throttler.mod
 import { ServerlessFunctionEntity } from 'src/engine/metadata-modules/serverless-function/serverless-function.entity';
 import { ServerlessFunctionResolver } from 'src/engine/metadata-modules/serverless-function/serverless-function.resolver';
 import { ServerlessFunctionService } from 'src/engine/metadata-modules/serverless-function/serverless-function.service';
+import { ServerlessFunctionTriggerJob } from 'src/engine/metadata-modules/serverless-function/jobs/serverless-function-trigger.job';
 
 @Module({
   imports: [
@@ -23,7 +24,11 @@ import { ServerlessFunctionService } from 'src/engine/metadata-modules/serverles
     AuditModule,
     FeatureFlagModule,
   ],
-  providers: [ServerlessFunctionService, ServerlessFunctionResolver],
+  providers: [
+    ServerlessFunctionService,
+    ServerlessFunctionTriggerJob,
+    ServerlessFunctionResolver,
+  ],
   exports: [ServerlessFunctionService],
 })
 export class ServerlessFunctionModule {}

packages/twenty-server/src/engine/metadata-modules/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/engine/metadata-modules/trigger/crons/jobs/cron-trigger.cron.job';
+@Command({
+  name: 'cron:trigger:start-cron-trigger',
+  description:
+    'Starts a cron job to trigger cron triggered serverless functions',
+})
+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/engine/metadata-modules/trigger/crons/jobs/cron-trigger.cron.job.ts

@@ -0,0 +1,78 @@
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { isDefined } from 'twenty-shared/utils';
+import { WorkspaceActivationStatus } from 'twenty-shared/workspace';
+import { Repository } from 'typeorm';
+
+import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator';
+import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator';
+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 { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service';
+import { CronTrigger } from 'src/engine/metadata-modules/trigger/entities/cron-trigger.entity';
+import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
+import {
+  ServerlessFunctionTriggerJob,
+  ServerlessFunctionTriggerJobData,
+} from 'src/engine/metadata-modules/serverless-function/jobs/serverless-function-trigger.job';
+import { shouldRunNow } from 'src/utils/should-run-now.utils';
+
+export const CRON_TRIGGER_CRON_PATTERN = '* * * * *';
+
+@Processor(MessageQueue.cronQueue)
+export class CronTriggerCronJob {
+  constructor(
+    @InjectMessageQueue(MessageQueue.serverlessFunctionQueue)
+    private readonly messageQueueService: MessageQueueService,
+    @InjectRepository(Workspace)
+    private readonly workspaceRepository: Repository<Workspace>,
+    @InjectRepository(CronTrigger)
+    private readonly cronTriggerRepository: Repository<CronTrigger>,
+  ) {}
+
+  @Process(CronTriggerCronJob.name)
+  @SentryCronMonitor(CronTriggerCronJob.name, CRON_TRIGGER_CRON_PATTERN)
+  async handle() {
+    const activeWorkspaces = await this.workspaceRepository.find({
+      where: {
+        activationStatus: WorkspaceActivationStatus.ACTIVE,
+      },
+      select: ['id'],
+    });
+
+    const now = new Date();
+
+    for (const activeWorkspace of activeWorkspaces) {
+      const cronTriggers = await this.cronTriggerRepository.find({
+        where: {
+          workspaceId: activeWorkspace.id,
+        },
+        select: ['settings'],
+        relations: ['serverlessFunction'],
+      });
+
+      for (const cronTrigger of cronTriggers) {
+        const settings = cronTrigger.settings;
+
+        if (!isDefined(settings.pattern)) {
+          continue;
+        }
+
+        if (!shouldRunNow(settings.pattern, now)) {
+          continue;
+        }
+
+        await this.messageQueueService.add<ServerlessFunctionTriggerJobData>(
+          ServerlessFunctionTriggerJob.name,
+          {
+            serverlessFunctionId: cronTrigger.serverlessFunction.id,
+            workspaceId: activeWorkspace.id,
+            payload: {},
+          },
+          { retryLimit: 3 },
+        );
+      }
+    }
+  }
+}

packages/twenty-server/src/engine/metadata-modules/trigger/entities/cron-trigger.entity.ts

@@ -0,0 +1,48 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  DeleteDateColumn,
+  JoinColumn,
+  ManyToOne,
+  PrimaryGeneratedColumn,
+  UpdateDateColumn,
+  Relation,
+  Index,
+} from 'typeorm';
+
+import { ServerlessFunctionEntity } from 'src/engine/metadata-modules/serverless-function/serverless-function.entity';
+
+export type CronTriggerSettings = {
+  pattern: string;
+};
+
+@Entity({ name: 'cronTrigger', schema: 'core' })
+@Index('IDX_CRON_TRIGGER_WORKSPACE_ID', ['workspaceId'])
+export class CronTrigger {
+  @PrimaryGeneratedColumn('uuid')
+  id: string;
+
+  @Column({ nullable: false, type: 'jsonb' })
+  settings: CronTriggerSettings;
+
+  @ManyToOne(
+    () => ServerlessFunctionEntity,
+    (serverlessFunction) => serverlessFunction.cronTriggers,
+    { onDelete: 'CASCADE' },
+  )
+  @JoinColumn({ name: 'serverlessFunctionId' })
+  serverlessFunction: Relation<ServerlessFunctionEntity>;
+
+  @Column({ nullable: false, type: 'uuid' })
+  workspaceId: string;
+
+  @CreateDateColumn({ type: 'timestamptz' })
+  createdAt: Date;
+
+  @UpdateDateColumn({ type: 'timestamptz' })
+  updatedAt: Date;
+
+  @DeleteDateColumn({ type: 'timestamptz' })
+  deletedAt?: Date;
+}

packages/twenty-server/src/engine/metadata-modules/trigger/trigger.module.ts

@@ -0,0 +1,14 @@
+import { Module } from '@nestjs/common';
+import { TypeOrmModule } from '@nestjs/typeorm';
+
+import { CronTriggerCronCommand } from 'src/engine/metadata-modules/trigger/crons/commands/cron-trigger.cron.command';
+import { CronTriggerCronJob } from 'src/engine/metadata-modules/trigger/crons/jobs/cron-trigger.cron.job';
+import { Workspace } from 'src/engine/core-modules/workspace/workspace.entity';
+import { CronTrigger } from 'src/engine/metadata-modules/trigger/entities/cron-trigger.entity';
+
+@Module({
+  imports: [TypeOrmModule.forFeature([Workspace, CronTrigger])],
+  providers: [CronTriggerCronJob, CronTriggerCronCommand],
+  exports: [CronTriggerCronCommand],
+})
+export class TriggerModule {}