Skip to content

Commit 1201fd3

Browse files
yash-pouranikyash-pouranik
authored
Merge pull request #157 from geturbackend/feat/mail-bullmq
feat: Migrate public mail API to BullMQ for asynchronous delivery
2 parents a79dc10 + 5cb0cee commit 1201fd3

5 files changed

Lines changed: 204 additions & 35 deletions

File tree

‎apps/public-api/src/__tests__/mail.controller.test.js‎

Lines changed: 81 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -3,15 +3,7 @@
33
process.env.REDIS_URL = process.env.REDIS_URL || "redis://localhost:6379/0";
44
process.env.ENCRYPTION_KEY = process.env.ENCRYPTION_KEY || "0123456789012345678901234567890a";
55

6-
jest.mock('resend', () => {
7-
const sendMock = jest.fn(() => Promise.resolve({ data: { id: 'mail-123' }, error: null }));
8-
return {
9-
Resend: jest.fn(() => ({
10-
emails: { send: sendMock },
11-
})),
12-
__sendMock: sendMock,
13-
};
14-
});
6+
// Removed resend mock because mail.controller no longer imports it
157

168
jest.mock('@urbackend/common', () => {
179
const { sendMailSchema } = require('../../../../packages/common/src/utils/input.validation');
@@ -34,11 +26,13 @@ jest.mock('@urbackend/common', () => {
3426
decrypt: jest.fn(),
3527
redis: redisMock,
3628
getPlanLimits: jest.fn(() => ({ mailPerMonth: 100 })),
29+
publicEmailQueue: {
30+
add: jest.fn(() => Promise.resolve({ id: 'job-123' }))
31+
}
3732
};
3833
});
3934

40-
const { Resend, __sendMock } = require('resend');
41-
const { Project, decrypt, redis } = require('@urbackend/common');
35+
const { Project, decrypt, redis, publicEmailQueue } = require('@urbackend/common');
4236
const mailController = require('../controllers/mail.controller');
4337

4438
const makeReq = () => ({
@@ -156,10 +150,82 @@ describe('mail.controller', () => {
156150
templateUsed: expect.objectContaining({ name: 'welcome', id: 'tpl_1', scope: 'project' }),
157151
}),
158152
}));
159-
expect(__sendMock).toHaveBeenCalledWith(expect.objectContaining({
160-
subject: 'Hello Yash',
161-
text: 'Welcome, Yash!',
162-
html: '<p>Welcome, Yash!</p>',
153+
expect(publicEmailQueue.add).toHaveBeenCalledWith("send-public-email", expect.objectContaining({
154+
payload: expect.objectContaining({
155+
subject: 'Hello Yash',
156+
text: 'Welcome, Yash!',
157+
html: '<p>Welcome, Yash!</p>',
158+
})
159+
}), expect.objectContaining({ attempts: expect.any(Number) }));
160+
});
161+
162+
test('refunds quota on terminal async worker failure', async () => {
163+
let failedHandler;
164+
jest.resetModules();
165+
jest.doMock('bullmq', () => ({
166+
Queue: jest.fn(),
167+
Worker: jest.fn(() => ({
168+
on: jest.fn((event, handler) => {
169+
if (event === 'failed') failedHandler = handler;
170+
}),
171+
removeAllListeners: jest.fn(),
172+
close: jest.fn()
173+
}))
163174
}));
175+
176+
const mockRedis = { eval: jest.fn().mockResolvedValue(0) };
177+
jest.doMock('../../../../packages/common/src/config/redis', () => mockRedis);
178+
179+
const { initPublicEmailWorker } = require('../../../../packages/common/src/queues/publicEmailQueue');
180+
const worker = initPublicEmailWorker();
181+
182+
const mockJob = {
183+
id: 'job-999',
184+
data: { consumedQuotaKey: 'project:mail:count:proj_1:2026-05' },
185+
opts: { attempts: 3 },
186+
attemptsMade: 3
187+
};
188+
189+
if (failedHandler) {
190+
await failedHandler(mockJob, new Error("Terminal failure"));
191+
}
192+
193+
expect(mockRedis.eval).toHaveBeenCalledWith(
194+
expect.any(String), 1, 'project:mail:count:proj_1:2026-05'
195+
);
196+
});
197+
198+
test('does not refund quota on non-terminal async worker failure', async () => {
199+
let failedHandler;
200+
jest.resetModules();
201+
jest.doMock('bullmq', () => ({
202+
Queue: jest.fn(),
203+
Worker: jest.fn(() => ({
204+
on: jest.fn((event, handler) => {
205+
if (event === 'failed') failedHandler = handler;
206+
}),
207+
removeAllListeners: jest.fn(),
208+
close: jest.fn()
209+
}))
210+
}));
211+
212+
const mockRedis = { eval: jest.fn().mockResolvedValue(0) };
213+
jest.doMock('../../../../packages/common/src/config/redis', () => mockRedis);
214+
215+
const { initPublicEmailWorker } = require('../../../../packages/common/src/queues/publicEmailQueue');
216+
const worker = initPublicEmailWorker();
217+
218+
const mockJob = {
219+
id: 'job-888',
220+
data: { consumedQuotaKey: 'project:mail:count:proj_1:2026-05' },
221+
opts: { attempts: 3 },
222+
attemptsMade: 1 // Not terminal yet
223+
};
224+
225+
if (failedHandler) {
226+
await failedHandler(mockJob, new Error("Temporary failure"));
227+
}
228+
229+
expect(mockRedis.eval).not.toHaveBeenCalled();
164230
});
165231
});

