Claude Code数据库全栈：PostgreSQL七步实战

mkdir pg-claude-demo && cd pg-claude-demo
git init
npm init -y
npm install drizzle-orm pg dotenv
npm install -D drizzle-kit tsx typescript @types/pg

# Database Project

## 技术栈
- PostgreSQL 16 + Drizzle ORM
- drizzle-kit 管理migration
- TypeScript strict mode

## 数据库规范
- 表名：snake_case复数（users, order_items）
- 列名：snake_case（created_at, user_id）
- 主键：id bigint GENERATED ALWAYS AS IDENTITY
- 外键：{关联表单数}_id（user_id, order_id）
- 时间列：created_at/updated_at TIMESTAMPTZ DEFAULT now()
- 软删除：deleted_at TIMESTAMPTZ NULL
- 金额：numeric(12,2) 不用float
- 枚举：PostgreSQL ENUM类型，不用varchar

## Migration规范
- 每个migration只做一件事
- 先加列后删列，分两步migration
- 新列必须有DEFAULT或允许NULL
- 禁止在migration中SELECT/INSERT大数据
- 文件名：0001_create_users.sql 顺序编号
- 回滚脚本与正向脚本同级目录

## 查询规范
- 禁止SELECT *，明确列出列
- JOIN超过3张表必须写注释
- 分页必须用OFFSET+LIMIT或cursor
- 大批量操作用COPY代替INSERT

pg-claude-demo/
├── CLAUDE.md          # 项目规范
├── package.json
├── tsconfig.json
├── drizzle.config.ts  # 下一步创建
├── src/
│   ├── db/
│   │   ├── schema.ts  # Drizzle schema定义
│   │   ├── index.ts   # 连接池
│   │   └── migrations/
│   └── app.ts
└── .env               # DATABASE_URL

// drizzle.config.ts
import { defineConfig } from "drizzle-kit";

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

基于CLAUDE.md的规范，为电商系统设计Drizzle ORM schema，包含以下表：
1. users - 用户（email唯一、密码哈希、角色enum）
2. products - 商品（名称、价格numeric(12,2)、库存、状态enum）
3. orders - 订单（用户外键、总额、状态enum、地址JSON）
4. order_items - 订单项（订单外键、商品外键、数量、单价）

要求：
- 使用pgTable，每个表都要created_at/updated_at
- 外键关系用relations()定义
- 用户和商品都支持软删除
- 导出所有类型（inferSelectModel/inferInsertModel）

// src/db/schema.ts
import {
  pgTable, bigint, varchar, text, numeric, integer,
  timestamp, pgEnum, jsonb,
} from "drizzle-orm/pg-core";
import { relations } from "drizzle-orm";

// 枚举定义
export const userRoleEnum = pgEnum("user_role", ["customer", "admin", "vendor"]);
export const productStatusEnum = pgEnum("product_status", ["draft", "active", "archived"]);
export const orderStatusEnum = pgEnum("order_status", ["pending", "paid", "shipped", "delivered", "cancelled"]);

// 用户表
export const users = pgTable("users", {
  id: bigint({ mode: "number" }).primaryKey().generatedAlwaysAsIdentity(),
  email: varchar({ length: 255 }).notNull().unique(),
  passwordHash: text("password_hash").notNull(),
  name: varchar({ length: 100 }),
  role: userRoleEnum().default("customer").notNull(),
  createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
  updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
  deletedAt: timestamp("deleted_at", { withTimezone: true }),
});

// 商品表
export const products = pgTable("products", {
  id: bigint({ mode: "number" }).primaryKey().generatedAlwaysAsIdentity(),
  name: varchar({ length: 255 }).notNull(),
  description: text(),
  price: numeric({ precision: 12, scale: 2 }).notNull(),
  stock: integer().default(0).notNull(),
  status: productStatusEnum().default("draft").notNull(),
  createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
  updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
  deletedAt: timestamp("deleted_at", { withTimezone: true }),
});

