From cebd009154ed3624794538a54459f60aaf4e1512 Mon Sep 17 00:00:00 2001 From: yash-pouranik Date: Fri, 8 May 2026 01:33:54 +0530 Subject: [PATCH 1/3] feat(mail): move /api/mail to bullmq, enforce rate-limit, and handle quota refunds --- .../src/__tests__/mail.controller.test.js | 94 ++++++++++++++++--- apps/public-api/src/app.js | 3 +- .../src/controllers/mail.controller.js | 28 ++---- packages/common/src/index.js | 3 + .../common/src/queues/publicEmailQueue.js | 91 ++++++++++++++++++ 5 files changed, 184 insertions(+), 35 deletions(-) create mode 100644 packages/common/src/queues/publicEmailQueue.js diff --git a/apps/public-api/src/__tests__/mail.controller.test.js b/apps/public-api/src/__tests__/mail.controller.test.js index 7666918d6..268ee08d3 100644 --- a/apps/public-api/src/__tests__/mail.controller.test.js +++ b/apps/public-api/src/__tests__/mail.controller.test.js @@ -3,15 +3,7 @@ process.env.REDIS_URL = process.env.REDIS_URL || "redis://localhost:6379/0"; process.env.ENCRYPTION_KEY = process.env.ENCRYPTION_KEY || "0123456789012345678901234567890a"; -jest.mock('resend', () => { - const sendMock = jest.fn(() => Promise.resolve({ data: { id: 'mail-123' }, error: null })); - return { - Resend: jest.fn(() => ({ - emails: { send: sendMock }, - })), - __sendMock: sendMock, - }; -}); +// Removed resend mock because mail.controller no longer imports it jest.mock('@urbackend/common', () => { const { sendMailSchema } = require('../../../../packages/common/src/utils/input.validation'); @@ -34,11 +26,13 @@ jest.mock('@urbackend/common', () => { decrypt: jest.fn(), redis: redisMock, getPlanLimits: jest.fn(() => ({ mailPerMonth: 100 })), + publicEmailQueue: { + add: jest.fn(() => Promise.resolve({ id: 'job-123' })) + } }; }); -const { Resend, __sendMock } = require('resend'); -const { Project, decrypt, redis } = require('@urbackend/common'); +const { Project, decrypt, redis, publicEmailQueue } = require('@urbackend/common'); const mailController = require('../controllers/mail.controller'); const makeReq = () => ({ @@ -156,10 +150,80 @@ describe('mail.controller', () => { templateUsed: expect.objectContaining({ name: 'welcome', id: 'tpl_1', scope: 'project' }), }), })); - expect(__sendMock).toHaveBeenCalledWith(expect.objectContaining({ - subject: 'Hello Yash', - text: 'Welcome, Yash!', - html: '

Welcome, Yash!

', + expect(publicEmailQueue.add).toHaveBeenCalledWith("send-public-email", expect.objectContaining({ + payload: expect.objectContaining({ + subject: 'Hello Yash', + text: 'Welcome, Yash!', + html: '

Welcome, Yash!

