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

Signed-off-by: Mouad BANI <mouad-mb@outlook.com>

Author: mbani01

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

@@ -1,3 +1,5 @@
+import { createHash } from 'crypto'
+
 import type { Request, Response } from 'express'
 
 import { NotFoundError } from '@crowd/common'
@@ -8,6 +10,7 @@ import {
   getStewardshipSummary,
   securityContactConfidenceBand,
 } from '@crowd/data-access-layer'
+import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal'
 
 import { getPackagesQx } from '@/db/packagesDb'
 import { ok } from '@/utils/api'
@@ -32,16 +35,38 @@ 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 {
+      await req.temporal.workflow.execute('ingestSecurityContactsForPurlWorkflow', {
+        taskQueue: 'packages-worker',
+        workflowId: ondemandWorkflowId(purl),
+        workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
+        workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING,
+        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,

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,91 @@
+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'
+
+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.
+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) AND r.host = 'github'
+    `,
+    { 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) — see security-contacts.md.
+ */
+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 github repo found — skipping')
+    return { found: false }
+  }
+
+  const target = toTarget(row)
+  const baseDeps = buildBaseDeps(config)
+  await processRepo(target, baseDeps, qx)
+
+  return { found: true, repoId: target.repoId }
+}

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

@@ -109,7 +109,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 +170,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'