inngest-steps

Inngest ステップ

Inngest のステップメソッドを使って、堅牢で持続可能なワークフローを構築しましょう。各ステップは個別の HTTP リクエストであり、独立して再試行および監視できます。

これらのスキルは TypeScript に焦点を当てています。 Python や Go については、言語固有のガイダンスとして Inngest ドキュメント を参照してください。コアコンセプトはすべての言語に共通して適用されます。

コアコンセプト

🔄 重要: 各ステップは関数を最初から再実行します。 すべての非決定的なコード（API 呼び出し、DB クエリ、ランダム性）はステップの内側に置き、決して外側に置かないでください。

📊 ステップの制限: すべての関数は最大 1,000 ステップ、ステップデータの合計は最大 4MB です。

// ❌ WRONG - will run 4 times
export default inngest.createFunction(
  { id: "bad-example", triggers: [{ event: "test" }] },
  async ({ step }) => {
    console.log("This logs 4 times!"); // Outside step = bad
    await step.run("a", () => console.log("a"));
    await step.run("b", () => console.log("b"));
    await step.run("c", () => console.log("c"));
  }
);

// ✅ CORRECT - logs once each
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 がカウンターを自動的に処理します。

// Basic usage
const result = await step.run("fetch-user", async () => {
  const user = await db.user.findById(userId);
  return user; // Always return useful data
});

// Synchronous code works too
const transformed = await step.run("transform-data", () => {
  return processData(result);
});

// Side effects (no return needed)
await step.run("send-notification", async () => {
  await sendEmail(user.email, "Welcome!");
});

✅ すべきこと:

• すべての非決定的なロジックをステップの内側に置く
• 後続のステップに役立つデータを返す
• ループ内でステップ ID を再利用する（カウンターは自動的に処理されます）

❌ すべきでないこと:

• 決定的なロジックを不必要にステップ内に置く
• 各ステップ = 個別の HTTP リクエストであることを忘れる

step.sleep()

コンピュート時間を使用せずに実行を一時停止します。

// Duration strings
await step.sleep("wait-24h", "24h");
await step.sleep("short-delay", "30s");
await step.sleep("weekly-pause", "7d");

// Use in workflows
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);

// From event data
const scheduledTime = new Date(event.data.remind_at);
await step.sleepUntil("wait-for-scheduled-time", scheduledTime);

step.waitForEvent()

🚨 重要: waitForEvent はこのステップが実行された後に送信されたイベントのみを捕捉します。

• ❌ waitForEvent の実行前に送信されたイベント → 捕捉されません
• ✅ waitForEvent の実行後に送信されたイベント → 捕捉されます
• 常に null の戻り値をチェックしてください（タイムアウト、つまりイベントが到達しなかったことを意味します）

// Basic event waiting with timeout
const approval = await step.waitForEvent("wait-for-approval", {
  event: "app/invoice.approved",
  timeout: "7d",
  match: "data.invoiceId" // Simple matching
});

// Expression-based matching (CEL syntax)
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'"
});

// Handle timeout
if (!approval) {
  await step.run("handle-timeout", () => {
    // Approval never came
    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" // Required: "replace" overwrites pending signal, "fail" throws an error
});

// Send signal elsewhere via Inngest API or SDK
// POST /v1/events with signal matching taskId

使用する場面:

• waitForEvent: 複数の関数が同じイベントを処理する可能性がある場合
• waitForSignal: 特定の関数実行への厳密な 1:1 シグナル

step.sendEvent()

結果を待たずに他の関数へファンアウトします。

// Trigger other functions
await step.sendEvent("notify-systems", {
  name: "user/profile.updated",
  data: { userId: user.id, changes: profileChanges }
});

// Multiple events at once
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 };
  }
);

// Invoke and use result
const square = await step.invoke("get-square", {
  function: computeSquare,
  data: { number: 4 }
});

console.log(square.result); // 16, fully typed!

// For cross-app invocation (when you can't import the function directly):
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 の破壊的変更: 文字列の関数 ID（例: function: "my-app-other-fn"）は step.invoke() でサポートされなくなりました。インポートした関数参照、またはクロスアプリ呼び出しには referenceFunction() を使用してください。

最適な用途:

• 複雑なワークフローをコンポーザブルな関数に分割する
• 複数のワークフロー間でロジックを再利用する
• マップリデュースパターン

パターン

ステップを使ったループ

ステップ ID を再利用します - Inngest がカウンターを自動的に処理します。

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

while (hasMore) {
  // Same ID "fetch-page" reused - counters handled automatically
  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 では、並列ステップ実行はデフォルトで最適化されています

// Create steps without awaiting
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);
});

// Run all in parallel
const [emailId, crmRecord, subscription] = await Promise.all([
  sendEmail,
  updateCRM,
  createSubscription
]);

// Parallel steps are optimized by default in 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() behavior with v4's optimized parallelism:
// All promises settle before race resolves. Use group.parallel() for true race:
const winner = await group.parallel(async () => {
  return Promise.race([
    step.run("fast-service", () => callFastService()),
    step.run("slow-service", () => callSlowService())
  ]);
});

// To disable optimized parallelism if needed:
// At the client level: new Inngest({ id: "app", optimizeParallelism: false })
// At the function level: { 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);

    // Process chunks in parallel
    const results = await Promise.all(
      chunks.map((chunk, index) =>
        step.run(`process-chunk-${index}`, () => processChunk(chunk))
      )
    );

    // Combine results
    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() を使用してください

一般的なユースケース

• 人間が介在するワークフロー： waitForEvent + Realtime UI
• 複数ステップのオンボーディング： ステップ間で sleep を使用し、ユーザーの操作には waitForEvent を使用します
• データ処理： チャンクに分割した作業に並列ステップを使用します
• 外部連携： 信頼性の高い API 呼び出しには step.run を使用します
• AI ワークフロー： 永続性のある LLM オーケストレーションには step.ai を使用します
• 関数の合成： 複雑なワークフローの構築には step.invoke を使用します

ステップを使うと、関数の永続性、可観測性、デバッグのしやすさが向上することを忘れないでください。ぜひ活用しましょう！