Build it

A provider-agnostic streaming chat proxy in NestJS 10: per-user token bucket by plan, a bounded queue when capacity is full, SSE to the client, and every message saved in Postgres. Two of its core pieces, the stream parser in step 2 and the token bucket in step 3, are exercises you can run in your browser in the Practice section.

1. Create the tables

   Every assistant message has a status, because streams end in three ways: finished, stopped by the user, or broken by the provider. Token counts let you add up cost later.

   SQLPrisma

   SQLCopy

   CREATE TABLE conversation (
     id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
     user_id     integer        NOT NULL,
     title       text,
     created_at  timestamptz(3) NOT NULL DEFAULT now()
   );
   CREATE INDEX conversation_user_id_created_at_idx ON conversation (user_id, created_at);
   
   CREATE TABLE chat_message (
     id               bigserial PRIMARY KEY,
     conversation_id  uuid NOT NULL REFERENCES conversation (id),
     role             text NOT NULL CHECK (role IN ('user', 'assistant')),
     content          text NOT NULL,
     status           text NOT NULL DEFAULT 'complete'
                      CHECK (status IN ('complete', 'stopped', 'failed')),
     input_tokens     integer,
     output_tokens    integer,
     created_at       timestamptz(3) NOT NULL DEFAULT now()
   );
   CREATE INDEX chat_message_conversation_id_id_idx ON chat_message (conversation_id, id);

2. Hide the provider behind one interface

   Many hosted providers, and self-hosted servers such as vLLM, accept the OpenAI-style Chat Completions format and stream lines like data: {"choices":[{"delta":{"content":"Hel"}}]}, ending with data: [DONE]. This class reads that stream and turns it into simple events. The signal matters: when it fires, fetch closes the connection and the provider stops generating, so you stop paying. For a provider with its own SDK, write a second class with the same interface.

   TypeScriptCopy

   // src/llm/provider.ts: the only file that knows which LLM you call
   export interface ChatTurn { role: 'user' | 'assistant'; content: string }
   export interface StreamOpts { system: string; maxOutputTokens: number; signal: AbortSignal }
   export type ChatEvent =
     | { type: 'delta'; text: string }                                // a piece of the answer
     | { type: 'done'; inputTokens: number; outputTokens: number };   // final usage
   export interface ChatProvider { stream(turns: ChatTurn[], o: StreamOpts): AsyncIterable<ChatEvent> }
   export const CHAT_PROVIDER = Symbol('CHAT_PROVIDER');
   
   export class OpenAICompatibleProvider implements ChatProvider {
     constructor(private baseUrl: string, private apiKey: string, private model: string) {}
   
     async *stream(turns: ChatTurn[], o: StreamOpts): AsyncIterable<ChatEvent> {
       const res = await fetch(`${this.baseUrl}/v1/chat/completions`, {
         method: 'POST', signal: o.signal,                 // user left or deadline hit: stop paying
         headers: { 'content-type': 'application/json', authorization: `Bearer ${this.apiKey}` },
         body: JSON.stringify({ model: this.model, stream: true, max_tokens: o.maxOutputTokens,
           stream_options: { include_usage: true },        // last chunk carries token counts
           messages: [{ role: 'system', content: o.system }, ...turns] }),
       });
       if (!res.ok || !res.body) throw new Error(`provider HTTP ${res.status}`);
       let buf = '', usage = { inputTokens: 0, outputTokens: 0 };
       for await (const chunk of res.body.pipeThrough(new TextDecoderStream())) {
         buf += chunk;
         const lines = buf.split('\n'); buf = lines.pop() ?? '';   // keep a half line for later
         for (const line of lines) {
           if (!line.startsWith('data:')) continue;                 // skip blanks and comments
           const data = line.slice(5).trim();
           if (data === '[DONE]') { yield { type: 'done', ...usage }; return; }
           const j = JSON.parse(data);
           if (j.choices?.[0]?.delta?.content) yield { type: 'delta', text: j.choices[0].delta.content };
           if (j.usage) usage = { inputTokens: j.usage.prompt_tokens, outputTokens: j.usage.completion_tokens };
         }
       }
       yield { type: 'done', ...usage };
     }
   }

