在NestJS与TypeORM中为PostgreSQL安全配置运行时变量以适配Supabase行级安全

我来给你几个能覆盖所有查询（包括SELECT）的通用方案，完美解决你在NestJS里结合Supabase行级安全设置运行时变量的问题：

方案一：NestJS拦截器 + 请求级QueryRunner（推荐） #

这个方案能在每个HTTP请求的生命周期内自动创建数据库会话、绑定用户变量，所有查询都复用这个会话，请求结束后自动清理，完全不用在Service里重复写模板代码。

步骤1：创建请求上下文存储Service #

用来保存当前请求对应的数据库会话：

// src/context/request-context.service.ts
import { Injectable, Scope, Inject } from '@nestjs/common';
import { REQUEST } from '@nestjs/core';
import { Request } from 'express';
import { QueryRunner } from 'typeorm';

@Injectable({ scope: Scope.REQUEST })
export class RequestContextService {
  queryRunner?: QueryRunner;

  constructor(@Inject(REQUEST) private readonly request: Request) {}
}

步骤2：实现数据库上下文拦截器 #

负责初始化会话、设置用户变量、请求结束后清理：

// src/interceptors/db-context.interceptor.ts
import { Injectable, NestInterceptor, ExecutionContext, CallHandler } from '@nestjs/common';
import { Observable, finalize } from 'rxjs';
import { DataSource } from 'typeorm';
import { RequestContextService } from '../context/request-context.service';
import { ContextService } from '../context/context.service'; // 你的用户上下文Service

@Injectable()
export class DbContextInterceptor implements NestInterceptor {
  constructor(
    private dataSource: DataSource,
    private requestContext: RequestContextService,
    private contextService: ContextService,
  ) {}

  async intercept(context: ExecutionContext, next: CallHandler): Promise<Observable<any>> {
    const userId = this.contextService.getRequest(); // 获取当前用户的sub
    if (!userId) {
      // 无用户的公共接口直接放行
      return next.handle();
    }

    // 创建并连接QueryRunner
    const queryRunner = this.dataSource.createQueryRunner();
    await queryRunner.connect();

    // 用参数化查询设置变量（避免SQL注入！）
    await queryRunner.query(`SELECT set_config('request.jwt.claim.sub', $1, true)`, [userId]);
    
    // 将QueryRunner存入请求上下文
    this.requestContext.queryRunner = queryRunner;

    // 请求结束后自动清理会话
    return next.handle().pipe(
      finalize(async () => {
        try {
          await queryRunner.query(`RESET request.jwt.claim.sub`);
          await queryRunner.release();
        } catch (err) {
          console.error('清理数据库会话失败:', err);
        }
      }),
    );
  }
}

步骤3：创建自定义EntityManager Provider #

让所有Service自动复用请求级的数据库会话：

// src/providers/custom-entity-manager.provider.ts
import { Provider } from '@nestjs/common';
import { EntityManager } from 'typeorm';
import { RequestContextService } from '../context/request-context.service';

export const CUSTOM_ENTITY_MANAGER = 'CUSTOM_ENTITY_MANAGER';

export const CustomEntityManagerProvider: Provider = {
  provide: CUSTOM_ENTITY_MANAGER,
  useFactory: (requestContext: RequestContextService): EntityManager => {
    // 优先使用请求级QueryRunner的EntityManager， fallback到默认实例
    return requestContext.queryRunner?.manager ?? new EntityManager(requestContext.queryRunner?.connection ?? null);
  },
  inject: [RequestContextService],
  scope: Scope.REQUEST,
};

步骤4：注册拦截器和Provider #

在根Module里全局注册：

// src/app.module.ts
import { Module } from '@nestjs/common';
import { APP_INTERCEPTOR } from '@nestjs/core';
import { DbContextInterceptor } from './interceptors/db-context.interceptor';
import { RequestContextService } from './context/request-context.service';
import { CustomEntityManagerProvider, CUSTOM_ENTITY_MANAGER } from './providers/custom-entity-manager.provider';

@Module({
  providers: [
    RequestContextService,
    CustomEntityManagerProvider,
    {
      provide: APP_INTERCEPTOR,
      useClass: DbContextInterceptor,
    },
  ],
  exports: [CUSTOM_ENTITY_MANAGER],
})
export class AppModule {}

步骤5：在Service中使用自定义EntityManager #

所有数据库操作都会自动带上用户变量：

// src/users/users.service.ts
import { Injectable, Inject } from '@nestjs/common';
import { Users } from '../common/entities';
import { EntityManager } from 'typeorm';
import { CUSTOM_ENTITY_MANAGER } from '../providers/custom-entity-manager.provider';

@Injectable()
export class UsersService {
  private userRepository;

  constructor(
    @Inject(CUSTOM_ENTITY_MANAGER) private readonly entityManager: EntityManager,
  ) {
    this.userRepository = this.entityManager.getRepository(Users);
  }

  async getAllUsers() {
    // 直接调用Repository方法，自动使用带用户变量的会话
    return this.userRepository.find();
  }
}

这个方案的优势：

• 覆盖所有数据库操作（SELECT/INSERT/UPDATE/DELETE）
• 完全无需在Service中手动处理会话逻辑
• 用参数化查询避免SQL注入，安全性拉满

方案二：扩展TypeORM Repository（轻量替代） #

如果不想用拦截器，可以自定义BaseRepository，在所有查询前自动设置变量：

// src/common/base.repository.ts
import { Repository, EntityManager } from 'typeorm';
import { ContextService } from '../context/context.service';

export class BaseRepository<T> extends Repository<T> {
  constructor(
    target: any,
    entityManager: EntityManager,
    private contextService: ContextService,
  ) {
    super(target, entityManager);
  }

  private async setUserVariable() {
    const userId = this.contextService.getRequest();
    if (userId) {
      await this.manager.query(`SELECT set_config('request.jwt.claim.sub', $1, true)`, [userId]);
    }
  }

  // 重写需要的查询方法
  async find(options?: any) {
    await this.setUserVariable();
    return super.find(options);
  }

  async findOne(options?: any) {
    await this.setUserVariable();
    return super.findOne(options);
  }

  async save(entity: T) {
    await this.setUserVariable();
    return super.save(entity);
  }

  // 按需重写update/delete等方法
}

然后在Module中注册自定义Repository：

// src/users/users.module.ts
import { Module } from '@nestjs/common';
import { TypeOrmModule } from '@nestjs/typeorm';
import { Users } from '../common/entities';
import { UsersService } from './users.service';
import { BaseRepository } from '../common/base.repository';
import { ContextService } from '../context/context.service';

@Module({
  imports: [
    TypeOrmModule.forFeature([
      {
        name: Users.name,
        repository: (entityManager) => new BaseRepository(Users, entityManager, new ContextService()),
      },
    ]),
  ],
  providers: [UsersService],
})
export class UsersModule {}

这个方案适合不想引入拦截器的场景，但需要手动重写Repository的所有查询方法，扩展性稍弱。

关于你之前QueryRunner方案失效的排查点 #

你之前的代码没生效，大概率是这几个原因：

1. SQL注入风险：直接拼接'${userId}'可能导致变量设置失败，一定要用参数化查询
2. RLS策略未正确配置：检查是否开启了表的RLS（ALTER TABLE users ENABLE ROW LEVEL SECURITY;），以及策略是否正确关联request.jwt.claim.sub
3. 会话不一致：确保设置变量和执行查询用的是同一个QueryRunner实例

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