字段	类型	说明	示例
`creditsPerGrant`	INTEGER	每次发放的积分数	1000 (Starter Yearly)
`intervalMonths`	INTEGER	发放间隔 (月)	1 (每月)
`grantsRemaining`	INTEGER	剩余发放次数	11 (首次已发放,还剩 11 次)
`totalCreditsRemaining`	INTEGER	剩余总积分	11000 (12000 - 1000)
`nextGrantAt`	TIMESTAMP	下次发放时间	2025-11-15 00:00:00

2. 初始化流程

用户购买年付订阅时 (app/api/payments/creem/webhook/route.ts):

// 1. 获取计划配置
const plan = subscriptionPlans["starter_yearly"];
// {
//   creditsPerCycle: 12000,
//   grantSchedule: {
//     mode: "installments",
//     grantsPerCycle: 12,      // 分 12 次
//     intervalMonths: 1,       // 每月一次
//     creditsPerGrant: 1000,   // 每次 1000
//     initialGrants: 1,        // 首次发放 1 次
//   }
// }

// 2. 立即发放首次积分 (1000)
await refundCredits(userId, 1000, "subscription_cycle", paymentId);

// 3. 创建调度记录
await db.insert(subscriptionCreditSchedule).values({
  id: randomUUID(),
  subscriptionId,
  userId,
  planKey: "starter_yearly",
  creditsPerGrant: 1000,
  intervalMonths: 1,
  grantsRemaining: 11,           // 12 - 1 (首次) = 11
  totalCreditsRemaining: 11000,  // 12000 - 1000 = 11000
  nextGrantAt: addMonths(new Date(), 1), // 一个月后
});

3. processDueSchedules 函数

文件: lib/billing/subscription.ts:140-241

这是定时任务的核心逻辑,负责处理所有到期的调度记录。

函数签名:

async function processDueSchedules(
  limit?: number,         // 每次处理的最大调度数 (默认 50)
  catchUpPerSchedule?: number  // 每个调度最多追补的次数 (默认 12)
): Promise<Array<{
  scheduleId: string;
  subscriptionId: string;
  userId: string;
  totalGranted: number;      // 本次发放的总积分
  grantsProcessed: number;   // 本次处理的发放次数
  remainingGrants: number;   // 剩余发放次数
}>>

执行流程:

export async function processDueSchedules(limit = 50, catchUpPerSchedule = 12) {
  const now = new Date();

  return db.transaction(async tx => {
    // 1. 查询到期的调度记录
    const schedules = await tx
      .select()
      .from(subscriptionCreditSchedule)
      .where(
        and(
          lte(subscriptionCreditSchedule.nextGrantAt, now),  // 发放时间 <= 当前时间
          gt(subscriptionCreditSchedule.grantsRemaining, 0), // 剩余次数 > 0
        ),
      )
      .orderBy(subscriptionCreditSchedule.nextGrantAt)
      .limit(limit)
      .for("update", { skipLocked: true }); // 行锁 + 跳过已锁定

    const results = [];

    // 2. 逐个处理调度
    for (const schedule of schedules) {
      let grantsRemaining = schedule.grantsRemaining;
      let creditsRemaining = schedule.totalCreditsRemaining;
      let nextGrantAt = schedule.nextGrantAt;
      let processed = 0;
      let grantedSum = 0;

      // 3. 追补逾期的发放 (最多 catchUpPerSchedule 次)
      while (
        grantsRemaining > 0 &&
        creditsRemaining > 0 &&
        nextGrantAt <= now &&
        processed < catchUpPerSchedule
      ) {
        // 计算本次发放量
        const grantAmount = grantsRemaining > 1
          ? Math.min(schedule.creditsPerGrant, creditsRemaining)
          : creditsRemaining; // 最后一次发放所有剩余积分

        // 4. 发放积分
        await tx
          .update(user)
          .set({ credits: sql`${user.credits} + ${grantAmount}` })
          .where(eq(user.id, schedule.userId));

        // 5. 记录账本
        await tx.insert(creditLedger).values({
          id: randomUUID(),
          userId: schedule.userId,
          delta: grantAmount,
          reason: "subscription_schedule", // 固定原因标识
        });

        // 6. 更新计数器
        grantsRemaining -= 1;
        creditsRemaining -= grantAmount;
        grantedSum += grantAmount;
        processed += 1;
        nextGrantAt = addMonths(nextGrantAt, schedule.intervalMonths);
      }

      // 7. 更新或删除调度记录
      if (grantsRemaining <= 0 || creditsRemaining <= 0) {
        // 已完成所有发放,删除调度
        await tx
          .delete(subscriptionCreditSchedule)
          .where(eq(subscriptionCreditSchedule.id, schedule.id));
      } else {
        // 更新下次发放时间
        await tx
          .update(subscriptionCreditSchedule)
          .set({
            grantsRemaining,
            totalCreditsRemaining: creditsRemaining,
            nextGrantAt,
            updatedAt: new Date(),
          })
          .where(eq(subscriptionCreditSchedule.id, schedule.id));
      }

      results.push({
        scheduleId: schedule.id,
        subscriptionId: schedule.subscriptionId,
        userId: schedule.userId,
        totalGranted: grantedSum,
        grantsProcessed: processed,
        remainingGrants: grantsRemaining,
      });
    }

    return results;
  });
}

