Inngest インテグレーション

サーバーレスファーストのバックグラウンドジョブ、イベント駆動型ワークフロー、キューやワーカーを管理しないダラブルな実行のための Inngest エキスパート。

原則

• イベントはプリミティブ - すべてはキューではなくイベントからトリガーされる
• ステップはあなたのチェックポイント - 各ステップの結果はダラブルに保存される
• スリープはハックではない - Inngest のスリープはリアルなもので、スレッドをブロックしない
• リトライは自動 - ただしポリシーはあなたが制御する
• 関数は単なる HTTP ハンドラー - HTTP を提供するどこにでもデプロイできる
• コンカレンシーはファーストクラスの関心事 - ダウンストリームサービスを保護する
• Idempotency キーは重複を防止 - クリティカルな操作に使用する
• ファンアウトはビルトイン - 1つのイベントは多くの関数をトリガーできる

機能

• inngest-functions
• event-driven-workflows
• step-functions
• serverless-background-jobs
• durable-sleep
• fan-out-patterns
• concurrency-control
• scheduled-functions

スコープ

• redis-queues -> bullmq-specialist
• workflow-orchestration -> temporal-craftsman
• message-streaming -> event-architect
• infrastructure -> infra-architect

ツーリング

コア

• inngest
• inngest-cli

フレームワーク

• nextjs
• express
• hono
• remix
• sveltekit

デプロイメント

• vercel
• cloudflare-workers
• netlify
• railway
• fly-io

パターン

• step-functions
• event-fan-out
• scheduled-cron
• webhook-handling

パターン

基本的な関数セットアップ

Next.js での型付きイベントを使用した Inngest 関数

いつ使うか: 任意の Next.js プロジェクトで Inngest を開始する場合

// lib/inngest/client.ts
import { Inngest } from 'inngest';

export const inngest = new Inngest({
  id: 'my-app',
  schemas: new EventSchemas().fromRecord<Events>(),
});

// イベントを型で定義
type Events = {
  'user/signed.up': { data: { userId: string; email: string } };
  'order/placed': { data: { orderId: string; total: number } };
};

// lib/inngest/functions.ts
import { inngest } from './client';

export const sendWelcomeEmail = inngest.createFunction(
  { id: 'send-welcome-email' },
  { event: 'user/signed.up' },
  async ({ event, step }) => {
    // ステップ 1: ユーザー詳細を取得
    const user = await step.run('get-user', async () => {
      return await db.users.findUnique({ where: { id: event.data.userId } });
    });

    // ステップ 2: ウェルカムメールを送信
    await step.run('send-email', async () => {
      await resend.emails.send({
        to: user.email,
        subject: 'Welcome!',
        template: 'welcome',
      });
    });

    // ステップ 3: 24時間待機、その後にヒントを送信
    await step.sleep('wait-for-tips', '24h');

    await step.run('send-tips', async () => {
      await resend.emails.send({
        to: user.email,
        subject: 'Getting Started Tips',
        template: 'tips',
      });
    });
  }
);

// app/api/inngest/route.ts (Next.js App Router)
import { serve } from 'inngest/next';
import { inngest } from '@/lib/inngest/client';
import { sendWelcomeEmail } from '@/lib/inngest/functions';

export const { GET, POST, PUT } = serve({
  client: inngest,
  functions: [sendWelcomeEmail],
});

マルチステップワークフロー

並列ステップとエラーハンドリングを含む複雑なワークフロー

いつ使うか: 複数のサービスやロングウェイトを含む処理

export const processOrder = inngest.createFunction(
  {
    id: 'process-order',
    retries: 3,
    concurrency: { limit: 10 },  // 最大10件の注文を同時処理
  },
  { event: 'order/placed' },
  async ({ event, step }) => {
    const { orderId } = event.data;

    // 並列ステップ - 両方同時に実行
    const [inventory, payment] = await Promise.all([
      step.run('check-inventory', () => checkInventory(orderId)),
      step.run('validate-payment', () => validatePayment(orderId)),
    ]);

    if (!inventory.available) {
      // 直接呼び出しの代わりにイベントを送信 (ファンアウトパターン)
      await step.sendEvent('notify-backorder', {
        name: 'order/backordered',
        data: { orderId, items: inventory.missing },
      });
      return { status: 'backordered' };
    }

    // 支払いを処理
    const charge = await step.run('charge-payment', async () => {
      return await stripe.charges.create({
        amount: event.data.total,
        customer: payment.customerId,
      });
    });

    // 注文を発送
    await step.run('ship-order', () => fulfillment.ship(orderId));

    return { status: 'completed', chargeId: charge.id };
  }
);

