libsql: Switch to hyper for local offline sync-switch-hyper

This commit switches our usage of reqwest to hyper. In the future, this
will allow us to pass in configuration for the connector to support both
custom TLS implementations (aka using the system tls impl over rustls)
and connecting to the sqld network simulation for future testing. This
also removes the extra dependency on reqwest which is not needed since
we already pull in the dependencies needed to implement the new
protocol for local offline writes.

Author: LucioFranco

Cargo.lock

@@ -42,7 +42,6 @@ fallible-iterator = { version = "0.3", optional = true }
 
 libsql_replication = { version = "0.6", path = "../libsql-replication", optional = true }
 async-stream = { version = "0.3.5", optional = true }
-reqwest = { version = "0.12.9", default-features = false, features = [ "rustls-tls", "json" ], optional = true }
 
 [dev-dependencies]
 criterion = { version = "0.5", features = ["html_reports", "async", "async_futures", "async_tokio"] }
@@ -105,7 +104,6 @@ sync = [
   "dep:bytes",
   "dep:tokio",
   "dep:futures",
-  "dep:reqwest",
   "dep:serde_json",
 ]
 hrana = [

libsql/Cargo.toml

@@ -476,21 +476,40 @@ impl Database {
     ) -> Result<u32> {
         let mut nr_retries = 0;
         loop {
-            let client = reqwest::Client::new();
-            let mut builder = client.post(uri.to_owned());
+            // TODO(lucio): add custom connector + tls support here
+            let client = hyper::client::Client::builder().build_http::<hyper::Body>();
+
+            let mut req = http::Request::post(uri.clone());
+
             match auth_token {
                 Some(ref auth_token) => {
-                    builder = builder
-                        .header("Authorization", format!("Bearer {}", auth_token.to_owned()));
+                    let auth_header =
+                        http::HeaderValue::try_from(format!("Bearer {}", auth_token.to_owned()))
+                            .unwrap();
+
+                    req.headers_mut()
+                        .expect("valid http request")
+                        .insert("Authorization", auth_header);
                 }
                 None => {}
             }
-            let res = builder.body(frame.to_vec()).send().await.unwrap();
+
+            // TODO(lucio): convert this to use bytes to make this clone cheap, it should be
+            // to possible use BytesMut when reading frames from the WAL and efficiently use Bytes
+            // from that.
+            let req = req.body(frame.clone().into()).expect("valid body");
+
+            let res = client.request(req).await.unwrap();
+
+            // TODO(lucio): only retry on server side errors
             if res.status().is_success() {
-                let resp = res.json::<serde_json::Value>().await.unwrap();
+                let res_body = hyper::body::to_bytes(res.into_body()).await.unwrap();
+                let resp = serde_json::from_slice::<serde_json::Value>(&res_body[..]).unwrap();
+
                 let max_frame_no = resp.get("max_frame_no").unwrap().as_u64().unwrap();
                 return Ok(max_frame_no as u32);
             }
+
             if nr_retries > max_retries {
                 return Err(crate::errors::Error::ConnectionFailed(format!(
                     "Failed to push frame: {}",