关键特性:

✅ 事务性: 所有操作在单个事务中完成
✅ 行锁机制: for("update", { skipLocked: true }) 防止并发冲突
✅ 追补逾期: 自动补发错过的发放 (最多 12 次)
✅ 精确计算: 最后一次发放时自动处理剩余积分

4. 追补机制 (catchUp)

如果定时任务中断 (服务器宕机、部署等),重启后会自动追补错过的发放。

示例:

用户购买时间: 2025-01-15
首次发放: 2025-01-15 (1000 积分)
计划发放时间:
  - 2025-02-15
  - 2025-03-15
  - 2025-04-15
  - ...

假设服务器在 2025-02-01 到 2025-04-20 期间宕机,
重启后 (2025-04-20) 定时任务会:
  1. 发现 nextGrantAt = 2025-02-15 <= now
  2. 自动发放 2025-02-15 的 1000 积分
  3. 更新 nextGrantAt = 2025-03-15
  4. 发现 2025-03-15 <= now
  5. 自动发放 2025-03-15 的 1000 积分
  6. 更新 nextGrantAt = 2025-04-15
  7. 发现 2025-04-15 <= now
  8. 自动发放 2025-04-15 的 1000 积分
  9. 更新 nextGrantAt = 2025-05-15 (未来)
  10. 停止追补

限制参数:

catchUpPerSchedule (默认 12): 每个调度最多追补 12 次,防止单次运行时间过长

API 端点详解

端点信息

路径: /api/cron/subscription-grants
方法: GET 或 POST
认证: Bearer Token 或 Basic Auth

认证配置

定时任务支持两种认证方式:

方式 1: Bearer Token (推荐)

环境变量:

CRON_SECRET="your-random-secret-key-at-least-32-chars"

请求示例:

curl -X GET https://your-domain.com/api/cron/subscription-grants \
  -H "Authorization: Bearer your-random-secret-key-at-least-32-chars"

方式 2: Basic Auth

环境变量:

CRON_JOBS_USERNAME="cron_admin"
CRON_JOBS_PASSWORD="your-secure-password"

请求示例:

curl -X GET https://your-domain.com/api/cron/subscription-grants \
  -u cron_admin:your-secure-password

或使用 Base64 编码:

# 编码: cron_admin:your-secure-password -> Y3Jvbl9hZG1pbjp5b3VyLXNlY3VyZS1wYXNzd29yZA==
curl -X GET https://your-domain.com/api/cron/subscription-grants \
  -H "Authorization: Basic Y3Jvbl9hZG1pbjp5b3VyLXNlY3VyZS1wYXNzd29yZA=="

核心实现

文件: app/api/cron/subscription-grants/route.ts

const CRON_SECRET = process.env.CRON_SECRET;
const CRON_JOBS_USERNAME = process.env.CRON_JOBS_USERNAME;
const CRON_JOBS_PASSWORD = process.env.CRON_JOBS_PASSWORD;
const DEFAULT_LIMIT = 50;
const MAX_LIMIT = 500;

