mcpField
T
trpc-type-safety

trpc-type-safety

github.com/bobmatnyc开发781.0k 次安装15 周前更新
代码仓库

关于 trpc-type-safety

面向全栈 TypeScript 开发者的 tRPC 实践指南，涵盖无需代码生成的端到端类型推断，并介绍路由器、校验、客户端集成与生产模式。

主要功能

  1. 01端到端 TypeScript 类型推断
  2. 02路由器与过程设计
  3. 03Zod 输入校验
  4. 04上下文与中间件模式
  5. 05React Query 客户端集成

适用场景

  1. 01为 TypeScript 应用设计内部 API
  2. 02添加经过校验的查询与变更操作
  3. 03连接 React 客户端与 tRPC 服务端

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/bobmatnyc/claude-mpm-skills --skill trpc-type-safety

安装后在 Claude Code 中输入/trpc-type-safety确认已加载。

附带文件metadata.json

详细介绍

tRPC - 端到端类型安全

progressive_disclosure: entry_point: summary sections: - id: summary title: "tRPC 概览" tokens: 70 next: [when_to_use, quick_start] - id: when_to_use title: "何时使用 tRPC" tokens: 150 next: [quick_start, core_concepts] - id: quick_start title: "快速入门" tokens: 300 next: [core_concepts, router_definition] - id: core_concepts title: "核心概念" tokens: 400 next: [router_definition, procedures] - id: router_definition title: "路由器定义" tokens: 350 next: [procedures, context] - id: procedures title: "过程（查询与修改）" tokens: 400 next: [input_validation, context] - id: input_validation title: "使用 Zod 验证输入" tokens: 350 next: [context, middleware] - id: context title: "上下文管理" tokens: 400 next: [middleware, error_handling] - id: middleware title: "中间件" tokens: 400 next: [error_handling, client_setup] - id: error_handling title: "错误处理" tokens: 350 next: [client_setup, react_integration] - id: client_setup title: "客户端设置" tokens: 400 next: [react_integration, nextjs_integration] - id: react_integration title: "React Query 集成" tokens: 450 next: [nextjs_integration, subscriptions] - id: nextjs_integration title: "Next.js App Router 集成" tokens: 500 next: [subscriptions, file_uploads] - id: subscriptions title: "实时订阅" tokens: 400 next: [file_uploads, batching] - id: file_uploads title: "文件上传" tokens: 300 next: [batching, typescript_inference] - id: batching title: "批量请求与数据加载器" tokens: 350 next: [typescript_inference, testing] - id: typescript_inference title: "TypeScript 类型推断模式" tokens: 300 next: [testing, production_patterns] - id: testing title: "测试策略" tokens: 400 next: [production_patterns, comparison] - id: production_patterns title: "生产环境模式" tokens: 450 next: [comparison, migration] - id: comparison title: "与 REST 和 GraphQL 的比较" tokens: 250 next: [migration, best_practices] - id: migration title: "从 REST 迁移" tokens: 300 next: [best_practices] - id: best_practices title: "最佳实践与性能" tokens: 400

概述

tRPC 让 TypeScript 客户端与服务端之间实现端到端类型安全，无需代码生成。只需定义一次 API，即可在各处自动推断类型。

主要优势：零代码生成、TypeScript 类型推断、React Query 集成、最少的样板代码。

何时使用 tRPC

✅ 最适合：

  • 全栈 TypeScript 应用（Next.js、T3 技术栈）
  • 客户端与服务端共享 TypeScript 代码库的项目
  • 希望获得类似 REST 的简洁性和类似 GraphQL 的类型安全的团队
  • 使用 React Query 获取数据的应用
  • 客户端与服务端都由你掌控的内部 API

❌ 应避免：

  • 由非 TypeScript 客户端使用的公共 API
  • 使用不同语言构建的微服务
  • 使用 Swift/Kotlin 的移动应用（应改用 REST/GraphQL）
  • 需要为外部开发者提供 API 文档（OpenAPI 更合适）

如何选择：

  • tRPC：全栈 TypeScript、单体代码库、内部工具
  • REST：公共 API、语言无关、兼容性广
  • GraphQL：复杂数据图、多种客户端、灵活查询

快速入门

安装
# Server dependencies
npm install @trpc/server zod

# React/Next.js client dependencies
npm install @trpc/client @trpc/react-query @tanstack/react-query
定义路由器（服务端）
// server/trpc.ts
import { initTRPC } from '@trpc/server';
import { z } from 'zod';

const t = initTRPC.create();

export const appRouter = t.router({
  hello: t.procedure
    .input(z.object({ name: z.string() }))
    .query(({ input }) => {
      return { greeting: `Hello ${input.name}` };
    }),

  createPost: t.procedure
    .input(z.object({ title: z.string(), content: z.string() }))
    .mutation(async ({ input }) => {
      // Save to database
      return { id: 1, ...input };
    }),
});

export type AppRouter = typeof appRouter;
在客户端中使用（React）
// client/trpc.ts
import { createTRPCReact } from '@trpc/react-query';
import type { AppRouter } from '../server/trpc';

export const trpc = createTRPCReact<AppRouter>();

// Component
function MyComponent() {
  const { data } = trpc.hello.useQuery({ name: 'World' });
  const createPost = trpc.createPost.useMutation();

  return <div>{data?.greeting}</div>; // Fully typed!
}

接下来：了解核心概念，或直接阅读路由器定义。

核心概念

tRPC 理念

tRPC 通过在客户端和服务端之间共享 TypeScript 类型，提供类型安全的远程过程调用。无需生成代码，只需利用 TypeScript 的类型推断。

核心组件
  1. 路由器：由多个过程（API 端点）组成的集合
  2. 过程：单个 API 操作（查询或变更）
  3. 上下文：请求范围内的数据（用户、数据库等）
  4. 中间件：拦截或修改请求（身份验证、日志记录）
  5. 输入/输出：使用 Zod 模式进行验证
