feat(quickbooks): add core webhook triggers

apps/sim/.env.example

@@ -125,6 +125,7 @@ CRON_SECRET=your_cron_secret # Use `openssl rand -hex 32` to generate. Authentic
 # QUICKBOOKS_CLIENT_ID=
 # QUICKBOOKS_CLIENT_SECRET=
 # QUICKBOOKS_ENV=sandbox                         # Required when QuickBooks is configured: sandbox or production
+# QUICKBOOKS_WEBHOOK_VERIFIER_TOKEN=              # Verifier token from the Intuit webhook configuration
 
 # Azure Blob Storage takes precedence over S3 if both are configured
 # AZURE_ACCOUNT_NAME=                             # Azure storage account name

apps/sim/app/api/webhooks/quickbooks/route.test.ts

@@ -0,0 +1,100 @@
+/** @vitest-environment node */
+
+import crypto from 'node:crypto'
+import { requestUtilsMockFns, resetEnvMock, setEnv } from '@sim/testing'
+import { NextRequest } from 'next/server'
+import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockEnqueue, mockRelease } = vi.hoisted(() => ({
+  mockEnqueue: vi.fn(),
+  mockRelease: vi.fn(),
+}))
+
+vi.mock('@/background/quickbooks-webhook-ingress', () => ({
+  enqueueQuickBooksWebhookIngress: mockEnqueue,
+}))
+vi.mock('@/lib/core/admission/gate', () => ({
+  admissionRejectedResponse: vi.fn(() => new Response(null, { status: 503 })),
+  tryAdmit: vi.fn(() => ({ release: mockRelease })),
+}))
+vi.mock('@/lib/core/utils/with-route-handler', () => ({
+  withRouteHandler:
+    (handler: (request: NextRequest) => Promise<Response>) => (request: NextRequest) =>
+      handler(request),
+}))
+
+import { POST } from '@/app/api/webhooks/quickbooks/route'
+
+const validEvent = {
+  specversion: '1.0',
+  id: 'event-1',
+  source: 'quickbooks-online',
+  type: 'qbo.invoice.created.v1',
+  time: '2026-08-03T12:00:00Z',
+  intuitentityid: '123',
+  intuitaccountid: '456',
+}
+
+function request(body: string, signature?: string): NextRequest {
+  return new NextRequest('http://localhost/api/webhooks/quickbooks', {
+    method: 'POST',
+    headers: {
+      'content-type': 'application/json',
+      ...(signature ? { 'intuit-signature': signature } : {}),
+    },
+    body,
+  })
+}
+
+function signedRequest(value: unknown): NextRequest {
+  const body = JSON.stringify(value)
+  const signature = crypto.createHmac('sha256', 'verifier').update(body).digest('base64')
+  return request(body, signature)
+}
+
+describe('QuickBooks webhook ingress route', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    setEnv({ QUICKBOOKS_WEBHOOK_VERIFIER_TOKEN: 'verifier' })
+    requestUtilsMockFns.mockGenerateRequestId.mockReturnValue('request-1')
+    mockEnqueue.mockResolvedValue('job-1')
+  })
+  afterAll(() => {
+    resetEnvMock()
+    requestUtilsMockFns.mockGenerateRequestId.mockReset()
+  })
+
+  it('accepts a signed multi-company batch only after durable enqueue', async () => {
+    const response = await POST(
+      signedRequest([validEvent, { ...validEvent, id: 'event-2', intuitaccountid: '789' }])
+    )
+    expect(response.status).toBe(200)
+    expect(mockEnqueue).toHaveBeenCalledWith(
+      expect.objectContaining({
+        events: [validEvent, { ...validEvent, id: 'event-2', intuitaccountid: '789' }],
+        requestId: 'request-1',
+      })
+    )
+    expect(mockRelease).toHaveBeenCalledOnce()
+  })
+
+  it('rejects missing signatures and malformed signed payloads before enqueue', async () => {
+    expect((await POST(request(JSON.stringify([validEvent])))).status).toBe(401)
+    expect((await POST(signedRequest({ invalid: true }))).status).toBe(400)
+    expect(mockEnqueue).not.toHaveBeenCalled()
+  })
+
+  it('rejects batches over the 1,000 event bound', async () => {
+    const events = Array.from({ length: 1001 }, (_, index) => ({
+      ...validEvent,
+      id: `event-${index}`,
+    }))
+    expect((await POST(signedRequest(events))).status).toBe(400)
+    expect(mockEnqueue).not.toHaveBeenCalled()
+  })
+
+  it('returns 503 when durable acceptance fails', async () => {
+    mockEnqueue.mockRejectedValue(new Error('queue unavailable'))
+    expect((await POST(signedRequest([validEvent]))).status).toBe(503)
+  })
+})

apps/sim/app/api/webhooks/quickbooks/route.ts

