fix: npm repository url gap (CM-1305)

services/apps/packages_worker/package.json

@@ -56,6 +56,8 @@
     "backfill:maven:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=maven LOG_LEVEL=info tsx src/bin/maven-backfill.ts",
     "backfill:maven-repo-url": "SERVICE=maven tsx src/bin/maven-repo-url-backfill.ts",
     "backfill:maven-repo-url:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=maven LOG_LEVEL=info tsx src/bin/maven-repo-url-backfill.ts",
+    "backfill:npm-repo-url": "SERVICE=npm tsx src/bin/npm-repo-url-backfill.ts",
+    "backfill:npm-repo-url:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=npm LOG_LEVEL=info tsx src/bin/npm-repo-url-backfill.ts",
     "import:maven-maintainers": "SERVICE=maven tsx src/maven/scripts/importMaintainersFromCsv.ts",
     "import:maven-maintainers:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=maven tsx src/maven/scripts/importMaintainersFromCsv.ts",
     "import:sonatype-popularity": "SERVICE=maven tsx src/maven/scripts/importSonatypePopularityFromCsv.ts",

services/apps/packages_worker/src/bin/npm-repo-url-backfill.ts

@@ -0,0 +1,61 @@
+import { getServiceLogger } from '@crowd/logging'
+
+import { getPackagesDb } from '../db'
+import { backfillNpmRepositoryUrls } from '../npm/backfillRepositoryUrl'
+
+const log = getServiceLogger()
+
+let shuttingDown = false
+
+// Graceful stop: finish the in-flight batch, then exit. Safe to interrupt — every
+// write is an idempotent UPDATE recomputed from declared_repository_url, so
+// re-running simply reprocesses and skips rows that already match.
+const shutdown = () => {
+  if (shuttingDown) return
+  shuttingDown = true
+  log.info('Shutting down npm repo-url backfill (stopping after the current batch)...')
+}
+
+process.on('SIGINT', shutdown)
+process.on('SIGTERM', shutdown)
+
+const DEFAULT_BATCH_SIZE = 5000
+
+const main = async () => {
+  const dryRun = process.argv.includes('--dry-run')
+  const criticalOnly = process.argv.includes('--critical-only')
+  const rawBatchSize = process.env.NPM_REPO_URL_BACKFILL_BATCH_SIZE
+  const parsedBatchSize = rawBatchSize === undefined ? DEFAULT_BATCH_SIZE : Number(rawBatchSize)
+  if (!Number.isInteger(parsedBatchSize) || parsedBatchSize <= 0) {
+    log.error(
+      { NPM_REPO_URL_BACKFILL_BATCH_SIZE: rawBatchSize },
+      'NPM_REPO_URL_BACKFILL_BATCH_SIZE must be a positive integer',
+    )
+    process.exit(1)
+  }
+  const batchSize = parsedBatchSize
+
+  log.info(
+    { dryRun, criticalOnly, batchSize },
+    'npm repo-url backfill starting (recompute canonicalizeRepoUrl from declared_repository_url, no packument fetch)...',
+  )
+
+  const qx = await getPackagesDb()
+  await qx.selectOne('SELECT 1')
+  log.info('Connected to packages-db.')
+
+  const totals = await backfillNpmRepositoryUrls(qx, {
+    batchSize,
+    dryRun,
+    criticalOnly,
+    isShuttingDown: () => shuttingDown,
+  })
+
+  log.info({ ...totals, dryRun }, 'npm repo-url backfill complete')
+  process.exit(0)
+}
+
+main().catch((err) => {
+  log.error({ err }, 'npm repo-url backfill fatal error')
+  process.exit(1)
+})

services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts

