API reference @cogover/sdk: Background job

Trang này thuộc API reference của @cogover/sdk.

Background job

Background job chạy code của project ngoài một HTTP request: được enqueue bởi script, trigger, job khác hoặc quản trị viên, hoặc được khởi chạy theo lịch cron. Khai báo từng job bằng defineJob và liệt kê kết quả trong named export jobs của project. Default export vẫn là HTTP handler và trở thành tuỳ chọn với project chỉ khai báo job và trigger.

Ví dụ này giả định workspace.d.ts đã khai báo Object order và các field được sử dụng bên dưới.

import { defineJob } from "@cogover/sdk";
 
interface RecalcPayload {
  cursor?: string;
}
 
export const jobs = [
  // Được enqueue từ route hoặc trigger; duyệt toàn bộ order, mỗi lần chạy một trang.
  defineJob<RecalcPayload>({
    key: "recalc_totals",
    name: "Recalculate order totals",
    timeoutMs: 60_000,
  }, async ({ payload, data, jobs, log }) => {
    const orders = data.object("order");
    const page = await orders.records.list({
      fields: ["subtotal", "tax"],
      limit: 200,
      ...(payload?.cursor ? { cursor: payload.cursor } : {}),
    });
    for (const order of page.items) {
      await orders.records.update(order.id, {
        total: (order.fields.subtotal ?? 0) + (order.fields.tax ?? 0),
      });
    }
    if (page.nextCursor) {
      // Tiếp tục trang kế tiếp trong một lần chạy mới thay vì một lần chạy dài.
      await jobs.enqueue("recalc_totals", { cursor: page.nextCursor });
      return;
    }
    log.info("Order totals recalculated");
  }),
 
  // Chạy hằng ngày lúc 02:00 theo múi giờ đã cho.
  defineJob({
    key: "cancel_stale_orders",
    schedule: { cron: "0 2 * * *", timezone: "Asia/Ho_Chi_Minh" },
  }, async ({ data }) => {
    const orders = data.object("order");
    const stale = await orders.records.list({
      where: orders.fields.status.eq("new"),
      fields: ["status"],
      limit: 200,
    });
    for (const order of stale.items) {
      await orders.records.update(order.id, { status: "cancelled" });
    }
  }),
];

Cogover đọc cấu hình job khi một version của project được publish và áp dụng cấu hình đó trong thời gian version ấy đang active: jobs.enqueue chỉ nhận các key do version active khai báo, và lịch chỉ chạy khi version của nó đang active. Deactivate project sẽ gỡ các lịch của project. Một project khai báo tối đa 50 job.

defineJob(config, handler): JobDefinition

function defineJob<
  Payload = unknown,
  Schema extends object = EffectiveWorkspaceObjects,
>(config: JobConfig, handler: JobHandler<Payload, Schema>): JobDefinition;

Payload là kiểu của giá trị truyền cho jobs.enqueue; giá trị này không được validate lúc chạy, nên hãy xử lý payload như input của request. defineJob validate cấu hình ngay lập tức và ném ValidationError với mọi vi phạm, với key cấu hình không xác định hoặc với handler không phải function. Hàm trả về một JobDefinition đã freeze.

interface JobConfig {
  readonly key: string;
  readonly name?: string;
  readonly timeoutMs?: number;
  readonly maxAttempts?: number;
  readonly schedule?: JobSchedule;
}
 
interface JobSchedule {
  readonly cron: string;
  readonly timezone?: string;
}

Một trường cron là *, một số, khoảng a-b, danh sách a,b, bước */n hoặc khoảng có bước a-b/n; các trường nhận phút 0–59, giờ 0–23, ngày 1–31, tháng 1–12 và thứ 0–7 trong đó cả 0 và 7 đều là Chủ nhật. Tên như MON, các ký tự L, W, #, ? và macro như @daily không được hỗ trợ. Chu kỳ ngắn nhất là một phút. Khoảng trắng giữa các trường được chuẩn hoá thành một dấu cách trong manifest, và biểu thức sau khi chuẩn hoá dài tối đa 128 ký tự; biểu thức dài hơn ném ValidationError. Cogover lưu time zone ở dạng chuẩn hoá, nên offset như UTC+7 được liệt kê là UTC+07:00 trong danh sách lịch của project.

JobDefinition và JobManifest

interface JobManifest {
  readonly key: string;
  readonly name?: string;
  readonly timeoutMs: number;
  readonly maxAttempts: number;
  readonly schedule?: { readonly cron: string; readonly timezone: string };
}
 
interface JobDefinition {
  readonly key: string;
  readonly config: JobManifest;
  readonly __cogoverJobHandler: SandboxHandler;
}

