db.rs (3301B)
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 compact_str::CompactString; 18 use jiff::Timestamp; 19 use sqlx::{PgPool, Row, postgres::PgRow, types::Json}; 20 use taler_api::{db::BindHelper as _, serialized}; 21 22 use crate::ebics::TaskStatus; 23 24 /** Register a pending transaction */ 25 pub async fn ebics_register(db: &PgPool, id: &str) -> sqlx::Result<()> { 26 serialized!( 27 sqlx::query( 28 "INSERT INTO pending_ebics_transactions (tx_id) VALUES ($1) ON CONFLICT DO NOTHING", 29 ) 30 .bind(id) 31 .execute(db) 32 )?; 33 Ok(()) 34 } 35 36 /** Register a pending transaction */ 37 pub async fn ebics_remove(db: &PgPool, id: &str) -> sqlx::Result<()> { 38 serialized!( 39 sqlx::query("DELETE FROM pending_ebics_transactions WHERE tx_id = $1") 40 .bind(id) 41 .execute(db) 42 )?; 43 Ok(()) 44 } 45 46 /** Register a pending transaction */ 47 pub async fn ebics_first(db: &PgPool) -> sqlx::Result<Option<CompactString>> { 48 serialized!( 49 sqlx::query("SELECT tx_id FROM pending_ebics_transactions LIMIT 1") 50 .try_map(|r: PgRow| r.try_get(0)) 51 .fetch_optional(db) 52 ) 53 } 54 55 /** Get current value for [key] */ 56 pub async fn get_task_status(db: &PgPool, key: &str) -> sqlx::Result<Option<TaskStatus>> { 57 serialized!( 58 sqlx::query_scalar::<_, Json<TaskStatus>>("SELECT value FROM kv WHERE key=$1") 59 .bind(key) 60 .fetch_optional(db) 61 ) 62 .map(|it| it.map(|it| it.0)) 63 } 64 65 /** Update a TaskStatus timestamp */ 66 pub async fn update_task_status( 67 db: &PgPool, 68 key: &str, 69 timestamp: &Timestamp, 70 success: bool, 71 ) -> sqlx::Result<()> { 72 serialized!( 73 sqlx::query(if success { 74 "INSERT INTO kv (key, value) VALUES ($1, jsonb_build_object('last_successfull', $2, 'last_trial', $3)) ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value" 75 } else { 76 "INSERT INTO kv (key, value) VALUES ($1, jsonb_build_object('last_trial', $2)) ON CONFLICT (key) DO UPDATE SET value=jsonb_set(kv.value, '{last_trial}'::text[], to_jsonb($3))" 77 }).bind(key).bind_timestamp(timestamp).bind_timestamp(timestamp).execute(db))?; 78 Ok(()) 79 } 80 81 pub mod test { 82 use sqlx::PgPool; 83 84 use crate::db::{ebics_first, ebics_register, ebics_remove}; 85 86 pub async fn ebics_routine(db: &PgPool) { 87 let ids = ["first", "second", "third"]; 88 89 for id in ids { 90 ebics_register(db, id).await.unwrap(); 91 } 92 for id in ids { 93 assert_eq!(Some(id), ebics_first(db).await.unwrap().as_deref()); 94 ebics_remove(db, id).await.unwrap(); 95 } 96 assert_eq!(ebics_first(db).await.unwrap(), None); 97 } 98 }