类型流转
// Server defines types
const router = t.router({
  getUser: t.procedure
    .input(z.string())
    .query(({ input }) => ({ id: input, name: 'Alice' })),
});

// Client gets automatic types
const user = await trpc.getUser.query('123');
// user is typed as { id: string, name: string }
架构模式
┌─────────────┐     Type-safe     ┌──────────────┐
│   Client    │ ←────────────────→ │    Server    │
│ (React)     │   No codegen!      │   (Node.js)  │
└─────────────┘                    └──────────────┘
      ↓                                    ↓
 React Query                          tRPC Router
 (caching)                            (procedures)

优势：

  • 更改会即时传播（无需构建步骤）
  • 重命名重构可跨客户端和服务端生效
  • 不可能调用错误类型
  • 所有 API 方法均支持自动补全

路由器定义

基本路由器结构
import { initTRPC } from '@trpc/server';

const t = initTRPC.create();

export const appRouter = t.router({
  // Procedures go here
});

export type AppRouter = typeof appRouter;
嵌套路由器（命名空间）
const userRouter = t.router({
  getById: t.procedure
    .input(z.string())
    .query(({ input }) => getUser(input)),

  create: t.procedure
    .input(z.object({ name: z.string(), email: z.string() }))
    .mutation(({ input }) => createUser(input)),
});

const postRouter = t.router({
  list: t.procedure.query(() => getPosts()),
  create: t.procedure
    .input(z.object({ title: z.string() }))
    .mutation(({ input }) => createPost(input)),
});

export const appRouter = t.router({
  user: userRouter,
  post: postRouter,
});

// Client usage:
// trpc.user.getById.useQuery('123')
// trpc.post.list.useQuery()
合并路由器
import { adminRouter } from './admin';
import { publicRouter } from './public';

export const appRouter = t.mergeRouters(publicRouter, adminRouter);
路由器组织最佳实践
server/
├── trpc.ts           # tRPC instance, context, middleware
├── routers/
│   ├── user.ts       # User-related procedures
│   ├── post.ts       # Post-related procedures
│   └── index.ts      # Combine all routers
└── index.ts          # Export AppRouter type

过程（查询与变更）

查询过程（读取操作）
const router = t.router({
  // Simple query
  getUser: t.procedure
    .input(z.string())
    .query(({ input }) => {
      return db.user.findUnique({ where: { id: input } });
    }),

  // Query with multiple inputs
  searchUsers: t.procedure
    .input(z.object({
      query: z.string(),
      limit: z.number().default(10),
    }))
    .query(({ input }) => {
      return db.user.findMany({
        where: { name: { contains: input.query } },
        take: input.limit,
      });
    }),
});
变更过程（写入操作）
const router = t.router({
  createUser: t.procedure
    .input(z.object({
      name: z.string().min(3),
      email: z.string().email(),
    }))
    .mutation(async ({ input }) => {
      return await db.user.create({ data: input });
    }),

  updateUser: t.procedure
    .input(z.object({
      id: z.string(),
      data: z.object({
        name: z.string().optional(),
        email: z.string().email().optional(),
      }),
    }))
    .mutation(async ({ input }) => {
      return await db.user.update({
        where: { id: input.id },
        data: input.data,
      });
    }),
});
查询与变更的区别
方面查询变更
用途读取数据修改数据
HTTP 方法GETPOST
缓存由 React Query 缓存不缓存
幂等是否
副作用无数据库写入、电子邮件等
输出类型
const router = t.router({
  getUser: t.procedure
    .input(z.string())
    .output(z.object({ id: z.string(), name: z.string() })) // Optional
    .query(({ input }) => {
      return { id: input, name: 'Alice' };
    }),
});

注意：输出验证会增加运行时开销——仅对关键数据使用。

使用 Zod 验证输入

为什么选择 Zod？

tRPC 使用 Zod 进行运行时类型验证和 TypeScript 类型推断。Zod 模式提供：

  • 运行时验证（防止无效数据）
  • TypeScript 类型（从模式自动推断）
  • 转换（解析、强制转换、默认值）
基本验证
import { z } from 'zod';

const router = t.router({
  createPost: t.procedure
    .input(z.object({
      title: z.string().min(5).max(100),
      content: z.string(),
      published: z.boolean().default(false),
      tags: z.array(z.string()).optional(),
    }))
    .mutation(({ input }) => {
      // input is fully typed and validated
      return createPost(input);
    }),
});
高级验证
const createUserInput = z.object({
  email: z.string().email(),
  password: z.string().min(8),
  age: z.number().int().min(18),
  role: z.enum(['user', 'admin']),
  metadata: z.record(z.string(), z.unknown()).optional(),
});

const router = t.router({
  createUser: t.procedure
    .input(createUserInput)
    .mutation(({ input }) => {
      // All validation passed
      return saveUser(input);
    }),
});
转换
const router = t.router({
  getUser: t.procedure
    .input(
      z.object({
        id: z.string().transform((id) => parseInt(id, 10)),
      })
    )
    .query(({ input }) => {
      // input.id is now a number
      return db.user.findUnique({ where: { id: input.id } });
    }),
});
可复用模式
// schemas/user.ts
export const CreateUserSchema = z.object({
  name: z.string(),
  email: z.string().email(),
});

export const UpdateUserSchema = CreateUserSchema.partial().extend({
  id: z.string(),
});

// routers/user.ts
const router = t.router({
  create: t.procedure.input(CreateUserSchema).mutation(/*...*/),
  update: t.procedure.input(UpdateUserSchema).mutation(/*...*/),
});

上下文管理

什么是上下文？

上下文为所有过程提供请求范围内的数据，例如身份验证信息、数据库连接、日志记录等。

