feat: add create or update workflow trigger (#14708)

### Description

This PR focuses on
https://github.com/twentyhq/core-team-issues/issues/1476

- introduces the handling of upserted events in the workflow trigger
system


## Visual Appearance
<img width="1792" height="1039" alt="Screenshot 2025-09-21 at 11 40
28 PM"
src="https://github.com/user-attachments/assets/d926b566-f7d3-412b-b1ab-ed2064425f3d"
/>

<img width="501" height="1025" alt="Screenshot 2025-09-21 at 11 40
39 PM"
src="https://github.com/user-attachments/assets/13c3031a-0762-4c80-8df1-fa74ca70a540"
/>


### Changes
- Added `UPSERTED` action to the `DatabaseEventAction` enum.
- Updated the `WorkflowEditTriggerDatabaseEventForm` to support upserted
events.

---------

Co-authored-by: Thomas Trompette <thomas.trompette@sfr.fr>

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/utils/__tests__/generate-fake-object-record-event.spec.ts

@@ -173,6 +173,37 @@ describe('generateFakeObjectRecordEvent', () => {
     });
   });
 
+  it('should generate record with "after" prefix for UPSERTED action', () => {
+    const result = generateFakeObjectRecordEvent(
+      objectMetadataInfo,
+      DatabaseEventAction.UPSERTED,
+    );
+
+    expect(result).toEqual({
+      object: {
+        isLeaf: true,
+        icon: 'test-company-icon',
+        label: 'Company',
+        value: 'A company',
+        fieldIdName: 'properties.after.id',
+        objectMetadataId: '20202020-c03c-45d6-a4b0-04afe1357c5c',
+      },
+      fields: {
+        'properties.after.field1': {
+          type: 'TEXT',
+          value: 'test',
+          fieldMetadataId: '123e4567-e89b-12d3-a456-426614174000',
+        },
+        'properties.after.field2': {
+          type: 'NUMBER',
+          value: 123,
+          fieldMetadataId: '123e4567-e89b-12d3-a456-426614174001',
+        },
+      },
+      _outputSchemaType: 'RECORD',
+    });
+  });
+
   it('should throw error for unknown action', () => {
     expect(() => {
       generateFakeObjectRecordEvent(

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/utils/generate-fake-object-record-event.ts

@@ -45,6 +45,7 @@ export const generateFakeObjectRecordEvent = (
   switch (action) {
     case DatabaseEventAction.CREATED:
     case DatabaseEventAction.UPDATED:
+    case DatabaseEventAction.UPSERTED:
       return generateFakeObjectRecordEventWithPrefix({
         objectMetadataInfo,
         prefix: 'properties.after',

packages/twenty-server/src/modules/workflow/workflow-executor/workflow-actions/record-crud/create-record.workflow-action.ts

@@ -1,17 +1,14 @@
 import { Injectable } from '@nestjs/common';
-import { InjectRepository } from '@nestjs/typeorm';
 
 import { isDefined } from 'class-validator';
 import { resolveInput } from 'twenty-shared/utils';
 import { canObjectBeManagedByWorkflow } from 'twenty-shared/workflow';
-import { Repository } from 'typeorm';
 
 import { type WorkflowAction } from 'src/modules/workflow/workflow-executor/interfaces/workflow-action.interface';
 
 import { RecordPositionService } from 'src/engine/core-modules/record-position/services/record-position.service';
 import { RecordInputTransformerService } from 'src/engine/core-modules/record-transformer/services/record-input-transformer.service';
 import { FieldActorSource } 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';
 import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
 import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
@@ -28,8 +25,6 @@ import { type WorkflowCreateRecordActionInput } from 'src/modules/workflow/workf
 export class CreateRecordWorkflowAction implements WorkflowAction {
   constructor(
     private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
-    @InjectRepository(ObjectMetadataEntity)
-    private readonly objectMetadataRepository: Repository<ObjectMetadataEntity>,
     private readonly scopedWorkspaceContextFactory: ScopedWorkspaceContextFactory,
     private readonly recordPositionService: RecordPositionService,
     private readonly recordInputTransformerService: RecordInputTransformerService,

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

@@ -4,12 +4,17 @@ export type BaseDatabaseEventTriggerSettings = {
 
 export type DatabaseEventTriggerSettings =
   | BaseDatabaseEventTriggerSettings
-  | UpdateEventTriggerSettings;
+  | UpdateEventTriggerSettings
+  | UpsertEventTriggerSettings;
 
 export type UpdateEventTriggerSettings = BaseDatabaseEventTriggerSettings & {
   fields: string[];
 };
 
+export type UpsertEventTriggerSettings = BaseDatabaseEventTriggerSettings & {
+  fields: string[];
+};
+
 export type CronTriggerSettings = {
   pattern: string;
 };

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

@@ -12,6 +12,7 @@ import { type ObjectRecordDeleteEvent } from 'src/engine/core-modules/event-emit
 import { type ObjectRecordDestroyEvent } from 'src/engine/core-modules/event-emitter/types/object-record-destroy.event';
 import { type ObjectRecordNonDestructiveEvent } from 'src/engine/core-modules/event-emitter/types/object-record-non-destructive-event';
 import { type ObjectRecordUpdateEvent } from 'src/engine/core-modules/event-emitter/types/object-record-update.event';
+import { type ObjectRecordUpsertEvent } from 'src/engine/core-modules/event-emitter/types/object-record-upsert.event';
 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';
@@ -22,7 +23,10 @@ import {
   type WorkflowAutomatedTriggerWorkspaceEntity,
 } 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 { type UpdateEventTriggerSettings } from 'src/modules/workflow/workflow-trigger/automated-trigger/constants/automated-trigger-settings';
+import {
+  type UpdateEventTriggerSettings,
+  type UpsertEventTriggerSettings,
+} from 'src/modules/workflow/workflow-trigger/automated-trigger/constants/automated-trigger-settings';
 import {
   WorkflowTriggerJob,
   type WorkflowTriggerJobData,
@@ -110,6 +114,22 @@ export class WorkflowDatabaseEventTriggerListener {
     });
   }
 
+  @OnDatabaseBatchEvent('*', DatabaseEventAction.UPSERTED)
+  async handleObjectRecordUpsertEvent(
+    payload: WorkspaceEventBatch<ObjectRecordUpsertEvent>,
+  ) {
+    if (await this.shouldIgnoreEvent(payload)) {
+      return;
+    }
+
+    const clonedPayload = structuredClone(payload);
+
+    await this.handleEvent({
+      payload: clonedPayload,
+      action: DatabaseEventAction.UPSERTED,
+    });
+  }
+
   private async enrichCreatedEvent(
     payload: WorkspaceEventBatch<ObjectRecordCreateEvent>,
   ) {
@@ -316,6 +336,19 @@ export class WorkflowDatabaseEventTriggerListener {
       );
     }
 
+    if (action === DatabaseEventAction.UPSERTED) {
+      const settings = eventListener.settings as UpsertEventTriggerSettings;
+      const upsertEventPayload = eventPayload as ObjectRecordUpsertEvent;
+
+      return (
+        !settings.fields ||
+        settings.fields.length === 0 ||
+        settings.fields.some((field) =>
+          upsertEventPayload?.properties?.updatedFields?.includes(field),
+        )
+      );
+    }
+
     return true;
   }
 }