Refresh connection flow

apps/web/.gitignore

@@ -8,6 +8,11 @@
 # testing
 /coverage
 
+# Cloudflare
+.dev.vars
+.open-next
+.wrangler
+
 # next.js
 /.next/
 /out/

kits/cdk/base-links.ts

@@ -1,7 +1,7 @@
 import type {
   AnyEntityPayload,
+  ConnectionUpdateData,
   Link,
-  ResoUpdateData,
   StateUpdateData,
   SyncOperation,
 } from '@openint/sync'
@@ -13,11 +13,11 @@ type OperationType = SyncOperation['type']
 export type OpHandlers<
   TRet,
   T = any,
-  TResoUpdate extends object = ResoUpdateData,
+  TConnUpdate extends object = ConnectionUpdateData,
   TStateUpdate extends object = StateUpdateData,
 > = Partial<{
   [k in OperationType]: (
-    op: Extract<SyncOperation<T, TResoUpdate, TStateUpdate>, {type: k}>,
+    op: Extract<SyncOperation<T, TConnUpdate, TStateUpdate>, {type: k}>,
   ) => TRet | Promise<TRet>
 }>
 
@@ -27,13 +27,13 @@ export type OpHandlers<
  */
 export function handlersLink<
   TData = any,
-  TResoUpdate extends object = ResoUpdateData,
+  TConnUpdate extends object = ConnectionUpdateData,
   TStateUpdate extends object = StateUpdateData,
 >(
   handlers: OpHandlers<
     rxjs.ObservableInput<SyncOperation<TData>> | void,
     TData,
-    TResoUpdate,
+    TConnUpdate,
     TStateUpdate
   >,
 ): Link<TData> {

kits/cdk/connector-meta.types.ts

@@ -1,7 +1,11 @@
 import type {NangoProvider} from '@opensdks/sdk-nango/src/nango.oas'
 import {z} from '@opensdks/util-zod'
 import type {oas30, oas31} from 'openapi3-ts'
-import type {AnyEntityPayload, ResoUpdateData, Source} from '@openint/sync'
+import type {
+  AnyEntityPayload,
+  ConnectionUpdateData,
+  Source,
+} from '@openint/sync'
 import {castIs} from '@openint/util'
 import type {ConnHelpers} from './connector.types'
 import type {CustomerId, ExtCustomerId, ExternalId, Id} from './id.types'
@@ -111,9 +115,12 @@ export const zCheckConnectionOptions = z.object({
 })
 
 /** Extra props not on ResoUpdateData */
-export interface ConnectionUpdate<TEntity = AnyEntityPayload, TSettings = unknown>
+export interface ConnectionUpdate<
+    TEntity = AnyEntityPayload,
+    TSettings = unknown,
+  >
   // make `ResoUpdateData.id` not prefixed so we can have better inheritance
-  extends Omit<ResoUpdateData<TSettings>, 'id'> {
+  extends Omit<ConnectionUpdateData<TSettings>, 'id'> {
   // Subset of connUpdate
   connectionExternalId: ExternalId
   // Can we inherit types used by metaLinks?

kits/cdk/connector.types.ts

@@ -1,9 +1,9 @@
 import type {Link as FetchLink} from '@opensdks/runtime'
 import type {z} from '@opensdks/util-zod'
 import type {
+  ConnectionUpdateData,
   Destination,
   EntityPayload,
-  ResoUpdateData,
   Source,
   StateUpdateData,
   SyncOperation,
@@ -14,10 +14,10 @@ import type {
   CheckConnectionContext,
   CheckConnectionOptions,
   ConnectContext,
+  ConnectionUpdate,
   ConnectOptions,
   ConnectorMetadata,
   OpenDialogFn,
-  ConnectionUpdate,
   WebhookReturnType,
   zPassthroughInput,
 } from './connector-meta.types'
@@ -312,7 +312,7 @@ export function connHelpers<TSchemas extends ConnectorSchemas>(
     }>,
     {type: 'data'}
   >
-  type connUpdate = ResoUpdateData<
+  type connUpdate = ConnectionUpdateData<
     _types['connectionSettings'],
     _types['integrationData']
   >

kits/cdk/internal/nangoProxyLink.ts

@@ -3,6 +3,7 @@ import {mergeHeaders, modifyRequest} from '@opensdks/fetch-links'
 import {initNangoSDK} from '@opensdks/sdk-nango'
 import {z} from '@opensdks/util-zod'
 import {isHttpError, NotAuthenticatedError} from '@openint/trpc'
+import {zOauthConnectionError} from './oauthConnector'
 
 const kBaseUrlOverride = 'base-url-override'
 
@@ -206,4 +207,5 @@ export const nangoConnectionWithCredentials = z.object({
   deleted_at: z.string().nullish(),
   last_fetched_at: z.string().nullish(),
   config_id: z.number().nullish(),
+  error: zOauthConnectionError.nullish(),
 })

kits/cdk/internal/oauthConnector.ts

@@ -22,6 +22,11 @@ export const zAuthMode = z
   .enum(['OAUTH2', 'OAUTH1', 'BASIC', 'API_KEY'])
   .openapi({ref: 'AuthMode'})
 
+export const zOauthConnectionError = z.object({
+  code: z.enum(['refresh_token_external_error']).or(z.string()),
+  message: z.string().nullish(),
+})
+
 export const oauthBaseSchema = {
   name: z.literal('__oauth__'), // TODO: This is a noop
   connectorConfig: z.object({
@@ -33,6 +38,7 @@ export const oauthBaseSchema = {
     }),
   }),
   connectionSettings: z.object({
+    // equivalent to nango /v1/connections data.connection object with certain fields removed like id
     oauth: z.object({
       credentials: z.object({
         type: zAuthMode,
@@ -61,6 +67,8 @@ export const oauthBaseSchema = {
         .nullish(),
       metadata: z.record(z.unknown()).nullable(),
     }),
+    // TODO: add error fields here or maybe at a higher level to capture when connection needs to be refreshed
+    error: zOauthConnectionError.nullish(),
   }),
   connectOutput: z.object({
     providerConfigKey: zId('ccfg'),
@@ -147,8 +155,16 @@ export function makeOauthConnectorServer({
             },
           },
         })
-        .then((r) => r.data)
-      return {connectionExternalId: extractId(connId)[2], settings: {oauth: res}}
+        .then((r) => r.data as OauthBaseTypes['connectionSettings'])
+      return {
+        connectionExternalId: extractId(connId)[2],
+        settings: {
+          oauth: res as any,
+          ...(res?.error?.code || res?.error?.message
+            ? {error: {code: res?.error?.code, message: res?.error?.message}}
+            : {}),
+        },
+      }
     },
   } satisfies ConnectorServer<typeof oauthBaseSchema>
   return {

kits/cdk/internal/remote-procedure.ts

@@ -63,7 +63,13 @@ export async function getRemoteContext(ctx: ProtectedContext) {
             },
           },
         })
-        .then((r) => nangoConnectionWithCredentials.parse(r.data)),
+        .then((r: any) => {
+          return nangoConnectionWithCredentials.parse(r.data)
+        })
+        .catch((error) => {
+          console.error('nangoConnectionWithCredentials error', error)
+          throw error
+        }),
     }),
   }
 