export async function GET(req: NextRequest) {
  const authHeader = req.headers.get("authorization") ?? "";

  // 1. 验证 Basic Auth
  const isBasicAuthorized = (() => {
    if (!CRON_JOBS_USERNAME || !CRON_JOBS_PASSWORD) return false;
    if (!authHeader.startsWith("Basic ")) return false;

    try {
      const decoded = Buffer.from(authHeader.slice(6), "base64").toString("utf8");
      const [username, password] = decoded.split(":");
      return username === CRON_JOBS_USERNAME && password === CRON_JOBS_PASSWORD;
    } catch (error) {
      return false;
    }
  })();

  // 2. 验证 Bearer Token
  const isBearerAuthorized = CRON_SECRET && authHeader === `Bearer ${CRON_SECRET}`;

  // 3. 任意一种认证通过即可
  if (!isBasicAuthorized && !isBearerAuthorized) {
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

  // 4. 解析查询参数
  const limitParam = req.nextUrl.searchParams.get("limit");
  const parsedLimit = limitParam ? Number.parseInt(limitParam, 10) : DEFAULT_LIMIT;
  const limit = Number.isFinite(parsedLimit)
    ? Math.min(Math.max(parsedLimit, 1), MAX_LIMIT)
    : DEFAULT_LIMIT;

  const catchUpParam = req.nextUrl.searchParams.get("catchUp");
  const parsedCatchUp = catchUpParam ? Number.parseInt(catchUpParam, 10) : undefined;
  const catchUp = parsedCatchUp && Number.isFinite(parsedCatchUp)
    ? Math.min(Math.max(parsedCatchUp, 1), 36)
    : undefined;

  // 5. 执行积分发放
  const results = await processDueSchedules(limit, catchUp);

  // 6. 返回结果
  return NextResponse.json({
    processed: results.reduce((sum, item) => sum + item.grantsProcessed, 0),
    schedulesTouched: results.length,
    grants: results,
  });
}

export async function POST(req: NextRequest) {
  return GET(req); // POST 和 GET 行为一致
}

查询参数

参数	类型	默认值	说明
`limit`	number	50	每次处理的最大调度数 (1-500)
`catchUp`	number	12	每个调度最多追补的次数 (1-36)

示例:

# 默认参数 (limit=50, catchUp=12)
curl -X GET https://your-domain.com/api/cron/subscription-grants \
  -H "Authorization: Bearer YOUR_SECRET"

# 自定义参数 (limit=100, catchUp=24)
curl -X GET "https://your-domain.com/api/cron/subscription-grants?limit=100&catchUp=24" \
  -H "Authorization: Bearer YOUR_SECRET"

响应格式

成功响应 (200 OK):

{
  "processed": 3,           // 总共处理的发放次数
  "schedulesTouched": 2,    // 总共涉及的调度数
  "grants": [
    {
      "scheduleId": "sched_abc123",
      "subscriptionId": "sub_xyz789",
      "userId": "user_456",
      "totalGranted": 2000,    // 本次发放的总积分
      "grantsProcessed": 2,    // 本次处理的发放次数 (追补了 2 次)
      "remainingGrants": 9     // 剩余发放次数
    },
    {
      "scheduleId": "sched_def456",
      "subscriptionId": "sub_uvw012",
      "userId": "user_789",
      "totalGranted": 1000,
      "grantsProcessed": 1,
      "remainingGrants": 10
    }
  ]
}

错误响应:

// 401 Unauthorized - 认证失败
{
  "error": "Unauthorized"
}

// 500 Internal Server Error - 未配置认证
{
  "error": "Cron auth not configured"
}

执行频率建议

部署配置

方案 1: Vercel Cron Jobs (推荐)

如果你的应用部署在 Vercel,可以使用内置的 Cron Jobs 功能。

配置文件

创建 vercel.json:

{
  "crons": [
    {
      "path": "/api/cron/subscription-grants",
      "schedule": "0 * * * *"
    }
  ]
}

Schedule 语法 (标准 Cron 格式):

 ┌───────────── 分钟 (0 - 59)
 │ ┌───────────── 小时 (0 - 23)
 │ │ ┌───────────── 日期 (1 - 31)
 │ │ │ ┌───────────── 月份 (1 - 12)
 │ │ │ │ ┌───────────── 星期 (0 - 7, 0 和 7 都代表星期日)
 │ │ │ │ │
 │ │ │ │ │
 * * * * *

常用示例:

{
  "crons": [
    {
      "path": "/api/cron/subscription-grants",
      "schedule": "0 * * * *"    // 每小时整点 (推荐)
    },
    {
      "path": "/api/cron/subscription-grants",
      "schedule": "*/15 * * * *" // 每 15 分钟
    },
    {
      "path": "/api/cron/subscription-grants",
      "schedule": "0 0 * * *"    // 每天午夜 (不推荐,延迟太高)
    }
  ]
}

环境变量配置

在 Vercel Dashboard 中设置:

Settings → Environment Variables

CRON_SECRET=your-random-secret-key-at-least-32-chars

重要: Vercel Cron Jobs 会自动在请求头中添加 Authorization: Bearer CRON_SECRET,因此必须设置 CRON_SECRET 环境变量。

部署

# 提交 vercel.json
git add vercel.json
git commit -m "Add Vercel Cron Jobs configuration"
git push

# Vercel 会自动检测并启用 Cron Jobs

监控

在 Vercel Dashboard 中查看 Cron 执行日志:

Deployments → [Your Deployment] → Functions → /api/cron/subscription-grants

方案 2: 外部 Cron 服务

如果不使用 Vercel,可以使用第三方 Cron 服务。

服务	免费额度	特点
cron-job.org	无限任务	简单易用,支持邮件通知
EasyCron	每月 1000 次	支持多种通知方式
Uptime Robot	50 个监控	主要用于健康检查,附带 Cron 功能

配置示例 (cron-job.org)

注册账号: https://console.cron-job.org/signup
创建任务:
- URL: https://your-domain.com/api/cron/subscription-grants
- Schedule: 0 * * * * (每小时)
- HTTP Method: GET
- Headers:
```
Authorization: Bearer your-random-secret-key-at-least-32-chars
```
启用通知:
- Email: 当任务失败时发送邮件

使用 GitHub Actions

创建 .github/workflows/cron.yml:

name: Subscription Credits Cron

on:
  schedule:
    # 每小时整点执行
    - cron: '0 * * * *'
  workflow_dispatch: # 支持手动触发

jobs:
  grant-credits:
    runs-on: ubuntu-latest
    steps:
      - name: Trigger subscription grants
        run: |
          curl -X GET "${{ secrets.APP_URL }}/api/cron/subscription-grants" \
            -H "Authorization: Bearer ${{ secrets.CRON_SECRET }}" \
            -f # 失败时返回非零退出码

      - name: Notify on failure
        if: failure()
        run: |
          echo "Cron job failed! Check logs."
          # 可添加邮件通知或 Slack 通知

环境变量 (GitHub Secrets):

Settings → Secrets and variables → Actions

APP_URL=https://your-domain.com
CRON_SECRET=your-random-secret-key-at-least-32-chars

方案 3: 自托管 Cron (服务器)

如果你有自己的 Linux 服务器,可以使用系统 Cron。

配置

# 编辑 crontab
crontab -e

# 添加任务 (每小时执行)
0 * * * * curl -X GET https://your-domain.com/api/cron/subscription-grants \
  -H "Authorization: Bearer your-random-secret-key-at-least-32-chars" \
  >> /var/log/sistine-cron.log 2>&1

日志管理

# 查看日志
tail -f /var/log/sistine-cron.log

# 日志轮转 (防止日志文件过大)
sudo nano /etc/logrotate.d/sistine-cron

# 内容:
/var/log/sistine-cron.log {
  daily
  rotate 7
  compress
  missingok
  notifempty
}

方案 4: AWS EventBridge

如果你使用 AWS,可以使用 EventBridge (原 CloudWatch Events)。

配置步骤

创建规则:
- 打开 AWS EventBridge Console
- 创建规则: sistine-subscription-grants
- 调度表达式: cron(0 * * * ? *) (每小时)
添加目标:
- 目标类型: API 目的地
- URL: https://your-domain.com/api/cron/subscription-grants
- HTTP Method: GET
- Headers:
```
{
  "Authorization": "Bearer your-random-secret-key-at-least-32-chars"
}
```
配置重试:
- 最大重试次数: 3
- 重试间隔: 60 秒

监控和日志

1. 应用日志

在定时任务端点中添加详细日志:

export async function GET(req: NextRequest) {
  const startTime = Date.now();

  console.log("[Cron] Subscription grants started");

  try {
    const results = await processDueSchedules(limit, catchUp);

    const duration = Date.now() - startTime;
    console.log(`[Cron] Completed in ${duration}ms`, {
      processed: results.reduce((sum, item) => sum + item.grantsProcessed, 0),
      schedulesTouched: results.length,
    });

    return NextResponse.json({ ... });
  } catch (error) {
    console.error("[Cron] Failed:", error);
    throw error;
  }
}

2. 数据库监控

定期检查调度表的状态:

-- 查看所有待发放的调度
SELECT
  s.id,
  s.user_id,
  u.email,
  s.plan_key,
  s.grants_remaining,
  s.total_credits_remaining,
  s.next_grant_at,
  s.next_grant_at - NOW() AS time_until_next_grant
FROM subscription_credit_schedule s
JOIN "user" u ON s.user_id = u.id
ORDER BY s.next_grant_at;

-- 统计逾期的调度
SELECT COUNT(*) AS overdue_count
FROM subscription_credit_schedule
WHERE next_grant_at <= NOW()
  AND grants_remaining > 0;

3. 告警机制

邮件告警 (失败时)

import { sendEmail } from "@/lib/email";

export async function GET(req: NextRequest) {
  try {
    const results = await processDueSchedules(limit, catchUp);

    // 检查是否有失败的调度
    if (results.length === 0 && hasOverdueSchedules()) {
      await sendEmail({
        to: "admin@example.com",
        subject: "⚠️  Cron Job Alert: No schedules processed",
        text: "定时任务未处理任何调度,请检查数据库或日志",
      });
    }

    return NextResponse.json({ ... });
  } catch (error) {
    // 发送错误告警
    await sendEmail({
      to: "admin@example.com",
      subject: "🚨 Cron Job Failed",
      text: `定时任务执行失败:\n${error.message}\n\n请立即检查!`,
    });

    throw error;
  }
}

Slack 告警

async function sendSlackAlert(message: string) {
  await fetch(process.env.SLACK_WEBHOOK_URL, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      text: message,
      channel: "#alerts",
      username: "Cron Monitor",
      icon_emoji: ":robot_face:",
    }),
  });
}