@@ -0,0 +1,163 @@
+import {
+  NpmRepoUrlRow,
+  getOrCreateRepoByUrl,
+  listNpmPackagesForRepoUrlRecompute,
+  updateNpmRepositoryUrls,
+  upsertPackageRepo,
+} from '@crowd/data-access-layer/src/packages'
+import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+import { getServiceChildLogger } from '@crowd/logging'
+
+import { canonicalizeRepoUrl } from '../utils/canonicalizeRepoUrl'
+
+const log = getServiceChildLogger('npm-repo-url-backfill')
+
+// Postgres deadlock (40P01) is transient: concurrent transactions upserting the same shared
+// rows (e.g. the same repo linked from many packages) can form a lock cycle. Re-running the
+// whole transaction resolves it — the upserts are idempotent. Mirrors Maven's
+// withDeadlockRetry (src/maven/runMavenEnrichmentLoop.ts).
+async function withDeadlockRetry<T>(fn: () => Promise<T>, maxAttempts = 4): Promise<T> {
+  for (let attempt = 1; ; attempt++) {
+    try {
+      return await fn()
+    } catch (err) {
+      const code = (err as { code?: string }).code
+      const isDeadlock =
+        code === '40P01' || /deadlock detected/i.test(String((err as Error)?.message))
+      if (isDeadlock && attempt < maxAttempts) {
+        await new Promise((r) => setTimeout(r, 50 * attempt + Math.random() * 100))
+        log.debug({ attempt }, 'Deadlock detected — retrying transaction')
+        continue
+      }
+      throw err
+    }
+  }
+}
+
+export type RepoUrlBackfillTotals = {
+  scanned: number
+  filled: number // NULL → canonical value
+  cleared: number // canonical value → NULL (e.g. a stricter normalizer no longer accepts it)
+  rewritten: number // non-canonical value → different canonical value
+  unchanged: number
+  linked: number // repos/package_repos link (re)written for a fill or rewrite
+  pruned: number // stale 'declared' link removed for a clear or rewrite
+}
+
+/**
+ * Recomputes `repository_url` for every npm row directly from the stored
+ * `declared_repository_url`, applying the current `canonicalizeRepoUrl`. No
+ * packument is re-fetched — the raw declared value is already in the DB. Fills
+ * recoverable NULLs (e.g. scp-form ssh:// URLs the previous normalizer dropped).
+ *
+ * `cleared` only accounts for values a *future* normalizer change stops
+ * accepting for a row that already has a repository_url — the current
+ * canonicalizeRepoUrl only ever adds recognized cases, so this backfill
+ * doesn't clear any values on its own today.
+ *
+ * Link table: package_repos is kept consistent with the recomputed
+ * repository_url. For a fill (previously null) or a rewrite (value changes) the
+ * canonical repo link is (re)written via getOrCreateRepoByUrl/upsertPackageRepo —
+ * the same pair the ingest path uses. A rewrite first prunes the stale
+ * source='declared' link so a package never carries two 'declared' links to
+ * different repos.
+ *
+ * Idempotent and resumable: the id cursor is derived from the scan, so a
+ * re-run after an interrupt simply reprocesses from the start and skips rows
+ * that already match.
+ */
+export async function backfillNpmRepositoryUrls(
+  qx: QueryExecutor,
+  options: {
+    batchSize: number
+    dryRun: boolean
+    criticalOnly: boolean
+    isShuttingDown: () => boolean
+  },
+): Promise<RepoUrlBackfillTotals> {
+  const { batchSize, dryRun, criticalOnly, isShuttingDown } = options
+  const totals: RepoUrlBackfillTotals = {
+    scanned: 0,
+    filled: 0,
+    cleared: 0,
+    rewritten: 0,
+    unchanged: 0,
+    linked: 0,
+    pruned: 0,
+  }
+
+  let afterId = '0'
+  for (;;) {
+    if (isShuttingDown()) {
+      log.info('Shutdown requested — stopping after the current batch.')
+      break
+    }
+
+    const rows: NpmRepoUrlRow[] = await listNpmPackagesForRepoUrlRecompute(qx, {
+      afterId,
+      limit: batchSize,
+      criticalOnly,
+    })
+    if (rows.length === 0) break
+
+    const updates: { id: string; repositoryUrl: string | null }[] = []
+    // Rows that had a link and now change to a different one — their stale 'declared' link is pruned.
+    // Kept as strings (packages.id is bigint) — Number() coercion would lose precision above 2^53.
+    const pruneTargets: string[] = []
+    // Rows that gained/changed a canonical URL — their repo link is (re)written after the update.
+    const linkTargets: { id: string; repositoryUrl: string; host: string }[] = []
+    for (const row of rows) {
+      totals.scanned++
+      // declaredRepositoryUrl is guaranteed non-null by the query's IS NOT NULL filter.
+      const canonical = canonicalizeRepoUrl(row.declaredRepositoryUrl as string)
+      const desired = canonical?.url ?? null
+      if (desired === row.repositoryUrl) {
+        totals.unchanged++
+        continue
+      }
+      if (row.repositoryUrl === null) totals.filled++
+      else if (desired === null) totals.cleared++
+      else totals.rewritten++
+      updates.push({ id: row.id, repositoryUrl: desired })
+      if (row.repositoryUrl !== null && desired !== row.repositoryUrl) {
+        pruneTargets.push(row.id)
+      }
+      if (canonical)
+        linkTargets.push({ id: row.id, repositoryUrl: canonical.url, host: canonical.host })
+    }
+
+    if (updates.length > 0 && !dryRun) {
+      // Atomic per batch: the repository_url UPDATE, the stale-link prune, and the
+      // relink must commit together — otherwise an interrupt between them leaves
+      // packages.repository_url updated but package_repos out of sync, and a
+      // re-run would skip the row (its repository_url already matches `desired`),
+      // so the inconsistency would never be repaired.
+      await withDeadlockRetry(() =>
+        qx.tx(async (t: QueryExecutor) => {
+          await updateNpmRepositoryUrls(t, updates)
+          if (pruneTargets.length > 0) {
+            await t.result(
+              `DELETE FROM package_repos
+               WHERE package_id = ANY($(packageIds)::bigint[]) AND source = 'declared'`,
+              { packageIds: pruneTargets },
+            )
+          }
+          for (const target of linkTargets) {
+            const { id: repoId } = await getOrCreateRepoByUrl(t, target.repositoryUrl, target.host)
+            await upsertPackageRepo(t, target.id, repoId, 'declared', 0.8)
+          }
+        }),
+      )
+      totals.pruned += pruneTargets.length
+      totals.linked += linkTargets.length
+    }
+
+    afterId = rows[rows.length - 1].id
+    log.info(
+      { afterId, changes: updates.length, dryRun, criticalOnly, ...totals },
+      'Backfill progress',
+    )
+  }
+
+  return totals
+}

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

