feat: cargo registry implementation [CM-1264]

scripts/builders/packages.env

@@ -1,4 +1,4 @@
 DOCKERFILE="./services/docker/Dockerfile.packages"
 CONTEXT="../"
 REPO="sjc.ocir.io/axbydjxa5zuh/packages"
-SERVICES="github-repos-enricher bq-dataset-ingest npm-worker maven-worker osv-worker dockerhub-sync"
+SERVICES="github-repos-enricher bq-dataset-ingest npm-worker maven-worker osv-worker dockerhub-sync cargo-worker"

scripts/services/cargo-worker.yaml

@@ -0,0 +1,65 @@
+version: '3.1'
+
+x-env-args: &env-args
+  DOCKER_BUILDKIT: 1
+  NODE_ENV: docker
+  SERVICE: cargo-worker
+  SHELL: /bin/sh
+  SUPPRESS_NO_CONFIG_WARNING: 'true'
+  CROWD_TEMPORAL_TASKQUEUE: cargo-worker
+
+services:
+  cargo-worker:
+    build:
+      context: ../../
+      dockerfile: ./scripts/services/docker/Dockerfile.packages
+    command: 'pnpm run start:cargo-worker'
+    working_dir: /usr/crowd/app/services/apps/packages_worker
+    env_file:
+      - ../../backend/.env.dist.local
+      - ../../backend/.env.dist.composed
+      - ../../backend/.env.override.local
+      - ../../backend/.env.override.composed
+    environment:
+      <<: *env-args
+    restart: always
+    networks:
+      - crowd-bridge
+
+  cargo-worker-dev:
+    build:
+      context: ../../
+      dockerfile: ./scripts/services/docker/Dockerfile.packages
+    command: 'pnpm run dev:cargo-worker'
+    working_dir: /usr/crowd/app/services/apps/packages_worker
+    env_file:
+      - ../../backend/.env.dist.local
+      - ../../backend/.env.dist.composed
+      - ../../backend/.env.override.local
+      - ../../backend/.env.override.composed
+    environment:
+      <<: *env-args
+    hostname: cargo-worker
+    networks:
+      - crowd-bridge
+    volumes:
+      - ../../services/libs/audit-logs/src:/usr/crowd/app/services/libs/audit-logs/src
+      - ../../services/libs/common/src:/usr/crowd/app/services/libs/common/src
+      - ../../services/libs/common_services/src:/usr/crowd/app/services/libs/common_services/src
+      - ../../services/libs/data-access-layer/src:/usr/crowd/app/services/libs/data-access-layer/src
+      - ../../services/libs/database/src:/usr/crowd/app/services/libs/database/src
+      - ../../services/libs/integrations/src:/usr/crowd/app/services/libs/integrations/src
+      - ../../services/libs/logging/src:/usr/crowd/app/services/libs/logging/src
+      - ../../services/libs/nango/src:/usr/crowd/app/services/libs/nango/src
+      - ../../services/libs/opensearch/src:/usr/crowd/app/services/libs/opensearch/src
+      - ../../services/libs/queue/src:/usr/crowd/app/services/libs/queue/src
+      - ../../services/libs/redis/src:/usr/crowd/app/services/libs/redis/src
+      - ../../services/libs/snowflake/src:/usr/crowd/app/services/libs/snowflake/src
+      - ../../services/libs/telemetry/src:/usr/crowd/app/services/libs/telemetry/src
+      - ../../services/libs/temporal/src:/usr/crowd/app/services/libs/temporal/src
+      - ../../services/libs/types/src:/usr/crowd/app/services/libs/types/src
+      - ../../services/apps/packages_worker/src:/usr/crowd/app/services/apps/packages_worker/src
+
+networks:
+  crowd-bridge:
+    external: true

services/apps/packages_worker/package.json

@@ -30,6 +30,9 @@
     "start:maven-worker": "CROWD_TEMPORAL_TASKQUEUE=packages-worker SERVICE=maven-worker tsx src/bin/maven-worker.ts",
     "dev:maven-worker": "CROWD_TEMPORAL_TASKQUEUE=packages-worker SERVICE=maven-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9236 src/bin/maven-worker.ts",
     "dev:maven-worker:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=packages-worker SERVICE=maven-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9236 src/bin/maven-worker.ts",
+    "start:cargo-worker": "CROWD_TEMPORAL_TASKQUEUE=cargo-worker SERVICE=cargo-worker tsx src/bin/cargo-worker.ts",
+    "dev:cargo-worker": "CROWD_TEMPORAL_TASKQUEUE=cargo-worker SERVICE=cargo-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9240 src/bin/cargo-worker.ts",
+    "dev:cargo-worker:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=cargo-worker SERVICE=cargo-worker LOG_LEVEL=trace nodemon --watch src --watch ../../libs --ext ts --exec tsx --inspect=0.0.0.0:9240 src/bin/cargo-worker.ts",
     "backfill:maven": "SERVICE=maven tsx src/bin/maven-backfill.ts",
     "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:stewardship": "SERVICE=stewardship-backfill tsx src/bin/stewardship-backfill.ts",
