feat(blob): Provide onUploadProgress({ loaded, total, percentage })

packages/blob/jest/setup.js

@@ -2,5 +2,11 @@
 // but they are available everywhere else.
 // See https://stackoverflow.com/questions/68468203/why-am-i-getting-textencoder-is-not-defined-in-jest
 const { TextEncoder, TextDecoder } = require('node:util');
+// eslint-disable-next-line import/order -- On purpose to make requiring undici work
+const { ReadableStream } = require('node:stream/web');
 
-Object.assign(global, { TextDecoder, TextEncoder });
+Object.assign(global, { TextDecoder, TextEncoder, ReadableStream });
+
+const { Request } = require('undici');
+
+Object.assign(global, { Request });

packages/blob/package.json

@@ -63,7 +63,8 @@
     "async-retry": "^1.3.3",
     "bytes": "^3.1.2",
     "is-buffer": "^2.0.5",
-    "undici": "^5.28.4"
+    "throttleit": "^2.1.0",
+    "undici": "^6.20.1"
   },
   "devDependencies": {
     "@edge-runtime/jest-environment": "2.3.10",

packages/blob/src/api.ts

@@ -1,9 +1,19 @@
-import type { RequestInit, Response } from 'undici';
+import type { BodyInit, RequestInit, Response } from 'undici';
 import { fetch } from 'undici';
 import retry from 'async-retry';
 import { debug } from './debug';
-import type { BlobCommandOptions } from './helpers';
-import { BlobError, getTokenFromOptionsOrEnv } from './helpers';
+import type { BlobCommandOptions, WithUploadProgress } from './helpers';
+import {
+  BlobError,
+  computeBodyLength,
+  createChunkTransformStream,
+  getApiUrl,
+  getTokenFromOptionsOrEnv,
+  isStream,
+  supportsRequestStreams,
+} from './helpers';
+import { toReadableStream } from './multipart/helpers';
+import type { PutBody } from './put-helpers';
 
 // maximum pathname length is:
 // 1024 (provider limit) - 26 chars (vercel  internal suffixes) - 31 chars (blob `-randomId` suffix) = 967
@@ -132,20 +142,6 @@ function getApiVersion(): string {
   return `${versionOverride ?? BLOB_API_VERSION}`;
 }
 
-function getApiUrl(pathname = ''): string {
-  let baseUrl = null;
-  try {
-    // wrapping this code in a try/catch as this function is used in the browser and Vite doesn't define the process.env.
-    // As this varaible is NOT used in production, it will always default to production endpoint
-    baseUrl =
-      process.env.VERCEL_BLOB_API_URL ||
-      process.env.NEXT_PUBLIC_VERCEL_BLOB_API_URL;
-  } catch {
-    // noop
-  }
-  return `${baseUrl || 'https://blob.vercel-storage.com'}${pathname}`;
-}
-
 function getRetries(): number {
   try {
     const retries = process.env.VERCEL_BLOB_RETRIES || '10';
@@ -175,7 +171,6 @@ async function getBlobError(
 
   try {
     const data = (await response.json()) as BlobApiError;
-
     code = data.error?.code ?? 'unknown_error';
     message = data.error?.message;
   } catch {
@@ -252,10 +247,12 @@ async function getBlobError(
   return { code, error };
 }
 
+const CHUNK_SIZE = 64 * 1024;
+
 export async function requestApi<TResponse>(
   pathname: string,
   init: RequestInit,
-  commandOptions: BlobCommandOptions | undefined,
+  commandOptions: (BlobCommandOptions & WithUploadProgress) | undefined,
 ): Promise<TResponse> {
   const apiVersion = getApiVersion();
   const token = getTokenFromOptionsOrEnv(commandOptions);
@@ -268,20 +265,104 @@ export async function requestApi<TResponse>(
   const apiResponse = await retry(
     async (bail) => {
       let res: Response;
+      let bodyLength: number | undefined;
+      let body: BodyInit | undefined;
+
+      if (
+        init.body &&
+        // 1. For upload progress we always need to know the total size of the body
+        // 2. In development we need the header for put() to work correctly when passing a stream
+        (commandOptions?.onUploadProgress || shouldUseXContentLength())
+      ) {
+        bodyLength = computeBodyLength(init.body);
+      }
+
+      if (init.body) {
+        if (commandOptions?.onUploadProgress) {
+          if (supportsRequestStreams) {
+            // We transform the body to a stream here instead of at the call site
+            // So that on retries we can reuse the original body, otherwise we would not be able to reuse it
+            const stream = await toReadableStream(init.body as PutBody);
+
+            let loaded = 0;
+
+            const chunkTransformStream = createChunkTransformStream(
+              CHUNK_SIZE,
+              (newLoaded: number) => {
+                loaded += newLoaded;
+                const total = bodyLength ?? loaded;
+                const percentage = Number(((loaded / total) * 100).toFixed(2));
+
+                // Leave percentage 100 to end of request
+                if (percentage === 100) {
+                  return;
+                }
+
+                commandOptions.onUploadProgress?.({
+                  loaded,
+                  // When passing a stream to put(), we have no way to know the total size of the body.
+                  // Instead of defining total as total?: number we decided to set the total to the currently
+                  // loaded number. This is not inaccurate and way more practical for DX.
+                  // Passing down a stream to put() is very rare
+                  total,
+                  percentage,
+                });
+              },
+            );
+
+            body = stream.pipeThrough(chunkTransformStream);
+          } else {
+            body = init.body;
+          }
+        } else {
+          body = init.body;
+        }
+      }
+
+      // Only set duplex option when supported and dealing with a stream body
+      const duplex =
+        supportsRequestStreams && body && isStream(body as PutBody)
+          ? 'half'
+          : undefined;
+
+      if (commandOptions?.onUploadProgress) {
+        commandOptions.onUploadProgress({
+          loaded: 0,
+          total: bodyLength ?? 0,
+          percentage: 0,
+        });
+      }
 
       // try/catch here to treat certain errors as not-retryable
       try {
         res = await fetch(getApiUrl(pathname), {
           ...init,
+          ...(init.body ? { body } : {}),
+          ...(duplex ? { duplex } : {}),
           headers: {
             'x-api-blob-request-id': requestId,
             'x-api-blob-request-attempt': String(retryCount),
             'x-api-version': apiVersion,
+            ...(bodyLength ? { 'x-content-length': String(bodyLength) } : {}),
             authorization: `Bearer ${token}`,
             ...extraHeaders,
             ...init.headers,
           },
         });
+
+        // Calling onUploadProgress here has two benefits:
+        // 1. It ensures 100% is only reached at the end of the request. While otherwise you can reach 100%
+        // before the request is fully done, as we only really measure what gets sent over the wire, not what
+        // has been processed by the server.
+        // 2. It makes the uploadProgress "work" even for browsers not supporting request streams like Safari.
+        // And in the case of multipart uploads it actually provides a simple progress indication (per part)
+        if (commandOptions?.onUploadProgress) {
+          commandOptions.onUploadProgress({
+            loaded: bodyLength ?? 0,
+            total: bodyLength ?? 0,
+            percentage: 100,
+          });
+        }
       } catch (error) {
         // if the request was aborted, don't retry
         if (error instanceof DOMException && error.name === 'AbortError') {
@@ -350,3 +431,11 @@ function getProxyThroughAlternativeApiHeaderFromEnv(): {
 
   return extraHeaders;
 }
+
+function shouldUseXContentLength(): boolean {
+  try {
+    return process.env.VERCEL_BLOB_USE_X_CONTENT_LENGTH === '1';
+  } catch {
+    return false;
+  }
+}

packages/blob/src/client.browser.test.ts

@@ -88,7 +88,6 @@ describe('client', () => {
         'https://blob.vercel-storage.com/foo.txt',
         {
           body: 'Test file data',
-          duplex: 'half',
           headers: {
             authorization: 'Bearer vercel_blob_client_fake_123',
             'x-api-blob-request-attempt': '0',
@@ -232,7 +231,6 @@ describe('client', () => {
             'x-mpu-part-number': '1',
           },
           method: 'POST',
-          duplex: 'half',
           signal: internalAbortSignal,
         },
       );
@@ -252,7 +250,6 @@ describe('client', () => {
             'x-mpu-part-number': '2',
           },
           method: 'POST',
-          duplex: 'half',
           signal: internalAbortSignal,
         },
       );
@@ -376,7 +373,6 @@ describe('client', () => {
             'x-mpu-part-number': '1',
           },
           method: 'POST',
-          duplex: 'half',
           signal: internalAbortSignal,
         },
       );
@@ -396,7 +392,6 @@ describe('client', () => {
             'x-mpu-part-number': '2',
           },
           method: 'POST',
-          duplex: 'half',
           signal: internalAbortSignal,
         },
       );

packages/blob/src/client.ts

@@ -5,7 +5,7 @@ import type { IncomingMessage } from 'node:http';
 // the `undici` module will be replaced with https://developer.mozilla.org/en-US/docs/Web/API/Fetch_API
 // for browser contexts. See ./undici-browser.js and ./package.json
 import { fetch } from 'undici';
-import type { BlobCommandOptions } from './helpers';
+import type { BlobCommandOptions, WithUploadProgress } from './helpers';
 import { BlobError, getTokenFromOptionsOrEnv } from './helpers';
 import { createPutMethod } from './put';
 import type { PutBlobResult } from './put-helpers';
@@ -42,7 +42,9 @@ export interface ClientTokenOptions {
 }
 
 // shared interface for put and upload
-interface ClientCommonPutOptions extends ClientCommonCreateBlobOptions {
+interface ClientCommonPutOptions
+  extends ClientCommonCreateBlobOptions,
+    WithUploadProgress {
   /**
    * Whether to use multipart upload. Use this when uploading large files. It will split the file into multiple parts, upload them in parallel and retry failed parts.
    */
@@ -89,7 +91,7 @@ export const put = createPutMethod<ClientPutCommandOptions>({
 // vercelBlob. createMultipartUpload()
 // vercelBlob. uploadPart()
 // vercelBlob. completeMultipartUpload()
-// vercelBlob. createMultipartUploaded()
+// vercelBlob. createMultipartUploader()
 
 export type ClientCreateMultipartUploadCommandOptions =
   ClientCommonCreateBlobOptions & ClientTokenOptions;
@@ -110,7 +112,8 @@ export const createMultipartUploader =
 
 type ClientMultipartUploadCommandOptions = ClientCommonCreateBlobOptions &
   ClientTokenOptions &
-  CommonMultipartUploadOptions;
+  CommonMultipartUploadOptions &
+  WithUploadProgress;
 
 export const uploadPart =
   createUploadPartMethod<ClientMultipartUploadCommandOptions>({