tanstack-ai

# Durable Streams — TanStack AI Connection adapter for TanStack AI's `useChat()`. Uses one stream per chat session: user messages are echoed into the stream alongside model responses, making it a complete transcript that supports multi-client sync and SSR hydration. ## The two auth layers — keep them separate | Auth | What | Where | How | | ------------------- | -------------------------------------- | ------------------------------------------------------- | ----------------------------------------------------------- | | **Durable Streams** | `DS_SECRET` | Server-only — POST `/api/chat` + GET `/api/chat-stream` | `Authorization: Bearer <DS_SECRET>` on upstream DS requests | | **AI model** | `ANTHROPIC_API_KEY` / `OPENAI_API_KEY` | Server-only (read by the adapter) | `process.env.ANTHROPIC_API_KEY` is picked up automatically | NEVER mix these. The DS secret authenticates your server to the DS service. The AI key authenticates your server to Anthropic/OpenAI. The client NEVER sees either. For apps that let users supply their own AI key, see "User-supplied AI key" below. ## Prerequisites Install the DS client + this package: ```bash pnpm add @durable-streams/client @durable-streams/tanstack-ai-transport @tanstack/ai-react @tanstack/ai-anthropic ``` Set env vars. You need a running Durable Streams service — self-hosted (see `server-deployment` skill) or Electric Cloud (see blog post for setup). ```bash ELECTRIC_URL= # e.g. https://api.electric-sql.cloud (root API URL) DS_SERVICE_ID= # Durable Streams service id, e.g. svc-abc-123 DS_SECRET= # Bearer token for DS auth ANTHROPIC_API_KEY= # AI model auth (OR let user supply their own, see below) ``` Build the stream base URL from `ELECTRIC_URL + DS_SERVICE_ID` rather than trusting a single `DS_URL` env var — different environments populate it differently, and a mismatch silently 404s your PUT. ```ts // src/lib/ds-stream.ts const electricUrl = process.env.ELECTRIC_URL || "https://api.electric-sql.cloud" const serviceId = process.env.DS_SERVICE_ID if (!serviceId) throw new Error("DS_SERVICE_ID is required") export const DS_BASE = `${electricUrl.replace(/\/+$/, "")}/v1/stream/${serviceId}` export const DS_AUTH = { Authorization: `Bearer ${process.env.DS_SECRET}` } ``` ## Client ```tsx import { useMemo } from "react" import { useChat } from "@tanstack/ai-react" import { durableStreamConnection } from "@durable-streams/tanstack-ai-transport" function Chat({ id, initialMessages, resumeOffset, }: { id: string initialMessages?: Array<any> resumeOffset?: string }) { const connection = useMemo( () => durableStreamConnection({ sendUrl: `/api/chat?id=${encodeURIComponent(id)}`, readUrl: `/api/chat-stream?id=${encodeURIComponent(id)}`, initialOffset: resumeOffset, // from SSR loader, skips replay }), [id, resumeOffset] ) const { messages, sendMessage } = useChat({ id, initialMessages, connection, live: true, // keeps read subscription open for multi-client sync }) // TanStack AI UIMessage has `parts: Array<MessagePart>`. TextPart uses // `.content` (NOT `.text` — that silently renders empty strings). return ( <> {messages.map((m) => ( <div key={m.id}> {m.parts .filter((p) => p.type === "text") .map((p, i) => ( <span key={i}>{p.content}</span> ))} </div> ))} </> ) } ``` ## Server — POST /api/chat ```ts import { chat } from "@tanstack/ai" import { anthropicText } from "@tanstack/ai-anthropic" import { toDurableChatSessionResponse } from "@durable-streams/tanstack-ai-transport" export async function POST(request: Request) { const url = new URL(request.url) const body = await request.json() const id = url.searchParams.get("id") ?? body.id if (!id) return Response.json({ error: "Missing chat id" }, { status: 400 }) const latestUserMessage = body.messages.findLast( (m: any) => m.role === "user" ) const responseStream = chat({ adapter: anthropicText("claude-sonnet-4-6"), messages: body.messages, }) return await toDurableChatSessionResponse({ stream: { writeUrl: `${DS_BASE}/chat-${id}`, headers: DS_AUTH, createIfMissing: true, }, newMessages: latestUserMessage ? [latestUserMessage] : [], responseStream, }) } ``` **Available adapters:** - `anthropicText("claude-sonnet-4-6")` from `@tanstack/ai-anthropic` - `openaiText("gpt-4o-mini")` from `@tanstack/ai-openai` Both read credentials from their standard env vars (`ANTHROPIC_API_KEY`, `OPENAI_API_KEY`). ## Server — GET /api/chat-stream (read proxy) Never expose the DS write URL to the client. Proxy reads through your server so the DS secret stays server-side. ```ts export async function GET(request: Request) { const url = new URL(request.url) const id = url.searchParams.get("id") if (!id) return Response.json({ error: "Missing id" }, { status: 400 }) const upstream = new URL(`${DS_BASE}/chat-${id}`) // Forward offset/live/sse params from the browser's DS client for (const [k, v] of url.searchParams) { if (k !== "id") upstream.searchParams.set(k, v) } const response = await fetch(upstream, { headers: { ...DS_AUTH, ...(request.headers.get("accept") && { Accept: request.headers.get("accept")!, }), }, }) // Strip hop-by-hop headers before forwarding const headers = new Headers() for (const [k, v] of response.headers) { const lk = k.toLowerCase() if ( lk === "connection" || lk === "transfer-encoding" || lk === "content-length" || lk === "content-encoding" ) continue headers.set(k, v) } return new Response(response.body, { status: response.status, headers }) } ``` Use the chat id as a **query parameter** — not a dynamic route segment. Segments like `/api/chat-stream/$id` break when the stream path contains slashes. ## SSR hydration + resume In your route loader, materialize the snapshot server-side and pass the offset down: ```ts import { materializeSnapshotFromDurableStream } from "@durable-streams/tanstack-ai-transport" export const loader = async ({ params }: { params: { id: string } }) => { const snapshot = await materializeSnapshotFromDurableStream({ readUrl: `${DS_BASE}/chat-${params.id}`, headers: DS_AUTH, }) return { messages: snapshot.messages, resumeOffset: snapshot.offset } } ``` Pass `resumeOffset` to `durableStreamConnection` — this skips replaying the history on first subscribe. ## User-supplied AI key If users enter their own AI key in a settings UI: 1. Store the key in a shared store (Context, Zustand, Jotai) — NOT per-hook `useState`, otherwise different components see different values. 2. Pass it via `headers` on `durableStreamConnection` (NOT on `useChat` — those headers aren't forwarded): ```tsx const { apiKey } = useSettings() // from Context/store, shared across components const connection = useMemo( () => durableStreamConnection({ sendUrl: `/api/chat?id=${encodeURIComponent(id)}`, readUrl: `/api/chat-stream?id=${encodeURIComponent(id)}`, headers: { "x-api-key": apiKey }, }), [id, apiKey] ) ``` Server reads the header and sets it for the adapter: ```ts const apiKey = request.headers.get("x-api-key") if (!apiKey) return Response.json({ error: "Missing API key" }, { status: 401 }) process.env.ANTHROPIC_API_KEY = apiKey // ... rest of handler ``` ## Common Mistakes ### CRITICAL Sending full message history as newMessages Wrong: `newMessages: messages` — echoes the entire conversation every request. Correct: only pass what's new since the last request: ```ts const latestUserMessage = messages.findLast((m) => m.role === "user") newMessages: latestUserMessage ? [latestUserMessage] : [] ``` ### CRITICAL Exposing the DS write URL to the client Setting `readUrl` on the server stream config to the durable stream's write URL leaks the secret in the `Location` header. Always use a read proxy route for `readUrl`. ### CRITICAL First assistant response invisible until refresh — dead subscription `durableStreamConnection` opens its live SSE read on mount. If the stream doesn't exist yet (new conversation), the read fails with `STREAM_NOT_FOUND` and **the subscription terminates — it does NOT retry on 404**. When the user's first POST then creates the stream and the server streams chunks, nothing is listening. After a refresh, the subscription is re-opened against an existing stream and everything works — which is exactly what the user describes when they say "the first response only shows after refresh". **Fix**: create the stream eagerly when you create the conversation row, so the client's subscription has something to attach to. PUT is idempotent — catch `CONFLICT_EXISTS` / `CONFLICT_SEQ` and treat as success. ```ts // src/routes/api/conversations.ts — after inserting the conversation row import { DurableStream, DurableStreamError } from "@durable-streams/client" async function ensureChatStream(streamId: string): Promise<void> { try { const stream = new DurableStream({ url: `${DS_BASE}/chat-${streamId}`, headers: DS_AUTH, contentType: "application/json", }) await stream.create({ contentType: "application/json" }) } catch (err) { if ( err instanceof DurableStreamError && err.status === 409 && (err.code === "CONFLICT_EXISTS" || err.code === "CONFLICT_SEQ") ) { return // already exists — fine } throw err } } ``` Do NOT rely on `toDurableChatSessionResponse`'s `createIfMissing` to cover this case. That handler runs during the first POST, which is AFTER the client's read subscription has already died. ### CRITICAL Switching conversations shows stale data — missing useLiveQuery deps + missing component key Two separate React pitfalls compound into the same symptom (header/messages don't update when the user clicks a different conversation): 1. `useLiveQuery` needs explicit deps — without them, the query closure captures the initial id and never re-runs: ```ts // WRONG — no deps, closure captures initial conversationId forever const { data } = useLiveQuery((q) => q.from({ conv }).where(({ conv }) => eq(conv.id, conversationId)) ) // RIGHT — deps array pins re-evaluation to the param const { data } = useLiveQuery( (q) => q.from({ conv }).where(({ conv }) => eq(conv.id, conversationId)), [conversationId] ) ``` 2. `useChat`'s internal `ChatClient` is memoized per-hook and keeps previous messages in a ref even when the `id` prop changes. Force a full remount by keying the component on the stream id: ```tsx return ( <ChatInner key={streamId} streamId={streamId} connection={connection} /> ) ``` ### CRITICAL sendMessage signature — crash on submit `useChat().sendMessage` takes either a string **or** `{ content: Array<ContentPart>, id? }` (for multimodal). The intuitive-looking `{ text: "hi" }` form is NOT supported — it normalizes to `{ content: undefined }` and crashes inside `StreamProcessor.addUserMessage` with: ``` TypeError: Cannot read properties of undefined (reading 'map') ``` ```ts // WRONG — passes { text } which is neither a string nor a valid object shape sendMessage({ text: input.trim() }) // RIGHT — pass the string directly sendMessage(input.trim()) // RIGHT — multimodal (explicit content parts) sendMessage({ content: [ { type: "text", content: input.trim() }, { type: "image", source: { type: "url", value: imageUrl } }, ], }) ``` ### CRITICAL Wrong field on UIMessage parts — empty bubbles