feat: add on-demand security-contacts ingest for a single purl (CM-1313)

backend/.env.dist.local

@@ -151,6 +151,7 @@ CROWD_TEMPORAL_ENCRYPTION_KEY_ID=local
 CROWD_TEMPORAL_ENCRYPTION_KEY=FweBMRnGCLshER8FlSvNusQA6G3MRUKt
 
 # Temporal — packages namespace
+CROWD_PACKAGES_TEMPORAL_SERVER_URL=localhost:7233
 CROWD_PACKAGES_TEMPORAL_NAMESPACE=default
 
 # Seach sync api

backend/config/custom-environment-variables.json

@@ -180,6 +180,12 @@
     "certificate": "CROWD_TEMPORAL_CERTIFICATE",
     "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY"
   },
+  "packagesTemporal": {
+    "serverUrl": "CROWD_PACKAGES_TEMPORAL_SERVER_URL",
+    "namespace": "CROWD_PACKAGES_TEMPORAL_NAMESPACE",
+    "certificate": "CROWD_TEMPORAL_CERTIFICATE",
+    "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY"
+  },
   "searchSyncApi": {
     "baseUrl": "CROWD_SEARCH_SYNC_API_URL"
   },

backend/config/default.json

@@ -43,6 +43,7 @@
     "isEnabled": "false"
   },
   "temporal": {},
+  "packagesTemporal": {},
   "searchSyncApi": {},
   "encryption": {},
   "openStatusApi": {},

backend/src/api/public/v1/packages/getPackage.ts

@@ -1,3 +1,4 @@
+import { createHash } from 'crypto'
 import type { Request, Response } from 'express'
 
 import { NotFoundError } from '@crowd/common'
@@ -8,8 +9,10 @@ import {
   getStewardshipSummary,
   securityContactConfidenceBand,
 } from '@crowd/data-access-layer'
+import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal'
 
 import { getPackagesQx } from '@/db/packagesDb'
+import { getPackagesTemporalClient } from '@/db/packagesTemporal'
 import { ok } from '@/utils/api'
 import { validateOrThrow } from '@/utils/validation'
 
@@ -32,16 +35,43 @@ function repoMappingLabel(confidence: number | null): 'High' | 'Medium' | 'Low'
   return 'Low'
 }
 
