🔌 API 设计

获取用户列表 → N 次请求获取每个用户的头像和角色。这是 REST 最被诟病的地方：

// 前端：拿到用户列表后，逐个获取关联数据
GET /api/v1/users                    // 1 次
GET /api/v1/users/1/avatar           // N 次
GET /api/v1/users/2/avatar
GET /api/v1/users/3/avatar
...

// 解决方案 1：include 参数
GET /api/v1/users?include=avatar,role

// 解决方案 2：字段展开
GET /api/v1/users?expand=avatar_url,role_name

// 解决方案 3：如果这真的是常态 → 考虑 GraphQL

# 没有 DataLoader：100 个用户 → 100 次 DB 查询获取头像
query {
  users(first: 100) {
    avatar { url }   # 每个 user 触发一次 DB 查询！
  }
}

# 解决：DataLoader 批量加载
const avatarLoader = new DataLoader(async (userIds) => {
  const avatars = await db.query(
    'SELECT * FROM avatars WHERE user_id IN (?)',
    [userIds]
  );
  return userIds.map(id => avatars.find(a => a.user_id === id));
});

# 恶意查询：递归嵌套耗尽服务器资源
query {
  users {          # 100 个
    orders {       # 每个 50 个 = 5000
      items {      # 每个 10 个 = 50000
        product {
          reviews { # 每个 100 个 = 5000000
            user { orders { items { ... } } }
          }
        }
      }
    }
  }
}

# 解决方案：
# 1. 查询深度限制 (depth limiting)
# 2. 复杂度分析 (query complexity analysis)
# 3. 持久化查询 (persisted queries) — 只允许预注册的查询
# 4. 超时 + 结果大小限制

# Apollo Server 示例
import { depthLimit } from 'graphql-depth-limit';
const server = new ApolloServer({
  typeDefs,
  resolvers,
  validationRules: [depthLimit(5)],  # 最大嵌套 5 层
});

# REST 可以直接用 HTTP 缓存：ETag、Cache-Control
GET /api/v1/users/123   →  200 OK, ETag: "abc", Cache-Control: max-age=300

# GraphQL 只有一个 endpoint，无法用 URL 区分缓存
POST /graphql  { query: "{ user(id: 123) { name } }" }

# 解决方案：
# 1. Apollo Client — normalized cache (按 __typename:id 缓存)
# 2. 持久化查询 — GET 请求 + hash → 可用 CDN 缓存
# 3. @cacheControl 指令 — 细粒度缓存控制

type User @cacheControl(maxAge: 300) {
  id: ID!
  name: String
  email: String @cacheControl(maxAge: 0, scope: PRIVATE)
}

3.3 何时选 GraphQL

04 tRPC：全栈 TypeScript 的终极方案

4.1 什么是 tRPC

tRPC 让你在零代码生成、零手动同步的情况下，实现前后端类型安全。后端定义 router，前端直接调用——像调用本地函数一样调用远程 API。

// ===== 后端：定义 router =====
import { initTRPC, TRPCError } from '@trpc/server';
import { z } from 'zod';

const t = initTRPC().create();

// 公开路由
const publicProcedure = t.procedure;
// 需认证路由
const authedProcedure = t.procedure.use(async ({ ctx, next }) => {
  if (!ctx.user) throw new TRPCError({ code: 'UNAUTHORIZED' });
  return next({ ctx: { user: ctx.user } });
});

const appRouter = t.router({
  // 查询用户列表
  listUsers: publicProcedure
    .input(z.object({
      limit: z.number().min(1).max(100).default(20),
      cursor: z.string().optional(),
    }))
    .query(async ({ input, ctx }) => {
      const users = await ctx.db.user.findMany({
        take: input.limit + 1,
        cursor: input.cursor ? { id: input.cursor } : undefined,
        orderBy: { created_at: 'desc' },
      });
      return {
        items: users.slice(0, input.limit),
        next_cursor: users.length > input.limit ? users[users.length - 1].id : undefined,
      };
    }),

  // 创建文章（需认证）
  createArticle: authedProcedure
    .input(z.object({
      title: z.string().min(1).max(200),
      content: z.string().min(1),
      tags: z.array(z.string()).max(5).default([]),
    }))
    .mutation(async ({ input, ctx }) => {
      return ctx.db.article.create({
        data: { ...input, author_id: ctx.user.id },
      });
    }),
});

export type AppRouter = typeof appRouter;

