vtex-io-events-and-workers

イベント、ワーカー、非同期処理

このスキルを使用する場面

VTEX IO アプリで、イベント、ワーカー、その他のバックグラウンド実行パターンを通じて非同期に処理する必要がある場合に、このスキルを使用してください。

• 他の VTEX サービスからブロードキャストされたイベントを消費する
• HTTP レスポンスをブロックしないバックグラウンド処理を実行する
• リトライに対応したハンドラーを設計する
• バッチや遅延ジョブを処理する
• 外部サービスと非同期連携する

次の場合は、このスキルを使用しないでください。

• HTTP ルートのコントラクトを定義する
• GraphQL スキーマやリゾルバーを設計する
• アプリレベルのポリシーを決定する
• 低レベルのクライアント構築

判断ルール

• 処理に時間がかかる、リトライが発生しやすい、またはリクエストとレスポンスのサイクル内で完了する必要がない場合は、イベントまたはワーカーを使用してください。
• VTEX は内部イベントブロードキャスターを使用して、プラットフォームやアプリのイベントをサービスに配信します。同じブロードキャスターを使って、アプリが発行したイベントを他のハンドラーにルーティングすることもできます。双方向とも少なくとも 1 回の配信が保証されるものと想定してください。イベントはリトライまたは再生される可能性があるため、ハンドラーは冪等で、重複に安全に対処できる必要があります。
• イベントハンドラーは冪等にしてください。同じイベントが複数回配信される可能性があるため、ハンドラーは再生に安全に対応できる必要があります。
• 重複、完了済みの処理、リトライ中の失敗をハンドラーが検出できるよう、キー付きマーカーには VBase、構造化レコードには Master Data など、適切なストアに冪等性と処理状態を永続化してください。
• イベントとワーカーを service.json で明示的に宣言して IO ランタイムに組み込み、HTTP ルートの前提に依存せず、入力コントラクトを安定した明示的なものにしてください。
• 他のアプリへの通知や処理の振り分けが必要な場合は、非同期配信を模倣するためだけに場当たり的な HTTP コールバックを作成するのではなく、適切な VTEX IO クライアントまたはイベントメカニズムを通じてイベントを発行してください。
• VTEX IO イベントバスを通じてイベントを発行するには、多くの場合、アプリの manifest.json に colossus-fire-event ポリシーが必要です。アプリが保護された他のリソースも実際に利用する場合に限り、追加のポリシーを設定してください。
• ハンドラーが小さく明確な単位の作業を超えて大きくなった場合は、イベントの受け取りとビジネス処理のオーケストレーションを分離してください。
• リトライは例外的な動作ではなく、想定される動作として扱ってください。繰り返し実行しても安全なハンドラーを設計してください。
• バックグラウンドハンドラーでは、書き込み、外部呼び出し、状態遷移などの副作用を明示してください。
• バッチ指向のハンドラーでは、明示的な小単位で項目を処理し、項目ごとに状態を記録してください。これにより、1 つの項目が失敗しても、バッチの残りの進捗が隠れないようにできます。

厳格な制約

制約：イベントハンドラーは冪等でなければなりません

すべてのイベントハンドラーまたはバックグラウンドハンドラーは、重複実行されても一貫性のない副作用を発生させないようにする必要があります。

重要な理由

非同期システムではリトライが発生します。冪等性がないと、重複した配信によってレコードの重複、パートナーへの複数回の呼び出し、または無効な状態遷移が発生する可能性があります。

検出方法

処理がすでに完了しているかを確認せずに、ハンドラーが書き込みや外部への副作用を実行する場合は、いったん停止し、処理を進める前に冪等性の保護を追加してください。

正しい例

export async function handleOrderCreated(ctx: Context) {
  const { orderId } = ctx.body
  const alreadyProcessed = await ctx.clients.statusStore.hasProcessed(orderId)

  if (alreadyProcessed) {
    return
  }

  await ctx.clients.partnerApi.sendOrder(orderId)
  await ctx.clients.statusStore.markProcessed(orderId)
}

誤った例

export async function handleOrderCreated(ctx: Context) {
  await ctx.clients.partnerApi.sendOrder(ctx.body.orderId)
}

制約：バックグラウンド処理はリクエスト時のみの前提に依存してはなりません

ワーカーとイベントハンドラーは、ルートのパラメーター、即時のユーザー操作、リクエストに紐づく可変状態など、HTTP 固有の前提に依存してはなりません。

重要な理由

非同期ハンドラーはルートのライフサイクル外で実行されます。HTTP の前提を再利用すると、コンテキストの欠落、脆弱な動作、同期処理と非同期処理の経路間の意図しない結合につながります。

検出方法

イベントハンドラーがリクエストヘッダー、ルートのパラメーター、またはルート固有の状態構造を想定している場合は、いったん停止し、ハンドラーが明示的な非同期データを受け取るように入力コントラクトを再設計してください。

正しい例

export async function handleImport(ctx: Context) {
  const { importId, account } = ctx.body
  await ctx.clients.importApi.process(importId, account)
}

誤った例

export async function handleImport(ctx: Context) {
  await ctx.clients.importApi.process(ctx.vtex.route.params.id, ctx.request.header.account)
}

制約：コストのかかる非同期フローでは部分的な失敗を明確に示す必要があります

