feat: migrate API client and codebase to native fetch

Author: joehan

npm-shrinkwrap.json

@@ -170,6 +170,7 @@
     "triple-beam": "^1.3.0",
     "universal-analytics": "^0.5.3",
     "update-notifier-cjs": "^5.1.6",
+    "undici": "^6.19.0",
     "winston": "^3.0.0",
     "winston-transport": "^4.4.0",
     "ws": "^7.5.10",
@@ -270,9 +271,6 @@
       "ajv-formats": "3.0.1",
       "ajv": "^8.17.1"
     },
-    "node-fetch": {
-      "whatwg-url": "^14.0.0"
-    },
     "hono": "^4.11.4",
     "@tootallnate/once": "^3.0.1",
     "protobufjs": "^7.4.0",

package.json

@@ -1,10 +1,7 @@
-import { AbortSignal } from "abort-controller";
 import { URL, URLSearchParams } from "url";
 import { Readable } from "stream";
-import { ProxyAgent } from "proxy-agent";
+import { ProxyAgent, request } from "undici";
 import * as retry from "retry";
-import AbortController from "abort-controller";
-import fetch, { HeadersInit, Response, RequestInit, Headers } from "node-fetch";
 import * as http from "http";
 import * as https from "https";
 import util from "util";
@@ -38,7 +35,6 @@ const GOOG_QUOTA_USER_HEADER = "x-goog-quota-user";
 
 // Header for specifying a quota project. See https://cloud.google.com/apis/docs/system-parameters#project-header
 export const GOOG_USER_PROJECT_HEADER = "x-goog-user-project";
-const GOOGLE_CLOUD_QUOTA_PROJECT = process.env.GOOGLE_CLOUD_QUOTA_PROJECT;
 export const CLI_OAUTH_PROJECT_NUMBER = "563584335869";
 
 export type HttpMethod =
@@ -150,13 +146,16 @@ export async function getAccessToken(): Promise<string> {
 }
 
 function proxyURIFromEnv(): string | undefined {
-  return (
+  const uri =
     process.env.HTTPS_PROXY ||
     process.env.https_proxy ||
     process.env.HTTP_PROXY ||
     process.env.http_proxy ||
-    undefined
-  );
+    undefined;
+  if (uri === "undefined" || uri === "null") {
+    return undefined;
+  }
+  return uri;
 }
 
 // Some networks (and recent Node.js security releases) interact badly with
@@ -198,6 +197,12 @@ export type ClientOptions = {
   auth?: boolean;
 };
 
