refactor(serverless,cli): redis pub/sub & rustls-tls

.changeset/nasty-weeks-lie.md

@@ -0,0 +1,5 @@
+---
+'@lagon/serverless': patch
+---
+
+Fix redis pub/sub sometimes disconnecting

.changeset/rude-toys-wait.md

@@ -0,0 +1,6 @@
+---
+'@lagon/cli': patch
+'@lagon/serverless': patch
+---
+
+Use rustls-ls instead of native-tls

crates/cli/Cargo.toml

@@ -14,12 +14,12 @@ indicatif = "0.17.5"
 dirs = "5.0.1"
 webbrowser = "0.8.10"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync"] }
-hyper = { version = "0.14.26", features = ["client", "server", "http1", "runtime", "stream"] }
+hyper = { version = "0.14.26", features = ["server", "http1", "runtime", "stream"] }
+reqwest = { version = "0.11.18", default-features = false, features = ["rustls-tls"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 walkdir = "2.3.3"
 pathdiff = "0.2.1"
-hyper-tls = { version = "0.5.0", features = ["vendored"] }
 flume = "0.10.14"
 chrono = "0.4.26"
 notify = "6.0.0"

crates/cli/src/utils/deployments.rs

@@ -5,7 +5,6 @@ use crate::utils::{get_theme, print_progress, TrpcClient};
 use anyhow::{anyhow, Result};
 use dialoguer::console::style;
 use dialoguer::{Confirm, Input};
-use hyper::{Body, Method, Request};
 use pathdiff::diff_paths;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -428,12 +427,12 @@ pub async fn create_deployment(
 
     let end_progress = print_progress("Uploading files");
 
-    let request = Request::builder()
-        .method(Method::PUT)
-        .uri(code_url)
-        .body(Body::from(index))?;
-
-    trpc_client.client.request(request).await?;
+    trpc_client
+        .client
+        .request("PUT".parse()?, code_url)
+        .body(index)
+        .send()
+        .await?;
 
     let mut join_set = tokio::task::JoinSet::new();
     for (asset, url) in assets_urls {
@@ -484,11 +483,12 @@ pub async fn create_deployment(
 }
 
 async fn upload_asset(trpc_client: Arc<TrpcClient>, asset: Vec<u8>, url: String) -> Result<()> {
-    let request = Request::builder()
-        .method(Method::PUT)
-        .uri(url)
-        .body(Body::from(asset))?;
+    trpc_client
+        .client
+        .request("PUT".parse()?, url)
+        .body(asset)
+        .send()
+        .await?;
 
-    trpc_client.client.request(request).await?;
     Ok(())
 }

crates/cli/src/utils/trpc.rs

@@ -1,7 +1,6 @@
 use super::Config;
 use anyhow::{anyhow, Result};
-use hyper::{body, client::HttpConnector, Body, Client, Method, Request};
-use hyper_tls::HttpsConnector;
+use reqwest::{Client, ClientBuilder};
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use urlencoding::encode;
 
@@ -26,13 +25,13 @@ pub struct TrpcErrorResult {
 }
 
 pub struct TrpcClient {
-    pub client: Client<HttpsConnector<HttpConnector>>,
+    pub client: Client,
     config: Config,
 }
 
 impl TrpcClient {
     pub fn new(config: Config) -> Self {
-        let client = Client::builder().build::<_, Body>(HttpsConnector::new());
+        let client = ClientBuilder::new().use_rustls_tls().build().unwrap();
 
         Self { client, config }
     }
@@ -47,21 +46,18 @@ impl TrpcClient {
             None => String::new(),
         };
 
-        let request = Request::builder()
-            .method(Method::GET)
-            .uri(format!(
-                "{}/api/trpc/{}{}",
-                self.config.site_url.clone(),
-                key,
-                input,
-            ))
+        let response = self
+            .client
+            .request(
+                "GET".parse()?,
+                format!("{}/api/trpc/{}{}", self.config.site_url.clone(), key, input,),
+            )
             .header("content-type", "application/json")
             .header("x-lagon-token", self.config.token.as_ref().unwrap())
-            .body(Body::empty())?;
+            .send()
+            .await?;
 
-        let response = self.client.request(request).await?;
-        let body = body::to_bytes(response.into_body()).await?;
-        let body = String::from_utf8(body.to_vec())?;
+        let body = response.text().await?;
 
         match serde_json::from_str::<TrpcResponse<R>>(&body) {
             Ok(response) => Ok(response),
@@ -79,16 +75,19 @@ impl TrpcClient {
     {
         let body = serde_json::to_string(&body)?;
 
-        let request = Request::builder()
-            .method(Method::POST)
-            .uri(format!("{}/api/trpc/{}", self.config.site_url.clone(), key))
+        let response = self
+            .client
+            .request(
+                "POST".parse()?,
+                format!("{}/api/trpc/{}", self.config.site_url.clone(), key),
+            )
             .header("content-type", "application/json")
             .header("x-lagon-token", self.config.token.as_ref().unwrap())
-            .body(Body::from(body))?;
+            .body(body)
+            .send()
+            .await?;
 
-        let response = self.client.request(request).await?;
-        let body = body::to_bytes(response.into_body()).await?;
-        let body = String::from_utf8(body.to_vec())?;
+        let body = response.text().await?;
 
         match serde_json::from_str::<TrpcResponse<R>>(&body) {
             Ok(response) => Ok(response),

crates/runtime_isolate/Cargo.toml

@@ -15,7 +15,7 @@ linked-hash-map = "0.5.6"
 lagon-runtime-v8-utils = { path = "../runtime_v8_utils" }
 lagon-runtime-http = { path = "../runtime_http" }
 lagon-runtime-crypto = { path = "../runtime_crypto" }
-reqwest = { version = "0.11.18", features = ["rustls-tls"] }
+reqwest = { version = "0.11.18", default-features = false, features = ["rustls-tls"] }
 
 [features]
 default = []

crates/serverless/Cargo.toml

@@ -15,7 +15,7 @@ lagon-serverless-logger = { path = "../serverless_logger" }
 lagon-serverless-downloader = { path = "../serverless_downloader" }
 lagon-serverless-pubsub = { path = "../serverless_pubsub" }
 flume = "0.10.14"
-mysql = "24.0.0"
+mysql = { version = "24.0.0", default-features = false, features = ["default-rustls"] }
 dotenv = "0.15.0"
 serde_json = "1.0"
 metrics = "0.21.0"
@@ -36,7 +36,7 @@ lagon-runtime-isolate = { path = "../runtime_isolate" }
 flume = "0.10.14"
 
 [dev-dependencies]
-reqwest = "0.11.18"
+reqwest = { version = "0.11.18", default-features = false, features = ["rustls-tls"] }
 serial_test = "2.0.0"
 clickhouse = { version = "0.11.4", features = ["test-util"] }