创建上下文
import { inferAsyncReturnType } from '@trpc/server';
import { CreateNextContextOptions } from '@trpc/server/adapters/next';

export async function createContext(opts: CreateNextContextOptions) {
  const session = await getSession(opts.req);

  return {
    session,
    db: prisma,
    req: opts.req,
    res: opts.res,
  };
}

export type Context = inferAsyncReturnType<typeof createContext>;

const t = initTRPC.context<Context>().create();
在过程中使用上下文
const router = t.router({
  getMe: t.procedure.query(({ ctx }) => {
    if (!ctx.session?.user) {
      throw new TRPCError({ code: 'UNAUTHORIZED' });
    }

    return ctx.db.user.findUnique({
      where: { id: ctx.session.user.id },
    });
  }),

  createPost: t.procedure
    .input(z.object({ title: z.string() }))
    .mutation(async ({ ctx, input }) => {
      return ctx.db.post.create({
        data: {
          title: input.title,
          authorId: ctx.session.user.id,
        },
      });
    }),
});
上下文最佳实践
// ✅ Good: Lazy database connection
export async function createContext(opts: CreateNextContextOptions) {
  return {
    getDB: () => prisma, // Lazy
    session: await getSession(opts.req),
  };
}

// ❌ Bad: Heavy computation in context
export async function createContext(opts: CreateNextContextOptions) {
  const allUsers = await prisma.user.findMany(); // Too expensive!
  return { allUsers };
}

中间件

什么是中间件？

中间件会拦截过程调用，以添加横切关注点：日志记录、计时、身份验证、速率限制。

基本中间件
const loggerMiddleware = t.middleware(async ({ path, type, next }) => {
  const start = Date.now();
  console.log(`→ ${type} ${path}`);

  const result = await next();

  const duration = Date.now() - start;
  console.log(`✓ ${type} ${path} - ${duration}ms`);

  return result;
});

const loggedProcedure = t.procedure.use(loggerMiddleware);
身份验证中间件
const isAuthed = t.middleware(({ ctx, next }) => {
  if (!ctx.session?.user) {
    throw new TRPCError({ code: 'UNAUTHORIZED' });
  }

  return next({
    ctx: {
      ...ctx,
      user: ctx.session.user, // Narrow type
    },
  });
});

// Protected procedure builder
const protectedProcedure = t.procedure.use(isAuthed);

const router = t.router({
  // Public
  getPublicPosts: t.procedure.query(() => getPosts()),

  // Protected - requires authentication
  getMyPosts: protectedProcedure.query(({ ctx }) => {
    // ctx.user is guaranteed to exist
    return getPostsByUser(ctx.user.id);
  }),
});
链式中间件
const timingMiddleware = t.middleware(async ({ next }) => {
  const start = performance.now();
  const result = await next();
  console.log(`Execution time: ${performance.now() - start}ms`);
  return result;
});

const rateLimitMiddleware = t.middleware(async ({ ctx, next }) => {
  await checkRateLimit(ctx.session?.user?.id);
  return next();
});

const protectedProcedure = t.procedure
  .use(timingMiddleware)
  .use(rateLimitMiddleware)
  .use(isAuthed);
上下文转换
const enrichContextMiddleware = t.middleware(async ({ ctx, next }) => {
  const user = ctx.session?.user
    ? await ctx.db.user.findUnique({ where: { id: ctx.session.user.id } })
    : null;

  return next({
    ctx: {
      ...ctx,
      user, // Full user object
    },
  });
});

错误处理

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

const router = t.router({
  getUser: t.procedure
    .input(z.string())
    .query(async ({ input }) => {
      const user = await db.user.findUnique({ where: { id: input } });

      if (!user) {
        throw new TRPCError({
          code: 'NOT_FOUND',
          message: `User ${input} not found`,
        });
      }

      return user;
    }),
});
错误代码
代码HTTP 状态使用场景
BAD_REQUEST400输入无效
UNAUTHORIZED401未通过身份验证
FORBIDDEN403未获授权
NOT_FOUND404未找到资源
TIMEOUT408请求超时
CONFLICT409资源冲突
PRECONDITION_FAILED412前置条件未满足
PAYLOAD_TOO_LARGE413请求过大
TOO_MANY_REQUESTS429超出速率限制
CLIENT_CLOSED_REQUEST499客户端已关闭连接
INTERNAL_SERVER_ERROR500服务端错误
自定义错误处理
const router = t.router({
  deleteUser: t.procedure
    .input(z.string())
    .mutation(async ({ input, ctx }) => {
      try {
        return await ctx.db.user.delete({ where: { id: input } });
      } catch (error) {
        if (error.code === 'P2025') { // Prisma not found
          throw new TRPCError({
            code: 'NOT_FOUND',
            message: 'User not found',
            cause: error,
          });
        }
        throw new TRPCError({
          code: 'INTERNAL_SERVER_ERROR',
          message: 'Failed to delete user',
          cause: error,
        });
      }
    }),
});
错误格式化
const t = initTRPC.context<Context>().create({
  errorFormatter({ shape, error }) {
    return {
      ...shape,
      data: {
        ...shape.data,
        zodError:
          error.code === 'BAD_REQUEST' && error.cause instanceof ZodError
            ? error.cause.flatten()
            : null,
      },
    };
  },
});
客户端错误处理
function MyComponent() {
  const mutation = trpc.createUser.useMutation({
    onError: (error) => {
      if (error.data?.code === 'UNAUTHORIZED') {
        router.push('/login');
      } else {
        toast.error(error.message);
      }
    },
  });
}

客户端设置

原生客户端
import { createTRPCProxyClient, httpBatchLink } from '@trpc/client';
import type { AppRouter } from './server';

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

// Usage
const user = await client.user.getById.query('123');
const newPost = await client.post.create.mutate({ title: 'Hello' });
React 客户端设置
// utils/trpc.ts
import { createTRPCReact } from '@trpc/react-query';
import type { AppRouter } from '../server/routers';