// 订单表
export const orders = pgTable("orders", {
  id: bigint({ mode: "number" }).primaryKey().generatedAlwaysAsIdentity(),
  userId: bigint("user_id", { mode: "number" }).notNull().references(() => users.id),
  totalAmount: numeric("total_amount", { precision: 12, scale: 2 }).notNull(),
  status: orderStatusEnum().default("pending").notNull(),
  shippingAddress: jsonb("shipping_address").$type<{
    line1: string; city: string; zip: string; country: string;
  }>(),
  createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
  updatedAt: timestamp("updated_at", { withTimezone: true }).defaultNow().notNull(),
});

// 订单项表
export const orderItems = pgTable("order_items", {
  id: bigint({ mode: "number" }).primaryKey().generatedAlwaysAsIdentity(),
  orderId: bigint("order_id", { mode: "number" }).notNull().references(() => orders.id),
  productId: bigint("product_id", { mode: "number" }).notNull().references(() => products.id),
  quantity: integer().notNull(),
  unitPrice: numeric("unit_price", { precision: 12, scale: 2 }).notNull(),
  createdAt: timestamp("created_at", { withTimezone: true }).defaultNow().notNull(),
});

// 关系定义
export const usersRelations = relations(users, ({ many }) => ({
  orders: many(orders),
}));

export const productsRelations = relations(products, ({ many }) => ({
  orderItems: many(orderItems),
}));

export const ordersRelations = relations(orders, ({ one, many }) => ({
  user: one(users, { fields: [orders.userId], references: [users.id] }),
  items: many(orderItems),
}));

export const orderItemsRelations = relations(orderItems, ({ one }) => ({
  order: one(orders, { fields: [orderItems.orderId], references: [orders.id] }),
  product: one(products, { fields: [orderItems.productId], references: [products.id] }),
}));

// 类型导出
export type User = typeof users.$inferSelectModel;
export type NewUser = typeof users.$inferInsertModel;
export type Product = typeof products.$inferSelectModel;
export type Order = typeof orders.$inferSelectModel;
export type OrderItem = typeof orderItems.$inferSelectModel;

// src/db/index.ts
import { drizzle } from "drizzle-orm/node-postgres";
import pg from "pg";
import * as schema from "./schema";

const pool = new pg.Pool({
  connectionString: process.env.DATABASE_URL,
  max: 10,
  idleTimeoutMillis: 30000,
});

export const db = drizzle(pool, { schema });
export { pool };

# 从schema生成migration SQL
npx drizzle-kit generate

# 输出结构：
# src/db/migrations/
# ├── 0000_marvelous_weapon_omega.sql   # 正向
# └── meta/
#     ├── _journal.json
#     └── 0000_snapshot.json

-- src/db/migrations/0001_add_product_category.sql

-- 幂等：IF NOT EXISTS
DO $$ BEGIN
  IF NOT EXISTS (
    SELECT 1 FROM pg_type WHERE typname = 'product_category'
  ) THEN
    CREATE TYPE product_category AS ENUM (
      'electronics', 'clothing', 'food', 'books', 'other'
    );
  END IF;
END $$;

-- 加列：带DEFAULT，对现有行安全
ALTER TABLE products
  ADD COLUMN IF NOT EXISTS category product_category
  DEFAULT 'other';

-- 加索引：CONCURRENTLY不锁表
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_products_category
  ON products (category);

-- 加注释
COMMENT ON COLUMN products.category IS 'Product classification for filtering';

-- src/db/migrations/0001_add_product_category_rollback.sql

DROP INDEX CONCURRENTLY IF EXISTS idx_products_category;
ALTER TABLE products DROP COLUMN IF EXISTS category;

-- 注意：不删除ENUM类型，因为可能有其他引用
-- 如果确认无引用：
-- DROP TYPE IF EXISTS product_category;

审查 src/db/migrations/0001_add_product_category.sql：

1. 这个migration在12M行的products表上执行是否安全？
2. 是否会造成锁表？预估执行时间？
3. 回滚脚本是否完整？
4. 是否遗漏了索引？
5. 对外键约束有没有影响？

# 开发环境：直接推送schema到数据库
npx drizzle-kit push

