inngest-realtime

Inngest Realtime

耐久性のある Inngest 関数からライブ UI へ更新をストリーミングします。チャンネルとトピックを使って進捗をブロードキャストしたり、ワークフローの実行をリアルタイムに描画したり、双方向の human-in-the-loop フローを構築したりできます。

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

⚠ 重要: v3 と v4 のパッケージ選択

Inngest v4 の Realtime は SDK のサブパス inngest/realtime にあります。単体の @inngest/realtime npm パッケージは v3 時代のパッケージであり、[email protected] とは互換性がありません。プロジェクトが v4（npm のデフォルト）を使用している場合は、@inngest/realtime をインストールしないでください。以下に示すインポートを使用してください。

v4 で誤ったパッケージを使用した場合の症状: すべての PUT /api/inngest で TypeError: Cls is not a constructor、サブスクリプショントークンで 401、new Inngest({ middleware: [...] }) での型の非互換。先に進む前に、package.json に "inngest": "^4.x" が記載されていることを確認してください。

前提条件

• Inngest v4 SDK がインストールされていること（npm install inngest） — inngest-setup スキルを参照してください
• ローカル開発用に .env.local で INNGEST_DEV=1 が設定されていること（これがないと、SDK はクラウド署名キーを要求し、トークンリクエストで 401 になります）
• ローカルの Inngest dev サーバーが実行中であること（npx inngest-cli@latest dev）
• オプション: トピックのスキーマ検証用の zod

Realtime を使うべき場面

問題の形	パターン
耐久性のあるワークフローのステップが完了するにつれて注文ステータスページがアニメーションする	実行ごとのチャンネル、ステップごとに publish、クライアントがサブスクライブ
AI エージェントがトークンをチャット UI にストリーミングする	会話ごとのチャンネル、チャンクを publish、ブラウザにストリーミング
長時間実行ジョブのログ追尾	単一チャンネル、ログトピック、UI に追記
Human-in-the-loop の承認	チャンネル + waitForEvent、プロンプトを publish、応答を待機
ライブ注文リストを備えた管理ダッシュボード	グローバル管理チャンネル、各関数からのファンアウト

アーキテクチャ

3 つの要素があります:

1. チャンネル定義 — 何が publish されるかについての型付きコントラクト。サーバーとクライアントの両方が同じチャンネル名を参照できるよう、共有モジュールに配置します。
2. Publish — すでにメモ化されたステップの内部にいるため、耐久性のある publish をラップするにはステップ間で step.realtime.publish を呼び出すか、step.run 内で inngest.realtime.publish を呼び出します。以下の「どの publish メソッドを使うか」を参照してください。
3. サブスクライブ — サーバーアクションがサブスクリプショントークンを発行し、React クライアントが useRealtime フック（または React 以外のコンシューマー向けの低レベル subscribe() API）を使用します。

ステップ 1: チャンネルを定義する

チャンネルは純粋なデータです — クラス階層も、zod ランタイムも必須ではありません（ただし型安全性のために推奨します）。一度定義して、必要な場所でインポートしてください。

// src/inngest/channels.ts
import { channel } from 'inngest/realtime';
import { z } from 'zod';

// Per-run channel: each fulfill-order run publishes step updates to its own channel.
export const orderChannel = channel({
  name: (orderId: string) => `order:${orderId}`,
  topics: {
    step: {
      schema: z.object({
        name: z.string(),
        status: z.enum(['running', 'complete', 'failed']),
        output: z.record(z.string(), z.unknown()).optional(),
        ts: z.number(),
      }),
    },
  },
});

// Global admin channel: fan-out for cross-cutting visibility.
export const adminChannel = channel({
  name: 'admin',
  topics: {
    order: {
      schema: z.object({
        orderId: z.string(),
        step: z.string(),
        status: z.enum(['running', 'complete', 'failed']),
        ts: z.number(),
      }),
    },
  },
});

2 つのチャンネル名の形:

• name: 'admin' — 静的チャンネル、adminChannel.order（トピック参照）としてアクセス
• name: (id) => 'channel:${id}' — パラメトリック、orderChannel(id).step としてアクセス（チャンネル定義を id で呼び出し、次にトピックにアクセス）

ステップ 2: 関数の内部から publish する

Inngest v4 は realtime サポートをネイティブに提供します — ミドルウェアは不要です。 ただし、publish をどこで呼び出すかが重要です。それは publish が耐久性を持つかどうかを決定し、realtime で最もよく間違えるポイントです。

どの公開メソッドを使うべきか

