convex-cron

用途

Convex Cron 用于在 Convex 项目中设置定时任务（cron jobs）和计划函数（scheduled functions）。当需要执行周期性任务、清理作业、数据同步、定时通知或其他后台任务时，可以使用该技能。

主要能力

基础 Cron 设置

在 convex/crons.ts 中导入 cronJobs 与生成的 internal API，创建 crons 实例并导出：

// convex/crons.ts
import { cronJobs } from "convex/server";
import { internal } from "./_generated/api";

const crons = cronJobs();

// Run every hour
crons.interval(
  "cleanup expired sessions",
  { hours: 1 },
  internal.tasks.cleanupExpiredSessions,
  {}
);

// Run every day at midnight UTC
crons.cron(
  "daily report",
  "0 0 * * *",
  internal.reports.generateDailyReport,
  {}
);

export default crons;

基于间隔的调度

使用 crons.interval 按固定时间间隔运行，支持秒、分钟、小时等单位，最小间隔为 30 秒：

// Every 5 minutes
crons.interval("sync data", { minutes: 5 }, internal.sync.fetchData, {});

// Every 2 hours
crons.interval("cleanup temp", { hours: 2 }, internal.files.cleanup, {});

// Every 30 seconds (minimum)
crons.interval("health check", { seconds: 30 }, internal.monitoring.check, {});

Cron 表达式调度

使用 crons.cron 按标准 Cron 表达式运行，表达式字段依次为分钟、小时、日、月、星期（Sunday=0）：

┌───────────── minute (0-59)
│ ┌───────────── hour (0-23)
│ │ ┌───────────── day of month (1-31)
│ │ │ ┌───────────── month (1-12)
│ │ │ │ ┌───────────── day of week (0-6, Sunday=0)
* * * * *

// Every day at 9 AM UTC
crons.cron("morning digest", "0 9 * * *", internal.notifications.morning, {});

// Every Monday at 8 AM UTC
crons.cron("weekly summary", "0 8 * * 1", internal.reports.weekly, {});

// First day of month at midnight
crons.cron("monthly billing", "0 0 1 * *", internal.billing.process, {});

// Every 15 minutes
crons.cron("frequent sync", "*/15 * * * *", internal.sync.run, {});

为 Cron 使用内部函数

Cron 任务应始终使用内部函数（internal functions）：

// convex/tasks.ts
import { internalMutation } from "./_generated/server";
import { v } from "convex/values";

export const cleanupExpiredSessions = internalMutation({
  args: {},
  returns: v.number(),
  handler: async (ctx) => {
    const oneHourAgo = Date.now() - 60 * 60 * 1000;

    const expired = await ctx.db
      .query("sessions")
      .withIndex("by_lastActive")
      .filter((q) => q.lt(q.field("lastActive"), oneHourAgo))
      .collect();

    for (const session of expired) {
      await ctx.db.delete(session._id);
    }

    return expired.length;
  },
});

带参数的 Cron

可以向 Cron 任务传递参数：

crons.interval(
  "cleanup temp files",
  { hours: 1 },
  internal.cleanup.byType,
  { fileType: "temp", maxAge: 3600000 }
);

crons.interval(
  "cleanup cache files",
  { hours: 24 },
  internal.cleanup.byType,
  { fileType: "cache", maxAge: 86400000 }
);

批处理大数据集

为避免超时，处理大数据集时应分批进行，并在未完成时调度下一批：

const BATCH_SIZE = 100;

export const processBatch = internalMutation({
  args: { cursor: v.optional(v.string()) },
  returns: v.null(),
  handler: async (ctx, args) => {
    const result = await ctx.db
      .query("items")
      .withIndex("by_status", (q) => q.eq("status", "pending"))
      .paginate({ numItems: BATCH_SIZE, cursor: args.cursor ?? null });

    for (const item of result.page) {
      await ctx.db.patch(item._id, { status: "processed", processedAt: Date.now() });
    }

    // Schedule next batch if more items
    if (!result.isDone) {
      await ctx.scheduler.runAfter(0, internal.tasks.processBatch, {
        cursor: result.continueCursor,
      });
    }
    return null;
  },
});

外部 API 调用

调用外部 API 时应使用 action：

// convex/sync.ts
"use node";

import { internalAction, internalMutation } from "./_generated/server";
import { internal } from "./_generated/api";
import { v } from "convex/values";

export const syncExternalData = internalAction({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    const response = await fetch("https://api.example.com/data", {
      headers: { Authorization: `Bearer ${process.env.API_KEY}` },
    });

    const data = await response.json();

    await ctx.runMutation(internal.sync.storeData, { data, syncedAt: Date.now() });
    return null;
  },
});

// In crons.ts
crons.interval("sync external", { minutes: 15 }, internal.sync.syncExternalData, {});

日志与监控

可将执行结果写入日志表以便调试：

export const cleanupWithLogging = internalMutation({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    const startTime = Date.now();
    let processedCount = 0;

    try {
      const items = await ctx.db.query("items")
        .filter((q) => q.lt(q.field("expiresAt"), Date.now()))
        .collect();

      for (const item of items) {
        await ctx.db.delete(item._id);
        processedCount++;
      }

      await ctx.db.insert("cronLogs", {
        jobName: "cleanup",
        duration: Date.now() - startTime,
        processedCount,
        status: "success",
      });
    } catch (error) {
      await ctx.db.insert("cronLogs", {
        jobName: "cleanup",
        duration: Date.now() - startTime,
        processedCount,
        status: "failed",
        error: String(error),
      });
      throw error;
    }
    return null;
  },
});

常见陷阱

• 使用公开函数：应始终使用内部函数。
• 忘记时区：所有 Cron 表达式均使用 UTC。
• 长时间运行的 mutation：应拆分为批次。
• 缺少错误处理：应记录失败以便调试。

参考

• Cron Jobs: https://docs.convex.dev/scheduling/cron-jobs
• Scheduling: https://docs.convex.dev/scheduling