perf(worker): Sprint 3 — broadcast/campaign scalability fixes

H4: Replace per-contact prisma.message.create() inside the loop with
a batch prisma.message.createMany() after each page — reduces N DB
round-trips to 1 per page of 200 contacts.

H5: Replace include:{contacts:true} (loads all contacts into memory) with
cursor-based pagination (PAGE_SIZE=200) using broadcastLists relation
filter. Large lists no longer OOM the worker.

H6 (usage check): Move UsageService.checkAndIncrement() from per-contact
to per-page with amount=page.length — reduces Redis INCRBY calls from
N to N/200.

H11: Add job.updateProgress() after each page in CampaignHandler so BullMQ's
stall detector sees the job as alive during long campaigns; also provides
live progress in BullBoard (0–99%).

H10: Confirmed resolved — wallet:ok:{orgId} cache key is consistent across
API middleware and worker since credit-guard.ts deletion in Sprint 2.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

apps/whatsapp-worker/src/handlers/BroadcastHandler.ts

@@ -8,6 +8,8 @@ import { getCachedOrganization } from '../services/organization';
 import { debitWallet } from '../services/wallet';
 import { CREDIT_PRICE } from '@repo/database';
 
+const PAGE_SIZE = 200;
+
 export class BroadcastHandler implements JobHandler {
     private async getTenantConfig(organizationId: string) {
         const org = await getCachedOrganization(organizationId);
@@ -18,81 +20,103 @@ export class BroadcastHandler implements JobHandler {
     async handle(job: Job<JobData>, connection: Redis): Promise<void> {
         if (job.name !== 'send-broadcast') return;
 
-        const { organizationId, listId, message, templateName, templateLanguage } = job.data as { 
-            organizationId: string, 
-            listId: string, 
+        const { organizationId, listId, message, templateName, templateLanguage } = job.data as {
+            organizationId: string,
+            listId: string,
             message: string,
             templateName?: string,
             templateLanguage?: string
         };
-        
+
         const tenantConfig = await this.getTenantConfig(organizationId);
         if (!tenantConfig) {
             logger.error({ organizationId }, '[BroadcastHandler] No WhatsApp config found for organization');
             return;
         }
 
-        // 1. Fetch the list with its contacts
         const list = await prisma.broadcastList.findUnique({
             where: { id: listId },
-            include: { contacts: true }
+            select: { name: true },
         });
-
         if (!list) {
             logger.error({ listId }, '[BroadcastHandler] Broadcast list not found');
             return;
         }
 
-        logger.info(`[BroadcastHandler] Starting broadcast for list "${list.name}" to ${list.contacts.length} contacts (Template: ${templateName || 'None'})`);
+        logger.info({ organizationId, listId }, `[BroadcastHandler] Starting broadcast "${list.name}" (Template: ${templateName || 'None'})`);
 
         const { sendTextMessage, sendTemplateMessage } = await import('../whatsapp-cloud');
 
         let successCount = 0;
         let failCount = 0;
+        let cursor: string | undefined;
 
-        for (const contact of list.contacts) {
-            try {
-                // 1.5 Check Usage Limit
-                const { allowed } = await UsageService.checkAndIncrement(organizationId, connection);
-                if (!allowed) {
-                    logger.warn({ organizationId, listId }, '[BroadcastHandler] Daily limit reached during broadcast. Halting.');
-                    break;
-                }
+        while (true) {
+            const contacts = await prisma.contact.findMany({
+                where: { broadcastLists: { some: { id: listId } } },
+                select: { id: true, phoneNumber: true },
+                take: PAGE_SIZE,
+                orderBy: { id: 'asc' },
+                ...(cursor ? { cursor: { id: cursor }, skip: 1 } : {}),
+            });
 
-                if (templateName) {
-                    await sendTemplateMessage(contact.phoneNumber, templateName, templateLanguage || 'fr', tenantConfig);
-                } else {
-                    await sendTextMessage(contact.phoneNumber, message, tenantConfig);
-                }
+            if (contacts.length === 0) break;
+            cursor = contacts[contacts.length - 1].id;
+
+            // Reserve quota for the whole page in one Redis call
+            const { allowed } = await UsageService.checkAndIncrement(organizationId, connection, contacts.length);
+            if (!allowed) {
+                logger.warn({ organizationId, listId }, '[BroadcastHandler] Daily limit reached. Halting.');
+                break;
+            }
 
-                successCount++;
-                // P0 Fix: Persist broadcast message for history
+            const messageRecords: Array<{
+                organizationId: string;
+                contactId: string;
+                content: string;
+                direction: 'OUTBOUND';
+                status: 'SENT';
+                channel: string;
+            }> = [];
+
+            for (const contact of contacts) {
                 try {
-                    await prisma.message.create({
-                        data: {
-                            organizationId,
-                            contactId: contact.id,
-                            content: templateName ? `[TEMPLATE: ${templateName}]` : message,
-                            direction: 'OUTBOUND',
-                            status: 'SENT',
-                            channel: 'WHATSAPP'
-                        }
+                    if (templateName) {
+                        await sendTemplateMessage(contact.phoneNumber, templateName, templateLanguage || 'fr', tenantConfig);
+                    } else {
+                        await sendTextMessage(contact.phoneNumber, message, tenantConfig);
+                    }
+
+                    successCount++;
+                    messageRecords.push({
+                        organizationId,
+                        contactId: contact.id,
+                        content: templateName ? `[TEMPLATE: ${templateName}]` : message,
+                        direction: 'OUTBOUND',
+                        status: 'SENT',
+                        channel: 'WHATSAPP',
                     });
-                } catch (dbErr) {
-                    logger.warn({ dbErr, contactId: contact.id }, '[BroadcastHandler] Message persisted failed but Meta send succeeded');
-                }
 
-                // 2. Rate limiting: sleep between 50ms and 100ms
-                const delay = Math.floor(Math.random() * (100 - 50 + 1) + 50);
-                await new Promise(resolve => setTimeout(resolve, delay));
+                    // Rate limiting: 50–100ms between sends
+                    const delay = Math.floor(Math.random() * 51) + 50;
+                    await new Promise(resolve => setTimeout(resolve, delay));
+                } catch (err) {
+                    logger.error({ phoneNumber: contact.phoneNumber, err }, '[BroadcastHandler] Failed to send to contact');
+                    failCount++;
+                }
+            }
 
-            } catch (err) {
-                logger.error({ phoneNumber: contact.phoneNumber, err }, '[BroadcastHandler] Failed to send to contact');
-                failCount++;
+            // Batch persist for the page — one DB round-trip instead of N
+            if (messageRecords.length > 0) {
+                await prisma.message.createMany({ data: messageRecords }).catch(err =>
+                    logger.warn({ err, organizationId }, '[BroadcastHandler] createMany failed')
+                );
             }
+
+            if (contacts.length < PAGE_SIZE) break;
         }
 
-        logger.info(`[BroadcastHandler] Broadcast finished for "${list.name}". Success: ${successCount}, Failed: ${failCount}`);
+        logger.info(`[BroadcastHandler] Broadcast "${list.name}" finished. Success: ${successCount}, Failed: ${failCount}`);
 
         if (successCount > 0) {
             try {

apps/whatsapp-worker/src/handlers/CampaignHandler.ts

@@ -8,6 +8,8 @@ import { getCachedOrganization } from '../services/organization';
 import { debitWallet } from '../services/wallet';
 import { CREDIT_PRICE } from '@repo/database';
 
+const PAGE_SIZE = 200;
+
 export class CampaignHandler implements JobHandler {
     private async getTenantConfig(organizationId: string) {
         const org = await getCachedOrganization(organizationId);
@@ -18,15 +20,15 @@ export class CampaignHandler implements JobHandler {
     async handle(job: Job<JobData>, connection: Redis): Promise<void> {
         if (job.name !== 'process-campaign') return;
 
-        const { organizationId, messageContent, listId, templateName, templateLanguage } = job.data as { 
-            organizationId: string, 
-            messageContent: string, 
+        const { organizationId, messageContent, listId, templateName, templateLanguage } = job.data as {
+            organizationId: string,
+            messageContent: string,
             listId?: string,
             templateName?: string,
             templateLanguage?: string
         };
-        
-        logger.info({ organizationId, listId }, `[CampaignHandler] Starting campaign processing (Template: ${templateName || 'None'})...`);
+
+        logger.info({ organizationId, listId }, `[CampaignHandler] Starting campaign (Template: ${templateName || 'None'})`);
 
         const tenantConfig = await this.getTenantConfig(organizationId);
         if (!tenantConfig) {
@@ -34,72 +36,95 @@ export class CampaignHandler implements JobHandler {
             return;
         }
 
-        // 1. Resolve contacts
-        let contacts: any[] = [];
-        if (listId) {
-            const list = await prisma.broadcastList.findUnique({
-                where: { id: listId },
-                include: { contacts: true }
-            });
-            contacts = list?.contacts || [];
-        } else {
-            // Fallback: send to all contacts of the organization
-            contacts = await prisma.contact.findMany({
-                where: { organizationId }
-            });
-        }
-
-        if (contacts.length === 0) {
-            logger.warn({ organizationId, listId }, '[CampaignHandler] No contacts found for campaign');
-            return;
-        }
-
-        logger.info(`[CampaignHandler] Broadcasting to ${contacts.length} contacts`);
-
         const { sendTextMessage, sendTemplateMessage } = await import('../whatsapp-cloud');
 
         let successCount = 0;
         let failCount = 0;
+        let cursor: string | undefined;
+        let pageIndex = 0;
+        const totalEstimate = await (listId
+            ? prisma.contact.count({ where: { broadcastLists: { some: { id: listId } } } })
+            : prisma.contact.count({ where: { organizationId } })
+        );
+
+        while (true) {
+            const contacts = await (listId
+                ? prisma.contact.findMany({
+                    where: { broadcastLists: { some: { id: listId } } },
+                    select: { id: true, phoneNumber: true },
+                    take: PAGE_SIZE,
+                    orderBy: { id: 'asc' },
+                    ...(cursor ? { cursor: { id: cursor }, skip: 1 } : {}),
+                })
+                : prisma.contact.findMany({
+                    where: { organizationId },
+                    select: { id: true, phoneNumber: true },
+                    take: PAGE_SIZE,
+                    orderBy: { id: 'asc' },
+                    ...(cursor ? { cursor: { id: cursor }, skip: 1 } : {}),
+                })
+            );
+
+            if (contacts.length === 0) break;
+            cursor = contacts[contacts.length - 1].id;
+            pageIndex++;
+
+            // Reserve quota for the whole page in one Redis call
+            const { allowed } = await UsageService.checkAndIncrement(organizationId, connection, contacts.length);
+            if (!allowed) {
+                logger.warn({ organizationId }, '[CampaignHandler] Daily limit reached. Halting campaign.');
+                break;
+            }
 
-        for (const contact of contacts) {
-            try {
-                // Usage Limit Check
-                const { allowed } = await UsageService.checkAndIncrement(organizationId, connection);
-                if (!allowed) {
-                    logger.warn({ organizationId }, '[CampaignHandler] Daily limit reached. Halting campaign.');
-                    break;
-                }
-
-                if (templateName) {
-                    await sendTemplateMessage(contact.phoneNumber, templateName, templateLanguage || 'fr', tenantConfig);
-                } else {
-                    await sendTextMessage(contact.phoneNumber, messageContent, tenantConfig);
-                }
+            const messageRecords: Array<{
+                organizationId: string;
+                contactId: string;
+                content: string;
+                direction: 'OUTBOUND';
+                status: 'SENT';
+                channel: string;
+            }> = [];
 
-                successCount++;
-                // Persist for history
+            for (const contact of contacts) {
                 try {
-                    await prisma.message.create({
-                        data: {
-                            organizationId,
-                            contactId: contact.id,
-                            content: templateName ? `[TEMPLATE: ${templateName}]` : messageContent,
-                            direction: 'OUTBOUND',
-                            status: 'SENT',
-                            channel: 'WHATSAPP'
-                        }
+                    if (templateName) {
+                        await sendTemplateMessage(contact.phoneNumber, templateName, templateLanguage || 'fr', tenantConfig);
+                    } else {
+                        await sendTextMessage(contact.phoneNumber, messageContent, tenantConfig);
+                    }
+
+                    successCount++;
+                    messageRecords.push({
+                        organizationId,
+                        contactId: contact.id,
+                        content: templateName ? `[TEMPLATE: ${templateName}]` : messageContent,
+                        direction: 'OUTBOUND',
+                        status: 'SENT',
+                        channel: 'WHATSAPP',
                     });
-                } catch (dbErr) {
-                    logger.warn({ dbErr, contactId: contact.id }, '[CampaignHandler] Message persistence failed');
-                }
 
-                // 2. REQUIRED DELAY: 1000ms as per instructions
-                await new Promise(resolve => setTimeout(resolve, 1000));
+                    // Required delay between sends to respect Meta rate limits
+                    await new Promise(resolve => setTimeout(resolve, 1000));
+                } catch (err) {
+                    logger.error({ phoneNumber: contact.phoneNumber, err }, '[CampaignHandler] Failed to send');
+                    failCount++;
+                }
+            }
 
-            } catch (err) {
-                logger.error({ phoneNumber: contact.phoneNumber, err }, '[CampaignHandler] Failed to send');
-                failCount++;
+            // Batch persist for the page — one DB round-trip instead of N
+            if (messageRecords.length > 0) {
+                await prisma.message.createMany({ data: messageRecords }).catch(err =>
+                    logger.warn({ err, organizationId }, '[CampaignHandler] createMany failed')
+                );
             }
+
+            // Signal liveness to BullMQ stall detector and update progress in BullBoard
+            const progress = totalEstimate > 0
+                ? Math.min(Math.floor(((pageIndex * PAGE_SIZE) / totalEstimate) * 100), 99)
+                : 0;
+            await job.updateProgress(progress);
+
+            if (contacts.length < PAGE_SIZE) break;
         }
 
         logger.info(`[CampaignHandler] Finished. Success: ${successCount}, Failed: ${failCount}`);