From 616ce56c753ec74048817242b7294d15d00f1560 Mon Sep 17 00:00:00 2001 From: Vlad Bisceanu <7993591+vladbisceanu@users.noreply.github.com> Date: Wed, 26 Aug 2026 19:07:22 +0200 Subject: [PATCH 1/2] fix(webhooks): make SNS processing replay-safe --- .env.self-host.example | 3 + .github/workflows/ci.yml | 2 + CLAUDE.md | 4 +- apps/api/.env.example | 2 + apps/api/src/app/constants.ts | 13 + apps/api/src/controllers/Webhooks.ts | 558 +++++++++++++----- .../Webhooks.sns-idempotency.test.ts | 458 ++++++++++++++ .../jobs/idempotency-key-cleanup-processor.ts | 63 +- apps/api/src/services/EventService.ts | 28 +- apps/api/src/services/SecurityService.ts | 11 +- .../__tests__/SecurityService.sns.test.ts | 75 +++ .../content/docs/self-hosting/email-setup.mdx | 9 +- .../self-hosting/environment-variables.mdx | 1 + docker-compose.yml | 1 + .../migration.sql | 33 ++ packages/db/prisma/schema.prisma | 36 ++ test/setup.ts | 7 +- turbo.json | 3 + 18 files changed, 1141 insertions(+), 166 deletions(-) create mode 100644 apps/api/src/controllers/__tests__/Webhooks.sns-idempotency.test.ts create mode 100644 apps/api/src/services/__tests__/SecurityService.sns.test.ts create mode 100644 packages/db/prisma/migrations/20260826120000_add_sns_webhook_receipts/migration.sql diff --git a/.env.self-host.example b/.env.self-host.example index 557adbd15..3df0b6219 100644 --- a/.env.self-host.example +++ b/.env.self-host.example @@ -42,6 +42,9 @@ USE_HTTPS=false AWS_SES_REGION=us-east-1 AWS_SES_ACCESS_KEY_ID= AWS_SES_SECRET_ACCESS_KEY= +# Exact SNS topics allowed to call /webhooks/sns. Include a separate inbound +# topic too, if used, as a comma-separated ARN. +SNS_TOPIC_ARNS=arn:aws:sns:us-east-1:123456789012:plunk-ses-events # Configuration sets for email tracking # SES_CONFIGURATION_SET: Default configuration with open/click tracking enabled diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 85f9ee795..3f2c710f8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -110,6 +110,7 @@ jobs: AWS_SES_REGION=us-east-1 AWS_SES_ACCESS_KEY_ID=mock AWS_SES_SECRET_ACCESS_KEY=mock + SNS_TOPIC_ARNS=arn:aws:sns:us-east-1:123456789012:plunk-ses-events SES_CONFIGURATION_SET=test SES_CONFIGURATION_SET_NO_TRACKING=test-no-tracking EOF @@ -195,6 +196,7 @@ jobs: AWS_SES_REGION=us-east-1 AWS_SES_ACCESS_KEY_ID=mock AWS_SES_SECRET_ACCESS_KEY=mock + SNS_TOPIC_ARNS=arn:aws:sns:us-east-1:123456789012:plunk-ses-events SES_CONFIGURATION_SET=test SES_CONFIGURATION_SET_NO_TRACKING=test EOF diff --git a/CLAUDE.md b/CLAUDE.md index deb33eb28..01489b879 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -140,8 +140,8 @@ Required for builds and deployment (see turbo.json and .env.example): optional) - S3-compatible Storage (Minio): `S3_ENDPOINT`, `S3_ACCESS_KEY_ID`, `S3_ACCESS_KEY_SECRET`, `S3_BUCKET`, `S3_PUBLIC_URL`, `S3_FORCE_PATH_STYLE` -- AWS SES: `AWS_SES_REGION`, `AWS_SES_ACCESS_KEY_ID`, `AWS_SES_SECRET_ACCESS_KEY`, `SES_CONFIGURATION_SET`, - `SES_CONFIGURATION_SET_NO_TRACKING` +- AWS SES: `AWS_SES_REGION`, `AWS_SES_ACCESS_KEY_ID`, `AWS_SES_SECRET_ACCESS_KEY`, `SNS_TOPIC_ARNS`, + `SES_CONFIGURATION_SET`, `SES_CONFIGURATION_SET_NO_TRACKING` - OAuth (optional): `GITHUB_OAUTH_CLIENT`, `GITHUB_OAUTH_SECRET`, `GOOGLE_OAUTH_CLIENT`, `GOOGLE_OAUTH_SECRET` - Stripe (optional): `STRIPE_SK`, `STRIPE_WEBHOOK_SECRET`, `STRIPE_PRICE_ONBOARDING`, `STRIPE_PRICE_EMAIL_USAGE`, `STRIPE_METER_EVENT_NAME` diff --git a/apps/api/.env.example b/apps/api/.env.example index 96966897f..ea94ad5eb 100644 --- a/apps/api/.env.example +++ b/apps/api/.env.example @@ -54,6 +54,8 @@ S3_FORCE_PATH_STYLE=true AWS_SES_REGION=eu-north-1 AWS_SES_ACCESS_KEY_ID= AWS_SES_SECRET_ACCESS_KEY= +# Exact SNS topics allowed to call /webhooks/sns (comma-separated when using more than one) +SNS_TOPIC_ARNS=arn:aws:sns:eu-north-1:123456789012:plunk-ses-events # Configuration sets for email tracking SES_CONFIGURATION_SET=plunk-configuration-set # Default: with open/click tracking diff --git a/apps/api/src/app/constants.ts b/apps/api/src/app/constants.ts index b2299776b..70e393bca 100644 --- a/apps/api/src/app/constants.ts +++ b/apps/api/src/app/constants.ts @@ -45,6 +45,19 @@ export const AWS_SES_REGION = validateEnv('AWS_SES_REGION'); export const AWS_SES_ACCESS_KEY_ID = validateEnv('AWS_SES_ACCESS_KEY_ID'); export const AWS_SES_SECRET_ACCESS_KEY = validateEnv('AWS_SES_SECRET_ACCESS_KEY'); +// Exact SNS topic ARNs authorized to deliver SES events to /webhooks/sns. +// Multiple topics support deployments that separate outbound and inbound SES. +const snsTopicArns = validateEnv('SNS_TOPIC_ARNS') + .split(',') + .map(topicArn => topicArn.trim()) + .filter(Boolean); + +if (snsTopicArns.length === 0) { + throw new Error('SNS_TOPIC_ARNS must contain at least one topic ARN'); +} + +export const SNS_TOPIC_ARNS: ReadonlySet = new Set(snsTopicArns); + // Custom MAIL FROM subdomain used to construct `.` // when a domain is added. Defaults to `plunk`. Override when `plunk.` // is already used for something else (e.g. a CDN), since the MAIL FROM hostname diff --git a/apps/api/src/controllers/Webhooks.ts b/apps/api/src/controllers/Webhooks.ts index 65045717c..9731c7656 100644 --- a/apps/api/src/controllers/Webhooks.ts +++ b/apps/api/src/controllers/Webhooks.ts @@ -1,6 +1,9 @@ +import {randomUUID} from 'node:crypto'; + import {Controller, Post} from '@overnightjs/core'; import type {Prisma} from '@plunk/db'; import {EmailSourceType, EmailStatus} from '@plunk/db'; +import {toPrismaJson} from '@plunk/types'; import type {Request, Response} from 'express'; import {simpleParser} from 'mailparser'; import sanitizeHtml from 'sanitize-html'; @@ -24,6 +27,157 @@ import {QueueService} from '../services/QueueService.js'; import {SecurityService} from '../services/SecurityService.js'; import {CatchAsync} from '../utils/asyncHandler.js'; +const SNS_CLAIM_LEASE_MS = 5 * 60 * 1000; +const SNS_CLAIM_HEARTBEAT_MS = 60 * 1000; +const SNS_CLAIM_ATTEMPTS = 3; +const SNS_RECEIPT_TTL_MS = 7 * 24 * 60 * 60 * 1000; + +type ActiveSnsClaim = { + messageId: string; + processingToken: string; +}; + +type SnsReceiptClient = Pick; + +type SnsClaimResult = + | {outcome: 'claimed'; claim: ActiveSnsClaim} + | {outcome: 'completed'} + | {outcome: 'in-flight'}; + +function isUniqueConstraintError(error: unknown): boolean { + return error instanceof Error && 'code' in error && error.code === 'P2002'; +} + +/** + * Claim a signed SNS MessageId before applying its side effects. + * + * The unique index decides concurrent races. Completed deliveries are safe to + * acknowledge, active deliveries receive a retryable response, and failed or + * abandoned deliveries are reclaimed with a compare-and-swap update. + */ +async function claimSnsNotification(messageId: string): Promise { + for (let attempt = 0; attempt < SNS_CLAIM_ATTEMPTS; attempt++) { + const processingToken = randomUUID(); + const processingStartedAt = new Date(); + + try { + await prisma.snsWebhookReceipt.create({ + data: { + messageId, + processingToken, + processingStartedAt, + expiresAt: new Date(processingStartedAt.getTime() + SNS_RECEIPT_TTL_MS), + }, + }); + return {outcome: 'claimed', claim: {messageId, processingToken}}; + } catch (error) { + if (!isUniqueConstraintError(error)) { + throw error; + } + } + + const existing = await prisma.snsWebhookReceipt.findUnique({ + where: {messageId}, + select: {status: true, processingToken: true, processingStartedAt: true}, + }); + + // The prior row may have been released between the unique conflict and read. + if (!existing) { + continue; + } + + if (existing.status === 'COMPLETED') { + return {outcome: 'completed'}; + } + + const leaseCutoff = Date.now() - SNS_CLAIM_LEASE_MS; + if (existing.status === 'PROCESSING' && existing.processingStartedAt.getTime() > leaseCutoff) { + return {outcome: 'in-flight'}; + } + + const reclaimed = await prisma.snsWebhookReceipt.updateMany({ + where: { + messageId, + status: existing.status, + processingToken: existing.processingToken, + processingStartedAt: existing.processingStartedAt, + }, + data: { + status: 'PROCESSING', + processingToken, + processingStartedAt, + completedAt: null, + }, + }); + + if (reclaimed.count === 1) { + return {outcome: 'claimed', claim: {messageId, processingToken}}; + } + } + + // Repeated claim races are transient. Do not acknowledge until one delivery + // has durably recorded completion. + return {outcome: 'in-flight'}; +} + +async function completeSnsNotification( + claim: ActiveSnsClaim, + client: SnsReceiptClient = prisma, +): Promise { + const completed = await client.snsWebhookReceipt.updateMany({ + where: { + messageId: claim.messageId, + processingToken: claim.processingToken, + status: 'PROCESSING', + }, + data: {status: 'COMPLETED', completedAt: new Date()}, + }); + + if (completed.count !== 1) { + throw new Error(`SNS claim ${claim.messageId} could not be completed`); + } +} + +async function failSnsNotification(claim: ActiveSnsClaim): Promise { + const failed = await prisma.snsWebhookReceipt.updateMany({ + where: { + messageId: claim.messageId, + processingToken: claim.processingToken, + status: 'PROCESSING', + }, + data: {status: 'FAILED'}, + }); + + if (failed.count !== 1) { + signale.warn(`[WEBHOOK] SNS claim ${claim.messageId} was no longer active while recording failure`); + } +} + +function startSnsClaimHeartbeat(claim: ActiveSnsClaim): () => void { + const timer = setInterval(() => { + void prisma.snsWebhookReceipt + .updateMany({ + where: { + messageId: claim.messageId, + processingToken: claim.processingToken, + status: 'PROCESSING', + }, + data: {processingStartedAt: new Date()}, + }) + .then(renewed => { + if (renewed.count !== 1) { + signale.warn(`[WEBHOOK] SNS claim ${claim.messageId} was no longer active during heartbeat`); + } + }) + .catch(error => { + signale.error(`[WEBHOOK] Failed to renew SNS claim ${claim.messageId}:`, error); + }); + }, SNS_CLAIM_HEARTBEAT_MS); + + timer.unref(); + return () => clearInterval(timer); +} + /** * Webhooks Controller * Handles incoming webhooks from external services (AWS SNS/SES) @@ -38,6 +192,16 @@ export class Webhooks { @Post('sns') @CatchAsync public async receiveSNSWebhook(req: Request, res: Response) { + let activeSnsClaim: ActiveSnsClaim | undefined; + let stopSnsClaimHeartbeat: (() => void) | undefined; + + const completeActiveClaim = async () => { + if (!activeSnsClaim) return; + + await completeSnsNotification(activeSnsClaim); + activeSnsClaim = undefined; + }; + try { // Verify SNS message signature before processing anything const signatureValid = await SecurityService.verifySnsSignature(req.body as Record); @@ -89,14 +253,14 @@ export class Webhooks { }); } else { signale.error('Failed to confirm SNS subscription:', confirmResponse.statusText); - return res.status(200).json({ + return res.status(502).json({ success: false, message: 'Failed to confirm subscription', }); } } catch (confirmError) { signale.error('Error confirming SNS subscription:', confirmError); - return res.status(200).json({ + return res.status(502).json({ success: false, message: 'Error confirming subscription', }); @@ -109,23 +273,75 @@ export class Webhooks { return res.status(200).json({success: false, message: 'Unknown message type'}); } + const snsMessageId: unknown = req.body.MessageId; + if (typeof snsMessageId !== 'string' || snsMessageId.length === 0) { + signale.warn('[WEBHOOK] SNS notification missing MessageId'); + return res.status(400).json({success: false, message: 'Missing SNS MessageId'}); + } + // Parse the nested SES event from the Message field const body = JSON.parse(req.body.Message); + const claimResult = await claimSnsNotification(snsMessageId); + if (claimResult.outcome === 'completed') { + return res.status(200).json({success: true, duplicate: true}); + } + if (claimResult.outcome === 'in-flight') { + return res.status(503).json({success: false, message: 'SNS notification is already being processed'}); + } + activeSnsClaim = claimResult.claim; + stopSnsClaimHeartbeat = startSnsClaimHeartbeat(activeSnsClaim); + // Check if this is an inbound email notification (SES Receiving) if (body.notificationType === 'Received') { signale.info('[WEBHOOK] Received inbound email notification from SES'); try { - // Extract recipient addresses from the inbound email const recipients = body.receipt?.recipients || []; if (recipients.length === 0) { signale.warn('[WEBHOOK] No recipients found in inbound email'); + await completeActiveClaim(); return res.status(200).json({success: true, message: 'No recipients found'}); } - // For each recipient, identify the domain and create events + const senderEmail = body.mail?.source; + if (typeof senderEmail !== 'string' || senderEmail.length === 0) { + throw new Error('Inbound SNS notification is missing mail.source'); + } + const normalizedSender = ContactService.normalizeEmail(senderEmail); + const senderFromHeader = body.mail?.commonHeaders?.from?.[0] || senderEmail; + let htmlBody: string | undefined; + + if (body.content && typeof body.content === 'string') { + try { + const isBase64 = body.receipt?.action?.encoding === 'BASE64'; + const emailBuffer = isBase64 ? Buffer.from(body.content, 'base64') : Buffer.from(body.content); + const parsed = await simpleParser(emailBuffer); + const raw = + (parsed.html ? String(parsed.html) : undefined) ?? parsed.textAsHtml ?? parsed.text ?? undefined; + + if (raw) { + htmlBody = sanitizeHtml(raw, { + allowedTags: sanitizeHtml.defaults.allowedTags.concat(['img']), + allowedAttributes: { + ...sanitizeHtml.defaults.allowedAttributes, + img: ['src', 'alt', 'width', 'height'], + '*': ['style'], + }, + allowedSchemes: ['http', 'https', 'mailto'], + }); + } + } catch (parseError) { + signale.error('[WEBHOOK] Failed to parse email content:', parseError); + } + } + + const targets: Array<{ + recipientEmail: string; + project: {id: string; name: string; customer: string | null}; + }> = []; + for (const recipient of recipients) { const recipientEmail = recipient as string; const domain = recipientEmail.split('@')[1]; @@ -152,140 +368,132 @@ export class Webhooks { continue; } - signale.info( - `[WEBHOOK] Found ${domainRecords.length} project(s) with verified domain ${domain}. Processing inbound email for all.`, - ); - - // Extract sender information (same for all projects) - const senderEmail = body.mail?.source; - const senderFromHeader = body.mail?.commonHeaders?.from?.[0] || senderEmail; - - // Parse email content if available - let htmlBody: string | undefined; - - if (body.content && typeof body.content === 'string') { - try { - const isBase64 = body.receipt?.action?.encoding === 'BASE64'; - const emailBuffer = isBase64 - ? Buffer.from(body.content, 'base64') - : Buffer.from(body.content); - - const parsed = await simpleParser(emailBuffer); - const raw = - (parsed.html ? String(parsed.html) : undefined) ?? - parsed.textAsHtml ?? - parsed.text ?? - undefined; - - if (raw) { - htmlBody = sanitizeHtml(raw, { - allowedTags: sanitizeHtml.defaults.allowedTags.concat(['img']), - allowedAttributes: { - ...sanitizeHtml.defaults.allowedAttributes, - img: ['src', 'alt', 'width', 'height'], - '*': ['style'], - }, - allowedSchemes: ['http', 'https', 'mailto'], - }); - } - } catch (parseError) { - signale.error('[WEBHOOK] Failed to parse email content:', parseError); - } - } - - // Process inbound email for each project that has this domain verified for (const domainRecord of domainRecords) { signale.info(`[WEBHOOK] Processing inbound email for project: ${domainRecord.project.name}`); - - // Check billing limits before processing inbound email const limitCheck = await BillingLimitService.checkLimit(domainRecord.projectId, EmailSourceType.INBOUND); if (!limitCheck.allowed) { signale.warn( `[WEBHOOK] Inbound email blocked for project ${domainRecord.project.name}: ${limitCheck.message}`, ); - continue; // Skip this project but continue processing for other projects + continue; } - // Find or create a contact for the sender in this project - let contact; - if (senderEmail) { - contact = await ContactService.upsert( - domainRecord.projectId, - senderEmail, - undefined, // No additional data - true, // Subscribe by default for inbound email senders - ); + targets.push({recipientEmail, project: domainRecord.project}); + } + } + + const claim = activeSnsClaim; + if (!claim) { + throw new Error(`SNS notification ${snsMessageId} lost its processing claim`); + } + + const committed = await prisma.$transaction(async tx => { + const effects: Array<{ + emailId: string; + eventIds: string[]; + project: {id: string; name: string; customer: string | null}; + recipientEmail: string; + }> = []; + + for (const target of targets) { + const existingContact = await tx.contact.findUnique({ + where: {projectId_email: {projectId: target.project.id, email: normalizedSender}}, + }); + const contact = existingContact + ? await tx.contact.update({ + where: {id: existingContact.id}, + data: {subscribed: true}, + }) + : await tx.contact.create({ + data: {projectId: target.project.id, email: normalizedSender, subscribed: true}, + }); + const eventIds: string[] = []; + + if (existingContact && !existingContact.subscribed) { + const subscribedEvent = await tx.event.create({ + data: { + projectId: target.project.id, + contactId: contact.id, + name: 'contact.subscribed', + }, + }); + eventIds.push(subscribedEvent.id); } - // Create an Email record for tracking with parsed content - const inboundEmail = await prisma.email.create({ + const inboundEmail = await tx.email.create({ data: { - projectId: domainRecord.projectId, - contactId: contact!.id, + projectId: target.project.id, + contactId: contact.id, subject: body.mail?.commonHeaders?.subject || '(No subject)', - body: htmlBody || '', // Store HTML body in the body field - from: recipientEmail, // The recipient address that received the email + body: htmlBody || '', + from: target.recipientEmail, sourceType: EmailSourceType.INBOUND, - status: EmailStatus.RECEIVED, // Inbound emails use RECEIVED status + status: EmailStatus.RECEIVED, deliveredAt: new Date(body.mail?.timestamp || new Date()), }, }); - - // Increment usage counter in cache - await BillingLimitService.incrementUsage(domainRecord.projectId, EmailSourceType.INBOUND); - - // Record Stripe metering if project has customer - if (domainRecord.project.customer) { - await MeterService.recordEmailSent( - domainRecord.project.customer, - 1, // Inbound emails count as 1 credit - `email_${inboundEmail.id}`, - ); - } - - // Prepare event data with all inbound email details including body content const eventData = { messageId: body.mail?.messageId, from: senderEmail, fromHeader: senderFromHeader, - to: recipientEmail, + to: target.recipientEmail, subject: body.mail?.commonHeaders?.subject, timestamp: body.mail?.timestamp, recipients: body.receipt?.recipients, hasContent: !!body.content, - // Email body content body: htmlBody, - // Security verdicts spamVerdict: body.receipt?.spamVerdict?.status, virusVerdict: body.receipt?.virusVerdict?.status, spfVerdict: body.receipt?.spfVerdict?.status, dkimVerdict: body.receipt?.dkimVerdict?.status, dmarcVerdict: body.receipt?.dmarcVerdict?.status, - // Processing metadata processingTimeMillis: body.receipt?.processingTimeMillis, }; + const receivedEvent = await tx.event.create({ + data: { + projectId: target.project.id, + contactId: contact.id, + emailId: inboundEmail.id, + name: 'email.received', + data: toPrismaJson(eventData), + }, + }); + eventIds.push(receivedEvent.id); + effects.push({ + emailId: inboundEmail.id, + eventIds, + project: target.project, + recipientEmail: target.recipientEmail, + }); + } - // Create the email.received event (this will trigger workflows) - await EventService.trackEvent( - domainRecord.projectId, - 'email.received', - contact?.id, - inboundEmail.id, // Link the event to the inbound email record - eventData, - ); + await completeSnsNotification(claim, tx); + return effects; + }); + activeSnsClaim = undefined; - signale.success( - `[WEBHOOK] Created email.received event for ${senderEmail} → ${recipientEmail} (project: ${domainRecord.project.name})`, - ); + for (const effect of committed) { + await BillingLimitService.incrementUsage(effect.project.id, EmailSourceType.INBOUND); + if (effect.project.customer) { + await MeterService.recordEmailSent(effect.project.customer, 1, `email_${effect.emailId}`); + } + for (const eventId of effect.eventIds) { + try { + await EventService.dispatchStoredEvent(eventId); + } catch (dispatchError) { + signale.error(`[WEBHOOK] Deferred workflow dispatch for event ${eventId}:`, dispatchError); + } } + signale.success( + `[WEBHOOK] Created email.received event for ${senderEmail} → ${effect.recipientEmail} (project: ${effect.project.name})`, + ); } return res.status(200).json({success: true, message: 'Inbound email processed'}); } catch (inboundError) { signale.error('[WEBHOOK] Error processing inbound email:', inboundError); - // Return 200 to acknowledge receipt even if processing failed - return res.status(200).json({success: true, message: 'Error processing inbound email'}); + throw inboundError; } } @@ -295,6 +503,7 @@ export class Webhooks { if (!messageId) { signale.warn('[WEBHOOK] No messageId found in SNS notification'); + await completeActiveClaim(); return res.status(400).json({success: false, error: 'No messageId found'}); } @@ -309,15 +518,22 @@ export class Webhooks { if (!email) { // Error level for the same reason as a signature failure: an event that matches no - // email row is silently lost, and SES gives up after its retries. A run of these - // means the send path is not stamping `messageId`, which is invisible from outside. - signale.error(`[WEBHOOK] ${eventType} event dropped — no email found for messageId: ${messageId}`); - return res.status(404).json({success: false, error: 'Email not found'}); + // email row is silently lost. A run of these means the send path is not stamping + // `messageId`, which is invisible from outside. + signale.error(`[WEBHOOK] ${eventType} event has no email for messageId: ${messageId}`); + // SES can publish before the sender has persisted its returned messageId. + // Keep the receipt retryable so that race cannot permanently lose the event. + throw new Error(`Email not found for messageId: ${messageId}`); } const now = new Date(); const updateData: Prisma.EmailUpdateInput = {}; const eventName = `email.${eventType.toLowerCase()}`; + let unsubscribeContact = false; + let bounceNotification = false; + let bounceNotificationType: string | undefined; + let complaintNotification = false; + let enforceSecurityLimits = false; // Base event data with email metadata const baseEventData = { @@ -350,14 +566,8 @@ export class Webhooks { if (!email.openedAt) { updateData.openedAt = now; } - updateData.opens = (email.opens || 0) + 1; + updateData.opens = {increment: 1}; updateData.status = EmailStatus.OPENED; - eventData = { - ...baseEventData, - openedAt: email.openedAt?.toISOString() || now.toISOString(), - opens: (email.opens || 0) + 1, - isFirstOpen: !email.openedAt, - }; break; case 'Click': { @@ -367,14 +577,11 @@ export class Webhooks { if (!email.clickedAt) { updateData.clickedAt = now; } - updateData.clicks = (email.clicks || 0) + 1; + updateData.clicks = {increment: 1}; updateData.status = EmailStatus.CLICKED; eventData = { ...baseEventData, link: clickedLink, - clickedAt: email.clickedAt?.toISOString() || now.toISOString(), - clicks: (email.clicks || 0) + 1, - isFirstClick: !email.clickedAt, }; break; } @@ -389,19 +596,15 @@ export class Webhooks { signale.warn(`[WEBHOOK] Permanent bounce received for ${email.contact.email} from ${email.project.name}`); updateData.status = EmailStatus.BOUNCED; updateData.bouncedAt = now; - // Unsubscribe contact on permanent bounce - await prisma.contact.update({ - where: {id: email.contactId}, - data: {subscribed: false}, - }); + unsubscribeContact = true; + bounceNotification = true; + bounceNotificationType = bounceType; + enforceSecurityLimits = true; eventData = { ...baseEventData, bounceType, bouncedAt: now.toISOString(), }; - - // Send notification about permanent bounce - await NtfyService.notifyEmailBounce(email.project.name, email.projectId, email.contact.email, bounceType); } else if (isTransientBounce) { // Soft bounce (e.g., out-of-office, mailbox full) - don't count toward bounce rate signale.info( @@ -421,17 +624,14 @@ export class Webhooks { ); updateData.status = EmailStatus.BOUNCED; updateData.bouncedAt = now; - await prisma.contact.update({ - where: {id: email.contactId}, - data: {subscribed: false}, - }); + unsubscribeContact = true; + bounceNotification = true; + bounceNotificationType = bounceType; eventData = { ...baseEventData, bounceType, bouncedAt: now.toISOString(), }; - - await NtfyService.notifyEmailBounce(email.project.name, email.projectId, email.contact.email, bounceType); } break; } @@ -440,30 +640,71 @@ export class Webhooks { signale.warn(`[WEBHOOK] Complaint received for ${email.contact.email} from ${email.project.name}`); updateData.status = EmailStatus.COMPLAINED; updateData.complainedAt = now; - // Unsubscribe contact on complaint - await prisma.contact.update({ - where: {id: email.contactId}, - data: {subscribed: false}, - }); + unsubscribeContact = true; + complaintNotification = true; + enforceSecurityLimits = true; eventData = { ...baseEventData, complainedAt: now.toISOString(), }; - - // Send notification about complaint - await NtfyService.notifyEmailComplaint(email.project.name, email.projectId, email.contact.email); break; default: signale.warn(`[WEBHOOK] Unknown event type: ${eventType}`); + await completeActiveClaim(); return res.status(200).json({success: true}); } - // Update email with new status and timestamps - await prisma.email.update({ - where: {id: email.id}, - data: updateData, + const claim = activeSnsClaim; + if (!claim) { + throw new Error(`SNS notification ${snsMessageId} lost its processing claim`); + } + + // The business mutation, durable event, and receipt completion share one + // commit. A database failure therefore leaves no partial effects for the + // SNS retry to duplicate. + const storedEvent = await prisma.$transaction(async tx => { + if (unsubscribeContact) { + await tx.contact.update({ + where: {id: email.contactId}, + data: {subscribed: false}, + }); + } + + const updatedEmail = await tx.email.update({ + where: {id: email.id}, + data: updateData, + }); + + if (eventType === 'Open') { + eventData = { + ...baseEventData, + openedAt: updatedEmail.openedAt?.toISOString(), + opens: updatedEmail.opens, + isFirstOpen: !email.openedAt, + }; + } else if (eventType === 'Click') { + eventData = { + ...eventData, + clickedAt: updatedEmail.clickedAt?.toISOString(), + clicks: updatedEmail.clicks, + isFirstClick: !email.clickedAt, + }; + } + + const event = await tx.event.create({ + data: { + projectId: email.projectId, + contactId: email.contactId, + emailId: email.id, + name: eventName, + data: toPrismaJson(eventData), + }, + }); + await completeSnsNotification(claim, tx); + return event; }); + activeSnsClaim = undefined; // The campaign counters the stats endpoint reads live on the campaign row, and this // event has just moved one of them. They are not incremented from here: this handler @@ -474,13 +715,30 @@ export class Webhooks { await CampaignService.markStatsDirty(email.campaignId); } - // Track event (this will trigger workflows) - await EventService.trackEvent(email.projectId, eventName, email.contactId, email.id, eventData); + if (bounceNotification) { + try { + await NtfyService.notifyEmailBounce( + email.project.name, + email.projectId, + email.contact.email, + bounceNotificationType, + ); + } catch (notificationError) { + signale.error('[WEBHOOK] Failed to notify about email bounce:', notificationError); + } + } else if (complaintNotification) { + await NtfyService.notifyEmailComplaint(email.project.name, email.projectId, email.contact.email); + } + + try { + await EventService.dispatchStoredEvent(storedEvent.id); + } catch (dispatchError) { + // The event row is the outbox. The maintenance worker can retry a null + // processedAt without asking SNS to replay committed email effects. + signale.error(`[WEBHOOK] Deferred workflow dispatch for event ${storedEvent.id}:`, dispatchError); + } - // Check security limits only for permanent bounces and complaints - // Transient bounces (soft bounces) don't count toward bounce rate - const isPermanentBounce = eventType === 'Bounce' && body.bounce?.bounceType === 'Permanent'; - if (isPermanentBounce || eventType === 'Complaint') { + if (enforceSecurityLimits) { await SecurityService.checkAndEnforceSecurityLimits(email.projectId); } @@ -488,8 +746,18 @@ export class Webhooks { return res.status(200).json({success: true}); } catch (error) { signale.error('[WEBHOOK] Error processing SNS webhook:', error); - // Always return 200 to prevent SNS from retrying - return res.status(200).json({success: true}); + if (activeSnsClaim) { + try { + await failSnsNotification(activeSnsClaim); + } catch (settleError) { + // The processing lease is the fallback if the database is unavailable + // while recording failure. The delivery still receives 5xx and retries. + signale.error(`[WEBHOOK] Failed to release SNS claim ${activeSnsClaim.messageId}:`, settleError); + } + } + return res.status(500).json({success: false, message: 'Failed to process SNS notification'}); + } finally { + stopSnsClaimHeartbeat?.(); } } diff --git a/apps/api/src/controllers/__tests__/Webhooks.sns-idempotency.test.ts b/apps/api/src/controllers/__tests__/Webhooks.sns-idempotency.test.ts new file mode 100644 index 000000000..9f4049352 --- /dev/null +++ b/apps/api/src/controllers/__tests__/Webhooks.sns-idempotency.test.ts @@ -0,0 +1,458 @@ +import type {Request, Response} from 'express'; +import type {Job} from 'bullmq'; +import type {Prisma} from '@plunk/db'; +import type {IdempotencyKeyCleanupJobData} from '@plunk/types'; +import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest'; + +import {factories, getPrismaClient} from '../../../../../test/helpers'; +import {prisma as controllerPrisma} from '../../database/prisma.js'; +import {processCleanup} from '../../jobs/idempotency-key-cleanup-processor.js'; +import {EventService} from '../../services/EventService.js'; +import {SecurityService} from '../../services/SecurityService.js'; +import {Webhooks} from '../Webhooks.js'; + +const SNS_TOPIC_ARN = 'arn:aws:sns:us-east-1:123456789012:plunk-ses-events'; + +function mockResponse() { + const captured = {status: 200, body: undefined as unknown}; + + let markSent: () => void; + const sent = new Promise(resolve => { + markSent = resolve; + }); + + const res = { + status(code: number) { + captured.status = code; + return this; + }, + json(body: unknown) { + captured.body = body; + markSent(); + return this; + }, + } as unknown as Response; + + return {res, captured, sent}; +} + +function notification( + snsMessageId: string, + sesMessageId: string, + eventType: 'Bounce' | 'Delivery' | 'Open' | 'Complaint' | 'Click' = 'Delivery', +): Request { + return { + body: { + Type: 'Notification', + MessageId: snsMessageId, + TopicArn: SNS_TOPIC_ARN, + Message: JSON.stringify({ + eventType, + mail: {messageId: sesMessageId}, + }), + }, + } as Request; +} + +function subscriptionConfirmation(): Request { + return { + body: { + Type: 'SubscriptionConfirmation', + MessageId: 'sns-subscription-confirmation', + TopicArn: SNS_TOPIC_ARN, + SubscribeURL: 'https://sns.eu-west-1.amazonaws.com/?Action=ConfirmSubscription', + }, + } as Request; +} + +function inboundNotification(snsMessageId: string, recipient: string): Request { + return { + body: { + Type: 'Notification', + MessageId: snsMessageId, + TopicArn: SNS_TOPIC_ARN, + Message: JSON.stringify({ + notificationType: 'Received', + mail: { + messageId: `ses-${snsMessageId}`, + source: 'Sender@external.example', + timestamp: new Date().toISOString(), + commonHeaders: {from: ['Sender '], subject: 'Inbound test'}, + }, + receipt: {recipients: [recipient]}, + }), + }, + } as Request; +} + +describe('SNS webhook delivery receipts', () => { + const prisma = getPrismaClient(); + const controller = new Webhooks(); + const next = vi.fn(); + + let projectId: string; + let contactId: string; + let sesMessageId: string; + + beforeEach(async () => { + const {project} = await factories.createUserWithProject(); + const contact = await factories.createContact({projectId: project.id}); + projectId = project.id; + contactId = contact.id; + sesMessageId = `ses-${project.id}`; + await factories.createEmail({projectId: project.id, contactId: contact.id, messageId: sesMessageId}); + + vi.spyOn(SecurityService, 'verifySnsSignature').mockResolvedValue(true); + vi.spyOn(EventService, 'dispatchStoredEvent').mockResolvedValue(undefined); + }); + + afterEach(async () => { + await prisma.snsWebhookReceipt.deleteMany(); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + async function invoke(req: Request) { + const {res, captured, sent} = mockResponse(); + const handler = controller.receiveSNSWebhook as unknown as ( + req: Request, + res: Response, + next: (error?: unknown) => void, + ) => void; + handler(req, res, next); + await sent; + expect(next).not.toHaveBeenCalled(); + return captured; + } + + function deliver( + snsMessageId: string, + innerMessageId = sesMessageId, + eventType: 'Bounce' | 'Delivery' | 'Open' | 'Complaint' | 'Click' = 'Delivery', + ) { + return invoke(notification(snsMessageId, innerMessageId, eventType)); + } + + function failNextTransactionAfterEventInsert() { + const transaction = controllerPrisma.$transaction.bind(controllerPrisma); + vi.spyOn(controllerPrisma, '$transaction').mockImplementationOnce( + (async (callback: (tx: Prisma.TransactionClient) => Promise) => + transaction(async tx => { + const createEvent = tx.event.create.bind(tx.event); + vi.spyOn(tx.event, 'create').mockImplementationOnce(async args => { + await createEvent(args); + throw new Error('injected database failure after event insert'); + }); + return callback(tx); + })) as typeof controllerPrisma.$transaction, + ); + } + + it('acknowledges a completed replay without applying its effects twice', async () => { + const first = await deliver('sns-sequential-replay'); + const replay = await deliver('sns-sequential-replay'); + + expect(first.status).toBe(200); + expect(replay).toMatchObject({status: 200, body: {success: true, duplicate: true}}); + expect(SecurityService.verifySnsSignature).toHaveBeenCalledTimes(2); + expect(EventService.dispatchStoredEvent).toHaveBeenCalledTimes(1); + const email = await prisma.email.findUniqueOrThrow({where: {messageId: sesMessageId}}); + expect(await prisma.event.count({where: {emailId: email.id}})).toBe(1); + expect(await prisma.email.findUniqueOrThrow({where: {messageId: sesMessageId}})).toMatchObject({ + status: 'DELIVERED', + }); + + const receipt = await prisma.snsWebhookReceipt.findUniqueOrThrow({ + where: {messageId: 'sns-sequential-replay'}, + }); + expect(receipt.status).toBe('COMPLETED'); + expect(receipt.completedAt).not.toBeNull(); + expect(receipt.expiresAt.getTime()).toBeGreaterThan(Date.now() + 6 * 24 * 60 * 60 * 1000); + }); + + it('rejects an untrusted topic before parsing its nested message or claiming a receipt', async () => { + vi.mocked(SecurityService.verifySnsSignature).mockRestore(); + + const response = await invoke({ + body: { + Type: 'Notification', + MessageId: 'sns-untrusted-topic', + TopicArn: 'arn:aws:sns:us-east-1:999999999999:plunk-ses-events', + Message: '{not-valid-json', + }, + } as Request); + + expect(response).toMatchObject({status: 403, body: {success: false}}); + expect(EventService.dispatchStoredEvent).not.toHaveBeenCalled(); + expect(await prisma.snsWebhookReceipt.count()).toBe(0); + }); + + it('rejects an untrusted subscription before following its confirmation URL', async () => { + vi.mocked(SecurityService.verifySnsSignature).mockRestore(); + const fetchMock = vi.fn(); + vi.stubGlobal('fetch', fetchMock); + + const response = await invoke({ + body: { + Type: 'SubscriptionConfirmation', + MessageId: 'sns-untrusted-subscription', + TopicArn: 'arn:aws:sns:us-east-1:999999999999:plunk-ses-events', + SubscribeURL: 'https://sns.us-east-1.amazonaws.com/?Action=ConfirmSubscription', + }, + } as Request); + + expect(response).toMatchObject({status: 403, body: {success: false}}); + expect(fetchMock).not.toHaveBeenCalled(); + expect(await prisma.snsWebhookReceipt.count()).toBe(0); + }); + + it('rolls back partial email and event effects when the database fails, then retries once', async () => { + failNextTransactionAfterEventInsert(); + + const failed = await deliver('sns-db-failure', sesMessageId, 'Open'); + + expect(failed.status).toBe(500); + expect(EventService.dispatchStoredEvent).not.toHaveBeenCalled(); + expect(await prisma.email.findUniqueOrThrow({where: {messageId: sesMessageId}})).toMatchObject({ + status: 'PENDING', + opens: 0, + openedAt: null, + }); + expect(await prisma.event.count({where: {projectId, name: 'email.open'}})).toBe(0); + expect( + await prisma.snsWebhookReceipt.findUniqueOrThrow({where: {messageId: 'sns-db-failure'}}), + ).toMatchObject({status: 'FAILED', completedAt: null}); + + const retry = await deliver('sns-db-failure', sesMessageId, 'Open'); + + expect(retry.status).toBe(200); + expect(EventService.dispatchStoredEvent).toHaveBeenCalledTimes(1); + expect(await prisma.email.findUniqueOrThrow({where: {messageId: sesMessageId}})).toMatchObject({ + status: 'OPENED', + opens: 1, + }); + expect(await prisma.event.count({where: {projectId, name: 'email.open'}})).toBe(1); + expect( + await prisma.snsWebhookReceipt.findUniqueOrThrow({where: {messageId: 'sns-db-failure'}}), + ).toMatchObject({status: 'COMPLETED'}); + }); + + it('acknowledges after commit when workflow dispatch fails and leaves the event recoverable', async () => { + vi.mocked(EventService.dispatchStoredEvent).mockRejectedValueOnce(new Error('injected workflow failure')); + + const first = await deliver('sns-dispatch-failure'); + const replay = await deliver('sns-dispatch-failure'); + + expect(first.status).toBe(200); + expect(replay).toMatchObject({status: 200, body: {success: true, duplicate: true}}); + const event = await prisma.event.findFirstOrThrow({where: {projectId, name: 'email.delivery'}}); + expect(event.processedAt).toBeNull(); + expect(await prisma.event.count({where: {projectId, name: 'email.delivery'}})).toBe(1); + expect( + await prisma.snsWebhookReceipt.findUniqueOrThrow({where: {messageId: 'sns-dispatch-failure'}}), + ).toMatchObject({status: 'COMPLETED'}); + + vi.mocked(EventService.dispatchStoredEvent).mockRestore(); + await EventService.dispatchStoredEvent(event.id); + expect((await prisma.event.findUniqueOrThrow({where: {id: event.id}})).processedAt).not.toBeNull(); + }); + + it('rolls back inbound contact, email, and event effects before retrying once', async () => { + await factories.createDomain({projectId, domain: 'inbound.example', verified: true}); + failNextTransactionAfterEventInsert(); + + const failed = await invoke(inboundNotification('sns-inbound-db-failure', 'reply@inbound.example')); + + expect(failed.status).toBe(500); + expect( + await prisma.contact.findUnique({ + where: {projectId_email: {projectId, email: 'sender@external.example'}}, + }), + ).toBeNull(); + expect(await prisma.email.count({where: {projectId, sourceType: 'INBOUND'}})).toBe(0); + expect(await prisma.event.count({where: {projectId, name: 'email.received'}})).toBe(0); + expect( + await prisma.snsWebhookReceipt.findUniqueOrThrow({where: {messageId: 'sns-inbound-db-failure'}}), + ).toMatchObject({status: 'FAILED'}); + + const retry = await invoke(inboundNotification('sns-inbound-db-failure', 'reply@inbound.example')); + + expect(retry.status).toBe(200); + expect( + await prisma.contact.findUnique({ + where: {projectId_email: {projectId, email: 'sender@external.example'}}, + }), + ).not.toBeNull(); + expect(await prisma.email.count({where: {projectId, sourceType: 'INBOUND'}})).toBe(1); + expect(await prisma.event.count({where: {projectId, name: 'email.received'}})).toBe(1); + }); + + it('retries an event that arrives before its SES messageId is persisted', async () => { + const early = await deliver('sns-send-persist-race', 'ses-not-persisted-yet'); + + expect(early.status).toBe(500); + expect( + await prisma.snsWebhookReceipt.findUniqueOrThrow({where: {messageId: 'sns-send-persist-race'}}), + ).toMatchObject({status: 'FAILED'}); + + await factories.createEmail({projectId, contactId, messageId: 'ses-not-persisted-yet'}); + const retry = await deliver('sns-send-persist-race', 'ses-not-persisted-yet'); + + expect(retry.status).toBe(200); + expect(EventService.dispatchStoredEvent).toHaveBeenCalledTimes(1); + expect( + await prisma.snsWebhookReceipt.findUniqueOrThrow({where: {messageId: 'sns-send-persist-race'}}), + ).toMatchObject({status: 'COMPLETED'}); + }); + + it('reclaims an abandoned processing receipt instead of losing the delivery', async () => { + await prisma.snsWebhookReceipt.create({ + data: { + messageId: 'sns-abandoned-claim', + processingToken: 'abandoned-worker', + processingStartedAt: new Date(Date.now() - 10 * 60 * 1000), + expiresAt: new Date(Date.now() + 7 * 24 * 60 * 60 * 1000), + }, + }); + + const retry = await deliver('sns-abandoned-claim'); + + expect(retry.status).toBe(200); + expect(EventService.dispatchStoredEvent).toHaveBeenCalledTimes(1); + expect( + await prisma.snsWebhookReceipt.findUniqueOrThrow({where: {messageId: 'sns-abandoned-claim'}}), + ).toMatchObject({status: 'COMPLETED'}); + }); + + it('does not reclaim a receipt after its owner renews the observed lease', async () => { + const originalToken = 'active-worker'; + await prisma.snsWebhookReceipt.create({ + data: { + messageId: 'sns-renewed-claim', + processingToken: originalToken, + processingStartedAt: new Date(Date.now() - 10 * 60 * 1000), + expiresAt: new Date(Date.now() + 7 * 24 * 60 * 60 * 1000), + }, + }); + + const findReceipt = controllerPrisma.snsWebhookReceipt.findUnique.bind(controllerPrisma.snsWebhookReceipt); + vi.spyOn(controllerPrisma.snsWebhookReceipt, 'findUnique').mockImplementationOnce(async args => { + const observed = await findReceipt(args); + await prisma.snsWebhookReceipt.update({ + where: {messageId: 'sns-renewed-claim'}, + data: {processingStartedAt: new Date()}, + }); + return observed; + }); + + const response = await deliver('sns-renewed-claim'); + + expect(response.status).toBe(503); + expect(EventService.dispatchStoredEvent).not.toHaveBeenCalled(); + expect( + await prisma.snsWebhookReceipt.findUniqueOrThrow({where: {messageId: 'sns-renewed-claim'}}), + ).toMatchObject({status: 'PROCESSING', processingToken: originalToken}); + }); + + it('returns a retryable response to a concurrent duplicate', async () => { + let releaseUpdate!: () => void; + const updateGate = new Promise(resolve => { + releaseUpdate = resolve; + }); + let markUpdateStarted!: () => void; + const updateStarted = new Promise(resolve => { + markUpdateStarted = resolve; + }); + + const transaction = controllerPrisma.$transaction.bind(controllerPrisma); + vi.spyOn(controllerPrisma, '$transaction').mockImplementationOnce( + (async (callback: (tx: Prisma.TransactionClient) => Promise) => + transaction(async tx => { + const updateEmail = tx.email.update.bind(tx.email); + vi.spyOn(tx.email, 'update').mockImplementationOnce(async args => { + markUpdateStarted(); + await updateGate; + return updateEmail(args); + }); + return callback(tx); + })) as typeof controllerPrisma.$transaction, + ); + + const firstDelivery = deliver('sns-concurrent-replay'); + await updateStarted; + + const concurrentReplay = await deliver('sns-concurrent-replay'); + expect(concurrentReplay.status).toBe(503); + expect(EventService.dispatchStoredEvent).not.toHaveBeenCalled(); + + releaseUpdate(); + expect((await firstDelivery).status).toBe(200); + expect(EventService.dispatchStoredEvent).toHaveBeenCalledTimes(1); + }); + + it('keeps successful subscription confirmation behavior', async () => { + const fetchMock = vi.fn().mockResolvedValue({ok: true}); + vi.stubGlobal('fetch', fetchMock); + + const response = await invoke(subscriptionConfirmation()); + + expect(response).toMatchObject({status: 200, body: {success: true, message: 'Subscription confirmed'}}); + expect(fetchMock).toHaveBeenCalledWith( + 'https://sns.eu-west-1.amazonaws.com/?Action=ConfirmSubscription', + ); + expect(await prisma.snsWebhookReceipt.count()).toBe(0); + }); + + it('returns 5xx when subscription confirmation fails so SNS can retry', async () => { + vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ok: false, statusText: 'upstream failed'})); + + const response = await invoke(subscriptionConfirmation()); + + expect(response).toMatchObject({status: 502, body: {success: false}}); + expect(await prisma.snsWebhookReceipt.count()).toBe(0); + }); + + it('removes expired delivery receipts while retaining the replay window', async () => { + const now = Date.now(); + const pendingEvent = await prisma.event.create({ + data: { + projectId, + contactId, + name: 'outbox.recovery', + createdAt: new Date(now - 10 * 60 * 1000), + }, + }); + await prisma.snsWebhookReceipt.createMany({ + data: [ + { + messageId: 'sns-expired', + processingToken: 'expired', + status: 'COMPLETED', + completedAt: new Date(now - 8 * 24 * 60 * 60 * 1000), + expiresAt: new Date(now - 1_000), + }, + { + messageId: 'sns-retained', + processingToken: 'retained', + status: 'COMPLETED', + completedAt: new Date(now), + expiresAt: new Date(now + 7 * 24 * 60 * 60 * 1000), + }, + ], + }); + + vi.mocked(EventService.dispatchStoredEvent).mockRestore(); + const updateProgress = vi.fn().mockResolvedValue(undefined); + const result = await processCleanup({updateProgress} as unknown as Job); + + expect(result.deleted).toBe(1); + expect(updateProgress).toHaveBeenCalledWith(100); + expect((await prisma.event.findUniqueOrThrow({where: {id: pendingEvent.id}})).processedAt).not.toBeNull(); + await expect( + prisma.snsWebhookReceipt.findUnique({where: {messageId: 'sns-expired'}}), + ).resolves.toBeNull(); + await expect( + prisma.snsWebhookReceipt.findUnique({where: {messageId: 'sns-retained'}}), + ).resolves.not.toBeNull(); + }); +}); diff --git a/apps/api/src/jobs/idempotency-key-cleanup-processor.ts b/apps/api/src/jobs/idempotency-key-cleanup-processor.ts index 8750f27c9..89610ac9e 100644 --- a/apps/api/src/jobs/idempotency-key-cleanup-processor.ts +++ b/apps/api/src/jobs/idempotency-key-cleanup-processor.ts @@ -6,20 +6,24 @@ import signale from 'signale'; import {REDIS_URL} from '../app/constants.js'; import {prisma} from '../database/prisma.js'; +import {EventService} from '../services/EventService.js'; /** - * Idempotency Key Cleanup Worker - * Deletes claims past their expiresAt, which both bounds table growth and is what - * makes an expired key reusable. Runs hourly, since the TTL is measured in hours. + * Durable-state maintenance worker. + * + * Deletes expired API/SNS claims and reconciles aged event-outbox rows. The + * grace window keeps the hourly sweep away from normal synchronous dispatch. */ const BATCH_SIZE = 10000; // Delete in batches to avoid long-held locks +const EVENT_DISPATCH_BATCH_SIZE = 100; +const EVENT_DISPATCH_GRACE_MS = 5 * 60 * 1000; /** * Process idempotency key cleanup job */ -async function processCleanup(job: Job): Promise<{deleted: number}> { - signale.info('[IDEMPOTENCY-CLEANUP] Starting cleanup of expired idempotency keys...'); +export async function processCleanup(job: Job): Promise<{deleted: number}> { + signale.info('[IDEMPOTENCY-CLEANUP] Starting durable-state maintenance...'); let totalDeleted = 0; @@ -46,7 +50,54 @@ async function processCleanup(job: Job): Promise<{ await new Promise(resolve => setTimeout(resolve, 100)); } - signale.success(`[IDEMPOTENCY-CLEANUP] Cleanup complete. Deleted ${totalDeleted} expired keys`); + for (;;) { + const deleted = await prisma.$executeRaw` + DELETE FROM "sns_webhook_receipts" + WHERE "id" IN ( + SELECT "id" FROM "sns_webhook_receipts" + WHERE "expiresAt" < NOW() + LIMIT ${BATCH_SIZE} + ) + `; + + totalDeleted += deleted; + + if (deleted < BATCH_SIZE) { + break; + } + + signale.info(`[IDEMPOTENCY-CLEANUP] Deleted ${totalDeleted} claims so far, continuing...`); + await new Promise(resolve => setTimeout(resolve, 100)); + } + + const pendingEvents = await prisma.event.findMany({ + where: { + processedAt: null, + createdAt: {lt: new Date(Date.now() - EVENT_DISPATCH_GRACE_MS)}, + }, + select: {id: true}, + orderBy: {createdAt: 'asc'}, + take: EVENT_DISPATCH_BATCH_SIZE, + }); + let dispatchedEvents = 0; + + for (const event of pendingEvents) { + try { + await EventService.dispatchStoredEvent(event.id); + dispatchedEvents += 1; + } catch (error) { + // Keep processedAt null. A later sweep can retry without replaying the + // external request that originally committed this event. + signale.error(`[EVENT-OUTBOX] Failed to dispatch event ${event.id}:`, error); + } + } + + signale.success(`[IDEMPOTENCY-CLEANUP] Cleanup complete. Deleted ${totalDeleted} expired records`); + if (pendingEvents.length > 0) { + signale.info( + `[EVENT-OUTBOX] Dispatched ${dispatchedEvents}/${pendingEvents.length} pending events; failures retry next sweep`, + ); + } await job.updateProgress(100); diff --git a/apps/api/src/services/EventService.ts b/apps/api/src/services/EventService.ts index 0a06b15d6..0704f6877 100644 --- a/apps/api/src/services/EventService.ts +++ b/apps/api/src/services/EventService.ts @@ -37,15 +37,33 @@ export class EventService { }, }); - // Trigger workflows that are listening for this event - await this.triggerWorkflows(projectId, eventName, contactId, data); - - // Resume workflows waiting for this event - await WorkflowExecutionService.handleEvent(projectId, eventName, contactId, data); + await this.dispatchStoredEvent(event.id); return event; } + /** + * Dispatch a durably stored event to workflow triggers and waits. A failed + * dispatch leaves processedAt null so the reconciliation worker can retry it. + */ + public static async dispatchStoredEvent(eventId: string): Promise { + const event = await prisma.event.findUnique({where: {id: eventId}}); + if (!event || event.processedAt) return; + + const data = + event.data && typeof event.data === 'object' && !Array.isArray(event.data) + ? (event.data as Record) + : undefined; + + await this.triggerWorkflows(event.projectId, event.name, event.contactId ?? undefined, data); + await WorkflowExecutionService.handleEvent(event.projectId, event.name, event.contactId ?? undefined, data); + + await prisma.event.updateMany({ + where: {id: event.id, processedAt: null}, + data: {processedAt: new Date()}, + }); + } + /** * Invalidate the workflow cache for a project * Should be called when workflows are enabled/disabled or updated diff --git a/apps/api/src/services/SecurityService.ts b/apps/api/src/services/SecurityService.ts index 4f3bad664..1512f36c6 100644 --- a/apps/api/src/services/SecurityService.ts +++ b/apps/api/src/services/SecurityService.ts @@ -21,6 +21,7 @@ import { PHISHING_CUMULATIVE_WINDOW_MS, PHISHING_DETECTION_ENABLED, PHISHING_DETECTION_SAMPLE_RATE, + SNS_TOPIC_ARNS, } from '../app/constants.js'; /** @@ -171,11 +172,17 @@ export class SecurityService { private static readonly CACHE_TTL = 300; // 5 minutes /** - * Verify an AWS SNS message signature. Returns false if the cert URL is - * untrusted, or the signature doesn't match. + * Authorize the signed topic and verify its AWS SNS signature. Returns false + * before certificate I/O when the topic is not configured for this deployment. */ public static async verifySnsSignature(body: Record): Promise { try { + const topicArn = body['TopicArn']; + if (typeof topicArn !== 'string' || !SNS_TOPIC_ARNS.has(topicArn)) { + signale.warn('[SNS] Missing or untrusted TopicArn'); + return false; + } + const certUrl = body['SigningCertURL']; const signature = body['Signature']; diff --git a/apps/api/src/services/__tests__/SecurityService.sns.test.ts b/apps/api/src/services/__tests__/SecurityService.sns.test.ts new file mode 100644 index 000000000..26f5d10a0 --- /dev/null +++ b/apps/api/src/services/__tests__/SecurityService.sns.test.ts @@ -0,0 +1,75 @@ +import {createSign, generateKeyPairSync} from 'node:crypto'; + +import {afterEach, describe, expect, it, vi} from 'vitest'; + +import {SecurityService} from '../SecurityService'; + +const {ALLOWED_SNS_TOPICS} = vi.hoisted(() => ({ + ALLOWED_SNS_TOPICS: [ + 'arn:aws:sns:us-east-1:123456789012:plunk-ses-events', + 'arn:aws:sns:eu-west-1:123456789012:plunk-ses-inbound', + ] as const, +})); + +const {privateKey: snsPrivateKey, publicKey: snsPublicKey} = generateKeyPairSync('rsa', {modulusLength: 2048}); +const snsPublicKeyPem = snsPublicKey.export({type: 'spki', format: 'pem'}).toString(); + +vi.mock('../../app/constants.js', async () => { + const actual = await vi.importActual('../../app/constants.js'); + return { + ...actual, + SNS_TOPIC_ARNS: new Set(ALLOWED_SNS_TOPICS), + }; +}); + +function signedSnsNotification(topicArn: string, certName: string): Record { + const body = { + Type: 'Notification', + MessageId: `message-${certName}`, + TopicArn: topicArn, + Message: 'test message', + Timestamp: '2026-08-26T18:00:00.000Z', + SignatureVersion: '2', + SigningCertURL: `https://sns.us-east-1.amazonaws.com/${certName}.pem`, + } as Record; + const stringToSign = ['Message', 'MessageId', 'Timestamp', 'TopicArn', 'Type'] + .map(key => `${key}\n${body[key]}\n`) + .join(''); + + body.Signature = createSign('RSA-SHA256').update(stringToSign, 'utf8').sign(snsPrivateKey, 'base64'); + return body; +} + +describe('SecurityService SNS topic authorization', () => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it.each(ALLOWED_SNS_TOPICS)('accepts a valid signature from configured topic %s', async topicArn => { + const certName = topicArn.endsWith('inbound') ? 'allowed-inbound' : 'allowed-outbound'; + const fetchMock = vi.fn().mockResolvedValue(new Response(snsPublicKeyPem, {status: 200})); + vi.stubGlobal('fetch', fetchMock); + + await expect(SecurityService.verifySnsSignature(signedSnsNotification(topicArn, certName))).resolves.toBe(true); + expect(fetchMock).toHaveBeenCalledOnce(); + }); + + it('rejects a valid AWS signature from an unconfigured account before fetching its certificate', async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response(snsPublicKeyPem, {status: 200})); + vi.stubGlobal('fetch', fetchMock); + const untrusted = signedSnsNotification('arn:aws:sns:us-east-1:999999999999:plunk-ses-events', 'untrusted-account'); + + await expect(SecurityService.verifySnsSignature(untrusted)).resolves.toBe(false); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it('rejects a missing TopicArn before fetching a signing certificate', async () => { + const fetchMock = vi.fn().mockResolvedValue(new Response(snsPublicKeyPem, {status: 200})); + vi.stubGlobal('fetch', fetchMock); + const missingTopic = signedSnsNotification(ALLOWED_SNS_TOPICS[0], 'missing-topic'); + delete missingTopic.TopicArn; + + await expect(SecurityService.verifySnsSignature(missingTopic)).resolves.toBe(false); + expect(fetchMock).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/wiki/content/docs/self-hosting/email-setup.mdx b/apps/wiki/content/docs/self-hosting/email-setup.mdx index 0756b012b..560114ced 100644 --- a/apps/wiki/content/docs/self-hosting/email-setup.mdx +++ b/apps/wiki/content/docs/self-hosting/email-setup.mdx @@ -39,10 +39,11 @@ description: Configure email delivery 2. Type: Standard 3. Name: `plunk-ses-events` 4. Create topic -5. Create subscription: +5. Copy the topic ARN. You will use it as `SNS_TOPIC_ARNS` in step 4. +6. Create subscription: - Protocol: HTTPS - Endpoint: `https://api.yourdomain.com/webhooks/sns` -6. Plunk automatically confirms the subscription. If it fails, check your logs for the confirmation URL. +7. Plunk automatically confirms the subscription. If it fails, check your logs for the confirmation URL. ## 3. Create Configuration Sets @@ -66,6 +67,7 @@ description: Configure email delivery AWS_SES_REGION="us-east-1" AWS_SES_ACCESS_KEY_ID="your-access-key" AWS_SES_SECRET_ACCESS_KEY="your-secret-key" +SNS_TOPIC_ARNS="arn:aws:sns:us-east-1:123456789012:plunk-ses-events" SES_CONFIGURATION_SET="plunk-tracking" SES_CONFIGURATION_SET_NO_TRACKING="plunk-no-tracking" ``` @@ -88,6 +90,8 @@ Not all AWS regions support SES inbound — confirm `inbound-smtp.. - **Actions**: Publish to Amazon SNS topic → select your `plunk-ses-events` topic (or create a separate topic that's also subscribed to `https://api.yourdomain.com/webhooks/sns`) 3. Set the rule set as **active** +If inbound email uses a separate topic, append its exact ARN to `SNS_TOPIC_ARNS`, separated by a comma. + ### Inbound IAM permissions Plunk doesn't call SES inbound APIs at runtime — the IAM policy from step 1 covers everything Plunk needs. Configuring receipt rules in AWS is a manual one-time action you do as an AWS admin. @@ -119,4 +123,3 @@ After approval, your sending quota will reflect a much higher daily and per-seco ## 8. (Optional) Configure a MAIL FROM Domain For better DMARC alignment, you can configure a custom MAIL FROM subdomain (e.g. `mail.yourdomain.com`). In the SES console under **Verified identities** → your domain → **MAIL FROM domain**, set a subdomain and add the additional MX and TXT records SES displays. Plunk's IAM policy already includes `ses:SetIdentityMailFromDomain` to support this. - diff --git a/apps/wiki/content/docs/self-hosting/environment-variables.mdx b/apps/wiki/content/docs/self-hosting/environment-variables.mdx index ebcc3b5da..e3ed59c4e 100644 --- a/apps/wiki/content/docs/self-hosting/environment-variables.mdx +++ b/apps/wiki/content/docs/self-hosting/environment-variables.mdx @@ -34,6 +34,7 @@ Set your subdomains here. The application automatically derives all internal and | `AWS_SES_REGION` | Yes | AWS region where SES is configured. | `us-east-1` | | `AWS_SES_ACCESS_KEY_ID` | Yes | AWS access key ID with SES send permissions. | `AKIA...` | | `AWS_SES_SECRET_ACCESS_KEY` | Yes | AWS secret access key for SES. | `wJalr...` | +| `SNS_TOPIC_ARNS` | Yes | Comma-separated exact SNS topic ARNs authorized to deliver SES events. Include each outbound and inbound topic subscribed to `/webhooks/sns`. | `arn:aws:sns:us-east-1:123456789012:plunk-ses-events` | | `SES_CONFIGURATION_SET` | No | SES configuration set name used for open/click tracking. | `plunk-configuration-set` (default) | | `SES_CONFIGURATION_SET_NO_TRACKING` | No | A second SES configuration set without tracking. When set, projects can toggle email tracking on/off. If omitted, the tracking toggle is hidden. | `plunk-no-tracking-configuration-set` (default) | | `MAIL_FROM_SUBDOMAIN` | No | Subdomain prefix used when constructing the MAIL FROM hostname for a verified domain (e.g. with default `plunk` and domain `yourdomain.com`, the MAIL FROM is `plunk.yourdomain.com`). Override when the default subdomain is already in use (e.g. by an R2/CDN custom domain), since the MAIL FROM hostname needs MX + TXT records that can't coexist with a CNAME. | `plunk` | diff --git a/docker-compose.yml b/docker-compose.yml index 18c0524fc..0546ae951 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -151,6 +151,7 @@ services: AWS_SES_REGION: ${AWS_SES_REGION} AWS_SES_ACCESS_KEY_ID: ${AWS_SES_ACCESS_KEY_ID} AWS_SES_SECRET_ACCESS_KEY: ${AWS_SES_SECRET_ACCESS_KEY} + SNS_TOPIC_ARNS: ${SNS_TOPIC_ARNS} SES_CONFIGURATION_SET: ${SES_CONFIGURATION_SET} SES_CONFIGURATION_SET_NO_TRACKING: ${SES_CONFIGURATION_SET_NO_TRACKING:-} diff --git a/packages/db/prisma/migrations/20260826120000_add_sns_webhook_receipts/migration.sql b/packages/db/prisma/migrations/20260826120000_add_sns_webhook_receipts/migration.sql new file mode 100644 index 000000000..ed04427a7 --- /dev/null +++ b/packages/db/prisma/migrations/20260826120000_add_sns_webhook_receipts/migration.sql @@ -0,0 +1,33 @@ +-- Durable SNS delivery receipts. The unique MessageId claim serializes concurrent +-- deliveries, while FAILED and stale PROCESSING rows remain reclaimable after a +-- handler or process failure. + +-- CreateEnum +CREATE TYPE "SnsWebhookReceiptStatus" AS ENUM ('PROCESSING', 'COMPLETED', 'FAILED'); + +-- CreateTable +CREATE TABLE "sns_webhook_receipts" ( + "id" TEXT NOT NULL, + "messageId" TEXT NOT NULL, + "status" "SnsWebhookReceiptStatus" NOT NULL DEFAULT 'PROCESSING', + "processingToken" TEXT NOT NULL, + "processingStartedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "completedAt" TIMESTAMP(3), + "expiresAt" TIMESTAMP(3) NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "sns_webhook_receipts_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "sns_webhook_receipts_messageId_key" ON "sns_webhook_receipts"("messageId"); + +-- CreateIndex +CREATE INDEX "sns_webhook_receipts_expiresAt_idx" ON "sns_webhook_receipts"("expiresAt"); + +-- Make event ingestion durable before workflow dispatch. Existing events have +-- already passed through the synchronous dispatcher and must not be replayed. +ALTER TABLE "events" ADD COLUMN "processedAt" TIMESTAMP(3); +UPDATE "events" SET "processedAt" = "createdAt"; +CREATE INDEX "events_processedAt_createdAt_idx" ON "events"("processedAt", "createdAt"); diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index 157efb7f6..842a9c308 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -634,6 +634,10 @@ model Event { email Email? @relation(fields: [emailId], references: [id], onDelete: Cascade) emailId String? + // Workflow dispatch is separate from durable event ingestion. Null means a + // committed event still needs dispatch or reconciliation. + processedAt DateTime? + // Timestamps createdAt DateTime @default(now()) @@ -641,6 +645,7 @@ model Event { @@index([contactId]) @@index([emailId]) @@index([createdAt]) + @@index([processedAt, createdAt]) @@index([projectId, contactId, name, createdAt]) // For event-based segment queries (fast!) @@index([contactId, name, createdAt]) // For per-contact event lookups @@index([projectId, name, createdAt]) // For event stats and analytics by type over time @@ -679,6 +684,31 @@ model IdempotencyKey { @@map("idempotency_keys") } +// ============================================ +// SNS WEBHOOK RECEIPTS +// ============================================ + +model SnsWebhookReceipt { + id String @id @default(uuid()) + + // AWS signs the outer SNS MessageId, making it the durable delivery identity. + messageId String @unique + + // A token-scoped claim prevents concurrent deliveries from processing together. + // Failed claims can be retried immediately; abandoned claims become reclaimable. + status SnsWebhookReceiptStatus @default(PROCESSING) + processingToken String + processingStartedAt DateTime @default(now()) + completedAt DateTime? + expiresAt DateTime + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@index([expiresAt]) + @@map("sns_webhook_receipts") +} + // ============================================ // ENUMS // ============================================ @@ -691,6 +721,12 @@ enum ProjectDisabledReason { MANUAL // Disabled by support/admin (e.g. directly in DB) } +enum SnsWebhookReceiptStatus { + PROCESSING + COMPLETED + FAILED +} + enum CardVerificationStatus { PENDING // Off-session charge queued or in flight VERIFIED // Card accepted a merchant-initiated charge diff --git a/test/setup.ts b/test/setup.ts index 8cf6c0913..2cc5f676b 100644 --- a/test/setup.ts +++ b/test/setup.ts @@ -38,13 +38,14 @@ process.env.NODE_ENV = 'test'; // with, since those two ship empty. Tests never reach SES, so fill in placeholders // rather than requiring every contributor to invent credentials. // -// Only these two: every other required var (JWT_SECRET, the *_URI values, -// AWS_SES_REGION, DATABASE_URL, REDIS_URL) ships with a value in .env.example and -// is set by the CI workflow, and the DB/Redis URLs must point at real services. +// Only supply safe test placeholders for required values that may be absent. +// Database and Redis URLs must still point at real isolated test services. const TEST_ENV_DEFAULTS: Record = { JWT_SECRET: 'test-jwt-secret-key-for-testing', AWS_SES_ACCESS_KEY_ID: 'test-ses-access-key-id', AWS_SES_SECRET_ACCESS_KEY: 'test-ses-secret-access-key', + SNS_TOPIC_ARNS: + 'arn:aws:sns:us-east-1:123456789012:plunk-ses-events,arn:aws:sns:eu-west-1:123456789012:plunk-ses-inbound', }; for (const [key, value] of Object.entries(TEST_ENV_DEFAULTS)) { diff --git a/turbo.json b/turbo.json index c99fb6d54..91510ec64 100644 --- a/turbo.json +++ b/turbo.json @@ -52,6 +52,7 @@ "AWS_SES_REGION", "AWS_SES_ACCESS_KEY_ID", "AWS_SES_SECRET_ACCESS_KEY", + "SNS_TOPIC_ARNS", "SES_CONFIGURATION_SET", "SES_CONFIGURATION_SET_NO_TRACKING", "EMAIL_RATE_LIMIT_PER_SECOND", @@ -118,6 +119,7 @@ "AWS_SES_REGION", "AWS_SES_ACCESS_KEY_ID", "AWS_SES_SECRET_ACCESS_KEY", + "SNS_TOPIC_ARNS", "SES_CONFIGURATION_SET", "SES_CONFIGURATION_SET_NO_TRACKING", "EMAIL_RATE_LIMIT_PER_SECOND", @@ -175,6 +177,7 @@ "AWS_SES_REGION", "AWS_SES_ACCESS_KEY_ID", "AWS_SES_SECRET_ACCESS_KEY", + "SNS_TOPIC_ARNS", "SES_CONFIGURATION_SET", "SES_CONFIGURATION_SET_NO_TRACKING", "EMAIL_RATE_LIMIT_PER_SECOND", From 9657c7e196f3cdda4354a7ddd0c1a7243a0e1650 Mon Sep 17 00:00:00 2001 From: Vlad Bisceanu <7993591+vladbisceanu@users.noreply.github.com> Date: Thu, 27 Aug 2026 02:07:23 +0200 Subject: [PATCH 2/2] test(webhooks): send complete SNS envelopes --- .../controllers/__tests__/Webhooks.sns.test.ts | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/apps/api/src/controllers/__tests__/Webhooks.sns.test.ts b/apps/api/src/controllers/__tests__/Webhooks.sns.test.ts index 658b8ac0a..8f393ffe7 100644 --- a/apps/api/src/controllers/__tests__/Webhooks.sns.test.ts +++ b/apps/api/src/controllers/__tests__/Webhooks.sns.test.ts @@ -1,4 +1,5 @@ import type {Request, Response} from 'express'; +import {randomUUID} from 'node:crypto'; import {beforeEach, describe, expect, it, vi} from 'vitest'; import {EmailStatus} from '@plunk/db'; @@ -71,10 +72,10 @@ describe('Webhooks - SES event notifications', () => { } /** Deliver one SES event notification, shaped the way SNS posts it. */ - async function post(event: Record) { + async function post(event: Record, snsMessageId = randomUUID()) { const {res, captured, sent} = mockResponse(); const req = { - body: {Type: 'Notification', Message: JSON.stringify(event)}, + body: {Type: 'Notification', MessageId: snsMessageId, Message: JSON.stringify(event)}, get: () => undefined, headers: {}, } as unknown as Request; @@ -242,8 +243,10 @@ describe('Webhooks - SES event notifications', () => { await sentEmail('ses-campaign-2', campaign.id); await prisma.campaign.update({where: {id: campaign.id}, data: {sentCount: 1}}); - await post(notification('Delivery', 'ses-campaign-2')); - await post(notification('Delivery', 'ses-campaign-2')); + const replayedNotification = notification('Delivery', 'ses-campaign-2'); + const replayedSnsMessageId = randomUUID(); + await post(replayedNotification, replayedSnsMessageId); + await post(replayedNotification, replayedSnsMessageId); await CampaignService.sweepDirtyStats(100); const stats = await CampaignService.getStats(projectId, campaign.id); @@ -272,10 +275,10 @@ describe('Webhooks - SES event notifications', () => { expect(updated?.deliveredAt).toBeNull(); }); - it('404s an event for a messageId it does not know', async () => { + it('retries an event for a messageId it does not know', async () => { const captured = await post(notification('Delivery', 'ses-never-sent')); - expect(captured.status).toBe(404); + expect(captured.status).toBe(500); }); }); });