feat: add refresh security contats api (CM-1348)

backend/src/api/public/v1/akrites-external/index.ts

@@ -12,26 +12,37 @@ import { getAkritesExternalContactDetailBatch } from '../packages/getAkritesExte
 import { getAkritesExternalPackageDetail } from '../packages/getAkritesExternalPackageDetail'
 import { getAkritesExternalPackageDetailBatch } from '../packages/getAkritesExternalPackageDetailBatch'
 import { getBlastRadiusJob } from '../packages/getBlastRadiusJob'
+import { ingestAkritesExternalContactDetail } from '../packages/ingestAkritesExternalContactDetail'
 import { submitBlastRadiusJob } from '../packages/submitBlastRadiusJob'
 
 const rateLimiter = createRateLimiter({ max: 60, windowMs: 60 * 1000 })
 
-// Blast-radius jobs kick off a Temporal workflow per request, so they get their own,
-// much stricter limiter — configurable via env so it can be tuned without a redeploy.
-// Defaults to 5 requests/hour.
-const blastRadiusRateLimitMax = Number(process.env.AKRITES_BLAST_RADIUS_RATE_LIMIT_MAX)
-const blastRadiusRateLimitWindowMs = Number(process.env.AKRITES_BLAST_RADIUS_RATE_LIMIT_WINDOW_MS)
+// Shared by every endpoint below that kicks off a Temporal workflow per request — those
+// get their own, much stricter limiter than plain reads, configurable via env so it can
+// be tuned without a redeploy.
+function envTunableRateLimiter(envPrefix: string, defaultMax: number, defaultWindowMs: number) {
+  const max = Number(process.env[`${envPrefix}_MAX`])
+  const windowMs = Number(process.env[`${envPrefix}_WINDOW_MS`])
+  return createRateLimiter({
+    max: Number.isSafeInteger(max) && max > 0 ? max : defaultMax,
+    windowMs: Number.isSafeInteger(windowMs) && windowMs > 0 ? windowMs : defaultWindowMs,
+  })
+}
+
+// Blast-radius jobs default to 5 requests/hour.
+const blastRadiusRateLimiter = envTunableRateLimiter(
+  'AKRITES_BLAST_RADIUS_RATE_LIMIT',
+  5,
+  60 * 60 * 1000,
+)
 
