T3应用中基于TRPC与Prisma实现组级数据权限控制的方案咨询

针对你的需求，目前有几种成熟的模式可以实现TRPC路由对用户组权限的双重验证（登录+记录访问权限），完全不需要为每个用户组创建独立数据库，以下是具体实现方案：

1. TRPC中间件封装权限校验 #

这是最常用的模式，通过自定义中间件结合protectedProcedure，封装出带组权限验证的专用Procedure，所有需要组权限的路由直接复用即可。

步骤1：定义用户组关联的Prisma Schema #

确保你的数据模型包含用户、组、记录之间的关联（以Post记录为例）：

model User {
  id        String   @id @default(cuid())
  clerkId   String   @unique
  groups    UserGroup[]
}

model Group {
  id        String   @id @default(cuid())
  name      String
  users     UserGroup[]
  posts     Post[]
}

model UserGroup {
  userId    String
  groupId   String
  user      User    @relation(fields: [userId], references: [id], onDelete: Cascade)
  group     Group   @relation(fields: [groupId], references: [id], onDelete: Cascade)

  @@id([userId, groupId])
}

model Post {
  id        String   @id @default(cuid())
  content   String
  groupId   String
  group     Group    @relation(fields: [groupId], references: [id])
}

步骤2：编写TRPC权限中间件 #

创建带组权限验证的中间件，并组合出专用Procedure：

import { TRPCError, initTRPC } from "@trpc/server";
import { type Context } from "./context";

const t = initTRPC.context<Context>().create();

// 基础登录验证中间件（等同于protectedProcedure）
const isAuthed = t.middleware(({ ctx, next }) => {
  if (!ctx.user) {
    throw new TRPCError({ code: "UNAUTHORIZED" });
  }
  return next({
    ctx: { user: ctx.user }, // 收窄用户类型，确保后续可安全访问
  });
});

// 组权限验证中间件：检查当前用户是否属于记录所属的组
const hasGroupAccess = t.middleware(async ({ ctx, next, rawInput }) => {
  // 从输入中获取记录ID（需确保所有带权限的路由输入都包含recordId字段）
  const { recordId } = rawInput as { recordId: string };

  // 查询记录所属的组ID
  const record = await ctx.prisma.post.findUnique({
    where: { id: recordId },
    select: { groupId: true },
  });

  if (!record) throw new TRPCError({ code: "NOT_FOUND" });

  // 验证用户是否为该组成员
  const isGroupMember = await ctx.prisma.userGroup.findFirst({
    where: { userId: ctx.user.id, groupId: record.groupId },
  });

  if (!isGroupMember) throw new TRPCError({ code: "FORBIDDEN" });

  // 将groupId传递给后续路由逻辑，方便复用
  return next({ ctx: { groupId: record.groupId } });
});

// 组合出带登录+组权限验证的Procedure
export const groupProtectedProcedure = t.procedure.use(isAuthed).use(hasGroupAccess);

步骤3：在路由中使用专用Procedure #

import { z } from "zod";

export const postRouter = t.router({
  // 获取单个Post（需组权限）
  getById: groupProtectedProcedure
    .input(z.object({ recordId: z.string() }))
    .query(async ({ ctx, input }) => {
      return ctx.prisma.post.findUnique({ where: { id: input.recordId } });
    }),

  // 更新Post（需组权限）
  update: groupProtectedProcedure
    .input(z.object({ recordId: z.string(), content: z.string() }))
    .mutation(async ({ ctx, input }) => {
      return ctx.prisma.post.update({
        where: { id: input.recordId },
        data: { content: input.content },
      });
    }),
});

2. 数据库层面的行级安全（RLS）兜底 #

为了防止绕过TRPC直接访问数据库的风险，可以结合PostgreSQL的行级安全（RLS）功能，在数据库层面实现权限过滤，作为应用层校验的兜底。

步骤1：在Prisma Schema中启用RLS #

model Post {
  id        String   @id @default(cuid())
  content   String
  groupId   String
  group     Group    @relation(fields: [groupId], references: [id])

  @@enableRowLevelSecurity // 启用行级安全
}

步骤2：创建RLS策略（SQL） #

执行以下SQL语句创建权限策略，限制用户只能访问所属组的记录：

-- 允许用户查询所属组的Post
CREATE POLICY "Users can view their group's posts" ON "Post"
FOR SELECT USING (
  EXISTS (
    SELECT 1 FROM "UserGroup"
    WHERE "UserGroup"."userId" = current_setting('app.current_user_id')::text
    AND "UserGroup"."groupId" = "Post"."groupId"
  )
);

-- 允许用户更新所属组的Post
CREATE POLICY "Users can update their group's posts" ON "Post"
FOR UPDATE USING (
  EXISTS (
    SELECT 1 FROM "UserGroup"
    WHERE "UserGroup"."userId" = current_setting('app.current_user_id')::text
    AND "UserGroup"."groupId" = "Post"."groupId"
  )
);

-- 允许用户删除所属组的Post
CREATE POLICY "Users can delete their group's posts" ON "Post"
FOR DELETE USING (
  EXISTS (
    SELECT 1 FROM "UserGroup"
    WHERE "UserGroup"."userId" = current_setting('app.current_user_id')::text
    AND "UserGroup"."groupId" = "Post"."groupId"
  )
);

步骤3：在TRPC上下文传递用户ID到数据库 #

每次请求时，将当前登录用户的ID设置到数据库会话变量中，供RLS策略使用：

// context.ts
import { PrismaClient } from "@prisma/client";
import { type AuthObject } from "@clerk/nextjs/api";

const prisma = new PrismaClient();

export async function createContext({ auth }: { auth: AuthObject }) {
  const user = auth.user;

  if (user) {
    // 设置数据库会话变量，用于RLS策略校验
    await prisma.$executeRaw`SET app.current_user_id = ${user.id}`;
  }

  return { prisma, user };
}

export type Context = Awaited<ReturnType<typeof createContext>>;

3. 工具函数封装校验逻辑（灵活场景） #

如果你的权限逻辑比较复杂（比如不同记录类型有不同权限规则），可以封装一个工具函数，在每个路由中手动调用：

// utils/permission.ts
import { type Context } from "../context";
import { TRPCError } from "@trpc/server";

// 通用组权限校验函数
export async function validateGroupAccess(ctx: Context, recordId: string) {
  if (!ctx.user) throw new TRPCError({ code: "UNAUTHORIZED" });

  const record = await ctx.prisma.post.findUnique({
    where: { id: recordId },
    select: { groupId: true },
  });

  if (!record) throw new TRPCError({ code: "NOT_FOUND" });

  const isMember = await ctx.prisma.userGroup.findFirst({
    where: { userId: ctx.user.id, groupId: record.groupId },
  });

  if (!isMember) throw new TRPCError({ code: "FORBIDDEN" });
}

在路由中使用：

export const postRouter = t.router({
  getById: protectedProcedure
    .input(z.object({ recordId: z.string() }))
    .query(async ({ ctx, input }) => {
      await validateGroupAccess(ctx, input.recordId);
      return ctx.prisma.post.findUnique({ where: { id: input.recordId } });
    }),
});

方案选择建议 #

• 推荐TRPC中间件 + RLS组合使用：应用层中间件提供友好的错误提示，数据库RLS提供最终的安全兜底，双重保障。
• 若权限逻辑简单，直接用TRPC中间件即可，代码更简洁。
• 复杂权限场景可采用工具函数，灵活适配不同业务规则。

内容的提问来源于stack exchange，提问作者empire29