⬡ API & SDK/2026-04-26Advanced

Building a Scalable Real-Time AI Chat Server with Claude API × WebSocket × Redis Pub/Sub — Node.js Production Architecture, Multi-User Management, and Cost Control

Running a real-time AI chat server on Claude API, WebSocket, and Redis Pub/Sub in production. SSE trade-offs, multi-instance routing, and how stop latency quietly corrupts per-user budgets — with the instrumentation code to measure it.

websocketredisnodejsstreaming²² realtime³ production¹¹¹ api³⁹

✦ Premium Article

When users press send in a chat UI and watch an AI response stream in word by word, that experience is often implemented with SSE (Server-Sent Events). But the moment requirements expand to "let users interrupt mid-generation," "manage multiple turns without reconnecting," or "show bidirectional typing indicators," SSE hits a hard wall.

SSE is an HTTP response extension — one-way, server to client only. Any client-to-server signal still needs a separate REST API call. Session state has to live somewhere. The moment you scale horizontally across multiple instances, routing messages to the right server becomes non-trivial.

I ran into exactly this when building the backend for a mobile app. The single-instance prototype worked fine, but the first time I tried to add "stop generation" functionality, I realized the architecture needed rethinking from the ground up.

WebSocket was designed for bidirectional communication. One persistent connection handles sending and receiving, which makes conversation state management much simpler. Add Redis Pub/Sub and you can route messages correctly regardless of how many server instances you're running.

This guide builds a production-quality real-time AI chat server using Claude API streaming + WebSocket + Redis Pub/Sub. Every section includes working code you can run.

SSE vs WebSocket: The Decision Framework

Getting this choice wrong early means a significant refactor later. Here's how to think through it.

SSE is sufficient when:

Users send a message and wait passively for a complete response
You need server-to-client push only (news feeds, progress indicators, notifications)
Your backend runs on Cloudflare Workers, Lambda, or another environment where WebSocket is awkward to manage
HTTP/2 is available and connection limits aren't a concern

WebSocket becomes necessary when:

Users need to interrupt AI generation mid-stream (the "stop" button that actually works)
You want to send "user is typing" signals from the client to influence AI behavior
You need to manage multi-turn conversations over a single connection to reduce reconnect overhead
Multiple users share a session for collaboration features
You're building for mobile, where connection lifecycle management benefits from explicit control

This guide targets the second set of use cases. If your requirements are firmly in the first group, SSE is simpler — don't add complexity you don't need.

System Architecture Overview

Here's the full picture of what we're building:

