fix: Split Webhook into Gateway (HuggingFace) and Processor (Railway), fix missing https:// in API_URL

apps/api/src/routes/internal.ts

@@ -1,11 +1,125 @@
 import { FastifyInstance } from 'fastify';
 import { WhatsAppService } from '../services/whatsapp';
+import { z } from 'zod';
+// ─── Zod Schema for WhatsApp Webhook Payload ─────────────────────────────────
+const WhatsAppMessageSchema = z.object({
+    from: z.string(),
+    id: z.string(),
+    timestamp: z.string(),
+    type: z.enum(['text', 'audio', 'image', 'video', 'document', 'sticker', 'reaction', 'interactive']),
+    text: z.object({ body: z.string() }).optional(),
+    audio: z.object({ id: z.string(), mime_type: z.string().optional() }).optional(),
+    image: z.object({ id: z.string() }).optional(),
+    interactive: z.object({
+        type: z.enum(['button_reply', 'list_reply']),
+        button_reply: z.object({
+            id: z.string(),
+            title: z.string(),
+        }).optional(),
+        list_reply: z.object({
+            id: z.string(),
+            title: z.string(),
+            description: z.string().optional()
+        }).optional(),
+    }).optional()
+});
+
+const WebhookPayloadSchema = z.object({
+    object: z.literal('whatsapp_business_account'),
+    entry: z.array(z.object({
+        id: z.string(),
+        changes: z.array(z.object({
+            value: z.object({
+                messaging_product: z.string().optional(),
+                metadata: z.object({ phone_number_id: z.string() }).optional(),
+                contacts: z.array(z.any()).optional(),
+                messages: z.array(WhatsAppMessageSchema).optional(),
+                statuses: z.array(z.any()).optional(),
+            }),
+            field: z.string(),
+        })),
+    })),
+});
 
 /**
  * Internal-only routes — protected by ADMIN_API_KEY, not exposed publicly.
  * Used by the Railway worker to call handleIncomingMessage after audio transcription.
  */
 export async function internalRoutes(fastify: FastifyInstance) {
+    // ── Handle Webhook Forwarding from Gateway (HF -> Railway) ───────────────
+    fastify.post('/v1/internal/whatsapp/inbound', {
+        config: { requireAuth: true } as any
+    }, async (request, reply) => {
+        // We received the raw webhook payload that was forwarded.
+        // Send a 200 immediately to release HF Gateway
+        reply.code(200).send({ ok: true });
+
+        // Process message parsing outside the request loop
+        setImmediate(async () => {
+            try {
+                const parsed = WebhookPayloadSchema.safeParse(request.body);
+
+                if (!parsed.success) {
+                    fastify.log.warn(`[INTERNAL-WEBHOOK] Invalid payload schema: ${JSON.stringify(parsed.error.flatten())}`);
+                    return;
+                }
+
+                const payload = parsed.data;
+
+                for (const entry of payload.entry) {
+                    for (const change of entry.changes) {
+                        const messages = change.value.messages ?? [];
+
+                        for (const message of messages) {
+                            const phone = message.from;
+                            let text = '';
+
+                            if (message.type === 'text' && message.text) {
+                                text = message.text.body;
+
+                            } else if (message.type === 'interactive' && message.interactive) {
+                                if (message.interactive.type === 'button_reply' && message.interactive.button_reply) {
+                                    text = message.interactive.button_reply.id;
+                                    fastify.log.info(`[INTERNAL-WEBHOOK] Button reply: ${text}`);
+                                } else if (message.interactive.type === 'list_reply' && message.interactive.list_reply) {
+                                    text = message.interactive.list_reply.id;
+                                    fastify.log.info(`[INTERNAL-WEBHOOK] List reply: ${text}`);
+                                }
+
+                            } else if (message.type === 'audio' && message.audio) {
+                                // ─── Audio inbound: delegate download to Railway worker ────────────
+                                const accessToken = process.env.WHATSAPP_ACCESS_TOKEN || '';
+                                const { Queue } = await import('bullmq');
+                                const Redis = (await import('ioredis')).default;
+                                const conn = process.env.REDIS_URL
+                                    ? new Redis(process.env.REDIS_URL, { maxRetriesPerRequest: null })
+                                    : new Redis({ host: process.env.REDIS_HOST || 'localhost', port: parseInt(process.env.REDIS_PORT || '6379'), maxRetriesPerRequest: null });
+                                const q = new Queue('whatsapp-queue', { connection: conn as any });
+
+                                await q.add('download-media', {
+                                    mediaId: message.audio.id,
+                                    mimeType: message.audio.mime_type || 'audio/ogg',
+                                    phone,
+                                    accessToken
+                                }, { priority: 1 });
+
+                                fastify.log.info(`[INTERNAL-WEBHOOK] Audio ${message.audio.id} enqueued for Railway download`);
+                                continue;
+                            }
+
+                            if (phone && text) {
+                                await WhatsAppService.handleIncomingMessage(phone, text);
+                            }
+                        }
+                    }
+                }
+            } catch (error) {
+                fastify.log.error(`[INTERNAL-WEBHOOK] Async processing error: ${String(error)}`);
+            }
+        });
+    });
+
+    // ── Handle standard transcribed messages from worker (Railway) ───────────
     fastify.post<{
         Body: { phone: string; text: string; audioUrl?: string }
     }>('/v1/internal/handle-message', {
@@ -35,3 +149,4 @@ export async function internalRoutes(fastify: FastifyInstance) {
         return reply.send({ ok: true });
     });
 }
+

apps/api/src/routes/whatsapp.ts

@@ -100,8 +100,10 @@ export async function whatsappRoutes(fastify: FastifyInstance) {
             const signature = request.headers['x-hub-signature-256'] as string;
             const rawBody = (request as any).rawBody as Buffer;
 
+            // Optional: verify webhook signature. Important for production on both HF and Railway.
             if (!verifyWebhookSignature(rawBody, signature, appSecret)) {
                 request.log.warn('[WEBHOOK] Invalid HMAC signature — request rejected');
+                // Meta won't retry if we return 401/403 directly for invalid signatures. 
                 return reply.code(403).send({ error: 'Invalid signature' });
             }
         } else {
@@ -109,7 +111,32 @@ export async function whatsappRoutes(fastify: FastifyInstance) {
             request.log.warn('[WEBHOOK] WHATSAPP_APP_SECRET not set — skipping signature verification');
         }
 
-        // ── 2. Return 200 IMMEDIATELY to satisfy Meta's < 20s timeout ───────
+        // ── 2. Forward to Railway Internal Worker (if configured as Gateway) ──
+        const railwayInternalUrl = process.env.RAILWAY_INTERNAL_URL;
+
+        if (railwayInternalUrl) {
+            request.log.info(`[WEBHOOK] Gateway Mode: Forwarding payload to Railway API (${railwayInternalUrl}).`);
+            try {
+                // Fire and forget (don't await) to ensure fast 200 response to Meta
+                fetch(`${railwayInternalUrl}/v1/internal/whatsapp/inbound`, {
+                    method: 'POST',
+                    headers: {
+                        'Content-Type': 'application/json',
+                        'x-admin-key': process.env.ADMIN_API_KEY || ''
+                    },
+                    body: request.body ? JSON.stringify(request.body) : ''
+                }).catch(err => {
+                    request.log.error(`[WEBHOOK] Forward to Railway failed: ${err.message}`);
+                });
+
+                return reply.code(200).send('EVENT_RECEIVED');
+            } catch (error: any) {
+                request.log.error(`[WEBHOOK] Forward throwing error: ${error?.message}`);
+                return reply.code(500).send({ error: 'Gateway forwarding failed' });
+            }
+        }
+
+        // ── 3. Return 200 IMMEDIATELY to satisfy Meta's < 20s timeout ───────
         reply.code(200).send('EVENT_RECEIVED');
 
         // ── 3. Validate payload with Zod (async, after reply sent) ──────────

apps/api/src/services/queue.ts

@@ -15,14 +15,26 @@ const connection = process.env.REDIS_URL
 export const whatsappQueue = new Queue('whatsapp-queue', { connection: connection as any });
 
 export async function scheduleMessage(userId: string, text: string, delayMs: number = 0) {
+    if (process.env.DISABLE_WHATSAPP_SEND === 'true') {
+        console.warn(`[QUEUE] DISABLE_WHATSAPP_SEND is true. Skipping 'send-message' for user ${userId}`);
+        return;
+    }
     await whatsappQueue.add('send-message', { userId, text }, { delay: delayMs });
 }
 
 export async function scheduleTrackDay(userId: string, trackId: string, dayNumber: number, delayMs: number = 0) {
+    if (process.env.DISABLE_WHATSAPP_SEND === 'true') {
+        console.warn(`[QUEUE] DISABLE_WHATSAPP_SEND is true. Skipping 'send-content' for user ${userId}`);
+        return;
+    }
     await whatsappQueue.add('send-content', { userId, trackId, dayNumber }, { delay: delayMs });
 }
 
 export async function enrollUser(userId: string, trackId: string) {
+    if (process.env.DISABLE_WHATSAPP_SEND === 'true') {
+        console.warn(`[QUEUE] DISABLE_WHATSAPP_SEND is true. Skipping 'enroll-user' for user ${userId}`);
+        return;
+    }
     await whatsappQueue.add('enroll-user', { userId, trackId });
 }
 
@@ -32,6 +44,10 @@ export async function scheduleInteractiveButtons(
     bodyText: string,
     buttons: Array<{ id: string; title: string }>
 ) {
+    if (process.env.DISABLE_WHATSAPP_SEND === 'true') {
+        console.warn(`[QUEUE] DISABLE_WHATSAPP_SEND is true. Skipping 'send-interactive-buttons' for user ${userId}`);
+        return;
+    }
     await whatsappQueue.add('send-interactive-buttons', { userId, bodyText, buttons });
 }
 
@@ -43,6 +59,10 @@ export async function scheduleInteractiveList(
     buttonLabel: string,
     sections: Array<{ title: string; rows: Array<{ id: string; title: string; description?: string }> }>
 ) {
+    if (process.env.DISABLE_WHATSAPP_SEND === 'true') {
+        console.warn(`[QUEUE] DISABLE_WHATSAPP_SEND is true. Skipping 'send-interactive-list' for user ${userId}`);
+        return;
+    }
     await whatsappQueue.add('send-interactive-list', { userId, headerText, bodyText, buttonLabel, sections });
 }
 

apps/whatsapp-worker/src/index.ts

@@ -63,7 +63,8 @@ const worker = new Worker('whatsapp-queue', async (job: Job) => {
             if (track.isPremium) {
                 console.log(`[WORKER] User ${userId} requested Premium Track ${trackId}. Generating Payment Link...`);
                 try {
-                    const API_URL = process.env.API_URL || 'http://localhost:3001';
+                    let API_URL = process.env.API_URL || 'http://localhost:3001';
+                    if (!API_URL.startsWith('http')) API_URL = `https://${API_URL}`;
                     const apiKey = process.env.ADMIN_API_KEY || '';
                     const checkoutRes = await fetch(`${API_URL}/v1/payments/checkout`, {
                         method: 'POST',
@@ -125,7 +126,8 @@ const worker = new Worker('whatsapp-queue', async (job: Job) => {
                 const { buffer } = await downloadMedia(mediaId, accessToken);
 
                 // Store audio on R2 via the API
-                const API_URL_STORE = process.env.API_URL || 'http://localhost:3001';
+                let API_URL_STORE = process.env.API_URL || 'http://localhost:3001';
+                if (!API_URL_STORE.startsWith('http')) API_URL_STORE = `https://${API_URL_STORE}`;
                 const apiKeyStore = process.env.ADMIN_API_KEY || '';
                 try {
                     const storeRes = await fetch(`${API_URL_STORE}/v1/ai/store-audio`, {
@@ -142,7 +144,8 @@ const worker = new Worker('whatsapp-queue', async (job: Job) => {
                 }
 
                 // Transcribe with Whisper via API
-                const API_URL = process.env.API_URL || 'http://localhost:3001';
+                let API_URL = process.env.API_URL || 'http://localhost:3001';
+                if (!API_URL.startsWith('http')) API_URL = `https://${API_URL}`;
                 const apiKey = process.env.ADMIN_API_KEY || '';
                 const transcribeRes = await fetch(`${API_URL}/v1/ai/transcribe`, {
                     method: 'POST',
@@ -174,7 +177,8 @@ const worker = new Worker('whatsapp-queue', async (job: Job) => {
 
             // Process the transcribed text as a normal incoming message via API
             if (transcribedText) {
-                const API_URL = process.env.API_URL || 'http://localhost:3001';
+                let API_URL = process.env.API_URL || 'http://localhost:3001';
+                if (!API_URL.startsWith('http')) API_URL = `https://${API_URL}`;
                 const apiKey = process.env.ADMIN_API_KEY || '';
                 await fetch(`${API_URL}/v1/internal/handle-message`, {
                     method: 'POST',
@@ -218,7 +222,8 @@ const worker = new Worker('whatsapp-queue', async (job: Job) => {
                     // Update userContext to explicitly reference the user's sector/activity
                     const userContext = `User ${userId} completed the Business Pitch track. Their business activity/sector is: ${user?.activity || 'Unknown'}. They want to build a business in this sector using the concepts learned in the track.`;
 
-                    const API_URL = process.env.API_URL || 'http://localhost:3001';
+                    let API_URL = process.env.API_URL || 'http://localhost:3001';
+                    if (!API_URL.startsWith('http')) API_URL = `https://${API_URL}`;
                     const apiKey = process.env.ADMIN_API_KEY || '';
                     const authHeaders = {
                         'Content-Type': 'application/json',

apps/whatsapp-worker/src/pedagogy.ts

@@ -27,7 +27,8 @@ export async function sendLessonDay(userId: string, trackId: string, dayNumber:
     if (user.activity && lessonText) {
         try {
             console.log(`[PEDAGOGY] Personalizing lesson for User ${userId}'s activity: ${user.activity}`);
-            const API_URL = process.env.API_URL || 'http://localhost:3001';
+            let API_URL = process.env.API_URL || 'http://localhost:3001';
+            if (!API_URL.startsWith('http')) API_URL = `https://${API_URL}`;
             const apiKey = process.env.ADMIN_API_KEY || '';
             const personalizeRes = await fetch(`${API_URL}/v1/ai/personalize-lesson`, {
                 method: 'POST',
@@ -55,7 +56,8 @@ export async function sendLessonDay(userId: string, trackId: string, dayNumber:
     if (!finalAudioUrl && lessonText) {
         try {
             console.log(`[PEDAGOGY] Generating TTS Audio for User ${userId}...`);
-            const API_URL = process.env.API_URL || 'http://localhost:3001';
+            let API_URL = process.env.API_URL || 'http://localhost:3001';
+            if (!API_URL.startsWith('http')) API_URL = `https://${API_URL}`;
             const apiKey = process.env.ADMIN_API_KEY || '';
             const ttsRes = await fetch(`${API_URL}/v1/ai/tts`, {
                 method: 'POST',