config là cấu hình đã chuẩn hoá dưới dạng JSON thuần được deep-freeze, với mọi giá trị mặc định đã áp dụng: timeoutMs: 30000, maxAttempts: 5 và schedule.timezone: "UTC". Các thiết lập tuỳ chọn không được cung cấp sẽ vắng mặt. config tách khỏi object đã truyền cho defineJob, nên sửa object đó về sau không có tác dụng. __cogoverJobHandler dành riêng cho Cogover; code của project không gọi nó.

Context của job handler

type JobHandler<
  Payload = unknown,
  Schema extends object = EffectiveWorkspaceObjects,
> = (context: JobContext<Payload, Schema>) => void | Promise<void>;
 
interface JobContext<Payload = unknown, Schema extends object = EffectiveWorkspaceObjects> {
  readonly job: JobInfo;
  readonly payload: Payload | null;
  readonly invocation: InvocationContext;
  readonly data: DataApi<Schema>;
  readonly schema: SchemaApi<Schema>;
  readonly log: ScriptLogger;
  readonly state: ProjectState;
  readonly locks: DistributedLocks;
  readonly notifications: NotificationsApi;
  readonly email: EmailApi<Schema>;
  readonly jobs: JobsApi;
  readonly secrets: SecretsApi;
  readonly crypto: CryptoApi;
  readonly org: OrgApi;
  readonly limits: LimitsApi;
  readonly processes: ProcessesApi;
  readonly agents: AgentsApi;
}
 
interface JobInfo {
  readonly id: string;
  readonly key: string;
  readonly attempt: number;
  readonly maxAttempts: number;
  readonly source: "enqueue" | "schedule";
  readonly runAt: number;
  readonly enqueuedAt?: number;
}

Handler có thể đồng bộ hoặc bất đồng bộ; giá trị trả về bị bỏ qua. data, schema, log, invocation, state, locks, notifications, email, jobs, secrets, crypto, org, limits, processes và agents là chính các API mà script nhận được; limits cho biết ngân sách lớn hơn của job (xem Giới hạn của một lần thực thi). Job context không có input, request, response hay push.

• job.id định danh lần chạy và không đổi qua các lần thử; dùng nó làm idempotency key cho các lời gọi ra ngoài. job.attempt bắt đầu từ 1.
• job.source là "enqueue" với lần chạy do jobs.enqueue hoặc quản trị viên tạo, hoặc để nhận kết quả của Process hay AI Agent, và "schedule" với lần chạy do lịch của job tạo. job.runAt là thời điểm lần chạy đến hạn và job.enqueuedAt là thời điểm được enqueue; cả hai là Unix millisecond, và enqueuedAt vắng mặt với lần chạy theo lịch.
• payload là giá trị JSON đã truyền cho jobs.enqueue, hoặc null khi không truyền gì hoặc lần chạy theo lịch. Đây là dữ liệu thuần, có thể sửa.

jobs.enqueue(jobKey, payload?, options?): Promise<EnqueueResult>

interface JobsApi {
  enqueue(jobKey: string, payload?: unknown, options?: EnqueueOptions): Promise<EnqueueResult>;
}
 
interface EnqueueOptions {
  readonly delayMs?: number;
  readonly runAt?: number;
  readonly idempotencyKey?: string;
}
 
interface EnqueueResult {
  readonly runId: string;
  readonly duplicate: boolean;
}

const { runId, duplicate } = await jobs.enqueue("sync_order", { orderId: order.id }, {
  delayMs: 60_000,
  idempotencyKey: `sync_order:${order.id}:${order.system.updatedAt}`,
});

jobKey phải là job do version active khai báo; nếu không, Cogover ném ValidationError. payload là giá trị JSON bất kỳ, tối đa 65.536 byte UTF-8 của dạng JSON.stringify, trong đó dấu ngoặc kép, dấu gạch chéo ngược hoặc ký tự điều khiển được tính theo chuỗi escape của nó; undefined và null enqueue lần chạy không có payload. Giá trị mà JSON sẽ âm thầm thay đổi, như NaN hoặc instance của class, bị từ chối bằng ValidationError. delayMs là số nguyên từ 0 đến 2.592.000.000 (30 ngày) và runAt là thời điểm Unix millisecond không quá 30 ngày tới; truyền một trong hai hoặc không truyền, không truyền cả hai. idempotencyKey bắt đầu bằng chữ cái hoặc chữ số, chứa chữ cái, chữ số, ., _, : hoặc -, tối đa 128 ký tự: lần enqueue thứ hai của cùng job với cùng key khi lần chạy trước còn được lưu sẽ trả về lần chạy đó với duplicate: true thay vì tạo thêm. Lịch sử lần chạy được giữ 7 ngày.