export const trpc = createTRPCReact<AppRouter>();

// _app.tsx
import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
import { httpBatchLink } from '@trpc/client';
import { useState } from 'react';
import { trpc } from '../utils/trpc';

export default function App({ Component, pageProps }: AppProps) {
  const [queryClient] = useState(() => new QueryClient());
  const [trpcClient] = useState(() =>
    trpc.createClient({
      links: [
        httpBatchLink({
          url: 'http://localhost:3000/api/trpc',
        }),
      ],
    })
  );

  return (
    <trpc.Provider client={trpcClient} queryClient={queryClient}>
      <QueryClientProvider client={queryClient}>
        <Component {...pageProps} />
      </QueryClientProvider>
    </trpc.Provider>
  );
}
Next.js API 路由
// pages/api/trpc/[trpc].ts
import { createNextApiHandler } from '@trpc/server/adapters/next';
import { appRouter } from '../../../server/routers';
import { createContext } from '../../../server/context';

export default createNextApiHandler({
  router: appRouter,
  createContext,
});
请求头与身份验证
const trpcClient = trpc.createClient({
  links: [
    httpBatchLink({
      url: 'http://localhost:3000/api/trpc',
      headers: async () => {
        const token = await getAuthToken();
        return {
          authorization: token ? `Bearer ${token}` : undefined,
        };
      },
    }),
  ],
});

React Query 集成

useQuery Hook
function UserProfile({ userId }: { userId: string }) {
  const { data, isLoading, error } = trpc.user.getById.useQuery(userId);

  if (isLoading) return <div>Loading...</div>;
  if (error) return <div>Error: {error.message}</div>;

  return <div>{data.name}</div>;
}
查询选项
const { data } = trpc.posts.list.useQuery(undefined, {
  refetchOnWindowFocus: false,
  staleTime: 5 * 60 * 1000, // 5 minutes
  cacheTime: 10 * 60 * 1000, // 10 minutes
  retry: 3,
  onSuccess: (data) => console.log('Fetched', data.length, 'posts'),
});
useMutation Hook
function CreatePostForm() {
  const utils = trpc.useContext();

  const createPost = trpc.post.create.useMutation({
    onSuccess: () => {
      // Invalidate and refetch
      utils.post.list.invalidate();
    },
  });

  const handleSubmit = (data: { title: string }) => {
    createPost.mutate(data);
  };

  return (
    <form onSubmit={handleSubmit}>
      <input name="title" />
      <button disabled={createPost.isLoading}>
        {createPost.isLoading ? 'Creating...' : 'Create'}
      </button>
      {createPost.error && <p>{createPost.error.message}</p>}
    </form>
  );
}
乐观更新
const createPost = trpc.post.create.useMutation({
  onMutate: async (newPost) => {
    // Cancel outgoing refetches
    await utils.post.list.cancel();

    // Snapshot previous value
    const previousPosts = utils.post.list.getData();

    // Optimistically update
    utils.post.list.setData(undefined, (old) => [
      ...(old ?? []),
      { id: 'temp', ...newPost },
    ]);

    return { previousPosts };
  },
  onError: (err, newPost, context) => {
    // Rollback on error
    utils.post.list.setData(undefined, context?.previousPosts);
  },
  onSettled: () => {
    // Refetch after success or error
    utils.post.list.invalidate();
  },
});
无限查询
// Server
const router = t.router({
  posts: t.procedure
    .input(z.object({
      cursor: z.number().optional(),
      limit: z.number().default(10),
    }))
    .query(({ input }) => {
      const posts = getPosts(input.cursor, input.limit);
      return {
        posts,
        nextCursor: posts.length === input.limit ? input.cursor + input.limit : undefined,
      };
    }),
});

// Client
function PostList() {
  const {
    data,
    fetchNextPage,
    hasNextPage,
    isFetchingNextPage,
  } = trpc.posts.useInfiniteQuery(
    { limit: 10 },
    {
      getNextPageParam: (lastPage) => lastPage.nextCursor,
    }
  );

  return (
    <div>
      {data?.pages.map((page) =>
        page.posts.map((post) => <PostCard key={post.id} post={post} />)
      )}
      {hasNextPage && (
        <button onClick={() => fetchNextPage()} disabled={isFetchingNextPage}>
          Load More
        </button>
      )}
    </div>
  );
}

Next.js App Router 集成

服务端组件
// app/users/page.tsx (Server Component)
import { createCaller } from '../server/routers';
import { createContext } from '../server/context';

export default async function UsersPage() {
  const ctx = await createContext({ req: null, res: null });
  const caller = createCaller(ctx);

  const users = await caller.user.list();

  return (
    <div>
      {users.map((user) => (
        <div key={user.id}>{user.name}</div>
      ))}
    </div>
  );
}
服务端操作
// app/actions.ts
'use server';

import { createCaller } from '../server/routers';
import { createContext } from '../server/context';

export async function createPost(formData: FormData) {
  const ctx = await createContext({ req: null, res: null });
  const caller = createCaller(ctx);

  return caller.post.create({
    title: formData.get('title') as string,
    content: formData.get('content') as string,
  });
}
App Router Provider
// app/providers.tsx
'use client';

import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
import { httpBatchLink } from '@trpc/client';
import { useState } from 'react';
import { trpc } from './trpc';

export function Providers({ children }: { children: React.ReactNode }) {
  const [queryClient] = useState(() => new QueryClient());
  const [trpcClient] = useState(() =>
    trpc.createClient({
      links: [
        httpBatchLink({
          url: '/api/trpc',
        }),
      ],
    })
  );

  return (
    <trpc.Provider client={trpcClient} queryClient={queryClient}>
      <QueryClientProvider client={queryClient}>
        {children}
      </QueryClientProvider>
    </trpc.Provider>
  );
}

