diff --git a/sim/app/api/webhooks/trigger/[path]/route.ts b/sim/app/api/webhooks/trigger/[path]/route.ts index 82d0379651..a6ab939ad7 100644 --- a/sim/app/api/webhooks/trigger/[path]/route.ts +++ b/sim/app/api/webhooks/trigger/[path]/route.ts @@ -4,7 +4,6 @@ import { v4 as uuidv4 } from 'uuid' import { createLogger } from '@/lib/logs/console-logger' import { db } from '@/db' import { webhook, workflow } from '@/db/schema' -import { acquireLock, hasProcessedMessage, markMessageAsProcessed } from '@/lib/redis' import { handleWhatsAppVerification, handleSlackChallenge, @@ -13,51 +12,45 @@ import { processWebhook, fetchAndProcessAirtablePayloads } from '@/lib/webhooks/utils' +import { getOAuthToken } from '@/app/api/auth/oauth/utils' +import { acquireLock, hasProcessedMessage, markMessageAsProcessed } from '@/lib/redis' + +// Set dynamic rendering and maximum duration for long-running webhook triggers +export const dynamic = 'force-dynamic' +export const maxDuration = 300 // 5 minutes const logger = createLogger('WebhookTriggerAPI') -// Ensure dynamic rendering to support real-time webhook processing -export const dynamic = 'force-dynamic' -export const maxDuration = 300 // 5 minutes max execution time for long-running webhooks - /** - * Webhook Verification Handler (GET) - * - * Handles verification requests from webhook providers: - * - WhatsApp: Responds to hub.challenge verification - * - Generic: Confirms webhook endpoint exists and is active - * - * @param request The incoming HTTP request - * @param params Route parameters containing the webhook path - * @returns HTTP response appropriate for the verification type + * GET handler for webhook verification. + * + * This handles provider-specific challenges (e.g. WhatsApp, Slack) as well as generic endpoint checks. */ -export async function GET(request: NextRequest, { params }: { params: Promise<{ path: string }> }) { +export async function GET( + request: NextRequest, + { params }: { params: Promise<{ path: string }> } +) { const requestId = crypto.randomUUID().slice(0, 8) - try { - const path = (await params).path + const { path } = await params const url = new URL(request.url) // --- WhatsApp Verification --- - // Extract WhatsApp challenge parameters const mode = url.searchParams.get('hub.mode') const token = url.searchParams.get('hub.verify_token') const challenge = url.searchParams.get('hub.challenge') - // Handle WhatsApp verification if applicable const whatsAppResponse = await handleWhatsAppVerification(requestId, path, mode, token, challenge) if (whatsAppResponse) { return whatsAppResponse } - // --- General Webhook Verification --- + // --- General Verification --- logger.debug(`[${requestId}] Looking for webhook with path: ${path}`) - // Find the webhook in the database + // Query the database for an active webhook matching the provided path const webhooks = await db - .select({ - webhook: webhook, - }) + .select({ webhook }) .from(webhook) .where(and(eq(webhook.path, path), eq(webhook.isActive, true))) .limit(1) @@ -67,171 +60,138 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{ return new NextResponse('Webhook not found', { status: 404 }) } - // For all other providers, confirm the webhook endpoint exists logger.info(`[${requestId}] Webhook verification successful for path: ${path}`) return new NextResponse('OK', { status: 200 }) } catch (error: any) { logger.error(`[${requestId}] Error processing webhook verification`, error) - return new NextResponse(`Internal Server Error: ${error.message}`, { - status: 500, - }) + return new NextResponse(`Internal Server Error: ${error.message}`, { status: 500 }) } } /** - * Webhook Payload Handler (POST) - * - * Processes incoming webhook payloads from all supported providers: - * - Validates and parses the request body - * - Performs provider-specific deduplication - * - Acquires distributed processing lock - * - Executes the associated workflow - * - * Performance optimizations: - * - Fast response time (2.5s timeout) to acknowledge receipt - * - Background processing for long-running operations - * - Robust deduplication to prevent duplicate executions - * - * @param request The incoming HTTP request with webhook payload - * @param params Route parameters containing the webhook path - * @returns HTTP response (may respond before processing completes) + * POST handler for processing incoming webhook payloads. + * + * This function handles: + * - Parsing and validation of the incoming request + * - Provider-specific deduplication (e.g. Slack and WhatsApp) + * - Immediate fast response via a timeout while processing continues in the background + * - Looking up the webhook & workflow in the database + * - For Airtable webhooks: + * • Augmenting the logic by ensuring that an Airtable subscription is set up + * • Deduplicating notifications (if a notificationId is provided) + * • Starting asynchronous payload fetching via polling + * - For other providers, executing the workflow as usual. */ export async function POST( request: NextRequest, { params }: { params: Promise<{ path: string }> } -) { +): Promise { const requestId = crypto.randomUUID().slice(0, 8) let foundWorkflow: any = null let foundWebhook: any = null - + // --- PHASE 1: Request validation and parsing --- - - // Extract and validate the raw request body let rawBody: string | null = null try { const requestClone = request.clone() rawBody = await requestClone.text() logger.debug(`[${requestId}] Captured raw request body, length: ${rawBody.length}`) - if (!rawBody || rawBody.length === 0) { logger.warn(`[${requestId}] Rejecting request with empty body`) return new NextResponse('Empty request body', { status: 400 }) } } catch (bodyError) { - logger.error(`[${requestId}] Failed to read request body`, { - error: bodyError instanceof Error ? bodyError.message : String(bodyError), - }) + logger.error(`[${requestId}] Failed to read request body`, { error: bodyError instanceof Error ? bodyError.message : String(bodyError) }) return new NextResponse('Failed to read request body', { status: 400 }) } - - // Parse the body as JSON + let body: any try { body = JSON.parse(rawBody) - if (Object.keys(body).length === 0) { logger.warn(`[${requestId}] Rejecting empty JSON object`) return new NextResponse('Empty JSON payload', { status: 400 }) } } catch (parseError) { - logger.error(`[${requestId}] Failed to parse JSON body`, { - error: parseError instanceof Error ? parseError.message : String(parseError), - }) + logger.error(`[${requestId}] Failed to parse JSON body`, { error: parseError instanceof Error ? parseError.message : String(parseError) }) return new NextResponse('Invalid JSON payload', { status: 400 }) } - + // --- PHASE 2: Early Slack deduplication --- - - // Handle Slack-specific message deduplication to prevent duplicates const messageId = body?.event_id const slackRetryNum = request.headers.get('x-slack-retry-num') const slackRetryReason = request.headers.get('x-slack-retry-reason') - + if (body?.type === 'event_callback') { - logger.debug(`[${requestId}] Slack event received with event_id: ${messageId || 'missing'}, retry: ${slackRetryNum || 'none'}`) - - // Create a robust deduplication key (works even if messageId is missing) - const dedupeKey = messageId ? - `slack:msg:${messageId}` : - `slack:${body?.team_id || ''}:${body?.event?.ts || body?.event?.event_ts || Date.now()}` - + logger.debug( + `[${requestId}] Slack event received with event_id: ${messageId || 'missing'}, retry: ${slackRetryNum || 'none'}` + ) + const dedupeKey = messageId + ? `slack:msg:${messageId}` + : `slack:${body?.team_id || ''}:${body?.event?.ts || body?.event?.event_ts || Date.now()}` + try { - // Check if this message was already processed const isDuplicate = await hasProcessedMessage(dedupeKey) if (isDuplicate) { logger.info(`[${requestId}] Duplicate Slack message detected: ${dedupeKey}, retry: ${slackRetryNum || 'none'}`) return new NextResponse('Duplicate message', { status: 200 }) } - - // Mark as processed immediately to prevent race conditions await markMessageAsProcessed(dedupeKey, 60 * 60 * 24) // 24 hour TTL logger.debug(`[${requestId}] Marked Slack message as processed with key: ${dedupeKey}`) - - // Log retry information if present if (slackRetryNum) { logger.info(`[${requestId}] Processing Slack retry #${slackRetryNum} for message, reason: ${slackRetryReason || 'unknown'}`) } } catch (error) { logger.error(`[${requestId}] Error in Slack deduplication`, error) - // Continue processing - better to risk a duplicate than fail to process } } - - // --- PHASE 3: Set up fast response timeout and distributed processing --- - - // Ensure we respond quickly to the webhook provider - // This prevents them from retrying the webhook and creating duplicates + + // --- PHASE 3: Fast response timeout and distributed lock --- const timeoutPromise = new Promise((resolve) => { setTimeout(() => { logger.warn(`[${requestId}] Request processing timeout, sending acknowledgment`) resolve(new NextResponse('Request received', { status: 200 })) - }, 2500) // 2.5 second timeout for fast response + }, 2500) // 2.5 second timeout }) - // Set up distributed processing lock to prevent duplicate processing let hasExecutionLock = false - - // Create a provider-specific lock key + // Build a provider-specific lock key let executionLockKey: string if (body?.type === 'event_callback') { - // For Slack events, use the same scheme as deduplication - executionLockKey = messageId ? - `execution:lock:slack:${messageId}` : - `execution:lock:slack:${body?.team_id || ''}:${body?.event?.ts || body?.event?.event_ts || Date.now()}` + executionLockKey = messageId + ? `execution:lock:slack:${messageId}` + : `execution:lock:slack:${body?.team_id || ''}:${body?.event?.ts || body?.event?.event_ts || Date.now()}` } else { - // Default fallback for other providers executionLockKey = `execution:lock:${requestId}:${crypto.randomUUID()}` } try { - // Attempt to acquire a distributed processing lock hasExecutionLock = await acquireLock(executionLockKey, requestId, 30) // 30 second TTL logger.debug(`[${requestId}] Execution lock acquisition ${hasExecutionLock ? 'successful' : 'failed'} for key: ${executionLockKey}`) } catch (lockError) { logger.error(`[${requestId}] Error acquiring execution lock`, lockError) - // Proceed without lock in case of Redis failure (fallback to best-effort) } // --- PHASE 4: Main processing logic --- const processingPromise = (async () => { try { - const path = (await params).path + const { path } = await params logger.info(`[${requestId}] Processing webhook request for path: ${path}`) - - // Handle Slack URL verification challenge (special fast path) + + // Handle Slack URL verification challenge (fast path) const slackChallengeResponse = handleSlackChallenge(body) if (slackChallengeResponse) { logger.info(`[${requestId}] Responding to Slack URL verification challenge`) return slackChallengeResponse } - - // Skip processing if another instance is already handling this request + + // If another instance is processing this request, skip duplicate processing. if (!hasExecutionLock) { logger.info(`[${requestId}] Skipping execution as lock was not acquired. Another instance is processing this request.`) return new NextResponse('Request is being processed by another instance', { status: 200 }) } - - // Look up the webhook and its associated workflow + + // --- PHASE 5: Look up the webhook & workflow --- const webhooks = await db .select({ webhook: webhook, @@ -249,10 +209,27 @@ export async function POST( foundWebhook = webhooks[0].webhook foundWorkflow = webhooks[0].workflow - - // --- PHASE 5: Provider-specific processing --- - - // --- Airtable Processing (Polling-based) --- + + // --- Augment Airtable logic: Ensure subscription exists --- + if (foundWebhook.provider === 'airtable') { + const providerConfig = foundWebhook.providerConfig || {} + if (!providerConfig.externalId) { + logger.info(`[${requestId}] Missing externalId for Airtable webhook ${foundWebhook.id}. Attempting to create subscription...`) + await createAirtableWebhookSubscription(request, foundWorkflow.userId, foundWebhook, requestId) + // Optionally refresh the webhook record after subscription creation + const refreshed = await db + .select({ webhook }) + .from(webhook) + .where(eq(webhook.id, foundWebhook.id)) + .limit(1) + if (refreshed.length > 0) { + foundWebhook = refreshed[0].webhook + } + } + } + + // --- Provider-specific processing --- + // Airtable Processing (Polling-based) if (foundWebhook.provider === 'airtable') { logger.info(`[${requestId}] Airtable webhook ping received for webhook: ${foundWebhook.id}`) @@ -261,8 +238,6 @@ export async function POST( if (notificationId) { try { const processedKey = `airtable-webhook-${foundWebhook.id}-${notificationId}` - - // Check if this notification was already processed const alreadyProcessed = await db .select({ id: webhook.id }) .from(webhook) @@ -273,36 +248,23 @@ export async function POST( ) ) .limit(1) - if (alreadyProcessed.length > 0) { - logger.info( - `[${requestId}] Duplicate Airtable notification detected: ${notificationId}`, - { webhookId: foundWebhook.id } - ) + logger.info(`[${requestId}] Duplicate Airtable notification detected: ${notificationId}`, { webhookId: foundWebhook.id }) return new NextResponse('Notification already processed', { status: 200 }) } - - // Store notification ID to prevent duplicate processing + // Append notification key to provider config (capping history to last 100 notifications) const providerConfig = foundWebhook.providerConfig || {} const processedNotifications = providerConfig.processedNotifications || [] processedNotifications.push(processedKey) - - // Keep only the last 100 notifications to prevent unlimited growth const limitedNotifications = processedNotifications.slice(-100) - - // Update the webhook record await db .update(webhook) .set({ - providerConfig: { - ...providerConfig, - processedNotifications: limitedNotifications, - }, + providerConfig: { ...providerConfig, processedNotifications: limitedNotifications }, updatedAt: new Date(), }) .where(eq(webhook.id, foundWebhook.id)) } catch (error) { - // If deduplication fails, log and continue processing logger.warn(`[${requestId}] Airtable deduplication check failed, continuing with processing`, { error: error instanceof Error ? error.message : String(error), webhookId: foundWebhook.id, @@ -310,73 +272,162 @@ export async function POST( } } - // Process Airtable payloads synchronously - try { - logger.info(`[${requestId}] Starting synchronous Airtable payload processing...`, { + // Start asynchronous Airtable payload processing in the background + logger.info(`[${requestId}] Starting Airtable payload processing...`, { + webhookId: foundWebhook.id, + workflowId: foundWorkflow.id, + }) + fetchAndProcessAirtablePayloads(foundWebhook, foundWorkflow, requestId).catch((err: any) => { + logger.error(`[${requestId}] Error during Airtable processing`, { webhookId: foundWebhook.id, - workflowId: foundWorkflow.id, + error: err.message, + stack: err.stack, }) - - // This function fetches all changes from Airtable API and triggers workflow execution - await fetchAndProcessAirtablePayloads( - foundWebhook, - foundWorkflow, - requestId // Pass the original request ID for consistent logging - ) - - logger.info(`[${requestId}] Synchronous Airtable payload processing finished.`, { - webhookId: foundWebhook.id, - }) - - return new NextResponse('Airtable ping processed successfully', { status: 200 }) - } catch (error: any) { - logger.error(`[${requestId}] Error during synchronous Airtable processing`, { - webhookId: foundWebhook.id, - error: error.message, - stack: error.stack, - }) - return new NextResponse(`Error processing Airtable webhook: ${error.message}`, { - status: 500, - }) - } + }) + + // Immediately return acknowledgment without waiting for background processing + return new NextResponse('Airtable ping acknowledged, processing started', { status: 200 }) } - - // --- WhatsApp & Other Providers Processing --- - - // Generate a unique execution ID for this webhook trigger + + // --- WhatsApp & other providers --- const executionId = uuidv4() - - // WhatsApp-specific deduplication if (foundWebhook.provider === 'whatsapp') { const data = body?.entry?.[0]?.changes?.[0]?.value const messages = data?.messages || [] - const whatsappDuplicateResponse = await processWhatsAppDeduplication(requestId, messages) if (whatsappDuplicateResponse) { return whatsappDuplicateResponse } - } - // Generic deduplication for other providers (excluding Slack which was handled earlier) - else if (foundWebhook.provider !== 'slack') { + } else if (foundWebhook.provider !== 'slack') { const genericDuplicateResponse = await processGenericDeduplication(requestId, path, body) if (genericDuplicateResponse) { return genericDuplicateResponse } } - - // --- Execute workflow for the webhook event --- + + // --- Execute the workflow for the webhook event --- logger.info(`[${requestId}] Executing workflow for ${foundWebhook.provider} webhook`) - - // Process the webhook and return the response - // This function handles formatting input, executing the workflow, and persisting results return await processWebhook(foundWebhook, foundWorkflow, body, request, executionId, requestId) - } catch (error: any) { logger.error(`[${requestId}] Error processing webhook:`, error) return new NextResponse(`Internal server error: ${error.message}`, { status: 500 }) } })() - - // Race the processing against the timeout to ensure fast response + + // Race processingPromise against the fast timeout response. return Promise.race([timeoutPromise, processingPromise]) } + +/** + * Helper function to create a subscription with Airtable. + * + * This logic is taken from the legacy route and attempts to register the webhook with Airtable. + * On success, it stores the externalId (Airtable webhook ID) in the provider configuration. + * + * @param request The original NextRequest. + * @param userId The user ID associated with the workflow. + * @param webhookData The webhook record from the database. + * @param requestId A short unique request ID for logging. + */ +async function createAirtableWebhookSubscription( + request: NextRequest, + userId: string, + webhookData: any, + requestId: string +) { + try { + const { path, providerConfig } = webhookData + const { baseId, tableId, includeCellValuesInFieldIds } = providerConfig || {} + + if (!baseId || !tableId) { + logger.warn(`[${requestId}] Missing baseId or tableId for Airtable webhook creation.`, { + webhookId: webhookData.id, + }) + return // Cannot proceed without essential IDs + } + + const accessToken = await getOAuthToken(userId, 'airtable') + if (!accessToken) { + logger.warn(`[${requestId}] Could not retrieve Airtable access token for user ${userId}. Cannot create webhook in Airtable.`) + return + } + + const requestOrigin = new URL(request.url).origin + // Remap localhost origins if necessary + const effectiveOrigin = requestOrigin.includes('localhost') + ? process.env.NEXT_PUBLIC_APP_URL || requestOrigin + : requestOrigin + + const notificationUrl = `${effectiveOrigin}/api/webhooks/trigger/${path}` + if (effectiveOrigin !== requestOrigin) { + logger.debug(`[${requestId}] Remapped localhost origin to ${effectiveOrigin} for notificationUrl`) + } + + const airtableApiUrl = `https://api.airtable.com/v0/bases/${baseId}/webhooks` + + const specification: any = { + options: { + filters: { + dataTypes: ['tableData'], + recordChangeScope: tableId, + }, + }, + } + + if (includeCellValuesInFieldIds === 'all') { + specification.options.includes = { + includeCellValuesInFieldIds: 'all', + } + } + + const requestBody: any = { + notificationUrl, + specification, + } + + const airtableResponse = await fetch(airtableApiUrl, { + method: 'POST', + headers: { + Authorization: `Bearer ${accessToken}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify(requestBody), + }) + + const responseBody = await airtableResponse.json() + + if (!airtableResponse.ok || responseBody.error) { + const errorMessage = + responseBody.error?.message || responseBody.error || 'Unknown Airtable API error' + const errorType = responseBody.error?.type + logger.error( + `[${requestId}] Failed to create webhook in Airtable for webhook ${webhookData.id}. Status: ${airtableResponse.status}`, + { type: errorType, message: errorMessage, response: responseBody } + ) + } else { + logger.info( + `[${requestId}] Successfully created webhook in Airtable for webhook ${webhookData.id}.`, + { airtableWebhookId: responseBody.id } + ) + // Store the externalId in the provider configuration + try { + const currentConfig = (webhookData.providerConfig as Record) || {} + const updatedConfig = { ...currentConfig, externalId: responseBody.id } + await db + .update(webhook) + .set({ providerConfig: updatedConfig, updatedAt: new Date() }) + .where(eq(webhook.id, webhookData.id)) + } catch (dbError: any) { + logger.error( + `[${requestId}] Failed to store externalId in providerConfig for webhook ${webhookData.id}.`, + dbError + ) + } + } + } catch (error: any) { + logger.error( + `[${requestId}] Exception during Airtable webhook creation for webhook ${webhookData.id}.`, + { message: error.message, stack: error.stack } + ) + } +}