[Mobile/Web Client]
       |  WebSocket (wss://)
       |
[Node.js WebSocket Server (horizontally scaled)]
       |  Claude API Streaming (HTTPS/SSE)
       |  Redis Pub/Sub (cross-instance message routing)
       |  Redis String/Hash (per-user token budget + history TTL)
       |
[Redis Cluster] --- [Anthropic Claude API]

Each component has a single responsibility:

Node.js WebSocket Server: Accepts connections, validates JWT tokens, forwards user messages to Claude API as streaming requests, and pipes response deltas back to the client in real time
Redis Pub/Sub: When running multiple server instances, acts as a message bus so any instance can deliver to any user regardless of which instance they're connected to
Redis Key/Value: Stores per-user monthly token usage counters with TTL, and optionally short-term conversation history for reconnect resumption

Install the required packages:

npm install ws @anthropic-ai/sdk ioredis jsonwebtoken dotenv
npm install -D typescript @types/ws @types/node tsx

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Instrumentation for measuring post-stop billing, and what 100ms vs 500ms vs 2,000ms of stop latency costs across 3,000 interruptions a month

✦Why a stop signal sent only over Redis Pub/Sub disappears during instance restarts, and the local-first fix

✦Two holes I found by actually running the code against a mock SSE server and Redis: aborts that never surface as AbortError, and a monthly cap that never fires — plus a reserve-then-settle fix

✦Complete implementation: WebSocket server, Claude API stream bridging, Redis routing, and JWT budget control (Node.js / TypeScript)

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Redis Client Setup

Before the application code, configure the Redis client carefully. Default ioredis settings can cause the entire server to hang when Redis becomes unavailable.

// src/redis.ts
import Redis from "ioredis";
 
let client: Redis | null = null;
let subscriber: Redis | null = null;
 
export function getRedisClient(): Redis {
  if (!client) {
    client = new Redis({
      host: process.env.REDIS_HOST ?? "localhost",
      port: parseInt(process.env.REDIS_PORT ?? "6379"),
      password: process.env.REDIS_PASSWORD,
      retryStrategy: (times) => {
        // Give up after 3 failed retries; null stops retrying
        if (times > 3) return null;
        return Math.min(times * 200, 2000); // 200ms, 400ms, 800ms
      },
      // Critical: fail immediately instead of queuing commands when disconnected
      enableOfflineQueue: false,
      connectTimeout: 3000,
      commandTimeout: 2000,
      lazyConnect: false,
    });
 
    client.on("error", (err) => {
      // Log but don't crash — let individual command failures bubble up
      console.error("Redis client error:", err.message);
    });
 
    client.on("reconnecting", () => {
      console.log("Redis reconnecting...");
    });
  }
  return client;
}
 
// Separate client for subscriptions — ioredis requires dedicated clients for subscribe mode
export function getSubscriberClient(): Redis {
  if (!subscriber) {
    subscriber = getRedisClient().duplicate();
  }
  return subscriber;
}

The enableOfflineQueue: false setting is critical. With the default true, commands queue up while Redis is unavailable, and WebSocket handlers block waiting for them. With false, commands throw immediately when disconnected, letting you handle failures gracefully per call site rather than having your whole server stall.

WebSocket Server: Handling Connections

// src/server.ts
import { WebSocketServer, WebSocket } from "ws";
import { createServer } from "http";
import { verifyToken, type JwtPayload } from "./auth.js";
import { ChatSession } from "./session.js";
import { initPubSub } from "./pubsub.js";
 
const httpServer = createServer();
const wss = new WebSocketServer({ server: httpServer });
const sessions = new Map<string, ChatSession>();
 
// Initialize Pub/Sub listener before accepting connections
await initPubSub();
 
wss.on("connection", async (ws: WebSocket, req) => {
  const url = new URL(req.url ?? "/", "http://localhost");
  const token = url.searchParams.get("token");
 
  if (!token) {
    ws.close(4001, "Authentication required");
    return;
  }
 
  let payload: JwtPayload;
  try {
    payload = verifyToken(token);
  } catch {
    ws.close(4001, "Invalid or expired token");
    return;
  }
 
  const { userId, plan } = payload;
 
  // Evict any existing session for this user (handles reconnects and duplicate tabs)
  const existingSession = sessions.get(userId);
  if (existingSession) {
    existingSession.ws.close(4002, "Replaced by new connection");
    existingSession.cleanup();
  }
 
  const session = new ChatSession(userId, plan, ws);
  sessions.set(userId, session);
 
  console.log(`✅ Connected: userId=${userId} plan=${plan} total=${wss.clients.size}`);
 
  ws.on("message", async (data) => {
    try {
      const message = JSON.parse(data.toString());
      await session.handleMessage(message);
    } catch {
      ws.send(JSON.stringify({ type: "error", message: "Invalid message format" }));
    }
  });
 
  ws.on("close", (code, reason) => {
    console.log(`🔌 Disconnected: userId=${userId} code=${code}`);
    if (session.isStreaming) {
      session.abortAndSavePartial();
    }
    sessions.delete(userId);
    session.cleanup();
  });
 
  ws.on("error", (err) => {
    console.error(`WebSocket error userId=${userId}:`, err.message);
    sessions.delete(userId);
    session.cleanup();
  });
});
 
// Periodic heartbeat to detect stale connections
setInterval(() => {
  wss.clients.forEach((ws) => {
    if ((ws as WebSocket & { isAlive?: boolean }).isAlive === false) {
      ws.terminate();
      return;
    }
    (ws as WebSocket & { isAlive?: boolean }).isAlive = false;
    ws.ping();
  });
 
  wss.clients.forEach((ws) => {
    ws.once("pong", () => {
      (ws as WebSocket & { isAlive?: boolean }).isAlive = true;
    });
  });
}, 30_000);
 
const PORT = process.env.PORT ?? 8080;
httpServer.listen(PORT, () => {
  console.log(`🚀 WebSocket server on port ${PORT}`);
});

The heartbeat interval (every 30s) terminates connections that are no longer alive — connections that dropped without a clean close handshake. Without this, your wss.clients set fills up with ghost connections over time.

ChatSession: Streaming Claude to WebSocket Clients

// src/session.ts
import { WebSocket } from "ws";
import Anthropic from "@anthropic-ai/sdk";
import { checkBudget, recordTokenUsage } from "./budget.js";
import { registerUserHandler, unregisterUserHandler } from "./pubsub.js";
import type { JwtPayload } from "./auth.js";
 
const anthropic = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY });
 
interface Message {
  role: "user" | "assistant";
  content: string;
}
 
export class ChatSession {
  userId: string;
  ws: WebSocket;
  isStreaming = false;
  private plan: JwtPayload["plan"];
  private history: Message[] = [];
  private abortController: AbortController | null = null;
  private partialContent = "";
 
  constructor(userId: string, plan: JwtPayload["plan"], ws: WebSocket) {
    this.userId = userId;
    this.plan = plan;
    this.ws = ws;
    // Register with Pub/Sub so cross-instance messages reach this client
    registerUserHandler(userId, (msg) => this.send(msg));
  }
 
