fix(webhook): process every message in a batched payload

.gitignore

@@ -3,3 +3,12 @@ coverage/
 *.lcov
 coverage-final.json
 clover.xml
+
+# Local environment files. .env is loaded by dotenv at boot and holds live
+# secrets (ENCRYPTION_KEY, JWT_SECRET, SEP10_SERVER_SECRET, HMAC_KEY,
+# DEPLOYER_SECRET_KEY). The committed example file is the only environment
+# file that belongs in git.
+.env
+.env.local
+.env.*.local
+!.env.example

src/__tests__/bot.controller.test.ts

@@ -2,6 +2,7 @@ import { BotController } from '../controllers/bot.controller';
 import { Request, Response } from 'express';
 
 const mockEnqueueMessage = jest.fn().mockResolvedValue({ id: 'mock-job-id' });
+const mockIsMessageProcessed = jest.fn().mockResolvedValue(false);
 
 jest.mock('../queue/message.queue', () => ({
     enqueueMessage: (...args: any[]) => mockEnqueueMessage(...args),
@@ -15,13 +16,18 @@ jest.mock('../services/observability.service', () => ({
     observabilityService: { logInfo: jest.fn(), alertCriticalFailure: jest.fn() },
 }));
 
+jest.mock('../services/deduplication.service', () => ({
+    isMessageProcessed: (...args: any[]) => mockIsMessageProcessed(...args),
+}));
+
 describe('BotController', () => {
     let botController: BotController;
     let mockReq: Partial<Request>;
     let mockRes: Partial<Response>;
 
     beforeEach(() => {
         jest.clearAllMocks();
+        mockIsMessageProcessed.mockResolvedValue(false);
         botController = new BotController();
         mockRes = { sendStatus: jest.fn(), status: jest.fn().mockReturnThis(), send: jest.fn() };
     });
@@ -105,6 +111,31 @@ describe('BotController', () => {
             expect(mockEnqueueMessage).not.toHaveBeenCalled();
         });
 
+        it('should enqueue every message in a two-message batched payload', async () => {
+            const payload = createWebhookPayload('SEND 10 @jane');
+            (payload.entry[0].changes[0].value as any).messages.push({
+                from: '12345',
+                id: 'wamid.456',
+                text: { body: 'BALANCE' },
+            });
+            mockReq = { body: payload };
+
+            await botController.handleMessage(mockReq as Request, mockRes as Response);
+
+            expect(mockEnqueueMessage).toHaveBeenCalledTimes(2);
+            expect(mockEnqueueMessage).toHaveBeenNthCalledWith(1, {
+                from: '12345',
+                msgBody: 'SEND 10 @jane',
+                whatsappMessageId: 'wamid.123',
+            });
+            expect(mockEnqueueMessage).toHaveBeenNthCalledWith(2, {
+                from: '12345',
+                msgBody: 'BALANCE',
+                whatsappMessageId: 'wamid.456',
+            });
+            expect(mockRes.sendStatus).toHaveBeenCalledWith(200);
+        });
+
         it('should return 200 even without messages array', async () => {
             mockReq = { body: { object: 'whatsapp_business_account', entry: [{ changes: [{ value: {} }] }] } };
             await botController.handleMessage(mockReq as Request, mockRes as Response);
@@ -137,6 +168,17 @@ describe('BotController', () => {
             expect(mockRes.sendStatus).toHaveBeenCalledWith(500);
             expect(mockEnqueueMessage).toHaveBeenCalledTimes(1);
         });
+
+        it('should not process a message that was already handled on redelivery', async () => {
+            mockIsMessageProcessed.mockResolvedValueOnce(true);
+            mockReq = { body: createWebhookPayload('SEND 10 @jane') };
+
+            await botController.handleMessage(mockReq as Request, mockRes as Response);
+
+            expect(mockIsMessageProcessed).toHaveBeenCalledWith('wamid.123');
+            expect(mockEnqueueMessage).not.toHaveBeenCalled();
+            expect(mockRes.sendStatus).toHaveBeenCalledWith(200);
+        });
     });
 
     describe('handleMessage - reliability edge cases', () => {
@@ -196,5 +238,23 @@ describe('BotController', () => {
                 jest.useRealTimers();
             }
         });
+
+        it('returns 500 when any message in a batch fails to enqueue', async () => {
+            const payload = createWebhookPayload('SEND 10 @jane');
+            (payload.entry[0].changes[0].value as any).messages.push({
+                from: '12345',
+                id: 'wamid.456',
+                text: { body: 'BALANCE' },
+            });
+            mockEnqueueMessage
+                .mockResolvedValueOnce({ id: 'job-1' })
+                .mockRejectedValueOnce(new Error('redis down'));
+            mockReq = { body: payload };
+
+            await botController.handleMessage(mockReq as Request, mockRes as Response);
+
+            expect(mockEnqueueMessage).toHaveBeenCalledTimes(2);
+            expect(mockRes.sendStatus).toHaveBeenCalledWith(500);
+        });
     });
 });

src/controllers/bot.controller.ts

@@ -20,6 +20,7 @@ interface WhatsAppContact {
     wa_id?: string;
     profile?: {
         name?: string;
+        locale?: string;
     };
 }
 
@@ -75,32 +76,52 @@ export class BotController {
     }
 
     /**
-     * Safely extract the first inbound text message from a (possibly malformed
+     * Safely extract every inbound text message from a (possibly malformed
      * or malicious) webhook body. Optional chaining is used end-to-end so a
-     * partial or garbage payload can never throw a TypeError.
-     * Returns null when there is no actionable text message.
+     * partial or garbage payload can never throw a TypeError. The WhatsApp
+     * envelope is an array at both the `entry` and `changes` levels, and each
+     * change can carry several messages, so we iterate all of them instead of
+     * reading the first element only.
      */
-    private extractTextMessage(body: WhatsAppWebhookBody): IncomingTextMessage | null {
-        const value = body.entry?.[0]?.changes?.[0]?.value;
-        const message = value?.messages?.[0];
-        const contacts = value?.contacts;
-
-        const from = message?.from;
-        const msgBody = message?.text?.body;
-        const id = message?.id;
-
-        if (typeof from !== 'string' || typeof msgBody !== 'string' || msgBody.length === 0 || typeof id !== 'string') {
-            return null;
+    private extractTextMessages(body: WhatsAppWebhookBody): IncomingTextMessage[] {
+        const result: IncomingTextMessage[] = [];
+        const entries = Array.isArray(body.entry) ? body.entry : [];
+
+        for (const entry of entries) {
+            const changes = Array.isArray(entry?.changes) ? entry.changes : [];
+
+            for (const change of changes) {
+                const value = change?.value;
+                const messages = Array.isArray(value?.messages) ? value.messages : [];
+                const contacts = value?.contacts;
+
+                for (const message of messages) {
+                    const from = message?.from;
+                    const msgBody = message?.text?.body;
+                    const id = message?.id;
+
+                    if (
+                        typeof from !== 'string' ||
+                        typeof msgBody !== 'string' ||
+                        msgBody.length === 0 ||
+                        typeof id !== 'string'
+                    ) {
+                        continue;
+                    }
+
+                    // Try to extract locale if available in contacts metadata for this sender
+                    let locale: string | undefined;
+                    const contact = contacts?.find((c) => c?.wa_id === from);
+                    if (contact && contact.profile?.locale) {
+                        locale = contact.profile.locale;
+                    }
+
+                    result.push({ from, msgBody, whatsappMessageId: id, locale });
+                }
+            }
         }
 
-        // Try to extract locale if available in contacts metadata for this sender (e.g., from WhatsApp profile)
-        let locale: string | undefined;
-        const contact = contacts?.find(c => c.wa_id === from);
-        if (contact && (contact as any).profile?.locale) {
-            locale = (contact as any).profile.locale;
-        }
-
-        return { from, msgBody, whatsappMessageId: id, locale };
+        return result;
     }
 
     public async handleMessage(req: Request, res: Response) {
@@ -111,9 +132,9 @@ export class BotController {
             return;
         }
 
-        const message = this.extractTextMessage(body);
+        const messages = this.extractTextMessages(body);
 
-        if (!message) {
+        if (messages.length === 0) {
             // Valid webhook event that carries nothing for us to act on
             // (status update, empty body, non-text message, etc.). Acknowledge.
             res.sendStatus(200);
@@ -122,20 +143,31 @@ export class BotController {
 
         observabilityService.logInfo('Received webhook message');
 
-        try {
-            await this.enqueueWithTimeout({
-                from: message.from,
-                msgBody: message.msgBody,
-                whatsappMessageId: message.whatsappMessageId,
-                locale: message.locale,
-            });
-        } catch (err) {
-            observabilityService.alertCriticalFailure('Failed to enqueue webhook message', err, {
-                from: message.from,
+        // Enqueue every message in the batch. We await all of them so the
+        // response reflects whether the whole delivery was handled.
+        const results = await Promise.allSettled(
+            messages.map((message) =>
+                this.enqueueWithTimeout({
+                    from: message.from,
+                    msgBody: message.msgBody,
+                    whatsappMessageId: message.whatsappMessageId,
+                    locale: message.locale,
+                }),
+            ),
+        );
+
+        const failures = results.filter((r) => r.status === 'rejected');
+
+        if (failures.length > 0) {
+            const firstFailure = failures[0] as PromiseRejectedResult;
+            observabilityService.alertCriticalFailure('Failed to enqueue webhook message', firstFailure.reason, {
+                from: messages[0].from,
+                failedCount: failures.length,
+                totalCount: messages.length,
             });
             // Let a queue/Redis failure surface as an unhandled promise
             // rejection. Returning 500 makes WhatsApp retry delivery rather
-            // than silently dropping the message — important for a financial
+            // than silently dropping the message -- important for a financial
             // app where a lost message can mean a lost transaction.
             res.sendStatus(500);
             return;
@@ -146,7 +178,7 @@ export class BotController {
 
     /**
      * Enqueue a message with a hard timeout so a stalled queue connection can
-     * never hang the webhook. Resolves once the job is queued; rejects on
+     * never hang the webhook. Resolves once the job is queued; rejects on the
      * enqueue error or when ENQUEUE_TIMEOUT_MS elapses. The timer is always
      * cleared so it cannot keep the event loop alive.
      */

src/middleware/command-rate-limit.middleware.ts

@@ -46,6 +46,35 @@ export function classifyCommand(text: string): CommandTier {
     return 'GENERAL';
 }
 
+/**
+ * Extracts every text message carried by a WhatsApp webhook payload.
+ * The envelope is an array at both the entry and changes levels, and each
+ * value may carry several messages, so all of them are collected.
+ */
+export function extractAllMessageDetails(body: any): Array<{ phoneNumber: string | null; commandText: string | null; messageId: string | null }> {
+    const results: Array<{ phoneNumber: string | null; commandText: string | null; messageId: string | null }> = [];
+    const entries = Array.isArray(body?.entry) ? body.entry : [];
+
+    for (const entry of entries) {
+        const changes = Array.isArray(entry?.changes) ? entry.changes : [];
+        for (const change of changes) {
+            const messages = Array.isArray(change?.value?.messages) ? change.value.messages : [];
+            for (const message of messages) {
+                const from = message?.from;
+                const msgBody = message?.text?.body;
+                const id = message?.id;
+                results.push({
+                    phoneNumber: typeof from === 'string' ? from : null,
+                    commandText: typeof msgBody === 'string' ? msgBody : null,
+                    messageId: typeof id === 'string' ? id : null,
+                });
+            }
+        }
+    }
+
+    return results;
+}
+
 /**
  * Minimally parses request to extract the WhatsApp message body and sender.
  * Useful if express.json() is not run yet, or as a defensive guard.
@@ -60,13 +89,12 @@ export function extractMessageDetails(req: any): { phoneNumber: string | null; c
         }
     }
 
-    const message = body?.entry?.[0]?.changes?.[0]?.value?.messages?.[0];
-    const from = message?.from;
-    const msgBody = message?.text?.body;
+    const all = extractAllMessageDetails(body);
+    const first = all[0];
 
     return {
-        phoneNumber: typeof from === 'string' ? from : null,
-        commandText: typeof msgBody === 'string' ? msgBody : null,
+        phoneNumber: first?.phoneNumber ?? null,
+        commandText: first?.commandText ?? null,
     };
 }
 
@@ -252,3 +280,36 @@ export const commandRateLimitMiddleware = async (
 
     next();
 };
+
+/**
+ * Applies the rate-limit accounting to every message in a batched payload so
+ * classification and abuse history see the same set the bot controller does.
+ * Returns the first message that exceeded its limit, if any.
+ */
+export async function checkBatchedMessages(
+    body: any,
+    windowMs: number = 60 * 1000
+): Promise<{ blocked: boolean; phoneNumber: string | null; commandText: string | null; remainingSeconds: number }> {
+    const messages = extractAllMessageDetails(body);
+
+    for (const { phoneNumber, commandText } of messages) {
+        if (!phoneNumber) continue;
+
+        const tier = commandText ? classifyCommand(commandText) : 'GENERAL';
+        const rateLimitKey = commandText ? `rate_limit:${tier}:${phoneNumber}` : `rate_limit:${phoneNumber}`;
+        const limit = tier === 'FINANCIAL' ? 5 : tier === 'READ' ? 10 : 30;
+
+        const { allowed, remainingSeconds } = await checkLimit(rateLimitKey, limit, windowMs);
+        if (!allowed) {
+            return { blocked: true, phoneNumber, commandText, remainingSeconds };
+        }
+
+        if (commandText) {
+            abuseDetectionService.recordCommand(phoneNumber, commandText).catch((err) => {
+                console.error('Failed to record command in abuse detection service:', err);
+            });
+        }
+    }
+
+    return { blocked: false, phoneNumber: null, commandText: null, remainingSeconds: 0 };
+}

src/queue/message.queue.ts

@@ -47,6 +47,20 @@ export async function enqueueMessage(data: MessageJobData): Promise<Job> {
     });
 }
 
+/**
+ * Enqueue every message in a batch. Each message is keyed by its WhatsApp
+ * message id, so a redelivered payload cannot process the same message twice.
+ *
+ * Returns the result for each message so callers can report success only when
+ * every message in the batch was handled.
+ */
+export async function enqueueMessages(
+    messages: MessageJobData[],
+): Promise<PromiseSettledResult<Job>[]> {
+    const results = await Promise.allSettled(messages.map((message) => enqueueMessage(message)));
+    return results;
+}
+
 export async function closeQueue(): Promise<void> {
     if (queueInstance) {
         await queueInstance.close();