如何实现 recurring billing 邮件提醒并避免重复发送？

2026-10-24支付老李

结论

可以继续使用每小时运行的 cron job，但不要只依赖“到期时间等于 5 天后”这类查询条件。建议采用下面的处理方式：

按订阅和提醒类型分别创建通知任务。
使用数据库唯一约束，防止重复创建任务。
记录任务状态、尝试次数和实际发送时间。
由 worker 异步发送邮件，失败后进行有限次数的重试。
发送前重新确认订阅仍然有效，并检查到期时间是否发生变化。

幂等键可以设计为：

subscription_id + expiration_at + reminder_type

其中，reminder_type 可以是 expires_in_5_days 或 expires_in_24_hours。

不要只在用户表中保存一个 reminder_sent 布尔值。用户可能续费，取消后重新订阅，也可能同时拥有多个订阅。单一布尔字段无法区分这些情况。

生成提醒任务

不要查询“恰好在 5 天后到期”的订阅，应改用时间区间。cron job 可能因部署、宕机或数据库故障晚执行几分钟，精确匹配时间点很容易漏掉任务。

如果调度器每小时执行一次，可以查询未来一小时内需要触发的提醒：

INSERT INTO subscription_reminders (
  subscription_id,
  user_id,
  expiration_at,
  reminder_type,
  scheduled_at
)
SELECT
  s.id,
  s.user_id,
  s.expiration_at,
  'expires_in_5_days',
  s.expiration_at - INTERVAL '5 days'
FROM subscriptions s
WHERE s.status = 'active'
  AND s.expiration_at - INTERVAL '5 days' >= NOW()
  AND s.expiration_at - INTERVAL '5 days' < NOW() + INTERVAL '1 hour'
ON CONFLICT (subscription_id, expiration_at, reminder_type)
DO NOTHING;

24 小时提醒可以采用相同的方式：

INSERT INTO subscription_reminders (
  subscription_id,
  user_id,
  expiration_at,
  reminder_type,
  scheduled_at
)
SELECT
  s.id,
  s.user_id,
  s.expiration_at,
  'expires_in_24_hours',
  s.expiration_at - INTERVAL '24 hours'
FROM subscriptions s
WHERE s.status = 'active'
  AND s.expiration_at - INTERVAL '24 hours' >= NOW()
  AND s.expiration_at - INTERVAL '24 hours' < NOW() + INTERVAL '1 hour'
ON CONFLICT (subscription_id, expiration_at, reminder_type)
DO NOTHING;

即使 cron job 重复执行，或者多个应用实例同时运行，数据库唯一约束也能阻止重复写入。相比先查询任务是否存在再插入，这种方式不会受到并发竞态的影响。

如果系统需要在停机恢复后补发提醒，可以将查询下界设为允许补偿的时间，例如过去 6 小时：

scheduled_at <= NOW()
AND scheduled_at >= NOW() - INTERVAL '6 hours'

是否补发已经错过的提醒取决于业务规则。例如，“5 天提醒”迟到两小时通常仍有价值。如果“24 小时提醒”执行时订阅已经到期，一般就不应再发送。

Node.js 发送 worker 示例

下面的示例使用 pg 和 SendGrid，主要演示任务认领与状态记录。生产环境还需要加入日志、监控和连接关闭逻辑。

import pg from 'pg';
import sgMail from '@sendgrid/mail';

const { Pool } = pg;

const pool = new Pool({
  connectionString: process.env.DATABASE_URL
});

sgMail.setApiKey(process.env.SENDGRID_API_KEY);

async function claimReminder() {
  const client = await pool.connect();

  try {
    await client.query('BEGIN');

    const result = await client.query(`
      SELECT id
      FROM subscription_reminders
      WHERE status = 'pending'
        AND scheduled_at <= NOW()
        AND attempts < 5
      ORDER BY scheduled_at
      FOR UPDATE SKIP LOCKED
      LIMIT 1
    `);

    if (result.rowCount === 0) {
      await client.query('COMMIT');
      return null;
    }

    const id = result.rows[0].id;

    const claimed = await client.query(
      `
        UPDATE subscription_reminders
        SET status = 'processing',
            attempts = attempts + 1,
            locked_at = NOW(),
            updated_at = NOW()
        WHERE id = $1
        RETURNING *
      `,
      [id]
    );

    await client.query('COMMIT');
    return claimed.rows[0];
  } catch (error) {
    await client.query('ROLLBACK');
    throw error;
  } finally {
    client.release();
  }
}

FOR UPDATE SKIP LOCKED 允许多个 worker 并行处理任务，同时避免它们认领同一条记录。

发送前需要重新读取订阅和用户信息，确认当前任务仍然有效：

async function loadReminderContext(reminder) {
  const result = await pool.query(
    `
      SELECT
        s.id AS subscription_id,
        s.status AS subscription_status,
        s.expiration_at,
        u.email,
        u.name
      FROM subscriptions s
      JOIN users u ON u.id = s.user_id
      WHERE s.id = $1
    `,
    [reminder.subscription_id]
  );

  if (result.rowCount === 0) {
    return null;
  }

  const context = result.rows[0];

  const expirationChanged =
    new Date(context.expiration_at).getTime() !==
    new Date(reminder.expiration_at).getTime();

  if (
    context.subscription_status !== 'active' ||
    expirationChanged ||
    !context.email
  ) {
    return null;
  }

  return context;
}

