fix: add bridge server to worker to handle forwarded webhooks

apps/whatsapp-worker/package.json

@@ -16,6 +16,7 @@
         "axios": "^1.13.5",
         "bullmq": "^4.0.0",
         "dotenv": "^16.0.0",
+        "fastify": "^4.24.3",
         "ioredis": "^5.9.3",
         "node-cron": "^4.2.1",
         "node-fetch": "^2.6.7",

apps/whatsapp-worker/src/index.ts

@@ -2,7 +2,8 @@ import { logger } from './logger';
 import dns from 'node:dns';
 dns.setDefaultResultOrder('ipv4first');
 
-import { Worker, Job } from 'bullmq';
+import fastify from 'fastify';
+import { Queue, Worker, Job } from 'bullmq';
 import dotenv from 'dotenv';
 import Redis from 'ioredis';
 import { validateEnvironment } from './config';
@@ -24,16 +25,21 @@ dotenv.config();
 validateEnvironment();
 startWorkerCleanupCron();
 
-const connection = process.env.REDIS_URL
-    ? new Redis(process.env.REDIS_URL, { maxRetriesPerRequest: null })
-    : new Redis({
+const redisConfig = process.env.REDIS_URL
+    ? { url: process.env.REDIS_URL }
+    : {
         host: process.env.REDIS_HOST || 'localhost',
         port: parseInt(process.env.REDIS_PORT || '6379'),
         username: process.env.REDIS_USERNAME || 'default',
         password: process.env.REDIS_PASSWORD || undefined,
         tls: process.env.REDIS_TLS === 'true' ? {} : undefined,
-        maxRetriesPerRequest: null
-    });
+    };
+
+const connection = process.env.REDIS_URL
+    ? new Redis(process.env.REDIS_URL, { maxRetriesPerRequest: null })
+    : new Redis({ ...redisConfig, maxRetriesPerRequest: null } as any);
+
+const whatsappQueue = new Queue('whatsapp-queue', { connection: connection as any });
 
 const handlers: Record<string, JobHandler> = {
     'send-message': new MessageHandler(),
@@ -49,6 +55,31 @@ const handlers: Record<string, JobHandler> = {
     'handle-inbound': new InboundHandler()
 };
 
+// ─── HTTP SERVER (Inbound Bridge) ─────────────────────────────────────────────
+// This server receives forwarded webhooks from the Hugging Face Gateway
+const server = fastify({ logger: false });
+
+server.post('/v1/internal/whatsapp/inbound', async (req, reply) => {
+    const organizationId = req.headers['x-organization-id'] as string || 'default-org-id';
+    const payload = req.body as any;
+
+    logger.info(`[BRIDGE] Received forwarded webhook for Org: ${organizationId}`);
+
+    // Schedule the job to be processed by the worker modularly
+    await whatsappQueue.add('handle-inbound', {
+        ...payload,
+        organizationId
+    }, {
+        attempts: 3,
+        backoff: { type: 'exponential', delay: 1000 }
+    });
+
+    return reply.code(200).send({ ok: true });
+});
+
+server.get('/health', async () => ({ status: 'ok' }));
+
+// ─── WORKER ──────────────────────────────────────────────────────────────────
 const worker = new Worker('whatsapp-queue', async (job: Job<JobData>) => {
     const organizationId = job.data.organizationId || 'default-org-id';
 
@@ -77,11 +108,24 @@ const worker = new Worker('whatsapp-queue', async (job: Job<JobData>) => {
     concurrency: parseInt(process.env.WORKER_CONCURRENCY || '5')
 });
 
-logger.info('🚀 WhatsApp Worker started (Modular Mode)...');
+// ─── STARTUP ─────────────────────────────────────────────────────────────────
+const PORT = parseInt(process.env.PORT || '7860');
+
+const start = async () => {
+    try {
+        await server.listen({ port: PORT, host: '0.0.0.0' });
+        logger.info(`🚀 WhatsApp Worker + Bridge listening on port ${PORT}`);
+        
+        // Start the daily cron scheduler
+        const { startDailyScheduler } = await import('./scheduler');
+        startDailyScheduler();
+    } catch (err) {
+        logger.error('Failed to start worker server:', err);
+        process.exit(1);
+    }
+};
 
-// Start the daily cron scheduler
-import { startDailyScheduler } from './scheduler';
-startDailyScheduler();
+start();
 
 worker.on('completed', job => {
     logger.info(`[WORKER] Job ${job.id} has completed!`);