redis_cloud/client.rs
1//! Redis Cloud API client core implementation
2//!
3//! This module contains the core HTTP client for interacting with the Redis Cloud REST API.
4//! It provides authentication handling, request/response processing, and error management.
5//!
6//! The client is designed around a builder pattern for flexible configuration and supports
7//! both typed and untyped API interactions.
8
9use crate::{CloudError as RestError, Result};
10use reqwest::Client;
11use reqwest::header::{HeaderMap, HeaderValue, USER_AGENT};
12use serde::Serialize;
13use std::sync::Arc;
14use tracing::{debug, instrument, trace};
15
16/// Default user agent for the Redis Cloud client
17const DEFAULT_USER_AGENT: &str = concat!("redis-cloud/", env!("CARGO_PKG_VERSION"));
18
19/// Builder for constructing a `CloudClient` with custom configuration
20///
21/// Provides a fluent interface for configuring API credentials, base URL, timeouts,
22/// and other client settings before creating the final `CloudClient` instance.
23///
24/// # Examples
25///
26/// ```rust,no_run
27/// use redis_cloud::CloudClient;
28///
29/// // Basic configuration
30/// let client = CloudClient::builder()
31/// .api_key("your-api-key")
32/// .api_secret("your-api-secret")
33/// .build()?;
34///
35/// // Advanced configuration
36/// let client = CloudClient::builder()
37/// .api_key("your-api-key")
38/// .api_secret("your-api-secret")
39/// .base_url("https://api.redislabs.com/v1".to_string())
40/// .timeout(std::time::Duration::from_secs(120))
41/// .build()?;
42/// # Ok::<(), Box<dyn std::error::Error>>(())
43/// ```
44#[derive(Debug, Clone)]
45pub struct CloudClientBuilder {
46 api_key: Option<String>,
47 api_secret: Option<String>,
48 base_url: String,
49 timeout: std::time::Duration,
50 user_agent: String,
51}
52
53impl Default for CloudClientBuilder {
54 fn default() -> Self {
55 Self {
56 api_key: None,
57 api_secret: None,
58 base_url: "https://api.redislabs.com/v1".to_string(),
59 timeout: std::time::Duration::from_secs(30),
60 user_agent: DEFAULT_USER_AGENT.to_string(),
61 }
62 }
63}
64
65impl CloudClientBuilder {
66 /// Create a new builder
67 #[must_use]
68 pub fn new() -> Self {
69 Self::default()
70 }
71
72 /// Set the API key
73 #[must_use]
74 pub fn api_key(mut self, key: impl Into<String>) -> Self {
75 self.api_key = Some(key.into());
76 self
77 }
78
79 /// Set the API secret
80 #[must_use]
81 pub fn api_secret(mut self, secret: impl Into<String>) -> Self {
82 self.api_secret = Some(secret.into());
83 self
84 }
85
86 /// Set the base URL
87 #[must_use]
88 pub fn base_url(mut self, url: impl Into<String>) -> Self {
89 self.base_url = url.into();
90 self
91 }
92
93 /// Set the timeout
94 #[must_use]
95 pub fn timeout(mut self, timeout: std::time::Duration) -> Self {
96 self.timeout = timeout;
97 self
98 }
99
100 /// Set the user agent string for HTTP requests
101 ///
102 /// The default user agent is `redis-cloud/{version}`.
103 /// This can be overridden to identify specific clients, for example:
104 /// `redisctl/1.2.3` or `my-app/1.0.0`.
105 #[must_use]
106 pub fn user_agent(mut self, user_agent: impl Into<String>) -> Self {
107 self.user_agent = user_agent.into();
108 self
109 }
110
111 /// Build the client
112 pub fn build(self) -> Result<CloudClient> {
113 let api_key = self
114 .api_key
115 .ok_or_else(|| RestError::ConnectionError("API key is required".to_string()))?;
116 let api_secret = self
117 .api_secret
118 .ok_or_else(|| RestError::ConnectionError("API secret is required".to_string()))?;
119
120 let mut default_headers = HeaderMap::new();
121 default_headers.insert(
122 USER_AGENT,
123 HeaderValue::from_str(&self.user_agent)
124 .map_err(|e| RestError::ConnectionError(format!("Invalid user agent: {e}")))?,
125 );
126
127 let client = Client::builder()
128 .timeout(self.timeout)
129 .default_headers(default_headers)
130 .build()
131 .map_err(|e| RestError::ConnectionError(e.to_string()))?;
132
133 Ok(CloudClient {
134 api_key,
135 api_secret,
136 base_url: self.base_url,
137 timeout: self.timeout,
138 client: Arc::new(client),
139 })
140 }
141}
142
143/// Redis Cloud API client
144#[derive(Clone)]
145pub struct CloudClient {
146 pub(crate) api_key: String,
147 pub(crate) api_secret: String,
148 pub(crate) base_url: String,
149 pub(crate) timeout: std::time::Duration,
150 pub(crate) client: Arc<Client>,
151}
152
153impl CloudClient {
154 /// Create a new builder for the client
155 #[must_use]
156 pub fn builder() -> CloudClientBuilder {
157 CloudClientBuilder::new()
158 }
159
160 /// Get the configured request timeout
161 ///
162 /// Returns the timeout duration that was set when building the client.
163 /// This timeout is applied to all HTTP requests made by this client.
164 #[must_use]
165 pub fn timeout(&self) -> std::time::Duration {
166 self.timeout
167 }
168
169 // ========================================================================
170 // Fluent API - Handler accessors
171 // ========================================================================
172
173 /// Get an account handler for account management operations
174 ///
175 /// # Example
176 ///
177 /// ```rust,no_run
178 /// # use redis_cloud::CloudClient;
179 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
180 /// let client = CloudClient::builder()
181 /// .api_key("key")
182 /// .api_secret("secret")
183 /// .build()?;
184 ///
185 /// let account = client.account().get_current_account().await?;
186 /// # Ok(())
187 /// # }
188 /// ```
189 #[must_use]
190 pub fn account(&self) -> crate::AccountHandler {
191 crate::AccountHandler::new(self.clone())
192 }
193
194 /// Get a subscription handler for Pro subscription operations
195 ///
196 /// # Example
197 ///
198 /// ```rust,no_run
199 /// # use redis_cloud::CloudClient;
200 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
201 /// let client = CloudClient::builder()
202 /// .api_key("key")
203 /// .api_secret("secret")
204 /// .build()?;
205 ///
206 /// let subscriptions = client.subscriptions().get_all_subscriptions().await?;
207 /// # Ok(())
208 /// # }
209 /// ```
210 #[must_use]
211 pub fn subscriptions(&self) -> crate::SubscriptionHandler {
212 crate::SubscriptionHandler::new(self.clone())
213 }
214
215 /// Get a database handler for Pro database operations
216 ///
217 /// # Example
218 ///
219 /// ```rust,no_run
220 /// # use redis_cloud::CloudClient;
221 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
222 /// let client = CloudClient::builder()
223 /// .api_key("key")
224 /// .api_secret("secret")
225 /// .build()?;
226 ///
227 /// let databases = client.databases().get_subscription_databases(123, None, None).await?;
228 /// # Ok(())
229 /// # }
230 /// ```
231 #[must_use]
232 pub fn databases(&self) -> crate::DatabaseHandler {
233 crate::DatabaseHandler::new(self.clone())
234 }
235
236 /// Get a fixed subscription handler for Essentials subscription operations
237 ///
238 /// # Example
239 ///
240 /// ```rust,no_run
241 /// # use redis_cloud::CloudClient;
242 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
243 /// let client = CloudClient::builder()
244 /// .api_key("key")
245 /// .api_secret("secret")
246 /// .build()?;
247 ///
248 /// let subscriptions = client.fixed_subscriptions().list().await?;
249 /// # Ok(())
250 /// # }
251 /// ```
252 #[must_use]
253 pub fn fixed_subscriptions(&self) -> crate::FixedSubscriptionHandler {
254 crate::FixedSubscriptionHandler::new(self.clone())
255 }
256
257 /// Get a fixed database handler for Essentials database operations
258 ///
259 /// # Example
260 ///
261 /// ```rust,no_run
262 /// # use redis_cloud::CloudClient;
263 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
264 /// let client = CloudClient::builder()
265 /// .api_key("key")
266 /// .api_secret("secret")
267 /// .build()?;
268 ///
269 /// let databases = client.fixed_databases().list(123, None, None).await?;
270 /// # Ok(())
271 /// # }
272 /// ```
273 #[must_use]
274 pub fn fixed_databases(&self) -> crate::FixedDatabaseHandler {
275 crate::FixedDatabaseHandler::new(self.clone())
276 }
277
278 /// Get an ACL handler for access control operations
279 ///
280 /// # Example
281 ///
282 /// ```rust,no_run
283 /// # use redis_cloud::CloudClient;
284 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
285 /// let client = CloudClient::builder()
286 /// .api_key("key")
287 /// .api_secret("secret")
288 /// .build()?;
289 ///
290 /// let users = client.acl().get_all_acl_users().await?;
291 /// # Ok(())
292 /// # }
293 /// ```
294 #[must_use]
295 pub fn acl(&self) -> crate::AclHandler {
296 crate::AclHandler::new(self.clone())
297 }
298
299 /// Get a users handler for user management operations
300 ///
301 /// # Example
302 ///
303 /// ```rust,no_run
304 /// # use redis_cloud::CloudClient;
305 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
306 /// let client = CloudClient::builder()
307 /// .api_key("key")
308 /// .api_secret("secret")
309 /// .build()?;
310 ///
311 /// let users = client.users().get_all_users().await?;
312 /// # Ok(())
313 /// # }
314 /// ```
315 #[must_use]
316 pub fn users(&self) -> crate::UserHandler {
317 crate::UserHandler::new(self.clone())
318 }
319
320 /// Get a tasks handler for async operation tracking
321 ///
322 /// # Example
323 ///
324 /// ```rust,no_run
325 /// # use redis_cloud::CloudClient;
326 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
327 /// let client = CloudClient::builder()
328 /// .api_key("key")
329 /// .api_secret("secret")
330 /// .build()?;
331 ///
332 /// let tasks = client.tasks().get_all_tasks().await?;
333 /// # Ok(())
334 /// # }
335 /// ```
336 #[must_use]
337 pub fn tasks(&self) -> crate::TaskHandler {
338 crate::TaskHandler::new(self.clone())
339 }
340
341 /// Get a cloud accounts handler for cloud provider integration
342 ///
343 /// # Example
344 ///
345 /// ```rust,no_run
346 /// # use redis_cloud::CloudClient;
347 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
348 /// let client = CloudClient::builder()
349 /// .api_key("key")
350 /// .api_secret("secret")
351 /// .build()?;
352 ///
353 /// let accounts = client.cloud_accounts().get_cloud_accounts().await?;
354 /// # Ok(())
355 /// # }
356 /// ```
357 #[must_use]
358 pub fn cloud_accounts(&self) -> crate::CloudAccountHandler {
359 crate::CloudAccountHandler::new(self.clone())
360 }
361
362 /// Get a VPC peering handler for VPC peering operations
363 ///
364 /// # Example
365 ///
366 /// ```rust,no_run
367 /// # use redis_cloud::CloudClient;
368 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
369 /// let client = CloudClient::builder()
370 /// .api_key("key")
371 /// .api_secret("secret")
372 /// .build()?;
373 ///
374 /// let peering = client.vpc_peering().get(123).await?;
375 /// # Ok(())
376 /// # }
377 /// ```
378 #[must_use]
379 pub fn vpc_peering(&self) -> crate::VpcPeeringHandler {
380 crate::VpcPeeringHandler::new(self.clone())
381 }
382
383 /// Get a transit gateway handler for AWS Transit Gateway operations
384 ///
385 /// # Example
386 ///
387 /// ```rust,no_run
388 /// # use redis_cloud::CloudClient;
389 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
390 /// let client = CloudClient::builder()
391 /// .api_key("key")
392 /// .api_secret("secret")
393 /// .build()?;
394 ///
395 /// let attachments = client.transit_gateway().get_attachments(123).await?;
396 /// # Ok(())
397 /// # }
398 /// ```
399 #[must_use]
400 pub fn transit_gateway(&self) -> crate::TransitGatewayHandler {
401 crate::TransitGatewayHandler::new(self.clone())
402 }
403
404 /// Get a Private Service Connect handler for GCP PSC operations
405 ///
406 /// # Example
407 ///
408 /// ```rust,no_run
409 /// # use redis_cloud::CloudClient;
410 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
411 /// let client = CloudClient::builder()
412 /// .api_key("key")
413 /// .api_secret("secret")
414 /// .build()?;
415 ///
416 /// let service = client.psc().get_service(123).await?;
417 /// # Ok(())
418 /// # }
419 /// ```
420 #[must_use]
421 pub fn psc(&self) -> crate::PscHandler {
422 crate::PscHandler::new(self.clone())
423 }
424
425 /// Get a `PrivateLink` handler for AWS `PrivateLink` operations
426 ///
427 /// # Example
428 ///
429 /// ```rust,no_run
430 /// # use redis_cloud::CloudClient;
431 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
432 /// let client = CloudClient::builder()
433 /// .api_key("key")
434 /// .api_secret("secret")
435 /// .build()?;
436 ///
437 /// let config = client.private_link().get(123).await?;
438 /// # Ok(())
439 /// # }
440 /// ```
441 #[must_use]
442 pub fn private_link(&self) -> crate::PrivateLinkHandler {
443 crate::PrivateLinkHandler::new(self.clone())
444 }
445
446 /// Get a cost report handler for generating cost reports
447 ///
448 /// # Example
449 ///
450 /// ```rust,no_run
451 /// # use redis_cloud::CloudClient;
452 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
453 /// let client = CloudClient::builder()
454 /// .api_key("key")
455 /// .api_secret("secret")
456 /// .build()?;
457 ///
458 /// let handler = client.cost_reports();
459 /// # Ok(())
460 /// # }
461 /// ```
462 #[must_use]
463 pub fn cost_reports(&self) -> crate::CostReportHandler {
464 crate::CostReportHandler::new(self.clone())
465 }
466
467 /// Get a Data Integration handler for workspace and proxy operations.
468 #[must_use]
469 pub fn data_integration(&self) -> crate::DataIntegrationHandler {
470 crate::DataIntegrationHandler::new(self.clone())
471 }
472
473 /// Get an endpoint redirections handler for database endpoint migrations.
474 #[must_use]
475 pub fn endpoint_redirections(&self) -> crate::EndpointRedirectionsHandler {
476 crate::EndpointRedirectionsHandler::new(self.clone())
477 }
478
479 /// Normalize URL path concatenation to avoid double slashes
480 fn normalize_url(&self, path: &str) -> String {
481 let base = self.base_url.trim_end_matches('/');
482 let path = path.trim_start_matches('/');
483 format!("{base}/{path}")
484 }
485
486 /// Convert HTTP status code and response text to appropriate error
487 ///
488 /// This is a helper to avoid duplicating the error handling pattern
489 /// across multiple methods.
490 fn status_to_error(status: reqwest::StatusCode, text: String) -> RestError {
491 match status.as_u16() {
492 400 => RestError::BadRequest { message: text },
493 401 => RestError::AuthenticationFailed { message: text },
494 403 => RestError::Forbidden { message: text },
495 404 => RestError::NotFound { message: text },
496 412 => RestError::PreconditionFailed,
497 429 => RestError::RateLimited { message: text },
498 500 => RestError::InternalServerError { message: text },
499 503 => RestError::ServiceUnavailable { message: text },
500 _ => RestError::ApiError {
501 code: status.as_u16(),
502 message: text,
503 },
504 }
505 }
506
507 /// Make a GET request with API key authentication
508 #[instrument(skip(self), fields(method = "GET"))]
509 pub async fn get<T: serde::de::DeserializeOwned>(&self, path: &str) -> Result<T> {
510 let url = self.normalize_url(path);
511 debug!("GET {}", url);
512
513 // Redis Cloud API uses these headers for authentication
514 let response = self
515 .client
516 .get(&url)
517 .header("x-api-key", &self.api_key)
518 .header("x-api-secret-key", &self.api_secret)
519 .send()
520 .await?;
521
522 trace!("Response status: {}", response.status());
523 self.handle_response(response).await
524 }
525
526 /// Make a POST request
527 #[instrument(skip(self, body), fields(method = "POST"))]
528 pub async fn post<B: Serialize, T: serde::de::DeserializeOwned>(
529 &self,
530 path: &str,
531 body: &B,
532 ) -> Result<T> {
533 let url = self.normalize_url(path);
534 debug!("POST {}", url);
535 trace!("Request body: {:?}", serde_json::to_value(body).ok());
536
537 // Same backwards header naming as GET
538 let response = self
539 .client
540 .post(&url)
541 .header("x-api-key", &self.api_key)
542 .header("x-api-secret-key", &self.api_secret)
543 .json(body)
544 .send()
545 .await?;
546
547 trace!("Response status: {}", response.status());
548 self.handle_response(response).await
549 }
550
551 /// Make a POST request without a request body.
552 pub(crate) async fn post_empty<T: serde::de::DeserializeOwned>(&self, path: &str) -> Result<T> {
553 let url = self.normalize_url(path);
554 debug!("POST {}", url);
555
556 let response = self
557 .client
558 .post(&url)
559 .header("x-api-key", &self.api_key)
560 .header("x-api-secret-key", &self.api_secret)
561 .send()
562 .await?;
563
564 trace!("Response status: {}", response.status());
565 self.handle_response(response).await
566 }
567
568 /// Make a PUT request
569 #[instrument(skip(self, body), fields(method = "PUT"))]
570 pub async fn put<B: Serialize, T: serde::de::DeserializeOwned>(
571 &self,
572 path: &str,
573 body: &B,
574 ) -> Result<T> {
575 let url = self.normalize_url(path);
576 debug!("PUT {}", url);
577 trace!("Request body: {:?}", serde_json::to_value(body).ok());
578
579 // Same backwards header naming as GET
580 let response = self
581 .client
582 .put(&url)
583 .header("x-api-key", &self.api_key)
584 .header("x-api-secret-key", &self.api_secret)
585 .json(body)
586 .send()
587 .await?;
588
589 trace!("Response status: {}", response.status());
590 self.handle_response(response).await
591 }
592
593 /// Make a DELETE request
594 #[instrument(skip(self), fields(method = "DELETE"))]
595 pub async fn delete(&self, path: &str) -> Result<()> {
596 let url = self.normalize_url(path);
597 debug!("DELETE {}", url);
598
599 // Same backwards header naming as GET
600 let response = self
601 .client
602 .delete(&url)
603 .header("x-api-key", &self.api_key)
604 .header("x-api-secret-key", &self.api_secret)
605 .send()
606 .await?;
607
608 trace!("Response status: {}", response.status());
609 if response.status().is_success() {
610 Ok(())
611 } else {
612 let status = response.status();
613 let text = response
614 .text()
615 .await
616 .unwrap_or_else(|e| format!("(failed to read response body: {e})"));
617 Err(Self::status_to_error(status, text))
618 }
619 }
620
621 /// Execute a bodyless DELETE request, deserializing the response body.
622 ///
623 /// Unlike [`Self::delete`] (which discards the body) this parses the
624 /// response into `T`. Connectivity deletes are asynchronous and the spec
625 /// returns a [`TaskStateUpdate`](crate::types::TaskStateUpdate) so callers
626 /// can poll the resulting task to completion.
627 #[instrument(skip(self), fields(method = "DELETE"))]
628 pub async fn delete_typed<T: serde::de::DeserializeOwned>(&self, path: &str) -> Result<T> {
629 let url = self.normalize_url(path);
630 debug!("DELETE {}", url);
631
632 let response = self
633 .client
634 .delete(&url)
635 .header("x-api-key", &self.api_key)
636 .header("x-api-secret-key", &self.api_secret)
637 .send()
638 .await?;
639
640 trace!("Response status: {}", response.status());
641 self.handle_response(response).await
642 }
643
644 /// Execute raw GET request returning JSON Value
645 #[instrument(skip(self), fields(method = "GET"))]
646 pub async fn get_raw(&self, path: &str) -> Result<serde_json::Value> {
647 self.get(path).await
648 }
649
650 /// Execute GET request returning raw bytes
651 ///
652 /// Useful for downloading binary content like cost reports or other files.
653 #[instrument(skip(self), fields(method = "GET"))]
654 pub async fn get_bytes(&self, path: &str) -> Result<Vec<u8>> {
655 let url = self.normalize_url(path);
656 debug!("GET {} (bytes)", url);
657
658 let response = self
659 .client
660 .get(&url)
661 .header("x-api-key", &self.api_key)
662 .header("x-api-secret-key", &self.api_secret)
663 .send()
664 .await?;
665
666 trace!("Response status: {}", response.status());
667 let status = response.status();
668
669 if status.is_success() {
670 response
671 .bytes()
672 .await
673 .map(|b| b.to_vec())
674 .map_err(|e| RestError::ConnectionError(format!("Failed to read response: {e}")))
675 } else {
676 let text = response
677 .text()
678 .await
679 .unwrap_or_else(|e| format!("(failed to read response body: {e})"));
680 Err(Self::status_to_error(status, text))
681 }
682 }
683
684 /// Execute raw POST request with JSON body
685 #[instrument(skip(self, body), fields(method = "POST"))]
686 pub async fn post_raw(&self, path: &str, body: serde_json::Value) -> Result<serde_json::Value> {
687 self.post(path, &body).await
688 }
689
690 /// Execute raw PUT request with JSON body
691 #[instrument(skip(self, body), fields(method = "PUT"))]
692 pub async fn put_raw(&self, path: &str, body: serde_json::Value) -> Result<serde_json::Value> {
693 self.put(path, &body).await
694 }
695
696 /// Execute raw PATCH request with JSON body
697 #[instrument(skip(self, body), fields(method = "PATCH"))]
698 pub async fn patch_raw(
699 &self,
700 path: &str,
701 body: serde_json::Value,
702 ) -> Result<serde_json::Value> {
703 let url = self.normalize_url(path);
704 debug!("PATCH {}", url);
705 trace!("Request body: {:?}", body);
706
707 // Use backwards header names for compatibility
708 let response = self
709 .client
710 .patch(&url)
711 .header("x-api-key", &self.api_key)
712 .header("x-api-secret-key", &self.api_secret)
713 .json(&body)
714 .send()
715 .await?;
716
717 trace!("Response status: {}", response.status());
718 self.handle_response(response).await
719 }
720
721 /// Execute raw DELETE request returning any response body
722 #[instrument(skip(self), fields(method = "DELETE"))]
723 pub async fn delete_raw(&self, path: &str) -> Result<serde_json::Value> {
724 let url = self.normalize_url(path);
725 debug!("DELETE {}", url);
726
727 // Use backwards header names for compatibility
728 let response = self
729 .client
730 .delete(&url)
731 .header("x-api-key", &self.api_key)
732 .header("x-api-secret-key", &self.api_secret)
733 .send()
734 .await?;
735
736 trace!("Response status: {}", response.status());
737 if response.status().is_success() {
738 if response.content_length() == Some(0) {
739 Ok(serde_json::json!({"status": "deleted"}))
740 } else {
741 response.json().await.map_err(Into::into)
742 }
743 } else {
744 let status = response.status();
745 let text = response
746 .text()
747 .await
748 .unwrap_or_else(|e| format!("(failed to read response body: {e})"));
749 Err(Self::status_to_error(status, text))
750 }
751 }
752
753 /// Execute DELETE request with JSON body (used by some endpoints like `PrivateLink` principals)
754 #[instrument(skip(self, body), fields(method = "DELETE"))]
755 pub async fn delete_with_body<T: serde::de::DeserializeOwned>(
756 &self,
757 path: &str,
758 body: serde_json::Value,
759 ) -> Result<T> {
760 let url = self.normalize_url(path);
761 debug!("DELETE {} (with body)", url);
762 trace!("Request body: {:?}", body);
763
764 let response = self
765 .client
766 .delete(&url)
767 .header("x-api-key", &self.api_key)
768 .header("x-api-secret-key", &self.api_secret)
769 .json(&body)
770 .send()
771 .await?;
772
773 trace!("Response status: {}", response.status());
774 self.handle_response(response).await
775 }
776
777 /// Handle HTTP response and return both status code and body as JSON
778 ///
779 /// This is used internally by the Tower service implementation to preserve
780 /// the actual HTTP status code in responses.
781 #[cfg(feature = "tower-integration")]
782 async fn handle_response_with_status(
783 &self,
784 response: reqwest::Response,
785 ) -> Result<(u16, serde_json::Value)> {
786 let status = response.status();
787 let status_code = status.as_u16();
788
789 if status.is_success() {
790 let bytes = response
791 .bytes()
792 .await
793 .map_err(|e| RestError::ConnectionError(format!("Failed to read response: {e}")))?;
794
795 let value: serde_json::Value = serde_json::from_slice(&bytes).map_err(|e| {
796 RestError::ConnectionError(format!("Failed to parse JSON response: {e}"))
797 })?;
798
799 Ok((status_code, value))
800 } else {
801 let text = response
802 .text()
803 .await
804 .unwrap_or_else(|e| format!("(failed to read response body: {e})"));
805 Err(Self::status_to_error(status, text))
806 }
807 }
808
809 /// Handle HTTP response
810 async fn handle_response<T: serde::de::DeserializeOwned>(
811 &self,
812 response: reqwest::Response,
813 ) -> Result<T> {
814 let status = response.status();
815
816 if status.is_success() {
817 // Get the response bytes for better error reporting
818 let bytes = response
819 .bytes()
820 .await
821 .map_err(|e| RestError::ConnectionError(format!("Failed to read response: {e}")))?;
822
823 // Treat an empty success body (e.g. HTTP 204 No Content from the
824 // traffic-resume endpoints) as JSON `null` so it deserializes
825 // cleanly into `()`, `Option<T>`, or `serde_json::Value::Null`.
826 let bytes: &[u8] = if bytes.is_empty() { b"null" } else { &bytes };
827
828 // Use serde_path_to_error for better deserialization error messages
829 let deserializer = &mut serde_json::Deserializer::from_slice(bytes);
830 serde_path_to_error::deserialize(deserializer).map_err(|err| {
831 let path = err.path().to_string();
832 // Use ConnectionError to provide detailed error message with field path
833 RestError::ConnectionError(format!(
834 "Failed to deserialize field '{}': {}",
835 path,
836 err.inner()
837 ))
838 })
839 } else {
840 let text = response
841 .text()
842 .await
843 .unwrap_or_else(|e| format!("(failed to read response body: {e})"));
844 Err(Self::status_to_error(status, text))
845 }
846 }
847}
848
849/// Tower Service integration for `CloudClient`
850///
851/// This module provides Tower Service implementations for `CloudClient`, enabling
852/// middleware composition with patterns like circuit breakers, retry, and rate limiting.
853///
854/// # Feature Flag
855///
856/// This module is only available when the `tower-integration` feature is enabled.
857///
858/// # Examples
859///
860/// ```rust,ignore
861/// use redis_cloud::CloudClient;
862/// use redis_cloud::tower_support::ApiRequest;
863/// use tower::ServiceExt;
864///
865/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
866/// let client = CloudClient::builder()
867/// .api_key("your-key")
868/// .api_secret("your-secret")
869/// .build()?;
870///
871/// // Convert to a Tower service
872/// let mut service = client.into_service();
873///
874/// // Use the service
875/// let response = service.oneshot(ApiRequest::get("/subscriptions")).await?;
876/// println!("Status: {}", response.status);
877/// # Ok(())
878/// # }
879/// ```
880#[cfg(feature = "tower-integration")]
881pub mod tower_support {
882 use super::{CloudClient, RestError, Result};
883 use std::future::Future;
884 use std::pin::Pin;
885 use std::task::{Context, Poll};
886 use tower::Service;
887
888 /// HTTP method for API requests
889 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
890 pub enum Method {
891 /// GET request
892 Get,
893 /// POST request
894 Post,
895 /// PUT request
896 Put,
897 /// PATCH request
898 Patch,
899 /// DELETE request
900 Delete,
901 }
902
903 /// Tower-compatible request type for Redis Cloud API
904 ///
905 /// This wraps the essential components of an API request in a format
906 /// suitable for Tower middleware composition.
907 #[derive(Debug, Clone)]
908 pub struct ApiRequest {
909 /// HTTP method
910 pub method: Method,
911 /// API endpoint path (e.g., "/subscriptions")
912 pub path: String,
913 /// Optional JSON body for POST/PUT/PATCH requests
914 pub body: Option<serde_json::Value>,
915 }
916
917 impl ApiRequest {
918 /// Create a GET request
919 pub fn get(path: impl Into<String>) -> Self {
920 Self {
921 method: Method::Get,
922 path: path.into(),
923 body: None,
924 }
925 }
926
927 /// Create a POST request with a JSON body
928 pub fn post(path: impl Into<String>, body: serde_json::Value) -> Self {
929 Self {
930 method: Method::Post,
931 path: path.into(),
932 body: Some(body),
933 }
934 }
935
936 /// Create a PUT request with a JSON body
937 pub fn put(path: impl Into<String>, body: serde_json::Value) -> Self {
938 Self {
939 method: Method::Put,
940 path: path.into(),
941 body: Some(body),
942 }
943 }
944
945 /// Create a PATCH request with a JSON body
946 pub fn patch(path: impl Into<String>, body: serde_json::Value) -> Self {
947 Self {
948 method: Method::Patch,
949 path: path.into(),
950 body: Some(body),
951 }
952 }
953
954 /// Create a DELETE request
955 pub fn delete(path: impl Into<String>) -> Self {
956 Self {
957 method: Method::Delete,
958 path: path.into(),
959 body: None,
960 }
961 }
962 }
963
964 /// Tower-compatible response type
965 ///
966 /// Contains the HTTP status code and response body as JSON.
967 #[derive(Debug, Clone)]
968 pub struct ApiResponse {
969 /// HTTP status code
970 pub status: u16,
971 /// Response body as JSON
972 pub body: serde_json::Value,
973 }
974
975 impl CloudClient {
976 /// Convert this client into a Tower service
977 ///
978 /// This consumes the client and returns it wrapped in a Tower service
979 /// implementation, enabling middleware composition.
980 ///
981 /// # Examples
982 ///
983 /// ```rust,ignore
984 /// use redis_cloud::CloudClient;
985 /// use tower::ServiceExt;
986 /// use redis_cloud::tower_support::ApiRequest;
987 ///
988 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
989 /// let client = CloudClient::builder()
990 /// .api_key("key")
991 /// .api_secret("secret")
992 /// .build()?;
993 ///
994 /// let mut service = client.into_service();
995 /// let response = service.oneshot(ApiRequest::get("/subscriptions")).await?;
996 /// # Ok(())
997 /// # }
998 /// ```
999 #[must_use]
1000 pub fn into_service(self) -> Self {
1001 self
1002 }
1003 }
1004
1005 impl Service<ApiRequest> for CloudClient {
1006 type Response = ApiResponse;
1007 type Error = RestError;
1008 type Future = Pin<Box<dyn Future<Output = Result<Self::Response>> + Send>>;
1009
1010 fn poll_ready(
1011 &mut self,
1012 _cx: &mut Context<'_>,
1013 ) -> Poll<std::result::Result<(), Self::Error>> {
1014 // CloudClient is always ready since it uses an internal connection pool
1015 Poll::Ready(Ok(()))
1016 }
1017
1018 fn call(&mut self, req: ApiRequest) -> Self::Future {
1019 let client = self.clone();
1020 Box::pin(async move {
1021 let url = client.normalize_url(&req.path);
1022
1023 let request_builder = match req.method {
1024 Method::Get => client.client.get(&url),
1025 Method::Post => {
1026 let body = req.body.ok_or_else(|| RestError::BadRequest {
1027 message: "POST request requires a body".to_string(),
1028 })?;
1029 client.client.post(&url).json(&body)
1030 }
1031 Method::Put => {
1032 let body = req.body.ok_or_else(|| RestError::BadRequest {
1033 message: "PUT request requires a body".to_string(),
1034 })?;
1035 client.client.put(&url).json(&body)
1036 }
1037 Method::Patch => {
1038 let body = req.body.ok_or_else(|| RestError::BadRequest {
1039 message: "PATCH request requires a body".to_string(),
1040 })?;
1041 client.client.patch(&url).json(&body)
1042 }
1043 Method::Delete => client.client.delete(&url),
1044 };
1045
1046 let response = request_builder
1047 .header("x-api-key", &client.api_key)
1048 .header("x-api-secret-key", &client.api_secret)
1049 .send()
1050 .await?;
1051
1052 let (status, body) = client.handle_response_with_status(response).await?;
1053
1054 Ok(ApiResponse { status, body })
1055 })
1056 }
1057 }
1058}