db.rs (4171B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2026 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU Affero General Public License as published by the Free Software 7 Foundation; either version 3, or (at your option) any later version. 8 9 TALER is distributed in the hope that it will be useful, but WITHOUT ANY 10 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 11 A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. 12 13 You should have received a copy of the GNU Affero General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 17 use jiff::Timestamp; 18 use sqlx::PgPool; 19 use taler_api::{db::BindHelper, serialized}; 20 use taler_common::config::Config; 21 22 use crate::config::parse_db_cfg; 23 24 const SCHEMA: &str = "apns_relay"; 25 26 pub async fn pool(cfg: &Config) -> anyhow::Result<PgPool> { 27 let db = parse_db_cfg(cfg)?; 28 let pool = taler_common::db::pool(db.cfg, SCHEMA).await?; 29 Ok(pool) 30 } 31 32 pub async fn dbinit(cfg: &Config, reset: bool) -> anyhow::Result<PgPool> { 33 let db_cfg = parse_db_cfg(cfg)?; 34 let pool = taler_common::db::pool(db_cfg.cfg, SCHEMA).await?; 35 let mut db = pool.acquire().await?; 36 taler_common::db::dbinit(&mut db, db_cfg.sql_dir.as_ref(), "apns-relay", reset).await?; 37 Ok(pool) 38 } 39 40 /// Register a new device 41 pub async fn register(db: &PgPool, token: &str, now: &Timestamp) -> sqlx::Result<()> { 42 serialized!( 43 sqlx::query( 44 " 45 INSERT INTO devices (token, registered_at) 46 VALUES ($1, $2) 47 ON CONFLICT (token) DO UPDATE 48 SET registered_at = GREATEST(devices.registered_at, EXCLUDED.registered_at) 49 ", 50 ) 51 .bind(token) 52 .bind_timestamp(now) 53 .execute(db) 54 )?; 55 Ok(()) 56 } 57 58 /// Unregister 59 pub async fn unregister(db: &PgPool, token: &str, before: &Timestamp) -> sqlx::Result<()> { 60 serialized!( 61 sqlx::query( 62 " 63 DELETE FROM devices 64 WHERE token = $1 AND registered_at <= $2 65 ", 66 ) 67 .bind(token) 68 .bind_timestamp(before) 69 .execute(db) 70 )?; 71 Ok(()) 72 } 73 74 /// List all registered devices 75 pub async fn all_registrations(db: &PgPool) -> sqlx::Result<Vec<String>> { 76 sqlx::query_scalar("SELECT token FROM devices") 77 .fetch_all(db) 78 .await 79 } 80 81 /// Remove all registered devices 82 pub async fn clear_registration(db: &PgPool) -> sqlx::Result<()> { 83 sqlx::query("TRUNCATE TABLE devices").execute(db).await?; 84 Ok(()) 85 } 86 87 #[cfg(test)] 88 pub mod test { 89 use jiff::{SignedDuration, Timestamp}; 90 use sqlx::PgPool; 91 use taler_macros::db_test; 92 93 use crate::db::{all_registrations, clear_registration, register, unregister}; 94 95 #[db_test] 96 async fn registration(db: PgPool) { 97 let token1 = "device_token1"; 98 let token2 = "device_token2"; 99 let now = Timestamp::now(); 100 101 // Empty 102 assert_eq!(all_registrations(&db).await.unwrap(), &[""; 0]); 103 clear_registration(&db).await.unwrap(); 104 105 // Register 106 register(&db, token1, &now).await.unwrap(); 107 assert_eq!(all_registrations(&db).await.unwrap(), &[token1]); 108 109 // Idempotent 110 register(&db, token1, &now).await.unwrap(); 111 assert_eq!(all_registrations(&db).await.unwrap(), &[token1]); 112 113 // Many 114 register(&db, token2, &(now + SignedDuration::from_mins(1))) 115 .await 116 .unwrap(); 117 assert_eq!(all_registrations(&db).await.unwrap(), &[token1, token2]); 118 119 // Unregister 120 unregister(&db, token1, &now).await.unwrap(); 121 assert_eq!(all_registrations(&db).await.unwrap(), &[token2]); 122 123 // Idempotent 124 unregister(&db, token1, &now).await.unwrap(); 125 assert_eq!(all_registrations(&db).await.unwrap(), &[token2]); 126 127 // Skip reregistered 128 unregister(&db, token2, &now).await.unwrap(); 129 assert_eq!(all_registrations(&db).await.unwrap(), &[token2]); 130 131 clear_registration(&db).await.unwrap(); 132 assert_eq!(all_registrations(&db).await.unwrap(), &[""; 0]); 133 } 134 }