スケジュール/Cron 関数

スケジュールで実行される関数

いつ使うか: 日次レポートやクリーンアップジョブのような繰り返しタスク

export const dailyDigest = inngest.createFunction(
  { id: 'daily-digest' },
  { cron: '0 9 * * *' },  // 毎日午前9時 UTC
  async ({ step }) => {
    // ダイジェストを望むすべてのユーザーを取得
    const users = await step.run('get-users', async () => {
      return await db.users.findMany({
        where: { digestEnabled: true },
      });
    });

    // 各ユーザーに送信 (子イベントを作成)
    await step.sendEvent(
      'send-digests',
      users.map(user => ({
        name: 'digest/send',
        data: { userId: user.id },
      }))
    );

    return { sent: users.length };
  }
);

// 個別のダイジェスト送信を処理する別の関数
export const sendDigest = inngest.createFunction(
  { id: 'send-digest', concurrency: { limit: 50 } },
  { event: 'digest/send' },
  async ({ event, step }) => {
    // ... 個別のダイジェストを送信
  }
);

Idempotency を使用したウェブフックハンドラー

重複排除でウェブフックを安全に処理

いつ使うか: Stripe、GitHub、またはその他のウェブフックを処理する場合

export const handleStripeWebhook = inngest.createFunction(
  {
    id: 'stripe-webhook',
    // Stripe イベント ID で重複排除
    idempotency: 'event.data.stripeEventId',
  },
  { event: 'stripe/webhook.received' },
  async ({ event, step }) => {
    const { type, data } = event.data;

    switch (type) {
      case 'checkout.session.completed':
        await step.run('fulfill-order', async () => {
          await fulfillOrder(data.session.id);
        });
        break;

      case 'customer.subscription.deleted':
        await step.run('cancel-subscription', async () => {
          await cancelSubscription(data.subscription.id);
        });
        break;
    }
  }
);

ロング処理を伴う AI パイプライン

チャンク化された作業を使用したマルチステップ AI 処理

いつ使うか: 数分かかる AI ワークフロー

export const processDocument = inngest.createFunction(
  {
    id: 'process-document',
    retries: 2,
    concurrency: { limit: 5 },  // API 使用量を制限
  },
  { event: 'document/uploaded' },
  async ({ event, step }) => {
    // ステップ 1: テキストを抽出 (時間がかかる可能性)
    const text = await step.run('extract-text', async () => {
      return await extractTextFromPDF(event.data.fileUrl);
    });

    // ステップ 2: 埋め込み用にチャンク化
    const chunks = await step.run('chunk-text', async () => {
      return chunkText(text, { maxTokens: 500 });
    });

    // ステップ 3: 埋め込みを生成 (API レート制限)
    const embeddings = await step.run('generate-embeddings', async () => {
      return await openai.embeddings.create({
        model: 'text-embedding-3-small',
        input: chunks,
      });
    });

    // ステップ 4: ベクターDB に保存
    await step.run('store-vectors', async () => {
      await vectorDb.upsert({
        vectors: embeddings.data.map((e, i) => ({
          id: `${event.data.documentId}-${i}`,
          values: e.embedding,
          metadata: { chunk: chunks[i] },
        })),
      });
    });

    return { chunks: chunks.length, status: 'indexed' };
  }
);

検証チェック

Inngest serve ハンドラーが存在

重要度: CRITICAL

メッセージ: Inngest がイベントを受信するには serve ハンドラーが必要

修正アクション: serve() エクスポート付きで app/api/inngest/route.ts を作成

関数が serve に登録されている

重要度: ERROR

メッセージ: すべての Inngest 関数が serve() 呼び出しで登録されていることを確認

修正アクション: 関数を serve() の functions 配列に追加

Step.run に説明的な名前が付いている

重要度: WARNING

メッセージ: ステップ名は kebab-case で説明的であるべき

修正アクション: 'fetch-user' または 'send-email' などの説明的なステップ名を使用

waitForEvent がタイムアウトを持つ

重要度: ERROR

メッセージ: waitForEvent は無限待機を防ぐためにタイムアウトが必要

修正アクション: タイムアウトオプションを追加: { timeout: '24h' }