@@ -22,6 +22,13 @@ describe('canonicalizeRepoUrl', () => {
     ['gitlab:group/project', 'https://gitlab.com/group/project', 'gitlab'],
     ['bitbucket:team/repo', 'https://bitbucket.org/team/repo', 'bitbucket'],
     ['https://example.com/owner/repo', 'https://example.com/owner/repo', 'other'],
+    ['git+https://github.com/1aGh/md-claude.git', 'https://github.com/1agh/md-claude', 'github'],
+    [
+      'ssh://git@github.com:1inch/limit-order-protocol-utils.git',
+      'https://github.com/1inch/limit-order-protocol-utils',
+      'github',
+    ],
+    ['ssh://git@github.com:2222/foo/bar.git', 'https://github.com/foo/bar', 'github'],
   ])('canonicalizes %s', (input, expectedUrl, expectedHost) => {
     expect(canonicalizeRepoUrl(input)).toEqual({ url: expectedUrl, host: expectedHost })
   })
@@ -44,10 +51,14 @@ describe('canonicalizeRepoUrl', () => {
     })
   })
 
-  it.each([['not a url'], ['https://github.com/onlyowner'], [''], ['   ']])(
-    'returns null for unparseable input %s',
-    (input) => {
-      expect(canonicalizeRepoUrl(input)).toBeNull()
-    },
-  )
+  it.each([
+    ['not a url'],
+    ['https://github.com/onlyowner'],
+    [''],
+    ['   '],
+    ['123'],
+    ['https://github.com/Wscats'],
+  ])('returns null for unparseable input %s', (input) => {
+    expect(canonicalizeRepoUrl(input)).toBeNull()
+  })
 })

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

@@ -54,7 +54,18 @@ export function canonicalizeRepoUrl(raw: string): CanonicalRepo | null {
     s = `https://${scp[1]}/${scp[2]}`
   }
 
-  s = s.replace(/^ssh:\/\/git@([^/]+)\//, 'https://$1/')
+  // ssh:// with an scp-style `host:path` (colon instead of slash) is not valid URL
+  // syntax — the part after `:` looks like a port to the URL parser and throws.
+  // Rewrite it before the generic ssh://git@host/path case below. A numeric-only
+  // segment before the next `/` is a real port (e.g. `ssh://git@host:2222/owner/repo`),
+  // not an scp-style owner — leave those for the generic case, which URL parses fine.
+  const sshScp = s.match(/^ssh:\/\/git@([^/:]+):(.+)$/)
+  const sshScpIsPort = sshScp ? /^\d+$/.test(sshScp[2].split('/')[0]) : false
+  if (sshScp && !sshScpIsPort) {
+    s = `https://${sshScp[1]}/${sshScp[2]}`
+  } else {
+    s = s.replace(/^ssh:\/\/git@([^/]+)\//, 'https://$1/')
+  }
   s = s.replace(/^git:\/\//, 'https://')
 
   let u: URL

services/libs/data-access-layer/src/packages/packages.ts

@@ -226,6 +226,99 @@ export async function getTrackedNpmPackages(
   }))
 }
 
