tRPC 高级特性 #

Links 机制 #

Links 是 tRPC 客户端的核心机制，用于控制请求的生命周期。类似于 GraphQL 的链接概念，tRPC 的链接允许你在请求发送前后插入自定义逻辑。

text

┌─────────────────────────────────────────────────────────────┐
│                    Links 执行流程                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   请求                                                      │
│     │                                                       │
│     ▼                                                       │
│   ┌─────────────┐                                          │
│   │ loggerLink  │  日志记录                                │
│   └──────┬──────┘                                          │
│          │                                                  │
│          ▼                                                  │
│   ┌─────────────┐                                          │
│   │ retryLink   │  重试逻辑                                │
│   └──────┬──────┘                                          │
│          │                                                  │
│          ▼                                                  │
│   ┌─────────────┐                                          │
│   │ httpBatchLink│  HTTP 批量请求                          │
│   └──────┬──────┘                                          │
│          │                                                  │
│          ▼                                                  │
│   服务端                                                    │
│                                                             │
└─────────────────────────────────────────────────────────────┘

内置 Links #

httpBatchLink #

将多个请求合并为一个 HTTP 请求：

typescript

import { createTRPCProxyClient, httpBatchLink } from '@trpc/client';

const client = createTRPCProxyClient<AppRouter>({
  links: [
    httpBatchLink({
      url: 'http://localhost:4000/trpc',
      maxURLLength: 2083,
      headers() {
        return {
          Authorization: `Bearer ${getToken()}`,
        };
      },
    }),
  ],
});

httpLink #

单个请求，不合并：

typescript

import { httpLink } from '@trpc/client';

const client = createTRPCProxyClient<AppRouter>({
  links: [
    httpLink({
      url: 'http://localhost:4000/trpc',
    }),
  ],
});

wsLink #

WebSocket 连接，用于订阅：

typescript

import { createWSClient, wsLink } from '@trpc/client';

const wsClient = createWSClient({
  url: 'ws://localhost:4000/trpc',
});

const client = createTRPCProxyClient<AppRouter>({
  links: [
    wsLink({
      client: wsClient,
    }),
  ],
});

loggerLink #

请求日志：

typescript

import { loggerLink } from '@trpc/client';

const client = createTRPCProxyClient<AppRouter>({
  links: [
    loggerLink({
      enabled: process.env.NODE_ENV === 'development',
    }),
    httpBatchLink({
      url: 'http://localhost:4000/trpc',
    }),
  ],
});

retryLink #

自动重试：

typescript

import { retryLink } from '@trpc/client';

const client = createTRPCProxyClient<AppRouter>({
  links: [
    retryLink({
      maxAttempts: 3,
      retry: (opts) => {
        if (opts.error.data?.code === 'UNAUTHORIZED') {
          return false;
        }
        return opts.attempts < 3;
      },
    }),
    httpBatchLink({
      url: 'http://localhost:4000/trpc',
    }),
  ],
});

自定义 Link #

typescript

import { TRPCLink } from '@trpc/client';
import { observable } from '@trpc/server/observable';

const customLink: TRPCLink<AppRouter> = () => {
  return ({ next, op }) => {
    return observable((observer) => {
      console.log('Request:', op);

      const unsubscribe = next(op).subscribe({
        next(result) {
          console.log('Response:', result);
          observer.next(result);
        },
        error(err) {
          console.error('Error:', err);
          observer.error(err);
        },
        complete() {
          observer.complete();
        },
      });

      return unsubscribe;
    });
  };
};

组合 Links #

typescript

import {
  createTRPCProxyClient,
  httpBatchLink,
  httpLink,
  splitLink,
  loggerLink,
  retryLink,
  wsLink,
  createWSClient,
} from '@trpc/client';

const wsClient = createWSClient({
  url: 'ws://localhost:4000/trpc',
});

