14064 extensibility add coretriggereventlistener table (#14138)

This Pr continues the extensibility journey
- adds a `core.databaseEventTrigger` table
- add a oneToMany relation between `core.serverlessFunction` and
`core.databaseEventTrigger`
- add a job `CallDatabaseEventTriggerJobsJob` triggered by
`EntityEventsToDbListener` triggering `serverlessFunction` based on the
`core.databaseEventTrigger.settings.eventName`
- add a new `trigger-queue` to carry this job
- renamed `DatabaseEventTriggerListener` into
`WorkflowDatabaseEventTriggerListener`

packages/twenty-server/src/database/typeorm/core/migrations/common/1756385539610-addDatabaseEventTriggerEntity.ts

@@ -0,0 +1,29 @@
+import { type MigrationInterface, type QueryRunner } from 'typeorm';
+
+export class AddDatabaseEventTriggerEntity1756385539610
+  implements MigrationInterface
+{
+  name = 'AddDatabaseEventTriggerEntity1756385539610';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `CREATE TABLE "core"."databaseEventTrigger" ("id" uuid NOT NULL DEFAULT uuid_generate_v4(), "settings" jsonb NOT NULL, "workspaceId" uuid NOT NULL, "createdAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), "updatedAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), "deletedAt" TIMESTAMP WITH TIME ZONE, "serverlessFunctionId" uuid, CONSTRAINT "PK_b3e1ceba9d36f8b5aac6342a267" PRIMARY KEY ("id"))`,
+    );
+    await queryRunner.query(
+      `CREATE INDEX "IDX_DATABASE_EVENT_TRIGGER_WORKSPACE_ID" ON "core"."databaseEventTrigger" ("workspaceId") `,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "core"."databaseEventTrigger" ADD CONSTRAINT "FK_fe492f1ecc81621b6d7cee1775b" FOREIGN KEY ("serverlessFunctionId") REFERENCES "core"."serverlessFunction"("id") ON DELETE CASCADE ON UPDATE NO ACTION`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "core"."databaseEventTrigger" DROP CONSTRAINT "FK_fe492f1ecc81621b6d7cee1775b"`,
+    );
+    await queryRunner.query(
+      `DROP INDEX "core"."IDX_DATABASE_EVENT_TRIGGER_WORKSPACE_ID"`,
+    );
+    await queryRunner.query(`DROP TABLE "core"."databaseEventTrigger"`);
+  }
+}

packages/twenty-server/src/database/typeorm/core/migrations/common/1756447211845-addUniqueIdentifierToTriggerEntities.ts

@@ -0,0 +1,49 @@
+import { type MigrationInterface, type QueryRunner } from 'typeorm';
+
+export class AddUniqueIdentifierToTriggerEntities1756447211845
+  implements MigrationInterface
+{
+  name = 'AddUniqueIdentifierToTriggerEntities1756447211845';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "core"."databaseEventTrigger" DROP CONSTRAINT "FK_fe492f1ecc81621b6d7cee1775b"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "core"."cronTrigger" ADD "universalIdentifier" uuid`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "core"."databaseEventTrigger" ADD "universalIdentifier" uuid`,
+    );
+    await queryRunner.query(
+      `CREATE UNIQUE INDEX "IDX_8adc1fd6cb0dad2fbfd945954d" ON "core"."cronTrigger" ("workspaceId", "universalIdentifier") `,
+    );
+    await queryRunner.query(
+      `CREATE UNIQUE INDEX "IDX_960465af116edf9ac501bfb3db" ON "core"."databaseEventTrigger" ("workspaceId", "universalIdentifier") `,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "core"."databaseEventTrigger" ADD CONSTRAINT "FK_7650f1b8b693cde204f44ab0aa4" FOREIGN KEY ("serverlessFunctionId") REFERENCES "core"."serverlessFunction"("id") ON DELETE CASCADE ON UPDATE NO ACTION`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "core"."databaseEventTrigger" DROP CONSTRAINT "FK_7650f1b8b693cde204f44ab0aa4"`,
+    );
+    await queryRunner.query(
+      `DROP INDEX "core"."IDX_960465af116edf9ac501bfb3db"`,
+    );
+    await queryRunner.query(
+      `DROP INDEX "core"."IDX_8adc1fd6cb0dad2fbfd945954d"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "core"."databaseEventTrigger" DROP COLUMN "universalIdentifier"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "core"."cronTrigger" DROP COLUMN "universalIdentifier"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "core"."databaseEventTrigger" ADD CONSTRAINT "FK_fe492f1ecc81621b6d7cee1775b" FOREIGN KEY ("serverlessFunctionId") REFERENCES "core"."serverlessFunction"("id") ON DELETE CASCADE ON UPDATE NO ACTION`,
+    );
+  }
+}

