Node Streams, Web Streams and SSE in One Model

Node has two stream implementations and the browser has one of them. Server-sent events is not a stream implementation at all, despite being discussed alongside them. Getting these three straight removes most of the confusion around streaming a model through a Node backend.

The three things

The confusion produces a specific class of unhelpful error. Node streams and Web Streams are two implementations of the same idea, one portable and one not. SSE is a text format that travels over either. When you see stream.pipe is not a function, you have a Web Stream and code expecting a Node one; when you see body.getReader is not a function, the reverse.

The practical guidance: write new code against Web Streams. It runs everywhere, it is what fetch hands you at both ends, and it means the SSE parser you wrote for the browser runs unchanged on the server.

Converting between them

Node 18 and later ship both directions as static methods on Readable. These are the two lines that let a Web-Streams codebase talk to an Express or Fastify handler.

import { Readable } from "node:stream";

// Web Stream -> Node stream. Use when a Node API must consume it,
// for example piping into an Express response.
const nodeReadable = Readable.fromWeb(webStream as any);

// Node stream -> Web Stream. Use when web-standard code must consume
// something a Node library produced.
const webReadable = Readable.toWeb(nodeReadable);

// The whole proxy, in a plain Node HTTP server.
import { Readable } from "node:stream";
import { createServer } from "node:http";

createServer(async (req, res) => {
  const upstream = await fetch(PROVIDER_URL, { method: "POST", headers: H, body: B });

  res.writeHead(200, {
    "Content-Type": "text/event-stream; charset=utf-8",
    "Cache-Control": "no-cache, no-transform",
    "Connection": "keep-alive",
    "X-Accel-Buffering": "no",
  });

  // fetch gives a Web Stream; res wants a Node one.
  Readable.fromWeb(upstream.body as any).pipe(res);
}).listen(3000);

SSE is a format, not an API

The format is small enough to state completely, and knowing it completely is what lets you debug a stream with curl instead of a library.

event: token                 <- optional; the client's event name
data: {"delta":"Hello"}      <- the payload; may repeat, joined with "\n"
id: 42                       <- optional; sent back as Last-Event-ID on reconnect
retry: 3000                  <- optional; reconnection delay in ms
                             <- BLANK LINE dispatches the event

: this is a comment and is ignored — the keepalive trick

data: [DONE]                 <- convention, not part of the specification

Four rules cover every SSE bug worth knowing about. The blank line is what dispatches an event, and forgetting the second newline is the single most common mistake — the data sits in the client’s buffer forever and nothing appears. A payload must not contain a raw newline, so JSON.stringify it or split it across multiple data: lines. The content type must be exactly text/event-stream. And [DONE] is a convention from the OpenAI API rather than something the specification knows about, so a strict SSE client hands it to you as a data payload and JSON.parse throws on it.

A complete Express endpoint

Everything above assembled: parse the upstream SSE, re-emit your own event shape, keep the connection alive, and clean up when the client disconnects.

// server.ts — Node 18+, Express 4
import express from "express";

const app = express();
app.use(express.json());

app.post("/api/chat", async (req, res) => {
  const { prompt } = req.body ?? {};
  if (typeof prompt !== "string" || prompt.length > 4000) {
    return res.status(400).json({ error: "bad prompt" });
  }

  res.writeHead(200, {
    "Content-Type": "text/event-stream; charset=utf-8",
    "Cache-Control": "no-cache, no-transform",
    "Connection": "keep-alive",
    "X-Accel-Buffering": "no",
  });
  res.flushHeaders();

  const send = (event: string, data: unknown) => {
    res.write("event: " + event + "\n");
    res.write("data: " + JSON.stringify(data) + "\n\n");
  };

  const keepalive = setInterval(() => res.write(": ping\n\n"), 15_000);
  const controller = new AbortController();
  req.on("close", () => controller.abort());     // the client went away

  try {
    const upstream = await fetch("https://api.multigrid.ai/v1/chat/completions", {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: "Bearer " + process.env.LLM_API_KEY,
      },
      body: JSON.stringify({
        model: "openai/gpt-4o-mini",
        messages: [{ role: "user", content: prompt }],
        stream: true,
        max_tokens: 800,
      }),
      signal: controller.signal,
    });

    if (!upstream.ok || !upstream.body) {
      send("error", { message: "upstream " + upstream.status });
      return res.end();
    }

    const reader = upstream.body.getReader();
    const decoder = new TextDecoder();
    let buffer = "";
    let chars = 0;

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const parts = buffer.split("\n\n");
      buffer = parts.pop() ?? "";

      for (const part of parts) {
        const line = part.split("\n").find((l) => l.startsWith("data:"));
        if (!line) continue;
        const payload = line.slice(5).trim();
        if (payload === "[DONE]") continue;

        try {
          const chunk = JSON.parse(payload);
          const delta = chunk.choices?.[0]?.delta?.content;
          if (delta) {
            chars += delta.length;
            send("token", { delta });
          }
        } catch {
          // A single malformed event is not worth killing the stream over.
        }
      }
    }

    send("done", { chars });
  } catch (err) {
    if ((err as Error).name !== "AbortError") {
      send("error", { message: "stream failed" });
    }
  } finally {
    clearInterval(keepalive);      // every exit path, including the abort
    res.end();
  }
});

app.listen(3000);

Two things there are worth copying even if you never use Express. res.flushHeaders() sends the headers before the first chunk, which is what tells intermediaries the response has started. And the req.on("close") handler is the Node equivalent of request.signal in a web-standard handler — without it, a user closing the tab leaves you reading tokens you are still paying for, exactly as in cancelling an in-flight request.

Backpressure, and how to lose it

Backpressure stops a fast producer exhausting memory in front of a slow consumer. Both stream implementations have it, and both lose it the same way: when you step outside the pipe and write manually.

// Backpressure preserved: pipe and pipeThrough handle it for you.
Readable.fromWeb(upstream.body).pipe(res);
return new Response(upstream.body.pipeThrough(transform));

// Backpressure lost: res.write() returns false when the buffer is full,
// and ignoring that return value buffers without limit in your process.
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  res.write(value);            // <- return value dropped
}

For a model stream this is usually harmless, because generation is slower than any network you are writing to. It stops being harmless when the consumer is a phone on a poor connection, or when you are replaying a stored generation at full speed rather than at generation speed. The correct manual form:

if (!res.write(value)) {
  await new Promise<void>((resolve) => res.once("drain", resolve));
}

Or, better, do not write manually. Use pipe or pipeThrough and the problem does not exist.

Where errors go once headers are sent

The awkward truth about streaming an HTTP response: once headers have been sent, the status code cannot change. A failure at token 400 of 800 arrives inside a response that already said 200 OK. There are exactly two options and you must pick one deliberately.

• Send an error event. The client sees event: error and renders a message. The right choice for anything a human is watching, and why the endpoint above defines an error event alongside token and done.
• Destroy the connection. The client’s read() rejects with a network error. Blunt, gives no explanation, and appropriate when the failure is such that you do not trust anything you would say about it.

Whichever you choose, the client must treat “stream ended without a done event” as a failure. Without that check a connection dropped by a proxy is indistinguishable from an answer the model chose to end, and the user is shown a truncated answer as if it were complete.