refactor(quickbooks): align with shared integration patterns

Author: Bill Leoutsakos

apps/sim/app/api/auth/oauth/utils.test.ts

@@ -7,20 +7,6 @@
 import { redisConfigMockFns } from '@sim/testing'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { capturedLeaderLockOptions } = vi.hoisted(() => ({
-  capturedLeaderLockOptions: [] as Array<Record<string, unknown>>,
-}))
-vi.mock('@/lib/concurrency/leader-lock', async (importOriginal) => {
-  const actual = await importOriginal<typeof import('@/lib/concurrency/leader-lock')>()
-  return {
-    ...actual,
-    withLeaderLock: vi.fn((options) => {
-      capturedLeaderLockOptions.push(options as unknown as Record<string, unknown>)
-      return actual.withLeaderLock(options)
-    }),
-  }
-})
-
 vi.mock('@/lib/oauth/oauth', () => ({
   refreshOAuthToken: vi.fn(),
   OAUTH_PROVIDERS: {},
@@ -84,7 +70,6 @@ function mockUpdateChain() {
 describe('OAuth Utils', () => {
   beforeEach(() => {
     vi.clearAllMocks()
-    capturedLeaderLockOptions.length = 0
     __resetCoalesceLocallyForTests()
     redisConfigMockFns.mockGetRedisClient.mockReturnValue(null)
     redisConfigMockFns.mockAcquireLock.mockResolvedValue(true)
@@ -435,62 +420,6 @@ describe('OAuth Utils', () => {
     })
   })
 
-  describe('QuickBooks refresh locking', () => {
-    it('keeps the lock and follower wait alive beyond the provider timeout', async () => {
-      const credential = {
-        id: 'quickbooks-row',
-        accessToken: 'expired-token',
-        refreshToken: 'rotating-refresh-token',
-        accessTokenExpiresAt: new Date(Date.now() - 3600 * 1000),
-        providerId: 'quickbooks',
-      }
-      mockRefreshOAuthToken.mockResolvedValueOnce({
-        ok: true,
-        accessToken: 'new-token',
-        expiresIn: 3600,
-        refreshToken: 'new-rotating-refresh-token',
-      })
-      mockUpdateChain()
-
-      const result = await refreshTokenIfNeeded('request-id', credential, credential.id)
-
-      expect(result).toEqual({ accessToken: 'new-token', refreshed: true })
-      expect(redisConfigMockFns.mockAcquireLock).toHaveBeenCalledWith(
-        'oauth:refresh:quickbooks-row',
-        expect.any(String),
-        30
-      )
-      expect(capturedLeaderLockOptions[0]).toMatchObject({ ttlSec: 30, maxWaitMs: 30_000 })
-    })
-
-    it('keeps the default refresh-lock budget for other providers', async () => {
-      const credential = {
-        id: 'google-row',
-        accessToken: 'expired-token',
-        refreshToken: 'refresh-token',
-        accessTokenExpiresAt: new Date(Date.now() - 3600 * 1000),
-        providerId: 'google',
-      }
-      mockRefreshOAuthToken.mockResolvedValueOnce({
-        ok: true,
-        accessToken: 'new-token',
-        expiresIn: 3600,
-        refreshToken: 'new-refresh-token',
-      })
-      mockUpdateChain()
-
-      await refreshTokenIfNeeded('request-id', credential, credential.id)
-
-      expect(redisConfigMockFns.mockAcquireLock).toHaveBeenCalledWith(
-        'oauth:refresh:google-row',
-        expect.any(String),
-        10
-      )
-      expect(capturedLeaderLockOptions[0]).not.toHaveProperty('ttlSec')
-      expect(capturedLeaderLockOptions[0]).not.toHaveProperty('maxWaitMs')
-    })
-  })
-
   describe('resolveServiceAccountToken', () => {
     it('throws loudly for an unknown provider (never silently attempts Google)', async () => {
       await expect(resolveServiceAccountToken('cred-1', 'mystery-provider')).rejects.toThrow(

apps/sim/app/api/auth/oauth/utils.ts

@@ -701,8 +701,6 @@ interface CoalescedRefreshOptions {
  */
 const SLACK_LOCK_TTL_SEC = 30
 const SLACK_FOLLOWER_MAX_WAIT_MS = SLACK_LOCK_TTL_SEC * 1000
-const QUICKBOOKS_LOCK_TTL_SEC = 30
-const QUICKBOOKS_FOLLOWER_MAX_WAIT_MS = QUICKBOOKS_LOCK_TTL_SEC * 1000
 
 async function performCoalescedRefresh({
   accountId,
@@ -718,7 +716,6 @@ async function performCoalescedRefresh({
    * dead-flagged, and written per installation rather than per row.
    */
   const slackTeamId = isSlackProvider(providerId) ? extractSlackTeamId(providerAccountId) : null
-  const isQuickBooks = providerId === 'quickbooks'
   const scopeKey = slackTeamId ? `slack:${slackTeamId}` : accountId
 
   const logContext = {
@@ -747,14 +744,7 @@ async function performCoalescedRefresh({
       // so their wait and the lock TTL must outlast the 15s provider timeout —
       // the 3s/10s defaults would fail followers early and let a second leader
       // start a concurrent rotation mid-refresh.
-      ...(slackTeamId
-        ? { maxWaitMs: SLACK_FOLLOWER_MAX_WAIT_MS, ttlSec: SLACK_LOCK_TTL_SEC }
-        : isQuickBooks
-          ? {
-              maxWaitMs: QUICKBOOKS_FOLLOWER_MAX_WAIT_MS,
-              ttlSec: QUICKBOOKS_LOCK_TTL_SEC,
-            }
-          : {}),
+      ...(slackTeamId ? { maxWaitMs: SLACK_FOLLOWER_MAX_WAIT_MS, ttlSec: SLACK_LOCK_TTL_SEC } : {}),
       onLeader: async () => {
         try {
           let refreshTokenToUse = refreshToken

apps/sim/app/api/tools/quickbooks/download-document/route.ts

@@ -19,7 +19,8 @@ import {
   readResponseToBufferWithLimit,
 } from '@/lib/core/utils/stream-limits'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { storeToolOutputFile } from '@/lib/uploads/store-tool-output-file'
+import { uploadCopilotFile } from '@/lib/uploads/contexts/copilot'
+import { uploadExecutionFile } from '@/lib/uploads/contexts/execution'
 import { buildQuickBooksCompanyUrl, buildQuickBooksHeaders } from '@/tools/quickbooks/client'
 import {
   getQuickBooksDocumentError,
@@ -203,18 +204,29 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
         : await downloadQuickBooksTransactionPdf(body, request.signal)
 
     request.signal.throwIfAborted()
+    const executionContext =
+      body.workspaceId && body.workflowId && body.executionId
+        ? {
+            workspaceId: body.workspaceId,
+            workflowId: body.workflowId,
+            executionId: body.executionId,
+          }
+        : null
     const storedFile = userFileSchema.parse(
-      await storeToolOutputFile({
-        buffer: downloaded.buffer,
-        fileName: downloaded.fileName,
-        contentType: downloaded.mimeType,
-        userId: authResult.userId,
-        context: {
-          workspaceId: body.workspaceId,
-          workflowId: body.workflowId,
-          executionId: body.executionId,
-        },
-      })
+      executionContext
+        ? await uploadExecutionFile(
+            executionContext,
+            downloaded.buffer,
+            downloaded.fileName,
+            downloaded.mimeType,
+            authResult.userId
+          )
+        : await uploadCopilotFile({
+            buffer: downloaded.buffer,
+            fileName: downloaded.fileName,
+            contentType: downloaded.mimeType,
+            userId: authResult.userId,
+          })
     )
 
     const shared = {

apps/sim/lib/auth/connectors/providers.ts

@@ -8,7 +8,6 @@ import { syntheticConnectorEmail } from '@/lib/auth/connector-email'
 import { env } from '@/lib/core/config/env'
 import { inspectConfiguredOAuthClient } from '@/lib/core/config/env-capabilities.server'
 import {
-  DEFAULT_MAX_ERROR_BODY_BYTES,
   readResponseJsonWithLimit,
   readResponseTextWithLimit,
 } from '@/lib/core/utils/stream-limits'
@@ -22,7 +21,6 @@ import {
   QUICKBOOKS_TOKEN_URL,
 } from '@/lib/oauth/quickbooks'
 import { getCanonicalScopesForProvider } from '@/lib/oauth/utils'
-import { QUICKBOOKS_OAUTH_REQUEST_TIMEOUT_MS } from '@/tools/quickbooks/client'
 import { deriveZohoDeskBaseFromApiDomain } from '@/tools/zoho_desk/host-allowlist'
 
 /**
@@ -2377,57 +2375,11 @@ export function buildConnectorProviders(): GenericOAuthConfig[] {
       responseType: 'code',
       accessType: 'offline',
       prompt: 'consent',
+      authentication: 'basic',
       redirectURI: `${getBaseUrl()}/api/auth/oauth2/callback/quickbooks`,
       authorizationUrlParams: {
         claims: JSON.stringify(QUICKBOOKS_OIDC_CLAIMS),
       },
-      getToken: async ({ code, redirectURI }) => {
-        const clientId = env.QUICKBOOKS_CLIENT_ID
-        const clientSecret = env.QUICKBOOKS_CLIENT_SECRET
-        if (!clientId || !clientSecret) {
-          throw new Error('QuickBooks OAuth client credentials are not configured')
-        }
-
-        const response = await fetch(QUICKBOOKS_TOKEN_URL, {
-          method: 'POST',
-          headers: {
-            Authorization: `Basic ${Buffer.from(`${clientId}:${clientSecret}`).toString('base64')}`,
-            'Content-Type': 'application/x-www-form-urlencoded',
-          },
-          body: new URLSearchParams({
-            code,
-            grant_type: 'authorization_code',
-            redirect_uri: redirectURI,
-          }),
-          signal: AbortSignal.timeout(QUICKBOOKS_OAUTH_REQUEST_TIMEOUT_MS),
-        })
-        if (!response.ok) {
-          await readResponseTextWithLimit(response, {
-            maxBytes: DEFAULT_MAX_ERROR_BODY_BYTES,
-            label: 'QuickBooks OAuth token error response',
-          }).catch(() => {})
-          throw new Error(`QuickBooks OAuth token exchange failed with HTTP ${response.status}`)
-        }
-
-        const data = await readResponseJsonWithLimit<Record<string, unknown>>(response, {
-          maxBytes: DEFAULT_MAX_ERROR_BODY_BYTES,
-          label: 'QuickBooks OAuth token response',
-        })
-        if (!data || typeof data !== 'object' || Array.isArray(data)) {
-          throw new Error('QuickBooks OAuth token exchange returned an invalid response')
-        }
-
-        const tokens = getOAuth2Tokens(data)
-        if (!tokens.accessToken || !tokens.refreshToken) {
-          throw new Error(
-            'QuickBooks OAuth token response did not include access and refresh tokens'
-          )
-        }
-        if (typeof data.scope === 'string') {
-          tokens.scopes = data.scope.split(/\s+/).filter(Boolean)
-        }
-        return tokens
-      },
       getUserInfo: async (tokens) => {
         if (!tokens.accessToken) {
           throw new Error('QuickBooks OAuth did not issue an access token')

apps/sim/lib/core/security/input-validation.server.ts

@@ -966,7 +966,6 @@ export async function secureFetchWithPinnedIP(
   options: SecureFetchOptions & { allowHttp?: boolean } = {},
   redirectCount = 0
 ): Promise<SecureFetchResponse> {
-  options.signal?.throwIfAborted()
   const maxRedirects = options.maxRedirects ?? DEFAULT_MAX_REDIRECTS
   const requestedMaxResponseBytes = options.maxResponseBytes
   const maxResponseBytes =
@@ -1015,7 +1014,6 @@ export async function secureFetchWithPinnedIP(
 
         validateUrlWithDNS(redirectUrl, 'redirectUrl', { allowHttp: options.allowHttp })
           .then((validation) => {
-            options.signal?.throwIfAborted()
             if (!validation.isValid) {
               settledReject(new Error(`Redirect blocked: ${validation.error}`))
               return

apps/sim/lib/core/security/secure-fetch-response-cap.server.test.ts

@@ -16,18 +16,15 @@ vi.mock('@/lib/core/config/env-flags', () => ({
   getProxyUrl: () => undefined,
 }))
 
-import { resolveHostAddresses } from '@sim/security/dns'
 import {
   DEFAULT_MAX_RESPONSE_BYTES,
   secureFetchWithPinnedIP,
 } from '@/lib/core/security/input-validation.server'
 
 const servers: http.Server[] = []
-const mockResolveHostAddresses = vi.mocked(resolveHostAddresses)
 
 afterEach(() => {
   for (const server of servers.splice(0)) server.close()
-  vi.clearAllMocks()
 })
 
 /** Starts a throwaway loopback server and returns its origin. */
@@ -39,49 +36,6 @@ async function startServer(handler: http.RequestListener): Promise<string> {
 }
 
 describe('secureFetchWithPinnedIP response cap', () => {
-  it('does not open a request when the signal is already aborted', async () => {
-    let requests = 0
-    const origin = await startServer((_req, res) => {
-      requests += 1
-      res.end('unexpected')
-    })
-    const controller = new AbortController()
-    controller.abort(new Error('cancelled'))
-
-    await expect(
-      secureFetchWithPinnedIP(origin, '127.0.0.1', {
-        allowHttp: true,
-        signal: controller.signal,
-      })
-    ).rejects.toThrow('cancelled')
-    expect(requests).toBe(0)
-  })
-
-  it('does not follow a redirect when cancellation arrives during redirect DNS validation', async () => {
-    let targetRequests = 0
-    const targetOrigin = await startServer((_req, res) => {
-      targetRequests += 1
-      res.end('unexpected')
-    })
-    const redirectOrigin = await startServer((_req, res) => {
-      res.writeHead(302, { Location: targetOrigin.replace('127.0.0.1', 'localhost') })
-      res.end()
-    })
-    const controller = new AbortController()
-    mockResolveHostAddresses.mockImplementationOnce(async () => {
-      controller.abort(new Error('cancelled during redirect DNS'))
-      return { addresses: ['127.0.0.1'], preferred: '127.0.0.1' }
-    })
-
-    await expect(
-      secureFetchWithPinnedIP(redirectOrigin, '127.0.0.1', {
-        allowHttp: true,
-        signal: controller.signal,
-      })
-    ).rejects.toThrow('cancelled during redirect DNS')
-    expect(targetRequests).toBe(0)
-  })
-
   it('rejects a body that exceeds an explicit cap instead of buffering it', async () => {
     const origin = await startServer((_req, res) => {
       res.writeHead(200, { 'Content-Type': 'application/octet-stream' })