pipedash_core/infrastructure/config/
sync.rs1use std::collections::HashSet;
2
3use indexmap::IndexMap;
4
5use super::schema::ProviderFileConfig;
6use super::token_ref::TokenReference;
7use crate::domain::{
8 DomainResult,
9 ProviderConfig,
10};
11use crate::infrastructure::{
12 ConfigBackend,
13 TokenStore,
14};
15
16#[derive(Debug, Clone, Default)]
17pub struct SyncResult {
18 pub added: Vec<String>,
19 pub updated: Vec<String>,
20 pub removed: Vec<String>,
21}
22
23pub struct ProviderSyncService;
24
25impl ProviderSyncService {
26 pub fn new() -> Self {
27 Self
28 }
29
30 pub async fn sync_toml_to_db(
31 &self, toml_providers: IndexMap<String, ProviderFileConfig>,
32 config_backend: &dyn ConfigBackend, token_store: &dyn TokenStore, delete_orphans: bool,
33 ) -> DomainResult<SyncResult> {
34 let mut result = SyncResult::default();
35
36 let db_providers = config_backend.list_providers().await?;
37
38 let toml_ids: HashSet<String> = toml_providers.keys().cloned().collect();
39
40 for (id, toml_provider) in toml_providers {
41 let existing_db = db_providers.iter().find(|p| p.name == id);
42
43 match existing_db {
44 None => {
45 let mut config = toml_provider.config.clone();
46 if let Some(display_name) = &toml_provider.name {
47 config.insert("display_name".to_string(), display_name.clone());
48 }
49
50 let provider_config = ProviderConfig {
51 id: None,
52 name: id.clone(), provider_type: toml_provider.provider_type.clone(),
54 token: toml_provider.token.clone(),
55 config,
56 refresh_interval: toml_provider.refresh_interval as i64,
57 version: None,
58 };
59
60 let new_provider_id = config_backend.create_provider(&provider_config).await?;
61
62 if let Some(resolved_token) = Self::resolve_token_ref(&toml_provider.token) {
63 if let Err(e) = token_store
64 .store_token(new_provider_id, &resolved_token)
65 .await
66 {
67 tracing::warn!(
68 provider = %id,
69 error = %e,
70 "Failed to store resolved token - provider may not work until token is re-added"
71 );
72 }
73 }
74
75 result.added.push(id.clone());
76
77 tracing::info!(
78 provider = %id,
79 "Synced provider from TOML to database (new)"
80 );
81 }
82 Some(db_provider) => {
83 if self.needs_update(&toml_provider, db_provider) {
84 let mut config = toml_provider.config.clone();
85 if let Some(display_name) = &toml_provider.name {
86 config.insert("display_name".to_string(), display_name.clone());
87 }
88
89 let provider_config = ProviderConfig {
90 id: db_provider.id,
91 name: id.clone(), provider_type: toml_provider.provider_type.clone(),
93 token: toml_provider.token.clone(),
94 config,
95 refresh_interval: toml_provider.refresh_interval as i64,
96 version: db_provider.version,
97 };
98
99 if let Some(db_id) = db_provider.id {
100 config_backend
101 .update_provider(db_id, &provider_config)
102 .await?;
103
104 if let Some(resolved_token) =
105 Self::resolve_token_ref(&toml_provider.token)
106 {
107 if let Err(e) =
108 token_store.store_token(db_id, &resolved_token).await
109 {
110 tracing::warn!(
111 provider = %id,
112 error = %e,
113 "Failed to store resolved token - provider may not work until token is re-added"
114 );
115 }
116 }
117
118 result.updated.push(id.clone());
119
120 tracing::info!(
121 provider = %id,
122 "Synced provider from TOML to database (updated)"
123 );
124 }
125 }
126 }
127 }
128 }
129
130 if delete_orphans {
131 for db_provider in db_providers {
132 if !toml_ids.contains(&db_provider.name) {
133 if let Some(id) = db_provider.id {
134 config_backend.delete_provider(id).await?;
135 result.removed.push(db_provider.name.clone());
136
137 tracing::info!(
138 provider = %db_provider.name,
139 "Removed provider from database (not in TOML)"
140 );
141 }
142 }
143 }
144 }
145
146 Ok(result)
147 }
148
149 fn needs_update(&self, toml: &ProviderFileConfig, db: &ProviderConfig) -> bool {
150 if toml.provider_type != db.provider_type
151 || toml.token != db.token
152 || toml.refresh_interval as i64 != db.refresh_interval
153 {
154 return true;
155 }
156
157 let db_display_name = db.config.get("display_name").map(|s| s.as_str());
158 let toml_display_name = toml.name.as_deref();
159 if db_display_name != toml_display_name {
160 return true;
161 }
162
163 let db_config_without_display: std::collections::HashMap<_, _> = db
164 .config
165 .iter()
166 .filter(|(k, _)| *k != "display_name")
167 .map(|(k, v)| (k.clone(), v.clone()))
168 .collect();
169
170 toml.config != db_config_without_display
171 }
172
173 fn resolve_token_ref(token_ref: &str) -> Option<String> {
174 if token_ref.is_empty() {
175 return None;
176 }
177
178 match TokenReference::parse(token_ref) {
179 Ok(TokenReference::EnvVar(var_name)) => match std::env::var(&var_name) {
180 Ok(value) => {
181 tracing::debug!(var_name = %var_name, "Resolved token from environment variable");
182 Some(value)
183 }
184 Err(_) => {
185 tracing::warn!(
186 var_name = %var_name,
187 "Environment variable not found for token reference"
188 );
189 None
190 }
191 },
192 Ok(TokenReference::None) => None,
193 Ok(_) => {
194 tracing::warn!(
195 token_ref = %token_ref,
196 "Unsupported token reference format in TOML config"
197 );
198 None
199 }
200 Err(_) => {
201 tracing::debug!("Using token value directly (not a reference)");
202 Some(token_ref.to_string())
203 }
204 }
205 }
206}
207
208impl Default for ProviderSyncService {
209 fn default() -> Self {
210 Self::new()
211 }
212}
213
214#[cfg(test)]
215mod tests {
216 use std::sync::Arc;
217
218 use tempfile::TempDir;
219
220 use super::*;
221 use crate::infrastructure::database::SqliteConfigBackend;
222 use crate::infrastructure::MemoryTokenStore;
223
224 async fn create_test_backend() -> (SqliteConfigBackend, Arc<MemoryTokenStore>, TempDir) {
225 let temp_dir = TempDir::new().unwrap();
226 let db_path = temp_dir.path().join("test.db");
227 let pool = crate::infrastructure::database::init_database(db_path)
228 .await
229 .unwrap();
230 let backend = SqliteConfigBackend::new(pool);
231 let token_store = Arc::new(MemoryTokenStore::new());
232 (backend, token_store, temp_dir)
233 }
234
235 #[tokio::test]
236 async fn test_sync_toml_to_db_new_provider() {
237 let (backend, token_store, _temp) = create_test_backend().await;
238 let sync_service = ProviderSyncService::new();
239
240 let mut toml_providers = IndexMap::new();
241 toml_providers.insert(
242 "test-github".to_string(),
243 ProviderFileConfig {
244 name: None,
245 provider_type: "github".to_string(),
246 token: "${GITHUB_TOKEN}".to_string(),
247 refresh_interval: 30,
248 config: Default::default(),
249 },
250 );
251
252 let result = sync_service
253 .sync_toml_to_db(toml_providers, &backend, token_store.as_ref(), true)
254 .await
255 .unwrap();
256
257 assert_eq!(result.added.len(), 1);
258 assert_eq!(result.added[0], "test-github");
259 assert!(result.updated.is_empty());
260 assert!(result.removed.is_empty());
261
262 let db_providers = backend.list_providers().await.unwrap();
263 assert_eq!(db_providers.len(), 1);
264 assert_eq!(db_providers[0].name, "test-github");
265 }
266
267 #[tokio::test]
268 async fn test_sync_toml_to_db_update_provider() {
269 let (backend, token_store, _temp) = create_test_backend().await;
270 let sync_service = ProviderSyncService::new();
271
272 let initial_provider = ProviderConfig {
273 id: None,
274 name: "test-github".to_string(),
275 provider_type: "github".to_string(),
276 token: "${OLD_TOKEN}".to_string(),
277 config: Default::default(),
278 refresh_interval: 30,
279 version: None,
280 };
281 backend.create_provider(&initial_provider).await.unwrap();
282
283 let mut updated_toml = IndexMap::new();
284 updated_toml.insert(
285 "test-github".to_string(),
286 ProviderFileConfig {
287 name: None,
288 provider_type: "github".to_string(),
289 token: "${NEW_TOKEN}".to_string(),
290 refresh_interval: 60,
291 config: Default::default(),
292 },
293 );
294
295 let result = sync_service
296 .sync_toml_to_db(updated_toml, &backend, token_store.as_ref(), true)
297 .await
298 .unwrap();
299
300 assert!(result.added.is_empty());
301 assert_eq!(result.updated.len(), 1);
302 assert_eq!(result.updated[0], "test-github");
303 assert!(result.removed.is_empty());
304
305 let db_providers = backend.list_providers().await.unwrap();
306 assert_eq!(db_providers[0].token, "${NEW_TOKEN}");
307 assert_eq!(db_providers[0].refresh_interval, 60);
308 }
309
310 #[tokio::test]
311 async fn test_sync_toml_to_db_remove_provider() {
312 let (backend, token_store, _temp) = create_test_backend().await;
313 let sync_service = ProviderSyncService::new();
314
315 let initial_provider = ProviderConfig {
316 id: None,
317 name: "test-github".to_string(),
318 provider_type: "github".to_string(),
319 token: "${TOKEN}".to_string(),
320 config: Default::default(),
321 refresh_interval: 30,
322 version: None,
323 };
324 backend.create_provider(&initial_provider).await.unwrap();
325
326 let result = sync_service
327 .sync_toml_to_db(IndexMap::new(), &backend, token_store.as_ref(), true)
328 .await
329 .unwrap();
330
331 assert!(result.added.is_empty());
332 assert!(result.updated.is_empty());
333 assert_eq!(result.removed.len(), 1);
334 assert_eq!(result.removed[0], "test-github");
335
336 let db_providers = backend.list_providers().await.unwrap();
337 assert!(db_providers.is_empty());
338 }
339
340 #[tokio::test]
341 async fn test_sync_toml_to_db_keep_orphans_for_postgres() {
342 let (backend, token_store, _temp) = create_test_backend().await;
343 let sync_service = ProviderSyncService::new();
344
345 let db_only_provider = ProviderConfig {
346 id: None,
347 name: "db-only-provider".to_string(),
348 provider_type: "github".to_string(),
349 token: "${DB_TOKEN}".to_string(),
350 config: Default::default(),
351 refresh_interval: 30,
352 version: None,
353 };
354 backend.create_provider(&db_only_provider).await.unwrap();
355
356 let result = sync_service
357 .sync_toml_to_db(IndexMap::new(), &backend, token_store.as_ref(), false)
358 .await
359 .unwrap();
360
361 assert!(result.added.is_empty());
362 assert!(result.updated.is_empty());
363 assert!(result.removed.is_empty());
364
365 let db_providers = backend.list_providers().await.unwrap();
366 assert_eq!(db_providers.len(), 1);
367 assert_eq!(db_providers[0].name, "db-only-provider");
368 }
369
370 #[tokio::test]
371 async fn test_sync_toml_to_db_postgres_mode_adds_from_toml() {
372 let (backend, token_store, _temp) = create_test_backend().await;
373 let sync_service = ProviderSyncService::new();
374
375 let db_only_provider = ProviderConfig {
376 id: None,
377 name: "db-only-provider".to_string(),
378 provider_type: "github".to_string(),
379 token: "${DB_TOKEN}".to_string(),
380 config: Default::default(),
381 refresh_interval: 30,
382 version: None,
383 };
384 backend.create_provider(&db_only_provider).await.unwrap();
385
386 let mut toml_providers = IndexMap::new();
387 toml_providers.insert(
388 "toml-provider".to_string(),
389 ProviderFileConfig {
390 name: None,
391 provider_type: "gitlab".to_string(),
392 token: "${TOML_TOKEN}".to_string(),
393 refresh_interval: 60,
394 config: Default::default(),
395 },
396 );
397
398 let result = sync_service
399 .sync_toml_to_db(toml_providers, &backend, token_store.as_ref(), false)
400 .await
401 .unwrap();
402
403 assert_eq!(result.added.len(), 1);
404 assert_eq!(result.added[0], "toml-provider");
405 assert!(result.updated.is_empty());
406 assert!(result.removed.is_empty());
407
408 let db_providers = backend.list_providers().await.unwrap();
409 assert_eq!(db_providers.len(), 2);
410 }
411}