chore: refactor cursors into a separate table

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

backend/src/database/migrations/U1770818540__createNangoCursorsTable.sql

@@ -0,0 +1,59 @@
+-- Backup settings before any modifications
+CREATE TABLE integration.integrations_settings_backup_02_13_2026 AS
+SELECT id, settings FROM integrations;
+
+CREATE TABLE integration.nango_cursors (
+    "integrationId"  UUID         NOT NULL,
+    "connectionId"   TEXT         NOT NULL,
+    platform         TEXT         NOT NULL,
+    model            TEXT         NOT NULL,
+    cursor           TEXT         NOT NULL,
+    "lastCheckedAt"  TIMESTAMPTZ,
+    "createdAt"      TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    "updatedAt"      TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    PRIMARY KEY ("integrationId", "connectionId", model),
+    FOREIGN KEY ("integrationId") REFERENCES integrations(id) ON DELETE CASCADE
+);
+
+CREATE INDEX ix_nango_cursors_lastCheckedAt ON integration.nango_cursors ("lastCheckedAt" NULLS FIRST);
+CREATE INDEX ix_nango_cursors_connectionId ON integration.nango_cursors ("connectionId");
+
+-- GitHub-nango: unnest nangoMapping keys x cursor models
+INSERT INTO integration.nango_cursors ("integrationId", "connectionId", platform, model, cursor)
+SELECT
+    i.id,
+    nm.key,
+    'github',
+    cm.key,
+    cm.value #>> '{}'
+FROM integrations i,
+     jsonb_each(i.settings->'nangoMapping') nm,
+     jsonb_each(COALESCE(i.settings->'cursors'->nm.key, '{}'::jsonb)) cm
+WHERE i.platform = 'github-nango'
+  AND i."deletedAt" IS NULL
+  AND i.settings->'nangoMapping' IS NOT NULL
+ON CONFLICT DO NOTHING;
+
+-- Non-GitHub nango: connectionId = integrationId, unnest cursor models
+INSERT INTO integration.nango_cursors ("integrationId", "connectionId", platform, model, cursor)
+SELECT
+    i.id,
+    i.id::text,
+    CASE
+        WHEN i.platform = 'gerrit' THEN 'gerrit'
+        WHEN i.platform = 'jira' THEN COALESCE(i.settings->>'nangoIntegrationName', 'jira-basic')
+        WHEN i.platform = 'confluence' THEN COALESCE(i.settings->>'nangoIntegrationName', 'confluence')
+        ELSE i.platform
+    END,
+    cm.key,
+    cm.value #>> '{}'
+FROM integrations i,
+     jsonb_each(COALESCE(i.settings->'cursors'->i.id::text, '{}'::jsonb)) cm
+WHERE i.platform IN ('gerrit', 'jira', 'confluence')
+  AND i."deletedAt" IS NULL
+ON CONFLICT DO NOTHING;
+
+-- Clean up settings.cursors
+UPDATE integrations
+SET settings = settings - 'cursors'
+WHERE settings->'cursors' IS NOT NULL;

backend/src/database/migrations/V1770818540__createNangoCursorsTable.sql

