feat: implement workflow system with enterprise init-cluster workflow

.env.example

@@ -0,0 +1,14 @@
+# Docker Compose environment variables for Redis Enterprise
+
+# Redis Enterprise Docker image
+# Default: redislabs/redis:latest
+REDIS_ENTERPRISE_IMAGE=redislabs/redis:latest
+
+# Docker platform
+# Default: linux/amd64
+# For ARM64 (Apple Silicon): linux/arm64
+REDIS_ENTERPRISE_PLATFORM=linux/amd64
+
+# Note: For ARM64 systems (Apple Silicon Macs), you'll need to:
+# 1. Use an ARM64-compatible Redis Enterprise image
+# 2. Set REDIS_ENTERPRISE_PLATFORM=linux/arm64

crates/redis-enterprise/src/client.rs

@@ -478,4 +478,26 @@ impl EnterpriseClient {
             }
         }
     }
+
+    /// Execute a Redis command on a specific database (internal use only)
+    /// This uses the /v1/bdbs/{uid}/command endpoint which may not be publicly documented
+    pub async fn execute_command(&self, db_uid: u32, command: &str) -> Result<serde_json::Value> {
+        let url = format!("{}/v1/bdbs/{}/command", self.base_url, db_uid);
+        let body = serde_json::json!({
+            "command": command
+        });
+
+        debug!("Executing command on database {}: {}", db_uid, command);
+
+        let response = self
+            .client
+            .post(&url)
+            .basic_auth(&self.username, Some(&self.password))
+            .json(&body)
+            .send()
+            .await
+            .map_err(|e| self.map_reqwest_error(e, &url))?;
+
+        self.handle_response(response).await
+    }
 }

crates/redisctl/src/cli.rs

@@ -1000,11 +1000,54 @@ pub enum EnterpriseCommands {
     #[command(subcommand)]
     Module(crate::commands::enterprise::module::ModuleCommands),
 
+    /// Workflow operations for multi-step tasks
+    #[command(subcommand)]
+    Workflow(EnterpriseWorkflowCommands),
+
     /// Statistics and metrics operations
     #[command(subcommand)]
     Stats(EnterpriseStatsCommands),
 }
 
+/// Enterprise workflow commands
+#[derive(Debug, Subcommand)]
+pub enum EnterpriseWorkflowCommands {
+    /// List available workflows
+    List,
+
+    /// Initialize a Redis Enterprise cluster
+    #[command(name = "init-cluster")]
+    InitCluster {
+        /// Cluster name
+        #[arg(long, default_value = "redis-cluster")]
+        name: String,
+
+        /// Admin username
+        #[arg(long, default_value = "admin@redis.local")]
+        username: String,
+
+        /// Admin password (required)
+        #[arg(long, env = "REDIS_ENTERPRISE_INIT_PASSWORD")]
+        password: String,
+
+        /// Skip creating a default database after initialization
+        #[arg(long)]
+        skip_database: bool,
+
+        /// Name for the default database
+        #[arg(long, default_value = "default-db")]
+        database_name: String,
+
+        /// Memory size for the default database in GB
+        #[arg(long, default_value = "1")]
+        database_memory_gb: i64,
+
+        /// Async operation options
+        #[command(flatten)]
+        async_ops: crate::commands::cloud::async_utils::AsyncOperationArgs,
+    },
+}
+
 // Placeholder command structures - will be expanded in later PRs
 
 #[derive(Subcommand, Debug)]

crates/redisctl/src/connection.rs

@@ -7,6 +7,7 @@ use tracing::{debug, info, trace};
 
 /// Connection manager for creating authenticated clients
 #[allow(dead_code)] // Used by binary target
+#[derive(Clone)]
 pub struct ConnectionManager {
     pub config: Config,
 }

crates/redisctl/src/main.rs

@@ -10,6 +10,7 @@ mod config;
 mod connection;
 mod error;
 mod output;
+mod workflows;
 
 use cli::{Cli, Commands};
 use config::Config;
@@ -255,6 +256,9 @@ async fn execute_enterprise_command(
             )
             .await
         }
