Databases for AI Apps: PostgreSQL, Drizzle, Redis, and pgvector

Design an AI app's data layer with PostgreSQL, Drizzle, pgvector for embeddings, and Redis for caching and rate limits, with a full schema and runnable queries.

Intermediate18 min readUpdated October 7, 2026

On this page· Where each kind of data lives

Why this matters

Models are stateless, so everything your AI feature "remembers" lives in your database: conversations, the exact messages sent to the model, token usage for billing, and the documents it can retrieve. Since pgvector, PostgreSQL also stores and searches embeddings, which means one database can hold your users, their chats, and their searchable knowledge, and enforce access rules across all three in a single query.

Where each kind of data lives#

Most AI apps need exactly two data stores. PostgreSQL is the system of record: anything that would be an incident to lose. Redis holds fast, disposable state: caches, rate-limit counters, sessions, and job queues.

Hover or tap a component to see what it does.

Data stores in an AI app: Postgres for durable records and pgvector embeddings, Redis for caches, rate limits, and queues.

Show text description

A rule of thumb: if losing the data would be an incident, it belongs in Postgres; if it can be rebuilt, Redis is fine.

API layer: Decides which store each piece of data belongs in.
Postgres: Durable, transactional data: users, conversations, messages, documents, billing.
pgvector: Embeddings stored next to the rows they describe, so one query can filter by user and rank by similarity.
Redis: Caches, rate-limit counters, sessions, and queues. Data you could rebuild if it disappeared.

A dedicated vector database is worth it at very large scale or for specialized features. Until then, keeping vectors in Postgres gives you transactions, joins, backups, and row-level filtering for free.

Run Postgres with pgvector and Redis locally#

Docker Compose gives every developer the same versions. The init script enables the vector extension once, when the database is created.

compose.yaml

services:
  postgres:
    image: pgvector/pgvector:pg17
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: app
      POSTGRES_DB: app
    ports: ["5432:5432"]
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./db/init.sql:/docker-entrypoint-initdb.d/init.sql:ro
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
volumes:
  pgdata:

db/init.sql

CREATE EXTENSION IF NOT EXISTS vector;

.env.local

DATABASE_URL=postgres://app:app@localhost:5432/app
REDIS_URL=redis://localhost:6379

docker compose up -d
npm install drizzle-orm pg ioredis @anthropic-ai/sdk
npm install -D drizzle-kit @types/pg

The schema#

Five tables cover a chat app with retrieval. Every table that holds user data carries a user_id, so every query can filter by owner.

Hover or tap a component to see what it does.

Database schema for an AI chat app with retrieval: users, conversations, messages, documents, and chunks with pgvector embeddings, all in one Postgres database.

Show text description

Each arrow is a one-to-many relationship enforced with a foreign key.

users: One row per account. Everything else references it, which makes per-user access checks simple.
conversations: A chat thread. Index user_id to list a user's conversations quickly.
messages: Each turn, stored as the exact content blocks sent to and received from the model, plus token usage.
documents: Source documents for retrieval, with the metadata you need for citations.
chunks: Passages of each document with a pgvector embedding column and an HNSW index for similarity search.

Drizzle defines the schema in TypeScript, generates SQL migrations from it, and infers row types for every query.

db/schema.ts

import type Anthropic from "@anthropic-ai/sdk";
import { index, integer, jsonb, pgEnum, pgTable, text, timestamp, uuid, vector } from "drizzle-orm/pg-core";
 
const createdAt = () => timestamp("created_at", { withTimezone: true }).notNull().defaultNow();
 
export const users = pgTable("users", {
  id: uuid("id").primaryKey().defaultRandom(),
  email: text("email").notNull().unique(),
  createdAt: createdAt(),
});
 
export const conversations = pgTable(
  "conversations",
  {
    id: uuid("id").primaryKey().defaultRandom(),
    userId: uuid("user_id")
      .notNull()
      .references(() => users.id, { onDelete: "cascade" }),
    title: text("title").notNull().default("New chat"),
    createdAt: createdAt(),
  },
  (t) => [index("conversations_user_idx").on(t.userId, t.createdAt)],
);
 
