fix: re-use canonicalizeRepoUrl & insert package_repos

Signed-off-by: Mouad BANI <mouad-mb@outlook.com>

Author: mbani01

services/apps/packages_worker/src/go/activities.ts

@@ -1,11 +1,16 @@
 import { Context } from '@temporalio/activity'
 
-import { logAuditFieldChanges } from '@crowd/data-access-layer/src/packages'
+import {
+  getOrCreateRepoByUrl,
+  logAuditFieldChanges,
+  upsertPackageRepo,
+} from '@crowd/data-access-layer/src/packages'
 import type { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
 import { getServiceChildLogger } from '@crowd/logging'
 
 import { getGoConfig } from '../config'
 import { getPackagesDb } from '../db'
+import { canonicalizeRepoUrl } from '../utils/canonicalizeRepoUrl'
 
 import { fetchStatus } from './pkgGoDevClient'
 import { fetchLatest } from './proxyClient'
@@ -21,7 +26,7 @@ export interface GoScanCursor {
   after: string
 }
 
-type GoRow = { purl: string; name: string }
+type GoRow = { id: string; purl: string; name: string }
 
 // Two independent purl-keyset cursors — one for critical packages, one for everything else —
 // each ordered/paginated purely by purl so WHERE and ORDER BY always match (no gaps, no
@@ -37,7 +42,7 @@ async function getGoBatch(
 ): Promise<GoRow[]> {
   if (batchSize <= 0) return []
   return qx.select(
-    `SELECT purl, name FROM packages
+    `SELECT id::text AS id, purl, name FROM packages
      WHERE ecosystem = 'go' AND is_critical = $(isCritical) AND purl > $(after)
      ORDER BY purl ASC
      LIMIT $(limit)`,
@@ -75,7 +80,7 @@ export async function enrichGoVersionsBatch(
 
   const { fetchTimeoutMs, proxyConcurrency } = getGoConfig()
 
-  const enrichOne = async (row: { purl: string; name: string }): Promise<void> => {
+  const enrichOne = async (row: GoRow): Promise<void> => {
     Context.current().heartbeat(row.purl)
     const result = await fetchLatest(row.name, fetchTimeoutMs)
     if (isFetchError(result)) {
@@ -85,37 +90,54 @@ export async function enrichGoVersionsBatch(
       )
       return
     }
-    const changed = await qx.selectOne(
-      `WITH old AS (
-         SELECT latest_version AS v, latest_release_at AS t, repository_url AS r
-         FROM packages WHERE purl = $(purl)
-       ),
-       upd AS (
-         UPDATE packages p SET
-           latest_version    = $(version),
-           latest_release_at = $(releaseAt),
-           repository_url    = COALESCE($(repoUrl), p.repository_url),
-           last_synced_at    = NOW()
-         WHERE p.purl = $(purl)
-         RETURNING latest_version AS v, latest_release_at AS t, repository_url AS r
-       )
-       SELECT
-         (SELECT v FROM old) IS DISTINCT FROM (SELECT v FROM upd) AS v_changed,
-         (SELECT t FROM old) IS DISTINCT FROM (SELECT t FROM upd) AS t_changed,
-         (SELECT r FROM old) IS DISTINCT FROM (SELECT r FROM upd) AS r_changed`,
-      {
-        version: result.version,
-        releaseAt: result.releaseAt,
-        repoUrl: result.repoUrl,
-        purl: row.purl,
-      },
-    )
-    const changedFields = [
-      changed?.v_changed ? 'packages.latest_version' : null,
-      changed?.t_changed ? 'packages.latest_release_at' : null,
-      changed?.r_changed ? 'packages.repository_url' : null,
-    ].filter(Boolean) as string[]
-    await logAuditFieldChanges(qx, PROXY_SOURCE, row.purl, changedFields)
+    const repo = result.repoUrl ? canonicalizeRepoUrl(result.repoUrl) : null
+
+    await qx.tx(async (t) => {
+      const changed = await t.selectOne(
+        `WITH old AS (
+           SELECT latest_version AS v, latest_release_at AS t, repository_url AS r
+           FROM packages WHERE purl = $(purl)
+         ),
+         upd AS (
+           UPDATE packages p SET
+             latest_version    = $(version),
+             latest_release_at = $(releaseAt),
+             repository_url    = COALESCE($(repoUrl), p.repository_url),
+             last_synced_at    = NOW()
+           WHERE p.purl = $(purl)
+           RETURNING latest_version AS v, latest_release_at AS t, repository_url AS r
+         )
+         SELECT
+           (SELECT v FROM old) IS DISTINCT FROM (SELECT v FROM upd) AS v_changed,
+           (SELECT t FROM old) IS DISTINCT FROM (SELECT t FROM upd) AS t_changed,
+           (SELECT r FROM old) IS DISTINCT FROM (SELECT r FROM upd) AS r_changed`,
+        {
+          version: result.version,
+          releaseAt: result.releaseAt,
+          repoUrl: repo?.url ?? null,
+          purl: row.purl,
+        },
+      )
+      const changedFields = [
+        changed?.v_changed ? 'packages.latest_version' : null,
+        changed?.t_changed ? 'packages.latest_release_at' : null,
+        changed?.r_changed ? 'packages.repository_url' : null,
+      ].filter(Boolean) as string[]
+
+      if (repo) {
+        const { id: repoId, changedFields: repoChanged } = await getOrCreateRepoByUrl(
+          t,
+          repo.url,
+          repo.host,
+        )
+        changedFields.push(...repoChanged)
+
+        const linkChanged = await upsertPackageRepo(t, row.id, repoId, 'declared', 0.8)
+        changedFields.push(...linkChanged)
+      }
+
+      await logAuditFieldChanges(t, PROXY_SOURCE, row.purl, changedFields)
+    })
   }
 
   for (let i = 0; i < rows.length; i += proxyConcurrency) {

services/apps/packages_worker/src/nuget/normalize.ts

@@ -1,3 +1,5 @@
+import { canonicalizeRepoUrl } from '../utils/canonicalizeRepoUrl'
+
 import {
   NormalizedNuGetPackage,
   NormalizedNuGetVersion,
@@ -28,16 +30,6 @@ function parsePublishedDate(published: string | undefined): Date | null {
 
 const SCM_HOSTS = ['github.com', 'gitlab.com', 'bitbucket.org']
 
-function normalizeRepoUrl(url: string | undefined): string | null {
-  if (!url) return null
-  return url
-    .trim()
-    .replace(/\.git$/, '')
-    .replace(/^git\+/, '')
-    .replace(/^git:\/\//, 'https://')
-    .replace(/^http:\/\/github\.com\//, 'https://github.com/')
-}
-
 function isScmUrl(url: string | undefined): boolean {
   if (!url) return false
   try {
@@ -98,8 +90,9 @@ export function normalizeNuGetPackage(
     : [...allEntries].reverse()
   const nuspecRepoUrl = entriesForRepo.find((e) => e.repository?.url)?.repository?.url
   const declaredRepositoryUrl = nuspecRepoUrl ?? null
-  const repoCandidate = nuspecRepoUrl ?? (isScmUrl(homepage) ? homepage : null)
-  const repositoryUrl = normalizeRepoUrl(repoCandidate ?? undefined)
+  const repo =
+    (nuspecRepoUrl ? canonicalizeRepoUrl(nuspecRepoUrl) : null) ??
+    (isScmUrl(homepage) ? canonicalizeRepoUrl(homepage) : null)
 
   const keywords = searchResult?.tags && searchResult.tags.length > 0 ? searchResult.tags : null
 
@@ -154,7 +147,7 @@ export function normalizeNuGetPackage(
     description,
     homepage: homepage || null,
     declaredRepositoryUrl,
-    repositoryUrl,
+    repo,
     licenses,
     licensesRaw,
     keywords,

services/apps/packages_worker/src/nuget/runNuGetEnrichmentLoop.ts

@@ -2,6 +2,7 @@ import {
   IDbNuGetVersionUpsert,
   NuGetPackageToSync,
   QueryExecutor,
+  getOrCreateRepoByUrl,
   listNuGetPackagesToSync,
   logAuditFieldChange,
   markNuGetPackageError,
@@ -10,6 +11,7 @@ import {
   upsertMaintainer,
   upsertNuGetPackage,
   upsertNuGetVersionsBatch,
+  upsertPackageRepo,
 } from '@crowd/data-access-layer'
 import { getServiceChildLogger } from '@crowd/logging'
 
@@ -108,7 +110,7 @@ async function processPackage(
         description: normalized.description,
         homepage: normalized.homepage,
         declaredRepositoryUrl: normalized.declaredRepositoryUrl,
-        repositoryUrl: normalized.repositoryUrl,
+        repositoryUrl: normalized.repo?.url ?? null,
         licenses: normalized.licenses,
         licensesRaw: normalized.licensesRaw,
         keywords: normalized.keywords,
@@ -122,6 +124,24 @@ async function processPackage(
       })
       pkgChanged.forEach((f) => changed.add(f))
 
+      if (normalized.repo) {
+        const { id: repoId, changedFields: repoChanged } = await getOrCreateRepoByUrl(
+          t,
+          normalized.repo.url,
+          normalized.repo.host,
+        )
+        repoChanged.forEach((f) => changed.add(f))
+
+        const linkChanged = await upsertPackageRepo(
+          t,
+          packageDbId.toString(),
+          repoId,
+          'declared',
+          0.8,
+        )
+        linkChanged.forEach((f) => changed.add(f))
+      }
+
       if (normalized.versions.length > 0) {
         const versionRows: IDbNuGetVersionUpsert[] = normalized.versions.map((v) => ({
           packageId: packageDbId,

services/apps/packages_worker/src/nuget/types.ts

@@ -1,3 +1,5 @@
+import { CanonicalRepo } from '../utils/canonicalizeRepoUrl'
+
 export interface NuGetConfig {
   batchSize: number
   concurrency: number
@@ -73,7 +75,7 @@ export interface NormalizedNuGetPackage {
   description: string | null
   homepage: string | null
   declaredRepositoryUrl: string | null
-  repositoryUrl: string | null
+  repo: CanonicalRepo | null
   licenses: string[] | null
   licensesRaw: string | null
   keywords: string[] | null