diff --git a/apps/api/src/controllers/Workflows.ts b/apps/api/src/controllers/Workflows.ts index 734ceebbd..f417701f3 100644 --- a/apps/api/src/controllers/Workflows.ts +++ b/apps/api/src/controllers/Workflows.ts @@ -311,7 +311,7 @@ export class Workflows { public async createTransition(req: Request, res: Response, _next: NextFunction) { const auth = res.locals.auth; const workflowId = req.params.id; - const {fromStepId, toStepId, condition, priority} = req.body; + const {fromStepId, toStepId, condition, waitOutcome, priority} = req.body; if (!workflowId) { return res.status(400).json({error: 'Workflow ID is required'}); @@ -325,6 +325,7 @@ export class Workflows { fromStepId, toStepId, condition, + waitOutcome, priority, }); diff --git a/apps/api/src/services/WorkflowExecutionService.ts b/apps/api/src/services/WorkflowExecutionService.ts index b62cf0f9b..c9d07df2c 100644 --- a/apps/api/src/services/WorkflowExecutionService.ts +++ b/apps/api/src/services/WorkflowExecutionService.ts @@ -7,7 +7,7 @@ import type { WorkflowStep, WorkflowStepExecution, } from '@plunk/db'; -import {StepExecutionStatus, WorkflowExecutionStatus} from '@plunk/db'; +import {StepExecutionStatus, WorkflowExecutionStatus, WorkflowWaitOutcome} from '@plunk/db'; import {toPrismaJson} from '@plunk/types'; import {renderTemplate, WorkflowStepConfigSchemas} from '@plunk/shared'; import dns from 'node:dns/promises'; @@ -30,6 +30,7 @@ type WorkflowStepWithTransitions = WorkflowStep & { outgoingTransitions?: Array<{ id: string; condition: Prisma.JsonValue; + waitOutcome: WorkflowWaitOutcome | null; priority: number; toStep: WorkflowStep; }>; @@ -328,71 +329,28 @@ export class WorkflowExecutionService { return; } + const result = { + waitOutcome: WorkflowWaitOutcome.TIMEOUT, + timedOut: true, + eventName: + stepExecution.step.config && + typeof stepExecution.step.config === 'object' && + 'eventName' in stepExecution.step.config + ? stepExecution.step.config.eventName + : undefined, + }; + // Mark step as completed with timeout await prisma.workflowStepExecution.update({ where: {id: stepExecution.id}, data: { status: StepExecutionStatus.COMPLETED, completedAt: new Date(), - output: { - timedOut: true, - eventName: - stepExecution.step.config && - typeof stepExecution.step.config === 'object' && - 'eventName' in stepExecution.step.config - ? stepExecution.step.config.eventName - : undefined, - }, + output: toPrismaJson(result), }, }); - // Continue workflow - find transitions with timeout/fallback logic - const transitions = stepExecution.step.outgoingTransitions || []; - const fallbackTransition = transitions.find( - t => - (t.condition && - typeof t.condition === 'object' && - 'branch' in t.condition && - t.condition.branch === 'timeout') || - (t.condition && typeof t.condition === 'object' && 'fallback' in t.condition && t.condition.fallback === true), - ); - - if (fallbackTransition) { - // Follow timeout branch - await prisma.workflowExecution.update({ - where: {id: stepExecution.executionId}, - data: { - status: WorkflowExecutionStatus.RUNNING, - currentStepId: fallbackTransition.toStep.id, - }, - }); - - await this.processStepExecution(stepExecution.executionId, fallbackTransition.toStep.id); - } else if (transitions.length > 0) { - // No timeout branch, follow first transition - const firstTransition = transitions[0]; - if (firstTransition?.toStep) { - const nextStep = firstTransition.toStep; - await prisma.workflowExecution.update({ - where: {id: stepExecution.executionId}, - data: { - status: WorkflowExecutionStatus.RUNNING, - currentStepId: nextStep.id, - }, - }); - - await this.processStepExecution(stepExecution.executionId, nextStep.id); - } - } else { - // No transitions, complete workflow - await prisma.workflowExecution.update({ - where: {id: stepExecution.executionId}, - data: { - status: WorkflowExecutionStatus.COMPLETED, - completedAt: new Date(), - }, - }); - } + await this.processNextSteps(stepExecution.execution, stepExecution.step, result); } /** @@ -438,25 +396,43 @@ export class WorkflowExecutionService { const config = stepExecution.step.config; if (config && typeof config === 'object' && 'eventName' in config && config.eventName === eventName) { - // Event matches, resume execution - await prisma.workflowStepExecution.update({ - where: {id: stepExecution.id}, - data: { - status: StepExecutionStatus.COMPLETED, - completedAt: new Date(), - output: toPrismaJson({ - eventName, - eventData: data ? toPrismaJson(data) : undefined, - receivedAt: new Date().toISOString(), - }), - }, - }); + const currentContext = + stepExecution.execution.context && + typeof stepExecution.execution.context === 'object' && + !Array.isArray(stepExecution.execution.context) + ? (stepExecution.execution.context as Record) + : {}; + const resumedContext = {...currentContext, ...(data ?? {})}; + const result = { + waitOutcome: WorkflowWaitOutcome.EVENT, + eventReceived: true, + eventName, + ...(data ? {eventData: data} : {}), + receivedAt: new Date().toISOString(), + }; + + // Persist the resume payload with the completed wait. A worker restart + // after this commit reads the same context for downstream steps. + await prisma.$transaction([ + prisma.workflowStepExecution.update({ + where: {id: stepExecution.id}, + data: { + status: StepExecutionStatus.COMPLETED, + completedAt: new Date(), + output: toPrismaJson(result), + }, + }), + prisma.workflowExecution.update({ + where: {id: stepExecution.executionId}, + data: {context: toPrismaJson(resumedContext)}, + }), + ]); // Cancel any pending timeout job await QueueService.cancelWorkflowTimeout(stepExecution.id); // Continue workflow - await this.processNextSteps(stepExecution.execution, stepExecution.step, {eventReceived: true}); + await this.processNextSteps(stepExecution.execution, stepExecution.step, result); } } } @@ -1192,7 +1168,7 @@ export class WorkflowExecutionService { * Process next steps based on transitions */ private static async processNextSteps( - execution: WorkflowExecutionWithRelations, + execution: Pick, currentStep: WorkflowStepWithTransitions, stepResult: StepResult, ): Promise { @@ -1211,10 +1187,24 @@ export class WorkflowExecutionService { return; } - // Find the appropriate transition based on conditions + // WAIT_FOR_EVENT owns a distinct routing dimension. It does not borrow the + // condition JSON used by CONDITION steps. let nextStep = null; - for (const transition of transitions) { + if (currentStep.type === 'WAIT_FOR_EVENT') { + const waitOutcome = stepResult.waitOutcome; + const outcomeTransition = transitions.find(t => t.waitOutcome === waitOutcome); + + if (outcomeTransition) { + nextStep = outcomeTransition.toStep; + } else if (transitions.every(t => t.waitOutcome === null)) { + // Compatibility for rows created before the wait-outcome migration and + // direct test fixtures: their single transition handled either result. + nextStep = transitions[0]?.toStep ?? null; + } + } + + for (const transition of currentStep.type === 'WAIT_FOR_EVENT' ? [] : transitions) { const condition = transition.condition; // If no condition, always follow @@ -1372,7 +1362,7 @@ export class WorkflowExecutionService { private static evaluateTransitionCondition( _condition: Prisma.JsonValue, _stepResult: StepResult, - _execution: WorkflowExecutionWithRelations, + _execution: Pick, ): boolean { // Implement custom transition condition logic here // For now, return false as default diff --git a/apps/api/src/services/WorkflowService.ts b/apps/api/src/services/WorkflowService.ts index 47aafb48b..c313ffde8 100644 --- a/apps/api/src/services/WorkflowService.ts +++ b/apps/api/src/services/WorkflowService.ts @@ -1,5 +1,5 @@ import type {Workflow, WorkflowExecution, WorkflowStep, WorkflowTransition} from '@plunk/db'; -import {Prisma, WorkflowExecutionStatus} from '@plunk/db'; +import {Prisma, WorkflowExecutionStatus, WorkflowWaitOutcome} from '@plunk/db'; import type {PaginatedResponse, WorkflowExecutionWithDetails, WorkflowWithDetails} from '@plunk/types'; import {toPrismaJson} from '@plunk/types'; import signale from 'signale'; @@ -14,6 +14,41 @@ import {NtfyService} from './NtfyService.js'; import {WorkflowExecutionService} from './WorkflowExecutionService.js'; export class WorkflowService { + private static waitHasTimeout(config: Prisma.JsonValue): boolean { + return ( + typeof config === 'object' && + config !== null && + !Array.isArray(config) && + typeof config.timeout === 'number' && + config.timeout > 0 + ); + } + + /** + * Build the default route(s) when a step is inserted into a linear path. + * WAIT_FOR_EVENT keeps event and timeout routing explicit even when both + * outcomes continue to the same next step. + */ + private static defaultOutgoingTransitions( + step: Pick, + toStepId: string, + ): Prisma.WorkflowTransitionCreateManyInput[] { + const base = {fromStepId: step.id, toStepId, priority: 0}; + + if (step.type === 'CONDITION') { + return [{...base, condition: toPrismaJson({branch: 'yes'})}]; + } + + if (step.type === 'WAIT_FOR_EVENT') { + return [ + {...base, waitOutcome: WorkflowWaitOutcome.EVENT}, + ...(this.waitHasTimeout(step.config) ? [{...base, waitOutcome: WorkflowWaitOutcome.TIMEOUT, priority: 1}] : []), + ]; + } + + return [base]; + } + /** * Get all workflows for a project with pagination */ @@ -469,6 +504,7 @@ export class WorkflowService { transition.condition === null ? Prisma.JsonNull : (transition.condition as Prisma.InputJsonValue), + waitOutcome: transition.waitOutcome, priority: transition.priority, }, }); @@ -530,12 +566,8 @@ export class WorkflowService { const lastStep = stepsWithoutOutgoing[stepsWithoutOutgoing.length - 1]; if (lastStep) { - await prisma.workflowTransition.create({ - data: { - fromStepId: lastStep.id, - toStepId: newStep.id, - priority: 0, - }, + await prisma.workflowTransition.createMany({ + data: this.defaultOutgoingTransitions(lastStep, newStep.id), }); } } @@ -746,7 +778,7 @@ export class WorkflowService { /** * Splice a step out of the flow: re-wire its parent(s) directly to its child, - * then delete only the step itself. Not allowed for CONDITION or TRIGGER steps. + * then delete only the step itself. Not allowed for branching or TRIGGER steps. */ public static async spliceStep(projectId: string, workflowId: string, stepId: string): Promise { await this.get(projectId, workflowId); @@ -767,8 +799,8 @@ export class WorkflowService { throw new HttpException(400, 'Cannot remove the trigger step.'); } - if (step.type === 'CONDITION') { - throw new HttpException(400, 'Cannot splice a condition step out of the flow.'); + if (step.type === 'CONDITION' || step.type === 'WAIT_FOR_EVENT') { + throw new HttpException(400, 'Cannot splice a branching step out of the flow.'); } // Check for active executions on this step @@ -812,10 +844,8 @@ export class WorkflowService { * so a condition branch keeps its label) and a fresh transition carries the * flow on to the original target. * - * When the inserted step is itself a CONDITION, the downstream transition is - * attached to its first branch (`yes`) rather than left unconditional — a - * condition step routes exclusively by branch, so an unconditional outgoing - * transition would never be taken and B would be stranded. + * A newly inserted CONDITION starts on `yes`. A newly inserted WAIT_FOR_EVENT + * gets explicit event and, when configured, timeout routes to B. * * Purely additive — no step is removed and nothing becomes unreachable — so * unlike splice/delete this does not need to guard against in-flight executions. @@ -879,15 +909,8 @@ export class WorkflowService { data: {toStepId: newStep.id}, }); - await tx.workflowTransition.create({ - data: { - fromStepId: newStep.id, - toStepId: transition.toStepId, - // A freshly created condition has no config yet, which the builder reads - // as the default binary yes/no pair — so `yes` is its first branch. - condition: data.type === 'CONDITION' ? toPrismaJson({branch: 'yes'}) : Prisma.JsonNull, - priority: 0, - }, + await tx.workflowTransition.createMany({ + data: this.defaultOutgoingTransitions(newStep, transition.toStepId), }); return newStep; @@ -940,6 +963,7 @@ export class WorkflowService { fromStepId: string; toStepId: string; condition?: Prisma.JsonValue; + waitOutcome?: WorkflowWaitOutcome; priority?: number; }, ): Promise { @@ -967,32 +991,61 @@ export class WorkflowService { throw new HttpException(400, 'An exit step ends the flow and cannot be connected to another step'); } - // For CONDITION steps, validate that this branch doesn't already have a transition - if (fromStep?.type === 'CONDITION' && data.condition) { - const conditionObj = - typeof data.condition === 'object' && data.condition !== null - ? (data.condition as Record) - : null; - if (conditionObj && 'branch' in conditionObj) { - // Check if a transition with this branch already exists - const existingTransition = await prisma.workflowTransition.findFirst({ - where: { - fromStepId: data.fromStepId, - condition: { - path: ['branch'], - equals: toPrismaJson(conditionObj.branch), + const hasCondition = data.condition !== undefined && data.condition !== null; + + if (fromStep?.type === 'WAIT_FOR_EVENT') { + if (hasCondition) { + throw new HttpException(400, 'Wait routes use waitOutcome, not transition conditions'); + } + if (!data.waitOutcome) { + throw new HttpException(400, 'WAIT_FOR_EVENT transitions require an EVENT or TIMEOUT outcome'); + } + if (data.waitOutcome === WorkflowWaitOutcome.TIMEOUT && !this.waitHasTimeout(fromStep.config)) { + throw new HttpException(400, 'A TIMEOUT transition requires a positive timeout on the wait step'); + } + + const existingOutcome = await prisma.workflowTransition.findFirst({ + where: {fromStepId: data.fromStepId, waitOutcome: data.waitOutcome}, + }); + if (existingOutcome) { + throw new HttpException(400, `The ${data.waitOutcome} route already has a transition`); + } + } else if (fromStep?.type === 'CONDITION') { + if (data.waitOutcome) { + throw new HttpException(400, 'Condition branches cannot declare a wait outcome'); + } + if (data.condition) { + const conditionObj = + typeof data.condition === 'object' && data.condition !== null + ? (data.condition as Record) + : null; + if (conditionObj && 'branch' in conditionObj) { + // Check if a transition with this branch already exists + const existingTransition = await prisma.workflowTransition.findFirst({ + where: { + fromStepId: data.fromStepId, + condition: { + path: ['branch'], + equals: toPrismaJson(conditionObj.branch), + }, }, - }, - }); + }); - if (existingTransition) { - throw new HttpException( - 400, - `A transition for the "${conditionObj.branch}" branch already exists from this step`, - ); + if (existingTransition) { + throw new HttpException( + 400, + `A transition for the "${conditionObj.branch}" branch already exists from this step`, + ); + } } } - } else if (fromStep?.type !== 'CONDITION') { + } else { + if (data.waitOutcome) { + throw new HttpException(400, 'Only WAIT_FOR_EVENT steps can declare a wait outcome'); + } + if (hasCondition) { + throw new HttpException(400, 'Only CONDITION steps can declare a transition condition'); + } // Every other step type routes to exactly one next step. A second outgoing // transition would make the next hop ambiguous at execution time. const existingOutgoing = await prisma.workflowTransition.findFirst({ @@ -1018,6 +1071,7 @@ export class WorkflowService { fromStepId: data.fromStepId, toStepId: data.toStepId, condition: data.condition ?? Prisma.JsonNull, + waitOutcome: data.waitOutcome, priority: data.priority ?? 0, }, }); diff --git a/apps/api/src/services/__tests__/WorkflowExecutionService.integration.test.ts b/apps/api/src/services/__tests__/WorkflowExecutionService.integration.test.ts index 5caedc15a..de3ee531f 100644 --- a/apps/api/src/services/__tests__/WorkflowExecutionService.integration.test.ts +++ b/apps/api/src/services/__tests__/WorkflowExecutionService.integration.test.ts @@ -1,5 +1,5 @@ import {beforeEach, describe, expect, it, vi} from 'vitest'; -import {StepExecutionStatus, WorkflowExecutionStatus, WorkflowStepType} from '@plunk/db'; +import {StepExecutionStatus, WorkflowExecutionStatus, WorkflowStepType, WorkflowWaitOutcome} from '@plunk/db'; import {toPrismaJson} from '@plunk/types'; import {WorkflowExecutionService} from '../WorkflowExecutionService'; import {factories, getPrismaClient} from '../../../../../test/helpers'; @@ -512,6 +512,173 @@ describe('WorkflowExecutionService - Integration Tests', () => { // Step should be completed after event arrives expect(waitStepExecution?.status).toBe(StepExecutionStatus.COMPLETED); }); + + it('should resume a persisted wait on the event route with merged event context', async () => { + const contact = await factories.createContact({projectId}); + const workflow = await factories.createWorkflow({projectId}); + const waitStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.WAIT_FOR_EVENT, + name: 'Wait for Purchase', + position: {x: 100, y: 0}, + config: toPrismaJson({eventName: 'purchase.completed', timeout: 3600}), + }, + }); + const eventCondition = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.CONDITION, + name: 'Inspect resumed plan', + position: {x: 200, y: -50}, + config: toPrismaJson({field: 'event.plan', operator: 'equals', value: 'pro'}), + }, + }); + const eventExit = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.EXIT, + name: 'Event path', + position: {x: 300, y: -75}, + config: toPrismaJson({}), + }, + }); + const wrongPlanExit = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.EXIT, + name: 'Wrong plan path', + position: {x: 300, y: -25}, + config: toPrismaJson({}), + }, + }); + const timeoutExit = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.EXIT, + name: 'Timeout path', + position: {x: 200, y: 50}, + config: toPrismaJson({}), + }, + }); + + await prisma.workflowTransition.createMany({ + data: [ + {fromStepId: waitStep.id, toStepId: eventCondition.id, waitOutcome: WorkflowWaitOutcome.EVENT}, + {fromStepId: waitStep.id, toStepId: timeoutExit.id, waitOutcome: WorkflowWaitOutcome.TIMEOUT}, + {fromStepId: eventCondition.id, toStepId: eventExit.id, condition: toPrismaJson({branch: 'yes'})}, + {fromStepId: eventCondition.id, toStepId: wrongPlanExit.id, condition: toPrismaJson({branch: 'no'})}, + ], + }); + + // These durable rows are all a new worker needs after a restart mid-wait. + const execution = await prisma.workflowExecution.create({ + data: { + workflowId: workflow.id, + contactId: contact.id, + status: WorkflowExecutionStatus.WAITING, + currentStepId: waitStep.id, + context: toPrismaJson({source: 'signup', plan: 'free'}), + }, + }); + const waitExecution = await prisma.workflowStepExecution.create({ + data: { + executionId: execution.id, + stepId: waitStep.id, + status: StepExecutionStatus.WAITING, + startedAt: new Date(), + }, + }); + + await WorkflowExecutionService.handleEvent(projectId, 'purchase.completed', contact.id, { + plan: 'pro', + orderId: 'order-123', + }); + + const resumed = await prisma.workflowExecution.findUniqueOrThrow({where: {id: execution.id}}); + expect(resumed.context).toEqual({source: 'signup', plan: 'pro', orderId: 'order-123'}); + + const stepExecutions = await prisma.workflowStepExecution.findMany({ + where: {executionId: execution.id}, + include: {step: true}, + }); + expect(stepExecutions.some(item => item.step.id === eventExit.id)).toBe(true); + expect(stepExecutions.some(item => item.step.id === timeoutExit.id)).toBe(false); + expect(stepExecutions.some(item => item.step.id === wrongPlanExit.id)).toBe(false); + expect( + (stepExecutions.find(item => item.step.id === eventCondition.id)?.output as {branch?: string} | null)?.branch, + ).toBe('yes'); + expect( + (await prisma.workflowStepExecution.findUniqueOrThrow({where: {id: waitExecution.id}})).output, + ).toMatchObject({waitOutcome: WorkflowWaitOutcome.EVENT, eventData: {plan: 'pro', orderId: 'order-123'}}); + }); + + it('should follow only the timeout route when a persisted wait expires', async () => { + const contact = await factories.createContact({projectId}); + const workflow = await factories.createWorkflow({projectId}); + const waitStep = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.WAIT_FOR_EVENT, + name: 'Wait for Purchase', + position: {x: 100, y: 0}, + config: toPrismaJson({eventName: 'purchase.completed', timeout: 3600}), + }, + }); + const eventExit = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.EXIT, + name: 'Event path', + position: {x: 200, y: -50}, + config: toPrismaJson({}), + }, + }); + const timeoutExit = await prisma.workflowStep.create({ + data: { + workflowId: workflow.id, + type: WorkflowStepType.EXIT, + name: 'Timeout path', + position: {x: 200, y: 50}, + config: toPrismaJson({}), + }, + }); + await prisma.workflowTransition.createMany({ + data: [ + {fromStepId: waitStep.id, toStepId: eventExit.id, waitOutcome: WorkflowWaitOutcome.EVENT}, + {fromStepId: waitStep.id, toStepId: timeoutExit.id, waitOutcome: WorkflowWaitOutcome.TIMEOUT}, + ], + }); + const execution = await prisma.workflowExecution.create({ + data: { + workflowId: workflow.id, + contactId: contact.id, + status: WorkflowExecutionStatus.WAITING, + currentStepId: waitStep.id, + context: toPrismaJson({source: 'signup'}), + }, + }); + const waitExecution = await prisma.workflowStepExecution.create({ + data: { + executionId: execution.id, + stepId: waitStep.id, + status: StepExecutionStatus.WAITING, + startedAt: new Date(), + }, + }); + + await WorkflowExecutionService.processTimeout(execution.id, waitStep.id, waitExecution.id); + + const stepExecutions = await prisma.workflowStepExecution.findMany({ + where: {executionId: execution.id}, + select: {stepId: true}, + }); + expect(stepExecutions.some(item => item.stepId === timeoutExit.id)).toBe(true); + expect(stepExecutions.some(item => item.stepId === eventExit.id)).toBe(false); + expect( + (await prisma.workflowStepExecution.findUniqueOrThrow({where: {id: waitExecution.id}})).output, + ).toMatchObject({waitOutcome: WorkflowWaitOutcome.TIMEOUT, timedOut: true}); + }); }); // ======================================== diff --git a/apps/api/src/services/__tests__/WorkflowService.test.ts b/apps/api/src/services/__tests__/WorkflowService.test.ts index 412bb5c1f..e4f645fcf 100644 --- a/apps/api/src/services/__tests__/WorkflowService.test.ts +++ b/apps/api/src/services/__tests__/WorkflowService.test.ts @@ -1,5 +1,5 @@ import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest'; -import {WorkflowExecutionStatus, WorkflowStepType, WorkflowTriggerType} from '@plunk/db'; +import {WorkflowExecutionStatus, WorkflowStepType, WorkflowTriggerType, WorkflowWaitOutcome} from '@plunk/db'; import {WorkflowService} from '../WorkflowService'; import {Keys} from '../keys'; import {factories, getPrismaClient} from '../../../../../test/helpers'; @@ -426,9 +426,9 @@ describe('WorkflowService', () => { const mine = await factories.createWorkflow({projectId}); const foreign = await factories.createWorkflow({projectId: otherProject.id}); - await expect( - WorkflowService.bulkUpdate(projectId, {ids: [mine.id, foreign.id], delete: true}), - ).rejects.toThrow(/not found in this project/i); + await expect(WorkflowService.bulkUpdate(projectId, {ids: [mine.id, foreign.id], delete: true})).rejects.toThrow( + /not found in this project/i, + ); // Both survive — atomic rollback, no cross-project leak. expect(await prisma.workflow.findUnique({where: {id: mine.id}})).not.toBeNull(); @@ -443,9 +443,9 @@ describe('WorkflowService', () => { status: WorkflowExecutionStatus.RUNNING, }); - await expect( - WorkflowService.bulkUpdate(projectId, {ids: [busy.id, idle.id], delete: true}), - ).rejects.toThrow(/active execution/i); + await expect(WorkflowService.bulkUpdate(projectId, {ids: [busy.id, idle.id], delete: true})).rejects.toThrow( + /active execution/i, + ); // Partial delete must be impossible — the idle workflow must survive too. expect(await prisma.workflow.findUnique({where: {id: busy.id}})).not.toBeNull(); @@ -754,6 +754,83 @@ describe('WorkflowService', () => { expect(noTransition.condition).toEqual({branch: 'no'}); }); + it('should create distinct event and timeout routes from WAIT_FOR_EVENT steps', async () => { + const workflow = await factories.createWorkflow({projectId}); + const waitStep = await factories.createWorkflowStep({ + workflowId: workflow.id, + type: WorkflowStepType.WAIT_FOR_EVENT, + config: {eventName: 'purchase.completed', timeout: 3600}, + }); + const eventStep = await factories.createWorkflowStep({workflowId: workflow.id}); + const timeoutStep = await factories.createWorkflowStep({workflowId: workflow.id}); + + const eventTransition = await WorkflowService.createTransition(projectId, workflow.id, { + fromStepId: waitStep.id, + toStepId: eventStep.id, + waitOutcome: WorkflowWaitOutcome.EVENT, + }); + const timeoutTransition = await WorkflowService.createTransition(projectId, workflow.id, { + fromStepId: waitStep.id, + toStepId: timeoutStep.id, + waitOutcome: WorkflowWaitOutcome.TIMEOUT, + }); + + expect(eventTransition.waitOutcome).toBe(WorkflowWaitOutcome.EVENT); + expect(timeoutTransition.waitOutcome).toBe(WorkflowWaitOutcome.TIMEOUT); + expect(eventTransition.condition).toBeNull(); + expect(timeoutTransition.condition).toBeNull(); + }); + + it('should reject duplicate or condition-based WAIT_FOR_EVENT routes', async () => { + const workflow = await factories.createWorkflow({projectId}); + const waitStep = await factories.createWorkflowStep({ + workflowId: workflow.id, + type: WorkflowStepType.WAIT_FOR_EVENT, + }); + const step2 = await factories.createWorkflowStep({workflowId: workflow.id}); + const step3 = await factories.createWorkflowStep({workflowId: workflow.id}); + + await WorkflowService.createTransition(projectId, workflow.id, { + fromStepId: waitStep.id, + toStepId: step2.id, + waitOutcome: WorkflowWaitOutcome.EVENT, + }); + + await expect( + WorkflowService.createTransition(projectId, workflow.id, { + fromStepId: waitStep.id, + toStepId: step3.id, + waitOutcome: WorkflowWaitOutcome.EVENT, + }), + ).rejects.toThrow(/already has a transition/i); + + await expect( + WorkflowService.createTransition(projectId, workflow.id, { + fromStepId: waitStep.id, + toStepId: step3.id, + condition: {branch: 'timeout'}, + }), + ).rejects.toThrow(/waitOutcome/i); + }); + + it('should reject a timeout route when the wait has no timeout', async () => { + const workflow = await factories.createWorkflow({projectId}); + const waitStep = await factories.createWorkflowStep({ + workflowId: workflow.id, + type: WorkflowStepType.WAIT_FOR_EVENT, + config: {eventName: 'purchase.completed'}, + }); + const timeoutStep = await factories.createWorkflowStep({workflowId: workflow.id}); + + await expect( + WorkflowService.createTransition(projectId, workflow.id, { + fromStepId: waitStep.id, + toStepId: timeoutStep.id, + waitOutcome: WorkflowWaitOutcome.TIMEOUT, + }), + ).rejects.toThrow(/positive timeout/i); + }); + it('should reject a second outgoing transition from a non-condition step', async () => { const workflow = await factories.createWorkflow({projectId}); const step1 = await factories.createWorkflowStep({workflowId: workflow.id, type: WorkflowStepType.DELAY}); diff --git a/apps/web/src/components/WorkflowBuilder.tsx b/apps/web/src/components/WorkflowBuilder.tsx index e772b05c4..35cd4da00 100644 --- a/apps/web/src/components/WorkflowBuilder.tsx +++ b/apps/web/src/components/WorkflowBuilder.tsx @@ -51,12 +51,14 @@ interface WorkflowBuilderProps { id: string; toStepId: string; condition: unknown; + waitOutcome: 'EVENT' | 'TIMEOUT' | null; priority: number; }>; incomingTransitions: Array<{ id: string; fromStepId: string; condition: unknown; + waitOutcome: 'EVENT' | 'TIMEOUT' | null; priority: number; }>; })[]; @@ -156,6 +158,37 @@ function getBranchColor(config: any, branchId: string): string { return branchId === 'yes' ? '#16a34a' : '#dc2626'; } +type WaitOutcome = 'EVENT' | 'TIMEOUT'; +const WAIT_OUTCOMES: WaitOutcome[] = ['EVENT', 'TIMEOUT']; + +function waitHasTimeout(config: any): boolean { + return typeof config?.timeout === 'number' && config.timeout > 0; +} + +function getWaitOutcomeHandles(config: any, transitions: Array<{waitOutcome: WaitOutcome | null}> = []): WaitOutcome[] { + const expected = waitHasTimeout(config) ? WAIT_OUTCOMES : WAIT_OUTCOMES.slice(0, 1); + const wired = transitions.map(t => t.waitOutcome).filter((outcome): outcome is WaitOutcome => Boolean(outcome)); + + return [...expected, ...wired.filter(outcome => !expected.includes(outcome))]; +} + +function getWaitOutcomeLabel(outcome: WaitOutcome): string { + return outcome === 'EVENT' ? 'Event received' : 'Timed out'; +} + +function getWaitOutcomeColor(outcome: WaitOutcome): string { + return outcome === 'EVENT' ? '#16a34a' : '#d97706'; +} + +function getRoutePriority(stepType: string | undefined, config: any, outcome?: string | null): number { + if (!outcome) return 0; + + const routes = + stepType === 'WAIT_FOR_EVENT' ? WAIT_OUTCOMES : stepType === 'CONDITION' ? getExpectedBranches(config) : []; + const index = routes.findIndex(route => route === outcome); + return index >= 0 ? index : 0; +} + // Dagre layout function function getLayoutedElements(nodes: Node[], edges: Edge[]) { const dagreGraph = new dagre.graphlib.Graph(); @@ -289,9 +322,9 @@ function CustomNode({ template?: {id: string; name: string}; config?: any; /** - * One entry per outgoing slot. Condition steps get a handle per branch so a - * dragged connection carries the branch it started from; everything else has - * a single unnamed slot. Empty for EXIT, which ends the path. + * One entry per outgoing slot. Conditions and event waits get named handles + * so a dragged connection carries its route; linear steps get one unnamed + * slot. Empty for EXIT, which ends the path. */ sourceHandles?: {id: string; color: string; connectable: boolean}[]; }; @@ -305,12 +338,7 @@ function CustomNode({ return ( <> - +
{ const step = steps.find(s => s.id === stepId); - const isCondition = step?.type === 'CONDITION'; + const isBranching = step?.type === 'CONDITION' || step?.type === 'WAIT_FOR_EVENT'; const hasChildren = (step?.outgoingTransitions?.length ?? 0) > 0; - setDeleteMode(isCondition || !hasChildren ? 'cascade' : 'splice'); + setDeleteMode(isBranching || !hasChildren ? 'cascade' : 'splice'); setStepToDelete(stepId); setShowDeleteDialog(true); }, @@ -577,8 +605,8 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr const color = STEP_TYPE_COLORS[step.type as keyof typeof STEP_TYPE_COLORS] || '#6b7280'; const bgColor = STEP_TYPE_BG[step.type as keyof typeof STEP_TYPE_BG] || '#f3f4f6'; - // EXIT ends the path, a condition routes per branch, everything else has a - // single next step. + // EXIT ends the path. Conditions and event waits expose named routes; + // every other step has one unnamed next step. const outgoing = step.outgoingTransitions ?? []; const sourceHandles = step.type === 'EXIT' @@ -589,7 +617,13 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr color: getBranchColor(step.config, branchId), connectable: !outgoing.some(t => getTransitionBranch(t.condition) === branchId), })) - : [{id: SOURCE_HANDLE_ID, color: '#94a3b8', connectable: outgoing.length === 0}]; + : step.type === 'WAIT_FOR_EVENT' + ? getWaitOutcomeHandles(step.config, outgoing).map(outcome => ({ + id: outcome, + color: getWaitOutcomeColor(outcome), + connectable: !outgoing.some(t => t.waitOutcome === outcome), + })) + : [{id: SOURCE_HANDLE_ID, color: '#94a3b8', connectable: outgoing.length === 0}]; return { id: step.id, @@ -631,13 +665,28 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr draggable: false, data: { label: getBranchLabel(step.config, branchId), - onClick: () => setPickerContext({mode: 'append', fromStepId: step.id, branch: branchId}), + onClick: () => setPickerContext({mode: 'append', fromStepId: step.id, outcome: branchId}), + }, + }); + } + }); + } else if (step.type === 'WAIT_FOR_EVENT') { + getWaitOutcomeHandles(step.config, step.outgoingTransitions ?? []).forEach(outcome => { + if (!step.outgoingTransitions?.some(t => t.waitOutcome === outcome)) { + nodes.push({ + id: `${step.id}-add-${outcome}`, + type: 'addStep', + position: {x: 0, y: 0}, + draggable: false, + data: { + label: getWaitOutcomeLabel(outcome), + onClick: () => setPickerContext({mode: 'append', fromStepId: step.id, outcome}), }, }); } }); } else { - // For non-condition steps, add + node if no outgoing transitions + // For linear steps, add + node if no outgoing transitions. if (!step.outgoingTransitions || step.outgoingTransitions.length === 0) { nodes.push({ id: `${step.id}-add`, @@ -664,26 +713,34 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr if (step.outgoingTransitions && step.outgoingTransitions.length > 0) { step.outgoingTransitions.forEach(transition => { const condition = transition.condition; - const isConditional = condition && typeof condition === 'object' && 'branch' in condition; - const branch = + const conditionBranch = condition && typeof condition === 'object' && 'branch' in condition ? (condition.branch as string) : undefined; - - const branchColor = branch ? getBranchColor(step.config, branch) : '#94a3b8'; - const branchLabel = branch ? getBranchLabel(step.config, branch) : undefined; + const waitOutcome = step.type === 'WAIT_FOR_EVENT' ? transition.waitOutcome : undefined; + const route = conditionBranch ?? waitOutcome ?? undefined; + const routeColor = conditionBranch + ? getBranchColor(step.config, conditionBranch) + : waitOutcome + ? getWaitOutcomeColor(waitOutcome) + : '#64748b'; + const routeLabel = conditionBranch + ? getBranchLabel(step.config, conditionBranch) + : waitOutcome + ? getWaitOutcomeLabel(waitOutcome) + : undefined; edges.push({ id: transition.id, source: step.id, target: transition.toStepId, - sourceHandle: branch ?? SOURCE_HANDLE_ID, + sourceHandle: route ?? SOURCE_HANDLE_ID, targetHandle: TARGET_HANDLE_ID, type: 'workflow', animated: false, data: { - branchLabel: isConditional ? branchLabel : undefined, - branchColor: isConditional ? branchColor : '#64748b', + branchLabel: routeLabel, + branchColor: routeColor, onInsert: () => setPickerContext({mode: 'insert', transitionId: transition.id}), onDisconnect: () => setTransitionToDisconnect(transition.id), }, @@ -738,6 +795,30 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr }); } }); + } else if (step.type === 'WAIT_FOR_EVENT') { + getWaitOutcomeHandles(step.config, step.outgoingTransitions ?? []).forEach(outcome => { + if (!step.outgoingTransitions?.some(t => t.waitOutcome === outcome)) { + const color = getWaitOutcomeColor(outcome); + const label = getWaitOutcomeLabel(outcome); + + edges.push({ + id: `${step.id}-add-${outcome}-edge`, + source: step.id, + target: `${step.id}-add-${outcome}`, + sourceHandle: outcome, + targetHandle: TARGET_HANDLE_ID, + type: 'smoothstep', + animated: false, + label, + labelStyle: {fill: color, fontWeight: 600, fontSize: 12}, + labelBgStyle: {fill: '#fff', fillOpacity: 0.95}, + labelBgPadding: [8, 4] as [number, number], + labelBgBorderRadius: 4, + style: {stroke: '#94a3b8', strokeWidth: 2, strokeDasharray: '5,5'}, + markerEnd: {type: MarkerType.ArrowClosed, color: '#94a3b8', width: 20, height: 20}, + }); + } + }); } else { if (!step.outgoingTransitions || step.outgoingTransitions.length === 0) { edges.push({ @@ -767,7 +848,6 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr const [nodes, setNodes, onNodesChange] = useNodesState(layoutedNodes); const [edges, setEdges, onEdgesChange] = useEdgesState(layoutedEdges); - // Update nodes/edges when layout changes useEffect(() => { setNodes(layoutedNodes); @@ -816,31 +896,38 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr const addStepContext = pickerContext; try { - // Validate that this branch doesn't already have a transition + // Validate that this named route doesn't already have a transition. const fromStep = steps.find(s => s.id === addStepContext.fromStepId); if (!fromStep) { toast.error('Parent step not found'); return; } - // For CONDITION steps, check if the branch already exists - if (fromStep.type === 'CONDITION' && addStepContext.branch) { + if (fromStep.type === 'CONDITION' && addStepContext.outcome) { const existingBranchTransition = fromStep.outgoingTransitions?.find(t => { const condition = t.condition; return ( condition && typeof condition === 'object' && 'branch' in condition && - condition.branch === addStepContext.branch + condition.branch === addStepContext.outcome ); }); if (existingBranchTransition) { - toast.error(`The ${addStepContext.branch} branch already has a connection`); + toast.error(`The ${addStepContext.outcome} branch already has a connection`); return; } } + if ( + fromStep.type === 'WAIT_FOR_EVENT' && + addStepContext.outcome && + fromStep.outgoingTransitions?.some(t => t.waitOutcome === addStepContext.outcome) + ) { + toast.error(`The ${addStepContext.outcome} route already has a connection`); + return; + } - // Create the new step (autoConnect: false because we manually create the transition with branch info) + // Create the new step; the named route is connected explicitly below. const newStep = await network.fetch( 'POST', `/workflows/${workflowId}/steps`, @@ -849,17 +936,17 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr name: `New ${stepType.toLowerCase().replace('_', ' ')}`, position: {x: 0, y: 0}, // Will be auto-positioned by dagre layout config: {}, - autoConnect: false, // We manually create transitions to preserve branch information + autoConnect: false, }, ); const newStepId = newStep.id; - // Create the transition with proper condition - const condition = addStepContext.branch ? {branch: addStepContext.branch} : null; - const expectedBranches = fromStep.type === 'CONDITION' ? getExpectedBranches(fromStep.config) : []; - const branchIndex = addStepContext.branch ? expectedBranches.indexOf(addStepContext.branch) : -1; - const priority = branchIndex >= 0 ? branchIndex : 0; + const condition = + fromStep.type === 'CONDITION' && addStepContext.outcome ? {branch: addStepContext.outcome} : null; + const waitOutcome = + fromStep.type === 'WAIT_FOR_EVENT' ? (addStepContext.outcome as WaitOutcome | undefined) : undefined; + const priority = getRoutePriority(fromStep.type, fromStep.config, addStepContext.outcome); await network.fetch( 'POST', @@ -868,6 +955,7 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr fromStepId: addStepContext.fromStepId, toStepId: newStepId, condition, + waitOutcome, priority, }, ); @@ -1013,6 +1101,8 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr // The slot being dragged from has to be free if (fromStep.type === 'CONDITION') { if (fromStep.outgoingTransitions?.some(t => getTransitionBranch(t.condition) === sourceHandle)) return false; + } else if (fromStep.type === 'WAIT_FOR_EVENT') { + if (fromStep.outgoingTransitions?.some(t => t.waitOutcome === sourceHandle)) return false; } else if ((fromStep.outgoingTransitions?.length ?? 0) > 0) { return false; } @@ -1030,8 +1120,8 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr const fromStep = steps.find(s => s.id === source); const branch = fromStep?.type === 'CONDITION' ? sourceHandle : null; - const expectedBranches = fromStep?.type === 'CONDITION' ? getExpectedBranches(fromStep.config) : []; - const branchIndex = branch ? expectedBranches.indexOf(branch) : -1; + const waitOutcome = fromStep?.type === 'WAIT_FOR_EVENT' ? (sourceHandle as WaitOutcome | null) : null; + const priority = getRoutePriority(fromStep?.type, fromStep?.config, branch ?? waitOutcome); try { await network.fetch( @@ -1041,7 +1131,8 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr fromStepId: source, toStepId: target, condition: branch ? {branch} : null, - priority: branchIndex >= 0 ? branchIndex : 0, + waitOutcome: waitOutcome ?? undefined, + priority, }, ); @@ -1060,8 +1151,11 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr try { const fromStep = steps.find(s => s.id === pickerContext.fromStepId); - const expectedBranches = fromStep?.type === 'CONDITION' ? getExpectedBranches(fromStep.config) : []; - const branchIndex = pickerContext.branch ? expectedBranches.indexOf(pickerContext.branch) : -1; + const priority = getRoutePriority(fromStep?.type, fromStep?.config, pickerContext.outcome); + const condition = + fromStep?.type === 'CONDITION' && pickerContext.outcome ? {branch: pickerContext.outcome} : null; + const waitOutcome = + fromStep?.type === 'WAIT_FOR_EVENT' ? (pickerContext.outcome as WaitOutcome | undefined) : undefined; await network.fetch( 'POST', @@ -1069,8 +1163,9 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr { fromStepId: pickerContext.fromStepId, toStepId, - condition: pickerContext.branch ? {branch: pickerContext.branch} : null, - priority: branchIndex >= 0 ? branchIndex : 0, + condition, + waitOutcome, + priority, }, ); @@ -1084,7 +1179,6 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr [pickerContext, steps, workflowId, onUpdate], ); - if (steps.length === 0) { return (
@@ -1097,12 +1191,7 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr return ( <> - {isExpanded && ( -
setIsExpanded(false)} - /> - )} + {isExpanded &&
setIsExpanded(false)} />}
- + { const step = steps.find(s => s.id === node.id); @@ -1150,10 +1236,7 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr className="bg-white border border-neutral-200 rounded-lg shadow-md" maskColor="rgba(0, 0, 0, 0.05)" /> - +
@@ -1185,8 +1268,8 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr
- Click a + to add a step, hover a connection to insert or disconnect, or drag between the dots to - connect two steps. + Click a + to add a step, hover a connection to insert or disconnect, or drag between the dots to connect + two steps.
@@ -1314,9 +1397,10 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr const affectedSteps = getAffectedSteps(stepToDelete); const stepToDeleteData = steps.find(s => s.id === stepToDelete); const downstreamSteps = affectedSteps.filter(s => s.id !== stepToDelete); - const isCondition = stepToDeleteData?.type === 'CONDITION'; + const isBranching = + stepToDeleteData?.type === 'CONDITION' || stepToDeleteData?.type === 'WAIT_FOR_EVENT'; const hasChildren = downstreamSteps.length > 0; - const canSplice = !isCondition && hasChildren; + const canSplice = !isBranching && hasChildren; return ( @@ -1352,7 +1436,8 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr }`} >

- Delete with {downstreamSteps.length} downstream {downstreamSteps.length === 1 ? 'step' : 'steps'} + Delete with {downstreamSteps.length} downstream{' '} + {downstreamSteps.length === 1 ? 'step' : 'steps'}

Permanently removes this step and everything below it. This cannot be undone. @@ -1360,11 +1445,11 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr

- ) : isCondition && hasChildren ? ( + ) : isBranching && hasChildren ? (

- Removing a condition step will also delete all {downstreamSteps.length} downstream{' '} - {downstreamSteps.length === 1 ? 'step' : 'steps'} across its branches: + Removing this branching step will also delete all {downstreamSteps.length} downstream{' '} + {downstreamSteps.length === 1 ? 'step' : 'steps'} across its routes:

    {downstreamSteps.map(step => ( @@ -1376,7 +1461,9 @@ export function WorkflowBuilder({workflowId, steps, onUpdate}: WorkflowBuilderPr

    This action cannot be undone.

) : ( -

This step is removed from the workflow. This can't be undone.

+

+ This step is removed from the workflow. This can't be undone. +

)} diff --git a/apps/web/src/components/WorkflowVisualizer.tsx b/apps/web/src/components/WorkflowVisualizer.tsx index ee61c5c8e..e051ec2ae 100644 --- a/apps/web/src/components/WorkflowVisualizer.tsx +++ b/apps/web/src/components/WorkflowVisualizer.tsx @@ -26,12 +26,14 @@ interface WorkflowVisualizerProps { id: string; toStepId: string; condition: unknown; + waitOutcome: 'EVENT' | 'TIMEOUT' | null; priority: number; }>; incomingTransitions: Array<{ id: string; fromStepId: string; condition: unknown; + waitOutcome: 'EVENT' | 'TIMEOUT' | null; priority: number; }>; })[]; @@ -325,6 +327,27 @@ export function WorkflowVisualizer({steps}: WorkflowVisualizerProps) { transition.condition && typeof transition.condition === 'object' && 'branch' in transition.condition ? transition.condition.branch : undefined; + const waitOutcome = step.type === 'WAIT_FOR_EVENT' ? transition.waitOutcome : null; + const routeLabel = isConditional + ? branch === 'yes' + ? 'Yes' + : 'No' + : waitOutcome === 'EVENT' + ? 'Event received' + : waitOutcome === 'TIMEOUT' + ? 'Timed out' + : undefined; + const routeColor = isConditional + ? branch === 'yes' + ? '#16a34a' + : branch === 'no' + ? '#dc2626' + : '#64748b' + : waitOutcome === 'EVENT' + ? '#16a34a' + : waitOutcome === 'TIMEOUT' + ? '#d97706' + : '#64748b'; edges.push({ id: transition.id, @@ -332,9 +355,9 @@ export function WorkflowVisualizer({steps}: WorkflowVisualizerProps) { target: transition.toStepId, type: 'smoothstep', animated: false, - label: isConditional ? (branch === 'yes' ? 'Yes' : 'No') : undefined, + label: routeLabel, labelStyle: { - fill: branch === 'yes' ? '#16a34a' : branch === 'no' ? '#dc2626' : '#64748b', + fill: routeColor, fontWeight: 600, fontSize: 12, }, diff --git a/apps/web/src/pages/workflows/[id].tsx b/apps/web/src/pages/workflows/[id].tsx index 2d4a2735e..4029d9a29 100644 --- a/apps/web/src/pages/workflows/[id].tsx +++ b/apps/web/src/pages/workflows/[id].tsx @@ -273,6 +273,23 @@ export default function WorkflowEditorPage() { } } } + + if (step.type === 'WAIT_FOR_EVENT' && step.outgoingTransitions) { + const config = step.config && typeof step.config === 'object' && !Array.isArray(step.config) ? step.config : {}; + const expectedOutcomes = + typeof config.timeout === 'number' && config.timeout > 0 ? ['EVENT', 'TIMEOUT'] : ['EVENT']; + const missingOutcomes = expectedOutcomes.filter( + outcome => !step.outgoingTransitions.some(transition => transition.waitOutcome === outcome), + ); + + if (missingOutcomes.length > 0) { + errors.push( + `"${step.name}" wait step is missing connections for: ${missingOutcomes + .map(outcome => (outcome === 'EVENT' ? 'Event received' : 'Timed out')) + .join(', ')}`, + ); + } + } }); return {valid: errors.length === 0, errors}; @@ -893,4 +910,3 @@ function SettingsDialog({workflow, open, onOpenChange, onSave}: SettingsDialogPr ); } - diff --git a/apps/wiki/content/docs/api-reference/overview.mdx b/apps/wiki/content/docs/api-reference/overview.mdx index c37e4f9ad..ffc0c0fa0 100644 --- a/apps/wiki/content/docs/api-reference/overview.mdx +++ b/apps/wiki/content/docs/api-reference/overview.mdx @@ -291,7 +291,7 @@ Each workflow consists of a **workflow** record, a graph of **steps**, **transit | PATCH | `/workflows/:id/steps/:stepId` | Update a step's config. | | DELETE | `/workflows/:id/steps/:stepId?splice=true` | Delete a step. Pass `splice=true` to auto-reconnect surrounding transitions. | | **Transitions** | | | -| POST | `/workflows/:id/transitions` | Add a transition between two steps. For `CONDITION` steps, include `branch: "yes" \| "no"`. | +| POST | `/workflows/:id/transitions` | Add a transition. Use `condition.branch` for a `CONDITION` route or `waitOutcome: "EVENT" \| "TIMEOUT"` for a `WAIT_FOR_EVENT` route. | | DELETE | `/workflows/:id/transitions/:transitionId` | Delete a transition. | | **Executions** | | | | POST | `/workflows/:id/executions` | Manually start an execution for a contact. Optional `context` JSON for per-execution variables. | diff --git a/apps/wiki/content/docs/concepts/workflows.mdx b/apps/wiki/content/docs/concepts/workflows.mdx index 3dc68ff20..2dbeb811c 100644 --- a/apps/wiki/content/docs/concepts/workflows.mdx +++ b/apps/wiki/content/docs/concepts/workflows.mdx @@ -35,7 +35,7 @@ A workflow always begins with a single auto-created `TRIGGER` step. You build th | `TRIGGER` | Auto-created entry point. Holds the trigger configuration. | `eventName` (for `EVENT` trigger) | | `SEND_EMAIL` | Sends an email to the contact using a template. Template variables resolve from contact data + execution context. | `templateId`, optional `from` override | | `DELAY` | Pauses the execution for a fixed duration before continuing. | `amount`, `unit` (`minutes` / `hours` / `days`) | -| `WAIT_FOR_EVENT` | Pauses until a specified event is tracked on the contact, with a timeout fallback. | `eventName`, `timeout` (seconds) | +| `WAIT_FOR_EVENT` | Pauses until a specified event is tracked on the contact. Event arrival and timeout have distinct outgoing routes. | `eventName`, optional `timeout` (seconds) | | `CONDITION` | Branches the execution based on contact data or event data. Each `CONDITION` step has two outgoing transitions tagged `yes` / `no`. | A filter expression (same shape as segment filters) | | `WEBHOOK` | Calls an external HTTPS endpoint with contact + execution context as the JSON body. `url`, header values, and `body` support `{{variables}}`. | `url`, optional `method`, `headers`, `body` | | `UPDATE_CONTACT` | Patches contact data — useful for tagging contacts as they progress (`{ stage: "activated" }`). | `data` object | @@ -43,9 +43,9 @@ A workflow always begins with a single auto-created `TRIGGER` step. You build th ## Transitions -Transitions are the edges of the workflow graph. For most step types a transition is a simple "next" pointer. For `CONDITION` steps, each transition carries a `branch` discriminator (`"yes"` or `"no"`) so the engine knows which path to take when the condition evaluates. +Transitions are the edges of the workflow graph. For most step types a transition is a simple "next" pointer. `CONDITION` transitions carry their branch in `condition`; `WAIT_FOR_EVENT` transitions use a separate `waitOutcome` of `EVENT` or `TIMEOUT`. A timeout route is available only when the wait step has a positive timeout. -When deleting a step in the middle of a chain, pass `?splice=true` on `DELETE /workflows/:id/steps/:stepId` to automatically reconnect the surrounding transitions; otherwise the deletion leaves the graph disconnected. +When deleting a non-branching step in the middle of a chain, pass `?splice=true` on `DELETE /workflows/:id/steps/:stepId` to automatically reconnect the surrounding transitions; otherwise the deletion leaves the graph disconnected. Condition and event-wait steps must be removed with their branches rather than spliced. ## Executions @@ -60,7 +60,7 @@ Each contact entering the workflow creates a `WorkflowExecution`. Executions mov | `FAILED` | An unrecoverable error occurred (e.g. webhook returned non-2xx after retries, template not found). | | `CANCELLED` | Cancelled manually via the API or dashboard. | -Each execution carries a `context` JSON object that's merged with the contact's `data` when rendering templates and evaluating conditions. When you start an execution manually with `POST /workflows/:id/executions`, you can pass an initial `context` — this is how you parameterise per-execution variables (a coupon code, a referrer name) without storing them on the contact. +Each execution carries a `context` JSON object that's merged with the contact's `data` when rendering templates and evaluating conditions. When you start an execution manually with `POST /workflows/:id/executions`, you can pass an initial `context` — this is how you parameterise per-execution variables (a coupon code, a referrer name) without storing them on the contact. When a `WAIT_FOR_EVENT` resumes, the arriving event data is shallow-merged into that context, replacing keys with the newer event values. Downstream email, condition, and webhook steps can read it through `event.*`. ## Locking active workflows diff --git a/packages/db/prisma/migrations/20260826213000_add_workflow_wait_outcomes/migration.sql b/packages/db/prisma/migrations/20260826213000_add_workflow_wait_outcomes/migration.sql new file mode 100644 index 000000000..0b5a1edf9 --- /dev/null +++ b/packages/db/prisma/migrations/20260826213000_add_workflow_wait_outcomes/migration.sql @@ -0,0 +1,59 @@ +-- CreateEnum +CREATE TYPE "WorkflowWaitOutcome" AS ENUM ('EVENT', 'TIMEOUT'); + +-- AlterTable +ALTER TABLE "workflow_transitions" ADD COLUMN "waitOutcome" "WorkflowWaitOutcome"; + +-- Existing WAIT_FOR_EVENT transitions were linear: the same next step handled +-- both an arriving event and a configured timeout. Preserve that behavior while +-- moving the route out of the generic condition JSON. +UPDATE "workflow_transitions" AS transition +SET + "waitOutcome" = CASE + WHEN transition."condition"->>'branch' = 'timeout' + OR transition."condition"->>'fallback' = 'true' + THEN 'TIMEOUT'::"WorkflowWaitOutcome" + ELSE 'EVENT'::"WorkflowWaitOutcome" + END, + "condition" = NULL +FROM "workflow_steps" AS step +WHERE transition."fromStepId" = step."id" + AND step."type" = 'WAIT_FOR_EVENT'; + +-- A legacy unconditional wait transition was taken for either outcome. When a +-- timeout is configured, duplicate its target as the explicit timeout route. +INSERT INTO "workflow_transitions" ( + "id", + "fromStepId", + "toStepId", + "condition", + "waitOutcome", + "priority", + "createdAt", + "updatedAt" +) +SELECT + gen_random_uuid()::text, + transition."fromStepId", + transition."toStepId", + NULL, + 'TIMEOUT'::"WorkflowWaitOutcome", + transition."priority" + 1, + transition."createdAt", + transition."updatedAt" +FROM "workflow_transitions" AS transition +JOIN "workflow_steps" AS step ON step."id" = transition."fromStepId" +WHERE step."type" = 'WAIT_FOR_EVENT' + AND transition."waitOutcome" = 'EVENT' + AND jsonb_typeof(step."config"->'timeout') = 'number' + AND (step."config"->>'timeout')::double precision > 0 + AND NOT EXISTS ( + SELECT 1 + FROM "workflow_transitions" AS timeout_transition + WHERE timeout_transition."fromStepId" = transition."fromStepId" + AND timeout_transition."waitOutcome" = 'TIMEOUT' + ); + +-- CreateIndex +CREATE UNIQUE INDEX "workflow_transitions_fromStepId_waitOutcome_key" +ON "workflow_transitions"("fromStepId", "waitOutcome"); diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index 157efb7f6..1bf08db40 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -438,13 +438,15 @@ model WorkflowTransition { toStepId String // Conditional routing - condition Json? // null = always follow, or { branch: "yes" } for condition steps - priority Int @default(0) // Order to evaluate transitions + condition Json? // CONDITION steps: { branch: "yes" }; null for other step types + waitOutcome WorkflowWaitOutcome? // WAIT_FOR_EVENT steps: event arrival or timeout + priority Int @default(0) // Order to evaluate transitions // Timestamps createdAt DateTime @default(now()) updatedAt DateTime @updatedAt + @@unique([fromStepId, waitOutcome]) @@index([fromStepId]) @@index([toStepId]) @@map("workflow_transitions") @@ -754,6 +756,11 @@ enum WorkflowStepType { UPDATE_CONTACT // Update contact fields } +enum WorkflowWaitOutcome { + EVENT + TIMEOUT +} + enum WorkflowExecutionStatus { RUNNING // Currently executing WAITING // Waiting for event or delay diff --git a/packages/shared/src/schemas/index.ts b/packages/shared/src/schemas/index.ts index 0087d8ffa..e3648daa4 100644 --- a/packages/shared/src/schemas/index.ts +++ b/packages/shared/src/schemas/index.ts @@ -1,4 +1,4 @@ -import {CampaignAudienceType, TemplateType, TrackingMode, WorkflowStepType, WorkflowTriggerType} from '@plunk/db'; +import {CampaignAudienceType, TemplateType, TrackingMode, WorkflowStepType, WorkflowTriggerType, WorkflowWaitOutcome} from '@plunk/db'; import type {FilterCondition, FilterGroup} from '@plunk/types'; import {z} from 'zod'; @@ -263,6 +263,7 @@ export const WorkflowSchemas = { fromStepId: uuid, toStepId: uuid, condition: jsonSchema.optional(), + waitOutcome: z.nativeEnum(WorkflowWaitOutcome).optional(), priority: z.number().int().min(0).default(0), }), startExecution: z.object({ diff --git a/packages/types/src/prisma/extended.ts b/packages/types/src/prisma/extended.ts index 40e0cd65c..523176bec 100644 --- a/packages/types/src/prisma/extended.ts +++ b/packages/types/src/prisma/extended.ts @@ -64,6 +64,7 @@ export type WorkflowStepWithTransitions = WorkflowStep & { outgoingTransitions?: Array<{ id: string; condition: Prisma.JsonValue; + waitOutcome: WorkflowTransition['waitOutcome']; priority: number; toStep: WorkflowStep; }>;