diff --git a/.agents/skills/add-trigger/SKILL.md b/.agents/skills/add-trigger/SKILL.md index 33473112aa0..7f8a3c7f155 100644 --- a/.agents/skills/add-trigger/SKILL.md +++ b/.agents/skills/add-trigger/SKILL.md @@ -423,6 +423,7 @@ export const {service}PollingHandler: PollingProviderHandler = { - First poll seeds state and emits nothing (avoids flooding with existing data) - Use `pollingIdempotency.executeWithIdempotency(provider, key, callback)` for dedup - Use `processPolledWebhookEvent(webhookData, workflowData, payload, requestId)` to fire the workflow +- Log through `ctx.logger` (already bound with `webhookId` and `pollId`, which equals `ctx.requestId`). Never add `{ requestId }` to a log line; the poll route's request context owns that field and overwrites it - Use `updateWebhookProviderConfig(webhookId, partialConfig, logger)` for read-merge-write on state - Use the latest server-side timestamp from API responses (not wall clock) to avoid clock skew diff --git a/apps/sim/lib/webhooks/polling/gmail.ts b/apps/sim/lib/webhooks/polling/gmail.ts index f17e86dcba9..7f8b75b7c88 100644 --- a/apps/sim/lib/webhooks/polling/gmail.ts +++ b/apps/sim/lib/webhooks/polling/gmail.ts @@ -77,12 +77,7 @@ export const gmailPollingHandler: PollingProviderHandler = { const config = getProviderConfig(webhookData.providerConfig) const now = new Date() - const { emails, latestHistoryId } = await fetchNewEmails( - accessToken, - config, - requestId, - logger - ) + const { emails, latestHistoryId } = await fetchNewEmails(accessToken, config, logger) const pollState = { lastCheckedTimestamp: now.toISOString(), // Omit unless this poll held a cursor, so it can't clear one an overlapping poll stored. @@ -92,11 +87,11 @@ export const gmailPollingHandler: PollingProviderHandler = { if (!emails || !emails.length) { await updateWebhookProviderConfig(webhookId, pollState, logger) await markWebhookSuccess(webhookId, logger) - logger.info(`No new emails found for webhook ${webhookId}`, { requestId }) + logger.info(`No new emails found for webhook ${webhookId}`) return 'success' } - logger.info(`Found ${emails.length} new emails for webhook ${webhookId}`, { requestId }) + logger.info(`Found ${emails.length} new emails for webhook ${webhookId}`) const { processedCount, failedCount } = await processEmails( emails, @@ -112,35 +107,27 @@ export const gmailPollingHandler: PollingProviderHandler = { if (failedCount > 0 && processedCount === 0) { await markWebhookFailed(webhookId, logger) - logger.warn(`All ${failedCount} emails failed to process for webhook ${webhookId}`, { - requestId, - }) + logger.warn(`All ${failedCount} emails failed to process for webhook ${webhookId}`) return 'failure' } await markWebhookSuccess(webhookId, logger) logger.info( - `Successfully processed ${processedCount} emails for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}`, - { requestId } + `Successfully processed ${processedCount} emails for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}` ) return 'success' } catch (error) { if (error instanceof PollAdmissionRefusedError) { - return skipAdmissionRefusedPoll(logger, requestId, webhookId) + return skipAdmissionRefusedPoll(logger, webhookId) } - logger.error(`Error processing Gmail webhook ${webhookId}:`, error, { requestId }) + logger.error(`Error processing Gmail webhook ${webhookId}:`, error) await markWebhookFailed(webhookId, logger) return 'failure' } }, } -async function fetchNewEmails( - accessToken: string, - config: GmailWebhookConfig, - requestId: string, - logger: Logger -) { +async function fetchNewEmails(accessToken: string, config: GmailWebhookConfig, logger: Logger) { try { const useHistoryApi = !!config.historyId let emails: GmailEmail[] = [] @@ -166,15 +153,13 @@ async function fetchNewEmails( if (status === 404 || status === 400) { await historyResponse.body?.cancel().catch(() => {}) logger.info( - `Gmail rejected history cursor ${config.historyId} (${status}); resyncing through search`, - { requestId } + `Gmail rejected history cursor ${config.historyId} (${status}); resyncing through search` ) - return resyncAfterExpiredHistory(accessToken, config, requestId, logger) + return resyncAfterExpiredHistory(accessToken, config, logger) } const errorData = await historyResponse.json().catch(() => ({})) logger.warn('Gmail history API error:', { - requestId, status, statusText: historyResponse.statusText, error: errorData, @@ -186,8 +171,8 @@ async function fetchNewEmails( ) } - logger.info(`Falling back to search API after history API error ${status}`, { requestId }) - return searchEmails(accessToken, config, requestId, logger) + logger.info(`Falling back to search API after history API error ${status}`) + return searchEmails(accessToken, config, logger) } const historyData = await historyResponse.json() @@ -215,14 +200,14 @@ async function fetchNewEmails( const sortedIds = [...messageIds].sort().reverse() const idsToFetch = sortedIds.slice(0, config.maxEmailsPerPoll || 25) - logger.info(`Processing ${idsToFetch.length} emails from history API`, { requestId }) + logger.info(`Processing ${idsToFetch.length} emails from history API`) const emailResults = await Promise.allSettled( idsToFetch.map((messageId) => getEmailDetails(accessToken, messageId)) ) const rejected = emailResults.filter((r) => r.status === 'rejected') if (rejected.length > 0) { - logger.warn(`Failed to fetch ${rejected.length} email details`, { requestId }) + logger.warn(`Failed to fetch ${rejected.length} email details`) } emails = emailResults .filter( @@ -232,13 +217,13 @@ async function fetchNewEmails( emails = filterEmailsByLabels(emails, config) } else { - return searchEmails(accessToken, config, requestId, logger) + return searchEmails(accessToken, config, logger) } return { emails, latestHistoryId } } catch (error) { const errorMessage = getErrorMessage(error, 'Unknown error') - logger.error('Error fetching new emails:', errorMessage, { requestId }) + logger.error('Error fetching new emails:', errorMessage) throw error } } @@ -283,12 +268,7 @@ function buildGmailSearchQuery(config: { return baseQuery } -async function searchEmails( - accessToken: string, - config: GmailWebhookConfig, - requestId: string, - logger: Logger -) { +async function searchEmails(accessToken: string, config: GmailWebhookConfig, logger: Logger) { try { const baseQuery = buildGmailSearchQuery(config) let timeConstraint = '' @@ -324,7 +304,6 @@ async function searchEmails( if (!searchResponse.ok) { const errorData = await searchResponse.json() logger.error('Gmail search API error:', { - requestId, status: searchResponse.status, statusText: searchResponse.statusText, query, @@ -338,7 +317,7 @@ async function searchEmails( const searchData = await searchResponse.json() if (!searchData.messages || !searchData.messages.length) { - logger.info(`No emails found matching query: ${query}`, { requestId }) + logger.info(`No emails found matching query: ${query}`) return { emails: [], latestHistoryId: config.historyId } } @@ -346,8 +325,7 @@ async function searchEmails( let latestHistoryId = config.historyId logger.info( - `Processing ${idsToFetch.length} emails from search API (total matches: ${searchData.messages.length})`, - { requestId } + `Processing ${idsToFetch.length} emails from search API (total matches: ${searchData.messages.length})` ) const emailResults = await Promise.allSettled( @@ -355,7 +333,7 @@ async function searchEmails( ) const rejected = emailResults.filter((r) => r.status === 'rejected') if (rejected.length > 0) { - logger.warn(`Failed to fetch ${rejected.length} email details`, { requestId }) + logger.warn(`Failed to fetch ${rejected.length} email details`) } const emails = emailResults .filter( @@ -370,7 +348,7 @@ async function searchEmails( return { emails, latestHistoryId } } catch (error) { const errorMessage = getErrorMessage(error, 'Unknown error') - logger.error('Error searching emails:', errorMessage, { requestId }) + logger.error('Error searching emails:', errorMessage) throw error } } @@ -383,24 +361,22 @@ async function searchEmails( async function resyncAfterExpiredHistory( accessToken: string, config: GmailWebhookConfig, - requestId: string, logger: Logger ) { - const searchResult = await searchEmails(accessToken, config, requestId, logger) + const searchResult = await searchEmails(accessToken, config, logger) const candidate = searchResult.latestHistoryId !== config.historyId ? searchResult.latestHistoryId - : await getGmailProfileHistoryId(accessToken, requestId, logger) + : await getGmailProfileHistoryId(accessToken, logger) const latestHistoryId = candidate && candidate !== config.historyId ? candidate : undefined if (!latestHistoryId) { - logger.info('No replacement Gmail history cursor yet; polling through search', { requestId }) + logger.info('No replacement Gmail history cursor yet; polling through search') } return { emails: searchResult.emails, latestHistoryId } } async function getGmailProfileHistoryId( accessToken: string, - requestId: string, logger: Logger ): Promise { try { @@ -408,15 +384,13 @@ async function getGmailProfileHistoryId( headers: { Authorization: `Bearer ${accessToken}` }, }) if (!response.ok) { - logger.warn(`Failed to fetch Gmail profile for fresh historyId: ${response.status}`, { - requestId, - }) + logger.warn(`Failed to fetch Gmail profile for fresh historyId: ${response.status}`) return null } const profile = await response.json() return (profile.historyId as string | undefined) ?? null } catch (error) { - logger.warn('Error fetching Gmail profile:', error, { requestId }) + logger.warn('Error fetching Gmail profile:', error) return null } } @@ -524,9 +498,7 @@ async function processEmails( const attachmentInfo = extractAttachmentInfo(payload) attachments = await downloadAttachments(email.id, attachmentInfo, accessToken) } catch (error) { - logger.error(`Error downloading attachments for email ${email.id}:`, error, { - requestId, - }) + logger.error(`Error downloading attachments for email ${email.id}:`, error) } } @@ -563,8 +535,7 @@ async function processEmails( logger.error( `Failed to process webhook for email ${email.id}:`, result.statusCode, - result.error, - { requestId } + result.error ) throw new Error(`Webhook processing failed (${result.statusCode}): ${result.error}`) } @@ -577,14 +548,12 @@ async function processEmails( } ) - logger.info(`Successfully processed email ${email.id} for webhook ${webhookData.id}`, { - requestId, - }) + logger.info(`Successfully processed email ${email.id} for webhook ${webhookData.id}`) processedCount++ } catch (error) { if (error instanceof PollAdmissionRefusedError && processedCount === 0) throw error const errorMessage = getErrorMessage(error, 'Unknown error') - logger.error(`Error processing email ${email.id}:`, errorMessage, { requestId }) + logger.error(`Error processing email ${email.id}:`, errorMessage) failedCount++ } } diff --git a/apps/sim/lib/webhooks/polling/google-calendar.ts b/apps/sim/lib/webhooks/polling/google-calendar.ts index ec6cf8beae4..72723f26cb4 100644 --- a/apps/sim/lib/webhooks/polling/google-calendar.ts +++ b/apps/sim/lib/webhooks/polling/google-calendar.ts @@ -119,7 +119,7 @@ export const googleCalendarPollingHandler: PollingProviderHandler = { logger ) await markWebhookSuccess(webhookId, logger) - logger.info(`First poll for webhook ${webhookId}, seeded timestamp`, { requestId }) + logger.info(`First poll for webhook ${webhookId}, seeded timestamp`) return 'success' } @@ -128,11 +128,11 @@ export const googleCalendarPollingHandler: PollingProviderHandler = { if (!events.length) { await markWebhookSuccess(webhookId, logger) - logger.info(`No changed events for webhook ${webhookId}`, { requestId }) + logger.info(`No changed events for webhook ${webhookId}`) return 'success' } - logger.info(`Found ${events.length} changed events for webhook ${webhookId}`, { requestId }) + logger.info(`Found ${events.length} changed events for webhook ${webhookId}`) const { processedCount, failedCount, latestUpdated } = await processEvents( events, @@ -162,23 +162,20 @@ export const googleCalendarPollingHandler: PollingProviderHandler = { if (failedCount > 0 && processedCount === 0) { await markWebhookFailed(webhookId, logger) - logger.warn(`All ${failedCount} events failed to process for webhook ${webhookId}`, { - requestId, - }) + logger.warn(`All ${failedCount} events failed to process for webhook ${webhookId}`) return 'failure' } await markWebhookSuccess(webhookId, logger) logger.info( - `Successfully processed ${processedCount} events for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}`, - { requestId } + `Successfully processed ${processedCount} events for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}` ) return 'success' } catch (error) { if (error instanceof PollAdmissionRefusedError) { - return skipAdmissionRefusedPoll(logger, requestId, webhookId) + return skipAdmissionRefusedPoll(logger, webhookId) } - logger.error(`Error processing Google Calendar webhook ${webhookId}:`, error, { requestId }) + logger.error(`Error processing Google Calendar webhook ${webhookId}:`, error) await markWebhookFailed(webhookId, logger) return 'failure' } @@ -341,8 +338,7 @@ async function processEvents( logger.error( `Failed to process webhook for event ${event.id}:`, result.statusCode, - result.error, - { requestId } + result.error ) throw new Error(`Webhook processing failed (${result.statusCode}): ${result.error}`) } @@ -351,14 +347,12 @@ async function processEvents( } ) - logger.info(`Successfully processed event ${event.id} for webhook ${webhookData.id}`, { - requestId, - }) + logger.info(`Successfully processed event ${event.id} for webhook ${webhookData.id}`) processedCount++ } catch (error) { if (error instanceof PollAdmissionRefusedError && processedCount === 0) throw error const errorMessage = getErrorMessage(error, 'Unknown error') - logger.error(`Error processing event ${event.id}:`, errorMessage, { requestId }) + logger.error(`Error processing event ${event.id}:`, errorMessage) failedCount++ } } diff --git a/apps/sim/lib/webhooks/polling/google-drive.ts b/apps/sim/lib/webhooks/polling/google-drive.ts index 67836766f35..cd98379d2f7 100644 --- a/apps/sim/lib/webhooks/polling/google-drive.ts +++ b/apps/sim/lib/webhooks/polling/google-drive.ts @@ -104,9 +104,7 @@ export const googleDrivePollingHandler: PollingProviderHandler = { logger ) await markWebhookSuccess(webhookId, logger) - logger.info(`First poll for webhook ${webhookId}, seeded pageToken: ${startPageToken}`, { - requestId, - }) + logger.info(`First poll for webhook ${webhookId}, seeded pageToken: ${startPageToken}`) return 'success' } @@ -115,7 +113,7 @@ export const googleDrivePollingHandler: PollingProviderHandler = { if (!changes.length) { await updateWebhookProviderConfig(webhookId, { pageToken: newStartPageToken }, logger) await markWebhookSuccess(webhookId, logger) - logger.info(`No changes found for webhook ${webhookId}`, { requestId }) + logger.info(`No changes found for webhook ${webhookId}`) return 'success' } @@ -125,15 +123,12 @@ export const googleDrivePollingHandler: PollingProviderHandler = { await updateWebhookProviderConfig(webhookId, { pageToken: newStartPageToken }, logger) await markWebhookSuccess(webhookId, logger) logger.info( - `${changes.length} changes found but none match filters for webhook ${webhookId}`, - { requestId } + `${changes.length} changes found but none match filters for webhook ${webhookId}` ) return 'success' } - logger.info(`Found ${filteredChanges.length} matching changes for webhook ${webhookId}`, { - requestId, - }) + logger.info(`Found ${filteredChanges.length} matching changes for webhook ${webhookId}`) const { processedCount, failedCount, newKnownFileIds } = await processChanges( filteredChanges, @@ -162,39 +157,33 @@ export const googleDrivePollingHandler: PollingProviderHandler = { if (failedCount > 0 && processedCount === 0) { await markWebhookFailed(webhookId, logger) - logger.warn(`All ${failedCount} changes failed to process for webhook ${webhookId}`, { - requestId, - }) + logger.warn(`All ${failedCount} changes failed to process for webhook ${webhookId}`) return 'failure' } await markWebhookSuccess(webhookId, logger) logger.info( - `Successfully processed ${processedCount} changes for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}`, - { requestId } + `Successfully processed ${processedCount} changes for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}` ) return 'success' } catch (error) { if (error instanceof PollAdmissionRefusedError) { - return skipAdmissionRefusedPoll(logger, requestId, webhookId) + return skipAdmissionRefusedPoll(logger, webhookId) } if (error instanceof Error && error.name === 'DrivePageTokenInvalidError') { await updateWebhookProviderConfig(webhookId, { pageToken: undefined }, logger) await markWebhookSuccess(webhookId, logger) - logger.warn(`Drive page token invalid for webhook ${webhookId}, re-seeding on next poll`, { - requestId, - }) + logger.warn(`Drive page token invalid for webhook ${webhookId}, re-seeding on next poll`) return 'success' } if (error instanceof Error && error.name === 'DriveRateLimitError') { await markWebhookSuccess(webhookId, logger) logger.warn( - `Drive API rate limited for webhook ${webhookId}, skipping to retry next poll cycle`, - { requestId } + `Drive API rate limited for webhook ${webhookId}, skipping to retry next poll cycle` ) return 'success' } - logger.error(`Error processing Google Drive webhook ${webhookId}:`, error, { requestId }) + logger.error(`Error processing Google Drive webhook ${webhookId}:`, error) await markWebhookFailed(webhookId, logger) return 'failure' } @@ -411,8 +400,7 @@ async function processChanges( logger.error( `Failed to process webhook for file ${change.fileId}:`, result.statusCode, - result.error, - { requestId } + result.error ) throw new Error(`Webhook processing failed (${result.statusCode}): ${result.error}`) } @@ -421,16 +409,13 @@ async function processChanges( }) logger.info( - `Successfully processed change for file ${change.fileId} for webhook ${webhookData.id}`, - { requestId } + `Successfully processed change for file ${change.fileId} for webhook ${webhookData.id}` ) processedCount++ } catch (error) { if (error instanceof PollAdmissionRefusedError && processedCount === 0) throw error const errorMessage = getErrorMessage(error, 'Unknown error') - logger.error(`Error processing change for file ${change.fileId}:`, errorMessage, { - requestId, - }) + logger.error(`Error processing change for file ${change.fileId}:`, errorMessage) failedCount++ } } diff --git a/apps/sim/lib/webhooks/polling/google-sheets.ts b/apps/sim/lib/webhooks/polling/google-sheets.ts index adbb7e7c649..111858e41a8 100644 --- a/apps/sim/lib/webhooks/polling/google-sheets.ts +++ b/apps/sim/lib/webhooks/polling/google-sheets.ts @@ -73,7 +73,7 @@ export const googleSheetsPollingHandler: PollingProviderHandler = { const now = new Date() if (!spreadsheetId || !sheetName) { - logger.error(`Missing spreadsheetId or sheetName for webhook ${webhookId}`, { requestId }) + logger.error(`Missing spreadsheetId or sheetName for webhook ${webhookId}`) await markWebhookFailed(webhookId, logger) return 'failure' } @@ -82,7 +82,6 @@ export const googleSheetsPollingHandler: PollingProviderHandler = { accessToken, spreadsheetId, config.lastModifiedTime, - requestId, logger ) @@ -93,7 +92,7 @@ export const googleSheetsPollingHandler: PollingProviderHandler = { logger ) await markWebhookSuccess(webhookId, logger) - logger.info(`Sheet not modified since last poll for webhook ${webhookId}`, { requestId }) + logger.info(`Sheet not modified since last poll for webhook ${webhookId}`) return 'success' } @@ -118,17 +117,14 @@ export const googleSheetsPollingHandler: PollingProviderHandler = { logger ) await markWebhookSuccess(webhookId, logger) - logger.info(`First poll for webhook ${webhookId}, seeded row index: ${currentRowCount}`, { - requestId, - }) + logger.info(`First poll for webhook ${webhookId}, seeded row index: ${currentRowCount}`) return 'success' } if (currentRowCount <= config.lastIndexChecked) { if (currentRowCount < config.lastIndexChecked) { logger.warn( - `Row count decreased from ${config.lastIndexChecked} to ${currentRowCount} for webhook ${webhookId}`, - { requestId } + `Row count decreased from ${config.lastIndexChecked} to ${currentRowCount} for webhook ${webhookId}` ) } await updateWebhookProviderConfig( @@ -141,7 +137,7 @@ export const googleSheetsPollingHandler: PollingProviderHandler = { logger ) await markWebhookSuccess(webhookId, logger) - logger.info(`No new rows for webhook ${webhookId}`, { requestId }) + logger.info(`No new rows for webhook ${webhookId}`) return 'success' } @@ -157,8 +153,7 @@ export const googleSheetsPollingHandler: PollingProviderHandler = { headerRowIndex > 0 ? Math.max(startRow, headerRowIndex + 1) : startRow logger.info( - `Found ${newRowCount} new rows for webhook ${webhookId}, processing rows ${adjustedStartRow}-${endRow}`, - { requestId } + `Found ${newRowCount} new rows for webhook ${webhookId}, processing rows ${adjustedStartRow}-${endRow}` ) // Entire batch is header/blank rows — advance pointer and skip fetch. @@ -177,8 +172,7 @@ export const googleSheetsPollingHandler: PollingProviderHandler = { ) await markWebhookSuccess(webhookId, logger) logger.info( - `Batch ${startRow}-${endRow} contained only header/blank rows for webhook ${webhookId}, advancing pointer`, - { requestId } + `Batch ${startRow}-${endRow} contained only header/blank rows for webhook ${webhookId}, advancing pointer` ) return 'success' } @@ -222,23 +216,20 @@ export const googleSheetsPollingHandler: PollingProviderHandler = { if (failedCount > 0 && processedCount === 0) { await markWebhookFailed(webhookId, logger) - logger.warn(`All ${failedCount} rows failed to process for webhook ${webhookId}`, { - requestId, - }) + logger.warn(`All ${failedCount} rows failed to process for webhook ${webhookId}`) return 'failure' } await markWebhookSuccess(webhookId, logger) logger.info( - `Successfully processed ${processedCount} rows for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}`, - { requestId } + `Successfully processed ${processedCount} rows for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}` ) return 'success' } catch (error) { if (error instanceof PollAdmissionRefusedError) { - return skipAdmissionRefusedPoll(logger, requestId, webhookId) + return skipAdmissionRefusedPoll(logger, webhookId) } - logger.error(`Error processing Google Sheets webhook ${webhookId}:`, error, { requestId }) + logger.error(`Error processing Google Sheets webhook ${webhookId}:`, error) await markWebhookFailed(webhookId, logger) return 'failure' } @@ -249,7 +240,6 @@ async function isDriveFileUnchanged( accessToken: string, spreadsheetId: string, lastModifiedTime: string | undefined, - requestId: string, logger: Logger ): Promise<{ unchanged: boolean; currentModifiedTime?: string }> { try { @@ -259,7 +249,7 @@ async function isDriveFileUnchanged( } return { unchanged: currentModifiedTime === lastModifiedTime, currentModifiedTime } } catch { - logger.warn('Drive modifiedTime check failed, proceeding with Sheets API', { requestId }) + logger.warn('Drive modifiedTime check failed, proceeding with Sheets API') return { unchanged: false } } } @@ -399,7 +389,7 @@ async function processRows( // Skip empty rows — don't fire a workflow run with no data. if (!row || row.length === 0) { - logger.info(`Skipping empty row ${rowNumber} for webhook ${webhookData.id}`, { requestId }) + logger.info(`Skipping empty row ${rowNumber} for webhook ${webhookData.id}`) continue } @@ -441,8 +431,7 @@ async function processRows( logger.error( `Failed to process webhook for row ${rowNumber}:`, result.statusCode, - result.error, - { requestId } + result.error ) throw new Error(`Webhook processing failed (${result.statusCode}): ${result.error}`) } @@ -451,14 +440,12 @@ async function processRows( } ) - logger.info(`Successfully processed row ${rowNumber} for webhook ${webhookData.id}`, { - requestId, - }) + logger.info(`Successfully processed row ${rowNumber} for webhook ${webhookData.id}`) processedCount++ } catch (error) { if (error instanceof PollAdmissionRefusedError && processedCount === 0) throw error const errorMessage = getErrorMessage(error, 'Unknown error') - logger.error(`Error processing row ${rowNumber}:`, errorMessage, { requestId }) + logger.error(`Error processing row ${rowNumber}:`, errorMessage) failedCount++ } } diff --git a/apps/sim/lib/webhooks/polling/hubspot.ts b/apps/sim/lib/webhooks/polling/hubspot.ts index 047d01831a0..4eb2c5fe303 100644 --- a/apps/sim/lib/webhooks/polling/hubspot.ts +++ b/apps/sim/lib/webhooks/polling/hubspot.ts @@ -210,9 +210,9 @@ export const hubspotPollingHandler: PollingProviderHandler = { return await pollSearchBased(ctx, config, accessToken) } catch (error) { if (error instanceof PollAdmissionRefusedError) { - return skipAdmissionRefusedPoll(logger, requestId, webhookId) + return skipAdmissionRefusedPoll(logger, webhookId) } - logger.error(`Error processing HubSpot webhook ${webhookId}:`, error, { requestId }) + logger.error(`Error processing HubSpot webhook ${webhookId}:`, error) await markWebhookFailed(webhookId, logger) return 'failure' } @@ -265,8 +265,7 @@ async function pollSearchBased( ) await markWebhookSuccess(webhookId, logger) logger.info( - `Seeded HubSpot webhook ${webhookId} watermark to ${nowMs} (${objectType}/${eventType}/${filterProperty})`, - { requestId } + `Seeded HubSpot webhook ${webhookId} watermark to ${nowMs} (${objectType}/${eventType}/${filterProperty})` ) return 'success' } @@ -278,7 +277,7 @@ async function pollSearchBased( ) } - const userFilters = buildUserFilters(config, logger, requestId) + const userFilters = buildUserFilters(config, logger) const properties = resolveRequestedProperties(config, objectType, filterProperty, userFilters) const maxRecords = Math.min( Math.max(config.maxRecordsPerPoll ?? DEFAULT_MAX_RECORDS, 1), @@ -295,7 +294,6 @@ async function pollSearchBased( properties, userFilters, maxRecords, - requestId, logger, }) @@ -306,13 +304,12 @@ async function pollSearchBased( logger ) await markWebhookSuccess(webhookId, logger) - logger.info(`No new HubSpot ${objectType} ${eventType} for webhook ${webhookId}`, { requestId }) + logger.info(`No new HubSpot ${objectType} ${eventType} for webhook ${webhookId}`) return 'success' } logger.info( - `Found ${records.length} HubSpot ${objectType} ${eventType} candidates for webhook ${webhookId}`, - { requestId } + `Found ${records.length} HubSpot ${objectType} ${eventType} candidates for webhook ${webhookId}` ) const targetProperty = config.targetPropertyName?.trim() || undefined @@ -350,16 +347,13 @@ async function pollSearchBased( if (failedCount > 0 && processedCount === 0) { await markWebhookFailed(webhookId, logger) - logger.warn(`All ${failedCount} HubSpot records failed to process for webhook ${webhookId}`, { - requestId, - }) + logger.warn(`All ${failedCount} HubSpot records failed to process for webhook ${webhookId}`) return 'failure' } await markWebhookSuccess(webhookId, logger) logger.info( - `Processed ${processedCount} HubSpot records${skippedCount ? `, skipped ${skippedCount} (no property change)` : ''}${failedCount ? `, ${failedCount} failed` : ''} for webhook ${webhookId}`, - { requestId } + `Processed ${processedCount} HubSpot records${skippedCount ? `, skipped ${skippedCount} (no property change)` : ''}${failedCount ? `, ${failedCount} failed` : ''} for webhook ${webhookId}` ) return 'success' } @@ -394,7 +388,6 @@ async function pollListMembership( accessToken, initialAfter: config.lastSeenMembershipCursor?.trim() || undefined, pageLimit: seedComplete ? maxRecords : MAX_PAGES_PER_POLL * HUBSPOT_PAGE_LIMIT, - requestId, logger, }) @@ -406,13 +399,10 @@ async function pollListMembership( } if (result.reachedEnd) { update.membershipSeedComplete = true - logger.info(`HubSpot list_membership ${webhookId} seed complete (list=${listId})`, { - requestId, - }) + logger.info(`HubSpot list_membership ${webhookId} seed complete (list=${listId})`) } else { logger.info( - `HubSpot list_membership ${webhookId} seed in progress (list=${listId}, scanned ${result.scanned})`, - { requestId } + `HubSpot list_membership ${webhookId} seed in progress (list=${listId}, scanned ${result.scanned})` ) } await updateWebhookProviderConfig(webhookId, update, logger) @@ -433,9 +423,7 @@ async function pollListMembership( return 'success' } - logger.info(`Found ${result.records.length} HubSpot list memberships for webhook ${webhookId}`, { - requestId, - }) + logger.info(`Found ${result.records.length} HubSpot list memberships for webhook ${webhookId}`) let processedCount = 0 let failedCount = 0 @@ -475,8 +463,7 @@ async function pollListMembership( failedCount++ logger.error( `Error processing HubSpot list membership ${member.recordId}:`, - getErrorMessage(error, 'Unknown error'), - { requestId } + getErrorMessage(error, 'Unknown error') ) } } @@ -551,11 +538,7 @@ function resolveRequestedProperties( return [...requested] } -export function buildUserFilters( - config: HubSpotWebhookConfig, - logger?: Logger, - requestId?: string -): FilterClause[] { +export function buildUserFilters(config: HubSpotWebhookConfig, logger?: Logger): FilterClause[] { const filters: FilterClause[] = [] if (config.pipelineId?.trim()) { @@ -591,7 +574,6 @@ export function buildUserFilters( } } catch (error) { logger?.warn('Could not parse user filters as JSON, ignoring:', { - requestId, error: getErrorMessage(error, 'parse error'), }) } @@ -603,7 +585,7 @@ export function buildUserFilters( // the webhook is marked failed and the user can trim their filter configuration. if (filters.length > MAX_USER_FILTERS) { throw new Error( - `[${requestId ?? ''}] HubSpot webhook has ${filters.length} combined filters (pipeline/stage/owner shortcuts + advanced filters), exceeding the ${MAX_USER_FILTERS}-filter limit HubSpot's Search API allows alongside the reserved cursor filters. Reduce the number of filters.` + `HubSpot webhook has ${filters.length} combined filters (pipeline/stage/owner shortcuts + advanced filters), exceeding the ${MAX_USER_FILTERS}-filter limit HubSpot's Search API allows alongside the reserved cursor filters. Reduce the number of filters.` ) } @@ -643,7 +625,6 @@ interface FetchArgs { properties: string[] userFilters: FilterClause[] maxRecords: number - requestId: string logger: Logger } @@ -657,7 +638,6 @@ async function fetchHubSpotChanges(args: FetchArgs): Promise '') logger.error('HubSpot search API error:', { - requestId, status: response.status, objectType, error: errorBody, @@ -725,8 +704,7 @@ async function fetchHubSpotChanges(args: FetchArgs): Promise= maxRecords) break if (pages >= MAX_PAGES_PER_POLL) { logger.warn( - `HubSpot poll hit MAX_PAGES_PER_POLL=${MAX_PAGES_PER_POLL} — remaining records will roll over to next poll`, - { requestId } + `HubSpot poll hit MAX_PAGES_PER_POLL=${MAX_PAGES_PER_POLL} — remaining records will roll over to next poll` ) break } @@ -866,8 +844,7 @@ async function processRecords( cursorFrozen = true logger.error( `Error processing HubSpot ${objectType} ${record.id}:`, - getErrorMessage(error, 'Unknown error'), - { requestId } + getErrorMessage(error, 'Unknown error') ) } } @@ -908,7 +885,6 @@ interface FetchListMembershipPagesArgs { initialAfter: string | undefined /** Soft cap on emitted records (normal mode) or pages × HUBSPOT_PAGE_LIMIT (seed mode). */ pageLimit: number - requestId: string logger: Logger } @@ -935,7 +911,7 @@ interface FetchListMembershipPagesResult { async function fetchListMembershipPages( args: FetchListMembershipPagesArgs ): Promise { - const { listId, accessToken, initialAfter, pageLimit, requestId, logger } = args + const { listId, accessToken, initialAfter, pageLimit, logger } = args const records: ListMembershipRecord[] = [] let after: string | undefined = initialAfter @@ -951,9 +927,7 @@ async function fetchListMembershipPages( if (!response.ok) { const errorText = await response.text().catch(() => '') - logger.error(`HubSpot list memberships fetch failed ${response.status}: ${errorText}`, { - requestId, - }) + logger.error(`HubSpot list memberships fetch failed ${response.status}: ${errorText}`) throw new Error( `HubSpot list memberships fetch ${response.status}: ${errorText.slice(0, 500)}` ) diff --git a/apps/sim/lib/webhooks/polling/imap.ts b/apps/sim/lib/webhooks/polling/imap.ts index 942f82475b9..993fb9db798 100644 --- a/apps/sim/lib/webhooks/polling/imap.ts +++ b/apps/sim/lib/webhooks/polling/imap.ts @@ -125,7 +125,7 @@ export const imapPollingHandler: PollingProviderHandler = { const config = getProviderConfig(webhookData.providerConfig) if (!config.host || !config.username || !config.password) { - logger.error(`Missing IMAP credentials for webhook ${webhookId}`, { requestId }) + logger.error(`Missing IMAP credentials for webhook ${webhookId}`) await markWebhookFailed(webhookId, logger) return 'failure' } @@ -140,7 +140,7 @@ export const imapPollingHandler: PollingProviderHandler = { try { await client.connect() - const result = await fetchNewEmails(client, config, requestId, logger) + const result = await fetchNewEmails(client, config, logger) emails = result.emails latestUidByMailbox = result.latestUidByMailbox uidValidityByMailbox = result.uidValidityByMailbox @@ -157,12 +157,12 @@ export const imapPollingHandler: PollingProviderHandler = { uidValidityByMailbox ) await markWebhookSuccess(webhookId, logger) - logger.info(`No new emails found for webhook ${webhookId}`, { requestId }) + logger.info(`No new emails found for webhook ${webhookId}`) await client.logout() return 'success' } - logger.info(`Found ${emails.length} new emails for webhook ${webhookId}`, { requestId }) + logger.info(`Found ${emails.length} new emails for webhook ${webhookId}`) const { processedCount, failedCount } = await processEmails( emails, @@ -187,16 +187,13 @@ export const imapPollingHandler: PollingProviderHandler = { if (failedCount > 0 && processedCount === 0) { await markWebhookFailed(webhookId, logger) - logger.warn(`All ${failedCount} emails failed to process for webhook ${webhookId}`, { - requestId, - }) + logger.warn(`All ${failedCount} emails failed to process for webhook ${webhookId}`) return 'failure' } await markWebhookSuccess(webhookId, logger) logger.info( - `Successfully processed ${processedCount} emails for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}`, - { requestId } + `Successfully processed ${processedCount} emails for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}` ) return 'success' } catch (innerError) { @@ -207,9 +204,9 @@ export const imapPollingHandler: PollingProviderHandler = { } } catch (error) { if (error instanceof PollAdmissionRefusedError) { - return skipAdmissionRefusedPoll(logger, requestId, webhookId) + return skipAdmissionRefusedPoll(logger, webhookId) } - logger.error(`Error processing IMAP webhook ${webhookId}`, { requestId }) + logger.error(`Error processing IMAP webhook ${webhookId}`) await markWebhookFailed(webhookId, logger) return 'failure' } @@ -263,12 +260,7 @@ async function updateImapState( ) } -async function fetchNewEmails( - client: ImapFlow, - config: ImapWebhookConfig, - requestId: string, - logger: Logger -) { +async function fetchNewEmails(client: ImapFlow, config: ImapWebhookConfig, logger: Logger) { const emails: Array<{ uid: number mailboxPath: string @@ -295,8 +287,7 @@ async function fetchNewEmails( if (storedUidValidity && storedUidValidity !== currentUidValidity) { logger.warn( - `UIDVALIDITY changed for ${mailboxPath} (${storedUidValidity} -> ${currentUidValidity}), discarding stored UID`, - { requestId } + `UIDVALIDITY changed for ${mailboxPath} (${storedUidValidity} -> ${currentUidValidity}), discarding stored UID` ) delete latestUidByMailbox[mailboxPath] } @@ -310,7 +301,7 @@ async function fetchNewEmails( try { searchCriteria = JSON.parse(config.searchCriteria) } catch { - logger.warn('Invalid search criteria JSON, using default', { requestId }) + logger.warn('Invalid search criteria JSON, using default') } } } @@ -362,7 +353,7 @@ async function fetchNewEmails( totalEmailsCollected++ } } catch { - logger.warn(`Error processing mailbox ${mailboxPath}`, { requestId }) + logger.warn(`Error processing mailbox ${mailboxPath}`) } } @@ -587,8 +578,7 @@ async function processEmails( logger.error( `Failed to process webhook for email ${email.uid}:`, result.statusCode, - result.error, - { requestId } + result.error ) throw new Error(`Webhook processing failed (${result.statusCode}): ${result.error}`) } @@ -605,7 +595,7 @@ async function processEmails( } await client.messageFlagsAdd(email.uid, ['\\Seen'], { uid: true }) } catch { - logger.warn(`Failed to mark message ${email.uid} as read`, { requestId }) + logger.warn(`Failed to mark message ${email.uid} as read`) } } @@ -614,13 +604,12 @@ async function processEmails( ) logger.info( - `Successfully processed email ${email.uid} from ${email.mailboxPath} for webhook ${webhookData.id}`, - { requestId } + `Successfully processed email ${email.uid} from ${email.mailboxPath} for webhook ${webhookData.id}` ) processedCount++ } catch (error) { if (error instanceof PollAdmissionRefusedError && processedCount === 0) throw error - logger.error(`Error processing email ${email.uid}`, { requestId }) + logger.error(`Error processing email ${email.uid}`) failedCount++ } } diff --git a/apps/sim/lib/webhooks/polling/orchestrator.ts b/apps/sim/lib/webhooks/polling/orchestrator.ts index fadfa6f21ef..af19b81ee52 100644 --- a/apps/sim/lib/webhooks/polling/orchestrator.ts +++ b/apps/sim/lib/webhooks/polling/orchestrator.ts @@ -47,12 +47,13 @@ export async function pollProvider(providerName: string): Promise { return 'skipped' } + const pollId = generateShortId() return withResourceOutboundScope({ workspaceId: entry.workflow.workspaceId }, () => handler.pollWebhook({ webhookData: entry.webhook, workflowData: entry.workflow, - requestId: generateShortId(), - logger: logger.withMetadata({ webhookId: entry.webhook.id }), + requestId: pollId, + logger: logger.withMetadata({ webhookId: entry.webhook.id, pollId }), }) ) }, diff --git a/apps/sim/lib/webhooks/polling/outlook.ts b/apps/sim/lib/webhooks/polling/outlook.ts index efa6a0b65a9..cb350ef9d1a 100644 --- a/apps/sim/lib/webhooks/polling/outlook.ts +++ b/apps/sim/lib/webhooks/polling/outlook.ts @@ -119,13 +119,13 @@ export const outlookPollingHandler: PollingProviderHandler = { const webhookId = webhookData.id try { - logger.info(`Processing Outlook webhook: ${webhookId}`, { requestId }) + logger.info(`Processing Outlook webhook: ${webhookId}`) const accessToken = await resolveOAuthCredential(webhookData, 'outlook', requestId) const config = getProviderConfig(webhookData.providerConfig) const now = new Date() - const { emails } = await fetchNewOutlookEmails(accessToken, config, requestId, logger) + const { emails } = await fetchNewOutlookEmails(accessToken, config, logger) if (!emails || !emails.length) { await updateWebhookProviderConfig( @@ -134,11 +134,11 @@ export const outlookPollingHandler: PollingProviderHandler = { logger ) await markWebhookSuccess(webhookId, logger) - logger.info(`No new emails found for webhook ${webhookId}`, { requestId }) + logger.info(`No new emails found for webhook ${webhookId}`) return 'success' } - logger.info(`Found ${emails.length} emails for webhook ${webhookId}`, { requestId }) + logger.info(`Found ${emails.length} emails for webhook ${webhookId}`) const { processedCount, failedCount } = await processOutlookEmails( emails, @@ -158,23 +158,20 @@ export const outlookPollingHandler: PollingProviderHandler = { if (failedCount > 0 && processedCount === 0) { await markWebhookFailed(webhookId, logger) - logger.warn(`All ${failedCount} emails failed to process for webhook ${webhookId}`, { - requestId, - }) + logger.warn(`All ${failedCount} emails failed to process for webhook ${webhookId}`) return 'failure' } await markWebhookSuccess(webhookId, logger) logger.info( - `Successfully processed ${processedCount} emails for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}`, - { requestId } + `Successfully processed ${processedCount} emails for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}` ) return 'success' } catch (error) { if (error instanceof PollAdmissionRefusedError) { - return skipAdmissionRefusedPoll(logger, requestId, webhookId) + return skipAdmissionRefusedPoll(logger, webhookId) } - logger.error(`Error processing Outlook webhook ${webhookId}:`, error, { requestId }) + logger.error(`Error processing Outlook webhook ${webhookId}:`, error) await markWebhookFailed(webhookId, logger) return 'failure' } @@ -190,7 +187,6 @@ const OUTLOOK_PAGE_SIZE = 50 async function fetchNewOutlookEmails( accessToken: string, config: OutlookWebhookConfig, - requestId: string, logger: Logger ) { try { @@ -212,7 +208,7 @@ async function fetchNewOutlookEmails( } const allEmails: OutlookEmail[] = [] let nextUrl: string | undefined = `${apiUrl}?${params.toString()}` - logger.info(`Fetching emails from: ${nextUrl}`, { requestId }) + logger.info(`Fetching emails from: ${nextUrl}`) while (nextUrl && allEmails.length < maxEmails) { const response = await fetchWithRetry(nextUrl, { @@ -226,7 +222,6 @@ async function fetchNewOutlookEmails( .json() .catch(() => ({ error: { message: 'Unknown error' } })) logger.error('Microsoft Graph API error:', { - requestId, status: response.status, statusText: response.statusText, error: errorData, @@ -247,7 +242,7 @@ async function fetchNewOutlookEmails( if (pageEmails.length === 0) break } - logger.info(`Fetched ${allEmails.length} emails total`, { requestId }) + logger.info(`Fetched ${allEmails.length} emails total`) const emails = allEmails @@ -256,16 +251,10 @@ async function fetchNewOutlookEmails( if (config.folderIds && config.folderIds.length > 0) { const wellKnownFolders = config.folderIds.filter(isWellKnownFolderName) if (wellKnownFolders.length > 0) { - resolvedFolderIds = await resolveWellKnownFolderIds( - accessToken, - config.folderIds, - requestId, - logger - ) + resolvedFolderIds = await resolveWellKnownFolderIds(accessToken, config.folderIds, logger) if (resolvedFolderIds.size < wellKnownFolders.length) { logger.warn( - `Could not resolve all well-known folders (${resolvedFolderIds.size}/${wellKnownFolders.length}) — skipping folder filter to avoid incorrect results`, - { requestId } + `Could not resolve all well-known folders (${resolvedFolderIds.size}/${wellKnownFolders.length}) — skipping folder filter to avoid incorrect results` ) skipFolderFilter = true } @@ -276,14 +265,12 @@ async function fetchNewOutlookEmails( ? emails : filterEmailsByFolder(emails, config, resolvedFolderIds) - logger.info(`Fetched ${emails.length} emails, ${filteredEmails.length} after filtering`, { - requestId, - }) + logger.info(`Fetched ${emails.length} emails, ${filteredEmails.length} after filtering`) return { emails: filteredEmails } } catch (error) { const errorMessage = getErrorMessage(error, 'Unknown error') - logger.error('Error fetching new Outlook emails:', errorMessage, { requestId }) + logger.error('Error fetching new Outlook emails:', errorMessage) throw error } } @@ -305,7 +292,6 @@ function isWellKnownFolderName(folderId: string): boolean { async function resolveWellKnownFolderId( accessToken: string, folderName: string, - requestId: string, logger: Logger ): Promise { try { @@ -319,16 +305,14 @@ async function resolveWellKnownFolderId( ) if (!response.ok) { - logger.warn(`Failed to resolve well-known folder '${folderName}': ${response.status}`, { - requestId, - }) + logger.warn(`Failed to resolve well-known folder '${folderName}': ${response.status}`) return null } const folder = await response.json() return folder.id || null } catch (error) { - logger.error(`Error resolving well-known folder '${folderName}':`, error, { requestId }) + logger.error(`Error resolving well-known folder '${folderName}':`, error) return null } } @@ -336,7 +320,6 @@ async function resolveWellKnownFolderId( async function resolveWellKnownFolderIds( accessToken: string, folderIds: string[], - requestId: string, logger: Logger ): Promise> { const resolvedIds = new Map() @@ -345,7 +328,7 @@ async function resolveWellKnownFolderIds( const resolutions = await Promise.all( wellKnownFolders.map(async (folderName) => { - const actualId = await resolveWellKnownFolderId(accessToken, folderName, requestId, logger) + const actualId = await resolveWellKnownFolderId(accessToken, folderName, logger) return { folderName, actualId } }) ) @@ -356,9 +339,7 @@ async function resolveWellKnownFolderIds( } } - logger.info(`Resolved ${resolvedIds.size}/${wellKnownFolders.length} well-known folders`, { - requestId, - }) + logger.info(`Resolved ${resolvedIds.size}/${wellKnownFolders.length} well-known folders`) return resolvedIds } @@ -407,16 +388,9 @@ async function processOutlookEmails( let attachments: OutlookAttachment[] = [] if (config.includeAttachments && email.hasAttachments) { try { - attachments = await downloadOutlookAttachments( - accessToken, - email.id, - requestId, - logger - ) + attachments = await downloadOutlookAttachments(accessToken, email.id, logger) } catch (error) { - logger.error(`Error downloading attachments for email ${email.id}:`, error, { - requestId, - }) + logger.error(`Error downloading attachments for email ${email.id}:`, error) } } @@ -454,8 +428,7 @@ async function processOutlookEmails( } logger.info( - `Processing email: ${email.subject} from ${email.from?.emailAddress?.address}`, - { requestId } + `Processing email: ${email.subject} from ${email.from?.emailAddress?.address}` ) const result = await processPolledWebhookEvent( @@ -470,8 +443,7 @@ async function processOutlookEmails( logger.error( `Failed to process webhook for email ${email.id}:`, result.statusCode, - result.error, - { requestId } + result.error ) throw new Error(`Webhook processing failed (${result.statusCode}): ${result.error}`) } @@ -484,13 +456,11 @@ async function processOutlookEmails( } ) - logger.info(`Successfully processed email ${email.id} for webhook ${webhookData.id}`, { - requestId, - }) + logger.info(`Successfully processed email ${email.id} for webhook ${webhookData.id}`) processedCount++ } catch (error) { if (error instanceof PollAdmissionRefusedError && processedCount === 0) throw error - logger.error(`Error processing email ${email.id}:`, error, { requestId }) + logger.error(`Error processing email ${email.id}:`, error) failedCount++ } } @@ -501,7 +471,6 @@ async function processOutlookEmails( async function downloadOutlookAttachments( accessToken: string, messageId: string, - requestId: string, logger: Logger ): Promise { const attachments: OutlookAttachment[] = [] @@ -517,7 +486,7 @@ async function downloadOutlookAttachments( ) if (!response.ok) { - logger.error(`Failed to fetch attachments for message ${messageId}`, { requestId }) + logger.error(`Failed to fetch attachments for message ${messageId}`) return attachments } @@ -541,17 +510,14 @@ async function downloadOutlookAttachments( } catch (error) { logger.error( `Error processing attachment ${attachment.id} for message ${messageId}:`, - error, - { requestId } + error ) } } - logger.info(`Downloaded ${attachments.length} attachments for message ${messageId}`, { - requestId, - }) + logger.info(`Downloaded ${attachments.length} attachments for message ${messageId}`) } catch (error) { - logger.error(`Error downloading attachments for message ${messageId}:`, error, { requestId }) + logger.error(`Error downloading attachments for message ${messageId}:`, error) } return attachments diff --git a/apps/sim/lib/webhooks/polling/rss.ts b/apps/sim/lib/webhooks/polling/rss.ts index 9da351d8f7b..5a4985a1c9c 100644 --- a/apps/sim/lib/webhooks/polling/rss.ts +++ b/apps/sim/lib/webhooks/polling/rss.ts @@ -106,7 +106,7 @@ export const rssPollingHandler: PollingProviderHandler = { const config = getProviderConfig(webhookData.providerConfig) if (!config?.feedUrl) { - logger.error(`Missing feedUrl for webhook ${webhookId}`, { requestId }) + logger.error(`Missing feedUrl for webhook ${webhookId}`) await markWebhookFailed(webhookId, logger) return 'failure' } @@ -117,16 +117,16 @@ export const rssPollingHandler: PollingProviderHandler = { items: newItems, etag, lastModified, - } = await fetchNewRssItems(config, webhookData.createdAt, requestId, logger) + } = await fetchNewRssItems(config, webhookData.createdAt, logger) if (!newItems.length) { await updateRssState(webhookId, now.toISOString(), [], config, logger, etag, lastModified) await markWebhookSuccess(webhookId, logger) - logger.info(`No new items found for webhook ${webhookId}`, { requestId }) + logger.info(`No new items found for webhook ${webhookId}`) return 'success' } - logger.info(`Found ${newItems.length} new items for webhook ${webhookId}`, { requestId }) + logger.info(`Found ${newItems.length} new items for webhook ${webhookId}`) const { processedCount, failedCount, admissionRejectedAt } = await processRssItems( newItems, @@ -148,8 +148,7 @@ export const rssPollingHandler: PollingProviderHandler = { await updateRssState(webhookId, now.toISOString(), newGuids, config, logger) } logger.info( - `Stopped polling webhook ${webhookId}: execution admission refused, ${newItems.length - attemptedItems.length} items left for a later poll`, - { requestId } + `Stopped polling webhook ${webhookId}: execution admission refused, ${newItems.length - attemptedItems.length} items left for a later poll` ) return 'skipped' } @@ -166,16 +165,13 @@ export const rssPollingHandler: PollingProviderHandler = { if (failedCount > 0 && processedCount === 0) { await markWebhookFailed(webhookId, logger) - logger.warn(`All ${failedCount} items failed to process for webhook ${webhookId}`, { - requestId, - }) + logger.warn(`All ${failedCount} items failed to process for webhook ${webhookId}`) return 'failure' } await markWebhookSuccess(webhookId, logger) logger.info( - `Successfully processed ${processedCount} items for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}`, - { requestId } + `Successfully processed ${processedCount} items for webhook ${webhookId}${failedCount > 0 ? ` (${failedCount} failed)` : ''}` ) return 'success' } catch (error) { @@ -225,7 +221,6 @@ async function updateRssState( async function fetchNewRssItems( config: RssWebhookConfig, subscriptionStartedAt: Date, - requestId: string, logger: Logger ): Promise<{ feed: RssFeed; items: RssItem[]; etag?: string; lastModified?: string }> { const urlValidation = await validateUrlWithDNS(config.feedUrl, 'feedUrl', 'requestTarget') @@ -252,7 +247,7 @@ async function fetchNewRssItems( }) if (response.status === 304) { - logger.info(`RSS feed not modified (304) for ${config.feedUrl}`, { requestId }) + logger.info(`RSS feed not modified (304) for ${config.feedUrl}`) return { feed: { items: [] } as RssFeed, items: [], @@ -310,9 +305,7 @@ async function fetchNewRssItems( const limitedItems = newItems.slice(0, 25) - logger.info(`Found ${newItems.length} new items (processing ${limitedItems.length})`, { - requestId, - }) + logger.info(`Found ${newItems.length} new items (processing ${limitedItems.length})`) return { feed: feed as RssFeed, @@ -338,9 +331,7 @@ async function processRssItems( const itemGuid = getRssItemGuid(item) if (!itemGuid) { - logger.warn(`Skipping RSS item with no identifiable GUID for webhook ${webhookData.id}`, { - requestId, - }) + logger.warn(`Skipping RSS item with no identifiable GUID for webhook ${webhookData.id}`) continue } @@ -385,8 +376,7 @@ async function processRssItems( logger.error( `Failed to process webhook for item ${itemGuid}:`, result.statusCode, - result.error, - { requestId } + result.error ) throw new Error(`Webhook processing failed (${result.statusCode}): ${result.error}`) } @@ -396,8 +386,7 @@ async function processRssItems( ) logger.info( - `Successfully processed item ${item.title || itemGuid} for webhook ${webhookData.id}`, - { requestId } + `Successfully processed item ${item.title || itemGuid} for webhook ${webhookData.id}` ) processedCount++ } catch (error) { @@ -405,7 +394,7 @@ async function processRssItems( return { processedCount, failedCount, admissionRejectedAt: index } } const errorMessage = getErrorMessage(error, 'Unknown error') - logger.error('Error processing item:', errorMessage, { requestId }) + logger.error('Error processing item:', errorMessage) failedCount++ } } diff --git a/apps/sim/lib/webhooks/polling/types.ts b/apps/sim/lib/webhooks/polling/types.ts index a208d279981..a3138141d79 100644 --- a/apps/sim/lib/webhooks/polling/types.ts +++ b/apps/sim/lib/webhooks/polling/types.ts @@ -18,7 +18,12 @@ export interface PollSummary { export interface PollWebhookContext { webhookData: WebhookRecord workflowData: WorkflowRecord + /** + * This poll's id, passed downstream as the request id of the executions it starts. + * `logger` already carries it as `pollId`; log lines never name it `requestId`. + */ requestId: string + /** Bound with `webhookId` and `pollId`. */ logger: Logger } diff --git a/apps/sim/lib/webhooks/polling/utils.ts b/apps/sim/lib/webhooks/polling/utils.ts index e1c3b5a1194..58d210971a1 100644 --- a/apps/sim/lib/webhooks/polling/utils.ts +++ b/apps/sim/lib/webhooks/polling/utils.ts @@ -63,14 +63,8 @@ export function throwIfAdmissionRefused(result: PolledWebhookEventResult): void } /** Logs a poll stopped by {@link PollAdmissionRefusedError} and reports it as skipped. */ -export function skipAdmissionRefusedPoll( - logger: Logger, - requestId: string, - webhookId: string -): 'skipped' { - logger.info(`Execution admission refused webhook ${webhookId}; left its items for a later poll`, { - requestId, - }) +export function skipAdmissionRefusedPoll(logger: Logger, webhookId: string): 'skipped' { + logger.info(`Execution admission refused webhook ${webhookId}; left its items for a later poll`) return 'skipped' } diff --git a/apps/sim/lib/workspace-events/emitter.ts b/apps/sim/lib/workspace-events/emitter.ts index f63c5296c9f..19c56f7fa85 100644 --- a/apps/sim/lib/workspace-events/emitter.ts +++ b/apps/sim/lib/workspace-events/emitter.ts @@ -46,14 +46,14 @@ export async function dispatchSimEvent( subscription: SimSubscription, payload: SimEventPayload ): Promise { - const requestId = generateShortId() + const dispatchId = generateShortId() try { const { processPolledWebhookEvent } = await import('@/lib/webhooks/processor') const result = await processPolledWebhookEvent( subscription.webhook, subscription.workflow, payload, - requestId + dispatchId ) if (!result.success) { @@ -61,12 +61,12 @@ export async function dispatchSimEvent( `Failed to fire sim trigger for workflow ${subscription.workflow.id}:`, result.statusCode, result.error, - { requestId } + { dispatchId } ) } } catch (error) { logger.error(`Error firing sim trigger for workflow ${subscription.workflow.id}:`, error, { - requestId, + dispatchId, }) } }