@@ -869,11 +869,6 @@ export default class IntegrationService {
             platform: PlatformType.GITHUB_NANGO,
             settings: {
               ...settings,
-              ...(integration.settings.cursors
-                ? {
-                    cursors: integration.settings.cursors,
-                  }
-                : {}),
               ...(integration.settings.nangoMapping
                 ? {
                     nangoMapping: integration.settings.nangoMapping,

backend/src/services/integrationService.ts

@@ -10,6 +10,7 @@ import {
 import { READ_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
 import {
   INangoIntegrationData,
+  fetchNangoCursorRowsForIntegration,
   fetchNangoIntegrationData,
 } from '@crowd/data-access-layer/src/integrations'
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
@@ -98,9 +99,12 @@ const job: IJobDefinition = {
 
         // then collect nango connection status checks for each connection
         if (int.settings.nangoMapping) {
+          const cursorRows = await fetchNangoCursorRowsForIntegration(pgpQx(dbConnection), int.id)
+          const connectionIdsWithCursors = new Set(cursorRows.map((r) => r.connectionId))
+
           for (const connectionId of Object.keys(int.settings.nangoMapping)) {
             // check if we have cursors already for this connection
-            if (!int.settings.cursors || !int.settings.cursors[connectionId]) {
+            if (!connectionIdsWithCursors.has(connectionId)) {
               if (ghNoCursorsYet.has(int.id)) {
                 ghNoCursorsYet.set(int.id, ghNoCursorsYet.get(int.id) + 1)
               } else {

services/apps/cron_service/src/jobs/nangoMonitoring.job.ts

@@ -2,7 +2,10 @@ import CronTime from 'cron-time-generator'
 
 import { ConcurrencyLimiter, IS_DEV_ENV } from '@crowd/common'
 import { READ_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
-import { fetchNangoIntegrationDataForCheck } from '@crowd/data-access-layer/src/integrations'
+import {
+  fetchNangoIntegrationDataForCheck,
+  fetchNangoLastCheckedAt,
+} from '@crowd/data-access-layer/src/integrations'
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
 import {
   ALL_NANGO_INTEGRATIONS,
@@ -12,7 +15,12 @@ import {
   nangoIntegrationToPlatform,
   platformToNangoIntegration,
 } from '@crowd/nango'
-import { TEMPORAL_CONFIG, WorkflowIdReusePolicy, getTemporalClient } from '@crowd/temporal'
+import {
+  TEMPORAL_CONFIG,
+  WorkflowIdConflictPolicy,
+  WorkflowIdReusePolicy,
+  getTemporalClient,
+} from '@crowd/temporal'
 import { PlatformType } from '@crowd/types'
 
 import { IJobDefinition } from '../types'
@@ -22,16 +30,15 @@ const AGE_THRESHOLD_MS = IS_DEV_ENV
   ? 20 * 60 * 1000 // 20 minutes for local testing
   : 30 * 24 * 60 * 60 * 1000 // 1 month
 
-// How often the cron runs (used to determine if old integrations should be triggered this run)
-const OLD_INTEGRATION_INTERVAL_HOURS = IS_DEV_ENV ? 0 : 6
-const OLD_INTEGRATION_INTERVAL_MINUTES = IS_DEV_ENV ? 15 : 0
+// Minimum interval between checks for new integrations
+const NEW_INTERVAL_MS = IS_DEV_ENV
+  ? 5 * 60 * 1000 // 5 minutes
+  : 60 * 60 * 1000 // 1 hour
 
-function shouldTriggerOldIntegrations(now: Date): boolean {
-  if (IS_DEV_ENV) {
-    return now.getMinutes() % OLD_INTEGRATION_INTERVAL_MINUTES === 0
-  }
-  return now.getHours() % OLD_INTEGRATION_INTERVAL_HOURS === 0
-}
+// Minimum interval between checks for old integrations
+const OLD_INTERVAL_MS = IS_DEV_ENV
+  ? 15 * 60 * 1000 // 15 minutes
+  : 6 * 60 * 60 * 1000 // 6 hours
 
 const job: IJobDefinition = {
   name: 'nango-trigger',
@@ -43,100 +50,63 @@ const job: IJobDefinition = {
     const temporal = await getTemporalClient(TEMPORAL_CONFIG())
 
     const dbConnection = await getDbConnection(READ_DB_CONFIG(), 3, 0)
+    const qx = pgpQx(dbConnection)
 
-    const allIntegrations = await fetchNangoIntegrationDataForCheck(pgpQx(dbConnection), [
-      ...new Set(ALL_NANGO_INTEGRATIONS.map(nangoIntegrationToPlatform)),
-    ])
+    const platforms = [...new Set(ALL_NANGO_INTEGRATIONS.map(nangoIntegrationToPlatform))]
 
-    const now = new Date()
-    const triggerOld = shouldTriggerOldIntegrations(now)
+    const allIntegrations = await fetchNangoIntegrationDataForCheck(qx, platforms)
 
-    const integrationsToTrigger = allIntegrations.filter((int) => {
-      const ageMs = now.getTime() - new Date(int.createdAt).getTime()
-      const isOld = ageMs >= AGE_THRESHOLD_MS
-      return !isOld || triggerOld
-    })
-
-    ctx.log.info(
-      `Total integrations: ${allIntegrations.length}, triggering: ${integrationsToTrigger.length} (old integrations ${triggerOld ? 'included' : 'skipped'})`,
-    )
+    // Batch-fetch lastCheckedAt for all connections
+    const lastCheckedAtRows = await fetchNangoLastCheckedAt(qx, platforms)
+    const lastCheckedAtMap = new Map<string, string | null>()
+    for (const row of lastCheckedAtRows) {
+      lastCheckedAtMap.set(`${row.integrationId}/${row.connectionId}`, row.lastCheckedAt)
+    }
 
+    const now = new Date()
     const limiter = new ConcurrencyLimiter(5)
-
-    // Collect all workflow start operations
     const workflowStarts: Array<() => Promise<void>> = []
+    let skippedConnections = 0
 
-    for (let i = 0; i < integrationsToTrigger.length; i++) {
-      const int = integrationsToTrigger[i]
-
+    for (let i = 0; i < allIntegrations.length; i++) {
+      const int = allIntegrations[i]
       const { id, settings } = int
 
-      ctx.log.info(
-        `${i + 1}/${integrationsToTrigger.length} Triggering nango integration check for ${id} (${int.platform})`,
-      )
-
       const platform = platformToNangoIntegration(int.platform as PlatformType, settings)
 
       if (platform === NangoIntegration.GITHUB && !settings.nangoMapping) {
         // ignore non-nango github integrations
         continue
       }
 
-      for (const model of Object.values(NANGO_INTEGRATION_CONFIG[platform].models)) {
-        ctx.log.debug(
-          {
-            integrationId: id,
-            platform,
-            model,
-          },
-          'Triggering nango integration check!',
-        )
+      const integrationAgeMs = now.getTime() - new Date(int.createdAt).getTime()
+      const isOld = integrationAgeMs >= AGE_THRESHOLD_MS
+      const requiredInterval = isOld ? OLD_INTERVAL_MS : NEW_INTERVAL_MS
 
-        if (platform === NangoIntegration.GITHUB) {
-          // trigger for each connection id - could be multiple because 1 integration can have multiple repositories and each repository has a connection id on nango
-          for (const connectionId of Object.keys(settings.nangoMapping)) {
-            const payload: INangoWebhookPayload = {
-              connectionId: connectionId,
-              providerConfigKey: platform,
-              syncName: 'not important',
-              model,
-              responseResults: { added: 1, updated: 1, deleted: 1 },
-              syncType: 'INCREMENTAL',
-              modifiedAfter: new Date().toISOString(),
-            }
-
-            workflowStarts.push(async () => {
-              try {
-                await temporal.workflow.start('processNangoWebhook', {
-                  taskQueue: 'nango',
-                  workflowId: `nango-webhook/${platform}/${id}/${connectionId}/${model}/cron-triggered`,
-                  workflowIdReusePolicy:
-                    WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
-                  retry: {
-                    maximumAttempts: 10,
-                  },
-                  args: [payload],
-                })
-              } catch (error) {
-                if (error.name === 'WorkflowExecutionAlreadyStartedError') {
-                  ctx.log.debug(
-                    {
-                      integrationId: id,
-                      platform,
-                      model,
-                      connectionId,
-                    },
-                    'Workflow already running, skipping...',
-                  )
-                  return
-                }
-                throw error
-              }
-            })
+      // Determine connectionIds for this integration
+      const connectionIds: string[] =
+        platform === NangoIntegration.GITHUB ? Object.keys(settings.nangoMapping) : [id]
+
+      for (const connectionId of connectionIds) {
+        const key = `${id}/${connectionId}`
+        const lastCheckedAt = lastCheckedAtMap.get(key)
+
+        // Skip if checked recently enough
+        if (lastCheckedAt) {
+          const elapsed = now.getTime() - new Date(lastCheckedAt).getTime()
+          if (elapsed < requiredInterval) {
+            skippedConnections++
+            continue
           }
-        } else {
+        }
+
+        ctx.log.info(
+          `${i + 1}/${allIntegrations.length} Triggering nango integration check for ${id} / ${connectionId} (${platform})`,
+        )
+
+        for (const model of Object.values(NANGO_INTEGRATION_CONFIG[platform].models)) {
           const payload: INangoWebhookPayload = {
-            connectionId: id,
+            connectionId,
             providerConfigKey: platform,
             syncName: 'not important',
             model,
@@ -145,38 +115,30 @@ const job: IJobDefinition = {
             modifiedAfter: new Date().toISOString(),
           }
 
+          const workflowId =
+            platform === NangoIntegration.GITHUB
+              ? `nango-webhook/${platform}/${id}/${connectionId}/${model}/cron-triggered`
+              : `nango-webhook/${platform}/${id}/${model}/cron-triggered`
+
           workflowStarts.push(async () => {
-            try {
-              await temporal.workflow.start('processNangoWebhook', {
-                taskQueue: 'nango',
-                workflowId: `nango-webhook/${platform}/${id}/${model}/cron-triggered`,
-                workflowIdReusePolicy:
-                  WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
-                retry: {
-                  maximumAttempts: 10,
-                },
-                args: [payload],
-              })
-            } catch (error) {
-              if (error.name === 'WorkflowExecutionAlreadyStartedError') {
-                ctx.log.debug(
-                  {
-                    integrationId: id,
-                    platform,
-                    model,
-                  },
-                  'Workflow already running, skipping...',
-                )
-                return
-              }
-              throw error
-            }
+            await temporal.workflow.start('processNangoWebhook', {
+              taskQueue: 'nango',
+              workflowId,
+              workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
+              workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING,
+              retry: {
+                maximumAttempts: 10,
+              },
+              args: [payload],
+            })
           })
         }
       }
     }
 
-    ctx.log.info(`Triggering nango integration checks with ${workflowStarts.length} workflows!`)
+    ctx.log.info(
+      `Triggering ${workflowStarts.length} workflows (skipped ${skippedConnections} connections due to recent checks)`,
+    )
 
     // Track completed workflows
     let completedWorkflows = 0