useRealtime

The useRealtime hook subscribes to realtime messages from Inngest functions in React components. It manages the WebSocket connection, reconnection, buffering, and provides typed access to messages by topic.

import { useRealtime } from "inngest/react";
import { pipelineChannel } from "../inngest/channels";

function Pipeline({ runId }: { runId: string }) {
  const ch = pipelineChannel({ runId });
  const topics = ["status", "tokens"] as const;

  const { connectionStatus, runStatus, messages, result } = useRealtime({
    channel: ch,
    topics,
    token: () =>
      fetch(`/api/realtime-token?runId=${runId}`).then((r) => r.json()),
  });

  return (
    <div>
      <p>Connection: {connectionStatus} | Run: {runStatus}</p>
      {messages.byTopic.status && (
        <p>{messages.byTopic.status.data.message}</p>
      )}
      <p>Messages received: {messages.all.length}</p>
      {result && <pre>{JSON.stringify(result, null, 2)}</pre>}
    </div>
  );
}

---

useRealtime(options)

• channel (ChannelInstance | string): The channel to subscribe to. Can be a channel instance from realtime.channel() or a plain string. When you use a channel instance, topic data is typed automatically.

• topics (string[]): The topics to subscribe to within the channel. Required when your token is a ClientSubscriptionToken from getClientSubscriptionToken() or when your token factory returns only a token key string.

• token (ClientSubscriptionToken | Token | () => Promise<string | ClientSubscriptionToken | Token>): Authentication for the subscription. Pass a ClientSubscriptionToken from getClientSubscriptionToken(), a full token object, or an async factory that returns a token key string, ClientSubscriptionToken, or full token object. The async factory pattern is recommended because it runs on mount and on reconnect.

• key (string): Optional subscription identity key. Change it to force the hook to reset its retained state and reconnect even if channel and topics are unchanged.

• enabled (boolean): Whether the subscription is active. Set to false to pause without unmounting. Defaults to true.

• validate (boolean): Enable subscriber-side schema validation on incoming messages. Defaults to true.

• historyLimit (number | null): Maximum number of messages to retain in messages.all. Set to null for unbounded history. Defaults to 100.

• bufferInterval (number): Milliseconds to buffer incoming messages before triggering a re-render. Useful for high-frequency streams to reduce render pressure. Defaults to 0 (immediate).

• reconnect (boolean): Automatically reconnect on disconnect. Defaults to true.

• reconnectMinMs (number): Minimum delay between reconnect attempts in milliseconds. Defaults to 250.

• reconnectMaxMs (number): Maximum delay between reconnect attempts in milliseconds (exponential backoff cap). Defaults to 5000.

• pauseOnHidden (boolean): Pause the subscription when the browser tab is hidden, resuming when it becomes visible. Defaults to true.

• autoCloseOnTerminal (boolean): Automatically close the subscription when the function run reaches a terminal status (completed, failed, or cancelled). Defaults to true.

Return value

• connectionStatus (`"idle" | "connecting" | "open" | "paused" | "closed" | "error"`): The WebSocket connection status.

• runStatus (`"unknown" | "running" | "completed" | "failed" | "cancelled"`): The lifecycle status of the Inngest function run. Updated from run-level messages on the channel.

• isPaused (boolean): Convenience boolean for connectionStatus === "paused".

• pauseReason (`"hidden" | "disabled" | null`): Why the hook is paused. hidden means the document is hidden and pauseOnHidden is enabled. disabled means enabled is false.

• messages.byTopic (Record<string, Message | undefined>): The latest message per subscribed topic. Access typed data with messages.byTopic.topicName?.data.

• messages.all (Message[]): All retained messages in chronological order, bounded by historyLimit.

• messages.last (Message | null): The most recently flushed message across all subscribed topics.

• messages.delta (Message[]): The newest batch of flushed messages. When buffering is disabled, this is a single-message array for the latest message.

• error (Error | null): The most recent connection error, or null if connected successfully.

• result (unknown): The function's return value, extracted from the terminal run message when the function completes.

• reset (() => void): Clears the retained messages, result, and error state and resets the hook back to its initial state.

Connection status

The connectionStatus field tracks the WebSocket connection lifecycle:

Status	Description
idle	Hook is mounted but hasn't started connecting (e.g., enabled: false)
connecting	Establishing the WebSocket connection
open	Connected and receiving messages
paused	The hook is intentionally paused because enabled is false or the tab is hidden
closed	Connection closed (manually or by autoCloseOnTerminal)
error	Connection failed. Check error for details

Run status

The runStatus field tracks the Inngest function's execution lifecycle:

Status	Description
unknown	No run status received yet
running	Function is actively executing
completed	Function finished successfully. result is available
failed	Function failed after exhausting retries
cancelled	Function was cancelled

When autoCloseOnTerminal is true (the default), the subscription closes automatically once runStatus reaches completed, failed, or cancelled.

Token factory pattern

The recommended approach is to pass an async factory function for token. This function is called when the hook mounts and on each reconnect, ensuring fresh tokens.

