db.rs (50173B)
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::fmt::Display; 18 19 use compact_str::CompactString; 20 use jiff::{Timestamp, civil::Date, tz::TimeZone}; 21 use serde::{Serialize, de::DeserializeOwned}; 22 use sqlx::{PgConnection, PgPool, QueryBuilder, Row, postgres::PgRow}; 23 use taler_api::{ 24 db::{BindHelper, TypeHelper, history, page}, 25 serialized, 26 subject::{IncomingSubject, OutgoingSubject, fmt_out_subject}, 27 }; 28 use taler_common::{ 29 api::{ 30 HashCode, ShortHashCode, 31 params::{History, Page}, 32 prepared::{RegistrationRequest, Unregistration}, 33 revenue::RevenueIncomingBankTransaction, 34 wire::{ 35 IncomingBankTransaction, OutgoingBankTransaction, TransferListStatus, TransferState, 36 TransferStatus, 37 }, 38 }, 39 config::Config, 40 db::IncomingType, 41 types::{ 42 amount::{Amount, Decimal}, 43 payto::PaytoImpl as _, 44 }, 45 }; 46 use tokio::sync::watch::{Receiver, Sender}; 47 use url::Url; 48 49 use crate::{FullHuPayto, config::parse_db_cfg, constants::CURR, magnet_api::types::TxStatus}; 50 51 const SCHEMA: &str = "magnet_bank"; 52 53 pub async fn pool(cfg: &Config) -> anyhow::Result<PgPool> { 54 let db = parse_db_cfg(cfg)?; 55 let pool = taler_common::db::pool(db.cfg, SCHEMA).await?; 56 Ok(pool) 57 } 58 59 pub async fn dbinit(cfg: &Config, reset: bool) -> anyhow::Result<PgPool> { 60 let db_cfg = parse_db_cfg(cfg)?; 61 let pool = taler_common::db::pool(db_cfg.cfg, SCHEMA).await?; 62 let mut db = pool.acquire().await?; 63 taler_common::db::dbinit(&mut db, db_cfg.sql_dir.as_ref(), "magnet-bank", reset).await?; 64 Ok(pool) 65 } 66 67 pub async fn notification_listener( 68 pool: PgPool, 69 in_channel: Sender<i64>, 70 taler_in_channel: Sender<i64>, 71 out_channel: Sender<i64>, 72 taler_out_channel: Sender<i64>, 73 ) { 74 taler_api::notification::notification_listener!(&pool, 75 "tx_in" => (row_id: i64) { 76 in_channel.send_replace(row_id); 77 }, 78 "taler_in" => (row_id: i64) { 79 taler_in_channel.send_replace(row_id); 80 }, 81 "tx_out" => (row_id: i64) { 82 out_channel.send_replace(row_id); 83 }, 84 "taler_out" => (row_id: i64) { 85 taler_out_channel.send_replace(row_id); 86 } 87 ) 88 } 89 90 #[derive(Debug, Clone)] 91 pub struct TxIn { 92 pub code: u64, 93 pub amount: Amount, 94 pub subject: Box<str>, 95 pub debtor: FullHuPayto, 96 pub value_date: Date, 97 pub status: TxStatus, 98 } 99 100 impl Display for TxIn { 101 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 102 let Self { 103 code, 104 amount, 105 subject, 106 debtor, 107 value_date, 108 status, 109 } = self; 110 write!( 111 f, 112 "{value_date} {code} {amount} ({} {}) {status:?} '{subject}'", 113 debtor.bban(), 114 debtor.name 115 ) 116 } 117 } 118 119 #[derive(Debug, Clone)] 120 pub struct TxOut { 121 pub code: u64, 122 pub amount: Amount, 123 pub subject: Box<str>, 124 pub creditor: FullHuPayto, 125 pub value_date: Date, 126 pub status: TxStatus, 127 } 128 129 impl Display for TxOut { 130 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 131 let Self { 132 code, 133 amount, 134 subject, 135 creditor, 136 value_date, 137 status, 138 } = self; 139 write!( 140 f, 141 "{value_date} {code} {amount} ({} {}) {status:?} '{subject}'", 142 creditor.bban(), 143 &creditor.name 144 ) 145 } 146 } 147 148 #[derive(Debug, PartialEq, Eq)] 149 pub struct Initiated { 150 pub id: u64, 151 pub amount: Amount, 152 pub subject: Box<str>, 153 pub creditor: FullHuPayto, 154 } 155 156 impl Display for Initiated { 157 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 158 let Self { 159 id, 160 amount, 161 subject, 162 creditor, 163 } = self; 164 write!( 165 f, 166 "{id} {amount} ({} {}) '{subject}'", 167 creditor.bban(), 168 &creditor.name 169 ) 170 } 171 } 172 173 #[derive(Debug, Clone)] 174 pub struct TxInAdmin { 175 pub amount: Amount, 176 pub subject: String, 177 pub debtor: FullHuPayto, 178 pub metadata: IncomingSubject, 179 } 180 181 /// Lock the database for worker execution 182 pub async fn worker_lock(e: &mut PgConnection) -> sqlx::Result<bool> { 183 sqlx::query("SELECT pg_try_advisory_lock(42)") 184 .try_map(|r: PgRow| r.try_get(0)) 185 .fetch_one(e) 186 .await 187 } 188 189 #[derive(Debug, PartialEq, Eq)] 190 pub enum AddIncomingResult { 191 Success { 192 new: bool, 193 pending: bool, 194 row_id: u64, 195 valued_at: Date, 196 }, 197 ReservePubReuse, 198 UnknownMapping, 199 MappingReuse, 200 } 201 202 pub async fn register_tx_in_admin( 203 db: &PgPool, 204 tx: &TxInAdmin, 205 now: &Timestamp, 206 ) -> sqlx::Result<AddIncomingResult> { 207 serialized!( 208 sqlx::query( 209 " 210 SELECT out_reserve_pub_reuse, out_mapping_reuse, out_unknown_mapping, out_tx_row_id, out_valued_at, out_new, out_pending 211 FROM register_tx_in(NULL, $1, $2, $3, $4, $5, $6, $7, $5) 212 ", 213 ) 214 .bind(tx.amount) 215 .bind(&tx.subject) 216 .bind(tx.debtor.iban()) 217 .bind(&tx.debtor.name) 218 .bind_date(&now.to_zoned(TimeZone::UTC).date()) 219 .bind(tx.metadata.ty()) 220 .bind(tx.metadata.key()) 221 .try_map(|r: PgRow| { 222 Ok(if r.try_get_flag(0)? { 223 AddIncomingResult::ReservePubReuse 224 } else if r.try_get_flag(1)? { 225 AddIncomingResult::MappingReuse 226 } else if r.try_get_flag(2)? { 227 AddIncomingResult::UnknownMapping 228 } else { 229 AddIncomingResult::Success { 230 row_id: r.try_get_u64(3)?, 231 valued_at: r.try_get_date(4)?, 232 new: r.try_get(5)?, 233 pending: r.try_get(6)? 234 } 235 }) 236 }) 237 .fetch_one(db) 238 ) 239 } 240 241 pub async fn register_tx_in( 242 db: &mut PgConnection, 243 tx: &TxIn, 244 subject: &Option<IncomingSubject>, 245 now: &Timestamp, 246 ) -> sqlx::Result<AddIncomingResult> { 247 serialized!( 248 sqlx::query( 249 " 250 SELECT out_reserve_pub_reuse, out_mapping_reuse, out_unknown_mapping, out_tx_row_id, out_valued_at, out_new, out_pending 251 FROM register_tx_in($1, $2, $3, $4, $5, $6, $7, $8, $9) 252 ", 253 ) 254 .bind(tx.code as i64) 255 .bind(tx.amount) 256 .bind(&tx.subject) 257 .bind(tx.debtor.iban()) 258 .bind(&tx.debtor.name) 259 .bind_date(&tx.value_date) 260 .bind(subject.as_ref().map(|it| it.ty())) 261 .bind(subject.as_ref().map(|it| it.key())) 262 .bind_timestamp(now) 263 .try_map(|r: PgRow| { 264 Ok(if r.try_get_flag(0)? { 265 AddIncomingResult::ReservePubReuse 266 } else if r.try_get_flag(1)? { 267 AddIncomingResult::MappingReuse 268 } else if r.try_get_flag(2)? { 269 AddIncomingResult::UnknownMapping 270 } else { 271 AddIncomingResult::Success { 272 row_id: r.try_get_u64(3)?, 273 valued_at: r.try_get_date(4)?, 274 new: r.try_get(5)?, 275 pending: r.try_get(6)? 276 } 277 }) 278 }) 279 .fetch_one(&mut *db) 280 ) 281 } 282 283 #[derive(Debug)] 284 pub enum TxOutKind { 285 Simple, 286 Bounce(u32), 287 Talerable(OutgoingSubject), 288 } 289 290 #[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type)] 291 #[allow(non_camel_case_types)] 292 #[sqlx(type_name = "register_result")] 293 pub enum RegisterResult { 294 /// Already registered 295 idempotent, 296 /// Initiated transaction 297 known, 298 /// Recovered unknown outgoing transaction 299 recovered, 300 } 301 302 #[derive(Debug, PartialEq, Eq)] 303 pub struct AddOutgoingResult { 304 pub result: RegisterResult, 305 pub row_id: u64, 306 } 307 308 pub async fn register_tx_out( 309 db: &mut PgConnection, 310 tx: &TxOut, 311 kind: &TxOutKind, 312 now: &Timestamp, 313 ) -> sqlx::Result<AddOutgoingResult> { 314 serialized!({ 315 let query = sqlx::query( 316 " 317 SELECT out_result, out_tx_row_id 318 FROM register_tx_out($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) 319 ", 320 ) 321 .bind(tx.code as i64) 322 .bind(tx.amount) 323 .bind(&tx.subject) 324 .bind(tx.creditor.iban()) 325 .bind(&tx.creditor.name) 326 .bind_date(&tx.value_date); 327 match kind { 328 TxOutKind::Simple => query 329 .bind(None::<&[u8]>) 330 .bind(None::<&str>) 331 .bind(None::<&str>) 332 .bind(None::<i64>), 333 TxOutKind::Bounce(bounced) => query 334 .bind(None::<&[u8]>) 335 .bind(None::<&str>) 336 .bind(None::<&str>) 337 .bind(*bounced as i64), 338 TxOutKind::Talerable(subject) => query 339 .bind(subject.wtid) 340 .bind(subject.exchange_base_url.as_str()) 341 .bind(&subject.metadata) 342 .bind(None::<i64>), 343 } 344 .bind_timestamp(now) 345 .try_map(|r: PgRow| { 346 Ok(AddOutgoingResult { 347 result: r.try_get(0)?, 348 row_id: r.try_get_u64(1)?, 349 }) 350 }) 351 .fetch_one(&mut *db) 352 }) 353 } 354 355 #[derive(Debug, PartialEq, Eq)] 356 pub struct OutFailureResult { 357 pub initiated_id: Option<u64>, 358 pub new: bool, 359 } 360 361 pub async fn register_tx_out_failure( 362 db: &mut PgConnection, 363 code: u64, 364 bounced: Option<u32>, 365 now: &Timestamp, 366 ) -> sqlx::Result<OutFailureResult> { 367 serialized!( 368 sqlx::query( 369 " 370 SELECT out_new, out_initiated_id 371 FROM register_tx_out_failure($1, $2, $3) 372 ", 373 ) 374 .bind(code as i64) 375 .bind(bounced.map(|i| i as i32)) 376 .bind_timestamp(now) 377 .try_map(|r: PgRow| { 378 Ok(OutFailureResult { 379 new: r.try_get(0)?, 380 initiated_id: r.try_get::<Option<i64>, _>(1)?.map(|i| i as u64), 381 }) 382 }) 383 .fetch_one(&mut *db) 384 ) 385 } 386 387 #[derive(Debug, PartialEq, Eq)] 388 pub enum TransferResult { 389 Success { id: u64, initiated_at: Timestamp }, 390 RequestUidReuse, 391 WtidReuse, 392 } 393 394 #[derive(Debug, Clone)] 395 pub struct Transfer { 396 pub request_uid: HashCode, 397 pub amount: Decimal, 398 pub exchange_base_url: Url, 399 pub metadata: Option<CompactString>, 400 pub wtid: ShortHashCode, 401 pub creditor: FullHuPayto, 402 } 403 404 pub async fn make_transfer( 405 db: &PgPool, 406 tx: &Transfer, 407 now: &Timestamp, 408 ) -> sqlx::Result<TransferResult> { 409 let subject = fmt_out_subject(&tx.wtid, &tx.exchange_base_url, tx.metadata.as_deref()); 410 serialized!( 411 sqlx::query( 412 " 413 SELECT out_request_uid_reuse, out_wtid_reuse, out_initiated_row_id, out_initiated_at 414 FROM taler_transfer($1, $2, $3, $4, $5, $6, $7, $8, $9) 415 ", 416 ) 417 .bind(tx.request_uid) 418 .bind(tx.wtid) 419 .bind(&subject) 420 .bind(tx.amount) 421 .bind(tx.exchange_base_url.as_str()) 422 .bind(&tx.metadata) 423 .bind(tx.creditor.iban()) 424 .bind(&tx.creditor.name) 425 .bind_timestamp(now) 426 .try_map(|r: PgRow| { 427 Ok(if r.try_get_flag(0)? { 428 TransferResult::RequestUidReuse 429 } else if r.try_get_flag(1)? { 430 TransferResult::WtidReuse 431 } else { 432 TransferResult::Success { 433 id: r.try_get_u64(2)?, 434 initiated_at: r.try_get_timestamp(3)?, 435 } 436 }) 437 }) 438 .fetch_one(db) 439 ) 440 } 441 442 #[derive(Debug, PartialEq, Eq)] 443 pub struct BounceResult { 444 pub tx_id: u64, 445 pub tx_new: bool, 446 pub bounce_id: u64, 447 pub bounce_new: bool, 448 } 449 450 pub async fn register_bounce_tx_in( 451 db: &mut PgConnection, 452 tx: &TxIn, 453 reason: &str, 454 now: &Timestamp, 455 ) -> sqlx::Result<BounceResult> { 456 serialized!( 457 sqlx::query( 458 " 459 SELECT out_tx_row_id, out_tx_new, out_bounce_row_id, out_bounce_new 460 FROM register_bounce_tx_in($1, $2, $3, $4, $5, $6, $7, $8) 461 ", 462 ) 463 .bind(tx.code as i64) 464 .bind(tx.amount) 465 .bind(&tx.subject) 466 .bind(tx.debtor.iban()) 467 .bind(&tx.debtor.name) 468 .bind_date(&tx.value_date) 469 .bind(reason) 470 .bind_timestamp(now) 471 .try_map(|r: PgRow| { 472 Ok(BounceResult { 473 tx_id: r.try_get_u64(0)?, 474 tx_new: r.try_get(1)?, 475 bounce_id: r.try_get_u64(2)?, 476 bounce_new: r.try_get(3)?, 477 }) 478 }) 479 .fetch_one(&mut *db) 480 ) 481 } 482 483 pub async fn transfer_page( 484 db: &PgPool, 485 status: &Option<TransferState>, 486 params: &Page, 487 ) -> sqlx::Result<Vec<TransferListStatus>> { 488 page( 489 db, 490 params, 491 "initiated_id", 492 || { 493 let mut builder = QueryBuilder::new( 494 " 495 SELECT 496 initiated_id, 497 status, 498 amount, 499 credit_account, 500 credit_name, 501 initiated_at 502 FROM transfer 503 JOIN initiated USING (initiated_id) 504 WHERE 505 ", 506 ); 507 if let Some(status) = status { 508 builder.push(" status = ").push_bind(status).push(" AND "); 509 } 510 builder 511 }, 512 |r: PgRow| { 513 Ok(TransferListStatus { 514 row_id: r.try_get_u64(0)?, 515 status: r.try_get(1)?, 516 amount: r.try_get_amount(2, &CURR)?, 517 credit_account: r.try_get_iban(3)?.as_full_uri(r.try_get(4)?), 518 timestamp: r.try_get_timestamp(5)?.into(), 519 }) 520 }, 521 ) 522 .await 523 } 524 525 pub async fn outgoing_history( 526 db: &PgPool, 527 params: &History, 528 listen: impl FnOnce() -> Receiver<i64>, 529 ) -> sqlx::Result<Vec<OutgoingBankTransaction>> { 530 history( 531 db, 532 "tx_out_id", 533 params, 534 listen, 535 || { 536 QueryBuilder::new( 537 " 538 SELECT 539 tx_out_id, 540 amount, 541 credit_account, 542 credit_name, 543 valued_at, 544 exchange_base_url, 545 metadata, 546 wtid 547 FROM taler_out 548 JOIN tx_out USING (tx_out_id) 549 WHERE 550 ", 551 ) 552 }, 553 |r: PgRow| { 554 Ok(OutgoingBankTransaction { 555 row_id: r.try_get_u64(0)?, 556 amount: r.try_get_amount(1, &CURR)?, 557 debit_fee: None, 558 credit_account: r.try_get_iban(2)?.as_full_uri(r.try_get(3)?), 559 date: r.try_get_timestamp(4)?.into(), 560 exchange_base_url: r.try_get_url(5)?, 561 metadata: r.try_get(6)?, 562 wtid: r.try_get(7)?, 563 }) 564 }, 565 ) 566 .await 567 } 568 569 pub async fn incoming_history( 570 db: &PgPool, 571 params: &History, 572 listen: impl FnOnce() -> Receiver<i64>, 573 ) -> sqlx::Result<Vec<IncomingBankTransaction>> { 574 history( 575 db, 576 "tx_in_id", 577 params, 578 listen, 579 || { 580 QueryBuilder::new( 581 " 582 SELECT 583 type, 584 tx_in_id, 585 amount, 586 debit_account, 587 debit_name, 588 valued_at, 589 metadata, 590 authorization_pub, 591 authorization_sig 592 FROM taler_in 593 JOIN tx_in USING (tx_in_id) 594 WHERE 595 ", 596 ) 597 }, 598 |r: PgRow| { 599 Ok(match r.try_get(0)? { 600 IncomingType::reserve => IncomingBankTransaction::Reserve { 601 row_id: r.try_get_u64(1)?, 602 amount: r.try_get_amount(2, &CURR)?, 603 credit_fee: None, 604 debit_account: r.try_get_iban(3)?.as_full_uri(r.try_get(4)?), 605 date: r.try_get_timestamp(5)?.into(), 606 reserve_pub: r.try_get(6)?, 607 authorization_pub: r.try_get(7)?, 608 authorization_sig: r.try_get(8)?, 609 }, 610 IncomingType::kyc => IncomingBankTransaction::Kyc { 611 row_id: r.try_get_u64(1)?, 612 amount: r.try_get_amount(2, &CURR)?, 613 credit_fee: None, 614 debit_account: r.try_get_iban(3)?.as_full_uri(r.try_get(4)?), 615 date: r.try_get_timestamp(5)?.into(), 616 account_pub: r.try_get(6)?, 617 authorization_pub: r.try_get(7)?, 618 authorization_sig: r.try_get(8)?, 619 }, 620 IncomingType::map => unimplemented!("MAP are never listed in the history"), 621 }) 622 }, 623 ) 624 .await 625 } 626 627 pub async fn revenue_history( 628 db: &PgPool, 629 params: &History, 630 listen: impl FnOnce() -> Receiver<i64>, 631 ) -> sqlx::Result<Vec<RevenueIncomingBankTransaction>> { 632 history( 633 db, 634 "tx_in_id", 635 params, 636 listen, 637 || { 638 QueryBuilder::new( 639 " 640 SELECT 641 tx_in_id, 642 valued_at, 643 amount, 644 debit_account, 645 debit_name, 646 subject 647 FROM tx_in 648 WHERE 649 ", 650 ) 651 }, 652 |r: PgRow| { 653 Ok(RevenueIncomingBankTransaction { 654 row_id: r.try_get_u64(0)?, 655 date: r.try_get_timestamp(1)?.into(), 656 amount: r.try_get_amount(2, &CURR)?, 657 credit_fee: None, 658 debit_account: r.try_get_iban(3)?.as_full_uri(r.try_get(4)?), 659 subject: r.try_get(5)?, 660 }) 661 }, 662 ) 663 .await 664 } 665 666 pub async fn transfer_by_id(db: &PgPool, id: u64) -> sqlx::Result<Option<TransferStatus>> { 667 serialized!( 668 sqlx::query( 669 " 670 SELECT 671 status, 672 status_msg, 673 amount, 674 exchange_base_url, 675 metadata, 676 wtid, 677 credit_account, 678 credit_name, 679 initiated_at 680 FROM transfer 681 JOIN initiated USING (initiated_id) 682 WHERE initiated_id = $1 683 ", 684 ) 685 .bind(id as i64) 686 .try_map(|r: PgRow| { 687 Ok(TransferStatus { 688 status: r.try_get(0)?, 689 status_msg: r.try_get(1)?, 690 amount: r.try_get_amount(2, &CURR)?, 691 exchange_base_url: r.try_get(3)?, 692 metadata: r.try_get(4)?, 693 wtid: r.try_get(5)?, 694 credit_account: r.try_get_iban(6)?.as_full_uri(r.try_get(7)?), 695 timestamp: r.try_get_timestamp(8)?.into(), 696 }) 697 }) 698 .fetch_optional(db) 699 ) 700 } 701 702 /** Get a batch of pending initiated transactions not attempted since [start] */ 703 pub async fn pending_batch( 704 db: &mut PgConnection, 705 start: &Timestamp, 706 ) -> sqlx::Result<Vec<Initiated>> { 707 serialized!( 708 sqlx::query( 709 " 710 SELECT initiated_id, amount, subject, credit_account, credit_name 711 FROM initiated 712 WHERE magnet_code IS NULL 713 AND status='pending' 714 AND (last_submitted IS NULL OR last_submitted < $1) 715 LIMIT 100 716 ", 717 ) 718 .bind_timestamp(start) 719 .try_map(|r: PgRow| { 720 Ok(Initiated { 721 id: r.try_get_u64(0)?, 722 amount: r.try_get_amount(1, &CURR)?, 723 subject: r.try_get(2)?, 724 creditor: FullHuPayto::new(r.try_get_parse(3)?, r.try_get(4)?), 725 }) 726 }) 727 .fetch_all(&mut *db) 728 ) 729 } 730 731 /** Get an initiated transaction matching the given magnet [code] */ 732 pub async fn initiated_by_code( 733 db: &mut PgConnection, 734 code: u64, 735 ) -> sqlx::Result<Option<Initiated>> { 736 serialized!( 737 sqlx::query( 738 " 739 SELECT initiated_id, amount, subject, credit_account, credit_name 740 FROM initiated 741 WHERE magnet_code IS $1 742 ", 743 ) 744 .bind(code as i64) 745 .try_map(|r: PgRow| { 746 Ok(Initiated { 747 id: r.try_get_u64(0)?, 748 amount: r.try_get_amount(1, &CURR)?, 749 subject: r.try_get(2)?, 750 creditor: FullHuPayto::new(r.try_get_parse(3)?, r.try_get(4)?), 751 }) 752 }) 753 .fetch_optional(&mut *db) 754 ) 755 } 756 757 /** Update status of a successful submitted initiated transaction */ 758 pub async fn initiated_submit_success( 759 db: &mut PgConnection, 760 id: u64, 761 timestamp: &Timestamp, 762 magnet_code: u64, 763 ) -> sqlx::Result<()> { 764 serialized!( 765 sqlx::query( 766 " 767 UPDATE initiated 768 SET status='pending', submission_counter=submission_counter+1, last_submitted=$1, magnet_code=$2 769 WHERE initiated_id=$3 770 " 771 ).bind_timestamp(timestamp) 772 .bind(magnet_code as i64) 773 .bind(id as i64) 774 .execute(&mut *db) 775 )?; 776 Ok(()) 777 } 778 779 /** Update status of a permanently failed initiated transaction */ 780 pub async fn initiated_submit_permanent_failure( 781 db: &mut PgConnection, 782 id: u64, 783 timestamp: &Timestamp, 784 msg: &str, 785 ) -> sqlx::Result<()> { 786 serialized!( 787 sqlx::query( 788 " 789 UPDATE initiated 790 SET status='permanent_failure', status_msg=$2 791 WHERE initiated_id=$3 792 ", 793 ) 794 .bind_timestamp(timestamp) 795 .bind(msg) 796 .bind(id as i64) 797 .execute(&mut *db) 798 )?; 799 Ok(()) 800 } 801 802 /** Check if an initiated transaction exist for a magnet code */ 803 pub async fn initiated_exists_for_code( 804 db: &mut PgConnection, 805 code: u64, 806 ) -> sqlx::Result<Option<u64>> { 807 serialized!( 808 sqlx::query("SELECT initiated_id FROM initiated WHERE magnet_code=$1") 809 .bind(code as i64) 810 .try_map(|r| Ok(r.try_get::<i64, _>(0)? as u64)) 811 .fetch_optional(&mut *db) 812 ) 813 } 814 815 /** Get JSON value from KV table */ 816 pub async fn kv_get<T: DeserializeOwned + Unpin + Send>( 817 db: &mut PgConnection, 818 key: &str, 819 ) -> sqlx::Result<Option<T>> { 820 serialized!( 821 sqlx::query("SELECT value FROM kv WHERE key=$1") 822 .bind(key) 823 .try_map(|r| Ok(r.try_get::<sqlx::types::Json<T>, _>(0)?.0)) 824 .fetch_optional(&mut *db) 825 ) 826 } 827 828 /** Set JSON value in KV table */ 829 pub async fn kv_set<T: Serialize>(db: &mut PgConnection, key: &str, value: &T) -> sqlx::Result<()> { 830 serialized!( 831 sqlx::query("INSERT INTO kv (key, value) VALUES ($1, $2) ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value") 832 .bind(key) 833 .bind(sqlx::types::Json(value)) 834 .execute(&mut *db) 835 )?; 836 Ok(()) 837 } 838 839 pub enum RegistrationResult { 840 Success, 841 ReservePubReuse, 842 } 843 844 pub async fn transfer_register( 845 db: &PgPool, 846 req: &RegistrationRequest, 847 ) -> sqlx::Result<RegistrationResult> { 848 let ty: IncomingType = req.r#type.into(); 849 serialized!( 850 sqlx::query( 851 "SELECT out_reserve_pub_reuse FROM register_prepared_transfers($1,$2,$3,$4,$5,$6)" 852 ) 853 .bind(ty) 854 .bind(req.account_pub) 855 .bind(req.authorization_pub) 856 .bind(req.authorization_sig) 857 .bind(req.recurrent) 858 .bind_timestamp(&Timestamp::now()) 859 .try_map(|r: PgRow| { 860 Ok(if r.try_get_flag("out_reserve_pub_reuse")? { 861 RegistrationResult::ReservePubReuse 862 } else { 863 RegistrationResult::Success 864 }) 865 }) 866 .fetch_one(db) 867 ) 868 } 869 870 pub async fn transfer_unregister(db: &PgPool, req: &Unregistration) -> sqlx::Result<bool> { 871 serialized!( 872 sqlx::query("SELECT out_found FROM delete_prepared_transfers($1,$2)") 873 .bind(req.authorization_pub) 874 .bind_timestamp(&Timestamp::now()) 875 .try_map(|r: PgRow| r.try_get_flag("out_found")) 876 .fetch_one(db) 877 ) 878 } 879 880 #[cfg(test)] 881 mod test { 882 use std::assert_matches; 883 884 use jiff::{Span, Timestamp, Zoned}; 885 use serde_json::json; 886 use sqlx::{PgPool, Postgres, pool::PoolConnection, postgres::PgRow}; 887 use taler_api::{ 888 db::TypeHelper, 889 notification::dummy_listen, 890 subject::{IncomingSubject, OutgoingSubject}, 891 }; 892 use taler_common::{ 893 api::{ 894 EddsaPublicKey, HashCode, ShortHashCode, 895 params::{History, Page}, 896 }, 897 types::{ 898 amount::{amount, decimal}, 899 url, 900 utils::now_sql_stable_ts, 901 }, 902 }; 903 904 use super::TxInAdmin; 905 use crate::{ 906 constants::CONFIG_SOURCE, 907 db::{ 908 self, AddIncomingResult, AddOutgoingResult, BounceResult, Initiated, OutFailureResult, 909 TransferResult, TxIn, TxOut, TxOutKind, kv_get, kv_set, make_transfer, 910 register_bounce_tx_in, register_tx_in, register_tx_in_admin, register_tx_out, 911 }, 912 magnet_api::types::TxStatus, 913 magnet_payto, 914 }; 915 916 async fn setup() -> (PoolConnection<Postgres>, PgPool) { 917 taler_test_utils::db::db_test_setup(CONFIG_SOURCE).await 918 } 919 920 #[tokio::test] 921 async fn kv() { 922 let (mut db, _) = setup().await; 923 924 let value = json!({ 925 "name": "Mr Smith", 926 "no way": 32 927 }); 928 929 assert_eq!( 930 kv_get::<serde_json::Value>(&mut db, "value").await.unwrap(), 931 None 932 ); 933 kv_set(&mut db, "value", &value).await.unwrap(); 934 kv_set(&mut db, "value", &value).await.unwrap(); 935 assert_eq!( 936 kv_get::<serde_json::Value>(&mut db, "value").await.unwrap(), 937 Some(value) 938 ); 939 } 940 941 #[tokio::test] 942 async fn tx_in() { 943 let (mut db, pool) = setup().await; 944 945 let mut routine = async |first: &Option<IncomingSubject>, 946 second: &Option<IncomingSubject>| { 947 let (id, code) = 948 sqlx::query("SELECT count(*) + 1, COALESCE(max(magnet_code), 0) + 20 FROM tx_in") 949 .try_map(|r: PgRow| Ok((r.try_get_u64(0)?, r.try_get_u64(1)?))) 950 .fetch_one(&mut *db) 951 .await 952 .unwrap(); 953 let now = now_sql_stable_ts(); 954 let date = Zoned::now().date(); 955 let later = date.tomorrow().unwrap(); 956 let tx = TxIn { 957 code, 958 amount: amount("EUR:10"), 959 subject: "subject".into(), 960 debtor: magnet_payto( 961 "payto://iban/HU30162000031000163100000000?receiver-name=name", 962 ), 963 value_date: date, 964 status: TxStatus::Completed, 965 }; 966 // Insert 967 assert_eq!( 968 register_tx_in(&mut db, &tx, first, &now) 969 .await 970 .expect("register tx in"), 971 AddIncomingResult::Success { 972 new: true, 973 pending: false, 974 row_id: id, 975 valued_at: date 976 } 977 ); 978 // Idempotent 979 assert_eq!( 980 register_tx_in( 981 &mut db, 982 &TxIn { 983 value_date: later, 984 ..tx.clone() 985 }, 986 first, 987 &now 988 ) 989 .await 990 .expect("register tx in"), 991 AddIncomingResult::Success { 992 new: false, 993 pending: false, 994 row_id: id, 995 valued_at: date 996 } 997 ); 998 // Many 999 assert_eq!( 1000 register_tx_in( 1001 &mut db, 1002 &TxIn { 1003 code: code + 1, 1004 value_date: later, 1005 ..tx 1006 }, 1007 second, 1008 &now 1009 ) 1010 .await 1011 .expect("register tx in"), 1012 AddIncomingResult::Success { 1013 new: true, 1014 pending: false, 1015 row_id: id + 1, 1016 valued_at: later 1017 } 1018 ); 1019 }; 1020 1021 // Empty db 1022 assert_eq!( 1023 db::revenue_history(&pool, &History::default(), dummy_listen) 1024 .await 1025 .unwrap(), 1026 Vec::new() 1027 ); 1028 assert_eq!( 1029 db::incoming_history(&pool, &History::default(), dummy_listen) 1030 .await 1031 .unwrap(), 1032 Vec::new() 1033 ); 1034 1035 // Regular transaction 1036 routine(&None, &None).await; 1037 1038 // Reserve transaction 1039 routine( 1040 &Some(IncomingSubject::Reserve(EddsaPublicKey::rand())), 1041 &Some(IncomingSubject::Reserve(EddsaPublicKey::rand())), 1042 ) 1043 .await; 1044 1045 // Kyc transaction 1046 routine( 1047 &Some(IncomingSubject::Kyc(EddsaPublicKey::rand())), 1048 &Some(IncomingSubject::Kyc(EddsaPublicKey::rand())), 1049 ) 1050 .await; 1051 1052 // History 1053 assert_eq!( 1054 db::revenue_history(&pool, &History::default(), dummy_listen) 1055 .await 1056 .unwrap() 1057 .len(), 1058 6 1059 ); 1060 assert_eq!( 1061 db::incoming_history(&pool, &History::default(), dummy_listen) 1062 .await 1063 .unwrap() 1064 .len(), 1065 4 1066 ); 1067 } 1068 1069 #[tokio::test] 1070 async fn tx_in_admin() { 1071 let (_, pool) = setup().await; 1072 1073 // Empty db 1074 assert_eq!( 1075 db::incoming_history(&pool, &History::default(), dummy_listen) 1076 .await 1077 .unwrap(), 1078 Vec::new() 1079 ); 1080 1081 let now = now_sql_stable_ts(); 1082 let later = now + Span::new().hours(2); 1083 let date = Zoned::now().date(); 1084 let tx = TxInAdmin { 1085 amount: amount("EUR:10"), 1086 subject: "subject".to_owned(), 1087 debtor: magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"), 1088 metadata: IncomingSubject::Reserve(EddsaPublicKey::rand()), 1089 }; 1090 // Insert 1091 assert_eq!( 1092 register_tx_in_admin(&pool, &tx, &now) 1093 .await 1094 .expect("register tx in"), 1095 AddIncomingResult::Success { 1096 new: true, 1097 pending: false, 1098 row_id: 1, 1099 valued_at: date 1100 } 1101 ); 1102 // Many 1103 assert_eq!( 1104 register_tx_in_admin( 1105 &pool, 1106 &TxInAdmin { 1107 subject: "Other".to_owned(), 1108 metadata: IncomingSubject::Reserve(EddsaPublicKey::rand()), 1109 ..tx.clone() 1110 }, 1111 &later 1112 ) 1113 .await 1114 .expect("register tx in"), 1115 AddIncomingResult::Success { 1116 new: true, 1117 pending: false, 1118 row_id: 2, 1119 valued_at: date 1120 } 1121 ); 1122 1123 // History 1124 assert_eq!( 1125 db::incoming_history(&pool, &History::default(), dummy_listen) 1126 .await 1127 .unwrap() 1128 .len(), 1129 2 1130 ); 1131 } 1132 1133 #[tokio::test] 1134 async fn tx_out() { 1135 let (mut db, pool) = setup().await; 1136 1137 let mut routine = async |first: &TxOutKind, second: &TxOutKind| { 1138 let (id, code) = 1139 sqlx::query("SELECT count(*) + 1, COALESCE(max(magnet_code), 0) + 20 FROM tx_out") 1140 .try_map(|r: PgRow| Ok((r.try_get_u64(0)?, r.try_get_u64(1)?))) 1141 .fetch_one(&mut *db) 1142 .await 1143 .unwrap(); 1144 let now = now_sql_stable_ts(); 1145 let date = Zoned::now().date(); 1146 let later = date.tomorrow().unwrap(); 1147 let tx = TxOut { 1148 code, 1149 amount: amount("HUF:10"), 1150 subject: "subject".into(), 1151 creditor: magnet_payto( 1152 "payto://iban/HU30162000031000163100000000?receiver-name=name", 1153 ), 1154 value_date: date, 1155 status: TxStatus::Completed, 1156 }; 1157 assert_matches!( 1158 make_transfer( 1159 &pool, 1160 &db::Transfer { 1161 request_uid: HashCode::rand(), 1162 amount: decimal("10"), 1163 exchange_base_url: url("https://exchange.test.com/"), 1164 metadata: None, 1165 wtid: ShortHashCode::rand(), 1166 creditor: tx.creditor.clone() 1167 }, 1168 &now 1169 ) 1170 .await 1171 .unwrap(), 1172 TransferResult::Success { .. } 1173 ); 1174 db::initiated_submit_success(&mut db, 1, &Timestamp::now(), tx.code) 1175 .await 1176 .expect("status success"); 1177 1178 // Insert 1179 assert_eq!( 1180 register_tx_out(&mut db, &tx, first, &now) 1181 .await 1182 .expect("register tx out"), 1183 AddOutgoingResult { 1184 result: db::RegisterResult::known, 1185 row_id: id, 1186 } 1187 ); 1188 // Idempotent 1189 assert_eq!( 1190 register_tx_out( 1191 &mut db, 1192 &TxOut { 1193 value_date: later, 1194 ..tx.clone() 1195 }, 1196 first, 1197 &now 1198 ) 1199 .await 1200 .expect("register tx out"), 1201 AddOutgoingResult { 1202 result: db::RegisterResult::idempotent, 1203 row_id: id, 1204 } 1205 ); 1206 // Recovered 1207 assert_eq!( 1208 register_tx_out( 1209 &mut db, 1210 &TxOut { 1211 code: code + 1, 1212 value_date: later, 1213 ..tx.clone() 1214 }, 1215 second, 1216 &now 1217 ) 1218 .await 1219 .expect("register tx out"), 1220 AddOutgoingResult { 1221 result: db::RegisterResult::recovered, 1222 row_id: id + 1, 1223 } 1224 ); 1225 }; 1226 1227 // Empty db 1228 assert_eq!( 1229 db::outgoing_history(&pool, &History::default(), dummy_listen) 1230 .await 1231 .unwrap(), 1232 Vec::new() 1233 ); 1234 1235 // Regular transaction 1236 routine(&TxOutKind::Simple, &TxOutKind::Simple).await; 1237 1238 // Talerable transaction 1239 routine( 1240 &TxOutKind::Talerable(OutgoingSubject::rand()), 1241 &TxOutKind::Talerable(OutgoingSubject::rand()), 1242 ) 1243 .await; 1244 1245 // Bounced transaction 1246 routine(&TxOutKind::Bounce(21), &TxOutKind::Bounce(42)).await; 1247 1248 // History 1249 assert_eq!( 1250 db::outgoing_history(&pool, &History::default(), dummy_listen) 1251 .await 1252 .unwrap() 1253 .len(), 1254 2 1255 ); 1256 } 1257 1258 #[tokio::test] 1259 async fn tx_out_failure() { 1260 let (mut db, pool) = setup().await; 1261 1262 let now = now_sql_stable_ts(); 1263 1264 // Unknown 1265 assert_eq!( 1266 db::register_tx_out_failure(&mut db, 42, None, &now) 1267 .await 1268 .unwrap(), 1269 OutFailureResult { 1270 initiated_id: None, 1271 new: false 1272 } 1273 ); 1274 assert_eq!( 1275 db::register_tx_out_failure(&mut db, 42, Some(12), &now) 1276 .await 1277 .unwrap(), 1278 OutFailureResult { 1279 initiated_id: None, 1280 new: false 1281 } 1282 ); 1283 1284 // Initiated 1285 let req = db::Transfer { 1286 request_uid: HashCode::rand(), 1287 amount: decimal("10"), 1288 exchange_base_url: url("https://exchange.test.com/"), 1289 metadata: None, 1290 wtid: ShortHashCode::rand(), 1291 creditor: magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"), 1292 }; 1293 let payto = magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"); 1294 assert_eq!( 1295 make_transfer(&pool, &req, &now).await.unwrap(), 1296 TransferResult::Success { 1297 id: 1, 1298 initiated_at: now 1299 } 1300 ); 1301 db::initiated_submit_success(&mut db, 1, &Timestamp::now(), 34) 1302 .await 1303 .expect("status success"); 1304 assert_eq!( 1305 db::register_tx_out_failure(&mut db, 34, None, &now) 1306 .await 1307 .unwrap(), 1308 OutFailureResult { 1309 initiated_id: Some(1), 1310 new: true 1311 } 1312 ); 1313 assert_eq!( 1314 db::register_tx_out_failure(&mut db, 34, None, &now) 1315 .await 1316 .unwrap(), 1317 OutFailureResult { 1318 initiated_id: Some(1), 1319 new: false 1320 } 1321 ); 1322 1323 // Recovered bounce 1324 let tx = TxIn { 1325 code: 12, 1326 amount: amount("HUF:11"), 1327 subject: "malformed transaction".into(), 1328 debtor: payto, 1329 value_date: Zoned::now().date(), 1330 status: TxStatus::Completed, 1331 }; 1332 assert_eq!( 1333 db::register_bounce_tx_in(&mut db, &tx, "no reason", &now) 1334 .await 1335 .unwrap(), 1336 BounceResult { 1337 tx_id: 1, 1338 tx_new: true, 1339 bounce_id: 2, 1340 bounce_new: true 1341 } 1342 ); 1343 assert_eq!( 1344 db::register_tx_out_failure(&mut db, 10, Some(12), &now) 1345 .await 1346 .unwrap(), 1347 OutFailureResult { 1348 initiated_id: Some(2), 1349 new: true 1350 } 1351 ); 1352 assert_eq!( 1353 db::register_tx_out_failure(&mut db, 10, Some(12), &now) 1354 .await 1355 .unwrap(), 1356 OutFailureResult { 1357 initiated_id: Some(2), 1358 new: false 1359 } 1360 ); 1361 } 1362 1363 #[tokio::test] 1364 async fn transfer() { 1365 let (_, pool) = setup().await; 1366 1367 // Empty db 1368 assert_eq!(db::transfer_by_id(&pool, 0).await.unwrap(), None); 1369 assert_eq!( 1370 db::transfer_page(&pool, &None, &Page::default()) 1371 .await 1372 .unwrap(), 1373 Vec::new() 1374 ); 1375 1376 let req = db::Transfer { 1377 request_uid: HashCode::rand(), 1378 amount: decimal("10"), 1379 exchange_base_url: url("https://exchange.test.com/"), 1380 metadata: None, 1381 wtid: ShortHashCode::rand(), 1382 creditor: magnet_payto("payto://iban/HU02162000031000164800000000?receiver-name=name"), 1383 }; 1384 let now = now_sql_stable_ts(); 1385 let later = now + Span::new().hours(2); 1386 // Insert 1387 assert_eq!( 1388 make_transfer(&pool, &req, &now).await.expect("transfer"), 1389 TransferResult::Success { 1390 id: 1, 1391 initiated_at: now 1392 } 1393 ); 1394 // Idempotent 1395 assert_eq!( 1396 make_transfer(&pool, &req, &later).await.expect("transfer"), 1397 TransferResult::Success { 1398 id: 1, 1399 initiated_at: now 1400 } 1401 ); 1402 // Request UID reuse 1403 assert_eq!( 1404 make_transfer( 1405 &pool, 1406 &db::Transfer { 1407 wtid: ShortHashCode::rand(), 1408 ..req.clone() 1409 }, 1410 &now 1411 ) 1412 .await 1413 .expect("transfer"), 1414 TransferResult::RequestUidReuse 1415 ); 1416 // wtid reuse 1417 assert_eq!( 1418 make_transfer( 1419 &pool, 1420 &db::Transfer { 1421 request_uid: HashCode::rand(), 1422 ..req.clone() 1423 }, 1424 &now 1425 ) 1426 .await 1427 .expect("transfer"), 1428 TransferResult::WtidReuse 1429 ); 1430 // Many 1431 assert_eq!( 1432 make_transfer( 1433 &pool, 1434 &db::Transfer { 1435 request_uid: HashCode::rand(), 1436 wtid: ShortHashCode::rand(), 1437 ..req 1438 }, 1439 &later 1440 ) 1441 .await 1442 .expect("transfer"), 1443 TransferResult::Success { 1444 id: 2, 1445 initiated_at: later 1446 } 1447 ); 1448 1449 // Get 1450 assert!(db::transfer_by_id(&pool, 1).await.unwrap().is_some()); 1451 assert!(db::transfer_by_id(&pool, 2).await.unwrap().is_some()); 1452 assert!(db::transfer_by_id(&pool, 3).await.unwrap().is_none()); 1453 assert_eq!( 1454 db::transfer_page(&pool, &None, &Page::default()) 1455 .await 1456 .unwrap() 1457 .len(), 1458 2 1459 ); 1460 } 1461 1462 #[tokio::test] 1463 async fn bounce() { 1464 let (mut db, _) = setup().await; 1465 1466 let amount = amount("HUF:10"); 1467 let payto = magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"); 1468 let now = now_sql_stable_ts(); 1469 let date = Zoned::now().date(); 1470 1471 // Empty db 1472 assert!(db::pending_batch(&mut db, &now).await.unwrap().is_empty()); 1473 1474 // Insert 1475 assert_eq!( 1476 register_tx_in( 1477 &mut db, 1478 &TxIn { 1479 code: 13, 1480 amount, 1481 subject: "subject".into(), 1482 debtor: payto.clone(), 1483 value_date: date, 1484 status: TxStatus::Completed 1485 }, 1486 &None, 1487 &now 1488 ) 1489 .await 1490 .expect("register tx in"), 1491 AddIncomingResult::Success { 1492 new: true, 1493 pending: false, 1494 row_id: 1, 1495 valued_at: date 1496 } 1497 ); 1498 1499 // Bounce 1500 assert_eq!( 1501 register_bounce_tx_in( 1502 &mut db, 1503 &TxIn { 1504 code: 12, 1505 amount, 1506 subject: "subject".into(), 1507 debtor: payto.clone(), 1508 value_date: date, 1509 status: TxStatus::Completed 1510 }, 1511 "good reason", 1512 &now 1513 ) 1514 .await 1515 .expect("bounce"), 1516 BounceResult { 1517 tx_id: 2, 1518 tx_new: true, 1519 bounce_id: 1, 1520 bounce_new: true 1521 } 1522 ); 1523 // Idempotent 1524 assert_eq!( 1525 register_bounce_tx_in( 1526 &mut db, 1527 &TxIn { 1528 code: 12, 1529 amount, 1530 subject: "subject".into(), 1531 debtor: payto.clone(), 1532 value_date: date, 1533 status: TxStatus::Completed 1534 }, 1535 "good reason", 1536 &now 1537 ) 1538 .await 1539 .expect("bounce"), 1540 BounceResult { 1541 tx_id: 2, 1542 tx_new: false, 1543 bounce_id: 1, 1544 bounce_new: false 1545 } 1546 ); 1547 1548 // Bounce registered 1549 assert_eq!( 1550 register_bounce_tx_in( 1551 &mut db, 1552 &TxIn { 1553 code: 13, 1554 amount, 1555 subject: "subject".into(), 1556 debtor: payto.clone(), 1557 value_date: date, 1558 status: TxStatus::Completed 1559 }, 1560 "good reason", 1561 &now 1562 ) 1563 .await 1564 .expect("bounce"), 1565 BounceResult { 1566 tx_id: 1, 1567 tx_new: false, 1568 bounce_id: 2, 1569 bounce_new: true 1570 } 1571 ); 1572 // Idempotent registered 1573 assert_eq!( 1574 register_bounce_tx_in( 1575 &mut db, 1576 &TxIn { 1577 code: 13, 1578 amount, 1579 subject: "subject".into(), 1580 debtor: payto.clone(), 1581 value_date: date, 1582 status: TxStatus::Completed 1583 }, 1584 "good reason", 1585 &now 1586 ) 1587 .await 1588 .expect("bounce"), 1589 BounceResult { 1590 tx_id: 1, 1591 tx_new: false, 1592 bounce_id: 2, 1593 bounce_new: false 1594 } 1595 ); 1596 1597 // Batch 1598 assert_eq!( 1599 db::pending_batch(&mut db, &now).await.unwrap(), 1600 &[ 1601 Initiated { 1602 id: 1, 1603 amount, 1604 subject: "bounce: 12".into(), 1605 creditor: payto.clone() 1606 }, 1607 Initiated { 1608 id: 2, 1609 amount, 1610 subject: "bounce: 13".into(), 1611 creditor: payto 1612 } 1613 ] 1614 ); 1615 } 1616 1617 #[tokio::test] 1618 async fn status() { 1619 let (mut db, _) = setup().await; 1620 1621 // Unknown transfer 1622 db::initiated_submit_permanent_failure(&mut db, 1, &Timestamp::now(), "msg") 1623 .await 1624 .unwrap(); 1625 db::initiated_submit_success(&mut db, 1, &Timestamp::now(), 12) 1626 .await 1627 .unwrap(); 1628 } 1629 1630 #[tokio::test] 1631 async fn batch() { 1632 let (mut db, pool) = setup().await; 1633 let start = Timestamp::now(); 1634 let magnet_payto = 1635 magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"); 1636 1637 // Empty db 1638 let pendings = db::pending_batch(&mut db, &start) 1639 .await 1640 .expect("pending_batch"); 1641 assert_eq!(pendings.len(), 0); 1642 1643 // Some transfers 1644 for i in 0..3 { 1645 make_transfer( 1646 &pool, 1647 &db::Transfer { 1648 request_uid: HashCode::rand(), 1649 amount: decimal(format!("{}", i + 1)), 1650 exchange_base_url: url("https://exchange.test.com/"), 1651 metadata: None, 1652 wtid: ShortHashCode::rand(), 1653 creditor: magnet_payto.clone(), 1654 }, 1655 &Timestamp::now(), 1656 ) 1657 .await 1658 .expect("transfer"); 1659 } 1660 let pendings = db::pending_batch(&mut db, &start) 1661 .await 1662 .expect("pending_batch"); 1663 assert_eq!(pendings.len(), 3); 1664 1665 // Max 100 txs in batch 1666 for i in 0..100 { 1667 make_transfer( 1668 &pool, 1669 &db::Transfer { 1670 request_uid: HashCode::rand(), 1671 amount: decimal(format!("{}", i + 1)), 1672 exchange_base_url: url("https://exchange.test.com/"), 1673 metadata: None, 1674 wtid: ShortHashCode::rand(), 1675 creditor: magnet_payto.clone(), 1676 }, 1677 &Timestamp::now(), 1678 ) 1679 .await 1680 .expect("transfer"); 1681 } 1682 let pendings = db::pending_batch(&mut db, &start) 1683 .await 1684 .expect("pending_batch"); 1685 assert_eq!(pendings.len(), 100); 1686 1687 // Skip uploaded 1688 for i in 0..=10 { 1689 db::initiated_submit_success(&mut db, i, &Timestamp::now(), i) 1690 .await 1691 .expect("status success"); 1692 } 1693 let pendings = db::pending_batch(&mut db, &start) 1694 .await 1695 .expect("pending_batch"); 1696 assert_eq!(pendings.len(), 93); 1697 1698 // Skip failed 1699 for i in 0..=10 { 1700 db::initiated_submit_permanent_failure(&mut db, 10 + i, &Timestamp::now(), "failure") 1701 .await 1702 .expect("status failure"); 1703 } 1704 let pendings = db::pending_batch(&mut db, &start) 1705 .await 1706 .expect("pending_batch"); 1707 assert_eq!(pendings.len(), 83); 1708 } 1709 }