+// ─── npm repository_url backfill ──────────────────────────────────────────────
+
+export type NpmRepoUrlRow = {
+  id: string
+  declaredRepositoryUrl: string | null
+  repositoryUrl: string | null
+}
+
+/**
+ * Keyset-paginated scan of npm rows that carry a declared repository value.
+ * The backfill recomputes repository_url *from* declared_repository_url, so rows
+ * with no declared value are skipped — there is nothing to recompute from, and a
+ * null declaration must never be used to clear an existing repository_url that a
+ * different source may have set.
+ *
+ * `criticalOnly` restricts the scan to is_critical rows — used for a fast,
+ * consumer-facing first pass.
+ */
+export async function listNpmPackagesForRepoUrlRecompute(
+  qx: QueryExecutor,
+  options: { afterId: string; limit: number; criticalOnly?: boolean },
+): Promise<NpmRepoUrlRow[]> {
+  const rows: Array<{
+    id: string
+    declared_repository_url: string | null
+    repository_url: string | null
+  }> = await qx.select(
+    `
+      SELECT
+        packages.id::text AS id,
+        declared_repository_url,
+        repository_url
+      FROM packages
+      WHERE ecosystem = 'npm'
+        ${options.criticalOnly ? 'AND is_critical' : ''}
+        AND id > $(afterId)::bigint
+        AND declared_repository_url IS NOT NULL
+      ORDER BY packages.id ASC
+      LIMIT $(limit)
+      `,
+    { afterId: options.afterId, limit: options.limit },
+  )
+  return rows.map((r) => ({
+    id: r.id,
+    declaredRepositoryUrl: r.declared_repository_url,
+    repositoryUrl: r.repository_url,
+  }))
+}
+
+/**
+ * Applies a batch of recomputed repository_url values via direct UPDATE — the
+ * only way to clear a stale value, since the enrichment upsert writes
+ * EXCLUDED.repository_url unconditionally but is only ever called with a fresh
+ * packument fetch. Splits clears (→ NULL) from sets to avoid NULLs inside a
+ * text[] array literal. Also bumps last_synced_at so the correction is picked
+ * up by any downstream export keyed off it.
+ */
+export async function updateNpmRepositoryUrls(
+  qx: QueryExecutor,
+  updates: { id: string; repositoryUrl: string | null }[],
+): Promise<void> {
+  if (updates.length === 0) return
+
+  const toClear = updates.filter((u) => u.repositoryUrl === null).map((u) => u.id)
+  const toSet = updates.filter(
+    (u): u is { id: string; repositoryUrl: string } => u.repositoryUrl !== null,
+  )
+
+  if (toClear.length > 0) {
+    await qx.result(
+      `UPDATE packages SET repository_url = NULL, last_synced_at = NOW()
+       WHERE id = ANY($(ids)::bigint[]) AND repository_url IS NOT NULL`,
+      { ids: toClear },
+    )
+  }
+
+  if (toSet.length > 0) {
+    await qx.result(
+      `
+      UPDATE packages p
+      SET repository_url = v.repository_url, last_synced_at = NOW()
+      FROM (
+        SELECT unnest($(ids)::bigint[]) AS id,
+               unnest($(urls)::text[])  AS repository_url
+      ) v
+      WHERE p.id = v.id
+        AND p.repository_url IS DISTINCT FROM v.repository_url
+      `,
+      { ids: toSet.map((u) => u.id), urls: toSet.map((u) => u.repositoryUrl) },
+    )
+  }
+}
+
 // How many critical PyPI packages exist — a cheap guard so the daily downloads workflow can
 // skip its BigQuery scan entirely when there are none to ingest (the merge scopes to is_critical).
 export async function getCriticalPypiPackageCount(qx: QueryExecutor): Promise<number> {