// app/layout.tsx
import { Providers } from './providers';

export default function RootLayout({ children }) {
  return (
    <html>
      <body>
        <Providers>{children}</Providers>
      </body>
    </html>
  );
}
App Router 中的客户端组件
// app/posts/create-button.tsx
'use client';

import { trpc } from '../trpc';

export function CreatePostButton() {
  const createPost = trpc.post.create.useMutation();

  return (
    <button onClick={() => createPost.mutate({ title: 'New Post' })}>
      Create Post
    </button>
  );
}
API 路由处理器（App Router）
// app/api/trpc/[trpc]/route.ts
import { fetchRequestHandler } from '@trpc/server/adapters/fetch';
import { appRouter } from '../../../../server/routers';
import { createContext } from '../../../../server/context';

const handler = (req: Request) =>
  fetchRequestHandler({
    endpoint: '/api/trpc',
    req,
    router: appRouter,
    createContext,
  });

export { handler as GET, handler as POST };

实时订阅

WebSocket 设置（服务端）
import { applyWSSHandler } from '@trpc/server/adapters/ws';
import ws from 'ws';

const wss = new ws.Server({ port: 3001 });

applyWSSHandler({
  wss,
  router: appRouter,
  createContext,
});

console.log('WebSocket server listening on port 3001');
订阅过程
import { observable } from '@trpc/server/observable';
import { EventEmitter } from 'events';

const ee = new EventEmitter();

const router = t.router({
  onPostAdd: t.procedure.subscription(() => {
    return observable<Post>((emit) => {
      const onAdd = (data: Post) => emit.next(data);

      ee.on('add', onAdd);

      return () => {
        ee.off('add', onAdd);
      };
    });
  }),

  createPost: t.procedure
    .input(z.object({ title: z.string() }))
    .mutation(({ input }) => {
      const post = { id: Date.now().toString(), ...input };
      ee.emit('add', post); // Emit to subscribers
      return post;
    }),
});
客户端 WebSocket 设置
import { createWSClient, wsLink } from '@trpc/client';

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

const trpcClient = trpc.createClient({
  links: [
    wsLink({
      client: wsClient,
    }),
  ],
});
useSubscription Hook
function PostFeed() {
  const [posts, setPosts] = useState<Post[]>([]);

  trpc.onPostAdd.useSubscription(undefined, {
    onData: (post) => {
      setPosts((prev) => [post, ...prev]);
    },
    onError: (err) => {
      console.error('Subscription error:', err);
    },
  });

  return (
    <div>
      {posts.map((post) => (
        <PostCard key={post.id} post={post} />
      ))}
    </div>
  );
}
带输入的订阅
// Server
const router = t.router({
  onUserStatusChange: t.procedure
    .input(z.string())
    .subscription(({ input }) => {
      return observable<UserStatus>((emit) => {
        const onChange = (userId: string, status: UserStatus) => {
          if (userId === input) {
            emit.next(status);
          }
        };

        ee.on('statusChange', onChange);
        return () => ee.off('statusChange', onChange);
      });
    }),
});

// Client
trpc.onUserStatusChange.useSubscription('user-123', {
  onData: (status) => console.log('Status:', status),
});

文件上传

Multipart 表单数据（服务端）
// Next.js API route with file upload
import { NextApiRequest, NextApiResponse } from 'next';
import formidable from 'formidable';
import fs from 'fs';

export const config = {
  api: { bodyParser: false },
};

export default async function handler(req: NextApiRequest, res: NextApiResponse) {
  const form = formidable({ multiples: false });

  form.parse(req, async (err, fields, files) => {
    if (err) return res.status(500).json({ error: 'Upload failed' });

    const file = files.file as formidable.File;
    const buffer = fs.readFileSync(file.filepath);

    // Upload to S3, etc.
    const url = await uploadToS3(buffer, file.originalFilename);

    res.json({ url });
  });
}
Base64 上传（tRPC）
// For small files only (<1MB)
const router = t.router({
  uploadAvatar: t.procedure
    .input(z.object({
      fileName: z.string(),
      fileData: z.string(), // Base64
    }))
    .mutation(async ({ input }) => {
      const buffer = Buffer.from(input.fileData, 'base64');
      const url = await uploadToS3(buffer, input.fileName);
      return { url };
    }),
});

// Client
const uploadAvatar = trpc.uploadAvatar.useMutation();

const handleFileChange = (e: React.ChangeEvent<HTMLInputElement>) => {
  const file = e.target.files?.[0];
  if (!file) return;

  const reader = new FileReader();
  reader.onload = () => {
    const base64 = reader.result as string;
    uploadAvatar.mutate({
      fileName: file.name,
      fileData: base64.split(',')[1], // Remove data:image/...;base64,
    });
  };
  reader.readAsDataURL(file);
};
签名 URL 模式（推荐）
// Step 1: Get signed upload URL from tRPC
const router = t.router({
  getUploadUrl: t.procedure
    .input(z.object({
      fileName: z.string(),
      fileType: z.string(),
    }))
    .mutation(async ({ input }) => {
      const signedUrl = await s3.getSignedUrl('putObject', {
        Bucket: 'my-bucket',
        Key: input.fileName,
        ContentType: input.fileType,
        Expires: 60, // 1 minute
      });

      return { uploadUrl: signedUrl, fileUrl: `https://cdn.example.com/${input.fileName}` };
    }),
});

// Step 2: Client uploads directly to S3
async function uploadFile(file: File) {
  // Get signed URL
  const { uploadUrl, fileUrl } = await trpc.getUploadUrl.mutate({
    fileName: file.name,
    fileType: file.type,
  });

  // Upload directly to S3
  await fetch(uploadUrl, {
    method: 'PUT',
    body: file,
    headers: { 'Content-Type': file.type },
  });

  // Save file URL to database via tRPC
  await trpc.user.updateAvatar.mutate({ url: fileUrl });
}

