systemprompt_sync/api_client/
mod.rs

1//! HTTP client used by sync push/pull/deploy.
2//!
3//! Handles direct-sync vs. cloud-relay endpoint selection, bearer-token
4//! auth, retryable failures with exponential backoff, and typed JSON /
5//! binary response handling.
6//!
7//! Copyright (c) systemprompt.io — Business Source License 1.1.
8//! See <https://systemprompt.io> for licensing details.
9
10mod response;
11mod retry;
12mod token;
13
14use std::sync::Arc;
15use std::time::Duration;
16
17use reqwest::Client;
18use serde::de::DeserializeOwned;
19use serde::{Deserialize, Serialize};
20use systemprompt_models::net::{HTTP_CONNECT_TIMEOUT, HTTP_SYNC_DEPLOY_TIMEOUT};
21use tokio::sync::Mutex;
22use tokio::time::sleep;
23
24use crate::error::{SyncError, SyncResult};
25pub use retry::RetryConfig;
26use token::is_unauthorized;
27pub use token::{exchange_subject_token, exchange_subject_token_at};
28
29#[derive(Clone, Debug)]
30pub struct SyncApiClient {
31    client: Client,
32    api_url: String,
33    token: String,
34    direct_sync_origin: Option<String>,
35    cached_sync_token: Arc<Mutex<Option<String>>>,
36    retry_config: RetryConfig,
37}
38
39#[derive(Debug, Deserialize)]
40pub struct RegistryToken {
41    pub registry: String,
42    pub username: String,
43    pub token: String,
44}
45
46#[derive(Debug, Clone, Copy, Deserialize)]
47pub struct UploadResponse {
48    pub files_uploaded: usize,
49}
50
51#[derive(Debug, Deserialize)]
52pub struct DeployResponse {
53    pub status: String,
54    pub app_url: Option<String>,
55}
56
57impl SyncApiClient {
58    pub fn new(api_url: &str, token: &str) -> SyncResult<Self> {
59        Ok(Self {
60            client: Client::builder()
61                .connect_timeout(HTTP_CONNECT_TIMEOUT)
62                .timeout(HTTP_SYNC_DEPLOY_TIMEOUT)
63                .build()?,
64            api_url: api_url.to_owned(),
65            token: token.to_owned(),
66            direct_sync_origin: None,
67            cached_sync_token: Arc::new(Mutex::new(None)),
68            retry_config: RetryConfig::default(),
69        })
70    }
71
72    pub fn with_direct_sync(self, hostname: Option<String>) -> Self {
73        let origin = hostname.map(|h| format!("https://{h}"));
74        self.with_direct_sync_origin(origin)
75    }
76
77    pub fn with_direct_sync_origin(mut self, origin: Option<String>) -> Self {
78        self.direct_sync_origin = origin;
79        self
80    }
81
82    pub const fn with_retry_config(mut self, retry_config: RetryConfig) -> Self {
83        self.retry_config = retry_config;
84        self
85    }
86
87    pub(crate) fn direct_sync_origin(&self) -> Option<&str> {
88        self.direct_sync_origin.as_deref()
89    }
90
91    const fn is_direct_sync(&self) -> bool {
92        self.direct_sync_origin.is_some()
93    }
94
95    fn files_url(&self, tenant_id: &systemprompt_identifiers::TenantId) -> String {
96        self.direct_sync_origin.as_ref().map_or_else(
97            || format!("{}/api/v1/cloud/tenants/{}/files", self.api_url, tenant_id),
98            |origin| format!("{origin}/api/v1/sync/files"),
99        )
100    }
101
102    fn calculate_next_delay(&self, current: Duration) -> Duration {
103        self.retry_config.next_delay(current)
104    }
105
106    pub async fn upload_files(
107        &self,
108        tenant_id: &systemprompt_identifiers::TenantId,
109        data: Vec<u8>,
110    ) -> SyncResult<UploadResponse> {
111        let url = self.files_url(tenant_id);
112        let direct = self.is_direct_sync();
113        let mut bearer = self.bearer_token(false).await?;
114        let mut reminted = false;
115
116        let mut current_delay = self.retry_config.initial_delay;
117
118        for attempt in 1..=self.retry_config.max_attempts {
119            let response = self
120                .client
121                .post(&url)
122                .header("Authorization", format!("Bearer {bearer}"))
123                .header("Content-Type", "application/octet-stream")
124                .body(data.clone())
125                .send()
126                .await?;
127
128            match response::handle_json::<UploadResponse>(response).await {
129                Ok(upload) => return Ok(upload),
130                Err(error) if direct && !reminted && is_unauthorized(&error) => {
131                    reminted = true;
132                    bearer = self.bearer_token(true).await?;
133                },
134                Err(error) if error.is_retryable() && attempt < self.retry_config.max_attempts => {
135                    tracing::warn!(
136                        attempt = attempt,
137                        max_attempts = self.retry_config.max_attempts,
138                        delay_ms = current_delay.as_millis() as u64,
139                        error = %error,
140                        "Retryable sync error, waiting before retry"
141                    );
142                    sleep(current_delay).await;
143                    current_delay = self.calculate_next_delay(current_delay);
144                },
145                Err(error) => return Err(error),
146            }
147        }
148
149        Err(SyncError::ApiError {
150            status: 503,
151            message: "Max retry attempts exceeded".to_owned(),
152        })
153    }
154
155    pub async fn download_files(
156        &self,
157        tenant_id: &systemprompt_identifiers::TenantId,
158    ) -> SyncResult<Vec<u8>> {
159        let url = self.files_url(tenant_id);
160        let direct = self.is_direct_sync();
161        let mut bearer = self.bearer_token(false).await?;
162        let mut reminted = false;
163
164        let mut current_delay = self.retry_config.initial_delay;
165
166        for attempt in 1..=self.retry_config.max_attempts {
167            let response = self
168                .client
169                .get(&url)
170                .header("Authorization", format!("Bearer {bearer}"))
171                .send()
172                .await?;
173
174            match response::handle_binary(response).await {
175                Ok(data) => return Ok(data),
176                Err(error) if direct && !reminted && is_unauthorized(&error) => {
177                    reminted = true;
178                    bearer = self.bearer_token(true).await?;
179                },
180                Err(error) if error.is_retryable() && attempt < self.retry_config.max_attempts => {
181                    tracing::warn!(
182                        attempt = attempt,
183                        max_attempts = self.retry_config.max_attempts,
184                        delay_ms = current_delay.as_millis() as u64,
185                        error = %error,
186                        "Retryable sync error, waiting before retry"
187                    );
188                    sleep(current_delay).await;
189                    current_delay = self.calculate_next_delay(current_delay);
190                },
191                Err(error) => return Err(error),
192            }
193        }
194
195        Err(SyncError::ApiError {
196            status: 503,
197            message: "Max retry attempts exceeded".to_owned(),
198        })
199    }
200
201    pub async fn get_registry_token(
202        &self,
203        tenant_id: &systemprompt_identifiers::TenantId,
204    ) -> SyncResult<RegistryToken> {
205        let url = format!(
206            "{}/api/v1/cloud/tenants/{}/registry-token",
207            self.api_url, tenant_id
208        );
209        self.get(&url).await
210    }
211
212    pub async fn deploy(
213        &self,
214        tenant_id: &systemprompt_identifiers::TenantId,
215        image: &str,
216    ) -> SyncResult<DeployResponse> {
217        let url = format!("{}/api/v1/cloud/tenants/{}/deploy", self.api_url, tenant_id);
218        self.post(&url, &serde_json::json!({ "image": image }))
219            .await
220    }
221
222    pub async fn get_tenant_app_id(
223        &self,
224        tenant_id: &systemprompt_identifiers::TenantId,
225    ) -> SyncResult<String> {
226        #[derive(Deserialize)]
227        struct TenantInfo {
228            fly_app_name: Option<String>,
229        }
230        let url = format!("{}/api/v1/cloud/tenants/{}", self.api_url, tenant_id);
231        let info: TenantInfo = self.get(&url).await?;
232        info.fly_app_name.ok_or(SyncError::TenantNoApp)
233    }
234
235    pub async fn get_database_url(
236        &self,
237        tenant_id: &systemprompt_identifiers::TenantId,
238    ) -> SyncResult<String> {
239        #[derive(Deserialize)]
240        struct DatabaseInfo {
241            database_url: Option<String>,
242        }
243        let url = format!(
244            "{}/api/v1/cloud/tenants/{}/database",
245            self.api_url, tenant_id
246        );
247        let info: DatabaseInfo = self.get(&url).await?;
248        info.database_url.ok_or_else(|| SyncError::ApiError {
249            status: 404,
250            message: "Database URL not available for tenant".to_owned(),
251        })
252    }
253
254    pub(crate) async fn get<T: DeserializeOwned>(&self, url: &str) -> SyncResult<T> {
255        let resp = self
256            .client
257            .get(url)
258            .header("Authorization", format!("Bearer {}", self.token))
259            .send()
260            .await?;
261        response::handle_json(resp).await
262    }
263
264    pub(crate) async fn post<T: DeserializeOwned, B: Serialize + Sync>(
265        &self,
266        url: &str,
267        body: &B,
268    ) -> SyncResult<T> {
269        let resp = self
270            .client
271            .post(url)
272            .header("Authorization", format!("Bearer {}", self.token))
273            .json(body)
274            .send()
275            .await?;
276        response::handle_json(resp).await
277    }
278}
systemprompt_sync/api_client/mod.rs

systemprompt_sync/api_client/
mod.rs