enqueue hoạt động trong script và route, trigger after-change, job, lời gọi inbound webhook và Custom Module Action có effect là "write". Trigger before-change bị từ chối bằng PermissionDeniedError (details.reason === "TRIGGER_READ_ONLY"), phiên phát triển local không có quyền ghi cũng vậy ("DEVELOPMENT_SESSION_READ_ONLY"). Một invocation enqueue tối đa 50 lần chạy, và một project có tối đa 10.000 lần chạy đang chờ hoặc đang chạy; vượt quá sẽ ném RateLimitError. Mỗi lần enqueue cũng tính là một capability call (xem Giới hạn của một lần thực thi). Khi job chưa được bật cho Workspace, lỗi là CogoverApiError với code: "JOBS_DISABLED".

Quy tắc thực thi job

• Danh tính. Lần chạy được enqueue từ script, route hoặc trigger chạy dưới danh tính người dùng của invocation đó: invocation mô tả người dùng ấy và data.object() thực hiện với quyền của người ấy. Lần chạy do Super Admin enqueue qua API quản lý job cũng chạy dưới danh tính Super Admin đó, và lịch sử lần chạy ghi enqueuedBy: "management". Lần chạy theo lịch và lần chạy được enqueue từ lời gọi inbound webhook không có người dùng: invocation.identity là "system", và data.object() chỉ hoạt động nếu identity policy đã duyệt của project đặt allowInternalSystem: true; nếu không, các lời gọi của nó ném PermissionDeniedError có details.reason là "IDENTITY_NOT_GRANTED". data.asUser() và data.asSystem() tuân theo identity policy như thông thường. Trong lần chạy có người dùng, records.aggregate qua data.object() có giới hạn của data.asUser(); dùng data.asSystem() để nhóm (xem Aggregate).
• Capability. Job được đọc và ghi record, gọi fetch, dùng state và locks, gửi notification và email, enqueue job, đọc secret, dùng crypto, và khởi chạy Process và AI Agent, trong giới hạn runtime của làn job.
• Kết quả của Process hoặc AI Agent. Lần chạy do onComplete của processes.start hoặc onResult của agents.start tạo nhận ProcessCompletionJobPayload hoặc AgentResultJobPayload làm payload, chạy dưới danh tính người dùng của lần thực thi đã gọi start, và được ghi trong lịch sử lần chạy với enqueuedBy là process:<instanceId> hoặc agent:<runId> (xem Quy tắc của job nhận kết quả). Lần chạy được enqueue từ Custom Module Action chạy dưới danh tính của lần gọi action đó.
• Thời gian. Mỗi lần thử phải hoàn tất trong timeoutMs. Việc cần lâu hơn phải được chia nhỏ: xử lý một trang, lưu cursor và enqueue lại chính job đó với cursor trong payload, như ví dụ ở trên.
• Ít nhất một lần. Cogover không bảo đảm một lần chạy được thực thi đúng một lần: lần thử bị gián đoạn, ví dụ do restart, sẽ được bắt đầu lại. Hãy viết mọi handler theo kiểu idempotent. job.id định danh lần chạy qua các lần thử, và idempotencyKey của enqueue ngăn tạo lần chạy trùng.
• Retry. Ném RetryableError khi handler thất bại vì lý do tạm thời, ví dụ dịch vụ bên ngoài không sẵn sàng. Cogover sẽ thử lại sau, mỗi lần chờ lâu hơn, từ vài giây đến khoảng nửa giờ, cho đến khi đã thử đủ maxAttempts lần. Lần thử vượt timeoutMs, hoặc lần thử Cogover không chạy được vì lý do tạm thời, cũng được thử lại theo cách đó. Mọi lỗi khác kết thúc lần chạy với trạng thái thất bại: lỗi được ghi log và handler không chạy lại cho lần chạy đó.
• Thứ tự và đồng thời. Các lần chạy thực thi không theo thứ tự bảo đảm, và hai lần chạy của cùng project, kể cả cùng job, có thể thực thi cùng lúc. Dùng locks khi một đoạn không được chạy đồng thời, và trì hoãn bằng delayMs hoặc runAt thay vì chờ bên trong handler.
• Vòng lặp luôn kết thúc. Thao tác ghi record do một lần chạy thực hiện mang ngân sách tự động hoá của invocation đã enqueue nó (lần chạy theo lịch và lần enqueue của quản trị viên bắt đầu với đủ ngân sách 10 bước), nên chuỗi job và trigger khởi phát từ thao tác ghi record kết thúc giống chuỗi trigger. Enqueue job không dùng bước nào: job tự enqueue chính nó phải tự dừng, ví dụ khi hết cursor.

Job theo lịch tạo một lần chạy cho mỗi thời điểm lịch khớp, theo múi giờ của lịch; lần chạy có thể bắt đầu muộn hơn phút đã lên lịch một chút khi tải cao. Thời điểm bị bỏ lỡ trong lúc project không active sẽ không được chạy bù. Lần chạy vẫn được tạo khi lần chạy theo lịch trước đó chưa xong, nên lịch có việc có thể kéo dài hơn chu kỳ của nó nên tự bảo vệ bằng lock.