letsbe/pn-new-crm
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(deps): adopt p-limit for unbounded mass-op fan-outs

Browse Source 
Cap concurrency on two services that were fanning out unbounded
requests to external systems:

1. email-compose.service.ts — attachment resolution. User attaches
   20 files → 20 simultaneous S3/MinIO GETs + 20 buffers in heap.
   Now capped at 4 concurrent reads; peak memory bounded by
   4 × max-attachment-size regardless of attachment count.

2. document-signing-emails.service.ts — sendSigningCompleted fanned
   out one SMTP send per recipient simultaneously. A Sales Contract
   with 10 recipients (client + 5 sellers + 4 witnesses) hit SMTP
   provider connection limits (Mailgun/SES/Postmark all cap concurrent
   connections in the single digits) and dropped overflow silently.
   Now capped at 3 concurrent sends.

Both use `pLimit(N)` from the Sindre Sorhus suite — well-tested at
scale, ~1kb gzip per service. Pattern is established for the
remaining audit-flagged mass-op services (brochures, backup, GDPR
export) to adopt as those files are touched.

Verified: tsc clean, vitest 1293/1293 pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt
2026-05-12 18:35:56 +02:00
parent ce662071f8
commit a65aadc530
4 changed files with 80 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
src/lib/services/document-signing-emails.service.ts
View File

@@ -28,6 +28,8 @@
* port (single-tenant deploys can keep using Documenso's hosted UI).
*/
import pLimit from 'p-limit';
import { sendEmail } from '@/lib/email';
import { getBrandingShell } from '@/lib/email/branding-resolver';
import {
@@ -231,33 +233,41 @@ export async function sendSigningReminder(args: SigningReminderArgs): Promise<vo
export async function sendSigningCompleted(args: SigningCompletedArgs): Promise<void> {
const branding = await getBrandingShell(args.portId);
// Cap concurrency at 3: a Sales Contract with 10 recipients (client +
// 5 sellers + 4 witnesses) shouldn't fan out 10 simultaneous SMTP
// sends. Most SMTP providers (Mailgun, SES, Postmark) cap concurrent
// connections in the single digits and silently drop the overflow.
const sendLimit = pLimit(3);
await Promise.all(
args.recipients.map(async (recipient) => {
const { subject, html, text } = signingCompletedEmail(
{
recipientName: recipient.name,
documentLabel: args.documentLabel,
clientName: args.clientName,
portName: args.portName,
completedAt: args.completedAt,
},
{ branding },
);
try {
await sendEmail(recipient.email, subject, html, undefined, text, args.portId, [
{ fileId: args.signedPdfFileId, filename: args.signedPdfFilename },
]);
logger.info(
{ portId: args.portId, recipient: recipient.email, documentLabel: args.documentLabel },
'Signing-completed email sent',
args.recipients.map((recipient) =>
sendLimit(async () => {
const { subject, html, text } = signingCompletedEmail(
{
recipientName: recipient.name,
documentLabel: args.documentLabel,
clientName: args.clientName,
portName: args.portName,
completedAt: args.completedAt,
},
{ branding },
);
} catch (err) {
logger.error(
{ err, portId: args.portId, recipient: recipient.email },
'Signing-completed email send failed',
);
// Don't throw — sending to one recipient shouldn't block the others.
}
}),
try {
await sendEmail(recipient.email, subject, html, undefined, text, args.portId, [
{ fileId: args.signedPdfFileId, filename: args.signedPdfFilename },
]);
logger.info(
{ portId: args.portId, recipient: recipient.email, documentLabel: args.documentLabel },
'Signing-completed email sent',
);
} catch (err) {
logger.error(
{ err, portId: args.portId, recipient: recipient.email },
'Signing-completed email send failed',
);
// Don't throw — sending to one recipient shouldn't block the others.
}
}),
),
);
}

44
src/lib/services/email-compose.service.ts
View File

@@ -1,5 +1,6 @@
import nodemailer from 'nodemailer';
import { and, eq, inArray, sql } from 'drizzle-orm';
import pLimit from 'p-limit';
import { db } from '@/lib/db';
import { emailAccounts, emailMessages, emailThreads } from '@/lib/db/schema/email';
@@ -112,26 +113,33 @@ export async function sendEmail(
}
}
// Resolve attachments for the user-path SMTP send.
// Resolve attachments for the user-path SMTP send. Cap concurrency so
// a user attaching 20 large files doesn't fan out 20 simultaneous
// S3/MinIO reads + 20 buffers in memory at once — bounded at 4 means
// peak memory tops out at ~4 × max-file-size irrespective of the
// attachment count.
const attachmentLimit = pLimit(4);
const resolvedAttachments = data.attachments
? await Promise.all(
data.attachments.map(async (ref) => {
const file = await db.query.files.findFirst({
where: eq(files.id, ref.fileId),
});
if (!file) throw new NotFoundError('File');
const { getStorageBackend } = await import('@/lib/storage');
const stream = await (await getStorageBackend()).get(file.storagePath);
const chunks: Buffer[] = [];
for await (const chunk of stream) {
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
}
return {
filename: ref.filename ?? file.originalName,
content: Buffer.concat(chunks),
...(file.mimeType ? { contentType: file.mimeType } : {}),
};
}),
data.attachments.map((ref) =>
attachmentLimit(async () => {
const file = await db.query.files.findFirst({
where: eq(files.id, ref.fileId),
});
if (!file) throw new NotFoundError('File');
const { getStorageBackend } = await import('@/lib/storage');
const stream = await (await getStorageBackend()).get(file.storagePath);
const chunks: Buffer[] = [];
for await (const chunk of stream) {
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
}
return {
filename: ref.filename ?? file.originalName,
content: Buffer.concat(chunks),
...(file.mimeType ? { contentType: file.mimeType } : {}),
};
}),
),
)
: undefined;