  async handleMessage(msg: { type: string; content?: string }) {
    switch (msg.type) {
      case "chat":
        await this.handleChat(msg.content ?? "");
        break;
      case "stop":
        this.abortAndSavePartial();
        break;
      case "clear":
        this.history = [];
        this.send({ type: "cleared" });
        break;
      default:
        this.send({ type: "error", message: `Unknown message type: ${msg.type}` });
    }
  }
 
  private async handleChat(userContent: string) {
    if (this.isStreaming) {
      this.send({ type: "error", message: "Generation in progress. Send 'stop' first." });
      return;
    }
 
    // Budget check before adding to history
    let budget;
    try {
      budget = await checkBudget(this.userId, this.plan);
    } catch {
      // Redis unavailable — degrade gracefully, allow the request
      console.warn(`Budget check failed for ${this.userId}, allowing request`);
      budget = { allowed: true, used: 0, limit: Infinity, remaining: Infinity };
    }
 
    if (!budget.allowed) {
      this.send({
        type: "error",
        message: `Monthly token limit reached. Used: ${budget.used.toLocaleString()} / ${budget.limit.toLocaleString()}`,
      });
      return;
    }
 
    this.history.push({ role: "user", content: userContent });
    this.isStreaming = true;
    this.partialContent = "";
    this.abortController = new AbortController();
 
    try {
      this.send({ type: "stream_start" });
 
      const stream = await anthropic.messages.stream(
        {
          model: "claude-sonnet-4-6",
          max_tokens: 2048,
          system: "You are a helpful, knowledgeable AI assistant.",
          messages: this.trimmedHistory(),
        },
        { signal: this.abortController.signal }
      );
 
      for await (const chunk of stream) {
        if (
          chunk.type === "content_block_delta" &&
          chunk.delta.type === "text_delta"
        ) {
          this.partialContent += chunk.delta.text;
          this.send({ type: "delta", content: chunk.delta.text });
        }
      }
 
      const finalMsg = await stream.finalMessage();
      const tokensUsed =
        finalMsg.usage.input_tokens + finalMsg.usage.output_tokens;
 
      this.history.push({ role: "assistant", content: this.partialContent });
      this.send({
        type: "stream_end",
        tokensUsed,
        remainingBudget: Math.max(0, budget.remaining - tokensUsed),
      });
 
      recordTokenUsage(this.userId, tokensUsed).catch((err) =>
        console.error("Usage recording failed:", err.message)
      );
 
    } catch (err: unknown) {
      // The SDK throws APIUserAbortError; its err.name is "Error" (see below)
      if (err instanceof Anthropic.APIUserAbortError) {
        if (this.partialContent) {
          this.history.push({
            role: "assistant",
            content: this.partialContent + " [stopped]",
          });
        } else {
          // Nothing generated — remove the user message too
          this.history.pop();
        }
        this.send({ type: "stream_stopped", partialContent: this.partialContent });
      } else {
        console.error("Claude API error:", err);
        // Remove user message — this turn failed entirely
        this.history.pop();
        this.send({ type: "error", message: "Generation failed. Please try again." });
      }
    } finally {
      this.isStreaming = false;
      this.partialContent = "";
      this.abortController = null;
    }
  }
 
  abortAndSavePartial() {
    this.abortController?.abort();
  }
 
  private trimmedHistory(maxMessages = 20): Message[] {
    if (this.history.length <= maxMessages) return this.history;
    return this.history.slice(-maxMessages);
  }
 
  send(data: object) {
    if (this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(data));
    }
  }
 
  cleanup() {
    unregisterUserHandler(this.userId);
  }
}

The clear message type is worth adding early. Users expect a "new conversation" button, and clearing server-side history ensures the AI doesn't carry stale context across sessions.

Redis Pub/Sub for Multi-Instance Routing

// src/pubsub.ts
import { getRedisClient, getSubscriberClient } from "./redis.js";
 
type MessageHandler = (message: object) => void;
const localHandlers = new Map<string, MessageHandler>();
 
export async function initPubSub() {
  const sub = getSubscriberClient();
  await sub.psubscribe("user:*:messages");
 
  sub.on("pmessage", (_pattern, channel, raw) => {
    // channel format: "user:{userId}:messages"
    const userId = channel.split(":")[1];
    const handler = localHandlers.get(userId);
    if (!handler) return; // This user isn't on this instance
 
    try {
      handler(JSON.parse(raw));
    } catch (err) {
      console.error("Pub/Sub parse error:", err);
    }
  });
 
  console.log("✅ Pub/Sub initialized");
}
 
export function registerUserHandler(userId: string, handler: MessageHandler) {
  localHandlers.set(userId, handler);
}
 
export function unregisterUserHandler(userId: string) {
  localHandlers.delete(userId);
}
 
// Callable from any instance to reach any user
export async function publishToUser(userId: string, message: object) {
  try {
    const pub = getRedisClient();
    await pub.publish(`user:${userId}:messages`, JSON.stringify(message));
  } catch (err) {
    console.error(`Failed to publish to user ${userId}:`, err);
  }
}