export const client = createTRPCProxyClient<AppRouter>({
  links: [
    loggerLink({
      enabled: process.env.NODE_ENV === 'development',
    }),
    retryLink({
      maxAttempts: 3,
    }),
    splitLink({
      condition: (op) => op.type === 'subscription',
      true: wsLink({
        client: wsClient,
      }),
      false: httpBatchLink({
        url: 'http://localhost:4000/trpc',
      }),
    }),
  ],
});

text

┌─────────────────────────────────────────────────────────────┐
│                    Links 组合策略                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  loggerLink                                                 │
│  └── 记录所有请求和响应                                     │
│                                                             │
│  retryLink                                                  │
│  └── 自动重试失败的请求                                     │
│                                                             │
│  splitLink                                                  │
│  ├── subscription → wsLink                                  │
│  └── query/mutation → httpBatchLink                         │
│                                                             │
│  请求流程：                                                  │
│  logger → retry → split → [ws/http] → server               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

数据转换 #

SuperJSON #

支持更多 JavaScript 类型：

typescript

import { initTRPC } from '@trpc/server';
import superjson from 'superjson';

export const t = initTRPC.create({
  transformer: superjson,
});

typescript

import { createTRPCProxyClient, httpBatchLink } from '@trpc/client';
import superjson from 'superjson';

export const client = createTRPCProxyClient<AppRouter>({
  transformer: superjson,
  links: [
    httpBatchLink({
      url: 'http://localhost:4000/trpc',
    }),
  ],
});

自定义 Transformer #

typescript

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

const transformer = {
  serialize: (data: any) => JSON.stringify(data),
  deserialize: (data: string) => JSON.parse(data),
};

export const t = initTRPC.create({
  transformer,
});

性能优化 #

批量请求 #

typescript

const client = createTRPCProxyClient<AppRouter>({
  links: [
    httpBatchLink({
      url: 'http://localhost:4000/trpc',
      maxURLLength: 2083,
    }),
  ],
});

async function loadDashboard() {
  const [users, posts, stats] = await Promise.all([
    client.user.list.query(),
    client.post.list.query(),
    client.stats.get.query(),
  ]);

  return { users, posts, stats };
}

缓存策略 #

tsx

function UserList() {
  const { data } = trpc.user.list.useQuery(undefined, {
    staleTime: 5 * 60 * 1000,
    cacheTime: 10 * 60 * 1000,
  });

  return <ul>{/* ... */}</ul>;
}

预取数据 #

tsx

function UserList() {
  const utils = trpc.useUtils();

  const handleHover = (id: string) => {
    utils.user.getById.prefetch({ id });
  };

  const { data } = trpc.user.list.useQuery();

  return (
    <ul>
      {data?.map((user) => (
        <li
          key={user.id}
          onMouseEnter={() => handleHover(user.id)}
        >
          {user.name}
        </li>
      ))}
    </ul>
  );
}

无限滚动 #

typescript

const userRouter = t.router({
  infiniteList: publicProcedure
    .input(z.object({
      limit: z.number().min(1).max(100).default(10),
      cursor: z.string().optional(),
    }))
    .query(async ({ input }) => {
      const items = await db.user.findMany({
        take: input.limit + 1,
        cursor: input.cursor ? { id: input.cursor } : undefined,
        orderBy: { createdAt: 'desc' },
      });

      let nextCursor: string | undefined;
      if (items.length > input.limit) {
        items.pop();
        nextCursor = items[items.length - 1]?.id;
      }

      return { items, nextCursor };
    }),
});

tsx

function InfiniteUserList() {
  const {
    data,
    fetchNextPage,
    hasNextPage,
    isFetchingNextPage,
  } = trpc.user.infiniteList.useInfiniteQuery(
    { limit: 10 },
    {
      getNextPageParam: (lastPage) => lastPage.nextCursor,
    }
  );

  return (
    <div>
      {data?.pages.map((page, i) => (
        <Fragment key={i}>
          {page.items.map((user) => (
            <div key={user.id}>{user.name}</div>
          ))}
        </Fragment>
      ))}
      <button
        onClick={() => fetchNextPage()}
        disabled={!hasNextPage || isFetchingNextPage}
      >
        {isFetchingNextPage ? 'Loading...' : 'Load More'}
      </button>
    </div>
  );
}