@@ -61,6 +64,7 @@
     "semver": "^7.6.0",
     "axios": "^1.16.1",
     "fast-xml-parser": "^5.8.0",
+    "pg-copy-streams": "^7.0.0",
     "tsx": "^4.7.1",
     "typescript": "^5.6.3",
     "undici": "^5.29.0",
@@ -69,6 +73,7 @@
   "devDependencies": {
     "@types/jsonwebtoken": "^9.0.0",
     "@types/node": "^20.8.2",
+    "@types/pg-copy-streams": "^1.2.5",
     "@types/semver": "^7.5.8",
     "@types/unzipper": "^0.10.10",
     "nodemon": "^3.0.1",

services/apps/packages_worker/src/activities.ts

@@ -14,3 +14,13 @@ export * from './deps-dev/activities'
 export { osvSyncEcosystem, osvDeriveCriticalFlag } from './osv/activities'
 export { processMavenCriticalBatch, processMavenNonCriticalBatch } from './maven/activities'
 export { criticalityComputePageRank, rankPackages } from './criticality/activities'
+export {
+  cargoDownloadAndLoad,
+  cargoEnrichPackages,
+  cargoEnrichVersions,
+  cargoEnrichRepos,
+  cargoEnrichMaintainers,
+  cargoEnrichDownloadsDaily,
+  cargoFlushAudit,
+  cargoCleanup,
+} from './cargo/activities'

services/apps/packages_worker/src/bin/cargo-worker.ts

@@ -0,0 +1,8 @@
+import { scheduleCargoSync } from '../cargo/schedule'
+import { svc } from '../service'
+
+setImmediate(async () => {
+  await svc.init()
+  await scheduleCargoSync()
+  await svc.start()
+})

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

@@ -0,0 +1,70 @@
+import { rm } from 'node:fs/promises'
+
+import { getServiceChildLogger } from '@crowd/logging'
+
+import { getCargoConfig } from '../config'
+import { getPackagesDb, getPackagesDbConnection } from '../db'
+
+import { DUMP_DIR, downloadAndExtractDump } from './dump'
+import {
+  enrichDownloadsDaily,
+  enrichMaintainers,
+  enrichPackages,
+  enrichRepos,
+  enrichVersions,
+  flushAudit,
+} from './enrich'
+import { STAGING_SCHEMA, loadDump } from './loadDump'
+import {
+  EnrichDownloadsDailyResult,
+  EnrichMaintainersResult,
+  EnrichPackagesResult,
+  EnrichReposResult,
+  EnrichVersionsResult,
+  LoadResult,
+} from './types'
+
+const log = getServiceChildLogger('cargo-activity')
+
+// Config read at call time — workflow bundle imports this file for type discovery without env.
+export async function cargoDownloadAndLoad(): Promise<LoadResult> {
+  const { dumpUrl } = getCargoConfig()
+  const dumpDir = await downloadAndExtractDump(dumpUrl)
+  const qx = await getPackagesDb()
+  const conn = await getPackagesDbConnection()
+  const result = await loadDump(qx, conn, dumpDir)
+  log.info({ ...result }, 'cargo dump loaded')
+  return result
+}
+
+export async function cargoEnrichPackages(): Promise<EnrichPackagesResult> {
+  return enrichPackages(await getPackagesDb())
+}
+
+export async function cargoEnrichVersions(): Promise<EnrichVersionsResult> {
+  return enrichVersions(await getPackagesDb())
+}
+
+export async function cargoEnrichRepos(): Promise<EnrichReposResult> {
+  return enrichRepos(await getPackagesDb())
+}
+
+export async function cargoEnrichMaintainers(): Promise<EnrichMaintainersResult> {
+  return enrichMaintainers(await getPackagesDb())
+}
+
+export async function cargoEnrichDownloadsDaily(): Promise<EnrichDownloadsDailyResult> {
+  return enrichDownloadsDaily(await getPackagesDb())
+}
+
+export async function cargoFlushAudit(): Promise<number> {
+  return flushAudit(await getPackagesDb())
+}
+
+// Best-effort: a crashed run self-heals on next run (schema rebuilt, DUMP_DIR cleared).
+export async function cargoCleanup(): Promise<void> {
+  const qx = await getPackagesDb()
+  await qx.result(`DROP SCHEMA IF EXISTS ${STAGING_SCHEMA} CASCADE`)
+  await rm(DUMP_DIR, { recursive: true, force: true })
+  log.info('cargo cleanup complete')
+}