fix: cargo repo gap (CM-1305)

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,

services/apps/packages_worker/src/cargo/activities.ts

@@ -15,13 +15,15 @@ import {
   flushAudit,
 } from './enrich'
 import { STAGING_SCHEMA, loadDump } from './loadDump'
+import { normalizeRepos } from './normalizeRepos'
 import {
   EnrichDownloadsDailyResult,
   EnrichMaintainersResult,
   EnrichPackagesResult,
   EnrichReposResult,
   EnrichVersionsResult,
   LoadResult,
+  NormalizeReposResult,
 } from './types'
 
 const log = getServiceChildLogger('cargo-activity')
@@ -37,6 +39,10 @@ export async function cargoDownloadAndLoad(): Promise<LoadResult> {
   return result
 }
 
+export async function cargoNormalizeRepos(): Promise<NormalizeReposResult> {
+  return normalizeRepos(await getPackagesDb())
+}
+
 export async function cargoEnrichPackages(): Promise<EnrichPackagesResult> {
   return enrichPackages(await getPackagesDb())
 }

services/apps/packages_worker/src/cargo/enrich.ts

@@ -48,6 +48,7 @@ export async function enrichPackages(qx: QueryExecutor): Promise<EnrichPackagesR
     tx.selectOne(
       `WITH snap AS (
          SELECT p.id, p.status, p.description, p.homepage, p.declared_repository_url,
+                p.repository_url,
                 p.licenses, p.licenses_raw, p.keywords, p.versions_count, p.latest_version,
                 p.first_release_at, p.latest_release_at, p.dependent_count,
                 p.dependent_repos_count, p.downloads_last_30d
@@ -60,6 +61,8 @@ export async function enrichPackages(qx: QueryExecutor): Promise<EnrichPackagesR
            description             = COALESCE(e.description, p.description),
            homepage                = COALESCE(e.homepage, p.homepage),
            declared_repository_url = COALESCE(e.declared_repository_url, p.declared_repository_url),
+           repository_url          = CASE WHEN e.declared_repository_url IS NOT NULL
+                                          THEN rn.repository_url ELSE p.repository_url END,
            licenses                = COALESCE(e.licenses, p.licenses),
            licenses_raw            = COALESCE(e.licenses_raw, p.licenses_raw),
            keywords                = COALESCE(e.keywords, p.keywords),
@@ -73,18 +76,22 @@ export async function enrichPackages(qx: QueryExecutor): Promise<EnrichPackagesR
            ingestion_source        = $(ingestionSource),
            last_synced_at          = NOW()
          FROM ${STAGING_SCHEMA}.enrich_packages e
+         LEFT JOIN ${STAGING_SCHEMA}.repo_norm rn ON rn.declared = e.declared_repository_url
          WHERE p.id = e.package_id
          RETURNING p.id
        ),
        diff AS (
          SELECT s.id AS package_id, f.field
          FROM snap s
          JOIN ${STAGING_SCHEMA}.enrich_packages e ON e.package_id = s.id
+         LEFT JOIN ${STAGING_SCHEMA}.repo_norm rn ON rn.declared = e.declared_repository_url
          CROSS JOIN LATERAL (VALUES
            ('packages.status',                  s.status                  IS DISTINCT FROM e.status),
            ('packages.description',             s.description             IS DISTINCT FROM COALESCE(e.description, s.description)),
            ('packages.homepage',                s.homepage                IS DISTINCT FROM COALESCE(e.homepage, s.homepage)),
            ('packages.declared_repository_url', s.declared_repository_url IS DISTINCT FROM COALESCE(e.declared_repository_url, s.declared_repository_url)),
+           ('packages.repository_url',          s.repository_url          IS DISTINCT FROM
+               CASE WHEN e.declared_repository_url IS NOT NULL THEN rn.repository_url ELSE s.repository_url END),
            ('packages.licenses',                s.licenses                IS DISTINCT FROM COALESCE(e.licenses, s.licenses)),
            ('packages.licenses_raw',            s.licenses_raw            IS DISTINCT FROM COALESCE(e.licenses_raw, s.licenses_raw)),
            ('packages.keywords',                s.keywords                IS DISTINCT FROM COALESCE(e.keywords, s.keywords)),
@@ -159,36 +166,65 @@ export async function enrichVersions(qx: QueryExecutor): Promise<EnrichVersionsR
   return { upserted: row.upserted }
 }
 
-// Writes only url + host — other repo fields belong to the GitHub enricher.
+// Writes only url + host — other repo fields belong to the GitHub enricher. Uses
+// repo_norm (built by normalizeRepos) so repos.url/package_repos always agree with
+// the canonical packages.repository_url — never the raw declared_repository_url.
 export async function enrichRepos(qx: QueryExecutor): Promise<EnrichReposResult> {
   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
        ),
        ins_audit AS (
          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<EnrichReposResult>
          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<EnrichReposResult>
       { 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 }
   })
 }
 

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://<host>/<owner>/<name>` 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<NormalizeReposResult> {
+  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
+}

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 {

services/apps/packages_worker/src/cargo/workflows.ts

@@ -14,6 +14,7 @@ const { cargoDownloadAndLoad } = proxyActivities<typeof activities>({
 })
 
 const {
+  cargoNormalizeRepos,
   cargoEnrichPackages,
   cargoEnrichVersions,
   cargoEnrichRepos,
@@ -31,6 +32,8 @@ export async function cargoSyncWorkflow(): Promise<void> {
   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<void> {
 
   log.info('cargoSync complete', {
     matched: load.matched,
+    normalizedRepos,
     packages,
     versions,
     repos,