4. 健康检查端点

创建 /api/cron/health 端点:

// app/api/cron/health/route.ts
import { db } from "@/lib/db";
import { subscriptionCreditSchedule } from "@/lib/db/schema";
import { sql } from "drizzle-orm";

export async function GET() {
  // 检查逾期调度
  const overdueSchedules = await db
    .select({ count: sql`count(*)` })
    .from(subscriptionCreditSchedule)
    .where(sql`${subscriptionCreditSchedule.nextGrantAt} <= NOW()`);

  const overdueCount = overdueSchedules[0].count;

  return NextResponse.json({
    status: overdueCount > 100 ? "warning" : "healthy",
    overdueSchedules: overdueCount,
    timestamp: new Date().toISOString(),
  });
}

使用 Uptime Robot 等服务监控此端点:

URL: https://your-domain.com/api/cron/health
检查间隔: 5 分钟
告警条件: status !== "healthy"

故障排查

问题 1: 定时任务未执行

症状:

用户未收到积分
subscriptionCreditSchedule 表中有逾期记录

原因排查:

检查 Cron 配置:

# Vercel
vercel env ls # 检查 CRON_SECRET 是否设置

# cron-job.org
# 登录控制台检查任务状态和执行历史

检查认证:

# 手动测试端点
curl -X GET https://your-domain.com/api/cron/subscription-grants \
  -H "Authorization: Bearer YOUR_SECRET" \
  -v