# 生产环境：执行已审查的migration文件
npx drizzle-kit migrate

# 验证：检查数据库是否与schema一致
npx drizzle-kit check

// .claude/settings.json 或 ~/.claude/settings.json
{
  "mcpServers": {
    "dbhub": {
      "command": "npx",
      "args": ["-y", "@anthropic/dbhub", "--url", "postgresql://user:pass@localhost:5432/mydb"]
    }
  }
}

{
  "mcpServers": {
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres",
               "postgresql://user:pass@localhost:5432/mydb"]
    }
  }
}

# 查看当前数据库状态
列出所有表和它们的行数

# 检查索引
products表上有哪些索引？有没有未使用的索引？

# 分析查询性能
对以下查询做EXPLAIN ANALYZE分析：
SELECT u.name, COUNT(o.id) as order_count
FROM users u
JOIN orders o ON u.id = o.user_id
WHERE o.created_at > '2025-01-01'
GROUP BY u.id
ORDER BY order_count DESC
LIMIT 20;

# 对比schema差异
对比 src/db/schema.ts 和数据库实际结构，告诉我哪些地方不一致

MCP数据库连接安全铁律：
1. 永远不要在生产数据库上配置读写MCP
2. 开发环境用只读连接（read-only user）
3. .env中的DATABASE_URL不要提交到git
4. MCP的SQL执行需要用户确认（Claude Code默认行为）
5. 定期轮换数据库密码

flowchart TD
    A[发现慢查询] --> B[获取EXPLAIN ANALYZE]
    B --> C{粘贴给Claude Code}
    C --> D[分析扫描类型]
    D --> E{Seq Scan?}
    E -->|是| F[检查缺失索引]
    E -->|否| G{Index Scan但仍慢?}
    G -->|是| H[检查索引选择性]
    G -->|否| I{Nested Loop?}
    I -->|是| J[检查JOIN条件索引]
    I -->|否| K[检查Sort/GroupBy]
    F --> L[Claude生成优化查询+索引]
    H --> L
    J --> L
    K --> L
    L --> M[验证优化效果]
    M --> N{查询时间降低?}
    N -->|是| O[创建migration添加索引]
    N -->|否| P[调整查询逻辑]

// 问题代码：N+1查询
// Claude Code会标记这段代码有问题
async function getUserOrders(userId: number) {
  const orders = await db.select().from(ordersTable)
    .where(eq(ordersTable.userId, userId));

  // 每个order都查一次product —— N+1！
  for (const order of orders) {
    const items = await db.select().from(orderItemsTable)
      .where(eq(orderItemsTable.orderId, order.id));
    order.items = items;
  }
  return orders;
}

// 修复：一次查询获取所有数据
async function getUserOrdersOptimized(userId: number) {
  const result = await db
    .select({
      order: ordersTable,
      item: orderItemsTable,
      product: productsTable,
    })
    .from(ordersTable)
    .innerJoin(orderItemsTable, eq(ordersTable.id, orderItemsTable.orderId))
    .innerJoin(productsTable, eq(orderItemsTable.productId, productsTable.id))
    .where(eq(ordersTable.userId, userId));

  // 在内存中组装
  const orderMap = new Map<number, Order & { items: OrderItem[] }>();
  for (const row of result) {
    if (!orderMap.has(row.order.id)) {
      orderMap.set(row.order.id, { ...row.order, items: [] });
    }
    orderMap.get(row.order.id)!.items.push(row.item);
  }
  return Array.from(orderMap.values());
}

-- 粘贴慢查询的EXPLAIN ANALYZE输出给Claude Code：
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT p.name, sum(oi.quantity * oi.unit_price) as revenue
FROM products p
JOIN order_items oi ON p.id = oi.product_id
JOIN orders o ON oi.order_id = o.id
WHERE o.created_at BETWEEN '2025-01-01' AND '2025-12-31'
  AND o.status IN ('paid', 'delivered')
GROUP BY p.id
ORDER BY revenue DESC
LIMIT 10;

