From 62aeca040f7e9175a71edf3a17d95b4781295ab9 Mon Sep 17 00:00:00 2001 From: Matthew Chambers Date: Tue, 28 Jul 2026 14:39:08 -0400 Subject: [PATCH 1/2] feat: add per-email tracking override to /v1/send Adds an optional `tracking` boolean to POST /v1/send that overrides the project's tracking mode for a single email. Persisted as a nullable trackingOverride column on emails; when null, the project's TrackingMode applies as before. Campaigns and workflows are unaffected. Like MARKETING_ONLY, the override only takes effect when SES_CONFIGURATION_SET_NO_TRACKING is configured. Co-Authored-By: Claude Fable 5 --- apps/api/src/controllers/Actions.ts | 4 +- apps/api/src/services/EmailService.ts | 6 +- .../services/__tests__/EmailService.test.ts | 102 +++++++++++++++++- apps/wiki/content/docs/guides/tracking.mdx | 17 ++- apps/wiki/openapi.json | 4 + .../migration.sql | 2 + packages/db/prisma/schema.prisma | 3 + packages/shared/src/schemas/index.ts | 1 + test/helpers/factories.ts | 4 + 9 files changed, 138 insertions(+), 5 deletions(-) create mode 100644 packages/db/prisma/migrations/20260728120000_add_email_tracking_override/migration.sql diff --git a/apps/api/src/controllers/Actions.ts b/apps/api/src/controllers/Actions.ts index 1266e807f..a471b72dc 100644 --- a/apps/api/src/controllers/Actions.ts +++ b/apps/api/src/controllers/Actions.ts @@ -125,6 +125,7 @@ export class Actions { * - name: string (optional) - Sender name (alternative to from.name) * - from: string | object (optional) - Sender email or {name, email} object (must be from verified domain) * - reply: string (optional) - Reply-to email + * - tracking: boolean (optional) - Override the project's open/click tracking mode for this email (uses project setting if omitted) * - headers: object (optional) - Additional email headers * - data: object (optional) - Contact data and template variables * - Simple values are saved to contact (persistent) @@ -185,7 +186,7 @@ export class Actions { const auth = res.locals.auth; // Zod validation - errors automatically handled by global error handler - const {to, subject, body, subscribed, name, from, reply, headers, data, template, attachments} = + const {to, subject, body, subscribed, name, from, reply, tracking, headers, data, template, attachments} = ActionSchemas.send.parse(req.body); // Normalize recipients to array and parse email/name @@ -336,6 +337,7 @@ export class Actions { headers: headers || undefined, attachments: attachments || undefined, templateId: templateId, + tracking: tracking, }); emailResults.push({ diff --git a/apps/api/src/services/EmailService.ts b/apps/api/src/services/EmailService.ts index 51ae7efa6..514e27dfd 100644 --- a/apps/api/src/services/EmailService.ts +++ b/apps/api/src/services/EmailService.ts @@ -40,6 +40,7 @@ interface SendEmailParams { workflowStepExecutionId?: string; recipientEmail?: string; // Optional custom recipient email (overrides contact.email) isTransactional?: boolean; // Override source type to TRANSACTIONAL (e.g. for transactional campaigns) + tracking?: boolean; // Per-email tracking override (undefined = use project tracking mode) } /** @@ -101,6 +102,7 @@ export class EmailService { attachments: params.attachments ? toPrismaJson(params.attachments) : undefined, sourceType: EmailSourceType.TRANSACTIONAL, templateId: params.templateId, + trackingOverride: params.tracking ?? null, status: EmailStatus.PENDING, }, }); @@ -414,8 +416,8 @@ export class EmailService { }>) : undefined; - // Determine tracking based on project settings and email type - const shouldTrack = this.shouldTrackEmail(email.project.tracking, email.sourceType); + // Determine tracking: per-email override wins, otherwise project settings and email type + const shouldTrack = email.trackingOverride ?? this.shouldTrackEmail(email.project.tracking, email.sourceType); // Send via AWS SES const result = await sendRawEmail({ diff --git a/apps/api/src/services/__tests__/EmailService.test.ts b/apps/api/src/services/__tests__/EmailService.test.ts index 1d5a1e0fa..3b5793410 100644 --- a/apps/api/src/services/__tests__/EmailService.test.ts +++ b/apps/api/src/services/__tests__/EmailService.test.ts @@ -1,5 +1,5 @@ import {beforeEach, describe, expect, it, vi, type Mock} from 'vitest'; -import {EmailSourceType, EmailStatus, TemplateType} from '@plunk/db'; +import {EmailSourceType, EmailStatus, TemplateType, TrackingMode} from '@plunk/db'; import {ActionSchemas} from '@plunk/shared'; import {EmailService} from '../EmailService'; import {sendRawEmail} from '../SESService'; @@ -763,6 +763,106 @@ describe('EmailService', () => { }); }); + // ======================================== + // PER-EMAIL TRACKING OVERRIDE + // ======================================== + describe('Per-Email Tracking Override', () => { + it('should disable tracking when trackingOverride is false, even with project tracking ENABLED', async () => { + const email = await factories.createEmail({ + projectId, + contactId, + status: EmailStatus.PENDING, + trackingOverride: false, + }); + + await EmailService.sendEmail(email.id); + + expect(vi.mocked(sendRawEmail)).toHaveBeenCalledWith(expect.objectContaining({tracking: false})); + }); + + it('should enable tracking when trackingOverride is true, even with project tracking DISABLED', async () => { + await prisma.project.update({ + where: {id: projectId}, + data: {tracking: TrackingMode.DISABLED}, + }); + + const email = await factories.createEmail({ + projectId, + contactId, + status: EmailStatus.PENDING, + trackingOverride: true, + }); + + await EmailService.sendEmail(email.id); + + expect(vi.mocked(sendRawEmail)).toHaveBeenCalledWith(expect.objectContaining({tracking: true})); + }); + + it('should fall back to project tracking mode when trackingOverride is null', async () => { + await prisma.project.update({ + where: {id: projectId}, + data: {tracking: TrackingMode.DISABLED}, + }); + + const email = await factories.createEmail({ + projectId, + contactId, + status: EmailStatus.PENDING, + }); + + await EmailService.sendEmail(email.id); + + expect(vi.mocked(sendRawEmail)).toHaveBeenCalledWith(expect.objectContaining({tracking: false})); + }); + + it('should persist the tracking param from sendTransactionalEmail', async () => { + const email = await EmailService.sendTransactionalEmail({ + projectId, + contactId, + subject: 'No tracking please', + body: '

Private

', + from: 'test@example.com', + tracking: false, + }); + + expect(email.trackingOverride).toBe(false); + }); + + it('should persist null when tracking param is omitted', async () => { + const email = await EmailService.sendTransactionalEmail({ + projectId, + contactId, + subject: 'Default tracking', + body: '

Hello

', + from: 'test@example.com', + }); + + expect(email.trackingOverride).toBeNull(); + }); + + it('should accept a boolean tracking field in the send schema', () => { + const result = ActionSchemas.send.safeParse({ + to: 'test@example.com', + subject: 'Test', + body: 'Test', + tracking: false, + }); + + expect(result.success).toBe(true); + }); + + it('should reject a non-boolean tracking field in the send schema', () => { + const result = ActionSchemas.send.safeParse({ + to: 'test@example.com', + subject: 'Test', + body: 'Test', + tracking: 'false', + }); + + expect(result.success).toBe(false); + }); + }); + // ======================================== // ATTACHMENT SCHEMA VALIDATION // ======================================== diff --git a/apps/wiki/content/docs/guides/tracking.mdx b/apps/wiki/content/docs/guides/tracking.mdx index a54b0d5f7..01ca3a456 100644 --- a/apps/wiki/content/docs/guides/tracking.mdx +++ b/apps/wiki/content/docs/guides/tracking.mdx @@ -37,6 +37,21 @@ Tracking is configured per project under **Settings → Tracking**. Three modes Tracking is **all or nothing** within a single send — there's no separate toggle for opens vs clicks. Pick the mode that fits your privacy and analytics requirements. +## Per-email override + +Transactional sends can override the project's tracking mode on a per-email basis by passing the optional `tracking` boolean to [`POST /v1/send`](/api-reference/public-api/sendEmail): + +```json +{ + "to": "user@example.com", + "subject": "Your invoice", + "body": "

...

", + "tracking": false +} +``` + +`false` sends the email without tracking, `true` sends it with tracking. When omitted, the project's tracking mode applies. + -On self-hosted instances, the Marketing-only mode is only available when a no-tracking configuration set is configured. See [Self-hosting → Email setup](/self-hosting/email-setup) for details. +On self-hosted instances, the Marketing-only mode and the per-email `tracking` override are only available when a no-tracking configuration set is configured. See [Self-hosting → Email setup](/self-hosting/email-setup) for details. diff --git a/apps/wiki/openapi.json b/apps/wiki/openapi.json index 0eb9ef99b..30bfc0f93 100644 --- a/apps/wiki/openapi.json +++ b/apps/wiki/openapi.json @@ -317,6 +317,10 @@ "format": "email", "description": "Reply-to address." }, + "tracking": { + "type": "boolean", + "description": "Override the project's open/click tracking mode for this email. `false` disables tracking, `true` enables it. Omit to use the project's tracking setting. Only effective when the instance has `SES_CONFIGURATION_SET_NO_TRACKING` configured." + }, "attachments": { "type": "array", "description": "Email attachments. Default cap: 10 attachments and 10 MB total. The full message size cannot exceed 40 MB.", diff --git a/packages/db/prisma/migrations/20260728120000_add_email_tracking_override/migration.sql b/packages/db/prisma/migrations/20260728120000_add_email_tracking_override/migration.sql new file mode 100644 index 000000000..930179423 --- /dev/null +++ b/packages/db/prisma/migrations/20260728120000_add_email_tracking_override/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "emails" ADD COLUMN "trackingOverride" BOOLEAN; diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index 3c2fd2c37..0c2f00a72 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -552,6 +552,9 @@ model Email { workflowStepExecution WorkflowStepExecution? @relation(fields: [workflowStepExecutionId], references: [id], onDelete: Cascade) workflowStepExecutionId String? // For WORKFLOW (specific step) + // Per-email tracking override (null = use project tracking mode) + trackingOverride Boolean? + // Delivery status EmailStatus @default(PENDING) diff --git a/packages/shared/src/schemas/index.ts b/packages/shared/src/schemas/index.ts index ab696e7e4..e496d8e14 100644 --- a/packages/shared/src/schemas/index.ts +++ b/packages/shared/src/schemas/index.ts @@ -506,6 +506,7 @@ export const ActionSchemas = { ) .optional(), reply: email.optional(), + tracking: z.boolean().optional(), headers: z .record( z.string().regex(/^[^\r\n]+$/, 'Header key contains invalid characters'), diff --git a/test/helpers/factories.ts b/test/helpers/factories.ts index b0f8ad6cf..b8e331be7 100644 --- a/test/helpers/factories.ts +++ b/test/helpers/factories.ts @@ -288,6 +288,7 @@ export class TestFactories { opens?: number; clicks?: number; error?: string | null; + trackingOverride?: boolean | null; }, contactId?: string, additionalOptions?: Partial<{ @@ -305,6 +306,7 @@ export class TestFactories { opens?: number; clicks?: number; error?: string | null; + trackingOverride?: boolean | null; }>, ) { // Support both calling conventions @@ -325,6 +327,7 @@ export class TestFactories { opens?: number; clicks?: number; error?: string | null; + trackingOverride?: boolean | null; }; if (typeof projectIdOrOptions === 'string') { @@ -372,6 +375,7 @@ export class TestFactories { opens: options.opens, clicks: options.clicks, error: options.error, + trackingOverride: options.trackingOverride, }, }); } From ab2ad0a698b940d53e40ab1a697d5abcca4ba3ae Mon Sep 17 00:00:00 2001 From: Matthew Chambers Date: Tue, 28 Jul 2026 15:12:33 -0400 Subject: [PATCH 2/2] fix: apply trackingOverride in the worker send path The tracking decision is duplicated in the BullMQ email processor, which is the path /v1/send actually takes (API enqueues, worker sends). The override was stored but never applied there. Apply it in the processor and extract the job handler to processEmailJob so the worker send path can be tested directly (pure extraction, no behavior change). Found during production verification of this feature. Co-Authored-By: Claude Fable 5 --- .../jobs/__tests__/email-processor.test.ts | 73 +++ apps/api/src/jobs/email-processor.ts | 434 +++++++++--------- 2 files changed, 293 insertions(+), 214 deletions(-) diff --git a/apps/api/src/jobs/__tests__/email-processor.test.ts b/apps/api/src/jobs/__tests__/email-processor.test.ts index 881b762bb..46d879534 100644 --- a/apps/api/src/jobs/__tests__/email-processor.test.ts +++ b/apps/api/src/jobs/__tests__/email-processor.test.ts @@ -1,6 +1,10 @@ import {beforeEach, describe, expect, it, vi} from 'vitest'; import {EmailSourceType, EmailStatus, TrackingMode} from '@plunk/db'; +import type {SendEmailJobData} from '@plunk/types'; import {toPrismaJson} from '@plunk/types'; +import type {Job} from 'bullmq'; +import {processEmailJob} from '../email-processor'; +import {sendRawEmail} from '../../services/SESService'; import {createServiceMocks, factories, getPrismaClient} from '../../../../../test/helpers'; // Mock MeterService @@ -10,6 +14,12 @@ vi.mock('../../services/MeterService.js', () => ({ }, })); +// Mock SES service so processEmailJob can run the real worker send path +vi.mock('../../services/SESService', () => ({ + sendRawEmail: vi.fn(), + getSendingQuota: vi.fn().mockResolvedValue(null), +})); + describe('Email Processor', () => { let projectId: string; const prisma = getPrismaClient(); @@ -337,4 +347,67 @@ describe('Email Processor', () => { expect(emailCountNoAttachments).toBe(1); }); }); + + describe('Per-Email Tracking Override (worker send path)', () => { + const runJob = (emailId: string) => processEmailJob({data: {emailId}} as Job); + + beforeEach(() => { + vi.mocked(sendRawEmail).mockClear(); + vi.mocked(sendRawEmail).mockResolvedValue({messageId: 'ses-worker-123'}); + }); + + it('should disable tracking when trackingOverride is false, even with project tracking ENABLED', async () => { + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail({ + projectId, + contactId: contact.id, + status: EmailStatus.PENDING, + trackingOverride: false, + }); + + await runJob(email.id); + + expect(vi.mocked(sendRawEmail)).toHaveBeenCalledWith(expect.objectContaining({tracking: false})); + + const sent = await prisma.email.findUnique({where: {id: email.id}}); + expect(sent?.status).toBe(EmailStatus.SENT); + }); + + it('should enable tracking when trackingOverride is true, even with project tracking DISABLED', async () => { + await prisma.project.update({ + where: {id: projectId}, + data: {tracking: TrackingMode.DISABLED}, + }); + + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail({ + projectId, + contactId: contact.id, + status: EmailStatus.PENDING, + trackingOverride: true, + }); + + await runJob(email.id); + + expect(vi.mocked(sendRawEmail)).toHaveBeenCalledWith(expect.objectContaining({tracking: true})); + }); + + it('should fall back to project tracking mode when trackingOverride is null', async () => { + await prisma.project.update({ + where: {id: projectId}, + data: {tracking: TrackingMode.DISABLED}, + }); + + const contact = await factories.createContact({projectId}); + const email = await factories.createEmail({ + projectId, + contactId: contact.id, + status: EmailStatus.PENDING, + }); + + await runJob(email.id); + + expect(vi.mocked(sendRawEmail)).toHaveBeenCalledWith(expect.objectContaining({tracking: false})); + }); + }); }); diff --git a/apps/api/src/jobs/email-processor.ts b/apps/api/src/jobs/email-processor.ts index 6d8b4a6c9..54b221c32 100644 --- a/apps/api/src/jobs/email-processor.ts +++ b/apps/api/src/jobs/email-processor.ts @@ -71,6 +71,225 @@ function deriveWorkerConcurrency(rateLimit: number): number { return Math.max(MIN_CONCURRENCY, Math.min(derived, EMAIL_WORKER_MAX_CONCURRENCY)); } +/** + * Process a single queued email send. + * Exported so the worker send path can be tested directly. + */ +export async function processEmailJob(job: Job): Promise { + const {emailId} = job.data; + + const email = await prisma.email.findUnique({ + where: {id: emailId}, + include: { + contact: true, + project: true, + template: {select: {type: true}}, + campaign: {select: {type: true}}, + }, + }); + + if (!email) { + throw new Error(`Email ${emailId} not found`); + } + + if (email.status !== EmailStatus.PENDING) { + return; + } + + // Check if project is disabled + if (email.project.disabled) { + signale.warn(`[EMAIL-PROCESSOR] Project ${email.projectId} is disabled, cancelling email ${emailId}`); + await prisma.email.update({ + where: {id: emailId}, + data: { + status: EmailStatus.FAILED, + error: 'Project is disabled', + }, + }); + + // Cancelled emails are terminal for the campaign — finalize so it doesn't + // stay stuck in SENDING forever waiting on emails that will never be sent. + if (email.campaignId) { + await CampaignService.finalizeIfDone(email.campaignId); + } + return; + } + + try { + // Update status to sending + await prisma.email.update({ + where: {id: emailId}, + data: {status: EmailStatus.SENDING}, + }); + + // Format template variables in subject and body + const contactData = (email.contact.data as Record) || {}; + const formattedEmail = EmailService.format({ + subject: email.subject, + body: email.body, + data: { + email: email.contact.email, + ...contactData, + data: contactData, + unsubscribeUrl: `${DASHBOARD_URI}/unsubscribe/${email.contact.id}`, + subscribeUrl: `${DASHBOARD_URI}/subscribe/${email.contact.id}`, + manageUrl: `${DASHBOARD_URI}/manage/${email.contact.id}`, + }, + }); + + // Classify the email once: it decides both the unsubscribe footer and the + // standards-based headers below. + const emailClass = classifyEmail({ + sourceType: email.sourceType, + templateType: email.template?.type, + campaignType: email.campaign?.type, + }); + + // Compile HTML with unsubscribe footer and badge. + // Only marketing emails get the Plunk unsubscribe footer. + const compiledHtml = EmailService.compile({ + content: formattedEmail.body, + contact: email.contact, + project: email.project, + includeUnsubscribe: emailClass === 'marketing', + }); + + // Use fromName from database if available, otherwise fall back to project name + // The 'from' field in the database is just the email address + const fromName = email.fromName || email.project.name; + const fromEmail = email.from; + + // Parse custom headers from JSON + const customHeaders = + email.headers && typeof email.headers === 'object' && !Array.isArray(email.headers) + ? (email.headers as Record) + : undefined; + + // Check for custom recipient override in headers + const recipientEmail = customHeaders?.['X-Plunk-Recipient-Override'] || 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']; + } + + // Build the outbound headers: standards-based defaults for the email class + // plus any caller-supplied headers (which override the defaults). + const outboundHeaders = buildEmailHeaders({ + emailClass, + isCampaign: email.campaignId != null, + hasListManagementLink: bodyHasListManagementLink(compiledHtml, email.contact.id), + unsubscribeId: email.contact.id, + customHeaders: publicHeaders, + }); + + // Build recipient with name if available + const recipient: {name?: string; email: string} | string = email.toName + ? {name: email.toName, email: recipientEmail} + : recipientEmail; + + // Determine tracking: per-email override wins, otherwise project settings and email type + const shouldTrack = email.trackingOverride ?? EmailService.shouldTrackEmail(email.project.tracking, email.sourceType); + + // Check for phishing/dangerous content before sending + const phishingCheck = await SecurityService.checkPhishingContent( + email.projectId, + email.project.name, + email.from, + formattedEmail.subject, + compiledHtml, + ); + + if (phishingCheck.shouldDisable) { + // Disable project immediately + await SecurityService.disableProjectForPhishing( + email.projectId, + formattedEmail.subject, + phishingCheck.confidence, + 'Phishing content detected', + ); + + // Mark email as failed + await prisma.email.update({ + where: {id: emailId}, + data: { + status: EmailStatus.FAILED, + error: 'This email could not be sent. The project has been disabled. Please contact support.', + }, + }); + + throw new Error(`Project ${email.projectId} has been disabled due to a policy violation`); + } + + // Send via AWS SES + const result = await sendRawEmail({ + from: { + name: fromName, + email: fromEmail, + }, + to: typeof recipient === 'string' ? [recipient] : [{name: recipient.name, email: recipient.email}], + content: { + subject: formattedEmail.subject, + html: compiledHtml, + }, + reply: email.replyTo || undefined, + headers: outboundHeaders, + tracking: shouldTrack, + attachments: email.attachments as {filename: string; content: string; contentType: string}[] | null, + }); + + // Mark as sent with SES message ID + await prisma.email.update({ + where: {id: emailId}, + data: { + status: EmailStatus.SENT, + sentAt: new Date(), + messageId: result.messageId, + }, + }); + + // Record usage for billing (pay-per-email) + // Uses email ID as idempotency key to prevent double-charging on retries + // Charge 2 emails if attachments are present + if (email.project.customer) { + const hasAttachments = email.attachments && Array.isArray(email.attachments) && email.attachments.length > 0; + const emailCount = hasAttachments ? 2 : 1; + await MeterService.recordEmailSent(email.project.customer, emailCount, `email_${emailId}`); + } + + // Track event (this will trigger workflows) + await EventService.trackEvent(email.projectId, 'email.sent', email.contactId, email.id, { + subject: formattedEmail.subject, + from: email.from, + fromName: email.fromName, + messageId: result.messageId, + emailId: email.id, + templateId: email.templateId, + campaignId: email.campaignId, + sourceType: email.sourceType, + sentAt: new Date().toISOString(), + }); + + if (email.campaignId) { + await CampaignService.finalizeIfDone(email.campaignId); + } + } catch (error) { + signale.error(`[EMAIL-PROCESSOR] Failed to send email ${emailId}:`, error); + + // Mark as failed + await prisma.email.update({ + where: {id: emailId}, + data: { + status: EmailStatus.FAILED, + error: error instanceof Error ? error.message : 'Unknown error', + }, + }); + + throw error; // Re-throw to trigger retry + } +} + export async function createEmailWorker() { // Fetch the rate limit (from env, AWS, or default) const rateLimit = await getEmailRateLimit(); @@ -80,220 +299,7 @@ export async function createEmailWorker() { ); const worker = new Worker( emailQueue.name, - async (job: Job) => { - const {emailId} = job.data; - - const email = await prisma.email.findUnique({ - where: {id: emailId}, - include: { - contact: true, - project: true, - template: {select: {type: true}}, - campaign: {select: {type: true}}, - }, - }); - - if (!email) { - throw new Error(`Email ${emailId} not found`); - } - - if (email.status !== EmailStatus.PENDING) { - return; - } - - // Check if project is disabled - if (email.project.disabled) { - signale.warn(`[EMAIL-PROCESSOR] Project ${email.projectId} is disabled, cancelling email ${emailId}`); - await prisma.email.update({ - where: {id: emailId}, - data: { - status: EmailStatus.FAILED, - error: 'Project is disabled', - }, - }); - - // Cancelled emails are terminal for the campaign — finalize so it doesn't - // stay stuck in SENDING forever waiting on emails that will never be sent. - if (email.campaignId) { - await CampaignService.finalizeIfDone(email.campaignId); - } - return; - } - - try { - // Update status to sending - await prisma.email.update({ - where: {id: emailId}, - data: {status: EmailStatus.SENDING}, - }); - - // Format template variables in subject and body - const contactData = (email.contact.data as Record) || {}; - const formattedEmail = EmailService.format({ - subject: email.subject, - body: email.body, - data: { - email: email.contact.email, - ...contactData, - data: contactData, - unsubscribeUrl: `${DASHBOARD_URI}/unsubscribe/${email.contact.id}`, - subscribeUrl: `${DASHBOARD_URI}/subscribe/${email.contact.id}`, - manageUrl: `${DASHBOARD_URI}/manage/${email.contact.id}`, - }, - }); - - // Classify the email once: it decides both the unsubscribe footer and the - // standards-based headers below. - const emailClass = classifyEmail({ - sourceType: email.sourceType, - templateType: email.template?.type, - campaignType: email.campaign?.type, - }); - - // Compile HTML with unsubscribe footer and badge. - // Only marketing emails get the Plunk unsubscribe footer. - const compiledHtml = EmailService.compile({ - content: formattedEmail.body, - contact: email.contact, - project: email.project, - includeUnsubscribe: emailClass === 'marketing', - }); - - // Use fromName from database if available, otherwise fall back to project name - // The 'from' field in the database is just the email address - const fromName = email.fromName || email.project.name; - const fromEmail = email.from; - - // Parse custom headers from JSON - const customHeaders = - email.headers && typeof email.headers === 'object' && !Array.isArray(email.headers) - ? (email.headers as Record) - : undefined; - - // Check for custom recipient override in headers - const recipientEmail = customHeaders?.['X-Plunk-Recipient-Override'] || 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']; - } - - // Build the outbound headers: standards-based defaults for the email class - // plus any caller-supplied headers (which override the defaults). - const outboundHeaders = buildEmailHeaders({ - emailClass, - isCampaign: email.campaignId != null, - hasListManagementLink: bodyHasListManagementLink(compiledHtml, email.contact.id), - unsubscribeId: email.contact.id, - customHeaders: publicHeaders, - }); - - // Build recipient with name if available - const recipient: {name?: string; email: string} | string = email.toName - ? {name: email.toName, email: recipientEmail} - : recipientEmail; - - // Determine tracking based on project settings and email type - const shouldTrack = EmailService.shouldTrackEmail(email.project.tracking, email.sourceType); - - // Check for phishing/dangerous content before sending - const phishingCheck = await SecurityService.checkPhishingContent( - email.projectId, - email.project.name, - email.from, - formattedEmail.subject, - compiledHtml, - ); - - if (phishingCheck.shouldDisable) { - // Disable project immediately - await SecurityService.disableProjectForPhishing( - email.projectId, - formattedEmail.subject, - phishingCheck.confidence, - 'Phishing content detected', - ); - - // Mark email as failed - await prisma.email.update({ - where: {id: emailId}, - data: { - status: EmailStatus.FAILED, - error: 'This email could not be sent. The project has been disabled. Please contact support.', - }, - }); - - throw new Error(`Project ${email.projectId} has been disabled due to a policy violation`); - } - - // Send via AWS SES - const result = await sendRawEmail({ - from: { - name: fromName, - email: fromEmail, - }, - to: typeof recipient === 'string' ? [recipient] : [{name: recipient.name, email: recipient.email}], - content: { - subject: formattedEmail.subject, - html: compiledHtml, - }, - reply: email.replyTo || undefined, - headers: outboundHeaders, - tracking: shouldTrack, - attachments: email.attachments as {filename: string; content: string; contentType: string}[] | null, - }); - - // Mark as sent with SES message ID - await prisma.email.update({ - where: {id: emailId}, - data: { - status: EmailStatus.SENT, - sentAt: new Date(), - messageId: result.messageId, - }, - }); - - // Record usage for billing (pay-per-email) - // Uses email ID as idempotency key to prevent double-charging on retries - // Charge 2 emails if attachments are present - if (email.project.customer) { - const hasAttachments = email.attachments && Array.isArray(email.attachments) && email.attachments.length > 0; - const emailCount = hasAttachments ? 2 : 1; - await MeterService.recordEmailSent(email.project.customer, emailCount, `email_${emailId}`); - } - - // Track event (this will trigger workflows) - await EventService.trackEvent(email.projectId, 'email.sent', email.contactId, email.id, { - subject: formattedEmail.subject, - from: email.from, - fromName: email.fromName, - messageId: result.messageId, - emailId: email.id, - templateId: email.templateId, - campaignId: email.campaignId, - sourceType: email.sourceType, - sentAt: new Date().toISOString(), - }); - - if (email.campaignId) { - await CampaignService.finalizeIfDone(email.campaignId); - } - } catch (error) { - signale.error(`[EMAIL-PROCESSOR] Failed to send email ${emailId}:`, error); - - // Mark as failed - await prisma.email.update({ - where: {id: emailId}, - data: { - status: EmailStatus.FAILED, - error: error instanceof Error ? error.message : 'Unknown error', - }, - }); - - throw error; // Re-throw to trigger retry - } - }, + processEmailJob, { connection: emailQueue.opts.connection, concurrency,