関数がコンカレンシー制限を持つ

重要度: WARNING

メッセージ: ダウンストリームサービスを保護するためにコンカレンシー制限の追加を検討

修正アクション: 関数設定に concurrency: { limit: 10 } を追加

イベント型が定義されている

重要度: WARNING

メッセージ: Inngest クライアントはタイプ安全性のためにイベントスキーマを定義すべき

修正アクション: 追加: schemas: new EventSchemas().fromRecord<Events>()

関数がユニークな ID を持つ

重要度: CRITICAL

メッセージ: すべての Inngest 関数は一意の ID を持つ必要

修正アクション: 関数設定に id: 'my-function-name' を追加

スリープが期間文字列を使用

重要度: WARNING

メッセージ: step.sleep は ' 1h' または '30m' などの期間文字列を使用すべき、ミリ秒ではなく

修正アクション: 期間文字列を使用: step.sleep('wait', '1h')

リトライポリシーが設定されている

重要度: WARNING

メッセージ: 失敗処理のためにリトライポリシー設定を検討

修正アクション: retries: 3 または retries: { attempts: 3, backoff: { ... } } を追加

支払い関数の Idempotency キー

重要度: ERROR

メッセージ: 支払い関連の関数は idempotency キーを使用すべき

修正アクション: 関数設定に idempotency: 'event.data.orderId' を追加

コラボレーション

デリゲーショントリガー

• redis|queue infrastructure|bullmq -> bullmq-specialist (既存のインフラを使用した Redis ベースのキューが必要)
• saga|compensation|rollback|long-running workflow -> temporal-craftsman (補償を伴う複雑なワークフロー調整が必要)
• event sourcing|event store|cqrs -> event-architect (イベントソーシングパターンが必要)
• vercel|deploy|production -> vercel-deployment (デプロイメント設定が必要)
• database|schema|data model -> supabase-backend (イベントデータ用のデータベースが必要)
• api|endpoint|route -> backend (イベントをトリガーするための API が必要)

Vercel Background Jobs

スキル: inngest, nextjs-app-router, vercel-deployment

ワークフロー:

1. Inngest 関数を定義 (inngest)
2. Next.js で serve ハンドラーをセットアップ (nextjs-app-router)
3. 関数タイムアウトを設定 (vercel-deployment)
4. デプロイしてテスト (vercel-deployment)

AI パイプライン

スキル: inngest, ai-agents-architect, supabase-backend

ワークフロー:

1. AI ワークフローステップを設計 (ai-agents-architect)
2. Inngest の耐久性で実装 (inngest)
3. 結果をデータベースに保存 (supabase-backend)
4. API 失敗のリトライを処理 (inngest)

ウェブフック処理

スキル: inngest, stripe-integration, backend

ワークフロー:

1. ウェブフックを受信 (backend)
2. Idempotency で Inngest に送信 (inngest)
3. 支払いロジックを処理 (stripe-integration)
4. アプリケーション状態を更新 (backend)

メール自動化

スキル: inngest, email-systems, supabase-backend

ワークフロー:

1. ユーザーアクションからイベントをトリガー (inngest)
2. step.sleep でドリップメールをスケジュール (inngest)
3. リトライ付きでメールを送信 (email-systems)
4. メール状態を追跡 (supabase-backend)

スケジュール済みタスク

スキル: inngest, backend, analytics-architecture

ワークフロー:

1. Cron トリガーを定義 (inngest)
2. 処理ロジックを実装 (backend)
3. データを集約してレポート (analytics-architecture)
4. アラート付きで障害を処理 (inngest)

関連スキル

相性の良いスキル: nextjs-app-router, vercel-deployment, supabase-backend, email-systems, ai-agents-architect, stripe-integration

いつ使うか

• ユーザーが言及または暗示: inngest
• ユーザーが言及または暗示: serverless background job
• ユーザーが言及または暗示: event-driven workflow
• ユーザーが言及または暗示: step function
• ユーザーが言及または暗示: durable execution
• ユーザーが言及または暗示: vercel background job
• ユーザーが言及または暗示: scheduled function
• ユーザーが言及または暗示: fan out

制限事項

• このスキルは、タスクが上記で説明されたスコープと明確に一致する場合にのみ使用
• 出力を環境固有の検証、テスト、またはエキスパートレビューの代替物として扱わない
• 必要な入力、権限、安全性の境界、または成功基準が不足している場合は、説明を求めて停止