feat: Implement passthrough endpoint

apps/web/pages/api/trpc/[...trpc].ts

@@ -1,17 +1,35 @@
 import '@usevenice/app-config/register.node'
 
 import * as trpcNext from '@trpc/server/adapters/next'
+import type {TRPCError} from '@trpc/server'
 import type {NextApiHandler} from 'next'
 
 import {contextFactory} from '@usevenice/app-config/backendConfig'
 import type {Id} from '@usevenice/cdk'
 import type {RouterContext} from '@usevenice/engine-backend'
 import {parseWebhookRequest} from '@usevenice/engine-backend'
-import {fromMaybeArray} from '@usevenice/util'
+import {fromMaybeArray, HTTPError} from '@usevenice/util'
 
 import {appRouter} from '@/lib-server/appRouter'
 import {respondToCORS, serverGetViewer} from '@/lib-server/server-helpers'
 
+/** https://trpc.io/docs/server/error-handling */
+const HTTP_CODE_TO_TRPC_CODE = {
+  400: 'BAD_REQUEST',
+  401: 'UNAUTHORIZED',
+  403: 'FORBIDDEN',
+  404: 'NOT_FOUND',
+  408: 'TIMEOUT',
+  409: 'CONFLICT',
+  412: 'PRECONDITION_FAILED',
+  413: 'PAYLOAD_TOO_LARGE',
+  405: 'METHOD_NOT_SUPPORTED',
+  422: 'UNPROCESSABLE_CONTENT',
+  429: 'TOO_MANY_REQUESTS',
+  499: 'CLIENT_CLOSED_REQUEST',
+  500: 'INTERNAL_SERVER_ERROR',
+} satisfies Record<number, TRPCError['code']>
+
 export const createContext: Parameters<
   typeof trpcNext.createNextApiHandler
 >[0]['createContext'] = async ({req, res}): Promise<RouterContext> => {
@@ -26,8 +44,17 @@ export const createContext: Parameters<
 
 export const onError: Parameters<
   typeof trpcNext.createNextApiHandler
->[0]['onError'] = ({error}) => {
-  console.warn('error', error)
+>[0]['onError'] = ({error, path}) => {
+  // Force passthrough the HTTP error code.
+  if (path === 'passthrough' && error.cause instanceof HTTPError) {
+    const newCode =
+      HTTP_CODE_TO_TRPC_CODE[
+        error.cause.code as keyof typeof HTTP_CODE_TO_TRPC_CODE
+      ]
+    Object.assign(error, {code: newCode ?? error.code})
+  } else {
+    console.warn('error', error)
+  }
 }
 
 const handler = trpcNext.createNextApiHandler({

integrations/integration-plaid/server.ts

@@ -4,11 +4,24 @@ import {CountryCode, Products} from 'plaid'
 
 import type {IntegrationServer} from '@usevenice/cdk'
 import {shouldSync} from '@usevenice/cdk'
-import type {DurationObjectUnits, IAxiosError} from '@usevenice/util'
-import {DateTime, R, RateLimit, Rx, rxjs} from '@usevenice/util'
+import type {
+  DurationObjectUnits,
+  IAxiosError,
+  InfoFromPaths} from '@usevenice/util';
+import {
+  DateTime,
+  makeOpenApiClient,
+  R,
+  RateLimit,
+  Rx,
+  rxjs,
+  safeJSONParse,
+} from '@usevenice/util'
 
 import type {plaidSchemas} from './def'
 import {helpers as def} from './def'
+import {inferPlaidEnvFromToken} from './plaid-utils'
+import type {paths} from './plaid.gen'
 import {makePlaidClient, zWebhook} from './PlaidClient'
 
 export const plaidServerIntegration = {
@@ -463,7 +476,50 @@ export const plaidServerIntegration = {
     console.warn('[plaid] Unhandled webhook', webhook)
     return {resourceUpdates: []}
   },
-} satisfies IntegrationServer<typeof plaidSchemas>
+  newInstance: ({config, settings}) => {
+    const env = inferPlaidEnvFromToken(settings.accessToken)
+    // https://plaid.com/docs/api/#api-host
+    return makeOpenApiClient<InfoFromPaths<paths>>({
+      baseUrl: `https://${env}.plaid.com`,
+      headers: {
+        'PLAID-CLIENT-ID': config.clientId,
+        'PLAID-SECRET': config.clientSecret,
+      },
+      middleware: (url, init) => {
+        if (init?.method?.toLowerCase() === 'post') {
+          const body =
+            typeof init?.body === 'string' ? safeJSONParse(init.body) : {}
+          if (typeof body === 'object') {
+            return [
+              url,
+              {
+                ...init,
+                body: JSON.stringify({
+                  ...body,
+                  access_token: settings.accessToken,
+                }),
+                headers: {
+                  ...init.headers,
+                  'Content-Type': 'application/json',
+                },
+              },
+            ]
+          }
+        }
+        return [url, init]
+      },
+    })
+  },
+  passthrough: (instance, input) =>
+    instance._request(input.method, input.path, {
+      header: input.headers,
+      query: input.query,
+      bodyJson: input.body,
+    }),
+} satisfies IntegrationServer<
+  typeof plaidSchemas,
+  ReturnType<typeof makeOpenApiClient<InfoFromPaths<paths>>>
+>
 
 // Per client limit.
 // TODO: Account for different rate limits for sandbox vs development & prduction

packages/cdk/integration.types.ts

@@ -21,6 +21,7 @@ import type {
   OpenDialogFn,
   ResourceUpdate,
   WebhookReturnType,
+  zPassthroughInput,
 } from './providers.types'
 import type {AccountingMethods, ZAccounting} from './verticals/accounting'
 import type {InvestmentMethods, ZInvestment} from './verticals/investment'
@@ -264,6 +265,11 @@ export interface IntegrationServer<
     ) => MaybePromise<void>
   }) => TInstance
 
