fix: re-use canonicalizeRepoUrl & insert package_repos in go and nuget (#4331)

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; declaredRepositoryUrl: string | null }
 
 // 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, declared_repository_url AS "declaredRepositoryUrl" 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,60 @@ 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
+    const declaredRepo = repo
+      ? null
+      : row.declaredRepositoryUrl
+        ? canonicalizeRepoUrl(row.declaredRepositoryUrl)
+        : 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 ?? declaredRepo?.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[]
+
+      const repoToLink = repo ?? declaredRepo
+      if (repoToLink) {
+        const { id: repoId, changedFields: repoChanged } = await getOrCreateRepoByUrl(
+          t,
+          repoToLink.url,
+          repoToLink.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/client.ts

@@ -8,6 +8,7 @@ import {
 } from './types'
 
 const SERVICE_INDEX_URL = 'https://api.nuget.org/v3/index.json'
+const FLAT_CONTAINER_BASE_URL = 'https://api.nuget.org/v3-flatcontainer'
 
 interface ServiceIndexResource {
   '@id': string
@@ -144,3 +145,29 @@ export async function fetchRegistration(
     throw err
   }
 }
+
+// The registration API never exposes the nuspec <repository> element — it must be read from the
+// raw nuspec XML, served by the flat-container endpoint.
+export async function fetchNuspec(
+  packageId: string,
+  version: string,
+): Promise<string | NuGetFetchError> {
+  const lowerId = packageId.toLowerCase()
+  const lowerVersion = version.toLowerCase()
+
+  try {
+    const resp = await axios.get<string>(
+      `${FLAT_CONTAINER_BASE_URL}/${lowerId}/${lowerVersion}/${lowerId}.nuspec`,
+      {
+        responseType: 'text',
+        transformResponse: (data) => data,
+        timeout: 15000,
+      },
+    )
+    return resp.data
+  } catch (err) {
+    const classified = classifyError(err)
+    if (classified) return classified
+    throw err
+  }
+}

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

@@ -1,3 +1,7 @@
+import { XMLParser } from 'fast-xml-parser'
+
+import { canonicalizeRepoUrl } from '../utils/canonicalizeRepoUrl'
+
 import {
   NormalizedNuGetPackage,
   NormalizedNuGetVersion,
@@ -26,18 +30,20 @@ function parsePublishedDate(published: string | undefined): Date | null {
   return !isNaN(date.getTime()) && date.getUTCFullYear() > 1900 ? date : null
 }
 
-const SCM_HOSTS = ['github.com', 'gitlab.com', 'bitbucket.org']
+const nuspecParser = new XMLParser({ ignoreAttributes: false, attributeNamePrefix: '@_' })
 
-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/')
+export function parseNuspecRepositoryUrl(nuspecXml: string): string | null {
+  try {
+    const doc = nuspecParser.parse(nuspecXml)
+    const url = doc?.package?.metadata?.repository?.['@_url']
+    return typeof url === 'string' && url.trim() !== '' ? url.trim() : null
+  } catch {
+    return null
+  }
 }
 
+const SCM_HOSTS = ['github.com', 'gitlab.com', 'bitbucket.org']
+
 function isScmUrl(url: string | undefined): boolean {
   if (!url) return false
   try {
@@ -64,6 +70,7 @@ export function normalizeNuGetPackage(
   packageId: string,
   searchResult: NuGetSearchItem | null,
   registration: NuGetRegistrationIndex,
+  nuspecXml?: string | null,
 ): NormalizedNuGetPackage {
   const allLeaves = registration.items.flatMap((page) => page.items ?? [])
   const allEntries: NuGetCatalogEntry[] = allLeaves.map((leaf) => leaf.catalogEntry)
@@ -96,10 +103,13 @@ export function normalizeNuGetPackage(
         ...(latestEntry ? [latestEntry] : []),
       ]
     : [...allEntries].reverse()
-  const nuspecRepoUrl = entriesForRepo.find((e) => e.repository?.url)?.repository?.url
+  const catalogRepoUrl = entriesForRepo.find((e) => e.repository?.url)?.repository?.url
+  const fetchedNuspecRepoUrl = nuspecXml ? parseNuspecRepositoryUrl(nuspecXml) : null
+  const nuspecRepoUrl = fetchedNuspecRepoUrl ?? catalogRepoUrl
   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 +164,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,12 +11,13 @@ import {
   upsertMaintainer,
   upsertNuGetPackage,
   upsertNuGetVersionsBatch,
+  upsertPackageRepo,
 } from '@crowd/data-access-layer'
 import { getServiceChildLogger } from '@crowd/logging'
 
 import { getNuGetConfig } from '../config'
 
-import { fetchRegistration, fetchSearch } from './client'
+import { fetchNuspec, fetchRegistration, fetchSearch } from './client'
 import { normalizeNuGetPackage } from './normalize'
 import { BatchResult, isNuGetFetchError } from './types'
 
@@ -96,7 +98,15 @@ async function processPackage(
 
   const searchItem = isNuGetFetchError(searchResult) ? null : searchResult
 
-  const normalized = normalizeNuGetPackage(packageId, searchItem, registrationResult)
+  const preliminary = normalizeNuGetPackage(packageId, searchItem, registrationResult)
+
+  let nuspecXml: string | null = null
+  if (preliminary.latestVersion) {
+    const nuspecResult = await fetchNuspec(packageId, preliminary.latestVersion)
+    nuspecXml = isNuGetFetchError(nuspecResult) ? null : nuspecResult
+  }
+
+  const normalized = normalizeNuGetPackage(packageId, searchItem, registrationResult, nuspecXml)
 
   await withDeadlockRetry(() =>
     qx.tx(async (t) => {
@@ -108,7 +118,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 +132,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

services/apps/packages_worker/src/utils/canonicalizeRepoUrl.ts

@@ -68,17 +68,18 @@ export function canonicalizeRepoUrl(raw: string): CanonicalRepo | null {
   const segments = u.pathname.split('/').filter(Boolean)
   if (segments.length < 2) return null
 
-  let owner = segments[0]
-  let name = segments[1].replace(/\.git$/, '')
-  if (!owner || !name) return null
+  const isKnownHost = hostname in HOST_ENUM
+  let ownerPath = isKnownHost ? [segments[0]] : segments.slice(0, -1)
+  let name = (isKnownHost ? segments[1] : segments[segments.length - 1]).replace(/\.git$/, '')
+  if (!name || ownerPath.length === 0 || ownerPath.some((seg) => !seg)) return null
 
   if (CASE_INSENSITIVE_HOSTS.has(hostname)) {
-    owner = owner.toLowerCase()
+    ownerPath = ownerPath.map((seg) => seg.toLowerCase())
     name = name.toLowerCase()
   }
 
   return {
-    url: `https://${hostname}/${owner}/${name}`,
+    url: `https://${hostname}/${[...ownerPath, name].join('/')}`,
     host: HOST_ENUM[hostname] ?? 'other',
   }
 }