feat: first version of blast-radius workflow

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

Author: ulemons

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

@@ -0,0 +1,127 @@
+import type {
+  AnalysisDetailRow,
+  VerdictResultRow,
+} from '@crowd/data-access-layer/src/packages/blastRadius'
+
+import type { BlastRadiusJobEcosystem, BlastRadiusJobStatus } from './blastRadius'
+
+export type BlastRadiusResultConfidence = 'high' | 'medium' | 'low'
+
+export interface BlastRadiusResultItem {
+  dependent: string
+  affected: boolean
+  confidence: BlastRadiusResultConfidence
+  evidence: string | null
+  downloadsLast30Days: string | null
+}
+
+export interface BlastRadiusAnalysisSummary {
+  totalDependentsInRange: number
+  dependentsExcludedUpfront: number
+  dependentsAnalyzed: number
+  dependentsAffected: number
+  affectedPercentage: number | null
+  affectedDependents: string[]
+}
+
+export interface BlastRadiusAnalysis {
+  analysisId: string
+  status: BlastRadiusJobStatus
+  advisoryId: string
+  package: string | null
+  ecosystem: BlastRadiusJobEcosystem
+  submittedAt: string | null
+  completedAt: string | null
+  errorMessage: string | null
+  summary: BlastRadiusAnalysisSummary | null
+  results: BlastRadiusResultItem[] | null
+}
+
+// Crosswalk from the DB's NUMERIC(3,2) 0-1 confidence score to the contract's
+// enum — matches the PoC methodology's confidence bands (report.py).
+function toResultConfidence(confidence: number): BlastRadiusResultConfidence {
+  if (confidence >= 0.8) return 'high'
+  if (confidence >= 0.4) return 'medium'
+  return 'low'
+}
+
+// npm purls use a literal (unencoded) `@` for the scope, matching the contract's
+// example response bodies — these are JSON fields, not URL query params, so the
+// %40-encoding normalizePurl applies elsewhere in this file group does not apply here.
+function toPurl(name: string): string {
+  return `pkg:npm/${name}`
+}
+
+function flattenEvidence(evidence: Record<string, unknown>[] | null): string | null {
+  if (!evidence || evidence.length === 0) return null
+  return evidence
+    .map((e) => {
+      const file = e.file ? String(e.file) : null
+      const line = e.line !== undefined && e.line !== null ? String(e.line) : null
+      const snippet = e.snippet ? String(e.snippet) : null
+      const location = [file, line].filter(Boolean).join(':')
+      return [location, snippet].filter(Boolean).join(' — ')
+    })
+    .filter(Boolean)
+    .join('\n')
+}
+
+function toResultItem(row: VerdictResultRow): BlastRadiusResultItem {
+  return {
+    dependent: toPurl(row.name),
+    affected: row.reachable_verdict === 'affected',
+    confidence: toResultConfidence(row.confidence),
+    evidence: flattenEvidence(row.evidence),
+    downloadsLast30Days: row.downloads !== null ? String(row.downloads) : null,
+  }
+}
+
+// Population-level summary, following the PoC's report.py ground truth:
+// totalDependentsInRange = candidates_considered (post phase-1-filter population),
+// dependentsAnalyzed = number of verdicts produced, dependentsExcludedUpfront =
+// the difference (range-excluded before reachability ran), dependentsAffected =
+// count with an 'affected' verdict, affectedPercentage rounded to 1 decimal
+// (null when nothing was analyzed, to avoid a misleading 0%).
+function toSummary(
+  candidatesConsidered: number | null,
+  results: BlastRadiusResultItem[],
+): BlastRadiusAnalysisSummary {
+  const totalDependentsInRange = candidatesConsidered ?? results.length
+  const dependentsAnalyzed = results.length
+  const affected = results.filter((r) => r.affected)
+
+  return {
+    totalDependentsInRange,
+    dependentsExcludedUpfront: Math.max(totalDependentsInRange - dependentsAnalyzed, 0),
+    dependentsAnalyzed,
+    dependentsAffected: affected.length,
+    affectedPercentage:
+      dependentsAnalyzed > 0
+        ? Math.round((affected.length / dependentsAnalyzed) * 1000) / 10
+        : null,
+    affectedDependents: affected.map((r) => r.dependent),
+  }
+}
+
+export function toBlastRadiusAnalysis(
+  analysis: AnalysisDetailRow,
+  verdictRows: VerdictResultRow[],
+): BlastRadiusAnalysis {
+  const status = analysis.status as BlastRadiusJobStatus
+  const done = status === 'done'
+
+  const results = done ? verdictRows.map(toResultItem) : null
+
+  return {
+    analysisId: analysis.id,
+    status,
+    advisoryId: analysis.advisory_osv_id,
+    package: analysis.package_name,
+    ecosystem: analysis.ecosystem as BlastRadiusJobEcosystem,
+    submittedAt: analysis.started_at,
+    completedAt: analysis.completed_at,
+    errorMessage: analysis.error,
+    summary: done && results ? toSummary(analysis.candidates_considered, results) : null,
+    results,
+  }
+}

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

