feat: pom extractor

Signed-off-by: Umberto Sgueglia <usgueglia@contractor.linuxfoundation.org>

Author: ulemons

scripts/cli

@@ -1060,7 +1060,7 @@ while test $# -gt 0; do
         exit
         ;;
     clean-start-dev)
-        # IGNORED_SERVICES=("python-worker" "job-generator" "discord-ws" "webhook-api" "profiles-worker" "organizations-enrichment-worker" "merge-suggestions-worker" "members-enrichment-worker" "exports-worker" "entity-merging-worker")
+        IGNORED_SERVICES=("python-worker" "job-generator" "discord-ws" "webhook-api" "profiles-worker" "organizations-enrichment-worker" "merge-suggestions-worker" "members-enrichment-worker" "exports-worker" "entity-merging-worker")
         CLEAN_START=1
         DEV=1
         start

scripts/services/packages-worker.yaml

@@ -8,6 +8,10 @@ x-env-args: &env-args
   CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
+  POM_FETCHER_BATCH_SIZE: '50'
+  POM_FETCHER_CONCURRENCY: '3'
+  POM_FETCHER_STALE_DAYS: '7'
+  POM_FETCHER_IDLE_SLEEP_SEC: '3600'
 
 services:
   packages-worker:

scripts/services/pom-fetcher.yaml

@@ -6,8 +6,9 @@ x-env-args: &env-args
   SERVICE: pom-fetcher
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
-  POM_FETCHER_BATCH_SIZE: '200'
-  POM_FETCHER_CONCURRENCY: '10'
+  LOG_LEVEL: 'info'
+  POM_FETCHER_BATCH_SIZE: '50'
+  POM_FETCHER_CONCURRENCY: '3'
   POM_FETCHER_STALE_DAYS: '7'
   POM_FETCHER_IDLE_SLEEP_SEC: '3600'
 

services/apps/packages_worker/package.json

@@ -17,6 +17,9 @@
     "export-to-bucket": "SERVICE=deps-dev-ingest tsx src/scripts/exportToBucket.ts",
     "export-to-bucket:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=deps-dev-ingest tsx src/scripts/exportToBucket.ts",
     "monitor:osspckgs:local": "bash -c 'set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && node ../../../scripts/monitor-osspckgs.mjs'",
+    "start:pom-fetcher": "SERVICE=pom-fetcher tsx src/bin/pom-fetcher.ts",
+    "dev:pom-fetcher": "SERVICE=pom-fetcher LOG_LEVEL=info nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9235 src/bin/pom-fetcher.ts",
+    "dev:pom-fetcher:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && SERVICE=pom-fetcher LOG_LEVEL=info nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9235 src/bin/pom-fetcher.ts",
     "lint": "npx eslint --ext .ts src --max-warnings=0",
     "format": "npx prettier --write \"src/**/*.ts\"",
     "format-check": "npx prettier --check .",

services/apps/packages_worker/src/config.ts

