feat: add xero sync (#56)

* feat(wip): dummy xero connector implementation

* remove xero-sdk reference till publish

* update connector image on dashboard

* feat: (wip) add sync logic for accounts

* upgrade sdk-xero

* revert: changes that are were needed

* feat: remove hard code in xero connector

* feat(xero): listCategories, sourceSync account and category

* feat(xero): bank transactions ync

---------

Co-authored-by: Tony Xiao <tonyx.ca@gmail.com>

apps/cli/package.json

@@ -27,6 +27,7 @@
     "@usevenice/connector-toggl": "workspace:*",
     "@usevenice/connector-venmo": "workspace:*",
     "@usevenice/connector-wise": "workspace:*",
+    "@usevenice/connector-xero": "workspace:*",
     "@usevenice/connector-yodlee": "workspace:*",
     "@usevenice/engine-backend": "workspace:*",
     "@usevenice/env": "workspace:*",

connectors/connector-xero/def.ts

@@ -1,21 +1,32 @@
-import type {
-  ConnectorDef,
-  ConnectorSchemas,
-  EntityPayloadWithRaw,
-} from '@usevenice/cdk'
-import {connHelpers, oauthBaseSchema} from '@usevenice/cdk'
-import {z, zCast} from '@usevenice/util'
+import type {components} from '@opensdks/sdk-xero/xero_accounting.oas.types'
+import type {ConnectorDef, ConnectorSchemas} from '@usevenice/cdk'
+import {connHelpers, oauthBaseSchema, zEntityPayload} from '@usevenice/cdk'
+import {R, z} from '@usevenice/util'
+
+export const zConfig = oauthBaseSchema.connectorConfig
+
+const oReso = oauthBaseSchema.resourceSettings
+export const zSettings = oReso.extend({
+  oauth: oReso.shape.oauth,
+})
+
+export type XERO = components['schemas']
+
+export const XERO_ENTITY_NAME = {
+  Account: 'Account',
+  BankTransaction: 'BankTransaction',
+} as const
 
 export const xeroSchemas = {
   name: z.literal('xero'),
-  resourceSettings: z.object({
-    access_token: z.string(),
-  }),
-  destinationInputEntity: zCast<EntityPayloadWithRaw>(),
-  connectorConfig: oauthBaseSchema.connectorConfig,
+  connectorConfig: zConfig,
+  resourceSettings: zSettings,
+  connectOutput: oauthBaseSchema.connectOutput,
+  sourceOutputEntity: zEntityPayload,
+  sourceOutputEntities: R.mapValues(XERO_ENTITY_NAME, () => z.unknown()),
 } satisfies ConnectorSchemas
 
-export const helpers = connHelpers(xeroSchemas)
+export const xeroHelpers = connHelpers(xeroSchemas)
 
 export const xeroDef = {
   metadata: {

connectors/connector-xero/index.ts

@@ -1,4 +1,10 @@
+import type {initXeroSDK} from '@opensdks/sdk-xero'
+
 // codegen:start {preset: barrel, include: "./{*.{ts,tsx},*/index.{ts,tsx}}", exclude: "./**/*.{d,spec,test,fixture,gen,node}.{ts,tsx}"}
 export * from './def'
 export * from './server'
 // codegen:end
+
+export * from '@opensdks/sdk-xero'
+
+export type XeroSDK = ReturnType<typeof initXeroSDK>

connectors/connector-xero/package.json

@@ -5,7 +5,8 @@
   "sideEffects": [],
   "module": "./index.ts",
   "dependencies": {
-    "@opensdks/sdk-xero": "^0.0.2",
+    "@opensdks/runtime": "^0.0.19",
+    "@opensdks/sdk-xero": "^0.0.7",
     "@opensdks/util-zod": "^0.0.15",
     "@usevenice/cdk": "workspace:*",
     "@usevenice/util": "workspace:*"

connectors/connector-xero/server.ts

@@ -1,16 +1,83 @@
 import {initXeroSDK} from '@opensdks/sdk-xero'
 import type {ConnectorServer} from '@usevenice/cdk'
-import type {xeroSchemas} from './def'
+import {nangoProxyLink} from '@usevenice/cdk'
+import {Rx, rxjs} from '@usevenice/util'
+import {XERO_ENTITY_NAME, xeroHelpers, type xeroSchemas} from './def'
 
 export const xeroServer = {
-  newInstance: ({settings}) => {
-    // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+  // Would be good if this was async...
+  newInstance: ({settings, fetchLinks}) => {
     const xero = initXeroSDK({
-      headers: {authorization: `Bearer ${settings.access_token}`},
+      headers: {
+        authorization: `Bearer ${settings.oauth.credentials.access_token}`,
+      },
+      links: (defaultLinks) => [
+        (req, next) => {
+          req.headers.set(
+            nangoProxyLink.kBaseUrlOverride,
+            'https://api.xero.com',
+          )
+          // nango's proxy endpoint is pretty annoying... Will only proxy
+          // if it is prefixed with nango-proxy. Might as well not proxy like this...
+          const tenantId = req.headers.get('xero-tenant-id')
+          if (tenantId) {
+            req.headers.delete('xero-tenant-id')
+            req.headers.set('nango-proxy-xero-tenant-id', tenantId)
+          }
+          return next(req)
+        },
+        ...fetchLinks,
+        ...defaultLinks,
+      ],
     })
-    // TODO(@jatin): Add logic here to handle sync.
     return xero
   },
-} satisfies ConnectorServer<typeof xeroSchemas>
+  sourceSync: ({instance: xero, streams}) => {
+    console.log('[xero] Starting sync')
+    async function* iterateEntities() {
+      // TODO: Should handle more than one tenant Id
+      const tenantId = await xero.identity
+        .GET('/Connections')
+        .then((r) => r.data?.[0]?.tenantId)
+      if (!tenantId) {
+        throw new Error(
+          'Missing access to any tenants. Check xero token permission',
+        )
+      }
+      for (const type of Object.values(XERO_ENTITY_NAME)) {
+        if (!streams[type]) {
+          continue
+        }
+
+        const singular = type as 'BankTransaction'
+        const plural = `${singular}s` as const
+        const kId = `${singular}ID` as const
+        let page = 1
+        while (true) {
+          const result = await xero.accounting.GET(`/${plural}`, {
+            params: {header: {'xero-tenant-id': tenantId}, query: {page}},
+          })
+          if (result.data[plural]?.length) {
+            yield result.data[plural]?.map((a) =>
+              xeroHelpers._opData(singular, a[kId]!, a),
+            )
+            // Account does not support pagination, all or nothing...
+            if (type !== 'Account') {
+              page++
+              continue
+            }
+          }
+          break
+        }
+      }
+    }
+
+    return rxjs
+      .from(iterateEntities())
+      .pipe(
+        Rx.mergeMap((ops) => rxjs.from([...ops, xeroHelpers._op('commit')])),
+      )
+  },
+} satisfies ConnectorServer<typeof xeroSchemas, ReturnType<typeof initXeroSDK>>
 
 export default xeroServer

docs/samples/banking-test.ts

@@ -3,7 +3,8 @@ import {createVeniceClient} from '@usevenice/sdk'
 const venice = createVeniceClient({
   apiKey: process.env['_VENICE_API_KEY'],
   apiHost: process.env['_VENICE_API_HOST'],
-  resourceId: process.env['_QBO_RESOURCE_ID'],
+  // resourceId: process.env['_QBO_RESOURCE_ID'],
+  resourceId: process.env['_XERO_RESOURCE_ID'],
 })
 
 void venice.GET('/verticals/banking/category').then((r) => {

packages/engine-backend/router/_base.ts

@@ -45,19 +45,18 @@ export const trpc = initTRPC
 
 export const publicProcedure = trpc.procedure
 // Enable me for logs
-// .use(
-//   ({next, ctx, input, rawInput, meta, path}) => {
-//     console.log('[trpc]', {
-//       input,
-//       rawInput,
-//       meta,
-//       path,
-//     })
-//     return next({ctx})
-//   },
-// )
+// .use(({next, ctx, input, rawInput, meta, path}) => {
+//   console.log('[trpc]', {
+//     input,
+//     rawInput,
+//     meta,
+//     path,
+//   })
+//   return next({ctx})
+// })
 
 export const protectedProcedure = publicProcedure.use(({next, ctx}) => {
+  console.log('DEBUG', ctx.viewer)
   if (!hasRole(ctx.viewer, ['end_user', 'user', 'org', 'system'])) {
     throw new TRPCError({
       code: ctx.viewer.role === 'anon' ? 'UNAUTHORIZED' : 'FORBIDDEN',

packages/engine-backend/router/endUserRouter.ts

@@ -205,6 +205,8 @@ export const endUserRouter = trpc.router({
         const int = await ctx.asOrgIfNeeded.getConnectorConfigOrFail(ccfgId)
         console.log('didConnect start', int.connector.name, input, connCtxInput)
 
+        // TODO: we should make it possible for oauth connectors to
+        // ALSO handle custom postConnect... This would be very handy for xero for instance
         const resoUpdate = await (async () => {
           if (
             !int.connector.postConnect &&

packages/engine-backend/router/index.ts

@@ -11,6 +11,7 @@ import {
   outreachAdapter,
   qboAdapter,
   salesloftAdapter,
+  xeroAdapter,
 } from '@usevenice/cdk/verticals'
 import {remoteProcedure, trpc} from './_base'
 import {adminRouter} from './adminRouter'
@@ -26,7 +27,7 @@ import {systemRouter} from './systemRouter'
 const bankingRouter = createBankingRouter({
   trpc,
   remoteProcedure,
-  adapterByName: {qbo: qboAdapter},
+  adapterByName: {qbo: qboAdapter, xero: xeroAdapter},
 })
 const accountingRouter = createAccountingRouter({
   trpc,

packages/engine-frontend/VeniceConnect.tsx

@@ -156,7 +156,6 @@ export function _VeniceConnect({
 }) {
   const nangoPublicKey =
     _trpcReact.getPublicEnv.useQuery().data?.NEXT_PUBLIC_NANGO_PUBLIC_KEY
-
   const nangoFrontend = React.useMemo(
     () =>
       nangoPublicKey &&

verticals/vertical-banking/adapters/xero-adapter.ts

@@ -0,0 +1,46 @@
+import type {Oas_accounting, XeroSDK} from 'connectors/connector-xero'
+import type {StrictObj} from '@usevenice/vdk'
+import {mapper, z, zCast} from '@usevenice/vdk'
+import type {VerticalBanking} from '../banking'
+import {zBanking} from '../banking'
+
+type Xero = Oas_accounting['components']['schemas']
+
+const mappers = {
+  category: mapper(
+    zCast<StrictObj<Xero['Account']>>(),
+    zBanking.category.extend({_raw: z.unknown().optional()}),
+    {
+      id: 'AccountID',
+      name: 'Name',
+      _raw: (a) => a,
+    },
+  ),
+}
+
+export const xeroAdapter = {
+  listCategories: async ({instance}) => {
+    // TODO: Abstract this away please...
+    const tenantId = await instance.identity
+      .GET('/Connections')
+      .then((r) => r.data?.[0]?.tenantId)
+    if (!tenantId) {
+      throw new Error(
+        'Missing access to any tenants. Check xero token permission',
+      )
+    }
+
+    const res = await instance.accounting.GET('/Accounts', {
+      params: {
+        header: {'xero-tenant-id': tenantId},
+        query: {
+          where: 'Class=="REVENUE"||Class=="EXPENSE"',
+        },
+      },
+    })
+    return {
+      hasNextPage: false,
+      items: (res.data.Accounts ?? []).map(mappers.category),
+    }
+  },
+} satisfies VerticalBanking<{instance: XeroSDK}>