OUT-4104: Skip the sync job when a Dropbox webhook has no relevant changes

src/app/api/webhook/dropbox/route.ts

@@ -4,6 +4,9 @@ import {
 } from '@/features/webhook/dropbox/api/webhook.controller'
 import { withErrorHandler } from '@/utils/withErrorHandler'
 
+// Background processing runs in after(), which Vercel bounds by maxDuration.
+export const maxDuration = 300
+
 /**
  * not used withErrorHander() as this is a sync function and has included its separate try catch block
  */

src/features/webhook/dropbox/api/webhook.controller.ts

@@ -1,6 +1,7 @@
 import crypto from 'node:crypto'
+import * as Sentry from '@sentry/nextjs'
 import status from 'http-status'
-import { type NextRequest, NextResponse } from 'next/server'
+import { after, type NextRequest, NextResponse } from 'next/server'
 import env from '@/config/server.env'
 import { DropboxWebhook } from '@/features/webhook/dropbox/lib/webhook.service'
 import { sleep } from '@/utils/sleep'
@@ -31,8 +32,6 @@ export const handleWebhookEvents = async (req: NextRequest) => {
 
   const body = await req.text()
 
-  await sleep(800) // prevent ping-pong case of webhooks
-
   const computedSignature = crypto
     .createHmac('sha256', env.DROPBOX_APP_SECRET)
     .update(body)
@@ -48,8 +47,17 @@ export const handleWebhookEvents = async (req: NextRequest) => {
   const { list_folder } = JSON.parse(body)
   const accounts = list_folder?.accounts ?? []
 
-  const dropboxWebhook = new DropboxWebhook()
-  await dropboxWebhook.handleDropboxEvents(accounts)
+  // Reply to Dropbox first, then process in the background so the check doesn't slow the reply.
+  after(async () => {
+    try {
+      await sleep(800) // let our own writes settle first
+      await new DropboxWebhook().handleDropboxEvents(accounts)
+    } catch (error) {
+      // Dropbox already got its 200, so it won't retry — report so this is visible.
+      console.error('Dropbox webhook :: background processing failed', { accounts }, error)
+      Sentry.captureException(error)
+    }
+  })
 
   // Dropbox expects a 200 OK with plain text body
   return new NextResponse('', {

src/features/webhook/dropbox/lib/webhook.service.ts

@@ -1,3 +1,4 @@
+import * as Sentry from '@sentry/nextjs'
 import { and, eq } from 'drizzle-orm'
 import { type Dropbox, DropboxResponseError } from 'dropbox'
 import httpStatus from 'http-status'
@@ -25,35 +26,120 @@ const DEBOUNCE_WINDOW_MS = 5 * 60 * 1000 // 5 minutes
 export class DropboxWebhook {
   async handleDropboxEvents(accounts: string[]) {
     for (const account of accounts) {
-      const connection = await db.query.dropboxConnections.findFirst({
-        where: (t, { eq, and }) => and(eq(t.accountId, account), eq(t.status, true)),
-        columns: { id: true, pendingWebhook: true, lastWebhookSyncStartedAt: true },
+      // Isolate per-account failures so one bad account doesn't block the rest. This only
+      // fires if the lookup itself threw (connection unknown, so nothing to flag) — log it;
+      // the next webhook re-processes the account since the cursor is untouched.
+      await this.processAccountWebhook(account).catch((error) => {
+        logger.error(`DropboxWebhook#handleDropboxEvents :: failed for ${account}`, error)
+        Sentry.captureException(error)
       })
+    }
+  }
 
-      if (!connection) continue
+  private async processAccountWebhook(account: string) {
+    const connection = await db.query.dropboxConnections.findFirst({
+      where: (t, { eq, and }) => and(eq(t.accountId, account), eq(t.status, true)),
+      columns: { id: true, pendingWebhook: true, lastWebhookSyncStartedAt: true },
+    })
 
-      // Skip if already pending — cron will handle it
-      if (connection.pendingWebhook) {
-        console.info(`Webhook skipped for account ${account}, already has pending webhook`)
-        continue
-      }
+    if (!connection) return
 
-      // Debounce: if the account was synced recently, defer to cron
-      const debounceThreshold = new Date(Date.now() - DEBOUNCE_WINDOW_MS)
-      const recentlySynced =
-        connection.lastWebhookSyncStartedAt &&
-        connection.lastWebhookSyncStartedAt >= debounceThreshold
+    // Skip if already pending — cron will handle it
+    if (connection.pendingWebhook) {
+      console.info(`Webhook skipped for account ${account}, already has pending webhook`)
+      return
+    }
 
+    // Debounce: if the account was synced recently, defer to cron
+    const debounceThreshold = new Date(Date.now() - DEBOUNCE_WINDOW_MS)
+    const recentlySynced =
+      connection.lastWebhookSyncStartedAt &&
+      connection.lastWebhookSyncStartedAt >= debounceThreshold
+
+    try {
       if (recentlySynced) {
-        await db
-          .update(dropboxConnections)
-          .set({ pendingWebhook: true })
-          .where(eq(dropboxConnections.id, connection.id))
+        await this.markConnectionPending(connection.id)
         console.info(`Webhook debounced for account ${account}, marked as pending`)
       } else {
-        await processDropboxChanges.trigger(account, { concurrencyKey: account })
+        await this.triggerIfPendingChanges(account)
       }
+    } catch (error) {
+      // The 200 already went back to Dropbox, so mark this specific connection pending
+      // for the catch-up cron to retry.
+      logger.error(`DropboxWebhook#processAccountWebhook :: failed for ${account}`, error)
+      Sentry.captureException(error)
+      await this.markConnectionPending(connection.id).catch((markError) =>
+        Sentry.captureException(markError),
+      )
+    }
+  }
+
+  private markConnectionPending(id: string) {
+    return db
+      .update(dropboxConnections)
+      .set({ pendingWebhook: true })
+      .where(eq(dropboxConnections.id, id))
+  }
+
+  // Only start the sync job when the account actually has changes to sync. Fail open:
+  // any error triggers the job so a real change is never dropped.
+  private async triggerIfPendingChanges(account: string) {
+    const shouldTrigger = await this.accountHasPendingChanges(account).catch((error) => {
+      logger.warn(
+        `DropboxWebhook#triggerIfPendingChanges :: pre-check failed for ${account}, triggering anyway`,
+        error,
+      )
+      return true // fail open — never drop a possibly-real change
+    })
+    if (shouldTrigger) {
+      await processDropboxChanges.trigger(account, { concurrencyKey: account })
+    } else {
+      console.info(`Webhook skipped for account ${account}, no relevant changes to sync`)
+    }
+  }
+
+  // Read-only peek of each channel's delta to decide if the sync job is worth starting.
+  // Does not persist the advanced cursor — the job re-fetches from the stored one.
+  async accountHasPendingChanges(account: string): Promise<boolean> {
+    const connection = await this.getActiveConnection(account)
+    if (!connection?.refreshToken) return false // no token → the job can't sync anyway, skip it
+
+    const channels = await db.query.channelSync.findMany({
+      where: (t, { eq, and }) => and(eq(t.dbxAccountId, account), eq(t.status, true)),
+      columns: { dbxRootPath: true, dbxCursor: true },
+    })
+    if (!channels.length) return false // nothing mapped for this account
+
+    const dbxClient = new DropboxClient(
+      connection.refreshToken,
+      connection.rootNamespaceId,
+    ).getDropboxClient()
+
+    for (const channel of channels) {
+      if (!channel.dbxCursor) return true // no baseline to peek → let the job run
+      const root = channel.dbxRootPath.toLowerCase()
+      if (await this.deltaHasRelevantEntry(dbxClient, channel.dbxCursor, root)) return true
     }
+
+    return false
+  }
+
+  // Peek delta pages from `cursor` (read-only — the cursor is never persisted), short-
+  // circuiting as soon as an entry under `root` appears. A missing path_display can't be
+  // placed, so it's treated as relevant (fail open).
+  private async deltaHasRelevantEntry(
+    dbxClient: Dropbox,
+    cursor: string,
+    root: string,
+  ): Promise<boolean> {
+    const { result } = await dbxClient.filesListFolderContinue({ cursor })
+    const relevant = result.entries.some((entry) => {
+      const path = entry.path_display?.toLowerCase()
+      return !path || path === root || path.startsWith(`${root}/`)
+    })
+    if (relevant) return true
+    if (!result.has_more) return false
+    return this.deltaHasRelevantEntry(dbxClient, result.cursor, root)
   }
 
   async fetchDropBoxChanges(accountId: string) {

test/flows/dropbox-webhook-debounce.integration.test.ts

@@ -37,12 +37,14 @@ describe('webhook debounce', () => {
   })
 
   it('triggers a sync when the last sync is older than the window', async () => {
-    // No channels seeded: fetchDropBoxChanges clears pending + stamps timestamps with no external calls.
     await dropboxConnectionSeeder.create({
       accountId: ACCOUNT,
       pendingWebhook: false,
       lastWebhookSyncStartedAt: minutesAgo(6),
     })
+    // Isolate debounce timing from the pre-check (covered in dropbox-webhook-precheck):
+    // assume the account has changes so the trigger path runs.
+    vi.spyOn(DropboxWebhook.prototype, 'accountHasPendingChanges').mockResolvedValue(true)
 
     await new DropboxWebhook().handleDropboxEvents([ACCOUNT])
 
@@ -88,4 +90,59 @@ describe('webhook debounce', () => {
     expect(row.pendingWebhook).toBe(true) // unchanged
     expect(row.lastWebhookSyncedAt).toBeNull() // no sync triggered
   })
+
+  it('marks pending when processing fails, so the catch-up cron retries', async () => {
+    await dropboxConnectionSeeder.create({
+      accountId: ACCOUNT,
+      pendingWebhook: false,
+      lastWebhookSyncStartedAt: minutesAgo(6),
+    })
+    vi.spyOn(DropboxWebhook.prototype, 'accountHasPendingChanges').mockResolvedValue(true)
+    vi.spyOn(processDropboxChanges, 'trigger').mockRejectedValue(new Error('trigger down'))
+
+    await new DropboxWebhook().handleDropboxEvents([ACCOUNT])
+
+    const row = await readConnection()
+    expect(row.pendingWebhook).toBe(true) // marked so the cron re-runs it
+  })
+
+  it('marks only the failed connection pending when an account has two connections', async () => {
+    for (const portalId of ['portal-a', 'portal-b']) {
+      await dropboxConnectionSeeder.create({
+        accountId: 'acc-multi',
+        portalId,
+        pendingWebhook: false,
+        lastWebhookSyncStartedAt: minutesAgo(6),
+      })
+    }
+    vi.spyOn(DropboxWebhook.prototype, 'accountHasPendingChanges').mockResolvedValue(true)
+    vi.spyOn(processDropboxChanges, 'trigger').mockRejectedValue(new Error('boom'))
+
+    await new DropboxWebhook().handleDropboxEvents(['acc-multi'])
+
+    const rows = await db
+      .select()
+      .from(dropboxConnections)
+      .where(eq(dropboxConnections.accountId, 'acc-multi'))
+    expect(rows.filter((r) => r.pendingWebhook)).toHaveLength(1) // only the processed one, not both
+  })
+
+  it('keeps processing later accounts when an earlier one fails', async () => {
+    for (const accountId of ['acc-fail', 'acc-ok']) {
+      await dropboxConnectionSeeder.create({
+        accountId,
+        pendingWebhook: false,
+        lastWebhookSyncStartedAt: minutesAgo(6),
+      })
+    }
+    vi.spyOn(DropboxWebhook.prototype, 'accountHasPendingChanges').mockResolvedValue(true)
+    const triggerSpy = vi
+      .spyOn(processDropboxChanges, 'trigger')
+      .mockRejectedValueOnce(new Error('boom')) // acc-fail
+      .mockResolvedValue(undefined as never) // acc-ok
+
+    await new DropboxWebhook().handleDropboxEvents(['acc-fail', 'acc-ok'])
+
+    expect(triggerSpy).toHaveBeenCalledWith('acc-ok', { concurrencyKey: 'acc-ok' })
+  })
 })

test/flows/dropbox-webhook-precheck.integration.test.ts

@@ -0,0 +1,124 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { DropboxWebhook } from '@/features/webhook/dropbox/lib/webhook.service'
+import { processDropboxChanges } from '@/trigger/processFileSync'
+import { dropboxEntryFactory } from '../factories'
+import { mockDropboxRpc, paginateDropboxListFolder, server } from '../msw'
+import { channelSeeder, dropboxConnectionSeeder } from '../seeders'
+
+const seedAccount = () =>
+  dropboxConnectionSeeder.create({ accountId: 'acc', rootNamespaceId: 'ns', refreshToken: 'rt' })
+
+afterEach(() => vi.restoreAllMocks())
+
+describe('DropboxWebhook#accountHasPendingChanges', () => {
+  it('returns false when no delta entry falls under a channel root', async () => {
+    const connection = await seedAccount()
+    await channelSeeder.create({
+      portalId: connection.portalId,
+      dbxRootPath: '/root',
+      dbxCursor: 'cursor:0',
+    })
+    // The whole delta is outside the synced root — nothing to sync.
+    server.use(
+      ...paginateDropboxListFolder([dropboxEntryFactory.build({ path_display: '/other/x.txt' })]),
+    )
+
+    expect(await new DropboxWebhook().accountHasPendingChanges('acc')).toBe(false)
+  })
+
+  it('returns true when a delta entry falls under a channel root', async () => {
+    const connection = await seedAccount()
+    await channelSeeder.create({
+      portalId: connection.portalId,
+      dbxRootPath: '/root',
+      dbxCursor: 'cursor:0',
+    })
+    server.use(
+      ...paginateDropboxListFolder([dropboxEntryFactory.build({ path_display: '/root/x.txt' })]),
+    )
+
+    expect(await new DropboxWebhook().accountHasPendingChanges('acc')).toBe(true)
+  })
+
+  it('returns true when a channel has no cursor (cannot peek safely)', async () => {
+    const connection = await seedAccount()
+    await channelSeeder.create({
+      portalId: connection.portalId,
+      dbxRootPath: '/root',
+      dbxCursor: null,
+    })
+
+    expect(await new DropboxWebhook().accountHasPendingChanges('acc')).toBe(true)
+  })
+
+  it('returns false when the account has no active channels', async () => {
+    await seedAccount()
+
+    expect(await new DropboxWebhook().accountHasPendingChanges('acc')).toBe(false)
+  })
+
+  it('returns false when the connection has no refresh token (job cannot sync)', async () => {
+    await dropboxConnectionSeeder.create({
+      accountId: 'acc',
+      rootNamespaceId: 'ns',
+      refreshToken: null,
+    })
+
+    expect(await new DropboxWebhook().accountHasPendingChanges('acc')).toBe(false)
+  })
+
+  it('throws on a Dropbox error so the caller can fail open and trigger', async () => {
+    const connection = await seedAccount()
+    await channelSeeder.create({
+      portalId: connection.portalId,
+      dbxRootPath: '/root',
+      dbxCursor: 'cursor:0',
+    })
+    mockDropboxRpc('/2/files/list_folder/continue', () =>
+      Response.json({ error_summary: 'reset/', error: { '.tag': 'reset' } }, { status: 409 }),
+    )
+
+    await expect(new DropboxWebhook().accountHasPendingChanges('acc')).rejects.toBeDefined()
+  })
+
+  it('returns true when an entry has no path_display (fail open)', async () => {
+    const connection = await seedAccount()
+    await channelSeeder.create({
+      portalId: connection.portalId,
+      dbxRootPath: '/root',
+      dbxCursor: 'cursor:0',
+    })
+    // Raw entry with no path_display (e.g. unmounted) — can't tell → treat as relevant.
+    server.use(...paginateDropboxListFolder([{ '.tag': 'deleted', name: 'x' }]))
+
+    expect(await new DropboxWebhook().accountHasPendingChanges('acc')).toBe(true)
+  })
+
+  it('does not match a sibling folder that shares the root prefix', async () => {
+    const connection = await seedAccount()
+    await channelSeeder.create({
+      portalId: connection.portalId,
+      dbxRootPath: '/root',
+      dbxCursor: 'cursor:0',
+    })
+    server.use(
+      ...paginateDropboxListFolder([dropboxEntryFactory.build({ path_display: '/rootbar/x.txt' })]),
+    )
+
+    expect(await new DropboxWebhook().accountHasPendingChanges('acc')).toBe(false)
+  })
+
+  it('fails open: handleDropboxEvents triggers the job when the pre-check throws', async () => {
+    await seedAccount() // lastWebhookSyncStartedAt null → not debounced → reaches the pre-check
+    vi.spyOn(DropboxWebhook.prototype, 'accountHasPendingChanges').mockRejectedValue(
+      new Error('boom'),
+    )
+    const triggerSpy = vi
+      .spyOn(processDropboxChanges, 'trigger')
+      .mockResolvedValue(undefined as never)
+
+    await new DropboxWebhook().handleDropboxEvents(['acc'])
+
+    expect(triggerSpy).toHaveBeenCalledWith('acc', { concurrencyKey: 'acc' })
+  })
+})