|
| 1 | +import { deleteMavenPackageRepoLinks } from '@crowd/data-access-layer' |
| 2 | +import { |
| 3 | + NpmRepoUrlRow, |
| 4 | + getOrCreateRepoByUrl, |
| 5 | + listNpmPackagesForRepoUrlRecompute, |
| 6 | + updateNpmRepositoryUrls, |
| 7 | + upsertPackageRepo, |
| 8 | +} from '@crowd/data-access-layer/src/packages' |
| 9 | +import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor' |
| 10 | +import { getServiceChildLogger } from '@crowd/logging' |
| 11 | + |
| 12 | +import { canonicalizeRepoUrl } from '../utils/canonicalizeRepoUrl' |
| 13 | + |
| 14 | +const log = getServiceChildLogger('npm-repo-url-backfill') |
| 15 | + |
| 16 | +// Postgres deadlock (40P01) is transient: concurrent transactions upserting the same shared |
| 17 | +// rows (e.g. the same repo linked from many packages) can form a lock cycle. Re-running the |
| 18 | +// whole transaction resolves it — the upserts are idempotent. Mirrors Maven's |
| 19 | +// withDeadlockRetry (src/maven/runMavenEnrichmentLoop.ts). |
| 20 | +async function withDeadlockRetry<T>(fn: () => Promise<T>, maxAttempts = 4): Promise<T> { |
| 21 | + for (let attempt = 1; ; attempt++) { |
| 22 | + try { |
| 23 | + return await fn() |
| 24 | + } catch (err) { |
| 25 | + const code = (err as { code?: string }).code |
| 26 | + const isDeadlock = |
| 27 | + code === '40P01' || /deadlock detected/i.test(String((err as Error)?.message)) |
| 28 | + if (isDeadlock && attempt < maxAttempts) { |
| 29 | + await new Promise((r) => setTimeout(r, 50 * attempt + Math.random() * 100)) |
| 30 | + log.debug({ attempt }, 'Deadlock detected — retrying transaction') |
| 31 | + continue |
| 32 | + } |
| 33 | + throw err |
| 34 | + } |
| 35 | + } |
| 36 | +} |
| 37 | + |
| 38 | +export type RepoUrlBackfillTotals = { |
| 39 | + scanned: number |
| 40 | + filled: number // NULL → canonical value |
| 41 | + cleared: number // canonical value → NULL (e.g. a stricter normalizer no longer accepts it) |
| 42 | + rewritten: number // non-canonical value → different canonical value |
| 43 | + unchanged: number |
| 44 | + linked: number // repos/package_repos link (re)written for a fill or rewrite |
| 45 | + pruned: number // stale 'declared' link removed for a clear or rewrite |
| 46 | +} |
| 47 | + |
| 48 | +/** |
| 49 | + * Recomputes `repository_url` for every npm row directly from the stored |
| 50 | + * `declared_repository_url`, applying the current `canonicalizeRepoUrl`. No |
| 51 | + * packument is re-fetched — the raw declared value is already in the DB. Fills |
| 52 | + * recoverable NULLs (e.g. scp-form ssh:// URLs the previous normalizer dropped). |
| 53 | + * |
| 54 | + * `cleared` only accounts for values a *future* normalizer change stops |
| 55 | + * accepting for a row that already has a repository_url — the current |
| 56 | + * canonicalizeRepoUrl only ever adds recognized cases, so this backfill |
| 57 | + * doesn't clear any values on its own today. |
| 58 | + * |
| 59 | + * Link table: package_repos is kept consistent with the recomputed |
| 60 | + * repository_url. For a fill (previously null) or a rewrite (value changes) the |
| 61 | + * canonical repo link is (re)written via getOrCreateRepoByUrl/upsertPackageRepo — |
| 62 | + * the same pair the ingest path uses. A rewrite first prunes the stale |
| 63 | + * source='declared' link so a package never carries two 'declared' links to |
| 64 | + * different repos. |
| 65 | + * |
| 66 | + * Idempotent and resumable: the id cursor is derived from the scan, so a |
| 67 | + * re-run after an interrupt simply reprocesses from the start and skips rows |
| 68 | + * that already match. |
| 69 | + */ |
| 70 | +export async function backfillNpmRepositoryUrls( |
| 71 | + qx: QueryExecutor, |
| 72 | + options: { |
| 73 | + batchSize: number |
| 74 | + dryRun: boolean |
| 75 | + criticalOnly: boolean |
| 76 | + isShuttingDown: () => boolean |
| 77 | + }, |
| 78 | +): Promise<RepoUrlBackfillTotals> { |
| 79 | + const { batchSize, dryRun, criticalOnly, isShuttingDown } = options |
| 80 | + const totals: RepoUrlBackfillTotals = { |
| 81 | + scanned: 0, |
| 82 | + filled: 0, |
| 83 | + cleared: 0, |
| 84 | + rewritten: 0, |
| 85 | + unchanged: 0, |
| 86 | + linked: 0, |
| 87 | + pruned: 0, |
| 88 | + } |
| 89 | + |
| 90 | + let afterId = '0' |
| 91 | + for (;;) { |
| 92 | + if (isShuttingDown()) { |
| 93 | + log.info('Shutdown requested — stopping after the current batch.') |
| 94 | + break |
| 95 | + } |
| 96 | + |
| 97 | + const rows: NpmRepoUrlRow[] = await listNpmPackagesForRepoUrlRecompute(qx, { |
| 98 | + afterId, |
| 99 | + limit: batchSize, |
| 100 | + criticalOnly, |
| 101 | + }) |
| 102 | + if (rows.length === 0) break |
| 103 | + |
| 104 | + const updates: { id: string; repositoryUrl: string | null }[] = [] |
| 105 | + // Rows that had a link and now change to a different one — their stale 'declared' link is pruned. |
| 106 | + const pruneTargets: number[] = [] |
| 107 | + // Rows that gained/changed a canonical URL — their repo link is (re)written after the update. |
| 108 | + const linkTargets: { id: string; repositoryUrl: string; host: string }[] = [] |
| 109 | + for (const row of rows) { |
| 110 | + totals.scanned++ |
| 111 | + // declaredRepositoryUrl is guaranteed non-null by the query's IS NOT NULL filter. |
| 112 | + const canonical = canonicalizeRepoUrl(row.declaredRepositoryUrl as string) |
| 113 | + const desired = canonical?.url ?? null |
| 114 | + if (desired === row.repositoryUrl) { |
| 115 | + totals.unchanged++ |
| 116 | + continue |
| 117 | + } |
| 118 | + if (row.repositoryUrl === null) totals.filled++ |
| 119 | + else if (desired === null) totals.cleared++ |
| 120 | + else totals.rewritten++ |
| 121 | + updates.push({ id: row.id, repositoryUrl: desired }) |
| 122 | + if (row.repositoryUrl !== null && desired !== row.repositoryUrl) { |
| 123 | + pruneTargets.push(Number(row.id)) |
| 124 | + } |
| 125 | + if (canonical) |
| 126 | + linkTargets.push({ id: row.id, repositoryUrl: canonical.url, host: canonical.host }) |
| 127 | + } |
| 128 | + |
| 129 | + if (updates.length > 0 && !dryRun) { |
| 130 | + // Atomic per batch: the repository_url UPDATE, the stale-link prune, and the |
| 131 | + // relink must commit together — otherwise an interrupt between them leaves |
| 132 | + // packages.repository_url updated but package_repos out of sync, and a |
| 133 | + // re-run would skip the row (its repository_url already matches `desired`), |
| 134 | + // so the inconsistency would never be repaired. |
| 135 | + await withDeadlockRetry(() => |
| 136 | + qx.tx(async (t: QueryExecutor) => { |
| 137 | + await updateNpmRepositoryUrls(t, updates) |
| 138 | + if (pruneTargets.length > 0) await deleteMavenPackageRepoLinks(t, pruneTargets) |
| 139 | + for (const target of linkTargets) { |
| 140 | + const { id: repoId } = await getOrCreateRepoByUrl(t, target.repositoryUrl, target.host) |
| 141 | + await upsertPackageRepo(t, target.id, repoId, 'declared', 0.8) |
| 142 | + } |
| 143 | + }), |
| 144 | + ) |
| 145 | + totals.pruned += pruneTargets.length |
| 146 | + totals.linked += linkTargets.length |
| 147 | + } |
| 148 | + |
| 149 | + afterId = rows[rows.length - 1].id |
| 150 | + log.info( |
| 151 | + { afterId, changes: updates.length, dryRun, criticalOnly, ...totals }, |
| 152 | + 'Backfill progress', |
| 153 | + ) |
| 154 | + } |
| 155 | + |
| 156 | + return totals |
| 157 | +} |
0 commit comments