feat: use SSE in forc-publish for long-running uploads (#7178)

This pull request introduces changes to enhance the functionality of the
`forc-publish` plugin, including improvements to error handling, support
for streaming server responses, and updates to dependencies. The most
significant changes include adding a new error variant, updating the
`upload` method to handle streaming responses, and modifying
dependencies to include `futures-util` and additional features for
`reqwest`.

### Enhancements to Error Handling:
* Added a new `ServerError` variant to the `Error` enum in
`forc-plugins/forc-publish/src/error.rs` to handle server-related
errors.

### Improvements to `upload` Method:
* Updated the `upload` method in `forc-publish/src/forc_pub_client.rs`
to process streaming server responses using `futures-util::StreamExt`.
This includes handling chunked data, parsing JSON responses, and
printing event updates.
[[1]](diffhunk://#diff-c706a808124a73081593a5e34fb999aa1a446f97d97da1657ddb4b3c55d24371R43)
[[2]](diffhunk://#diff-c706a808124a73081593a5e34fb999aa1a446f97d97da1657ddb4b3c55d24371L53-R97)

### Dependency Updates:
* Added `futures-util` as a dependency in `Cargo.toml` and
`forc-plugins/forc-publish/Cargo.toml` to support streaming operations.
[[1]](diffhunk://#diff-2e9d962a08321605940b5a657135052fbcef87b5e360662bb527c96d9a615542R157)
[[2]](diffhunk://#diff-2bd34a7b1fa0f1850ad25c6e080d410b2df98c16b6cd75445a138d04e617311aL16-R17)
* Enabled the `stream` feature for `reqwest` in
`forc-plugins/forc-publish/Cargo.toml` to facilitate streaming server
responses.

### Minor Adjustments:
* Added empty `println!` calls in `main.rs` to improve output formatting
during error handling and publishing operations.
[[1]](diffhunk://#diff-2c9e261f1e02238f33a9403387f40481b5b372475c0de3782146b614210675b3R32)
[[2]](diffhunk://#diff-2c9e261f1e02238f33a9403387f40481b5b372475c0de3782146b614210675b3R52)

![May-16-2025
03-56-49](https://github.com/user-attachments/assets/f3fd0a36-a3e1-41df-9039-b0d489d9e3b5)

Author: sdankel

Cargo.lock

@@ -154,6 +154,7 @@ filecheck = "0.5"
 flate2 = "1.0"
 fs_extra = "1.2"
 futures = { version = "0.3", default-features = false }
+futures-util = "0.3"
 gag = "1.0"
 gimli = "0.31"
 git2 = "0.19"

Cargo.toml

@@ -13,7 +13,8 @@ clap = { workspace = true, features = ["derive", "env"] }
 flate2.workspace = true
 forc-tracing.workspace = true
 forc-util.workspace = true
-reqwest = { workspace = true, features = ["json"] }
+futures-util.workspace = true
+reqwest = { workspace = true, features = ["json", "stream"] }
 semver = { workspace = true, features = ["serde"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true

forc-plugins/forc-publish/Cargo.toml

@@ -28,6 +28,9 @@ pub enum Error {
 
     #[error("Forc.toml not found in the current directory")]
     ForcTomlNotFound,
+
+    #[error("Server error")]
+    ServerError,
 }
 
 #[derive(Deserialize)]

forc-plugins/forc-publish/src/error.rs

@@ -1,5 +1,6 @@
 use crate::error::Error;
 use crate::error::Result;
+use reqwest::StatusCode;
 use semver::Version;
 use serde::{Deserialize, Serialize};
 use std::fs;
@@ -39,6 +40,8 @@ impl ForcPubClient {
 
     /// Uploads the given file to the server
     pub async fn upload<P: AsRef<Path>>(&self, file_path: P, forc_version: &str) -> Result<Uuid> {
+        use futures_util::StreamExt;
+        use std::io::{stdout, Write};
         let url = self
             .uri
             .join(&format!("upload_project?forc_version={}", forc_version))?;
@@ -50,16 +53,55 @@ impl ForcPubClient {
             .header("Content-Type", "application/gzip")
             .body(file_bytes)
             .send()
-            .await?;
-
-        let status = response.status();
+            .await;
 
-        if status.is_success() {
-            // Extract `upload_id` from the response if available
-            let upload_response: UploadResponse = response.json().await?;
-            Ok(upload_response.upload_id)
+        if let Ok(response) = response {
+            let mut stream = response.bytes_stream();
+
+            // Process the SSE stream.
+            // The server sends events in the format: "data: <event>\n\n" or
+            // ": <event>\n\n" for keep-alive events.
+            // The first event is usually a progress event, and the last one contains the upload_id
+            // or an error message. If the stream is open for more than 60 seconds, it will be closed
+            // by the server, and we will return an HTTPError.
+            while let Some(chunk) = stream.next().await {
+                match chunk {
+                    Ok(bytes) => {
+                        let event_str = String::from_utf8_lossy(&bytes);
+                        for event in event_str.split("\n\n") {
+                            if let Some(stripped) = event.strip_prefix("data:") {
+                                let data = &stripped.trim();
+                                if let Ok(upload_response) =
+                                    serde_json::from_str::<UploadResponse>(data)
+                                {
+                                    return Ok(upload_response.upload_id);
+                                } else if data.starts_with("{") {
+                                    // Attempt to parse error from JSON
+                                    return Err(Error::ApiResponseError {
+                                        status: StatusCode::INTERNAL_SERVER_ERROR,
+                                        error: data.to_string(),
+                                    });
+                                } else {
+                                    // Print the event data, replacing the previous message.
+                                    print!("\r\x1b[2K  =>  {}", data);
+                                    stdout().flush().unwrap();
+                                }
+                            }
+                            // else if event.starts_with(":") {
+                            // These are keep-alive events. Uncomment if you need to debug them.
+                            // println!("Keep-alive event: {}", event);
+                            // }
+                        }
+                    }
+                    Err(e) => {
+                        return Err(Error::HttpError(e));
+                    }
+                }
+            }
+            Err(Error::ServerError)
         } else {
-            Err(Error::from_response(response).await)
+            eprintln!("Error during upload initiation: {:?}", response);
+            Err(Error::ServerError)
         }
     }
 
@@ -110,12 +152,22 @@ mod test {
     async fn test_upload_success() {
         let (client, mock_server) = get_mock_client_server().await;
         let upload_id = Uuid::new_v4();
-        let success_response = serde_json::json!({ "upload_id": upload_id });
+
+        // Simulate SSE response with a progress event and a final upload_id event
+        let sse_body = format!(
+            "data: uploading...\n\n\
+             data: {{\"upload_id\":\"{}\"}}\n\n",
+            upload_id
+        );
 
         Mock::given(method("POST"))
             .and(path("/upload_project"))
             .and(query_param("forc_version", "0.66.5"))
-            .respond_with(ResponseTemplate::new(200).set_body_json(&success_response))
+            .respond_with(
+                ResponseTemplate::new(200)
+                    .insert_header("Content-Type", "text/event-stream")
+                    .set_body_string(sse_body),
+            )
             .mount(&mock_server)
             .await;
 
@@ -133,11 +185,15 @@ mod test {
     async fn test_upload_server_error() {
         let (client, mock_server) = get_mock_client_server().await;
 
+        // Simulate SSE error event
+        let sse_body = "data: {\"error\":\"Internal Server Error\"}\n\n";
+
         Mock::given(method("POST"))
             .and(path("/upload_project"))
             .respond_with(
-                ResponseTemplate::new(500)
-                    .set_body_json(serde_json::json!({ "error": "Internal Server Error" })),
+                ResponseTemplate::new(200)
+                    .insert_header("Content-Type", "text/event-stream")
+                    .set_body_string(sse_body),
             )
             .mount(&mock_server)
             .await;
@@ -151,7 +207,7 @@ mod test {
         match result {
             Err(Error::ApiResponseError { status, error }) => {
                 assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
-                assert_eq!(error, "Internal Server Error");
+                assert_eq!(error, "{\"error\":\"Internal Server Error\"}");
             }
             _ => panic!("Expected ApiResponseError"),
         }

forc-plugins/forc-publish/src/forc_pub_client.rs

@@ -29,6 +29,7 @@ async fn main() {
     init_tracing_subscriber(TracingSubscriberOptions::default());
 
     if let Err(err) = run().await {
+        println!();
         println_error(&format!("{err}"));
         std::process::exit(1);
     }
@@ -48,6 +49,7 @@ async fn run() -> Result<()> {
     let upload_id = client.upload(file_path, forc_version).await?;
     let published = client.publish(upload_id, &auth_token).await?;
 
+    println!();
     println_action_green(
         "Published",
         &format!("{} {}", published.name, published.version),