From 3ac9cf5d8cf27cb0a74207f5debef2492fdd4372 Mon Sep 17 00:00:00 2001 From: Emir Karabeg Date: Tue, 11 Mar 2025 03:31:34 -0700 Subject: [PATCH] fix: duplicate whatsapp messasge fix --- app/api/webhooks/trigger/[path]/route.ts | 145 ++++++++++++----------- 1 file changed, 76 insertions(+), 69 deletions(-) diff --git a/app/api/webhooks/trigger/[path]/route.ts b/app/api/webhooks/trigger/[path]/route.ts index ee2dfa6452..152aea338b 100644 --- a/app/api/webhooks/trigger/[path]/route.ts +++ b/app/api/webhooks/trigger/[path]/route.ts @@ -11,6 +11,11 @@ import { Serializer } from '@/serializer' export const dynamic = 'force-dynamic' +// Store for tracking processed message IDs in memory +// This is a simple in-memory solution that works for a single instance +// For multi-instance deployments, consider using Redis or another distributed cache +const processedMessageIds = new Set() + /** * Consolidated webhook trigger endpoint for all providers * Handles both WhatsApp verification and other webhook providers @@ -125,6 +130,74 @@ export async function POST( const { webhook: foundWebhook, workflow: workflowData } = webhooks[0] foundWorkflow = workflowData + // For WhatsApp, check for duplicate messages before processing + if (foundWebhook.provider === 'whatsapp') { + const data = body?.entry?.[0]?.changes?.[0]?.value + const messages = data?.messages || [] + + if (messages.length > 0) { + const message = messages[0] + const messageId = message.id + + // Check if we've already processed this message + if (messageId && processedMessageIds.has(messageId)) { + console.log( + `Duplicate WhatsApp message detected with ID: ${messageId}. Skipping processing.` + ) + return new NextResponse('OK - Duplicate message', { status: 200 }) + } + + // Store the message ID to prevent duplicate processing in future requests + if (messageId) { + processedMessageIds.add(messageId) + + // Clean up old message IDs periodically (keep last ~1000 messages) + if (processedMessageIds.size > 1000) { + const idsArray = Array.from(processedMessageIds) + for (let i = 0; i < idsArray.length - 1000; i++) { + processedMessageIds.delete(idsArray[i]) + } + } + } + + // Process the webhook synchronously - complete the workflow before returning + const result = await processWebhook(foundWebhook, foundWorkflow, body, request, executionId) + + // After workflow execution is complete, return 200 OK + console.log(`Workflow execution complete for WhatsApp message ID: ${messageId}`) + return result + } else { + // This might be a different type of notification (e.g., status update) + console.log('No messages in WhatsApp payload, might be a status update') + return new NextResponse('OK', { status: 200 }) + } + } + + // For other providers, continue with synchronous processing + return await processWebhook(foundWebhook, foundWorkflow, body, request, executionId) + } catch (error: any) { + console.error('Error processing webhook:', error) + + // Log the error if we have a workflow ID + if (foundWorkflow?.id) { + await persistExecutionError(foundWorkflow.id, executionId, error, 'webhook') + } + + return new NextResponse(`Internal Server Error: ${error.message}`, { status: 500 }) + } +} + +/** + * Process a webhook synchronously + */ +async function processWebhook( + foundWebhook: any, + foundWorkflow: any, + body: any, + request: NextRequest, + executionId: string +): Promise { + try { // Handle provider-specific verification and authentication if (foundWebhook.provider && foundWebhook.provider !== 'whatsapp') { const authHeader = request.headers.get('authorization') @@ -167,7 +240,7 @@ export async function POST( const text = message.text?.body console.log( - `Received WhatsApp message: ${text ? text.substring(0, 50) : '[no text]'} from ${from}` + `Processing WhatsApp message: ${text ? text.substring(0, 50) : '[no text]'} from ${from}` ) input = { @@ -202,7 +275,7 @@ export async function POST( input = { webhook: { data: { - path, + path: foundWebhook.path, provider: foundWebhook.provider, providerConfig: foundWebhook.providerConfig, payload: body, @@ -256,32 +329,6 @@ export async function POST( decryptedEnvVars = Object.fromEntries(decryptedEntries) } - // Log environment variables for debugging (masking sensitive values) - console.log('Available environment variables before execution:') - Object.keys(decryptedEnvVars).forEach((key) => { - const value = decryptedEnvVars[key] - // Show full value for OpenAI API key for debugging, mask others - if (key === 'OPENAI_API_KEY') { - console.log(` ${key}: ${value} (FULL KEY FOR DEBUGGING)`) - } else { - // Mask sensitive values like API keys, showing only first and last few characters - const maskedValue = - key.toLowerCase().includes('key') || - key.toLowerCase().includes('secret') || - key.toLowerCase().includes('token') - ? `${value.substring(0, 4)}...${value.substring(value.length - 4)}` - : value - console.log(` ${key}: ${maskedValue}`) - } - }) - - // Check specifically for OpenAI API key - if (!decryptedEnvVars['OPENAI_API_KEY']) { - console.warn('WARNING: OPENAI_API_KEY is missing from environment variables!') - } else { - console.log('OPENAI_API_KEY is present in environment variables') - } - // Process the block states to extract values from subBlocks const currentBlockStates = Object.entries(mergedStates).reduce( (acc, [id, block]) => { @@ -297,22 +344,6 @@ export async function POST( {} as Record> ) - // Log block states for debugging - console.log(`[Webhook Debug] Block states after async merging:`) - Object.entries(currentBlockStates).forEach(([blockId, state]) => { - // Check for agent blocks specifically - if (blocks[blockId]?.type === 'agent') { - console.log(`[Webhook Debug] Agent block ${blockId} state:`, JSON.stringify(state, null, 2)) - // Check for null values - const nullKeys = Object.entries(state) - .filter(([_, value]) => value === null) - .map(([key]) => key) - if (nullKeys.length > 0) { - console.warn(`[Webhook Debug] Agent block ${blockId} has null values for:`, nullKeys) - } - } - }) - // Serialize and execute the workflow const serializedWorkflow = new Serializer().serializeWorkflow(mergedStates as any, edges, loops) @@ -322,26 +353,13 @@ export async function POST( workflowId: foundWorkflow.id, } - console.log('Executing workflow with workflowId:', foundWorkflow.id) - // Process the block states to ensure response formats are properly parsed - // This is crucial for agent blocks with response format const processedBlockStates = Object.entries(currentBlockStates).reduce( (acc, [blockId, blockState]) => { // Check if this block has a responseFormat that needs to be parsed if (blockState.responseFormat && typeof blockState.responseFormat === 'string') { try { - console.log( - `[Webhook Debug] Block ${blockId} has responseFormat as string:`, - blockState.responseFormat - ) - // Attempt to parse the responseFormat if it's a string const parsedResponseFormat = JSON.parse(blockState.responseFormat) - console.log( - `[Webhook Debug] Successfully parsed responseFormat for block ${blockId}:`, - parsedResponseFormat - ) - acc[blockId] = { ...blockState, responseFormat: parsedResponseFormat, @@ -358,20 +376,9 @@ export async function POST( {} as Record> ) - // Remove the timeout and use the already awaited states - console.log('[Webhook Debug] Final processed block states before execution:') - Object.entries(processedBlockStates).forEach(([blockId, state]) => { - if (blocks[blockId]?.type === 'agent') { - console.log( - `[Webhook Debug] Agent block ${blockId} final state:`, - JSON.stringify(state, null, 2) - ) - } - }) - const executor = new Executor( serializedWorkflow, - processedBlockStates, // Use the processed block states + processedBlockStates, decryptedEnvVars, enrichedInput )