Every instance subscribes to user:*:messages via pattern subscription. When a message arrives, each instance checks its own localHandlers map. If the target user isn't on this instance, localHandlers.get(userId) returns undefined and nothing happens. No cross-instance coordination needed.

JWT Authentication and Token Budget Control

// src/auth.ts
import jwt from "jsonwebtoken";
 
export interface JwtPayload {
  userId: string;
  plan: "free" | "pro" | "premium";
  iat: number;
  exp: number;
}
 
export function verifyToken(token: string): JwtPayload {
  return jwt.verify(token, process.env.JWT_SECRET!) as JwtPayload;
}
 
export function createToken(userId: string, plan: JwtPayload["plan"]): string {
  return jwt.sign({ userId, plan }, process.env.JWT_SECRET!, { expiresIn: "24h" });
}

// src/budget.ts
import { getRedisClient } from "./redis.js";
import type { JwtPayload } from "./auth.js";
 
const PLAN_LIMITS: Record<JwtPayload["plan"], number> = {
  free: 50_000,
  pro: 500_000,
  premium: 5_000_000,
};
 
export interface BudgetResult {
  allowed: boolean;
  used: number;
  limit: number;
  remaining: number;
}
 
function monthKey(userId: string): string {
  return `budget:${userId}:${new Date().toISOString().slice(0, 7)}`;
}
 
export async function checkBudget(
  userId: string,
  plan: JwtPayload["plan"]
): Promise<BudgetResult> {
  const redis = getRedisClient();
  const used = parseInt((await redis.get(monthKey(userId))) ?? "0", 10);
  const limit = PLAN_LIMITS[plan];
  return {
    allowed: used < limit,
    used,
    limit,
    remaining: Math.max(0, limit - used),
  };
}
 
export async function recordTokenUsage(userId: string, tokens: number): Promise<void> {
  const redis = getRedisClient();
  const key = monthKey(userId);
  await redis.incrby(key, tokens);
  await redis.expire(key, 60 * 60 * 24 * 45); // 45-day TTL covers billing period overlaps
}

For Claude Sonnet 4.6 ($3 input / $15 output per 1M tokens, per the official pricing page), a 50,000-token free tier costs about $0.15 if every token is input and about $0.75 if every token is output; real conversations land in between. Adjust based on your pricing model.

Client Implementation (React / React Native)

// hooks/useChatWebSocket.ts
import { useState, useEffect, useRef, useCallback } from "react";
 
interface ChatMessage {
  role: "user" | "assistant";
  content: string;
  partial?: boolean;
}
 
export function useChatWebSocket(token: string) {
  const [messages, setMessages] = useState<ChatMessage[]>([]);
  const [streamingContent, setStreamingContent] = useState("");
  const [isStreaming, setIsStreaming] = useState(false);
  const [isConnected, setIsConnected] = useState(false);
  const wsRef = useRef<WebSocket | null>(null);
  const retryRef = useRef(0);
  const maxRetries = 5;
 
  const connect = useCallback(() => {
    if (wsRef.current?.readyState === WebSocket.OPEN) return;
 
    const wsUrl = `${process.env.NEXT_PUBLIC_WS_URL}?token=${encodeURIComponent(token)}`;
    const ws = new WebSocket(wsUrl);
 
    ws.onopen = () => {
      setIsConnected(true);
      retryRef.current = 0;
    };
 
    ws.onmessage = ({ data }) => {
      const msg = JSON.parse(data);
 
      switch (msg.type) {
        case "stream_start":
          setIsStreaming(true);
          setStreamingContent("");
          break;
 
        case "delta":
          setStreamingContent((prev) => prev + msg.content);
          break;
 
        case "stream_end":
          setIsStreaming(false);
          // Move accumulated content into the messages array
          setMessages((prev) => {
            const latest = prev[prev.length - 1];
            if (latest?.partial) {
              // Replace the in-progress placeholder
              return [
                ...prev.slice(0, -1),
                { role: "assistant", content: streamingContent },
              ];
            }
            return [...prev, { role: "assistant", content: streamingContent }];
          });
          setStreamingContent("");
          break;
 
        case "stream_stopped":
          setIsStreaming(false);
          if (msg.partialContent) {
            setMessages((prev) => [
              ...prev,
              { role: "assistant", content: msg.partialContent + " [stopped]" },
            ]);
          }
          setStreamingContent("");
          break;
 
        case "error":
          setIsStreaming(false);
          console.error("Server:", msg.message);
          break;
 
        case "cleared":
          setMessages([]);
          setStreamingContent("");
          break;
      }
    };
 
    ws.onclose = ({ code }) => {
      setIsConnected(false);
      wsRef.current = null;
 
      if (code === 4001) return; // Auth failure — don't retry
      if (code === 4002) {
        // Replaced by new connection elsewhere — reconnect immediately
        connect();
        return;
      }
 
      if (retryRef.current < maxRetries) {
        const delay = Math.min(1000 * 2 ** retryRef.current, 30_000);
        retryRef.current++;
        setTimeout(connect, delay);
      }
    };
 
    wsRef.current = ws;
  }, [token]);
 
  useEffect(() => {
    connect();
    return () => wsRef.current?.close();
  }, [connect]);
 
  const sendMessage = useCallback((content: string) => {
    if (wsRef.current?.readyState !== WebSocket.OPEN) return;
    setMessages((prev) => [...prev, { role: "user", content }]);
    wsRef.current.send(JSON.stringify({ type: "chat", content }));
  }, []);
 
  const stopGeneration = useCallback(() => {
    wsRef.current?.send(JSON.stringify({ type: "stop" }));
  }, []);
 
  const clearHistory = useCallback(() => {
    wsRef.current?.send(JSON.stringify({ type: "clear" }));
  }, []);
 
  return {
    messages,
    streamingContent,
    isStreaming,
    isConnected,
    sendMessage,
    stopGeneration,
    clearHistory,
  };
}

