feat(enterprise): add database watch command for real-time status monitoring (#458)

Implements Phase 2 of issue #405 - Database Lifecycle Event Monitoring

## Library Changes (redis-enterprise)
- Add watch_database() method to DatabaseHandler/BdbHandler
- Polls database endpoint and tracks status changes
- Returns stream of (DatabaseInfo, Option<String>) tuples
- Second element is Some(old_status) when status changes occur

## CLI Changes (redisctl)
- New command: redisctl enterprise database watch <id>
- Displays timestamped status updates with key metrics
- Highlights status transitions prominently
- Supports JMESPath filtering via -q flag
- Ctrl+C for graceful shutdown
- Configurable poll interval (default: 5 seconds)

## Use Cases
- Monitor database during upgrades/migrations
- Watch backup/restore operations
- Track scaling operations
- Detect and alert on status changes

## Example Usage
```bash
# Watch database with default 5-second polling
redisctl enterprise database watch 1

# Faster polling for critical operations
redisctl enterprise database watch 1 --poll-interval 2

# Filter output with JMESPath
redisctl enterprise database watch 1 -q '{name: name, status: status}'
```

## Example Output
```
Watching database 1 (Ctrl+C to stop)...

[10:30:00] Database 1: active | mem: 2.10GB | shards: 1
[10:30:15] Database 1: active -> updating (TRANSITION)
  Memory: 2.10GB / 5.00GB (42.0%)
  Shards: 1
[10:32:30] Database 1: updating -> active (TRANSITION)
  Memory: 2.15GB / 5.00GB (43.0%)
  Shards: 1
```

Addresses #405 (Phase 2)

Author: joshrotenberg

crates/redis-enterprise/src/bdb.rs

@@ -74,13 +74,19 @@
 
 use crate::client::RestClient;
 use crate::error::Result;
+use futures::stream::Stream;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
+use std::pin::Pin;
+use std::time::Duration;
+use tokio::time::sleep;
 use typed_builder::TypedBuilder;
 
 // Aliases for easier use
 pub type Database = DatabaseInfo;
 pub type BdbHandler = DatabaseHandler;
+pub type DatabaseWatchStream<'a> =
+    Pin<Box<dyn Stream<Item = Result<(DatabaseInfo, Option<String>)>> + Send + 'a>>;
 
 /// Response from database action operations
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -981,4 +987,84 @@ impl DatabaseHandler {
     pub async fn create_v2(&self, request: Value) -> Result<DatabaseInfo> {
         self.client.post("/v2/bdbs", &request).await
     }
+
+    /// Watch database status changes in real-time
+    ///
+    /// Polls the database endpoint and yields updates when status changes occur.
+    /// Useful for monitoring database operations like upgrades, migrations, backups, etc.
+    ///
+    /// # Arguments
+    /// * `uid` - Database ID to watch
+    /// * `poll_interval` - Time to wait between polls
+    ///
+    /// # Returns
+    /// A stream of `(DatabaseInfo, Option<String>)` tuples where:
+    /// - `DatabaseInfo` - Current database state
+    /// - `Option<String>` - Previous status (None on first poll, Some on status change)
+    ///
+    /// # Example
+    /// ```no_run
+    /// use redis_enterprise::{EnterpriseClient, BdbHandler as DatabaseHandler};
+    /// use futures::StreamExt;
+    /// use std::time::Duration;
+    ///
+    /// # async fn example(client: EnterpriseClient) -> Result<(), Box<dyn std::error::Error>> {
+    /// let handler = DatabaseHandler::new(client);
+    /// let mut stream = handler.watch_database(1, Duration::from_secs(5));
+    ///
+    /// while let Some(result) = stream.next().await {
+    ///     match result {
+    ///         Ok((db_info, prev_status)) => {
+    ///             if let Some(old_status) = prev_status {
+    ///                 println!("Status changed: {} -> {}", old_status, db_info.status.unwrap_or_default());
+    ///             } else {
+    ///                 println!("Initial status: {}", db_info.status.unwrap_or_default());
+    ///             }
+    ///         }
+    ///         Err(e) => eprintln!("Error: {}", e),
+    ///     }
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn watch_database(&self, uid: u32, poll_interval: Duration) -> DatabaseWatchStream<'_> {
+        Box::pin(async_stream::stream! {
+            let mut last_status: Option<String> = None;
+
+            loop {
+                match self.info(uid).await {
+                    Ok(db_info) => {
+                        let current_status = db_info.status.clone();
+
+                        // Check if status changed
+                        let status_changed = match (&last_status, &current_status) {
+                            (Some(old), Some(new)) => old != new,
+                            (None, Some(_)) => false, // First poll, not a change
+                            (Some(_), None) => true,  // Status disappeared
+                            (None, None) => false,
+                        };
+
+                        // Yield the database info with previous status if changed
+                        if status_changed {
+                            yield Ok((db_info, last_status.clone()));
+                        } else if last_status.is_none() {
+                            // First poll - always yield
+                            yield Ok((db_info, None));
+                        } else {
+                            // Status unchanged - yield current state for monitoring
+                            yield Ok((db_info, None));
+                        }
+
+                        last_status = current_status;
+                    }
+                    Err(e) => {
+                        yield Err(e);
+                        break;
+                    }
+                }
+
+                sleep(poll_interval).await;
+            }
+        })
+    }
 }

