refactor(quickbooks): align webhook trigger routing

Author: Bill Leoutsakos

apps/sim/background/quickbooks-webhook-ingress.test.ts

@@ -2,15 +2,17 @@
 
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { mockDispatch, mockEnqueue, mockFindPage } = vi.hoisted(() => ({
+const { mockDispatch, mockEnqueue, mockFindWebhooks } = vi.hoisted(() => ({
   mockDispatch: vi.fn(),
   mockEnqueue: vi.fn(),
-  mockFindPage: vi.fn(),
+  mockFindWebhooks: vi.fn(),
 }))
-vi.mock('@trigger.dev/sdk', () => ({ task: vi.fn((config: unknown) => config) }))
-vi.mock('@/lib/webhooks/processor', () => ({ dispatchResolvedWebhookTarget: mockDispatch }))
-vi.mock('@/background/quickbooks-webhook-targets', () => ({
-  findQuickBooksWebhookTargetPage: mockFindPage,
+vi.mock('@trigger.dev/sdk', () => ({
+  task: vi.fn((config: unknown) => config),
+}))
+vi.mock('@/lib/webhooks/processor', () => ({
+  dispatchResolvedWebhookTarget: mockDispatch,
+  findWebhooksByRoutingKey: mockFindWebhooks,
 }))
 vi.mock('@/lib/core/async-jobs', () => ({
   getJobQueue: vi.fn(async () => ({ enqueue: mockEnqueue })),
@@ -46,45 +48,33 @@ describe('QuickBooks webhook ingress job', () => {
 
   it('routes the batch by company and dispatches targets sequentially', async () => {
     const order: string[] = []
-    mockFindPage
-      .mockResolvedValueOnce({
-        hasMore: true,
-        nextCursor: 'w2',
-        targets: [
-          { webhook: { id: 'w1' }, workflow: { id: 'wf1' } },
-          { webhook: { id: 'w2' }, workflow: { id: 'wf2' } },
-        ],
-      })
-      .mockResolvedValueOnce({
-        hasMore: false,
-        nextCursor: null,
-        targets: [{ webhook: { id: 'w3' }, workflow: { id: 'wf3' } }],
-      })
-      .mockResolvedValueOnce({
-        hasMore: false,
-        nextCursor: null,
-        targets: [{ webhook: { id: 'w4' }, workflow: { id: 'wf4' } }],
-      })
+    mockFindWebhooks
+      .mockResolvedValueOnce([
+        { webhook: { id: 'w1' }, workflow: { id: 'wf1' } },
+        { webhook: { id: 'w2' }, workflow: { id: 'wf2' } },
+      ])
+      .mockResolvedValueOnce([{ webhook: { id: 'w3' }, workflow: { id: 'wf3' } }])
     mockDispatch.mockImplementation(async (webhook: { id: string }) => {
       order.push(webhook.id)
       return { outcome: 'queued' }
     })
     await expect(executeQuickBooksWebhookIngress(payload)).resolves.toEqual({
       failed: 0,
       ignored: 0,
-      processed: 4,
-      targetCount: 4,
+      processed: 3,
+      targetCount: 3,
     })
-    expect(mockFindPage).toHaveBeenCalledWith('456', 'request-1', undefined)
-    expect(mockFindPage).toHaveBeenCalledWith('456', 'request-1', 'w2')
-    expect(mockFindPage).toHaveBeenCalledWith('789', 'request-1', undefined)
-    expect(order).toEqual(['w1', 'w2', 'w3', 'w4'])
+    expect(mockFindWebhooks).toHaveBeenNthCalledWith(1, '456', 'request-1', 'quickbooks')
+    expect(mockFindWebhooks).toHaveBeenNthCalledWith(2, '789', 'request-1', 'quickbooks')
+    expect(order).toEqual(['w1', 'w2', 'w3'])
   })
 
   it('enqueues the bounded delivery once without copying it into continuation jobs', async () => {
-    mockFindPage.mockResolvedValue({ hasMore: false, nextCursor: null, targets: [] })
+    mockFindWebhooks.mockResolvedValue([])
     await enqueueQuickBooksWebhookIngress(payload)
-    const options = mockEnqueue.mock.calls[0][2] as { runner: () => Promise<void> }
+    const options = mockEnqueue.mock.calls[0][2] as {
+      runner: () => Promise<void>
+    }
     await options.runner()
     expect(mockEnqueue).toHaveBeenCalledOnce()
     expect(mockEnqueue).toHaveBeenCalledWith(
@@ -97,46 +87,40 @@ describe('QuickBooks webhook ingress job', () => {
   })
 
   it('continues later events before retrying a delivery with failed targets', async () => {
-    mockFindPage
-      .mockResolvedValueOnce({
-        hasMore: false,
-        nextCursor: null,
-        targets: [
-          { webhook: { id: 'w1' }, workflow: { id: 'wf1' } },
-          { webhook: { id: 'w2' }, workflow: { id: 'wf2' } },
-        ],
-      })
-      .mockResolvedValueOnce({
-        hasMore: false,
-        nextCursor: null,
-        targets: [{ webhook: { id: 'w3' }, workflow: { id: 'wf3' } }],
-      })
+    mockFindWebhooks
+      .mockResolvedValueOnce([
+        { webhook: { id: 'w1' }, workflow: { id: 'wf1' } },
+        { webhook: { id: 'w2' }, workflow: { id: 'wf2' } },
+      ])
+      .mockResolvedValueOnce([{ webhook: { id: 'w3' }, workflow: { id: 'wf3' } }])
     mockDispatch
       .mockRejectedValueOnce(new Error('dispatch unavailable'))
       .mockResolvedValueOnce({ outcome: 'failed' })
       .mockResolvedValueOnce({ outcome: 'queued' })
 
     await enqueueQuickBooksWebhookIngress(payload)
-    const options = mockEnqueue.mock.calls[0][2] as { runner: () => Promise<void> }
+    const options = mockEnqueue.mock.calls[0][2] as {
+      runner: () => Promise<void>
+    }
     await expect(options.runner()).rejects.toThrow(
       'QuickBooks webhook delivery completed with 2 failures'
     )
-    expect(mockFindPage).toHaveBeenCalledWith('789', 'request-1', undefined)
+    expect(mockFindWebhooks).toHaveBeenCalledWith('789', 'request-1', 'quickbooks')
     expect(mockDispatch).toHaveBeenCalledTimes(3)
     expect(mockEnqueue).toHaveBeenCalledOnce()
   })
 
-  it('continues later events when a target page cannot be resolved', async () => {
-    mockFindPage
+  it('continues later events when targets cannot be resolved', async () => {
+    mockFindWebhooks
       .mockRejectedValueOnce(new Error('database unavailable'))
-      .mockResolvedValueOnce({ hasMore: false, nextCursor: null, targets: [] })
+      .mockResolvedValueOnce([])
 
     await expect(executeQuickBooksWebhookIngress(payload)).resolves.toEqual({
       failed: 1,
       ignored: 0,
       processed: 0,
       targetCount: 0,
     })
-    expect(mockFindPage).toHaveBeenCalledWith('789', 'request-1', undefined)
+    expect(mockFindWebhooks).toHaveBeenCalledWith('789', 'request-1', 'quickbooks')
   })
 })

apps/sim/background/quickbooks-webhook-ingress.ts

@@ -3,8 +3,7 @@ import { task } from '@trigger.dev/sdk'
 import { NextRequest } from 'next/server'
 import type { QuickBooksWebhookEvent } from '@/lib/api/contracts/webhooks'
 import { getJobQueue } from '@/lib/core/async-jobs'
-import { dispatchResolvedWebhookTarget } from '@/lib/webhooks/processor'
-import { findQuickBooksWebhookTargetPage } from '@/background/quickbooks-webhook-targets'
+import { dispatchResolvedWebhookTarget, findWebhooksByRoutingKey } from '@/lib/webhooks/processor'
 
 const logger = createLogger('QuickBooksWebhookIngressTask')
 
@@ -25,7 +24,7 @@ export interface QuickBooksWebhookIngressResult {
   targetCount: number
 }
 
-/** Process the bounded delivery sequentially, retaining at most one target page at a time. */
+/** Process the bounded delivery sequentially through the shared routing-key dispatcher. */
 export async function executeQuickBooksWebhookIngress(
   payload: QuickBooksWebhookIngressPayload
 ): Promise<QuickBooksWebhookIngressResult> {
@@ -40,61 +39,51 @@ export async function executeQuickBooksWebhookIngress(
       headers: payload.headers,
       body: JSON.stringify(event),
     })
-    let afterWebhookId: string | undefined
 
-    while (true) {
-      try {
-        const page = await findQuickBooksWebhookTargetPage(
-          event.intuitaccountid,
-          payload.requestId,
-          afterWebhookId
-        )
-        const nextCursor = page.hasMore ? page.nextCursor : null
-        if (page.hasMore && (!nextCursor || nextCursor === afterWebhookId)) {
-          throw new Error('QuickBooks webhook target pagination did not advance')
-        }
+    try {
+      const targets = await findWebhooksByRoutingKey(
+        event.intuitaccountid,
+        payload.requestId,
+        'quickbooks'
+      )
+      targetCount += targets.length
 
-        targetCount += page.targets.length
-        for (const { webhook, workflow } of page.targets) {
-          try {
-            const result = await dispatchResolvedWebhookTarget(webhook, workflow, event, request, {
-              requestId: payload.requestId,
-              path: webhook.path ?? undefined,
-              receivedAt: payload.receivedAt,
-              triggerTimestampMs: Date.parse(event.time),
-            })
-            if (result.outcome === 'queued') processed += 1
-            else if (result.outcome === 'ignored') ignored += 1
-            else failed += 1
-          } catch (error) {
-            failed += 1
-            logger.error(`[${payload.requestId}] QuickBooks webhook target dispatch failed`, {
-              error,
-              eventId: event.id,
-              eventIndex,
-              webhookId: webhook.id,
-            })
-          }
+      for (const { webhook, workflow } of targets) {
+        try {
+          const result = await dispatchResolvedWebhookTarget(webhook, workflow, event, request, {
+            requestId: payload.requestId,
+            path: webhook.path ?? undefined,
+            receivedAt: payload.receivedAt,
+            triggerTimestampMs: Date.parse(event.time),
+          })
+          if (result.outcome === 'queued') processed += 1
+          else if (result.outcome === 'ignored') ignored += 1
+          else failed += 1
+        } catch (error) {
+          failed += 1
+          logger.error(`[${payload.requestId}] QuickBooks webhook target dispatch failed`, {
+            error,
+            eventId: event.id,
+            eventIndex,
+            webhookId: webhook.id,
+          })
         }
-
-        logger.info(`[${payload.requestId}] QuickBooks webhook page completed`, {
-          eventId: event.id,
-          eventIndex,
-          ignored,
-          processed,
-          targetCount: page.targets.length,
-        })
-        if (!nextCursor) break
-        afterWebhookId = nextCursor
-      } catch (error) {
-        failed += 1
-        logger.error(`[${payload.requestId}] QuickBooks webhook event page failed`, {
-          error,
-          eventId: event.id,
-          eventIndex,
-        })
-        break
       }
+
+      logger.info(`[${payload.requestId}] QuickBooks webhook event completed`, {
+        eventId: event.id,
+        eventIndex,
+        ignored,
+        processed,
+        targetCount: targets.length,
+      })
+    } catch (error) {
+      failed += 1
+      logger.error(`[${payload.requestId}] QuickBooks webhook target lookup failed`, {
+        error,
+        eventId: event.id,
+        eventIndex,
+      })
     }
   }