@@ -0,0 +1,142 @@
+import type { Request, Response } from 'express'
+import { describe, expect, it, vi } from 'vitest'
+
+import { getBlastRadiusJob } from './getBlastRadiusJob'
+
+const { getAnalysisDetail, getVerdictResults } = vi.hoisted(() => ({
+  getAnalysisDetail: vi.fn(),
+  getVerdictResults: vi.fn(),
+}))
+
+vi.mock('@crowd/data-access-layer/src/packages/blastRadius', () => ({
+  getAnalysisDetail,
+  getVerdictResults,
+}))
+
+vi.mock('@/db/packagesDb', () => ({
+  getPackagesQx: vi.fn().mockResolvedValue({}),
+}))
+
+const ANALYSIS_ID = '11111111-1111-1111-1111-111111111111'
+
+function mockReqRes(params: unknown) {
+  getAnalysisDetail.mockClear()
+  getVerdictResults.mockClear()
+
+  const req = { params } 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('getBlastRadiusJob', () => {
+  it('returns a pending analysis with no results/summary', async () => {
+    getAnalysisDetail.mockResolvedValue({
+      id: ANALYSIS_ID,
+      advisory_osv_id: 'GHSA-jf85-cpcp-j695',
+      package_name: null,
+      ecosystem: 'npm',
+      status: 'pending',
+      error: null,
+      candidates_considered: null,
+      started_at: '2026-07-01T00:00:00.000Z',
+      completed_at: null,
+    })
+
+    const { req, res, json } = mockReqRes({ analysisId: ANALYSIS_ID })
+
+    await getBlastRadiusJob(req, res)
+
+    expect(getVerdictResults).not.toHaveBeenCalled()
+    expect(json).toHaveBeenCalledWith(
+      expect.objectContaining({
+        analysisId: ANALYSIS_ID,
+        status: 'pending',
+        summary: null,
+        results: null,
+      }),
+    )
+  })
+
+  it('returns summary and results for a done analysis', async () => {
+    getAnalysisDetail.mockResolvedValue({
+      id: ANALYSIS_ID,
+      advisory_osv_id: 'GHSA-jf85-cpcp-j695',
+      package_name: 'lodash',
+      ecosystem: 'npm',
+      status: 'done',
+      error: null,
+      candidates_considered: 10,
+      started_at: '2026-07-01T00:00:00.000Z',
+      completed_at: '2026-07-01T01:00:00.000Z',
+    })
+    getVerdictResults.mockResolvedValue([
+      {
+        name: 'benchmark.js',
+        version: '2.1.4',
+        downloads: 500000,
+        reachable_verdict: 'affected',
+        confidence: 0.9,
+        evidence: [{ file: 'index.js', line: 10, snippet: 'require("lodash").merge' }],
+        reasoning: 'uses merge',
+      },
+      {
+        name: 'other-pkg',
+        version: '1.0.0',
+        downloads: 100,
+        reachable_verdict: 'not_affected',
+        confidence: 0.5,
+        evidence: null,
+        reasoning: 'unused',
+      },
+    ])
+
+    const { req, res, json } = mockReqRes({ analysisId: ANALYSIS_ID })
+
+    await getBlastRadiusJob(req, res)
+
+    expect(json).toHaveBeenCalledWith(
+      expect.objectContaining({
+        status: 'done',
+        summary: expect.objectContaining({
+          totalDependentsInRange: 10,
+          dependentsAnalyzed: 2,
+          dependentsExcludedUpfront: 8,
+          dependentsAffected: 1,
+          affectedPercentage: 50,
+          affectedDependents: ['pkg:npm/benchmark.js'],
+        }),
+        results: [
+          expect.objectContaining({
+            dependent: 'pkg:npm/benchmark.js',
+            affected: true,
+            confidence: 'high',
+          }),
+          expect.objectContaining({
+            dependent: 'pkg:npm/other-pkg',
+            affected: false,
+            confidence: 'medium',
+          }),
+        ],
+      }),
+    )
+  })
+
+  it('404s when the analysis does not exist', async () => {
+    getAnalysisDetail.mockResolvedValue(null)
+
+    const { req, res } = mockReqRes({ analysisId: ANALYSIS_ID })
+
+    await expect(getBlastRadiusJob(req, res)).rejects.toThrow()
+  })
+
+  it('rejects a non-uuid analysisId', async () => {
+    const { req, res } = mockReqRes({ analysisId: 'not-a-uuid' })
+
+    await expect(getBlastRadiusJob(req, res)).rejects.toThrow()
+    expect(getAnalysisDetail).not.toHaveBeenCalled()
+  })
+})

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

@@ -0,0 +1,32 @@
+import type { Request, Response } from 'express'
+import { z } from 'zod'
+
+import { NotFoundError } from '@crowd/common'
+import * as blastRadiusDal from '@crowd/data-access-layer/src/packages/blastRadius'
+
+import { getPackagesQx } from '@/db/packagesDb'
+import { ok } from '@/utils/api'
+import { validateOrThrow } from '@/utils/validation'
+
+import { toBlastRadiusAnalysis } from './blastRadiusAnalysis'
+
+const paramsSchema = z.object({
+  analysisId: z.uuid(),
+})
+
+// 2b — poll a blast-radius analysis job. results/summary are only populated once
+// status is 'done' — see toBlastRadiusAnalysis.
+export async function getBlastRadiusJob(req: Request, res: Response): Promise<void> {
+  const { analysisId } = validateOrThrow(paramsSchema, req.params)
+
+  const qx = await getPackagesQx()
+  const analysis = await blastRadiusDal.getAnalysisDetail(qx, analysisId)
+  if (!analysis) {
+    throw new NotFoundError()
+  }
+
+  const verdictRows =
+    analysis.status === 'done' ? await blastRadiusDal.getVerdictResults(qx, analysisId) : []
+
+  ok(res, toBlastRadiusAnalysis(analysis, verdictRows))
+}

services/apps/packages_worker/src/blast-radius/activities.ts

@@ -0,0 +1,84 @@
+import { Context } from '@temporalio/activity'
+
+import * as blastRadiusDal from '@crowd/data-access-layer/src/packages/blastRadius'
+import { getServiceChildLogger } from '@crowd/logging'
+
+import { getPackagesDb } from '../db'
+
+import { runDependentsStage } from './stages/dependents'
+import { runIntelStage } from './stages/intel'
+import { runReachabilityStage } from './stages/reachability'
+import { runReportStage } from './stages/report'
+
+const log = getServiceChildLogger('blast-radius')
+
+export interface BlastRadiusActivityInput {
+  analysisId: string
+  advisoryOsvId: string
+}
+
+export interface BlastRadiusStartInput {
+  analysisId: string
+  advisoryOsvId: string
+  packageName: string | null
+  ecosystem: string
+  force: boolean
+}
+
+// Activity: create (or reuse, on retrigger) the analysis row and mark it running.
+// All DB access for the workflow lives in activities — Temporal workflow code
+// must stay deterministic and cannot talk to the database directly.
+export async function blastRadiusStart(input: BlastRadiusStartInput): Promise<void> {
+  log.info({ analysisId: input.analysisId }, 'blast-radius: starting analysis')
+  const qx = await getPackagesDb()
+  await blastRadiusDal.createAnalysis(qx, {
+    id: input.analysisId,
+    advisoryOsvId: input.advisoryOsvId,
+    packageName: input.packageName,
+    ecosystem: input.ecosystem,
+    force: input.force,
+  })
+  await blastRadiusDal.markAnalysisRunning(qx, input.analysisId)
+  log.info({ analysisId: input.analysisId }, 'blast-radius: analysis marked running')
+}
+
+// Activity: mark the analysis row failed with the given error message.
+export async function blastRadiusFail(input: { analysisId: string; error: string }): Promise<void> {
+  log.warn({ analysisId: input.analysisId, error: input.error }, 'blast-radius: analysis failed')
+  const qx = await getPackagesDb()
+  await blastRadiusDal.failAnalysis(qx, input.analysisId, input.error)
+}
+
+// Activity: Stage 1 — vulnerability intelligence
+export async function blastRadiusIntel(input: BlastRadiusActivityInput): Promise<void> {
+  log.info({ analysisId: input.analysisId }, 'blast-radius: intel stage starting')
+  const qx = await getPackagesDb()
+  await runIntelStage(qx, input.analysisId, input.advisoryOsvId)
+  Context.current().heartbeat()
+  log.info({ analysisId: input.analysisId }, 'blast-radius: intel stage done')
+}
+
+// Activity: Stage 2 — dependents scan
+export async function blastRadiusDependents(input: BlastRadiusActivityInput): Promise<void> {
+  log.info({ analysisId: input.analysisId }, 'blast-radius: dependents stage starting')
+  const qx = await getPackagesDb()
+  await runDependentsStage(qx, input.analysisId, () => Context.current().heartbeat())
+  log.info({ analysisId: input.analysisId }, 'blast-radius: dependents stage done')
+}
+
+// Activity: Stage 3 — reachability analysis
+export async function blastRadiusReachability(input: BlastRadiusActivityInput): Promise<void> {
+  log.info({ analysisId: input.analysisId }, 'blast-radius: reachability stage starting')
+  const qx = await getPackagesDb()
+  await runReachabilityStage(qx, input.analysisId, () => Context.current().heartbeat())
+  log.info({ analysisId: input.analysisId }, 'blast-radius: reachability stage done')
+}
+
+// Activity: Stage 4 — report aggregation
+export async function blastRadiusReport(input: BlastRadiusActivityInput): Promise<void> {
+  log.info({ analysisId: input.analysisId }, 'blast-radius: report stage starting')
+  const qx = await getPackagesDb()
+  await runReportStage(qx, input.analysisId)
+  Context.current().heartbeat()
+  log.info({ analysisId: input.analysisId }, 'blast-radius: report stage done')
+}