現在の場所	使用するもの	理由
ステップの外（トップレベルのハンドラーコード、step.run 呼び出しの間）	step.realtime.publish(id, topicRef, data)	公開処理を独自のステップでラップするため、永続的で、id によって重複排除され、リトライしても安全です。
ステップの内側（step.run に渡されたコールバック内）	inngest.realtime.publish(topicRef, data)	すでにメモ化されたステップの内側にいます。step.realtime.publish はステップの中にステップを作成してしまいます。ここでは素のクライアント公開が正しい選択です。
関数の外（単発のルート、スクリプトなど）	inngest.realtime.publish(topicRef, data)	許可されていますが、リトライしても安全ではありません — クライアントの受信側で重複を処理する必要があります。

90% のルール: ハンドラーコードを書いていて publish を使いたくなったら、step.realtime.publish を使ってください。step.run ブロックの内側のコードを書いていて publish を使いたくなったら、inngest.realtime.publish を使ってください。

例: 1 つの関数で両方のパターン

// src/inngest/functions/fulfill-order.ts
import { inngest } from '../client';
import { orderChannel, adminChannel } from '../channels';

export const fulfillOrder = inngest.createFunction(
  {
    id: 'fulfill-order',
    retries: 3,
    triggers: [{ event: 'store/order.placed' }],
  },
  async ({ event, step }) => {
    const { orderId, customerEmail, lineItems } = event.data;

    // Outside any step.run — use step.realtime.publish for a durable wrapper.
    const emit = async (
      name: string,
      status: 'running' | 'complete' | 'failed',
      output?: Record<string, unknown>,
    ) => {
      const ts = Date.now();
      await step.realtime.publish(
        `emit-order-${name}-${status}`,
        orderChannel(orderId).step,
        { name, status, output, ts },
      );
      await step.realtime.publish(
        `emit-admin-${name}-${status}`,
        adminChannel.order,
        { orderId, step: name, status, ts },
      );
    };

    await emit('capture-payment', 'running');

    // Inside step.run — use inngest.realtime.publish (already in a memoized step).
    const payment = await step.run('capture-payment', async () => {
      const intent = await stripe.paymentIntents.create({ /* ... */ });

      // Stream a partial update mid-step. No step-in-step wrapping needed.
      await inngest.realtime.publish(orderChannel(orderId).step, {
        name: 'capture-payment',
        status: 'running',
        output: { stage: 'intent-created', intentId: intent.id },
        ts: Date.now(),
      });

      return await stripe.paymentIntents.confirm(intent.id);
    });

    await emit('capture-payment', 'complete', payment);

    await emit('reserve-inventory', 'running');
    const inventory = await step.run('reserve-inventory', async () => {
      // ...
    });
    await emit('reserve-inventory', 'complete', inventory);

    // ...
  },
);

ミドルウェアが不要な理由: 以前のバージョンでは @inngest/realtime の realtimeMiddleware() を使ってハンドラーに publish 引数を注入していました。v4 ではこれを step.realtime と inngest.realtime に直接配置しています。

ステップ 3: サブスクリプショントークンを発行する（サーバーアクション）

Next.js App Router では、サーバーアクションを使って、ステップ 4 の React フック用の短命トークンを安全に発行します。トークンがないと、クライアントはサブスクライブできません。

// src/app/orders/[orderId]/actions.ts
'use server';

import { getClientSubscriptionToken } from 'inngest/react';
import { inngest } from '@/inngest/client';
import { orderChannel } from '@/inngest/channels';

export async function fetchOrderSubscriptionToken(orderId: string) {
  // ⚠ AUTHORIZATION GATE: verify the current user owns this orderId
  // before minting a token. Channels are addressable by ID, so without
  // an ownership check, anyone can subscribe to any order's stream by
  // guessing IDs.
  //
  //   const session = await getServerSession();
  //   if (!session) throw new Error('Unauthenticated');
  //   const order = await db.order.findUnique({ where: { id: orderId } });
  //   if (order?.userId !== session.userId) throw new Error('Forbidden');

  return getClientSubscriptionToken(inngest, {
    channel: orderChannel(orderId),
    topics: ['step'],
  });
}

inngest/react の getClientSubscriptionToken は、ステップ 4 の useRealtime フックがそのまま消费するトークンの形を返します。ChannelInstance の除去は不要です — この注意点は、より低レベルの getSubscriptionToken + 手動 subscribe() のパスにのみ当てはまります（後述の「パターン: 手動サブスクライブ」を参照）。

ステップ 4: useRealtime フックでサブスクライブする

React/Next.js で推奨されるコンシューマーは、inngest/react の useRealtime フックです。これはサブスクリプションのライフサイクル、再接続、トピックごとの型の絞り込み、クリーンアップを処理します。

// src/components/OrderStatusClient.tsx
'use client';

import { useRealtime } from 'inngest/react';
import { orderChannel } from '@/inngest/channels';
import { fetchOrderSubscriptionToken } from '@/app/orders/[orderId]/actions';

