diff --git a/services/apps/packages_worker/src/activities.ts b/services/apps/packages_worker/src/activities.ts index ddce9cd27a..7375896cd9 100644 --- a/services/apps/packages_worker/src/activities.ts +++ b/services/apps/packages_worker/src/activities.ts @@ -16,6 +16,7 @@ export { processMavenCriticalBatch } from './maven/activities' export { criticalityComputePageRank, rankPackages } from './criticality/activities' export { cargoDownloadAndLoad, + cargoNormalizeRepos, cargoEnrichPackages, cargoEnrichVersions, cargoEnrichRepos, diff --git a/services/apps/packages_worker/src/cargo/activities.ts b/services/apps/packages_worker/src/cargo/activities.ts index d5e3439021..9c913a0be9 100644 --- a/services/apps/packages_worker/src/cargo/activities.ts +++ b/services/apps/packages_worker/src/cargo/activities.ts @@ -15,6 +15,7 @@ import { flushAudit, } from './enrich' import { STAGING_SCHEMA, loadDump } from './loadDump' +import { normalizeRepos } from './normalizeRepos' import { EnrichDownloadsDailyResult, EnrichMaintainersResult, @@ -22,6 +23,7 @@ import { EnrichReposResult, EnrichVersionsResult, LoadResult, + NormalizeReposResult, } from './types' const log = getServiceChildLogger('cargo-activity') @@ -37,6 +39,10 @@ export async function cargoDownloadAndLoad(): Promise { return result } +export async function cargoNormalizeRepos(): Promise { + return normalizeRepos(await getPackagesDb()) +} + export async function cargoEnrichPackages(): Promise { return enrichPackages(await getPackagesDb()) } diff --git a/services/apps/packages_worker/src/cargo/enrich.ts b/services/apps/packages_worker/src/cargo/enrich.ts index 58e1115103..78aed8aaef 100644 --- a/services/apps/packages_worker/src/cargo/enrich.ts +++ b/services/apps/packages_worker/src/cargo/enrich.ts @@ -48,6 +48,7 @@ export async function enrichPackages(qx: QueryExecutor): Promise { return withTunedSession(qx, 'repos', async (tx) => { const repoRow = await tx.selectOne( `WITH new_repos AS ( INSERT INTO repos (url, host, updated_at) - SELECT DISTINCT e.declared_repository_url, - CASE - WHEN e.declared_repository_url ~* '://([^/]+\\.)?github\\.com(/|$)' THEN 'github' - WHEN e.declared_repository_url ~* '://[^/]*gitlab' THEN 'gitlab' - WHEN e.declared_repository_url ~* '://([^/]+\\.)?bitbucket\\.org(/|$)' THEN 'bitbucket' - ELSE 'other' - END, - NOW() + SELECT DISTINCT rn.repository_url, rn.host, NOW() FROM ${STAGING_SCHEMA}.enrich_packages e - WHERE e.declared_repository_url IS NOT NULL AND e.declared_repository_url LIKE 'http%' + JOIN ${STAGING_SCHEMA}.repo_norm rn ON rn.declared = e.declared_repository_url ON CONFLICT (url) DO NOTHING RETURNING url ), @@ -182,13 +184,47 @@ export async function enrichRepos(qx: QueryExecutor): Promise INSERT INTO ${STAGING_SCHEMA}.audit_changes (package_id, field) SELECT e.package_id, f.field FROM ${STAGING_SCHEMA}.enrich_packages e - JOIN new_repos nr ON nr.url = e.declared_repository_url + JOIN ${STAGING_SCHEMA}.repo_norm rn ON rn.declared = e.declared_repository_url + JOIN new_repos nr ON nr.url = rn.repository_url CROSS JOIN LATERAL (VALUES ('repos.url'), ('repos.host')) AS f(field) RETURNING 1 ) SELECT (SELECT COUNT(*) FROM new_repos)::int AS repos`, ) + // Prunes stale 'declared' links before relinking — covers junk/unparseable declared + // values, rewrites (declared URL now maps elsewhere), and removals (this dump's + // declared_repository_url is NULL, meaning the crate no longer declares a repo at + // all — loadDump.ts stages every matched crate every run, so NULL here is + // authoritative, not "no data this run"). Safe to always prune on that signal because + // the DELETE is scoped to source = 'declared' — cargo only ever removes links it owns. + // Without this, package_repos would accumulate a link to a repo no crate declares + // anymore, and consumers such as security-contacts (which join through + // repos ⋈ package_repos, not packages.repository_url) would keep reading it. + const pruneRow = await tx.selectOne( + `WITH targets AS ( + SELECT e.package_id, r.id AS repo_id + FROM ${STAGING_SCHEMA}.enrich_packages e + LEFT JOIN ${STAGING_SCHEMA}.repo_norm rn ON rn.declared = e.declared_repository_url + LEFT JOIN repos r ON r.url = rn.repository_url + ), + del AS ( + DELETE FROM package_repos pr + USING targets t + WHERE pr.package_id = t.package_id + AND pr.source = $(source) + AND (t.repo_id IS NULL OR pr.repo_id IS DISTINCT FROM t.repo_id) + RETURNING pr.package_id + ), + ins_audit AS ( + INSERT INTO ${STAGING_SCHEMA}.audit_changes (package_id, field) + SELECT package_id, 'package_repos.repo_id' FROM del + RETURNING 1 + ) + SELECT (SELECT COUNT(*) FROM del)::int AS pruned`, + { source: REPO_LINK_SOURCE }, + ) + const linkRow = await tx.selectOne( `WITH old AS ( SELECT pr.package_id, pr.repo_id, pr.source, pr.confidence @@ -201,11 +237,13 @@ export async function enrichRepos(qx: QueryExecutor): Promise INSERT INTO package_repos (package_id, repo_id, source, confidence, created_at, verified_at) SELECT e.package_id, r.id, $(source), $(confidence), NOW(), NOW() FROM ${STAGING_SCHEMA}.enrich_packages e - JOIN repos r ON r.url = e.declared_repository_url - WHERE e.declared_repository_url IS NOT NULL + JOIN ${STAGING_SCHEMA}.repo_norm rn ON rn.declared = e.declared_repository_url + JOIN repos r ON r.url = rn.repository_url + -- Leaves source untouched on conflict — a link another enricher already owns for + -- this (package_id, repo_id) keeps its provenance instead of being reassigned to + -- 'declared', matching upsertMavenPackageRepo's confidence-only merge. ON CONFLICT (package_id, repo_id) DO UPDATE SET - source = EXCLUDED.source, - confidence = EXCLUDED.confidence, + confidence = GREATEST(EXCLUDED.confidence, package_repos.confidence), verified_at = NOW() RETURNING package_id, repo_id, source, confidence ), @@ -228,7 +266,7 @@ export async function enrichRepos(qx: QueryExecutor): Promise { source: REPO_LINK_SOURCE, confidence: REPO_LINK_CONFIDENCE }, ) - return { repos: repoRow.repos, links: linkRow.links } + return { repos: repoRow.repos, links: linkRow.links, pruned: pruneRow.pruned } }) } diff --git a/services/apps/packages_worker/src/cargo/normalizeRepos.ts b/services/apps/packages_worker/src/cargo/normalizeRepos.ts new file mode 100644 index 0000000000..f32b7aff4d --- /dev/null +++ b/services/apps/packages_worker/src/cargo/normalizeRepos.ts @@ -0,0 +1,72 @@ +import { QueryExecutor } from '@crowd/data-access-layer' +import { getServiceChildLogger } from '@crowd/logging' + +import { CanonicalRepo, canonicalizeRepoUrl } from '../utils/canonicalizeRepoUrl' + +import { STAGING_SCHEMA } from './loadDump' +import { NormalizeReposResult } from './types' + +const log = getServiceChildLogger('cargo-normalize') + +// Batch size for the mapping upsert — keeps parameter arrays well under Postgres limits. +const INSERT_BATCH = 5000 + +/** + * Builds `cargo_sync.repo_norm(declared, repository_url, host)`, mapping every + * distinct `declared_repository_url` staged in `enrich_packages` to its canonical + * `https:////` form via the shared `canonicalizeRepoUrl` + * (same normalizer npm/pypi use — no per-ecosystem fork). + * + * Only inputs `canonicalizeRepoUrl` can reduce to an owner/name pair are stored; + * unparseable URLs or those with fewer than two path segments are omitted, so a + * LEFT JOIN from `enrich_packages` yields NULL — that both recovers derivable + * URLs (Gap A) and clears that class of junk from `packages.repository_url` + * (Gap C). Note this does not validate that the result is an actual repository + * host — a URL-shaped homepage with ≥2 path segments (e.g. + * `https://example.com/owner/repo`) still canonicalizes and is stored. + * + * Normalization runs in TypeScript because the bulk set-based cargo pipeline + * cannot call the parser per row; the resulting mapping table lets the SQL + * enrich phases join back to it. Idempotent: the table is rebuilt each run. + */ +export async function normalizeRepos(qx: QueryExecutor): Promise { + const rows: Array<{ declared_repository_url: string }> = await qx.select( + `SELECT DISTINCT declared_repository_url + FROM ${STAGING_SCHEMA}.enrich_packages + WHERE declared_repository_url IS NOT NULL`, + ) + + await qx.result( + `DROP TABLE IF EXISTS ${STAGING_SCHEMA}.repo_norm CASCADE; + CREATE TABLE ${STAGING_SCHEMA}.repo_norm ( + declared text PRIMARY KEY, + repository_url text NOT NULL, + host text NOT NULL + )`, + ) + + const mapped = rows + .map(({ declared_repository_url }) => ({ + declared: declared_repository_url, + canonical: canonicalizeRepoUrl(declared_repository_url), + })) + .filter((r): r is { declared: string; canonical: CanonicalRepo } => r.canonical !== null) + + for (let i = 0; i < mapped.length; i += INSERT_BATCH) { + const batch = mapped.slice(i, i + INSERT_BATCH) + await qx.result( + `INSERT INTO ${STAGING_SCHEMA}.repo_norm (declared, repository_url, host) + SELECT * FROM unnest($(declared)::text[], $(urls)::text[], $(hosts)::text[]) + ON CONFLICT (declared) DO NOTHING`, + { + declared: batch.map((r) => r.declared), + urls: batch.map((r) => r.canonical.url), + hosts: batch.map((r) => r.canonical.host), + }, + ) + } + + const result: NormalizeReposResult = { scanned: rows.length, normalized: mapped.length } + log.info(result, 'cargo repo normalization complete') + return result +} diff --git a/services/apps/packages_worker/src/cargo/types.ts b/services/apps/packages_worker/src/cargo/types.ts index bd1ac08d06..aafa402340 100644 --- a/services/apps/packages_worker/src/cargo/types.ts +++ b/services/apps/packages_worker/src/cargo/types.ts @@ -12,6 +12,11 @@ export interface LoadResult { durationMs: number } +export interface NormalizeReposResult { + scanned: number + normalized: number +} + export interface EnrichPackagesResult { updated: number } @@ -23,6 +28,7 @@ export interface EnrichVersionsResult { export interface EnrichReposResult { repos: number links: number + pruned: number } export interface EnrichMaintainersResult { diff --git a/services/apps/packages_worker/src/cargo/workflows.ts b/services/apps/packages_worker/src/cargo/workflows.ts index 1edb6dbf81..44ad1c4bd7 100644 --- a/services/apps/packages_worker/src/cargo/workflows.ts +++ b/services/apps/packages_worker/src/cargo/workflows.ts @@ -14,6 +14,7 @@ const { cargoDownloadAndLoad } = proxyActivities({ }) const { + cargoNormalizeRepos, cargoEnrichPackages, cargoEnrichVersions, cargoEnrichRepos, @@ -31,6 +32,8 @@ export async function cargoSyncWorkflow(): Promise { const load = await cargoDownloadAndLoad() log.info('cargoSync loaded dump', { ...load }) + const normalizedRepos = await cargoNormalizeRepos() + log.info('cargoSync normalized repos', { ...normalizedRepos }) const packages = await cargoEnrichPackages() const versions = await cargoEnrichVersions() const repos = await cargoEnrichRepos() @@ -41,6 +44,7 @@ export async function cargoSyncWorkflow(): Promise { log.info('cargoSync complete', { matched: load.matched, + normalizedRepos, packages, versions, repos,