Update to Inngest

{
  "description": "Use when streaming durable workflow updates to a UI in real time — live order status pages that animate as steps complete, AI agent token streaming from a function to the browser, log tailing for long-running jobs, or human-in-the-loop approval flows that publish a prompt and wait for a user reply. Covers Inngest v4 native realtime: defining typed channels, publishing from inside step.run, minting subscription tokens via server actions, and consuming the stream from React/Next.js client components.",
  "included_files": [],
  "name": "inngest-realtime",
  "skill_md_contents": "---\nname: inngest-realtime\ndescription: \"Use when streaming durable workflow updates to a UI in real time — live order status pages that animate as steps complete, AI agent token streaming from a function to the browser, log tailing for long-running jobs, or human-in-the-loop approval flows that publish a prompt and wait for a user reply. Covers Inngest v4 native realtime: defining typed channels, publishing from inside step.run, minting subscription tokens via server actions, and consuming the stream from React/Next.js client components.\"\n---\n\n# Inngest Realtime\n\nStream 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.\n\n> **These skills are focused on TypeScript.** For Python or Go, refer to the [Inngest documentation](https://www.inngest.com/llms.txt) for language-specific guidance. Core concepts apply across all languages.\n\n> **⚠ CRITICAL: v3 vs v4 package selection**\n>\n> 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.\n>\n> 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.\n\n## Prerequisites\n\n- Inngest v4 SDK installed (`npm install inngest`) — see the `inngest-setup` skill\n- `INNGEST_DEV=1` set in `.env.local` for local development (without it, the SDK demands cloud signing keys and 401s on token requests)\n- Local Inngest dev server running (`npx inngest-cli@latest dev`)\n- Optional: `zod` for schema validation on topics\n\n## When to use Realtime\n\n| Problem shape | Pattern |\n|---|---|\n| Order status page animates as durable workflow steps complete | Per-run channel, publish per step, client subscribes |\n| AI agent streams tokens to a chat UI | Per-conversation channel, publish chunks, stream to browser |\n| Log tail for a long-running job | Single channel, log topic, append to UI |\n| Human-in-the-loop approval | Channel + waitForEvent, publish prompt, wait for response |\n| Admin dashboard with live order list | Global admin channel, fan-out from each function |\n\n## Architecture\n\nThree pieces:\n\n1. **Channel definition** — a typed contract for what gets published. Lives in shared module so both server and client can reference the same channel name.\n2. **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.\n3. **Subscribing** — server action mints a subscription token; React client uses the `useRealtime` hook (or the lower-level `subscribe()` API for non-React consumers).\n\n## Step 1: Define a channel\n\nChannels are pure data — no class hierarchy, no zod runtime required (but recommended for type safety). Define them once and import where needed.\n\n```typescript\n// src/inngest/channels.ts\nimport { channel } from 'inngest/realtime';\nimport { z } from 'zod';\n\n// Per-run channel: each fulfill-order run publishes step updates to its own channel.\nexport const orderChannel = channel({\n  name: (orderId: string) => `order:${orderId}`,\n  topics: {\n    step: {\n      schema: z.object({\n        name: z.string(),\n        status: z.enum(['running', 'complete', 'failed']),\n        output: z.record(z.string(), z.unknown()).optional(),\n        ts: z.number(),\n      }),\n    },\n  },\n});\n\n// Global admin channel: fan-out for cross-cutting visibility.\nexport const adminChannel = channel({\n  name: 'admin',\n  topics: {\n    order: {\n      schema: z.object({\n        orderId: z.string(),\n        step: z.string(),\n        status: z.enum(['running', 'complete', 'failed']),\n        ts: z.number(),\n      }),\n    },\n  },\n});\n```\n\n**Two channel name shapes:**\n- `name: 'admin'` — static channel, accessed as `adminChannel.order` (topic ref)\n- `name: (id) => 'channel:${id}'` — parametric, accessed as `orderChannel(id).step` (call the channel def with the id, then access topic)\n\n## Step 2: Publish from inside a function\n\nInngest 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.\n\n### Which publish method to use\n\n| Where you are | Use this | Why |\n|---|---|---|\n| **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. |\n| **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. |\n| **Outside a function** (one-off route, script, etc.) | `inngest.realtime.publish(topicRef, data)` | Allowed, but **not retry-safe** — your client receiver must handle duplicates. |\n\nThe 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`.\n\n### Example: both patterns in one function\n\n```typescript\n// src/inngest/functions/fulfill-order.ts\nimport { inngest } from '../client';\nimport { orderChannel, adminChannel } from '../channels';\n\nexport const fulfillOrder = inngest.createFunction(\n  {\n    id: 'fulfill-order',\n    retries: 3,\n    triggers: [{ event: 'store/order.placed' }],\n  },\n  async ({ event, step }) => {\n    const { orderId, customerEmail, lineItems } = event.data;\n\n    // Outside any step.run — use step.realtime.publish for a durable wrapper.\n    const emit = async (\n      name: string,\n      status: 'running' | 'complete' | 'failed',\n      output?: Record<string, unknown>,\n    ) => {\n      const ts = Date.now();\n      await step.realtime.publish(\n        `emit-order-${name}-${status}`,\n        orderChannel(orderId).step,\n        { name, status, output, ts },\n      );\n      await step.realtime.publish(\n        `emit-admin-${name}-${status}`,\n        adminChannel.order,\n        { orderId, step: name, status, ts },\n      );\n    };\n\n    await emit('capture-payment', 'running');\n\n    // Inside step.run — use inngest.realtime.publish (already in a memoized step).\n    const payment = await step.run('capture-payment', async () => {\n      const intent = await stripe.paymentIntents.create({ /* ... */ });\n\n      // Stream a partial update mid-step. No step-in-step wrapping needed.\n      await inngest.realtime.publish(orderChannel(orderId).step, {\n        name: 'capture-payment',\n        status: 'running',\n        output: { stage: 'intent-created', intentId: intent.id },\n        ts: Date.now(),\n      });\n\n      return await stripe.paymentIntents.confirm(intent.id);\n    });\n\n    await emit('capture-payment', 'complete', payment);\n\n    await emit('reserve-inventory', 'running');\n    const inventory = await step.run('reserve-inventory', async () => {\n      // ...\n    });\n    await emit('reserve-inventory', 'complete', inventory);\n\n    // ...\n  },\n);\n```\n\n**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.\n\n## Step 3: Mint a subscription token (server action)\n\nIn 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.\n\n```typescript\n// src/app/orders/[orderId]/actions.ts\n'use server';\n\nimport { getClientSubscriptionToken } from 'inngest/react';\nimport { inngest } from '@/inngest/client';\nimport { orderChannel } from '@/inngest/channels';\n\nexport async function fetchOrderSubscriptionToken(orderId: string) {\n  // ⚠ AUTHORIZATION GATE: verify the current user owns this orderId\n  // before minting a token. Channels are addressable by ID, so without\n  // an ownership check, anyone can subscribe to any order's stream by\n  // guessing IDs.\n  //\n  //   const session = await getServerSession();\n  //   if (!session) throw new Error('Unauthenticated');\n  //   const order = await db.order.findUnique({ where: { id: orderId } });\n  //   if (order?.userId !== session.userId) throw new Error('Forbidden');\n\n  return getClientSubscriptionToken(inngest, {\n    channel: orderChannel(orderId),\n    topics: ['step'],\n  });\n}\n```\n\n`getClientSubscriptionToken` from `inngest/react` returns a token shape that the `useRealtime` hook in Step 4 consumes directly. No ChannelInstance stripping needed — that gotcha only applies to the lower-level `getSubscriptionToken` + manual `subscribe()` path (see \"Pattern: Manual subscribe\" below).\n\n## Step 4: Subscribe with the `useRealtime` hook\n\nThe recommended consumer for React/Next.js is the `useRealtime` hook from `inngest/react`. It handles the subscription lifecycle, reconnect, type narrowing per topic, and cleanup.\n\n```typescript\n// src/components/OrderStatusClient.tsx\n'use client';\n\nimport { useRealtime } from 'inngest/react';\nimport { orderChannel } from '@/inngest/channels';\nimport { fetchOrderSubscriptionToken } from '@/app/orders/[orderId]/actions';\n\nexport function OrderStatusClient({ orderId }: { orderId: string }) {\n  const { messages, connectionStatus, error } = useRealtime({\n    channel: orderChannel(orderId),\n    topics: ['step'] as const,\n    token: () => fetchOrderSubscriptionToken(orderId),\n  });\n\n  if (error) return <div>Error: {error.message}</div>;\n\n  return (\n    <div>\n      <div>Status: {connectionStatus}</div>\n      <ul>\n        {messages.all.map((m, i) => (\n          <li key={i}>\n            {(m.data as { name: string }).name}: {(m.data as { status: string }).status}\n          </li>\n        ))}\n      </ul>\n    </div>\n  );\n}\n```\n\n**Useful options on the hook:**\n\n| Option | Default | Use it when |\n|---|---|---|\n| `enabled` | `true` | Delay the subscription until you have an ID (e.g., `enabled: !!runId`). |\n| `bufferInterval` | `0` | Batch updates from a fast stream so React doesn't re-render per message. |\n| `pauseOnHidden` | `false` | Pause the stream when the tab isn't visible (saves bandwidth). |\n| `autoCloseOnTerminal` | `true` | Disconnect when the run completes — turn off to keep the stream open for fan-out channels. |\n| `historyLimit` | unbounded | Cap how many messages are retained in `messages.all`. |\n\nThe hook returns `messages.byTopic` (latest per topic), `messages.all` (full history), `messages.last` (most recent), and `messages.delta` (new since last render).\n\n## Pattern: Manual subscribe (non-React or custom transport)\n\nThe `useRealtime` hook covers the React case. If you're not using React, or you need a custom subscription lifecycle (server-side streaming, background workers, custom protocols), use the lower-level `subscribe()` API directly.\n\n### Server action: mint a token with the lower-level helper\n\n```typescript\n// src/app/orders/[orderId]/actions.ts\n'use server';\n\nimport { getSubscriptionToken } from 'inngest/realtime';\nimport { inngest } from '@/inngest/client';\nimport { orderChannel } from '@/inngest/channels';\n\nexport async function fetchOrderSubscriptionTokenLowLevel(orderId: string) {\n  // ⚠ AUTHORIZATION GATE: same as Step 3 — verify ownership before minting.\n\n  const token = await getSubscriptionToken(inngest, {\n    channel: orderChannel(orderId),\n    topics: ['step'],\n  });\n\n  // ⚠ CRITICAL: strip the ChannelInstance from the response.\n  // getSubscriptionToken returns { channel: ChannelInstance, ... } where\n  // ChannelInstance contains zod schema methods (a class with prototypes).\n  // Next.js refuses to serialize classes across the server-action → client-component\n  // boundary, so return ONLY primitives.\n  return {\n    channel: orderChannel(orderId).name as string,\n    topics: ['step'] as const,\n    key: token.key,\n    apiBaseUrl: token.apiBaseUrl,\n  };\n}\n```\n\n### Manual client subscription\n\n```typescript\n// src/components/OrderStatusManual.tsx\n'use client';\n\nimport * as React from 'react';\nimport { subscribe } from 'inngest/realtime';\nimport { fetchOrderSubscriptionTokenLowLevel } from '@/app/orders/[orderId]/actions';\n\nexport function OrderStatusManual({ orderId }: { orderId: string }) {\n  const [messages, setMessages] = React.useState<unknown[]>([]);\n\n  React.useEffect(() => {\n    let cancelled = false;\n    let sub: { close?: (reason?: string) => void } | undefined;\n\n    (async () => {\n      const token = await fetchOrderSubscriptionTokenLowLevel(orderId);\n      if (cancelled) return;\n\n      sub = await subscribe(\n        {\n          channel: token.channel,\n          topics: [...token.topics],\n          key: token.key,\n          apiBaseUrl: token.apiBaseUrl,\n        },\n        (message) => {\n          if (cancelled) return;\n          setMessages((prev) => [...prev, message.data]);\n        },\n      );\n    })();\n\n    return () => {\n      cancelled = true;\n      sub?.close?.('unmount');\n    };\n  }, [orderId]);\n\n  // ... render ...\n}\n```\n\n### SSE streaming from a route handler\n\nSubscribe inside a Next.js API route and pipe the stream to the client via SSE:\n\n```typescript\n// src/app/api/orders/[orderId]/stream/route.ts\nimport { inngest } from '@/inngest/client';\nimport { subscribe } from 'inngest/realtime';\nimport { orderChannel } from '@/inngest/channels';\n\nexport async function GET(req: Request, { params }: { params: { orderId: string } }) {\n  // ⚠ AUTHORIZATION GATE: same rule as the server-action token mint above.\n  // Authenticate the request and confirm the caller owns params.orderId\n  // before opening the SSE stream. Skipping this leaks every order's\n  // step events to anyone with a URL.\n\n  const stream = await subscribe({\n    app: inngest,\n    channel: orderChannel(params.orderId),\n    topics: ['step'],\n  });\n\n  return new Response(stream.getEncodedStream(), {\n    headers: {\n      'Content-Type': 'text/event-stream',\n      'Cache-Control': 'no-cache',\n      Connection: 'keep-alive',\n    },\n  });\n}\n```\n\nClient consumes via `fetch().getReader()` rather than the `subscribe()` callback. Use this when you want the SSE behavior or when the client-side `subscribe()` API doesn't fit your component lifecycle.\n\n## Pattern: Human-in-the-loop\n\nCombine `step.realtime.publish` with `step.waitForEvent`:\n\n```typescript\nimport crypto from 'crypto';\n\nexport const reviewWorkflow = inngest.createFunction(\n  { id: 'review-workflow', triggers: [{ event: 'review/start' }] },\n  async ({ event, step }) => {\n    const confirmationId = await step.run('gen-id', () => crypto.randomUUID());\n\n    // Publish a prompt — the client subscribes and renders an approval UI\n    await step.realtime.publish(\n      'publish-prompt',\n      reviewChannel.message,\n      { message: 'Confirm to proceed?', confirmationId },\n    );\n\n    // Wait up to 15 minutes for the user to send the matching event back\n    const confirmation = await step.waitForEvent('await-confirmation', {\n      event: 'review/confirmation',\n      timeout: '15m',\n      if: `async.data.confirmationId == \"${confirmationId}\"`,\n    });\n\n    if (!confirmation) {\n      // user didn't respond — abort or escalate\n      return { decision: 'timed_out' };\n    }\n    // continue workflow...\n  },\n);\n```\n\nThe `confirmationId` links the published prompt to the matching reply, so the workflow knows which response to act on.\n\n## Common pitfalls\n\n### Don't use `@inngest/realtime` on v4\n\nThe standalone `@inngest/realtime` package is for Inngest v3 only. On v4, all realtime APIs are in the SDK subpath `inngest/realtime`. Mixing them produces:\n- `TypeError: Cls is not a constructor` on `PUT /api/inngest` (v3 middleware class signature mismatch)\n- 401 Unauthorized on subscription tokens\n- TypeScript errors casting middleware\n\n**Verify with:** `grep '\"inngest\"' package.json` — if it's `^4.x`, use `inngest/realtime`. Period.\n\n### Don't return ChannelInstance from a Next.js server action (manual subscribe path only)\n\n`getSubscriptionToken` returns `{ channel: ChannelInstance, ... }` where ChannelInstance has zod schema methods (a class). Next.js refuses to serialize classes across the server-action → client-component boundary. Strip to primitives before returning. See \"Pattern: Manual subscribe\" above.\n\nThis gotcha does **not** apply when you use `getClientSubscriptionToken` from `inngest/react` (Step 3 — the recommended path). That helper returns a serialization-safe shape directly.\n\n### `INNGEST_DEV=1` is required for local dev\n\nWithout it, the SDK assumes cloud mode and demands `INNGEST_SIGNING_KEY` + `INNGEST_EVENT_KEY`. All realtime operations 401 / 500. Add to `.env.local`. Hard restart the dev server (Next.js does not hot-reload `.env.local` changes).\n\n### Channel topic schemas validate on publish, not on consume\n\nIf your published payload doesn't match the zod schema, the publish fails server-side. Subscriber receives nothing. Catch publish errors during step execution, or run with `validate: false` in `subscribe()` if you have a reason to skip schema validation client-side.\n\n## Reference\n\n- v4 entry points:\n  - `import { channel } from 'inngest/realtime'` — channel definitions\n  - `import { useRealtime, getClientSubscriptionToken } from 'inngest/react'` — React hook + matching token helper (Step 3 + Step 4)\n  - `import { getSubscriptionToken, subscribe } from 'inngest/realtime'` — lower-level helpers for non-React or custom transport\n- Publish methods:\n  - **Outside a step:** `step.realtime.publish(id, topicRef, data)` — wraps in a durable step\n  - **Inside `step.run`:** `inngest.realtime.publish(topicRef, data)` — already inside a memoized step, no wrapping needed\n  - **Outside a function:** `inngest.realtime.publish(topicRef, data)` — allowed but not retry-safe\n- Subscribe overloads: `subscribe(token)` returns a stream; `subscribe(token, callback)` invokes callback per message\n- Next.js Server Action gotcha (manual path only): strip `ChannelInstance` → return `{ channel: string, topics, key, apiBaseUrl }`. Not needed with `getClientSubscriptionToken`.\n"
}