redisctl_core/enterprise/
progress.rs

1//! Progress tracking and action polling for async Enterprise operations
2//!
3//! Enterprise API operations that are asynchronous return an `Action` which must be polled
4//! until completion. This module provides utilities for that polling
5//! with optional progress callbacks for UI updates.
6
7use crate::error::{CoreError, Result};
8use redis_enterprise::EnterpriseClient;
9use redis_enterprise::actions::Action;
10use std::time::{Duration, Instant};
11
12/// Progress events emitted during async Enterprise operations
13#[derive(Debug, Clone)]
14pub enum EnterpriseProgressEvent {
15    /// Action has been created/started
16    Started { action_uid: String },
17    /// Polling iteration with current status
18    Polling {
19        action_uid: String,
20        status: String,
21        progress: Option<f32>,
22        elapsed: Duration,
23    },
24    /// Action completed successfully
25    Completed { action_uid: String },
26    /// Action failed
27    Failed { action_uid: String, error: String },
28}
29
30/// Callback type for Enterprise progress updates
31///
32/// CLI can use this to update spinners/progress bars.
33/// MCP typically doesn't need this.
34pub type EnterpriseProgressCallback = Box<dyn Fn(EnterpriseProgressEvent) + Send + Sync>;
35
36/// Poll an Enterprise action until completion
37///
38/// # Arguments
39///
40/// * `client` - The Enterprise API client
41/// * `action_uid` - The action UID to poll
42/// * `timeout` - Maximum time to wait for completion
43/// * `interval` - Time between polling attempts
44/// * `on_progress` - Optional callback for progress updates
45///
46/// # Returns
47///
48/// The completed action, or an error if the action failed or timed out.
49///
50/// # Example
51///
52/// ```rust,ignore
53/// use redisctl_core::enterprise::{poll_action, EnterpriseProgressEvent};
54/// use std::time::Duration;
55///
56/// // Start an async operation (returns an action_uid)
57/// let action_uid = "some-action-uid";
58///
59/// // Poll with progress callback
60/// let completed = poll_action(
61///     &client,
62///     action_uid,
63///     Duration::from_secs(600),
64///     Duration::from_secs(5),
65///     Some(Box::new(|event| {
66///         match event {
67///             EnterpriseProgressEvent::Polling { status, progress, elapsed, .. } => {
68///                 println!("Status: {} ({:?}%) ({:.0}s)", status, progress, elapsed.as_secs());
69///             }
70///             EnterpriseProgressEvent::Completed { .. } => {
71///                 println!("Done!");
72///             }
73///             _ => {}
74///         }
75///     })),
76/// ).await?;
77/// ```
78pub async fn poll_action(
79    client: &EnterpriseClient,
80    action_uid: &str,
81    timeout: Duration,
82    interval: Duration,
83    on_progress: Option<EnterpriseProgressCallback>,
84) -> Result<Action> {
85    let start = Instant::now();
86    let handler = client.actions();
87
88    emit(
89        &on_progress,
90        EnterpriseProgressEvent::Started {
91            action_uid: action_uid.to_string(),
92        },
93    );
94
95    loop {
96        let elapsed = start.elapsed();
97        if elapsed > timeout {
98            return Err(CoreError::TaskTimeout(timeout));
99        }
100
101        let action = handler.get(action_uid).await?;
102        let status = action.status.clone();
103
104        emit(
105            &on_progress,
106            EnterpriseProgressEvent::Polling {
107                action_uid: action_uid.to_string(),
108                status: status.clone(),
109                progress: action.progress,
110                elapsed,
111            },
112        );
113
114        match status.as_str() {
115            "completed" => {
116                emit(
117                    &on_progress,
118                    EnterpriseProgressEvent::Completed {
119                        action_uid: action_uid.to_string(),
120                    },
121                );
122                return Ok(action);
123            }
124            "failed" | "cancelled" => {
125                let error = action
126                    .error
127                    .clone()
128                    .unwrap_or_else(|| format!("Action {}", status));
129
130                emit(
131                    &on_progress,
132                    EnterpriseProgressEvent::Failed {
133                        action_uid: action_uid.to_string(),
134                        error: error.clone(),
135                    },
136                );
137                return Err(CoreError::TaskFailed(error));
138            }
139            // 'queued', 'starting', 'running', 'cancelling' - still in progress
140            _ => {
141                tokio::time::sleep(interval).await;
142            }
143        }
144    }
145}
146
147/// Helper to emit progress events
148fn emit(callback: &Option<EnterpriseProgressCallback>, event: EnterpriseProgressEvent) {
149    if let Some(cb) = callback {
150        cb(event);
151    }
152}
redisctl_core/enterprise/progress.rs

redisctl_core/enterprise/
progress.rs