@@ -0,0 +1,97 @@
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import { type NextRequest, NextResponse } from 'next/server'
+import { quickBooksWebhookEventsSchema } from '@/lib/api/contracts/webhooks'
+import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate'
+import { generateRequestId } from '@/lib/core/utils/request'
+import {
+  assertContentLengthWithinLimit,
+  isPayloadSizeLimitError,
+  readStreamToBufferWithLimit,
+} from '@/lib/core/utils/stream-limits'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { WEBHOOK_MAX_BODY_BYTES } from '@/lib/webhooks/constants'
+import { verifyQuickBooksSignature } from '@/lib/webhooks/providers/quickbooks'
+import {
+  enqueueQuickBooksWebhookIngress,
+  type QuickBooksWebhookIngressPayload,
+} from '@/background/quickbooks-webhook-ingress'
+
+const logger = createLogger('QuickBooksWebhookIngress')
+const BODY_LABEL = 'QuickBooks webhook body'
+
+export const dynamic = 'force-dynamic'
+export const runtime = 'nodejs'
+export const maxDuration = 60
+
+async function readBody(request: Request): Promise<string> {
+  assertContentLengthWithinLimit(request.headers, WEBHOOK_MAX_BODY_BYTES, BODY_LABEL)
+  const buffer = await readStreamToBufferWithLimit(request.body, {
+    maxBytes: WEBHOOK_MAX_BODY_BYTES,
+    label: BODY_LABEL,
+  })
+  return new TextDecoder().decode(buffer)
+}
+
+/** App-level Intuit callback. Verifies raw bytes and durably accepts before fanout. */
+export const POST = withRouteHandler(async (request: NextRequest) => {
+  const ticket = tryAdmit()
+  if (!ticket) return admissionRejectedResponse()
+
+  const requestId = generateRequestId()
+  const receivedAt = Date.now()
+  try {
+    let rawBody: string
+    try {
+      rawBody = await readBody(request)
+    } catch (error) {
+      if (isPayloadSizeLimitError(error)) {
+        return NextResponse.json({ error: 'Request body too large' }, { status: 413 })
+      }
+      throw error
+    }
+
+    const authError = verifyQuickBooksSignature(
+      rawBody,
+      request.headers.get('intuit-signature'),
+      requestId
+    )
+    if (authError) return authError
+
+    let json: unknown
+    try {
+      json = JSON.parse(rawBody)
+    } catch {
+      return NextResponse.json({ error: 'Invalid JSON' }, { status: 400 })
+    }
+    const parsed = quickBooksWebhookEventsSchema.safeParse(json)
+    if (!parsed.success) {
+      logger.warn(`[${requestId}] Invalid QuickBooks webhook envelope`, {
+        issues: parsed.error.issues,
+      })
+      return NextResponse.json({ error: 'Invalid webhook envelope' }, { status: 400 })
+    }
+
+    const payload: QuickBooksWebhookIngressPayload = {
+      events: parsed.data,
+      headers: {
+        'content-type': request.headers.get('content-type') ?? 'application/json',
+      },
+      requestId,
+      receivedAt,
+    }
+    const jobId = await enqueueQuickBooksWebhookIngress(payload)
+    logger.info(`[${requestId}] Accepted QuickBooks webhook delivery`, {
+      eventCount: parsed.data.length,
+      jobId,
+    })
+    return NextResponse.json({ ok: true })
+  } catch (error) {
+    logger.error(`[${requestId}] QuickBooks webhook ingress error`, {
+      error: getErrorMessage(error, 'Unknown error'),
+    })
+    return NextResponse.json({ error: 'Temporarily unable to accept webhook' }, { status: 503 })
+  } finally {
+    ticket.release()
+  }
+})

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

@@ -0,0 +1,126 @@
+/** @vitest-environment node */
+
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockDispatch, mockEnqueue, mockFindWebhooks } = vi.hoisted(() => ({
+  mockDispatch: vi.fn(),
+  mockEnqueue: vi.fn(),
+  mockFindWebhooks: vi.fn(),
+}))
+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 })),
+}))
+
+import {
+  enqueueQuickBooksWebhookIngress,
+  executeQuickBooksWebhookIngress,
+  type QuickBooksWebhookIngressPayload,
+} from '@/background/quickbooks-webhook-ingress'
+
+const event = {
+  specversion: '1.0',
+  id: 'event-1',
+  source: 'quickbooks-online',
+  type: 'qbo.invoice.created.v1',
+  time: '2026-08-03T12:00:00Z',
+  intuitentityid: '123',
+  intuitaccountid: '456',
+}
+const payload: QuickBooksWebhookIngressPayload = {
+  events: [event, { ...event, id: 'event-2', intuitaccountid: '789' }],
+  headers: { 'content-type': 'application/json' },
+  requestId: 'request-1',
+  receivedAt: 1,
+}
+
+describe('QuickBooks webhook ingress job', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockEnqueue.mockResolvedValue('job-1')
+  })
+
+  it('routes the batch by company and dispatches targets sequentially', async () => {
+    const order: string[] = []
+    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: 3,
+      targetCount: 3,
+    })
+    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 () => {
+    mockFindWebhooks.mockResolvedValue([])
+    await enqueueQuickBooksWebhookIngress(payload)
+    const options = mockEnqueue.mock.calls[0][2] as {
+      runner: () => Promise<void>
+    }
+    await options.runner()
+    expect(mockEnqueue).toHaveBeenCalledOnce()
+    expect(mockEnqueue).toHaveBeenCalledWith(
+      'quickbooks-webhook-ingress',
+      payload,
+      expect.objectContaining({
+        jobId: 'quickbooks-webhook-ingress:request-1',
+      })
+    )
+  })
+
+  it('continues later events before retrying a delivery with failed targets', async () => {
+    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>
+    }
+    await expect(options.runner()).rejects.toThrow(
+      'QuickBooks webhook delivery completed with 2 failures'
+    )
+    expect(mockFindWebhooks).toHaveBeenCalledWith('789', 'request-1', 'quickbooks')
+    expect(mockDispatch).toHaveBeenCalledTimes(3)
+    expect(mockEnqueue).toHaveBeenCalledOnce()
+  })
+
+  it('continues later events when targets cannot be resolved', async () => {
+    mockFindWebhooks
+      .mockRejectedValueOnce(new Error('database unavailable'))
+      .mockResolvedValueOnce([])
+
+    await expect(executeQuickBooksWebhookIngress(payload)).resolves.toEqual({
+      failed: 1,
+      ignored: 0,
+      processed: 0,
+      targetCount: 0,
+    })
+    expect(mockFindWebhooks).toHaveBeenCalledWith('789', 'request-1', 'quickbooks')
+  })
+})