kits/cdk/models.ts

@@ -63,7 +63,7 @@ export const zStandard = {
   }),
   connection: z.object({
     id: zId('conn'),
-    displayName: z.string(),
+    displayName: z.string().nullish(),
     /**
      * This correspond to the connection status.
      * Pipeline shall have a separate syncStatus */

kits/sdk/openapi.json

@@ -8063,7 +8063,7 @@
             "type": ["object", "null"],
             "properties": {
               "displayName": {
-                "type": "string"
+                "type": ["string", "null"]
               },
               "status": {
                 "type": ["string", "null"],
@@ -8078,8 +8078,7 @@
                   "type": "string"
                 }
               }
-            },
-            "required": ["displayName"]
+            }
           },
           "disabled": {
             "type": "boolean"

kits/sdk/openapi.types.d.ts

@@ -547,7 +547,7 @@ export interface components {
         [key: string]: unknown
       } | null
       standard?: {
-        displayName: string
+        displayName?: string | null
         /** @enum {string|null} */
         status?: 'healthy' | 'disconnected' | 'error' | 'manual'
         statusMessage?: string | null

kits/sync/protocol.ts

@@ -29,7 +29,7 @@ export interface AnyEntityPayload {
   id: string // ExternalId
 }
 
-export interface ResoUpdateData<
+export interface ConnectionUpdateData<
   TSettings = {},
   TInsData = {},
   TVariant extends 'partial' | 'complete' = 'partial',
@@ -43,6 +43,9 @@ export interface ResoUpdateData<
     externalId: ExternalId
     data: TInsData
   }
+  // QQ: Extend this or calculate it before displaying based on settings?
+  // status?: ZStandard['connection']['status']
+  // statusMessage?: string
 }
 export interface StateUpdateData<TSrcOptions = {}, TDestOptions = {}> {
   sourceState?: ObjectPartialDeep<NoInfer<TSrcOptions>>
@@ -55,10 +58,10 @@ type NullableEntity<T> = T extends AnyEntityPayload
 
 export type SyncOperation<
   TData = any,
-  TResoUpdate extends object = ResoUpdateData,
+  TConnUpdate extends object = ConnectionUpdateData,
   TStateUpdate extends object = StateUpdateData,
 > =
-  | (TResoUpdate & {type: 'connUpdate'})
+  | (TConnUpdate & {type: 'connUpdate'})
   // TODO: We should separate state from options, and perhaps make state
   // less black box also, see airbyte protocol v2 for inspiration
   // Also consider merging fields below into a single field
@@ -71,9 +74,9 @@ export type AnySyncOperation = NonDiscriminatedUnion<SyncOperation>
 
 export type Source<
   T,
-  TResoUpdate extends object = ResoUpdateData,
+  TConnUpdate extends object = ConnectionUpdateData,
   TStateUpdate extends object = StateUpdateData,
-> = rxjs.Observable<SyncOperation<T, TResoUpdate, TStateUpdate>>
+> = rxjs.Observable<SyncOperation<T, TConnUpdate, TStateUpdate>>
 
 /**
  * Adapted from TRPC link and Apollo Link
@@ -82,30 +85,30 @@ export type Source<
 export type Link<
   TDataIn = any,
   TDataOut = TDataIn,
-  TResoUpdate extends object = ResoUpdateData,
+  TConnUpdate extends object = ConnectionUpdateData,
   TStateUpdate extends object = StateUpdateData,
 > = (
-  obs: rxjs.Observable<SyncOperation<TDataIn, TResoUpdate, TStateUpdate>>,
-) => rxjs.Observable<SyncOperation<TDataOut, TResoUpdate, TStateUpdate>>
+  obs: rxjs.Observable<SyncOperation<TDataIn, TConnUpdate, TStateUpdate>>,
+) => rxjs.Observable<SyncOperation<TDataOut, TConnUpdate, TStateUpdate>>
 
 export type LinkFactory<
   TDataIn = any,
   TDataOut = TDataIn,
-  TResoUpdate extends object = ResoUpdateData,
+  TConnUpdate extends object = ConnectionUpdateData,
   TStateUpdate extends object = StateUpdateData,
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   TArg = any,
-> = (arg: TArg) => Link<TDataIn, TDataOut, TResoUpdate, TStateUpdate>
+> = (arg: TArg) => Link<TDataIn, TDataOut, TConnUpdate, TStateUpdate>
 
 /**
  * Terminating link is just a link... It can still emit things like ready event
  * for the engine to listen to. The resulting event may not be the same as the input events
  */
 export type Destination<
   T = any,
-  TResoUpdate extends object = ResoUpdateData,
+  TConnUpdate extends object = ConnectionUpdateData,
   TStateUpdate extends object = StateUpdateData,
-> = Link<T, T, TResoUpdate, TStateUpdate>
+> = Link<T, T, TConnUpdate, TStateUpdate>
 
 // @deprecated?
 

kits/sync/sync.ts

@@ -1,9 +1,9 @@
 import {R, Rx, toCompletion} from '@openint/util'
 import type {
   AnyEntityPayload,
+  ConnectionUpdateData,
   Destination,
   Link,
-  ResoUpdateData,
   Source,
   StateUpdateData,
   SyncOperation,
@@ -17,12 +17,12 @@ const COMMIT: Extract<SyncOperation, {type: 'commit'}> = {type: 'commit'}
 /** The most fundamental implementation of using using protocol */
 export async function sync<
   T extends Data = Data,
-  TResoUpdate extends object = ResoUpdateData,
+  TConnUpdate extends object = ConnectionUpdateData,
   TStateUpdate extends object = StateUpdateData,
 >(input: {
-  source: Source<T, TResoUpdate, TStateUpdate>
-  destination: Destination<T, TResoUpdate, TStateUpdate>
-  links?: Array<Link<T, T, TResoUpdate, TStateUpdate>>
+  source: Source<T, TConnUpdate, TStateUpdate>
+  destination: Destination<T, TConnUpdate, TStateUpdate>
+  links?: Array<Link<T, T, TConnUpdate, TStateUpdate>>
   watch?: boolean
 }) {
   const start = Date.now()

packages/api/proxyHandler.ts

@@ -20,6 +20,7 @@ export const proxyHandler = async (req: Request) => {
       new Date()
     : false
 
+  // TODO: this logic is potentially being duplicated in the getRemoteContext call
   if (credentialsExpired && remoteContext.remoteConnectionId) {
     const nango = initNangoSDK({
       headers: {authorization: `Bearer ${process.env['NANGO_SECRET_KEY']}`},

packages/engine-backend/router/connectionRouter.ts

@@ -34,7 +34,11 @@ const zExpandConnector = z.object({
   logoUrl: z.string().url(),
 })
 
-async function performConnectionCheck(ctx: any, connId: string, opts: any) {
+export async function performConnectionCheck(
+  ctx: any,
+  connId: string,
+  opts: any,
+) {
   const remoteCtx = await getRemoteContext({
     ...ctx,
     remoteConnectionId: connId,
@@ -51,8 +55,13 @@ async function performConnectionCheck(ctx: any, connId: string, opts: any) {
       webhookBaseUrl: joinPath(ctx.apiUrl, parseWebhookRequest.pathOf(int.id)),
     },
   })
-
-  if (connUpdate || opts?.import !== false) {
+  if (
+    conn?.settings?.error ||
+    connUpdate ||
+    opts?.import !== false ||
+    remoteCtx.remote.settings?.oauth?.error
+  ) {
+    /** Do not update the `customerId` here... */
     await ctx.asOrgIfNeeded._syncConnectionUpdate(int, {
       customerId: conn.customerId ?? undefined,
       integration: conn.integrationId

packages/engine-backend/router/customerRouter.ts

@@ -307,6 +307,10 @@ export const customerRouter = trpc.router({
             customerId:
               ctx.viewer.role === 'customer' ? ctx.viewer.customerId : null,
             triggerDefaultSync,
+            settings: {
+              ...connUpdate?.settings,
+              error: connUpdate?.settings?.['error'] || null,
+            },
           },
         )