// app/actions.ts
"use server";
import { getClientSubscriptionToken } from "inngest/react";
import { inngest } from "../inngest/client";
import { pipelineChannel } from "../inngest/channels";

export async function getToken(runId: string) {
  return getClientSubscriptionToken(inngest, {
    channel: pipelineChannel({ runId }),
    topics: ["status", "tokens"],
  });
}

// app/page.tsx
"use client";
import { useRealtime } from "inngest/react";
import { pipelineChannel } from "../inngest/channels";
import { getToken } from "./actions";

function Pipeline({ runId }: { runId: string }) {
  const topics = ["status", "tokens"] as const;
  const { connectionStatus, messages } = useRealtime({
    channel: pipelineChannel({ runId }),
    topics,
    token: () => getToken(runId),
  });

  return (
    <p>
      {connectionStatus}: {messages.byTopic.status?.data.message}
    </p>
  );
}

// app/api/realtime-token/route.ts
import { getClientSubscriptionToken } from "inngest/react";
import { inngest } from "../../inngest/client";
import { pipelineChannel } from "../../inngest/channels";

export async function GET(req: Request) {
  const { searchParams } = new URL(req.url);
  const runId = searchParams.get("runId")!;

  const token = await getClientSubscriptionToken(inngest, {
    channel: pipelineChannel({ runId }),
    topics: ["status", "tokens"],
  });

  return Response.json(token);
}

// app/page.tsx
"use client";
import { useRealtime } from "inngest/react";
import { pipelineChannel } from "../inngest/channels";

function Pipeline({ runId }: { runId: string }) {
  const topics = ["status", "tokens"] as const;
  const { messages } = useRealtime({
    channel: pipelineChannel({ runId }),
    topics,
    token: () =>
      fetch(`/api/realtime-token?runId=${runId}`).then((r) => r.json()),
  });

  return <p>{messages.byTopic.status?.data.message}</p>;
}

Pre-minted loader tokens

If your framework passes data from a server loader into a client component, you can pass the ClientSubscriptionToken from getClientSubscriptionToken() directly. Keep channel and topics as top-level useRealtime options; the client token intentionally contains only key and apiBaseUrl, and the top-level channel and topics are what preserve typed message inference.

import { useLoaderData } from "react-router";
import { useRealtime } from "inngest/react";
import { pipelineChannel } from "../inngest/channels";
import type { loader } from "./route";

export function Pipeline() {
  const { runId, realtimeToken } = useLoaderData<typeof loader>();
  const channel = pipelineChannel({ runId });
  const topics = ["status", "tokens"] as const;

  const { connectionStatus, messages } = useRealtime({
    channel,
    topics,
    token: realtimeToken,
  });

  return <p>{connectionStatus}: {messages.byTopic.status?.data.message}</p>;
}

For long-lived subscriptions, prefer a token factory or route handler so useRealtime can request a fresh token when it reconnects.

Typed topic access

When you pass a channel instance instead of a plain string, messages.byTopic, messages.all, messages.last, and messages.delta all stay typed to the subscribed topics:

const ch = pipelineChannel({ runId });
const topics = ["status", "tokens"] as const;

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

messages.byTopic.status?.data.message; // string
messages.byTopic.tokens?.data.token; // string

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

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

History management

By default, messages.all retains the last 100 messages. Adjust with historyLimit:

// Keep last 500 messages
useRealtime({
  channel: ch,
  topics: ["status"],
  token: () => getToken(runId),
  historyLimit: 500,
});

// Keep all messages (unbounded, use with caution)
useRealtime({
  channel: ch,
  topics: ["status"],
  token: () => getToken(runId),
  historyLimit: null,
});

// Keep only 10 messages
useRealtime({
  channel: ch,
  topics: ["status"],
  token: () => getToken(runId),
  historyLimit: 10,
});

Buffering

For high-frequency streams like token-by-token AI output, use bufferInterval to batch re-renders:

const { messages } = useRealtime({
  channel: ch,
  topics: ["tokens"],
  token: () => getToken(runId),
  bufferInterval: 100, // Batch messages, re-render at most every 100ms
});

messages.delta; // up to 100ms of new messages per flush
messages.all; // retained history after each flush

messages.byTopic still tracks the latest message for each topic, while messages.all, messages.last, and messages.delta are flushed on the buffer interval.

Conditional subscription

Use enabled to start or stop the subscription without unmounting the component:

const [runId, setRunId] = useState<string | null>(null);

const channel = runId ? pipelineChannel({ runId }) : undefined;

const { connectionStatus, isPaused, pauseReason } = useRealtime({
  channel,
  topics: ["status"],
  token: runId ? () => getToken(runId) : undefined,
  enabled: !!runId,
});

connectionStatus; // "idle" until a run exists, then "connecting" | "open" | ...
isPaused; // true when disabled or hidden
pauseReason; // "disabled" | "hidden" | null

If your token is a ClientSubscriptionToken, or your token factory returns either a ClientSubscriptionToken or only a string key, make sure channel and topics are present. If it returns a full token object, the hook can derive them from the token instead, though top-level channel and topics are still recommended for typed message inference.