+// Deterministic per-purl id: concurrent requests for the same purl attach to the
+// same workflow run (WorkflowIdConflictPolicy.USE_EXISTING below) instead of each
+// kicking off its own ingest.
+function ondemandWorkflowId(purl: string): string {
+  return `security-contacts-ondemand/${createHash('sha1').update(purl).digest('hex')}`
+}
+
 export async function getPackage(req: Request, res: Response): Promise<void> {
   const { purl } = validateOrThrow(purlQuerySchema, req.query)
 
   const qx = await getPackagesQx()
-  const pkg = await getPackageDetailByPurl(qx, purl)
+  let pkg = await getPackageDetailByPurl(qx, purl)
 
   if (!pkg) {
     throw new NotFoundError()
   }
 
+  if (pkg.contactsLastRefreshed == null) {
+    try {
+      const packagesTemporal = await getPackagesTemporalClient()
+      await packagesTemporal.workflow.execute('ingestSecurityContactsForPurlWorkflow', {
+        taskQueue: 'packages-worker',
+        workflowId: ondemandWorkflowId(purl),
+        workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
+        workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING,
+        // Bounds the request's wait if packages-worker is down or its task queue is
+        // backed up — the activity's startToCloseTimeout only starts counting once a
+        // worker picks the task up, so without this the request could hang indefinitely.
+        workflowExecutionTimeout: '60 seconds',
+        args: [purl],
+      })
+      pkg = (await getPackageDetailByPurl(qx, purl)) ?? pkg
+    } catch (err) {
+      req.log.warn(err, 'On-demand security contacts ingest failed — serving cached detail')
+    }
+  }
+
   const [{ rows: advisories }, stewardshipSummary] = await Promise.all([
     getAdvisoriesByPackageId(qx, pkg.id),
     pkg.stewardshipId ? getStewardshipSummary(qx, Number(pkg.stewardshipId)) : null,

backend/src/conf/index.ts

@@ -86,6 +86,14 @@ export const PACKAGES_DB_CONFIG: IDatabaseConfig | undefined = config.has('packa
   ? config.get<IDatabaseConfig>('packagesDb')
   : undefined
 
+// packages_worker (npm/maven/pypi/osv/security-contacts/...) runs in its own Temporal
+// namespace, separate from the API's default namespace — see CROWD_PACKAGES_TEMPORAL_NAMESPACE.
+export const PACKAGES_TEMPORAL_CONFIG: ITemporalConfig | undefined = config.has(
+  'packagesTemporal.namespace',
+)
+  ? config.get<ITemporalConfig>('packagesTemporal')
+  : undefined
+
 export const SEGMENT_CONFIG: SegmentConfiguration = config.get<SegmentConfiguration>('segment')
 
 export const COMPREHEND_CONFIG: ComprehendConfiguration =

backend/src/db/packagesTemporal.ts

@@ -0,0 +1,45 @@
+import { IS_DEV_ENV, SERVICE } from '@crowd/common'
+import { Client, Connection, getDataConverter } from '@crowd/temporal'
+
+import { PACKAGES_TEMPORAL_CONFIG } from '@/conf'
+
+let _init: Promise<Client> | undefined
+
+// Separate connection from the API's default req.temporal client — packages_worker
+// (npm/maven/pypi/osv/security-contacts/...) polls task queues in its own Temporal
+// namespace (CROWD_PACKAGES_TEMPORAL_NAMESPACE), not the API's default namespace.
+export function getPackagesTemporalClient(): Promise<Client> {
+  if (!_init) {
+    if (!PACKAGES_TEMPORAL_CONFIG?.serverUrl) {
+      throw new Error('Packages Temporal is not configured — set CROWD_PACKAGES_TEMPORAL_NAMESPACE')
+    }
+
+    const cfg = PACKAGES_TEMPORAL_CONFIG
+    _init = Connection.connect({
+      address: cfg.serverUrl,
+      tls:
+        cfg.certificate && cfg.privateKey
+          ? {
+              clientCertPair: {
+                crt: Buffer.from(cfg.certificate, 'base64'),
+                key: Buffer.from(cfg.privateKey, 'base64'),
+              },
+            }
+          : undefined,
+    })
+      .then(
+        async (connection) =>
+          new Client({
+            connection,
+            namespace: cfg.namespace,
+            identity: SERVICE,
+            dataConverter: IS_DEV_ENV ? undefined : await getDataConverter(),
+          }),
+      )
+      .catch((err) => {
+        _init = undefined
+        throw err
+      })
+  }
+  return _init
+}

scripts/services/security-contacts-worker.yaml

@@ -7,6 +7,7 @@ x-env-args: &env-args
   SHELL: /bin/sh
   SUPPRESS_NO_CONFIG_WARNING: 'true'
   CROWD_TEMPORAL_TASKQUEUE: packages-worker
+  CROWD_TEMPORAL_NAMESPACE: ${CROWD_PACKAGES_TEMPORAL_NAMESPACE}
 
 services:
   security-contacts-worker:

services/apps/packages_worker/src/activities.ts

@@ -32,4 +32,7 @@ export {
 } from './pypi/activities'
 export { getCriticalPypiCount } from './pypi/downloads/getCriticalPypiCount'
 export { processNuGetBatch } from './nuget/activities'
-export { processSecurityContactsBatch } from './security-contacts/activities'
+export {
+  processSecurityContactsBatch,
+  ingestSecurityContactsForPurlActivity,
+} from './security-contacts/activities'

services/apps/packages_worker/src/security-contacts/activities.ts

@@ -3,6 +3,7 @@ import { getServiceChildLogger } from '@crowd/logging'
 import { getSecurityContactsConfig } from '../config'
 import { getPackagesDb } from '../db'
 
+import { IngestSingleResult, ingestSecurityContactsForPurl } from './ingestSingle'
 import { BatchResult, processBatch } from './processBatch'
 
 const log = getServiceChildLogger('security-contacts-activity')
@@ -15,3 +16,14 @@ export async function processSecurityContactsBatch(): Promise<BatchResult> {
   log.info({ ...result }, 'Security contacts batch activity complete')
   return result
 }
+
+export async function ingestSecurityContactsForPurlActivity(
+  purl: string,
+): Promise<IngestSingleResult> {
+  const config = getSecurityContactsConfig()
+  const qx = await getPackagesDb()
+
+  const result = await ingestSecurityContactsForPurl(qx, config, purl)
+  log.info({ purl, ...result }, 'On-demand security contacts ingest activity complete')
+  return result
+}

services/apps/packages_worker/src/security-contacts/ingestSingle.ts

@@ -0,0 +1,102 @@
+import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+import { getServiceChildLogger } from '@crowd/logging'
+
+import { getSecurityContactsConfig } from '../config'
+
+import { buildBaseDeps, processRepo } from './processBatch'
+import { RepoPackage, RepoTarget } from './types'
+import { markRepoAttempted } from './writeContacts'
+
+const log = getServiceChildLogger('security-contacts-ondemand')
+
+type Config = ReturnType<typeof getSecurityContactsConfig>
+
+export interface IngestSingleResult {
+  found: boolean
+  repoId?: string
+}
+
+interface SingleRepoRow {
+  repoId: string
+  url: string
+  homepage: string | null
+  archived: boolean | null
+  packages: RepoPackage[] | null
+}
+
+// Mirrors the best-repo LATERAL selection in getPackageDetailByPurl
+// (services/libs/data-access-layer/src/osspckgs/api.ts) so the repo we ingest is the
+// one the read side surfaces. No `is_critical` filter — non-critical purls are exactly
+// what this on-demand path exists for. `packages` aggregates every package linked to
+// the repo (not just the requested purl) so extractors see every ecosystem, matching
+// the shape the batch sweep builds.
+//
+// No host filter: processRepo already degrades gracefully for non-github repos (the
+// github-specific extractors no-op, security.txt/registry-manifest extractors still run) —
+// filtering here would leave non-github repos permanently NULL, re-triggering this
+// on-demand path on every single request.
+async function findBestRepoForPurl(qx: QueryExecutor, purl: string): Promise<SingleRepoRow | null> {
+  return qx.selectOneOrNone(
+    `
+    SELECT r.id::text AS "repoId",
+           r.url,
+           r.homepage,
+           r.archived,
+           (
+             SELECT json_agg(json_build_object('purl', p2.purl, 'ecosystem', p2.ecosystem))
+             FROM package_repos pr2
+             JOIN packages p2 ON p2.id = pr2.package_id
+             WHERE pr2.repo_id = r.id
+           ) AS packages
+    FROM packages p
+    JOIN LATERAL (
+      SELECT pr.repo_id
+      FROM package_repos pr
+      WHERE pr.package_id = p.id
+      ORDER BY pr.confidence DESC, (pr.source = 'declared') DESC
+      LIMIT 1
+    ) best ON true
+    JOIN repos r ON r.id = best.repo_id
+    WHERE p.purl = $(purl)
+    `,
+    { purl },
+  )
+}
+
+function toTarget(row: SingleRepoRow): RepoTarget {
+  return {
+    repoId: row.repoId,
+    url: row.url,
+    homepage: row.homepage,
+    archived: row.archived,
+    packages: row.packages ?? [],
+  }
+}
+
+/**
+ * On-demand ingest for a single purl, bypassing the daily critical-only sweep.
+ * Used when the akrites API hits a repo that has never been evaluated
+ * (repos.contacts_last_refreshed IS NULL).
+ */
+export async function ingestSecurityContactsForPurl(
+  qx: QueryExecutor,
+  config: Config,
+  purl: string,
+): Promise<IngestSingleResult> {
+  const row = await findBestRepoForPurl(qx, purl)
+  if (!row) {
+    log.info({ purl }, 'On-demand security contacts: no linked repo found — skipping')
+    return { found: false }
+  }
+
+  const target = toTarget(row)
+  const baseDeps = buildBaseDeps(config)
+  try {
+    await processRepo(target, baseDeps, qx)
+  } catch (err) {
+    log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed')
+    await markRepoAttempted(qx, target.repoId).catch(() => undefined)
+  }
+
+  return { found: true, repoId: target.repoId }
+}

services/apps/packages_worker/src/security-contacts/processBatch.ts

@@ -72,8 +72,7 @@ async function fetchBatch(qx: QueryExecutor): Promise<SweepRow[]> {
     FROM repos r
     JOIN package_repos pr ON pr.repo_id = r.id
     JOIN packages p ON p.id = pr.package_id AND p.is_critical
-    WHERE r.host = 'github'
-      AND (
+    WHERE (
         -- never evaluated → always eligible
         r.contacts_last_refreshed IS NULL
         -- evaluated but no contacts found yet → retry on the daily cadence
@@ -109,7 +108,15 @@ function toTarget(row: SweepRow): RepoTarget {
   }
 }
 
-async function processRepo(
+export function buildBaseDeps(config: Config): Omit<ExtractorDeps, 'repoTree'> {
+  return {
+    fetchTimeoutMs: FETCH_TIMEOUT_MS,
+    userAgent: config.userAgent,
+    githubGet: (path, opts) => githubApiGet(path, FETCH_TIMEOUT_MS, opts),
+  }
+}
+
+export async function processRepo(
   target: RepoTarget,
   baseDeps: Omit<ExtractorDeps, 'repoTree'>,
   qx: QueryExecutor,
@@ -162,11 +169,7 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise<B
   const batch = await fetchBatch(qx)
   if (batch.length === 0) return { processed: 0 }
 
-  const deps: Omit<ExtractorDeps, 'repoTree'> = {
-    fetchTimeoutMs: FETCH_TIMEOUT_MS,
-    userAgent: config.userAgent,
-    githubGet: (path, opts) => githubApiGet(path, FETCH_TIMEOUT_MS, opts),
-  }
+  const deps = buildBaseDeps(config)
 
   const targets = batch.map(toTarget)
   // Fixed-cadence heartbeat: a slow repo can outlast the 2-minute heartbeatTimeout even while

services/apps/packages_worker/src/security-contacts/workflows.ts

@@ -1,6 +1,7 @@
 import { continueAsNew, log, proxyActivities } from '@temporalio/workflow'
 
 import type * as activities from './activities'
+import type { IngestSingleResult } from './ingestSingle'
 
 const acts = proxyActivities<typeof activities>({
   startToCloseTimeout: '30 minutes',
@@ -14,6 +15,17 @@ const acts = proxyActivities<typeof activities>({
   },
 })
 
+// Single-repo, synchronous, on-demand path: short bound (no continueAsNew/heartbeat —
+// one repo's extractors, not a whole sweep), so a caller awaiting the result (e.g. the
+// akrites API on a cache miss) doesn't hang.
+const singleActs = proxyActivities<typeof activities>({
+  startToCloseTimeout: '45 seconds',
+  retry: {
+    initialInterval: '5 seconds',
+    maximumAttempts: 2,
+  },
+})
+
 export async function ingestSecurityContacts(): Promise<void> {
   const result = await acts.processSecurityContactsBatch()
   if (result.processed === 0) {
@@ -22,3 +34,9 @@ export async function ingestSecurityContacts(): Promise<void> {
   }
   await continueAsNew<typeof ingestSecurityContacts>()
 }
+
+export async function ingestSecurityContactsForPurlWorkflow(
+  purl: string,
+): Promise<IngestSingleResult> {
+  return singleActs.ingestSecurityContactsForPurlActivity(purl)
+}

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

@@ -25,4 +25,7 @@ export {
   ingestPypiDownloadsDaily,
 } from '../pypi/downloads/ingestPypiDownloads'
 export { ingestNuGetPackages } from '../nuget/workflows'
-export { ingestSecurityContacts } from '../security-contacts/workflows'
+export {
+  ingestSecurityContacts,
+  ingestSecurityContactsForPurlWorkflow,
+} from '../security-contacts/workflows'