# 预期: 200 OK + JSON 响应
# 错误: 401 Unauthorized → 检查密钥是否正确

检查数据库连接:

// 添加日志
console.log("DATABASE_URL:", process.env.DATABASE_URL ? "✓ Set" : "✗ Missing");

问题 2: 部分用户未收到积分

症状:

部分调度正常,部分调度失败
日志显示部分用户 ID 不存在

原因排查:

-- 检查孤立的调度记录 (用户已删除)
SELECT s.*
FROM subscription_credit_schedule s
LEFT JOIN "user" u ON s.user_id = u.id
WHERE u.id IS NULL;

-- 清理孤立记录
DELETE FROM subscription_credit_schedule
WHERE user_id NOT IN (SELECT id FROM "user");

问题 3: 调度记录未删除

症状:

grantsRemaining = 0 但记录仍存在
数据库中有过期的调度

解决:

-- 手动清理已完成的调度
DELETE FROM subscription_credit_schedule
WHERE grants_remaining <= 0
   OR total_credits_remaining <= 0;

预防:

确保 processDueSchedules 中的删除逻辑正确:

if (grantsRemaining <= 0 || creditsRemaining <= 0) {
  await tx
    .delete(subscriptionCreditSchedule)
    .where(eq(subscriptionCreditSchedule.id, schedule.id));
}

