feat(cloud): complete --wait flag implementation for connectivity commands

- Updated all PSC handler functions to use handle_async_response
- Updated all VPC Peering handlers (already done in previous commit)
- Updated TGW handler function signatures to accept ConnectionManager
- Replaced manual task ID handling with handle_async_response utility
- All connectivity commands now support --wait, --wait-timeout, and --wait-interval flags

Completes implementation for:
- VPC Peering: 6 commands (create, update, delete + Active-Active variants)
- PSC: 9 commands (service and endpoint operations + Active-Active variants)
- TGW: 7 commands (attachment operations + Active-Active variants)

Part of #196
Closes #196

Author: joshrotenberg

crates/redisctl/src/commands/cloud/connectivity/psc.rs

@@ -284,13 +284,23 @@ async fn delete_service(
     }
 
     let handler = PscHandler::new(client.clone());
-    handler
+    let response = handler
         .delete_service(subscription_id)
         .await
         .context("Failed to delete PSC service")?;
 
-    eprintln!("PSC service deleted successfully");
-    Ok(())
+    let json_response = serde_json::to_value(&response).context("Failed to serialize response")?;
+
+    handle_async_response(
+        conn_mgr,
+        profile_name,
+        json_response,
+        async_ops,
+        output_format,
+        query,
+        "PSC service deleted successfully",
+    )
+    .await
 }
 
 // ============================================================================