批量请求与 DataLoader

自动批处理
// Client configuration
const trpcClient = trpc.createClient({
  links: [
    httpBatchLink({
      url: 'http://localhost:3000/api/trpc',
      maxBatchSize: 10, // Batch up to 10 requests
    }),
  ],
});

// Multiple calls made close together are batched into one HTTP request
const user1 = trpc.user.getById.useQuery('1');
const user2 = trpc.user.getById.useQuery('2');
const user3 = trpc.user.getById.useQuery('3');
// → Single HTTP request with 3 procedure calls
DataLoader 模式
import DataLoader from 'dataloader';

// Create DataLoader in context
export async function createContext() {
  const userLoader = new DataLoader(async (ids: readonly string[]) => {
    const users = await db.user.findMany({
      where: { id: { in: [...ids] } },
    });

    // Return in same order as input
    return ids.map((id) => users.find((u) => u.id === id));
  });

  return { userLoader };
}

// Use in procedures
const router = t.router({
  getUser: t.procedure
    .input(z.string())
    .query(({ ctx, input }) => {
      return ctx.userLoader.load(input); // Batched!
    }),

  getPosts: t.procedure.query(async ({ ctx }) => {
    const posts = await db.post.findMany({ take: 10 });

    // N+1 problem solved—all authors fetched in one query
    const postsWithAuthors = await Promise.all(
      posts.map(async (post) => ({
        ...post,
        author: await ctx.userLoader.load(post.authorId),
      }))
    );

    return postsWithAuthors;
  }),
});
条件批处理
import { httpBatchLink, httpLink, splitLink } from '@trpc/client';

const trpcClient = trpc.createClient({
  links: [
    splitLink({
      // Batch queries, don't batch mutations
      condition: (op) => op.type === 'query',
      true: httpBatchLink({ url: '/api/trpc' }),
      false: httpLink({ url: '/api/trpc' }),
    }),
  ],
});

TypeScript 类型推断模式

从 Router 推断类型
import type { inferRouterInputs, inferRouterOutputs } from '@trpc/server';
import type { AppRouter } from './server';

// Input types
type RouterInputs = inferRouterInputs<AppRouter>;
type CreateUserInput = RouterInputs['user']['create'];

// Output types
type RouterOutputs = inferRouterOutputs<AppRouter>;
type User = RouterOutputs['user']['getById'];

// Use in components
function UserCard({ user }: { user: User }) {
  return <div>{user.name}</div>;
}
过程辅助工具
import type { inferProcedureInput, inferProcedureOutput } from '@trpc/server';

type CreatePostInput = inferProcedureInput<AppRouter['post']['create']>;
type Post = inferProcedureOutput<AppRouter['post']['getById']>;
Context 类型推断
import { inferAsyncReturnType } from '@trpc/server';

export async function createContext() {
  return {
    db: prisma,
    user: null as User | null,
  };
}

export type Context = inferAsyncReturnType<typeof createContext>;

const t = initTRPC.context<Context>().create();
泛型过程
// Reusable pagination
function createPaginatedProcedure<T>(
  getData: (cursor: number, limit: number) => Promise<T[]>
) {
  return t.procedure
    .input(z.object({
      cursor: z.number().optional(),
      limit: z.number().default(10),
    }))
    .query(async ({ input }) => {
      const items = await getData(input.cursor ?? 0, input.limit);
      return {
        items,
        nextCursor: items.length === input.limit
          ? (input.cursor ?? 0) + input.limit
          : undefined,
      };
    });
}

const router = t.router({
  posts: createPaginatedProcedure((cursor, limit) =>
    db.post.findMany({ skip: cursor, take: limit })
  ),
  users: createPaginatedProcedure((cursor, limit) =>
    db.user.findMany({ skip: cursor, take: limit })
  ),
});

测试策略

过程单元测试
import { createCaller } from '../routers';

describe('User Router', () => {
  it('should create user', async () => {
    const ctx = {
      db: mockDb,
      session: null,
    };

    const caller = createCaller(ctx);

    const result = await caller.user.create({
      name: 'Alice',
      email: '[email protected]',
    });

    expect(result).toMatchObject({
      name: 'Alice',
      email: '[email protected]',
    });
  });
});
集成测试
import { httpBatchLink } from '@trpc/client';
import { createTRPCProxyClient } from '@trpc/client';
import type { AppRouter } from '../server';

describe('tRPC Integration', () => {
  const client = createTRPCProxyClient<AppRouter>({
    links: [
      httpBatchLink({
        url: 'http://localhost:3000/api/trpc',
      }),
    ],
  });

  it('should fetch user', async () => {
    const user = await client.user.getById.query('123');
    expect(user.id).toBe('123');
  });
});
模拟 Context
import { createCaller } from '../routers';

const mockContext = {
  db: {
    user: {
      findUnique: vi.fn().mockResolvedValue({ id: '1', name: 'Alice' }),
      create: vi.fn(),
    },
  },
  session: {
    user: { id: '1', email: '[email protected]' },
  },
};

it('should get current user', async () => {
  const caller = createCaller(mockContext);
  const user = await caller.user.getMe();

  expect(mockContext.db.user.findUnique).toHaveBeenCalledWith({
    where: { id: '1' },
  });
  expect(user.name).toBe('Alice');
});
测试 React Hooks
import { renderHook, waitFor } from '@testing-library/react';
import { createWrapper } from './test-utils';

it('should fetch posts', async () => {
  const { result } = renderHook(() => trpc.post.list.useQuery(), {
    wrapper: createWrapper(),
  });

  await waitFor(() => expect(result.current.isSuccess).toBe(true));

  expect(result.current.data).toHaveLength(10);
});

// test-utils.ts
import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
import { httpBatchLink } from '@trpc/client';
import { trpc } from '../utils/trpc';

