inngest-steps

Inngest 步驟

使用 Inngest 的步驟方法建立穩固、持久的工作流程。每個步驟都是一個獨立的 HTTP 請求，可以獨立重試和監控。

這些技能專注於 TypeScript。 如需 Python 或 Go 的語言特定指引，請參閱 Inngest 文件。核心概念適用於所有語言。

核心概念

🔄 重要：每個步驟都會從頭開始重新執行你的函式。 將所有非確定性程式碼（API 呼叫、資料庫查詢、隨機性）放在步驟內，絕對不要放在外面。

📊 步驟限制： 每個函式最多 1,000 個步驟，總步驟資料 4MB。

// ❌ 錯誤 - 會執行 4 次
export default inngest.createFunction(
  { id: "bad-example", triggers: [{ event: "test" }] },
  async ({ step }) => {
    console.log("這會記錄 4 次！"); // 在步驟外 = 不好
    await step.run("a", () => console.log("a"));
    await step.run("b", () => console.log("b"));
    await step.run("c", () => console.log("c"));
  }
);

// ✅ 正確 - 各記錄一次
export default inngest.createFunction(
  { id: "good-example", triggers: [{ event: "test" }] },
  async ({ step }) => {
    await step.run("log-hello", () => console.log("hello"));
    await step.run("a", () => console.log("a"));
    await step.run("b", () => console.log("b"));
    await step.run("c", () => console.log("c"));
  }
);

step.run()

將可重試的程式碼作為步驟執行。每個步驟 ID 可以重複使用 - Inngest 會自動處理計數器。

// 基本用法
const result = await step.run("fetch-user", async () => {
  const user = await db.user.findById(userId);
  return user; // 總是回傳有用的資料
});

// 同步程式碼也可以
const transformed = await step.run("transform-data", () => {
  return processData(result);
});

// 副作用（不需要回傳）
await step.run("send-notification", async () => {
  await sendEmail(user.email, "Welcome!");
});

✅ 應該：

• 將所有非確定性邏輯放在步驟內
• 回傳有用的資料給後續步驟
• 在迴圈中重複使用步驟 ID（計數器會自動處理）

❌ 不應該：

• 不必要地將確定性邏輯放在步驟中
• 忘記每個步驟 = 獨立的 HTTP 請求

step.sleep()

暫停執行而不消耗運算時間。

// 持續時間字串
await step.sleep("wait-24h", "24h");
await step.sleep("short-delay", "30s");
await step.sleep("weekly-pause", "7d");

// 在工作流程中使用
await step.run("send-welcome", () => sendEmail(email));
await step.sleep("wait-for-engagement", "3d");
await step.run("send-followup", () => sendFollowupEmail(email));

step.sleepUntil()

休眠直到指定的日期時間。

const reminderDate = new Date("2024-12-25T09:00:00Z");
await step.sleepUntil("wait-for-christmas", reminderDate);

// 從事件資料
const scheduledTime = new Date(event.data.remind_at);
await step.sleepUntil("wait-for-scheduled-time", scheduledTime);

step.waitForEvent()

🚨 重要：waitForEvent 只會捕捉在此步驟執行後才發送的事件。

• ❌ 在 waitForEvent 執行前發送的事件 → 不會被捕捉
• ✅ 在 waitForEvent 執行後發送的事件 → 會被捕捉
• 務必檢查 null 回傳（表示超時，事件從未到達）

// 基本事件等待並設定超時
const approval = await step.waitForEvent("wait-for-approval", {
  event: "app/invoice.approved",
  timeout: "7d",
  match: "data.invoiceId" // 簡單比對
});

// 表達式比對（CEL 語法）
const subscription = await step.waitForEvent("wait-for-subscription", {
  event: "app/subscription.created",
  timeout: "30d",
  if: "event.data.userId == async.data.userId && async.data.plan == 'pro'"
});

// 處理超時
if (!approval) {
  await step.run("handle-timeout", () => {
    // 審核從未到來
    return notifyAccountingTeam();
  });
}

✅ 應該：

• 使用唯一 ID 進行比對（userId、sessionId、requestId）
• 總是設定合理的超時時間
• 處理 null 回傳（超時情況）
• 搭配 Realtime 用於人機協作流程

❌ 不應該：

• 預期在此步驟之前發送的事件會被處理
• 在生產環境中使用時不設定超時

表達式語法

在表達式中，event = 原始觸發事件，async = 被比對的新事件。完整語法、運算子和模式請參閱表達式語法參考。

step.waitForSignal()

等待唯一的訊號（而非事件）。更適合 1:1 比對。

const taskId = "task-" + crypto.randomUUID();

const signal = await step.waitForSignal("wait-for-task-completion", {
  signal: taskId,
  timeout: "1h",
  onConflict: "replace" // 必要："replace" 覆蓋待處理訊號，"fail" 擲出錯誤
});

// 透過 Inngest API 或 SDK 在其他地方發送訊號
// POST /v1/events 並使用與 taskId 相符的訊號

使用時機：

• waitForEvent：多個函式可能處理同一個事件
• waitForSignal：精確的 1:1 訊號對應特定函式執行

step.sendEvent()

扇出到其他函式，不等待結果。

// 觸發其他函式
await step.sendEvent("notify-systems", {
  name: "user/profile.updated",
  data: { userId: user.id, changes: profileChanges }
});

