Allow bulk records for manual trigger (#14725)

https://github.com/user-attachments/assets/d6c565eb-9a29-4830-9396-5f979c8caa7b

- Added a new component for manual trigger (mostly duplicated from
previous one). Will remove the old one once all data are migrated
- Updated schema output so the current item of the iterator can be typed

Todo left:
- migrate old triggers
- add an util to search iterator output. Today current item fields will
be displayed as not found
- set new manual triggers for workflow runs

packages/twenty-server/src/modules/workflow/workflow-builder/utils/__tests__/compute-workflow-version-step-changes.util.spec.ts

@@ -13,7 +13,6 @@ describe('computeWorkflowVersionStepChanges', () => {
     name: 'Test Manual Trigger',
     type: WorkflowTriggerType.MANUAL,
     settings: {
-      input: {},
       outputSchema: {},
     },
     nextStepIds: ['step-1'],

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/constants/default-iterator-current-item.const.ts

@@ -0,0 +1,8 @@
+import { type Leaf } from 'src/modules/workflow/workflow-builder/workflow-schema/types/output-schema.type';
+
+export const DEFAULT_ITERATOR_CURRENT_ITEM: Leaf = {
+  label: 'Current Item',
+  isLeaf: true,
+  type: 'unknown',
+  value: null,
+};

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/utils/__tests__/generate-fake-array-item.spec.ts

@@ -0,0 +1,38 @@
+import { DEFAULT_ITERATOR_CURRENT_ITEM } from 'src/modules/workflow/workflow-builder/workflow-schema/constants/default-iterator-current-item.const';
+import { generateFakeArrayItem } from 'src/modules/workflow/workflow-builder/workflow-schema/utils/generate-fake-array-item';
+
+describe('generateFakeArrayItem', () => {
+  it('should return default iterator when input cannot be parsed', () => {
+    const result = generateFakeArrayItem({ items: 'invalid json' });
+
+    expect(result).toEqual(DEFAULT_ITERATOR_CURRENT_ITEM);
+  });
+
+  it('should handle string array input', () => {
+    const result = generateFakeArrayItem({ items: '["test1", "test2"]' });
+
+    expect(result).toEqual({
+      label: 'Current Item',
+      isLeaf: true,
+      type: 'string',
+      value: expect.any(String),
+    });
+  });
+
+  it('should handle regular array input', () => {
+    const result = generateFakeArrayItem({ items: [1, 2, 3] });
+
+    expect(result).toEqual({
+      label: 'Current Item',
+      isLeaf: true,
+      type: 'number',
+      value: expect.any(Number),
+    });
+  });
+
+  it('should return default iterator when input is parsed but not an array', () => {
+    const result = generateFakeArrayItem({ items: '{"key": "value"}' });
+
+    expect(result).toEqual(DEFAULT_ITERATOR_CURRENT_ITEM);
+  });
+});

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/utils/generate-fake-array-item.spec.ts

@@ -0,0 +1,39 @@
+import { DEFAULT_ITERATOR_CURRENT_ITEM } from 'src/modules/workflow/workflow-builder/workflow-schema/constants/default-iterator-current-item.const';
+
+import { generateFakeArrayItem } from './generate-fake-array-item';
+
+describe('generateFakeArrayItem', () => {
+  it('should return default iterator when input cannot be parsed', () => {
+    const result = generateFakeArrayItem({ items: 'invalid json' });
+
+    expect(result).toEqual(DEFAULT_ITERATOR_CURRENT_ITEM);
+  });
+
+  it('should handle string array input', () => {
+    const result = generateFakeArrayItem({ items: '["test1", "test2"]' });
+
+    expect(result).toEqual({
+      label: 'Current Item',
+      isLeaf: true,
+      type: 'string',
+      value: expect.any(String),
+    });
+  });
+
+  it('should handle regular array input', () => {
+    const result = generateFakeArrayItem({ items: [1, 2, 3] });
+
+    expect(result).toEqual({
+      label: 'Current Item',
+      isLeaf: true,
+      type: 'number',
+      value: expect.any(Number),
+    });
+  });
+
+  it('should return default iterator when input is parsed but not an array', () => {
+    const result = generateFakeArrayItem({ items: '{"key": "value"}' });
+
+    expect(result).toEqual(DEFAULT_ITERATOR_CURRENT_ITEM);
+  });
+});

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/utils/generate-fake-array-item.ts

@@ -0,0 +1,34 @@
+import { isArray } from '@sniptt/guards';
+import { isString } from 'class-validator';
+
+import { generateFakeValue } from 'src/engine/utils/generate-fake-value';
+import { DEFAULT_ITERATOR_CURRENT_ITEM } from 'src/modules/workflow/workflow-builder/workflow-schema/constants/default-iterator-current-item.const';
+import { type InputSchemaPropertyType } from 'src/modules/workflow/workflow-builder/workflow-schema/types/input-schema.type';
+import { type Leaf } from 'src/modules/workflow/workflow-builder/workflow-schema/types/output-schema.type';
+
+export const generateFakeArrayItem = ({
+  items,
+}: {
+  items: unknown[] | string;
+}): Leaf => {
+  let parsedItems: unknown[] | string;
+
+  try {
+    parsedItems = isString(items) ? JSON.parse(items) : items;
+  } catch {
+    return DEFAULT_ITERATOR_CURRENT_ITEM;
+  }
+
+  if (isArray(parsedItems) && parsedItems.length > 0) {
+    const type = typeof parsedItems[0];
+
+    return {
+      label: 'Current Item',
+      isLeaf: true,
+      type: type as InputSchemaPropertyType,
+      value: generateFakeValue(type),
+    };
+  }
+
+  return DEFAULT_ITERATOR_CURRENT_ITEM;
+};

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-schema/workflow-schema.workspace-service.ts

@@ -1,10 +1,26 @@
 import { Injectable } from '@nestjs/common';
 
+import { isString } from '@sniptt/guards';
+import { isDefined, isValidVariable } from 'twenty-shared/utils';
+import {
+  BulkRecordsAvailability,
+  extractRawVariableNamePart,
+  GlobalAvailability,
+  SingleRecordAvailability,
+  TRIGGER_STEP_ID,
+} from 'twenty-shared/workflow';
+
 import { type DatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/enums/database-event-action';
 import { checkStringIsDatabaseEventAction } from 'src/engine/api/graphql/graphql-query-runner/utils/check-string-is-database-event-action';
 import { generateFakeValue } from 'src/engine/utils/generate-fake-value';
 import { WorkflowCommonWorkspaceService } from 'src/modules/workflow/common/workspace-services/workflow-common.workspace-service';
-import { type OutputSchema } from 'src/modules/workflow/workflow-builder/workflow-schema/types/output-schema.type';
+import { DEFAULT_ITERATOR_CURRENT_ITEM } from 'src/modules/workflow/workflow-builder/workflow-schema/constants/default-iterator-current-item.const';
+import {
+  Leaf,
+  Node,
+  type OutputSchema,
+} from 'src/modules/workflow/workflow-builder/workflow-schema/types/output-schema.type';
+import { generateFakeArrayItem } from 'src/modules/workflow/workflow-builder/workflow-schema/utils/generate-fake-array-item';
 import { generateFakeFormResponse } from 'src/modules/workflow/workflow-builder/workflow-schema/utils/generate-fake-form-response';
 import { generateFakeObjectRecord } from 'src/modules/workflow/workflow-builder/workflow-schema/utils/generate-fake-object-record';
 import { generateFakeObjectRecordEvent } from 'src/modules/workflow/workflow-builder/workflow-schema/utils/generate-fake-object-record-event';
@@ -14,7 +30,7 @@ import {
   WorkflowActionType,
 } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
 import {
-  type WorkflowTrigger,
+  WorkflowTrigger,
   WorkflowTriggerType,
 } from 'src/modules/workflow/workflow-trigger/types/workflow-trigger.type';
 
@@ -27,9 +43,11 @@ export class WorkflowSchemaWorkspaceService {
   async computeStepOutputSchema({
     step,
     workspaceId,
+    workflowVersionId,
   }: {
     step: WorkflowTrigger | WorkflowAction;
     workspaceId: string;
+    workflowVersionId: string;
   }): Promise<OutputSchema> {
     const stepType = step.type;
 
@@ -41,16 +59,24 @@ export class WorkflowSchemaWorkspaceService {
         });
       }
       case WorkflowTriggerType.MANUAL: {
-        const { objectType } = step.settings;
+        const { objectType, availability } = step.settings;
 
-        if (!objectType) {
-          return {};
+        // TODO: to be deprecated once all triggers are migrated to the new availability type
+        if (isDefined(objectType)) {
+          return this.computeRecordOutputSchema({
+            objectType,
+            workspaceId,
+          });
         }
 
-        return this.computeRecordOutputSchema({
-          objectType,
-          workspaceId,
-        });
+        if (isDefined(availability)) {
+          return this.computeTriggerOutputSchemaFromAvailability({
+            availability,
+            workspaceId,
+          });
+        }
+
+        return {};
       }
       case WorkflowTriggerType.WEBHOOK:
       case WorkflowTriggerType.CRON: {
@@ -77,13 +103,14 @@ export class WorkflowSchemaWorkspaceService {
           workspaceId,
         });
       case WorkflowActionType.ITERATOR: {
+        const items = step.settings.input.items;
+
         return {
-          currentItem: {
-            label: 'Current Item',
-            isLeaf: true,
-            type: 'unknown',
-            value: generateFakeValue('unknown'),
-          },
+          currentItem: await this.computeLoopCurrentItemOutputSchema({
+            items,
+            workflowVersionId,
+            workspaceId,
+          }),
           currentItemIndex: {
             label: 'Current Item Index',
             isLeaf: true,
@@ -107,9 +134,11 @@ export class WorkflowSchemaWorkspaceService {
   async enrichOutputSchema({
     step,
     workspaceId,
+    workflowVersionId,
   }: {
     step: WorkflowAction;
     workspaceId: string;
+    workflowVersionId: string;
   }): Promise<WorkflowAction> {
     // We don't enrich on the fly for code and HTTP request workflow actions.
     // For code actions, OutputSchema is computed and updated when testing the serverless function.
@@ -128,6 +157,7 @@ export class WorkflowSchemaWorkspaceService {
     const outputSchema = await this.computeStepOutputSchema({
       step,
       workspaceId,
+      workflowVersionId,
     });
 
     result.settings = {
@@ -231,4 +261,137 @@ export class WorkflowSchemaWorkspaceService {
       objectMetadataMaps,
     });
   }
+
+  private async computeTriggerOutputSchemaFromAvailability({
+    availability,
+    workspaceId,
+  }: {
+    availability:
+      | GlobalAvailability
+      | SingleRecordAvailability
+      | BulkRecordsAvailability;
+    workspaceId: string;
+  }): Promise<OutputSchema> {
+    if (availability.type === 'GLOBAL') {
+      return {};
+    }
+
+    if (availability.type === 'SINGLE_RECORD') {
+      return this.computeRecordOutputSchema({
+        objectType: availability.objectNameSingular,
+        workspaceId,
+      });
+    }
+
+    if (availability.type === 'BULK_RECORDS') {
+      const objectMetadataInfo =
+        await this.workflowCommonWorkspaceService.getObjectMetadataItemWithFieldsMaps(
+          availability.objectNameSingular,
+          workspaceId,
+        );
+
+      return {
+        [availability.objectNameSingular]: {
+          label:
+            objectMetadataInfo.objectMetadataItemWithFieldsMaps.labelPlural,
+          isLeaf: true,
+          type: 'array',
+          value:
+            'Array of ' +
+            objectMetadataInfo.objectMetadataItemWithFieldsMaps.labelPlural,
+        },
+      };
+    }
+
+    return {};
+  }
+
+  private async computeLoopCurrentItemOutputSchema({
+    items,
+    workflowVersionId,
+    workspaceId,
+  }: {
+    items: string | undefined | unknown[];
+    workflowVersionId: string;
+    workspaceId: string;
+  }): Promise<Leaf | Node> {
+    if (!isDefined(items)) {
+      return DEFAULT_ITERATOR_CURRENT_ITEM;
+    }
+
+    if (isString(items) && isValidVariable(items)) {
+      return this.computeIteratorCurrentItemFromVariable({
+        items,
+        workflowVersionId,
+        workspaceId,
+      });
+    }
+
+    return generateFakeArrayItem({ items });
+  }
+
+  private async computeIteratorCurrentItemFromVariable({
+    items,
+    workflowVersionId,
+    workspaceId,
+  }: {
+    items: string;
+    workflowVersionId: string;
+    workspaceId: string;
+  }): Promise<Leaf | Node> {
+    const workflowVersion =
+      await this.workflowCommonWorkspaceService.getWorkflowVersionOrFail({
+        workflowVersionId,
+        workspaceId,
+      });
+
+    const stepId = extractRawVariableNamePart({
+      rawVariableName: items,
+      part: 'stepId',
+    });
+
+    if (stepId === TRIGGER_STEP_ID) {
+      const trigger = workflowVersion.trigger;
+
+      if (!isDefined(trigger)) {
+        return DEFAULT_ITERATOR_CURRENT_ITEM;
+      }
+
+      switch (trigger.type) {
+        case WorkflowTriggerType.MANUAL: {
+          if (trigger.settings.availability?.type === 'BULK_RECORDS') {
+            const objectMetadataInfo =
+              await this.workflowCommonWorkspaceService.getObjectMetadataItemWithFieldsMaps(
+                trigger.settings.availability.objectNameSingular,
+                workspaceId,
+              );
+
+            return {
+              label:
+                'Current Item (' +
+                objectMetadataInfo.objectMetadataItemWithFieldsMaps
+                  .labelSingular +
+                ')',
+              isLeaf: false,
+              type: 'object',
+              value: await this.computeRecordOutputSchema({
+                objectType: trigger.settings.availability.objectNameSingular,
+                workspaceId,
+              }),
+            };
+          }
+
+          return DEFAULT_ITERATOR_CURRENT_ITEM;
+        }
+        // TODO(t.trompette): handle other trigger types
+        default: {
+          return DEFAULT_ITERATOR_CURRENT_ITEM;
+        }
+      }
+    }
+
+    // TODO(t.trompette): handle other step types
+
+    return DEFAULT_ITERATOR_CURRENT_ITEM;
+  }
 }

packages/twenty-server/src/modules/workflow/workflow-builder/workflow-version-step/workflow-version-step.workspace-service.ts

@@ -74,6 +74,7 @@ export class WorkflowVersionStepWorkspaceService {
       await this.workflowSchemaWorkspaceService.enrichOutputSchema({
         step: builtStep,
         workspaceId,
+        workflowVersionId,
       });
 
     const { updatedSteps, updatedTrigger } = insertStep({
@@ -156,6 +157,7 @@ export class WorkflowVersionStepWorkspaceService {
       : await this.updateWorkflowVersionStepSettings({
           newStep: step,
           workspaceId,
+          workflowVersionId,
         });
 
     const updatedSteps = workflowVersion.steps.map((existingStep) => {
@@ -378,19 +380,23 @@ export class WorkflowVersionStepWorkspaceService {
         position: existingStep.position,
       },
       workspaceId,
+      workflowVersionId,
     });
   }
 
   private async updateWorkflowVersionStepSettings({
     newStep,
     workspaceId,
+    workflowVersionId,
   }: {
     newStep: WorkflowAction;
     workspaceId: string;
+    workflowVersionId: string;
   }): Promise<WorkflowAction> {
     return this.workflowSchemaWorkspaceService.enrichOutputSchema({
       step: newStep,
       workspaceId,
+      workflowVersionId,
     });
   }
 }

packages/twenty-server/src/modules/workflow/workflow-tools/schemas/workflow-tool-schemas.ts

@@ -96,6 +96,7 @@ export const computeStepOutputSchemaSchema = z.object({
   step: z
     .union([workflowTriggerSchema, workflowActionSchema])
     .describe('The workflow step configuration'),
+  workflowVersionId: z.string().describe('The ID of the workflow version'),
 });
 
 export const createCompleteWorkflowSchema = z.object({

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

@@ -376,11 +376,13 @@ This is the most efficient way for AI to create workflows as it handles all the
       inputSchema: computeStepOutputSchemaSchema,
       execute: async (parameters: {
         step: WorkflowTrigger | WorkflowAction;
+        workflowVersionId: string;
       }) => {
         try {
           return await this.workflowSchemaService.computeStepOutputSchema({
             step: parameters.step,
             workspaceId,
+            workflowVersionId: parameters.workflowVersionId,
           });
         } catch (error) {
           return {

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

@@ -1,3 +1,9 @@
+import {
+  type BulkRecordsAvailability,
+  type GlobalAvailability,
+  type SingleRecordAvailability,
+} from 'twenty-shared/workflow';
+
 import { type OutputSchema } from 'src/modules/workflow/workflow-builder/workflow-schema/types/output-schema.type';
 
 export enum WorkflowTriggerType {
@@ -8,7 +14,6 @@ export enum WorkflowTriggerType {
 }
 
 type BaseWorkflowTriggerSettings = {
-  input?: object;
   outputSchema: OutputSchema;
 };
 
@@ -35,6 +40,10 @@ export type WorkflowManualTrigger = BaseTrigger & {
   settings: BaseWorkflowTriggerSettings & {
     objectType?: string;
     icon?: string;
+    availability?:
+      | GlobalAvailability
+      | SingleRecordAvailability
+      | BulkRecordsAvailability;
   };
 };