The exponential backoff on reconnect (1s → 2s → 4s → 8s → 16s → 30s max) matters most on mobile. Apps that get backgrounded, switch networks, or wake from sleep will have their WebSocket connection dropped without a clean close event. Without backoff logic, aggressive reconnect attempts can trigger rate limiting.

Three Pitfalls That Will Catch You in Production

Pitfall 1: Context Window Exhaustion from Growing History

Long sessions accumulate history, driving up input tokens on every turn. Even a 200K context window eventually fills. The trimmedHistory(20) function handles the immediate problem, but for real sessions you want a summarization step:

async function buildContextWithSummary(
  history: Message[],
  keepRecent = 10
): Promise<Message[]> {
  if (history.length <= keepRecent) return history;
 
  const toSummarize = history.slice(0, -keepRecent);
  const recent = history.slice(-keepRecent);
 
  // Use Haiku for cheap summarization
  const response = await anthropic.messages.create({
    model: "claude-haiku-4-5-20251001",
    max_tokens: 300,
    messages: [
      {
        role: "user",
        content:
          `Summarize the key points from this conversation in 2-3 sentences:\n\n` +
          toSummarize.map((m) => `${m.role}: ${m.content}`).join("\n"),
      },
    ],
  });
 
  const summary =
    response.content[0].type === "text" ? response.content[0].text : "";
 
  return [
    { role: "assistant", content: `[Earlier conversation: ${summary}]` },
    ...recent,
  ];
}

Haiku summarization costs a fraction of Sonnet, and a 300-token summary replacing 20+ messages is a significant net saving.

Pitfall 2: Race Condition on Rapid Message Sends

If a client sends two messages before the first stream finishes, the second should be rejected cleanly — not queued, not processed out of order. The isStreaming guard handles this, but make sure the error message tells the client what to do:

if (this.isStreaming) {
  this.send({
    type: "error",
    message: "Generation in progress. Send { type: 'stop' } to interrupt.",
  });
  return;
}

Clear error messages prevent clients from implementing their own queuing logic in wrong ways.

Pitfall 3: Memory Leak from History on Long-Lived Connections

In-process history arrays on ChatSession objects grow without bound if sessions are never cleared. Set a max turn count or a session TTL, and implement the clear command on your client's "new chat" button:

// Add to ChatSession
private readonly MAX_HISTORY_TURNS = 50;
 
private async handleChat(content: string) {
  // Auto-clear extremely long sessions
  if (this.history.length > this.MAX_HISTORY_TURNS * 2) {
    this.history = this.history.slice(-this.MAX_HISTORY_TURNS * 2);
  }
  // ...rest of implementation
}

What Actually Happens Between the Stop Button and the Last Billed Token

When I moved this backend to WebSocket, what I expected was speed.

It wasn't. Time-to-first-token barely moved. That window is dominated by how long Claude API takes to start generating, and swapping the transport underneath it doesn't shorten that.

What changed was whether generation actually stops when you ask it to.

I hit this while working on the support chat backend for a wallpaper app I build on my own. With SSE plus a REST cancel endpoint, pressing stop means opening a second connection and waiting for the round trip. Throughout that round trip, Claude API is still writing. Everything it writes is billed.

Measure the gap before you argue about it

Don't reason about this from first principles — instrument it. Record how many tokens arrive between the stop request and the moment the stream actually ends.

// src/abort-metrics.ts
type StopTrace = { requestedAt: number; tokensAfterStop: number };
 
const traces = new Map<string, StopTrace>();
 
// Called the instant a stop message arrives over the socket
export function markStopRequested(streamId: string) {
  traces.set(streamId, { requestedAt: Date.now(), tokensAfterStop: 0 });
}
 