packages/twenty-server/src/engine/api/graphql/workspace-query-runner/listeners/entity-events-to-db.listener.ts

@@ -18,6 +18,7 @@ import { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/wo
 import { UpsertTimelineActivityFromInternalEvent } from 'src/modules/timeline/jobs/upsert-timeline-activity-from-internal-event.job';
 import { CallWebhookJobsJob } from 'src/engine/core-modules/webhook/jobs/call-webhook-jobs.job';
 import { type ObjectRecordEventForWebhook } from 'src/engine/core-modules/webhook/types/object-record-event-for-webhook.type';
+import { CallDatabaseEventTriggerJobsJob } from 'src/engine/metadata-modules/trigger/jobs/call-database-event-trigger-jobs.job';
 
 @Injectable()
 export class EntityEventsToDbListener {
@@ -28,6 +29,8 @@ export class EntityEventsToDbListener {
     private readonly webhookQueueService: MessageQueueService,
     @InjectMessageQueue(MessageQueue.subscriptionsQueue)
     private readonly subscriptionsQueueService: MessageQueueService,
+    @InjectMessageQueue(MessageQueue.triggerQueue)
+    private readonly triggerQueueService: MessageQueueService,
   ) {}
 
   @OnDatabaseBatchEvent('*', DatabaseEventAction.CREATED)
@@ -82,6 +85,11 @@ export class EntityEventsToDbListener {
         batchEvent,
         { retryLimit: 3 },
       ),
+      this.triggerQueueService.add<WorkspaceEventBatch<T>>(
+        CallDatabaseEventTriggerJobsJob.name,
+        batchEvent,
+        { retryLimit: 3 },
+      ),
       this.webhookQueueService.add<
         WorkspaceEventBatch<ObjectRecordEventForWebhook>
       >(

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

@@ -17,4 +17,5 @@ export enum MessageQueue {
   deleteCascadeQueue = 'delete-cascade-queue',
   subscriptionsQueue = 'subscriptions-queue',
   serverlessFunctionQueue = 'serverless-function-queue',
+  triggerQueue = 'trigger-queue',
 }

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

@@ -12,6 +12,7 @@ import {
 
 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';
+import { DatabaseEventTrigger } from 'src/engine/metadata-modules/trigger/entities/database-event-trigger.entity';
 
 const DEFAULT_SERVERLESS_TIMEOUT_SECONDS = 300; // 5 minutes
 
@@ -63,6 +64,15 @@ export class ServerlessFunctionEntity {
   )
   cronTriggers: CronTrigger[];
 
+  @OneToMany(
+    () => DatabaseEventTrigger,
+    (databaseEventTrigger) => databaseEventTrigger.serverlessFunction,
+    {
+      cascade: true,
+    },
+  )
+  databaseEventTriggers: DatabaseEventTrigger[];
+
   @CreateDateColumn({ type: 'timestamptz' })
   createdAt: Date;
 

packages/twenty-server/src/engine/metadata-modules/trigger/crons/jobs/cron-trigger.cron.job.ts

@@ -48,7 +48,7 @@ export class CronTriggerCronJob {
         where: {
           workspaceId: activeWorkspace.id,
         },
-        select: ['settings'],
+        select: ['id', 'settings', 'workspaceId'],
         relations: ['serverlessFunction'],
       });
 
@@ -67,7 +67,7 @@ export class CronTriggerCronJob {
           ServerlessFunctionTriggerJob.name,
           {
             serverlessFunctionId: cronTrigger.serverlessFunction.id,
-            workspaceId: activeWorkspace.id,
+            workspaceId: cronTrigger.workspaceId,
             payload: {},
           },
           { retryLimit: 3 },

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

@@ -11,6 +11,8 @@ import {
   Index,
 } from 'typeorm';
 
+import { SyncableEntity } from 'src/engine/workspace-manager/workspace-sync/interfaces/syncable-entity.interface';
+
 import { ServerlessFunctionEntity } from 'src/engine/metadata-modules/serverless-function/serverless-function.entity';
 
 export type CronTriggerSettings = {
@@ -19,7 +21,7 @@ export type CronTriggerSettings = {
 
 @Entity({ name: 'cronTrigger', schema: 'core' })
 @Index('IDX_CRON_TRIGGER_WORKSPACE_ID', ['workspaceId'])
-export class CronTrigger {
+export class CronTrigger extends SyncableEntity {
   @PrimaryGeneratedColumn('uuid')
   id: string;
 

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

@@ -0,0 +1,50 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  DeleteDateColumn,
+  JoinColumn,
+  ManyToOne,
+  PrimaryGeneratedColumn,
+  UpdateDateColumn,
+  Relation,
+  Index,
+} from 'typeorm';
+
+import { SyncableEntity } from 'src/engine/workspace-manager/workspace-sync/interfaces/syncable-entity.interface';
+
+import { ServerlessFunctionEntity } from 'src/engine/metadata-modules/serverless-function/serverless-function.entity';
+
+export type DatabaseEventTriggerSettings = {
+  eventName: string;
+};
+
+@Entity({ name: 'databaseEventTrigger', schema: 'core' })
+@Index('IDX_DATABASE_EVENT_TRIGGER_WORKSPACE_ID', ['workspaceId'])
+export class DatabaseEventTrigger extends SyncableEntity {
+  @PrimaryGeneratedColumn('uuid')
+  id: string;
+
+  @Column({ nullable: false, type: 'jsonb' })
+  settings: DatabaseEventTriggerSettings;
+
+  @ManyToOne(
+    () => ServerlessFunctionEntity,
+    (serverlessFunction) => serverlessFunction.databaseEventTriggers,
+    { 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/jobs/call-database-event-trigger-jobs.job.ts

@@ -0,0 +1,80 @@
+import { InjectRepository } from '@nestjs/typeorm';
+
+import { Repository } from 'typeorm';
+
+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 { WorkspaceEventBatch } from 'src/engine/workspace-event-emitter/types/workspace-event.type';
+import type { ObjectRecordEvent } from 'src/engine/core-modules/event-emitter/types/object-record-event.event';
+import { DatabaseEventTrigger } from 'src/engine/metadata-modules/trigger/entities/database-event-trigger.entity';
+import {
+  ServerlessFunctionTriggerJob,
+  ServerlessFunctionTriggerJobData,
+} from 'src/engine/metadata-modules/serverless-function/jobs/serverless-function-trigger.job';
+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';
+
+@Processor(MessageQueue.triggerQueue)
+export class CallDatabaseEventTriggerJobsJob {
+  constructor(
+    @InjectMessageQueue(MessageQueue.serverlessFunctionQueue)
+    private readonly messageQueueService: MessageQueueService,
+    @InjectRepository(DatabaseEventTrigger)
+    private readonly databaseEventTriggerRepository: Repository<DatabaseEventTrigger>,
+  ) {}
+
+  @Process(CallDatabaseEventTriggerJobsJob.name)
+  async handle(workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>) {
+    const databaseEventListeners =
+      await this.databaseEventTriggerRepository.find({
+        where: {
+          workspaceId: workspaceEventBatch.workspaceId,
+        },
+        select: ['id', 'settings', 'workspaceId'],
+        relations: ['serverlessFunction'],
+      });
+
+    for (const databaseEventListener of databaseEventListeners) {
+      if (
+        !this.shouldTriggerJob({
+          workspaceEventBatch,
+          eventName: databaseEventListener.settings.eventName,
+        })
+      ) {
+        continue;
+      }
+
+      for (const eventData of workspaceEventBatch.events) {
+        await this.messageQueueService.add<ServerlessFunctionTriggerJobData>(
+          ServerlessFunctionTriggerJob.name,
+          {
+            serverlessFunctionId: databaseEventListener.serverlessFunction.id,
+            workspaceId: databaseEventListener.workspaceId,
+            payload: eventData,
+          },
+          { retryLimit: 3 },
+        );
+      }
+    }
+  }
+
+  private shouldTriggerJob({
+    workspaceEventBatch,
+    eventName,
+  }: {
+    workspaceEventBatch: WorkspaceEventBatch<ObjectRecordEvent>;
+    eventName: string;
+  }) {
+    const [nameSingular, operation] = workspaceEventBatch.name.split('.');
+
+    const validEventNames = [
+      `${nameSingular}.${operation}`,
+      `*.${operation}`,
+      `${nameSingular}.*`,
+      '*.*',
+    ];
+
+    return validEventNames.includes(eventName);
+  }
+}

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

@@ -5,10 +5,18 @@ import { CronTriggerCronCommand } from 'src/engine/metadata-modules/trigger/cron
 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';
+import { CallDatabaseEventTriggerJobsJob } from 'src/engine/metadata-modules/trigger/jobs/call-database-event-trigger-jobs.job';
+import { DatabaseEventTrigger } from 'src/engine/metadata-modules/trigger/entities/database-event-trigger.entity';
 
 @Module({
-  imports: [TypeOrmModule.forFeature([Workspace, CronTrigger])],
-  providers: [CronTriggerCronJob, CronTriggerCronCommand],
+  imports: [
+    TypeOrmModule.forFeature([Workspace, CronTrigger, DatabaseEventTrigger]),
+  ],
+  providers: [
+    CronTriggerCronJob,
+    CronTriggerCronCommand,
+    CallDatabaseEventTriggerJobsJob,
+  ],
   exports: [CronTriggerCronCommand],
 })
 export class TriggerModule {}

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

@@ -7,7 +7,7 @@ import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-commo
 import { AutomatedTriggerWorkspaceService } from 'src/modules/workflow/workflow-trigger/automated-trigger/automated-trigger.workspace-service';
 import { WorkflowCronTriggerCronCommand } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/commands/workflow-cron-trigger.cron.command';
 import { WorkflowCronTriggerCronJob } from 'src/modules/workflow/workflow-trigger/automated-trigger/crons/jobs/workflow-cron-trigger-cron.job';
-import { DatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener';
+import { WorkflowDatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trigger/automated-trigger/listeners/workflow-database-event-trigger.listener';
 
 @Module({
   imports: [
@@ -17,7 +17,7 @@ import { DatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trig
   ],
   providers: [
     AutomatedTriggerWorkspaceService,
-    DatabaseEventTriggerListener,
+    WorkflowDatabaseEventTriggerListener,
     WorkflowCronTriggerCronJob,
     WorkflowCronTriggerCronCommand,
   ],

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

@@ -4,13 +4,13 @@ import { MessageQueueService } from 'src/engine/core-modules/message-queue/servi
 import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
 import { AutomatedTriggerType } from 'src/modules/workflow/common/standard-objects/workflow-automated-trigger.workspace-entity';
 import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
-import { DatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trigger/automated-trigger/listeners/database-event-trigger.listener';
+import { WorkflowDatabaseEventTriggerListener } from 'src/modules/workflow/workflow-trigger/automated-trigger/listeners/workflow-database-event-trigger.listener';
 import { WorkflowTriggerJob } from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
 import { getMockObjectMetadataEntity } from 'src/utils/__test__/get-object-metadata-entity.mock';
 import { getMockObjectMetadataItemWithFieldsMaps } from 'src/utils/__test__/get-object-metadata-item-with-fields-maps.mock';
 
-describe('DatabaseEventTriggerListener', () => {
-  let listener: DatabaseEventTriggerListener;
+describe('WorkflowDatabaseEventTriggerListener', () => {
+  let listener: WorkflowDatabaseEventTriggerListener;
   let twentyORMGlobalManager: jest.Mocked<TwentyORMGlobalManager>;
   let messageQueueService: jest.Mocked<MessageQueueService>;
 
@@ -29,7 +29,7 @@ describe('DatabaseEventTriggerListener', () => {
 
     const module: TestingModule = await Test.createTestingModule({
       providers: [
-        DatabaseEventTriggerListener,
+        WorkflowDatabaseEventTriggerListener,
         {
           provide: TwentyORMGlobalManager,
           useValue: twentyORMGlobalManager,
@@ -83,8 +83,8 @@ describe('DatabaseEventTriggerListener', () => {
       ],
     }).compile();
 
-    listener = module.get<DatabaseEventTriggerListener>(
-      DatabaseEventTriggerListener,
+    listener = module.get<WorkflowDatabaseEventTriggerListener>(
+      WorkflowDatabaseEventTriggerListener,
     );
   });
 

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

@@ -29,8 +29,10 @@ import {
 } from 'src/modules/workflow/workflow-trigger/jobs/workflow-trigger.job';
 
 @Injectable()
-export class DatabaseEventTriggerListener {
-  private readonly logger = new Logger('DatabaseEventTriggerListener');
+export class WorkflowDatabaseEventTriggerListener {
+  private readonly logger = new Logger(
+    WorkflowDatabaseEventTriggerListener.name,
+  );
 
   constructor(
     private readonly twentyORMGlobalManager: TwentyORMGlobalManager,