', + }) + })); + }); + + test('refunds quota on terminal async worker failure', async () => { + let failedHandler; + jest.resetModules(); + jest.doMock('bullmq', () => ({ + Queue: jest.fn(), + Worker: jest.fn(() => ({ + on: jest.fn((event, handler) => { + if (event === 'failed') failedHandler = handler; + }), + removeAllListeners: jest.fn(), + close: jest.fn() + })) + })); + + const mockRedis = { decr: jest.fn().mockResolvedValue(1) }; + jest.doMock('../../../../packages/common/src/config/redis', () => mockRedis); + + const { initPublicEmailWorker } = require('../../../../packages/common/src/queues/publicEmailQueue'); + const worker = initPublicEmailWorker(); + + const mockJob = { + id: 'job-999', + data: { consumedQuotaKey: 'project:mail:count:proj_1:2026-05' }, + opts: { attempts: 3 }, + attemptsMade: 3 + }; + + if (failedHandler) { + await failedHandler(mockJob, new Error("Terminal failure")); + } + + expect(mockRedis.decr).toHaveBeenCalledWith('project:mail:count:proj_1:2026-05'); + }); + + test('does not refund quota on non-terminal async worker failure', async () => { + let failedHandler; + jest.resetModules(); + jest.doMock('bullmq', () => ({ + Queue: jest.fn(), + Worker: jest.fn(() => ({ + on: jest.fn((event, handler) => { + if (event === 'failed') failedHandler = handler; + }), + removeAllListeners: jest.fn(), + close: jest.fn() + })) })); + + const mockRedis = { decr: jest.fn().mockResolvedValue(1) }; + jest.doMock('../../../../packages/common/src/config/redis', () => mockRedis); + + const { initPublicEmailWorker } = require('../../../../packages/common/src/queues/publicEmailQueue'); + const worker = initPublicEmailWorker(); + + const mockJob = { + id: 'job-888', + data: { consumedQuotaKey: 'project:mail:count:proj_1:2026-05' }, + opts: { attempts: 3 }, + attemptsMade: 1 // Not terminal yet + }; + + if (failedHandler) { + await failedHandler(mockJob, new Error("Temporary failure")); + } + + expect(mockRedis.decr).not.toHaveBeenCalled(); }); }); diff --git a/apps/public-api/src/app.js b/apps/public-api/src/app.js index 9040f6eaa..54b96b894 100644 --- a/apps/public-api/src/app.js +++ b/apps/public-api/src/app.js @@ -22,12 +22,13 @@ const { capture } = require('@kiroo/sdk'); const {emailQueue} = require('@urbackend/common'); const {authEmailQueue} = require('@urbackend/common'); const {initWebhookWorker} = require('@urbackend/common'); -const {initAuthEmailWorker} = require('@urbackend/common'); +const {initAuthEmailWorker, initPublicEmailWorker} = require('@urbackend/common'); // Initialize webhook worker if (process.env.NODE_ENV !== 'test') { initWebhookWorker(); initAuthEmailWorker(); + initPublicEmailWorker(); } app.use(express.json()); diff --git a/apps/public-api/src/controllers/mail.controller.js b/apps/public-api/src/controllers/mail.controller.js index 724607d70..ab960b149 100644 --- a/apps/public-api/src/controllers/mail.controller.js +++ b/apps/public-api/src/controllers/mail.controller.js @@ -1,13 +1,11 @@ -const { Resend } = require("resend"); const { z } = require("zod"); -const { Project, MailTemplate, decrypt, redis, sendMailSchema } = require("@urbackend/common"); +const { Project, MailTemplate, decrypt, redis, sendMailSchema, publicEmailQueue } = require("@urbackend/common"); const { getMonthKey, getEndOfMonthTtlSeconds, getMonthlyMailLimit, } = require("../utils/mailLimit"); -const DEFAULT_FROM = process.env.EMAIL_FROM || "urBackend "; const getMailCountKey = (projectId, monthKey) => `project:mail:count:${projectId}:${monthKey}`; @@ -300,38 +298,30 @@ module.exports.sendMail = async (req, res) => { const { count, key } = await reserveMonthlyMailSlot(projectId, limit); consumedQuotaKey = key; - const resend = new Resend(clientKey); - - let fromAddress = DEFAULT_FROM; - if (usingByok) { - fromAddress = project.resendFromEmail && project.resendFromEmail.trim() - ? project.resendFromEmail.trim() - : "onboarding@resend.dev"; - } - const payload = { - from: fromAddress, to, subject: resolvedSubject, }; if (typeof resolvedHtml === "string" && resolvedHtml.trim()) payload.html = resolvedHtml; if (typeof resolvedText === "string" && resolvedText.trim()) payload.text = resolvedText; - const { data, error } = await resend.emails.send(payload); - if (error) { - throw new Error(error.message || "Failed to send mail."); - } + const job = await publicEmailQueue.add("send-public-email", { + projectId, + payload, + usingByok, + consumedQuotaKey + }); return res.status(200).json({ success: true, data: { - id: data?.id || null, + id: job.id ? String(job.id) : null, provider: usingByok ? "byok" : "default", monthlyUsage: count, monthlyLimit: limit, ...(templateUsed ? { templateUsed } : {}), }, - message: "Mail sent successfully.", + message: "Mail queued successfully.", }); } catch (err) { if (consumedQuotaKey) { diff --git a/packages/common/src/index.js b/packages/common/src/index.js index c196c951e..8292d58c0 100644 --- a/packages/common/src/index.js +++ b/packages/common/src/index.js @@ -27,6 +27,7 @@ const ApiAnalytics = require("./models/ApiAnalytics"); // Queues const { authEmailQueue, initAuthEmailWorker } = require("./queues/authEmailQueue"); +const { publicEmailQueue, initPublicEmailWorker } = require("./queues/publicEmailQueue"); const { emailQueue } = require("./queues/emailQueue"); const { webhookQueue, @@ -170,6 +171,8 @@ module.exports = { validateUpdateData, userSignupSchema, initAuthEmailWorker, + publicEmailQueue, + initPublicEmailWorker, ...sessionManager, ...planLimits, AppError, diff --git a/packages/common/src/queues/publicEmailQueue.js b/packages/common/src/queues/publicEmailQueue.js new file mode 100644 index 000000000..63efd3eec --- /dev/null +++ b/packages/common/src/queues/publicEmailQueue.js @@ -0,0 +1,91 @@ +const { Queue, Worker } = require('bullmq'); +const connection = require('../config/redis'); +const Project = require('../models/Project'); +const { decrypt } = require('../utils/encryption'); +const { Resend } = require('resend'); + +// Create the email queue for public API +const publicEmailQueue = new Queue('public-email-queue', { connection }); + +let worker = null; + +const initPublicEmailWorker = () => { + if (worker) return worker; + + // Initialize Worker with Rate Limiting (10 per second to respect Resend limits) + worker = new Worker('public-email-queue', async (job) => { + const { projectId, payload, usingByok, consumedQuotaKey } = job.data; + + try { + + let clientKey = process.env.RESEND_API_KEY_2 || process.env.RESEND_API_KEY; + let fromAddress = process.env.EMAIL_FROM || "urBackend "; + + try { + if (projectId && usingByok) { + const project = await Project.findById(projectId).select('+resendApiKey.encrypted +resendApiKey.iv +resendApiKey.tag resendFromEmail').lean(); + if (project && project.resendApiKey) { + const decrypted = decrypt(project.resendApiKey); + if (typeof decrypted === 'string' && decrypted.trim().length > 0) { + clientKey = decrypted.trim(); + fromAddress = project.resendFromEmail && project.resendFromEmail.trim() + ? project.resendFromEmail.trim() + : "onboarding@resend.dev"; + } + } + } + } catch (err) { + console.error(`[Queue] Failed to load BYOK config for project ${projectId}:`, err); + // Fallback to global key + } + + if (!clientKey) { + throw new Error("Resend API key is not configured."); + } + + const resend = new Resend(clientKey); + + const finalPayload = { + ...payload, + from: fromAddress + }; + + console.log(`[Queue] Processing public email to: ${finalPayload.to}`); + + const { data, error } = await resend.emails.send(finalPayload); + + if (error) { + console.error(`[Queue] Failed to send public email to ${finalPayload.to}:`, error); + throw new Error(error.message || "Failed to send email"); + } + + return { data }; + } catch (err) { + throw err; + } + }, { + connection, + limiter: { + max: 10, + duration: 1000, // 10 per second + } + }); + + worker.on('completed', (job) => { + console.log(`[Queue] Job ${job.id} (public email) completed successfully`); + }); + + worker.on('failed', async (job, err) => { + console.error(`[Queue] Job ${job?.id} (public email) failed:`, err); + if (job && job.data && job.data.consumedQuotaKey) { + const maxAttempts = job.opts?.attempts || 1; + if (job.attemptsMade >= maxAttempts) { + await connection.decr(job.data.consumedQuotaKey).catch(() => {}); + } + } + }); + + return worker; +}; + +module.exports = { publicEmailQueue, initPublicEmailWorker }; From 8b3385705d1382dc6a6a9c3d14d428c8eca4ff30 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 8 May 2026 08:23:46 +0000 Subject: [PATCH 2/3] fix: mask PII in logs, safe Lua quota refund, add retry/backoff on enqueue Agent-Logs-Url: https://github.com/geturbackend/urBackend/sessions/091132d9-772a-45c6-9e9f-74b313afd713 Co-authored-by: yash-pouranik <172860064+yash-pouranik@users.noreply.github.com> --- .../public-api/src/__tests__/mail.controller.test.js | 12 +++++++----- apps/public-api/src/controllers/mail.controller.js | 3 +++ packages/common/src/queues/publicEmailQueue.js | 11 ++++++++--- 3 files changed, 18 insertions(+), 8 deletions(-) diff --git a/apps/public-api/src/__tests__/mail.controller.test.js b/apps/public-api/src/__tests__/mail.controller.test.js index 268ee08d3..a1297982b 100644 --- a/apps/public-api/src/__tests__/mail.controller.test.js +++ b/apps/public-api/src/__tests__/mail.controller.test.js @@ -156,7 +156,7 @@ describe('mail.controller', () => { text: 'Welcome, Yash!', html: '