测试策略 #

单元测试 #

typescript

import { initTRPC } from '@trpc/server';
import { z } from 'zod';

const t = initTRPC.create();

const router = t.router({
  add: t.procedure
    .input(z.object({ a: z.number(), b: z.number() }))
    .query(({ input }) => input.a + input.b),
});

describe('Router', () => {
  it('should add two numbers', async () => {
    const caller = router.createCaller({});
    const result = await caller.add({ a: 1, b: 2 });
    expect(result).toBe(3);
  });
});

集成测试 #

typescript

import { createTRPCProxyClient, httpBatchLink } from '@trpc/client';
import { initTRPC } from '@trpc/server';
import express from 'express';
import * as trpcExpress from '@trpc/server/adapters/express';

const t = initTRPC.create();

const router = t.router({
  hello: t.procedure.query(() => 'Hello World'),
});

const app = express();
app.use('/trpc', trpcExpress.createExpressMiddleware({ router }));

let server: any;

beforeAll((done) => {
  server = app.listen(4000, done);
});

afterAll((done) => {
  server.close(done);
});

describe('API', () => {
  it('should return hello', async () => {
    const client = createTRPCProxyClient<typeof router>({
      links: [
        httpBatchLink({
          url: 'http://localhost:4000/trpc',
        }),
      ],
    });

    const result = await client.hello.query();
    expect(result).toBe('Hello World');
  });
});

Mock 测试 #

typescript

import { mock } from 'jest-mock-extended';

interface Context {
  db: Database;
  user: User | null;
}

const createMockContext = () => ({
  db: mock<Database>(),
  user: null,
});

describe('User Router', () => {
  it('should get user by id', async () => {
    const ctx = createMockContext();
    ctx.db.user.findUnique.mockResolvedValue({
      id: '1',
      name: 'John',
    });

    const caller = router.createCaller(ctx);
    const result = await caller.user.getById({ id: '1' });

    expect(result).toEqual({ id: '1', name: 'John' });
    expect(ctx.db.user.findUnique).toHaveBeenCalledWith({
      where: { id: '1' },
    });
  });
});

生产部署 #

环境变量 #

typescript

const getBaseUrl = () => {
  if (typeof window !== 'undefined') return '';
  if (process.env.VERCEL_URL) return `https://${process.env.VERCEL_URL}`;
  if (process.env.RENDER_INTERNAL_HOSTNAME)
    return `http://${process.env.RENDER_INTERNAL_HOSTNAME}:${process.env.PORT}`;
  return `http://localhost:${process.env.PORT ?? 3000}`;
};

export const trpc = createTRPCNext<AppRouter>({
  config() {
    return {
      links: [
        httpBatchLink({
          url: `${getBaseUrl()}/api/trpc`,
        }),
      ],
    };
  },
});

健康检查 #

typescript

import express from 'express';

const app = express();

app.get('/health', (req, res) => {
  res.json({ status: 'ok', timestamp: new Date().toISOString() });
});

app.use('/trpc', trpcExpress.createExpressMiddleware({ router }));

优雅关闭 #

typescript

const server = app.listen(4000);

process.on('SIGTERM', () => {
  console.log('SIGTERM received, shutting down...');
  server.close(() => {
    console.log('Server closed');
    process.exit(0);
  });
});

Docker 部署 #

dockerfile

FROM node:18-alpine

WORKDIR /app

COPY package*.json ./
RUN npm ci --only=production

COPY . .
RUN npm run build

EXPOSE 4000

CMD ["npm", "start"]

Kubernetes 部署 #

yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: trpc-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: trpc-server
  template:
    metadata:
      labels:
        app: trpc-server
    spec:
      containers:
        - name: trpc-server
          image: trpc-server:latest
          ports:
            - containerPort: 4000
          env:
            - name: NODE_ENV
              value: production
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: trpc-secrets
                  key: database-url

