inngest-realtime

Inngest Realtime

Stream updates from durable Inngest functions to live UIs. Use channels and topics to broadcast progress, render workflow execution as it happens, or build bi-directional human-in-the-loop flows.

These skills are focused on TypeScript. For Python or Go, refer to the Inngest documentation for language-specific guidance. Core concepts apply across all languages.

⚠ CRITICAL: v3 vs v4 package selection

Realtime in Inngest v4 lives at the SDK subpath inngest/realtime. The standalone @inngest/realtime npm package is a v3-era package and is NOT compatible with inngest@4.x. If your project is on v4 (the npm default), do not install @inngest/realtime. Use the imports below.

Symptoms of using the wrong package on v4: TypeError: Cls is not a constructor on every PUT /api/inngest, 401 on subscription tokens, type incompatibility on new Inngest({ middleware: [...] }). Verify your package.json shows "inngest": "^4.x" before reading further.

Prerequisites

• Inngest v4 SDK installed (npm install inngest) — see the inngest-setup skill
• INNGEST_DEV=1 set in .env.local for local development (without it, the SDK demands cloud signing keys and 401s on token requests)
• Local Inngest dev server running (npx inngest-cli@latest dev)
• Optional: zod for schema validation on topics

When to use Realtime

| Problem shape | Pattern | |---|---| | Order status page animates as durable workflow steps complete | Per-run channel, publish per step, client subscribes | | AI agent streams tokens to a chat UI | Per-conversation channel, publish chunks, stream to browser | | Log tail for a long-running job | Single channel, log topic, append to UI | | Human-in-the-loop approval | Channel + waitForEvent, publish prompt, wait for response | | Admin dashboard with live order list | Global admin channel, fan-out from each function |

Architecture

Three pieces:

1. Channel definition — a typed contract for what gets published. Lives in shared module so both server and client can reference the same channel name.
2. Publishing — call step.realtime.publish between steps to wrap a durable publish, or inngest.realtime.publish inside step.run because you're already inside a memoized step. See "Which publish method to use" below.
3. Subscribing — server action mints a subscription token; React client uses the useRealtime hook (or the lower-level subscribe() API for non-React consumers).

Step 1: Define a channel

Channels are pure data — no class hierarchy, no zod runtime required (but recommended for type safety). Define them once and import where needed.

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

Two channel name shapes:

• name: 'admin' — static channel, accessed as adminChannel.order (topic ref)
• name: (id) => 'channel:${id}' — parametric, accessed as orderChannel(id).step (call the channel def with the id, then access topic)

Step 2: Publish from inside a function

Inngest v4 ships realtime support natively — no middleware required. But where you call publish matters: it determines whether the publish is durable, and it's the most common place to get realtime wrong.

Which publish method to use

| Where you are | Use this | Why | |---|---|---| | Outside a step (top-level handler code, between step.run calls) | step.realtime.publish(id, topicRef, data) | Wraps the publish in its own step so it's durable, deduplicated by id, and retry-safe. | | Inside a step (inside the callback passed to step.run) | inngest.realtime.publish(topicRef, data) | You're already inside a memoized step. step.realtime.publish would create a step inside a step. The bare client publish is the right call here. | | Outside a function (one-off route, script, etc.) | inngest.realtime.publish(topicRef, data) | Allowed, but not retry-safe — your client receiver must handle duplicates. |

The 90% rule: if you're writing handler code and you reach for publish, use step.realtime.publish. If you're writing code inside a step.run block and you reach for publish, use inngest.realtime.publish.

Example: both patterns in one function

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

    // ...
  },
);

Why no middleware: Earlier versions used @inngest/realtime's realtimeMiddleware() to inject a publish arg into the handler. v4 puts it on step.realtime and inngest.realtime directly.

Step 3: Mint a subscription token (server action)

In Next.js App Router, use a Server Action to securely mint a short-lived token for the React hook in Step 4. Without a token, clients can't subscribe.

// 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('Unau