诊断结果：
1. orders表在created_at上做了Seq Scan → 需要复合索引 (status, created_at)
2. order_items的JOIN用Nested Loop → product_id上缺少索引
3. 排序在内存中完成但用了external merge → work_mem可能不够

建议索引：
CREATE INDEX CONCURRENTLY idx_orders_status_created
  ON orders (status, created_at)
  WHERE status IN ('paid', 'delivered');  -- 部分索引，更小更快

CREATE INDEX CONCURRENTLY idx_order_items_product
  ON order_items (product_id)
  INCLUDE (quantity, unit_price);  -- 覆盖索引，避免回表

## 索引设计规则（加入CLAUDE.md）

1. **等值条件列在前，范围条件列在后**
   WHERE status = 'active' AND created_at > '2025-01-01'
   → INDEX (status, created_at) ✓
   → INDEX (created_at, status) ✗

2. **高选择性列在前**
   WHERE country = 'CN' AND status = 'active'
   country选择性低（大量CN），status选择性高
   → INDEX (status, country) ✓

3. **能用部分索引就不建全表索引**
   WHERE status = 'active'
   → WHERE status = 'active' 的部分索引更小

4. **覆盖索引避免回表**
   SELECT id, name FROM users WHERE email = ?
   → INDEX (email) INCLUDE (name)

5. **监控索引使用率**
   每月检查pg_stat_user_indexes的idx_scan
   idx_scan = 0 的索引考虑删除

// src/db/seed.ts
import { db } from "./index";
import { users, products, orders, orderItems } from "./schema";
import { faker } from "@faker-js/faker/locale/zh_CN";

async function seed() {
  console.log("🌱 开始播种...");

  // 1. 用户
  const userList = Array.from({ length: 100 }, () => ({
    email: faker.internet.email(),
    passwordHash: "$2b$10$dummy.hash.for.seed.data",
    name: faker.person.fullName(),
    role: faker.helpers.arrayElement(["customer", "customer", "customer", "vendor"]),
  }));
  const insertedUsers = await db.insert(users).values(userList).returning();

  // 2. 商品
  const productList = Array.from({ length: 500 }, () => ({
    name: faker.commerce.productName(),
    description: faker.commerce.productDescription(),
    price: faker.commerce.price({ min: 1, max: 9999, dec: 2 }),
    stock: faker.number.int({ min: 0, max: 1000 }),
    status: faker.helpers.arrayElement(["draft", "active", "active", "active", "archived"]),
  }));
  const insertedProducts = await db.insert(products).values(productList).returning();

  // 3. 订单 + 订单项（每个用户1-5个订单，每个订单1-5个商品）
  for (const user of insertedUsers) {
    const orderCount = faker.number.int({ min: 1, max: 5 });
    for (let i = 0; i < orderCount; i++) {
      const itemCount = faker.number.int({ min: 1, max: 5 });
      const selectedProducts = faker.helpers.arrayElements(insertedProducts, itemCount);

      const orderResult = await db.insert(orders).values({
        userId: user.id,
        totalAmount: "0", // 先插入，后面更新
        status: faker.helpers.arrayElement(["pending", "paid", "shipped", "delivered"]),
        shippingAddress: {
          line1: faker.location.streetAddress(),
          city: faker.location.city(),
          zip: faker.location.zipCode(),
          country: "CN",
        },
      }).returning();

      let total = 0;
      for (const product of selectedProducts) {
        const qty = faker.number.int({ min: 1, max: 3 });
        await db.insert(orderItems).values({
          orderId: orderResult[0].id,
          productId: product.id,
          quantity: qty,
          unitPrice: product.price,
        });
        total += parseFloat(product.price) * qty;
      }

      // 更新订单总额
      await db.update(orders)
        .set({ totalAmount: total.toFixed(2) })
        .where(eq(orders.id, orderResult[0].id));
    }
  }

  console.log("✅ 播种完成！");
}

seed().catch(console.error);

为电商数据库生成种子数据：