// Called on every delta pushed to the client, counting only post-stop output
export function countDeltaAfterStop(streamId: string, text: string) {
  const t = traces.get(streamId);
  if (!t) return;
  // A rough estimate is fine here; exact counts come from message_delta usage
  t.tokensAfterStop += Math.ceil(text.length / 4);
}
 
// Called when the stream genuinely terminates
export function finalizeStop(streamId: string): StopTrace | null {
  const t = traces.get(streamId);
  if (!t) return null;
  traces.delete(streamId);
  console.info(
    JSON.stringify({
      event: "stop_latency",
      streamId,
      latencyMs: Date.now() - t.requestedAt,
      tokensAfterStop: t.tokensAfterStop,
    })
  );
  return t;
}

Collect latencyMs and tokensAfterStop for a week before deciding anything. Every judgment below rests on those two numbers.

Stop latency converts directly into money

Once you know your stop latency, unit price and generation speed turn it into a dollar figure.

The assumptions here are explicit: output priced at $10 per 1M tokens (the Claude Sonnet 5 output rate listed on the official pricing page, checked in October 2026), generation at roughly 50 tokens per second, and 3,000 user-initiated interruptions per month. Generation speed varies by model and by what's being written, so substitute your own logged values.

Stop path	Typical stop latency	Wasted tokens per stop	Added cost at 3,000 stops/month
Call abort() directly on the socket	~100ms	~5	~$0.15
Separate REST cancel endpoint	~500ms	~25	~$0.75
Buffering proxy sitting in front	~2,000ms	~100	~$3.00
No stop path wired at all (generation runs to completion)	—	~400	~$12.00

In pure dollars that's a difference of a few tens per month. That alone is not a reason to choose WebSocket, and I'd rather say so plainly than oversell it.

The reason sits one layer further in. Tokens billed after a stop drift your per-user TokenBudget away from reality. A user rejected by a budget that quietly overcounted doesn't file a code bug — they lose trust. That, more than the invoice, is why I care about the stop path.

Running the code turned up two holes

I had assumed the stop path worked because the happy path did. This time I ran the code against a mock SSE server (Node.js 22, @anthropic-ai/sdk 0.133.0) and ioredis-mock, and only then saw that I was wrong in two places. (The Japanese edition also had a budget key mismatch, with usage recorded under one key and read from another, which I have fixed there.)

Hole 1: a stop never arrives as AbortError. I called abort() on the fifth delta and inspected the exception:

What I checked	Result
err.name	"Error"
err.constructor.name	APIUserAbortError
err instanceof Anthropic.APIUserAbortError	true

A branch on err.name === "AbortError" is never taken, so every stop falls into the generic failure path: the partial text is dropped and the client never gets stream_stopped. The code above now uses instanceof Anthropic.APIUserAbortError.

Hole 2: reserve before you spend. Even with a single key, "read now, record later" lets concurrent requests all pass. With 50,000 tokens remaining, I fired 20 simultaneous requests of 10,000 tokens each:

Approach	Requests allowed	Final usage
Read first, record afterwards	20	200,000
Reserve first, decide on the return value	5	50,000

The same user sending from two devices or two instances at once is ordinary in production. INCRBY returns a distinct value per call, so "reserve, and give it back if over the limit" needs no Lua script:

// src/budget.ts (addition): reserve, then settle
export async function reserveBudget(
  userId: string,
  plan: JwtPayload["plan"],
  reserveTokens: number
): Promise<{ ok: boolean }> {
  const redis = getRedisClient();
  const key = monthKey(userId);
  const after = await redis.incrby(key, reserveTokens);
  if (after > PLAN_LIMITS[plan]) {
    await redis.decrby(key, reserveTokens); // give back what we over-took
    return { ok: false };
  }
  await redis.expire(key, 60 * 60 * 24 * 45);
  return { ok: true };
}
 
// Once the real usage is known, apply only the difference
// (on stop or error, settle with whatever actually arrived)
export async function settleBudget(
  userId: string,
  reserved: number,
  actual: number
): Promise<void> {
  const redis = getRedisClient();
  await redis.incrby(monthKey(userId), actual - reserved);
}

Reserving max_tokens plus a rough input estimate is enough. Call settleBudget from finally so a stop or an error never leaves the reservation stuck. I have not run the settle path under load in these tests, so please try it with two simultaneous runs in your own environment before relying on it.

Check the key that was written, not the number on screen. That is the rule I now apply to every function whose name promises more than I have read.

Three assumptions the docs never spell out

1. Without abort() in your close handler, generation keeps going.

A dropped client connection does not stop the Claude API stream. Tokens continue to be produced with nowhere to go, and they are billed. The abortController.abort() inside ws.on("close") above isn't only about history integrity.

2. Redis Pub/Sub does not retain undelivered messages.

Pub/Sub drops messages when no subscriber is listening. Put your stop signal exclusively on Pub/Sub and it will vanish during the few seconds a subscribing instance is restarting. Generation won't stop.

