exchange.rs (13564B)
1 /* 2 * This file is part of LibEuFin. 3 * Copyright (C) 2026 Taler Systems S.A. 4 5 * LibEuFin is free software; you can redistribute it and/or modify 6 * it under the terms of the GNU Affero General Public License as 7 * published by the Free Software Foundation; either version 3, or 8 * (at your option) any later version. 9 10 * LibEuFin is distributed in the hope that it will be useful, but 11 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY 12 * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General 13 * Public License for more details. 14 15 * You should have received a copy of the GNU Affero General Public 16 * License along with LibEuFin; see the file COPYING. If not, see 17 * <http://www.gnu.org/licenses/> 18 */ 19 20 //! Data access logic for exchange specific logic 21 22 use jiff::Timestamp; 23 use sqlx::{PgPool, QueryBuilder, Row, postgres::PgRow}; 24 use taler_api::{ 25 db::{BindHelper, TypeHelper, history, page}, 26 notification::NotificationChannel, 27 serialized, 28 subject::{IncomingKey, fmt_out_subject}, 29 }; 30 use taler_common::{ 31 api::{ 32 params::{History, Page}, 33 wire::{ 34 IncomingBankTransaction, OutgoingBankTransaction, TransferListStatus, TransferRequest, 35 TransferState, TransferStatus, 36 }, 37 }, 38 db::IncomingType, 39 types::amount::{Amount, Currency}, 40 }; 41 42 use crate::payto::{BankPayto, PaytoCtx, sql_bank_payto, sql_bank_simple_payto}; 43 44 /** Result of taler transfer transaction creation */ 45 pub enum TransferResult { 46 /** Transaction [id] and wire transfer [timestamp] */ 47 Success { 48 id: u64, 49 timestamp: Timestamp, 50 }, 51 NotAnExchange, 52 UnknownExchange, 53 BothPartyAreExchange, 54 BalanceInsufficient, 55 ReserveUidReuse, 56 WtidReuse, 57 AdminCreditor, 58 } 59 60 /** Perform a Taler transfer */ 61 pub async fn transfer( 62 db: &PgPool, 63 username: &str, 64 req: &TransferRequest, 65 payto: &BankPayto, 66 timestamp: &Timestamp, 67 conversion: bool, 68 ) -> sqlx::Result<TransferResult> { 69 let subject = fmt_out_subject(&req.wtid, &req.exchange_base_url, req.metadata.as_deref()); 70 serialized!( 71 sqlx::query( 72 " 73 SELECT 74 out_debtor_not_found, 75 out_debtor_not_exchange, 76 out_both_exchanges, 77 out_request_uid_reuse, 78 out_wtid_reuse, 79 out_exchange_balance_insufficient, 80 out_creditor_admin, 81 out_tx_row_id, 82 out_timestamp 83 FROM taler_transfer($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) 84 ", 85 ) 86 .bind(req.request_uid) 87 .bind(req.wtid) 88 .bind(&subject) 89 .bind(req.amount) 90 .bind(req.exchange_base_url.as_str()) 91 .bind(&req.metadata) 92 .bind(payto.canonical()) 93 .bind(username) 94 .bind_timestamp(timestamp) 95 .bind(conversion) 96 .try_map(|r: PgRow| { 97 Ok(if r.try_get_flag("out_debtor_not_found")? { 98 TransferResult::UnknownExchange 99 } else if r.try_get_flag("out_debtor_not_exchange")? { 100 TransferResult::NotAnExchange 101 } else if r.try_get_flag("out_both_exchanges")? { 102 TransferResult::BothPartyAreExchange 103 } else if r.try_get_flag("out_exchange_balance_insufficient")? { 104 TransferResult::BalanceInsufficient 105 } else if r.try_get_flag("out_request_uid_reuse")? { 106 TransferResult::ReserveUidReuse 107 } else if r.try_get_flag("out_wtid_reuse")? { 108 TransferResult::WtidReuse 109 } else if r.try_get_flag("out_creditor_admin")? { 110 TransferResult::AdminCreditor 111 } else { 112 TransferResult::Success { 113 id: r.try_get_u64("out_tx_row_id")?, 114 timestamp: r.try_get_timestamp("out_timestamp")?, 115 } 116 }) 117 }) 118 .fetch_one(db) 119 ) 120 } 121 122 /** Get status of transfer [txId] of account [exchangeId] */ 123 pub async fn transfer_by_id( 124 db: &PgPool, 125 ctx: &PaytoCtx, 126 currency: &Currency, 127 exchange_id: u64, 128 tx_id: u64, 129 ) -> sqlx::Result<Option<TransferStatus>> { 130 serialized!( 131 sqlx::query( 132 " 133 SELECT 134 wtid, 135 exchange_base_url, 136 metadata, 137 transfer_date, 138 amount, 139 creditor_payto, 140 status, 141 status_msg 142 FROM transfer_operations 143 WHERE transfer_operation_id=$1 AND exchange_id=$2 144 ", 145 ) 146 .bind(tx_id as i64) 147 .bind(exchange_id as i64) 148 .try_map(|r: PgRow| { 149 Ok(TransferStatus { 150 status: r.try_get("status")?, 151 status_msg: r.try_get("status_msg")?, 152 amount: r.try_get_amount("amount", currency)?, 153 exchange_base_url: r.try_get("exchange_base_url")?, 154 metadata: r.try_get("metadata")?, 155 wtid: r.try_get("wtid")?, 156 credit_account: sql_bank_simple_payto(&r, ctx, "creditor_payto")?.as_uri(), 157 timestamp: r.try_get_timestamp("transfer_date")?.into(), 158 }) 159 }) 160 .fetch_optional(db) 161 ) 162 } 163 164 /** Get a page of transfers status of account [exchangeId] */ 165 pub async fn page_transfer( 166 db: &PgPool, 167 ctx: &PaytoCtx, 168 currency: &Currency, 169 params: &Page, 170 exchange_id: u64, 171 status: Option<TransferState>, 172 ) -> sqlx::Result<Vec<TransferListStatus>> { 173 if matches!( 174 status, 175 Some(TransferState::pending | TransferState::transient_failure) 176 ) { 177 return Ok(Vec::new()); 178 } 179 page( 180 db, 181 params, 182 "transfer_operation_id", 183 || { 184 let mut query = QueryBuilder::new( 185 " 186 SELECT 187 transfer_operation_id 188 ,transfer_date 189 ,amount 190 ,creditor_payto 191 ,status 192 FROM transfer_operations 193 WHERE exchange_id=", 194 ); 195 query.push_bind(exchange_id as i64).push(" AND "); 196 if let Some(status) = status { 197 query.push("status=").push_bind(status).push(" AND "); 198 } 199 query 200 }, 201 |r| { 202 Ok(TransferListStatus { 203 row_id: r.try_get_u64("transfer_operation_id")?, 204 status: r.try_get("status")?, 205 amount: r.try_get_amount("amount", currency)?, 206 credit_account: sql_bank_simple_payto(&r, ctx, "creditor_payto")?.as_uri(), 207 timestamp: r.try_get_timestamp("transfer_date")?.into(), 208 }) 209 }, 210 ) 211 .await 212 } 213 214 /** Result of taler add incoming transaction creation */ 215 pub enum AddIncomingResult { 216 /** Transaction [id] and wire transfer [timestamp] */ 217 Success { 218 id: u64, 219 timestamp: Timestamp, 220 pending: bool, 221 }, 222 NotAnExchange, 223 UnknownExchange, 224 UnknownDebtor, 225 BothPartyAreExchange, 226 ReservePubReuse, 227 UnknownMapping, 228 MappingReuse, 229 BalanceInsufficient, 230 } 231 232 /** Add a new taler incoming transaction */ 233 pub async fn add_incoming( 234 db: &PgPool, 235 amount: &Amount, 236 debtor: &BankPayto, 237 subject: &str, 238 username: &str, 239 timestamp: &Timestamp, 240 metadata: &IncomingKey, 241 ) -> sqlx::Result<AddIncomingResult> { 242 serialized!( 243 sqlx::query( 244 " 245 SELECT 246 out_creditor_not_found 247 ,out_creditor_not_exchange 248 ,out_debtor_not_found 249 ,out_both_exchanges 250 ,out_reserve_pub_reuse 251 ,out_mapping_reuse 252 ,out_unknown_mapping 253 ,out_debitor_balance_insufficient 254 ,out_tx_row_id 255 ,out_pending 256 FROM taler_add_incoming($1,$2,$3,$4,$5,$6,$7::taler_incoming_type) 257 ", 258 ) 259 .bind(metadata.key) 260 .bind(subject) 261 .bind(amount) 262 .bind(debtor.canonical()) 263 .bind(username) 264 .bind_timestamp(timestamp) 265 .bind(metadata.ty.as_ref()) 266 .try_map(|r: PgRow| { 267 Ok(if r.try_get_flag("out_creditor_not_found")? { 268 AddIncomingResult::UnknownExchange 269 } else if r.try_get_flag("out_creditor_not_exchange")? { 270 AddIncomingResult::NotAnExchange 271 } else if r.try_get_flag("out_debtor_not_found")? { 272 AddIncomingResult::UnknownDebtor 273 } else if r.try_get_flag("out_both_exchanges")? { 274 AddIncomingResult::BothPartyAreExchange 275 } else if r.try_get_flag("out_debitor_balance_insufficient")? { 276 AddIncomingResult::BalanceInsufficient 277 } else if r.try_get_flag("out_reserve_pub_reuse")? { 278 AddIncomingResult::ReservePubReuse 279 } else if r.try_get_flag("out_mapping_reuse")? { 280 AddIncomingResult::MappingReuse 281 } else if r.try_get_flag("out_unknown_mapping")? { 282 AddIncomingResult::UnknownMapping 283 } else { 284 AddIncomingResult::Success { 285 id: r.try_get_u64("out_tx_row_id")?, 286 timestamp: *timestamp, 287 pending: r.try_get_flag("out_pending")?, 288 } 289 }) 290 }) 291 .fetch_one(db) 292 ) 293 } 294 295 /** Query [exchangeId] history of taler incoming transactions */ 296 pub async fn incoming_history( 297 db: &PgPool, 298 ctx: &PaytoCtx, 299 channel: &NotificationChannel<u64, i64>, 300 currency: &Currency, 301 params: &History, 302 exchange_id: u64, 303 ) -> sqlx::Result<Vec<IncomingBankTransaction>> { 304 history( 305 db, 306 "exchange_incoming_id", 307 params, 308 || channel.subscribe(exchange_id), 309 || { 310 let mut query = QueryBuilder::new( 311 "SELECT 312 exchange_incoming_id 313 ,transaction_date 314 ,amount 315 ,debtor_payto 316 ,debtor_name 317 ,type::text 318 ,metadata 319 ,authorization_pub 320 ,authorization_sig 321 FROM taler_exchange_incoming AS tfr 322 JOIN bank_account_transactions AS txs 323 ON bank_transaction=txs.bank_transaction_id 324 WHERE bank_account_id=", 325 ); 326 query.push_bind(exchange_id as i64).push(" AND "); 327 query 328 }, 329 |r| { 330 Ok(match r.try_get_parse("type")? { 331 IncomingType::reserve => IncomingBankTransaction::Reserve { 332 row_id: r.try_get_u64("exchange_incoming_id")?, 333 date: r.try_get_timestamp("transaction_date")?.into(), 334 amount: r.try_get_amount("amount", currency)?, 335 credit_fee: None, 336 debit_account: sql_bank_payto(&r, ctx, "debtor_payto", "debtor_name")?.as_uri(), 337 reserve_pub: r.try_get("metadata")?, 338 authorization_pub: r.try_get("authorization_pub")?, 339 authorization_sig: r.try_get("authorization_sig")?, 340 }, 341 IncomingType::kyc => IncomingBankTransaction::Kyc { 342 row_id: r.try_get_u64("exchange_incoming_id")?, 343 date: r.try_get_timestamp("transaction_date")?.into(), 344 amount: r.try_get_amount("amount", currency)?, 345 credit_fee: None, 346 debit_account: sql_bank_payto(&r, ctx, "debtor_payto", "debtor_name")?.as_uri(), 347 account_pub: r.try_get("metadata")?, 348 authorization_pub: r.try_get("authorization_pub")?, 349 authorization_sig: r.try_get("authorization_sig")?, 350 }, 351 IncomingType::map => unreachable!(), 352 }) 353 }, 354 ) 355 .await 356 } 357 358 /** Query [exchangeId] history of taler incoming transactions */ 359 pub async fn outgoing_history( 360 db: &PgPool, 361 ctx: &PaytoCtx, 362 channel: &NotificationChannel<u64, i64>, 363 currency: &Currency, 364 params: &History, 365 exchange_id: u64, 366 ) -> sqlx::Result<Vec<OutgoingBankTransaction>> { 367 history( 368 db, 369 "bank_transaction_id", 370 params, 371 || channel.subscribe(exchange_id), 372 || { 373 let mut query = QueryBuilder::new( 374 "SELECT 375 bank_transaction_id 376 ,transaction_date 377 ,txs.amount 378 ,txs.creditor_payto 379 ,txs.creditor_name 380 ,wtid 381 ,exchange_base_url 382 ,transfer_operations.metadata 383 FROM taler_exchange_outgoing AS tfr 384 JOIN transfer_operations USING (exchange_outgoing_id) 385 JOIN bank_account_transactions AS txs 386 ON bank_transaction=txs.bank_transaction_id 387 WHERE bank_account_id=", 388 ); 389 query.push_bind(exchange_id as i64).push(" AND "); 390 query 391 }, 392 |r| { 393 Ok(OutgoingBankTransaction { 394 row_id: r.try_get_u64("bank_transaction_id")?, 395 date: r.try_get_timestamp("transaction_date")?.into(), 396 amount: r.try_get_amount("amount", currency)?, 397 debit_fee: None, 398 credit_account: sql_bank_payto(&r, ctx, "creditor_payto", "creditor_name")? 399 .as_uri(), 400 wtid: r.try_get("wtid")?, 401 metadata: r.try_get("metadata")?, 402 exchange_base_url: r.try_get_parse("exchange_base_url")?, 403 }) 404 }, 405 ) 406 .await 407 }