3. Limit each user by plan

   Reuse the atomic token bucket Lua script from Rate limiting (save it as token_bucket.lua next to this file and copy it into dist on build). Free users get a burst of 5 and then about 3 messages a minute; paid users a burst of 20 and 30 a minute. The limiter runs before the response starts, so a refused request is a normal HTTP 429.

   TypeScriptCopy

   // src/chat/chat-limiter.ts
   import { HttpException, HttpStatus, Injectable } from '@nestjs/common';
   import Redis from 'ioredis';
   import { readFileSync } from 'node:fs';
   import { join } from 'node:path';
   
   const PLANS = { free: { burst: 5, perS: 0.05 }, plus: { burst: 20, perS: 0.5 } }; // 3 vs 30 a minute
   export type Plan = keyof typeof PLANS;
   
   @Injectable()
   export class ChatLimiter {
     private lua = readFileSync(join(__dirname, 'token_bucket.lua'), 'utf8');
     constructor(private redis: Redis) {}
   
     /** Throws 429 with a retry time. Call it before any SSE bytes are sent. */
     async admit(userId: number, plan: Plan): Promise<void> {
       const { burst, perS } = PLANS[plan];
       const [allowed, , retryMs] =
         (await this.redis.eval(this.lua, 1, `chat:${userId}`, burst, perS, 1)) as number[];
       if (allowed !== 1) throw new HttpException(
         { error: 'rate_limited', retryAfterS: Math.ceil(retryMs / 1000) }, HttpStatus.TOO_MANY_REQUESTS);
     }
   }

4. Add a capacity gate with a short, bounded queue

   The provider, or your own model server, can run only so many answers at once. This gate lets maxActive calls run, keeps up to maxWaiting in line for at most maxWaitMs, and refuses the rest at once. A finished call hands its slot straight to the next in line. If the user leaves while waiting, they leave the line.

   TypeScriptCopy

   // src/chat/capacity-gate.ts
   export class GateFullError extends Error {}
   
   export class CapacityGate {
     private active = 0;
     private line: Array<() => void> = [];
   
     constructor(private maxActive: number, private maxWaiting: number, private maxWaitMs: number) {}
   
     /** Resolves with a release function. onQueued tells the caller its place in line. */
     async acquire(signal: AbortSignal, onQueued: (position: number) => void): Promise<() => void> {
       if (this.active < this.maxActive) { this.active++; return this.releaser(); }
       if (this.line.length >= this.maxWaiting) throw new GateFullError();   // shed load now
       onQueued(this.line.length + 1);
       await new Promise<void>((resolve, reject) => {
         const leave = (err: Error) => {
           clearTimeout(timer); this.line = this.line.filter((f) => f !== go); reject(err);
         };
         const go = () => { clearTimeout(timer); signal.removeEventListener('abort', onAbort); resolve(); };
         const onAbort = () => leave(new Error('client left while queued'));
         const timer = setTimeout(() => leave(new GateFullError()), this.maxWaitMs);
         signal.addEventListener('abort', onAbort, { once: true });
         this.line.push(go);
       });
       return this.releaser();                       // the slot was handed straight to us
     }
   
     private releaser(): () => void {
       let done = false;
       return () => {
         if (done) return; done = true;
         const next = this.line.shift();
         if (next) next(); else this.active--;       // hand the slot on, or free it
       };
     }
   }

   This gate counts per API copy: with 4 copies and maxActive 8, up to 32 calls run. For one shared limit across copies, use a Redis-based semaphore, or let your model server queue requests itself.

