payment.rs (36500B)
1 /* 2 This file is part of TALER 3 Copyright (C) 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 compact_str::CompactString; 18 use jiff::Timestamp; 19 use libeufin_ebics::iso20022::model::{InTx, OutTx}; 20 use sqlx::{PgPool, Row as _, postgres::PgRow}; 21 use taler_api::{ 22 db::{BindHelper as _, TypeHelper as _}, 23 serialized, 24 subject::{IncomingKey, OutgoingSubject}, 25 }; 26 use taler_common::types::amount::Amount; 27 28 #[derive(Debug, PartialEq, Eq)] 29 pub struct OutgoingRegistrationResult { 30 pub id: u64, 31 pub initiated: bool, 32 pub new: bool, 33 } 34 35 /** Register an outgoing payment reconciling it with its initiated payment counterpart if present */ 36 pub async fn register_out_tx( 37 pool: &PgPool, 38 payment: &OutTx, 39 subject: Option<&OutgoingSubject>, 40 ) -> sqlx::Result<OutgoingRegistrationResult> { 41 serialized!( 42 sqlx::query( 43 " 44 SELECT out_tx_id, out_initiated, out_found 45 FROM register_outgoing($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) 46 ", 47 ) 48 .bind(payment.amount) 49 .bind(payment.debit_fee) 50 .bind(&payment.subject) 51 .bind_timestamp(&payment.execution_time) 52 .bind(payment.creditor.as_ref().map(|it| it.as_ref().as_str())) 53 .bind(&payment.id.e2e_id) 54 .bind(&payment.id.msg_id) 55 .bind(&payment.id.sref) 56 .bind(subject.as_ref().map(|s| &s.wtid)) 57 .bind(subject.as_ref().map(|s| s.exchange_base_url.as_str())) 58 .bind(subject.as_ref().map(|s| &s.metadata)) 59 .try_map(|r: PgRow| { 60 Ok(OutgoingRegistrationResult { 61 id: r.try_get_u64(0)?, 62 initiated: r.try_get_flag(1)?, 63 new: !r.try_get_flag(2)?, 64 }) 65 }) 66 .fetch_one(pool) 67 ) 68 } 69 70 /// Register an outgoing batch 71 pub async fn register_out_batch( 72 pool: &PgPool, 73 payment: &OutTx, 74 subject: Option<&OutgoingSubject>, 75 ) -> sqlx::Result<OutgoingRegistrationResult> { 76 serialized!( 77 sqlx::query( 78 " 79 SELECT out_tx_id, out_initiated, out_found 80 FROM register_outgoing($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) 81 ", 82 ) 83 .bind(payment.amount) 84 .bind(payment.debit_fee) 85 .bind(&payment.subject) 86 .bind_timestamp(&payment.execution_time) 87 .bind(payment.creditor.as_ref().map(|it| it.as_ref().as_str())) 88 .bind(&payment.id.e2e_id) 89 .bind(&payment.id.msg_id) 90 .bind(&payment.id.sref) 91 .bind(subject.as_ref().map(|s| &s.wtid)) 92 .bind(subject.as_ref().map(|s| s.exchange_base_url.as_str())) 93 .bind(subject.as_ref().map(|s| &s.metadata)) 94 .try_map(|r: PgRow| { 95 Ok(OutgoingRegistrationResult { 96 id: r.try_get_u64(0)?, 97 initiated: r.try_get_flag(1)?, 98 new: !r.try_get_flag(2)?, 99 }) 100 }) 101 .fetch_one(pool) 102 ) 103 } 104 105 #[derive(Debug, Clone, PartialEq, Eq)] 106 pub struct InResult { 107 pub id: u64, 108 pub new: bool, 109 pub completed: bool, 110 pub pending: bool, 111 pub bounce_id: Option<CompactString>, 112 } 113 114 /** Incoming payments registration result */ 115 #[derive(Debug, PartialEq, Eq)] 116 pub enum IncomingRegistrationResult { 117 Success(InResult), 118 ReservePubReuse, 119 MappingReuse, 120 UnknownMapping, 121 } 122 123 /** Register an incoming payment */ 124 pub async fn register_in(pool: &PgPool, payment: &InTx) -> sqlx::Result<InResult> { 125 serialized!( 126 sqlx::query( 127 " 128 SELECT out_found, out_completed, out_tx_id, out_bounce_id 129 FROM register_incoming($1,$2,$3,$4,$5,$6,$7,$8,NULL,NULL,NULL) 130 ", 131 ) 132 .bind(payment.amount) 133 .bind(payment.credit_fee) 134 .bind(&payment.subject) 135 .bind_timestamp(&payment.execution_time) 136 .bind(payment.debtor.as_ref().map(|it| it.as_ref().as_str())) 137 .bind(payment.id.uetr) 138 .bind(&payment.id.tx_id) 139 .bind(&payment.id.sref) 140 .try_map(|r: PgRow| { 141 Ok(InResult { 142 id: r.try_get_u64("out_tx_id")?, 143 new: !r.try_get_flag("out_found")?, 144 completed: r.try_get_flag("out_completed")?, 145 bounce_id: r.try_get("out_bounce_id")?, 146 pending: false, 147 }) 148 }) 149 .fetch_one(pool) 150 ) 151 } 152 153 /** Register an talerable incoming payment */ 154 pub async fn register_in_talerable( 155 pool: &PgPool, 156 payment: &InTx, 157 subject: &IncomingKey, 158 ) -> sqlx::Result<IncomingRegistrationResult> { 159 serialized!( 160 sqlx::query( 161 " 162 SELECT 163 out_reserve_pub_reuse, 164 out_mapping_reuse, 165 out_unknown_mapping, 166 out_found, 167 out_completed, 168 out_pending, 169 out_tx_id, 170 out_bounce_id 171 FROM register_incoming($1,$2,$3,$4,$5,$6,$7,$8,$9::taler_incoming_type,$10,NULL) 172 ", 173 ) 174 .bind(payment.amount) 175 .bind(payment.credit_fee) 176 .bind(&payment.subject) 177 .bind_timestamp(&payment.execution_time) 178 .bind(payment.debtor.as_ref().map(|it| it.as_ref().as_str())) 179 .bind(payment.id.uetr) 180 .bind(&payment.id.tx_id) 181 .bind(&payment.id.sref) 182 .bind(subject.ty.as_ref()) 183 .bind(subject.key) 184 .try_map(|r: PgRow| { 185 Ok(if r.try_get_flag("out_reserve_pub_reuse")? { 186 IncomingRegistrationResult::ReservePubReuse 187 } else if r.try_get_flag("out_mapping_reuse")? { 188 IncomingRegistrationResult::MappingReuse 189 } else if r.try_get_flag("out_unknown_mapping")? { 190 IncomingRegistrationResult::UnknownMapping 191 } else { 192 IncomingRegistrationResult::Success(InResult { 193 id: r.try_get_u64("out_tx_id")?, 194 new: !r.try_get_flag("out_found")?, 195 completed: r.try_get_flag("out_completed")?, 196 bounce_id: r.try_get("out_bounce_id")?, 197 pending: r.try_get("out_pending")?, 198 }) 199 }) 200 }) 201 .fetch_one(pool) 202 ) 203 } 204 205 /** Register an talerable incoming payment */ 206 pub async fn register_in_qr_bill( 207 pool: &PgPool, 208 payment: &InTx, 209 reference: &str, 210 ) -> sqlx::Result<IncomingRegistrationResult> { 211 serialized!( 212 sqlx::query( 213 " 214 SELECT 215 out_reserve_pub_reuse, 216 out_mapping_reuse, 217 out_unknown_mapping, 218 out_found, 219 out_completed, 220 out_pending, 221 out_tx_id, 222 out_bounce_id 223 FROM register_incoming($1,$2,$3,$4,$5,$6,$7,$8,NULL,NULL,$9) 224 ", 225 ) 226 .bind(payment.amount) 227 .bind(payment.credit_fee) 228 .bind(&payment.subject) 229 .bind_timestamp(&payment.execution_time) 230 .bind(payment.debtor.as_ref().map(|it| it.as_ref().as_str())) 231 .bind(payment.id.uetr) 232 .bind(&payment.id.tx_id) 233 .bind(&payment.id.sref) 234 .bind(reference) 235 .try_map(|r: PgRow| { 236 Ok(if r.try_get_flag("out_reserve_pub_reuse")? { 237 IncomingRegistrationResult::ReservePubReuse 238 } else if r.try_get_flag("out_mapping_reuse")? { 239 IncomingRegistrationResult::MappingReuse 240 } else if r.try_get_flag("out_unknown_mapping")? { 241 IncomingRegistrationResult::UnknownMapping 242 } else { 243 IncomingRegistrationResult::Success(InResult { 244 id: r.try_get_u64("out_tx_id")?, 245 new: !r.try_get_flag("out_found")?, 246 completed: r.try_get_flag("out_completed")?, 247 bounce_id: r.try_get("out_bounce_id")?, 248 pending: r.try_get("out_pending")?, 249 }) 250 }) 251 }) 252 .fetch_one(pool) 253 ) 254 } 255 256 #[derive(Debug, Clone, PartialEq, Eq)] 257 /** Incoming payments bounce registration result */ 258 pub enum IncomingBounceRegistrationResult { 259 Success(InResult), 260 Talerable, 261 } 262 263 /** Register an incoming payment and bounce it */ 264 pub async fn register_in_malformed( 265 pool: &PgPool, 266 payment: &InTx, 267 bounce_amount: &Amount, 268 bounce_end_to_end_id: &str, 269 timestamp: &Timestamp, 270 cause: &str, 271 ) -> sqlx::Result<IncomingBounceRegistrationResult> { 272 serialized!( 273 sqlx::query( 274 " 275 SELECT out_found, out_tx_id, out_completed, out_bounce_id, out_talerable 276 FROM register_and_bounce_incoming($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) 277 ", 278 ) 279 .bind(payment.amount) 280 .bind(payment.credit_fee) 281 .bind(&payment.subject) 282 .bind_timestamp(&payment.execution_time) 283 .bind(payment.debtor.as_ref().map(|it| it.as_ref().as_str())) 284 .bind(payment.id.uetr) 285 .bind(&payment.id.tx_id) 286 .bind(&payment.id.sref) 287 .bind(bounce_amount) 288 .bind_timestamp(timestamp) 289 .bind(bounce_end_to_end_id) 290 .bind(cause) 291 .try_map(|r: PgRow| { 292 Ok(if r.try_get_flag("out_talerable")? { 293 IncomingBounceRegistrationResult::Talerable 294 } else { 295 IncomingBounceRegistrationResult::Success(InResult { 296 id: r.try_get_u64("out_tx_id")?, 297 new: !r.try_get_flag("out_found")?, 298 completed: r.try_get_flag("out_completed")?, 299 bounce_id: r.try_get("out_bounce_id")?, 300 pending: false, 301 }) 302 }) 303 }) 304 .fetch_one(pool) 305 ) 306 } 307 308 #[cfg(test)] 309 mod test { 310 use jiff::Timestamp; 311 use libeufin_ebics::{ 312 ebics::rand_ebics_id, 313 iso20022::model::{InId, InTx, OutBatch, OutId, OutTx}, 314 }; 315 use sqlx::{PgPool, postgres::PgRow}; 316 use taler_api::{db::TypeHelper as _, subject::subject_fmt_qr_bill}; 317 use taler_common::{ 318 api::{EddsaPublicKey, EddsaSignature, ShortHashCode}, 319 db::IncomingType, 320 types::amount::amount, 321 }; 322 use taler_macros::db_test; 323 use taler_test_utils::routine::Status::*; 324 use uuid::Uuid; 325 326 use crate::{ 327 config::{AccountType, NexusIngestCfg}, 328 db::{ 329 initiated::{PaymentInitiationResult, batch_initiated, initiated_ack}, 330 payment::{ 331 InResult, IncomingBounceRegistrationResult, OutgoingRegistrationResult, 332 register_in_malformed, 333 }, 334 test::{check_in_count, check_in_state, check_out_count}, 335 transfer::{RegistrationResult, transfer_register}, 336 }, 337 fetch::{register_incoming, register_outgoing, register_outgoing_batch}, 338 test::{CURR, gen_in_pay, gen_initiate, gen_out_pay}, 339 }; 340 341 #[db_test] 342 async fn out_tx(db: PgPool) { 343 // Register initiated transactions 344 for subject in [ 345 "initiated by nexus".to_owned(), 346 format!("{} https://exchange.com/", ShortHashCode::rand()), 347 ] { 348 let payment = gen_out_pay(subject.clone()); 349 assert!(matches!( 350 gen_initiate(&db, payment.id.e2e_id.clone().unwrap(), subject).await, 351 PaymentInitiationResult::Success(_) 352 )); 353 let first = register_outgoing(&db, &payment).await.unwrap(); 354 assert_eq!( 355 first, 356 OutgoingRegistrationResult { 357 id: first.id, 358 initiated: true, 359 new: true 360 } 361 ); 362 assert_eq!( 363 register_outgoing(&db, &payment).await.unwrap(), 364 OutgoingRegistrationResult { 365 id: first.id, 366 initiated: true, 367 new: false 368 } 369 ); 370 let payment = OutTx { 371 id: OutId { 372 msg_id: None, 373 e2e_id: None, 374 sref: payment.id.e2e_id, 375 }, 376 ..payment 377 }; 378 let second = register_outgoing(&db, &payment).await.unwrap(); 379 assert_eq!( 380 second, 381 OutgoingRegistrationResult { 382 id: first.id + 1, 383 initiated: false, 384 new: true 385 } 386 ); 387 assert_eq!( 388 register_outgoing(&db, &payment).await.unwrap(), 389 OutgoingRegistrationResult { 390 id: second.id, 391 initiated: false, 392 new: false 393 } 394 ); 395 } 396 check_out_count(&db, 4, 1).await; 397 398 // Register unknown 399 for subject in [ 400 "initiated by nexus".to_owned(), 401 format!("{} https://exchange.com/", ShortHashCode::rand()), 402 ] { 403 let payment = gen_out_pay(subject.clone()); 404 let res = register_outgoing(&db, &payment).await.unwrap(); 405 assert_eq!( 406 res, 407 OutgoingRegistrationResult { 408 id: res.id, 409 initiated: false, 410 new: true 411 } 412 ); 413 assert_eq!( 414 register_outgoing(&db, &payment).await.unwrap(), 415 OutgoingRegistrationResult { 416 id: res.id, 417 initiated: false, 418 new: false 419 } 420 ); 421 } 422 check_out_count(&db, 6, 2).await; 423 424 // Register wtid reuse 425 let wtid = ShortHashCode::rand(); 426 for subject in [ 427 format!("{wtid} https://exchange.com/"), 428 format!("{wtid} https://exchange.com/"), 429 ] { 430 let payment = gen_out_pay(subject.clone()); 431 let res = register_outgoing(&db, &payment).await.unwrap(); 432 assert_eq!( 433 res, 434 OutgoingRegistrationResult { 435 id: res.id, 436 initiated: false, 437 new: true 438 } 439 ); 440 assert_eq!( 441 register_outgoing(&db, &payment).await.unwrap(), 442 OutgoingRegistrationResult { 443 id: res.id, 444 initiated: false, 445 new: false 446 } 447 ); 448 } 449 check_out_count(&db, 8, 3).await 450 } 451 452 #[db_test] 453 async fn out_batch(db: PgPool) { 454 // Init batch 455 let wtid = ShortHashCode::rand(); 456 for subject in [ 457 "initiated by nexus".to_string(), 458 format!("{} https://exchange.com/", ShortHashCode::rand()), 459 format!("{wtid} https://exchange.com/"), 460 format!("{wtid} https://exchange.com/"), 461 ] { 462 assert!(matches!( 463 gen_initiate(&db, rand_ebics_id(), subject).await, 464 PaymentInitiationResult::Success(_) 465 )); 466 } 467 batch_initiated(&db, &Timestamp::now(), "BATCH", false) 468 .await 469 .unwrap(); 470 471 // Register batch 472 register_outgoing_batch( 473 &db, 474 &CURR, 475 &OutBatch { 476 msg_id: "BATCH".into(), 477 execution_time: Timestamp::now(), 478 }, 479 ) 480 .await 481 .unwrap(); 482 check_out_count(&db, 4, 2).await; 483 484 // Test manual ack 485 let mut txs = Vec::new(); 486 for nb in 0..3 { 487 let res = gen_initiate(&db, rand_ebics_id(), format!("tx {nb}")).await; 488 if let PaymentInitiationResult::Success(id) = &res { 489 txs.push(*id); 490 } else { 491 panic!("Expected success got {res:?}"); 492 } 493 } 494 495 // Check not sent without ack 496 batch_initiated(&db, &Timestamp::now(), "BATCH_MANUAL", true) 497 .await 498 .unwrap(); 499 register_outgoing_batch( 500 &db, 501 &CURR, 502 &OutBatch { 503 msg_id: "BATCH_MANUAL".into(), 504 execution_time: Timestamp::now(), 505 }, 506 ) 507 .await 508 .unwrap(); 509 check_out_count(&db, 4, 2).await; 510 511 // Check sent with ack 512 for tx in txs { 513 initiated_ack(&db, tx).await.unwrap(); 514 } 515 batch_initiated(&db, &Timestamp::now(), "BATCH_MANUAL", true) 516 .await 517 .unwrap(); 518 register_outgoing_batch( 519 &db, 520 &CURR, 521 &OutBatch { 522 msg_id: "BATCH_MANUAL".into(), 523 execution_time: Timestamp::now(), 524 }, 525 ) 526 .await 527 .unwrap(); 528 check_out_count(&db, 7, 2).await; 529 } 530 531 #[db_test] 532 async fn in_bounce(db: PgPool) { 533 // Creating and bouncing one incoming transaction 534 let payment = gen_in_pay("incoming and bounce"); 535 let id = rand_ebics_id(); 536 537 let bounce_amount = amount("KUDOS:2.53"); 538 let res = register_in_malformed( 539 &db, 540 &payment, 541 &bounce_amount, 542 &id, 543 &Timestamp::now(), 544 "manual bounce", 545 ) 546 .await 547 .unwrap(); 548 assert!( 549 matches!( 550 res, 551 IncomingBounceRegistrationResult::Success(InResult { 552 new: true, 553 id: _, 554 completed: false, 555 pending: false, 556 ref bounce_id 557 }) if bounce_id.as_ref() == Some(&id)), 558 "{res:?}" 559 ); 560 // Idempotent 561 let res = register_in_malformed( 562 &db, 563 &payment, 564 &amount("KUDOS:2.5"), 565 &rand_ebics_id(), 566 &Timestamp::now(), 567 "other reason to bounce", 568 ) 569 .await 570 .unwrap(); 571 assert!( 572 matches!( 573 res, 574 IncomingBounceRegistrationResult::Success(InResult { 575 new: false, 576 id: _, 577 completed: false, 578 pending: false, 579 ref bounce_id 580 }) if bounce_id.as_ref() == Some(&id), 581 ), 582 "{res:?}" 583 ); 584 585 // Checking one incoming got created and bounced 586 sqlx::query( 587 " 588 SELECT 589 incoming_transactions.amount as in_amount, 590 initiated_outgoing_transactions.amount as bounce_amount 591 FROM incoming_transactions 592 JOIN bounced_transactions USING (incoming_transaction_id) 593 JOIN initiated_outgoing_transactions USING (initiated_outgoing_transaction_id) 594 ", 595 ) 596 .try_map(|r: PgRow| { 597 assert_eq!(r.try_get_amount("in_amount", &CURR)?, payment.amount); 598 assert_eq!(r.try_get_amount("bounce_amount", &CURR)?, bounce_amount); 599 Ok(()) 600 }) 601 .fetch_one(&db) 602 .await 603 .unwrap(); 604 } 605 606 #[db_test] 607 async fn in_simple(db: PgPool) { 608 let cfg = NexusIngestCfg::simple(AccountType::exchange, &CURR); 609 610 // Register 611 let incoming = gen_in_pay("test"); 612 register_incoming(&db, &cfg, &incoming).await.unwrap(); 613 check_in_state(&db, &[Bounced]).await; 614 615 // Idempotent 616 register_incoming(&db, &cfg, &incoming).await.unwrap(); 617 check_in_state(&db, &[Bounced]).await; 618 619 // Many 620 register_incoming(&db, &cfg, &gen_in_pay("another subject")) 621 .await 622 .unwrap(); 623 check_in_state(&db, &[Bounced, Bounced]).await; 624 625 // Admin balance adjust is ignored 626 register_incoming(&db, &cfg, &gen_in_pay("ADMIN BALANCE ADJUST")) 627 .await 628 .unwrap(); 629 630 check_in_state(&db, &[Bounced, Bounced, Simple]).await; 631 632 let original = gen_in_pay("test 2"); 633 let incomplete = InTx { 634 subject: None, 635 debtor: None, 636 ..original.clone() 637 }; 638 639 // Register incomplete transaction 640 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 641 check_in_state(&db, &[Bounced, Bounced, Simple, Incomplete]).await; 642 // Idempotent 643 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 644 check_in_state(&db, &[Bounced, Bounced, Simple, Incomplete]).await; 645 // Recover info when completed 646 register_incoming(&db, &cfg, &original).await.unwrap(); 647 check_in_state(&db, &[Bounced, Bounced, Simple, Bounced]).await; 648 } 649 650 #[db_test] 651 async fn in_talerable(db: PgPool) { 652 let cfg = NexusIngestCfg::simple(AccountType::exchange, &CURR); 653 let key = EddsaPublicKey::rand(); 654 let subject = format!("test with {key} reserve pub"); 655 656 // Register 657 let incoming = gen_in_pay(&subject); 658 register_incoming(&db, &cfg, &incoming).await.unwrap(); 659 check_in_state(&db, &[Reserve(key)]).await; 660 661 // Idempotent 662 register_incoming(&db, &cfg, &incoming).await.unwrap(); 663 check_in_state(&db, &[Reserve(key)]).await; 664 665 // Key reuse is bounced 666 register_incoming(&db, &cfg, &gen_in_pay(&subject)) 667 .await 668 .unwrap(); 669 register_incoming(&db, &cfg, &gen_in_pay(format_args!("another {subject}"))) 670 .await 671 .unwrap(); 672 check_in_state(&db, &[Reserve(key), Bounced, Bounced]).await; 673 674 // Admin balance adjust is ignored 675 register_incoming(&db, &cfg, &gen_in_pay("ADMIN BALANCE ADJUST")) 676 .await 677 .unwrap(); 678 check_in_state(&db, &[Reserve(key), Bounced, Bounced, Simple]).await; 679 680 let new = EddsaPublicKey::rand(); 681 let original = gen_in_pay(format_args!("test 2 with {new} reserve pub")); 682 let incomplete = InTx { 683 subject: None, 684 debtor: None, 685 ..original.clone() 686 }; 687 688 // Register incomplete transaction 689 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 690 check_in_state(&db, &[Reserve(key), Bounced, Bounced, Simple, Incomplete]).await; 691 // Idempotent 692 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 693 check_in_state(&db, &[Reserve(key), Bounced, Bounced, Simple, Incomplete]).await; 694 // Recover info when completed 695 register_incoming(&db, &cfg, &original).await.unwrap(); 696 check_in_state(&db, &[Reserve(key), Bounced, Bounced, Simple, Reserve(new)]).await; 697 } 698 699 #[db_test] 700 async fn in_mapping(db: PgPool) { 701 let cfg = NexusIngestCfg::simple(AccountType::exchange, &CURR); 702 let first = EddsaPublicKey::rand(); 703 let auth_pub = EddsaPublicKey::rand(); 704 let auth_sig = EddsaSignature::rand(); 705 let reference_number = subject_fmt_qr_bill(auth_pub.as_slice()); 706 let subject = format!("test with MAP:{auth_pub} auth pub"); 707 708 assert_eq!( 709 transfer_register( 710 &db, 711 IncomingType::reserve, 712 &first, 713 &auth_pub, 714 &auth_sig, 715 false, 716 None, 717 &Timestamp::now() 718 ) 719 .await 720 .unwrap(), 721 RegistrationResult::Success 722 ); 723 724 // Register 725 let incoming = gen_in_pay(&subject); 726 register_incoming(&db, &cfg, &incoming).await.unwrap(); 727 check_in_state(&db, &[Reserve(first)]).await; 728 729 // Idempotent 730 register_incoming(&db, &cfg, &incoming).await.unwrap(); 731 check_in_state(&db, &[Reserve(first)]).await; 732 733 // Admin balance adjust is ignored 734 register_incoming(&db, &cfg, &gen_in_pay("ADMIN BALANCE ADJUST")) 735 .await 736 .unwrap(); 737 check_in_state(&db, &[Reserve(first), Simple]).await; 738 739 let original = gen_in_pay(format_args!("test 2 for {subject}")); 740 let incomplete = InTx { 741 subject: None, 742 debtor: None, 743 ..original.clone() 744 }; 745 // Register incomplete transaction 746 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 747 check_in_state(&db, &[Reserve(first), Simple, Incomplete]).await; 748 // Idempotent 749 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 750 check_in_state(&db, &[Reserve(first), Simple, Incomplete]).await; 751 // Recover info when completed 752 register_incoming(&db, &cfg, &original).await.unwrap(); 753 check_in_state(&db, &[Reserve(first), Simple, Bounced]).await; 754 755 let second = EddsaPublicKey::rand(); 756 assert_eq!( 757 transfer_register( 758 &db, 759 IncomingType::reserve, 760 &second, 761 &auth_pub, 762 &auth_sig, 763 true, 764 Some(&reference_number), 765 &Timestamp::now() 766 ) 767 .await 768 .unwrap(), 769 RegistrationResult::Success 770 ); 771 check_in_state(&db, &[Reserve(first), Simple, Bounced]).await; 772 773 // Key reuse is pending 774 for _ in 0..3 { 775 register_incoming(&db, &cfg, &gen_in_pay(&subject)) 776 .await 777 .unwrap(); 778 } 779 check_in_state( 780 &db, 781 &[ 782 Reserve(first), 783 Simple, 784 Bounced, 785 Reserve(second), 786 Pending, 787 Pending, 788 ], 789 ) 790 .await; 791 792 // Finish pending 793 let third = EddsaPublicKey::rand(); 794 assert_eq!( 795 transfer_register( 796 &db, 797 IncomingType::reserve, 798 &third, 799 &auth_pub, 800 &auth_sig, 801 true, 802 Some(&reference_number), 803 &Timestamp::now() 804 ) 805 .await 806 .unwrap(), 807 RegistrationResult::Success 808 ); 809 check_in_state( 810 &db, 811 &[ 812 Reserve(first), 813 Simple, 814 Bounced, 815 Reserve(second), 816 Reserve(third), 817 Pending, 818 ], 819 ) 820 .await; 821 } 822 823 #[db_test] 824 async fn in_reference(db: PgPool) { 825 let cfg = NexusIngestCfg::simple(AccountType::exchange, &CURR); 826 let first = EddsaPublicKey::rand(); 827 let auth_pub = EddsaPublicKey::rand(); 828 let auth_sig = EddsaSignature::rand(); 829 let reference_number = subject_fmt_qr_bill(auth_pub.as_slice()); 830 831 assert_eq!( 832 transfer_register( 833 &db, 834 IncomingType::reserve, 835 &first, 836 &auth_pub, 837 &auth_sig, 838 false, 839 Some(&reference_number), 840 &Timestamp::now() 841 ) 842 .await 843 .unwrap(), 844 RegistrationResult::Success 845 ); 846 847 // Register 848 let incoming = gen_in_pay(reference_number.clone()); 849 register_incoming(&db, &cfg, &incoming).await.unwrap(); 850 check_in_state(&db, &[Reserve(first)]).await; 851 852 // Idempotent 853 register_incoming(&db, &cfg, &incoming).await.unwrap(); 854 check_in_state(&db, &[Reserve(first)]).await; 855 856 // Admin balance adjust is ignored 857 register_incoming(&db, &cfg, &gen_in_pay("ADMIN BALANCE ADJUST")) 858 .await 859 .unwrap(); 860 check_in_state(&db, &[Reserve(first), Simple]).await; 861 862 let original = gen_in_pay(reference_number.clone()); 863 let incomplete = InTx { 864 subject: None, 865 debtor: None, 866 ..original.clone() 867 }; 868 // Register incomplete transaction 869 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 870 check_in_state(&db, &[Reserve(first), Simple, Incomplete]).await; 871 // Idempotent 872 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 873 check_in_state(&db, &[Reserve(first), Simple, Incomplete]).await; 874 // Recover info when completed 875 register_incoming(&db, &cfg, &original).await.unwrap(); 876 check_in_state(&db, &[Reserve(first), Simple, Bounced]).await; 877 878 let second = EddsaPublicKey::rand(); 879 assert_eq!( 880 transfer_register( 881 &db, 882 IncomingType::reserve, 883 &second, 884 &auth_pub, 885 &auth_sig, 886 true, 887 Some(&reference_number), 888 &Timestamp::now() 889 ) 890 .await 891 .unwrap(), 892 RegistrationResult::Success 893 ); 894 check_in_state(&db, &[Reserve(first), Simple, Bounced]).await; 895 896 // Key reuse is pending 897 for _ in 0..3 { 898 register_incoming(&db, &cfg, &gen_in_pay(&reference_number)) 899 .await 900 .unwrap(); 901 } 902 check_in_state( 903 &db, 904 &[ 905 Reserve(first), 906 Simple, 907 Bounced, 908 Reserve(second), 909 Pending, 910 Pending, 911 ], 912 ) 913 .await; 914 915 // Finish pending 916 let third = EddsaPublicKey::rand(); 917 assert_eq!( 918 transfer_register( 919 &db, 920 IncomingType::reserve, 921 &third, 922 &auth_pub, 923 &auth_sig, 924 true, 925 Some(&reference_number), 926 &Timestamp::now() 927 ) 928 .await 929 .unwrap(), 930 RegistrationResult::Success 931 ); 932 check_in_state( 933 &db, 934 &[ 935 Reserve(first), 936 Simple, 937 Bounced, 938 Reserve(second), 939 Reserve(third), 940 Pending, 941 ], 942 ) 943 .await; 944 } 945 946 #[db_test] 947 async fn in_recover_info(db: PgPool) { 948 let cfg = NexusIngestCfg::simple(AccountType::exchange, &CURR); 949 950 async fn check_content(db: &PgPool, p: &InTx) { 951 sqlx::query( 952 " 953 SELECT 954 uetr IS NOT DISTINCT FROM $1 AND 955 tx_id IS NOT DISTINCT FROM $2 AND 956 acct_svcr_ref IS NOT DISTINCT FROM $3 AND 957 subject IS NOT DISTINCT FROM $4 AND 958 debit_payto IS NOT DISTINCT FROM $5 959 FROM incoming_transactions ORDER BY incoming_transaction_id DESC LIMIT 1 960 ", 961 ) 962 .bind(p.id.uetr) 963 .bind(&p.id.tx_id) 964 .bind(&p.id.sref) 965 .bind(&p.subject) 966 .bind(p.debtor.as_ref().map(|it| it.as_ref().as_str())) 967 .try_map(|r: PgRow| { 968 assert!(r.try_get_flag(0)?); 969 Ok(()) 970 }) 971 .fetch_one(db) 972 .await 973 .unwrap(); 974 } 975 976 // Non talerable 977 for (i, id) in [ 978 InId::new(Some(Uuid::new_v4()), None, None), 979 InId::new(None, Some(rand_ebics_id()), None), 980 InId::new(None, None, Some(rand_ebics_id())), 981 ] 982 .iter() 983 .enumerate() 984 { 985 let payment = gen_in_pay("subject"); 986 987 // Register minimal 988 let partial = InTx { 989 id: id.clone(), 990 subject: None, 991 debtor: None, 992 ..payment.clone() 993 }; 994 register_incoming(&db, &cfg, &partial).await.unwrap(); 995 check_content(&db, &partial).await; 996 check_in_count(&db, i + 1, i, 0).await; 997 998 // Recover ID 999 let full_id = InId::new( 1000 Some(id.uetr.unwrap_or_else(Uuid::new_v4)), 1001 Some(id.tx_id.clone().unwrap_or_else(rand_ebics_id)), 1002 Some(id.sref.clone().unwrap_or_else(rand_ebics_id)), 1003 ); 1004 let full = InTx { 1005 id: full_id.clone(), 1006 ..partial.clone() 1007 }; 1008 register_incoming(&db, &cfg, &full).await.unwrap(); 1009 check_content(&db, &full).await; 1010 check_in_count(&db, i + 1, i, 0).await; 1011 1012 // Recover subject & debtor 1013 let full = InTx { 1014 id: full_id, 1015 ..payment.clone() 1016 }; 1017 register_incoming(&db, &cfg, &full).await.unwrap(); 1018 check_content(&db, &full).await; 1019 check_in_count(&db, i + 1, i + 1, 0).await; 1020 } 1021 1022 // Talerable 1023 for (i, id) in [ 1024 InId::new(Some(Uuid::new_v4()), None, None), 1025 InId::new(None, Some(rand_ebics_id()), None), 1026 InId::new(None, None, Some(rand_ebics_id())), 1027 ] 1028 .iter() 1029 .enumerate() 1030 { 1031 let key = EddsaPublicKey::rand(); 1032 let payment = gen_in_pay(format_args!("test with {key} reserve pub")); 1033 1034 // Register minimal 1035 let partial = InTx { 1036 id: id.clone(), 1037 subject: None, 1038 debtor: None, 1039 ..payment.clone() 1040 }; 1041 register_incoming(&db, &cfg, &partial).await.unwrap(); 1042 check_content(&db, &partial).await; 1043 check_in_count(&db, i + 4, 3, i).await; 1044 1045 // Recover ID 1046 let full_id = InId::new( 1047 Some(id.uetr.unwrap_or_else(Uuid::new_v4)), 1048 Some(id.tx_id.clone().unwrap_or_else(rand_ebics_id)), 1049 Some(id.sref.clone().unwrap_or_else(rand_ebics_id)), 1050 ); 1051 let full = InTx { 1052 id: full_id.clone(), 1053 ..partial.clone() 1054 }; 1055 register_incoming(&db, &cfg, &full).await.unwrap(); 1056 check_content(&db, &full).await; 1057 check_in_count(&db, i + 4, 3, i).await; 1058 1059 // Recover subject & debtor 1060 let full = InTx { 1061 id: full_id, 1062 ..payment.clone() 1063 }; 1064 register_incoming(&db, &cfg, &full).await.unwrap(); 1065 check_content(&db, &full).await; 1066 check_in_count(&db, i + 4, 3, i + 1).await; 1067 } 1068 } 1069 1070 #[db_test] 1071 async fn in_horror(db: PgPool) { 1072 let cfg = NexusIngestCfg::simple(AccountType::exchange, &CURR); 1073 1074 // Check we do not bounce already registered talerable transaction 1075 let key = EddsaPublicKey::rand(); 1076 let payment = gen_in_pay(format_args!("test with {key} reserve pub")); 1077 register_incoming(&db, &cfg, &payment).await.unwrap(); 1078 assert_eq!( 1079 register_in_malformed( 1080 &db, 1081 &payment, 1082 &amount("KUDOS:2.53"), 1083 &rand_ebics_id(), 1084 &Timestamp::now(), 1085 "manual bounce", 1086 ) 1087 .await 1088 .unwrap(), 1089 IncomingBounceRegistrationResult::Talerable 1090 ); 1091 let incomplete = InTx { 1092 subject: None, 1093 ..payment.clone() 1094 }; 1095 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 1096 register_incoming(&db, &cfg, &payment).await.unwrap(); 1097 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 1098 check_in_state(&db, &[Reserve(key)]).await; 1099 1100 // Check we do not register as talerable bounced transaction 1101 let new_key = EddsaPublicKey::rand(); 1102 let payment = gen_in_pay(format_args!("bounced {new_key}")); 1103 let incomplete = InTx { 1104 subject: None, 1105 ..payment.clone() 1106 }; 1107 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 1108 register_incoming(&db, &cfg, &payment).await.unwrap(); 1109 register_incoming(&db, &cfg, &incomplete).await.unwrap(); 1110 register_incoming(&db, &cfg, &payment).await.unwrap(); 1111 check_in_state(&db, &[Reserve(key), Bounced]).await; 1112 } 1113 }