diff --git a/apps/api/src/jobs/email-processor.ts b/apps/api/src/jobs/email-processor.ts index 6d8d73afd..d56965429 100644 --- a/apps/api/src/jobs/email-processor.ts +++ b/apps/api/src/jobs/email-processor.ts @@ -28,6 +28,7 @@ import { bodyHasListManagementLink, buildEmailHeaders, classifyEmail, + RECIPIENT_OVERRIDE_HEADER, senderOneClickUrl, withSourceEmail, } from '../services/EmailHeaderService.js'; @@ -242,12 +243,12 @@ export async function processEmailJob(emailId: string): Promise { : undefined; // Check for custom recipient override in headers - const recipientEmail = customHeaders?.['X-Plunk-Recipient-Override'] || email.contact.email; + const recipientEmail = customHeaders?.[RECIPIENT_OVERRIDE_HEADER] || email.contact.email; // Remove internal headers before sending const publicHeaders = customHeaders ? {...customHeaders} : undefined; - if (publicHeaders && 'X-Plunk-Recipient-Override' in publicHeaders) { - delete publicHeaders['X-Plunk-Recipient-Override']; + if (publicHeaders && RECIPIENT_OVERRIDE_HEADER in publicHeaders) { + delete publicHeaders[RECIPIENT_OVERRIDE_HEADER]; } // Build the outbound headers: standards-based defaults for the email class diff --git a/apps/api/src/services/EmailHeaderService.ts b/apps/api/src/services/EmailHeaderService.ts index 8c4efb8a4..47d03d136 100644 --- a/apps/api/src/services/EmailHeaderService.ts +++ b/apps/api/src/services/EmailHeaderService.ts @@ -3,6 +3,13 @@ import {EmailSourceType} from '@plunk/db'; import {API_URI} from '../app/constants.js'; +/** + * Internal header on an email that a workflow step sends to a custom recipient instead of the + * contact. The email is still recorded under the contact; the email processor addresses it to this + * header's value and strips the header before sending. + */ +export const RECIPIENT_OVERRIDE_HEADER = 'X-Plunk-Recipient-Override'; + /** * Classification of an outbound email, derived by the caller from the email's * source type and template/campaign type. This is the single axis that decides diff --git a/apps/api/src/services/EmailService.ts b/apps/api/src/services/EmailService.ts index 07b33732c..1ead44934 100644 --- a/apps/api/src/services/EmailService.ts +++ b/apps/api/src/services/EmailService.ts @@ -9,7 +9,7 @@ import {ConflictError, ErrorCode, HttpException, NotFound} from '../exceptions/i import {createTranslatorSync, renderTemplate} from '@plunk/shared'; import {BillingLimitService} from './BillingLimitService.js'; -import {withSourceEmail} from './EmailHeaderService.js'; +import {RECIPIENT_OVERRIDE_HEADER, withSourceEmail} from './EmailHeaderService.js'; import {QueueService} from './QueueService.js'; import {ThrottleService} from './ThrottleService.js'; @@ -332,7 +332,7 @@ export class EmailService { // If custom recipient email is provided, store it in headers for later use const emailHeaders = params.headers ? {...params.headers} : {}; if (params.recipientEmail) { - emailHeaders['X-Plunk-Recipient-Override'] = params.recipientEmail; + emailHeaders[RECIPIENT_OVERRIDE_HEADER] = params.recipientEmail; } const email = await prisma.email.create({ diff --git a/apps/api/src/services/EventService.ts b/apps/api/src/services/EventService.ts index f326175c1..44526893b 100644 --- a/apps/api/src/services/EventService.ts +++ b/apps/api/src/services/EventService.ts @@ -42,7 +42,7 @@ export class EventService { await this.triggerWorkflows(projectId, eventName, contactId, data); // Resume workflows waiting for this event - await WorkflowExecutionService.handleEvent(projectId, eventName, contactId, data); + await WorkflowExecutionService.handleEvent(projectId, eventName, contactId, data, emailId); return event; } diff --git a/apps/api/src/services/WorkflowExecutionService.ts b/apps/api/src/services/WorkflowExecutionService.ts index c43e36d42..e9214f698 100644 --- a/apps/api/src/services/WorkflowExecutionService.ts +++ b/apps/api/src/services/WorkflowExecutionService.ts @@ -1,5 +1,6 @@ import type { Contact, + Email, Prisma, Template, Workflow, @@ -9,7 +10,7 @@ import type { } from '@plunk/db'; import {StepExecutionStatus, WorkflowExecutionStatus} from '@plunk/db'; import {toPrismaJson} from '@plunk/types'; -import {renderTemplate, WorkflowStepConfigSchemas} from '@plunk/shared'; +import {canScopeWaitToExecution, renderTemplate, WorkflowStepConfigSchemas} from '@plunk/shared'; import dns from 'node:dns/promises'; import net from 'node:net'; import signale from 'signale'; @@ -19,6 +20,7 @@ import {HttpException} from '../exceptions/index.js'; import {DASHBOARD_URI} from '../app/constants.js'; import {ContactFieldService} from './ContactFieldService.js'; +import {RECIPIENT_OVERRIDE_HEADER} from './EmailHeaderService.js'; import {EmailService} from './EmailService.js'; import {NtfyService} from './NtfyService.js'; import {QueueService} from './QueueService.js'; @@ -471,6 +473,7 @@ export class WorkflowExecutionService { eventName: string, contactId?: string, data?: Record, + emailId?: string, ): Promise { // Find workflows waiting for this event (a cancelled execution keeps its wait step WAITING) const waitingExecutions = await prisma.workflowStepExecution.findMany({ @@ -503,10 +506,29 @@ export class WorkflowExecutionService { }, }); + // The email behind the event, loaded at most once and only for a wait scoped to its execution + let sourceEmail: Pick | null | undefined; + for (const stepExecution of waitingExecutions) { const config = stepExecution.step.config; if (config && typeof config === 'object' && 'eventName' in config && config.eventName === eventName) { + // A wait scoped to its execution counts only the emails that execution sent to the contact + if ('scope' in config && config.scope === 'execution' && canScopeWaitToExecution(eventName)) { + if (sourceEmail === undefined) { + sourceEmail = emailId + ? await prisma.email.findUnique({ + where: {id: emailId}, + select: {workflowExecutionId: true, headers: true}, + }) + : null; + } + + if (!this.sentByExecution(sourceEmail, stepExecution.executionId)) { + continue; + } + } + // Event matches. Claim the wait before resuming it: the list above can be stale by now, // as the same event may be delivered twice at once (SES notifications, repeated clicks) // or tracked again while an earlier wait of this loop was resumed. Whoever moves the @@ -543,6 +565,24 @@ export class WorkflowExecutionService { } } + /** + * Whether an email was sent to the contact by the given execution. An email a step sent to a + * custom recipient is recorded under the contact too, but it is not the contact's engagement. + */ + private static sentByExecution( + email: Pick | null, + executionId: string, + ): boolean { + if (email?.workflowExecutionId !== executionId) { + return false; + } + + const headers = + email.headers && typeof email.headers === 'object' && !Array.isArray(email.headers) ? email.headers : {}; + + return !(RECIPIENT_OVERRIDE_HEADER in headers); + } + /** * Execute a specific step based on its type */ diff --git a/apps/api/src/services/__tests__/WorkflowExecutionService.waitScope.test.ts b/apps/api/src/services/__tests__/WorkflowExecutionService.waitScope.test.ts new file mode 100644 index 000000000..5713b18fa --- /dev/null +++ b/apps/api/src/services/__tests__/WorkflowExecutionService.waitScope.test.ts @@ -0,0 +1,143 @@ +import {beforeEach, describe, expect, it, vi} from 'vitest'; +import {EmailSourceType, EmailStatus, StepExecutionStatus, WorkflowExecutionStatus, WorkflowStepType} from '@plunk/db'; +import {prisma as servicePrisma} from '../../database/prisma'; +import {EventService} from '../EventService'; +import {WorkflowExecutionService} from '../WorkflowExecutionService'; +import {factories, getPrismaClient} from '../../../../../test/helpers'; + +vi.mock('../QueueService', () => ({ + QueueService: { + queueWorkflowStep: vi.fn(async () => ({id: 'mock-job-id'})), + queueEmail: vi.fn(async () => ({id: 'mock-email-job-id'})), + queueWorkflowTimeout: vi.fn(async () => ({id: 'mock-timeout-job-id'})), + cancelWorkflowTimeout: vi.fn(async () => true), + }, +})); + +/** + * By default a WAIT_FOR_EVENT step resumes on the event from any email of the contact. + * `scope: "execution"` limits email events to the emails the waiting execution sent the contact. + */ +describe('WorkflowExecutionService - WAIT_FOR_EVENT scope', () => { + const prisma = getPrismaClient(); + let projectId: string; + let contactId: string; + + beforeEach(async () => { + const {project} = await factories.createUserWithProject(); + projectId = project.id; + contactId = (await factories.createContact({projectId, subscribed: true})).id; + }); + + /** TRIGGER → SEND_EMAIL → WAIT_FOR_EVENT → EXIT, run until it waits. */ + async function runUntilWaiting(waitConfig: Record, sendConfig: Record = {}) { + const template = await factories.createTemplate({projectId}); + const workflow = await factories.createWorkflow({projectId}); + const trigger = await prisma.workflowStep.findFirstOrThrow({ + where: {workflowId: workflow.id, type: WorkflowStepType.TRIGGER}, + }); + const send = await factories.createWorkflowStep({ + workflowId: workflow.id, + type: WorkflowStepType.SEND_EMAIL, + templateId: template.id, + config: {templateId: template.id, ...sendConfig}, + }); + const wait = await factories.createWorkflowStep({ + workflowId: workflow.id, + type: WorkflowStepType.WAIT_FOR_EVENT, + config: {timeout: 3600, ...waitConfig}, + }); + const exit = await factories.createWorkflowStep({workflowId: workflow.id, type: WorkflowStepType.EXIT}); + await prisma.workflowTransition.createMany({ + data: [ + {fromStepId: trigger.id, toStepId: send.id}, + {fromStepId: send.id, toStepId: wait.id}, + {fromStepId: wait.id, toStepId: exit.id}, + ], + }); + + const execution = await prisma.workflowExecution.create({ + data: {workflowId: workflow.id, contactId, status: WorkflowExecutionStatus.RUNNING, currentStepId: trigger.id}, + }); + await WorkflowExecutionService.processStepExecution(execution.id, trigger.id); + + const sent = await prisma.email.findFirstOrThrow({where: {workflowExecutionId: execution.id}}); + const waitStatus = async () => + (await prisma.workflowStepExecution.findFirstOrThrow({where: {executionId: execution.id, stepId: wait.id}})) + .status; + expect(await waitStatus()).toBe(StepExecutionStatus.WAITING); + + return {sent, waitStatus}; + } + + const newsletter = () => + factories.createEmail({ + projectId, + contactId, + sourceType: EmailSourceType.CAMPAIGN, + status: EmailStatus.SENT, + sentAt: new Date(), + }); + + /** The call the SES webhook makes for an open (Webhooks.receiveSNSWebhook). */ + const open = (emailId: string) => EventService.trackEvent(projectId, 'email.open', contactId, emailId, {emailId}); + + it('without a scope, resumes on an open of any email of the contact, and looks no email up', async () => { + const run = await runUntilWaiting({eventName: 'email.open'}); + const emailLookup = vi.spyOn(servicePrisma.email, 'findUnique'); + + await open((await newsletter()).id); + + expect(await run.waitStatus()).toBe(StepExecutionStatus.COMPLETED); + expect(emailLookup).not.toHaveBeenCalled(); + }); + + it("with scope execution, ignores other emails and resumes on the execution's own email", async () => { + const run = await runUntilWaiting({eventName: 'email.open', scope: 'execution'}); + const emailLookup = vi.spyOn(servicePrisma.email, 'findUnique'); + + await open((await newsletter()).id); + expect(await run.waitStatus()).toBe(StepExecutionStatus.WAITING); + expect(emailLookup).toHaveBeenCalledTimes(1); + + await open(run.sent.id); + expect(await run.waitStatus()).toBe(StepExecutionStatus.COMPLETED); + }); + + it('with scope execution, ignores an email the execution sent to a custom recipient', async () => { + const run = await runUntilWaiting( + {eventName: 'email.open', scope: 'execution'}, + {recipient: {type: 'CUSTOM', customEmail: 'team@plunk.test'}}, + ); + + await open(run.sent.id); + + expect(await run.waitStatus()).toBe(StepExecutionStatus.WAITING); + }); + + it('with scope execution, ignores an email event that names no email', async () => { + const run = await runUntilWaiting({eventName: 'email.open', scope: 'execution'}); + + await EventService.trackEvent(projectId, 'email.open', contactId); + + expect(await run.waitStatus()).toBe(StepExecutionStatus.WAITING); + }); + + it('with scope execution, resumes only the execution that sent the email', async () => { + const first = await runUntilWaiting({eventName: 'email.click', scope: 'execution'}); + const second = await runUntilWaiting({eventName: 'email.click', scope: 'execution'}); + + await EventService.trackEvent(projectId, 'email.click', contactId, second.sent.id, {emailId: second.sent.id}); + + expect(await first.waitStatus()).toBe(StepExecutionStatus.WAITING); + expect(await second.waitStatus()).toBe(StepExecutionStatus.COMPLETED); + }); + + it('applies no scope to events an execution cannot have produced', async () => { + const run = await runUntilWaiting({eventName: 'order.shipped', scope: 'execution'}); + + await EventService.trackEvent(projectId, 'order.shipped', contactId); + + expect(await run.waitStatus()).toBe(StepExecutionStatus.COMPLETED); + }); +}); diff --git a/apps/mcp/src/tools/workflow-graph.ts b/apps/mcp/src/tools/workflow-graph.ts index 26bcdc4d4..b8a17d7fa 100644 --- a/apps/mcp/src/tools/workflow-graph.ts +++ b/apps/mcp/src/tools/workflow-graph.ts @@ -112,6 +112,7 @@ const STEP_CONFIG_SCHEMAS: Record = { .object({amount: z.number().positive(), unit: z.enum(DELAY_UNITS)}) .refine((data) => data.amount <= MAX_DELAY[data.unit], {message: 'Delay cannot exceed 365 days'}), WAIT_FOR_EVENT: z.object({ + scope: z.enum(['contact', 'execution']).optional(), eventName: z.string().min(1), timeout: z.number().nonnegative().max(MAX_WAIT_SECONDS, 'Timeout cannot exceed 365 days').optional(), }), diff --git a/apps/web/src/components/workflow-steps/WaitForEventStepDialog.tsx b/apps/web/src/components/workflow-steps/WaitForEventStepDialog.tsx index 551f25dce..fce08cb38 100644 --- a/apps/web/src/components/workflow-steps/WaitForEventStepDialog.tsx +++ b/apps/web/src/components/workflow-steps/WaitForEventStepDialog.tsx @@ -1,4 +1,18 @@ -import {Command, CommandGroup, CommandItem, CommandList, Input, Label, Select, SelectContent, SelectItem, SelectTrigger, SelectValue} from '@plunk/ui'; +import {canScopeWaitToExecution} from '@plunk/shared'; +import { + Command, + CommandGroup, + CommandItem, + CommandList, + Input, + Label, + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, + Switch, +} from '@plunk/ui'; import {useState} from 'react'; import {toast} from 'sonner'; import useSWR from 'swr'; @@ -49,6 +63,7 @@ export function WaitForEventStepDialog({step, workflowId, open, onOpenChange, on const [eventPopoverOpen, setEventPopoverOpen] = useState(false); const [timeoutAmount, setTimeoutAmount] = useState(initialTimeout.amount); const [timeoutUnit, setTimeoutUnit] = useState(initialUnit); + const [onlyThisExecution, setOnlyThisExecution] = useState(config.scope === 'execution'); const {data: eventNamesData} = useSWR<{eventNames: string[]}>(open ? '/events/names' : null, { revalidateOnFocus: false, @@ -72,9 +87,18 @@ export function WaitForEventStepDialog({step, workflowId, open, onOpenChange, on const timeoutSeconds = amount > 0 ? amount * SECONDS_PER_UNIT[timeoutUnit] : 0; + // The scope only applies to email events an execution's own emails produce + const scoped = onlyThisExecution && canScopeWaitToExecution(eventName); + + // A scoped wait whose email is never sent or opened would otherwise wait forever + if (scoped && timeoutSeconds === 0) { + toast.error('Set a timeout when waiting only for emails from this run'); + return; + } + const ok = await update({ name, - config: {eventName, timeout: timeoutSeconds}, + config: {eventName, timeout: timeoutSeconds, ...(scoped ? {scope: 'execution'} : {})}, }); if (ok) { @@ -156,6 +180,21 @@ export function WaitForEventStepDialog({step, workflowId, open, onOpenChange, on + {canScopeWaitToExecution(eventName) && ( +
+ +
+ +

+ Count the event only for emails this workflow sent to the contact in this run. Otherwise any email the + contact receives counts. Requires a timeout, in case the email is never sent or opened. +

+
+
+ )} +
diff --git a/apps/wiki/content/docs/concepts/workflows.mdx b/apps/wiki/content/docs/concepts/workflows.mdx index 2774eb9a6..75ab6be58 100644 --- a/apps/wiki/content/docs/concepts/workflows.mdx +++ b/apps/wiki/content/docs/concepts/workflows.mdx @@ -66,6 +66,8 @@ A workflow always begins with a single auto-created `TRIGGER` step. You build th | `UPDATE_CONTACT` | Patches contact data — useful for tagging contacts as they progress (`{ stage: "activated" }`). | `data` object | | `EXIT` | Terminates the execution. Optionally records an `exitReason` for analytics. | optional `reason` | +A `WAIT_FOR_EVENT` step on an email event resumes when that event comes from any email of the contact, including emails sent by campaigns or other workflows (`scope: "contact"`, the default). For the events a workflow email produces (`email.sent`, `email.delivery`, `email.open`, `email.click`, `email.bounce`, `email.complaint`), set `scope: "execution"` in the step config (**Only emails from this run** in the editor) to count only the emails this execution sent to the contact. Emails a step sends to a custom recipient are recorded under the contact but never count for a scoped wait. If the execution sent several emails, an event from any of them resumes the wait. The editor requires a timeout for a scoped wait: if its email is never sent (for example, to an unsubscribed contact) or never opened, the execution would otherwise wait forever. + ## 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. diff --git a/packages/shared/src/__tests__/workflow-events.test.ts b/packages/shared/src/__tests__/workflow-events.test.ts new file mode 100644 index 000000000..d9db34e09 --- /dev/null +++ b/packages/shared/src/__tests__/workflow-events.test.ts @@ -0,0 +1,19 @@ +import {describe, expect, it} from 'vitest'; + +import {canScopeWaitToExecution} from '../workflow-events.js'; + +describe('canScopeWaitToExecution', () => { + it.each(['email.sent', 'email.delivery', 'email.open', 'email.click', 'email.bounce', 'email.complaint'])( + 'accepts %s, which a workflow email produces', + eventName => { + expect(canScopeWaitToExecution(eventName)).toBe(true); + }, + ); + + it.each(['email.received', 'email.opened', 'contact.subscribed', 'order.shipped', ''])( + 'refuses %s, which cannot be traced back to an execution', + eventName => { + expect(canScopeWaitToExecution(eventName)).toBe(false); + }, + ); +}); diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index ac3064f4e..073a99ee6 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -5,3 +5,4 @@ export * from './i18n/index.js'; export * from './mailbox-simulator.js'; export * from './unsubscribe.js'; export * from './delivery.js'; +export * from './workflow-events.js'; diff --git a/packages/shared/src/schemas/index.ts b/packages/shared/src/schemas/index.ts index d00a25a91..eb8459e2e 100644 --- a/packages/shared/src/schemas/index.ts +++ b/packages/shared/src/schemas/index.ts @@ -360,6 +360,8 @@ export const WorkflowStepConfigSchemas = { }, ), waitForEvent: z.object({ + // "execution": count only emails this execution sent, for the events canScopeWaitToExecution accepts + scope: z.enum(['contact', 'execution']).optional(), eventName: z.string().min(1), // 0 or omitted waits with no time limit timeout: z.number().nonnegative().max(31536000, 'Timeout cannot exceed 365 days (31,536,000 seconds)').optional(), diff --git a/packages/shared/src/workflow-events.ts b/packages/shared/src/workflow-events.ts new file mode 100644 index 000000000..0bfb64554 --- /dev/null +++ b/packages/shared/src/workflow-events.ts @@ -0,0 +1,17 @@ +/** + * Email events a workflow's own sends produce. A WAIT_FOR_EVENT step waiting for one of these can + * count only the emails its execution sent (`scope: "execution"`). Other events cannot be traced + * back to an execution, so the scope does not apply to them. + */ +const EXECUTION_SCOPED_EVENTS = new Set([ + 'email.sent', + 'email.delivery', + 'email.open', + 'email.click', + 'email.bounce', + 'email.complaint', +]); + +export function canScopeWaitToExecution(eventName: string): boolean { + return EXECUTION_SCOPED_EVENTS.has(eventName); +}