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