-const blastRadiusRateLimiter = createRateLimiter({
-  max:
-    Number.isSafeInteger(blastRadiusRateLimitMax) && blastRadiusRateLimitMax > 0
-      ? blastRadiusRateLimitMax
-      : 5,
-  windowMs:
-    Number.isSafeInteger(blastRadiusRateLimitWindowMs) && blastRadiusRateLimitWindowMs > 0
-      ? blastRadiusRateLimitWindowMs
-      : 60 * 60 * 1000,
-})
+// /contacts/ingest blocks ~10-20s per request (vs. the read-only /contacts/detail
+// endpoints), so it gets its own limiter. Defaults to 20 requests/hour.
+const contactIngestRateLimiter = envTunableRateLimiter(
+  'AKRITES_CONTACT_INGEST_RATE_LIMIT',
+  20,
+  60 * 60 * 1000,
+)
 
 export function akritesExternalRouter(): Router {
   const router = Router()
@@ -63,10 +74,19 @@ export function akritesExternalRouter(): Router {
   // closest issued one — READ_MAINTAINER_ROLES (maintainer data) — NOT READ_PACKAGES.
   // TODO: swap for cdp:maintainers:read once issued.
   const contactsSubRouter = Router()
+  // rateLimiter first so requests failing the scope check still count against the quota
+  // (throttles repeated invalid-auth attempts, not just successful ones).
   contactsSubRouter.use(rateLimiter)
   contactsSubRouter.use(requireScopes([SCOPES.READ_MAINTAINER_ROLES]))
   contactsSubRouter.get('/detail', safeWrap(getAkritesExternalContactDetail))
   contactsSubRouter.post(/^\/detail:batch\/?$/, safeWrap(getAkritesExternalContactDetailBatch))
+  // Sync, single-purl on-demand ingest — starts a Temporal workflow and blocks ~10-20s,
+  // so it stacks the dedicated contactIngestRateLimiter on top of the shared rateLimiter above.
+  contactsSubRouter.post(
+    '/ingest',
+    contactIngestRateLimiter,
+    safeWrap(ingestAkritesExternalContactDetail),
+  )
   router.use('/contacts', contactsSubRouter)
 
   // TODO: the contract gates blast-radius behind a dedicated read:advisories scope

backend/src/api/public/v1/akrites-external/openapi.yaml

@@ -1246,3 +1246,81 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/Error'
+
+  /akrites-external/contacts/ingest:
+    post:
+      operationId: ingestContactDetail
+      summary: On-demand security-contacts ingest for a single PURL
+      description: >
+        Synchronous, single-purl only — no batch variant, since fanning this out over
+        many purls would multiply concurrent Temporal workflow starts. Blocks ~10-20s
+        while a single-repo Temporal workflow (ingestSecurityContactsForPurlWorkflow)
+        ingests the linked repo's security contacts, then returns the same
+        ContactDetail payload as GET /contacts/detail. Concurrent requests for the
+        same purl attach to the same running workflow instead of starting duplicates.
+
+
+        Note: the underlying workflow only ingests contacts it has never seen before —
+        it does not refresh contacts already ingested for a purl.
+
+
+        Rate limited independently of the other /contacts endpoints — default
+        20 requests/hour, tunable via AKRITES_CONTACT_INGEST_RATE_LIMIT_MAX /
+        AKRITES_CONTACT_INGEST_RATE_LIMIT_WINDOW_MS.
+
+
+        Not yet implemented: this is a synchronous, blocking call. Future work should
+        make this async — an accept-and-poll job, similar to POST /blast-radius/jobs —
+        so callers aren't held open for 10-20s per request.
+      tags: [Contacts]
+      security:
+        - M2MBearer:
+            - read:maintainer-roles
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              required: [purl]
+              properties:
+                purl:
+                  type: string
+                  example: pkg:npm/%40angular/core
+      responses:
+        '200':
+          description: Security contact detail, freshly ingested.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ContactDetail'
+        '400':
+          description: Malformed purl.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
+        '401':
+          description: Missing or invalid bearer token.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
+        '403':
+          description: Token missing read:maintainer-roles scope.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
+        '404':
+          description: Purl has no linked repo to ingest contacts from.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
+        '429':
+          description: Too many requests — rate-limited independently of the other endpoints.
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'

backend/src/api/public/v1/packages/ingestAkritesExternalContactDetail.test.ts

@@ -0,0 +1,119 @@
+import type { Request, Response } from 'express'
+import { describe, expect, it, vi } from 'vitest'
+
+import type { AkritesExternalContactDetailRow } from '@crowd/data-access-layer'
+import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal'
+
+import { ingestAkritesExternalContactDetail } from './ingestAkritesExternalContactDetail'
+
+const { execute, getContactDetailsByPurls } = vi.hoisted(() => ({
+  execute: vi.fn(),
+  getContactDetailsByPurls: vi.fn(),
+}))
+
+vi.mock('@/db/packagesTemporal', () => ({
+  getPackagesTemporalClient: vi.fn().mockResolvedValue({ workflow: { execute } }),
+}))
+
+vi.mock('@/db/packagesDb', () => ({
+  getPackagesQx: vi.fn().mockResolvedValue({}),
+}))
+
+vi.mock('@crowd/data-access-layer', () => ({
+  getContactDetailsByPurls,
+}))
+
+function baseRow(
+  overrides: Partial<AkritesExternalContactDetailRow> = {},
+): AkritesExternalContactDetailRow {
+  return {
+    purl: 'pkg:npm/lodash',
+    name: 'lodash',
+    ecosystem: 'npm',
+    securityPolicyUrl: null,
+    vulnerabilityReportingUrl: null,
+    bugBountyUrl: null,
+    pvrEnabled: null,
+    declaredRepositoryUrl: null,
+    resolvedRepositoryUrl: null,
+    repoMappingConfidence: null,
+    securityContacts: [],
+    ...overrides,
+  }
+}
+
+function mockReqRes(body: unknown) {
+  execute.mockClear()
+  getContactDetailsByPurls.mockClear()
+
+  const req = { body } as unknown as Request
+
+  const json = vi.fn()
+  const status = vi.fn().mockReturnValue({ json })
+  const res = { status, json } as unknown as Response
+
+  return { req, res, status, json }
+}
+
+describe('ingestAkritesExternalContactDetail', () => {
+  it('executes ingestSecurityContactsForPurlWorkflow and returns the re-read contact detail', async () => {
+    execute.mockResolvedValue({ found: true, repoId: 'repo-1' })
+    getContactDetailsByPurls.mockResolvedValue([baseRow()])
+
+    const { req, res, json } = mockReqRes({ purl: 'pkg:npm/lodash' })
+
+    await ingestAkritesExternalContactDetail(req, res)
+
+    expect(execute).toHaveBeenCalledTimes(1)
+    const [workflowType, options] = execute.mock.calls[0]
+    expect(workflowType).toBe('ingestSecurityContactsForPurlWorkflow')
+    expect(options.taskQueue).toBe('security-contacts-worker')
+    expect(options.workflowId).toMatch(/^security-contacts-ondemand:[0-9a-f]{64}$/)
+    expect(options.workflowIdConflictPolicy).toBe(WorkflowIdConflictPolicy.USE_EXISTING)
+    expect(options.workflowIdReusePolicy).toBe(WorkflowIdReusePolicy.ALLOW_DUPLICATE)
+    expect(options.args).toEqual(['pkg:npm/lodash'])
+
+    expect(getContactDetailsByPurls).toHaveBeenCalledWith(expect.anything(), ['pkg:npm/lodash'])
+    expect(json).toHaveBeenCalledWith(expect.objectContaining({ purl: 'pkg:npm/lodash' }))
+  })
+
+  it('derives the same deterministic workflowId for the same purl', async () => {
+    execute.mockResolvedValue({ found: true })
+    getContactDetailsByPurls.mockResolvedValue([baseRow()])
+
+    const { req: req1, res: res1 } = mockReqRes({ purl: 'pkg:npm/lodash' })
+    await ingestAkritesExternalContactDetail(req1, res1)
+    const id1 = execute.mock.calls[0][1].workflowId
+
+    const { req: req2, res: res2 } = mockReqRes({ purl: 'pkg:npm/lodash' })
+    await ingestAkritesExternalContactDetail(req2, res2)
+    const id2 = execute.mock.calls[0][1].workflowId
+
+    expect(id1).toBe(id2)
+  })
+
+  it('throws NotFoundError when the workflow reports no linked repo', async () => {
+    execute.mockResolvedValue({ found: false })
+
+    const { req, res } = mockReqRes({ purl: 'pkg:npm/left-pad' })
+
+    await expect(ingestAkritesExternalContactDetail(req, res)).rejects.toThrow()
+    expect(getContactDetailsByPurls).not.toHaveBeenCalled()
+  })
+
+  it('throws NotFoundError when the re-read finds no row', async () => {
+    execute.mockResolvedValue({ found: true })
+    getContactDetailsByPurls.mockResolvedValue([])
+
+    const { req, res } = mockReqRes({ purl: 'pkg:npm/lodash' })
+
+    await expect(ingestAkritesExternalContactDetail(req, res)).rejects.toThrow()
+  })
+
+  it('rejects a request missing purl without executing a workflow', async () => {
+    const { req, res } = mockReqRes({})
+
+    await expect(ingestAkritesExternalContactDetail(req, res)).rejects.toThrow()
+    expect(execute).not.toHaveBeenCalled()
+  })
+})

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

@@ -0,0 +1,65 @@
+import { createHash } from 'crypto'
+import type { Request, Response } from 'express'
+
+import { NotFoundError } from '@crowd/common'
+import { getContactDetailsByPurls } from '@crowd/data-access-layer'
+import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@crowd/temporal'
+import { TemporalWorkflowId } from '@crowd/types'
+
+import { getPackagesQx } from '@/db/packagesDb'
+import { getPackagesTemporalClient } from '@/db/packagesTemporal'
+import { ok } from '@/utils/api'
+import { validateOrThrow } from '@/utils/validation'
+
+import { toAkritesExternalContactDetail } from './akritesExternalContactDetail'
+import { purlBodySchema } from './purl'
+
+interface IngestSecurityContactsForPurlResult {
+  found: boolean
+  repoId?: string
+}
+
+// Deterministic, purl-derived workflowId: concurrent callers hitting the same purl attach
+// to the same running workflow (USE_EXISTING) instead of each starting their own ingest —
+// same pattern as integrationService.ts's github-nango-sync workflow start.
+function ingestWorkflowId(purl: string): string {
+  return `${TemporalWorkflowId.SECURITY_CONTACTS_ONDEMAND}:${createHash('sha256').update(purl).digest('hex')}`
+}
+
+// Sync, single-purl on-demand ingest. Blocks ~10-20s (the worker's single-repo bound —
+// see security-contacts/workflows.ts's singleActs timeout) — no batch variant, since
+// fanning this out over many purls would multiply concurrent Temporal workflow starts.
+//
+// Note: the workflow only ingests contacts it has never seen before — it does not
+// refresh/update contacts already ingested for a purl. That's why this endpoint is
+// named "ingest", not "refresh".
+export async function ingestAkritesExternalContactDetail(
+  req: Request,
+  res: Response,
+): Promise<void> {
+  const { purl } = validateOrThrow(purlBodySchema, req.body)
+
+  const packagesTemporal = await getPackagesTemporalClient()
+  const result = await packagesTemporal.workflow.execute<
+    (purl: string) => Promise<IngestSecurityContactsForPurlResult>
+  >('ingestSecurityContactsForPurlWorkflow', {
+    taskQueue: 'security-contacts-worker',
+    workflowId: ingestWorkflowId(purl),
+    workflowIdConflictPolicy: WorkflowIdConflictPolicy.USE_EXISTING,
+    workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE,
+    args: [purl],
+  })
+
+  if (!result.found) {
+    throw new NotFoundError('Purl has no linked repository to ingest security contacts from')
+  }
+
+  const qx = await getPackagesQx()
+  const [row] = await getContactDetailsByPurls(qx, [purl])
+
+  if (!row) {
+    throw new NotFoundError('Contact detail not found after ingest')
+  }
+
+  ok(res, toAkritesExternalContactDetail(row))
+}

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

@@ -39,6 +39,10 @@ export const purlFieldSchema = z
 
 export const purlQuerySchema = z.object({ purl: purlFieldSchema })
 
+// Single-purl body (as opposed to purlsBodySchema's array) — normalizes like
+// purlFieldSchema/purlQuerySchema, for endpoints that take exactly one purl in the body.
+export const purlBodySchema = z.object({ purl: purlFieldSchema })
+
 // Loose schema for search filters: normalizes without requiring the pkg: prefix,
 // so partial inputs (e.g. "@babel/core" or "lodash") are accepted.
 export const purlFilterSchema = z.string().trim().transform(normalizePurl).optional()

services/libs/types/src/enums/temporal.ts

@@ -12,4 +12,5 @@ export enum TemporalWorkflowId {
   DELETE_ORPHAN_MEMBER = 'delete-orphan-member',
 
   BLAST_RADIUS_ANALYSIS = 'blast-radius-analysis',
+  SECURITY_CONTACTS_ONDEMAND = 'security-contacts-ondemand',
 }