From ca4055130f524c2d89dbd7e8d0ce80b317e6275f Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 6 Apr 2025 22:24:09 -0700 Subject: [PATCH] fix(airtable): add logs to debug airtable webhook failure in production only --- sim/app/api/webhooks/trigger/[path]/route.ts | 443 ++++++++++--------- 1 file changed, 245 insertions(+), 198 deletions(-) diff --git a/sim/app/api/webhooks/trigger/[path]/route.ts b/sim/app/api/webhooks/trigger/[path]/route.ts index 26b101ad85..c47f7da168 100644 --- a/sim/app/api/webhooks/trigger/[path]/route.ts +++ b/sim/app/api/webhooks/trigger/[path]/route.ts @@ -4,19 +4,12 @@ import { v4 as uuidv4 } from 'uuid' import { createLogger } from '@/lib/logs/console-logger' import { persistExecutionError, persistExecutionLogs } from '@/lib/logs/execution-logger' import { buildTraceSpans } from '@/lib/logs/trace-spans' -import { - acquireLock, - closeRedisConnection, - getLockValue, - hasProcessedMessage, - markMessageAsProcessed, - releaseLock, -} from '@/lib/redis' import { decryptSecret } from '@/lib/utils' import { updateWorkflowRunCounts } from '@/lib/workflows/utils' import { mergeSubblockStateAsync } from '@/stores/workflows/utils' +import { getOAuthToken } from '@/app/api/auth/oauth/utils' import { db } from '@/db' -import { account, environment, userStats, webhook, workflow } from '@/db/schema' +import { environment, userStats, webhook, workflow } from '@/db/schema' import { Executor } from '@/executor' import { Serializer } from '@/serializer' import { validateSlackSignature } from '../../utils' @@ -26,29 +19,6 @@ const logger = createLogger('WebhookTriggerAPI') export const dynamic = 'force-dynamic' export const maxDuration = 300 // 5 minutes max execution time for long-running webhooks -// --- Placeholder Function to Get OAuth Token (copied from webhooks/route.ts) --- -// NOTE: Replace this with your actual implementation for fetching stored OAuth tokens -async function getOAuthToken(userId: string, providerId: string): Promise { - logger.debug(`Attempting to fetch OAuth token for user ${userId}, provider ${providerId}`) - const connections = await db - .select({ accessToken: account.accessToken }) - .from(account) - .where(and(eq(account.userId, userId), eq(account.providerId, providerId))) - .orderBy(account.createdAt) // Consider filtering by expiration or using a specific 'active' flag - .limit(1) - - if (connections.length > 0 && connections[0].accessToken) { - logger.info(`Found OAuth token for user ${userId}, provider ${providerId}`) - // TODO: Add token decryption if necessary - // TODO: Add token refresh logic if necessary - return connections[0].accessToken - } - - logger.warn(`No valid OAuth token found for user ${userId}, provider ${providerId}`) - return null -} -// --- End Placeholder --- - /** * Consolidated webhook trigger endpoint for all providers * Handles both WhatsApp verification and other webhook providers @@ -131,9 +101,6 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{ return new NextResponse(`Internal Server Error: ${error.message}`, { status: 500, }) - } finally { - // Ensure Redis connection is properly closed in serverless environment - await closeRedisConnection() } } @@ -182,70 +149,65 @@ export async function POST( // --- Provider Specific Logic --- - // --- Airtable Ping Handling with Debounce --- + // --- Airtable Ping Handling (Simplified, no locking) --- if (foundWebhook.provider === 'airtable') { logger.info(`[${requestId}] Airtable webhook ping received for webhook: ${foundWebhook.id}`) - // Define lock key and expiry (e.g., 30 seconds) - const lockKey = `airtable_processing:${foundWebhook.id}` - const lockExpirySeconds = 30 + // Acknowledge the ping immediately - Now we process synchronously + // logger.debug(`[${requestId}] Acknowledging Airtable ping for webhook ${foundWebhook.id}.`) - // Attempt to acquire the lock - const lockAcquired = await acquireLock(lockKey, requestId, lockExpirySeconds) - - if (!lockAcquired) { - logger.info( - `[${requestId}] Processing already in progress for webhook ${foundWebhook.id}. Skipping this ping.` + // Process the ping SYNCHRONOUSLY + try { + logger.info(`[${requestId}] Starting synchronous Airtable payload processing...`, { + webhookId: foundWebhook.id, + workflowId: foundWorkflow.id, + }) + await fetchAndProcessAirtablePayloads( + foundWebhook, + foundWorkflow, + requestId // Pass the original request ID ) - return new NextResponse('Processing already in progress', { status: 200 }) // Acknowledge ping but don't process - } - - logger.info(`[${requestId}] Acquired processing lock for webhook ${foundWebhook.id}.`) - - // TODO: Implement Airtable Signature Verification if needed - // ... (Signature verification logic) ... - - // Acknowledge the ping immediately AFTER acquiring lock - logger.debug(`[${requestId}] Acknowledging Airtable ping for webhook ${foundWebhook.id}.`) - // Use request.waitUntil for Vercel edge functions to allow background tasks - const backgroundTaskPromise = fetchAndProcessAirtablePayloads( - foundWebhook, - foundWorkflow, - requestId, // Pass the original request ID - lockKey // Pass the lock key to release it later - ) - if (typeof (request as any).waitUntil === 'function') { - ;(request as any).waitUntil(backgroundTaskPromise) - logger.info(`[${requestId}] Scheduled Airtable payload fetch via waitUntil.`) - } else { - // For non-edge environments, run async but don't await the main handler - logger.info(`[${requestId}] Running Airtable payload fetch asynchronously (non-edge).`) - backgroundTaskPromise.catch((err) => { - logger.error(`[${requestId}] Uncaught error in background Airtable processing.`, err) - // Persist internal polling error - use original requestId as a prefix - persistExecutionError(foundWorkflow.id, `poll-${requestId}-uncaught`, err, 'webhook') - // Ensure lock is released even on uncaught error - releaseLock(lockKey).catch((releaseErr) => - logger.error(`Failed to release lock on uncaught error for ${lockKey}`, releaseErr) - ) + logger.info(`[${requestId}] Synchronous Airtable payload processing finished.`, { + webhookId: foundWebhook.id, + }) + // Return success after processing is complete + 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, + }) + // Persist the error if processing fails + await persistExecutionError( + foundWorkflow.id, + `airtable-sync-process-${requestId}`, + error, + 'webhook' + ) + return new NextResponse(`Error processing Airtable webhook: ${error.message}`, { + status: 500, }) } - // Return immediately after scheduling the background task - return new NextResponse('Airtable ping acknowledged, processing started', { status: 200 }) + // REMOVED Asynchronous handling logic: + // const backgroundTaskPromise = fetchAndProcessAirtablePayloads(...) + // if (typeof (request as any).waitUntil === 'function') { ... } + // else { ... } + // return new NextResponse('Airtable ping acknowledged, processing started', { status: 200 }) } // --- Existing Deduplication and Processing for other providers --- // Generate hash *only* for non-Airtable requests now - const requestHash = await generateRequestHash(path, body) - if (await hasProcessedMessage(requestHash)) { - logger.info( - `[${requestId}] Duplicate webhook request detected (non-Airtable) with hash: ${requestHash}` - ) - return new NextResponse('Duplicate request', { status: 200 }) - } + // const requestHash = await generateRequestHash(path, body) // Commented out or removed hash generation + // if (await hasProcessedMessage(requestHash)) { // Removed check + // logger.info( + // `[${requestId}] Duplicate webhook request detected (non-Airtable) with hash: ${requestHash}` + // ) + // return new NextResponse('Duplicate request', { status: 200 }) + // } // Mark as processed *only* for non-Airtable requests needing this body-based dedup - await markMessageAsProcessed(`req:${requestHash}`, 60 * 60 * 24) // Use prefix for clarity + // await markMessageAsProcessed(`req:${requestHash}`, 60 * 60 * 24) // Removed marking as processed // --- Slack Specific Handling --- if (foundWebhook.provider === 'slack') { @@ -285,14 +247,14 @@ export async function POST( } const messageId = body?.event?.event_id // Slack message ID // Use full prefix for clarity and avoid collision with general keys - const slackMsgKey = `slack:msg:${messageId}` - if (messageId && (await hasProcessedMessage(slackMsgKey))) { - logger.info(`[${requestId}] Duplicate Slack message detected with ID: ${messageId}`) - return new NextResponse('Duplicate message', { status: 200 }) - } - if (messageId) { - await markMessageAsProcessed(slackMsgKey) // Use specific key - } + // const slackMsgKey = `slack:msg:${messageId}` // Removed key generation + // if (messageId && (await hasProcessedMessage(slackMsgKey))) { // Removed check + // logger.info(`[${requestId}] Duplicate Slack message detected with ID: ${messageId}`) + // return new NextResponse('Duplicate message', { status: 200 }) + // } + // if (messageId) { + // await markMessageAsProcessed(slackMsgKey) // Removed marking as processed + // } // Process Slack webhook using the existing function return await processWebhook( foundWebhook, @@ -314,14 +276,14 @@ export async function POST( const message = messages[0] const messageId = message.id // WhatsApp message ID // Use full prefix for clarity - const whatsappMsgKey = `whatsapp:msg:${messageId}` - if (messageId && (await hasProcessedMessage(whatsappMsgKey))) { - logger.info(`[${requestId}] Duplicate WhatsApp message detected with ID: ${messageId}`) - return new NextResponse('Duplicate message', { status: 200 }) - } - if (messageId) { - await markMessageAsProcessed(whatsappMsgKey) // Use specific key - } + // const whatsappMsgKey = `whatsapp:msg:${messageId}` // Removed key generation + // if (messageId && (await hasProcessedMessage(whatsappMsgKey))) { // Removed check + // logger.info(`[${requestId}] Duplicate WhatsApp message detected with ID: ${messageId}`) + // return new NextResponse('Duplicate message', { status: 200 }) + // } + // if (messageId) { + // await markMessageAsProcessed(whatsappMsgKey) // Removed marking as processed + // } // Process WhatsApp webhook using the existing function const result = await processWebhook( foundWebhook, @@ -360,12 +322,6 @@ export async function POST( return new NextResponse(`Internal Server Error: ${error.message}`, { status: 500, }) - } finally { - // Close Redis connection ONLY if not processing Airtable in background - // If Airtable is handled, the background task will close it. - if (foundWebhook?.provider !== 'airtable') { - await closeRedisConnection() - } } } @@ -374,13 +330,12 @@ export async function POST( /** * Fetches all available payloads from Airtable API after a ping * and triggers a single workflow execution with the batch of payloads. - * Releases the processing lock upon completion or error. + * Simplified version without locking mechanism. */ async function fetchAndProcessAirtablePayloads( webhookData: any, workflowData: any, - requestId: string, // Original request ID from the ping, used for the final execution log & lock value - lockKey: string // Lock key to check and release + requestId: string // Original request ID from the ping, used for the final execution log ) { // Use a prefix derived from requestId for *internal* polling logs/errors const internalPollIdPrefix = `poll-${requestId}` @@ -392,6 +347,12 @@ async function fetchAndProcessAirtablePayloads( const consolidatedChangesMap = new Map() let localProviderConfig = { ...((webhookData.providerConfig as Record) || {}) } // Local copy + // Add log at the very beginning of the try block + logger.info(`[${requestId}] Entering fetchAndProcessAirtablePayloads try block`, { + webhookId: webhookData.id, + workflowId: workflowData.id, + }) + try { // --- Essential IDs & Config from localProviderConfig --- const baseId = localProviderConfig.baseId @@ -407,30 +368,92 @@ async function fetchAndProcessAirtablePayloads( new Error('Missing Airtable baseId or externalId in providerConfig'), 'webhook' ) - return // Exit early, lock will be released in finally block + return // Exit early } // --- Retrieve Stored Cursor from localProviderConfig --- const storedCursor = localProviderConfig.externalWebhookCursor + logger.info(`[${requestId}] Retrieved providerConfig for cursor check`, { + webhookId: webhookData.id, + storedCursorValue: storedCursor, + type: typeof storedCursor, + }) + + // IMPORTANT FIX: Initialize cursor in provider config if missing + if (storedCursor === undefined || storedCursor === null) { + logger.info(`[${requestId}] Cursor is missing in providerConfig, initializing it`, { + webhookId: webhookData.id, + }) + + // Update the local copy + localProviderConfig.externalWebhookCursor = null + + // Add cursor to the database immediately to fix the configuration + try { + // Add log before the DB update + logger.info(`[${requestId}] Attempting to initialize cursor in DB`, { + webhookId: webhookData.id, + configToSave: { ...localProviderConfig, externalWebhookCursor: null }, + }) + await db + .update(webhook) + .set({ + providerConfig: { ...localProviderConfig, externalWebhookCursor: null }, + updatedAt: new Date(), + }) + .where(eq(webhook.id, webhookData.id)) + + localProviderConfig.externalWebhookCursor = null // Update local copy too + logger.info(`[${requestId}] Successfully initialized cursor in DB`, { + webhookId: webhookData.id, + }) + } catch (initError: any) { + logger.error(`[${requestId}] Failed to initialize cursor in DB`, { + webhookId: webhookData.id, + error: initError.message, + stack: initError.stack, + }) + // Persist the error specifically for cursor initialization failure + await persistExecutionError( + workflowData.id, + `${internalPollIdPrefix}-cursor-init-error`, + initError, + 'webhook' + ) + } + } + if (storedCursor && typeof storedCursor === 'number') { currentCursor = storedCursor - logger.info( - `[${requestId}] Retrieved stored cursor from providerConfig for Airtable webhook ${webhookData.id}: ${currentCursor}` - ) + logger.info(`[${requestId}] Using stored cursor`, { + webhookId: webhookData.id, + cursor: currentCursor, + }) } else { - logger.info( - `[${requestId}] No valid stored cursor in providerConfig for Airtable webhook ${webhookData.id}, starting poll.` - ) + logger.info(`[${requestId}] No valid stored cursor found, starting poll`, { + webhookId: webhookData.id, + }) currentCursor = null // Airtable API defaults to 1 if omitted } // --- Get OAuth Token --- + logger.info(`[${requestId}] Attempting to get OAuth token`, { + userId: workflowData.userId, + provider: 'airtable', + }) let accessToken: string | null = null try { accessToken = await getOAuthToken(workflowData.userId, 'airtable') if (!accessToken) { + logger.error( + `[${requestId}] Failed to obtain valid Airtable access token. Cannot proceed.`, + { userId: workflowData.userId } + ) throw new Error('Airtable access token not found.') } + logger.info(`[${requestId}] Successfully obtained Airtable access token`, { + userId: workflowData.userId, + }) } catch (tokenError: any) { logger.error( `[${requestId}] Failed to get Airtable OAuth token for user ${workflowData.userId}`, @@ -442,7 +465,7 @@ async function fetchAndProcessAirtablePayloads( tokenError, 'webhook' ) - return // Exit early, lock will be released in finally block + return // Exit early } const airtableApiBase = 'https://api.airtable.com/v0' @@ -452,9 +475,10 @@ async function fetchAndProcessAirtablePayloads( apiCallCount++ // Safety break if (apiCallCount > 10) { - logger.warn( - `[${requestId}] Reached maximum polling limit (10 calls) for webhook ${webhookData.id}. Processing ${consolidatedChangesMap.size} consolidated changes.` - ) + logger.warn(`[${requestId}] Reached maximum polling limit (10 calls)`, { + webhookId: webhookData.id, + consolidatedCount: consolidatedChangesMap.size, + }) mightHaveMore = false break } @@ -483,8 +507,8 @@ async function fetchAndProcessAirtablePayloads( responseBody.error || `Airtable API error Status ${response.status}` logger.error( - `[${requestId}] Failed to fetch Airtable payloads (Call ${apiCallCount}) for webhook ${webhookData.id}. Status: ${response.status}`, - { error: errorMessage } + `[${requestId}] Airtable API request to /payloads failed (Call ${apiCallCount})`, + { webhookId: webhookData.id, status: response.status, error: errorMessage } ) await persistExecutionError( workflowData.id, @@ -498,7 +522,12 @@ async function fetchAndProcessAirtablePayloads( const receivedPayloads = responseBody.payloads || [] logger.info( - `[${requestId}] Received ${receivedPayloads.length} payloads from Airtable (Call ${apiCallCount}). mightHaveMore: ${responseBody.mightHaveMore}` + `[${requestId}] Received response from Airtable /payloads (Call ${apiCallCount})`, + { + payloadCount: receivedPayloads.length, + mightHaveMore: responseBody.mightHaveMore, + nextCursor: responseBody.cursor, + } ) // --- Process and Consolidate Changes --- @@ -580,22 +609,41 @@ async function fetchAndProcessAirtablePayloads( mightHaveMore = responseBody.mightHaveMore || false if (nextCursor && typeof nextCursor === 'number' && nextCursor !== currentCursor) { - logger.info( - `[${requestId}] Updating cursor in providerConfig for Airtable webhook ${webhookData.id} to ${nextCursor}` - ) + logger.info(`[${requestId}] Updating cursor for next potential iteration`, { + webhookId: webhookData.id, + previousCursor: currentCursor, + newCursor: nextCursor, + mightHaveMore, + }) currentCursor = nextCursor + // --- Add logging before and after DB update --- + const updatedConfig = { ...localProviderConfig, externalWebhookCursor: currentCursor } + logger.info(`[${requestId}] Attempting to persist new cursor to DB`, { + webhookId: webhookData.id, + cursor: currentCursor, + configToSave: updatedConfig, // Log the object being saved + }) try { - const updatedConfig = { ...localProviderConfig, externalWebhookCursor: currentCursor } + // Force a complete object update to ensure consistency in serverless env await db .update(webhook) - .set({ providerConfig: updatedConfig, updatedAt: new Date() }) + .set({ + providerConfig: updatedConfig, // Use full object + updatedAt: new Date(), + }) .where(eq(webhook.id, webhookData.id)) + localProviderConfig.externalWebhookCursor = currentCursor // Update local copy too + logger.info(`[${requestId}] Successfully persisted new cursor to DB`, { + webhookId: webhookData.id, + newCursor: currentCursor, + }) } catch (dbError: any) { - logger.error( - `[${requestId}] Failed to persist updated cursor ${currentCursor} in providerConfig for Airtable webhook ${webhookData.id}`, - dbError - ) + logger.error(`[${requestId}] Failed to persist Airtable cursor to DB`, { + webhookId: webhookData.id, + cursor: currentCursor, + error: dbError.message, + }) await persistExecutionError( workflowData.id, `${internalPollIdPrefix}-cursor-persist-error`, @@ -603,16 +651,21 @@ async function fetchAndProcessAirtablePayloads( 'webhook' ) mightHaveMore = false + throw new Error('Failed to save Airtable cursor, stopping processing.') // Re-throw to break loop clearly } } else if (!nextCursor || typeof nextCursor !== 'number') { - logger.warn( - `[${requestId}] Invalid or missing cursor from Airtable API (Call ${apiCallCount}). Stopping poll. Cursor: ${nextCursor}` - ) + logger.warn(`[${requestId}] Invalid or missing cursor received, stopping poll`, { + webhookId: webhookData.id, + apiCall: apiCallCount, + receivedCursor: nextCursor, + }) mightHaveMore = false } else { - logger.debug( - `[${requestId}] Cursor unchanged (${currentCursor}) after call ${apiCallCount}. mightHaveMore is ${mightHaveMore}.` + logger.info( + `[${requestId}] Cursor unchanged or no new cursor, ending payload fetch loop`, + { webhookId: webhookData.id, finalCursor: currentCursor, apiCall: apiCallCount } ) + mightHaveMore = false // Explicitly stop if cursor hasn't changed } } catch (fetchError: any) { logger.error( @@ -634,46 +687,48 @@ async function fetchAndProcessAirtablePayloads( // Convert map values to array for final processing const finalConsolidatedChanges = Array.from(consolidatedChangesMap.values()) - logger.info( - `[${requestId}] Finished polling Airtable for webhook ${webhookData.id}. Processed ${payloadsFetched} payloads into ${finalConsolidatedChanges.length} consolidated change entries.` - ) + logger.info(`[${requestId}] Finished polling Airtable`, { + webhookId: webhookData.id, + totalPayloadsFetched: payloadsFetched, + consolidatedCount: finalConsolidatedChanges.length, + finalCursor: currentCursor, + }) - // --- Verify Lock and Execute Workflow ONCE --- + // --- Execute Workflow if we have changes (simplified - no lock check) --- if (finalConsolidatedChanges.length > 0) { - // Check if our lock is still valid before executing - const currentLockValue = await getLockValue(lockKey) - if (currentLockValue === requestId) { - logger.info( - `[${requestId}] Lock verified. Triggering single workflow execution for ${finalConsolidatedChanges.length} consolidated Airtable changes.` - ) - try { - // Format the input for the executor using the consolidated changes - const input = { airtableChanges: finalConsolidatedChanges } // Use the consolidated array - // Execute using the original requestId as the executionId - await executeWorkflowFromPayload(workflowData, input, requestId, requestId) - } catch (executionError: any) { - // Errors logged within executeWorkflowFromPayload - logger.error( - `[${requestId}] Error during the final workflow execution after polling`, - executionError - ) + logger.info( + `[${requestId}] Triggering workflow execution with ${finalConsolidatedChanges.length} changes`, + { + webhookId: webhookData.id, } - } else { - logger.info( - `[${requestId}] Lock was overwritten or expired (current value: ${currentLockValue}). Aborting execution for stale data.` + ) + try { + // Format the input for the executor using the consolidated changes + const input = { airtableChanges: finalConsolidatedChanges } // Use the consolidated array + // Execute using the original requestId as the executionId + await executeWorkflowFromPayload(workflowData, input, requestId, requestId) + } catch (executionError: any) { + // Errors logged within executeWorkflowFromPayload + logger.error( + `[${requestId}] Error during workflow execution triggered by Airtable polling`, + executionError ) - // Do not execute the workflow as a newer ping likely started processing. } } else { - logger.info( - `[${requestId}] No new changes collected from Airtable for webhook ${webhookData.id}. Workflow not executed.` - ) + logger.info(`[${requestId}] No new changes collected from Airtable, workflow not executed`, { + webhookId: webhookData.id, + }) } } catch (error) { // Catch any unexpected errors during the setup/polling logic itself logger.error( - `[${requestId}] Unexpected error during Airtable payload processing for webhook ${webhookData.id}`, - error + `[${requestId}] Unexpected error during asynchronous Airtable payload processing task`, + { + webhookId: webhookData.id, + workflowId: workflowData.id, + error: (error as Error).message, + stack: (error as Error).stack, + } ) // Persist this higher-level error await persistExecutionError( @@ -682,21 +737,6 @@ async function fetchAndProcessAirtablePayloads( error as Error, 'webhook' ) - } finally { - // Ensure the lock is released regardless of success or failure - // Only release if WE still hold the lock (value matches our requestId) - // This prevents releasing a lock acquired by a newer process. - const finalLockValue = await getLockValue(lockKey) - if (finalLockValue === requestId) { - logger.info(`[${requestId}] Releasing processing lock for webhook ${webhookData.id}.`) - await releaseLock(lockKey) - } else { - logger.warn( - `[${requestId}] Did not release lock ${lockKey} as its value (${finalLockValue}) does not match our requestId.` - ) - } - // Ensure Redis connection is closed after background task is done - await closeRedisConnection() } } @@ -719,6 +759,12 @@ async function executeWorkflowFromPayload( executionId: string, // Unique ID for this specific execution requestId: string // Original request ID for logging context ): Promise { + // Add log at the beginning of this function for clarity + logger.info(`[${requestId}] Preparing to execute workflow`, { + workflowId: foundWorkflow.id, + executionId, + triggerSource: 'webhook-payload', + }) // Returns void as errors are handled internally try { // Get the workflow state @@ -829,9 +875,10 @@ async function executeWorkflowFromPayload( } } - logger.debug( - `[${requestId}] Starting workflow execution ${executionId} with ${Object.keys(processedBlockStates).length} blocks` - ) + logger.debug(`[${requestId}] Starting workflow execution`, { + executionId, + blockCount: Object.keys(processedBlockStates).length, + }) const executor = new Executor( serializedWorkflow, processedBlockStates, @@ -841,13 +888,11 @@ async function executeWorkflowFromPayload( ) const result = await executor.execute(foundWorkflow.id) - logger.info( - `[${requestId}] Successfully executed workflow ${foundWorkflow.id} (Execution: ${executionId})`, - { - success: result.success, - executionTime: result.metadata?.duration, - } - ) + logger.info(`[${requestId}] Workflow execution finished`, { + executionId, + success: result.success, + durationMs: result.metadata?.duration, + }) // Update counts and stats if successful if (result.success) { @@ -868,10 +913,12 @@ async function executeWorkflowFromPayload( // Persist logs for this execution using the standard 'webhook' trigger type await persistExecutionLogs(foundWorkflow.id, executionId, enrichedResult, 'webhook') } catch (error: any) { - logger.error( - `[${requestId}] Error executing workflow ${foundWorkflow.id} (Execution: ${executionId})`, - error - ) + logger.error(`[${requestId}] Error executing workflow`, { + workflowId: foundWorkflow.id, + executionId, + error: error.message, + stack: error.stack, + }) // Persist the error for this execution using the standard 'webhook' trigger type await persistExecutionError(foundWorkflow.id, executionId, error, 'webhook') // Re-throw the error so the caller (fetchAndProcessAirtablePayloads) knows it failed