React hooks / Next.js

Info: In v4, React subscriptions use useRealtime from inngest/react together with getClientSubscriptionToken() on the server.

Warning: Need the old useInngestSubscription() docs? See the archived v3 docs here.

useRealtime manages token refresh, reconnection, buffering, pause behavior, and typed message access per topic. The typical Next.js flow is:

Define a shared channel with realtime.channel().
Mint a scoped token in a server action or route handler.
Pass the channel, topics, and token factory into useRealtime().

Next.js example

tsx

"use server";

import { getClientSubscriptionToken } from "inngest/react";
import { inngest } from "@/inngest/client";
import { helloChannel } from "@/inngest/channels";

export async function fetchRealtimeSubscriptionToken(runId: string) {
  return getClientSubscriptionToken(inngest, {
    channel: helloChannel({ runId }),
    topics: ["logs"],
  });
}

tsx

"use client";

import { useRealtime } from "inngest/react";
import { helloChannel } from "@/inngest/channels";
import { fetchRealtimeSubscriptionToken } from "./actions";

export default function Home({ runId }: { runId: string }) {
  const topics = ["logs"] as const;
  const channel = helloChannel({ runId });

  const {
    connectionStatus,
    runStatus,
    messages,
    error,
    reset,
  } = useRealtime({
    channel,
    topics,
    token: () => fetchRealtimeSubscriptionToken(runId),
    bufferInterval: 100,
  });

  return (
    <div>
      <p>Connection: {connectionStatus}</p>
      <p>Run: {runStatus}</p>
      <p>Latest log: {messages.byTopic.logs?.data}</p>
      <p>Buffered messages: {messages.delta.length}</p>
      {error && <p>{error.message}</p>}
      <button onClick={() => reset()}>Reset</button>
    </div>
  );
}

What `useRealtime` gives you

connectionStatus for websocket lifecycle state
runStatus for function lifecycle state
messages.byTopic for the latest typed message per topic
messages.all for retained message history
messages.last and messages.delta for incremental UI updates
error and reset() for operational handling

Common options

enabled to delay the connection until you have a runId
bufferInterval to batch UI updates for fast streams
historyLimit to cap retained messages
pauseOnHidden to pause when the tab is hidden
autoCloseOnTerminal to disconnect when the run finishes

Typed topic access

When you pass a channel instance and a literal topics array, TypeScript narrows message payloads per topic:

tsx

const topics = ["status", "artifact"] as const;

const { messages } = useRealtime({
  channel,
  topics,
  token: () => fetchRealtimeSubscriptionToken(runId),
});

messages.byTopic.status?.data.message;
messages.byTopic.artifact?.data.title;

for (const message of messages.delta) {
  if (message.kind === "run") continue;

  if (message.topic === "status") {
    message.data.message;
  }
}