export function createWrapper() {
  const queryClient = new QueryClient();
  const trpcClient = trpc.createClient({
    links: [httpBatchLink({ url: 'http://localhost:3000/api/trpc' })],
  });

  return ({ children }) => (
    <trpc.Provider client={trpcClient} queryClient={queryClient}>
      <QueryClientProvider client={queryClient}>
        {children}
      </QueryClientProvider>
    </trpc.Provider>
  );
}

生产环境模式

错误监控
import * as Sentry from '@sentry/node';

const t = initTRPC.context<Context>().create({
  errorFormatter({ shape, error }) {
    // Log to Sentry
    if (error.code === 'INTERNAL_SERVER_ERROR') {
      Sentry.captureException(error);
    }

    return {
      ...shape,
      data: {
        ...shape.data,
        // Don't expose internal errors in production
        message: process.env.NODE_ENV === 'production' && error.code === 'INTERNAL_SERVER_ERROR'
          ? 'Internal server error'
          : shape.message,
      },
    };
  },
});
速率限制
import { Ratelimit } from '@upstash/ratelimit';
import { Redis } from '@upstash/redis';

const ratelimit = new Ratelimit({
  redis: Redis.fromEnv(),
  limiter: Ratelimit.slidingWindow(10, '10 s'),
});

const rateLimitMiddleware = t.middleware(async ({ ctx, next }) => {
  const identifier = ctx.session?.user?.id ?? ctx.req.ip;
  const { success } = await ratelimit.limit(identifier);

  if (!success) {
    throw new TRPCError({
      code: 'TOO_MANY_REQUESTS',
      message: 'Rate limit exceeded',
    });
  }

  return next();
});
缓存
import { Redis } from 'ioredis';

const redis = new Redis(process.env.REDIS_URL);

const router = t.router({
  getUser: t.procedure
    .input(z.string())
    .query(async ({ input }) => {
      // Check cache
      const cached = await redis.get(`user:${input}`);
      if (cached) return JSON.parse(cached);

      // Fetch from database
      const user = await db.user.findUnique({ where: { id: input } });

      // Cache for 5 minutes
      await redis.setex(`user:${input}`, 300, JSON.stringify(user));

      return user;
    }),
});
请求日志记录
const loggingMiddleware = t.middleware(async ({ path, type, next, input }) => {
  const start = Date.now();

  console.log(`→ ${type} ${path}`, { input });

  try {
    const result = await next();
    const duration = Date.now() - start;
    console.log(`✓ ${type} ${path} - ${duration}ms`);
    return result;
  } catch (error) {
    const duration = Date.now() - start;
    console.error(`✗ ${type} ${path} - ${duration}ms`, { error });
    throw error;
  }
});
OpenTelemetry 集成
import { trace } from '@opentelemetry/api';

const tracingMiddleware = t.middleware(async ({ path, type, next }) => {
  const tracer = trace.getTracer('trpc');

  return tracer.startActiveSpan(`trpc.${type}.${path}`, async (span) => {
    try {
      const result = await next();
      span.setStatus({ code: 0 }); // OK
      return result;
    } catch (error) {
      span.setStatus({ code: 2, message: error.message }); // ERROR
      span.recordException(error);
      throw error;
    } finally {
      span.end();
    }
  });
});

与 REST 和 GraphQL 的比较

功能比较
功能tRPCRESTGraphQL
类型安全完整（TypeScript）手动／代码生成手动／代码生成
代码生成无可选（OpenAPI）必需
学习曲线低低中／高
客户端库仅 TypeScript任意语言任意语言
API 文档TypeScript 类型OpenAPI／SwaggerSchema／内省
公共 API❌ 否✅ 是✅ 是
灵活查询❌ 固定❌ 固定✅ 是
过度获取极少常见无
缓存React QueryHTTP 缓存复杂
实时性WebSocketSSE／WebSocket订阅
文件上传变通方案原生支持复杂
如何选择

tRPC：

  • ✅ 全栈 TypeScript 单体仓库
  • ✅ 内部工具和仪表板
  • ✅ Next.js 应用
  • ✅ 小团队快速开发
  • ❌ 面向外部使用者的公共 API
  • ❌ 多语言客户端

REST：

  • ✅ 兼容性广泛的公共 API
  • ✅ 多语言服务
  • ✅ HTTP 缓存需求
  • ✅ 文件上传和下载
  • ❌ 复杂的嵌套数据结构
  • ❌ 无需代码生成即可实现类型安全

GraphQL：

  • ✅ 复杂的数据图
  • ✅ 多种客户端类型（Web、移动端等）
  • ✅ 需要灵活查询
  • ✅ 避免过度获取
  • ❌ 简单的 CRUD 操作
  • ❌ 小团队（复杂度开销）
迁移路径

tRPC 可以与 REST／GraphQL 共存：

// Use tRPC for internal, REST for public
const router = t.router({
  internal: internalRouter, // tRPC only
});

// Expose REST endpoints separately
app.get('/api/public/users', publicRestHandler);

从 REST 迁移

渐进式迁移策略
  1. 在 REST 旁添加 tRPC：不要一次性重写所有内容
  2. 使用 tRPC 开发新功能：从新的端点开始
  3. 迁移高价值端点：优先处理复杂或频繁使用的 API
  4. 将公共 API 保留在 REST 中：只迁移内部调用
将 REST 转换为 tRPC

之前（REST）：

// pages/api/users/[id].ts
export default async function handler(req, res) {
  if (req.method === 'GET') {
    const user = await db.user.findUnique({ where: { id: req.query.id } });
    res.json(user);
  } else if (req.method === 'PATCH') {
    const user = await db.user.update({
      where: { id: req.query.id },
      data: req.body,
    });
    res.json(user);
  }
}

