db.rs (50131B)
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::{IncomingKey, 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: IncomingKey, 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<IncomingKey>, 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 jiff::{Span, Timestamp, Zoned}; 883 use serde_json::json; 884 use sqlx::{PgPool, Postgres, pool::PoolConnection, postgres::PgRow}; 885 use taler_api::{ 886 db::TypeHelper, 887 notification::dummy_listen, 888 subject::{IncomingKey, OutgoingSubject}, 889 }; 890 use taler_common::{ 891 api::{ 892 EddsaPublicKey, HashCode, ShortHashCode, 893 params::{History, Page}, 894 }, 895 types::{ 896 amount::{amount, decimal}, 897 url, 898 utils::now_sql_stable_ts, 899 }, 900 }; 901 902 use super::TxInAdmin; 903 use crate::{ 904 constants::CONFIG_SOURCE, 905 db::{ 906 self, AddIncomingResult, AddOutgoingResult, BounceResult, Initiated, OutFailureResult, 907 TransferResult, TxIn, TxOut, TxOutKind, kv_get, kv_set, make_transfer, 908 register_bounce_tx_in, register_tx_in, register_tx_in_admin, register_tx_out, 909 }, 910 magnet_api::types::TxStatus, 911 magnet_payto, 912 }; 913 914 async fn setup() -> (PoolConnection<Postgres>, PgPool) { 915 taler_test_utils::db::db_test_setup(CONFIG_SOURCE).await 916 } 917 918 #[tokio::test] 919 async fn kv() { 920 let (mut db, _) = setup().await; 921 922 let value = json!({ 923 "name": "Mr Smith", 924 "no way": 32 925 }); 926 927 assert_eq!( 928 kv_get::<serde_json::Value>(&mut db, "value").await.unwrap(), 929 None 930 ); 931 kv_set(&mut db, "value", &value).await.unwrap(); 932 kv_set(&mut db, "value", &value).await.unwrap(); 933 assert_eq!( 934 kv_get::<serde_json::Value>(&mut db, "value").await.unwrap(), 935 Some(value) 936 ); 937 } 938 939 #[tokio::test] 940 async fn tx_in() { 941 let (mut db, pool) = setup().await; 942 943 let mut routine = async |first: &Option<IncomingKey>, second: &Option<IncomingKey>| { 944 let (id, code) = 945 sqlx::query("SELECT count(*) + 1, COALESCE(max(magnet_code), 0) + 20 FROM tx_in") 946 .try_map(|r: PgRow| Ok((r.try_get_u64(0)?, r.try_get_u64(1)?))) 947 .fetch_one(&mut *db) 948 .await 949 .unwrap(); 950 let now = now_sql_stable_ts(); 951 let date = Zoned::now().date(); 952 let later = date.tomorrow().unwrap(); 953 let tx = TxIn { 954 code, 955 amount: amount("EUR:10"), 956 subject: "subject".into(), 957 debtor: magnet_payto( 958 "payto://iban/HU30162000031000163100000000?receiver-name=name", 959 ), 960 value_date: date, 961 status: TxStatus::Completed, 962 }; 963 // Insert 964 assert_eq!( 965 register_tx_in(&mut db, &tx, first, &now) 966 .await 967 .expect("register tx in"), 968 AddIncomingResult::Success { 969 new: true, 970 pending: false, 971 row_id: id, 972 valued_at: date 973 } 974 ); 975 // Idempotent 976 assert_eq!( 977 register_tx_in( 978 &mut db, 979 &TxIn { 980 value_date: later, 981 ..tx.clone() 982 }, 983 first, 984 &now 985 ) 986 .await 987 .expect("register tx in"), 988 AddIncomingResult::Success { 989 new: false, 990 pending: false, 991 row_id: id, 992 valued_at: date 993 } 994 ); 995 // Many 996 assert_eq!( 997 register_tx_in( 998 &mut db, 999 &TxIn { 1000 code: code + 1, 1001 value_date: later, 1002 ..tx 1003 }, 1004 second, 1005 &now 1006 ) 1007 .await 1008 .expect("register tx in"), 1009 AddIncomingResult::Success { 1010 new: true, 1011 pending: false, 1012 row_id: id + 1, 1013 valued_at: later 1014 } 1015 ); 1016 }; 1017 1018 // Empty db 1019 assert_eq!( 1020 db::revenue_history(&pool, &History::default(), dummy_listen) 1021 .await 1022 .unwrap(), 1023 Vec::new() 1024 ); 1025 assert_eq!( 1026 db::incoming_history(&pool, &History::default(), dummy_listen) 1027 .await 1028 .unwrap(), 1029 Vec::new() 1030 ); 1031 1032 // Regular transaction 1033 routine(&None, &None).await; 1034 1035 // Reserve transaction 1036 routine( 1037 &Some(IncomingKey::reserve(EddsaPublicKey::rand())), 1038 &Some(IncomingKey::reserve(EddsaPublicKey::rand())), 1039 ) 1040 .await; 1041 1042 // Kyc transaction 1043 routine( 1044 &Some(IncomingKey::kyc(EddsaPublicKey::rand())), 1045 &Some(IncomingKey::kyc(EddsaPublicKey::rand())), 1046 ) 1047 .await; 1048 1049 // History 1050 assert_eq!( 1051 db::revenue_history(&pool, &History::default(), dummy_listen) 1052 .await 1053 .unwrap() 1054 .len(), 1055 6 1056 ); 1057 assert_eq!( 1058 db::incoming_history(&pool, &History::default(), dummy_listen) 1059 .await 1060 .unwrap() 1061 .len(), 1062 4 1063 ); 1064 } 1065 1066 #[tokio::test] 1067 async fn tx_in_admin() { 1068 let (_, pool) = setup().await; 1069 1070 // Empty db 1071 assert_eq!( 1072 db::incoming_history(&pool, &History::default(), dummy_listen) 1073 .await 1074 .unwrap(), 1075 Vec::new() 1076 ); 1077 1078 let now = now_sql_stable_ts(); 1079 let later = now + Span::new().hours(2); 1080 let date = Zoned::now().date(); 1081 let tx = TxInAdmin { 1082 amount: amount("EUR:10"), 1083 subject: "subject".to_owned(), 1084 debtor: magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"), 1085 metadata: IncomingKey::reserve(EddsaPublicKey::rand()), 1086 }; 1087 // Insert 1088 assert_eq!( 1089 register_tx_in_admin(&pool, &tx, &now) 1090 .await 1091 .expect("register tx in"), 1092 AddIncomingResult::Success { 1093 new: true, 1094 pending: false, 1095 row_id: 1, 1096 valued_at: date 1097 } 1098 ); 1099 // Many 1100 assert_eq!( 1101 register_tx_in_admin( 1102 &pool, 1103 &TxInAdmin { 1104 subject: "Other".to_owned(), 1105 metadata: IncomingKey::reserve(EddsaPublicKey::rand()), 1106 ..tx.clone() 1107 }, 1108 &later 1109 ) 1110 .await 1111 .expect("register tx in"), 1112 AddIncomingResult::Success { 1113 new: true, 1114 pending: false, 1115 row_id: 2, 1116 valued_at: date 1117 } 1118 ); 1119 1120 // History 1121 assert_eq!( 1122 db::incoming_history(&pool, &History::default(), dummy_listen) 1123 .await 1124 .unwrap() 1125 .len(), 1126 2 1127 ); 1128 } 1129 1130 #[tokio::test] 1131 async fn tx_out() { 1132 let (mut db, pool) = setup().await; 1133 1134 let mut routine = async |first: &TxOutKind, second: &TxOutKind| { 1135 let (id, code) = 1136 sqlx::query("SELECT count(*) + 1, COALESCE(max(magnet_code), 0) + 20 FROM tx_out") 1137 .try_map(|r: PgRow| Ok((r.try_get_u64(0)?, r.try_get_u64(1)?))) 1138 .fetch_one(&mut *db) 1139 .await 1140 .unwrap(); 1141 let now = now_sql_stable_ts(); 1142 let date = Zoned::now().date(); 1143 let later = date.tomorrow().unwrap(); 1144 let tx = TxOut { 1145 code, 1146 amount: amount("HUF:10"), 1147 subject: "subject".into(), 1148 creditor: magnet_payto( 1149 "payto://iban/HU30162000031000163100000000?receiver-name=name", 1150 ), 1151 value_date: date, 1152 status: TxStatus::Completed, 1153 }; 1154 // TODO revert to assert_matches! once rust-lang#82775 is stable 1155 assert!(matches!( 1156 make_transfer( 1157 &pool, 1158 &db::Transfer { 1159 request_uid: HashCode::rand(), 1160 amount: decimal("10"), 1161 exchange_base_url: url("https://exchange.test.com/"), 1162 metadata: None, 1163 wtid: ShortHashCode::rand(), 1164 creditor: tx.creditor.clone() 1165 }, 1166 &now 1167 ) 1168 .await 1169 .unwrap(), 1170 TransferResult::Success { .. } 1171 )); 1172 db::initiated_submit_success(&mut db, 1, &Timestamp::now(), tx.code) 1173 .await 1174 .expect("status success"); 1175 1176 // Insert 1177 assert_eq!( 1178 register_tx_out(&mut db, &tx, first, &now) 1179 .await 1180 .expect("register tx out"), 1181 AddOutgoingResult { 1182 result: db::RegisterResult::known, 1183 row_id: id, 1184 } 1185 ); 1186 // Idempotent 1187 assert_eq!( 1188 register_tx_out( 1189 &mut db, 1190 &TxOut { 1191 value_date: later, 1192 ..tx.clone() 1193 }, 1194 first, 1195 &now 1196 ) 1197 .await 1198 .expect("register tx out"), 1199 AddOutgoingResult { 1200 result: db::RegisterResult::idempotent, 1201 row_id: id, 1202 } 1203 ); 1204 // Recovered 1205 assert_eq!( 1206 register_tx_out( 1207 &mut db, 1208 &TxOut { 1209 code: code + 1, 1210 value_date: later, 1211 ..tx.clone() 1212 }, 1213 second, 1214 &now 1215 ) 1216 .await 1217 .expect("register tx out"), 1218 AddOutgoingResult { 1219 result: db::RegisterResult::recovered, 1220 row_id: id + 1, 1221 } 1222 ); 1223 }; 1224 1225 // Empty db 1226 assert_eq!( 1227 db::outgoing_history(&pool, &History::default(), dummy_listen) 1228 .await 1229 .unwrap(), 1230 Vec::new() 1231 ); 1232 1233 // Regular transaction 1234 routine(&TxOutKind::Simple, &TxOutKind::Simple).await; 1235 1236 // Talerable transaction 1237 routine( 1238 &TxOutKind::Talerable(OutgoingSubject::rand()), 1239 &TxOutKind::Talerable(OutgoingSubject::rand()), 1240 ) 1241 .await; 1242 1243 // Bounced transaction 1244 routine(&TxOutKind::Bounce(21), &TxOutKind::Bounce(42)).await; 1245 1246 // History 1247 assert_eq!( 1248 db::outgoing_history(&pool, &History::default(), dummy_listen) 1249 .await 1250 .unwrap() 1251 .len(), 1252 2 1253 ); 1254 } 1255 1256 #[tokio::test] 1257 async fn tx_out_failure() { 1258 let (mut db, pool) = setup().await; 1259 1260 let now = now_sql_stable_ts(); 1261 1262 // Unknown 1263 assert_eq!( 1264 db::register_tx_out_failure(&mut db, 42, None, &now) 1265 .await 1266 .unwrap(), 1267 OutFailureResult { 1268 initiated_id: None, 1269 new: false 1270 } 1271 ); 1272 assert_eq!( 1273 db::register_tx_out_failure(&mut db, 42, Some(12), &now) 1274 .await 1275 .unwrap(), 1276 OutFailureResult { 1277 initiated_id: None, 1278 new: false 1279 } 1280 ); 1281 1282 // Initiated 1283 let req = db::Transfer { 1284 request_uid: HashCode::rand(), 1285 amount: decimal("10"), 1286 exchange_base_url: url("https://exchange.test.com/"), 1287 metadata: None, 1288 wtid: ShortHashCode::rand(), 1289 creditor: magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"), 1290 }; 1291 let payto = magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"); 1292 assert_eq!( 1293 make_transfer(&pool, &req, &now).await.unwrap(), 1294 TransferResult::Success { 1295 id: 1, 1296 initiated_at: now 1297 } 1298 ); 1299 db::initiated_submit_success(&mut db, 1, &Timestamp::now(), 34) 1300 .await 1301 .expect("status success"); 1302 assert_eq!( 1303 db::register_tx_out_failure(&mut db, 34, None, &now) 1304 .await 1305 .unwrap(), 1306 OutFailureResult { 1307 initiated_id: Some(1), 1308 new: true 1309 } 1310 ); 1311 assert_eq!( 1312 db::register_tx_out_failure(&mut db, 34, None, &now) 1313 .await 1314 .unwrap(), 1315 OutFailureResult { 1316 initiated_id: Some(1), 1317 new: false 1318 } 1319 ); 1320 1321 // Recovered bounce 1322 let tx = TxIn { 1323 code: 12, 1324 amount: amount("HUF:11"), 1325 subject: "malformed transaction".into(), 1326 debtor: payto, 1327 value_date: Zoned::now().date(), 1328 status: TxStatus::Completed, 1329 }; 1330 assert_eq!( 1331 db::register_bounce_tx_in(&mut db, &tx, "no reason", &now) 1332 .await 1333 .unwrap(), 1334 BounceResult { 1335 tx_id: 1, 1336 tx_new: true, 1337 bounce_id: 2, 1338 bounce_new: true 1339 } 1340 ); 1341 assert_eq!( 1342 db::register_tx_out_failure(&mut db, 10, Some(12), &now) 1343 .await 1344 .unwrap(), 1345 OutFailureResult { 1346 initiated_id: Some(2), 1347 new: true 1348 } 1349 ); 1350 assert_eq!( 1351 db::register_tx_out_failure(&mut db, 10, Some(12), &now) 1352 .await 1353 .unwrap(), 1354 OutFailureResult { 1355 initiated_id: Some(2), 1356 new: false 1357 } 1358 ); 1359 } 1360 1361 #[tokio::test] 1362 async fn transfer() { 1363 let (_, pool) = setup().await; 1364 1365 // Empty db 1366 assert_eq!(db::transfer_by_id(&pool, 0).await.unwrap(), None); 1367 assert_eq!( 1368 db::transfer_page(&pool, &None, &Page::default()) 1369 .await 1370 .unwrap(), 1371 Vec::new() 1372 ); 1373 1374 let req = db::Transfer { 1375 request_uid: HashCode::rand(), 1376 amount: decimal("10"), 1377 exchange_base_url: url("https://exchange.test.com/"), 1378 metadata: None, 1379 wtid: ShortHashCode::rand(), 1380 creditor: magnet_payto("payto://iban/HU02162000031000164800000000?receiver-name=name"), 1381 }; 1382 let now = now_sql_stable_ts(); 1383 let later = now + Span::new().hours(2); 1384 // Insert 1385 assert_eq!( 1386 make_transfer(&pool, &req, &now).await.expect("transfer"), 1387 TransferResult::Success { 1388 id: 1, 1389 initiated_at: now 1390 } 1391 ); 1392 // Idempotent 1393 assert_eq!( 1394 make_transfer(&pool, &req, &later).await.expect("transfer"), 1395 TransferResult::Success { 1396 id: 1, 1397 initiated_at: now 1398 } 1399 ); 1400 // Request UID reuse 1401 assert_eq!( 1402 make_transfer( 1403 &pool, 1404 &db::Transfer { 1405 wtid: ShortHashCode::rand(), 1406 ..req.clone() 1407 }, 1408 &now 1409 ) 1410 .await 1411 .expect("transfer"), 1412 TransferResult::RequestUidReuse 1413 ); 1414 // wtid reuse 1415 assert_eq!( 1416 make_transfer( 1417 &pool, 1418 &db::Transfer { 1419 request_uid: HashCode::rand(), 1420 ..req.clone() 1421 }, 1422 &now 1423 ) 1424 .await 1425 .expect("transfer"), 1426 TransferResult::WtidReuse 1427 ); 1428 // Many 1429 assert_eq!( 1430 make_transfer( 1431 &pool, 1432 &db::Transfer { 1433 request_uid: HashCode::rand(), 1434 wtid: ShortHashCode::rand(), 1435 ..req 1436 }, 1437 &later 1438 ) 1439 .await 1440 .expect("transfer"), 1441 TransferResult::Success { 1442 id: 2, 1443 initiated_at: later 1444 } 1445 ); 1446 1447 // Get 1448 assert!(db::transfer_by_id(&pool, 1).await.unwrap().is_some()); 1449 assert!(db::transfer_by_id(&pool, 2).await.unwrap().is_some()); 1450 assert!(db::transfer_by_id(&pool, 3).await.unwrap().is_none()); 1451 assert_eq!( 1452 db::transfer_page(&pool, &None, &Page::default()) 1453 .await 1454 .unwrap() 1455 .len(), 1456 2 1457 ); 1458 } 1459 1460 #[tokio::test] 1461 async fn bounce() { 1462 let (mut db, _) = setup().await; 1463 1464 let amount = amount("HUF:10"); 1465 let payto = magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"); 1466 let now = now_sql_stable_ts(); 1467 let date = Zoned::now().date(); 1468 1469 // Empty db 1470 assert!(db::pending_batch(&mut db, &now).await.unwrap().is_empty()); 1471 1472 // Insert 1473 assert_eq!( 1474 register_tx_in( 1475 &mut db, 1476 &TxIn { 1477 code: 13, 1478 amount, 1479 subject: "subject".into(), 1480 debtor: payto.clone(), 1481 value_date: date, 1482 status: TxStatus::Completed 1483 }, 1484 &None, 1485 &now 1486 ) 1487 .await 1488 .expect("register tx in"), 1489 AddIncomingResult::Success { 1490 new: true, 1491 pending: false, 1492 row_id: 1, 1493 valued_at: date 1494 } 1495 ); 1496 1497 // Bounce 1498 assert_eq!( 1499 register_bounce_tx_in( 1500 &mut db, 1501 &TxIn { 1502 code: 12, 1503 amount, 1504 subject: "subject".into(), 1505 debtor: payto.clone(), 1506 value_date: date, 1507 status: TxStatus::Completed 1508 }, 1509 "good reason", 1510 &now 1511 ) 1512 .await 1513 .expect("bounce"), 1514 BounceResult { 1515 tx_id: 2, 1516 tx_new: true, 1517 bounce_id: 1, 1518 bounce_new: true 1519 } 1520 ); 1521 // Idempotent 1522 assert_eq!( 1523 register_bounce_tx_in( 1524 &mut db, 1525 &TxIn { 1526 code: 12, 1527 amount, 1528 subject: "subject".into(), 1529 debtor: payto.clone(), 1530 value_date: date, 1531 status: TxStatus::Completed 1532 }, 1533 "good reason", 1534 &now 1535 ) 1536 .await 1537 .expect("bounce"), 1538 BounceResult { 1539 tx_id: 2, 1540 tx_new: false, 1541 bounce_id: 1, 1542 bounce_new: false 1543 } 1544 ); 1545 1546 // Bounce registered 1547 assert_eq!( 1548 register_bounce_tx_in( 1549 &mut db, 1550 &TxIn { 1551 code: 13, 1552 amount, 1553 subject: "subject".into(), 1554 debtor: payto.clone(), 1555 value_date: date, 1556 status: TxStatus::Completed 1557 }, 1558 "good reason", 1559 &now 1560 ) 1561 .await 1562 .expect("bounce"), 1563 BounceResult { 1564 tx_id: 1, 1565 tx_new: false, 1566 bounce_id: 2, 1567 bounce_new: true 1568 } 1569 ); 1570 // Idempotent registered 1571 assert_eq!( 1572 register_bounce_tx_in( 1573 &mut db, 1574 &TxIn { 1575 code: 13, 1576 amount, 1577 subject: "subject".into(), 1578 debtor: payto.clone(), 1579 value_date: date, 1580 status: TxStatus::Completed 1581 }, 1582 "good reason", 1583 &now 1584 ) 1585 .await 1586 .expect("bounce"), 1587 BounceResult { 1588 tx_id: 1, 1589 tx_new: false, 1590 bounce_id: 2, 1591 bounce_new: false 1592 } 1593 ); 1594 1595 // Batch 1596 assert_eq!( 1597 db::pending_batch(&mut db, &now).await.unwrap(), 1598 &[ 1599 Initiated { 1600 id: 1, 1601 amount, 1602 subject: "bounce: 12".into(), 1603 creditor: payto.clone() 1604 }, 1605 Initiated { 1606 id: 2, 1607 amount, 1608 subject: "bounce: 13".into(), 1609 creditor: payto 1610 } 1611 ] 1612 ); 1613 } 1614 1615 #[tokio::test] 1616 async fn status() { 1617 let (mut db, _) = setup().await; 1618 1619 // Unknown transfer 1620 db::initiated_submit_permanent_failure(&mut db, 1, &Timestamp::now(), "msg") 1621 .await 1622 .unwrap(); 1623 db::initiated_submit_success(&mut db, 1, &Timestamp::now(), 12) 1624 .await 1625 .unwrap(); 1626 } 1627 1628 #[tokio::test] 1629 async fn batch() { 1630 let (mut db, pool) = setup().await; 1631 let start = Timestamp::now(); 1632 let magnet_payto = 1633 magnet_payto("payto://iban/HU30162000031000163100000000?receiver-name=name"); 1634 1635 // Empty db 1636 let pendings = db::pending_batch(&mut db, &start) 1637 .await 1638 .expect("pending_batch"); 1639 assert_eq!(pendings.len(), 0); 1640 1641 // Some transfers 1642 for i in 0..3 { 1643 make_transfer( 1644 &pool, 1645 &db::Transfer { 1646 request_uid: HashCode::rand(), 1647 amount: decimal(format!("{}", i + 1)), 1648 exchange_base_url: url("https://exchange.test.com/"), 1649 metadata: None, 1650 wtid: ShortHashCode::rand(), 1651 creditor: magnet_payto.clone(), 1652 }, 1653 &Timestamp::now(), 1654 ) 1655 .await 1656 .expect("transfer"); 1657 } 1658 let pendings = db::pending_batch(&mut db, &start) 1659 .await 1660 .expect("pending_batch"); 1661 assert_eq!(pendings.len(), 3); 1662 1663 // Max 100 txs in batch 1664 for i in 0..100 { 1665 make_transfer( 1666 &pool, 1667 &db::Transfer { 1668 request_uid: HashCode::rand(), 1669 amount: decimal(format!("{}", i + 1)), 1670 exchange_base_url: url("https://exchange.test.com/"), 1671 metadata: None, 1672 wtid: ShortHashCode::rand(), 1673 creditor: magnet_payto.clone(), 1674 }, 1675 &Timestamp::now(), 1676 ) 1677 .await 1678 .expect("transfer"); 1679 } 1680 let pendings = db::pending_batch(&mut db, &start) 1681 .await 1682 .expect("pending_batch"); 1683 assert_eq!(pendings.len(), 100); 1684 1685 // Skip uploaded 1686 for i in 0..=10 { 1687 db::initiated_submit_success(&mut db, i, &Timestamp::now(), i) 1688 .await 1689 .expect("status success"); 1690 } 1691 let pendings = db::pending_batch(&mut db, &start) 1692 .await 1693 .expect("pending_batch"); 1694 assert_eq!(pendings.len(), 93); 1695 1696 // Skip failed 1697 for i in 0..=10 { 1698 db::initiated_submit_permanent_failure(&mut db, 10 + i, &Timestamp::now(), "failure") 1699 .await 1700 .expect("status failure"); 1701 } 1702 let pendings = db::pending_batch(&mut db, &start) 1703 .await 1704 .expect("pending_batch"); 1705 assert_eq!(pendings.len(), 83); 1706 } 1707 }