‎apps/public-api/src/app.js‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,13 @@ const { capture } = require('@kiroo/sdk');
2222
const {emailQueue} = require('@urbackend/common');
2323
const {authEmailQueue} = require('@urbackend/common');
2424
const {initWebhookWorker} = require('@urbackend/common');
25-
const {initAuthEmailWorker} = require('@urbackend/common');
25+
const {initAuthEmailWorker, initPublicEmailWorker} = require('@urbackend/common');
2626

2727
// Initialize webhook worker
2828
if (process.env.NODE_ENV !== 'test') {
2929
initWebhookWorker();
3030
initAuthEmailWorker();
31+
initPublicEmailWorker();
3132
}
3233

3334
app.use(express.json());

‎apps/public-api/src/controllers/mail.controller.js‎

Lines changed: 12 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,11 @@
1-
const { Resend } = require("resend");
21
const { z } = require("zod");
3-
const { Project, MailTemplate, decrypt, redis, sendMailSchema } = require("@urbackend/common");
2+
const { Project, MailTemplate, decrypt, redis, sendMailSchema, publicEmailQueue } = require("@urbackend/common");
43
const {
54
getMonthKey,
65
getEndOfMonthTtlSeconds,
76
getMonthlyMailLimit,
87
} = require("../utils/mailLimit");
98

10-
const DEFAULT_FROM = process.env.EMAIL_FROM || "urBackend <urbackend@apps.bitbros.in>";
119