+  passthrough?: (
+    instance: TInstance,
+    input: z.infer<typeof zPassthroughInput>,
+  ) => unknown
+
   verticals?: {
     [v in keyof T['_verticals']]: v extends keyof Verticals
       ? Verticals<TDef, TInstance>[v]['methods']

packages/cdk/providers.types.ts

@@ -190,3 +190,11 @@ export interface WebhookReturnType<
     body: Record<string, unknown>
   }
 }
+
+export const zPassthroughInput = z.object({
+  method: z.enum(['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS']),
+  path: z.string(),
+  query: z.record(z.unknown()).optional(),
+  headers: z.record(z.unknown()).optional(),
+  body: z.record(z.unknown()).optional(),
+})

packages/engine-backend/router/_base.ts

@@ -2,6 +2,7 @@ import {initTRPC, TRPCError} from '@trpc/server'
 import type {OpenApiMeta} from 'trpc-openapi'
 
 import {getExtEndUserId, hasRole} from '@usevenice/cdk'
+import {HTTPError} from '@usevenice/util'
 
 import type {RouterContext} from '../context'
 
@@ -23,7 +24,26 @@ export const trpc = initTRPC
   .context<RouterContext>()
   .meta<RouterMeta>()
   // For client side to be able to import runtime schema from server side also
-  .create({allowOutsideOfServer: true})
+  .create({
+    allowOutsideOfServer: true,
+    errorFormatter(opts) {
+      const {shape, error} = opts
+      if (!(error.cause instanceof HTTPError)) {
+        return shape
+      }
+      return {
+        ...shape,
+        data: error.cause.response
+          ? {
+              ...error.cause.response,
+              // Renaming body to be nicer. otherwise we end up with data.data
+              data: undefined,
+              body: error.cause.response.data,
+            }
+          : shape.data,
+      }
+    },
+  })
 
 export const publicProcedure = trpc.procedure
 

packages/engine-backend/router/resourceRouter.ts

@@ -1,11 +1,35 @@
-import {zId} from '@usevenice/cdk'
+import {TRPCError} from '@trpc/server'
+
+import {zId, zPassthroughInput} from '@usevenice/cdk'
 import {Rx, rxjs, z} from '@usevenice/util'
 
