diff --git a/.gitignore b/.gitignore index bea4619..ac0c2db 100644 --- a/.gitignore +++ b/.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 diff --git a/src/__tests__/bot.controller.test.ts b/src/__tests__/bot.controller.test.ts index 3c2ee3d..1642388 100644 --- a/src/__tests__/bot.controller.test.ts +++ b/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,6 +16,10 @@ 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; @@ -22,6 +27,7 @@ describe('BotController', () => { 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); + }); }); }); diff --git a/src/controllers/bot.controller.ts b/src/controllers/bot.controller.ts index 8e546f5..cbe6fb0 100644 --- a/src/controllers/bot.controller.ts +++ b/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. */ diff --git a/src/middleware/command-rate-limit.middleware.ts b/src/middleware/command-rate-limit.middleware.ts index 3bb23e2..14c2d2f 100644 --- a/src/middleware/command-rate-limit.middleware.ts +++ b/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 }; +} diff --git a/src/queue/message.queue.ts b/src/queue/message.queue.ts index 9368ac0..6720408 100644 --- a/src/queue/message.queue.ts +++ b/src/queue/message.queue.ts @@ -47,6 +47,20 @@ export async function enqueueMessage(data: MessageJobData): Promise { }); } +/** + * 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[]> { + const results = await Promise.allSettled(messages.map((message) => enqueueMessage(message))); + return results; +} + export async function closeQueue(): Promise { if (queueInstance) { await queueInstance.close();