Merge pull request #1830 from bluesky-social/xrpc-server-blob-limit

xrpc server blob limit
bluesky-social · Nov 7, 2023 · ce1baba · ce1baba
2 parents b37ff3c + 28655dd
commit ce1baba
Show file tree

Hide file tree

Showing 5 changed files with 19 additions and 7 deletions.
diff --git a/packages/pds/src/api/com/atproto/temp/importRepo.ts b/packages/pds/src/api/com/atproto/temp/importRepo.ts
@@ -15,6 +15,9 @@ import { AtprotoData } from '@atproto/identity'
 
 export default function (server: Server, ctx: AppContext) {
   server.com.atproto.temp.importRepo({
+    opts: {
+      blobLimit: 5 * 1024 * 1024, // 5GB
+    },
     handler: async ({ params, input, req }) => {
       const { did } = params
       const outBuffer = new AsyncBuffer<string>()

diff --git a/packages/pds/src/lexicon/index.ts b/packages/pds/src/lexicon/index.ts
@@ -1614,6 +1614,7 @@ type HandlerRateLimitOpts<T> = SharedRateLimitOpts<T> | RouteRateLimitOpts<T>
 type ConfigOf<Auth, Handler, ReqCtx> =
   | Handler
   | {
+      opts?: { blobLimit?: number }
       auth?: Auth
       rateLimit?: HandlerRateLimitOpts<ReqCtx> | HandlerRateLimitOpts<ReqCtx>[]
       handler: Handler

diff --git a/packages/xrpc-server/src/server.ts b/packages/xrpc-server/src/server.ts
@@ -173,7 +173,7 @@ export class Server {
     this.routes[verb](
       `/xrpc/${nsid}`,
       ...middleware,
-      this.createHandler(nsid, def, config.handler),
+      this.createHandler(nsid, def, config),
     )
   }
 
@@ -206,10 +206,13 @@ export class Server {
   createHandler(
     nsid: string,
     def: LexXrpcQuery | LexXrpcProcedure,
-    handler: XRPCHandler,
+    routeCfg: XRPCHandlerConfig,
   ): RequestHandler {
+    const routeOpts = {
+      blobLimit: routeCfg.opts?.blobLimit ?? this.options.payload?.blobLimit,
+    }
     const validateReqInput = (req: express.Request) =>
-      validateInput(nsid, def, req, this.options, this.lex)
+      validateInput(nsid, def, req, routeOpts, this.lex)
     const validateResOutput =
       this.options.validateResponse === false
         ? (output?: HandlerSuccess) => output
@@ -256,7 +259,7 @@ export class Server {
         }
 
         // run the handler
-        const outputUnvalidated = await handler(reqCtx)
+        const outputUnvalidated = await routeCfg.handler(reqCtx)
 
         if (isHandlerError(outputUnvalidated)) {
           throw XRPCError.fromError(outputUnvalidated)

diff --git a/packages/xrpc-server/src/types.ts b/packages/xrpc-server/src/types.ts
@@ -143,7 +143,12 @@ export type RateLimiterStatus = {
   isFirstInDuration: boolean
 }
 
+export type RouteOpts = {
+  blobLimit?: number
+}
+
 export type XRPCHandlerConfig = {
+  opts?: RouteOpts
   rateLimit?: HandlerRateLimitOpts | HandlerRateLimitOpts[]
   auth?: AuthVerifier
   handler: XRPCHandler

diff --git a/packages/xrpc-server/src/util.ts b/packages/xrpc-server/src/util.ts
@@ -19,8 +19,8 @@ import {
   handlerSuccess,
   InvalidRequestError,
   InternalServerError,
-  Options,
   XRPCError,
+  RouteOpts,
 } from './types'
 
 export function decodeQueryParams(
@@ -82,7 +82,7 @@ export function validateInput(
   nsid: string,
   def: LexXrpcProcedure | LexXrpcQuery,
   req: express.Request,
-  opts: Options,
+  opts: RouteOpts,
   lexicons: Lexicons,
 ): HandlerInput | undefined {
   // request expectation
@@ -139,7 +139,7 @@ export function validateInput(
   if (req.readableEnded) {
     body = req.body
   } else {
-    body = decodeBodyStream(req, opts.payload?.blobLimit)
+    body = decodeBodyStream(req, opts.blobLimit)
   }
 
   return {