From a19af957bb95e5d42775f4915cce8000c01dae2c Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 7 Jul 2026 11:05:01 +0100 Subject: [PATCH 1/9] feat: add on-demand security-contacts ingest for a single purl (CM-1313) Signed-off-by: Mouad BANI --- .../src/api/public/v1/packages/getPackage.ts | 27 +++++- .../apps/packages_worker/src/activities.ts | 5 +- .../src/security-contacts/activities.ts | 12 +++ .../src/security-contacts/ingestSingle.ts | 91 +++++++++++++++++++ .../src/security-contacts/processBatch.ts | 16 ++-- .../src/security-contacts/workflows.ts | 18 ++++ .../packages_worker/src/workflows/index.ts | 5 +- 7 files changed, 165 insertions(+), 9 deletions(-) create mode 100644 services/apps/packages_worker/src/security-contacts/ingestSingle.ts diff --git a/backend/src/api/public/v1/packages/getPackage.ts b/backend/src/api/public/v1/packages/getPackage.ts index 44c61ca50f..25137cbb21 100644 --- a/backend/src/api/public/v1/packages/getPackage.ts +++ b/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 { 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, diff --git a/services/apps/packages_worker/src/activities.ts b/services/apps/packages_worker/src/activities.ts index b747a3bd3d..5b035b52cb 100644 --- a/services/apps/packages_worker/src/activities.ts +++ b/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' diff --git a/services/apps/packages_worker/src/security-contacts/activities.ts b/services/apps/packages_worker/src/security-contacts/activities.ts index e6be0fecde..454d10e32f 100644 --- a/services/apps/packages_worker/src/security-contacts/activities.ts +++ b/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 { log.info({ ...result }, 'Security contacts batch activity complete') return result } + +export async function ingestSecurityContactsForPurlActivity( + purl: string, +): Promise { + 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 +} diff --git a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts new file mode 100644 index 0000000000..840abfde3b --- /dev/null +++ b/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 + +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 { + 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 { + 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 } +} diff --git a/services/apps/packages_worker/src/security-contacts/processBatch.ts b/services/apps/packages_worker/src/security-contacts/processBatch.ts index 9033a8dfa5..ec8aca07d4 100644 --- a/services/apps/packages_worker/src/security-contacts/processBatch.ts +++ b/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 { + 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, qx: QueryExecutor, @@ -162,11 +170,7 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise = { - 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 diff --git a/services/apps/packages_worker/src/security-contacts/workflows.ts b/services/apps/packages_worker/src/security-contacts/workflows.ts index fa9c64b072..2bfff86514 100644 --- a/services/apps/packages_worker/src/security-contacts/workflows.ts +++ b/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({ startToCloseTimeout: '30 minutes', @@ -14,6 +15,17 @@ const acts = proxyActivities({ }, }) +// 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({ + startToCloseTimeout: '45 seconds', + retry: { + initialInterval: '5 seconds', + maximumAttempts: 2, + }, +}) + export async function ingestSecurityContacts(): Promise { const result = await acts.processSecurityContactsBatch() if (result.processed === 0) { @@ -22,3 +34,9 @@ export async function ingestSecurityContacts(): Promise { } await continueAsNew() } + +export async function ingestSecurityContactsForPurlWorkflow( + purl: string, +): Promise { + return singleActs.ingestSecurityContactsForPurlActivity(purl) +} diff --git a/services/apps/packages_worker/src/workflows/index.ts b/services/apps/packages_worker/src/workflows/index.ts index 91cf431947..9b762282b4 100644 --- a/services/apps/packages_worker/src/workflows/index.ts +++ b/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' From 471ad1df1f213a89f0dbcc957af8877ae6ded23f Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 7 Jul 2026 12:26:16 +0100 Subject: [PATCH 2/9] fix: use packages temporal on api Signed-off-by: Mouad BANI --- .../config/custom-environment-variables.json | 6 +++ backend/config/default.json | 1 + .../src/api/public/v1/packages/getPackage.ts | 4 +- backend/src/conf/index.ts | 8 ++++ backend/src/db/packagesTemporal.ts | 47 +++++++++++++++++++ .../services/security-contacts-worker.yaml | 1 + 6 files changed, 66 insertions(+), 1 deletion(-) create mode 100644 backend/src/db/packagesTemporal.ts diff --git a/backend/config/custom-environment-variables.json b/backend/config/custom-environment-variables.json index 4d1d01b6f5..c79dab0671 100644 --- a/backend/config/custom-environment-variables.json +++ b/backend/config/custom-environment-variables.json @@ -180,6 +180,12 @@ "certificate": "CROWD_TEMPORAL_CERTIFICATE", "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY" }, + "packagesTemporal": { + "serverUrl": "CROWD_TEMPORAL_SERVER_URL", + "namespace": "CROWD_PACKAGES_TEMPORAL_NAMESPACE", + "certificate": "CROWD_TEMPORAL_CERTIFICATE", + "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY" + }, "searchSyncApi": { "baseUrl": "CROWD_SEARCH_SYNC_API_URL" }, diff --git a/backend/config/default.json b/backend/config/default.json index e60af044e2..de0a7a9e53 100644 --- a/backend/config/default.json +++ b/backend/config/default.json @@ -43,6 +43,7 @@ "isEnabled": "false" }, "temporal": {}, + "packagesTemporal": {}, "searchSyncApi": {}, "encryption": {}, "openStatusApi": {}, diff --git a/backend/src/api/public/v1/packages/getPackage.ts b/backend/src/api/public/v1/packages/getPackage.ts index 25137cbb21..491e10f75e 100644 --- a/backend/src/api/public/v1/packages/getPackage.ts +++ b/backend/src/api/public/v1/packages/getPackage.ts @@ -13,6 +13,7 @@ import { 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' @@ -54,7 +55,8 @@ export async function getPackage(req: Request, res: Response): Promise { if (pkg.contactsLastRefreshed == null) { try { - await req.temporal.workflow.execute('ingestSecurityContactsForPurlWorkflow', { + const packagesTemporal = await getPackagesTemporalClient() + await packagesTemporal.workflow.execute('ingestSecurityContactsForPurlWorkflow', { taskQueue: 'packages-worker', workflowId: ondemandWorkflowId(purl), workflowIdReusePolicy: WorkflowIdReusePolicy.ALLOW_DUPLICATE, diff --git a/backend/src/conf/index.ts b/backend/src/conf/index.ts index 553167cde5..938ffb66c2 100644 --- a/backend/src/conf/index.ts +++ b/backend/src/conf/index.ts @@ -86,6 +86,14 @@ export const PACKAGES_DB_CONFIG: IDatabaseConfig | undefined = config.has('packa ? config.get('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('packagesTemporal') + : undefined + export const SEGMENT_CONFIG: SegmentConfiguration = config.get('segment') export const COMPREHEND_CONFIG: ComprehendConfiguration = diff --git a/backend/src/db/packagesTemporal.ts b/backend/src/db/packagesTemporal.ts new file mode 100644 index 0000000000..8d2561a01b --- /dev/null +++ b/backend/src/db/packagesTemporal.ts @@ -0,0 +1,47 @@ +import { Client, Connection, getDataConverter } from '@crowd/temporal' +import { IS_DEV_ENV, SERVICE } from '@crowd/common' + +import { PACKAGES_TEMPORAL_CONFIG } from '@/conf' + +let _init: Promise | 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 { + 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 +} diff --git a/scripts/services/security-contacts-worker.yaml b/scripts/services/security-contacts-worker.yaml index 934afe2c69..4706d39425 100644 --- a/scripts/services/security-contacts-worker.yaml +++ b/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: From ffcd3ac21169b7c4619ededcc706e180ce73017a Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 7 Jul 2026 14:52:18 +0100 Subject: [PATCH 3/9] fix: missing temporal config Signed-off-by: Mouad BANI --- backend/.env.dist.local | 1 + backend/config/custom-environment-variables.json | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/backend/.env.dist.local b/backend/.env.dist.local index adca4976c3..3fdede3c01 100755 --- a/backend/.env.dist.local +++ b/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 diff --git a/backend/config/custom-environment-variables.json b/backend/config/custom-environment-variables.json index c79dab0671..93de9c67da 100644 --- a/backend/config/custom-environment-variables.json +++ b/backend/config/custom-environment-variables.json @@ -181,7 +181,7 @@ "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY" }, "packagesTemporal": { - "serverUrl": "CROWD_TEMPORAL_SERVER_URL", + "serverUrl": "CROWD_PACKAGES_TEMPORAL_SERVER_URL", "namespace": "CROWD_PACKAGES_TEMPORAL_NAMESPACE", "certificate": "CROWD_TEMPORAL_CERTIFICATE", "privateKey": "CROWD_TEMPORAL_PRIVATE_KEY" From 1402d42bc71dec4875ae195f20b3077445ad7beb Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 7 Jul 2026 15:33:09 +0100 Subject: [PATCH 4/9] fix: stop excluding non-github repos from security contacts ingest Signed-off-by: Mouad BANI --- .../packages_worker/src/security-contacts/ingestSingle.ts | 7 ++++++- .../packages_worker/src/security-contacts/processBatch.ts | 3 +-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts index 840abfde3b..5020a73dd0 100644 --- a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts +++ b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts @@ -29,6 +29,11 @@ interface SingleRepoRow { // 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) +// and always stamps contacts_last_refreshed — 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 { return qx.selectOneOrNone( ` @@ -51,7 +56,7 @@ async function findBestRepoForPurl(qx: QueryExecutor, purl: string): Promise { 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 From 59fdeee193bbf408f45ff5967bd7d9eb10d06458 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 7 Jul 2026 15:34:50 +0100 Subject: [PATCH 5/9] fix: catch processRepo failures in on-demand security contacts ingest Signed-off-by: Mouad BANI --- .../packages_worker/src/security-contacts/ingestSingle.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts index 5020a73dd0..e20a472921 100644 --- a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts +++ b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts @@ -5,6 +5,7 @@ import { getSecurityContactsConfig } from '../config' import { buildBaseDeps, processRepo } from './processBatch' import { RepoPackage, RepoTarget } from './types' +import { markRepoAttempted } from './writeContacts' const log = getServiceChildLogger('security-contacts-ondemand') @@ -90,7 +91,12 @@ export async function ingestSecurityContactsForPurl( const target = toTarget(row) const baseDeps = buildBaseDeps(config) - await processRepo(target, baseDeps, qx) + 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 } } From 418e40bae838621aecf9809af8a77f4395b96e6e Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 7 Jul 2026 15:44:22 +0100 Subject: [PATCH 6/9] chore: update comments and timeout Signed-off-by: Mouad BANI --- backend/src/api/public/v1/packages/getPackage.ts | 4 ++++ .../packages_worker/src/security-contacts/ingestSingle.ts | 8 ++++---- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/backend/src/api/public/v1/packages/getPackage.ts b/backend/src/api/public/v1/packages/getPackage.ts index 491e10f75e..e66f7be986 100644 --- a/backend/src/api/public/v1/packages/getPackage.ts +++ b/backend/src/api/public/v1/packages/getPackage.ts @@ -61,6 +61,10 @@ export async function getPackage(req: Request, res: Response): Promise { 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 diff --git a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts index e20a472921..e53ef374e1 100644 --- a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts +++ b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts @@ -32,9 +32,9 @@ interface SingleRepoRow { // 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) -// and always stamps contacts_last_refreshed — filtering here would leave non-github repos -// permanently NULL, re-triggering this on-demand path on every single request. +// 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 { return qx.selectOneOrNone( ` @@ -76,7 +76,7 @@ function toTarget(row: SingleRepoRow): RepoTarget { /** * 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. + * (repos.contacts_last_refreshed IS NULL). */ export async function ingestSecurityContactsForPurl( qx: QueryExecutor, From a536c7f3bb5bc28a9e570b929c48822e74c1f1fe Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 7 Jul 2026 15:45:46 +0100 Subject: [PATCH 7/9] chore: update log Signed-off-by: Mouad BANI --- .../apps/packages_worker/src/security-contacts/ingestSingle.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts index e53ef374e1..61e162f0a2 100644 --- a/services/apps/packages_worker/src/security-contacts/ingestSingle.ts +++ b/services/apps/packages_worker/src/security-contacts/ingestSingle.ts @@ -85,7 +85,7 @@ export async function ingestSecurityContactsForPurl( ): Promise { const row = await findBestRepoForPurl(qx, purl) if (!row) { - log.info({ purl }, 'On-demand security contacts: no linked github repo found — skipping') + log.info({ purl }, 'On-demand security contacts: no linked repo found — skipping') return { found: false } } From 24bb6c17447bf4dc3a545e50ccbf1eb8a9c73465 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Tue, 7 Jul 2026 15:46:56 +0100 Subject: [PATCH 8/9] chore: fix lint Signed-off-by: Mouad BANI --- backend/src/api/public/v1/packages/getPackage.ts | 1 - backend/src/db/packagesTemporal.ts | 6 ++---- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/backend/src/api/public/v1/packages/getPackage.ts b/backend/src/api/public/v1/packages/getPackage.ts index e66f7be986..a8b435d121 100644 --- a/backend/src/api/public/v1/packages/getPackage.ts +++ b/backend/src/api/public/v1/packages/getPackage.ts @@ -1,5 +1,4 @@ import { createHash } from 'crypto' - import type { Request, Response } from 'express' import { NotFoundError } from '@crowd/common' diff --git a/backend/src/db/packagesTemporal.ts b/backend/src/db/packagesTemporal.ts index 8d2561a01b..4e5a981d1c 100644 --- a/backend/src/db/packagesTemporal.ts +++ b/backend/src/db/packagesTemporal.ts @@ -1,5 +1,5 @@ -import { Client, Connection, getDataConverter } from '@crowd/temporal' import { IS_DEV_ENV, SERVICE } from '@crowd/common' +import { Client, Connection, getDataConverter } from '@crowd/temporal' import { PACKAGES_TEMPORAL_CONFIG } from '@/conf' @@ -11,9 +11,7 @@ let _init: Promise | undefined export function getPackagesTemporalClient(): Promise { if (!_init) { if (!PACKAGES_TEMPORAL_CONFIG?.serverUrl) { - throw new Error( - 'Packages Temporal is not configured — set CROWD_PACKAGES_TEMPORAL_NAMESPACE', - ) + throw new Error('Packages Temporal is not configured — set CROWD_PACKAGES_TEMPORAL_NAMESPACE') } const cfg = PACKAGES_TEMPORAL_CONFIG From cde9fdac914c2f1d072df7fcce3e72bc7b343321 Mon Sep 17 00:00:00 2001 From: Mouad BANI Date: Wed, 8 Jul 2026 11:37:06 +0100 Subject: [PATCH 9/9] chore: revert enabling on-demand on packages api Signed-off-by: Mouad BANI --- .../src/api/public/v1/packages/getPackage.ts | 32 +------------------ 1 file changed, 1 insertion(+), 31 deletions(-) diff --git a/backend/src/api/public/v1/packages/getPackage.ts b/backend/src/api/public/v1/packages/getPackage.ts index a8b435d121..44c61ca50f 100644 --- a/backend/src/api/public/v1/packages/getPackage.ts +++ b/backend/src/api/public/v1/packages/getPackage.ts @@ -1,4 +1,3 @@ -import { createHash } from 'crypto' import type { Request, Response } from 'express' import { NotFoundError } from '@crowd/common' @@ -9,10 +8,8 @@ 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' @@ -35,43 +32,16 @@ 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 { const { purl } = validateOrThrow(purlQuerySchema, req.query) const qx = await getPackagesQx() - let pkg = await getPackageDetailByPurl(qx, purl) + const 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,