要求：
1. 50个用户，email不重复，密码统一用测试哈希
2. 200个商品，价格在10-5000之间
3. 每个用户2-8个订单，订单状态分布：
   - pending 10%, paid 30%, shipped 25%, delivered 30%, cancelled 5%
4. 订单项的unit_price必须等于下单时商品的价格
5. 订单的total_amount必须等于所有订单项的(quantity * unit_price)之和
6. 所有外键关系必须正确

输出为 drizzle-orm 的 insert 语句

# 运行种子脚本
npx tsx src/db/seed.ts

# 验证数据
npx tsx -e "
const { db } = require('./src/db/index');
const { users, products, orders } = require('./src/db/schema');
import { sql } from 'drizzle-orm';

async function check() {
  const counts = await Promise.all([
    db.select({ count: sql\`count(*)\` }).from(users),
    db.select({ count: sql\`count(*)\` }).from(products),
    db.select({ count: sql\`count(*)\` }).from(orders),
  ]);
  console.log('Users:', counts[0][0].count);
  console.log('Products:', counts[1][0].count);
  console.log('Orders:', counts[2][0].count);
}
check();
"

实现商品搜索接口：
- 支持按名称模糊搜索（LIKE '%keyword%'）
- 支持按分类、价格区间、状态筛选
- 支持按价格/销量/创建时间排序
- 分页：cursor-based，每页20条
- 目标：< 50ms响应时间

使用Drizzle ORM，不要原始SQL。
配合必要的索引和EXPLAIN验证。

// src/repositories/product.search.ts
import { db } from "../db";
import { products, orderItems, sql, and, or, ilike, gte, lte, desc, asc, eq } from "drizzle-orm";
import type { SQL } from "drizzle-orm";

interface ProductSearchParams {
  keyword?: string;
  category?: string;
  minPrice?: string;
  maxPrice?: string;
  status?: string;
  sortBy?: "price" | "sales" | "created_at";
  sortOrder?: "asc" | "desc";
  cursor?: string; // base64 encoded JSON: { field: value }
  limit?: number;
}

interface ProductSearchResult {
  items: Product[];
  nextCursor: string | null;
  total: number;
}

export async function searchProducts(
  params: ProductSearchParams
): Promise<ProductSearchResult> {
  const {
    keyword, category, minPrice, maxPrice, status,
    sortBy = "created_at", sortOrder = "desc",
    cursor, limit = 20,
  } = params;

  // 构建WHERE条件
  const conditions: SQL[] = [];

  // 软删除过滤
  conditions.push(sql`${products.deletedAt} IS NULL`);

  if (keyword) {
    conditions.push(ilike(products.name, `%${keyword}%`));
  }
  if (category) {
    conditions.push(eq(products.category, category as any));
  }
  if (minPrice) {
    conditions.push(gte(products.price, minPrice));
  }
  if (maxPrice) {
    conditions.push(lte(products.price, maxPrice));
  }
  if (status) {
    conditions.push(eq(products.status, status as any));
  }

  // Cursor分页
  if (cursor) {
    const cursorData = JSON.parse(Buffer.from(cursor, "base64").toString());
    if (sortOrder === "desc") {
      conditions.push(sql`${products[sortBy]} < ${cursorData.value}`);
    } else {
      conditions.push(sql`${products[sortBy]} > ${cursorData.value}`);
    }
  }

  const whereClause = and(...conditions);

  // 按销量排序需要子查询
  if (sortBy === "sales") {
    const salesSubquery = db
      .select({
        productId: orderItems.productId,
        totalSold: sql<number>`coalesce(sum(${orderItems.quantity}), 0)`,
      })
      .from(orderItems)
      .groupBy(orderItems.productId)
      .as("sales");

    const result = await db
      .select({
        id: products.id,
        name: products.name,
        price: products.price,
        stock: products.stock,
        status: products.status,
        createdAt: products.createdAt,
        totalSold: salesSubquery.totalSold,
      })
      .from(products)
      .leftJoin(salesSubquery, eq(products.id, salesSubquery.productId))
      .where(whereClause)
      .orderBy(
        sortOrder === "desc"
          ? desc(salesSubquery.totalSold)
          : asc(salesSubquery.totalSold)
      )
      .limit(limit + 1);

    const hasNext = result.length > limit;
    const items = hasNext ? result.slice(0, -1) : result;
    const lastItem = items[items.length - 1];

    return {
      items,
      nextCursor: hasNext
        ? Buffer.from(JSON.stringify({
            field: "sales",
            value: lastItem.totalSold,
          })).toString("base64")
        : null,
      total: -1, // cursor分页不返回总数，单独count
    };
  }

  // 其他排序：直接查products表
  const result = await db
    .select()
    .from(products)
    .where(whereClause)
    .orderBy(
      sortOrder === "desc"
        ? desc(products[sortBy])
        : asc(products[sortBy])
    )
    .limit(limit + 1);

  const hasNext = result.length > limit;
  const items = hasNext ? result.slice(0, -1) : result;
  const lastItem = items[items.length - 1];

  return {
    items,
    nextCursor: hasNext
      ? Buffer.from(JSON.stringify({
          field: sortBy,
          value: lastItem[sortBy],
        })).toString("base64")
      : null,
    total: -1,
  };
}

-- src/db/migrations/0002_product_search_indexes.sql

-- 名称模糊搜索：pg_trgm三元组索引
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_products_name_trgm
  ON products USING gin (name gin_trgm_ops);

-- 分类+状态组合筛选
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_products_category_status
  ON products (category, status)
  WHERE deleted_at IS NULL;

-- 价格范围查询
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_products_price
  ON products (price)
  WHERE deleted_at IS NULL;

-- 创建时间排序（分页cursor）
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_products_created_desc
  ON products (created_at DESC)
  WHERE deleted_at IS NULL;

-- 订单项按商品聚合（销量排序）
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_order_items_product_qty
  ON order_items (product_id)
  INCLUDE (quantity);

# 执行migration
npx drizzle-kit migrate

# 用Claude Code验证查询计划

在已连接的PostgreSQL上执行以下验证：

1. EXPLAIN ANALYZE 搜索 "手机" 按销量排序前20条
2. EXPLAIN ANALYZE 价格100-500区间按创建时间排序
3. 检查所有新建索引是否被使用
4. 如果有Seq Scan，分析原因并给出修复建议

flowchart LR
    A[CLAUDE.md规范] --> B[Schema定义]
    B --> C[Drizzle生成Migration]
    C --> D[MCP直连验证]
    D --> E[种子数据填充]
    E --> F[查询开发]
    F --> G[EXPLAIN优化]
    G --> H[索引Migration]
    H --> D

    style A fill:#e1f5fe
    style D fill:#fff3e0
    style G fill:#e8f5e9

对比 git HEAD~1 和 HEAD 的 schema.ts 变更：
1. 哪些表被修改了？
2. 是否有破坏性变更（删列、改类型）？
3. 需要几个migration？给出SQL。

我用MCP连接了一个已有的PostgreSQL数据库。
请：
1. 列出所有表的结构
2. 生成对应的Drizzle schema.ts
3. 生成relations()关系定义
4. 导出所有TypeScript类型

对当前数据库做全面性能巡检：
1. 列出所有缺失索引的查询模式（从pg_stat_statements）
2. 找出未使用的索引（idx_scan = 0）
3. 检查膨胀率超标的表（dead_tuples ratio > 10%）
4. 建议需要VACUUM的表
5. 检查外键是否都有对应索引

我准备在生产库执行以下migration：

[paste SQL here]

请检查：
1. 是否会锁表？锁类型和预估时间
2. 在1000万行表上的执行时间估算
3. 是否有数据丢失风险
4. 回滚方案是否可行
5. 是否需要SET statement_timeout

编写SQL校验以下数据一致性规则：
1. 订单总额 = SUM(订单项.数量 * 单价)
2. 用户不能有重复email
3. 订单项的商品必须存在且未删除
4. 库存不能为负数
5. 外键无孤立记录

输出为可重复执行的检查脚本，每条规则输出通过/失败+违规数量。