inngest-realtime

Inngest Realtime

将持久化 Inngest 函数的更新实时流式传输到 UI。使用频道和主题广播进度、实时呈现工作流执行过程，或构建双向的人机协作流程。

这些技能专注于 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 开发服务器正在运行（npx inngest-cli@latest dev）
• 可选：使用 zod 对主题进行架构验证

何时使用 Realtime

问题场景	模式
订单状态页面随持久化工作流步骤完成而动态更新	每次运行使用一个频道，按步骤发布，客户端订阅
AI Agent 向聊天 UI 流式传输令牌	每次对话使用一个频道，发布分块内容，流式传输到浏览器
长时间运行的任务的日志尾部	单个频道，日志主题，将日志追加到 UI
人机协作审批	频道 + waitForEvent，发布提示词，等待响应
显示实时订单列表的管理后台	全局管理频道，从每个函数扇出

架构

由三部分组成：

1. 频道定义 — 已发布内容的类型化约定。放在共享模块中，以便服务端和客户端都能引用同一个频道名称。
2. 发布 — 在步骤之间调用 step.realtime.publish，以封装持久化发布；或者在 step.run 中调用 inngest.realtime.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(),
      }),
    },
  },
});

两种频道名称形式：

• name: 'admin' — 静态频道，通过 adminChannel.order（主题引用）访问
• name: (id) => 'channel:${id}' — 参数化频道，通过 orderChannel(id).step 访问（使用 id 调用频道定义，然后访问主题）

步骤 2：在函数内部发布

Inngest v4 原生支持 realtime——无需中间件。 但 publish 的调用位置很重要：它决定发布是否持久化，也是 realtime 最容易出错的地方。

应使用哪种发布方法

所在位置	使用此方法	原因
步骤外（顶层处理程序代码、step.run 调用之间）	step.realtime.publish(id, topicRef, data)	会将发布封装在独立步骤中，因此具有持久性、可通过 id 去重，并且可安全重试。
步骤内（传递给 step.run 的回调内部）	inngest.realtime.publish(topicRef, data)	你已经在一个带记忆的步骤中。step.realtime.publish 会在步骤内再创建一个步骤。此处应调用不带 step 的客户端发布方法。
函数外（一次性路由、脚本等）	inngest.realtime.publish(topicRef, data)	允许这样做，但不可安全重试——客户端接收方必须处理重复消息。

90% 的情况适用以下规则：如果你在编写处理程序代码，并准备调用 publish，请使用 step.realtime.publish。如果你在 step.run 代码块内编写代码，并准备调用 publish，请使用 inngest.realtime.publish。

示例：在一个函数中同时使用这两种模式

// 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 Hook 使用。没有令牌，客户端就无法订阅。

// 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 Hook 直接使用。无需剥离 ChannelInstance——这个注意事项仅适用于较低层级的 getSubscriptionToken + 手动 subscribe() 路径（请参阅下文的“模式：手动订阅”）。

第 4 步：使用 useRealtime Hook 订阅

React/Next.js 推荐使用者是来自 inngest/react 的 useRealtime Hook。它负责处理订阅生命周期、重新连接、按主题进行类型收窄，以及清理工作。

// 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>
  );
}

Hook 的实用选项：

选项	默认值	使用场景
enabled	true	在获得 ID 之前延迟订阅（例如，enabled: !!runId）。
bufferInterval	0	对快速数据流的更新进行批处理，避免 React 每收到一条消息就重新渲染。
pauseOnHidden	false	标签页不可见时暂停数据流（节省带宽）。
autoCloseOnTerminal	true	运行完成时断开连接——关闭此选项可使数据流为扇出通道保持打开。
historyLimit	无限制	限制 messages.all 中保留的消息数量。

该 Hook 会返回 messages.byTopic（每个主题的最新消息）、messages.all（完整历史记录）、messages.last（最新消息）和 messages.delta（自上次渲染后新增的消息）。

模式：手动订阅（非 React 或自定义传输）

useRealtime Hook 已涵盖 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',
    },
  });
}

客户端通过 fetch().getReader() 消费数据，而不是使用 subscribe() 回调。需要 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 中。混用会导致：

• TypeError: Cls is not a constructor on PUT /api/inngest（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 时无需此操作。