Reland "refactor(fetch): reimplement fetch with hyper instead of reqwest" (#24593)

Originally landed in
https://github.com/denoland/deno/commit/f6fd6619e708a515831f707438368d81b0c9aa56.
Reverted in https://github.com/denoland/deno/pull/24574.

This reland contains a fix that sends "Accept: */*" header for calls made
from "FileFetcher". Absence of this header made downloading source code
from JSR broken. This is tested by ensuring this header is present in the
test server that servers JSR packages.

---------

Co-authored-by: Sean McArthur <sean@seanmonstar.com>

2 files changed, 24 insertions, 16 deletions

diff --git a/ext/kv/Cargo.toml b/ext/kv/Cargo.toml
index e4249afea..4556eb23c 100644
--- a/ext/kv/Cargo.toml
+++ b/ext/kv/Cargo.toml
@@ -29,6 +29,7 @@ denokv_remote.workspace = true
 denokv_sqlite.workspace = true
 faster-hex.workspace = true
 http.workspace = true
+http-body-util.workspace = true
 log.workspace = true
 num-bigint.workspace = true
 prost.workspace = true
diff --git a/ext/kv/remote.rs b/ext/kv/remote.rs
index 7541b5a06..922853588 100644
--- a/ext/kv/remote.rs
+++ b/ext/kv/remote.rs
@@ -12,10 +12,8 @@ use bytes::Bytes;
 use deno_core::error::type_error;
 use deno_core::error::AnyError;
 use deno_core::futures::Stream;
-use deno_core::futures::TryStreamExt as _;
 use deno_core::OpState;
 use deno_fetch::create_http_client;
-use deno_fetch::reqwest;
 use deno_fetch::CreateHttpClientOptions;
 use deno_tls::rustls::RootCertStore;
 use deno_tls::Proxy;
@@ -25,6 +23,7 @@ use denokv_remote::MetadataEndpoint;
 use denokv_remote::Remote;
 use denokv_remote::RemoteResponse;
 use denokv_remote::RemoteTransport;
+use http_body_util::BodyExt;
 use url::Url;
 
 #[derive(Clone)]
@@ -109,35 +108,43 @@ impl<P: RemoteDbHandlerPermissions + 'static> denokv_remote::RemotePermissions
 }
 
 #[derive(Clone)]
-pub struct ReqwestClient(reqwest::Client);
-pub struct ReqwestResponse(reqwest::Response);
+pub struct FetchClient(deno_fetch::Client);
+pub struct FetchResponse(http::Response<deno_fetch::ResBody>);
 
-impl RemoteTransport for ReqwestClient {
-  type Response = ReqwestResponse;
+impl RemoteTransport for FetchClient {
+  type Response = FetchResponse;
   async fn post(
     &self,
     url: Url,
     headers: http::HeaderMap,
     body: Bytes,
   ) -> Result<(Url, http::StatusCode, Self::Response), anyhow::Error> {
-    let res = self.0.post(url).headers(headers).body(body).send().await?;
-    let url = res.url().clone();
+    let body = http_body_util::Full::new(body)
+      .map_err(|never| match never {})
+      .boxed();
+    let mut req = http::Request::new(body);
+    *req.method_mut() = http::Method::POST;
+    *req.uri_mut() = url.as_str().parse()?;
+    *req.headers_mut() = headers;
+
+    let res = self.0.clone().send(req).await?;
     let status = res.status();
-    Ok((url, status, ReqwestResponse(res)))
+    Ok((url, status, FetchResponse(res)))
   }
 }
 
-impl RemoteResponse for ReqwestResponse {
+impl RemoteResponse for FetchResponse {
   async fn bytes(self) -> Result<Bytes, anyhow::Error> {
-    Ok(self.0.bytes().await?)
+    Ok(self.0.collect().await?.to_bytes())
   }
   fn stream(
     self,
   ) -> impl Stream<Item = Result<Bytes, anyhow::Error>> + Send + Sync {
-    self.0.bytes_stream().map_err(|e| e.into())
+    self.0.into_body().into_data_stream()
   }
   async fn text(self) -> Result<String, anyhow::Error> {
-    Ok(self.0.text().await?)
+    let bytes = self.bytes().await?;
+    Ok(std::str::from_utf8(&bytes)?.into())
   }
 }
 
@@ -145,7 +152,7 @@ impl RemoteResponse for ReqwestResponse {
 impl<P: RemoteDbHandlerPermissions + 'static> DatabaseHandler
   for RemoteDbHandler<P>
 {
-  type DB = Remote<PermissionChecker<P>, ReqwestClient>;
+  type DB = Remote<PermissionChecker<P>, FetchClient>;
 
   async fn open(
     &self,
@@ -201,14 +208,14 @@ impl<P: RemoteDbHandlerPermissions + 'static> DatabaseHandler
         http2: true,
       },
     )?;
-    let reqwest_client = ReqwestClient(client);
+    let fetch_client = FetchClient(client);
 
     let permissions = PermissionChecker {
       state: state.clone(),
       _permissions: PhantomData,
     };
 
-    let remote = Remote::new(reqwest_client, permissions, metadata_endpoint);
+    let remote = Remote::new(fetch_client, permissions, metadata_endpoint);
 
     Ok(remote)
   }