crates/redisctl/src/cli/enterprise.rs

@@ -451,6 +451,15 @@ NOTE: Memory size is in bytes. Common values:
         force: bool,
     },
 
+    /// Watch database status changes in real-time
+    Watch {
+        /// Database ID
+        id: u32,
+        /// Poll interval in seconds
+        #[arg(long, default_value = "5")]
+        poll_interval: u64,
+    },
+
     /// Export database
     Export {
         /// Database ID

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

@@ -74,6 +74,9 @@ pub async fn handle_database_command(
             )
             .await
         }
+        EnterpriseDatabaseCommands::Watch { id, poll_interval } => {
+            database_impl::watch_database(conn_mgr, profile_name, *id, *poll_interval, query).await
+        }
         EnterpriseDatabaseCommands::Export { id, data } => {
             database_impl::export_database(conn_mgr, profile_name, *id, data, output_format, query)
                 .await

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

@@ -208,6 +208,108 @@ pub async fn delete_database(
     Ok(())
 }
 
+/// Watch database status changes
+pub async fn watch_database(
+    conn_mgr: &ConnectionManager,
+    profile_name: Option<&str>,
+    id: u32,
+    poll_interval: u64,
+    query: Option<&str>,
+) -> CliResult<()> {
+    use futures::StreamExt;
+    use tokio::signal;
+
+    let client = conn_mgr.create_enterprise_client(profile_name).await?;
+    let handler = redis_enterprise::BdbHandler::new(client);
+    let mut stream = handler.watch_database(id, std::time::Duration::from_secs(poll_interval));
+
+    println!("Watching database {} (Ctrl+C to stop)...\n", id);
+
+    loop {
+        tokio::select! {
+            _ = signal::ctrl_c() => {
+                println!("\nStopping database watch...");
+                break;
+            }
+            result = stream.next() => {
+                match result {
+                    Some(Ok((db_info, prev_status))) => {
+                        let current_status = db_info.status.as_deref().unwrap_or("unknown");
+                        let timestamp = chrono::Utc::now().format("%H:%M:%S");
+
+                        // Check if this is a status change
+                        if let Some(old_status) = prev_status {
+                            // Status transition detected
+                            println!(
+                                "[{}] Database {}: {} -> {} (TRANSITION)",
+                                timestamp, id, old_status, current_status
+                            );
+
+                            // Show key metrics during transition
+                            if let Some(memory_used) = db_info.memory_used
+                                && let Some(memory_size) = db_info.memory_size {
+                                    let usage_pct = (memory_used as f64 / memory_size as f64) * 100.0;
+                                    println!("  Memory: {} / {} ({:.1}%)",
+                                        format_bytes(memory_used),
+                                        format_bytes(memory_size),
+                                        usage_pct
+                                    );
+                                }
+
+                            if let Some(shards) = db_info.shards_count {
+                                println!("  Shards: {}", shards);
+                            }
+                        } else {
+                            // Regular status update (no change)
+                            print!("[{}] Database {}: {}", timestamp, id, current_status);
+
+                            // Apply JMESPath query if provided
+                            if query.is_some() {
+                                let db_json = serde_json::to_value(&db_info)
+                                    .map_err(|e| RedisCtlError::from(anyhow::anyhow!("Serialization error: {}", e)))?;
+                                let filtered = handle_output(db_json, OutputFormat::Json, query)?;
+                                print!(" | {}", serde_json::to_string(&filtered)?);
+                            } else {
+                                // Show brief metrics
+                                if let Some(memory_used) = db_info.memory_used {
+                                    print!(" | mem: {}", format_bytes(memory_used));
+                                }
+                                if let Some(shards) = db_info.shards_count {
+                                    print!(" | shards: {}", shards);
+                                }
+                            }
+                            println!();
+                        }
+                    }
+                    Some(Err(e)) => {
+                        eprintln!("Error watching database: {}", e);
+                        break;
+                    }
+                    None => {
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    Ok(())
+}
+
+/// Format bytes into human-readable format
+fn format_bytes(bytes: u64) -> String {
+    const UNITS: &[&str] = &["B", "KB", "MB", "GB", "TB"];
+    let mut size = bytes as f64;
+    let mut unit_index = 0;
+
+    while size >= 1024.0 && unit_index < UNITS.len() - 1 {
+        size /= 1024.0;
+        unit_index += 1;
+    }
+
+    format!("{:.2}{}", size, UNITS[unit_index])
+}
+
 /// Export database
 pub async fn export_database(
     conn_mgr: &ConnectionManager,