diff --git a/apps/public-api/src/__tests__/mail.controller.test.js b/apps/public-api/src/__tests__/mail.controller.test.js
index 7666918d6..a1297982b 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,82 @@ 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!
',
+ })
+ }), expect.objectContaining({ attempts: expect.any(Number) }));
+ });
+
+ 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 = { eval: jest.fn().mockResolvedValue(0) };
+ 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.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 () => {
+ 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 = { eval: jest.fn().mockResolvedValue(0) };
+ 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.eval).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..1f04563d2 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,33 @@ 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
+ }, {
+ attempts: 3,
+ backoff: { type: 'exponential', delay: 5000 }
+ });
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..c028eedf7
--- /dev/null
+++ b/packages/common/src/queues/publicEmailQueue.js
@@ -0,0 +1,106 @@
+const { Queue, Worker } = require('bullmq');
+const connection = require('../config/redis');
+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 });
+
+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
+ };
+
+ 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(', ');
+
+ 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 ${maskedTo}:`, 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) {
+ const luaScript = DECR_IF_EXISTS_SCRIPT;
+ await connection.eval(luaScript, 1, job.data.consumedQuotaKey).catch(() => {});
+ }
+ }
+ });
+
+ return worker;
+};
+
+module.exports = { publicEmailQueue, initPublicEmailWorker };