Skip to content

Commit 28c04b1

Browse files
Bill LeoutsakosBill Leoutsakos
Bill Leoutsakos
authored and
Bill Leoutsakos
committed
fix(quickbooks): bound webhook ingress jobs
1 parent 2982466 commit 28c04b1

2 files changed

Lines changed: 138 additions & 113 deletions

File tree

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

Lines changed: 67 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -44,78 +44,99 @@ describe('QuickBooks webhook ingress job', () => {
4444
mockEnqueue.mockResolvedValue('job-1')
4545
})
4646

47-
it('routes one event to one company and dispatches targets sequentially', async () => {
47+
it('routes the batch by company and dispatches targets sequentially', async () => {
4848
const order: string[] = []
49-
mockFindPage.mockResolvedValue({
50-
hasMore: false,
51-
nextCursor: null,
52-
targets: [
53-
{ webhook: { id: 'w1' }, workflow: { id: 'wf1' } },
54-
{ webhook: { id: 'w2' }, workflow: { id: 'wf2' } },
55-
],
56-
})
49+
mockFindPage
50+
.mockResolvedValueOnce({
51+
hasMore: true,
52+
nextCursor: 'w2',
53+
targets: [
54+
{ webhook: { id: 'w1' }, workflow: { id: 'wf1' } },
55+
{ webhook: { id: 'w2' }, workflow: { id: 'wf2' } },
56+
],
57+
})
58+
.mockResolvedValueOnce({
59+
hasMore: false,
60+
nextCursor: null,
61+
targets: [{ webhook: { id: 'w3' }, workflow: { id: 'wf3' } }],
62+
})
63+
.mockResolvedValueOnce({
64+
hasMore: false,
65+
nextCursor: null,
66+
targets: [{ webhook: { id: 'w4' }, workflow: { id: 'wf4' } }],
67+
})
5768
mockDispatch.mockImplementation(async (webhook: { id: string }) => {
5869
order.push(webhook.id)
5970
return { outcome: 'queued' }
6071
})
6172
await expect(executeQuickBooksWebhookIngress(payload)).resolves.toEqual({
6273
failed: 0,
6374
ignored: 0,
64-
processed: 2,
65-
targetCount: 2,
75+
processed: 4,
76+
targetCount: 4,
6677
})
6778
expect(mockFindPage).toHaveBeenCalledWith('456', 'request-1', undefined)
68-
expect(order).toEqual(['w1', 'w2'])
79+
expect(mockFindPage).toHaveBeenCalledWith('456', 'request-1', 'w2')
80+
expect(mockFindPage).toHaveBeenCalledWith('789', 'request-1', undefined)
81+
expect(order).toEqual(['w1', 'w2', 'w3', 'w4'])
6982
})
7083

71-
it('continues target pages before advancing to the next event', async () => {
72-
mockFindPage.mockResolvedValue({
73-
hasMore: true,
74-
nextCursor: 'webhook-100',
75-
targets: [],
76-
})
84+
it('enqueues the bounded delivery once without copying it into continuation jobs', async () => {
85+
mockFindPage.mockResolvedValue({ hasMore: false, nextCursor: null, targets: [] })
7786
await enqueueQuickBooksWebhookIngress(payload)
7887
const options = mockEnqueue.mock.calls[0][2] as { runner: () => Promise<void> }
7988
await options.runner()
80-
expect(mockEnqueue).toHaveBeenNthCalledWith(
81-
2,
89+
expect(mockEnqueue).toHaveBeenCalledOnce()
90+
expect(mockEnqueue).toHaveBeenCalledWith(
8291
'quickbooks-webhook-ingress',
83-
expect.objectContaining({ afterWebhookId: 'webhook-100' }),
92+
payload,
8493
expect.objectContaining({
85-
jobId: 'quickbooks-webhook-ingress:request-1:0:webhook-100',
94+
jobId: 'quickbooks-webhook-ingress:request-1',
8695
})
8796
)
8897
})
8998

90-
it('advances to the next event only after the current event finishes', async () => {
91-
mockFindPage.mockResolvedValue({ hasMore: false, nextCursor: null, targets: [] })
99+
it('continues later events before retrying a delivery with failed targets', async () => {
100+
mockFindPage
101+
.mockResolvedValueOnce({
102+
hasMore: false,
103+
nextCursor: null,
104+
targets: [
105+
{ webhook: { id: 'w1' }, workflow: { id: 'wf1' } },
106+
{ webhook: { id: 'w2' }, workflow: { id: 'wf2' } },
107+
],
108+
})
109+
.mockResolvedValueOnce({
110+
hasMore: false,
111+
nextCursor: null,
112+
targets: [{ webhook: { id: 'w3' }, workflow: { id: 'wf3' } }],
113+
})
114+
mockDispatch
115+
.mockRejectedValueOnce(new Error('dispatch unavailable'))
116+
.mockResolvedValueOnce({ outcome: 'failed' })
117+
.mockResolvedValueOnce({ outcome: 'queued' })
118+
92119
await enqueueQuickBooksWebhookIngress(payload)
93120
const options = mockEnqueue.mock.calls[0][2] as { runner: () => Promise<void> }
94-
await options.runner()
95-
expect(mockEnqueue).toHaveBeenNthCalledWith(
96-
2,
97-
'quickbooks-webhook-ingress',
98-
expect.objectContaining({ eventIndex: 1, afterWebhookId: undefined }),
99-
expect.objectContaining({ jobId: 'quickbooks-webhook-ingress:request-1:1:root' })
121+
await expect(options.runner()).rejects.toThrow(
122+
'QuickBooks webhook delivery completed with 2 failures'
100123
)
124+
expect(mockFindPage).toHaveBeenCalledWith('789', 'request-1', undefined)
125+
expect(mockDispatch).toHaveBeenCalledTimes(3)
126+
expect(mockEnqueue).toHaveBeenCalledOnce()
101127
})
102128

103-
it('durably continues the batch before retrying a failed target page', async () => {
104-
mockFindPage.mockResolvedValue({
105-
hasMore: false,
106-
nextCursor: null,
107-
targets: [{ webhook: { id: 'w1' }, workflow: { id: 'wf1' } }],
108-
})
109-
mockDispatch.mockResolvedValue({ outcome: 'failed' })
129+
it('continues later events when a target page cannot be resolved', async () => {
130+
mockFindPage
131+
.mockRejectedValueOnce(new Error('database unavailable'))
132+
.mockResolvedValueOnce({ hasMore: false, nextCursor: null, targets: [] })
110133

111-
await enqueueQuickBooksWebhookIngress(payload)
112-
const options = mockEnqueue.mock.calls[0][2] as { runner: () => Promise<void> }
113-
await expect(options.runner()).rejects.toThrow('Failed to dispatch 1 of 1 QuickBooks targets')
114-
expect(mockEnqueue).toHaveBeenNthCalledWith(
115-
2,
116-
'quickbooks-webhook-ingress',
117-
expect.objectContaining({ eventIndex: 1, afterWebhookId: undefined }),
118-
expect.objectContaining({ jobId: 'quickbooks-webhook-ingress:request-1:1:root' })
119-
)
134+
await expect(executeQuickBooksWebhookIngress(payload)).resolves.toEqual({
135+
failed: 1,
136+
ignored: 0,
137+
processed: 0,
138+
targetCount: 0,
139+
})
140+
expect(mockFindPage).toHaveBeenCalledWith('789', 'request-1', undefined)
120141
})
121142
})

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

Lines changed: 71 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,6 @@ export const QUICKBOOKS_WEBHOOK_INGRESS_CONCURRENCY_LIMIT = 50
1212
export const QUICKBOOKS_WEBHOOK_INGRESS_MAX_ATTEMPTS = 3
1313

1414
export interface QuickBooksWebhookIngressPayload {
15-
afterWebhookId?: string
16-
eventIndex?: number
1715
events: QuickBooksWebhookEvent[]
1816
headers: { 'content-type': string }
1917
requestId: string
@@ -23,102 +21,108 @@ export interface QuickBooksWebhookIngressPayload {
2321
export interface QuickBooksWebhookIngressResult {
2422
failed: number
2523
ignored: number
26-
nextCursor?: string
2724
processed: number
2825
targetCount: number
2926
}
3027

31-
/** Process one event against one bounded target page. */
28+
/** Process the bounded delivery sequentially, retaining at most one target page at a time. */
3229
export async function executeQuickBooksWebhookIngress(
3330
payload: QuickBooksWebhookIngressPayload
3431
): Promise<QuickBooksWebhookIngressResult> {
35-
const eventIndex = payload.eventIndex ?? 0
36-
const event = payload.events[eventIndex]
37-
if (!event) return { failed: 0, ignored: 0, processed: 0, targetCount: 0 }
38-
39-
const request = new NextRequest('http://internal/api/webhooks/quickbooks', {
40-
method: 'POST',
41-
headers: payload.headers,
42-
body: JSON.stringify(event),
43-
})
44-
const page = await findQuickBooksWebhookTargetPage(
45-
event.intuitaccountid,
46-
payload.requestId,
47-
payload.afterWebhookId
48-
)
49-
const nextCursor = page.hasMore ? page.nextCursor : null
50-
if (page.hasMore && (!nextCursor || nextCursor === payload.afterWebhookId)) {
51-
throw new Error('QuickBooks webhook target pagination did not advance')
52-
}
53-
5432
let ignored = 0
5533
let processed = 0
5634
let failed = 0
57-
for (const { webhook, workflow } of page.targets) {
58-
const result = await dispatchResolvedWebhookTarget(webhook, workflow, event, request, {
59-
requestId: payload.requestId,
60-
path: webhook.path ?? undefined,
61-
receivedAt: payload.receivedAt,
62-
triggerTimestampMs: Date.parse(event.time),
35+
let targetCount = 0
36+
37+
for (const [eventIndex, event] of payload.events.entries()) {
38+
const request = new NextRequest('http://internal/api/webhooks/quickbooks', {
39+
method: 'POST',
40+
headers: payload.headers,
41+
body: JSON.stringify(event),
6342
})
64-
if (result.outcome === 'queued') processed += 1
65-
else if (result.outcome === 'ignored') ignored += 1
66-
else failed += 1
43+
let afterWebhookId: string | undefined
44+
45+
while (true) {
46+
try {
47+
const page = await findQuickBooksWebhookTargetPage(
48+
event.intuitaccountid,
49+
payload.requestId,
50+
afterWebhookId
51+
)
52+
const nextCursor = page.hasMore ? page.nextCursor : null
53+
if (page.hasMore && (!nextCursor || nextCursor === afterWebhookId)) {
54+
throw new Error('QuickBooks webhook target pagination did not advance')
55+
}
56+
57+
targetCount += page.targets.length
58+
for (const { webhook, workflow } of page.targets) {
59+
try {
60+
const result = await dispatchResolvedWebhookTarget(webhook, workflow, event, request, {
61+
requestId: payload.requestId,
62+
path: webhook.path ?? undefined,
63+
receivedAt: payload.receivedAt,
64+
triggerTimestampMs: Date.parse(event.time),
65+
})
66+
if (result.outcome === 'queued') processed += 1
67+
else if (result.outcome === 'ignored') ignored += 1
68+
else failed += 1
69+
} catch (error) {
70+
failed += 1
71+
logger.error(`[${payload.requestId}] QuickBooks webhook target dispatch failed`, {
72+
error,
73+
eventId: event.id,
74+
eventIndex,
75+
webhookId: webhook.id,
76+
})
77+
}
78+
}
79+
80+
logger.info(`[${payload.requestId}] QuickBooks webhook page completed`, {
81+
eventId: event.id,
82+
eventIndex,
83+
ignored,
84+
processed,
85+
targetCount: page.targets.length,
86+
})
87+
if (!nextCursor) break
88+
afterWebhookId = nextCursor
89+
} catch (error) {
90+
failed += 1
91+
logger.error(`[${payload.requestId}] QuickBooks webhook event page failed`, {
92+
error,
93+
eventId: event.id,
94+
eventIndex,
95+
})
96+
break
97+
}
98+
}
6799
}
68100

69-
logger.info(`[${payload.requestId}] QuickBooks webhook page completed`, {
70-
eventId: event.id,
71-
eventIndex,
72-
ignored,
73-
processed,
74-
targetCount: page.targets.length,
75-
})
76-
return {
101+
logger.info(`[${payload.requestId}] QuickBooks webhook delivery completed`, {
102+
eventCount: payload.events.length,
77103
failed,
78104
ignored,
79105
processed,
80-
targetCount: page.targets.length,
81-
...(nextCursor ? { nextCursor } : {}),
82-
}
83-
}
84-
85-
async function enqueueQuickBooksWebhookContinuation(
86-
payload: QuickBooksWebhookIngressPayload,
87-
result: QuickBooksWebhookIngressResult
88-
): Promise<void> {
89-
const eventIndex = payload.eventIndex ?? 0
90-
if (result.nextCursor) {
91-
await enqueueQuickBooksWebhookIngress({ ...payload, afterWebhookId: result.nextCursor })
92-
} else if (eventIndex + 1 < payload.events.length) {
93-
await enqueueQuickBooksWebhookIngress({
94-
...payload,
95-
eventIndex: eventIndex + 1,
96-
afterWebhookId: undefined,
97-
})
98-
}
106+
targetCount,
107+
})
108+
return { failed, ignored, processed, targetCount }
99109
}
100110

101111
async function runQuickBooksWebhookIngressJob(
102112
payload: QuickBooksWebhookIngressPayload
103113
): Promise<void> {
104114
const result = await executeQuickBooksWebhookIngress(payload)
105-
// The continuation has a deterministic job id, so retries cannot duplicate it. Enqueue it
106-
// before retrying this page to avoid stranding later events in an already-acknowledged batch.
107-
await enqueueQuickBooksWebhookContinuation(payload, result)
108115
if (result.failed > 0) {
109-
throw new Error(
110-
`Failed to dispatch ${result.failed} of ${result.targetCount} QuickBooks targets`
111-
)
116+
throw new Error(`QuickBooks webhook delivery completed with ${result.failed} failures`)
112117
}
113118
}
114119

115120
export async function enqueueQuickBooksWebhookIngress(
116121
payload: QuickBooksWebhookIngressPayload
117122
): Promise<string> {
118123
const jobQueue = await getJobQueue()
119-
const eventIndex = payload.eventIndex ?? 0
120124
return jobQueue.enqueue('quickbooks-webhook-ingress', payload, {
121-
jobId: `quickbooks-webhook-ingress:${payload.requestId}:${eventIndex}:${payload.afterWebhookId ?? 'root'}`,
125+
jobId: `quickbooks-webhook-ingress:${payload.requestId}`,
122126
maxAttempts: QUICKBOOKS_WEBHOOK_INGRESS_MAX_ATTEMPTS,
123127
concurrencyKey: 'quickbooks-webhook-ingress',
124128
concurrencyLimit: QUICKBOOKS_WEBHOOK_INGRESS_CONCURRENCY_LIMIT,

0 commit comments

Comments
 (0)