fix: correct chunked write params, casts, heartbeat lifetime, and plus-tag noreply detection

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

Author: mbani01

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

@@ -223,33 +223,36 @@ export async function processBatch(
         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)
   }
 
-  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/writeContacts.ts

@@ -122,7 +122,7 @@ 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}))`,
+      `($(id${i})::bigint, $(securityPolicyUrl${i}), $(vulnerabilityReportingUrl${i}), $(pvrResolved${i})::boolean, $(bugBountyUrl${i}), $(securityTxtUrl${i}), $(pvrEnabled${i})::boolean)`,
   )
 
   const params = chunk.reduce(
@@ -186,6 +186,7 @@ export async function writeContactsBatch(
     .map((o) => o.repoId)
   const ok = outcomes.filter((o): o is OkResult => o.status === 'ok')
 
+  let firstError: Error | undefined
   for (let i = 0; i < ok.length; i += WRITE_CHUNK_SIZE) {
     const chunk = ok.slice(i, i + WRITE_CHUNK_SIZE)
     try {
@@ -194,13 +195,18 @@ export async function writeContactsBatch(
       // 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.
+      // Every remaining chunk is still attempted; the error is re-thrown only after everything
+      // that can be persisted has been, so the activity attempt still surfaces as failed.
       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))
+      firstError = firstError ?? (err as Error)
     }
   }
 
   await markReposAttempted(qx, attemptedOnlyIds)
+
+  if (firstError) throw firstError
 }

services/libs/common/src/email.ts

@@ -148,8 +148,10 @@ export const classifyEmailReachability = (email: string): EmailReachability => {
   if (!isValidEmail(value)) return fail('invalid')
   if (PLACEHOLDER_EMAIL_DOMAINS.has(emailDomain(value))) return fail('placeholder-domain')
 
-  if (NOREPLY_LOCAL_PARTS.has(getEmailLocalPart(value))) return fail('noreply')
-  if (looksLikeBot(getEmailLocalPart(value))) return fail('bot')
+  const localPart = getEmailLocalPart(value)
+  const baseLocalPart = localPart.split('+')[0]
+  if (NOREPLY_LOCAL_PARTS.has(baseLocalPart)) return fail('noreply')
+  if (looksLikeBot(localPart)) return fail('bot')
   if (disposableEmailDomains.has(emailDomain(value))) return fail('disposable')
 
   return { email, reachable: true, reason: null }