feat: workflow delay action (Pause - Wait/Sleep/Delay) (#14915)
## Description - This PR focuses on issue https://github.com/orgs/twentyhq/projects/1/views/33?pane=issue&itemId=93150683&issue=twentyhq%7Ccore-team-issues%7C20 - added Workflow delay as a Flow action - for V1 added Type 1: Resume at a specific date or time ## Visual Appearance <img width="1792" height="1038" alt="Screenshot 2025-10-09 at 5 46 18 PM" src="https://github.com/user-attachments/assets/e62980e9-59c7-4e5a-b8ec-1e848a462d3f" /> <img width="1792" height="1037" alt="Screenshot 2025-10-09 at 5 46 35 PM" src="https://github.com/user-attachments/assets/7c3f4e39-ab0a-40ed-97a8-4f0cdb86f295" /> --------- Co-authored-by: Thomas Trompette <thomas.trompette@sfr.fr>
This commit is contained in:
@@ -23,6 +23,7 @@ import {
|
||||
type workflowTriggerSchema,
|
||||
type workflowUpdateRecordActionSchema,
|
||||
type workflowWebhookTriggerSchema,
|
||||
type workflowDelayActionSchema,
|
||||
} from 'twenty-shared/workflow';
|
||||
import { type z } from 'zod';
|
||||
|
||||
@@ -42,6 +43,7 @@ export type WorkflowDeleteRecordAction = z.infer<
|
||||
export type WorkflowFindRecordsAction = z.infer<
|
||||
typeof workflowFindRecordsActionSchema
|
||||
>;
|
||||
export type WorkflowDelayAction = z.infer<typeof workflowDelayActionSchema>;
|
||||
export type WorkflowFilterAction = z.infer<typeof workflowFilterActionSchema>;
|
||||
export type WorkflowFormAction = z.infer<typeof workflowFormActionSchema>;
|
||||
export type WorkflowHttpRequestAction = z.infer<
|
||||
@@ -65,6 +67,7 @@ export type WorkflowAction =
|
||||
| WorkflowHttpRequestAction
|
||||
| WorkflowAiAgentAction
|
||||
| WorkflowIteratorAction
|
||||
| WorkflowDelayAction
|
||||
| WorkflowEmptyAction;
|
||||
|
||||
export type WorkflowActionType = WorkflowAction['type'];
|
||||
|
||||
+5
-2
@@ -47,9 +47,12 @@ export const WorkflowDiagramStepNodeIcon = ({
|
||||
case 'AI_AGENT': {
|
||||
return <Icon size={theme.icon.size.md} color={theme.color.pink} />;
|
||||
}
|
||||
case 'EMPTY': {
|
||||
case 'EMPTY':
|
||||
return null;
|
||||
}
|
||||
case 'DELAY':
|
||||
case 'FILTER':
|
||||
case 'ITERATOR':
|
||||
return <Icon size={theme.icon.size.md} color={theme.color.green60} />;
|
||||
default: {
|
||||
return (
|
||||
<Icon
|
||||
|
||||
+12
@@ -11,6 +11,7 @@ import { WorkflowEditActionDeleteRecord } from '@/workflow/workflow-steps/workfl
|
||||
import { WorkflowEditActionEmpty } from '@/workflow/workflow-steps/workflow-actions/components/WorkflowEditActionEmpty';
|
||||
import { WorkflowEditActionSendEmail } from '@/workflow/workflow-steps/workflow-actions/components/WorkflowEditActionSendEmail';
|
||||
import { WorkflowEditActionUpdateRecord } from '@/workflow/workflow-steps/workflow-actions/components/WorkflowEditActionUpdateRecord';
|
||||
import { WorkflowEditActionDelay } from '@/workflow/workflow-steps/workflow-actions/delay-actions/components/WorkflowEditActionDelay';
|
||||
import { WorkflowEditActionFilter } from '@/workflow/workflow-steps/workflow-actions/filter-action/components/WorkflowEditActionFilter';
|
||||
import { WorkflowEditActionFindRecords } from '@/workflow/workflow-steps/workflow-actions/find-records-action/components/WorkflowEditActionFindRecords';
|
||||
import { WorkflowEditActionFormFiller } from '@/workflow/workflow-steps/workflow-actions/form-action/components/WorkflowEditActionFormFiller';
|
||||
@@ -237,6 +238,17 @@ export const WorkflowRunStepNodeDetail = ({
|
||||
/>
|
||||
);
|
||||
}
|
||||
case 'DELAY': {
|
||||
return (
|
||||
<WorkflowEditActionDelay
|
||||
key={stepId}
|
||||
action={stepDefinition.definition}
|
||||
actionOptions={{
|
||||
readonly: true,
|
||||
}}
|
||||
/>
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+10
@@ -10,6 +10,7 @@ import { WorkflowEditActionDeleteRecord } from '@/workflow/workflow-steps/workfl
|
||||
import { WorkflowEditActionEmpty } from '@/workflow/workflow-steps/workflow-actions/components/WorkflowEditActionEmpty';
|
||||
import { WorkflowEditActionSendEmail } from '@/workflow/workflow-steps/workflow-actions/components/WorkflowEditActionSendEmail';
|
||||
import { WorkflowEditActionUpdateRecord } from '@/workflow/workflow-steps/workflow-actions/components/WorkflowEditActionUpdateRecord';
|
||||
import { WorkflowEditActionDelay } from '@/workflow/workflow-steps/workflow-actions/delay-actions/components/WorkflowEditActionDelay';
|
||||
import { WorkflowEditActionFilter } from '@/workflow/workflow-steps/workflow-actions/filter-action/components/WorkflowEditActionFilter';
|
||||
import { WorkflowEditActionFindRecords } from '@/workflow/workflow-steps/workflow-actions/find-records-action/components/WorkflowEditActionFindRecords';
|
||||
import { WorkflowEditActionFormBuilder } from '@/workflow/workflow-steps/workflow-actions/form-action/components/WorkflowEditActionFormBuilder';
|
||||
@@ -210,6 +211,15 @@ export const WorkflowStepDetail = ({
|
||||
case 'EMPTY': {
|
||||
return <WorkflowEditActionEmpty key={stepId} actionOptions={props} />;
|
||||
}
|
||||
case 'DELAY': {
|
||||
return (
|
||||
<WorkflowEditActionDelay
|
||||
key={stepId}
|
||||
action={stepDefinition.definition}
|
||||
actionOptions={props}
|
||||
/>
|
||||
);
|
||||
}
|
||||
default:
|
||||
return assertUnreachable(
|
||||
stepDefinition.definition,
|
||||
|
||||
+6
-1
@@ -2,7 +2,7 @@ import { type WorkflowActionType } from '@/workflow/types/Workflow';
|
||||
|
||||
export const FLOW_ACTIONS: Array<{
|
||||
label: string;
|
||||
type: Extract<WorkflowActionType, 'ITERATOR' | 'FILTER'>;
|
||||
type: Extract<WorkflowActionType, 'ITERATOR' | 'FILTER' | 'DELAY'>;
|
||||
icon: string;
|
||||
}> = [
|
||||
{
|
||||
@@ -15,4 +15,9 @@ export const FLOW_ACTIONS: Array<{
|
||||
type: 'FILTER',
|
||||
icon: 'IconFilter',
|
||||
},
|
||||
{
|
||||
label: 'Delay',
|
||||
type: 'DELAY',
|
||||
icon: 'IconPlayerPause',
|
||||
},
|
||||
];
|
||||
|
||||
+225
@@ -0,0 +1,225 @@
|
||||
import { SidePanelHeader } from '@/command-menu/components/SidePanelHeader';
|
||||
import { FormDateTimeFieldInput } from '@/object-record/record-field/ui/form-types/components/FormDateTimeFieldInput';
|
||||
import { FormNumberFieldInput } from '@/object-record/record-field/ui/form-types/components/FormNumberFieldInput';
|
||||
import { Select } from '@/ui/input/components/Select';
|
||||
import { GenericDropdownContentWidth } from '@/ui/layout/dropdown/constants/GenericDropdownContentWidth';
|
||||
import { type WorkflowDelayAction } from '@/workflow/types/Workflow';
|
||||
import { WorkflowActionFooter } from '@/workflow/workflow-steps/components/WorkflowActionFooter';
|
||||
import { WorkflowStepBody } from '@/workflow/workflow-steps/components/WorkflowStepBody';
|
||||
import { useWorkflowActionHeader } from '@/workflow/workflow-steps/workflow-actions/hooks/useWorkflowActionHeader';
|
||||
import { WorkflowVariablePicker } from '@/workflow/workflow-variables/components/WorkflowVariablePicker';
|
||||
import { t } from '@lingui/core/macro';
|
||||
import {
|
||||
HorizontalSeparator,
|
||||
IconCalendar,
|
||||
IconHourglassHigh,
|
||||
} from 'twenty-ui/display';
|
||||
import { type SelectOption } from 'twenty-ui/input';
|
||||
type WorkflowEditActionDelayProps = {
|
||||
action: WorkflowDelayAction;
|
||||
actionOptions:
|
||||
| {
|
||||
readonly: true;
|
||||
}
|
||||
| {
|
||||
readonly?: false;
|
||||
onActionUpdate: (action: WorkflowDelayAction) => void;
|
||||
};
|
||||
};
|
||||
|
||||
export const WorkflowEditActionDelay = ({
|
||||
action,
|
||||
actionOptions,
|
||||
}: WorkflowEditActionDelayProps) => {
|
||||
const { headerTitle, headerIcon, headerIconColor, headerType, getIcon } =
|
||||
useWorkflowActionHeader({
|
||||
action,
|
||||
defaultTitle: 'Delay',
|
||||
});
|
||||
|
||||
const delayOptions: Array<SelectOption<'SCHEDULED_DATE' | 'DURATION'>> = [
|
||||
{
|
||||
label: t`At a specific date or time`,
|
||||
value: 'SCHEDULED_DATE',
|
||||
Icon: IconCalendar,
|
||||
},
|
||||
{
|
||||
label: t`After a set amount of time`,
|
||||
value: 'DURATION',
|
||||
Icon: IconHourglassHigh,
|
||||
},
|
||||
];
|
||||
|
||||
const handleDelayTypeChange = (
|
||||
newDelayType: 'SCHEDULED_DATE' | 'DURATION',
|
||||
) => {
|
||||
if (
|
||||
actionOptions.readonly === true ||
|
||||
newDelayType === action.settings.input.delayType
|
||||
) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (newDelayType === 'SCHEDULED_DATE') {
|
||||
actionOptions.onActionUpdate({
|
||||
...action,
|
||||
settings: {
|
||||
...action.settings,
|
||||
input: {
|
||||
delayType: 'SCHEDULED_DATE',
|
||||
},
|
||||
},
|
||||
});
|
||||
} else {
|
||||
actionOptions.onActionUpdate({
|
||||
...action,
|
||||
settings: {
|
||||
...action.settings,
|
||||
input: {
|
||||
delayType: 'DURATION',
|
||||
duration: undefined,
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
const handleDateTimeChange = (value: string | null) => {
|
||||
if (actionOptions.readonly === true) {
|
||||
return;
|
||||
}
|
||||
|
||||
actionOptions.onActionUpdate({
|
||||
...action,
|
||||
settings: {
|
||||
...action.settings,
|
||||
input: {
|
||||
delayType: 'SCHEDULED_DATE',
|
||||
scheduledDateTime: value ?? '',
|
||||
},
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
const handleDurationChange = (
|
||||
field: 'days' | 'hours' | 'minutes' | 'seconds',
|
||||
value: number | string | null,
|
||||
) => {
|
||||
if (actionOptions.readonly === true) {
|
||||
return;
|
||||
}
|
||||
|
||||
actionOptions.onActionUpdate({
|
||||
...action,
|
||||
settings: {
|
||||
...action.settings,
|
||||
input: {
|
||||
delayType: 'DURATION',
|
||||
duration: {
|
||||
days:
|
||||
field === 'days'
|
||||
? (value ?? undefined)
|
||||
: action.settings.input.duration?.days,
|
||||
hours:
|
||||
field === 'hours'
|
||||
? (value ?? undefined)
|
||||
: action.settings.input.duration?.hours,
|
||||
minutes:
|
||||
field === 'minutes'
|
||||
? (value ?? undefined)
|
||||
: action.settings.input.duration?.minutes,
|
||||
seconds:
|
||||
field === 'seconds'
|
||||
? (value ?? undefined)
|
||||
: action.settings.input.duration?.seconds,
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
const HeaderIcon = getIcon(headerIcon ?? 'IconPlayerPause');
|
||||
|
||||
return (
|
||||
<>
|
||||
<SidePanelHeader
|
||||
initialTitle={headerTitle}
|
||||
Icon={HeaderIcon}
|
||||
iconColor={headerIconColor}
|
||||
headerType={headerType}
|
||||
onTitleChange={(newTitle: string) => {
|
||||
if (actionOptions.readonly === true) {
|
||||
return;
|
||||
}
|
||||
|
||||
actionOptions.onActionUpdate({
|
||||
...action,
|
||||
name: newTitle,
|
||||
});
|
||||
}}
|
||||
/>
|
||||
|
||||
<WorkflowStepBody>
|
||||
<Select
|
||||
dropdownId="workflow-edit-action-delay-type"
|
||||
label={t`Resume`}
|
||||
options={delayOptions}
|
||||
dropdownWidth={GenericDropdownContentWidth.Large}
|
||||
value={action.settings.input.delayType}
|
||||
onChange={handleDelayTypeChange}
|
||||
disabled={actionOptions.readonly}
|
||||
/>
|
||||
<HorizontalSeparator noMargin />
|
||||
|
||||
{action.settings.input.delayType === 'SCHEDULED_DATE' && (
|
||||
<FormDateTimeFieldInput
|
||||
label={t`Delay until date`}
|
||||
defaultValue={action.settings.input.scheduledDateTime ?? undefined}
|
||||
onChange={handleDateTimeChange}
|
||||
readonly={actionOptions.readonly}
|
||||
VariablePicker={WorkflowVariablePicker}
|
||||
placeholder="Select a date"
|
||||
/>
|
||||
)}
|
||||
{action.settings.input.delayType === 'DURATION' && (
|
||||
<>
|
||||
<FormNumberFieldInput
|
||||
label={t`Days`}
|
||||
defaultValue={action.settings.input.duration?.days}
|
||||
onChange={(value) => handleDurationChange('days', value)}
|
||||
readonly={actionOptions.readonly}
|
||||
VariablePicker={WorkflowVariablePicker}
|
||||
placeholder="0"
|
||||
/>
|
||||
<FormNumberFieldInput
|
||||
label={t`Hours`}
|
||||
defaultValue={action.settings.input.duration?.hours}
|
||||
onChange={(value) => handleDurationChange('hours', value)}
|
||||
readonly={actionOptions.readonly}
|
||||
VariablePicker={WorkflowVariablePicker}
|
||||
placeholder="0"
|
||||
/>
|
||||
<FormNumberFieldInput
|
||||
label={t`Minutes`}
|
||||
defaultValue={action.settings.input.duration?.minutes}
|
||||
onChange={(value) => handleDurationChange('minutes', value)}
|
||||
readonly={actionOptions.readonly}
|
||||
VariablePicker={WorkflowVariablePicker}
|
||||
placeholder="0"
|
||||
/>
|
||||
<FormNumberFieldInput
|
||||
label={t`Seconds`}
|
||||
defaultValue={action.settings.input.duration?.seconds}
|
||||
onChange={(value) => handleDurationChange('seconds', value)}
|
||||
readonly={actionOptions.readonly}
|
||||
VariablePicker={WorkflowVariablePicker}
|
||||
placeholder="0"
|
||||
/>
|
||||
</>
|
||||
)}
|
||||
</WorkflowStepBody>
|
||||
|
||||
<WorkflowActionFooter stepId={action.id} />
|
||||
</>
|
||||
);
|
||||
};
|
||||
+2
-2
@@ -79,13 +79,13 @@ describe('getActionIconColorOrThrow', () => {
|
||||
});
|
||||
|
||||
describe('FILTER action type', () => {
|
||||
it('should throw an error for FILTER action type', () => {
|
||||
it('should return green color for FILTER action type', () => {
|
||||
const result = getActionIconColorOrThrow({
|
||||
theme: mockTheme,
|
||||
actionType: 'FILTER',
|
||||
});
|
||||
|
||||
expect(result).toBe(mockTheme.font.color.tertiary);
|
||||
expect(result).toBe(mockTheme.color.green60);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
+1
@@ -12,6 +12,7 @@ export const getActionHeaderTypeOrThrow = (actionType: WorkflowActionType) => {
|
||||
case 'FIND_RECORDS':
|
||||
case 'FORM':
|
||||
case 'SEND_EMAIL':
|
||||
case 'DELAY':
|
||||
return msg`Action`;
|
||||
case 'HTTP_REQUEST':
|
||||
return msg`HTTP Request`;
|
||||
|
||||
+4
-3
@@ -1,6 +1,7 @@
|
||||
import { type WorkflowActionType } from '@/workflow/types/Workflow';
|
||||
import { AI_ACTIONS } from '@/workflow/workflow-steps/workflow-actions/constants/AiActions';
|
||||
import { CORE_ACTIONS } from '@/workflow/workflow-steps/workflow-actions/constants/CoreActions';
|
||||
import { FLOW_ACTIONS } from '@/workflow/workflow-steps/workflow-actions/constants/FlowActions';
|
||||
import { HUMAN_INPUT_ACTIONS } from '@/workflow/workflow-steps/workflow-actions/constants/HumanInputActions';
|
||||
import { RECORD_ACTIONS } from '@/workflow/workflow-steps/workflow-actions/constants/RecordActions';
|
||||
|
||||
@@ -11,8 +12,6 @@ export const getActionIcon = (actionType: WorkflowActionType) => {
|
||||
case 'DELETE_RECORD':
|
||||
case 'FIND_RECORDS':
|
||||
return RECORD_ACTIONS.find((item) => item.type === actionType)?.icon;
|
||||
case 'FILTER':
|
||||
return 'IconFilter';
|
||||
case 'AI_AGENT':
|
||||
return AI_ACTIONS.find((item) => item.type === actionType)?.icon;
|
||||
case 'CODE':
|
||||
@@ -22,7 +21,9 @@ export const getActionIcon = (actionType: WorkflowActionType) => {
|
||||
case 'FORM':
|
||||
return HUMAN_INPUT_ACTIONS.find((item) => item.type === actionType)?.icon;
|
||||
case 'ITERATOR':
|
||||
return 'IconRepeat';
|
||||
case 'DELAY':
|
||||
case 'FILTER':
|
||||
return FLOW_ACTIONS.find((item) => item.type === actionType)?.icon;
|
||||
case 'EMPTY':
|
||||
return 'IconSettingsAutomation';
|
||||
default:
|
||||
|
||||
+3
-2
@@ -22,9 +22,10 @@ export const getActionIconColorOrThrow = ({
|
||||
case 'FORM':
|
||||
return theme.color.orange;
|
||||
case 'ITERATOR':
|
||||
case 'FILTER':
|
||||
case 'EMPTY':
|
||||
return theme.font.color.tertiary;
|
||||
case 'FILTER':
|
||||
case 'DELAY':
|
||||
return theme.color.green60;
|
||||
case 'AI_AGENT':
|
||||
return theme.color.pink;
|
||||
default:
|
||||
|
||||
+3
-1
@@ -20,6 +20,7 @@ import { type MessageQueueWorkerOptions } from 'src/engine/core-modules/message-
|
||||
|
||||
import { type MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
|
||||
import { getJobKey } from 'src/engine/core-modules/message-queue/utils/get-job-key.util';
|
||||
import { MESSAGE_QUEUE_PRIORITY } from 'src/engine/core-modules/message-queue/message-queue-priority.constant';
|
||||
|
||||
export type BullMQDriverOptions = QueueOptions;
|
||||
|
||||
@@ -167,10 +168,11 @@ export class BullMQDriver implements MessageQueueDriver, OnModuleDestroy {
|
||||
|
||||
const queueOptions: JobsOptions = {
|
||||
jobId: options?.id ? `${options.id}-${v4()}` : undefined, // We add V4() to id to make sure ids are uniques so we can add a waiting job when a job related with the same option.id is running
|
||||
priority: options?.priority,
|
||||
priority: options?.priority ?? MESSAGE_QUEUE_PRIORITY[queueName],
|
||||
attempts: 1 + (options?.retryLimit || 0),
|
||||
removeOnComplete: true,
|
||||
removeOnFail: 100,
|
||||
delay: options?.delay,
|
||||
};
|
||||
|
||||
await this.queueMap[queueName].add(jobName, data, queueOptions);
|
||||
|
||||
+1
@@ -2,6 +2,7 @@ export interface QueueJobOptions {
|
||||
id?: string;
|
||||
priority?: number;
|
||||
retryLimit?: number;
|
||||
delay?: number;
|
||||
}
|
||||
|
||||
export interface QueueCronJobOptions extends QueueJobOptions {
|
||||
|
||||
+19
@@ -0,0 +1,19 @@
|
||||
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
|
||||
|
||||
export const MESSAGE_QUEUE_PRIORITY = {
|
||||
[MessageQueue.billingQueue]: 1,
|
||||
[MessageQueue.entityEventsToDbQueue]: 1,
|
||||
[MessageQueue.emailQueue]: 1,
|
||||
[MessageQueue.workflowQueue]: 2,
|
||||
[MessageQueue.webhookQueue]: 2,
|
||||
[MessageQueue.messagingQueue]: 2,
|
||||
[MessageQueue.delayedJobsQueue]: 3,
|
||||
[MessageQueue.calendarQueue]: 4,
|
||||
[MessageQueue.contactCreationQueue]: 4,
|
||||
[MessageQueue.taskAssignedQueue]: 4,
|
||||
[MessageQueue.serverlessFunctionQueue]: 4,
|
||||
[MessageQueue.workspaceQueue]: 5,
|
||||
[MessageQueue.triggerQueue]: 5,
|
||||
[MessageQueue.deleteCascadeQueue]: 6,
|
||||
[MessageQueue.cronQueue]: 7,
|
||||
};
|
||||
+1
@@ -14,6 +14,7 @@ export enum MessageQueue {
|
||||
workspaceQueue = 'workspace-queue',
|
||||
entityEventsToDbQueue = 'entity-events-to-db-queue',
|
||||
workflowQueue = 'workflow-queue',
|
||||
delayedJobsQueue = 'delayed-jobs-queue',
|
||||
deleteCascadeQueue = 'delete-cascade-queue',
|
||||
serverlessFunctionQueue = 'serverless-function-queue',
|
||||
triggerQueue = 'trigger-queue',
|
||||
|
||||
+21
@@ -353,6 +353,27 @@ export class WorkflowVersionStepOperationsWorkspaceService {
|
||||
additionalCreatedSteps: [emptyNodeStep],
|
||||
};
|
||||
}
|
||||
case WorkflowActionType.DELAY: {
|
||||
return {
|
||||
builtStep: {
|
||||
...baseStep,
|
||||
name: 'Delay',
|
||||
type: WorkflowActionType.DELAY,
|
||||
settings: {
|
||||
...BASE_STEP_DEFINITION,
|
||||
input: {
|
||||
delayType: 'DURATION',
|
||||
duration: {
|
||||
days: 0,
|
||||
hours: 0,
|
||||
minutes: 0,
|
||||
seconds: 0,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
default:
|
||||
throw new WorkflowVersionStepException(
|
||||
`WorkflowActionType '${type}' unknown`,
|
||||
|
||||
+4
@@ -8,6 +8,7 @@ import {
|
||||
} from 'src/modules/workflow/workflow-executor/exceptions/workflow-step-executor.exception';
|
||||
import { AiAgentWorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/ai-agent/ai-agent.workflow-action';
|
||||
import { CodeWorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/code/code.workflow-action';
|
||||
import { DelayWorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/delay.workflow-action';
|
||||
import { EmptyWorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/empty/empty.workflow-action';
|
||||
import { FilterWorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/filter.workflow-action';
|
||||
import { FormWorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/form/form.workflow-action';
|
||||
@@ -33,6 +34,7 @@ export class WorkflowActionFactory {
|
||||
private readonly toolExecutorWorkflowAction: ToolExecutorWorkflowAction,
|
||||
private readonly aiAgentWorkflowAction: AiAgentWorkflowAction,
|
||||
private readonly emptyWorkflowAction: EmptyWorkflowAction,
|
||||
private readonly delayWorkflowAction: DelayWorkflowAction,
|
||||
) {}
|
||||
|
||||
get(stepType: WorkflowActionType): WorkflowAction {
|
||||
@@ -61,6 +63,8 @@ export class WorkflowActionFactory {
|
||||
return this.aiAgentWorkflowAction;
|
||||
case WorkflowActionType.EMPTY:
|
||||
return this.emptyWorkflowAction;
|
||||
case WorkflowActionType.DELAY:
|
||||
return this.delayWorkflowAction;
|
||||
default:
|
||||
throw new WorkflowStepExecutorException(
|
||||
`Workflow step executor not found for step type '${stepType}'`,
|
||||
|
||||
+1
@@ -0,0 +1 @@
|
||||
export const RESUME_DELAYED_WORKFLOW_JOB_NAME = 'ResumeDelayedWorkflowJob';
|
||||
+13
@@ -0,0 +1,13 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
|
||||
import { DelayWorkflowAction } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/delay.workflow-action';
|
||||
import { ResumeDelayedWorkflowJob } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/jobs/resume-delayed-workflow.job';
|
||||
import { WorkflowRunQueueModule } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workflow-run-queue.module';
|
||||
import { WorkflowRunModule } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.module';
|
||||
|
||||
@Module({
|
||||
imports: [WorkflowRunModule, WorkflowRunQueueModule],
|
||||
providers: [DelayWorkflowAction, ResumeDelayedWorkflowJob],
|
||||
exports: [DelayWorkflowAction],
|
||||
})
|
||||
export class DelayActionModule {}
|
||||
+116
@@ -0,0 +1,116 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
|
||||
import { resolveInput } from 'twenty-shared/utils';
|
||||
|
||||
import { type WorkflowAction } from 'src/modules/workflow/workflow-executor/interfaces/workflow-action.interface';
|
||||
|
||||
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 {
|
||||
WorkflowStepExecutorException,
|
||||
WorkflowStepExecutorExceptionCode,
|
||||
} from 'src/modules/workflow/workflow-executor/exceptions/workflow-step-executor.exception';
|
||||
import { type WorkflowActionInput } from 'src/modules/workflow/workflow-executor/types/workflow-action-input';
|
||||
import { type WorkflowActionOutput } from 'src/modules/workflow/workflow-executor/types/workflow-action-output.type';
|
||||
import { findStepOrThrow } from 'src/modules/workflow/workflow-executor/utils/find-step-or-throw.util';
|
||||
import { RESUME_DELAYED_WORKFLOW_JOB_NAME } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/contants/resume-delayed-workflow-job-name';
|
||||
import { isWorkflowDelayAction } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/guards/is-workflow-delay-action.guard';
|
||||
import { ResumeDelayedWorkflowJobData } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/types/resume-delayed-workflow-job-data.type';
|
||||
import { WorkflowDelayActionInput } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/types/workflow-delay-action-input.type';
|
||||
|
||||
@Injectable()
|
||||
export class DelayWorkflowAction implements WorkflowAction {
|
||||
constructor(
|
||||
@InjectMessageQueue(MessageQueue.delayedJobsQueue)
|
||||
private readonly messageQueueService: MessageQueueService,
|
||||
) {}
|
||||
|
||||
async execute({
|
||||
currentStepId,
|
||||
steps,
|
||||
runInfo,
|
||||
context,
|
||||
}: WorkflowActionInput): Promise<WorkflowActionOutput> {
|
||||
const step = findStepOrThrow({
|
||||
stepId: currentStepId,
|
||||
steps,
|
||||
});
|
||||
|
||||
if (!isWorkflowDelayAction(step)) {
|
||||
throw new WorkflowStepExecutorException(
|
||||
'Step is not a delay action',
|
||||
WorkflowStepExecutorExceptionCode.INVALID_STEP_TYPE,
|
||||
);
|
||||
}
|
||||
|
||||
const workflowActionInput = resolveInput(
|
||||
step.settings.input,
|
||||
context,
|
||||
) as WorkflowDelayActionInput;
|
||||
|
||||
let delayInMs: number;
|
||||
|
||||
if (workflowActionInput.delayType === 'SCHEDULED_DATE') {
|
||||
if (!workflowActionInput.scheduledDateTime) {
|
||||
throw new WorkflowStepExecutorException(
|
||||
'Scheduled date time is required for scheduled date delay',
|
||||
WorkflowStepExecutorExceptionCode.INVALID_STEP_TYPE,
|
||||
);
|
||||
}
|
||||
|
||||
const scheduledDate = new Date(workflowActionInput.scheduledDateTime);
|
||||
const now = new Date();
|
||||
|
||||
delayInMs = scheduledDate.getTime() - now.getTime();
|
||||
|
||||
if (delayInMs < 0) {
|
||||
throw new WorkflowStepExecutorException(
|
||||
'Scheduled date cannot be in the past',
|
||||
WorkflowStepExecutorExceptionCode.INVALID_STEP_TYPE,
|
||||
);
|
||||
}
|
||||
} else if (workflowActionInput.delayType === 'DURATION') {
|
||||
if (!workflowActionInput.duration) {
|
||||
throw new WorkflowStepExecutorException(
|
||||
'Duration is required for duration delay',
|
||||
WorkflowStepExecutorExceptionCode.INVALID_STEP_TYPE,
|
||||
);
|
||||
}
|
||||
|
||||
const {
|
||||
days = 0,
|
||||
hours = 0,
|
||||
minutes = 0,
|
||||
seconds = 0,
|
||||
} = workflowActionInput.duration;
|
||||
|
||||
delayInMs =
|
||||
days * 24 * 60 * 60 * 1000 +
|
||||
hours * 60 * 60 * 1000 +
|
||||
minutes * 60 * 1000 +
|
||||
seconds * 1000;
|
||||
} else {
|
||||
throw new WorkflowStepExecutorException(
|
||||
'Invalid delay type',
|
||||
WorkflowStepExecutorExceptionCode.INVALID_STEP_TYPE,
|
||||
);
|
||||
}
|
||||
|
||||
await this.messageQueueService.add<ResumeDelayedWorkflowJobData>(
|
||||
RESUME_DELAYED_WORKFLOW_JOB_NAME,
|
||||
{
|
||||
workspaceId: runInfo.workspaceId,
|
||||
workflowRunId: runInfo.workflowRunId,
|
||||
stepId: currentStepId,
|
||||
},
|
||||
{
|
||||
delay: delayInMs,
|
||||
},
|
||||
);
|
||||
|
||||
return {
|
||||
pendingEvent: true,
|
||||
};
|
||||
}
|
||||
}
|
||||
+11
@@ -0,0 +1,11 @@
|
||||
import {
|
||||
type WorkflowAction,
|
||||
WorkflowActionType,
|
||||
type WorkflowDelayAction,
|
||||
} from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action.type';
|
||||
|
||||
export const isWorkflowDelayAction = (
|
||||
action: WorkflowAction,
|
||||
): action is WorkflowDelayAction => {
|
||||
return action.type === WorkflowActionType.DELAY;
|
||||
};
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
import { Scope } from '@nestjs/common';
|
||||
|
||||
import { StepStatus } from 'twenty-shared/workflow';
|
||||
|
||||
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 { WorkflowRunStatus } from 'src/modules/workflow/common/standard-objects/workflow-run.workspace-entity';
|
||||
import { RESUME_DELAYED_WORKFLOW_JOB_NAME } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/contants/resume-delayed-workflow-job-name';
|
||||
import { isWorkflowDelayAction } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/guards/is-workflow-delay-action.guard';
|
||||
import { ResumeDelayedWorkflowJobData } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/types/resume-delayed-workflow-job-data.type';
|
||||
import {
|
||||
WorkflowRunException,
|
||||
WorkflowRunExceptionCode,
|
||||
} from 'src/modules/workflow/workflow-runner/exceptions/workflow-run.exception';
|
||||
import { RunWorkflowJob } from 'src/modules/workflow/workflow-runner/jobs/run-workflow.job';
|
||||
import { type RunWorkflowJobData } from 'src/modules/workflow/workflow-runner/types/run-workflow-job-data.type';
|
||||
import { WorkflowRunQueueWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run-queue/workspace-services/workflow-run-queue.workspace-service';
|
||||
import { WorkflowRunWorkspaceService } from 'src/modules/workflow/workflow-runner/workflow-run/workflow-run.workspace-service';
|
||||
|
||||
@Processor({
|
||||
queueName: MessageQueue.delayedJobsQueue,
|
||||
scope: Scope.REQUEST,
|
||||
})
|
||||
export class ResumeDelayedWorkflowJob {
|
||||
constructor(
|
||||
@InjectMessageQueue(MessageQueue.workflowQueue)
|
||||
private readonly messageQueueService: MessageQueueService,
|
||||
private readonly workflowRunWorkspaceService: WorkflowRunWorkspaceService,
|
||||
private readonly workflowRunQueueWorkspaceService: WorkflowRunQueueWorkspaceService,
|
||||
) {}
|
||||
|
||||
@Process(RESUME_DELAYED_WORKFLOW_JOB_NAME)
|
||||
async handle({
|
||||
workspaceId,
|
||||
workflowRunId,
|
||||
stepId,
|
||||
}: ResumeDelayedWorkflowJobData): Promise<void> {
|
||||
try {
|
||||
const workflowRun =
|
||||
await this.workflowRunWorkspaceService.getWorkflowRunOrFail({
|
||||
workflowRunId,
|
||||
workspaceId,
|
||||
});
|
||||
|
||||
if (workflowRun.status !== WorkflowRunStatus.RUNNING) {
|
||||
return;
|
||||
}
|
||||
|
||||
const step = workflowRun.state?.flow?.steps?.find(
|
||||
(step) => step.id === stepId,
|
||||
);
|
||||
|
||||
const stepInfo = workflowRun.state?.stepInfos[stepId];
|
||||
|
||||
if (!step || !isWorkflowDelayAction(step)) {
|
||||
throw new WorkflowRunException(
|
||||
'Step not found or is not a delay action',
|
||||
WorkflowRunExceptionCode.INVALID_INPUT,
|
||||
);
|
||||
}
|
||||
|
||||
if (stepInfo?.status !== StepStatus.PENDING) {
|
||||
throw new WorkflowRunException(
|
||||
'Step is not pending',
|
||||
WorkflowRunExceptionCode.INVALID_INPUT,
|
||||
);
|
||||
}
|
||||
|
||||
await this.workflowRunWorkspaceService.updateWorkflowRunStepInfo({
|
||||
stepId,
|
||||
stepInfo: {
|
||||
status: StepStatus.SUCCESS,
|
||||
result: {
|
||||
success: true,
|
||||
},
|
||||
},
|
||||
workspaceId,
|
||||
workflowRunId,
|
||||
});
|
||||
|
||||
await this.messageQueueService.add<RunWorkflowJobData>(
|
||||
RunWorkflowJob.name,
|
||||
{
|
||||
workspaceId,
|
||||
workflowRunId,
|
||||
lastExecutedStepId: stepId,
|
||||
},
|
||||
);
|
||||
|
||||
await this.workflowRunQueueWorkspaceService.increaseWorkflowRunQueuedCount(
|
||||
workspaceId,
|
||||
);
|
||||
} catch (error) {
|
||||
await this.workflowRunWorkspaceService.endWorkflowRun({
|
||||
workflowRunId,
|
||||
workspaceId,
|
||||
status: WorkflowRunStatus.FAILED,
|
||||
error:
|
||||
error instanceof Error
|
||||
? error.message
|
||||
: 'Unknown error during delay resume',
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
+5
@@ -0,0 +1,5 @@
|
||||
export type ResumeDelayedWorkflowJobData = {
|
||||
workspaceId: string;
|
||||
workflowRunId: string;
|
||||
stepId: string;
|
||||
};
|
||||
+18
@@ -0,0 +1,18 @@
|
||||
export type WorkflowDelayActionInput =
|
||||
| WorkflowScheduledDateActionInput
|
||||
| WorkflowDurationDelayActionInput;
|
||||
|
||||
export type WorkflowScheduledDateActionInput = {
|
||||
delayType: 'SCHEDULED_DATE';
|
||||
scheduledDateTime: string;
|
||||
};
|
||||
|
||||
export type WorkflowDurationDelayActionInput = {
|
||||
delayType: 'DURATION';
|
||||
duration: {
|
||||
days?: number;
|
||||
hours?: number;
|
||||
minutes?: number;
|
||||
seconds?: number;
|
||||
};
|
||||
};
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
import { type WorkflowDelayActionInput } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/types/workflow-delay-action-input.type';
|
||||
import { type BaseWorkflowActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action-settings.type';
|
||||
|
||||
export type WorkflowDelayActionSettings = BaseWorkflowActionSettings & {
|
||||
input: WorkflowDelayActionInput;
|
||||
};
|
||||
+2
@@ -1,6 +1,7 @@
|
||||
import { type OutputSchema } from 'src/modules/workflow/workflow-builder/workflow-schema/types/output-schema.type';
|
||||
import { type WorkflowAiAgentActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/ai-agent/types/workflow-ai-agent-action-settings.type';
|
||||
import { type WorkflowCodeActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/code/types/workflow-code-action-settings.type';
|
||||
import { type WorkflowDelayActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/types/workflow-delay-action-settings.type';
|
||||
import { type WorkflowFilterActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/types/workflow-filter-action-settings.type';
|
||||
import { type WorkflowFormActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/form/types/workflow-form-action-settings.type';
|
||||
import { type WorkflowHttpRequestActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/http-request/types/workflow-http-request-action-settings.type';
|
||||
@@ -36,4 +37,5 @@ export type WorkflowActionSettings =
|
||||
| WorkflowFilterActionSettings
|
||||
| WorkflowHttpRequestActionSettings
|
||||
| WorkflowAiAgentActionSettings
|
||||
| WorkflowDelayActionSettings
|
||||
| WorkflowIteratorActionSettings;
|
||||
|
||||
+9
-1
@@ -12,6 +12,7 @@ import {
|
||||
type WorkflowUpdateRecordActionSettings,
|
||||
} from 'src/modules/workflow/workflow-executor/workflow-actions/record-crud/types/workflow-record-crud-action-settings.type';
|
||||
import { type WorkflowActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/types/workflow-action-settings.type';
|
||||
import { type WorkflowDelayActionSettings } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/types/workflow-delay-action-settings.type';
|
||||
|
||||
export enum WorkflowActionType {
|
||||
CODE = 'CODE',
|
||||
@@ -26,6 +27,7 @@ export enum WorkflowActionType {
|
||||
AI_AGENT = 'AI_AGENT',
|
||||
ITERATOR = 'ITERATOR',
|
||||
EMPTY = 'EMPTY',
|
||||
DELAY = 'DELAY',
|
||||
}
|
||||
|
||||
type BaseWorkflowAction = {
|
||||
@@ -100,6 +102,11 @@ export type WorkflowEmptyAction = BaseWorkflowAction & {
|
||||
type: WorkflowActionType.EMPTY;
|
||||
};
|
||||
|
||||
export type WorkflowDelayAction = BaseWorkflowAction & {
|
||||
type: WorkflowActionType.DELAY;
|
||||
settings: WorkflowDelayActionSettings;
|
||||
};
|
||||
|
||||
export type WorkflowAction =
|
||||
| WorkflowCodeAction
|
||||
| WorkflowSendEmailAction
|
||||
@@ -112,4 +119,5 @@ export type WorkflowAction =
|
||||
| WorkflowHttpRequestAction
|
||||
| WorkflowAiAgentAction
|
||||
| WorkflowIteratorAction
|
||||
| WorkflowEmptyAction;
|
||||
| WorkflowEmptyAction
|
||||
| WorkflowDelayAction;
|
||||
|
||||
+2
@@ -8,6 +8,7 @@ import { WorkflowCommonModule } from 'src/modules/workflow/common/workflow-commo
|
||||
import { WorkflowActionFactory } from 'src/modules/workflow/workflow-executor/factories/workflow-action.factory';
|
||||
import { AiAgentActionModule } from 'src/modules/workflow/workflow-executor/workflow-actions/ai-agent/ai-agent-action.module';
|
||||
import { CodeActionModule } from 'src/modules/workflow/workflow-executor/workflow-actions/code/code-action.module';
|
||||
import { DelayActionModule } from 'src/modules/workflow/workflow-executor/workflow-actions/delay/delay-action.module';
|
||||
import { EmptyActionModule } from 'src/modules/workflow/workflow-executor/workflow-actions/empty/empty-action.module';
|
||||
import { FilterActionModule } from 'src/modules/workflow/workflow-executor/workflow-actions/filter/filter-action.module';
|
||||
import { FormActionModule } from 'src/modules/workflow/workflow-executor/workflow-actions/form/form-action.module';
|
||||
@@ -22,6 +23,7 @@ import { WorkflowRunModule } from 'src/modules/workflow/workflow-runner/workflow
|
||||
imports: [
|
||||
WorkflowCommonModule,
|
||||
CodeActionModule,
|
||||
DelayActionModule,
|
||||
RecordCRUDActionModule,
|
||||
FormActionModule,
|
||||
WorkflowRunModule,
|
||||
|
||||
@@ -43,6 +43,8 @@ export { workflowUpdateRecordActionSchema } from './schemas/update-record-action
|
||||
export { workflowUpdateRecordActionSettingsSchema } from './schemas/update-record-action-settings-schema';
|
||||
export { workflowWebhookTriggerSchema } from './schemas/webhook-trigger-schema';
|
||||
export { workflowActionSchema } from './schemas/workflow-action-schema';
|
||||
export { workflowDelayActionSchema } from './schemas/workflow-delay-action-schema';
|
||||
export { workflowDelayActionSettingsSchema } from './schemas/workflow-delay-action-settings-schema';
|
||||
export { workflowRunSchema } from './schemas/workflow-run-schema';
|
||||
export { workflowRunStateSchema } from './schemas/workflow-run-state-schema';
|
||||
export { workflowRunStateStepInfoSchema } from './schemas/workflow-run-state-step-info-schema';
|
||||
|
||||
@@ -11,6 +11,7 @@ import { workflowHttpRequestActionSchema } from './http-request-action-schema';
|
||||
import { workflowIteratorActionSchema } from './iterator-action-schema';
|
||||
import { workflowSendEmailActionSchema } from './send-email-action-schema';
|
||||
import { workflowUpdateRecordActionSchema } from './update-record-action-schema';
|
||||
import { workflowDelayActionSchema } from './workflow-delay-action-schema';
|
||||
|
||||
export const workflowActionSchema = z.discriminatedUnion('type', [
|
||||
workflowCodeActionSchema,
|
||||
@@ -24,5 +25,6 @@ export const workflowActionSchema = z.discriminatedUnion('type', [
|
||||
workflowAiAgentActionSchema,
|
||||
workflowFilterActionSchema,
|
||||
workflowIteratorActionSchema,
|
||||
workflowDelayActionSchema,
|
||||
workflowEmptyActionSchema,
|
||||
]);
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
import { z } from 'zod';
|
||||
import { baseWorkflowActionSchema } from './base-workflow-action-schema';
|
||||
import { workflowDelayActionSettingsSchema } from './workflow-delay-action-settings-schema';
|
||||
|
||||
export const workflowDelayActionSchema = baseWorkflowActionSchema.extend({
|
||||
type: z.literal('DELAY'),
|
||||
settings: workflowDelayActionSettingsSchema,
|
||||
});
|
||||
@@ -0,0 +1,18 @@
|
||||
import { z } from 'zod';
|
||||
import { baseWorkflowActionSettingsSchema } from './base-workflow-action-settings-schema';
|
||||
|
||||
export const workflowDelayActionSettingsSchema =
|
||||
baseWorkflowActionSettingsSchema.extend({
|
||||
input: z.object({
|
||||
delayType: z.enum(['SCHEDULED_DATE', 'DURATION']),
|
||||
scheduledDateTime: z.string().nullable().optional(),
|
||||
duration: z
|
||||
.object({
|
||||
days: z.union([z.number().min(0), z.string()]).optional(),
|
||||
hours: z.union([z.number().min(0), z.string()]).optional(),
|
||||
minutes: z.union([z.number().min(0), z.string()]).optional(),
|
||||
seconds: z.union([z.number().min(0), z.string()]).optional(),
|
||||
})
|
||||
.optional(),
|
||||
}),
|
||||
});
|
||||
@@ -194,6 +194,7 @@ export {
|
||||
IconHistory,
|
||||
IconHistoryToggle,
|
||||
IconHome,
|
||||
IconHourglassHigh,
|
||||
IconHours24,
|
||||
IconHttpGet,
|
||||
IconHttpPost,
|
||||
|
||||
@@ -257,6 +257,7 @@ export {
|
||||
IconHistory,
|
||||
IconHistoryToggle,
|
||||
IconHome,
|
||||
IconHourglassHigh,
|
||||
IconHours24,
|
||||
IconHttpGet,
|
||||
IconHttpPost,
|
||||
|
||||
Reference in New Issue
Block a user