refactor(fetch): reimplement fetch with hyper instead of reqwest (#24237)

This commit re-implements `ext/fetch` and all dependent crates
using `hyper` and `hyper-util`, instead of `reqwest`.

The reasoning is that we want to have greater control and access
to low level `hyper` APIs when implementing `fetch` API as well
as `node:http` module.

---------

Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>

5 files changed, 76 insertions, 49 deletions

diff --git a/cli/tools/registry/api.rs b/cli/tools/registry/api.rs
index ee9579a19..519800660 100644
--- a/cli/tools/registry/api.rs
+++ b/cli/tools/registry/api.rs
@@ -1,8 +1,9 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
+use crate::http_util;
 use deno_core::error::AnyError;
 use deno_core::serde_json;
-use deno_runtime::deno_fetch::reqwest;
+use deno_runtime::deno_fetch;
 use lsp_types::Url;
 use serde::de::DeserializeOwned;
 
@@ -82,7 +83,7 @@ impl std::fmt::Debug for ApiError {
 impl std::error::Error for ApiError {}
 
 pub async fn parse_response<T: DeserializeOwned>(
-  response: reqwest::Response,
+  response: http::Response<deno_fetch::ResBody>,
 ) -> Result<T, ApiError> {
   let status = response.status();
   let x_deno_ray = response
@@ -90,7 +91,7 @@ pub async fn parse_response<T: DeserializeOwned>(
     .get("x-deno-ray")
     .and_then(|value| value.to_str().ok())
     .map(|s| s.to_string());
-  let text = response.text().await.unwrap();
+  let text = http_util::body_to_string(response).await.unwrap();
 
   if !status.is_success() {
     match serde_json::from_str::<ApiError>(&text) {
@@ -122,9 +123,9 @@ pub async fn get_scope(
   client: &HttpClient,
   registry_api_url: &Url,
   scope: &str,
-) -> Result<reqwest::Response, AnyError> {
+) -> Result<http::Response<deno_fetch::ResBody>, AnyError> {
   let scope_url = format!("{}scopes/{}", registry_api_url, scope);
-  let response = client.get(&scope_url).send().await?;
+  let response = client.get(scope_url.parse()?)?.send().await?;
   Ok(response)
 }
 
@@ -141,9 +142,9 @@ pub async fn get_package(
   registry_api_url: &Url,
   scope: &str,
   package: &str,
-) -> Result<reqwest::Response, AnyError> {
+) -> Result<http::Response<deno_fetch::ResBody>, AnyError> {
   let package_url = get_package_api_url(registry_api_url, scope, package);
-  let response = client.get(&package_url).send().await?;
+  let response = client.get(package_url.parse()?)?.send().await?;
   Ok(response)
 }
 
diff --git a/cli/tools/registry/mod.rs b/cli/tools/registry/mod.rs
index 8e4d97897..a22384a52 100644
--- a/cli/tools/registry/mod.rs
+++ b/cli/tools/registry/mod.rs
@@ -23,8 +23,8 @@ use deno_core::futures::StreamExt;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
-use deno_runtime::deno_fetch::reqwest;
 use deno_terminal::colors;
+use http_body_util::BodyExt;
 use lsp_types::Url;
 use serde::Deserialize;
 use serde::Serialize;
@@ -539,11 +539,13 @@ async fn get_auth_headers(
       let challenge = BASE64_STANDARD.encode(sha2::Sha256::digest(&verifier));
 
       let response = client
-        .post(format!("{}authorizations", registry_url))
-        .json(&serde_json::json!({
-          "challenge": challenge,
-          "permissions": permissions,
-        }))
+        .post_json(
+          format!("{}authorizations", registry_url).parse()?,
+          &serde_json::json!({
+            "challenge": challenge,
+            "permissions": permissions,
+          }),
+        )?
         .send()
         .await
         .context("Failed to create interactive authorization")?;
@@ -573,11 +575,13 @@ async fn get_auth_headers(
       loop {
         tokio::time::sleep(interval).await;
         let response = client
-          .post(format!("{}authorizations/exchange", registry_url))
-          .json(&serde_json::json!({
-            "exchangeToken": auth.exchange_token,
-            "verifier": verifier,
-          }))
+          .post_json(
+            format!("{}authorizations/exchange", registry_url).parse()?,
+            &serde_json::json!({
+              "exchangeToken": auth.exchange_token,
+              "verifier": verifier,
+            }),
+          )?
           .send()
           .await
           .context("Failed to exchange authorization")?;
@@ -634,15 +638,20 @@ async fn get_auth_headers(
         );
 
         let response = client
-          .get(url)
-          .bearer_auth(&oidc_config.token)
+          .get(url.parse()?)?
+          .header(
+            http::header::AUTHORIZATION,
+            format!("Bearer {}", oidc_config.token).parse()?,
+          )
           .send()
           .await
           .context("Failed to get OIDC token")?;
         let status = response.status();
-        let text = response.text().await.with_context(|| {
-          format!("Failed to get OIDC token: status {}", status)
-        })?;
+        let text = crate::http_util::body_to_string(response)
+          .await
+          .with_context(|| {
+            format!("Failed to get OIDC token: status {}", status)
+          })?;
         if !status.is_success() {
           bail!(
             "Failed to get OIDC token: status {}, response: '{}'",
@@ -770,7 +779,7 @@ async fn ensure_scopes_and_packages_exist(
 
     loop {
       tokio::time::sleep(std::time::Duration::from_secs(3)).await;
-      let response = client.get(&package_api_url).send().await?;
+      let response = client.get(package_api_url.parse()?)?.send().await?;
       if response.status() == 200 {
         let name = format!("@{}/{}", package.scope, package.package);
         log::info!("Package {} created", colors::green(name));
@@ -894,11 +903,19 @@ async fn publish_package(
     package.config
   );
 
+  let body = http_body_util::Full::new(package.tarball.bytes.clone())
+    .map_err(|never| match never {})
+    .boxed();
   let response = http_client
-    .post(url)
-    .header(reqwest::header::AUTHORIZATION, authorization)
-    .header(reqwest::header::CONTENT_ENCODING, "gzip")
-    .body(package.tarball.bytes.clone())
+    .post(url.parse()?, body)?
+    .header(
+      http::header::AUTHORIZATION,
+      authorization.parse().map_err(http::Error::from)?,
+    )
+    .header(
+      http::header::CONTENT_ENCODING,
+      "gzip".parse().map_err(http::Error::from)?,
+    )
     .send()
     .await?;
 
@@ -943,7 +960,7 @@ async fn publish_package(
   while task.status != "success" && task.status != "failure" {
     tokio::time::sleep(interval).await;
     let resp = http_client
-      .get(format!("{}publish_status/{}", registry_api_url, task.id))
+      .get(format!("{}publish_status/{}", registry_api_url, task.id).parse()?)?
       .send()
       .await
       .with_context(|| {
@@ -992,7 +1009,8 @@ async fn publish_package(
       package.scope, package.package, package.version
     ))?;
 
-    let meta_bytes = http_client.get(meta_url).send().await?.bytes().await?;
+    let resp = http_client.get(meta_url)?.send().await?;
+    let meta_bytes = resp.collect().await?.to_bytes();
 
     if std::env::var("DISABLE_JSR_MANIFEST_VERIFICATION_FOR_TESTING").is_err() {
       verify_version_manifest(&meta_bytes, &package)?;
@@ -1023,9 +1041,8 @@ async fn publish_package(
       registry_api_url, package.scope, package.package, package.version
     );
     http_client
-      .post(provenance_url)
-      .header(reqwest::header::AUTHORIZATION, authorization)
-      .json(&json!({ "bundle": bundle }))
+      .post_json(provenance_url.parse()?, &json!({ "bundle": bundle }))?
+      .header(http::header::AUTHORIZATION, authorization.parse()?)
       .send()
       .await?;
   }
diff --git a/cli/tools/registry/provenance.rs b/cli/tools/registry/provenance.rs
index 622e483d6..ce3d6ff8a 100644
--- a/cli/tools/registry/provenance.rs
+++ b/cli/tools/registry/provenance.rs
@@ -1,5 +1,6 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
+use crate::http_util;
 use crate::http_util::HttpClient;
 
 use super::api::OidcTokenResponse;
@@ -12,6 +13,8 @@ use deno_core::anyhow;
 use deno_core::anyhow::bail;
 use deno_core::error::AnyError;
 use deno_core::serde_json;
+use deno_core::url::Url;
+use http_body_util::BodyExt;
 use once_cell::sync::Lazy;
 use p256::elliptic_curve;
 use p256::pkcs8::AssociatedOid;
@@ -504,12 +507,12 @@ impl<'a> FulcioSigner<'a> {
 
     let response = self
       .http_client
-      .post(url)
-      .json(&request_body)
+      .post_json(url.parse()?, &request_body)?
       .send()
       .await?;
 
-    let body: SigningCertificateResponse = response.json().await?;
+    let body: SigningCertificateResponse =
+      http_util::body_to_json(response).await?;
 
     let key = body
       .signed_certificate_embedded_sct
@@ -527,15 +530,23 @@ impl<'a> FulcioSigner<'a> {
       bail!("No OIDC token available");
     };
 
-    let res = self
+    let mut url = req_url.parse::<Url>()?;
+    url.query_pairs_mut().append_pair("audience", aud);
+    let res_bytes = self
       .http_client
-      .get(&req_url)
-      .bearer_auth(token)
-      .query(&[("audience", aud)])
+      .get(url)?
+      .header(
+        http::header::AUTHORIZATION,
+        format!("Bearer {}", token)
+          .parse()
+          .map_err(http::Error::from)?,
+      )
       .send()
       .await?
-      .json::<OidcTokenResponse>()
-      .await?;
+      .collect()
+      .await?
+      .to_bytes();
+    let res: OidcTokenResponse = serde_json::from_slice(&res_bytes)?;
     Ok(res.value)
   }
 }
@@ -685,11 +696,10 @@ async fn testify(
 
   let url = format!("{}/api/v1/log/entries", *DEFAULT_REKOR_URL);
   let res = http_client
-    .post(&url)
-    .json(&proposed_intoto_entry)
+    .post_json(url.parse()?, &proposed_intoto_entry)?
     .send()
     .await?;
-  let body: RekorEntry = res.json().await?;
+  let body: RekorEntry = http_util::body_to_json(res).await?;
 
   Ok(body)
 }
diff --git a/cli/tools/test/mod.rs b/cli/tools/test/mod.rs
index 81dc36a89..587b737d6 100644
--- a/cli/tools/test/mod.rs
+++ b/cli/tools/test/mod.rs
@@ -881,12 +881,11 @@ async fn run_tests_for_worker_inner(
     // failing. If we don't do this, a connection to a test server we just tore down might be re-used in
     // the next test.
     // TODO(mmastrac): this should be some sort of callback that we can implement for any subsystem
-    #[allow(clippy::disallowed_types)] // allow using reqwest::Client here
     worker
       .js_runtime
       .op_state()
       .borrow_mut()
-      .try_take::<deno_runtime::deno_fetch::reqwest::Client>();
+      .try_take::<deno_runtime::deno_fetch::Client>();
 
     if desc.ignore {
       send_test_event(
diff --git a/cli/tools/upgrade.rs b/cli/tools/upgrade.rs
index 2afeffc92..fd8394efa 100644
--- a/cli/tools/upgrade.rs
+++ b/cli/tools/upgrade.rs
@@ -571,7 +571,7 @@ async fn get_latest_version(
   check_kind: UpgradeCheckKind,
 ) -> Result<String, AnyError> {
   let url = get_url(release_kind, env!("TARGET"), check_kind);
-  let text = client.download_text(url).await?;
+  let text = client.download_text(url.parse()?).await?;
   Ok(normalize_version_from_server(release_kind, &text))
 }
 
@@ -624,7 +624,7 @@ async fn download_package(
     // text above which will stay alive after the progress bars are complete
     let progress = progress_bar.update("");
     client
-      .download_with_progress(download_url, None, &progress)
+      .download_with_progress(download_url.parse()?, None, &progress)
       .await?
   };
   match maybe_bytes {