@@ -35,8 +35,8 @@ export function getEnricherConfig() {
 
 export function getPomFetcherConfig() {
   return {
-    batchSize: parseInt(process.env.POM_FETCHER_BATCH_SIZE ?? '200', 10),
-    concurrency: parseInt(process.env.POM_FETCHER_CONCURRENCY ?? '10', 10),
+    batchSize: parseInt(process.env.POM_FETCHER_BATCH_SIZE ?? '50', 10),
+    concurrency: parseInt(process.env.POM_FETCHER_CONCURRENCY ?? '3', 10),
     staleDays: parseInt(process.env.POM_FETCHER_STALE_DAYS ?? '7', 10),
     idleSleepSec: parseInt(process.env.POM_FETCHER_IDLE_SLEEP_SEC ?? '3600', 10),
   }

services/apps/packages_worker/src/pom-fetcher/runPomEnrichmentLoop.ts

@@ -27,24 +27,24 @@ interface BatchResult {
 
 async function processBatch(
   qx: QueryExecutor,
-  offset: number,
   config: ReturnType<typeof getPomFetcherConfig>,
 ): Promise<BatchResult> {
   const packages = await listMavenPackagesToEnrich(qx, {
     limit: config.batchSize,
-    offset,
+    offset: 0,
     staleDays: config.staleDays,
   })
 
   if (packages.length === 0) {
     return { processed: 0, skipped: 0, errors: 0 }
   }
 
-  log.info({ offset, count: packages.length }, 'Processing POM batch...')
+  log.info({ count: packages.length }, 'Processing POM batch...')
 
   let processed = 0
   let skipped = 0
   let errors = 0
+  const PROGRESS_EVERY = 25
 
   // Process in small concurrent groups to be polite to Maven Central
   for (let i = 0; i < packages.length; i += config.concurrency) {
@@ -62,16 +62,30 @@ async function processBatch(
         }
 
         try {
-          log.info({ groupId, artifactId }, 'Fetching POM...')
-
           // Step 1: resolve latest version from maven-metadata.xml
           const version = await resolveLatestVersion(groupId, artifactId)
           if (!version) {
             log.warn({ groupId, artifactId }, 'Could not resolve latest version, skipping')
+            // Upsert a minimal record so last_synced_at is set — prevents this package
+            // from re-appearing in every batch within the same pass.
+            // ingestionSource 'pom_fetcher_no_version' marks that it was tried but had no
+            // resolvable version on Maven Central (404 on maven-metadata.xml).
+            await upsertPackage(qx, {
+              purl: `pkg:maven/${groupId}/${artifactId}`,
+              ecosystem: 'maven',
+              namespace: groupId,
+              name: artifactId,
+              description: null,
+              homepage: null,
+              declaredRepositoryUrl: null,
+              licenses: null,
+              licensesRaw: null,
+              latestVersion: null,
+              ingestionSource: 'pom_fetcher_no_version',
+            })
             skipped++
             return
           }
-          log.info({ groupId, artifactId, version }, 'Version resolved, extracting POM...')
 
           // Step 2: fetch + resolve POM (follows parent chain)
           const result = await extractArtifact(groupId, artifactId, version, (msg) => {
@@ -83,19 +97,6 @@ async function processBatch(
             errors++
             return
           }
-          log.info(
-            {
-              groupId,
-              artifactId,
-              version,
-              licenses: result.licenses,
-              scmUrl: result.scmUrl,
-              developers: result.developers.length,
-              contributors: result.contributors.length,
-              parentHops: result.parentHops,
-            },
-            'POM extracted, upserting...',
-          )
 
           // Step 3: upsert into `packages`
           // purl at package level has no version (package-level identifier)
@@ -151,6 +152,17 @@ async function processBatch(
         }
       }),
     )
+
+    // done = packages processed so far (based on loop index, always accurate)
+    const done = i + group.length
+    const prevDone = i
+    const crossedBoundary = Math.floor(done / PROGRESS_EVERY) > Math.floor(prevDone / PROGRESS_EVERY)
+    if (crossedBoundary || done === packages.length) {
+      log.info(
+        { done, total: packages.length, processed, skipped, errors },
+        `Progress: ${done}/${packages.length}`,
+      )
+    }
   }
 
   return { processed, skipped, errors }
@@ -170,21 +182,20 @@ export async function runPomEnrichmentLoop(
   config: ReturnType<typeof getPomFetcherConfig>,
   isShuttingDown: () => boolean,
 ): Promise<void> {
-  let offset = 0
   let totalProcessed = 0
   let totalSkipped = 0
   let totalErrors = 0
   let passNumber = 0
   let passStartedAt = Date.now()
 
   while (!isShuttingDown()) {
-    if (offset === 0) {
+    if (totalProcessed + totalSkipped + totalErrors === 0) {
       passNumber++
       passStartedAt = Date.now()
       log.info({ pass: passNumber }, 'Starting pass')
     }
 
-    const result = await processBatch(qx, offset, config)
+    const result = await processBatch(qx, config)
 
     if (result.processed + result.skipped + result.errors === 0) {
       // Nothing left in this pass — log summary and sleep
@@ -200,21 +211,18 @@ export async function runPomEnrichmentLoop(
         `Pass complete. Sleeping ${config.idleSleepSec}s before next pass.`,
       )
       await new Promise((r) => setTimeout(r, config.idleSleepSec * 1000))
-      offset = 0
       totalProcessed = 0
       totalSkipped = 0
       totalErrors = 0
-      passStartedAt = Date.now()
       continue
     }
 
     totalProcessed += result.processed
     totalSkipped += result.skipped
     totalErrors += result.errors
-    offset += config.batchSize
 
     log.info(
-      { offset, processed: result.processed, skipped: result.skipped, errors: result.errors },
+      { processed: result.processed, skipped: result.skipped, errors: result.errors, totalProcessed, totalSkipped, totalErrors },
       'Batch complete',
     )
   }