发送邮件并更新任务状态：

async function processReminder(reminder) {
  const context = await loadReminderContext(reminder);

  if (!context) {
    await pool.query(
      `
        UPDATE subscription_reminders
        SET status = 'failed',
            last_error = 'Reminder is no longer applicable',
            updated_at = NOW()
        WHERE id = $1
      `,
      [reminder.id]
    );
    return;
  }

  try {
    const [response] = await sgMail.send({
      to: context.email,
      from: process.env.BILLING_EMAIL_FROM,
      subject:
        reminder.reminder_type === 'expires_in_5_days'
          ? 'Your subscription expires in 5 days'
          : 'Your subscription expires in 24 hours',
      text: `Hello ${context.name}, your subscription expires at ${context.expiration_at}.`
    });

    await pool.query(
      `
        UPDATE subscription_reminders
        SET status = 'sent',
            sent_at = NOW(),
            provider_message_id = $2,
            last_error = NULL,
            updated_at = NOW()
        WHERE id = $1
      `,
      [reminder.id, response.headers?.['x-message-id'] ?? null]
    );
  } catch (error) {
    const message =
      error instanceof Error ? error.message : String(error);

    await pool.query(
      `
        UPDATE subscription_reminders
        SET status = CASE
              WHEN attempts >= 5 THEN 'failed'
              ELSE 'pending'
            END,
            locked_at = NULL,
            last_error = $2,
            updated_at = NOW()
        WHERE id = $1
      `,
      [reminder.id, message]
    );

    throw error;
  }
}

worker 可以循环认领并处理任务：

async function runWorker() {
  while (true) {
    const reminder = await claimReminder();

    if (!reminder) {
      break;
    }

    try {
      await processReminder(reminder);
    } catch (error) {
      console.error('Failed to send reminder', {
        reminderId: reminder.id,
        error
      });
    }
  }
}

runWorker().catch(console.error);

处理“邮件已发送，但状态未写入”的情况

数据库幂等可以防止任务被重复创建，但无法绝对保证邮件只发送一次。例如：

SendGrid 已接受邮件。
应用在更新 status = 'sent' 之前崩溃。
任务恢复后再次执行。

这是外部系统一致性问题。数据库事务无法同时覆盖本地数据库和 SendGrid。

实际工程中通常只能做到“尽量避免重复”，很难实现严格的端到端 exactly once。可以采取以下措施降低重复发送的风险：

为每个提醒生成稳定的业务幂等键，并将其保存到发送记录中。
保存 SendGrid 返回的 message ID。
对长时间停留在 processing 状态的任务进行人工核查，或谨慎重试。
接收 SendGrid Event Webhook，使用投递事件补充发送状态。
网络请求超时且结果未知时，不要立即无限重试。
如果邮件服务支持请求级幂等，应按照官方接口说明使用。不能假设所有 SendGrid 发送接口都有严格的幂等语义。

可以定期释放已确认未完成的任务：

UPDATE subscription_reminders
SET status = 'pending',
    locked_at = NULL,
    updated_at = NOW()
WHERE status = 'processing'
  AND locked_at < NOW() - INTERVAL '15 minutes'
  AND attempts < 5;

超时时间应根据实际发送耗时和监控数据确定，不要直接将 15 分钟用于所有场景。

调度方式的选择

数据量较小时，cron job 配合数据库任务表已经够用，也比较容易维护。数据量增加后，可以将任务表作为 outbox，由 worker 把任务投递到 BullMQ、RabbitMQ 或 SQS 等队列系统。

无论使用哪种队列，数据库中的业务幂等约束都不能省略。队列通常采用至少一次投递，消息本身也可能重复。使用队列并不等于任务不会重复。

如果订阅由 Stripe、Braintree 等支付平台管理，还需要消费相应的 webhook，及时同步这些变化：

续费成功，到期时间延后；
自动续费失败；
用户取消订阅；
订阅暂停或恢复；
套餐或计费周期变更。

cron job 用于触发定时提醒，webhook 用于保持订阅状态准确，两者可以同时使用。

需要特别注意的细节

所有时间建议使用 UTC 和 TIMESTAMPTZ 存储，只在生成邮件内容时转换为用户所在时区。
查询应使用时间范围，不要依赖日期字符串或时间点精确相等。
为 status 和 scheduled_at 建立索引，避免每小时扫描全部订阅。
发送前重新校验订阅状态，防止用户续费或取消后仍收到旧提醒。
重试应设置最大次数和退避时间。永久失败的任务需要进入告警或死信处理流程。
测试环境必须使用独立收件地址或发送抑制机制，避免向真实用户发送测试邮件。
如果提醒针对“自动续费扣款”而非“订阅到期”，邮件措辞和触发条件应以下一次扣款时间为准，不能直接使用 expiration_at。
用户退订营销邮件与账单服务通知可能适用不同规则，需要根据业务所在地和邮件性质确认合规要求。

对于大多数 Node.js 项目，一个实用的起点是：cron job 创建任务，唯一约束去重，worker 发送邮件，状态表负责审计和重试。这套结构不复杂，也能逐步扩展到队列和多实例部署。