OUT-4101: Process webhook changes per channel instead of a global serializer

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

@@ -231,14 +231,17 @@ export class DropboxWebhook {
     }
 
     if (allChanges.length > 0) {
-      const result = await handleChannelFileChanges.triggerAndWait({
-        files: allChanges,
-        channelSyncId,
-        dbxRootPath,
-        assemblyChannelId,
-        user,
-        connectionToken,
-      })
+      const result = await handleChannelFileChanges.triggerAndWait(
+        {
+          files: allChanges,
+          channelSyncId,
+          dbxRootPath,
+          assemblyChannelId,
+          user,
+          connectionToken,
+        },
+        { concurrencyKey: channelSyncId },
+      )
       // Don't advance the cursor if change processing failed, or these deltas move
       // past the cursor and are never re-fetched. Throwing lets the run retry from
       // the same cursor.

src/trigger/processFileSync.ts

@@ -232,13 +232,15 @@ export const handleChannelFileChanges = task({
      * After partial unique index: First delete and then create the files. This ensures that constraints are not violated.
      * This section should handle file rename, folder rename cases
      */
+    // Keyed per channel so different channels run in parallel instead of the whole
+    // system going one-at-a-time; delete-before-create ordering is preserved by
+    // awaiting each step in turn.
     if (deleted.length)
-      await deleteDropboxFileInAssembly.batchTriggerAndWait(deleted.map(toPayload))
-    if (created.length) await syncDropboxFileToAssembly.batchTriggerAndWait(created.map(toPayload))
-
-    for (const entry of contentUpdated) {
-      await updateDropboxFileInAssembly.triggerAndWait({ opts, entry })
-    }
+      await fanOutAndWait(deleteDropboxFileInAssembly, deleted.map(toPayload), channelSyncId)
+    if (created.length)
+      await fanOutAndWait(syncDropboxFileToAssembly, created.map(toPayload), channelSyncId)
+    if (contentUpdated.length)
+      await fanOutAndWait(updateDropboxFileInAssembly, contentUpdated.map(toPayload), channelSyncId)
   },
 })
 
@@ -265,8 +267,12 @@ export const updateDropboxFileInAssembly = task({
   },
   retry: RETRY_CONFIG,
   run: async (payload: DropboxToAssemblySyncFilesPayload) => {
-    await deleteDropboxFileInAssembly.triggerAndWait(payload)
-    await syncDropboxFileToAssembly.trigger(payload)
+    const { channelSyncId } = payload.opts
+    // Await the recreate (not fire-and-forget) so the mapping row is committed before
+    // this update reports done — otherwise a follow-up webhook for the same file can
+    // read stale state and dispatch a second concurrent sync of the same id.
+    await deleteDropboxFileInAssembly.triggerAndWait(payload, { concurrencyKey: channelSyncId })
+    await syncDropboxFileToAssembly.triggerAndWait(payload, { concurrencyKey: channelSyncId })
   },
 })
 

src/utils/__tests__/classify-dbx-changes.test.ts

@@ -133,4 +133,54 @@ describe('classifyDbxChanges', () => {
     expect(ids(result.created)).toEqual(['add'])
     expect(ids(result.contentUpdated)).toEqual(['upd'])
   })
+
+  it('dedupes duplicate ids within a bucket, keeping the last occurrence', () => {
+    const entries = [
+      makeEntry({ id: 'dup', content_hash: 'new-1' }),
+      makeEntry({ id: 'dup', content_hash: 'new-2' }),
+    ]
+    const rows = [makeRow('dup', 'old')]
+
+    const result = classifyDbxChanges(entries, rows)
+
+    expect(ids(result.contentUpdated)).toEqual(['dup'])
+    expect(result.contentUpdated[0].content_hash).toBe('new-2')
+  })
+
+  it('keeps a rename pair intact (same id in delete and create buckets)', () => {
+    const entries = [
+      makeEntry({ id: 'x', '.tag': 'deleted' }),
+      makeEntry({ id: 'x', '.tag': 'file', path_display: '/folder/renamed.txt' }),
+    ]
+    const rows = [makeRow('x')]
+
+    const result = classifyDbxChanges(entries, rows)
+
+    expect(ids(result.deleted)).toEqual(['x'])
+    expect(ids(result.created)).toEqual(['x'])
+  })
+
+  it('dedupes duplicate ids within the created bucket', () => {
+    const entries = [
+      makeEntry({ id: 'new', path_display: '/folder/a.txt' }),
+      makeEntry({ id: 'new', path_display: '/folder/b.txt' }),
+    ]
+
+    const result = classifyDbxChanges(entries, [])
+
+    expect(ids(result.created)).toEqual(['new'])
+    expect(result.created[0].path_display).toBe('/folder/b.txt')
+  })
+
+  it('dedupes duplicate ids within the deleted bucket', () => {
+    const entries = [
+      makeEntry({ id: 'gone', '.tag': 'deleted' }),
+      makeEntry({ id: 'gone', '.tag': 'deleted' }),
+    ]
+    const rows = [makeRow('gone')]
+
+    const result = classifyDbxChanges(entries, rows)
+
+    expect(ids(result.deleted)).toEqual(['gone'])
+  })
 })

src/utils/classify-dbx-changes.ts

@@ -10,6 +10,18 @@ export type DbxChangeClassification = {
   contentUpdated: DropboxFileListFolderSingleEntry[]
 }
 
+// Keep only the last entry per id, preserving order. Guards against a delta carrying
+// duplicate ids in one bucket, which would otherwise fan out concurrent same-file syncs
+// and double-create or trip the partial unique index. Applied per bucket so a rename's
+// delete+create pair (same id, different buckets) is left intact.
+const dedupeByIdKeepLast = (
+  entries: DropboxFileListFolderSingleEntry[],
+): DropboxFileListFolderSingleEntry[] => {
+  const byId = new Map<string, DropboxFileListFolderSingleEntry>()
+  for (const entry of entries) byId.set(entry.id, entry)
+  return [...byId.values()]
+}
+
 /**
  * Splits Dropbox delta entries into deletes, creates, and content updates by comparing
  * them against the already-mapped rows. The caller must process `deleted` before `created`
@@ -42,5 +54,9 @@ export const classifyDbxChanges = (
     return !!existing?.contentHash && existing.contentHash !== entry.content_hash
   })
 
-  return { deleted, created, contentUpdated }
+  return {
+    deleted: dedupeByIdKeepLast(deleted),
+    created: dedupeByIdKeepLast(created),
+    contentUpdated: dedupeByIdKeepLast(contentUpdated),
+  }
 }