export const messageRole = pgEnum("message_role", ["user", "assistant"]);
 
export const messages = pgTable(
  "messages",
  {
    id: uuid("id").primaryKey().defaultRandom(),
    conversationId: uuid("conversation_id")
      .notNull()
      .references(() => conversations.id, { onDelete: "cascade" }),
    role: messageRole("role").notNull(),
    // Store the exact content sent to or received from the model, so history replays byte-for-byte.
    content: jsonb("content").$type<Anthropic.MessageParam["content"]>().notNull(),
    inputTokens: integer("input_tokens"),
    outputTokens: integer("output_tokens"),
    createdAt: createdAt(),
  },
  (t) => [index("messages_conversation_idx").on(t.conversationId, t.createdAt)],
);
 
export const documents = pgTable("documents", {
  id: uuid("id").primaryKey().defaultRandom(),
  userId: uuid("user_id")
    .notNull()
    .references(() => users.id, { onDelete: "cascade" }),
  url: text("url").notNull(),
  title: text("title").notNull(),
  createdAt: createdAt(),
});
 
export const chunks = pgTable(
  "chunks",
  {
    id: uuid("id").primaryKey().defaultRandom(),
    documentId: uuid("document_id")
      .notNull()
      .references(() => documents.id, { onDelete: "cascade" }),
    userId: uuid("user_id").notNull(), // copied from the document so searches can filter without a join
    content: text("content").notNull(),
    embedding: vector("embedding", { dimensions: 1024 }).notNull(),
  },
  (t) => [
    index("chunks_embedding_idx").using("hnsw", t.embedding.op("vector_cosine_ops")),
    index("chunks_user_idx").on(t.userId),
  ],
);

Three design choices are worth copying:

Messages store content blocks as jsonb, not just text. Tool calls, tool results, and reasoning blocks must be replayed exactly on the next request, so you need the full structure.
Token counts live on each message. Billing, quotas, and cost dashboards become simple SUM queries.
The embedding dimension is fixed (1024 here) and must match your embedding model's output. An HNSW index with cosine distance makes similarity search fast.

drizzle.config.ts

import { defineConfig } from "drizzle-kit";
 
export default defineConfig({
  schema: "./db/schema.ts",
  out: "./db/migrations",
  dialect: "postgresql",
  dbCredentials: { url: process.env.DATABASE_URL! },
});

npx drizzle-kit generate   # writes SQL migrations from schema.ts
npx drizzle-kit migrate    # applies them to the database

Commit the generated migrations. Reviewing the SQL in a pull request is how you catch an accidental DROP COLUMN before production does.

Queries#

db/client.ts

import { drizzle } from "drizzle-orm/node-postgres";
import { Pool } from "pg";
import * as schema from "./schema";
 
// One pool per process. Each connection costs memory on the database server.
export const pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 10 });
export const db = drizzle({ client: pool, schema });

db/queries.ts

import { and, asc, cosineDistance, eq, sql } from "drizzle-orm";
import { db } from "./client";
import { chunks, conversations, messages } from "./schema";
 
export async function createConversation(userId: string, title: string) {
  const [conversation] = await db.insert(conversations).values({ userId, title }).returning();
  return conversation;
}
 
/** Save the user's message and the model's reply atomically: either both are stored or neither is. */
export async function saveTurn(
  conversationId: string,
  user: typeof messages.$inferInsert.content,
  assistant: typeof messages.$inferInsert.content,
  usage: { input_tokens: number; output_tokens: number },
) {
  await db.transaction(async (tx) => {
    await tx.insert(messages).values({ conversationId, role: "user", content: user });
    await tx.insert(messages).values({
      conversationId,
      role: "assistant",
      content: assistant,
      inputTokens: usage.input_tokens,
      outputTokens: usage.output_tokens,
    });
  });
}
 