// ===== 前端：直接调用，完整类型推断 =====
import { createTRPCProxyClient, httpBatchLink } from '@trpc/client';
import type { AppRouter } from '../server/router';

const trpc = createTRPCProxyClient<AppRouter>({
  links: [httpBatchLink({ url: '/api/trpc' })],
});

// 调用后端函数——自动补全、类型检查、编译时错误
const result = await trpc.listUsers.query({
  limit: 10,
  cursor: undefined,
});
// result.items: 自动推断为 User[]
// result.next_cursor: 自动推断为 string | undefined

const article = await trpc.createArticle.mutate({
  title: 'tRPC 入门',
  content: '端到端类型安全...',
});
// 如果写错字段名 → 编译时直接报错！

4.2 tRPC vs GraphQL 对比

05 gRPC：微服务内部的高速公路

5.1 什么是 gRPC

gRPC 是 Google 开源的高性能 RPC 框架，使用 Protocol Buffers 序列化，支持双向流、流控、超时、认证。它是微服务间通信的黄金标准。

Proto 定义

// order.proto
syntax = "proto3";
package order;

service OrderService {
  // 普通 RPC
  rpc GetOrder(GetOrderRequest) returns (Order);
  rpc ListOrders(ListOrdersRequest) returns (stream Order);
  
  // 客户端流
  rpc BulkCreateOrders(stream CreateOrderRequest) returns (BulkCreateResponse);
  
  // 双向流
  rpc OrderChat(stream OrderMessage) returns (stream OrderMessage);
}

message GetOrderRequest {
  string order_id = 1;
}

message ListOrdersRequest {
  int32 page_size = 1;
  string page_token = 2;
  string filter = 3;
}

message Order {
  string id = 1;
  string user_id = 2;
  repeated OrderItem items = 3;
  double total = 4;
  string status = 5;
  int64 created_at = 6;  // Unix timestamp
}

message OrderItem {
  string product_id = 1;
  int32 quantity = 2;
  double price = 3;
}

Python 服务端实现

import grpc
from concurrent import futures
import order_pb2
import order_pb2_grpc

