convex-cron-jobs

Convex Cron Jobs

在 Convex 應用程式中，排程週期性函式以執行背景任務、清理工作、資料同步，以及自動化工作流程。

文件來源

在實作之前，請勿假設；請取得最新文件：

• 主要：https://docs.convex.dev/scheduling/cron-jobs
• 排程概覽：https://docs.convex.dev/scheduling
• 排程函式：https://docs.convex.dev/scheduling/scheduled-functions
• 更廣泛的上下文：https://docs.convex.dev/llms.txt

說明

Cron 任務概覽

Convex cron 任務允許您排程函式在固定間隔或特定時間執行。主要功能：

• 在固定排程上執行函式
• 支援基於間隔和 cron 表達式的排程
• 失敗時自動重試
• 透過 Convex 儀表板進行監控

基本 Cron 設定

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

const crons = cronJobs();

// 每小時執行
crons.interval(
  "cleanup expired sessions",
  { hours: 1 },
  internal.tasks.cleanupExpiredSessions,
  {}
);

// 每天 UTC 午夜執行
crons.cron(
  "daily report",
  "0 0 * * *",
  internal.reports.generateDailyReport,
  {}
);

export default crons;

基於間隔的排程

使用 crons.interval 進行簡單的週期性任務：

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

const crons = cronJobs();

// 每 5 分鐘
crons.interval(
  "sync external data",
  { minutes: 5 },
  internal.sync.fetchExternalData,
  {}
);

// 每 2 小時
crons.interval(
  "cleanup temp files",
  { hours: 2 },
  internal.files.cleanupTempFiles,
  {}
);

// 每 30 秒（最小間隔）
crons.interval(
  "health check",
  { seconds: 30 },
  internal.monitoring.healthCheck,
  {}
);

export default crons;

Cron 表達式排程

使用 crons.cron 進行精確的 cron 表達式排程：

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

const crons = cronJobs();

// 每天 UTC 上午 9 點
crons.cron(
  "morning notifications",
  "0 9 * * *",
  internal.notifications.sendMorningDigest,
  {}
);

// 每週一 UTC 上午 8 點
crons.cron(
  "weekly summary",
  "0 8 * * 1",
  internal.reports.generateWeeklySummary,
  {}
);

// 每月第一天午夜
crons.cron(
  "monthly billing",
  "0 0 1 * *",
  internal.billing.processMonthlyBilling,
  {}
);

// 每 15 分鐘
crons.cron(
  "frequent sync",
  "*/15 * * * *",
  internal.sync.syncData,
  {}
);

export default crons;

Cron 表達式參考

┌───────────── 分鐘 (0-59)
│ ┌───────────── 小時 (0-23)
│ │ ┌───────────── 月份中的日期 (1-31)
│ │ │ ┌───────────── 月份 (1-12)
│ │ │ │ ┌───────────── 星期幾 (0-6，星期日=0)
│ │ │ │ │
* * * * *

常見模式：

• * * * * * - 每分鐘
• 0 * * * * - 每小時
• 0 0 * * * - 每天午夜
• 0 0 * * 0 - 每週日午夜
• 0 0 1 * * - 每月第一天
• */5 * * * * - 每 5 分鐘
• 0 9-17 * * 1-5 - 週一至週五，上午 9 點到下午 5 點，每小時

用於 Cron 的內部函式

Cron 任務應呼叫內部函式以確保安全性：

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

// 清理過期的工作階段
// 清理過期的 session
export const cleanupExpiredSessions = internalMutation({
  args: {},
  returns: v.number(),
  handler: async (ctx) => {
    const oneHourAgo = Date.now() - 60 * 60 * 1000;
    
    const expiredSessions = await ctx.db
      .query("sessions")
      .withIndex("by_lastActive")
      .filter((q) => q.lt(q.field("lastActive"), oneHourAgo))
      .collect();

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

    return expiredSessions.length;
  },
});

// 處理待處理的任務
export const processPendingTasks = internalMutation({
  args: {},
  returns: v.null(),
  handler: async (ctx) => {
    const pendingTasks = await ctx.db
      .query("tasks")
      .withIndex("by_status", (q) => q.eq("status", "pending"))
      .take(100);

    for (const task of pendingTasks) {
      await ctx.db.patch(task._id, {
        status: "processing",
        startedAt: Date.now(),
      });
      
      // 排程實際處理
      await ctx.scheduler.runAfter(0, internal.tasks.processTask, {
        taskId: task._id,
      });
    }

    return null;
  },
});

帶有參數的 Cron 任務

將靜態參數傳遞給 cron 任務：

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

const crons = cronJobs();

// 不同類型的清理間隔
crons.interval(
  "cleanup temp files",
  { hours: 1 },
  internal.cleanup.cleanupByType,
  { fileType: "temp", maxAge: 3600000 }
);

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

export default crons;

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

export const cleanupByType = internalMutation({
  args: {
    fileType: v.string(),
    maxAge: v.number(),
  },
  returns: v.number(),
  handler: async (ctx, args) => {
    const cutoff = Date.now() - args.maxAge;
    
    const oldFiles = await ctx.db
      .query("files")
      .withIndex("by_type_and_created", (q) => 
        q.eq("type", args.fileType).lt("createdAt", cutoff)
      )
      .collect();

    for (const file of oldFiles) {
      await ctx.storage.delete(file.storageId);
      await ctx.db.delete(file._id);
    }

    return oldFiles.length;
  },
});

監控與記錄

