chore: sync github repos if fullSync

themarolt · themarolt · commit 1de115963834 · 2025-11-05T09:20:10.000+01:00
Signed-off-by: Uroš Marolt &lt;uros@marolt.me&gt;
diff --git a/backend/src/bin/jobs/refreshGithubRepoSettings.ts b/backend/src/bin/jobs/refreshGithubRepoSettings.ts
@@ -1,7 +1,7 @@
 /* eslint-disable no-continue */
 import cronGenerator from 'cron-time-generator'
 
-import { timeout } from '@crowd/common'
+import { IS_DEV_ENV, timeout } from '@crowd/common'
 import { getServiceChildLogger } from '@crowd/logging'
 
 import SequelizeRepository from '../../database/repositories/sequelizeRepository'
@@ -59,7 +59,7 @@ export const refreshGithubRepoSettings = async () => {
 const job: CrowdJob = {
   name: 'Refresh Github repo settings',
   // every day
-  cronTime: cronGenerator.every(1).days(),
+  cronTime: IS_DEV_ENV ? cronGenerator.every(5).minutes() : cronGenerator.every(1).days(),
   onTrigger: async () => {
     await refreshGithubRepoSettings()
   },
diff --git a/backend/src/segment/track.ts b/backend/src/segment/track.ts
@@ -1,7 +1,7 @@
 import { getServiceChildLogger } from '@crowd/logging'
 import { Edition } from '@crowd/types'
 
-import { API_CONFIG, IS_TEST_ENV, SEGMENT_CONFIG } from '../conf'
+import { API_CONFIG, IS_DEV_ENV, IS_TEST_ENV, SEGMENT_CONFIG } from '../conf'
 import SequelizeRepository from '../database/repositories/sequelizeRepository'
 
 import { CROWD_ANALYTICS_PLATORM_NAME } from './addProductDataToCrowdTenant'
@@ -21,6 +21,7 @@ export default async function identify(
   }).email
   if (
     !IS_TEST_ENV &&
+    !IS_DEV_ENV &&
     SEGMENT_CONFIG.writeKey &&
     // This is only for events in the hosted version. Self-hosted has less telemetry.
     (API_CONFIG.edition === Edition.CROWD_HOSTED || API_CONFIG.edition === Edition.LFX) &&
@@ -41,6 +42,10 @@ export default async function identify(
 
     const { userIdOut, tenantIdOut } = getTenatUser(userId, options)
 
+    if (!userIdOut) {
+      return
+    }
+
     const payload = {
       userId: userIdOut,
       event,
diff --git a/backend/src/services/integrationService.ts b/backend/src/services/integrationService.ts
@@ -4,8 +4,7 @@ import { request } from '@octokit/request'
 import axios, { AxiosRequestConfig, AxiosResponse } from 'axios'
 import lodash from 'lodash'
 import moment from 'moment'
-import { QueryTypes, Transaction } from 'sequelize'
-import { v4 as uuidv4 } from 'uuid'
+import { Transaction } from 'sequelize'
 
 import { EDITION, Error400, Error404, Error542 } from '@crowd/common'
 import { getGithubInstallationToken } from '@crowd/common_services'
@@ -15,6 +14,7 @@ import {
   deleteSegmentRepositories,
   upsertSegmentRepositories,
 } from '@crowd/data-access-layer/src/collections'
+import { syncRepositoriesToGitV2 } from '@crowd/data-access-layer/src/integrations'
 import {
   NangoIntegration,
   connectNangoIntegration,
@@ -908,7 +908,7 @@ export default class IntegrationService {
         args: [{ integrationIds: [integration.id] }],
       })
 
-      return await txService.findById(integrationId)
+      return await this.findById(integration.id)
     } catch (err) {
       this.options.log.error(err, 'Error while creating or updating GitHub integration!')
       if (!existingTransaction) {
@@ -1340,11 +1340,15 @@ export default class IntegrationService {
       )
 
       // upsert repositories to git.repositories in order to be processed by git-integration V2
-      await this.syncRepositoriesToGitV2(
-        remotes,
-        options || this.options,
+      const qx = SequelizeRepository.getQueryExecutor({
+        ...(options || this.options),
         transaction,
+      })
+      await syncRepositoriesToGitV2(
+        qx,
+        remotes,
         integration.id,
+        (options || this.options).currentSegments[0].id,
       )
 
       // Only commit if we created the transaction ourselves
@@ -1362,95 +1366,6 @@ export default class IntegrationService {
     return integration
   }
 
-  /**
-   * Syncs repositories to git.repositories table (git-integration V2)
-   * @param remotes Array of repository objects with url and optional forkedFrom
-   * @param options Repository options
-   * @param transaction Database transaction
-   * @param integrationId The integration ID from the git integration
-   * @param inheritFromExistingRepos If true, queries githubRepos and gitlabRepos for IDs; if false, generates new UUIDs
-   *
-   * TODO: @Mouad After migration is complete, simplify this function by:
-   * 1. Using an object parameter instead of multiple parameters for better maintainability
-   * 2. Removing the inheritFromExistingRepos parameter since git.repositories will be the source of truth
-   * 3. Simplifying the logic to only handle git.repositories operations
-   */
-  private async syncRepositoriesToGitV2(
-    remotes: Array<{ url: string; forkedFrom?: string | null }>,
-    options: IRepositoryOptions,
-    transaction: Transaction,
-    integrationId: string,
-  ) {
-    const seq = SequelizeRepository.getSequelize(options)
-
-    let repositoriesToSync: Array<{
-      id: string
-      url: string
-      integrationId: string
-      segmentId: string
-      forkedFrom?: string | null
-    }> = []
-    // check GitHub repos first, fallback to GitLab repos if none found
-    const existingRepos: Array<{
-      id: string
-      url: string
-    }> = await seq.query(
-      `
-          WITH github_repos AS (
-            SELECT id, url FROM "githubRepos" 
-            WHERE url IN (:urls) AND "deletedAt" IS NULL
-          ),
-          gitlab_repos AS (
-            SELECT id, url FROM "gitlabRepos" 
-            WHERE url IN (:urls) AND "deletedAt" IS NULL
-          )
-          SELECT id, url FROM github_repos
-          UNION ALL
-          SELECT id, url FROM gitlab_repos
-          WHERE NOT EXISTS (SELECT 1 FROM github_repos)
-        `,
-      {
-        replacements: {
-          urls: remotes.map((r) => r.url),
-        },
-        type: QueryTypes.SELECT,
-        transaction,
-      },
-    )
-
-    // Create a map of url to forkedFrom for quick lookup
-    const forkedFromMap = new Map(remotes.map((r) => [r.url, r.forkedFrom]))
-
-    repositoriesToSync = existingRepos.map((repo) => ({
-      id: repo.id,
-      url: repo.url,
-      integrationId,
-      segmentId: options.currentSegments[0].id,
-      forkedFrom: forkedFromMap.get(repo.url) || null,
-    }))
-
-    if (repositoriesToSync.length === 0) {
-      this.options.log.warn(
-        'No existing repos found in githubRepos or gitlabRepos - inserting new to git.repositories with new uuid',
-      )
-      repositoriesToSync = remotes.map((remote) => ({
-        id: uuidv4(), // Generate new UUID
-        url: remote.url,
-        integrationId,
-        segmentId: options.currentSegments[0].id,
-        forkedFrom: remote.forkedFrom || null,
-      }))
-    }
-
-    // Sync to git.repositories v2
-    await GitReposRepository.upsert(repositoriesToSync, {
-      ...options,
-      transaction,
-    })
-
-    this.options.log.info(`Synced ${repositoriesToSync.length} repos to git v2`)
-  }
-
   async atlassianAdminConnect(adminApi: string, organizationId: string) {
     const nangoPayload = {
       params: {
diff --git a/services/apps/cron_service/src/jobs/nangoGithubSync.job.ts b/services/apps/cron_service/src/jobs/nangoGithubSync.job.ts
@@ -1,5 +1,6 @@
 import CronTime from 'cron-time-generator'
 
+import { IS_DEV_ENV } from '@crowd/common'
 import { READ_DB_CONFIG, getDbConnection } from '@crowd/data-access-layer/src/database'
 import { fetchNangoIntegrationData } from '@crowd/data-access-layer/src/integrations'
 import { pgpQx } from '@crowd/data-access-layer/src/queryExecutor'
@@ -12,7 +13,7 @@ import { IJobDefinition } from '../types'
 const job: IJobDefinition = {
   name: 'nango-github-sync',
   cronTime: CronTime.every(
-    Number(process.env.CROWD_GH_NANGO_SYNC_INTERVAL_MINUTES || 60),
+    Number(process.env.CROWD_GH_NANGO_SYNC_INTERVAL_MINUTES || IS_DEV_ENV ? 5 : 60),
   ).minutes(),
   timeout: 10 * 60,
   process: async (ctx) => {
diff --git a/services/apps/nango_worker/src/activities.ts b/services/apps/nango_worker/src/activities.ts
@@ -2,12 +2,14 @@ import {
   analyzeGithubIntegration,
   createGithubConnection,
   deleteConnection,
+  mapGithubRepo,
   numberOfGithubConnectionsToCreate,
   processNangoWebhook,
   removeGithubConnection,
   setGithubConnection,
   startNangoSync,
   unmapGithubRepo,
+  updateGitIntegrationWithRepo,
 } from './activities/nangoActivities'
 
 export {
@@ -18,6 +20,8 @@ export {
   removeGithubConnection,
   setGithubConnection,
   startNangoSync,
+  mapGithubRepo,
   unmapGithubRepo,
   numberOfGithubConnectionsToCreate,
+  updateGitIntegrationWithRepo,
 }
diff --git a/services/apps/nango_worker/src/activities/nangoActivities.ts b/services/apps/nango_worker/src/activities/nangoActivities.ts
@@ -1,7 +1,9 @@
 import { IS_DEV_ENV, IS_STAGING_ENV, singleOrDefault } from '@crowd/common'
 import GithubIntegrationService from '@crowd/common_services/src/services/github.integration.service'
 import {
+  addGitHubRepoMapping,
   addGithubNangoConnection,
+  addRepoToGitIntegration,
   fetchIntegrationById,
   findIntegrationDataForNangoWebhookProcessing,
   removeGitHubRepoMapping,
@@ -55,7 +57,7 @@ export async function numberOfGithubConnectionsToCreate(): Promise<number> {
 
   if (IS_DEV_ENV || IS_STAGING_ENV) {
     svc.log.info('[GITHUB] Number of github connections to create: 5')
-    return 5
+    return 10
   }
 
   const lastConnectDate = await getLastConnectTs()
@@ -186,6 +188,10 @@ export async function analyzeGithubIntegration(
     repo: IGithubRepoData
     connectionId: string
   }[] = []
+  const duplicatesToDelete: {
+    repo: IGithubRepoData
+    connectionId: string
+  }[] = []
   const reposToSync: IGithubRepoData[] = []
 
   const integration = await fetchIntegrationById(dbStoreQx(svc.postgres.writer), integrationId)
@@ -238,19 +244,44 @@ export async function analyzeGithubIntegration(
       if (settings.nangoMapping) {
         const nangoMapping = settings.nangoMapping as Record<string, IGithubRepoData>
 
-        for (const connectionId of Object.keys(nangoMapping)) {
+        const connectionIds = Object.keys(nangoMapping)
+
+        // check for duplicates as well by tracking which repos have connectionIds
+        const existingConnectedRepos = []
+        for (const connectionId of connectionIds) {
           const mappedRepo = nangoMapping[connectionId]
-          const found = singleOrDefault(
-            finalRepos,
-            (r) => r.owner === mappedRepo.owner && r.repoName === mappedRepo.repoName,
-          )
 
-          // if repo is in nangoMapping but not in settings delete the connection
-          if (!found) {
-            reposToDelete.push({
+          if (
+            existingConnectedRepos.some(
+              (r) => r.owner === mappedRepo.owner && r.repoName === mappedRepo.repoName,
+            )
+          ) {
+            // found duplicate connectionId for the same repo
+            duplicatesToDelete.push({
               repo: mappedRepo,
               connectionId,
             })
+
+            // just so that later singleOrDefault doesn't find it
+            delete nangoMapping[connectionId]
+          } else {
+            const found = singleOrDefault(
+              finalRepos,
+              (r) => r.owner === mappedRepo.owner && r.repoName === mappedRepo.repoName,
+            )
+
+            // if repo is in nangoMapping but not in settings delete the connection
+            if (!found) {
+              reposToDelete.push({
+                repo: mappedRepo,
+                connectionId,
+              })
+
+              // just so that later singleOrDefault doesn't find it
+              delete nangoMapping[connectionId]
+            } else {
+              existingConnectedRepos.push(mappedRepo)
+            }
           }
         }
       }
@@ -290,6 +321,7 @@ export async function analyzeGithubIntegration(
   return {
     providerConfigKey: NangoIntegration.GITHUB,
     reposToDelete,
+    duplicatesToDelete,
     reposToSync,
   }
 }
@@ -382,6 +414,15 @@ export async function deleteConnection(
   await deleteNangoConnection(providerConfigKey as NangoIntegration, connectionId)
 }
 
+export async function mapGithubRepo(integrationId: string, repo: IGithubRepoData): Promise<void> {
+  await addGitHubRepoMapping(
+    dbStoreQx(svc.postgres.writer),
+    integrationId,
+    repo.owner,
+    repo.repoName,
+  )
+}
+
 export async function unmapGithubRepo(integrationId: string, repo: IGithubRepoData): Promise<void> {
   // remove repo from githubRepos mapping
   await removeGitHubRepoMapping(
@@ -393,6 +434,14 @@ export async function unmapGithubRepo(integrationId: string, repo: IGithubRepoDa
   )
 }
 
+export async function updateGitIntegrationWithRepo(
+  integrationId: string,
+  repo: IGithubRepoData,
+): Promise<void> {
+  const repoUrl = `https://github.com/${repo.owner}/${repo.repoName}`
+  await addRepoToGitIntegration(dbStoreQx(svc.postgres.writer), integrationId, repoUrl)
+}
+
 function parseGithubUrl(url: string): IGithubRepoData {
   // Create URL object
   const parsedUrl = new URL(url)
diff --git a/services/apps/nango_worker/src/types.ts b/services/apps/nango_worker/src/types.ts
@@ -17,6 +17,10 @@ export interface IGithubIntegrationSyncInstructions {
     repo: IGithubRepoData
     connectionId: string
   }[]
+  duplicatesToDelete: {
+    repo: IGithubRepoData
+    connectionId: string
+  }[]
   reposToSync: IGithubRepoData[]
 }
 
diff --git a/services/apps/nango_worker/src/workflows/syncGithubIntegration.ts b/services/apps/nango_worker/src/workflows/syncGithubIntegration.ts
@@ -26,6 +26,17 @@ export async function syncGithubIntegration(args: ISyncGithubIntegrationArgument
       await activity.unmapGithubRepo(integrationId, repo.repo)
     }
 
+    // delete duplicate connections
+    for (const repo of result.duplicatesToDelete) {
+      // delete nango connection
+      await activity.deleteConnection(result.providerConfigKey, repo.connectionId)
+
+      // delete connection from integrations.settings.nangoMapping object
+      await activity.removeGithubConnection(integrationId, repo.connectionId)
+
+      // we don't unmap because this one was duplicated
+    }
+
     // create connections for repos that are not already connected
     for (const repo of result.reposToSync) {
       if (created >= limit) {
@@ -38,6 +49,12 @@ export async function syncGithubIntegration(args: ISyncGithubIntegrationArgument
       // add connection to integrations.settings.nangoMapping object
       await activity.setGithubConnection(integrationId, repo, connectionId)
 
+      // add repo to githubRepos mapping if it's not already mapped
+      await activity.mapGithubRepo(integrationId, repo)
+
+      // add repo to git integration
+      await activity.updateGitIntegrationWithRepo(integrationId, repo)
+
       // start nango sync
       await activity.startNangoSync(result.providerConfigKey, connectionId)
 
diff --git a/services/archetypes/worker/src/worker.ts b/services/archetypes/worker/src/worker.ts
diff --git a/services/libs/data-access-layer/src/integrations/index.ts b/services/libs/data-access-layer/src/integrations/index.ts
diff --git a/services/libs/nango/src/client.ts b/services/libs/nango/src/client.ts
diff --git a/services/libs/temporal/src/index.ts b/services/libs/temporal/src/index.ts
