Is there any way to use Pipes with Grpc in Nestjs?

So I'm building an httpgateway which sends messages to a microservice made with nestjs/grpc.

Problem is that, once I decorate my controller with @UsePipes(....) it throws an error for the gateway. I tried to log data entering into pipe and found out that grpc sends not only payload but also metadata and ServerDuplexStream prior to payload itself. So, my consumer throws an error because it faces with the ServerDuplexStream at first and cannot validate the args inside it.

I further tried to use my pipes in app.service but it doesnt make any sense since pipes receive data from the request. So it doesnt work as expected.

Is there a workaround like putting all three in a call in my gateway prior to sending request?

You can see an example of a pipe that im trying to implement:

@Injectable()
export class ValidateSingleBalanceByUser implements PipeTransform {
  transform(value: SingleBalanceDto) {
    if (!value.user) throw new RpcException('Provide user value to query!');
    if (!value.asset) throw new RpcException('Provide asset value to query!');
    return value;
  }
}

and an example of a controller that im trying to implement to

  @UsePipes(new ValidateSingleBalanceByUser())
  @GrpcMethod('BridgeService', 'getSingleBalanceByUser')
  singleBalanceByUser(data: SingleBalanceDto): Promise<Balance> {
    return this.balancesService.handleSingleBalanceByUser(data);
  }

I was getting the same problem, cause the Validation its a pipe and does not treat grpc exceptions. I fixed it using this solution:

controller.ts

@GrpcMethod('service','action')
@UsePipes(new ValidationPipe({ transform: true }))  
@UseFilters(new TranslateHttpToGrpcExceptionFilter())
method(){}

translate.ts

@Catch(HttpException)
export class TranslateHttpToGrpcExceptionFilter implements ExceptionFilter {
static HttpStatusCode: Record<number, number> = {
[HttpStatus.BAD_REQUEST]: status.INVALID_ARGUMENT,
[HttpStatus.UNAUTHORIZED]: status.UNAUTHENTICATED,
[HttpStatus.FORBIDDEN]: status.PERMISSION_DENIED,
[HttpStatus.NOT_FOUND]: status.NOT_FOUND,
[HttpStatus.CONFLICT]: status.ALREADY_EXISTS,
[HttpStatus.GONE]: status.ABORTED,
[HttpStatus.TOO_MANY_REQUESTS]: status.RESOURCE_EXHAUSTED,
499: status.CANCELLED,
[HttpStatus.INTERNAL_SERVER_ERROR]: status.INTERNAL,
[HttpStatus.NOT_IMPLEMENTED]: status.UNIMPLEMENTED,
[HttpStatus.BAD_GATEWAY]: status.UNKNOWN,
[HttpStatus.SERVICE_UNAVAILABLE]: status.UNAVAILABLE,
[HttpStatus.GATEWAY_TIMEOUT]: status.DEADLINE_EXCEEDED,
[HttpStatus.HTTP_VERSION_NOT_SUPPORTED]: status.UNAVAILABLE,
[HttpStatus.PAYLOAD_TOO_LARGE]: status.OUT_OF_RANGE,
[HttpStatus.UNSUPPORTED_MEDIA_TYPE]: status.CANCELLED,
[HttpStatus.UNPROCESSABLE_ENTITY]: status.CANCELLED,
[HttpStatus.I_AM_A_TEAPOT]: status.UNKNOWN,
[HttpStatus.METHOD_NOT_ALLOWED]: status.CANCELLED,
[HttpStatus.PRECONDITION_FAILED]: status.FAILED_PRECONDITION
}

catch(exception: HttpException): Observable<never> | void {
const httpStatus = exception.getStatus()
const httpRes = exception.getResponse() as { details?: unknown, message: unknown }

return throwError(() => ({
  code: TranslateHttpToGrpcExceptionFilter.HttpStatusCode[httpStatus] ?? status.UNKNOWN,
  message: httpRes.message || exception.message,
  details: Array.isArray(httpRes.details) ? httpRes.details : httpRes.message
}))

  }
}

Hope it helps!