class OrderServicer(order_pb2_grpc.OrderServiceServicer):
    def GetOrder(self, request, context):
        order = db.get_order(request.order_id)
        if not order:
            context.abort(grpc.StatusCode.NOT_FOUND, "Order not found")
        return order_pb2.Order(
            id=order.id,
            user_id=order.user_id,
            total=order.total,
            status=order.status,
        )

    def ListOrders(self, request, context):
        # 服务端流：逐个推送订单
        orders = db.list_orders(
            page_size=request.page_size,
            page_token=request.page_token,
        )
        for order in orders:
            yield order_pb2.Order(
                id=order.id,
                user_id=order.user_id,
                total=order.total,
                status=order.status,
            )

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    order_pb2_grpc.add_OrderServiceServicer_to_server(
        OrderServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

5.2 gRPC 性能对比

06 API 版本管理

版本管理是 API 设计中最容易被忽视、但影响最深远的决策。选错策略，未来的每次变更都是噩梦。

6.1 四种版本管理策略

推荐：URI 版本 + 向后兼容变更

# 大版本变更：URI
/api/v1/users   →  /api/v2/users

# v2 内的向后兼容变更：不改版本号
# ✅ 新增可选字段
# ✅ 新增 endpoint
# ✅ 新增查询参数
# ❌ 删除字段 → 需要新大版本
# ❌ 改变字段类型 → 需要新大版本
# ❌ 重命名字段 → 需要新大版本

# v1 → v2 迁移期：同时维护两个版本
# 迁移时间：至少 6 个月
# 通知方式：响应头 Sunset: Sat, 1 Nov 2025 00:00:00 GMT
# 通知方式：响应头 Deprecation: true

6.2 向后兼容变更的实战技巧

// ✅ 好的做法：新增字段，旧客户端忽略
// v1 响应
{ "id": 1, "name": "Alice" }
// v1 + 新字段（向后兼容）
{ "id": 1, "name": "Alice", "avatar_url": "..." }

// ❌ 坏的做法：重命名字段
// 旧
{ "name": "Alice" }
// 新（破坏性变更！）
{ "full_name": "Alice" }

// ✅ 修复：同时返回两个字段，旧字段标记 deprecated
{ "name": "Alice", "full_name": "Alice" }
// 几个月后移除 name 字段，升版本到 v2

07 认证与授权模式

API 安全的核心问题：你是谁（认证）+ 你能做什么（授权）。这部分与 认证授权 模块深度关联，这里聚焦 API 层面的实现。

7.1 API 认证方式对比

JWT 实战：最佳实践

// JWT 结构
// Header: { alg: "RS256", typ: "JWT" }
// Payload: { sub: "usr_abc123", exp: 1700000000, iat: 1699999000, scope: "read write" }
// Signature: RS256(header.payload, private_key)

// Access Token + Refresh Token 双令牌模式
const accessToken = jwtSign(
  { sub: user.id, scope: 'read write' },
  { expiresIn: '15m' }   // 短命：15 分钟
);
const refreshToken = jwtSign(
  { sub: user.id, type: 'refresh' },
  { expiresIn: '7d' }     // 长命：7 天
);

// 中间件：验证 Access Token
function authMiddleware(req, res, next) {
  const token = req.headers.authorization?.replace('Bearer ', '');
  if (!token) return res.status(401).json({ error: 'Missing token' });

  try {
    const decoded = jwtVerify(token, PUBLIC_KEY, { algorithms: ['RS256'] });

    // 检查 Token 是否被吊销（黑名单）
    const isRevoked = await redis.get(`revoked:${decoded.jti}`);
    if (isRevoked) return res.status(401).json({ error: 'Token revoked' });

    req.user = decoded;
    next();
  } catch (err) {
    return res.status(401).json({ error: 'Invalid token' });
  }
}

7.2 API Key 设计

// API Key 格式
// 前缀 + 随机部分，方便识别来源
sk_live_1a2b3c4d5e6f7g8h9i0j   // Stripe 风格
key_xxxx_yyyy_zzzz              // 自定义风格

// 存储：只存 hash，不存明文
const crypto = require('crypto');
const keyHash = crypto.createHash('sha256').update(apiKey).digest('hex');

// 请求时验证
async function validateApiKey(key) {
  const hash = sha256(key);
  const apiKeyRecord = await db.apiKey.findUnique({ where: { hash } });
  if (!apiKeyRecord) return null;
  if (apiKeyRecord.expires_at && apiKeyRecord.expires_at < new Date()) return null;
  return apiKeyRecord;
}

08 错误处理设计

好的错误处理让 API 从"能用"升级到"好用"。差的错误处理让集成者痛苦不堪。

8.1 错误响应标准格式

// 标准错误响应（RFC 7807 inspired）
{
  "error": {
    "type": "https://example.com/errors/validation-error",
    "title": "请求验证失败",
    "status": 400,
    "detail": "邮箱格式不正确",
    "instance": "/api/v1/users",
    "request_id": "req_abc123",
    "timestamp": "2025-01-15T08:30:00Z",
    "details": [
      {
        "field": "email",
        "message": "邮箱格式不正确",
        "value": "not-an-email"
      }
    ]
  }
}

8.2 HTTP 状态码的正确使用

8.3 tRPC 错误处理

import { TRPCError } from '@trpc/server';

// 抛出类型安全错误
throw new TRPCError({
  code: 'NOT_FOUND',           // 自动映射到 HTTP 404
  message: '文章不存在',
  cause: originalError,
});

// 错误码映射
// PARSE_ERROR       → 400
// BAD_REQUEST       → 400
// UNAUTHORIZED      → 401
// FORBIDDEN         → 403
// NOT_FOUND         → 404
// METHOD_NOT_SUPPORTED → 405
// TIMEOUT           → 408
// CONFLICT          → 409
// PRECONDITION_FAILED → 412
// PAYLOAD_TOO_LARGE → 413
// TOO_MANY_REQUESTS → 429
// INTERNAL_SERVER_ERROR → 500

09 限流与配额

每个公开 API 都需要限流。没有限流 = 一个恶意用户就能打垮你的服务。

9.1 限流算法对比

令牌桶实现 (Redis + Lua)

-- rate_limit.lua
local key = KEYS[1]
local capacity = tonumber(ARGV[1])   -- 桶容量
local refill_rate = tonumber(ARGV[2]) -- 每秒补充令牌数
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])  -- 本次请求令牌数

local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1])
local last_refill = tonumber(bucket[2])

if tokens == nil then
  tokens = capacity
  last_refill = now
end

-- 补充令牌
local elapsed = now - last_refill
local new_tokens = math.min(capacity, tokens + elapsed * refill_rate)

-- 消耗令牌
local allowed = false
if new_tokens >= requested then
  new_tokens = new_tokens - requested
  allowed = true
