Architecting an AI Chat SaaS with Next.js, FastAPI, Credits and SSE
Walk through a full-stack AI chat architecture with a FastAPI gatekeeper, credit ledger, Redis rate limits and SSE streaming, plus the gaps to close first.
This article is published in English.
Sources: Build a Production-Ready AI SaaS with React 19 + FastAPI: The Complete Architecture (2026)
A chat UI around a model API is easy; a product must also keep keys off the client, meter usage, throttle abuse, stream tokens cleanly, and deploy reproducibly. This walkthrough builds those pieces with Next.js on React 19, FastAPI, PostgreSQL with pgvector, and Redis. Treat the code as a sketch: each section notes what to harden first.
The backend is the gatekeeper
The browser never calls the model provider. Your API authenticates, checks limits and credits, calls the model, and records the cost.
┌─────────────────────────────────────────────────────────────┐
│ React 19 App │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Server │ │ Client │ │ Chat Interface │ │
│ │ Components │ │ Components │ │ (Streaming) │ │
│ │ (Initial │ │ (Interactivity│ │ │ │
│ │ data fetch)│ │ + State) │ │ │ │
│ └──────┬───────┘ └──────┬───────┘ └────────┬─────────┘ │
│ │ │ │ │
│ └─────────────────┴────────────────────┘ │
│ │ │
│ TanStack Query │
│ (Auth + Sync) │
└───────────────────────────┬─────────────────────────────────┘
│ HTTPS / JSON
┌───────────────────────────┴─────────────────────────────────┐
│ FastAPI Backend │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Auth │ │ Rate │ │ OpenAI │ │
│ │ Middleware │ │ Limiter │ │ Service │ │
│ │ (JWT + │ │ (Redis) │ │ (Structured │ │
│ │ Credits) │ │ │ │ Output) │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────┐ │
│ │ Background │ │ PostgreSQL │ │ pgvector │ │
│ │ Worker │ │ (Users + │ │ (Embeddings) │ │
│ │ (Celery) │ │ Conversations│ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────────┘ │
└─────────────────────────────────────────────────────────────┘
Server Components load history without extra JavaScript, async FastAPI handles slow model calls without blocking threads, pgvector leaves room for retrieval, Redis holds rate-limit counters and cached embeddings, and a worker takes PDF parsing and embedding jobs off the request path.
Data model: append to a ledger
The schema defines users, conversations, messages and a credit ledger.
# backend/models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text, Numeric, Index
from sqlalchemy.dialects.postgresql import UUID, JSONB, VECTOR
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
email = Column(String, unique=True, nullable=False)
hashed_password = Column(String, nullable=False)
credits_balance = Column(Numeric(10, 2), default=0.00)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
conversations = relationship("Conversation", back_populates="user")
__table_args__ = (
Index('ix_users_email', 'email'),
)
class Conversation(Base):
__tablename__ = "conversations"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
title = Column(String, nullable=True)
model = Column(String, default="gpt-4o")
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
user = relationship("User", back_populates="conversations")
messages = relationship("Message", back_populates="conversation", order_by="Message.created_at")
class Message(Base):
__tablename__ = "messages"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
conversation_id = Column(UUID(as_uuid=True), ForeignKey("conversations.id"), nullable=False)
role = Column(String, nullable=False) # 'user', 'assistant', 'system'
content = Column(Text, nullable=False)
token_count = Column(Integer, default=0)
metadata = Column(JSONB, default={}) # Store model, latency, finish_reason
created_at = Column(DateTime, default=datetime.utcnow)
conversation = relationship("Conversation", back_populates="messages")
class CreditLedger(Base):
__tablename__ = "credit_ledger"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
amount = Column(Numeric(10, 2), nullable=False) # Negative for usage
description = Column(String, nullable=False)
reference_id = Column(String, nullable=True) # conversation_id or message_id
created_at = Column(DateTime, default=datetime.utcnow)
Instead of mutating a balance, every purchase or usage is appended and the balance is their sum, giving an audit trail and no lost updates. Before using it, fix a few things: Message.metadata clashes with an attribute SQLAlchemy's declarative base reserves, so rename it (for example meta); imports such as Boolean and declarative_base are missing; and pgvector's column type comes from the pgvector package, not SQLAlchemy's PostgreSQL dialect.
FastAPI services and routes
A conventional layout separates auth, AI logic, routers and middleware:
backend/
├── main.py
├── config.py
├── database.py
├── models.py
├── schemas.py
├── auth/
│ ├── utils.py
│ ├── dependencies.py
│ └── router.py
├── ai/
│ ├── service.py
│ └── schemas.py
├── api/
│ ├── conversations.py
│ └── credits.py
├── middleware/
│ └── rate_limit.py
└── worker.py
A streaming service that meters usage
This service allows 20 requests per minute per user via Redis, checks the balance, streams Server-Sent Events, then commits the assistant message and a negative ledger entry together.
# backend/ai/service.py
from openai import AsyncOpenAI
from typing import AsyncGenerator
import json
from sqlalchemy.ext.asyncio import AsyncSession
from models import Message, CreditLedger
from schemas import ChatCompletionRequest, ChatStreamChunk
import redis.asyncio as redis
client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
redis_client = redis.from_url(settings.REDIS_URL)
class AIService:
def __init__(self, db: AsyncSession):
self.db = db
async def stream_chat(
self,
user_id: str,
conversation_id: str,
messages: list[dict],
model: str = "gpt-4o"
) -> AsyncGenerator[str, None]:
"""
Streams chat completion and deducts credits.
Yields SSE-formatted strings.
"""
# 1. Check rate limit
key = f"ratelimit:chat:{user_id}"
current = await redis_client.incr(key)
if current == 1:
await redis_client.expire(key, 60)
if current > 20: # 20 requests per minute
yield f"data: {json.dumps({'error': 'Rate limit exceeded'})}\n\n"
return
# 2. Check credits (simplified; use ledger sum in production)
balance = await self.get_credit_balance(user_id)
if balance < 0.01:
yield f"data: {json.dumps({'error': 'Insufficient credits'})}\n\n"
return
# 3. Stream from OpenAI
try:
stream = await client.chat.completions.create(
model=model,
messages=messages,
stream=True,
max_tokens=4096,
)
full_content = ""
total_tokens = 0
async for chunk in stream:
delta = chunk.choices[0].delta.content or ""
full_content += delta
total_tokens += 1 # Approximate; use tiktoken for accuracy
yield f"data: {json.dumps({'chunk': delta, 'done': False})}\n\n"
# 4. Persist assistant message
assistant_msg = Message(
conversation_id=conversation_id,
role="assistant",
content=full_content,
token_count=total_tokens,
metadata={"model": model, "finish_reason": "stop"}
)
self.db.add(assistant_msg)
# 5. Deduct credits (approximate $0.002 per 1K tokens for GPT-4o)
cost = (total_tokens / 1000) * 0.002
ledger_entry = CreditLedger(
user_id=user_id,
amount=-cost,
description=f"Chat completion: {conversation_id}",
reference_id=str(conversation_id)
)
self.db.add(ledger_entry)
await self.db.commit()
yield f"data: {json.dumps({'done': True, 'tokens': total_tokens})}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"
async def get_credit_balance(self, user_id: str) -> float:
# In production, use a SQL SUM query
result = await self.db.execute(
select(func.sum(CreditLedger.amount)).where(CreditLedger.user_id == user_id)
)
return result.scalar() or 0.00
Caveats: one token per chunk is an approximation, so use provider-reported usage or a tokenizer. The hardcoded price is illustrative, not current; load pricing from configuration. And checking the balance before streaming and deducting afterward is not atomic; reserve credits up front or lock the user row.
The message endpoint
The router verifies ownership (404, not 403, to avoid revealing IDs), saves the user message, and returns a StreamingResponse:
# backend/api/conversations.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
from auth.dependencies import get_current_user
from database import get_db
from ai.service import AIService
from models import Conversation, Message
from schemas import CreateMessageRequest
router = APIRouter(prefix="/conversations", tags=["conversations"])
@router.post("/{conversation_id}/messages")
async def create_message(
conversation_id: str,
request: CreateMessageRequest,
current_user = Depends(get_current_user),
db: AsyncSession = Depends(get_db)
):
# Verify ownership
conversation = await db.get(Conversation, conversation_id)
if not conversation or conversation.user_id != current_user.id:
raise HTTPException(status_code=404, detail="Conversation not found")
# Persist user message
user_msg = Message(
conversation_id=conversation_id,
role="user",
content=request.content
)
db.add(user_msg)
await db.commit()
# Build message history
messages = [
{"role": m.role, "content": m.content}
for m in conversation.messages
]
messages.append({"role": "user", "content": request.content})
# Stream response
ai_service = AIService(db)
return StreamingResponse(
ai_service.stream_chat(
user_id=str(current_user.id),
conversation_id=conversation_id,
messages=messages,
model=conversation.model
),
media_type="text/event-stream"
)
With async SQLAlchemy, touching conversation.messages lazily raises an error, so load the relationship eagerly.
A global rate limit
A second, coarser limit of 100 requests per minute runs as middleware:
# backend/middleware/rate_limit.py
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
import redis.asyncio as redis
redis_client = redis.from_url("redis://localhost:6379")
class RateLimitMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# Skip rate limiting for auth routes
if request.url.path.startswith("/auth"):
return await call_next(request)
# Identify user (from JWT cookie or header)
user_id = request.state.user_id if hasattr(request.state, "user_id") else "anonymous"
key = f"ratelimit:global:{user_id}"
current = await redis_client.incr(key)
if current == 1:
await redis_client.expire(key, 60)
if current > 100: # 100 requests per minute globally
raise HTTPException(status_code=429, detail="Too many requests")
response = await call_next(request)
response.headers["X-RateLimit-Remaining"] = str(100 - current)
return response
An HTTPException raised inside BaseHTTPMiddleware is not converted into a clean 429 by FastAPI's handlers, so return a JSONResponse with status 429 instead. Fixed windows also allow edge bursts.
The frontend
The App Router tree pairs Server Component pages with a client component for the live chat:
src/
├── app/
│ ├── layout.tsx
│ ├── page.tsx
│ ├── (dashboard)/
│ │ ├── layout.tsx
│ │ ├── conversations/
│ │ │ ├── page.tsx # Server Component: list conversations
│ │ │ └── [id]/
│ │ │ ├── page.tsx # Server Component: load initial messages
│ │ │ └── ChatClient.tsx # Client Component: streaming + input
│ │ └── settings/
│ │ └── page.tsx
│ └── api/ # Route handlers if needed
├── components/
│ ├── ui/ # shadcn/ui components
│ └── chat/
│ ├── MessageList.tsx
│ ├── MessageItem.tsx
│ └── StreamingText.tsx
├── lib/
│ ├── api.ts # Axios instance with interceptors
│ └── auth.ts
├── hooks/
│ ├── useChatStream.ts # Custom hook for SSE
│ └── useConversations.ts
└── providers.tsx
Loading history on the server
The page reads the cookie and fetches conversation and messages in parallel, so the first paint contains the chat:
// app/(dashboard)/conversations/[id]/page.tsx
import { notFound } from "next/navigation";
import { cookies } from "next/headers";
import { ChatClient } from "./ChatClient";
async function getConversation(id: string, token: string) {
const res = await fetch(`${process.env.API_URL}/conversations/${id}`, {
headers: { Authorization: `Bearer ${token}` },
next: { revalidate: 0 } // Dynamic data
});
if (!res.ok) return null;
return res.json();
}async function getMessages(id: string, token: string) {
const res = await fetch(`${process.env.API_URL}/conversations/${id}/messages`, {
headers: { Authorization: `Bearer ${token}` },
});
if (!res.ok) return [];
return res.json();
}export default async function ConversationPage({
params
}: {
params: { id: string }
}) {
const cookieStore = cookies();
const token = cookieStore.get("access_token")?.value;
if (!token) return notFound();
// Parallel data fetching on the server
const [conversation, messages] = await Promise.all([
getConversation(params.id, token),
getMessages(params.id, token)
]);
if (!conversation) return notFound();
return (
<div className="flex h-[calc(100vh-4rem)] flex-col">
<header className="border-b px-6 py-4">
<h1 className="text-lg font-semibold">{conversation.title || "New Chat"}</h1>
</header>
{/* Client component handles interactivity and streaming */}
<ChatClient
conversationId={params.id}
initialMessages={messages}
/>
</div>
);
}
In Next.js 15 and later, cookies() and params are asynchronous; await them.
Streaming into the UI
The client adds the user message and an empty assistant placeholder optimistically, POSTs, and fills the placeholder from the response stream:
// app/(dashboard)/conversations/[id]/ChatClient.tsx
"use client";import { useState, useRef, useCallback } from "react";
import { useChatStream } from "@/hooks/useChatStream";
import { MessageList } from "@/components/chat/MessageList";
import { ChatInput } from "@/components/chat/ChatInput";
interface Message {
id: string;
role: "user" | "assistant";
content: string;
isStreaming?: boolean;
}
export function ChatClient({
conversationId,
initialMessages
}: {
conversationId: string;
initialMessages: Message[];
}) {
const [messages, setMessages] = useState<Message[]>(initialMessages);
const [input, setInput] = useState("");
const [isLoading, setIsLoading] = useState(false);
const handleSubmit = useCallback(async (content: string) => {
if (!content.trim() || isLoading) return;
// 1. Optimistic UI: Add user message immediately
const userMessage: Message = {
id: `temp-${Date.now()}`,
role: "user",
content
};
const assistantMessage: Message = {
id: `temp-assistant-${Date.now()}`,
role: "assistant",
content: "",
isStreaming: true
};
setMessages(prev => [...prev, userMessage, assistantMessage]);
setInput("");
setIsLoading(true);
// 2. Open SSE connection
const response = await fetch(`/api/conversations/${conversationId}/messages`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ content })
});
if (!response.body) {
setIsLoading(false);
return;
}
const reader = response.body.getReader();
const decoder = new TextDecoder();
let assistantContent = "";
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
const lines = chunk.split("\n");
for (const line of lines) {
if (line.startsWith("data: ")) {
const data = JSON.parse(line.slice(6));
if (data.error) {
throw new Error(data.error);
}
if (data.chunk) {
assistantContent += data.chunk;
setMessages(prev =>
prev.map(msg =>
msg.id === assistantMessage.id
? { ...msg, content: assistantContent }
: msg
)
);
}
if (data.done) {
setMessages(prev =>
prev.map(msg =>
msg.id === assistantMessage.id
? { ...msg, isStreaming: false, tokenCount: data.tokens }
: msg
)
);
}
}
}
}
} catch (error) {
setMessages(prev =>
prev.map(msg =>
msg.id === assistantMessage.id
? { ...msg, content: "Error: Failed to generate response", isStreaming: false }
: msg
)
);
} finally {
setIsLoading(false);
reader.releaseLock();
}
}, [conversationId, isLoading]);
return (
<div className="flex flex-1 flex-col">
<MessageList messages={messages} />
<ChatInput
value={input}
onChange={setInput}
onSubmit={handleSubmit}
disabled={isLoading}
/>
</div>
);
}
Despite the comment, this uses fetch with a stream reader, not EventSource, which cannot send POST bodies. Buffer partial lines, since chunks can split them, and add an AbortController to cancel on unmount. The relative /api/... URL also assumes a route handler or rewrite that proxies to FastAPI.
An Axios client for everything else
With httpOnly cookies, the client sends credentials, adds a tracing ID, and retries once after refreshing on 401:
// lib/api.ts
import axios from "axios";
export const api = axios.create({
baseURL: process.env.NEXT_PUBLIC_API_URL,
withCredentials: true, // Send cookies
});
// Request interceptor: add auth token
api.interceptors.request.use((config) => {
// Token is in httpOnly cookie, sent automatically
// Add request ID for tracing
config.headers["X-Request-ID"] = crypto.randomUUID();
return config;
});
// Response interceptor: handle 401 and token refresh
api.interceptors.response.use(
(response) => response,
async (error) => {
const originalRequest = error.config;
if (error.response?.status === 401 && !originalRequest._retry) {
originalRequest._retry = true;
try {
await axios.post(
`${process.env.NEXT_PUBLIC_API_URL}/auth/refresh`,
{},
{ withCredentials: true }
);
return api(originalRequest);
} catch (refreshError) {
window.location.href = "/login";
return Promise.reject(refreshError);
}
}
return Promise.reject(error);
}
);
Parallel 401s each trigger a refresh; see the single-flight refresh fix for a race-free version.
Enforcing credits as a dependency
A FastAPI dependency sums the ledger and rejects the request with 402 Payment Required when the balance is too low, before any model call:
# backend/api/deps.py
from fastapi import Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from auth.dependencies import get_current_user
from database import get_db
from models import CreditLedger
from sqlalchemy import func, select
async def require_credits(
min_credits: float = 0.01,
current_user = Depends(get_current_user),
db: AsyncSession = Depends(get_db)
):
result = await db.execute(
select(func.sum(CreditLedger.amount)).where(
CreditLedger.user_id == current_user.id
)
)
balance = result.scalar() or 0.00
if balance < min_credits:
raise HTTPException(
status_code=402, # Payment Required
detail=f"Insufficient credits. Balance: ${balance:.4f}"
)
return current_user
# Usage
@router.post("/conversations/{id}/messages")
async def create_message(
...,
user = Depends(require_credits) # Enforces payment before AI call
):
...
It shares the non-atomic caveat above.
Testing without calling the model
MSW 2.x returns a fake event stream, and a Testing Library test checks the optimistic and streamed text:
// tests/mocks/openai.ts
import { http, HttpResponse } from "msw";
export const openaiHandlers = [
http.post("https://api.openai.com/v1/chat/completions", async () => {
const encoder = new TextEncoder();
const stream = new ReadableStream({
start(controller) {
const chunks = [
'data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n',
'data: {"choices":[{"delta":{"content":" world"}}]}\n\n',
'data: {"choices":[{"delta":{}}],"done":true}\n\n',
];
chunks.forEach(chunk => {
controller.enqueue(encoder.encode(chunk));
});
controller.close();
}
});
return new HttpResponse(stream, {
headers: {
"Content-Type": "text/event-stream",
},
});
}),
];
// tests/ChatClient.test.tsx
import { render, screen, waitFor } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import { ChatClient } from "@/app/(dashboard)/conversations/[id]/ChatClient";
it("streams assistant response", async () => {
const user = userEvent.setup();
render(
<ChatClient
conversationId="test-id"
initialMessages={[]}
/>
);
await user.type(screen.getByPlaceholderText("Ask anything..."), "Hello");
await user.click(screen.getByRole("button", { name: /send/i }));
// Optimistic user message appears immediately
expect(screen.getByText("Hello")).toBeInTheDocument();
// Streaming response appears
await waitFor(() => {
expect(screen.getByText("Hello world")).toBeInTheDocument();
});
});
The handler mocks the provider URL, but the component calls your API, so mock your endpoint instead. The backend test patches the client and asserts one ledger row:
# tests/test_chat.py
import pytest
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
async def test_chat_deducts_credits(client, auth_headers, db_session):
with patch("ai.service.client.chat.completions.create") as mock_create:
mock_create.return_value = AsyncMock()
mock_create.return_value.__aiter__.return_value = [
MockChunk(delta=MockContent("Hi"))
]
response = await client.post(
"/conversations/test-id/messages",
json={"content": "Hello"},
headers=auth_headers
)
assert response.status_code == 200
# Verify credit deduction
ledger = await db_session.execute(
select(CreditLedger).where(CreditLedger.user_id == test_user_id)
)
assert len(ledger.scalars().all()) == 1
Deploying with containers
A multi-stage Dockerfile builds Next.js in standalone mode and runs it as a non-root user:
# frontend/Dockerfile
FROM node:20-alpine AS builder
WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build
FROM node:20-alpine AS runner
WORKDIR /app
ENV NODE_ENV=production
RUN addgroup --system --gid 1001 nodejs
RUN adduser --system --uid 1001 nextjs
COPY --from=builder /app/public ./public
COPY --from=builder --chown=nextjs:nodejs /app/.next/standalone ./
COPY --from=builder --chown=nextjs:nodejs /app/.next/static ./.next/static
USER nextjs
EXPOSE 3000
ENV PORT=3000
CMD ["node", "server.js"]
Compose runs pgvector Postgres with a health check, Redis, the API, the frontend and nginx:
version: "3.8"
services:
postgres:
image: pgvector/pgvector:pg16
environment:
POSTGRES_USER: app
POSTGRES_PASSWORD: ${DB_PASSWORD}
POSTGRES_DB: saas_db
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U app"]
interval: 5s
timeout: 5s
retries: 5
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
backend:
build: ./backend
environment:
DATABASE_URL: postgresql+asyncpg://app:${DB_PASSWORD}@postgres/saas_db
REDIS_URL: redis://redis:6379/0
OPENAI_API_KEY: ${OPENAI_API_KEY}
SECRET_KEY: ${SECRET_KEY}
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_started
command: uvicorn main:app --host 0.0.0.0 --port 8000
frontend:
build: ./frontend
environment:
NEXT_PUBLIC_API_URL: http://backend:8000
depends_on:
- backend
nginx:
image: nginx:alpine
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
depends_on:
- frontend
- backend
volumes:
postgres_data:
redis_data:
NEXT_PUBLIC_ values are baked in at build time for the browser, which cannot resolve backend:8000; use the public nginx route.
Key takeaways
- Keep keys, pricing and limits server-side; the model is a dependency behind your API.
- Record usage in an append-only ledger and make the check-and-deduct step atomic.
- Stream with explicit cancellation and line buffering.
- Mock your own API contract in tests, not the vendor's.