db.rs (14913B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2025, 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 std::{collections::BTreeMap, fmt::Display, sync::Arc}; 18 19 use compact_str::CompactString; 20 use jiff::Timestamp; 21 use sqlx::{PgPool, QueryBuilder, Row, postgres::PgRow}; 22 use taler_api::{ 23 db::{BindHelper, TypeHelper, history}, 24 serialized, 25 subject::IncomingSubject, 26 }; 27 use taler_common::{ 28 api::{ 29 params::History, 30 prepared::{RegistrationRequest, Unregistration}, 31 revenue::RevenueIncomingBankTransaction, 32 wire::IncomingBankTransaction, 33 }, 34 config::Config, 35 db::IncomingType, 36 types::{ 37 amount::{Amount, Currency}, 38 payto::PaytoImpl as _, 39 }, 40 }; 41 use tokio::sync::watch::{Receiver, Sender}; 42 43 use crate::{ 44 config::parse_db_cfg, 45 payto::{WiseAccount, WisePayto}, 46 }; 47 48 const SCHEMA: &str = "wise"; 49 50 pub async fn pool(cfg: &Config) -> anyhow::Result<PgPool> { 51 let db = parse_db_cfg(cfg)?; 52 let pool = taler_common::db::pool(db.cfg, SCHEMA).await?; 53 Ok(pool) 54 } 55 56 pub async fn dbinit(cfg: &Config, reset: bool) -> anyhow::Result<PgPool> { 57 let db_cfg = parse_db_cfg(cfg)?; 58 let pool = taler_common::db::pool(db_cfg.cfg, SCHEMA).await?; 59 let mut db = pool.acquire().await?; 60 taler_common::db::dbinit(&mut db, db_cfg.sql_dir.as_ref(), SCHEMA, reset).await?; 61 Ok(pool) 62 } 63 64 pub async fn notification_listener( 65 pool: PgPool, 66 in_channels: Arc<BTreeMap<u32, Sender<i64>>>, 67 taler_in_channels: Arc<BTreeMap<u32, Sender<i64>>>, 68 ) { 69 taler_api::notification::notification_listener!(&pool, 70 "tx_in" => (balance_id: u32, row_id: i64) { 71 if let Some(channel) = in_channels.get(&balance_id) { 72 channel.send_replace(row_id); 73 } 74 }, 75 "taler_in" => (balance_id: u32, row_id: i64) { 76 if let Some(channel) = taler_in_channels.get(&balance_id) { 77 channel.send_replace(row_id); 78 } 79 } 80 ) 81 } 82 83 #[derive(Debug, Clone)] 84 pub struct TxIn { 85 pub balance_id: u32, 86 pub wise_ref: Option<CompactString>, 87 pub amount: Amount, 88 pub subject: CompactString, 89 pub name: CompactString, 90 pub debtor: Option<WiseAccount>, 91 pub value_at: Timestamp, 92 } 93 94 impl Display for TxIn { 95 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 96 let Self { 97 balance_id, 98 wise_ref, 99 amount, 100 subject, 101 name, 102 debtor, 103 value_at: value_date, 104 } = self; 105 106 write!( 107 f, 108 "{balance_id} {value_date} {} {amount} ({} {name}) '{subject}'", 109 wise_ref.as_deref().unwrap_or("admin"), 110 std::fmt::from_fn(|w| { 111 match debtor { 112 Some(account) => write!(w, "{account}"), 113 None => write!(w, "unsupported"), 114 } 115 }) 116 ) 117 } 118 } 119 120 #[derive(Debug, PartialEq, Eq)] 121 pub enum AddIncomingResult { 122 Success { 123 new: bool, 124 pending: bool, 125 row_id: u64, 126 valued_at: Timestamp, 127 }, 128 ReservePubReuse, 129 UnknownMapping, 130 MappingReuse, 131 } 132 133 pub async fn register_tx_in( 134 db: &PgPool, 135 tx: &TxIn, 136 subject: &Option<IncomingSubject>, 137 now: &Timestamp, 138 ) -> sqlx::Result<AddIncomingResult> { 139 let payto = tx.debtor.as_ref().map(|it| it.as_uri()); 140 serialized!( 141 sqlx::query( 142 " 143 SELECT out_reserve_pub_reuse, out_mapping_reuse, out_unknown_mapping, out_tx_row_id, out_valued_at, out_new, out_pending 144 FROM register_tx_in($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) 145 ", 146 ) 147 .bind(tx.balance_id as i64) 148 .bind(&tx.wise_ref) 149 .bind(tx.amount) 150 .bind(&tx.subject) 151 .bind(payto.as_ref().map(|it| it.as_ref().as_str())) 152 .bind(&tx.name) 153 .bind_timestamp(&tx.value_at) 154 .bind(subject.as_ref().map(|it| it.ty())) 155 .bind(subject.as_ref().map(|it| it.key())) 156 .bind_timestamp(now) 157 .try_map(|r: PgRow| { 158 Ok(if r.try_get_flag(0)? { 159 AddIncomingResult::ReservePubReuse 160 } else if r.try_get_flag(1)? { 161 AddIncomingResult::MappingReuse 162 } else if r.try_get_flag(2)? { 163 AddIncomingResult::UnknownMapping 164 } else { 165 AddIncomingResult::Success { 166 row_id: r.try_get_u64(3)?, 167 valued_at: r.try_get_timestamp(4)?, 168 new: r.try_get(5)?, 169 pending: r.try_get(6)? 170 } 171 }) 172 }) 173 .fetch_one(db) 174 ) 175 } 176 177 pub async fn incoming_history( 178 db: &PgPool, 179 balance_id: u32, 180 currency: &Currency, 181 params: &History, 182 listen: impl FnOnce() -> Receiver<i64>, 183 ) -> sqlx::Result<Vec<IncomingBankTransaction>> { 184 history( 185 db, 186 "tx_in_id", 187 params, 188 listen, 189 || { 190 let mut query = QueryBuilder::new( 191 " 192 SELECT 193 type, 194 tx_in_id, 195 amount, 196 debit_payto, 197 debit_name, 198 valued_at, 199 metadata, 200 authorization_pub, 201 authorization_sig 202 FROM taler_in 203 JOIN tx_in USING (tx_in_id) 204 WHERE balance_id = 205 ", 206 ); 207 query.push_bind(balance_id as i32).push(" AND "); 208 query 209 }, 210 |r: PgRow| { 211 Ok(match r.try_get(0)? { 212 IncomingType::reserve => IncomingBankTransaction::Reserve { 213 row_id: r.try_get_u64(1)?, 214 amount: r.try_get_amount(2, currency)?, 215 credit_fee: None, 216 debit_account: r 217 .try_get_parse::<_, _, WisePayto>(3)? 218 .as_full_uri(r.try_get(4)?), 219 date: r.try_get_timestamp(5)?.into(), 220 reserve_pub: r.try_get(6)?, 221 authorization_pub: r.try_get(7)?, 222 authorization_sig: r.try_get(8)?, 223 }, 224 IncomingType::kyc => IncomingBankTransaction::Kyc { 225 row_id: r.try_get_u64(1)?, 226 amount: r.try_get_amount(2, currency)?, 227 credit_fee: None, 228 debit_account: r 229 .try_get_parse::<_, _, WisePayto>(3)? 230 .as_full_uri(r.try_get(4)?), 231 date: r.try_get_timestamp(5)?.into(), 232 account_pub: r.try_get(6)?, 233 authorization_pub: r.try_get(7)?, 234 authorization_sig: r.try_get(8)?, 235 }, 236 IncomingType::map => unimplemented!("MAP are never listed in the history"), 237 }) 238 }, 239 ) 240 .await 241 } 242 243 pub async fn revenue_history( 244 db: &PgPool, 245 balance_id: u32, 246 currency: &Currency, 247 params: &History, 248 listen: impl FnOnce() -> Receiver<i64>, 249 ) -> sqlx::Result<Vec<RevenueIncomingBankTransaction>> { 250 history( 251 db, 252 "tx_in_id", 253 params, 254 listen, 255 || { 256 let mut query = QueryBuilder::new( 257 " 258 SELECT 259 tx_in_id, 260 valued_at, 261 amount, 262 debit_payto, 263 debit_name, 264 subject 265 FROM tx_in 266 WHERE balance_id = 267 ", 268 ); 269 query.push_bind(balance_id as i32).push(" AND "); 270 query 271 }, 272 |r: PgRow| { 273 Ok(RevenueIncomingBankTransaction { 274 row_id: r.try_get_u64(0)?, 275 date: r.try_get_timestamp(1)?.into(), 276 amount: r.try_get_amount(2, currency)?, 277 credit_fee: None, 278 debit_account: r 279 .try_get_parse::<_, _, WisePayto>(3)? 280 .as_full_uri(r.try_get(4)?), 281 subject: r.try_get(5)?, 282 }) 283 }, 284 ) 285 .await 286 } 287 288 pub enum RegistrationResult { 289 Success, 290 ReservePubReuse, 291 } 292 293 pub async fn transfer_register( 294 db: &PgPool, 295 req: &RegistrationRequest, 296 ) -> sqlx::Result<RegistrationResult> { 297 let ty: IncomingType = req.r#type.into(); 298 serialized!( 299 sqlx::query( 300 "SELECT out_reserve_pub_reuse FROM register_prepared_transfers($1,$2,$3,$4,$5,$6)" 301 ) 302 .bind(ty) 303 .bind(req.account_pub) 304 .bind(req.authorization_pub) 305 .bind(req.authorization_sig) 306 .bind(req.recurrent) 307 .bind_timestamp(&Timestamp::now()) 308 .try_map(|r: PgRow| { 309 Ok(if r.try_get_flag("out_reserve_pub_reuse")? { 310 RegistrationResult::ReservePubReuse 311 } else { 312 RegistrationResult::Success 313 }) 314 }) 315 .fetch_one(db) 316 ) 317 } 318 319 pub async fn transfer_unregister(db: &PgPool, req: &Unregistration) -> sqlx::Result<bool> { 320 serialized!( 321 sqlx::query("SELECT out_found FROM delete_prepared_transfers($1,$2)") 322 .bind(req.authorization_pub) 323 .bind_timestamp(&Timestamp::now()) 324 .try_map(|r: PgRow| r.try_get_flag("out_found")) 325 .fetch_one(db) 326 ) 327 } 328 329 #[cfg(test)] 330 mod test { 331 332 use compact_str::{CompactString, format_compact}; 333 use jiff::Span; 334 use sqlx::{PgPool, Postgres, pool::PoolConnection, postgres::PgRow}; 335 use taler_api::{db::TypeHelper, notification::dummy_listen, subject::IncomingSubject}; 336 use taler_common::{ 337 api::{EddsaPublicKey, params::History}, 338 types::{ 339 amount::{Currency, amount}, 340 iban::iban, 341 payto::BankID, 342 utils::now_sql_stable_ts, 343 }, 344 }; 345 346 use crate::{ 347 CONFIG_SOURCE, 348 db::{AddIncomingResult, TxIn, incoming_history, register_tx_in, revenue_history}, 349 payto::WiseAccount, 350 }; 351 352 async fn setup() -> (PoolConnection<Postgres>, PgPool) { 353 taler_test_utils::db::db_test_setup(CONFIG_SOURCE).await 354 } 355 356 const BALANCE_ID: u32 = 42; 357 const CURR: Currency = Currency::TEST; 358 359 #[tokio::test] 360 async fn tx_in() { 361 let (mut db, pool) = setup().await; 362 363 let mut routine = async |first: &Option<IncomingSubject>, 364 second: &Option<IncomingSubject>| { 365 let id = sqlx::query("SELECT count(*) + 1 FROM tx_in") 366 .try_map(|r: PgRow| r.try_get_u64(0)) 367 .fetch_one(&mut *db) 368 .await 369 .unwrap(); 370 let now = now_sql_stable_ts(); 371 let later = now + Span::new().hours(1); 372 let tx = TxIn { 373 balance_id: 42, 374 name: CompactString::const_new("Name"), 375 wise_ref: Some(format_compact!("TRANSFER-{id}")), 376 amount: amount("EUR:10"), 377 subject: "subject".into(), 378 debtor: Some(WiseAccount::IBAN(BankID { 379 iban: iban("HU30162000031000163100000000"), 380 bic: None, 381 })), 382 value_at: now, 383 }; 384 // Insert 385 assert_eq!( 386 register_tx_in(&pool, &tx, first, &now) 387 .await 388 .expect("register tx in"), 389 AddIncomingResult::Success { 390 new: true, 391 pending: false, 392 row_id: id, 393 valued_at: now 394 } 395 ); 396 // Idempotent 397 assert_eq!( 398 register_tx_in( 399 &pool, 400 &TxIn { 401 value_at: later, 402 ..tx.clone() 403 }, 404 first, 405 &now 406 ) 407 .await 408 .expect("register tx in"), 409 AddIncomingResult::Success { 410 new: false, 411 pending: false, 412 row_id: id, 413 valued_at: now 414 } 415 ); 416 // Many 417 assert_eq!( 418 register_tx_in( 419 &pool, 420 &TxIn { 421 wise_ref: Some(format_compact!("TRANSFER-{}", id + 1)), 422 value_at: later, 423 ..tx 424 }, 425 second, 426 &now 427 ) 428 .await 429 .expect("register tx in"), 430 AddIncomingResult::Success { 431 new: true, 432 pending: false, 433 row_id: id + 1, 434 valued_at: later 435 } 436 ); 437 }; 438 439 // Empty db 440 assert_eq!( 441 revenue_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 442 .await 443 .unwrap(), 444 Vec::new() 445 ); 446 assert_eq!( 447 incoming_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 448 .await 449 .unwrap(), 450 Vec::new() 451 ); 452 453 // Regular transaction 454 routine(&None, &None).await; 455 456 // Reserve transaction 457 routine( 458 &Some(IncomingSubject::Reserve(EddsaPublicKey::rand())), 459 &Some(IncomingSubject::Reserve(EddsaPublicKey::rand())), 460 ) 461 .await; 462 463 // Kyc transaction 464 routine( 465 &Some(IncomingSubject::Kyc(EddsaPublicKey::rand())), 466 &Some(IncomingSubject::Kyc(EddsaPublicKey::rand())), 467 ) 468 .await; 469 470 // History 471 assert_eq!( 472 revenue_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 473 .await 474 .unwrap() 475 .len(), 476 6 477 ); 478 assert_eq!( 479 incoming_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 480 .await 481 .unwrap() 482 .len(), 483 4 484 ); 485 } 486 }