Merge branch 'main' into support/CM-2298

Author: skwowet

services/apps/nango_worker/src/activities/nangoActivities.ts

@@ -61,7 +61,16 @@ export async function processNangoWebhook(
     integrationId: integration.id,
   })
 
-  const cursor = integration.settings.cursors ? integration.settings.cursors[args.model] : undefined
+  const settings = integration.settings
+  let cursor = args.nextPageCursor
+  if (
+    !cursor &&
+    settings.cursors &&
+    settings.cursors[args.connectionId] &&
+    settings.cursors[args.connectionId][args.model]
+  ) {
+    cursor = settings.cursors[args.connectionId][args.model]
+  }
 
   await initNangoCloudClient()
 
@@ -96,6 +105,7 @@ export async function processNangoWebhook(
       await setNangoIntegrationCursor(
         dbStoreQx(svc.postgres.writer),
         integration.id,
+        args.connectionId,
         args.model,
         records.nextCursor,
       )
@@ -105,6 +115,7 @@ export async function processNangoWebhook(
       await setNangoIntegrationCursor(
         dbStoreQx(svc.postgres.writer),
         integration.id,
+        args.connectionId,
         args.model,
         records.records[records.records.length - 1].metadata.cursor,
       )

services/apps/nango_worker/src/types.ts

@@ -3,6 +3,7 @@ export interface IProcessNangoWebhookArguments {
   providerConfigKey: string
   model: string
   syncType: 'INITIAL' | 'INCREMENTAL'
+  nextPageCursor?: string
 }
 
 export interface ISyncGithubIntegrationArguments {

services/apps/nango_worker/src/workflows/processNangoWebhook.ts

@@ -11,6 +11,9 @@ export async function processNangoWebhook(args: IProcessNangoWebhookArguments):
   let nextCursor = await activity.processNangoWebhook(args)
 
   while (nextCursor) {
-    nextCursor = await activity.processNangoWebhook(args)
+    nextCursor = await activity.processNangoWebhook({
+      ...args,
+      nextPageCursor: nextCursor,
+    })
   }
 }

services/libs/data-access-layer/src/integrations/index.ts

@@ -275,21 +275,42 @@ export async function findIntegrationDataForNangoWebhookProcessing(
 export async function setNangoIntegrationCursor(
   qx: QueryExecutor,
   integrationId: string,
+  connectionId: string,
   model: string,
   cursor: string,
 ): Promise<void> {
   await qx.result(
     `
       update integrations
-      set settings = jsonb_set(
-        settings,
-        '{cursors}',
-        coalesce(settings -> 'cursors', '{}') || jsonb_build_object($(model), $(cursor))
-      )
-      where id = $(integrationId)
+      set settings = case
+          -- when we don't have any cursors yet
+                        when settings -> 'cursors' is null then
+                            jsonb_set(
+                                    settings,
+                                    array['cursors'],
+                                    jsonb_build_object($(connectionId), jsonb_build_object($(model), $(cursor)))
+                            )
+          -- when we have cursors but not yet for this connectionId
+                        when settings -> 'cursors' -> $(connectionId) is null then
+                            jsonb_set(
+                                    settings,
+                                    array['cursors'],
+                                    (settings -> 'cursors') ||
+                                    jsonb_build_object($(connectionId), jsonb_build_object($(model), $(cursor)))
+                            )
+          -- when we have cursors and entries for this connectionId
+                        else
+                            jsonb_set(
+                                    settings,
+                                    array['cursors', $(connectionId)],
+                                    (settings -> 'cursors' -> $(connectionId)) || jsonb_build_object($(model), $(cursor))
+                            )
+          end
+      where id = $(integrationId);
     `,
     {
       integrationId,
+      connectionId,
       model,
       cursor,
     },