feat: Migrate public mail API to BullMQ for asynchronous delivery

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: '<p>Welcome, Yash!</p>',
+        expect(publicEmailQueue.add).toHaveBeenCalledWith("send-public-email", expect.objectContaining({
+            payload: expect.objectContaining({
+                subject: 'Hello Yash',
+                text: 'Welcome, Yash!',
+                html: '<p>Welcome, Yash!</p>',
+            })
+        }), 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();
     });
 });

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());

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 <urbackend@apps.bitbros.in>";
 
 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) {

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,

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 <urbackend@apps.bitbros.in>";
+
+        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 };