improvement(polling): fix correctness and efficiency across all polling handlers

- Gmail: paginate history API, add historyTypes filter, differentiate 403/429,
  fetch fresh historyId on fallback to break 404 retry loop
- Outlook: follow @odata.nextLink pagination, use fetchWithRetry for all Graph
  calls, fix $top alignment, skip folder filter on partial resolution failure,
  remove Content-Type from GET requests
- RSS: add conditional GET (ETag/If-None-Match), raise GUID cap to 500, fix 304
  ETag capture per RFC 9111, align GUID tracking with idempotency fallback key
- IMAP: single connection reuse, UIDVALIDITY tracking per mailbox, advance UID
  only on successful fetch, fix messageFlagsAdd range type, remove cross-mailbox
  legacy UID fallback
- Dispatch polling via trigger.dev task with per-provider concurrency key;
  fall back to synchronous Redis-locked polling for self-hosted

Author: waleedlatif1

apps/sim/app/api/webhooks/poll/[provider]/route.ts

@@ -1,9 +1,11 @@
 import { createLogger } from '@sim/logger'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
+import { isTriggerDevEnabled } from '@/lib/core/config/feature-flags'
 import { acquireLock, releaseLock } from '@/lib/core/config/redis'
 import { generateShortId } from '@/lib/core/utils/uuid'
 import { pollProvider, VALID_POLLING_PROVIDERS } from '@/lib/webhooks/polling'
+import { providerPolling } from '@/background/provider-polling'
 
 const logger = createLogger('PollingAPI')
 
