From f6d8ec4938977d5f64d2959a8f72d3ed59a94a91 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Mon, 13 Jul 2026 18:05:23 +0200 Subject: [PATCH 1/4] fix: npm repository url gap Signed-off-by: Umberto Sgueglia --- services/apps/packages_worker/package.json | 2 + .../src/bin/npm-repo-url-backfill.ts | 61 +++++++ .../src/npm/backfillRepositoryUrl.ts | 157 ++++++++++++++++++ .../__tests__/canonicalizeRepoUrl.test.ts | 22 ++- .../src/utils/canonicalizeRepoUrl.ts | 10 +- .../src/packages/packages.ts | 93 +++++++++++ 6 files changed, 338 insertions(+), 7 deletions(-) create mode 100644 services/apps/packages_worker/src/bin/npm-repo-url-backfill.ts create mode 100644 services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts diff --git a/services/apps/packages_worker/package.json b/services/apps/packages_worker/package.json index 9a490af507..3b567e3345 100644 --- a/services/apps/packages_worker/package.json +++ b/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", diff --git a/services/apps/packages_worker/src/bin/npm-repo-url-backfill.ts b/services/apps/packages_worker/src/bin/npm-repo-url-backfill.ts new file mode 100644 index 0000000000..8c6b995803 --- /dev/null +++ b/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) +}) diff --git a/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts b/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts new file mode 100644 index 0000000000..e7fa67c339 --- /dev/null +++ b/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts @@ -0,0 +1,157 @@ +import { deleteMavenPackageRepoLinks } from '@crowd/data-access-layer' +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(fn: () => Promise, maxAttempts = 4): Promise { + 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 { + 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. + const pruneTargets: number[] = [] + // 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(Number(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 deleteMavenPackageRepoLinks(t, 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 +} diff --git a/services/apps/packages_worker/src/utils/__tests__/canonicalizeRepoUrl.test.ts b/services/apps/packages_worker/src/utils/__tests__/canonicalizeRepoUrl.test.ts index e986512efd..5e1bc58151 100644 --- a/services/apps/packages_worker/src/utils/__tests__/canonicalizeRepoUrl.test.ts +++ b/services/apps/packages_worker/src/utils/__tests__/canonicalizeRepoUrl.test.ts @@ -22,6 +22,12 @@ 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', + ], ])('canonicalizes %s', (input, expectedUrl, expectedHost) => { expect(canonicalizeRepoUrl(input)).toEqual({ url: expectedUrl, host: expectedHost }) }) @@ -44,10 +50,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() + }) }) diff --git a/services/apps/packages_worker/src/utils/canonicalizeRepoUrl.ts b/services/apps/packages_worker/src/utils/canonicalizeRepoUrl.ts index 6e24e3218c..078da2a75c 100644 --- a/services/apps/packages_worker/src/utils/canonicalizeRepoUrl.ts +++ b/services/apps/packages_worker/src/utils/canonicalizeRepoUrl.ts @@ -54,7 +54,15 @@ 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. + const sshScp = s.match(/^ssh:\/\/git@([^/:]+):(.+)$/) + if (sshScp) { + s = `https://${sshScp[1]}/${sshScp[2]}` + } else { + s = s.replace(/^ssh:\/\/git@([^/]+)\//, 'https://$1/') + } s = s.replace(/^git:\/\//, 'https://') let u: URL diff --git a/services/libs/data-access-layer/src/packages/packages.ts b/services/libs/data-access-layer/src/packages/packages.ts index 3a18d0cd82..65b597606c 100644 --- a/services/libs/data-access-layer/src/packages/packages.ts +++ b/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 { + const rows: Array<{ + id: string + declared_repository_url: string | null + repository_url: string | null + }> = await qx.select( + ` + SELECT + 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 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 { + 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 { From cb5773920ad0345f6a66d1d964a653b9cea05eeb Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Mon, 13 Jul 2026 18:22:43 +0200 Subject: [PATCH 2/4] fix: inconsistencies Signed-off-by: Umberto Sgueglia --- .../src/npm/backfillRepositoryUrl.ts | 14 ++++++++++---- .../utils/__tests__/canonicalizeRepoUrl.test.ts | 1 + .../src/utils/canonicalizeRepoUrl.ts | 7 +++++-- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts b/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts index e7fa67c339..2efa0cded9 100644 --- a/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts +++ b/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts @@ -1,4 +1,3 @@ -import { deleteMavenPackageRepoLinks } from '@crowd/data-access-layer' import { NpmRepoUrlRow, getOrCreateRepoByUrl, @@ -103,7 +102,8 @@ export async function backfillNpmRepositoryUrls( 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. - const pruneTargets: number[] = [] + // 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) { @@ -120,7 +120,7 @@ export async function backfillNpmRepositoryUrls( else totals.rewritten++ updates.push({ id: row.id, repositoryUrl: desired }) if (row.repositoryUrl !== null && desired !== row.repositoryUrl) { - pruneTargets.push(Number(row.id)) + pruneTargets.push(row.id) } if (canonical) linkTargets.push({ id: row.id, repositoryUrl: canonical.url, host: canonical.host }) @@ -135,7 +135,13 @@ export async function backfillNpmRepositoryUrls( await withDeadlockRetry(() => qx.tx(async (t: QueryExecutor) => { await updateNpmRepositoryUrls(t, updates) - if (pruneTargets.length > 0) await deleteMavenPackageRepoLinks(t, pruneTargets) + 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) diff --git a/services/apps/packages_worker/src/utils/__tests__/canonicalizeRepoUrl.test.ts b/services/apps/packages_worker/src/utils/__tests__/canonicalizeRepoUrl.test.ts index 5e1bc58151..544c368b30 100644 --- a/services/apps/packages_worker/src/utils/__tests__/canonicalizeRepoUrl.test.ts +++ b/services/apps/packages_worker/src/utils/__tests__/canonicalizeRepoUrl.test.ts @@ -28,6 +28,7 @@ describe('canonicalizeRepoUrl', () => { '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 }) }) diff --git a/services/apps/packages_worker/src/utils/canonicalizeRepoUrl.ts b/services/apps/packages_worker/src/utils/canonicalizeRepoUrl.ts index 078da2a75c..a308697a45 100644 --- a/services/apps/packages_worker/src/utils/canonicalizeRepoUrl.ts +++ b/services/apps/packages_worker/src/utils/canonicalizeRepoUrl.ts @@ -56,9 +56,12 @@ export function canonicalizeRepoUrl(raw: string): CanonicalRepo | null { // 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. + // 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@([^/:]+):(.+)$/) - if (sshScp) { + 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/') From 05a58458a73d03633a0dc532767febe160610dba Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Tue, 14 Jul 2026 10:08:20 +0200 Subject: [PATCH 3/4] fix: filter by packages id Signed-off-by: Umberto Sgueglia --- services/libs/data-access-layer/src/packages/packages.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/services/libs/data-access-layer/src/packages/packages.ts b/services/libs/data-access-layer/src/packages/packages.ts index 65b597606c..1fc9bb0f2f 100644 --- a/services/libs/data-access-layer/src/packages/packages.ts +++ b/services/libs/data-access-layer/src/packages/packages.ts @@ -255,7 +255,7 @@ export async function listNpmPackagesForRepoUrlRecompute( }> = await qx.select( ` SELECT - id::text AS id, + packages.id::text AS id, declared_repository_url, repository_url FROM packages @@ -263,7 +263,7 @@ export async function listNpmPackagesForRepoUrlRecompute( ${options.criticalOnly ? 'AND is_critical' : ''} AND id > $(afterId)::bigint AND declared_repository_url IS NOT NULL - ORDER BY id ASC + ORDER BY packages.id ASC LIMIT $(limit) `, { afterId: options.afterId, limit: options.limit }, From 054077db4b69f1509726964ff5e4e1d7b0cafb92 Mon Sep 17 00:00:00 2001 From: Umberto Sgueglia Date: Tue, 14 Jul 2026 10:43:23 +0200 Subject: [PATCH 4/4] fix: remove dead code for backfill Signed-off-by: Umberto Sgueglia --- services/apps/packages_worker/package.json | 2 - .../src/bin/npm-repo-url-backfill.ts | 61 ------- .../src/npm/backfillRepositoryUrl.ts | 163 ------------------ .../src/packages/packages.ts | 93 ---------- 4 files changed, 319 deletions(-) delete mode 100644 services/apps/packages_worker/src/bin/npm-repo-url-backfill.ts delete mode 100644 services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts diff --git a/services/apps/packages_worker/package.json b/services/apps/packages_worker/package.json index 3b567e3345..9a490af507 100644 --- a/services/apps/packages_worker/package.json +++ b/services/apps/packages_worker/package.json @@ -56,8 +56,6 @@ "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", diff --git a/services/apps/packages_worker/src/bin/npm-repo-url-backfill.ts b/services/apps/packages_worker/src/bin/npm-repo-url-backfill.ts deleted file mode 100644 index 8c6b995803..0000000000 --- a/services/apps/packages_worker/src/bin/npm-repo-url-backfill.ts +++ /dev/null @@ -1,61 +0,0 @@ -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) -}) diff --git a/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts b/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts deleted file mode 100644 index 2efa0cded9..0000000000 --- a/services/apps/packages_worker/src/npm/backfillRepositoryUrl.ts +++ /dev/null @@ -1,163 +0,0 @@ -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(fn: () => Promise, maxAttempts = 4): Promise { - 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 { - 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 -} diff --git a/services/libs/data-access-layer/src/packages/packages.ts b/services/libs/data-access-layer/src/packages/packages.ts index 1fc9bb0f2f..3a18d0cd82 100644 --- a/services/libs/data-access-layer/src/packages/packages.ts +++ b/services/libs/data-access-layer/src/packages/packages.ts @@ -226,99 +226,6 @@ 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 { - 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 { - 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 {