export function OrderStatusClient({ orderId }: { orderId: string }) {
  const { messages, connectionStatus, error } = useRealtime({
    channel: orderChannel(orderId),
    topics: ['step'] as const,
    token: () => fetchOrderSubscriptionToken(orderId),
  });

  if (error) return <div>Error: {error.message}</div>;

  return (
    <div>
      <div>Status: {connectionStatus}</div>
      <ul>
        {messages.all.map((m, i) => (
          <li key={i}>
            {(m.data as { name: string }).name}: {(m.data as { status: string }).status}
          </li>
        ))}
      </ul>
    </div>
  );
}

フックの便利なオプション:

オプション	デフォルト	使うべきとき
enabled	true	ID を取得するまでサブスクリプションを遅延させます（例: enabled: !!runId）。
bufferInterval	0	高速なストリームからの更新をバッチ処理し、React がメッセージごとに再レンダリングしないようにします。
pauseOnHidden	false	タブが表示されていないときにストリームを一時停止します（帯域幅を節約）。
autoCloseOnTerminal	true	実行が完了したら接続を切断します — ファンアウトチャネルでストリームを開いたままにするにはオフにします。
historyLimit	無制限	messages.all に保持されるメッセージの数を制限します。

フックは messages.byTopic（トピックごとの最新）、messages.all（全履歴）、messages.last（最新）、messages.delta（前回のレンダリング以降の新規）を返します。

パターン: 手動サブスクライブ（非 React またはカスタムトランスポート）

useRealtime フックは React の場合をカバーします。React を使っていない場合、またはカスタムのサブスクリプションライフサイクル（サーバーサイドストリーミング、バックグラウンドワーカー、カスタムプロトコル）が必要な場合は、より低レベルの subscribe() API を直接使用してください。

サーバーアクション: 低レベルヘルパーでトークンを発行する

// src/app/orders/[orderId]/actions.ts
'use server';

import { getSubscriptionToken } from 'inngest/realtime';
import { inngest } from '@/inngest/client';
import { orderChannel } from '@/inngest/channels';

export async function fetchOrderSubscriptionTokenLowLevel(orderId: string) {
  // ⚠ AUTHORIZATION GATE: same as Step 3 — verify ownership before minting.

  const token = await getSubscriptionToken(inngest, {
    channel: orderChannel(orderId),
    topics: ['step'],
  });

  // ⚠ CRITICAL: strip the ChannelInstance from the response.
  // getSubscriptionToken returns { channel: ChannelInstance, ... } where
  // ChannelInstance contains zod schema methods (a class with prototypes).
  // Next.js refuses to serialize classes across the server-action → client-component
  // boundary, so return ONLY primitives.
  return {
    channel: orderChannel(orderId).name as string,
    topics: ['step'] as const,
    key: token.key,
    apiBaseUrl: token.apiBaseUrl,
  };
}

手動クライアントサブスクリプション

// src/components/OrderStatusManual.tsx
'use client';

import * as React from 'react';
import { subscribe } from 'inngest/realtime';
import { fetchOrderSubscriptionTokenLowLevel } from '@/app/orders/[orderId]/actions';

export function OrderStatusManual({ orderId }: { orderId: string }) {
  const [messages, setMessages] = React.useState<unknown[]>([]);

  React.useEffect(() => {
    let cancelled = false;
    let sub: { close?: (reason?: string) => void } | undefined;

    (async () => {
      const token = await fetchOrderSubscriptionTokenLowLevel(orderId);
      if (cancelled) return;

      sub = await subscribe(
        {
          channel: token.channel,
          topics: [...token.topics],
          key: token.key,
          apiBaseUrl: token.apiBaseUrl,
        },
        (message) => {
          if (cancelled) return;
          setMessages((prev) => [...prev, message.data]);
        },
      );
    })();

    return () => {
      cancelled = true;
      sub?.close?.('unmount');
    };
  }, [orderId]);

  // ... render ...
}

ルートハンドラーからの SSE ストリーミング

Next.js API ルート内でサブスクライブし、SSE 経由でストリームをクライアントにパイプします:

// src/app/api/orders/[orderId]/stream/route.ts
import { inngest } from '@/inngest/client';
import { subscribe } from 'inngest/realtime';
import { orderChannel } from '@/inngest/channels';

export async function GET(req: Request, { params }: { params: { orderId: string } }) {
  // ⚠ AUTHORIZATION GATE: same rule as the server-action token mint above.
  // Authenticate the request and confirm the caller owns params.orderId
  // before opening the SSE stream. Skipping this leaks every order's
  // step events to anyone with a URL.

  const stream = await subscribe({
    app: inngest,
    channel: orderChannel(params.orderId),
    topics: ['step'],
  });

  return new Response(stream.getEncodedStream(), {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      Connection: 'keep-alive',
    },
  });
}

