1use std::collections::HashMap;
2use std::time::Duration;
3
4use async_trait::async_trait;
5use sqlx::{
6 Row as SqlxRow,
7 SqlitePool,
8};
9use tokio::time::sleep;
10
11use crate::domain::{
12 DomainError,
13 DomainResult,
14 ProviderConfig,
15};
16use crate::infrastructure::config_backend::{
17 ConfigBackend,
18 ConfigExport,
19 StoredPermissions,
20};
21
22async fn retry_on_busy<F, Fut, T>(operation: F) -> DomainResult<T>
23where
24 F: Fn() -> Fut,
25 Fut: std::future::Future<Output = DomainResult<T>>,
26{
27 const MAX_RETRIES: u32 = 3; const INITIAL_DELAY_MS: u64 = 10;
30
31 let mut attempt = 0;
32 loop {
33 match operation().await {
34 Ok(result) => {
35 if attempt > 0 {
36 tracing::warn!(
37 "SQLite operation succeeded after {} retries ({}ms total)",
38 attempt,
39 INITIAL_DELAY_MS * (2_u64.pow(attempt) - 1)
40 );
41 }
42 return Ok(result);
43 }
44 Err(DomainError::DatabaseError(ref msg))
45 if (msg.contains("database is locked")
46 || msg.contains("SQLITE_BUSY")
47 || msg.contains("locked"))
48 && attempt < MAX_RETRIES =>
49 {
50 attempt += 1;
51 let delay = INITIAL_DELAY_MS * 2_u64.pow(attempt - 1);
52
53 if attempt > 3 {
54 tracing::warn!(
55 "SQLite busy, retry {}/{} (waiting {}ms)",
56 attempt,
57 MAX_RETRIES,
58 delay
59 );
60 }
61
62 sleep(Duration::from_millis(delay)).await;
63 }
64 Err(e) => {
65 if attempt > 0 {
66 tracing::error!("SQLite operation failed after {} retries: {}", attempt, e);
67 }
68 return Err(e);
69 }
70 }
71 }
72}
73
74pub struct SqliteConfigBackend {
75 pool: SqlitePool,
76}
77
78impl SqliteConfigBackend {
79 pub fn new(pool: SqlitePool) -> Self {
80 Self { pool }
81 }
82
83 pub fn pool(&self) -> &SqlitePool {
84 &self.pool
85 }
86}
87
88#[async_trait]
89impl ConfigBackend for SqliteConfigBackend {
90 async fn list_providers(&self) -> DomainResult<Vec<ProviderConfig>> {
91 retry_on_busy(|| async {
92 let rows = sqlx::query(
93 r#"SELECT id, name, provider_type, token_encrypted, config_json, refresh_interval, version FROM providers"#,
94 )
95 .fetch_all(&self.pool)
96 .await
97 .map_err(|e| DomainError::DatabaseError(format!("Failed to list providers: {}", e)))?;
98
99 let mut providers = Vec::new();
100 for row in rows {
101 let id: i64 = row.get("id");
102 let config_json: String = row.get("config_json");
103 let config: HashMap<String, String> = if config_json.trim().is_empty() {
104 HashMap::new()
105 } else {
106 serde_json::from_str(&config_json).map_err(|e| {
107 DomainError::DatabaseError(format!(
108 "Failed to parse config for {}: {}",
109 id, e
110 ))
111 })?
112 };
113
114 let token_ref: String = row.get("token_encrypted");
115
116 providers.push(ProviderConfig {
117 id: Some(id),
118 name: row.get("name"),
119 provider_type: row.get("provider_type"),
120 config,
121 token: token_ref,
122 refresh_interval: row.get("refresh_interval"),
123 version: Some(row.get("version")),
124 });
125 }
126
127 Ok(providers)
128 })
129 .await
130 }
131
132 async fn get_provider(&self, id: i64) -> DomainResult<Option<ProviderConfig>> {
133 retry_on_busy(|| async {
134 let row = sqlx::query(
135 r#"SELECT id, name, provider_type, token_encrypted, config_json, refresh_interval, version FROM providers WHERE id = ?"#,
136 )
137 .bind(id)
138 .fetch_optional(&self.pool)
139 .await
140 .map_err(|e| DomainError::DatabaseError(format!("Failed to get provider: {}", e)))?;
141
142 match row {
143 Some(row) => {
144 let config_json: String = row.get("config_json");
145 let config: HashMap<String, String> = if config_json.trim().is_empty() {
146 HashMap::new()
147 } else {
148 serde_json::from_str(&config_json).map_err(|e| {
149 DomainError::DatabaseError(format!("Failed to parse config: {}", e))
150 })?
151 };
152
153 let token_ref: String = row.get("token_encrypted");
154
155 Ok(Some(ProviderConfig {
156 id: Some(id),
157 name: row.get("name"),
158 provider_type: row.get("provider_type"),
159 config,
160 token: token_ref,
161 refresh_interval: row.get("refresh_interval"),
162 version: Some(row.get("version")),
163 }))
164 }
165 None => Ok(None),
166 }
167 })
168 .await
169 }
170
171 async fn create_provider(&self, config: &ProviderConfig) -> DomainResult<i64> {
172 retry_on_busy(|| async {
173 let existing = sqlx::query_scalar::<_, i64>("SELECT id FROM providers WHERE name = ?")
174 .bind(&config.name)
175 .fetch_optional(&self.pool)
176 .await
177 .map_err(|e| {
178 DomainError::DatabaseError(format!("Failed to check existing: {}", e))
179 })?;
180
181 if existing.is_some() {
182 return Err(DomainError::InvalidConfig(format!(
183 "Provider '{}' already exists",
184 config.name
185 )));
186 }
187
188 let config_json = serde_json::to_string(&config.config)
189 .map_err(|e| DomainError::DatabaseError(format!("Failed to serialize config: {}", e)))?;
190
191 let result = sqlx::query(
192 r#"INSERT INTO providers (name, provider_type, token_encrypted, config_json, refresh_interval)
193 VALUES (?, ?, ?, ?, ?)"#,
194 )
195 .bind(&config.name)
196 .bind(&config.provider_type)
197 .bind(&config.token)
198 .bind(&config_json)
199 .bind(config.refresh_interval)
200 .execute(&self.pool)
201 .await
202 .map_err(|e| DomainError::DatabaseError(format!("Failed to insert provider: {}", e)))?;
203
204 Ok(result.last_insert_rowid())
205 })
206 .await
207 }
208
209 async fn update_provider(&self, id: i64, config: &ProviderConfig) -> DomainResult<()> {
210 retry_on_busy(|| async {
211 let config_json = serde_json::to_string(&config.config)
212 .map_err(|e| DomainError::DatabaseError(format!("Failed to serialize config: {}", e)))?;
213
214 let result = sqlx::query(
215 r#"UPDATE providers
216 SET name = ?, provider_type = ?, token_encrypted = ?, config_json = ?, refresh_interval = ?, updated_at = datetime('now')
217 WHERE id = ?"#,
218 )
219 .bind(&config.name)
220 .bind(&config.provider_type)
221 .bind(&config.token)
222 .bind(&config_json)
223 .bind(config.refresh_interval)
224 .bind(id)
225 .execute(&self.pool)
226 .await
227 .map_err(|e| DomainError::DatabaseError(format!("Failed to update provider: {}", e)))?;
228
229 if result.rows_affected() == 0 {
230 return Err(DomainError::ProviderNotFound(format!("Provider {} not found", id)));
231 }
232
233 Ok(())
234 })
235 .await
236 }
237
238 async fn delete_provider(&self, id: i64) -> DomainResult<()> {
239 retry_on_busy(|| async {
240 let _ = sqlx::query("DELETE FROM pipelines_cache WHERE provider_id = ?")
241 .bind(id)
242 .execute(&self.pool)
243 .await;
244
245 let _ = sqlx::query("DELETE FROM provider_permissions WHERE provider_id = ?")
246 .bind(id)
247 .execute(&self.pool)
248 .await;
249
250 let _ = sqlx::query("DELETE FROM table_preferences WHERE provider_id = ?")
251 .bind(id)
252 .execute(&self.pool)
253 .await;
254
255 let result = sqlx::query("DELETE FROM providers WHERE id = ?")
256 .bind(id)
257 .execute(&self.pool)
258 .await
259 .map_err(|e| {
260 DomainError::DatabaseError(format!("Failed to delete provider: {}", e))
261 })?;
262
263 if result.rows_affected() == 0 {
264 return Err(DomainError::ProviderNotFound(format!(
265 "Provider {} not found",
266 id
267 )));
268 }
269
270 Ok(())
271 })
272 .await
273 }
274
275 async fn get_table_preferences(
276 &self, provider_id: i64, table_id: &str,
277 ) -> DomainResult<Option<String>> {
278 retry_on_busy(|| async {
279 let result = sqlx::query_scalar::<_, String>(
280 "SELECT preferences_json FROM table_preferences WHERE provider_id = ? AND table_id = ?",
281 )
282 .bind(provider_id)
283 .bind(table_id)
284 .fetch_optional(&self.pool)
285 .await
286 .map_err(|e| DomainError::DatabaseError(format!("Failed to get preferences: {}", e)))?;
287
288 Ok(result)
289 })
290 .await
291 }
292
293 async fn set_table_preferences(
294 &self, provider_id: i64, table_id: &str, preferences_json: &str,
295 ) -> DomainResult<()> {
296 retry_on_busy(|| async {
297 sqlx::query(
298 r#"INSERT INTO table_preferences (provider_id, table_id, preferences_json, created_at, updated_at)
299 VALUES (?, ?, ?, datetime('now'), datetime('now'))
300 ON CONFLICT(provider_id, table_id) DO UPDATE SET
301 preferences_json = excluded.preferences_json,
302 updated_at = datetime('now')"#,
303 )
304 .bind(provider_id)
305 .bind(table_id)
306 .bind(preferences_json)
307 .execute(&self.pool)
308 .await
309 .map_err(|e| DomainError::DatabaseError(format!("Failed to save preferences: {}", e)))?;
310
311 Ok(())
312 })
313 .await
314 }
315
316 async fn store_permissions(
317 &self, provider_id: i64, permissions: &StoredPermissions,
318 ) -> DomainResult<()> {
319 let start = std::time::Instant::now();
320 let permissions_count = permissions.permissions.len();
321
322 retry_on_busy(|| async {
323 let checked_at_str = permissions.last_checked.to_rfc3339();
324
325 sqlx::query("DELETE FROM provider_permissions WHERE provider_id = ?")
326 .bind(provider_id)
327 .execute(&self.pool)
328 .await
329 .map_err(|e| {
330 DomainError::DatabaseError(format!("Failed to clear permissions: {}", e))
331 })?;
332
333 if permissions.permissions.is_empty() {
334 return Ok(());
335 }
336
337 const BATCH_SIZE: usize = 100;
338
339 let perms_vec: Vec<_> = permissions.permissions.iter().collect();
340
341 for chunk in perms_vec.chunks(BATCH_SIZE) {
342 let values_clause = chunk
343 .iter()
344 .map(|_| "(?, ?, ?, ?)")
345 .collect::<Vec<_>>()
346 .join(", ");
347
348 let sql = format!(
349 r#"INSERT INTO provider_permissions (provider_id, permission_name, granted, checked_at)
350 VALUES {}"#,
351 values_clause
352 );
353
354 let mut query = sqlx::query(&sql);
355 for (permission_name, granted) in chunk {
356 query = query
357 .bind(provider_id)
358 .bind(permission_name)
359 .bind(**granted)
360 .bind(&checked_at_str);
361 }
362
363 query
364 .execute(&self.pool)
365 .await
366 .map_err(|e| {
367 DomainError::DatabaseError(format!("Failed to store permission: {}", e))
368 })?;
369 }
370
371 Ok(())
372 })
373 .await?;
374
375 let elapsed = start.elapsed();
376 tracing::debug!(
377 provider_id = provider_id,
378 permissions_count = permissions_count,
379 elapsed_ms = elapsed.as_millis(),
380 "Stored provider permissions (batch insert)"
381 );
382
383 Ok(())
384 }
385
386 async fn get_permissions(&self, provider_id: i64) -> DomainResult<Option<StoredPermissions>> {
387 retry_on_busy(|| async {
388 let rows = sqlx::query(
389 r#"SELECT permission_name, granted, checked_at
390 FROM provider_permissions
391 WHERE provider_id = ?"#,
392 )
393 .bind(provider_id)
394 .fetch_all(&self.pool)
395 .await
396 .map_err(|e| DomainError::DatabaseError(format!("Failed to get permissions: {}", e)))?;
397
398 if rows.is_empty() {
399 return Ok(None);
400 }
401
402 let checked_at_str: String = rows[0].get("checked_at");
403 let last_checked = chrono::DateTime::parse_from_rfc3339(&checked_at_str)
404 .map_err(|e| {
405 DomainError::DatabaseError(format!("Failed to parse timestamp: {}", e))
406 })?
407 .with_timezone(&chrono::Utc);
408
409 let mut permissions = HashMap::new();
410 for row in rows {
411 let permission_name: String = row.get("permission_name");
412 let granted: bool = row.get("granted");
413 permissions.insert(permission_name, granted);
414 }
415
416 Ok(Some(StoredPermissions {
417 permissions,
418 last_checked,
419 }))
420 })
421 .await
422 }
423
424 async fn export_all(&self) -> DomainResult<ConfigExport> {
425 let providers = self.list_providers().await?;
426
427 let pref_rows =
428 sqlx::query(r#"SELECT provider_id, table_id, preferences_json FROM table_preferences"#)
429 .fetch_all(&self.pool)
430 .await
431 .map_err(|e| {
432 DomainError::DatabaseError(format!("Failed to export preferences: {}", e))
433 })?;
434
435 let mut table_preferences = HashMap::new();
436 for row in pref_rows {
437 let provider_id: i64 = row.get("provider_id");
438 let table_id: String = row.get("table_id");
439 let preferences_json: String = row.get("preferences_json");
440 let key = format!("{}:{}", provider_id, table_id);
441 table_preferences.insert(key, preferences_json);
442 }
443
444 let perm_rows = sqlx::query(
445 r#"SELECT provider_id, permission_name, granted, checked_at FROM provider_permissions"#,
446 )
447 .fetch_all(&self.pool)
448 .await
449 .map_err(|e| DomainError::DatabaseError(format!("Failed to export permissions: {}", e)))?;
450
451 let mut permissions: HashMap<i64, StoredPermissions> = HashMap::new();
452 for row in perm_rows {
453 let provider_id: i64 = row.get("provider_id");
454 let permission_name: String = row.get("permission_name");
455 let granted: bool = row.get("granted");
456 let checked_at_str: String = row.get("checked_at");
457
458 let last_checked = chrono::DateTime::parse_from_rfc3339(&checked_at_str)
459 .map(|dt| dt.with_timezone(&chrono::Utc))
460 .unwrap_or_else(|_| chrono::Utc::now());
461
462 let entry = permissions
463 .entry(provider_id)
464 .or_insert_with(|| StoredPermissions {
465 permissions: HashMap::new(),
466 last_checked,
467 });
468 entry.permissions.insert(permission_name, granted);
469 }
470
471 Ok(ConfigExport {
472 version: "1.0".to_string(),
473 providers,
474 table_preferences,
475 permissions,
476 })
477 }
478
479 async fn import_all(&self, data: &ConfigExport) -> DomainResult<HashMap<i64, i64>> {
480 let mut id_mapping: HashMap<i64, i64> = HashMap::new();
481
482 for provider in &data.providers {
483 let old_id = provider.id.unwrap_or(0);
484
485 let existing = sqlx::query_scalar::<_, i64>("SELECT id FROM providers WHERE name = ?")
486 .bind(&provider.name)
487 .fetch_optional(&self.pool)
488 .await
489 .map_err(|e| {
490 DomainError::DatabaseError(format!("Failed to check existing: {}", e))
491 })?;
492
493 let new_id = if let Some(existing_id) = existing {
494 existing_id
495 } else {
496 self.create_provider(provider).await?
497 };
498
499 if old_id != 0 {
500 id_mapping.insert(old_id, new_id);
501 }
502 }
503
504 for (key, preferences_json) in &data.table_preferences {
505 let parts: Vec<&str> = key.splitn(2, ':').collect();
506 if parts.len() == 2 {
507 if let Ok(old_provider_id) = parts[0].parse::<i64>() {
508 let table_id = parts[1];
509 let new_provider_id = id_mapping
510 .get(&old_provider_id)
511 .copied()
512 .unwrap_or(old_provider_id);
513 if self.get_provider(new_provider_id).await?.is_some() {
514 self.set_table_preferences(new_provider_id, table_id, preferences_json)
515 .await?;
516 }
517 }
518 }
519 }
520
521 for (old_provider_id, perms) in &data.permissions {
522 let new_provider_id = id_mapping
523 .get(old_provider_id)
524 .copied()
525 .unwrap_or(*old_provider_id);
526 if self.get_provider(new_provider_id).await?.is_some() {
527 self.store_permissions(new_provider_id, perms).await?;
528 }
529 }
530
531 Ok(id_mapping)
532 }
533}
534
535#[cfg(test)]
536mod tests {
537 use super::*;
538 use crate::infrastructure::database::init_database;
539
540 #[tokio::test]
541 async fn test_sqlite_backend_providers() {
542 let dir = tempfile::tempdir().unwrap();
543 let db_path = dir.path().join("test.db");
544 let pool = init_database(db_path).await.unwrap();
545 let backend = SqliteConfigBackend::new(pool);
546
547 let config = ProviderConfig {
548 id: None,
549 name: "Test Provider".to_string(),
550 provider_type: "github".to_string(),
551 config: HashMap::from([("owner".to_string(), "test".to_string())]),
552 token: String::new(),
553 refresh_interval: 30,
554 version: None,
555 };
556
557 let id = backend.create_provider(&config).await.unwrap();
558 assert!(id > 0);
559
560 let retrieved = backend.get_provider(id).await.unwrap().unwrap();
561 assert_eq!(retrieved.name, "Test Provider");
562 assert_eq!(retrieved.provider_type, "github");
563
564 let all = backend.list_providers().await.unwrap();
565 assert_eq!(all.len(), 1);
566
567 let updated_config = ProviderConfig {
568 id: Some(id),
569 name: "Updated Provider".to_string(),
570 provider_type: "github".to_string(),
571 config: HashMap::from([("owner".to_string(), "updated".to_string())]),
572 token: String::new(),
573 refresh_interval: 60,
574 version: None,
575 };
576 backend.update_provider(id, &updated_config).await.unwrap();
577
578 let retrieved = backend.get_provider(id).await.unwrap().unwrap();
579 assert_eq!(retrieved.name, "Updated Provider");
580 assert_eq!(retrieved.refresh_interval, 60);
581
582 backend.delete_provider(id).await.unwrap();
583 assert!(backend.get_provider(id).await.unwrap().is_none());
584 }
585
586 #[tokio::test]
587 async fn test_sqlite_backend_table_preferences() {
588 let dir = tempfile::tempdir().unwrap();
589 let db_path = dir.path().join("test.db");
590 let pool = init_database(db_path).await.unwrap();
591 let backend = SqliteConfigBackend::new(pool);
592
593 let config = ProviderConfig {
594 id: None,
595 name: "Prefs Provider".to_string(),
596 provider_type: "github".to_string(),
597 config: HashMap::new(),
598 token: String::new(),
599 refresh_interval: 30,
600 version: None,
601 };
602 let provider_id = backend.create_provider(&config).await.unwrap();
603
604 let prefs_json = r#"{"columns":["name","status"],"sort":"name"}"#;
605 backend
606 .set_table_preferences(provider_id, "pipelines", prefs_json)
607 .await
608 .unwrap();
609
610 let retrieved = backend
611 .get_table_preferences(provider_id, "pipelines")
612 .await
613 .unwrap();
614 assert_eq!(retrieved, Some(prefs_json.to_string()));
615
616 let missing = backend
617 .get_table_preferences(provider_id, "other")
618 .await
619 .unwrap();
620 assert!(missing.is_none());
621 }
622
623 #[tokio::test]
624 async fn test_sqlite_backend_permissions() {
625 let dir = tempfile::tempdir().unwrap();
626 let db_path = dir.path().join("test.db");
627 let pool = init_database(db_path).await.unwrap();
628 let backend = SqliteConfigBackend::new(pool);
629
630 let config = ProviderConfig {
631 id: None,
632 name: "Perms Provider".to_string(),
633 provider_type: "github".to_string(),
634 config: HashMap::new(),
635 token: String::new(),
636 refresh_interval: 30,
637 version: None,
638 };
639 let provider_id = backend.create_provider(&config).await.unwrap();
640
641 let mut perms_map = HashMap::new();
642 perms_map.insert("repo:read".to_string(), true);
643 perms_map.insert("repo:write".to_string(), false);
644 let permissions = StoredPermissions {
645 permissions: perms_map,
646 last_checked: chrono::Utc::now(),
647 };
648 backend
649 .store_permissions(provider_id, &permissions)
650 .await
651 .unwrap();
652
653 let retrieved = backend.get_permissions(provider_id).await.unwrap().unwrap();
654 assert_eq!(retrieved.permissions.get("repo:read"), Some(&true));
655 assert_eq!(retrieved.permissions.get("repo:write"), Some(&false));
656 }
657
658 #[tokio::test]
659 async fn test_sqlite_backend_export_import() {
660 let dir = tempfile::tempdir().unwrap();
661 let db_path = dir.path().join("test.db");
662 let pool = init_database(db_path).await.unwrap();
663 let backend = SqliteConfigBackend::new(pool);
664
665 let config = ProviderConfig {
666 id: None,
667 name: "Export Test".to_string(),
668 provider_type: "gitlab".to_string(),
669 config: HashMap::from([("host".to_string(), "gitlab.com".to_string())]),
670 token: String::new(),
671 refresh_interval: 45,
672 version: None,
673 };
674 backend.create_provider(&config).await.unwrap();
675
676 let export = backend.export_all().await.unwrap();
677 assert_eq!(export.providers.len(), 1);
678 assert_eq!(export.providers[0].name, "Export Test");
679
680 let dir2 = tempfile::tempdir().unwrap();
681 let db_path2 = dir2.path().join("test2.db");
682 let pool2 = init_database(db_path2).await.unwrap();
683 let backend2 = SqliteConfigBackend::new(pool2);
684
685 backend2.import_all(&export).await.unwrap();
686
687 let imported = backend2.list_providers().await.unwrap();
688 assert_eq!(imported.len(), 1);
689 assert_eq!(imported[0].name, "Export Test");
690 }
691}