Welcome, Yash!

', }) - })); + }), expect.objectContaining({ attempts: expect.any(Number) })); }); test('refunds quota on terminal async worker failure', async () => { @@ -173,7 +173,7 @@ describe('mail.controller', () => { })) })); - const mockRedis = { decr: jest.fn().mockResolvedValue(1) }; + const mockRedis = { eval: jest.fn().mockResolvedValue(0) }; jest.doMock('../../../../packages/common/src/config/redis', () => mockRedis); const { initPublicEmailWorker } = require('../../../../packages/common/src/queues/publicEmailQueue'); @@ -190,7 +190,9 @@ describe('mail.controller', () => { await failedHandler(mockJob, new Error("Terminal failure")); } - expect(mockRedis.decr).toHaveBeenCalledWith('project:mail:count:proj_1:2026-05'); + expect(mockRedis.eval).toHaveBeenCalledWith( + expect.any(String), 1, 'project:mail:count:proj_1:2026-05' + ); }); test('does not refund quota on non-terminal async worker failure', async () => { @@ -207,7 +209,7 @@ describe('mail.controller', () => { })) })); - const mockRedis = { decr: jest.fn().mockResolvedValue(1) }; + const mockRedis = { eval: jest.fn().mockResolvedValue(0) }; jest.doMock('../../../../packages/common/src/config/redis', () => mockRedis); const { initPublicEmailWorker } = require('../../../../packages/common/src/queues/publicEmailQueue'); @@ -224,6 +226,6 @@ describe('mail.controller', () => { await failedHandler(mockJob, new Error("Temporary failure")); } - expect(mockRedis.decr).not.toHaveBeenCalled(); + expect(mockRedis.eval).not.toHaveBeenCalled(); }); }); diff --git a/apps/public-api/src/controllers/mail.controller.js b/apps/public-api/src/controllers/mail.controller.js index ab960b149..1f04563d2 100644 --- a/apps/public-api/src/controllers/mail.controller.js +++ b/apps/public-api/src/controllers/mail.controller.js @@ -310,6 +310,9 @@ module.exports.sendMail = async (req, res) => { payload, usingByok, consumedQuotaKey + }, { + attempts: 3, + backoff: { type: 'exponential', delay: 5000 } }); return res.status(200).json({ diff --git a/packages/common/src/queues/publicEmailQueue.js b/packages/common/src/queues/publicEmailQueue.js index 63efd3eec..d594af597 100644 --- a/packages/common/src/queues/publicEmailQueue.js +++ b/packages/common/src/queues/publicEmailQueue.js @@ -50,12 +50,16 @@ const initPublicEmailWorker = () => { from: fromAddress }; - console.log(`[Queue] Processing public email to: ${finalPayload.to}`); + const redact = (e) => e.replace(/(.{2})(.*)(?=@)/, (_, keep, mask) => keep + '*'.repeat(mask.length)); + const toList = Array.isArray(finalPayload.to) ? finalPayload.to : [finalPayload.to]; + const maskedTo = toList.map(redact).join(', '); + + console.log(`[Queue] Processing public email to: ${maskedTo}`); const { data, error } = await resend.emails.send(finalPayload); if (error) { - console.error(`[Queue] Failed to send public email to ${finalPayload.to}:`, error); + console.error(`[Queue] Failed to send public email to ${maskedTo}:`, error); throw new Error(error.message || "Failed to send email"); } @@ -80,7 +84,8 @@ const initPublicEmailWorker = () => { if (job && job.data && job.data.consumedQuotaKey) { const maxAttempts = job.opts?.attempts || 1; if (job.attemptsMade >= maxAttempts) { - await connection.decr(job.data.consumedQuotaKey).catch(() => {}); + const luaScript = `if redis.call('EXISTS', KEYS[1]) == 1 then return redis.call('DECR', KEYS[1]) else return 0 end`; + await connection.eval(luaScript, 1, job.data.consumedQuotaKey).catch(() => {}); } } }); From 5cb0cee82627e98c3a53bf2ed370128a5173939f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 8 May 2026 08:25:31 +0000 Subject: [PATCH 3/3] fix: extract Lua script as module constant, robust email redaction for short local parts Agent-Logs-Url: https://github.com/geturbackend/urBackend/sessions/091132d9-772a-45c6-9e9f-74b313afd713 Co-authored-by: yash-pouranik <172860064+yash-pouranik@users.noreply.github.com> --- packages/common/src/queues/publicEmailQueue.js | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/packages/common/src/queues/publicEmailQueue.js b/packages/common/src/queues/publicEmailQueue.js index d594af597..c028eedf7 100644 --- a/packages/common/src/queues/publicEmailQueue.js +++ b/packages/common/src/queues/publicEmailQueue.js @@ -4,6 +4,9 @@ const Project = require('../models/Project'); const { decrypt } = require('../utils/encryption'); const { Resend } = require('resend'); +// Lua script: only decrement quota key if it exists (prevents phantom negative keys with no TTL) +const DECR_IF_EXISTS_SCRIPT = `if redis.call('EXISTS', KEYS[1]) == 1 then return redis.call('DECR', KEYS[1]) else return 0 end`; + // Create the email queue for public API const publicEmailQueue = new Queue('public-email-queue', { connection }); @@ -50,7 +53,14 @@ const initPublicEmailWorker = () => { from: fromAddress }; - const redact = (e) => e.replace(/(.{2})(.*)(?=@)/, (_, keep, mask) => keep + '*'.repeat(mask.length)); + const redact = (e) => { + const atIdx = e.indexOf('@'); + if (atIdx <= 0) return e; + const local = e.slice(0, atIdx); + const domain = e.slice(atIdx); + if (local.length <= 2) return '*'.repeat(local.length) + domain; + return local.slice(0, 2) + '*'.repeat(local.length - 2) + domain; + }; const toList = Array.isArray(finalPayload.to) ? finalPayload.to : [finalPayload.to]; const maskedTo = toList.map(redact).join(', '); @@ -84,7 +94,7 @@ const initPublicEmailWorker = () => { if (job && job.data && job.data.consumedQuotaKey) { const maxAttempts = job.opts?.attempts || 1; if (job.attemptsMade >= maxAttempts) { - const luaScript = `if redis.call('EXISTS', KEYS[1]) == 1 then return redis.call('DECR', KEYS[1]) else return 0 end`; + const luaScript = DECR_IF_EXISTS_SCRIPT; await connection.eval(luaScript, 1, job.data.consumedQuotaKey).catch(() => {}); } }