-import {protectedProcedure, trpc} from './_base'
+import {protectedProcedure, remoteProcedure, trpc} from './_base'
 
 export {type inferProcedureInput} from '@trpc/server'
 
 export const resourceRouter = trpc.router({
+  // TODO: maybe we should allow resourceId to be part of the path rather than only in the headers
+  passthrough: remoteProcedure
+    .meta({openapi: {method: 'POST', path: '/passthrough'}})
+    .input(zPassthroughInput)
+    .output(z.any())
+    .mutation(async ({input, ctx}) => {
+      if (!ctx.remote.provider.passthrough) {
+        throw new TRPCError({
+          code: 'NOT_IMPLEMENTED',
+          message: `${ctx.remote.providerName} does not implement passthrough`,
+        })
+      }
+
+      // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+      const instance = ctx.remote.provider.newInstance?.({
+        config: ctx.remote.config,
+        settings: ctx.remote.settings,
+        onSettingsChange: () => {}, // not implemented
+      })
+      return await ctx.remote.provider.passthrough(instance, input)
+    }),
+
   sourceSync: protectedProcedure
     .meta({openapi: {method: 'POST', path: '/resources/{id}/source_sync'}})
     .input(z.object({id: zId('reso'), state: z.record(z.unknown()).optional()}))

packages/util/http/makeHttpClient.ts

@@ -38,7 +38,7 @@ export type HttpClientOptions = RequestInit & {
     bearerToken?: string
     basic?: {username: string; password: string}
   }
-
+  middleware?: (...args: Parameters<typeof fetch>) => Parameters<typeof fetch>
   fetch?: typeof fetch
   URL?: typeof URL
 }
@@ -58,6 +58,7 @@ export function makeHttpClient(options: HttpClientOptions) {
     URL = globalThis.URL,
     baseUrl,
     auth,
+    middleware = (url, init) => [url, init],
     ...defaults
   } = options
 
@@ -106,12 +107,15 @@ export function makeHttpClient(options: HttpClientOptions) {
     // NOTE: Implement proxyAgent as a middleware
     // This way we can transparently use reverse proxies also in addition to forward proxy
     // as well as just simple in-app logging.
-    return _fetch(url, {
-      ...defaults,
-      method,
-      headers,
-      body,
-    }).then(async (res) => {
+
+    return _fetch(
+      ...middleware(url, {
+        ...defaults,
+        method,
+        headers,
+        body,
+      }),
+    ).then(async (res) => {
       const text = await res.text()
       const json = safeJSONParse(text)
       if (res.status < 200 || res.status >= 300) {
@@ -136,7 +140,7 @@ export function makeHttpClient(options: HttpClientOptions) {
       request(method.toUpperCase() as Uppercase<typeof method>, path, input),
   ])
 
-  return {...methods, request, _fetch}
+  return {...methods, request, _request: request, _fetch}
 }
 
 // TODO: build url from utils?

packages/util/http/makeOpenApiClient.ts

@@ -82,6 +82,7 @@ export function makeOpenApiClient<Info extends OpenApiInfo>(
       input: Get<Info[M][Path], 'input'>,
     ) => Promise<Get<Info[M][Path], 'output'>>
     _fetch: NonNullable<HttpClientOptions['fetch']>
+    _request: ReturnType<typeof makeHttpClient>['_request']
   }
 }
 

patches/trpc-openapi@1.2.0.patch

@@ -1,3 +1,15 @@
+diff --git a/dist/adapters/node-http/core.js b/dist/adapters/node-http/core.js
+index ad1b3e5d8fc9d5a9dcdcb5c719f2a3a66fc5f29a..d55f2fcff198112d7faca321d886f60eb1514fcc 100644
+--- a/dist/adapters/node-http/core.js
++++ b/dist/adapters/node-http/core.js
+@@ -122,6 +122,7 @@ const createOpenApiNodeHttpHandler = (opts) => {
+             const statusCode = (_h = (_g = meta === null || meta === void 0 ? void 0 : meta.status) !== null && _g !== void 0 ? _g : errors_1.TRPC_ERROR_CODE_HTTP_STATUS[error.code]) !== null && _h !== void 0 ? _h : 500;
+             const headers = (_j = meta === null || meta === void 0 ? void 0 : meta.headers) !== null && _j !== void 0 ? _j : {};
+             const body = {
++                ...errorShape, // Pass the formatting through...
+                 message: isInputValidationError
+                     ? 'Input validation failed'
+                     : (_l = (_k = errorShape === null || errorShape === void 0 ? void 0 : errorShape.message) !== null && _k !== void 0 ? _k : error.message) !== null && _l !== void 0 ? _l : 'An error occurred',
 diff --git a/dist/types.d.ts b/dist/types.d.ts
 index d93e38e020d89bef3e615c44fe687b10c8874417..956b349a0dda132bcdd7c255e8ae9946f41ccffd 100644
 --- a/dist/types.d.ts