feat(security-contacts): emails reachability & noise filtering [CM-1324]

backend/src/osspckgs/migrations/V1784073600__security_contacts_reachability.sql

@@ -0,0 +1,2 @@
+ALTER TABLE security_contacts ADD COLUMN IF NOT EXISTS reachable           BOOLEAN NOT NULL DEFAULT TRUE;
+ALTER TABLE security_contacts ADD COLUMN IF NOT EXISTS reachability_reason TEXT;

backend/src/osspckgs/migrations/V1784080000__repos_contacts_last_refreshed_idx.sql

@@ -0,0 +1,4 @@
+-- Supports fetchBatch's eligibility WHERE clause (contacts_last_refreshed IS NULL,
+-- and the daily/weekly age comparisons), which previously had no supporting index.
+CREATE INDEX IF NOT EXISTS repos_contacts_last_refreshed_idx
+    ON repos (contacts_last_refreshed);

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

@@ -5,7 +5,7 @@ import { getSecurityContactsConfig } from '../config'
 
 import { buildBaseDeps, processRepo } from './processBatch'
 import { RepoPackage, RepoTarget } from './types'
-import { markRepoAttempted } from './writeContacts'
+import { markRepoAttempted, writeContacts } from './writeContacts'
 
 const log = getServiceChildLogger('security-contacts-ondemand')
 
@@ -93,7 +93,12 @@ export async function ingestSecurityContactsForPurl(
   const target = toTarget(row)
   const baseDeps = buildBaseDeps(config)
   try {
-    await processRepo(target, baseDeps, qx, cdpQx)
+    const result = await processRepo(target, baseDeps, cdpQx)
+    if (result.status === 'ok') {
+      await writeContacts(qx, result.repoId, result.contacts, result.policies)
+    } else {
+      await markRepoAttempted(qx, result.repoId)
+    }
   } catch (err) {
     log.error({ repoId: target.repoId, errMsg: (err as Error).message }, 'Repo processing failed')
     await markRepoAttempted(qx, target.repoId).catch(() => undefined)

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

@@ -20,12 +20,13 @@ import { resolveCdpEmails } from './resolveCdpEmails'
 import {
   Extractor,
   ExtractorDeps,
+  ProcessRepoResult,
   RawContact,
   RepoPackage,
   RepoPolicies,
   RepoTarget,
 } from './types'
-import { markRepoAttempted, writeContacts } from './writeContacts'
+import { writeContactsBatch } from './writeContacts'
 
 const log = getServiceChildLogger('security-contacts')
 
@@ -120,9 +121,8 @@ export function buildBaseDeps(config: Config): Omit<ExtractorDeps, 'repoTree'> {
 export async function processRepo(
   target: RepoTarget,
   baseDeps: Omit<ExtractorDeps, 'repoTree'>,
-  qx: QueryExecutor,
   cdpQx: QueryExecutor,
-): Promise<void> {
+): Promise<ProcessRepoResult> {
   // One tree fetch per repo, shared by extractors that probe well-known paths.
   let repoTree: ExtractorDeps['repoTree'] = { paths: null }
   try {
@@ -142,8 +142,7 @@ export async function processRepo(
       { repoId: target.repoId, errMsg: failed.reason?.message },
       'Extractor failed — preserving existing data',
     )
-    await markRepoAttempted(qx, target.repoId)
-    return
+    return { repoId: target.repoId, status: 'extractor-failed' }
   }
 
   let contacts: RawContact[] = []
@@ -176,7 +175,7 @@ export async function processRepo(
   }
 
   const scored = reconcile(contacts)
-  await writeContacts(qx, target.repoId, scored, policies)
+  return { repoId: target.repoId, status: 'ok', contacts: scored, policies }
 }
 
 export async function processBatch(
@@ -199,30 +198,58 @@ export async function processBatch(
       log.warn({ errMsg: (err as Error).message }, 'Heartbeat failed')
     }
   }, 30_000)
+  let outcomes: ProcessRepoResult[]
+  const extractionStartedAt = Date.now()
   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) => {
+    //
+    // Extraction is collected here and persisted afterward in one batched call (below) rather
+    // than per-repo, since per-repo writes meant up to CONCURRENCY concurrent transactions
+    // against a packages-db pool sized for far fewer connections — the actual sweep bottleneck.
+    outcomes = 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, cdpQx)
+        return await processRepo(target, deps, cdpQx)
       } 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)