新增記錄以追蹤 cron 任務的執行：

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

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

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

      for (const item of expiredItems) {
        try {
          await ctx.db.delete(item._id);
          processedCount++;
        } catch (error) {
          errorCount++;
          console.error(`Failed to delete item ${item._id}:`, error);
        }
      }

      // 記錄任務完成
      await ctx.db.insert("cronLogs", {
        jobName: "cleanup",
        startTime,
        endTime: Date.now(),
        duration: Date.now() - startTime,
        processedCount,
        errorCount,
        status: errorCount === 0 ? "success" : "partial",
      });
    } catch (error) {
      // 記錄任務失敗
      await ctx.db.insert("cronLogs", {
        jobName: "cleanup",
        startTime,
        endTime: Date.now(),
        duration: Date.now() - startTime,
        processedCount,
        errorCount,
        status: "failed",
        error: String(error),
      });
      throw error;
    }

    return null;
  },
});

大型資料集的批次處理

分批處理大型資料集以避免逾時：

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

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(),
      });
    }

    // 如果有更多項目，排程下一批
    if (!result.isDone) {
      await ctx.scheduler.runAfter(0, internal.tasks.processBatch, {
        cursor: result.continueCursor,
      });
    }

    return null;
  },
});

Cron 中的外部 API 呼叫

使用 actions 進行外部 API 呼叫：

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

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

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

    if (!response.ok) {
      throw new Error(`API request failed: ${response.status}`);
    }

    const data = await response.json();

    // 使用 mutation 儲存資料
    await ctx.runMutation(internal.sync.storeExternalData, {
      data,
      syncedAt: Date.now(),
    });

    return null;
  },
});

export const storeExternalData = internalMutation({
  args: {
    data: v.any(),
    syncedAt: v.number(),
  },
  returns: v.null(),
  handler: async (ctx, args) => {
    await ctx.db.insert("externalData", {
      data: args.data,
      syncedAt: args.syncedAt,
    });
    return null;
  },
});

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

const crons = cronJobs();

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

export default crons;

範例

Cron 任務記錄的 Schema

// convex/schema.ts
import { defineSchema, defineTable } from "convex/server";
import { v } from "convex/values";

export default defineSchema({
  cronLogs: defineTable({
    jobName: v.string(),
    startTime: v.number(),
    endTime: v.number(),
    duration: v.number(),
    processedCount: v.number(),
    errorCount: v.number(),
    status: v.union(
      v.literal("success"),
      v.literal("partial"),
      v.literal("failed")
    ),
    error: v.optional(v.string()),
  })
    .index("by_job", ["jobName"])
    .index("by_status", ["status"])
    .index("by_startTime", ["startTime"]),

  sessions: defineTable({
    userId: v.id("users"),
    token: v.string(),
    lastActive: v.number(),
    expiresAt: v.number(),
  })
    .index("by_user", ["userId"])
    .index("by_lastActive", ["lastActive"])
    .index("by_expiresAt", ["expiresAt"]),

  tasks: defineTable({
    type: v.string(),
    status: v.union(
      v.literal("pending"),
      v.literal("processing"),
      v.literal("completed"),
      v.literal("failed")
    ),
    data: v.any(),
    createdAt: v.number(),
    startedAt: v.optional(v.number()),
    completedAt: v.optional(v.number()),
  })
    .index("by_status", ["status"])
    .index("by_type_and_status", ["type", "status"]),
});

完整的 Cron 設定範例

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

const crons = cronJobs();

// 清理任務
crons.interval(
  "cleanup expired sessions",
  { hours: 1 },
  internal.cleanup.expiredSessions,
  {}
);

crons.interval(
  "cleanup old logs",
  { hours: 24 },
  internal.cleanup.oldLogs,
  { maxAgeDays: 30 }
);

// 同步任務
crons.interval(
  "sync user data",
  { minutes: 15 },
  internal.sync.userData,
  {}
);

// 報告任務
crons.cron(
  "daily analytics",
  "0 1 * * *",
  internal.reports.dailyAnalytics,
  {}
);

crons.cron(
  "weekly summary",
  "0 9 * * 1",
  internal.reports.weeklySummary,
  {}
);

// 健康檢查
crons.interval(
  "service health check",
  { minutes: 5 },
  internal.monitoring.healthCheck,
  {}
);

export default crons;

最佳實踐

• 除非明確指示，否則切勿執行 npx convex deploy
• 除非明確指示，否則切勿執行任何 git 指令
• 僅使用 crons.interval 或 crons.cron 方法，不要使用已棄用的輔助函式
• 為了安全性，cron 任務應一律呼叫內部函式
• 即使函式在同一個檔案中，也應從 _generated/api 匯入 internal
• 為生產環境的 cron 任務新增記錄和監控
• 對處理大型資料集的操作使用批次處理
• 妥善處理錯誤以防止任務失敗
• 使用有意義的任務名稱以便在儀表板中檢視
• 使用 cron 表達式時考慮時區（Convex 使用 UTC）

常見陷阱

1. 使用公開函式 - Cron 任務應僅呼叫內部函式
2. 長時間執行的 mutation - 將大型操作拆分成批次
3. 缺少錯誤處理 - 未處理的錯誤會導致整個任務失敗
4. 忘記時區 - 所有 cron 表達式都使用 UTC
5. 使用已棄用的輔助函式 - 避免使用 crons.hourly、crons.daily 等
6. 未記錄執行 - 使得除錯生產問題變得困難

參考資料

• Convex 文件：https://docs.convex.dev/
• Convex LLMs.txt：https://docs.convex.dev/llms.txt
• Cron 任務：https://docs.convex.dev/scheduling/cron-jobs
• 排程概覽：https://docs.convex.dev/scheduling
• 排程函式：https://docs.convex.dev/scheduling/scheduled-functions