安全最佳实践 #

CORS 配置 #

typescript

import cors from 'cors';

app.use(cors({
  origin: process.env.ALLOWED_ORIGINS?.split(',') || ['http://localhost:3000'],
  credentials: true,
}));

速率限制 #

typescript

import rateLimit from 'express-rate-limit';

const limiter = rateLimit({
  windowMs: 15 * 60 * 1000,
  max: 100,
  message: 'Too many requests, please try again later.',
});

app.use('/trpc', limiter, trpcExpress.createExpressMiddleware({ router }));

输入验证 #

typescript

const router = t.router({
  create: t.procedure
    .input(z.object({
      name: z.string().min(1).max(100),
      email: z.string().email(),
      content: z.string().max(10000),
    }))
    .mutation(async ({ input }) => {
      return db.post.create({ data: input });
    }),
});

敏感数据处理 #

typescript

const router = t.router({
  user: t.router({
    getProfile: protectedProcedure.query(({ ctx }) => {
      const { password, ...safeUser } = ctx.user;
      return safeUser;
    }),
  }),
});

监控与日志 #

结构化日志 #

typescript

import pino from 'pino';

const logger = pino({
  level: process.env.LOG_LEVEL || 'info',
});

const loggingMiddleware = t.middleware(async ({ path, type, next }) => {
  const start = Date.now();
  const result = await next();
  
  logger.info({
    type,
    path,
    duration: Date.now() - start,
    ok: result.ok,
  });
  
  return result;
});

错误追踪 #

typescript

import * as Sentry from '@sentry/node';

Sentry.init({
  dsn: process.env.SENTRY_DSN,
  environment: process.env.NODE_ENV,
});

const sentryMiddleware = t.middleware(async ({ path, type, next }) => {
  try {
    return await next();
  } catch (error) {
    Sentry.captureException(error, {
      tags: { trpc_path: path, trpc_type: type },
    });
    throw error;
  }
});

性能监控 #

typescript

import { performance } from 'perf_hooks';

const performanceMiddleware = t.middleware(async ({ path, next }) => {
  const start = performance.now();
  const result = await next();
  const duration = performance.now() - start;

  if (duration > 1000) {
    console.warn(`Slow query: ${path} took ${duration.toFixed(2)}ms`);
  }

  return result;
});

最佳实践总结 #

1. 项目结构 #

text

src/
├── server/
│   ├── routers/
│   │   ├── index.ts
│   │   ├── user.ts
│   │   └── post.ts
│   ├── context.ts
│   ├── trpc.ts
│   └── index.ts
├── client/
│   ├── trpc.ts
│   └── hooks/
│       ├── useUser.ts
│       └── usePost.ts
└── shared/
    ├── types.ts
    └── schemas.ts

2. 类型安全 #

typescript

import type { inferRouterInputs, inferRouterOutputs } from '@trpc/server';

type Input = inferRouterInputs<AppRouter>;
type Output = inferRouterOutputs<AppRouter>;

3. 错误处理 #

typescript

const errorHandlerMiddleware = t.middleware(async ({ next }) => {
  try {
    return await next();
  } catch (error) {
    if (error instanceof TRPCError) throw error;
    throw new TRPCError({
      code: 'INTERNAL_SERVER_ERROR',
      message: 'An unexpected error occurred',
      cause: error,
    });
  }
});

4. 性能优化 #

使用批量请求
实现缓存策略
预取数据
无限滚动

5. 安全措施 #

CORS 配置
速率限制
输入验证
敏感数据过滤

总结 #

恭喜你完成了 tRPC 的学习之旅！从基础概念到高级特性，你已经掌握了：

tRPC 的核心概念和优势
路由器和过程的定义
客户端的使用方法
类型系统的深入理解
中间件的实现和应用
错误处理机制
高级特性和最佳实践

现在你可以开始构建自己的 tRPC 应用了！