At-least-once delivery with deduplication guard

At-least-once delivery guarantees that a job is processed at least once — it does not guarantee it is processed exactly once. If a worker sends a password-reset email, then crashes before acknowledging the job, BullMQ re-queues it and the next worker pickup sends an identical email. For transactional messages, duplicates erode user trust immediately. CWE-362 covers race conditions on shared resources; duplicate sends are precisely that class of failure. ISO-25010:2011 reliability.fault-tolerance requires the system to behave correctly despite component failures.

Add a pre-flight dedup check at the start of the job processor in workers/email.worker.ts, before calling the ESP:

async function processEmailJob(job: Job<EmailJobData>) {
  const { campaignId, recipientId } = job.data

  const existing = await db.emailSendLog.findFirst({
    where: { campaignId, recipientId, status: 'sent' }
  })
  if (existing) {
    logger.info({ campaignId, recipientId }, 'Duplicate job — skipping')
    return
  }

  await db.emailSendLog.create({
    data: { campaignId, recipientId, status: 'sending', jobId: job.id }
  })

  const result = await esp.send(job.data)

  await db.emailSendLog.update({
    where: { campaignId_recipientId: { campaignId, recipientId } },
    data: { status: 'sent', espMessageId: result.messageId, sentAt: new Date() }
  })
}

• ID: sending-pipeline-infrastructure.queue-architecture.at-least-once-dedup

• Severity: high

• What to look for: Enumerate all deduplication mechanisms in the job processing logic: (1) sent_at timestamp check before sending, (2) idempotency key stored in a persistent database, (3) dedup ID passed to the ESP API. Count how many of these mechanisms are implemented. In an at-least-once delivery system, a job can be processed more than once if the worker crashes mid-execution before acknowledging.

• Pass criteria: Before sending an email, the worker checks a persistent store to verify the message has not already been sent (by campaign ID + recipient ID, or a stored idempotency key). At least 1 deduplication mechanism must be present. Report even on pass: describe which dedup mechanism is used and where it is implemented.

• Fail criteria: No deduplication guard exists. If a worker crashes after sending but before acknowledging the job, the next worker pickup sends the same email again. Transactional emails (password resets, order confirmations) would be sent twice.

• Skip (N/A) when: The queue explicitly provides exactly-once delivery guarantees AND that guarantee is verified in the configuration — note the mechanism in the detail field.

• Cross-reference: The retry-error-handling category's idempotent-retries check verifies idempotency at the retry level — this check verifies the initial dedup guard at the queue level.

• Detail on fail: "Worker marks job complete only after ESP responds, but no dedup check before sending — worker crash causes duplicate send" or "No idempotency key passed to SendGrid — same job retried after network timeout sends email twice"

• Remediation: Add a deduplication check using a database record before each send:

  // workers/email.worker.ts
  async function processEmailJob(job: Job<EmailJobData>) {
    const { campaignId, recipientId, templateId, mergeFields } = job.data
  
    // Dedup check — was this already sent?
    const existing = await db.emailSendLog.findFirst({
      where: { campaignId, recipientId, status: 'sent' }
    })
  
    if (existing) {
      logger.info({ campaignId, recipientId }, 'Duplicate job detected — skipping send')
      return // Acknowledge job without resending
    }
  
    // Mark as in-flight before calling ESP
    await db.emailSendLog.create({
      data: { campaignId, recipientId, templateId, status: 'sending', jobId: job.id }
    })
  
    const result = await esp.send({ to, templateId, mergeFields })
  
    await db.emailSendLog.update({
      where: { campaignId_recipientId: { campaignId, recipientId } },
      data: { status: 'sent', espMessageId: result.messageId, sentAt: new Date() }
    })
  }