// Client
const response = await fetch(`/api/users/${id}`);
const user = await response.json(); // No types!

之后（tRPC）：

// server/routers/user.ts
export const userRouter = t.router({
  getById: t.procedure
    .input(z.string())
    .query(({ input }) => db.user.findUnique({ where: { id: input } })),

  update: t.procedure
    .input(z.object({
      id: z.string(),
      data: z.object({ name: z.string().optional() }),
    }))
    .mutation(({ input }) => db.user.update({
      where: { id: input.id },
      data: input.data,
    })),
});

// Client
const user = await trpc.user.getById.query(id); // Fully typed!
共享验证
// Reuse Zod schemas across REST and tRPC during migration
import { createUserSchema } from '../schemas/user';

// tRPC
const router = t.router({
  createUser: t.procedure
    .input(createUserSchema)
    .mutation(({ input }) => createUser(input)),
});

// REST (validate with same schema)
export default async function handler(req, res) {
  const parsed = createUserSchema.safeParse(req.body);
  if (!parsed.success) {
    return res.status(400).json({ errors: parsed.error });
  }
  const user = await createUser(parsed.data);
  res.json(user);
}

最佳实践与性能

代码组织
server/
├── trpc.ts              # tRPC instance, base procedures
├── context.ts           # Context creation
├── middleware/
│   ├── auth.ts          # Authentication middleware
│   ├── logging.ts       # Logging middleware
│   └── rateLimit.ts     # Rate limiting
├── routers/
│   ├── _app.ts          # Root router
│   ├── user.ts          # User procedures
│   ├── post.ts          # Post procedures
│   └── admin/
│       └── index.ts     # Admin-only procedures
└── schemas/
    ├── user.ts          # User Zod schemas
    └── post.ts          # Post Zod schemas
性能提示

  1. 对多个查询使用批处理：

    httpBatchLink({ url: '/api/trpc', maxBatchSize: 10 })

  2. 为 N+1 查询实现 DataLoader：

    const userLoader = new DataLoader(batchLoadUsers);

  3. 缓存开销较大的查询：

    trpc.posts.list.useQuery(undefined, { staleTime: 5 * 60 * 1000 });

  4. 优化数据库查询：

    // ❌ Bad: N+1 query
const posts = await db.post.findMany();
const postsWithAuthors = await Promise.all(
  posts.map((p) => db.user.findUnique({ where: { id: p.authorId } }))
);

// ✅ Good: Single query with include
const posts = await db.post.findMany({
  include: { author: true },
});

  5. 使用 React Query 的去重功能：

    // Multiple components can call same query—React Query deduplicates
const { data } = trpc.user.getMe.useQuery();
安全最佳实践
  1. 始终使用 Zod 验证输入
  2. 使用中间件进行身份验证： 
    const protectedProcedure = t.procedure.use(isAuthed);
  3. 在生产环境中清理错误消息
  4. 实施速率限制
  5. 在生产环境中使用 HTTPS
  6. 正确设置 CORS： 
    createNextApiHandler({
  router: appRouter,
  createContext,
  onError: ({ error }) => {
    if (error.code === 'INTERNAL_SERVER_ERROR') {
      console.error('Internal error:', error);
    }
  },
});
类型安全提示

  1. 导出 Router 类型，而非实现：

    export type AppRouter = typeof appRouter; // ✅
// Don't export `appRouter` itself to client

  2. 使用 satisfies 改进类型推断：

    const input = {
  name: 'Alice',
  age: 30,
} satisfies CreateUserInput;

  3. 避免在 Context 中使用 any：

    // ❌ Bad
ctx: { user: any }

// ✅ Good
ctx: { user: User | null }
开发工作流
  1. 先定义 Schema：在编写过程之前编写 Zod Schema
  2. 单独测试过程：使用 createCaller 进行单元测试
  3. 使用 TypeScript 严格模式：及早发现类型错误
  4. 启用 React Query DevTools： 
    import { ReactQueryDevtools } from '@tanstack/react-query-devtools';

<ReactQueryDevtools initialIsOpen={false} />
常见陷阱

❌ 不要返回敏感数据：

// Bad: Exposes password hash
.query(() => db.user.findMany())

// Good: Select specific fields
.query(() => db.user.findMany({ select: { id: true, name: true } }))

❌ 读取操作不要使用变更：

// Bad: Side-effect-free operation as mutation
getMostRecentPost: t.procedure.mutation(() => getPost())

// Good: Use query for reads
getMostRecentPost: t.procedure.query(() => getPost())

❌ 不要跳过输入验证：

// Bad: No validation
.input(z.any())

// Good: Strict validation
.input(z.object({ id: z.string().uuid() }))
监控与可观测性
const t = initTRPC.context<Context>().create({
  errorFormatter({ shape, error }) {
    // Log metrics
    metrics.increment('trpc.error', { code: error.code });

    // Send to error tracking
    if (error.code === 'INTERNAL_SERVER_ERROR') {
      Sentry.captureException(error);
    }

    return shape;
  },
});

const loggingMiddleware = t.middleware(async ({ path, type, next }) => {
  const start = Date.now();
  const result = await next();

  // Log performance metrics
  metrics.timing('trpc.duration', Date.now() - start, { path, type });

  return result;
});

总结

tRPC 以最少的样板代码实现类型安全的 API：

  • ✅ 无需代码生成：从 TypeScript 推断类型
  • ✅ 集成 React Query：内置缓存和乐观更新
  • ✅ 一流的 Next.js 支持：App Router、Server Components
  • ✅ 开发者体验：自动补全、重构、类型错误

最适合：全栈 TypeScript 应用、Next.js 项目、内部工具 不适合：公共 API、多语言服务

快速开始：安装 → 定义路由器 → 在客户端使用 → 享受类型安全！

相关技能：Zod（验证）、React Query（缓存）、Next.js（集成）

评论

支持 Markdown