fix: security contacts ingestion improvements [CM-1297] (#4295)

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

Author: mbani01

backend/src/osspckgs/migrations/V1783036800__security_contacts_soft_delete.sql

@@ -0,0 +1,5 @@
+ALTER TABLE security_contacts ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ;
+
+CREATE INDEX IF NOT EXISTS security_contacts_repo_active_idx
+   ON security_contacts (repo_id)
+   WHERE deleted_at IS NULL;

services/apps/packages_worker/src/security-contacts/extractors/gitTree.ts

@@ -0,0 +1,39 @@
+import { ExtractorDeps } from '../types'
+
+/** null means unresolved (empty repo, truncated, fetch failure) — callers must fall back to probing. */
+export interface RepoTree {
+  paths: Set<string> | null
+}
+
+interface GitTreeEntry {
+  path?: unknown
+  type?: unknown
+}
+
+interface GitTreeResponse {
+  tree?: GitTreeEntry[]
+  truncated?: boolean
+}
+
+// `HEAD` resolves to the default branch on the git data API.
+export async function fetchRepoTree(
+  owner: string,
+  name: string,
+  githubGet: ExtractorDeps['githubGet'],
+): Promise<RepoTree> {
+  try {
+    const { text } = await githubGet(`/repos/${owner}/${name}/git/trees/HEAD?recursive=1`)
+    if (!text) return { paths: null }
+
+    const doc = JSON.parse(text) as GitTreeResponse
+    if (doc.truncated || !Array.isArray(doc.tree)) return { paths: null }
+
+    const paths = new Set<string>()
+    for (const entry of doc.tree) {
+      if (entry.type === 'blob' && typeof entry.path === 'string') paths.add(entry.path)
+    }
+    return { paths }
+  } catch {
+    return { paths: null }
+  }
+}

services/apps/packages_worker/src/security-contacts/extractors/http.ts

@@ -4,14 +4,11 @@ export function registryHeaders(userAgent: string): Record<string, string> {
   return { 'User-Agent': userAgent }
 }
 
-// Genuinely-absent / not-determinable → null body; every other non-200 throws so transient
-// failures (5xx/...) are treated as failures and the pipeline preserves data instead of
-// wiping it. 422 is included because GitHub's PVR endpoint returns it (per-repo, non-transient)
-// when the flag can't be determined — that must read as "unknown", not block the whole repo.
-const ABSENT_STATUSES = new Set([404, 410, 422])
+// Genuinely-absent/not-determinable → null body; every other non-200 throws so the pipeline
+// preserves existing data instead of wiping it. 422 covers GitHub's PVR "can't determine".
+const ABSENT_STATUSES = new Set([404, 410, 422, 451])
 
-// Registry rate-limit / overload responses: retried in-process (honoring Retry-After) so a brief
-// throttle doesn't fail the extractor and cost the repo a whole refresh cadence.
+// Retried in-process (honoring Retry-After) so a brief throttle doesn't cost a whole cadence.
 const RATE_LIMIT_STATUSES = new Set([429, 503])
 const MAX_RATE_LIMIT_RETRIES = 3
 

services/apps/packages_worker/src/security-contacts/extractors/securityContactsFile.ts

@@ -55,6 +55,8 @@ export const extractSecurityContactsFile: Extractor = async (target, deps) => {
     return { contacts: [], policies: {} }
   }
 
+  if (deps.repoTree.paths && !deps.repoTree.paths.has(PATH)) return { contacts: [], policies: {} }
+
   const { text } = await deps.githubGet(`/repos/${owner}/${name}/contents/${PATH}`, { raw: true })
   if (!text) return { contacts: [], policies: {} }
 

services/apps/packages_worker/src/security-contacts/extractors/securityInsights.ts

@@ -211,8 +211,10 @@ export const extractSecurityInsights: Extractor = async (target, deps) => {
   }
 
   const fetchedAt = new Date().toISOString()
+  const { paths: treePaths } = deps.repoTree
 
   for (const path of PATHS) {
+    if (treePaths && !treePaths.has(path)) continue
     const { text } = await deps.githubGet(`/repos/${owner}/${name}/contents/${path}`, { raw: true })
     if (!text) continue
 

services/apps/packages_worker/src/security-contacts/extractors/securityMd.ts

@@ -109,7 +109,9 @@ export const extractSecurityMd: Extractor = async (target, deps) => {
   }
 
   const fetchedAt = new Date().toISOString()
+  const { paths: treePaths } = deps.repoTree
   for (const path of PATHS) {
+    if (treePaths && !treePaths.has(path)) continue
     const { text } = await deps.githubGet(`/repos/${owner}/${name}/contents/${path}`, { raw: true })
     if (text) return parseSecurityMd(text, owner, name, path, fetchedAt)
   }

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

@@ -15,17 +15,13 @@ const log = getServiceChildLogger('security-contacts:github-token')
 
 const GITHUB_API = 'https://api.github.com'
 
-// Genuinely-absent / not-determinable → null body (see http.ts for the same set). 422 covers the
-// PVR endpoint returning "can't determine" per-repo; it must read as unknown, not a hard failure.
-const ABSENT_STATUSES = new Set([404, 410, 422])
+// Genuinely-absent / not-determinable → null body, not a failure (see http.ts for the same set).
+// 422 covers GitHub's PVR endpoint returning "can't determine" per-repo; 451 is permanent removal.
+const ABSENT_STATUSES = new Set([404, 410, 422, 451])
 
-// App-wide ceiling on concurrent GitHub requests. GitHub's secondary limit rejects bursts of
-// >100 concurrent requests from one app; staying under that (across all installations) is the
-// single most effective guard against secondary-limit 429s at high repo concurrency.
+// App-wide ceiling on concurrent requests — GitHub's secondary limit rejects bursts >100/app.
 const MAX_CONCURRENT_GITHUB_REQUESTS = 50
 
-// Bound the park/switch/backoff retry loop so a persistently-limited request eventually surfaces
-// as a failure (which the pipeline treats as transient and preserves existing data).
 const MAX_RATE_LIMIT_RETRIES = 6
 
 /** Minimal async semaphore with fair FIFO hand-off, used to cap concurrent GitHub requests. */
@@ -126,13 +122,9 @@ async function fetchOnce(
 }
 
 /**
- * Rate-limit-safe GitHub API GET. Selects an installation from the pool, sleeps if all are parked,
- * feeds response budget headers back so exhausted installations get parked before they 403, and on
- * a rate-limit response parks (primary) or waits out Retry-After (secondary, app-wide) then retries
- * on another installation. Falls back to a single unauthenticated request when no App is configured.
- *
- * Returns text on 200; null body for absent resources (404/410/422); throws on other non-200s and
- * once the retry budget is exhausted (callers treat throws as transient and preserve existing data).
+ * Rate-limit-safe GitHub API GET. Rotates across installations in the pool, parking exhausted
+ * or rate-limited ones and retrying, up to MAX_RATE_LIMIT_RETRIES. Falls back to a single
+ * unauthenticated request when no App is configured.
  */
 export async function githubApiGet(
   path: string,

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

@@ -1,9 +1,13 @@
+import { cancellationSignal, heartbeat } from '@temporalio/activity'
+
 import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
 import { getServiceChildLogger } from '@crowd/logging'
 
 import { getSecurityContactsConfig } from '../config'
+import { parseGithubUrl } from '../enricher/fetchLightRepo'
 import { mapWithConcurrency } from '../utils/concurrency'
 
+import { fetchRepoTree } from './extractors/gitTree'
 import { extractPvr } from './extractors/pvr'
 import { extractManifest } from './extractors/registry'
 import { extractSecurityContactsFile } from './extractors/securityContactsFile'
@@ -31,18 +35,11 @@ export interface BatchResult {
   processed: number
 }
 
-// Two-tier refresh cadence (the worker runs on a daily cron). Just under 24h/168h so a repo
-// processed at ~06:00 is eligible again at the next daily/weekly tick rather than slipping a day.
-const DAILY_INTERVAL_HOURS = 20 // repos never evaluated or with no contacts yet
-const WEEKLY_INTERVAL_HOURS = 156 // already-enriched repos (have contacts)
-
-// Tuned for throughput within the platform ceilings:
-//  - CONCURRENCY: parallel repos. GitHub calls go through the authed Contents API via a
-//    rate-limit-aware pool (per-installation budget parking + app-wide concurrency gate +
-//    Retry-After backoff in githubToken), so high repo concurrency won't trip GitHub limits.
-//  - FETCH_TIMEOUT_MS: generous enough for slow registries (Maven metadata/POM) without hanging slots.
-//  - BATCH_SIZE: bounded by the 30-min activity timeout (worst case: an all-cargo batch throttled
-//    to crates.io's 1 req/s finishes ~8 min).
+// Just under 24h/168h so a repo processed at ~06:00 is still eligible at the next tick.
+const DAILY_INTERVAL_HOURS = 20 // no contacts found yet
+const WEEKLY_INTERVAL_HOURS = 156 // already has contacts
+
+// GitHub calls are rate-limit-aware (see githubToken.ts), so high repo concurrency is safe.
 const CONCURRENCY = 100
 const FETCH_TIMEOUT_MS = 15000
 const BATCH_SIZE = 500
@@ -81,12 +78,12 @@ async function fetchBatch(qx: QueryExecutor): Promise<SweepRow[]> {
         r.contacts_last_refreshed IS NULL
         -- evaluated but no contacts found yet → retry on the daily cadence
         OR (
-          NOT EXISTS (SELECT 1 FROM security_contacts sc WHERE sc.repo_id = r.id)
+          NOT EXISTS (SELECT 1 FROM security_contacts sc WHERE sc.repo_id = r.id AND sc.deleted_at IS NULL)
           AND r.contacts_last_refreshed < NOW() - INTERVAL '$(dailyIntervalHours) hours'
         )
         -- already enriched (has contacts) → refresh on the weekly cadence
         OR (
-          EXISTS (SELECT 1 FROM security_contacts sc WHERE sc.repo_id = r.id)
+          EXISTS (SELECT 1 FROM security_contacts sc WHERE sc.repo_id = r.id AND sc.deleted_at IS NULL)
           AND r.contacts_last_refreshed < NOW() - INTERVAL '$(weeklyIntervalHours) hours'
         )
       )
@@ -114,14 +111,22 @@ function toTarget(row: SweepRow): RepoTarget {
 
 async function processRepo(
   target: RepoTarget,
-  deps: ExtractorDeps,
+  baseDeps: Omit<ExtractorDeps, 'repoTree'>,
   qx: QueryExecutor,
 ): Promise<void> {
+  // One tree fetch per repo, shared by extractors that probe well-known paths.
+  let repoTree: ExtractorDeps['repoTree'] = { paths: null }
+  try {
+    const { owner, name } = parseGithubUrl(target.url)
+    repoTree = await fetchRepoTree(owner, name, baseDeps.githubGet)
+  } catch {
+    // not a github.com URL
+  }
+  const deps: ExtractorDeps = { ...baseDeps, repoTree }
+
   const results = await Promise.allSettled(EXTRACTORS.map((extract) => extract(target, deps)))
 
-  // Replace the repo's contacts only when every extractor succeeded. If any failed (transient
-  // error — non-200s throw), a destructive rewrite would drop contacts a failed tier-A/B extractor
-  // still has, so preserve existing data and just record the attempt; retried next cadence.
+  // Write only when every extractor succeeded, so a failed one can't wipe contacts it didn't see.
   const failed = results.find((r) => r.status === 'rejected') as PromiseRejectedResult | undefined
   if (failed) {
     log.warn(
@@ -144,8 +149,7 @@ async function processRepo(
     }
   }
 
-  // A2 veto: B1 may emit a github-pvr contact from redirect language; drop it when A2
-  // authoritatively reports PVR disabled (Option C from the design discussion).
+  // A2 veto: drop B1's github-pvr guess when A2 authoritatively reports PVR disabled.
   if (policies.pvrEnabled === false) {
     contacts = contacts.filter((c) => c.channel !== 'github-pvr')
   }
@@ -158,26 +162,45 @@ export async function processBatch(qx: QueryExecutor, config: Config): Promise<B
   const batch = await fetchBatch(qx)
   if (batch.length === 0) return { processed: 0 }
 
-  const deps: ExtractorDeps = {
+  const deps: Omit<ExtractorDeps, 'repoTree'> = {
     fetchTimeoutMs: FETCH_TIMEOUT_MS,
     userAgent: config.userAgent,
     githubGet: (path, opts) => githubApiGet(path, FETCH_TIMEOUT_MS, opts),
   }
 
   const targets = batch.map(toTarget)
-  // Isolate per-repo failures: mapWithConcurrency is fail-fast, so a single repo's DB error must
-  // not abort the batch (and, across Temporal retries, halt the whole sweep). Unmarked repos on
-  // error are simply retried next sweep.
-  await mapWithConcurrency(targets, CONCURRENCY, async (target) => {
+  // Fixed-cadence heartbeat: a slow repo can outlast the 2-minute heartbeatTimeout even while
+  // every concurrency slot is still busy, so this can't rely on task completions alone.
+  const heartbeatTimer = setInterval(() => {
     try {
-      await processRepo(target, deps, qx)
+      heartbeat()
     } catch (err) {
-      log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed')
-      // Best-effort mark so a persistently-failing repo drains on cadence rather than making the
-      // sweep hot-loop (it would otherwise stay eligible and keep processed > 0 forever).
-      await markRepoAttempted(qx, target.repoId).catch(() => undefined)
+      log.warn({ errMsg: (err as Error).message }, 'Heartbeat failed')
     }
-  })
+  }, 30_000)
+  try {
+    // A cancelled task (superseded by a newer activity attempt) is left to throw so it stops
+    // scheduling further repos instead of racing the new attempt.
+    await mapWithConcurrency(targets, CONCURRENCY, async (target) => {
+      if (cancellationSignal().aborted) {
+        throw new Error(
+          'Security contacts batch cancelled — superseded by a newer activity attempt',
+        )
+      }
+      try {
+        await processRepo(target, deps, qx)
+      } catch (err) {
+        log.error(
+          { repoId: target.repoId, errMsg: (err as Error).message },
+          'Repo processing failed',
+        )
+        // Best-effort: keeps a persistently-failing repo from hot-looping the sweep.
+        await markRepoAttempted(qx, target.repoId).catch(() => undefined)
+      }
+    })
+  } finally {
+    clearInterval(heartbeatTimer)
+  }
 
   log.info({ processed: targets.length }, 'Security contacts batch complete')
   return { processed: targets.length }

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

@@ -20,6 +20,44 @@ const ROLE_PRIORITY: Record<ContactRole, number> = {
 
 const TIER_RANK: Record<SourceTier, number> = { A: 4, B: 3, C: 2, D: 1 }
 
+// RFC 2606 reserved domains — always unedited template placeholders (e.g. security@example.com).
+const PLACEHOLDER_EMAIL_DOMAINS = new Set([
+  'example.com',
+  'example.org',
+  'example.net',
+  'example.edu',
+])
+
+// Generic GitHub/Dependabot help pages that templated SECURITY.md files link to — never a
+// project-specific contact, unlike an actual github.com/<owner>/<repo>/... URL.
+const GENERIC_URL_HOSTS = new Set(['docs.github.com', 'dependabot.com', 'www.dependabot.com'])
+
+const HAS_SCHEME_RE = /^[a-z][a-z0-9+.-]*:\/\//i
+
+// Some upstream sources (e.g. SECURITY-INSIGHTS {type:"url"} entries) don't enforce a scheme —
+// only add one to resolve the host, never mutate the stored contact value.
+function urlHost(value: string): string | null {
+  const candidate = HAS_SCHEME_RE.test(value) ? value : `https://${value}`
+  try {
+    return new URL(candidate).hostname.toLowerCase()
+  } catch {
+    return null
+  }
+}
+
+function isJunkContact(c: RawContact): boolean {
+  if (c.channel === 'email') {
+    const domain = c.value.split('@')[1]?.toLowerCase().trim()
+    return domain != null && PLACEHOLDER_EMAIL_DOMAINS.has(domain)
+  }
+  if (c.channel === 'url' || c.channel === 'web-form') {
+    const host = urlHost(c.value)
+    if (host == null) return true
+    return GENERIC_URL_HOSTS.has(host) || host === 'localhost' || host.startsWith('127.')
+  }
+  return false
+}
+
 function normalizeValue(channel: ContactChannel, value: string): string {
   const v = value.trim()
   return channel === 'email' || channel === 'github-handle' ? v.toLowerCase() : v
@@ -90,7 +128,7 @@ function identityLinkMerge(contacts: RawContact[]): RawContact[] {
 }
 
 export function reconcile(contacts: RawContact[], now: Date = new Date()): ScoredContact[] {
-  const merged = identityLinkMerge(exactMatchMerge(contacts))
+  const merged = identityLinkMerge(exactMatchMerge(contacts.filter((c) => !isJunkContact(c))))
 
   const scored: ScoredContact[] = merged.map((c) => {
     const contact = { ...c, provenance: dedupeProvenance(c.provenance) }