end

redis.call('HMSET', key, 'tokens', new_tokens, 'last_refill', now)
redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 1)

return { allowed and 1 or 0, math.floor(new_tokens) }

9.2 限流响应头

# 响应头告诉客户端限流状态
X-RateLimit-Limit: 100          # 窗口内允许的请求数
X-RateLimit-Remaining: 67       # 剩余请求数
X-RateLimit-Reset: 1700001000   # 窗口重置时间 (Unix)
Retry-After: 30                 # 被限流时，多久后重试 (秒)

10 从 Agent 架构学到的 API 设计

在研究 20 个 AI Agent 框架时，我们发现 API 设计有一些独特的挑战和模式，值得所有 SaaS 开发者学习。

10.1 流式响应是刚需

LLM 生成是流式的，API 必须支持。这不只是 AI 场景——任何长耗时操作都应该考虑流式。

// Server-Sent Events (SSE) — 最简单的流式方案
// 服务端
app.get('/api/v1/chat/stream', async (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  const stream = await llm.chat({
    messages: req.body.messages,
    stream: true,
  });

  for await (const chunk of stream) {
    res.write(`data: ${JSON.stringify(chunk)}\n\n`);
  }

  res.write('data: [DONE]\n\n');
  res.end();
});

// 客户端
const eventSource = new EventSource('/api/v1/chat/stream');
eventSource.onmessage = (event) => {
  if (event.data === '[DONE]') return;
  const chunk = JSON.parse(event.data);
  appendToken(chunk.content);
};

10.2 工具调用 API（Function Calling）

Agent 需要通过 API 调用工具。OpenAI 的 Function Calling 已经成为事实标准：

// 工具定义
{
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "get_weather",
        "description": "获取指定城市的天气信息",
        "parameters": {
          "type": "object",
          "properties": {
            "city": { "type": "string", "description": "城市名称" },
            "unit": { "type": "string", "enum": ["celsius", "fahrenheit"] }
          },
          "required": ["city"]
        }
      }
    }
  ]
}

// LLM 返回工具调用请求
{
  "choices": [{
    "message": {
      "role": "assistant",
      "content": null,
      "tool_calls": [{
        "id": "call_abc123",
        "type": "function",
        "function": {
          "name": "get_weather",
          "arguments": '{"city":"北京","unit":"celsius"}'
        }
      }]
    }
  }]
}

11 API 文档与可观测性

11.1 OpenAPI / Swagger

// openapi.yaml
openapi: 3.1.0
info:
  title: SaaS Platform API
  version: 1.0.0
paths:
  /api/v1/users:
    get:
      summary: 获取用户列表
      parameters:
        - name: limit
          in: query
          schema: { type: integer, default: 20, maximum: 100 }
        - name: cursor
          in: query
          schema: { type: string }
      responses:
        '200':
          description: 成功
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/UserListResponse'

11.2 自动文档工具

11.3 API 可观测性

// 请求日志格式
{
  "request_id": "req_abc123",
  "method": "GET",
  "path": "/api/v1/users",
  "status": 200,
  "duration_ms": 45,
  "user_id": "usr_xyz",
  "ip": "1.2.3.4",
  "user_agent": "Mozilla/5.0...",
  "timestamp": "2025-01-15T08:30:00.123Z"
}

// 响应头注入可追踪信息
X-Request-Id: req_abc123       // 每个请求唯一 ID
X-Response-Time: 45ms          // 响应耗时
X-RateLimit-Limit: 100         // 限流信息
X-RateLimit-Remaining: 67

12 决策框架：选哪种 API 风格？

13 混合架构实战：REST + tRPC + gRPC

真实世界中，很多成功的 SaaS 同时使用多种 API 风格。这是一个典型的混合架构：

// Next.js App Router 混合路由
// app/api/v1/[...path]/route.ts  → REST handler
// app/api/trpc/[trpc]/route.ts    → tRPC handler

// Nginx 路由配置
# /api/v1/*  → Node.js REST server
# /api/trpc/* → tRPC server
# 内部 gRPC 不暴露，只在 K8s 集群内通信

server {
  location /api/v1/ {
    proxy_pass http://rest-server:3001;
  }
  location /api/trpc/ {
    proxy_pass http://trpc-server:3002;
  }
  # gRPC 服务不在 Nginx 暴露
}

14 检查清单与替代方案

14.1 API 设计检查清单

14.2 替代方案速查

15 关键链接