db.rs (18542B)
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 /// Durable transaction high-water marks and completed scan schedules. 65 #[derive(Debug, Clone, PartialEq, Eq)] 66 pub(crate) struct SyncState { 67 pub execution: Option<Timestamp>, 68 pub early: Timestamp, 69 pub sync: Timestamp, 70 pub checkpoint: Timestamp, 71 pub early_success: Option<Timestamp>, 72 pub sync_success: Option<Timestamp>, 73 pub checkpoint_success: Option<Timestamp>, 74 } 75 76 pub(crate) async fn load_sync_state( 77 db: &PgPool, 78 balance_id: u32, 79 ) -> sqlx::Result<Option<SyncState>> { 80 sqlx::query( 81 "SELECT early_cursor, sync_cursor, checkpoint_cursor, 82 early_success_at, sync_success_at, checkpoint_success_at, sync_execution_at 83 FROM balances WHERE balance_id = $1", 84 ) 85 .bind(i64::from(balance_id)) 86 .try_map(|r: PgRow| { 87 Ok(SyncState { 88 execution: r.try_get_opt_timestamp(6)?, 89 early: r.try_get_timestamp(0)?, 90 sync: r.try_get_timestamp(1)?, 91 checkpoint: r.try_get_timestamp(2)?, 92 early_success: r.try_get_opt_timestamp(3)?, 93 sync_success: r.try_get_opt_timestamp(4)?, 94 checkpoint_success: r.try_get_opt_timestamp(5)?, 95 }) 96 }) 97 .fetch_optional(db) 98 .await 99 } 100 101 pub(crate) async fn initialize_sync_state( 102 db: &PgPool, 103 balance_id: u32, 104 created_at: Timestamp, 105 ) -> sqlx::Result<()> { 106 serialized!( 107 sqlx::query( 108 "INSERT INTO balances (balance_id, created_at, early_cursor, sync_cursor, checkpoint_cursor) 109 VALUES ($1, $2, $2, $2, $2) 110 ON CONFLICT (balance_id) DO NOTHING", 111 ) 112 .bind(i64::from(balance_id)) 113 .bind_timestamp(&created_at) 114 .execute(db) 115 )?; 116 Ok(()) 117 } 118 119 pub(crate) async fn reset_sync_state(db: &PgPool) -> sqlx::Result<()> { 120 serialized!( 121 sqlx::query( 122 "UPDATE balances SET 123 sync_execution_at = NULL, 124 early_cursor = created_at, 125 sync_cursor = created_at, 126 checkpoint_cursor = created_at, 127 early_success_at = NULL, 128 sync_success_at = NULL, 129 checkpoint_success_at = NULL", 130 ) 131 .execute(db) 132 )?; 133 Ok(()) 134 } 135 136 /// Called after all registrations in the chunk have committed. A failed write 137 /// leaves the old cursor available for an idempotent replay on restart. 138 pub(crate) async fn save_sync_state( 139 db: &PgPool, 140 balance_id: u32, 141 state: &SyncState, 142 ) -> sqlx::Result<()> { 143 serialized!( 144 sqlx::query( 145 "UPDATE balances SET 146 early_cursor = GREATEST(early_cursor, $2), 147 sync_cursor = GREATEST(sync_cursor, $3), 148 checkpoint_cursor = GREATEST(checkpoint_cursor, $4), 149 early_success_at = GREATEST(early_success_at, $5), 150 sync_success_at = GREATEST(sync_success_at, $6), 151 checkpoint_success_at = GREATEST(checkpoint_success_at, $7) 152 WHERE balance_id = $1", 153 ) 154 .bind(i64::from(balance_id)) 155 .bind_timestamp(&state.early) 156 .bind_timestamp(&state.sync) 157 .bind_timestamp(&state.checkpoint) 158 .bind(state.early_success.map(|t| t.as_microsecond())) 159 .bind(state.sync_success.map(|t| t.as_microsecond())) 160 .bind(state.checkpoint_success.map(|t| t.as_microsecond())) 161 .execute(db) 162 )?; 163 Ok(()) 164 } 165 166 pub async fn notification_listener( 167 pool: PgPool, 168 in_channels: Arc<BTreeMap<u32, Sender<i64>>>, 169 taler_in_channels: Arc<BTreeMap<u32, Sender<i64>>>, 170 ) { 171 taler_api::notification::notification_listener!(&pool, 172 "tx_in" => (balance_id: u32, row_id: i64) { 173 if let Some(channel) = in_channels.get(&balance_id) { 174 channel.send_replace(row_id); 175 } 176 }, 177 "taler_in" => (balance_id: u32, row_id: i64) { 178 if let Some(channel) = taler_in_channels.get(&balance_id) { 179 channel.send_replace(row_id); 180 } 181 } 182 ) 183 } 184 185 #[derive(Debug, Clone)] 186 pub struct TxIn { 187 pub balance_id: u32, 188 pub wise_ref: Option<CompactString>, 189 pub amount: Amount, 190 pub subject: CompactString, 191 pub name: CompactString, 192 pub debtor: Option<WiseAccount>, 193 pub value_at: Timestamp, 194 } 195 196 impl Display for TxIn { 197 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 198 let Self { 199 balance_id, 200 wise_ref, 201 amount, 202 subject, 203 name, 204 debtor, 205 value_at: value_date, 206 } = self; 207 208 write!( 209 f, 210 "{balance_id} {value_date} {} {amount} ({} {name}) '{subject}'", 211 wise_ref.as_deref().unwrap_or("admin"), 212 std::fmt::from_fn(|w| { 213 match debtor { 214 Some(account) => write!(w, "{account}"), 215 None => write!(w, "unsupported"), 216 } 217 }) 218 ) 219 } 220 } 221 222 #[derive(Debug, PartialEq, Eq)] 223 pub enum AddIncomingResult { 224 Success { 225 new: bool, 226 pending: bool, 227 row_id: u64, 228 valued_at: Timestamp, 229 }, 230 ReservePubReuse, 231 UnknownMapping, 232 MappingReuse, 233 } 234 235 pub async fn register_tx_in( 236 db: &PgPool, 237 tx: &TxIn, 238 subject: &Option<IncomingKey>, 239 now: &Timestamp, 240 ) -> sqlx::Result<AddIncomingResult> { 241 let payto = tx.debtor.as_ref().map(|it| it.as_uri()); 242 serialized!( 243 sqlx::query( 244 " 245 SELECT out_reserve_pub_reuse, out_mapping_reuse, out_unknown_mapping, out_tx_row_id, out_valued_at, out_new, out_pending 246 FROM register_tx_in($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) 247 ", 248 ) 249 .bind(tx.balance_id as i64) 250 .bind(&tx.wise_ref) 251 .bind(tx.amount) 252 .bind(&tx.subject) 253 .bind(payto.as_ref().map(|it| it.as_ref().as_str())) 254 .bind(&tx.name) 255 .bind_timestamp(&tx.value_at) 256 .bind(subject.as_ref().map(|it| it.ty)) 257 .bind(subject.as_ref().map(|it| it.key)) 258 .bind_timestamp(now) 259 .try_map(|r: PgRow| { 260 Ok(if r.try_get_flag(0)? { 261 AddIncomingResult::ReservePubReuse 262 } else if r.try_get_flag(1)? { 263 AddIncomingResult::MappingReuse 264 } else if r.try_get_flag(2)? { 265 AddIncomingResult::UnknownMapping 266 } else { 267 AddIncomingResult::Success { 268 row_id: r.try_get_u64(3)?, 269 valued_at: r.try_get_timestamp(4)?, 270 new: r.try_get(5)?, 271 pending: r.try_get(6)? 272 } 273 }) 274 }) 275 .fetch_one(db) 276 ) 277 } 278 279 pub async fn incoming_history( 280 db: &PgPool, 281 balance_id: u32, 282 currency: &Currency, 283 params: &History, 284 listen: impl FnOnce() -> Receiver<i64>, 285 ) -> sqlx::Result<Vec<IncomingBankTransaction>> { 286 history( 287 db, 288 "taler_in_id", 289 params, 290 listen, 291 || { 292 let mut query = QueryBuilder::new( 293 " 294 SELECT 295 type, 296 taler_in_id, 297 amount, 298 debit_payto, 299 debit_name, 300 valued_at, 301 metadata, 302 authorization_pub, 303 authorization_sig 304 FROM taler_in 305 JOIN tx_in USING (tx_in_id, balance_id) 306 WHERE balance_id = 307 ", 308 ); 309 query.push_bind(balance_id as i32).push(" AND "); 310 query 311 }, 312 |r: PgRow| { 313 Ok(match r.try_get(0)? { 314 IncomingType::reserve => IncomingBankTransaction::Reserve { 315 row_id: r.try_get_u64(1)?, 316 amount: r.try_get_amount(2, currency)?, 317 credit_fee: None, 318 debit_account: r 319 .try_get_parse::<_, _, WisePayto>(3)? 320 .as_full_uri(r.try_get(4)?), 321 date: r.try_get_timestamp(5)?.into(), 322 reserve_pub: r.try_get(6)?, 323 authorization_pub: r.try_get(7)?, 324 authorization_sig: r.try_get(8)?, 325 }, 326 IncomingType::kyc => IncomingBankTransaction::Kyc { 327 row_id: r.try_get_u64(1)?, 328 amount: r.try_get_amount(2, currency)?, 329 credit_fee: None, 330 debit_account: r 331 .try_get_parse::<_, _, WisePayto>(3)? 332 .as_full_uri(r.try_get(4)?), 333 date: r.try_get_timestamp(5)?.into(), 334 account_pub: r.try_get(6)?, 335 authorization_pub: r.try_get(7)?, 336 authorization_sig: r.try_get(8)?, 337 }, 338 IncomingType::map => unimplemented!("MAP are never listed in the history"), 339 }) 340 }, 341 ) 342 .await 343 } 344 345 pub async fn revenue_history( 346 db: &PgPool, 347 balance_id: u32, 348 currency: &Currency, 349 params: &History, 350 listen: impl FnOnce() -> Receiver<i64>, 351 ) -> sqlx::Result<Vec<RevenueIncomingBankTransaction>> { 352 history( 353 db, 354 "tx_in_id", 355 params, 356 listen, 357 || { 358 let mut query = QueryBuilder::new( 359 " 360 SELECT 361 tx_in_id, 362 valued_at, 363 amount, 364 debit_payto, 365 debit_name, 366 subject 367 FROM tx_in 368 WHERE debit_payto IS NOT NULL AND balance_id = 369 ", 370 ); 371 query.push_bind(balance_id as i32).push(" AND "); 372 query 373 }, 374 |r: PgRow| { 375 Ok(RevenueIncomingBankTransaction { 376 row_id: r.try_get_u64(0)?, 377 date: r.try_get_timestamp(1)?.into(), 378 amount: r.try_get_amount(2, currency)?, 379 credit_fee: None, 380 debit_account: r 381 .try_get_parse::<_, _, WisePayto>(3)? 382 .as_full_uri(r.try_get(4)?), 383 subject: r.try_get(5)?, 384 }) 385 }, 386 ) 387 .await 388 } 389 390 pub enum RegistrationResult { 391 Success, 392 ReservePubReuse, 393 } 394 395 pub async fn transfer_register( 396 db: &PgPool, 397 balance_id: u32, 398 req: &RegistrationRequest, 399 ) -> sqlx::Result<RegistrationResult> { 400 let ty: IncomingType = req.r#type.into(); 401 serialized!( 402 sqlx::query( 403 "SELECT out_reserve_pub_reuse FROM register_prepared_transfers($1,$2,$3,$4,$5,$6,$7)" 404 ) 405 .bind(i64::from(balance_id)) 406 .bind(ty) 407 .bind(req.account_pub) 408 .bind(req.authorization_pub) 409 .bind(req.authorization_sig) 410 .bind(req.recurrent) 411 .bind_timestamp(&Timestamp::now()) 412 .try_map(|r: PgRow| { 413 Ok(if r.try_get_flag("out_reserve_pub_reuse")? { 414 RegistrationResult::ReservePubReuse 415 } else { 416 RegistrationResult::Success 417 }) 418 }) 419 .fetch_one(db) 420 ) 421 } 422 423 pub async fn transfer_unregister( 424 db: &PgPool, 425 balance_id: u32, 426 req: &Unregistration, 427 ) -> sqlx::Result<bool> { 428 serialized!( 429 sqlx::query("SELECT out_found FROM delete_prepared_transfers($1,$2,$3)") 430 .bind(i64::from(balance_id)) 431 .bind(req.authorization_pub) 432 .bind_timestamp(&Timestamp::now()) 433 .try_map(|r: PgRow| r.try_get_flag("out_found")) 434 .fetch_one(db) 435 ) 436 } 437 438 pub async fn record_sync_execution(db: &PgPool, balance_id: u32) -> sqlx::Result<()> { 439 serialized!( 440 sqlx::query("UPDATE balances SET sync_execution_at = $2 WHERE balance_id = $1") 441 .bind(i64::from(balance_id)) 442 .bind_timestamp(&Timestamp::now()) 443 .execute(db) 444 )?; 445 Ok(()) 446 } 447 448 #[cfg(test)] 449 mod test { 450 451 use compact_str::{CompactString, format_compact}; 452 use sqlx::{PgConnection, PgPool, postgres::PgRow}; 453 use taler_api::{db::TypeHelper, notification::dummy_listen, subject::IncomingKey}; 454 use taler_common::{ 455 api::{EddsaPublicKey, params::History}, 456 types::{ 457 amount::{Currency, amount}, 458 iban::iban, 459 payto::BankID, 460 utils::now_sql_stable_ts, 461 }, 462 }; 463 use taler_macros::db_test; 464 465 use crate::{ 466 constants::HOUR, 467 db::{AddIncomingResult, TxIn, incoming_history, register_tx_in, revenue_history}, 468 payto::WiseAccount, 469 }; 470 471 const BALANCE_ID: u32 = 42; 472 const CURR: Currency = Currency::TEST; 473 474 #[db_test] 475 async fn tx_in(mut db: PgConnection, pool: PgPool) { 476 let mut routine = async |first: &Option<IncomingKey>, second: &Option<IncomingKey>| { 477 let id = sqlx::query("SELECT count(*) + 1 FROM tx_in") 478 .try_map(|r: PgRow| r.try_get_u64(0)) 479 .fetch_one(&mut db) 480 .await 481 .unwrap(); 482 let now = now_sql_stable_ts(); 483 let later = now + HOUR; 484 let tx = TxIn { 485 balance_id: 42, 486 name: CompactString::const_new("Name"), 487 wise_ref: Some(format_compact!("TRANSFER-{id}")), 488 amount: amount("EUR:10"), 489 subject: "subject".into(), 490 debtor: Some(WiseAccount::IBAN(BankID { 491 iban: iban("HU30162000031000163100000000"), 492 bic: None, 493 })), 494 value_at: now, 495 }; 496 // Insert 497 assert_eq!( 498 register_tx_in(&pool, &tx, first, &now) 499 .await 500 .expect("register tx in"), 501 AddIncomingResult::Success { 502 new: true, 503 pending: false, 504 row_id: id, 505 valued_at: now 506 } 507 ); 508 // Idempotent 509 assert_eq!( 510 register_tx_in( 511 &pool, 512 &TxIn { 513 value_at: later, 514 ..tx.clone() 515 }, 516 first, 517 &now 518 ) 519 .await 520 .expect("register tx in"), 521 AddIncomingResult::Success { 522 new: false, 523 pending: false, 524 row_id: id, 525 valued_at: now 526 } 527 ); 528 // Many 529 assert_eq!( 530 register_tx_in( 531 &pool, 532 &TxIn { 533 wise_ref: Some(format_compact!("TRANSFER-{}", id + 1)), 534 value_at: later, 535 ..tx 536 }, 537 second, 538 &now 539 ) 540 .await 541 .expect("register tx in"), 542 AddIncomingResult::Success { 543 new: true, 544 pending: false, 545 row_id: id + 1, 546 valued_at: later 547 } 548 ); 549 }; 550 551 // Empty db 552 assert_eq!( 553 revenue_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 554 .await 555 .unwrap(), 556 Vec::new() 557 ); 558 assert_eq!( 559 incoming_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 560 .await 561 .unwrap(), 562 Vec::new() 563 ); 564 565 // Regular transaction 566 routine(&None, &None).await; 567 568 // Reserve transaction 569 routine( 570 &Some(IncomingKey::reserve(EddsaPublicKey::rand())), 571 &Some(IncomingKey::reserve(EddsaPublicKey::rand())), 572 ) 573 .await; 574 575 // Kyc transaction 576 routine( 577 &Some(IncomingKey::kyc(EddsaPublicKey::rand())), 578 &Some(IncomingKey::kyc(EddsaPublicKey::rand())), 579 ) 580 .await; 581 582 // History 583 assert_eq!( 584 revenue_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 585 .await 586 .unwrap() 587 .len(), 588 6 589 ); 590 assert_eq!( 591 incoming_history(&pool, BALANCE_ID, &CURR, &History::default(), dummy_listen) 592 .await 593 .unwrap() 594 .len(), 595 4 596 ); 597 } 598 }