クライアントは subscribe() コールバックではなく fetch().getReader() 経由で消费します。SSE の挙動が必要な場合や、クライアント側の subscribe() API がコンポーネントのライフサイクルに合わない場合にこれを使用してください。

パターン: ヒューマン・イン・ザ・ループ

step.realtime.publish と step.waitForEvent を組み合わせます:

import crypto from 'crypto';

export const reviewWorkflow = inngest.createFunction(
  { id: 'review-workflow', triggers: [{ event: 'review/start' }] },
  async ({ event, step }) => {
    const confirmationId = await step.run('gen-id', () => crypto.randomUUID());

    // Publish a prompt — the client subscribes and renders an approval UI
    await step.realtime.publish(
      'publish-prompt',
      reviewChannel.message,
      { message: 'Confirm to proceed?', confirmationId },
    );

    // Wait up to 15 minutes for the user to send the matching event back
    const confirmation = await step.waitForEvent('await-confirmation', {
      event: 'review/confirmation',
      timeout: '15m',
      if: `async.data.confirmationId == "${confirmationId}"`,
    });

    if (!confirmation) {
      // user didn't respond — abort or escalate
      return { decision: 'timed_out' };
    }
    // continue workflow...
  },
);

confirmationId は、発行されたプロンプトと一致する返信をリンクするため、ワークフローはどのレスポンスに基づいて行動すべきかを把握できます。

よくある落とし穴

v4 では @inngest/realtime を使用しないでください

スタンドアロンの @inngest/realtime パッケージは Inngest v3 専用です。v4 では、すべてのリアルタイム API は SDK のサブパス inngest/realtime にあります。これらを混在させると、以下が発生します:

• PUT /api/inngest での TypeError: Cls is not a constructor (v3 ミドルウェアクラスのシグネチャ不一致)
• サブスクリプショントークンでの 401 Unauthorized
• ミドルウェアのキャスト時の TypeScript エラー

確認方法: grep '"inngest"' package.json — もし ^4.x であれば、inngest/realtime を使用してください。以上です。

Next.js のサーバーアクションから ChannelInstance を返さないでください (手動サブスクライブパスのみ)

getSubscriptionToken は { channel: ChannelInstance, ... } を返しますが、ここで ChannelInstance は zod スキーマメソッド (クラス) を持ちます。Next.js はサーバーアクション → クライアントコンポーネントの境界を越えてクラスをシリアライズすることを拒否します。返す前にプリミティブに変換してください。上記の「パターン: 手動サブスクライブ」を参照してください。

この注意点は、inngest/react の getClientSubscriptionToken (ステップ 3 — 推奨パス) を使用する場合には該当しません。このヘルパーはシリアライズ安全な形を直接返します。

ローカル開発には INNGEST_DEV=1 が必要です

これがないと、SDK はクラウドモードを想定し、INNGEST_SIGNING_KEY と INNGEST_EVENT_KEY を要求します。すべてのリアルタイム操作が 401 / 500 になります。.env.local に追加してください。開発サーバーをハード再起動してください (Next.js は .env.local の変更をホットリロードしません)。

チャネルトピックのスキーマはパブリッシュ時に検証され、コンシューム時には検証されません

パブリッシュしたペイロードが zod スキーマに一致しない場合、パブリッシュはサーバー側で失敗します。サブスクライバーは何も受信しません。ステップ実行中にパブリッシュのエラーを捕捉するか、クライアント側でスキーマ検証をスキップする理由がある場合は subscribe() で validate: false を指定して実行してください。

リファレンス

• v4 のエントリポイント:
  • import { channel } from 'inngest/realtime' — チャネル定義
  • import { useRealtime, getClientSubscriptionToken } from 'inngest/react' — React フック + 対応するトークンヘルパー (ステップ 3 + ステップ 4)
  • import { getSubscriptionToken, subscribe } from 'inngest/realtime' — 非 React またはカスタムトランスポート向けの低レベルヘルパー
• パブリッシュメソッド:
  • ステップの外: step.realtime.publish(id, topicRef, data) — 永続ステップでラップします
  • step.run の中: inngest.realtime.publish(topicRef, data) — すでにメモ化されたステップの中なので、ラップは不要です
  • 関数の外: inngest.realtime.publish(topicRef, data) — 許可されていますが、リトライセーフではありません
• サブスクライブのオーバーロード: subscribe(token) はストリームを返します; subscribe(token, callback) はメッセージごとにコールバックを呼び出します
• Next.js サーバーアクションの注意点 (手動パスのみ): ChannelInstance を除去 → { channel: string, topics, key, apiBaseUrl } を返します。getClientSubscriptionToken では不要です。