Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .agents/skills/add-trigger/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
91 changes: 30 additions & 61 deletions apps/sim/lib/webhooks/polling/gmail.ts
Original file line number Diff line number Diff line change
Expand Up @@ -77,12 +77,7 @@ export const gmailPollingHandler: PollingProviderHandler = {
const config = getProviderConfig<GmailWebhookConfig>(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.
Expand All @@ -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,
Expand All @@ -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[] = []
Expand All @@ -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,
Expand All @@ -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()
Expand Down Expand Up @@ -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(
Expand All @@ -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
}
}
Expand Down Expand Up @@ -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 = ''
Expand Down Expand Up @@ -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,
Expand All @@ -338,24 +317,23 @@ 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 }
}

const idsToFetch = searchData.messages.slice(0, config.maxEmailsPerPoll || 25)
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(
idsToFetch.map((message: { id: string }) => getEmailDetails(accessToken, message.id))
)
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(
Expand All @@ -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
}
}
Expand All @@ -383,40 +361,36 @@ 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<string | null> {
try {
const response = await fetch('https://gmail.googleapis.com/gmail/v1/users/me/profile', {
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
}
}
Expand Down Expand Up @@ -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)
}
}

Expand Down Expand Up @@ -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}`)
}
Expand All @@ -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++
}
}
Expand Down
26 changes: 10 additions & 16 deletions apps/sim/lib/webhooks/polling/google-calendar.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'
}

Expand All @@ -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,
Expand Down Expand Up @@ -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'
}
Expand Down Expand Up @@ -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}`)
}
Expand All @@ -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++
}
}
Expand Down
Loading
Loading