The fix is straightforward: abort locally when the connection lives on this instance, and use Pub/Sub only for cross-instance routing.

async function requestStop(userId: string) {
  const local = sessions.get(userId);
  if (local) {
    local.abortController?.abort(); // Same instance: the reliable path
    return;
  }
  // Only reach for Pub/Sub when the socket is on another instance
  await publisher.publish(`user:${userId}`, JSON.stringify({ type: "stop" }));
}

3. Load balancer idle timeouts apply to WebSocket connections too.

AWS ALB defaults to a 60-second idle timeout, and it applies to persistent WebSocket connections. A user who pauses to think gets disconnected, and their next message silently goes nowhere.

Keep your heartbeat interval below the idle timeout — 30 seconds against a 60-second default gives you margin. If Nginx sits in front, raise proxy_read_timeout explicitly for the same reason.

Which setup fits which situation

Situation	Recommended setup	Why
One-shot Q&A, no interruption	Stay on SSE	WebSocket's operational cost buys you nothing here
Interruption needed, single instance is enough	WebSocket only (no Redis)	The stop path stays in-process, which is the most reliable it gets
Interruption needed, multiple instances	WebSocket + Redis Pub/Sub	Local abort first, Pub/Sub as the cross-instance fallback
Strict per-user budgets	WebSocket + continuous stop-latency measurement	Budget accuracy is bounded by how fast your stop path is

Production Deployment Checklist

[ ] JWT_SECRET must be a strong random value (32+ chars) in production — never a hardcoded string
[ ] Terminate SSL at Nginx / ALB; enforce wss:// only in production
[ ] ANTHROPIC_API_KEY stored as a secret in your deployment platform — not in .env committed to source control
[ ] Redis password enabled (requirepass) and TLS configured for Redis connections
[ ] Load balancer configured for WebSocket sticky sessions or Redis Pub/Sub routing active — both work, pick one
[ ] PM2 or systemd for process management with automatic restart
[ ] Heartbeat interval active to clear stale connections
[ ] Monitoring: active connection count, Claude API error rate, Redis latency percentiles
[ ] Per-user concurrent connection limit enforced (one active WebSocket per userId)

Looking down this list, most items share one purpose: making a problem visible before it becomes an incident. Connection counts and error rates tell you nothing if no graph exists to read them from, so setting up the monitoring view before the first deploy may save you a long night later.

Performance Optimization: Reducing Latency and Cost

Once your basic implementation is working, there are several optimizations worth considering before handling production traffic.

Reduce Input Tokens with Prompt Caching

If your system prompt is more than a few hundred tokens, enable prompt caching for it. On Claude Sonnet 4.6, cached tokens cost 10% of the standard input price:

const stream = await anthropic.messages.stream({
  model: "claude-sonnet-4-6",
  max_tokens: 2048,
  system: [
    {
      type: "text",
      text: yourSystemPromptText, // Must be 1024+ tokens to qualify for caching
      cache_control: { type: "ephemeral" },
    },
  ],
  messages: this.trimmedHistory(),
});

For a system prompt of 2,000 tokens, this reduces the system prompt cost by 90% on cache hits. Cache lifetime is 5 minutes — in a WebSocket session where messages arrive frequently, hit rates are very high.

Avoid Blocking the Event Loop

Claude API requests are async and handled well by Node.js's event loop, but watch out for synchronous operations in your message handlers. JSON parsing of large messages, complex history trimming logic, or synchronous crypto operations can block other connections from being served.

If you're doing heavy processing in a message handler, offload it:

ws.on("message", (data) => {
  // Don't await synchronously in the message handler
  // Let it return immediately and process asynchronously
  setImmediate(async () => {
    try {
      const message = JSON.parse(data.toString());
      await session.handleMessage(message);
    } catch (err) {
      session.send({ type: "error", message: "Processing failed" });
    }
  });
});

Connection Pool Sizing

Each Node.js WebSocket server instance can handle thousands of concurrent connections. The practical limit is determined by memory (each ChatSession holds its history in memory), CPU (Anthropic SDK processing), and your Claude API rate limits.

A rough guideline: with max_tokens: 2048 and histories capped at 20 messages, each ChatSession uses roughly 50-200KB of memory. For 1,000 concurrent users, that's 50-200MB — well within a standard 1GB instance. API rate limits are typically the binding constraint long before memory is.

Testing Your Implementation

Testing WebSocket servers requires a slightly different approach than testing REST APIs.

Integration Test with ws Client

// test/chat.integration.ts
import WebSocket from "ws";
import { createToken } from "../src/auth.js";
import { describe, it, before, after } from "node:test";
import assert from "node:assert";
 