+        Workflow(workflow_cmd) => {
+            handle_enterprise_workflow_command(conn_mgr, profile, workflow_cmd, output).await
+        }
         Stats(stats_cmd) => {
             commands::enterprise::stats::handle_stats_command(
                 conn_mgr, profile, stats_cmd, output, query,
@@ -264,6 +268,127 @@ async fn execute_enterprise_command(
     }
 }
 
+async fn handle_enterprise_workflow_command(
+    conn_mgr: &ConnectionManager,
+    profile: Option<&str>,
+    workflow_cmd: &cli::EnterpriseWorkflowCommands,
+    output: cli::OutputFormat,
+) -> Result<(), RedisCtlError> {
+    use cli::EnterpriseWorkflowCommands::*;
+    use workflows::{WorkflowArgs, WorkflowContext, WorkflowRegistry};
+
+    match workflow_cmd {
+        List => {
+            let registry = WorkflowRegistry::new();
+            let workflows = registry.list();
+
+            match output {
+                cli::OutputFormat::Json | cli::OutputFormat::Yaml => {
+                    let workflow_list: Vec<serde_json::Value> = workflows
+                        .into_iter()
+                        .map(|(name, description)| {
+                            serde_json::json!({
+                                "name": name,
+                                "description": description
+                            })
+                        })
+                        .collect();
+                    let output_format = match output {
+                        cli::OutputFormat::Json => output::OutputFormat::Json,
+                        cli::OutputFormat::Yaml => output::OutputFormat::Yaml,
+                        _ => output::OutputFormat::Table,
+                    };
+                    crate::output::print_output(
+                        serde_json::json!(workflow_list),
+                        output_format,
+                        None,
+                    )?;
+                }
+                _ => {
+                    println!("Available Enterprise Workflows:");
+                    println!();
+                    for (name, description) in workflows {
+                        println!("  {} - {}", name, description);
+                    }
+                }
+            }
+            Ok(())
+        }
+        InitCluster {
+            name,
+            username,
+            password,
+            skip_database,
+            database_name,
+            database_memory_gb,
+            async_ops,
+        } => {
+            let mut args = WorkflowArgs::new();
+            args.insert("name", name);
+            args.insert("username", username);
+            args.insert("password", password);
+            args.insert("create_database", !skip_database);
+            args.insert("database_name", database_name);
+            args.insert("database_memory_gb", database_memory_gb);
+
+            let output_format = match output {
+                cli::OutputFormat::Json => output::OutputFormat::Json,
+                cli::OutputFormat::Yaml => output::OutputFormat::Yaml,
+                cli::OutputFormat::Table | cli::OutputFormat::Auto => output::OutputFormat::Table,
+            };
+
+            let context = WorkflowContext {
+                conn_mgr: conn_mgr.clone(),
+                profile_name: profile.map(String::from),
+                output_format,
+                wait_timeout: if async_ops.wait {
+                    async_ops.wait_timeout
+                } else {
+                    0
+                },
+            };
+
+            let registry = WorkflowRegistry::new();
+            let workflow = registry
+                .get("init-cluster")
+                .ok_or_else(|| RedisCtlError::ApiError {
+                    message: "Workflow not found".to_string(),
+                })?;
+
+            let result =
+                workflow
+                    .execute(context, args)
+                    .await
+                    .map_err(|e| RedisCtlError::ApiError {
+                        message: e.to_string(),
+                    })?;
+
+            if !result.success {
+                return Err(RedisCtlError::ApiError {
+                    message: result.message,
+                });
+            }
+
+            // Print result as JSON/YAML if requested
+            match output {
+                cli::OutputFormat::Json | cli::OutputFormat::Yaml => {
+                    let result_json = serde_json::json!({
+                        "success": result.success,
+                        "message": result.message,
+                        "outputs": result.outputs,
+                    });
+                    crate::output::print_output(&result_json, output_format, None)?;
+                }
+                _ => {
+                    // Human output was already printed by the workflow
+                }
+            }
+
+            Ok(())
+        }
+    }
+}
+
 async fn execute_profile_command(
     profile_cmd: &cli::ProfileCommands,
     conn_mgr: &ConnectionManager,