975 lines
28 KiB
TypeScript
975 lines
28 KiB
TypeScript
import type {Workflow, WorkflowExecution, WorkflowStep, WorkflowTransition} from '@plunk/db';
|
|
import {Prisma, WorkflowExecutionStatus} from '@plunk/db';
|
|
import type {PaginatedResponse, WorkflowExecutionWithDetails, WorkflowWithDetails} from '@plunk/types';
|
|
import {toPrismaJson} from '@plunk/types';
|
|
import signale from 'signale';
|
|
|
|
import {prisma} from '../database/prisma.js';
|
|
import {HttpException} from '../exceptions/index.js';
|
|
|
|
import {ContactService} from './ContactService.js';
|
|
import {EventService} from './EventService.js';
|
|
import {NtfyService} from './NtfyService.js';
|
|
import {WorkflowExecutionService} from './WorkflowExecutionService.js';
|
|
|
|
export class WorkflowService {
|
|
/**
|
|
* Get all workflows for a project with pagination
|
|
*/
|
|
public static async list(
|
|
projectId: string,
|
|
page = 1,
|
|
pageSize = 20,
|
|
search?: string,
|
|
): Promise<PaginatedResponse<Workflow>> {
|
|
const skip = (page - 1) * pageSize;
|
|
|
|
const where: Prisma.WorkflowWhereInput = {
|
|
projectId,
|
|
...(search
|
|
? {
|
|
OR: [
|
|
{name: {contains: search, mode: 'insensitive' as const}},
|
|
{description: {contains: search, mode: 'insensitive' as const}},
|
|
],
|
|
}
|
|
: {}),
|
|
};
|
|
|
|
const [workflows, total] = await Promise.all([
|
|
prisma.workflow.findMany({
|
|
where,
|
|
skip,
|
|
take: pageSize,
|
|
orderBy: {createdAt: 'desc'},
|
|
include: {
|
|
_count: {
|
|
select: {
|
|
steps: true,
|
|
executions: true,
|
|
},
|
|
},
|
|
},
|
|
}),
|
|
prisma.workflow.count({where}),
|
|
]);
|
|
|
|
return {
|
|
data: workflows as Workflow[],
|
|
total,
|
|
page,
|
|
pageSize,
|
|
totalPages: Math.ceil(total / pageSize),
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get a single workflow by ID with all steps and transitions
|
|
*/
|
|
public static async get(projectId: string, workflowId: string): Promise<WorkflowWithDetails> {
|
|
const workflow = await prisma.workflow.findFirst({
|
|
where: {
|
|
id: workflowId,
|
|
projectId,
|
|
},
|
|
include: {
|
|
steps: {
|
|
include: {
|
|
template: {
|
|
select: {
|
|
id: true,
|
|
name: true,
|
|
},
|
|
},
|
|
outgoingTransitions: true,
|
|
incomingTransitions: true,
|
|
},
|
|
orderBy: {createdAt: 'asc'},
|
|
},
|
|
},
|
|
});
|
|
|
|
if (!workflow) {
|
|
throw new HttpException(404, 'Workflow not found');
|
|
}
|
|
|
|
return workflow;
|
|
}
|
|
|
|
/**
|
|
* Create a new workflow
|
|
*/
|
|
public static async create(
|
|
projectId: string,
|
|
data: {
|
|
name: string;
|
|
description?: string;
|
|
eventName: string;
|
|
enabled?: boolean;
|
|
allowReentry?: boolean;
|
|
},
|
|
): Promise<Workflow> {
|
|
if (!data.eventName?.trim()) {
|
|
throw new HttpException(400, 'Event name is required');
|
|
}
|
|
|
|
const workflow = await prisma.$transaction(async tx => {
|
|
const newWorkflow = await tx.workflow.create({
|
|
data: {
|
|
projectId,
|
|
name: data.name,
|
|
description: data.description,
|
|
triggerType: 'EVENT',
|
|
triggerConfig: {eventName: data.eventName.trim()},
|
|
enabled: data.enabled ?? false,
|
|
allowReentry: data.allowReentry ?? false,
|
|
},
|
|
});
|
|
|
|
await tx.workflowStep.create({
|
|
data: {
|
|
workflowId: newWorkflow.id,
|
|
type: 'TRIGGER',
|
|
name: `Trigger: ${data.eventName.trim()}`,
|
|
position: {x: 100, y: 100},
|
|
config: {eventName: data.eventName.trim()},
|
|
},
|
|
});
|
|
|
|
return newWorkflow;
|
|
});
|
|
|
|
if (workflow.enabled) {
|
|
await EventService.invalidateWorkflowCache(projectId);
|
|
}
|
|
|
|
// Get project name for notification
|
|
const project = await prisma.project.findUnique({
|
|
where: {id: projectId},
|
|
select: {name: true},
|
|
});
|
|
|
|
if (project) {
|
|
// Send notification about workflow creation
|
|
await NtfyService.notifyWorkflowCreated(workflow.name, project.name, projectId);
|
|
}
|
|
|
|
return workflow;
|
|
}
|
|
|
|
/**
|
|
* Update a workflow
|
|
*/
|
|
public static async update(
|
|
projectId: string,
|
|
workflowId: string,
|
|
data: {
|
|
name?: string;
|
|
description?: string;
|
|
triggerType?: Workflow['triggerType'];
|
|
triggerConfig?: Prisma.JsonValue;
|
|
enabled?: boolean;
|
|
allowReentry?: boolean;
|
|
},
|
|
): Promise<Workflow> {
|
|
// Verify workflow exists and belongs to project
|
|
const workflow = await this.get(projectId, workflowId);
|
|
|
|
// Check if workflow is enabled and has active executions
|
|
if (workflow.enabled) {
|
|
const activeExecutions = await this.hasActiveExecutions(workflowId);
|
|
|
|
if (activeExecutions > 0) {
|
|
// Block changes to trigger configuration while executions are running
|
|
const hasCriticalChanges = data.triggerType !== undefined || data.triggerConfig !== undefined;
|
|
|
|
if (hasCriticalChanges) {
|
|
throw new HttpException(
|
|
409,
|
|
`Cannot modify workflow trigger while workflow has ${activeExecutions} active execution(s). ` +
|
|
'Please disable the workflow first or wait for executions to complete. ' +
|
|
'You can still update name, description, and re-entry settings.',
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
const updateData: Prisma.WorkflowUpdateInput = {};
|
|
|
|
if (data.name !== undefined) updateData.name = data.name;
|
|
if (data.description !== undefined) updateData.description = data.description;
|
|
if (data.triggerType !== undefined) updateData.triggerType = data.triggerType;
|
|
if (data.triggerConfig !== undefined) {
|
|
updateData.triggerConfig = data.triggerConfig === null ? Prisma.JsonNull : data.triggerConfig;
|
|
}
|
|
if (data.enabled !== undefined) updateData.enabled = data.enabled;
|
|
if (data.allowReentry !== undefined) updateData.allowReentry = data.allowReentry;
|
|
|
|
const updated = await prisma.workflow.update({
|
|
where: {id: workflowId},
|
|
data: updateData,
|
|
include: {
|
|
project: {
|
|
select: {name: true},
|
|
},
|
|
},
|
|
});
|
|
|
|
// Invalidate workflow cache if enabled status changed or workflow is enabled
|
|
if (data.enabled !== undefined || updated.enabled) {
|
|
await EventService.invalidateWorkflowCache(projectId);
|
|
}
|
|
|
|
// Send notification if enabled status changed
|
|
if (data.enabled !== undefined && data.enabled !== workflow.enabled) {
|
|
if (data.enabled) {
|
|
await NtfyService.notifyWorkflowEnabled(updated.name, updated.project.name, projectId);
|
|
} else {
|
|
await NtfyService.notifyWorkflowDisabled(updated.name, updated.project.name, projectId);
|
|
}
|
|
}
|
|
|
|
return updated;
|
|
}
|
|
|
|
/**
|
|
* Delete a workflow
|
|
*/
|
|
public static async delete(projectId: string, workflowId: string): Promise<void> {
|
|
// Verify workflow exists and belongs to project - get with project name
|
|
const workflow = await prisma.workflow.findUnique({
|
|
where: {id: workflowId, projectId},
|
|
include: {
|
|
project: {
|
|
select: {name: true},
|
|
},
|
|
},
|
|
});
|
|
|
|
if (!workflow) {
|
|
throw new HttpException(404, 'Workflow not found');
|
|
}
|
|
|
|
// Check if workflow has active executions
|
|
const activeExecutions = await this.hasActiveExecutions(workflowId);
|
|
|
|
if (activeExecutions > 0) {
|
|
throw new HttpException(
|
|
409,
|
|
`Cannot delete workflow: it has ${activeExecutions} active execution(s). Please wait for them to complete or cancel them first.`,
|
|
);
|
|
}
|
|
|
|
await prisma.workflow.delete({
|
|
where: {id: workflowId},
|
|
});
|
|
|
|
// Invalidate workflow cache if workflow was enabled
|
|
if (workflow.enabled) {
|
|
await EventService.invalidateWorkflowCache(projectId);
|
|
}
|
|
|
|
// Send notification about workflow deletion
|
|
await NtfyService.notifyWorkflowDeleted(workflow.name, workflow.project.name, projectId);
|
|
}
|
|
|
|
/**
|
|
* Add a step to a workflow
|
|
*/
|
|
public static async addStep(
|
|
projectId: string,
|
|
workflowId: string,
|
|
data: {
|
|
type: WorkflowStep['type'];
|
|
name: string;
|
|
position: Prisma.JsonValue;
|
|
config: Prisma.JsonValue;
|
|
templateId?: string;
|
|
autoConnect?: boolean; // If true, automatically connect from the last step
|
|
},
|
|
): Promise<WorkflowStep> {
|
|
// Verify workflow exists and belongs to project
|
|
const workflow = await this.get(projectId, workflowId);
|
|
|
|
// Prevent adding duplicate TRIGGER steps
|
|
if (data.type === 'TRIGGER') {
|
|
const existingTrigger = workflow.steps.find(step => step.type === 'TRIGGER');
|
|
if (existingTrigger) {
|
|
throw new HttpException(400, 'Workflow already has a trigger step. Only one trigger is allowed per workflow.');
|
|
}
|
|
}
|
|
|
|
// Create the new step
|
|
const newStep = await prisma.workflowStep.create({
|
|
data: {
|
|
workflowId,
|
|
type: data.type,
|
|
name: data.name,
|
|
position: toPrismaJson(data.position),
|
|
config: toPrismaJson(data.config),
|
|
templateId: data.templateId,
|
|
},
|
|
});
|
|
|
|
// Auto-connect: If enabled (default true), create a transition from the last step to this new step
|
|
// This is useful for linear workflows where steps are added sequentially
|
|
const shouldAutoConnect = data.autoConnect !== false; // Default to true
|
|
|
|
if (shouldAutoConnect && workflow.steps.length > 0) {
|
|
// Find the last step that doesn't have any outgoing transitions (the "leaf" step)
|
|
// This is typically the most recently added step
|
|
const stepsWithoutOutgoing = workflow.steps.filter(step => step.outgoingTransitions.length === 0);
|
|
|
|
if (stepsWithoutOutgoing.length > 0) {
|
|
// Connect from the last leaf step to the new step
|
|
const lastStep = stepsWithoutOutgoing[stepsWithoutOutgoing.length - 1];
|
|
|
|
if (lastStep) {
|
|
await prisma.workflowTransition.create({
|
|
data: {
|
|
fromStepId: lastStep.id,
|
|
toStepId: newStep.id,
|
|
priority: 0,
|
|
},
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
return newStep;
|
|
}
|
|
|
|
/**
|
|
* Update a workflow step
|
|
*/
|
|
public static async updateStep(
|
|
projectId: string,
|
|
workflowId: string,
|
|
stepId: string,
|
|
data: {
|
|
name?: string;
|
|
position?: Prisma.JsonValue;
|
|
config?: Prisma.JsonValue;
|
|
templateId?: string | null;
|
|
},
|
|
): Promise<WorkflowStep> {
|
|
// First verify workflow belongs to project
|
|
const workflow = await this.get(projectId, workflowId);
|
|
|
|
// Then verify step exists and belongs to workflow
|
|
const step = await prisma.workflowStep.findUnique({
|
|
where: {id: stepId},
|
|
});
|
|
|
|
if (step?.workflowId !== workflowId) {
|
|
throw new HttpException(404, 'Workflow step not found');
|
|
}
|
|
|
|
// Check if workflow is enabled and has active executions
|
|
if (workflow.enabled) {
|
|
const activeExecutions = await this.hasActiveExecutions(workflowId);
|
|
|
|
if (activeExecutions > 0) {
|
|
// Only allow safe changes: name and position updates
|
|
const hasCriticalChanges = data.config !== undefined || data.templateId !== undefined;
|
|
|
|
if (hasCriticalChanges) {
|
|
throw new HttpException(
|
|
409,
|
|
`Cannot modify step configuration while workflow has ${activeExecutions} active execution(s). ` +
|
|
'Please disable the workflow first or wait for executions to complete. ' +
|
|
'You can still update the step name and position.',
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
const updateData: Prisma.WorkflowStepUpdateInput = {};
|
|
|
|
if (data.name !== undefined) updateData.name = data.name;
|
|
if (data.position !== undefined) updateData.position = toPrismaJson(data.position);
|
|
if (data.config !== undefined) updateData.config = toPrismaJson(data.config);
|
|
if (data.templateId !== undefined) {
|
|
if (data.templateId === null) {
|
|
updateData.template = {disconnect: true};
|
|
} else {
|
|
updateData.template = {connect: {id: data.templateId}};
|
|
}
|
|
}
|
|
|
|
return prisma.workflowStep.update({
|
|
where: {id: stepId},
|
|
data: updateData,
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Delete a workflow step
|
|
*/
|
|
public static async deleteStep(projectId: string, workflowId: string, stepId: string): Promise<void> {
|
|
const workflow = await this.get(projectId, workflowId);
|
|
|
|
const step = await prisma.workflowStep.findUnique({
|
|
where: {id: stepId},
|
|
include: {
|
|
outgoingTransitions: true,
|
|
},
|
|
});
|
|
|
|
if (step?.workflowId !== workflowId) {
|
|
throw new HttpException(404, 'Workflow step not found');
|
|
}
|
|
|
|
// Prevent deletion of TRIGGER steps
|
|
if (step.type === 'TRIGGER') {
|
|
throw new HttpException(400, 'Cannot delete the trigger step. Every workflow must have a trigger.');
|
|
}
|
|
|
|
// Check if workflow is enabled and has active executions on this step or downstream
|
|
if (workflow.enabled) {
|
|
// Check if any active executions are currently on this step
|
|
const executionsOnStep = await prisma.workflowExecution.count({
|
|
where: {
|
|
workflowId,
|
|
currentStepId: stepId,
|
|
status: {
|
|
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
|
|
},
|
|
},
|
|
});
|
|
|
|
if (executionsOnStep > 0) {
|
|
throw new HttpException(
|
|
409,
|
|
`Cannot delete step "${step.name}" while ${executionsOnStep} execution(s) are currently on this step. ` +
|
|
'Please disable the workflow first or wait for executions to complete.',
|
|
);
|
|
}
|
|
|
|
// Also check downstream steps for active executions
|
|
const allSteps = await prisma.workflowStep.findMany({
|
|
where: {workflowId},
|
|
include: {outgoingTransitions: true},
|
|
});
|
|
|
|
// Build adjacency map
|
|
const adjacencyMap = new Map<string, string[]>();
|
|
for (const s of allSteps) {
|
|
adjacencyMap.set(
|
|
s.id,
|
|
s.outgoingTransitions.map(t => t.toStepId),
|
|
);
|
|
}
|
|
|
|
// Find all downstream steps
|
|
const downstreamSteps = new Set<string>([stepId]);
|
|
const queue = [stepId];
|
|
|
|
while (queue.length > 0) {
|
|
const currentStepId = queue.shift()!;
|
|
const outgoingStepIds = adjacencyMap.get(currentStepId) || [];
|
|
|
|
for (const nextStepId of outgoingStepIds) {
|
|
if (!downstreamSteps.has(nextStepId)) {
|
|
downstreamSteps.add(nextStepId);
|
|
queue.push(nextStepId);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Check if any active executions are on downstream steps
|
|
const executionsOnDownstream = await prisma.workflowExecution.count({
|
|
where: {
|
|
workflowId,
|
|
currentStepId: {in: Array.from(downstreamSteps)},
|
|
status: {
|
|
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
|
|
},
|
|
},
|
|
});
|
|
|
|
if (executionsOnDownstream > 0) {
|
|
throw new HttpException(
|
|
409,
|
|
`Cannot delete step "${step.name}" while ${executionsOnDownstream} execution(s) are on downstream steps. ` +
|
|
'Deleting this step would orphan those executions. ' +
|
|
'Please disable the workflow first or wait for executions to complete.',
|
|
);
|
|
}
|
|
}
|
|
|
|
// Find all downstream steps that need to be deleted (cascade)
|
|
// First, get all steps and transitions for this workflow to build a graph
|
|
const allSteps = await prisma.workflowStep.findMany({
|
|
where: {workflowId},
|
|
include: {outgoingTransitions: true},
|
|
});
|
|
|
|
// Build adjacency map for efficient traversal
|
|
const adjacencyMap = new Map<string, string[]>();
|
|
for (const s of allSteps) {
|
|
adjacencyMap.set(
|
|
s.id,
|
|
s.outgoingTransitions.map(t => t.toStepId),
|
|
);
|
|
}
|
|
|
|
// Use BFS to traverse the workflow graph and find all downstream steps
|
|
const stepsToDelete = new Set<string>([stepId]);
|
|
const queue = [stepId];
|
|
|
|
while (queue.length > 0) {
|
|
const currentStepId = queue.shift()!;
|
|
const outgoingStepIds = adjacencyMap.get(currentStepId) || [];
|
|
|
|
for (const nextStepId of outgoingStepIds) {
|
|
if (!stepsToDelete.has(nextStepId)) {
|
|
stepsToDelete.add(nextStepId);
|
|
queue.push(nextStepId);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Delete all affected steps (Prisma will cascade delete the transitions)
|
|
await prisma.workflowStep.deleteMany({
|
|
where: {
|
|
id: {in: Array.from(stepsToDelete)},
|
|
workflowId, // Safety check to ensure we only delete steps from this workflow
|
|
},
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Create a transition between two steps
|
|
*/
|
|
public static async createTransition(
|
|
projectId: string,
|
|
workflowId: string,
|
|
data: {
|
|
fromStepId: string;
|
|
toStepId: string;
|
|
condition?: Prisma.JsonValue;
|
|
priority?: number;
|
|
},
|
|
): Promise<WorkflowTransition> {
|
|
// Verify both steps belong to the workflow
|
|
const steps = await prisma.workflowStep.findMany({
|
|
where: {
|
|
id: {in: [data.fromStepId, data.toStepId]},
|
|
workflowId,
|
|
workflow: {projectId},
|
|
},
|
|
});
|
|
|
|
if (steps.length !== 2) {
|
|
throw new HttpException(404, 'One or both steps not found');
|
|
}
|
|
|
|
const fromStep = steps.find(s => s.id === data.fromStepId);
|
|
|
|
// 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<string, unknown>)
|
|
: 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`,
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
const newTransition = await prisma.workflowTransition.create({
|
|
data: {
|
|
fromStepId: data.fromStepId,
|
|
toStepId: data.toStepId,
|
|
condition: data.condition ?? Prisma.JsonNull,
|
|
priority: data.priority ?? 0,
|
|
},
|
|
});
|
|
|
|
return newTransition;
|
|
}
|
|
|
|
/**
|
|
* Delete a transition
|
|
*/
|
|
public static async deleteTransition(projectId: string, workflowId: string, transitionId: string): Promise<void> {
|
|
// Get workflow to check if it's enabled
|
|
const workflow = await this.get(projectId, workflowId);
|
|
|
|
// Verify transition exists and belongs to workflow
|
|
const transition = await prisma.workflowTransition.findFirst({
|
|
where: {
|
|
id: transitionId,
|
|
fromStep: {
|
|
workflowId,
|
|
workflow: {projectId},
|
|
},
|
|
},
|
|
include: {
|
|
fromStep: true,
|
|
toStep: true,
|
|
},
|
|
});
|
|
|
|
if (!transition) {
|
|
throw new HttpException(404, 'Transition not found');
|
|
}
|
|
|
|
// Check if workflow is enabled and has active executions that could be affected
|
|
if (workflow.enabled) {
|
|
// Get all steps that would become orphaned by removing this transition
|
|
const allSteps = await prisma.workflowStep.findMany({
|
|
where: {workflowId},
|
|
include: {outgoingTransitions: true, incomingTransitions: true},
|
|
});
|
|
|
|
// Build adjacency map without this transition
|
|
const adjacencyMap = new Map<string, string[]>();
|
|
for (const s of allSteps) {
|
|
adjacencyMap.set(
|
|
s.id,
|
|
s.outgoingTransitions.filter(t => t.id !== transitionId).map(t => t.toStepId),
|
|
);
|
|
}
|
|
|
|
// Find all steps reachable from the toStep (downstream)
|
|
const downstreamSteps = new Set<string>([transition.toStepId]);
|
|
const queue = [transition.toStepId];
|
|
|
|
while (queue.length > 0) {
|
|
const currentStepId = queue.shift()!;
|
|
const outgoingStepIds = adjacencyMap.get(currentStepId) || [];
|
|
|
|
for (const nextStepId of outgoingStepIds) {
|
|
if (!downstreamSteps.has(nextStepId)) {
|
|
downstreamSteps.add(nextStepId);
|
|
queue.push(nextStepId);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Check if any active executions are on the toStep or downstream steps
|
|
const executionsAffected = await prisma.workflowExecution.count({
|
|
where: {
|
|
workflowId,
|
|
currentStepId: {in: Array.from(downstreamSteps)},
|
|
status: {
|
|
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
|
|
},
|
|
},
|
|
});
|
|
|
|
if (executionsAffected > 0) {
|
|
throw new HttpException(
|
|
409,
|
|
`Cannot delete transition from "${transition.fromStep.name}" to "${transition.toStep.name}" ` +
|
|
`while ${executionsAffected} execution(s) are on affected steps. ` +
|
|
'Please disable the workflow first or wait for executions to complete.',
|
|
);
|
|
}
|
|
}
|
|
|
|
await prisma.workflowTransition.delete({
|
|
where: {id: transitionId},
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Start a workflow execution for a contact
|
|
*/
|
|
public static async startExecution(
|
|
projectId: string,
|
|
workflowId: string,
|
|
contactId: string,
|
|
context?: Prisma.JsonValue,
|
|
): Promise<WorkflowExecution> {
|
|
// Verify workflow exists, is enabled, and belongs to project
|
|
const workflow = await this.get(projectId, workflowId);
|
|
|
|
if (!workflow.enabled) {
|
|
throw new HttpException(400, 'Workflow is not enabled');
|
|
}
|
|
|
|
// Verify contact belongs to project
|
|
const contact = await prisma.contact.findFirst({
|
|
where: {
|
|
id: contactId,
|
|
projectId,
|
|
},
|
|
});
|
|
|
|
if (!contact) {
|
|
throw new HttpException(404, 'Contact not found');
|
|
}
|
|
|
|
// Check re-entry rules
|
|
if (!workflow.allowReentry) {
|
|
// If re-entry is not allowed, check if contact has ANY execution (regardless of status)
|
|
const existingExecution = await prisma.workflowExecution.findFirst({
|
|
where: {
|
|
workflowId,
|
|
contactId,
|
|
},
|
|
});
|
|
|
|
if (existingExecution) {
|
|
throw new HttpException(
|
|
409,
|
|
`Workflow does not allow re-entry. Contact already has execution (${existingExecution.status})`,
|
|
);
|
|
}
|
|
} else {
|
|
// If re-entry is allowed, only check if there's a currently RUNNING execution
|
|
const runningExecution = await prisma.workflowExecution.findFirst({
|
|
where: {
|
|
workflowId,
|
|
contactId,
|
|
status: WorkflowExecutionStatus.RUNNING,
|
|
},
|
|
});
|
|
|
|
if (runningExecution) {
|
|
throw new HttpException(409, 'Workflow is already running for this contact');
|
|
}
|
|
}
|
|
|
|
// Find the trigger step
|
|
const triggerStep = workflow.steps.find(step => step.type === 'TRIGGER');
|
|
|
|
if (!triggerStep) {
|
|
throw new HttpException(400, 'Workflow has no trigger step');
|
|
}
|
|
|
|
// Create workflow execution
|
|
const execution = await prisma.workflowExecution.create({
|
|
data: {
|
|
workflowId,
|
|
contactId,
|
|
status: WorkflowExecutionStatus.RUNNING,
|
|
currentStepId: triggerStep.id,
|
|
context: context ?? Prisma.JsonNull,
|
|
},
|
|
});
|
|
|
|
// Start executing the workflow asynchronously
|
|
// Don't await - let it run in background
|
|
WorkflowExecutionService.processStepExecution(execution.id, triggerStep.id).catch(error => {
|
|
signale.error('Error executing workflow:', error);
|
|
});
|
|
|
|
return execution;
|
|
}
|
|
|
|
/**
|
|
* Get workflow executions with filtering
|
|
*/
|
|
public static async listExecutions(
|
|
projectId: string,
|
|
workflowId: string,
|
|
page = 1,
|
|
pageSize = 20,
|
|
status?: WorkflowExecutionStatus,
|
|
) {
|
|
// Verify workflow belongs to project
|
|
await this.get(projectId, workflowId);
|
|
|
|
const skip = (page - 1) * pageSize;
|
|
|
|
const where: Prisma.WorkflowExecutionWhereInput = {
|
|
workflowId,
|
|
...(status ? {status} : {}),
|
|
};
|
|
|
|
const [executions, total] = await Promise.all([
|
|
prisma.workflowExecution.findMany({
|
|
where,
|
|
skip,
|
|
take: pageSize,
|
|
orderBy: {startedAt: 'desc'},
|
|
include: {
|
|
contact: {
|
|
select: {
|
|
id: true,
|
|
email: true,
|
|
},
|
|
},
|
|
currentStep: {
|
|
select: {
|
|
id: true,
|
|
name: true,
|
|
type: true,
|
|
},
|
|
},
|
|
},
|
|
}),
|
|
prisma.workflowExecution.count({where}),
|
|
]);
|
|
|
|
return {
|
|
executions,
|
|
total,
|
|
page,
|
|
pageSize,
|
|
totalPages: Math.ceil(total / pageSize),
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get a single execution with details
|
|
*/
|
|
public static async getExecution(
|
|
projectId: string,
|
|
workflowId: string,
|
|
executionId: string,
|
|
): Promise<WorkflowExecutionWithDetails> {
|
|
const execution = await prisma.workflowExecution.findFirst({
|
|
where: {
|
|
id: executionId,
|
|
workflowId,
|
|
workflow: {projectId},
|
|
},
|
|
include: {
|
|
workflow: true,
|
|
contact: {
|
|
select: {
|
|
id: true,
|
|
email: true,
|
|
},
|
|
},
|
|
currentStep: true,
|
|
stepExecutions: {
|
|
include: {
|
|
step: {
|
|
select: {
|
|
id: true,
|
|
name: true,
|
|
type: true,
|
|
},
|
|
},
|
|
},
|
|
orderBy: {createdAt: 'asc'},
|
|
},
|
|
},
|
|
});
|
|
|
|
if (!execution) {
|
|
throw new HttpException(404, 'Workflow execution not found');
|
|
}
|
|
|
|
return execution;
|
|
}
|
|
|
|
/**
|
|
* Cancel a workflow execution
|
|
*/
|
|
public static async cancelExecution(
|
|
projectId: string,
|
|
workflowId: string,
|
|
executionId: string,
|
|
): Promise<WorkflowExecution> {
|
|
// Verify execution exists
|
|
await this.getExecution(projectId, workflowId, executionId);
|
|
|
|
return prisma.workflowExecution.update({
|
|
where: {id: executionId},
|
|
data: {
|
|
status: WorkflowExecutionStatus.CANCELLED,
|
|
completedAt: new Date(),
|
|
exitReason: 'Cancelled by user',
|
|
},
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Cancel all active executions for a workflow
|
|
*/
|
|
public static async cancelAllExecutions(projectId: string, workflowId: string): Promise<{cancelled: number}> {
|
|
// Verify workflow exists and belongs to project
|
|
await this.get(projectId, workflowId);
|
|
|
|
// Cancel all running and waiting executions
|
|
const result = await prisma.workflowExecution.updateMany({
|
|
where: {
|
|
workflowId,
|
|
status: {
|
|
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
|
|
},
|
|
},
|
|
data: {
|
|
status: WorkflowExecutionStatus.CANCELLED,
|
|
completedAt: new Date(),
|
|
exitReason: 'Cancelled by user (bulk cancel)',
|
|
},
|
|
});
|
|
|
|
return {cancelled: result.count};
|
|
}
|
|
|
|
/**
|
|
* Get all available fields for workflow conditions (contact fields + event fields)
|
|
*/
|
|
public static async getAvailableFields(projectId: string, eventName?: string) {
|
|
// Get contact fields with types (standard + custom data fields)
|
|
const contactFieldsWithTypes = await ContactService.getAvailableFields(projectId);
|
|
|
|
// Build typed field list with 'contact.' prefix
|
|
const contactFields = contactFieldsWithTypes.map(f => ({
|
|
field: `contact.${f.field}`,
|
|
type: f.type,
|
|
category: f.field.startsWith('data.') ? 'Custom Data' : 'Contact Fields',
|
|
}));
|
|
|
|
// Get event fields by analyzing actual event data
|
|
// Event fields are treated as dynamic (unknown type at runtime)
|
|
const eventFieldNames = await EventService.getAvailableEventFields(projectId, eventName);
|
|
const eventFields = eventFieldNames.map(field => ({
|
|
field,
|
|
type: 'string' as const, // Event fields default to string, can contain any JSON value
|
|
category: 'Event Data',
|
|
}));
|
|
|
|
// Combine all fields
|
|
const allFields = [...contactFields, ...eventFields].sort((a, b) => a.field.localeCompare(b.field));
|
|
|
|
// Also return legacy format for backwards compatibility
|
|
const fieldNames = allFields.map(f => f.field);
|
|
|
|
return {
|
|
fields: fieldNames,
|
|
typedFields: allFields,
|
|
count: allFields.length,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Check if a workflow has active executions
|
|
*/
|
|
private static async hasActiveExecutions(workflowId: string): Promise<number> {
|
|
return prisma.workflowExecution.count({
|
|
where: {
|
|
workflowId,
|
|
status: {
|
|
in: [WorkflowExecutionStatus.RUNNING, WorkflowExecutionStatus.WAITING],
|
|
},
|
|
},
|
|
});
|
|
}
|
|
}
|