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