describe("ChatSession WebSocket", () => {
  let ws: WebSocket;
  const token = createToken("test-user", "free");
 
  before((done) => {
    ws = new WebSocket(`ws://localhost:8080?token=${token}`);
    ws.on("open", done);
  });
 
  after(() => ws.close());
 
  it("should receive stream_start, delta, and stream_end for a chat message", async () => {
    const events: string[] = [];
 
    ws.send(JSON.stringify({ type: "chat", content: "Say 'hello' and nothing else." }));
 
    await new Promise<void>((resolve) => {
      ws.on("message", (data) => {
        const msg = JSON.parse(data.toString());
        events.push(msg.type);
        if (msg.type === "stream_end") resolve();
      });
    });
 
    assert.equal(events[0], "stream_start");
    assert.ok(events.includes("delta"), "Expected at least one delta");
    assert.equal(events[events.length - 1], "stream_end");
  });
 
  it("should stop generation when stop message is sent", async () => {
    ws.send(JSON.stringify({ type: "chat", content: "Count from 1 to 1000 slowly." }));
 
    // Wait for stream to start, then stop it
    await new Promise<void>((resolve) => {
      ws.once("message", (data) => {
        const msg = JSON.parse(data.toString());
        if (msg.type === "stream_start") {
          setTimeout(() => {
            ws.send(JSON.stringify({ type: "stop" }));
            resolve();
          }, 200);
        }
      });
    });
 
    const stopEvent = await new Promise<string>((resolve) => {
      ws.on("message", (data) => {
        const msg = JSON.parse(data.toString());
        if (msg.type === "stream_stopped" || msg.type === "stream_end") {
          resolve(msg.type);
        }
      });
    });
 
    assert.equal(stopEvent, "stream_stopped");
  });
});

Running integration tests against a real WebSocket server (with a test API key) catches timing issues and protocol mismatches that unit tests miss. Budget extra time for this — WebSocket test infrastructure is less mature than REST testing tooling.

Monitoring: What to Watch in Production

Beyond basic uptime monitoring, a few WebSocket-specific metrics are worth tracking:

Active connection count: Should scale smoothly with traffic. Sudden drops suggest connection handling bugs; slow growth that never decreases suggests cleanup bugs.
Message round-trip time: Time from stream_start to stream_end. High P95 latency often means Claude API load, not your server.
Abort rate: What percentage of streams get stopped by users? High abort rates suggest responses are too long or off-topic — a prompt engineering issue, not a technical one.
Reconnection rate: How often clients reconnect? High rates indicate network instability or server-side crash loops.
Token usage per user per day: Catch anomalous usage before it becomes a billing surprise.

These can all be tracked with simple Redis counters and exposed via a /metrics endpoint for Prometheus or DataDog.

Setting Up Basic Metrics with Redis

Here's a lightweight metrics collector that doesn't require external dependencies:

// src/metrics.ts
import { getRedisClient } from "./redis.js";
 
export async function recordMetric(name: string, value = 1) {
  try {
    const redis = getRedisClient();
    const key = `metrics:${name}:${new Date().toISOString().slice(0, 13)}`; // hourly buckets
    await redis.incrby(key, value);
    await redis.expire(key, 60 * 60 * 72); // Keep 3 days of hourly data
  } catch {
    // Never let metrics fail silently through to the caller
  }
}
 
// Usage in ChatSession
// In stream_end handler:
await recordMetric("stream_completed");
await recordMetric("tokens_used", tokensUsed);
 
// In stream_stopped handler:
await recordMetric("stream_aborted");
 
// In connection handler:
await recordMetric("connections_opened");

Expose an aggregated view via a health endpoint:

// In your HTTP server
httpServer.on("request", async (req, res) => {
  if (req.url === "/health") {
    res.writeHead(200, { "Content-Type": "application/json" });
    res.end(
      JSON.stringify({
        status: "ok",
        connections: wss.clients.size,
        uptime: process.uptime(),
      })
    );
  }
});

This gives your load balancer a reliable health check endpoint and gives you a quick way to verify connection counts from outside the process.

Where to Go Next

This architecture — WebSocket for bidirectional transport, Redis Pub/Sub for cross-instance routing, Claude API streaming for generation, JWT plus budget control for cost management — is a solid production foundation.

The natural next steps are persistent conversation storage (persisting history beyond in-memory and Redis TTL to a proper database) and prompt caching to reduce input token costs on long conversations. For the caching implementation, see the notes on halving monthly Claude API cost with prompt caching.

For handling the full range of production resilience concerns — model fallbacks, circuit breakers, rate limit handling — the Claude API Production Resilience Patterns guide covers what you'll need before your first real traffic spike.

Start with a single instance and no Redis. Get the chat loop running end-to-end, verify the stop button works, then layer in Redis when you're ready to scale. The implementation above is designed so Pub/Sub is an additive change — not a rewrite.

Thank You for Reading

Claude Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.

✦Copy-paste ready implementation code
✦New advanced guides published daily
✦$5/mo or $10 for lifetime access

View Membership →