// 一次發送多個事件
await step.sendEvent("batch-notifications", [
  { name: "billing/invoice.created", data: { invoiceId } },
  { name: "email/invoice.send", data: { email: user.email, invoiceId } }
]);

使用時機： 你想觸發其他函式，但不需要在當前函式中取得它們的結果。

step.invoke()

呼叫其他函式並處理其結果。非常適合組合。

const computeSquare = inngest.createFunction(
  { id: "compute-square", triggers: [{ event: "calculate/square" }] },
  async ({ event }) => {
    return { result: event.data.number * event.data.number };
  }
);

// 呼叫並使用結果
const square = await step.invoke("get-square", {
  function: computeSquare,
  data: { number: 4 }
});

console.log(square.result); // 16，完整型別！

// 跨應用程式呼叫（當無法直接匯入函式時）：
import { referenceFunction } from "inngest";

const externalFn = referenceFunction({
  appId: "other-app",
  functionId: "other-fn"
});

const result = await step.invoke("call-external", {
  function: externalFn,
  data: { key: "value" }
});

警告：v4 重大變更： step.invoke() 不再支援字串函式 ID（例如 function: "my-app-other-fn"）。請使用匯入的函式參考或 referenceFunction() 進行跨應用程式呼叫。

適用於：

• 將複雜工作流程拆解為可組合的函式
• 在多個工作流程中重複使用邏輯
• Map-reduce 模式

模式

迴圈與步驟

重複使用步驟 ID - Inngest 會自動處理計數器。

const allProducts = [];
let cursor = null;
let hasMore = true;

while (hasMore) {
  // 重複使用相同 ID "fetch-page" - 計數器自動處理
  const page = await step.run("fetch-page", async () => {
    return shopify.products.list({ cursor, limit: 50 });
  });

  allProducts.push(...page.products);

  if (page.products.length < 50) {
    hasMore = false;
  } else {
    cursor = page.products[49].id;
  }
}

await step.run("process-products", () => {
  return processAllProducts(allProducts);
});

平行執行

使用 Promise.all 進行平行步驟。在 v4 中，平行步驟執行預設已最佳化

// 建立步驟但不 await
const sendEmail = step.run("send-email", async () => {
  return await sendWelcomeEmail(user.email);
});

const updateCRM = step.run("update-crm", async () => {
  return await crmService.addUser(user);
});

const createSubscription = step.run("create-subscription", async () => {
  return await subscriptionService.create(user.id);
});

// 全部平行執行
const [emailId, crmRecord, subscription] = await Promise.all([
  sendEmail,
  updateCRM,
  createSubscription
]);

// 在 v4 中，平行步驟預設已最佳化
export default inngest.createFunction(
  {
    id: "parallel-heavy-function",
    triggers: [{ event: "process/batch" }]
  },
  async ({ event, step }) => {
    const results = await Promise.all(
      event.data.items.map((item, i) =>
        step.run(`process-item-${i}`, () => processItem(item))
      )
    );
  }
);

// ⚠️ Promise.race() 在 v4 最佳化平行處理中的行為：
// 所有 promise 在 race 解析前都會完成。使用 group.parallel() 獲得真正的 race：
const winner = await group.parallel(async () => {
  return Promise.race([
    step.run("fast-service", () => callFastService()),
    step.run("slow-service", () => callSlowService())
  ]);
});

// 如果需要停用最佳化平行處理：
// 在客戶端層級：new Inngest({ id: "app", optimizeParallelism: false })
// 在函式層級：{ id: "fn", optimizeParallelism: false, triggers: [...] }

請參閱 inngest-flow-control 了解並行和節流選項。

分塊作業

非常適合批次處理與平行步驟。

export default inngest.createFunction(
  { id: "process-large-dataset", triggers: [{ event: "data/process.large" }] },
  async ({ event, step }) => {
    const chunks = chunkArray(event.data.items, 10);

    // 平行處理區塊
    const results = await Promise.all(
      chunks.map((chunk, index) =>
        step.run(`process-chunk-${index}`, () => processChunk(chunk))
      )
    );

    // 合併結果
    await step.run("combine-results", () => {
      return aggregateResults(results);
    });
  }
);

關鍵陷阱

🔄 函式重新執行： 步驟外的程式碼會在每次步驟執行時執行
⏰ 事件時機： waitForEvent 只會捕捉在步驟執行後發送的事件
🔢 步驟限制： 每個函式最多 1,000 個步驟，每個步驟輸出 4MB，每個函式執行總計 32MB
📨 HTTP 請求： 在 v4 中，檢查點預設啟用，減少 HTTP 開銷。對於無伺服器平台，請在客戶端設定 maxRuntime
🔁 步驟 ID： 可在迴圈中重複使用 - Inngest 會處理計數器
⚡ 平行處理： 使用 Promise.all 進行平行步驟（v4 中預設最佳化）。請注意 Promise.race() 會等待所有 promise 完成 — 使用 group.parallel() 獲得真正的 race 語意

常見使用案例

• 人機協作： waitForEvent + Realtime UI
• 多步驟入職： 步驟之間使用 sleep，waitForEvent 等待使用者操作
• 資料處理： 使用平行步驟進行分塊工作
• 外部整合： step.run 用於可靠的 API 呼叫
• AI 工作流程： step.ai 用於持久的 LLM 編排
• 函式組合： step.invoke 用於建立複雜工作流程

記住：步驟讓你的函式更持久、可觀察且可除錯。擁抱它們吧！