@@ -20,9 +22,6 @@ export async function GET(
   const { provider } = await params
   const requestId = generateShortId()
 
-  const LOCK_KEY = `${provider}-polling-lock`
-  let lockValue: string | undefined
-
   try {
     const authError = verifyCronAuth(request, `${provider} webhook polling`)
     if (authError) return authError
@@ -31,29 +30,75 @@ export async function GET(
       return NextResponse.json({ error: `Unknown polling provider: ${provider}` }, { status: 404 })
     }
 
-    lockValue = requestId
-    const locked = await acquireLock(LOCK_KEY, lockValue, LOCK_TTL_SECONDS)
-    if (!locked) {
-      return NextResponse.json(
-        {
+    // When trigger.dev is enabled, dispatch polling as an async task and return immediately.
+    // Per-provider concurrency (concurrencyKey) ensures only one poll per provider runs at a time,
+    // while different providers (gmail vs outlook) can poll in parallel.
+    if (isTriggerDevEnabled) {
+      try {
+        const handle = await providerPolling.trigger(
+          { provider, requestId },
+          {
+            concurrencyKey: provider,
+            tags: [`provider:${provider}`],
+          }
+        )
+
+        logger.info(`[${requestId}] Dispatched ${provider} polling to trigger.dev`, {
+          runId: handle.id,
+        })
+
+        return NextResponse.json({
           success: true,
-          message: 'Polling already in progress – skipped',
+          message: `${provider} polling dispatched`,
           requestId,
-          status: 'skip',
-        },
-        { status: 202 }
-      )
+          runId: handle.id,
+          status: 'dispatched',
+        })
+      } catch (triggerError) {
+        // If trigger.dev is unavailable, fall through to synchronous polling below.
+        logger.warn(
+          `[${requestId}] Trigger.dev dispatch failed for ${provider}, falling back to synchronous polling`,
+          {
+            error: triggerError instanceof Error ? triggerError.message : String(triggerError),
+          }
+        )
+      }
     }
 
-    const results = await pollProvider(provider)
+    // Fallback: synchronous polling when trigger.dev is not enabled (self-hosted).
+    // Redis lock prevents concurrent polls for the same provider.
+    const LOCK_KEY = `${provider}-polling-lock`
+    let lockValue: string | undefined
+
+    try {
+      lockValue = requestId
+      const locked = await acquireLock(LOCK_KEY, lockValue, LOCK_TTL_SECONDS)
+      if (!locked) {
+        return NextResponse.json(
+          {
+            success: true,
+            message: 'Polling already in progress – skipped',
+            requestId,
+            status: 'skip',
+          },
+          { status: 202 }
+        )
+      }
 
-    return NextResponse.json({
-      success: true,
-      message: `${provider} polling completed`,
-      requestId,
-      status: 'completed',
-      ...results,
-    })
+      const results = await pollProvider(provider)
+
+      return NextResponse.json({
+        success: true,
+        message: `${provider} polling completed`,
+        requestId,
+        status: 'completed',
+        ...results,
+      })
+    } finally {
+      if (lockValue) {
+        await releaseLock(LOCK_KEY, lockValue).catch(() => {})
+      }
+    }
   } catch (error) {
     logger.error(`Error during ${provider} polling (${requestId}):`, error)
     return NextResponse.json(
@@ -65,9 +110,5 @@ export async function GET(
       },
       { status: 500 }
     )
-  } finally {
-    if (lockValue) {
-      await releaseLock(LOCK_KEY, lockValue).catch(() => {})
-    }
   }
 }

apps/sim/background/provider-polling.ts

@@ -0,0 +1,34 @@
+import { createLogger } from '@sim/logger'
+import { task } from '@trigger.dev/sdk'
+import { pollProvider } from '@/lib/webhooks/polling'
+
+const logger = createLogger('TriggerProviderPolling')
+
+export type ProviderPollingPayload = {
+  provider: string
+  requestId: string
+}
+
+export const providerPolling = task({
+  id: 'provider-polling',
+  machine: 'medium-1x',
+  maxDuration: 300,
+  retry: {
+    maxAttempts: 1,
+  },
+  queue: {
+    name: 'provider-polling',
+    concurrencyLimit: 1,
+  },
+  run: async (payload: ProviderPollingPayload) => {
+    const { provider, requestId } = payload
+
+    logger.info(`[${requestId}] Starting ${provider} polling`)
+
+    const result = await pollProvider(provider)
+
+    logger.info(`[${requestId}] ${provider} polling completed`, result)
+
+    return result
+  },
+})

apps/sim/lib/webhooks/polling/gmail.ts

@@ -151,44 +151,70 @@ async function fetchNewEmails(
     let latestHistoryId = config.historyId
 
     if (useHistoryApi) {
-      const historyUrl = `https://gmail.googleapis.com/gmail/v1/users/me/history?startHistoryId=${config.historyId}`
+      const messageIds = new Set<string>()
+      let pageToken: string | undefined
 
-      const historyResponse = await fetch(historyUrl, {
-        headers: { Authorization: `Bearer ${accessToken}` },
-      })
+      do {
+        let historyUrl = `https://gmail.googleapis.com/gmail/v1/users/me/history?startHistoryId=${config.historyId}&historyTypes=messageAdded`
+        if (pageToken) {
+          historyUrl += `&pageToken=${pageToken}`
+        }
 
-      if (!historyResponse.ok) {
-        const errorData = await historyResponse.json()
-        logger.error(`[${requestId}] Gmail history API error:`, {
-          status: historyResponse.status,
-          statusText: historyResponse.statusText,
-          error: errorData,
+        const historyResponse = await fetch(historyUrl, {
+          headers: { Authorization: `Bearer ${accessToken}` },
         })
 
-        logger.info(`[${requestId}] Falling back to search API after history API failure`)
-        return searchEmails(accessToken, config, requestId, logger)
-      }
+        if (!historyResponse.ok) {
+          const status = historyResponse.status
+          const errorData = await historyResponse.json().catch(() => ({}))
+          logger.error(`[${requestId}] Gmail history API error:`, {
+            status,
+            statusText: historyResponse.statusText,
+            error: errorData,
+          })
+
+          if (status === 403 || status === 429) {
+            throw new Error(
+              `Gmail API error ${status} — skipping to retry next poll cycle: ${JSON.stringify(errorData)}`
+            )
+          }
 
-      const historyData = await historyResponse.json()
+          logger.info(`[${requestId}] Falling back to search API after history API error ${status}`)
+          const searchResult = await searchEmails(accessToken, config, requestId, logger)
+          // When search finds 0 emails after a history API failure, the stored historyId is likely
+          // invalid. Fetch a fresh one from the profile API to break the potential 404 retry loop.
+          if (searchResult.emails.length === 0) {
+            const freshHistoryId = await getGmailProfileHistoryId(accessToken, requestId, logger)
+            if (freshHistoryId) {
+              logger.info(
+                `[${requestId}] Fetched fresh historyId ${freshHistoryId} after invalid historyId (was: ${config.historyId})`
+              )
+              return { emails: [], latestHistoryId: freshHistoryId }
+            }
+          }
+          return searchResult
+        }
 
-      if (!historyData.history || !historyData.history.length) {
-        return { emails: [], latestHistoryId }
-      }
+        const historyData = await historyResponse.json()
 
-      if (historyData.historyId) {
-        latestHistoryId = historyData.historyId
-      }
+        if (historyData.historyId) {
+          latestHistoryId = historyData.historyId
+        }
 
-      const messageIds = new Set<string>()
-      for (const history of historyData.history) {
-        if (history.messagesAdded) {
-          for (const messageAdded of history.messagesAdded) {
-            messageIds.add(messageAdded.message.id)
+        if (historyData.history) {
+          for (const history of historyData.history) {
+            if (history.messagesAdded) {
+              for (const messageAdded of history.messagesAdded) {
+                messageIds.add(messageAdded.message.id)
+              }
+            }
           }
         }
-      }
 
-      if (messageIds.size === 0) {
+        pageToken = historyData.nextPageToken
+      } while (pageToken)
+
+      if (!messageIds.size) {
         return { emails: [], latestHistoryId }
       }
 
@@ -352,6 +378,29 @@ async function searchEmails(
   }
 }
 
+async function getGmailProfileHistoryId(
+  accessToken: string,
+  requestId: string,
+  logger: ReturnType<typeof import('@sim/logger').createLogger>
+): 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(
+        `[${requestId}] 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(`[${requestId}] Error fetching Gmail profile:`, error)
+    return null
+  }
+}
+
 async function getEmailDetails(accessToken: string, messageId: string): Promise<GmailEmail> {
   const messageUrl = `https://gmail.googleapis.com/gmail/v1/users/me/messages/${messageId}?format=full`