@@ -338,19 +348,18 @@ async fn create_endpoint(
         .await
         .context("Failed to create PSC endpoint")?;
 
-    // Convert response to JSON and check for task ID
     let json_response = serde_json::to_value(&response).context("Failed to serialize response")?;
-    if let Some(task_id) = json_response.get("taskId").and_then(|v| v.as_str()) {
-        eprintln!("PSC endpoint creation initiated. Task ID: {}", task_id);
-        eprintln!(
-            "Use 'redisctl cloud task wait {}' to monitor progress",
-            task_id
-        );
-    }
 
-    let data = handle_output(json_response, output_format, query)?;
-    print_formatted_output(data, output_format)?;
-    Ok(())
+    handle_async_response(
+        conn_mgr,
+        profile_name,
+        json_response,
+        async_ops,
+        output_format,
+        query,
+        "PSC endpoint created successfully",
+    )
+    .await
 }
 
 async fn update_endpoint(
@@ -378,19 +387,18 @@ async fn update_endpoint(
         .await
         .context("Failed to update PSC endpoint")?;
 
-    // Convert response to JSON and check for task ID
     let json_response = serde_json::to_value(&response).context("Failed to serialize response")?;
-    if let Some(task_id) = json_response.get("taskId").and_then(|v| v.as_str()) {
-        eprintln!("PSC endpoint update initiated. Task ID: {}", task_id);
-        eprintln!(
-            "Use 'redisctl cloud task wait {}' to monitor progress",
-            task_id
-        );
-    }
 
-    let data = handle_output(json_response, output_format, query)?;
-    print_formatted_output(data, output_format)?;
-    Ok(())
+    handle_async_response(
+        conn_mgr,
+        profile_name,
+        json_response,
+        async_ops,
+        output_format,
+        query,
+        "PSC endpoint updated successfully",
+    )
+    .await
 }
 
 async fn delete_endpoint(
@@ -416,13 +424,23 @@ async fn delete_endpoint(
     }
 
     let handler = PscHandler::new(client.clone());
-    handler
+    let response = handler
         .delete_endpoint(subscription_id, endpoint_id)
         .await
         .context("Failed to delete PSC endpoint")?;
 
-    eprintln!("PSC endpoint deleted successfully");
-    Ok(())
+    let json_response = serde_json::to_value(&response).context("Failed to serialize response")?;
+
+    handle_async_response(
+        conn_mgr,
+        profile_name,
+        json_response,
+        async_ops,
+        output_format,
+        query,
+        "PSC endpoint deleted successfully",
+    )
+    .await
 }
 
 async fn get_endpoint_creation_script(
@@ -494,22 +512,18 @@ async fn create_service_aa(
         .await
         .context("Failed to create Active-Active PSC service")?;
 
-    // Convert response to JSON and check for task ID
     let json_response = serde_json::to_value(&response).context("Failed to serialize response")?;
-    if let Some(task_id) = json_response.get("taskId").and_then(|v| v.as_str()) {
-        eprintln!(
-            "Active-Active PSC service creation initiated. Task ID: {}",
-            task_id
-        );
-        eprintln!(
-            "Use 'redisctl cloud task wait {}' to monitor progress",
-            task_id
-        );
-    }
 
-    let data = handle_output(json_response, output_format, query)?;
-    print_formatted_output(data, output_format)?;
-    Ok(())
+    handle_async_response(
+        conn_mgr,
+        profile_name,
+        json_response,
+        async_ops,
+        output_format,
+        query,
+        "Active-Active PSC service created successfully",
+    )
+    .await
 }
 
 async fn delete_service_aa(
@@ -534,13 +548,23 @@ async fn delete_service_aa(
     }
 
     let handler = PscHandler::new(client.clone());
-    handler
+    let response = handler
         .delete_service_active_active(subscription_id)
         .await
         .context("Failed to delete Active-Active PSC service")?;
 
-    eprintln!("Active-Active PSC service deleted successfully");
-    Ok(())
+    let json_response = serde_json::to_value(&response).context("Failed to serialize response")?;
+
+    handle_async_response(
+        conn_mgr,
+        profile_name,
+        json_response,
+        async_ops,
+        output_format,
+        query,
+        "Active-Active PSC service deleted successfully",
+    )
+    .await
 }
 
 // ============================================================================
@@ -588,22 +612,18 @@ async fn create_endpoint_aa(
         .await
         .context("Failed to create Active-Active PSC endpoint")?;
 
-    // Convert response to JSON and check for task ID
     let json_response = serde_json::to_value(&response).context("Failed to serialize response")?;
-    if let Some(task_id) = json_response.get("taskId").and_then(|v| v.as_str()) {
-        eprintln!(
-            "Active-Active PSC endpoint creation initiated. Task ID: {}",
-            task_id
-        );
-        eprintln!(
-            "Use 'redisctl cloud task wait {}' to monitor progress",
-            task_id
-        );
-    }
 
-    let data = handle_output(json_response, output_format, query)?;
-    print_formatted_output(data, output_format)?;
-    Ok(())
+    handle_async_response(
+        conn_mgr,
+        profile_name,
+        json_response,
+        async_ops,
+        output_format,
+        query,
+        "Active-Active PSC endpoint created successfully",
+    )
+    .await
 }
 
 async fn delete_endpoint_aa(
@@ -630,11 +650,21 @@ async fn delete_endpoint_aa(
     }
 
     let handler = PscHandler::new(client.clone());
-    handler
+    let response = handler
         .delete_endpoint_active_active(subscription_id, region_id, endpoint_id)
         .await
         .context("Failed to delete Active-Active PSC endpoint")?;
 
-    eprintln!("Active-Active PSC endpoint deleted successfully");
-    Ok(())
+    let json_response = serde_json::to_value(&response).context("Failed to serialize response")?;
+
+    handle_async_response(
+        conn_mgr,
+        profile_name,
+        json_response,
+        async_ops,
+        output_format,
+        query,
+        "Active-Active PSC endpoint deleted successfully",
+    )
+    .await
 }

crates/redisctl/src/commands/cloud/connectivity/tgw.rs

@@ -37,6 +37,8 @@ pub async fn handle_tgw_command(
             async_ops,
         } => {
             create_attachment(
+                conn_mgr,
+                profile_name,
                 &client,
                 *subscription_id,
                 file,
@@ -243,9 +245,12 @@ async fn list_attachments(
 }
 
 async fn create_attachment(
+    conn_mgr: &ConnectionManager,
+    profile_name: Option<&str>,
     client: &CloudClient,
     subscription_id: i32,
     file: &str,
+    async_ops: &AsyncOperationArgs,
     output_format: OutputFormat,
     query: Option<&str>,
 ) -> CliResult<()> {
@@ -259,25 +264,25 @@ async fn create_attachment(
         .await
         .context("Failed to create TGW attachment")?;
 
-    // Convert response to JSON and check for task ID
     let json_response = serde_json::to_value(&response).context("Failed to serialize response")?;
-    if let Some(task_id) = json_response.get("taskId").and_then(|v| v.as_str()) {
-        eprintln!("TGW attachment creation initiated. Task ID: {}", task_id);
-        eprintln!(
-            "Use 'redisctl cloud task wait {}' to monitor progress",
-            task_id
-        );
-    }
 
-    let data = handle_output(json_response, output_format, query)?;
-    print_formatted_output(data, output_format)?;
-    Ok(())
+    handle_async_response(
+        conn_mgr,
+        profile_name,
+        json_response,
+        async_ops,
+        output_format,
+        query,
+        "TGW attachment created successfully",
+    )
+    .await
 }
 
 async fn create_attachment_with_id(
     client: &CloudClient,
     subscription_id: i32,
     tgw_id: &str,
+    async_ops: &AsyncOperationArgs,
     output_format: OutputFormat,
     query: Option<&str>,
 ) -> CliResult<()> {
@@ -307,6 +312,7 @@ async fn update_attachment_cidrs(
     subscription_id: i32,
     attachment_id: &str,
     file: &str,
+    async_ops: &AsyncOperationArgs,
     output_format: OutputFormat,
     query: Option<&str>,
 ) -> CliResult<()> {
@@ -340,6 +346,9 @@ async fn delete_attachment(
     subscription_id: i32,
     attachment_id: &str,
     yes: bool,
+    async_ops: &AsyncOperationArgs,
+    output_format: OutputFormat,
+    query: Option<&str>,
 ) -> CliResult<()> {
     if !yes {
         let prompt = format!(
@@ -454,6 +463,7 @@ async fn create_attachment_aa(
     subscription_id: i32,
     region_id: i32,
     file: &str,
+    async_ops: &AsyncOperationArgs,
     output_format: OutputFormat,
     query: Option<&str>,
 ) -> CliResult<()> {
@@ -491,6 +501,7 @@ async fn update_attachment_cidrs_aa(
     region_id: i32,
     attachment_id: &str,
     file: &str,
+    async_ops: &AsyncOperationArgs,
     output_format: OutputFormat,
     query: Option<&str>,
 ) -> CliResult<()> {
@@ -533,6 +544,9 @@ async fn delete_attachment_aa(
     region_id: i32,
     attachment_id: &str,
     yes: bool,
+    async_ops: &AsyncOperationArgs,
+    output_format: OutputFormat,
+    query: Option<&str>,
 ) -> CliResult<()> {
     if !yes {
         let prompt = format!(