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

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

Author: mbani01

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

@@ -16,16 +16,17 @@ import { extractSecurityMd } from './extractors/securityMd'
 import { extractSecurityTxt } from './extractors/securityTxt'
 import { githubApiGet } from './githubToken'
 import { reconcile } from './reconcile'
-import { resolveCdpEmails } from './resolveCdpEmails'
+import { deriveGithubHandlesFromNoreplyEmails, 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[] = []
@@ -163,6 +162,8 @@ export async function processRepo(
     contacts = contacts.filter((c) => c.channel !== 'github-pvr')
   }
 
+  contacts.push(...deriveGithubHandlesFromNoreplyEmails(contacts))
+
   const handleContacts = contacts.filter((c) => c.channel === 'github-handle')
   if (handleContacts.length > 0) {
     try {
@@ -176,7 +177,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,26 +200,57 @@ 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 }
       }
     })
+
+    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',
+    )
+
+    // Heartbeat is kept alive through persistence too: a slow/lock-blocked write can outlast the
+    // 2-minute heartbeat timeout just like extraction can, and letting the timer stop early would
+    // let Temporal retry this activity while the original attempt is still writing.
+    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',
+    )
   } finally {
     clearInterval(heartbeatTimer)
   }

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/resolveCdpEmails.ts

@@ -1,3 +1,4 @@
+import { parseGitHubNoreplyEmail } from '@crowd/common'
 import {
   findMembersByGithubHandles,
   findResolvableEmailsForMembers,
@@ -13,6 +14,28 @@ function latestTimestamp(provenance: ProvenanceEntry[]): string {
     : times.reduce((a, b) => (new Date(b).getTime() > new Date(a).getTime() ? b : a))
 }
 
+// A GitHub noreply email already encodes its owner's handle — derive it so the handle can be
+// resolved to a real address the same way any other github-handle contact is. The noreply email
+// itself is kept as the provenance source, so the resolved contact's origin stays traceable.
+export function deriveGithubHandlesFromNoreplyEmails(contacts: RawContact[]): RawContact[] {
+  const derived: RawContact[] = []
+  for (const c of contacts) {
+    if (c.channel !== 'email') continue
+    const handle = parseGitHubNoreplyEmail(c.value)
+    if (!handle) continue
+    derived.push({
+      channel: 'github-handle',
+      value: handle,
+      role: c.role,
+      tier: c.tier,
+      provenance: [
+        { source: c.value, sourceTier: c.tier, fetchedAt: latestTimestamp(c.provenance) },
+      ],
+    })
+  }
+  return derived
+}
+
 export async function resolveCdpEmails(
   cdpQx: QueryExecutor,
   handleContacts: RawContact[],
@@ -51,8 +74,9 @@ export async function resolveCdpEmails(
         channel: 'email' as const,
         value: email,
         role: contact.role,
+        handle: contact.value,
         tier: contact.tier,
-        provenance: [{ source, sourceTier: contact.tier, fetchedAt }],
+        provenance: [...contact.provenance, { source, sourceTier: contact.tier, fetchedAt }],
       }))
     })
   })

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' }