5. Check, wait, stream and save

   The service finds the conversation with the user's ID in the query, so a guessed ID returns nothing (the March 2023 lesson). It loads the last 20 complete messages, saves the new one, waits for a slot, streams, and in finally always releases the slot and saves whatever was written, with the right status.

   TypeScriptCopy

   // src/chat/chat.service.ts
   import { Inject, Injectable } from '@nestjs/common';
   import { PrismaClient } from '@prisma/client';
   import { CHAT_PROVIDER, ChatProvider, ChatTurn } from '../llm/provider';
   import { CapacityGate, GateFullError } from './capacity-gate';
   
   export type Send = (event: string, data: object) => void;          // writes one SSE event
   const SYSTEM = 'You help Mangrove Market store staff. Answer briefly.';
   
   @Injectable()
   export class ChatService {
     // at most MAX_ACTIVE model calls from this API copy; 50 may wait, each up to 20 s
     private gate = new CapacityGate(Number(process.env.MAX_ACTIVE ?? 8), 50, 20_000);
     constructor(private prisma: PrismaClient, @Inject(CHAT_PROVIDER) private llm: ChatProvider) {}
   
     async reply(userId: number, conversationId: string, text: string, send: Send, signal: AbortSignal) {
       const owned = await this.prisma.conversation.findFirst({ where: { id: conversationId, userId } });
       if (!owned) return send('error', { error: 'not_found' });          // never another user's chat
       const history = await this.prisma.chatMessage.findMany({ where: { conversationId, status: 'complete' },
         orderBy: { id: 'desc' }, take: 20, select: { role: true, content: true } });
       await this.prisma.chatMessage.create({ data: { conversationId, role: 'user', content: text } });
   
       let release: () => void;
       try { release = await this.gate.acquire(signal, (position) => send('queued', { position })); }
       catch (e) { if (e instanceof GateFullError) send('error', { error: 'busy', retryAfterS: 30 }); return; }
   
       let answer = '', status = 'complete', usage = { inputTokens: 0, outputTokens: 0 };
       try {
         send('start', {});
         const turns = [...history.reverse(), { role: 'user', content: text }] as ChatTurn[];
         for await (const ev of this.llm.stream(turns, { system: SYSTEM, maxOutputTokens: 800, signal })) {
           if (ev.type === 'delta') { answer += ev.text; send('delta', { text: ev.text }); }
           else usage = { inputTokens: ev.inputTokens, outputTokens: ev.outputTokens };
         }
         send('done', usage);
       } catch {
         status = signal.aborted ? 'stopped' : 'failed';                 // user left, or provider broke
         if (!signal.aborted) send('error', { error: 'model_failed' });
       } finally {
         release();                                                      // always give the slot back
         await this.prisma.chatMessage.create({ data: { conversationId, role: 'assistant', content: answer,
           status, inputTokens: usage.inputTokens || null, outputTokens: usage.outputTokens || null } });
       }
     }
   }

   The service does not retry a provider failure mid-answer: half an answer is already on the user's screen, and a retry would repeat it. Retry only before the first token, with backoff (Timeouts and retries, and the resilient client in Claude).

6. Write the SSE endpoint

   The message comes in a POST body, so this uses @Res() and writes the events by hand instead of the GET-only @Sse() from Real-time updates. In a browser, read it with fetch and a stream reader, since EventSource only does GET. When the user closes the tab, close fires, the signal aborts, and the provider call stops.

   TypeScriptCopy

   // src/chat/chat.controller.ts
   import { Body, Controller, Headers, Param, ParseUUIDPipe, Post, Res } from '@nestjs/common';
   import type { Response } from 'express';
   import { ChatLimiter, Plan } from './chat-limiter';
   import { ChatService } from './chat.service';
   
   @Controller('conversations')
   export class ChatController {
     constructor(private chat: ChatService, private limiter: ChatLimiter) {}
   
     @Post(':id/messages')
     async send(@Param('id', ParseUUIDPipe) id: string, @Body('text') text: string,
                @Headers('x-user-id') user: string, @Headers('x-plan') plan: Plan,  // stand-ins for
                @Res() res: Response) {                                            // your auth guard
       const userId = Number(user);
       await this.limiter.admit(userId, plan === 'plus' ? 'plus' : 'free');        // 429 happens here
       res.set({ 'content-type': 'text/event-stream', 'cache-control': 'no-cache',
                 'x-accel-buffering': 'no' });                                     // Nginx: don't buffer
       res.flushHeaders();
   
       const ac = new AbortController();
       res.on('close', () => { if (!res.writableEnded) ac.abort(); });            // tab closed: stop
       const send = (event: string, data: object) =>
         res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
       const ping = setInterval(() => res.write(': ping\n\n'), 15_000);           // keep proxies open
       try { await this.chat.reply(userId, id, String(text ?? '').slice(0, 4000), send, ac.signal); }
       finally { clearInterval(ping); res.end(); }
     }
   }

   The two headers are stand-ins so you can test with curl. In the real app, take the user and plan from your auth guard (Security essentials), never from headers the client controls.