问题 4: 并发冲突

症状:

同一调度被处理两次
积分发放重复

原因:

多个 Cron 任务同时执行
未启用行锁

解决:

确保代码中使用了行锁:

const schedules = await tx
  .select()
  .from(subscriptionCreditSchedule)
  .where(...)
  .for("update", { skipLocked: true }); // ✅ 必须启用

部署建议:

只配置一个 Cron 服务
避免手动触发与自动任务重叠

问题 5: 追补次数不足

症状:

服务器宕机后,用户积分未完全补发
日志显示 grantsProcessed < 逾期次数

原因:

catchUpPerSchedule 默认值 12 不足以覆盖长时间宕机

解决:

# 临时增加追补次数
curl -X GET "https://your-domain.com/api/cron/subscription-grants?catchUp=36" \
  -H "Authorization: Bearer YOUR_SECRET"

或修改默认值:

// app/api/cron/subscription-grants/route.ts
const catchUp = parsedCatchUp ?? 36; // 从 12 改为 36

最佳实践

1. 定期备份调度表

# PostgreSQL 备份
pg_dump -h localhost -U postgres -d sistine_db \
  -t subscription_credit_schedule \
  > schedule_backup_$(date +%Y%m%d).sql

2. 监控执行时间

如果执行时间过长,考虑:

增加 limit 参数
优化数据库索引
分批处理调度

-- 为 nextGrantAt 添加索引
CREATE INDEX idx_subscription_credit_schedule_next_grant
ON subscription_credit_schedule(next_grant_at)
WHERE grants_remaining > 0;

3. 模拟测试

在生产环境部署前,先在本地测试:

# 启动本地服务
pnpm dev

# 手动触发
curl -X GET http://localhost:3000/api/cron/subscription-grants \
  -H "Authorization: Bearer YOUR_SECRET"

4. 日志保留策略

避免日志文件过大:

// 只保留关键信息
console.log(`[Cron] Processed ${results.length} schedules`);

// 详细日志仅在开发环境
if (process.env.NODE_ENV === "development") {
  console.log("[Cron] Details:", results);
}

5. 异常重试机制

export async function GET(req: NextRequest) {
  const maxRetries = 3;
  let attempt = 0;

  while (attempt < maxRetries) {
    try {
      const results = await processDueSchedules(limit, catchUp);
      return NextResponse.json({ ... });
    } catch (error) {
      attempt++;
      console.error(`[Cron] Attempt ${attempt} failed:`, error);

      if (attempt >= maxRetries) {
        throw error;
      }

      // 等待后重试
      await new Promise(resolve => setTimeout(resolve, 5000));
    }
  }
}

总结

定时任务是年付订阅分期发放积分的核心机制:

工作原理: 通过 subscriptionCreditSchedule 表管理发放计划
核心逻辑: processDueSchedules 函数自动处理到期调度
认证方式: Bearer Token 或 Basic Auth
推荐频率: 每小时执行一次
部署方案: Vercel Cron Jobs (最简单) 或第三方服务

关键特性:

✅ 事务性操作,确保数据一致性
✅ 行锁机制,防止并发冲突
✅ 自动追补,处理宕机等异常情况
✅ 灵活配置,支持自定义 limit 和 catchUp 参数

遵循本文档的最佳实践,你的定时任务将稳定可靠地为用户发放积分。

定时任务

On this page