[事后分析] 队列重试导致生产环境 1,284 封活动邮件重复发送

// Delivery ledger schema (D1)
await db.createTable('delivery_ledger', {
  campaignId: text().notNull(),
  recipientId: text().notNull(),
  idempotencyKey: text().notNull(),
  status: text().notNull(), // 'reserved' | 'sent' | 'failed'
  resendMessageId: text(),
  createdAt: integer().notNull(),
  updatedAt: integer().notNull(),
});

// 复合唯一索引：每个活动每名收件人仅一次发送
await db.createIndex('delivery_ledger_unique').on(['campaignId', 'recipientId']).unique();

// 发送前预留
async function reserveAndSend(params: {
  campaignId: string;
  recipientId: string;
  email: string;
  subject: string;
  html: string;
}): Promise<SendResult> {
  const idempotencyKey = `${params.campaignId}:${params.recipientId}`;
  
  // 尝试预留
  try {
    await db.insert(deliveryLedger).values({
      campaignId: params.campaignId,
      recipientId: params.recipientId,
      idempotencyKey,
      status: 'reserved',
      createdAt: Date.now(),
    });
  } catch (err) {
    if (err.code === 'SQLITE_CONSTRAINT_UNIQUE') {
      // 已预留 — 跳过
      return { skipped: true, reason: 'already_reserved' };
    }
    throw err;
  }
  
  // 带幂等性密钥通过 Resend 发送
  const resendResponse = await resend.emails.send({
    from: 'TanStack Ship <noreply@tanstackship.com>',
    to: params.email,
    subject: params.subject,
    html: params.html,
    idempotencyKey, // Resend 用此密钥去重
  });
  
  // 更新账本
  await db
    .update(deliveryLedger)
    .set({ 
      status: 'sent',
      resendMessageId: resendResponse.data?.id,
      updatedAt: Date.now(),
    })
    .where(eq(deliveryLedger.idempotencyKey, idempotencyKey));
  
  return { sent: true, messageId: resendResponse.data?.id };
}

// 带 DLQ 处理的队列消费者
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    const results: SendResult[] = [];
    
    for (const message of batch.messages) {
      const params = JSON.parse(message.body) as CampaignSendParams;
      
      try {
        const result = await reserveAndSend(params);
        results.push(result);
        message.ack(); // 确认成功处理
      } catch (err) {
        // 最大重试次数后，消息进入 DLQ
        console.error('重试后发送失败：', err, params);
        results.push({ failed: true, error: String(err) });
        // 不确认 — 由 Queue 处理 DLQ 路由
      }
    }
    
    // 记录批次结果以供监控
    await logBatchResults(batch.queue, results);
  },
};

# 暂停活动（将状态设为 'paused'）
curl -X POST /api/campaigns/{id}/pause

# 检查投递状态
curl /api/campaigns/{id}/delivery-status | jq '.stats'
# {
#   "total": 12480,
#   "sent": 8520,
#   "reserved": 120,
#   "failed": 40,
#   "remaining": 3800
# }

# 从上一个检查点恢复（跳过已发送的）
curl -X POST /api/campaigns/{id}/resume
# 只有剩余的 3,800 名收件人会收到邮件

// 通用模式（非幂等）
await resend.emails.send({
  to: recipient.email,
  subject: campaign.subject,
  html: campaign.body,
});

// Resend 支持幂等性密钥
await resend.emails.send({
  to: recipient.email,
  subject: campaign.subject,
  headers: {
    'X-Resend-Idempotency-Key': `${campaignId}:${recipientId}`
  }
})

// 追踪重复检测
const duplicateKey = `${campaignId}:${recipientId}`
const exists = await deliveryLedger.findUnique({ where: { duplicateKey } })

if (exists) {
  metrics.increment('email.duplicate_detected')
  logger.info('已阻止重复邮件', { campaignId, recipientId })
}

// 设置告警
if (metrics.get('email.duplicate_detected') > 10) {
  alert('检测到重复率异常', { count: metrics.get('email.duplicate_detected') })
}

// 数据库 schema
import { sqliteTable, text, integer } from 'drizzle-orm/sqlite-core'

export const deliveryLedger = sqliteTable('delivery_ledger', {
  id: integer('id').primaryKey(),
  campaignId: text('campaign_id').notNull(),
  recipientId: text('recipient_id').notNull(),
  emailId: text('email_id'), // Resend 邮件 ID
  status: text('status').notNull(), // pending, sent, failed
  sentAt: integer('sent_at'),
  createdAt: integer('created_at').notNull()
})

// 唯一约束防止重复
export const deliveryLedgerKey = {
  campaignId: text('campaign_id').notNull(),
  recipientId: text('recipient_id').notNull()
}

// 调用邮件提供商前
const duplicateKey = `${campaignId}:${recipientId}`
const existing = await db.query.deliveryLedger.findFirst({
  where: eq(deliveryLedger.campaignId, campaignId),
  where: eq(deliveryLedger.recipientId, recipientId)
})

if (existing) {
  logger.info('已阻止重复', { campaignId, recipientId })
  return // 跳过发送
}

// 死信队列处理器
export const dlqHandler = async (message: QueueMessage) => {
  const { campaignId, recipientId, error } = message.body

  await db.insert(dlqLogs).values({
    campaignId,
    recipientId,
    errorMessage: error.message,
    stack: error.stack,
    timestamp: Date.now()
  })

  // DLQ 超过 100 条时告警
  const count = await db.query.dlqLogs.count()
  if (count > 100) {
    await notifyTeam('DLQ 阈值超出', { count })
  }
}