非同期ハンドラーは、失敗を黙って握りつぶすのではなく、状態、ログ、または永続的なマーカーを通じて部分的な失敗を必ず可視化する必要があります。

重要な理由

バックグラウンドでの失敗は、ルートでの失敗よりも見つけにくいものです。明示的な失敗通知がないと、運用チームは処理がスキップされたのか、再試行されたのか、部分的に完了したのかを判断できません。

検出方法

ハンドラーが失敗状態を記録せず、十分なコンテキストをログに残さず、適切な場合にエラーを再スローもしないままエラーを捕捉している場合は、処理を停止し、失敗処理を明示してください。

適切な例

export async function handleSync(ctx: Context) {
  try {
    await ctx.clients.partnerApi.syncCatalog(ctx.body.catalogId)
    await ctx.clients.statusStore.markSuccess(ctx.body.catalogId)
  } catch (error) {
    await ctx.clients.statusStore.markFailure(ctx.body.catalogId)
    throw error
  }
}

不適切な例

export async function handleSync(ctx: Context) {
  try {
    await ctx.clients.partnerApi.syncCatalog(ctx.body.catalogId)
  } catch (_) {
    return
  }
}

推奨パターン

推奨するファイル構成：

node/
├── events/
│   ├── index.ts
│   ├── catalog.ts
│   └── orders.ts
└── workers/
    └── sync.ts

非同期ハンドラーの最小パターン：

export async function handleCatalogChanged(ctx: Context) {
  const { skuId } = ctx.body
  const alreadyDone = await ctx.clients.syncState.isProcessed(skuId)

  if (alreadyDone) {
    return
  }

  await ctx.clients.catalogSync.syncSku(skuId)
  await ctx.clients.syncState.markProcessed(skuId)
}

イベント発行の例：

export async function broadcast(ctx: Context, next: () => Promise<void>) {
  const {
    clients: { events },
    body: { payload, senderAppId, clientAppId },
  } = ctx

  for (const row of payload as unknown[]) {
    await events.sendEvent(clientAppId, 'my-app.event-name', {
      data: row,
      senderAppId,
    })
  }

  await next()
}

イベント発行に必要な最小限のマニフェストポリシー：

{
  "policies": [
    {
      "name": "colossus-fire-event"
    }
  ]
}

ルートは処理の受け付けや開始に使用し、時間のかかる、繰り返し可能な、失敗を考慮した処理にはイベントやワーカーを使用してください。

非同期状態を保存する際は、ストレージを意図的に選択してください。

• 外部識別子をキーとするシンプルな冪等性マーカーには、VBase を使用します
• ステータスとタイムスタンプを含む構造化された処理レコードには、Master Data を使用します

HTTP リクエスト向けのルート固有の前提をそのまま再利用するのではなく、非同期ペイロード自体を独立したコントラクトとして扱ってください。

ファンアウトやアプリ間通知では、ID と最小限のメタデータを含む、小さく明確に定義されたイベントを発行してください。その後、必要に応じて下流のハンドラーが信頼できる唯一の情報源から詳細を取得するようにし、大きなペイロードを埋め込んだり、独自のコールバック URL に依存したりしないでください。

開発では、その場しのぎのパブリックルートやテスト専用の配信フローを作る代わりに、Broadcaster アプリの Admin にある Notify Target Workspace 設定を使って、イベントを特定のワークスペースにルーティングしてください。ハンドラーは、イベントの受信先がどのワークスペースであっても正しく動作する必要があります。

よくある失敗パターン

• イベント配信が少なくとも1回ではなく、必ず1回だけ行われるものとして扱う。
• ワーカーやイベントハンドラー内で HTTP ルートの前提を再利用する。
• 明示的な失敗状態を示さずに、バックグラウンドエラーを握りつぶす。
• 1つのイベントハンドラーに、互いに無関係な副作用を過剰に調整させる。
• 高コストの処理を非同期処理に移さず、ルート内で同期的に実行する。
• 相関関係の把握に ID とメタデータを使わず、シークレットやトークンを含むイベントペイロード全体をログに記録する。

レビューチェックリスト

• この処理には非同期処理が適切な仕組みですか？
• 配信が重複しても、ハンドラーは冪等に動作しますか？
• 冪等性または処理状態は、VBase や Master Data などの適切なバックエンドに保存されていますか？
• イベントとワーカーは service.json に明示的に宣言されていますか？
• バックグラウンド処理への入力は明示され、HTTP ルートの前提に依存していませんか？
• 再試行やトラブルシューティングに十分な形で失敗が明示されていますか？
• バッチ処理の場合、項目ごと、または小さな作業単位ごとにステータスを確認できますか？
• 大きなハンドラーを、より小さな非同期処理単位やオーケストレーション手順に分割する必要がありますか？

関連するスキル

• vtex-io-data-access-patterns - 冪等性キー、同期状態、または処理レコードの保存先を選ぶ場合に使用します
• vtex-io-observability-and-ops - 非同期処理の失敗をどのようにログに記録し、測定し、監視するかが主な検討事項である場合に使用します

参考資料

• Service - イベント宣言とサービスの実行モデル
• Node Builder - サービスのバックエンドファイル構成
• Broadcaster - 内部イベント配信の概要と、開発用の Notify Target Workspace 設定