pipedash_core/infrastructure/database/
device_repository.rs

1use std::collections::HashMap;
2
3use chrono::{
4    DateTime,
5    Utc,
6};
7
8use super::DatabasePool;
9use crate::domain::{
10    DomainError,
11    DomainResult,
12    NotificationPreferences,
13    PairedDevice,
14};
15
16type Row = (
17    String,
18    String,
19    String,
20    String,
21    DateTime<Utc>,
22    Option<DateTime<Utc>>,
23);
24
25const COLUMNS: &str = "id, token_hash, name, platform, created_at, last_seen_at";
26const UPDATE_ATTEMPTS: usize = 8;
27
28fn db_error(error: sqlx::Error) -> DomainError {
29    DomainError::DatabaseError(error.to_string())
30}
31
32fn device(row: Row) -> PairedDevice {
33    let (id, token_hash, name, platform, created_at, last_seen_at) = row;
34
35    PairedDevice {
36        id,
37        token_hash,
38        name,
39        platform,
40        created_at,
41        last_seen_at,
42    }
43}
44
45pub struct DeviceRepository {
46    pool: DatabasePool,
47}
48
49impl DeviceRepository {
50    pub fn new(pool: DatabasePool) -> Self {
51        Self { pool }
52    }
53
54    fn placeholder(&self, index: usize) -> String {
55        match &self.pool {
56            DatabasePool::Sqlite(_) => "?".to_string(),
57            DatabasePool::Postgres(_) => format!("${index}"),
58        }
59    }
60
61    pub async fn insert(&self, device: &PairedDevice) -> DomainResult<()> {
62        let sql = format!(
63            "INSERT INTO paired_devices ({COLUMNS}) VALUES ({}, {}, {}, {}, {}, {})",
64            self.placeholder(1),
65            self.placeholder(2),
66            self.placeholder(3),
67            self.placeholder(4),
68            self.placeholder(5),
69            self.placeholder(6),
70        );
71        self.execute_insert(sql, device).await
72    }
73
74    pub async fn list(&self) -> DomainResult<Vec<PairedDevice>> {
75        let sql = format!("SELECT {COLUMNS} FROM paired_devices ORDER BY created_at, id");
76        let rows = match &self.pool {
77            DatabasePool::Sqlite(p) => {
78                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
79                    .fetch_all(p)
80                    .await
81            }
82            DatabasePool::Postgres(p) => {
83                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
84                    .fetch_all(p)
85                    .await
86            }
87        }
88        .map_err(db_error)?;
89
90        Ok(rows.into_iter().map(device).collect())
91    }
92
93    pub async fn find_by_token(&self, token: &str) -> DomainResult<Option<PairedDevice>> {
94        let sql = format!(
95            "SELECT {COLUMNS} FROM paired_devices WHERE token_hash = {}",
96            self.placeholder(1)
97        );
98        let token_hash = PairedDevice::hash_token(token);
99        let row = match &self.pool {
100            DatabasePool::Sqlite(p) => {
101                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
102                    .bind(token_hash)
103                    .fetch_optional(p)
104                    .await
105            }
106            DatabasePool::Postgres(p) => {
107                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
108                    .bind(token_hash)
109                    .fetch_optional(p)
110                    .await
111            }
112        }
113        .map_err(db_error)?;
114
115        Ok(row.map(device))
116    }
117
118    pub async fn touch(&self, id: &str) -> DomainResult<()> {
119        let sql = format!(
120            "UPDATE paired_devices SET last_seen_at = {} WHERE id = {}",
121            self.placeholder(1),
122            self.placeholder(2)
123        );
124        let now = Utc::now();
125        match &self.pool {
126            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
127                .bind(now)
128                .bind(id)
129                .execute(p)
130                .await
131                .map(|_| ()),
132            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
133                .bind(now)
134                .bind(id)
135                .execute(p)
136                .await
137                .map(|_| ()),
138        }
139        .map_err(db_error)
140    }
141
142    pub async fn set_notification_preferences(
143        &self, id: &str, preferences: &NotificationPreferences,
144    ) -> DomainResult<bool> {
145        let sql = format!(
146            "UPDATE paired_devices SET notification_preferences = {} WHERE id = {}",
147            self.placeholder(1),
148            self.placeholder(2)
149        );
150        let json = serde_json::to_string(preferences)
151            .map_err(|error| DomainError::DatabaseError(error.to_string()))?;
152        let updated = match &self.pool {
153            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
154                .bind(json)
155                .bind(id)
156                .execute(p)
157                .await
158                .map(|result| result.rows_affected()),
159            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
160                .bind(json)
161                .bind(id)
162                .execute(p)
163                .await
164                .map(|result| result.rows_affected()),
165        }
166        .map_err(db_error)?;
167
168        Ok(updated > 0)
169    }
170
171    pub async fn update_notification_preferences(
172        &self, id: &str, mut change: impl FnMut(&mut NotificationPreferences) -> bool,
173    ) -> DomainResult<Option<NotificationPreferences>> {
174        for _ in 0..UPDATE_ATTEMPTS {
175            let Some(stored) = self.preferences_json(id).await? else {
176                return Ok(None);
177            };
178            let mut preferences: NotificationPreferences = stored
179                .as_deref()
180                .and_then(|json| serde_json::from_str(json).ok())
181                .unwrap_or_default();
182            if !change(&mut preferences) {
183                return Ok(Some(preferences));
184            }
185            let json = serde_json::to_string(&preferences)
186                .map_err(|error| DomainError::DatabaseError(error.to_string()))?;
187            if self.swap_preferences(id, stored.as_deref(), &json).await? {
188                return Ok(Some(preferences));
189            }
190        }
191
192        Err(DomainError::DatabaseError(
193            "Notification preferences kept changing while being saved".into(),
194        ))
195    }
196
197    async fn preferences_json(&self, id: &str) -> DomainResult<Option<Option<String>>> {
198        let sql = format!(
199            "SELECT notification_preferences FROM paired_devices WHERE id = {}",
200            self.placeholder(1)
201        );
202        let row = match &self.pool {
203            DatabasePool::Sqlite(p) => {
204                sqlx::query_as::<_, (Option<String>,)>(sqlx::AssertSqlSafe(sql))
205                    .bind(id)
206                    .fetch_optional(p)
207                    .await
208            }
209            DatabasePool::Postgres(p) => {
210                sqlx::query_as::<_, (Option<String>,)>(sqlx::AssertSqlSafe(sql))
211                    .bind(id)
212                    .fetch_optional(p)
213                    .await
214            }
215        }
216        .map_err(db_error)?;
217
218        Ok(row.map(|(json,)| json))
219    }
220
221    async fn swap_preferences(
222        &self, id: &str, expected: Option<&str>, json: &str,
223    ) -> DomainResult<bool> {
224        let same = match &self.pool {
225            DatabasePool::Sqlite(_) => "IS",
226            DatabasePool::Postgres(_) => "IS NOT DISTINCT FROM",
227        };
228        let sql = format!(
229            "UPDATE paired_devices SET notification_preferences = {} \
230             WHERE id = {} AND notification_preferences {same} {}",
231            self.placeholder(1),
232            self.placeholder(2),
233            self.placeholder(3)
234        );
235        let updated = match &self.pool {
236            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
237                .bind(json)
238                .bind(id)
239                .bind(expected)
240                .execute(p)
241                .await
242                .map(|result| result.rows_affected()),
243            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
244                .bind(json)
245                .bind(id)
246                .bind(expected)
247                .execute(p)
248                .await
249                .map(|result| result.rows_affected()),
250        }
251        .map_err(db_error)?;
252
253        Ok(updated > 0)
254    }
255
256    pub async fn notification_preferences(
257        &self,
258    ) -> DomainResult<HashMap<String, NotificationPreferences>> {
259        let sql = "SELECT id, notification_preferences FROM paired_devices \
260                   WHERE notification_preferences IS NOT NULL";
261        let rows = match &self.pool {
262            DatabasePool::Sqlite(p) => {
263                sqlx::query_as::<_, (String, String)>(sql)
264                    .fetch_all(p)
265                    .await
266            }
267            DatabasePool::Postgres(p) => {
268                sqlx::query_as::<_, (String, String)>(sql)
269                    .fetch_all(p)
270                    .await
271            }
272        }
273        .map_err(db_error)?;
274
275        Ok(rows
276            .into_iter()
277            .filter_map(|(id, json)| Some((id, serde_json::from_str(&json).ok()?)))
278            .collect())
279    }
280
281    pub async fn delete(&self, id: &str) -> DomainResult<bool> {
282        let sql = format!(
283            "DELETE FROM paired_devices WHERE id = {}",
284            self.placeholder(1)
285        );
286        let deleted = match &self.pool {
287            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
288                .bind(id)
289                .execute(p)
290                .await
291                .map(|result| result.rows_affected()),
292            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
293                .bind(id)
294                .execute(p)
295                .await
296                .map(|result| result.rows_affected()),
297        }
298        .map_err(db_error)?;
299
300        Ok(deleted > 0)
301    }
302
303    pub async fn delete_all(&self) -> DomainResult<u64> {
304        let sql = "DELETE FROM paired_devices";
305        match &self.pool {
306            DatabasePool::Sqlite(p) => sqlx::query(sql)
307                .execute(p)
308                .await
309                .map(|result| result.rows_affected()),
310            DatabasePool::Postgres(p) => sqlx::query(sql)
311                .execute(p)
312                .await
313                .map(|result| result.rows_affected()),
314        }
315        .map_err(db_error)
316    }
317
318    pub async fn import(&self, devices: &[PairedDevice]) -> DomainResult<()> {
319        let sql = format!(
320            "INSERT INTO paired_devices ({COLUMNS}) VALUES ({}, {}, {}, {}, {}, {})
321             ON CONFLICT (id) DO NOTHING",
322            self.placeholder(1),
323            self.placeholder(2),
324            self.placeholder(3),
325            self.placeholder(4),
326            self.placeholder(5),
327            self.placeholder(6),
328        );
329        for device in devices {
330            self.execute_insert(sql.clone(), device).await?;
331        }
332
333        Ok(())
334    }
335
336    async fn execute_insert(&self, sql: String, device: &PairedDevice) -> DomainResult<()> {
337        match &self.pool {
338            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
339                .bind(&device.id)
340                .bind(&device.token_hash)
341                .bind(&device.name)
342                .bind(&device.platform)
343                .bind(device.created_at)
344                .bind(device.last_seen_at)
345                .execute(p)
346                .await
347                .map(|_| ()),
348            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
349                .bind(&device.id)
350                .bind(&device.token_hash)
351                .bind(&device.name)
352                .bind(&device.platform)
353                .bind(device.created_at)
354                .bind(device.last_seen_at)
355                .execute(p)
356                .await
357                .map(|_| ()),
358        }
359        .map_err(db_error)
360    }
361}
362
363#[cfg(test)]
364mod tests {
365    use super::*;
366    use crate::infrastructure::database::init_database;
367
368    #[tokio::test]
369    async fn sqlite_devices_are_found_by_token_touched_and_deleted() {
370        let dir = tempfile::tempdir().unwrap();
371        let pool = init_database(dir.path().join("devices.db")).await.unwrap();
372        let repository = DeviceRepository::new(DatabasePool::Sqlite(pool));
373        let (device, token) = PairedDevice::issue("Pixel", "android");
374
375        repository.insert(&device).await.unwrap();
376
377        let found = repository.find_by_token(&token).await.unwrap().unwrap();
378        assert_eq!(found.id, device.id);
379        assert!(found.last_seen_at.is_some());
380        assert_eq!(repository.find_by_token("wrong").await.unwrap(), None);
381        assert_eq!(
382            repository.find_by_token(&device.token_hash).await.unwrap(),
383            None
384        );
385
386        assert!(repository
387            .notification_preferences()
388            .await
389            .unwrap()
390            .is_empty());
391        let muted = NotificationPreferences {
392            long_run: false,
393            muted_pipelines: vec!["github__1__api".into()],
394            ..NotificationPreferences::default()
395        };
396        assert!(repository
397            .set_notification_preferences(&device.id, &muted)
398            .await
399            .unwrap());
400        assert!(!repository
401            .set_notification_preferences("unknown", &muted)
402            .await
403            .unwrap());
404        assert_eq!(
405            repository.notification_preferences().await.unwrap(),
406            HashMap::from([(device.id.clone(), muted)])
407        );
408
409        repository.touch(&device.id).await.unwrap();
410        let listed = repository.list().await.unwrap();
411        assert_eq!(listed.len(), 1);
412        assert!(listed[0].last_seen_at.is_some());
413
414        assert!(repository.delete(&device.id).await.unwrap());
415        assert!(!repository.delete(&device.id).await.unwrap());
416        assert_eq!(repository.find_by_token(&token).await.unwrap(), None);
417
418        repository.import(&listed).await.unwrap();
419        repository.import(&listed).await.unwrap();
420        assert_eq!(repository.list().await.unwrap(), listed);
421        assert_eq!(repository.delete_all().await.unwrap(), 1);
422        assert!(repository.list().await.unwrap().is_empty());
423    }
424
425    #[tokio::test(flavor = "multi_thread")]
426    async fn a_change_saved_between_read_and_write_is_kept_and_reapplied() {
427        let dir = tempfile::tempdir().unwrap();
428        let pool = init_database(dir.path().join("devices.db")).await.unwrap();
429        let repository = DeviceRepository::new(DatabasePool::Sqlite(pool));
430        let (device, _) = PairedDevice::issue("Pixel", "android");
431        repository.insert(&device).await.unwrap();
432        let request = crate::domain::OnceWatch {
433            pipeline_id: "github__1__api".into(),
434            since: Utc::now(),
435            fired_at: None,
436        };
437        let finish = crate::domain::RunFinish {
438            at: request.since + chrono::Duration::minutes(1),
439            previous: None,
440        };
441        let incoming = NotificationPreferences {
442            outage: false,
443            watch_once: vec![request],
444            ..NotificationPreferences::default()
445        };
446        repository
447            .set_notification_preferences(&device.id, &incoming)
448            .await
449            .unwrap();
450        let mut attempts = 0;
451
452        let saved = repository
453            .update_notification_preferences(&device.id, |stored| {
454                attempts += 1;
455                if attempts == 1 {
456                    let mut fired = stored.clone();
457                    assert!(!fired.fire_once("github__1__api", &finish).is_empty());
458                    tokio::task::block_in_place(|| {
459                        tokio::runtime::Handle::current()
460                            .block_on(repository.set_notification_preferences(&device.id, &fired))
461                    })
462                    .unwrap();
463                }
464                let mut next = incoming.clone();
465                next.keep_fired(stored);
466                *stored = next;
467                true
468            })
469            .await
470            .unwrap()
471            .unwrap();
472
473        assert_eq!(attempts, 2);
474        assert_eq!(saved.watch_once[0].fired_at, Some(finish.at));
475        assert_eq!(
476            repository.notification_preferences().await.unwrap()[&device.id],
477            saved
478        );
479        assert_eq!(
480            repository
481                .update_notification_preferences("unknown", |_| true)
482                .await
483                .unwrap(),
484            None
485        );
486    }
487}
pipedash_core/infrastructure/database/device_repository.rs

pipedash_core/infrastructure/database/
device_repository.rs
