feat: add database upgrade command for Redis version upgrades (#442)

* feat: add database upgrade command for Redis version upgrades

- Add DatabaseUpgradeRequest and ModuleUpgrade types to redis-enterprise library
- Add upgrade_redis_version() method to DatabaseHandler with full parameter support
- Implement 'redisctl enterprise database upgrade' CLI command
- Add safety checks for database status, persistence, and replication
- Support --version, --preserve-roles, --force-restart and other upgrade options
- Add structured JSON output support
- Add CLI test for upgrade command help
- Addresses issue #422

* test: add integration test for database upgrade_redis_version method

* test: update upgrade test with real API response from Docker cluster

The actual API returns the full BDB object with action_uid embedded,
not just action_uid + description. Updated mock to match real response
captured from Docker cluster test.

* fix(clippy): use derive for Default impl on OutputFormat

Rust 1.91 introduced a new lint 'derivable_impls' that flags manual
Default implementations that can be derived. Changed OutputFormat to
use #[derive(Default)] with #[default] attribute on the Json variant.

Same fix as applied to PR #444 to pass CI on Rust 1.91.0.

* fix(redis-enterprise): correct doctest imports for upgrade_redis_version

Use EnterpriseClient and BdbHandler which are properly exported from the crate root, instead of internal Client type.

Author: joshrotenberg

crates/redis-enterprise/src/bdb.rs

@@ -133,6 +133,55 @@ pub struct ExportResponse {
     pub extra: Value,
 }
 
+/// Module information for database upgrade
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ModuleUpgrade {
+    /// Module name
+    pub module_name: String,
+    /// Module version
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub new_version: Option<String>,
+    /// Module arguments
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub module_args: Option<String>,
+}
+
+/// Request for database upgrade operation
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct DatabaseUpgradeRequest {
+    /// Target Redis version (optional, defaults to latest)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub redis_version: Option<String>,
+
+    /// Preserve master/replica roles (requires extra failover)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub preserve_roles: Option<bool>,
+
+    /// Restart shards even if no version change
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub force_restart: Option<bool>,
+
+    /// Allow data loss in non-replicated, non-persistent databases
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub may_discard_data: Option<bool>,
+
+    /// Force data discard even if replicated/persistent
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub force_discard: Option<bool>,
+
+    /// Keep current CRDT protocol version
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub keep_crdt_protocol_version: Option<bool>,
+
+    /// Maximum parallel shard upgrades (default: all shards)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub parallel_shards_upgrade: Option<u32>,
+
+    /// Modules to upgrade alongside Redis
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub modules: Option<Vec<ModuleUpgrade>>,
+}
+
 /// Database information from the REST API - 100% field coverage (152/152 fields)
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DatabaseInfo {
@@ -856,6 +905,50 @@ impl DatabaseHandler {
             .await
     }
 
+    /// Upgrade database Redis version and/or modules (BDB.UPGRADE)
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # use redis_enterprise::EnterpriseClient;
+    /// # use redis_enterprise::bdb::{BdbHandler, DatabaseUpgradeRequest};
+    /// # async fn example() -> redis_enterprise::Result<()> {
+    /// let client = EnterpriseClient::builder()
+    ///     .base_url("https://localhost:9443")
+    ///     .username("admin")
+    ///     .password("password")
+    ///     .insecure(true)
+    ///     .build()?;
+    /// let db_handler = BdbHandler::new(client);
+    ///
+    /// // Upgrade to latest Redis version
+    /// let request = DatabaseUpgradeRequest {
+    ///     redis_version: None,  // defaults to latest
+    ///     preserve_roles: Some(true),
+    ///     ..Default::default()
+    /// };
+    /// db_handler.upgrade_redis_version(1, request).await?;
+    ///
+    /// // Upgrade to specific Redis version
+    /// let request = DatabaseUpgradeRequest {
+    ///     redis_version: Some("7.4.2".to_string()),
+    ///     preserve_roles: Some(true),
+    ///     ..Default::default()
+    /// };
+    /// db_handler.upgrade_redis_version(1, request).await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn upgrade_redis_version(
+        &self,
+        uid: u32,
+        request: DatabaseUpgradeRequest,
+    ) -> Result<DatabaseActionResponse> {
+        self.client
+            .post(&format!("/v1/bdbs/{}/upgrade", uid), &request)
+            .await
+    }
+
     /// Reset database password (BDB.RESET_PASSWORD)
     pub async fn reset_password(
         &self,

crates/redis-enterprise/tests/database_tests.rs

@@ -453,3 +453,62 @@ async fn test_passwords_delete_and_reset_status() {
     let reset = handler.backup_reset_status(1).await.unwrap();
     assert_eq!(reset["status"], "reset");
 }
+
+#[tokio::test]
+async fn test_database_upgrade_redis_version() {
+    let mock_server = MockServer::start().await;
+
+    // Mock data captured from: curl -k -u "[email protected]:Redis123!" -X POST \
+    //   -H "Content-Type: application/json" -d '{"force_restart": true}' \
+    //   https://localhost:9443/v1/bdbs/1/upgrade
+    // Real API returns full BDB object with action_uid embedded
+    Mock::given(method("POST"))
+        .and(path("/v1/bdbs/1/upgrade"))
+        .and(basic_auth("admin", "password"))
+        .respond_with(success_response(json!({
+            "action_uid": "591d9dcb-ddd7-48a9-a04d-bd5d4d6834d0",
+            "uid": 1,
+            "name": "test-db",
+            "status": "active",
+            "redis_version": "7.4",
+            "version": "7.4.2",
+            "memory_size": 1073741824,
+            "type": "redis",
+            "replication": false,
+            "persistence": "disabled",
+            "port": 18367,
+            "shards_count": 1,
+            "oss_cluster": false
+        })))
+        .mount(&mock_server)
+        .await;
+
+    let client = EnterpriseClient::builder()
+        .base_url(mock_server.uri())
+        .username("admin")
+        .password("password")
+        .build()
+        .unwrap();
+
+    let handler = BdbHandler::new(client);
+
+    let request = redis_enterprise::bdb::DatabaseUpgradeRequest {
+        redis_version: Some("7.4.2".to_string()),
+        preserve_roles: Some(true),
+        force_restart: Some(false),
+        may_discard_data: Some(false),
+        force_discard: Some(false),
+        keep_crdt_protocol_version: Some(false),
+        parallel_shards_upgrade: None,
+        modules: None,
+    };
+
+    let result = handler.upgrade_redis_version(1, request).await;
+    assert!(result.is_ok());
+    let response = result.unwrap();
+    assert_eq!(response.action_uid, "591d9dcb-ddd7-48a9-a04d-bd5d4d6834d0");
+    // Verify the flattened extra fields are captured
+    assert_eq!(response.extra["uid"], 1);
+    assert_eq!(response.extra["name"], "test-db");
+    assert_eq!(response.extra["redis_version"], "7.4");
+}

crates/redisctl/src/cli.rs

@@ -2196,6 +2196,36 @@ pub enum EnterpriseDatabaseCommands {
         data: String,
     },
 
+    /// Upgrade database Redis version
+    Upgrade {
+        /// Database ID
+        id: u32,
+        /// Target Redis version (defaults to latest)
+        #[arg(long)]
+        version: Option<String>,
+        /// Preserve master/replica roles (requires extra failover)
+        #[arg(long)]
+        preserve_roles: bool,
+        /// Restart shards even if no version change
+        #[arg(long)]
+        force_restart: bool,
+        /// Allow data loss in non-replicated, non-persistent databases
+        #[arg(long)]
+        may_discard_data: bool,
+        /// Force data discard even if replicated/persistent
+        #[arg(long)]
+        force_discard: bool,
+        /// Keep current CRDT protocol version
+        #[arg(long)]
+        keep_crdt_protocol_version: bool,
+        /// Maximum parallel shard upgrades
+        #[arg(long)]
+        parallel_shards_upgrade: Option<u32>,
+        /// Skip confirmation prompt
+        #[arg(long)]
+        force: bool,
+    },
+
     /// Get ACL configuration
     GetAcl {
         /// Database ID

crates/redisctl/src/commands/enterprise/database.rs

@@ -98,6 +98,34 @@ pub async fn handle_database_command(
             )
             .await
         }
+        EnterpriseDatabaseCommands::Upgrade {
+            id,
+            version,
+            preserve_roles,
+            force_restart,
+            may_discard_data,
+            force_discard,
+            keep_crdt_protocol_version,
+            parallel_shards_upgrade,
+            force,
+        } => {
+            database_impl::upgrade_database(
+                conn_mgr,
+                profile_name,
+                *id,
+                version.as_deref(),
+                *preserve_roles,
+                *force_restart,
+                *may_discard_data,
+                *force_discard,
+                *keep_crdt_protocol_version,
+                *parallel_shards_upgrade,
+                *force,
+                output_format,
+                query,
+            )
+            .await
+        }
         EnterpriseDatabaseCommands::GetAcl { id } => {
             database_impl::get_database_acl(conn_mgr, profile_name, *id, output_format, query).await
         }

crates/redisctl/src/commands/enterprise/database_impl.rs

@@ -440,3 +440,129 @@ pub async fn get_database_clients(
     print_formatted_output(data, output_format)?;
     Ok(())
 }
+
+/// Upgrade database Redis version
+#[allow(clippy::too_many_arguments)]
+pub async fn upgrade_database(
+    conn_mgr: &ConnectionManager,
+    profile_name: Option<&str>,
+    id: u32,
+    version: Option<&str>,
+    preserve_roles: bool,
+    force_restart: bool,
+    may_discard_data: bool,
+    force_discard: bool,
+    keep_crdt_protocol_version: bool,
+    parallel_shards_upgrade: Option<u32>,
+    force: bool,
+    output_format: OutputFormat,
+    query: Option<&str>,
+) -> CliResult<()> {
+    use redis_enterprise::bdb::{DatabaseHandler, DatabaseInfo, DatabaseUpgradeRequest};
+
+    let client = conn_mgr.create_enterprise_client(profile_name).await?;
+
+    // Get current database info
+    let db_handler = DatabaseHandler::new(client);
+    let db: DatabaseInfo = db_handler.get(id).await?;
+    let current_version = db.redis_version.as_deref().unwrap_or("unknown");
+
+    // Determine target version
+    let target_version = if let Some(v) = version {
+        v.to_string()
+    } else {
+        // Get latest version from cluster - for now just use current
+        // TODO: Get from cluster info when we add that endpoint
+        current_version.to_string()
+    };
+
+    // Safety checks unless --force
+    if !force {
+        // Check if database is active
+        if db.status.as_deref() != Some("active") {
+            return Err(RedisCtlError::InvalidInput {
+                message: format!(
+                    "Database is not active (status: {}). Use --force to upgrade anyway.",
+                    db.status.as_deref().unwrap_or("unknown")
+                ),
+            });
+        }
+
+        // Warn about persistence (check if persistence is disabled/none)
+        let has_persistence = db
+            .persistence
+            .as_deref()
+            .map(|p| p != "disabled")
+            .unwrap_or(false);
+        if !has_persistence && !may_discard_data {
+            eprintln!("Warning: Database has no persistence enabled.");
+            eprintln!("If upgrade fails, data may be lost.");
+            eprintln!("Use --may-discard-data to proceed.");
+            return Err(RedisCtlError::InvalidInput {
+                message: "Upgrade cancelled for safety".to_string(),
+            });
+        }
+
+        // Warn about replication (check if replication is enabled)
+        let has_replication = db.replication.unwrap_or(false);
+        if !has_replication {
+            eprintln!("Warning: Database has no replication enabled.");
+            eprintln!("Upgrade will cause downtime.");
+            eprintln!("Use --force to proceed.");
+            return Err(RedisCtlError::InvalidInput {
+                message: "Upgrade cancelled for safety".to_string(),
+            });
+        }
+    }
+
+    // Display upgrade info
+    if matches!(output_format, OutputFormat::Table | OutputFormat::Auto) {
+        println!("Upgrading database '{}' (db:{})...", db.name, id);
+        println!("  Current version: {}", current_version);
+        println!("  Target version: {}", target_version);
+    }
+
+    // Build upgrade request
+    let request = DatabaseUpgradeRequest {
+        redis_version: Some(target_version.clone()),
+        preserve_roles: Some(preserve_roles),
+        force_restart: Some(force_restart),
+        may_discard_data: Some(may_discard_data),
+        force_discard: Some(force_discard),
+        keep_crdt_protocol_version: Some(keep_crdt_protocol_version),
+        parallel_shards_upgrade,
+        modules: None,
+    };
+
+    // Call upgrade API
+    let response = db_handler.upgrade_redis_version(id, request).await?;
+
+    // Handle output
+    match output_format {
+        OutputFormat::Json => {
+            let output = serde_json::json!({
+                "database_id": id,
+                "database_name": db.name,
+                "old_version": current_version,
+                "new_version": target_version,
+                "action_uid": response.action_uid,
+                "status": "upgrade_initiated"
+            });
+            println!("{}", serde_json::to_string_pretty(&output)?);
+        }
+        OutputFormat::Table | OutputFormat::Auto => {
+            println!("Upgrade initiated (action_uid: {})", response.action_uid);
+            println!(
+                "Use 'redisctl enterprise database get {}' to check status",
+                id
+            );
+        }
+        _ => {
+            let data = serde_json::to_value(&response)?;
+            let filtered = handle_output(data, output_format, query)?;
+            print_formatted_output(filtered, output_format)?;
+        }
+    }
+
+    Ok(())
+}

crates/redisctl/tests/cli_basic_tests.rs

@@ -258,6 +258,20 @@ fn test_profile_remove_missing_name() {
         .stderr(predicate::str::contains("required"));
 }
 
+#[test]
+fn test_enterprise_database_upgrade_help() {
+    redisctl()
+        .arg("enterprise")
+        .arg("database")
+        .arg("upgrade")
+        .arg("--help")
+        .assert()
+        .success()
+        .stdout(predicate::str::contains("Upgrade database Redis version"))
+        .stdout(predicate::str::contains("--version"))
+        .stdout(predicate::str::contains("--preserve-roles"));
+}
+
 #[test]
 fn test_payment_method_help() {
     redisctl()