pipedash_core/infrastructure/database/
push_repository.rs

1use chrono::{
2    DateTime,
3    Utc,
4};
5
6use super::DatabasePool;
7use crate::domain::{
8    DomainError,
9    DomainResult,
10    PushSubscription,
11};
12
13type Row = (
14    String,
15    String,
16    String,
17    String,
18    DateTime<Utc>,
19    Option<DateTime<Utc>>,
20);
21
22const COLUMNS: &str =
23    "id, relay_subscription_id, device_name, platform, created_at, last_success_at";
24
25fn db_error(error: sqlx::Error) -> DomainError {
26    DomainError::DatabaseError(error.to_string())
27}
28
29fn subscription(row: Row) -> PushSubscription {
30    let (id, relay_subscription_id, device_name, platform, created_at, last_success_at) = row;
31
32    PushSubscription {
33        id,
34        relay_subscription_id,
35        device_name,
36        platform,
37        created_at,
38        last_success_at,
39    }
40}
41
42pub struct PushRepository {
43    pool: DatabasePool,
44}
45
46impl PushRepository {
47    pub fn new(pool: DatabasePool) -> Self {
48        Self { pool }
49    }
50
51    fn placeholder(&self, index: usize) -> String {
52        match &self.pool {
53            DatabasePool::Sqlite(_) => "?".to_string(),
54            DatabasePool::Postgres(_) => format!("${index}"),
55        }
56    }
57
58    pub async fn list(&self) -> DomainResult<Vec<PushSubscription>> {
59        let sql = format!("SELECT {COLUMNS} FROM push_subscriptions ORDER BY created_at, id");
60        let rows = match &self.pool {
61            DatabasePool::Sqlite(p) => {
62                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
63                    .fetch_all(p)
64                    .await
65            }
66            DatabasePool::Postgres(p) => {
67                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
68                    .fetch_all(p)
69                    .await
70            }
71        }
72        .map_err(db_error)?;
73
74        Ok(rows.into_iter().map(subscription).collect())
75    }
76
77    pub async fn get(&self, id: &str) -> DomainResult<Option<PushSubscription>> {
78        let sql = format!(
79            "SELECT {COLUMNS} FROM push_subscriptions WHERE id = {}",
80            self.placeholder(1)
81        );
82        let row = match &self.pool {
83            DatabasePool::Sqlite(p) => {
84                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
85                    .bind(id)
86                    .fetch_optional(p)
87                    .await
88            }
89            DatabasePool::Postgres(p) => {
90                sqlx::query_as::<_, Row>(sqlx::AssertSqlSafe(sql))
91                    .bind(id)
92                    .fetch_optional(p)
93                    .await
94            }
95        }
96        .map_err(db_error)?;
97
98        Ok(row.map(subscription))
99    }
100
101    pub async fn register(
102        &self, relay_subscription_id: &str, device_name: &str, platform: &str,
103    ) -> DomainResult<PushSubscription> {
104        let id = PushSubscription::local_id(relay_subscription_id);
105        let sql = format!(
106            "INSERT INTO push_subscriptions (id, relay_subscription_id, device_name, platform, created_at)
107             VALUES ({}, {}, {}, {}, {})
108             ON CONFLICT (id) DO NOTHING",
109            self.placeholder(1),
110            self.placeholder(2),
111            self.placeholder(3),
112            self.placeholder(4),
113            self.placeholder(5),
114        );
115        let now = Utc::now();
116        match &self.pool {
117            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
118                .bind(&id)
119                .bind(relay_subscription_id)
120                .bind(device_name)
121                .bind(platform)
122                .bind(now)
123                .execute(p)
124                .await
125                .map(|_| ()),
126            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
127                .bind(&id)
128                .bind(relay_subscription_id)
129                .bind(device_name)
130                .bind(platform)
131                .bind(now)
132                .execute(p)
133                .await
134                .map(|_| ()),
135        }
136        .map_err(db_error)?;
137
138        self.get(&id).await?.ok_or_else(|| {
139            DomainError::DatabaseError("Push subscription disappeared after insert".to_string())
140        })
141    }
142
143    pub async fn delete(&self, id: &str) -> DomainResult<bool> {
144        let sql = format!(
145            "DELETE FROM push_subscriptions WHERE id = {}",
146            self.placeholder(1)
147        );
148        let deleted = match &self.pool {
149            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
150                .bind(id)
151                .execute(p)
152                .await
153                .map(|result| result.rows_affected()),
154            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
155                .bind(id)
156                .execute(p)
157                .await
158                .map(|result| result.rows_affected()),
159        }
160        .map_err(db_error)?;
161
162        Ok(deleted > 0)
163    }
164
165    pub async fn mark_success(&self, id: &str) -> DomainResult<()> {
166        let sql = format!(
167            "UPDATE push_subscriptions SET last_success_at = {} WHERE id = {}",
168            self.placeholder(1),
169            self.placeholder(2)
170        );
171        let now = Utc::now();
172        match &self.pool {
173            DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
174                .bind(now)
175                .bind(id)
176                .execute(p)
177                .await
178                .map(|_| ()),
179            DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql))
180                .bind(now)
181                .bind(id)
182                .execute(p)
183                .await
184                .map(|_| ()),
185        }
186        .map_err(db_error)
187    }
188
189    pub async fn delete_all(&self) -> DomainResult<u64> {
190        let sql = "DELETE FROM push_subscriptions";
191        match &self.pool {
192            DatabasePool::Sqlite(p) => sqlx::query(sql)
193                .execute(p)
194                .await
195                .map(|result| result.rows_affected()),
196            DatabasePool::Postgres(p) => sqlx::query(sql)
197                .execute(p)
198                .await
199                .map(|result| result.rows_affected()),
200        }
201        .map_err(db_error)
202    }
203
204    pub async fn import(&self, subscriptions: &[PushSubscription]) -> DomainResult<()> {
205        let sql = format!(
206            "INSERT INTO push_subscriptions
207                 (id, relay_subscription_id, device_name, platform, created_at, last_success_at)
208             VALUES ({}, {}, {}, {}, {}, {})
209             ON CONFLICT (id) DO UPDATE SET
210                 relay_subscription_id = excluded.relay_subscription_id,
211                 device_name = excluded.device_name,
212                 platform = excluded.platform,
213                 created_at = excluded.created_at,
214                 last_success_at = excluded.last_success_at",
215            self.placeholder(1),
216            self.placeholder(2),
217            self.placeholder(3),
218            self.placeholder(4),
219            self.placeholder(5),
220            self.placeholder(6),
221        );
222
223        for subscription in subscriptions {
224            match &self.pool {
225                DatabasePool::Sqlite(p) => sqlx::query(sqlx::AssertSqlSafe(sql.clone()))
226                    .bind(&subscription.id)
227                    .bind(&subscription.relay_subscription_id)
228                    .bind(&subscription.device_name)
229                    .bind(&subscription.platform)
230                    .bind(subscription.created_at)
231                    .bind(subscription.last_success_at)
232                    .execute(p)
233                    .await
234                    .map(|_| ()),
235                DatabasePool::Postgres(p) => sqlx::query(sqlx::AssertSqlSafe(sql.clone()))
236                    .bind(&subscription.id)
237                    .bind(&subscription.relay_subscription_id)
238                    .bind(&subscription.device_name)
239                    .bind(&subscription.platform)
240                    .bind(subscription.created_at)
241                    .bind(subscription.last_success_at)
242                    .execute(p)
243                    .await
244                    .map(|_| ()),
245            }
246            .map_err(db_error)?;
247        }
248
249        Ok(())
250    }
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256    use crate::infrastructure::database::init_database;
257
258    async fn exercise(repository: &PushRepository) {
259        repository
260            .delete(&PushSubscription::local_id("relay-a"))
261            .await
262            .unwrap();
263        repository
264            .delete(&PushSubscription::local_id("relay-b"))
265            .await
266            .unwrap();
267
268        let first = repository
269            .register("relay-a", "Henrique's iPhone", "ios")
270            .await
271            .unwrap();
272        assert_eq!(first.id, PushSubscription::local_id("relay-a"));
273        assert_ne!(first.id, first.relay_subscription_id);
274        assert_eq!(first.relay_subscription_id, "relay-a");
275        assert_eq!(first.device_name, "Henrique's iPhone");
276        assert_eq!(first.platform, "ios");
277        assert_eq!(first.last_success_at, None);
278
279        let again = repository
280            .register("relay-a", "Renamed", "ios")
281            .await
282            .unwrap();
283        assert_eq!(again, first);
284
285        let second = repository.register("relay-b", "iPad", "ios").await.unwrap();
286        let listed: Vec<String> = repository
287            .list()
288            .await
289            .unwrap()
290            .into_iter()
291            .map(|s| s.id)
292            .filter(|id| *id == first.id || *id == second.id)
293            .collect();
294        assert_eq!(listed, vec![first.id.clone(), second.id.clone()]);
295
296        repository.mark_success(&first.id).await.unwrap();
297        let marked = repository.get(&first.id).await.unwrap().unwrap();
298        assert!(marked.last_success_at.is_some());
299        assert_eq!(repository.get("missing").await.unwrap(), None);
300
301        assert!(repository.delete(&first.id).await.unwrap());
302        assert!(!repository.delete(&first.id).await.unwrap());
303        assert_eq!(repository.get(&first.id).await.unwrap(), None);
304        assert!(repository.delete(&second.id).await.unwrap());
305    }
306
307    #[tokio::test]
308    async fn sqlite_subscriptions_register_idempotently_and_can_be_marked_and_deleted() {
309        let dir = tempfile::tempdir().unwrap();
310        let pool = init_database(dir.path().join("push.db")).await.unwrap();
311
312        exercise(&PushRepository::new(DatabasePool::Sqlite(pool))).await;
313    }
314
315    #[cfg(feature = "postgres")]
316    #[tokio::test]
317    #[ignore = "requires TEST_DATABASE_URL"]
318    async fn postgres_subscriptions_register_idempotently_and_can_be_marked_and_deleted() {
319        let url = std::env::var("TEST_DATABASE_URL")
320            .expect("TEST_DATABASE_URL must be set to run the PostgreSQL tests");
321        let pool = crate::infrastructure::database::init_postgres_database(&url)
322            .await
323            .unwrap();
324
325        exercise(&PushRepository::new(DatabasePool::Postgres(pool))).await;
326    }
327
328    #[tokio::test]
329    async fn imported_subscriptions_keep_their_ids_capabilities_and_timestamps() {
330        let dir = tempfile::tempdir().unwrap();
331        let source = PushRepository::new(DatabasePool::Sqlite(
332            init_database(dir.path().join("source.db")).await.unwrap(),
333        ));
334        let target = PushRepository::new(DatabasePool::Sqlite(
335            init_database(dir.path().join("target.db")).await.unwrap(),
336        ));
337        source.register("relay-a", "iPhone", "ios").await.unwrap();
338        let paired = source.register("relay-b", "iPad", "ios").await.unwrap();
339        source.mark_success(&paired.id).await.unwrap();
340        let exported = source.list().await.unwrap();
341
342        target.import(&exported).await.unwrap();
343        target.import(&exported).await.unwrap();
344
345        assert_eq!(target.list().await.unwrap(), exported);
346        assert_eq!(target.delete_all().await.unwrap(), 2);
347        assert!(target.list().await.unwrap().is_empty());
348    }
349}
pipedash_core/infrastructure/database/push_repository.rs

pipedash_core/infrastructure/database/
push_repository.rs