/** History for one conversation, only if it belongs to this user. */
export async function getHistory(conversationId: string, userId: string) {
  return db
    .select({ role: messages.role, content: messages.content })
    .from(messages)
    .innerJoin(conversations, eq(messages.conversationId, conversations.id))
    .where(and(eq(messages.conversationId, conversationId), eq(conversations.userId, userId)))
    .orderBy(asc(messages.createdAt));
}
 
/** Nearest chunks to a query embedding, restricted to the user's own documents. */
export async function searchChunks(userId: string, queryEmbedding: number[], limit = 5) {
  const distance = cosineDistance(chunks.embedding, queryEmbedding);
  return db
    .select({ id: chunks.id, content: chunks.content, similarity: sql<number>`1 - (${distance})` })
    .from(chunks)
    .where(eq(chunks.userId, userId))
    .orderBy(distance)
    .limit(limit);
}

The userId filter inside searchChunks is not optional decoration. Without it, one user's question can retrieve another user's documents, and the model will happily quote them.

Redis: rate limits and caching#

Two Redis patterns appear in almost every AI app. A rate limiter protects your budget from a runaway client. A cache avoids paying twice for identical work, such as re-embedding the same text.

lib/redis.ts

import { Redis } from "ioredis";
 
export const redis = new Redis(process.env.REDIS_URL ?? "redis://localhost:6379");
 
/** Fixed-window limiter: at most `limit` calls per `windowSeconds` for this key. */
export async function rateLimit(key: string, limit: number, windowSeconds: number) {
  const window = Math.floor(Date.now() / 1000 / windowSeconds);
  const bucket = `ratelimit:${key}:${window}`;
  // MULTI runs both commands atomically, so the counter always gets an expiry.
  const results = await redis.multi().incr(bucket).expire(bucket, windowSeconds).exec();
  const count = Number(results?.[0]?.[1] ?? 0);
  return { allowed: count <= limit, remaining: Math.max(0, limit - count) };
}
 
/** Return the cached value for `key`, or compute it, store it with a TTL, and return it. */
export async function cached<T>(key: string, ttlSeconds: number, compute: () => Promise<T>): Promise<T> {
  const hit = await redis.get(key);
  if (hit !== null) return JSON.parse(hit) as T;
  const value = await compute();
  await redis.set(key, JSON.stringify(value), "EX", ttlSeconds);
  return value;
}

In a route handler, call rateLimit(`chat:${userId}`, 20, 60) before calling the model and return HTTP 429 when allowed is false. A fixed window allows short bursts at window edges; that is usually fine for protecting a budget, and much simpler than a sliding window.

Common mistakes#

Storing only the text of model responses. Tool calls and reasoning blocks are lost, and the next request fails or behaves differently. Store the content blocks.
Vector search without an owner filter. Retrieval is a data access path; apply the same authorization as every other query.
Mismatched embedding dimensions. Changing embedding models changes the vector size; the column, the index, and every stored vector must change together. Plan re-indexing as a migration.
Creating a database client per request. Each new Pool() opens fresh connections; under load you exhaust the database's connection limit. Create one per process.
Using Redis as the only copy of important data. Redis is configured for speed; treat what it holds as rebuildable.

Try it

Add per-user usage totals

Write a query that returns the total input and output tokens per conversation for one user, newest conversations first.
Add a monthlyUsage(userId) query that sums tokens since the first day of the current month.
Cache monthlyUsage in Redis for 60 seconds with the cached helper, and explain why a short TTL is acceptable for a usage meter but not for a billing invoice.

Check your understanding

Key takeaways

Postgres is the system of record, including embeddings via pgvector; Redis holds fast, disposable state.
Store messages as the exact content blocks you exchange with the model, plus token usage.
Every vector search filters by owner, exactly like every other query.
Define the schema in Drizzle, review generated migrations, use one connection pool per process, and wrap related writes in a transaction.

Next steps#

APIs and authentication

Expose this data safely over REST and Server-Sent Events.

RAG

Fill the chunks table with real embeddings and retrieve from it.

Finished this article? Track it to see your progress on the roadmap.