+        return { repoId: target.repoId, status: 'extractor-failed' as const }
       }
     })
   } finally {
     clearInterval(heartbeatTimer)
   }
 
+  const extractionDurationMs = Date.now() - extractionStartedAt
+  const ok = outcomes.filter((o) => o.status === 'ok').length
+  log.info(
+    {
+      ok,
+      failed: outcomes.length - ok,
+      durationMs: extractionDurationMs,
+      reposPerSec: Number((outcomes.length / (extractionDurationMs / 1000)).toFixed(1)),
+    },
+    'Security contacts batch extraction complete — persisting',
+  )
+
+  const writeStartedAt = Date.now()
+  await writeContactsBatch(qx, outcomes)
+  const writeDurationMs = Date.now() - writeStartedAt
+  log.info(
+    {
+      durationMs: writeDurationMs,
+      reposPerSec: Number((outcomes.length / (writeDurationMs / 1000)).toFixed(1)),
+    },
+    'Security contacts batch persistence complete',
+  )
+
   log.info({ processed: targets.length }, 'Security contacts batch complete')
   return { processed: targets.length }
 }

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

@@ -1,3 +1,5 @@
+import { type EmailReachabilityReason, classifyEmailReachability } from '@crowd/common'
+
 import { scoreContact } from './score'
 import {
   ContactChannel,
@@ -56,6 +58,15 @@ function isJunkContact(c: RawContact): boolean {
   return false
 }
 
+function classifyReachability(c: RawContact): {
+  reachable: boolean
+  reachabilityReason: EmailReachabilityReason | null
+} {
+  if (c.channel !== 'email') return { reachable: true, reachabilityReason: null }
+  const r = classifyEmailReachability(c.value)
+  return { reachable: r.reachable, reachabilityReason: r.reason }
+}
+
 function normalizeValue(channel: ContactChannel, value: string): string {
   const v = value.trim()
   return channel === 'email' || channel === 'github-handle' ? v.toLowerCase() : v
@@ -130,7 +141,7 @@ export function reconcile(contacts: RawContact[], now: Date = new Date()): Score
 
   const scored: ScoredContact[] = merged.map((c) => {
     const contact = { ...c, provenance: dedupeProvenance(c.provenance) }
-    return { ...contact, ...scoreContact(contact, now) }
+    return { ...contact, ...scoreContact(contact, now), ...classifyReachability(contact) }
   })
 
   scored.sort(

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

@@ -1,3 +1,4 @@
+import type { EmailReachabilityReason } from '@crowd/common'
 import type { SecurityContactConfidence } from '@crowd/data-access-layer/src/osspckgs/api'
 
 export type ContactChannel = 'email' | 'github-pvr' | 'url' | 'github-handle' | 'web-form'
@@ -30,6 +31,8 @@ export interface RawContact {
 export interface ScoredContact extends RawContact {
   score: number
   confidence: SecurityContactConfidence
+  reachable: boolean
+  reachabilityReason: EmailReachabilityReason | null
 }
 
 export interface RepoPolicies {
@@ -75,3 +78,7 @@ export interface ExtractorDeps {
 }
 
 export type Extractor = (target: RepoTarget, deps: ExtractorDeps) => Promise<ExtractorResult>
+
+export type ProcessRepoResult =
+  | { repoId: string; status: 'ok'; contacts: ScoredContact[]; policies: Partial<RepoPolicies> }
+  | { repoId: string; status: 'extractor-failed' }

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

@@ -1,7 +1,50 @@
-import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
+import { QueryExecutor, formatQuery } from '@crowd/data-access-layer/src/queryExecutor'
 import { prepareBulkInsert } from '@crowd/data-access-layer/src/utils'
+import { getServiceChildLogger } from '@crowd/logging'
 
-import { RepoPolicies, ScoredContact } from './types'
+import { ProcessRepoResult, RepoPolicies, ScoredContact } from './types'
+
+const log = getServiceChildLogger('security-contacts')
+
+const CONTACT_COLUMNS = [
+  'repo_id',
+  'channel',
+  'value',
+  'role',
+  'name',
+  'score',
+  'confidence',
+  'provenance',
+  'reachable',
+  'reachability_reason',
+]
+
+const CONTACT_UPSERT_SET = `(repo_id, channel, value) DO UPDATE SET
+     role = EXCLUDED.role,
+     name = EXCLUDED.name,
+     score = EXCLUDED.score,
+     confidence = EXCLUDED.confidence,
+     provenance = EXCLUDED.provenance,
+     reachable = EXCLUDED.reachable,
+     reachability_reason = EXCLUDED.reachability_reason,
+     last_refreshed = NOW(),
+     updated_at = NOW(),
+     deleted_at = NULL`
+
+function toContactRow(repoId: string, c: ScoredContact) {
+  return {
+    repo_id: repoId,
+    channel: c.channel,
+    value: c.value,
+    role: c.role,
+    name: c.name ?? null,
+    score: c.score,
+    confidence: c.confidence,
+    provenance: JSON.stringify(c.provenance),
+    reachable: c.reachable,
+    reachability_reason: c.reachabilityReason,
+  }
+}
 
 // Advances contacts_last_refreshed only, so a failed pass isn't reprocessed this sweep.
 export async function markRepoAttempted(qx: QueryExecutor, repoId: string): Promise<void> {
@@ -30,26 +73,9 @@ export async function writeContacts(
       await tx.result(
         prepareBulkInsert(
           'security_contacts',
-          ['repo_id', 'channel', 'value', 'role', 'name', 'score', 'confidence', 'provenance'],
-          contacts.map((c) => ({
-            repo_id: repoId,
-            channel: c.channel,
-            value: c.value,
-            role: c.role,
-            name: c.name ?? null,
-            score: c.score,
-            confidence: c.confidence,
-            provenance: JSON.stringify(c.provenance),
-          })),
-          `(repo_id, channel, value) DO UPDATE SET
-             role = EXCLUDED.role,
-             name = EXCLUDED.name,
-             score = EXCLUDED.score,
-             confidence = EXCLUDED.confidence,
-             provenance = EXCLUDED.provenance,
-             last_refreshed = NOW(),
-             updated_at = NOW(),
-             deleted_at = NULL`,
+          CONTACT_COLUMNS,
+          contacts.map((c) => toContactRow(repoId, c)),
+          CONTACT_UPSERT_SET,
         ),
       )
     }
@@ -79,3 +105,102 @@ export async function writeContacts(
     )
   })
 }
+
+export async function markReposAttempted(qx: QueryExecutor, repoIds: string[]): Promise<void> {
+  if (repoIds.length === 0) return
+  await qx.result(
+    'UPDATE repos SET contacts_last_refreshed = NOW() WHERE id = ANY($(repoIds)::bigint[])',
+    { repoIds },
+  )
+}
+
+type OkResult = Extract<ProcessRepoResult, { status: 'ok' }>
+
+// Bounds blast radius: a bad row only forces a re-extract of its chunk next sweep, not the whole batch.
+const WRITE_CHUNK_SIZE = 100
+
+function prepareBulkPolicyUpdate(chunk: OkResult[]): string {
+  const rows = chunk.map(
+    (_, i) =>
+      `($(rows.id${i}), $(rows.securityPolicyUrl${i}), $(rows.vulnerabilityReportingUrl${i}), $(rows.pvrResolved${i}), $(rows.bugBountyUrl${i}), $(rows.securityTxtUrl${i}), $(rows.pvrEnabled${i}))`,
+  )
+
+  const params = chunk.reduce(
+    (acc, r, i) => {
+      acc[`id${i}`] = r.repoId
+      acc[`securityPolicyUrl${i}`] = r.policies.securityPolicyUrl ?? null
+      acc[`vulnerabilityReportingUrl${i}`] = r.policies.vulnerabilityReportingUrl ?? null
+      acc[`pvrResolved${i}`] = r.policies.pvrEnabled !== undefined
+      acc[`bugBountyUrl${i}`] = r.policies.bugBountyUrl ?? null
+      acc[`securityTxtUrl${i}`] = r.policies.securityTxtUrl ?? null
+      acc[`pvrEnabled${i}`] = r.policies.pvrEnabled ?? null
+      return acc
+    },
+    {} as Record<string, unknown>,
+  )
+
+  return formatQuery(
+    `UPDATE repos AS r SET
+       security_policy_url         = COALESCE(v.security_policy_url, r.security_policy_url),
+       vulnerability_reporting_url = CASE WHEN v.pvr_resolved
+                                      THEN v.vulnerability_reporting_url
+                                      ELSE COALESCE(v.vulnerability_reporting_url, r.vulnerability_reporting_url)
+                                    END,
+       bug_bounty_url              = COALESCE(v.bug_bounty_url, r.bug_bounty_url),
+       security_txt_url            = COALESCE(v.security_txt_url, r.security_txt_url),
+       pvr_enabled                 = COALESCE(v.pvr_enabled, r.pvr_enabled),
+       contacts_last_refreshed     = NOW()
+     FROM (VALUES ${rows.join(',')}) AS v(id, security_policy_url, vulnerability_reporting_url, pvr_resolved, bug_bounty_url, security_txt_url, pvr_enabled)
+     WHERE r.id = v.id::bigint`,
+    params,
+  )
+}
+
+async function writeContactsChunk(qx: QueryExecutor, chunk: OkResult[]): Promise<void> {
+  if (chunk.length === 0) return
+  const repoIds = chunk.map((r) => r.repoId)
+
+  await qx.tx(async (tx) => {
+    await tx.result(
+      'UPDATE security_contacts SET deleted_at = NOW(), updated_at = NOW() WHERE repo_id = ANY($(repoIds)::bigint[]) AND deleted_at IS NULL',
+      { repoIds },
+    )
+
+    const rows = chunk.flatMap((r) => r.contacts.map((c) => toContactRow(r.repoId, c)))
+    if (rows.length > 0) {
+      await tx.result(
+        prepareBulkInsert('security_contacts', CONTACT_COLUMNS, rows, CONTACT_UPSERT_SET),
+      )
+    }
+
+    await tx.result(prepareBulkPolicyUpdate(chunk))
+  })
+}
+
+export async function writeContactsBatch(
+  qx: QueryExecutor,
+  outcomes: ProcessRepoResult[],
+): Promise<void> {
+  const attemptedOnlyIds = outcomes
+    .filter((o) => o.status === 'extractor-failed')
+    .map((o) => o.repoId)
+  const ok = outcomes.filter((o): o is OkResult => o.status === 'ok')
+
+  for (let i = 0; i < ok.length; i += WRITE_CHUNK_SIZE) {
+    const chunk = ok.slice(i, i + WRITE_CHUNK_SIZE)
+    try {
+      await writeContactsChunk(qx, chunk)
+    } catch (err) {
+      // Isolates blast radius to this chunk: without this, one bad chunk would throw out of
+      // the whole batch, leaving every repo's contacts_last_refreshed untouched and the sweep
+      // stuck re-fetching + re-extracting the same batch forever instead of ever advancing.
+      log.error(
+        { errMsg: (err as Error).message, repoIds: chunk.map((r) => r.repoId) },
+        'Batched contacts write failed for chunk — marking attempted without contacts update',
+      )
+      attemptedOnlyIds.push(...chunk.map((r) => r.repoId))
+    }
+  }
+
+  await markReposAttempted(qx, attemptedOnlyIds)
+}