db.rs (14681B)
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::IncomingKey, 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<IncomingKey>, 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::{PgConnection, PgPool, postgres::PgRow}; 335 use taler_api::{db::TypeHelper, notification::dummy_listen, subject::IncomingKey}; 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 use taler_macros::db_test; 346 347 use crate::{ 348 db::{AddIncomingResult, TxIn, incoming_history, register_tx_in, revenue_history}, 349 payto::WiseAccount, 350 }; 351 352 const BALANCE_ID: u32 = 42; 353 const CURR: Currency = Currency::TEST; 354 355 #[db_test] 356 async fn tx_in(mut db: PgConnection, pool: PgPool) { 357 let mut routine = async |first: &Option<IncomingKey>, second: &Option<IncomingKey>| { 358 let id = sqlx::query("SELECT count(*) + 1 FROM tx_in") 359 .try_map(|r: PgRow| r.try_get_u64(0)) 360 .fetch_one(&mut db) 361 .await 362 .unwrap(); 363 let now = now_sql_stable_ts(); 364 let later = now + Span::new().hours(1); 365 let tx = TxIn { 366 balance_id: 42, 367 name: CompactString::const_new("Name"), 368 wise_ref: Some(format_compact!("TRANSFER-{id}")), 369 amount: amount("EUR:10"), 370 subject: "subject".into(), 371 debtor: Some(WiseAccount::IBAN(BankID { 372 iban: iban("HU30162000031000163100000000"), 373 bic: None, 374 })), 375 value_at: now, 376 }; 377 // Insert 378 assert_eq!( 379 register_tx_in(&pool, &tx, first, &now) 380 .await 381 .expect("register tx in"), 382 AddIncomingResult::Success { 383 new: true, 384 pending: false, 385 row_id: id, 386 valued_at: now 387 } 388 ); 389 // Idempotent 390 assert_eq!( 391 register_tx_in( 392 &pool, 393 &TxIn { 394 value_at: later, 395 ..tx.clone() 396 }, 397 first, 398 &now 399 ) 400 .await 401 .expect("register tx in"), 402 AddIncomingResult::Success { 403 new: false, 404 pending: false, 405 row_id: id, 406 valued_at: now 407 } 408 ); 409 // Many 410 assert_eq!( 411 register_tx_in( 412 &pool, 413 &TxIn { 414 wise_ref: Some(format_compact!("TRANSFER-{}", id + 1)), 415 value_at: later, 416 ..tx 417 }, 418 second, 419 &now 420 ) 421 .await 422 .expect("register tx in"), 423 AddIncomingResult::Success { 424 new: true, 425 pending: false, 426 row_id: id + 1, 427 valued_at: later 428 } 429 ); 430 }; 431 432 // Empty db 433 assert_eq!( 434 revenue_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 435 .await 436 .unwrap(), 437 Vec::new() 438 ); 439 assert_eq!( 440 incoming_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 441 .await 442 .unwrap(), 443 Vec::new() 444 ); 445 446 // Regular transaction 447 routine(&None, &None).await; 448 449 // Reserve transaction 450 routine( 451 &Some(IncomingKey::reserve(EddsaPublicKey::rand())), 452 &Some(IncomingKey::reserve(EddsaPublicKey::rand())), 453 ) 454 .await; 455 456 // Kyc transaction 457 routine( 458 &Some(IncomingKey::kyc(EddsaPublicKey::rand())), 459 &Some(IncomingKey::kyc(EddsaPublicKey::rand())), 460 ) 461 .await; 462 463 // History 464 assert_eq!( 465 revenue_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 466 .await 467 .unwrap() 468 .len(), 469 6 470 ); 471 assert_eq!( 472 incoming_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 473 .await 474 .unwrap() 475 .len(), 476 4 477 ); 478 } 479 }