1210
const getMailCountKey = (projectId, monthKey) =>
1311
`project:mail:count:${projectId}:${monthKey}`;
@@ -300,38 +298,33 @@ module.exports.sendMail = async (req, res) => {
300298
const { count, key } = await reserveMonthlyMailSlot(projectId, limit);
301299
consumedQuotaKey = key;
302300

303-
const resend = new Resend(clientKey);
304-
305-
let fromAddress = DEFAULT_FROM;
306-
if (usingByok) {
307-
fromAddress = project.resendFromEmail && project.resendFromEmail.trim()
308-
? project.resendFromEmail.trim()
309-
: "onboarding@resend.dev";
310-
}
311-
312301
const payload = {
313-
from: fromAddress,
314302
to,
315303
subject: resolvedSubject,
316304
};
317305
if (typeof resolvedHtml === "string" && resolvedHtml.trim()) payload.html = resolvedHtml;
318306
if (typeof resolvedText === "string" && resolvedText.trim()) payload.text = resolvedText;
319307

320-
const { data, error } = await resend.emails.send(payload);
321-
if (error) {
322-
throw new Error(error.message || "Failed to send mail.");
323-
}
308+
const job = await publicEmailQueue.add("send-public-email", {
309+
projectId,
310+
payload,
311+
usingByok,
312+
consumedQuotaKey
313+
}, {
314+
attempts: 3,
315+
backoff: { type: 'exponential', delay: 5000 }
316+
});
324317

325318
return res.status(200).json({
326319
success: true,
327320
data: {
328-
id: data?.id || null,
321+
id: job.id ? String(job.id) : null,
329322
provider: usingByok ? "byok" : "default",
330323
monthlyUsage: count,
331324
monthlyLimit: limit,
332325
...(templateUsed ? { templateUsed } : {}),
333326
},
334-
message: "Mail sent successfully.",
327+
message: "Mail queued successfully.",
335328
});
336329
} catch (err) {
337330
if (consumedQuotaKey) {

‎packages/common/src/index.js‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ const ApiAnalytics = require("./models/ApiAnalytics");
2727

2828
// Queues
2929
const { authEmailQueue, initAuthEmailWorker } = require("./queues/authEmailQueue");
30+
const { publicEmailQueue, initPublicEmailWorker } = require("./queues/publicEmailQueue");
3031
const { emailQueue } = require("./queues/emailQueue");
3132
const {
3233
webhookQueue,
@@ -170,6 +171,8 @@ module.exports = {
170171
validateUpdateData,
171172
userSignupSchema,
172173
initAuthEmailWorker,
174+
publicEmailQueue,
175+
initPublicEmailWorker,
173176
...sessionManager,
174177
...planLimits,
175178
AppError,

‎packages/common/src/queues/publicEmailQueue.js‎

Lines changed: 106 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,106 @@
1+
const { Queue, Worker } = require('bullmq');
2+
const connection = require('../config/redis');
3+
const Project = require('../models/Project');
4+
const { decrypt } = require('../utils/encryption');
5+
const { Resend } = require('resend');
6+
7+
// Lua script: only decrement quota key if it exists (prevents phantom negative keys with no TTL)
8+
const DECR_IF_EXISTS_SCRIPT = `if redis.call('EXISTS', KEYS[1]) == 1 then return redis.call('DECR', KEYS[1]) else return 0 end`;
9+
10+
// Create the email queue for public API
11+
const publicEmailQueue = new Queue('public-email-queue', { connection });
12+
13+
let worker = null;
14+
15+
const initPublicEmailWorker = () => {
16+
if (worker) return worker;
17+
18+
// Initialize Worker with Rate Limiting (10 per second to respect Resend limits)
19+
worker = new Worker('public-email-queue', async (job) => {
20+
const { projectId, payload, usingByok, consumedQuotaKey } = job.data;
21+
22+
try {
23+
24+
let clientKey = process.env.RESEND_API_KEY_2 || process.env.RESEND_API_KEY;
25+
let fromAddress = process.env.EMAIL_FROM || "urBackend <urbackend@apps.bitbros.in>";
26+
27+
try {
28+
if (projectId && usingByok) {
29+
const project = await Project.findById(projectId).select('+resendApiKey.encrypted +resendApiKey.iv +resendApiKey.tag resendFromEmail').lean();
30+
if (project && project.resendApiKey) {
31+
const decrypted = decrypt(project.resendApiKey);
32+
if (typeof decrypted === 'string' && decrypted.trim().length > 0) {
33+
clientKey = decrypted.trim();
34+
fromAddress = project.resendFromEmail && project.resendFromEmail.trim()
35+
? project.resendFromEmail.trim()
36+
: "onboarding@resend.dev";
37+
}
38+
}
39+
}
40+
} catch (err) {
41+
console.error(`[Queue] Failed to load BYOK config for project ${projectId}:`, err);
42+
// Fallback to global key
43+
}
44+
45+
if (!clientKey) {
46+
throw new Error("Resend API key is not configured.");
47+
}
48+
49+
const resend = new Resend(clientKey);
50+
51+
const finalPayload = {
52+
...payload,
53+
from: fromAddress
54+
};
55+
56+
const redact = (e) => {
57+
const atIdx = e.indexOf('@');
58+
if (atIdx <= 0) return e;
59+
const local = e.slice(0, atIdx);
60+
const domain = e.slice(atIdx);
61+
if (local.length <= 2) return '*'.repeat(local.length) + domain;
62+
return local.slice(0, 2) + '*'.repeat(local.length - 2) + domain;
63+
};
64+
const toList = Array.isArray(finalPayload.to) ? finalPayload.to : [finalPayload.to];
65+
const maskedTo = toList.map(redact).join(', ');
66+
67+
console.log(`[Queue] Processing public email to: ${maskedTo}`);
68+
69+
const { data, error } = await resend.emails.send(finalPayload);
70+
71+
if (error) {
72+
console.error(`[Queue] Failed to send public email to ${maskedTo}:`, error);
73+
throw new Error(error.message || "Failed to send email");
74+
}
75+
76+
return { data };
77+
} catch (err) {
78+
throw err;
79+
}
80+
}, {
81+
connection,
82+
limiter: {
83+
max: 10,
84+
duration: 1000, // 10 per second
85+
}
86+
});
87+
88+
worker.on('completed', (job) => {
89+
console.log(`[Queue] Job ${job.id} (public email) completed successfully`);
90+
});
91+
92+
worker.on('failed', async (job, err) => {
93+
console.error(`[Queue] Job ${job?.id} (public email) failed:`, err);
94+
if (job && job.data && job.data.consumedQuotaKey) {
95+
const maxAttempts = job.opts?.attempts || 1;
96+
if (job.attemptsMade >= maxAttempts) {
97+
const luaScript = DECR_IF_EXISTS_SCRIPT;
98+
await connection.eval(luaScript, 1, job.data.consumedQuotaKey).catch(() => {});
99+
}
100+
}
101+
});
102+
103+
return worker;
104+
};
105+
106+
module.exports = { publicEmailQueue, initPublicEmailWorker };

0 commit comments

Comments
 (0)