7. Wire the module

   TypeScriptCopy

   // src/app.module.ts
   import { Module } from '@nestjs/common';
   import { PrismaClient } from '@prisma/client';
   import Redis from 'ioredis';
   import { ChatController } from './chat/chat.controller';
   import { ChatLimiter } from './chat/chat-limiter';
   import { ChatService } from './chat/chat.service';
   import { CHAT_PROVIDER, OpenAICompatibleProvider } from './llm/provider';
   
   @Module({
     controllers: [ChatController],
     providers: [
       ChatService, ChatLimiter,
       { provide: Redis, useFactory: () => new Redis(process.env.REDIS_URL ?? 'redis://localhost:6379') },
       { provide: PrismaClient, useFactory: () => new PrismaClient() },
       { provide: CHAT_PROVIDER, useFactory: () => new OpenAICompatibleProvider(
           process.env.LLM_BASE_URL!, process.env.LLM_API_KEY ?? '', process.env.LLM_MODEL!) },
     ],
   })
   export class AppModule {}

8. Test it with a pretend model

   You do not need a real model or an API key to test the plumbing. This fake server streams in the same format at about 7 tokens a second and logs when the client goes away.

   TypeScriptCopy

   // scripts/fake-llm.ts: a pretend model server that streams like the real thing
   import { createServer } from 'node:http';
   
   createServer((req, res) => {
     let body = '';
     req.on('data', (c) => (body += c));
     req.on('end', async () => {
       const { messages } = JSON.parse(body);
       const words = `You asked: ${messages.at(-1).content}. Check the GRN screen, then press Post.`.split(' ');
       res.writeHead(200, { 'content-type': 'text/event-stream' });
       let closed = false;
       res.on('close', () => { closed = true; });
       for (const w of words) {
         if (closed) { console.log('fake-llm: client went away, stopped generating'); return; }
         await new Promise((r) => setTimeout(r, 150));
         res.write(`data: ${JSON.stringify({ choices: [{ delta: { content: w + ' ' } }] })}\n\n`);
       }
       res.write(`data: ${JSON.stringify({ choices: [], usage: { prompt_tokens: 42, completion_tokens: words.length } })}\n\n`);
       res.end('data: [DONE]\n\n');
     });
   }).listen(8001, () => console.log('fake LLM on :8001'));

   Start the fake model and the API with only 2 slots, create a conversation, then send four messages at once, and seven quick ones from a free user.

   TerminalCopy

   npx ts-node scripts/fake-llm.ts &                     # the pretend model on :8001
   LLM_BASE_URL=http://localhost:8001 LLM_MODEL=fake MAX_ACTIVE=2 npm run start   # DATABASE_URL, REDIS_URL in .env
   
   CID=$(psql "$DATABASE_URL" -tA -c "INSERT INTO conversation (user_id) VALUES (4) RETURNING id" | head -1)
   
   curl -N -X POST localhost:3000/conversations/$CID/messages \
     -H 'x-user-id: 4' -H 'x-plan: plus' -H 'content-type: application/json' \
     -d '{"text":"How do I post a GRN?"}'

   OutputCopy

   event: start
   data: {}
   
   event: delta
   data: {"text":"You "}
   
   event: delta
   data: {"text":"asked: "}
   ...
   event: done
   data: {"inputTokens":42,"outputTokens":15}

   With four users sending at once and 2 slots, two start at once and two receive a place in line first. A free user's sixth and seventh quick messages are refused before any streaming begins:

   OutputCopy

   user 6:  event: start ... event: done
   user 10: event: start ... event: done
   user 8:  event: queued data: {"position":1}  event: start ... event: done
   user 4:  event: queued data: {"position":2}  event: start ... event: done
   
   free user, 7 quick messages:  201 201 201 201 201 429 429
   HTTP/1.1 429 Too Many Requests
   {"error":"rate_limited","retryAfterS":19}

   Finally, close a request early (curl --max-time 0.8). The fake model logs that the client went away and stops, and the partial answer is saved as stopped:

   SQLCopy

   SELECT m.role, m.status, left(m.content, 28) AS content, m.output_tokens AS out
   FROM chat_message m JOIN conversation c ON c.id = m.conversation_id
   WHERE c.user_id IN (2, 4) ORDER BY m.id;

   OutputCopy

      role    |  status  |           content            | out
   -----------+----------+------------------------------+-----
    user      | complete | stock?                       |
    assistant | complete | You asked: stock?. Check the |  10
    user      | complete | long answer please           |
    assistant | stopped  | You asked: long answer pleas |