feat(cli): parallel upload of assets

.changeset/fair-tomatoes-unite.md

@@ -0,0 +1,5 @@
+---
+'@lagon/cli': patch
+---
+
+Allow parallel upload for assets in lagon deploy command.

crates/cli/src/commands/deploy.rs

@@ -73,7 +73,7 @@ pub async fn deploy(
         println!("{}", debug("No deployment config found..."));
         println!();
 
-        let trpc_client = TrpcClient::new(&config);
+        let trpc_client = TrpcClient::new(config.clone());
         let response = trpc_client
             .query::<(), OrganizationsResponse>("organizationsList", None)
             .await?;
@@ -107,7 +107,7 @@ pub async fn deploy(
                 function_config.organization_id = organization.id.clone();
                 function_config.write(&root)?;
 
-                create_deployment(&config, &function_config, prod, &root).await?;
+                create_deployment(config, &function_config, prod, &root).await?;
             }
             false => {
                 let name = Input::<String>::new()
@@ -137,11 +137,11 @@ pub async fn deploy(
                 function_config.organization_id = organization.id.clone();
                 function_config.write(&root)?;
 
-                create_deployment(&config, &function_config, prod, &root).await?;
+                create_deployment(config, &function_config, prod, &root).await?;
             }
         }
     } else {
-        create_deployment(&config, &function_config, prod, &root).await?;
+        create_deployment(config, &function_config, prod, &root).await?;
     }
 
     Ok(())

crates/cli/src/commands/link.rs

@@ -23,7 +23,7 @@ pub async fn link(directory: Option<PathBuf>) -> Result<()> {
     match root.join(".lagon").join("config.json").exists() {
         true => Err(anyhow!("This directory is already linked to a Function.")),
         false => {
-            let trpc_client = TrpcClient::new(&config);
+            let trpc_client = TrpcClient::new(config);
             let response = trpc_client
                 .query::<(), OrganizationsResponse>("organizationsList", None)
                 .await?;

crates/cli/src/commands/login.rs

@@ -46,7 +46,7 @@ pub async fn login() -> Result<()> {
 
     config.set_token(Some(code.clone()));
 
-    let client = TrpcClient::new(&config);
+    let client = TrpcClient::new(config.clone());
     let request = CliRequest { code };
 
     match client

crates/cli/src/commands/ls.rs

@@ -41,7 +41,7 @@ pub async fn ls(directory: Option<PathBuf>) -> Result<()> {
     let function_config = FunctionConfig::load(&root, None, None)?;
     let end_progress = print_progress("Fetching deployments...");
 
-    let function = TrpcClient::new(&config)
+    let function = TrpcClient::new(config)
         .query::<FunctionRequest, FunctionResponse>(
             "functionGet",
             Some(FunctionRequest {

crates/cli/src/commands/promote.rs

@@ -39,7 +39,7 @@ pub async fn promote(deployment_id: String, directory: Option<PathBuf>) -> Resul
     {
         true => {
             let end_progress = print_progress("Promoting Deployment...");
-            TrpcClient::new(&config)
+            TrpcClient::new(config)
                 .mutation::<PromoteDeploymentRequest, PromoteDeploymentResponse>(
                     "deploymentPromote",
                     PromoteDeploymentRequest {

crates/cli/src/commands/rm.rs

@@ -39,7 +39,7 @@ pub async fn rm(directory: Option<PathBuf>) -> Result<()> {
     {
         true => {
             let end_progress = print_progress("Deleting Function...");
-            TrpcClient::new(&config)
+            TrpcClient::new(config)
                 .mutation::<DeleteFunctionRequest, DeleteFunctionResponse>(
                     "functionDelete",
                     DeleteFunctionRequest {

crates/cli/src/commands/undeploy.rs

@@ -38,7 +38,7 @@ pub async fn undeploy(deployment_id: String, directory: Option<PathBuf>) -> Resu
     {
         true => {
             let end_progress = print_progress("Deleting Deployment...");
-            TrpcClient::new(&config)
+            TrpcClient::new(config)
                 .mutation::<UndeployDeploymentRequest, UndeployDeploymentResponse>(
                     "deploymentUndeploy",
                     UndeployDeploymentRequest {

crates/cli/src/utils/config.rs

@@ -20,7 +20,7 @@ fn get_config_path() -> Result<PathBuf> {
         .join("config.json"))
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct Config {
     pub token: Option<String>,
     pub site_url: String,

crates/cli/src/utils/deployments.rs

@@ -9,6 +9,7 @@ use std::{
     path::{Path, PathBuf},
     process::Command,
 };
+use std::sync::Arc;
 use walkdir::{DirEntry, WalkDir};
 
 use pathdiff::diff_paths;
@@ -343,7 +344,7 @@ struct DeployDeploymentResponse {
 }
 
 pub async fn create_deployment(
-    config: &Config,
+    config: Config,
     function_config: &FunctionConfig,
     prod: bool,
     root: &Path,
@@ -352,7 +353,7 @@ pub async fn create_deployment(
 
     let end_progress = print_progress("Creating deployment...");
 
-    let trpc_client = TrpcClient::new(config);
+    let trpc_client = Arc::new(TrpcClient::new(config));
     let response = trpc_client
         .mutation::<CreateDeploymentRequest, CreateDeploymentResponse>(
             "deploymentCreate",
@@ -387,18 +388,17 @@ pub async fn create_deployment(
 
     trpc_client.client.request(request).await?;
 
-    // TODO upload in parallel
+    let mut join_set = tokio::task::JoinSet::new();
     for (asset, url) in assets_urls {
         let asset = assets
             .get(&asset)
             .unwrap_or_else(|| panic!("Couldn't find asset {asset}"));
 
-        let request = Request::builder()
-            .method(Method::PUT)
-            .uri(url)
-            .body(Body::from(asset.clone()))?;
+        join_set.spawn(upload_asset(trpc_client.clone(), asset.clone(), url));
+    }
 
-        trpc_client.client.request(request).await?;
+    while let Some(res) = join_set.join_next().await {
+        res?.expect("Couldn't upload asset");
     }
 
     end_progress();
@@ -430,3 +430,13 @@ pub async fn create_deployment(
 
     Ok(())
 }
+
+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(request).await?;
+    Ok(())
+}

crates/cli/src/utils/trpc.rs

@@ -26,13 +26,13 @@ pub struct TrpcErrorResult {
     error: TrpcError,
 }
 
-pub struct TrpcClient<'a> {
+pub struct TrpcClient {
     pub client: Client<HttpsConnector<HttpConnector>>,
-    config: &'a Config,
+    config:  Config,
 }
 
-impl<'a> TrpcClient<'a> {
-    pub fn new(config: &'a Config) -> Self {
+impl TrpcClient {
+    pub fn new(config: Config) -> Self {
         let client = Client::builder().build::<_, Body>(HttpsConnector::new());
 
         Self { client, config }