+interface FetchOptions extends RequestInit {
+  dispatcher?: ProxyAgent;
+  duplex?: "half";
+  agent?: any;
+}
+
 export class Client {
   constructor(private opts: ClientOptions) {
     if (this.opts.auth === undefined) {
@@ -357,10 +362,10 @@ export class Client {
     }
     if (
       !reqOptions.ignoreQuotaProject &&
-      GOOGLE_CLOUD_QUOTA_PROJECT &&
-      GOOGLE_CLOUD_QUOTA_PROJECT !== ""
+      process.env.GOOGLE_CLOUD_QUOTA_PROJECT &&
+      process.env.GOOGLE_CLOUD_QUOTA_PROJECT !== ""
     ) {
-      reqOptions.headers.set(GOOG_USER_PROJECT_HEADER, GOOGLE_CLOUD_QUOTA_PROJECT);
+      reqOptions.headers.set(GOOG_USER_PROJECT_HEADER, process.env.GOOGLE_CLOUD_QUOTA_PROJECT);
     }
     return reqOptions;
   }
@@ -409,24 +414,19 @@ export class Client {
       }
     }
 
-    const fetchOptions: RequestInit = {
+    const fetchOptions: FetchOptions = {
       headers: options.headers,
       method: options.method,
       redirect: options.redirect,
-      compress: options.compress,
     };
 
-    if (proxyURIFromEnv()) {
-      fetchOptions.agent = new ProxyAgent();
+    const proxyURI = proxyURIFromEnv();
+    if (proxyURI) {
+      fetchOptions.dispatcher = new ProxyAgent({ uri: proxyURI });
     }
 
     if (options.signal) {
-      const signal = options.signal as any;
-      signal.reason = "";
-      signal.throwIfAborted = () => {
-        throw new FirebaseError("Aborted");
-      };
-      fetchOptions.signal = signal;
+      fetchOptions.signal = options.signal;
     }
 
     let reqTimeout: NodeJS.Timeout | undefined;
@@ -435,12 +435,7 @@ export class Client {
       reqTimeout = setTimeout(() => {
         controller.abort();
       }, options.timeout);
-      const signal = controller.signal as any;
-      signal.reason = "";
-      signal.throwIfAborted = () => {
-        throw new FirebaseError("Aborted");
-      };
-      fetchOptions.signal = signal;
+      fetchOptions.signal = controller.signal;
     }
 
     // A request can only be safely retried if its body can be sent again.
@@ -449,11 +444,12 @@ export class Client {
     // Raw streams cannot be replayed and therefore disable the keep-alive retry.
     let bodyReplayable = true;
     if (typeof options.body === "string" || Buffer.isBuffer(options.body)) {
-      fetchOptions.body = options.body;
+      fetchOptions.body = options.body as any;
     } else if (options.body instanceof FormData) {
-      fetchOptions.body = options.body.getBuffer();
+      fetchOptions.body = options.body.getBuffer() as any;
     } else if (isStream(options.body)) {
-      fetchOptions.body = options.body;
+      fetchOptions.body = options.body as any;
+      fetchOptions.duplex = "half";
       bodyReplayable = false;
     } else if (options.body !== undefined) {
       fetchOptions.body = JSON.stringify(options.body);
@@ -493,9 +489,38 @@ export class Client {
           }
           this.logRequest(options);
           try {
-            res = await fetch(fetchURL, fetchOptions);
+            if (options.compress === false) {
+              const undiciOptions: any = {
+                method: fetchOptions.method,
+                headers: {},
+                decompress: false,
+              };
+              if (fetchOptions.dispatcher) {
+                undiciOptions.dispatcher = fetchOptions.dispatcher;
+              }
+              if (fetchOptions.signal) {
+                undiciOptions.signal = fetchOptions.signal;
+              }
+              if (fetchOptions.body) {
+                undiciOptions.body = fetchOptions.body;
+              }
+              if (fetchOptions.headers) {
+                for (const [key, value] of (fetchOptions.headers as any).entries()) {
+                  undiciOptions.headers[key] = value;
+                }
+              }
+              const undiciRes = await request(fetchURL, undiciOptions);
+              res = new UndiciResponseCompat(undiciRes) as any;
+            } else {
+              res = await fetch(fetchURL, fetchOptions);
+            }
           } catch (thrown: any) {
-            const err = thrown instanceof Error ? thrown : new Error(thrown);
+            const err =
+              thrown && typeof thrown === "object" && thrown.cause instanceof Error
+                ? thrown.cause
+                : thrown instanceof Error
+                  ? thrown
+                  : new Error(thrown);
             logger.debug(
               `*** [apiv2] error from fetch(${fetchURL}, ${JSON.stringify(fetchOptions)}): ${err}`,
             );
@@ -532,7 +557,17 @@ export class Client {
           } else if (options.responseType === "xml") {
             body = (await res.text()) as unknown as ResT;
           } else if (options.responseType === "stream") {
-            body = res.body as unknown as ResT;
+            if (res.body) {
+              if (typeof (res.body as any).getReader === "function") {
+                body = Readable.fromWeb(res.body as any) as unknown as ResT;
+              } else {
+                body = res.body as unknown as ResT;
+              }
+            } else {
+              const emptyStream = new Readable();
+              emptyStream.push(null);
+              body = emptyStream as unknown as ResT;
+            }
           } else {
             throw new FirebaseError(`Unable to interpret response. Please set responseType.`, {
               exit: 2,
@@ -650,6 +685,34 @@ function isLocalInsecureRequest(urlPrefix: string): boolean {
   return u.protocol === "http:";
 }
 
+class UndiciResponseCompat {
+  readonly status: number;
+  readonly headers: Headers;
+  readonly body: any;
+  readonly ok: boolean;
+
+  constructor(private undiciRes: any) {
+    this.status = undiciRes.statusCode;
+    this.ok = undiciRes.statusCode >= 200 && undiciRes.statusCode < 300;
+    this.headers = new Headers();
+    for (const [key, value] of Object.entries(undiciRes.headers)) {
+      if (Array.isArray(value)) {
+        for (const v of value) {
+          this.headers.append(key, v);
+        }
+      } else if (value !== undefined) {
+        this.headers.set(key, String(value));
+      }
+    }
+    this.body = undiciRes.body;
+  }
+
+  async text(): Promise<string> {
+    const { streamToString } = require("./utils");
+    return streamToString(this.body);
+  }
+}
+
 function bodyToString(body: unknown): string {
   if (isStream(body)) {
     // Don't attempt to read any stream type, in case the caller needs it.

src/apiv2.ts

@@ -25,7 +25,6 @@ import { DeepOmit } from "../metaprogramming";
 import { webApps } from "./app";
 import { GitRepositoryLink } from "../gcp/devConnect";
 import * as ora from "ora";
-import fetch from "node-fetch";
 import { orchestrateRollout } from "./rollout";
 import * as fuzzy from "fuzzy";
 import { isEnabled } from "../experiments";

src/apphosting/backend.ts

@@ -6,7 +6,6 @@ import * as StreamObject from "stream-json/streamers/StreamObject";
 
 import { URL } from "url";
 import { Client, ClientResponse } from "../apiv2";
-import { FetchError } from "node-fetch";
 import { FirebaseError } from "../error";
 import { pLimit, Limit } from "../utils";
 
@@ -240,8 +239,9 @@ export default class DatabaseImporter {
       } catch (err: any) {
         const isTimeoutErr =
           err instanceof FirebaseError &&
-          err.original instanceof FetchError &&
-          err.original.code === "ETIMEDOUT";
+          (err.original?.name === "AbortError" ||
+            (err.original as any)?.code === "ETIMEDOUT" ||
+            (err.original as any)?.cause?.code === "ETIMEDOUT");
         if (isTimeoutErr) {
           // RTDB connection timeouts are transient and can be retried
           await new Promise((res) => setTimeout(res, this.nonFatalRetryTimeout));

src/database/import.ts

@@ -2,9 +2,7 @@
  * Webhook send API used to notify VSCode of states within
  */
 
-import fetch from "node-fetch";
 import { logger } from "../logger";
-import { AbortSignal } from "node-fetch/externals";
 
 export enum VSCODE_MESSAGE {
   EMULATORS_STARTED = "EMULATORS_STARTED",
@@ -33,7 +31,7 @@ export async function sendVSCodeMessage(body: WebhookBody) {
         "x-mantle-admin": "all",
       },
       body: jsonBody,
-      signal: AbortSignal.timeout(3000) as unknown as AbortSignal, // necessary due to https://github.com/node-fetch/node-fetch/issues/1652
+      signal: AbortSignal.timeout(3000),
     });
   } catch (e) {
     logger.debug(

src/dataconnect/webhook.ts

@@ -1,4 +1,3 @@
-import fetch, { Response } from "node-fetch";
 import * as fs from "fs";
 import * as path from "path";
 import * as yaml from "yaml";
@@ -97,9 +96,11 @@ export async function detectFromPort(
       res = await Promise.race([fetch(url), timedOut]);
       break;
     } catch (err: any) {
+      const realErr = err?.cause || err;
       if (
         err?.name === "FetchError" ||
-        ["ECONNREFUSED", "ECONNRESET", "ETIMEDOUT"].includes(err?.code)
+        realErr?.name === "FetchError" ||
+        ["ECONNREFUSED", "ECONNRESET", "ETIMEDOUT"].includes(realErr?.code)
       ) {
         continue;
       }

src/deploy/functions/runtimes/discovery/index.ts

@@ -4,7 +4,6 @@ import * as path from "path";
 import * as portfinder from "portfinder";
 import * as semver from "semver";
 import * as spawn from "cross-spawn";
-import fetch from "node-fetch";
 import { ChildProcess } from "child_process";
 
 import { FirebaseError } from "../../../../error";

src/deploy/functions/runtimes/node/index.ts

@@ -1,6 +1,5 @@
 import * as fs from "fs";
 import * as path from "path";
-import fetch from "node-fetch";
 import { promisify } from "util";
 
 import * as portfinder from "portfinder";

src/deploy/functions/runtimes/python/index.ts

@@ -1,5 +1,4 @@
 import { size } from "lodash";
-import AbortController from "abort-controller";
 import * as clc from "colorette";
 import * as crypto from "crypto";
 import * as fs from "fs";
@@ -250,7 +249,7 @@ export class Uploader {
       logger.debug(
         `[hosting][upload] ${this.hashMap[toUpload]} (${toUpload}) HTTP ERROR ${
           res.status
-        }: headers=${JSON.stringify(res.response.headers.raw())} ${errorMessage}`,
+        }: headers=${JSON.stringify(Object.fromEntries((res.response.headers as any).entries()))} ${errorMessage}`,
       );
       throw new Error(`Unexpected error while uploading file: ${errorMessage}`);
     }