pipedash_core/infrastructure/database/
device_repository.rs

1use chrono::{
2    DateTime,
3    Utc,
4};
5
6use super::DatabasePool;
7use crate::domain::{
8    DomainError,
9    DomainResult,
10    PairedDevice,
11};
12
13type Row = (
14    String,
15    String,
16    String,
17    String,
18    DateTime<Utc>,
19    Option<DateTime<Utc>>,
20);
21
22const COLUMNS: &str = "id, token_hash, name, platform, created_at, last_seen_at";
23
24fn db_error(error: sqlx::Error) -> DomainError {
25    DomainError::DatabaseError(error.to_string())
26}
27
28fn device(row: Row) -> PairedDevice {
29    let (id, token_hash, name, platform, created_at, last_seen_at) = row;
30
31    PairedDevice {
32        id,
33        token_hash,
34        name,
35        platform,
36        created_at,
37        last_seen_at,
38    }
39}
40
41pub struct DeviceRepository {
42    pool: DatabasePool,
43}
44
45impl DeviceRepository {
46    pub fn new(pool: DatabasePool) -> Self {
47        Self { pool }
48    }
49
50    fn placeholder(&self, index: usize) -> String {
51        match &self.pool {
52            DatabasePool::Sqlite(_) => "?".to_string(),
53            DatabasePool::Postgres(_) => format!("${index}"),
54        }
55    }
56
57    pub async fn insert(&self, device: &PairedDevice) -> DomainResult<()> {
58        let sql = format!(
59            "INSERT INTO paired_devices ({COLUMNS}) VALUES ({}, {}, {}, {}, {}, {})",
60            self.placeholder(1),
61            self.placeholder(2),
62            self.placeholder(3),
63            self.placeholder(4),
64            self.placeholder(5),
65            self.placeholder(6),
66        );
67        self.execute_insert(sql, device).await
68    }
69
70    pub async fn list(&self) -> DomainResult<Vec<PairedDevice>> {
71        let sql = format!("SELECT {COLUMNS} FROM paired_devices ORDER BY created_at, id");
72        let rows = match &self.pool {
73            DatabasePool::Sqlite(p) => {
74                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
75                    .fetch_all(p)
76                    .await
77            }
78            DatabasePool::Postgres(p) => {
79                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
80                    .fetch_all(p)
81                    .await
82            }
83        }
84        .map_err(db_error)?;
85
86        Ok(rows.into_iter().map(device).collect())
87    }
88
89    pub async fn find_by_token(&self, token: &str) -> DomainResult<Option<PairedDevice>> {
90        let sql = format!(
91            "SELECT {COLUMNS} FROM paired_devices WHERE token_hash = {}",
92            self.placeholder(1)
93        );
94        let token_hash = PairedDevice::hash_token(token);
95        let row = match &self.pool {
96            DatabasePool::Sqlite(p) => {
97                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
98                    .bind(token_hash)
99                    .fetch_optional(p)
100                    .await
101            }
102            DatabasePool::Postgres(p) => {
103                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
104                    .bind(token_hash)
105                    .fetch_optional(p)
106                    .await
107            }
108        }
109        .map_err(db_error)?;
110
111        Ok(row.map(device))
112    }
113
114    pub async fn touch(&self, id: &str) -> DomainResult<()> {
115        let sql = format!(
116            "UPDATE paired_devices SET last_seen_at = {} WHERE id = {}",
117            self.placeholder(1),
118            self.placeholder(2)
119        );
120        let now = Utc::now();
121        match &self.pool {
122            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
123                .bind(now)
124                .bind(id)
125                .execute(p)
126                .await
127                .map(|_| ()),
128            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
129                .bind(now)
130                .bind(id)
131                .execute(p)
132                .await
133                .map(|_| ()),
134        }
135        .map_err(db_error)
136    }
137
138    pub async fn delete(&self, id: &str) -> DomainResult<bool> {
139        let sql = format!(
140            "DELETE FROM paired_devices WHERE id = {}",
141            self.placeholder(1)
142        );
143        let deleted = match &self.pool {
144            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
145                .bind(id)
146                .execute(p)
147                .await
148                .map(|result| result.rows_affected()),
149            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
150                .bind(id)
151                .execute(p)
152                .await
153                .map(|result| result.rows_affected()),
154        }
155        .map_err(db_error)?;
156
157        Ok(deleted > 0)
158    }
159
160    pub async fn delete_all(&self) -> DomainResult<u64> {
161        let sql = "DELETE FROM paired_devices";
162        match &self.pool {
163            DatabasePool::Sqlite(p) => sqlx::query(sql)
164                .execute(p)
165                .await
166                .map(|result| result.rows_affected()),
167            DatabasePool::Postgres(p) => sqlx::query(sql)
168                .execute(p)
169                .await
170                .map(|result| result.rows_affected()),
171        }
172        .map_err(db_error)
173    }
174
175    pub async fn import(&self, devices: &[PairedDevice]) -> DomainResult<()> {
176        let sql = format!(
177            "INSERT INTO paired_devices ({COLUMNS}) VALUES ({}, {}, {}, {}, {}, {})
178             ON CONFLICT (id) DO NOTHING",
179            self.placeholder(1),
180            self.placeholder(2),
181            self.placeholder(3),
182            self.placeholder(4),
183            self.placeholder(5),
184            self.placeholder(6),
185        );
186        for device in devices {
187            self.execute_insert(sql.clone(), device).await?;
188        }
189
190        Ok(())
191    }
192
193    async fn execute_insert(&self, sql: String, device: &PairedDevice) -> DomainResult<()> {
194        match &self.pool {
195            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
196                .bind(&device.id)
197                .bind(&device.token_hash)
198                .bind(&device.name)
199                .bind(&device.platform)
200                .bind(device.created_at)
201                .bind(device.last_seen_at)
202                .execute(p)
203                .await
204                .map(|_| ()),
205            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
206                .bind(&device.id)
207                .bind(&device.token_hash)
208                .bind(&device.name)
209                .bind(&device.platform)
210                .bind(device.created_at)
211                .bind(device.last_seen_at)
212                .execute(p)
213                .await
214                .map(|_| ()),
215        }
216        .map_err(db_error)
217    }
218}
219
220#[cfg(test)]
221mod tests {
222    use super::*;
223    use crate::infrastructure::database::init_database;
224
225    #[tokio::test]
226    async fn sqlite_devices_are_found_by_token_touched_and_deleted() {
227        let dir = tempfile::tempdir().unwrap();
228        let pool = init_database(dir.path().join("devices.db")).await.unwrap();
229        let repository = DeviceRepository::new(DatabasePool::Sqlite(pool));
230        let (device, token) = PairedDevice::issue("Pixel", "android");
231
232        repository.insert(&device).await.unwrap();
233
234        let found = repository.find_by_token(&token).await.unwrap().unwrap();
235        assert_eq!(found.id, device.id);
236        assert!(found.last_seen_at.is_some());
237        assert_eq!(repository.find_by_token("wrong").await.unwrap(), None);
238        assert_eq!(
239            repository.find_by_token(&device.token_hash).await.unwrap(),
240            None
241        );
242
243        repository.touch(&device.id).await.unwrap();
244        let listed = repository.list().await.unwrap();
245        assert_eq!(listed.len(), 1);
246        assert!(listed[0].last_seen_at.is_some());
247
248        assert!(repository.delete(&device.id).await.unwrap());
249        assert!(!repository.delete(&device.id).await.unwrap());
250        assert_eq!(repository.find_by_token(&token).await.unwrap(), None);
251
252        repository.import(&listed).await.unwrap();
253        repository.import(&listed).await.unwrap();
254        assert_eq!(repository.list().await.unwrap(), listed);
255        assert_eq!(repository.delete_all().await.unwrap(), 1);
256        assert!(repository.list().await.unwrap().is_empty());
257    }
258}
pipedash_core/infrastructure/database/device_repository.rs

pipedash_core/infrastructure/database/
device_repository.rs
