Type-safe NestJS controllers for streaming chat

Building a nestjs typed streaming chat controller forces you to reconcile TypeScript’s compile-time guarantees with the messy reality of token-by-token HTTP streams. This guide gives an ordered path from shared DTOs to a production-ready POST+SSE endpoint that talks to any OpenAI-compatible LLM API.

1. Define the contract first

Type safety starts at the edges. Before writing controllers, declare the request shape and the streaming chunk shape.

export type ChatRole = 'system' | 'user' | 'assistant';

export interface ChatMessage {
  role: ChatRole;
  content: string;
}

export interface ChatCompletionRequest {
  model: string;
  messages: ChatMessage[];
  stream: true;
}

export interface ChatCompletionChunk {
  id: string;
  object: 'chat.completion.chunk';
  choices: Array<{
    index: number;
    delta: { role?: ChatRole; content?: string };
    finish_reason: null | 'stop' | 'length';
  }>;
}

These interfaces are your single source of truth. The controller consumes ChatCompletionRequest; the service emits MessageEvent whose data is a stringified ChatCompletionChunk.

Why not reuse the SDK types

The official SDKs ship their own types, but they change across versions and include fields you don’t need. Defining a minimal contract keeps your nestjs typed streaming chat controller decoupled from provider churn.

2. Pick the transport

NestJS offers @Sse() (GET-only) but chat requires a body. Use @Post() with manual headers. This keeps the request JSON-typed and avoids query-string hacks.

import { Controller, Post, Body, Header } from '@nestjs/common';
import { Observable } from 'rxjs';
import { MessageEvent } from '@nestjs/common';

@Controller('v1/chat')
export class ChatController {
  constructor(private readonly chat: ChatService) {}

  @Post('stream')
  @Header('Content-Type', 'text/event-stream')
  @Header('Cache-Control', 'no-cache')
  @Header('Connection', 'keep-alive')
  streamChat(@Body() body: ChatCompletionRequest): Observable<MessageEvent> {
    return this.chat.stream(body);
  }
}

The return type Observable<MessageEvent> is enforced by Nest’s response handling. If ChatService.stream returns the wrong generic, the build fails.

3. Implement the service stream

The service opens a fetch to an OpenAI-compatible endpoint, reads the raw byte stream, and maps each parsed chunk to a MessageEvent.

import { Injectable } from '@nestjs/common';
import { Observable, from, mergeMap } from 'rxjs';
import { MessageEvent } from '@nestjs/common';

@Injectable()
export class ChatService {
  stream(req: ChatCompletionRequest): Observable<MessageEvent> {
    return from(
      fetch('https://api.openai.com/v1/chat/completions', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          Authorization: `Bearer ${process.env.OPENAI_API_KEY}`,
        },
        body: JSON.stringify(req),
      })
    ).pipe(
      mergeMap(async (res) => {
        if (!res.body) throw new Error('empty body');
        return this.readStream(res.body);
      }),
      mergeMap((obs) => obs)
    );
  }

  private readStream(body: ReadableStream<Uint8Array>): Observable<MessageEvent> {
    const reader = body.getReader();
    const decoder = new TextDecoder();
    return new Observable<MessageEvent>((subscriber) => {
      const pump = async () => {
        while (true) {
          const { done, value } = await reader.read();
          if (done) {
            subscriber.complete();
            return;
          }
          const lines = decoder.decode(value).split('\n');
          for (const line of lines) {
            if (line.startsWith('data: ')) {
              const json = line.slice(6).trim();
              if (json === '[DONE]') {
                subscriber.complete();
                return;
              }
              const chunk = JSON.parse(json) as ChatCompletionChunk;
              subscriber.next({ data: JSON.stringify(chunk) });
            }
          }
        }
      };
      pump().catch((err) => subscriber.error(err));
    });
  }
}

If you route through n4n.ai, the single OpenAI-compatible endpoint exposes 240+ models and automatically falls back when a provider is degraded, so the controller above works without per-provider branching.

4. Keep types at the boundary

The JSON.parse call returns any. Cast it to ChatCompletionChunk as shown, but add a runtime guard in production:

function isChunk(x: unknown): x is ChatCompletionChunk {
  return typeof x === 'object' && x !== null && 'choices' in x;
}

Without this, a malformed provider response crashes the stream mid-flight. Tradeoff: the guard adds CPU per chunk, but at token latency it’s negligible.

5. Handle client disconnect

NestJS does not auto-unsubscribe your observable on socket close unless you use @Sse with RxJS takeUntil. For POST+SSE, inject Req and watch close.

import { Request } from 'express';
import { fromEvent, takeUntil } from 'rxjs';

@Post('stream')
@Header('Content-Type', 'text/event-stream')
streamChat(@Body() body: ChatCompletionRequest, @Req() req: Request) {
  const closed$ = fromEvent(req, 'close');
  return this.chat.stream(body).pipe(takeUntil(closed$));
}

This cancels the upstream fetch and frees memory. Forgetting this leaks connections under load.

6. Test without timers

Mock the fetch and push synthetic chunks:

const fakeStream = new ReadableStream({
  start(c) {
    c.enqueue(new TextEncoder().encode('data: {"choices":[]}\n\n'));
    c.close();
  },
});
// jest.mock('fetch', () => async () => ({ body: fakeStream }));

Assert the controller emits MessageEvent with correctly typed data. Use rxjs toArray() to collect the stream.

7. Pitfalls and tradeoffs

CORS: SSE needs Access-Control-Allow-Origin if called from browser. Set via Nest interceptor.

Buffering proxies: Nginx or Cloudflare may buffer the stream. Disable proxy buffering or set X-Accel-Buffering: no.

Type erosion: Avoid any in the controller. If you must adapt multiple providers, define a discriminated union and map each to your ChatCompletionChunk.

SSE vs WebSocket: SSE is simpler and natively reconnects, but is unidirectional. If you need mid-stream user corrections, use WebSocket and lose the free EventSource client.

Backpressure: Node’s fetch stream is pull-based; if the client is slow, the kernel socket buffers. Monitor memory.

A nestjs typed streaming chat controller is not just a wrapper around fetch. It is a typed boundary that survives provider differences, network flaps, and disconnecting clients.