feat: implement WhatsApp broadcast system with BullMQ worker and rate limiting

apps/admin/src/pages/CrmConversationalDashboard.tsx

@@ -148,12 +148,44 @@ export default function CrmConversationalDashboard() {
         }
     };
 
-    const handleValidateAndSend = (_message: string) => {
-        if (!uploadedFile) {
+    const handleValidateAndSend = async (message: string) => {
+        if (!uploadedFile || !token || !selectedOrgId) {
             alert("Veuillez d'abord importer une liste de contacts.");
             return;
         }
-        alert(`Campagne validée ! Envoi du message vers la liste "${uploadedFile.listName}"... (Action à implémenter)`);
+
+        try {
+            const res = await fetch(`${import.meta.env.VITE_API_URL}/v1/organizations/${selectedOrgId}/campaigns/send`, {
+                method: 'POST',
+                headers: {
+                    'Content-Type': 'application/json',
+                    'Authorization': `Bearer ${token}`
+                },
+                body: JSON.stringify({ 
+                    listId: uploadedFile.listId,
+                    message: message 
+                })
+            });
+
+            if (res.ok) {
+                alert("🚀 Votre campagne est en cours de distribution !");
+                // Reset state for a new campaign
+                setUploadedFile(null);
+                setMessages([
+                    {
+                        id: 'initial-' + Date.now(),
+                        role: 'assistant',
+                        content: "Campagne lancée avec succès ! Souhaitez-vous préparer une autre liste ou une nouvelle campagne ?",
+                        timestamp: new Date()
+                    }
+                ]);
+            } else {
+                alert("Erreur lors de l'envoi de la campagne.");
+            }
+        } catch (err) {
+            console.error("Send failed:", err);
+            alert("Une erreur technique est survenue.");
+        }
     };
 
     return (

apps/api/src/routes/campaigns.ts

@@ -24,4 +24,28 @@ export default async function campaignRoutes(fastify: FastifyInstance) {
             return reply.code(500).send({ error: 'AI Generation failed' });
         }
     });
+
+    // Send Campaign to Broadcast List
+    fastify.post('/:id/campaigns/send', async (req, reply) => {
+        const { id: organizationId } = req.params as { id: string };
+        const { listId, message } = req.body as { listId: string, message: string };
+
+        if (!listId || !message) {
+            return reply.code(400).send({ error: 'listId and message are required' });
+        }
+
+        try {
+            const { scheduleBroadcast } = await import('../services/queue');
+            await scheduleBroadcast({ organizationId, listId, message });
+
+            return reply.code(202).send({ 
+                ok: true, 
+                status: 'queued', 
+                message: 'Campagne en cours d\'envoi en arrière-plan' 
+            });
+        } catch (err) {
+            fastify.log.error(err);
+            return reply.code(500).send({ error: 'Failed to enqueue campaign' });
+        }
+    });
 }

apps/api/src/services/queue.ts

@@ -127,3 +127,11 @@ export async function scheduleInboundMessage(payload: { phone: string, text: str
     });
 }
 
+/** 📢 BROADCAST: Enqueue a mass message task. */
+export async function scheduleBroadcast(payload: { organizationId: string, listId: string, message: string }) {
+    await whatsappQueue.add('send-broadcast', payload, {
+        attempts: 1, // We handle retry logic within the loop if needed, but the whole job shouldn't necessarily retry
+        removeOnComplete: true
+    });
+}
+

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

@@ -0,0 +1,97 @@
+import { Job } from 'bullmq';
+import Redis from 'ioredis';
+import { JobHandler, JobData } from './types';
+import { prisma } from '../services/prisma';
+import { logger } from '../logger';
+
+export class BroadcastHandler implements JobHandler {
+    private async getTenantConfig(organizationId: string, connection: Redis) {
+        const cacheKey = `org:config:${organizationId}`;
+        const cached = await connection.get(cacheKey);
+        if (cached) return JSON.parse(cached);
+
+        const org = await prisma.organization.findUnique({
+            where: { id: organizationId },
+            include: { phoneNumbers: true }
+        });
+        
+        if (!org || !org.systemUserToken || !org.phoneNumbers?.[0]?.id) return undefined;
+
+        const config = {
+            accessToken: org.systemUserToken,
+            phoneNumberId: org.phoneNumbers[0].id
+        };
+        await connection.set(cacheKey, JSON.stringify(config), 'EX', 3600);
+        return config;
+    }
+
+    async handle(job: Job<JobData>, connection: Redis): Promise<void> {
+        if (job.name !== 'send-broadcast') return;
+
+        const { organizationId, listId, message } = job.data as { organizationId: string, listId: string, message: string };
+        
+        const tenantConfig = await this.getTenantConfig(organizationId, connection);
+        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 }
+        });
+
+        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`);
+
+        const baseUrl = `https://graph.facebook.com/v19.0/${tenantConfig.phoneNumberId}/messages`;
+        const headers = {
+            'Content-Type': 'application/json',
+            'Authorization': `Bearer ${tenantConfig.accessToken}`
+        };
+
+        let successCount = 0;
+        let failCount = 0;
+
+        for (const contact of list.contacts) {
+            try {
+                const body = {
+                    messaging_product: 'whatsapp',
+                    recipient_type: 'individual',
+                    to: contact.phoneNumber,
+                    type: 'text',
+                    text: { preview_url: false, body: message }
+                };
+
+                const response = await fetch(baseUrl, {
+                    method: 'POST',
+                    headers,
+                    body: JSON.stringify(body)
+                });
+
+                if (response.ok) {
+                    successCount++;
+                } else {
+                    const errorData = await response.json();
+                    logger.warn({ phoneNumber: contact.phoneNumber, errorData }, '[BroadcastHandler] Meta API returned error');
+                    failCount++;
+                }
+
+                // 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));
+
+            } catch (err) {
+                logger.error({ phoneNumber: contact.phoneNumber, err }, '[BroadcastHandler] Failed to send to contact');
+                failCount++;
+            }
+        }
+
+        logger.info(`[BroadcastHandler] Broadcast finished for "${list.name}". Success: ${successCount}, Failed: ${failCount}`);
+    }
+}

apps/whatsapp-worker/src/index.ts

@@ -26,6 +26,7 @@ import { InboundHandler } from './handlers/InboundHandler';
 import { WebhookHandler } from './handlers/WebhookHandler';
 import { KBProcessor } from './handlers/KBProcessor';
 import { EmailHandler } from './handlers/EmailHandler';
+import { BroadcastHandler } from './handlers/BroadcastHandler';
 import { UsageService } from './services/usage';
 
 dotenv.config();
@@ -65,7 +66,8 @@ const handlers: Record<string, JobHandler> = {
     'handle-inbound': new InboundHandler(),
     'send-webhook': new WebhookHandler(),
     'process-kb': new KBProcessor(),
-    'send-email': new EmailHandler()
+    'send-email': new EmailHandler(),
+    'send-broadcast': new BroadcastHandler()
 };
 
 // ─── HTTP SERVER (Inbound Bridge) ─────────────────────────────────────────────
@@ -165,7 +167,7 @@ const worker = new Worker('whatsapp-queue', async (job: Job<JobData>) => {
     const outboundJobNames = [
         'send-message', 'send-message-direct', 'send-image', 
         'send-interactive-buttons', 'send-interactive-list', 
-        'send-content', 'send-nudge'
+        'send-content', 'send-nudge', 'send-broadcast'
     ];
 
     if (outboundJobNames.includes(job.name)) {