api.rs (14530B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2026 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU Affero General Public License as published by the Free Software 7 Foundation; either version 3, or (at your option) any later version. 8 9 TALER is distributed in the hope that it will be useful, but WITHOUT ANY 10 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 11 A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. 12 13 You should have received a copy of the GNU Affero General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 17 use std::{collections::BTreeMap, sync::Arc}; 18 19 use compact_str::format_compact; 20 use jiff::Timestamp; 21 use taler_api::{ 22 api::{ 23 TalerApi, TalerRouter as _, prepared::PreparedTransfer, revenue::Revenue, wire::WireGateway, 24 }, 25 config::ApiCfg, 26 error::{ApiResult, failure_code, not_implemented}, 27 subject::{IncomingSubject, fmt_in_subject}, 28 }; 29 use taler_common::{ 30 api::{ 31 params::{History, Page}, 32 prepared::{ 33 RegistrationRequest, RegistrationResponse, SubjectFormat, TransferSubject, 34 Unregistration, 35 }, 36 revenue::RevenueIncomingHistory, 37 wire::{ 38 AddIncomingRequest, AddIncomingResponse, AddKycauthRequest, AddMappedRequest, 39 IncomingHistory, OutgoingHistory, TransferList, TransferRequest, TransferResponse, 40 TransferState, TransferStatus, 41 }, 42 }, 43 db::IncomingType, 44 error_code::ErrorCode, 45 types::{amount::Currency, payto::PaytoImpl, timestamp::TalerTimestamp}, 46 }; 47 use taler_test_utils::Router; 48 use tokio::sync::watch::Sender; 49 50 use crate::{ 51 config::WiseBalance, 52 db::{self, AddIncomingResult, TxIn}, 53 payto::FullWisePayto, 54 }; 55 56 pub async fn start( 57 pool: sqlx::PgPool, 58 name: &str, 59 balances: Vec<WiseBalance>, 60 wire_gateway: &Option<ApiCfg>, 61 revenue: &Option<ApiCfg>, 62 ) -> Router { 63 let apis: Vec<_> = balances 64 .into_iter() 65 .map(|b| WiseBalanceApi { 66 id: b.id, 67 currency: b.currency, 68 pool: pool.clone(), 69 payto: b.payto.full(name), 70 in_channel: Sender::new(0), 71 taler_in_channel: Sender::new(0), 72 }) 73 .collect(); 74 let in_channels = Arc::new(BTreeMap::from_iter( 75 apis.iter().map(|b| (b.id, b.in_channel.clone())), 76 )); 77 let taler_in_channels = Arc::new(BTreeMap::from_iter( 78 apis.iter().map(|b| (b.id, b.taler_in_channel.clone())), 79 )); 80 tokio::spawn(db::notification_listener( 81 pool, 82 in_channels, 83 taler_in_channels, 84 )); 85 let mut main_router = Router::new(); 86 for api in apis { 87 let api = Arc::new(api); 88 let mut balance_router = Router::new(); 89 if let Some(cfg) = wire_gateway { 90 balance_router = balance_router 91 .wire_gateway(api.clone(), cfg.auth.method()) 92 .prepared_transfer(api.clone()); 93 } 94 if let Some(cfg) = revenue { 95 balance_router = balance_router.revenue(api.clone(), cfg.auth.method()); 96 } 97 main_router = main_router.nest(&format!("/balances/{}", api.id), balance_router); 98 } 99 main_router 100 } 101 102 pub struct WiseBalanceApi { 103 pub id: u32, 104 pub currency: Currency, 105 pub pool: sqlx::PgPool, 106 pub payto: FullWisePayto, 107 pub in_channel: Sender<i64>, 108 pub taler_in_channel: Sender<i64>, 109 } 110 111 impl TalerApi for WiseBalanceApi { 112 fn currency(&self) -> Currency { 113 self.currency 114 } 115 116 fn implementation(&self) -> &'static str { 117 "urn:net:taler:specs:taler-wise:taler-rust" 118 } 119 } 120 121 impl WiseBalanceApi { 122 async fn add_incoming( 123 &self, 124 tx: &TxIn, 125 subject: IncomingSubject, 126 ) -> ApiResult<AddIncomingResponse> { 127 let now = Timestamp::now(); 128 match db::register_tx_in(&self.pool, tx, &Some(subject), &now).await? { 129 AddIncomingResult::Success { 130 row_id, valued_at, .. 131 } => Ok(AddIncomingResponse { 132 row_id, 133 timestamp: valued_at.into(), 134 }), 135 AddIncomingResult::ReservePubReuse => { 136 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 137 } 138 AddIncomingResult::UnknownMapping => { 139 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_UNKNOWN)) 140 } 141 AddIncomingResult::MappingReuse => { 142 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_REUSED)) 143 } 144 } 145 } 146 } 147 148 impl WireGateway for WiseBalanceApi { 149 async fn transfer(&self, _: TransferRequest) -> ApiResult<TransferResponse> { 150 Err(not_implemented()) 151 } 152 153 async fn transfer_page(&self, _: Page, _: Option<TransferState>) -> ApiResult<TransferList> { 154 Ok(TransferList { 155 transfers: Vec::new(), 156 debit_account: self.payto.as_uri(), 157 }) 158 } 159 160 async fn transfer_by_id(&self, _: u64) -> ApiResult<Option<TransferStatus>> { 161 Ok(None) 162 } 163 164 async fn outgoing_history(&self, _: History) -> ApiResult<OutgoingHistory> { 165 Ok(OutgoingHistory { 166 outgoing_transactions: Vec::new(), 167 debit_account: self.payto.as_uri(), 168 }) 169 } 170 171 async fn incoming_history(&self, params: History) -> ApiResult<IncomingHistory> { 172 Ok(IncomingHistory { 173 incoming_transactions: db::incoming_history( 174 &self.pool, 175 self.id, 176 &self.currency, 177 ¶ms, 178 || self.taler_in_channel.subscribe(), 179 ) 180 .await?, 181 credit_account: self.payto.as_uri(), 182 }) 183 } 184 185 async fn add_incoming_reserve( 186 &self, 187 req: AddIncomingRequest, 188 ) -> ApiResult<AddIncomingResponse> { 189 let (account, name) = FullWisePayto::try_from(&req.debit_account)?.into_inner(); 190 let subject = format_compact!("Admin incoming {}", req.reserve_pub); 191 self.add_incoming( 192 &TxIn { 193 balance_id: self.id, 194 wise_ref: None, 195 amount: req.amount, 196 subject, 197 name, 198 debtor: Some(account), 199 value_at: Timestamp::now(), 200 }, 201 IncomingSubject::Reserve(req.reserve_pub), 202 ) 203 .await 204 } 205 206 async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddIncomingResponse> { 207 let (account, name) = FullWisePayto::try_from(&req.debit_account)?.into_inner(); 208 let subject = format_compact!("Admin incoming KYC:{}", req.account_pub); 209 self.add_incoming( 210 &TxIn { 211 balance_id: self.id, 212 wise_ref: None, 213 amount: req.amount, 214 subject, 215 name, 216 debtor: Some(account), 217 value_at: Timestamp::now(), 218 }, 219 IncomingSubject::Kyc(req.account_pub), 220 ) 221 .await 222 } 223 224 async fn add_incoming_mapped(&self, req: AddMappedRequest) -> ApiResult<AddIncomingResponse> { 225 let (account, name) = FullWisePayto::try_from(&req.debit_account)?.into_inner(); 226 let subject = format_compact!("Admin incoming MAP:{}", req.authorization_pub); 227 self.add_incoming( 228 &TxIn { 229 balance_id: self.id, 230 wise_ref: None, 231 amount: req.amount, 232 subject, 233 name, 234 debtor: Some(account), 235 value_at: Timestamp::now(), 236 }, 237 IncomingSubject::Map(req.authorization_pub), 238 ) 239 .await 240 } 241 242 fn support_account_check(&self) -> bool { 243 false 244 } 245 } 246 247 impl Revenue for WiseBalanceApi { 248 async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> { 249 Ok(RevenueIncomingHistory { 250 incoming_transactions: db::revenue_history( 251 &self.pool, 252 self.id, 253 &self.currency, 254 ¶ms, 255 || self.in_channel.subscribe(), 256 ) 257 .await?, 258 credit_account: self.payto.as_uri(), 259 }) 260 } 261 } 262 263 impl PreparedTransfer for WiseBalanceApi { 264 fn supported_formats(&self) -> &[SubjectFormat] { 265 &[SubjectFormat::SIMPLE] 266 } 267 268 async fn registration(&self, req: RegistrationRequest) -> ApiResult<RegistrationResponse> { 269 let creditor = FullWisePayto::try_from(&req.credit_account)?; 270 if creditor != self.payto { 271 return Err(failure_code(ErrorCode::BANK_UNKNOWN_CREDITOR)); 272 } 273 match db::transfer_register(&self.pool, &req).await? { 274 db::RegistrationResult::Success => { 275 let simple = TransferSubject::Simple { 276 credit_amount: req.credit_amount, 277 subject: if req.authorization_pub == req.account_pub && !req.recurrent { 278 fmt_in_subject(req.r#type.into(), &req.account_pub).to_string() 279 } else { 280 fmt_in_subject(IncomingType::map, &req.authorization_pub).to_string() 281 }, 282 }; 283 ApiResult::Ok(RegistrationResponse { 284 subjects: vec![simple], 285 expiration: TalerTimestamp::Never, 286 }) 287 } 288 db::RegistrationResult::ReservePubReuse => { 289 ApiResult::Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 290 } 291 } 292 } 293 294 async fn unregistration(&self, req: Unregistration) -> ApiResult<bool> { 295 Ok(db::transfer_unregister(&self.pool, &req).await?) 296 } 297 } 298 299 #[cfg(test)] 300 mod test { 301 302 use std::sync::LazyLock; 303 304 use compact_str::CompactString; 305 use jiff::Timestamp; 306 use sqlx::PgPool; 307 use taler_api::{ 308 api::TalerRouter as _, 309 config::{ApiCfg, AuthCfg}, 310 subject::IncomingSubject, 311 }; 312 use taler_common::{ 313 api::{ 314 EddsaPublicKey, prepared::PreparedTransferConfig, revenue::RevenueConfig, 315 wire::WireConfig, 316 }, 317 types::{ 318 amount::{Currency, amount}, 319 iban::iban, 320 payto::{BankID, PaytoURI}, 321 }, 322 }; 323 use taler_test_utils::{ 324 Router, 325 db::db_test_setup, 326 routine::{admin_add_incoming_routine, in_history_routine, revenue_routine}, 327 server::TestServer, 328 tasks, 329 }; 330 331 use crate::{ 332 CONFIG_SOURCE, api, 333 config::WiseBalance, 334 db::{TxIn, register_tx_in}, 335 payto::{FullWisePayto, WiseAccount}, 336 }; 337 338 static PAYTO: LazyLock<FullWisePayto> = LazyLock::new(|| { 339 "payto://ach/021000021/024030222?receiver-name=Exchange" 340 .parse() 341 .unwrap() 342 }); 343 static EXCHANGE: LazyLock<PaytoURI> = LazyLock::new(|| PAYTO.as_uri()); 344 345 async fn setup() -> (Router, PgPool) { 346 let (_, pool) = db_test_setup(CONFIG_SOURCE).await; 347 let balances = vec![ 348 WiseBalance { 349 id: 42, 350 currency: Currency::TEST, 351 payto: PAYTO.clone().into_inner().0, 352 }, 353 WiseBalance { 354 id: 34, 355 currency: Currency::KUDOS, 356 payto: PAYTO.clone().into_inner().0, 357 }, 358 ]; 359 let server = api::start( 360 pool.clone(), 361 "Exchange", 362 balances, 363 &Some(ApiCfg { 364 auth: AuthCfg::None, 365 }), 366 &Some(ApiCfg { 367 auth: AuthCfg::None, 368 }), 369 ) 370 .await 371 .finalize(); 372 373 (server, pool) 374 } 375 376 #[tokio::test] 377 async fn config() { 378 let (server, _) = setup().await; 379 for id in [42, 34] { 380 server 381 .get(format!("/balances/{id}/taler-wire-gateway/config")) 382 .await 383 .assert_ok_json::<WireConfig>(); 384 server 385 .get(format!("/balances/{id}/taler-prepared-transfer/config")) 386 .await 387 .assert_ok_json::<PreparedTransferConfig>(); 388 server 389 .get(format!("/balances/{id}/taler-revenue/config")) 390 .await 391 .assert_ok_json::<RevenueConfig>(); 392 } 393 } 394 395 async fn r#in(db: &PgPool, subject: Option<IncomingSubject>) { 396 register_tx_in( 397 db, 398 &TxIn { 399 balance_id: 42, 400 wise_ref: None, 401 amount: amount("EUR:10"), 402 subject: "subject".into(), 403 name: CompactString::const_new("Name"), 404 debtor: Some(WiseAccount::IBAN(BankID { 405 iban: iban("HU30162000031000163100000000"), 406 bic: None, 407 })), 408 value_at: Timestamp::now(), 409 }, 410 &subject, 411 &Timestamp::now(), 412 ) 413 .await 414 .unwrap(); 415 } 416 417 async fn in_malformed(db: &PgPool) { 418 r#in(db, None).await 419 } 420 421 async fn in_talerable(db: &PgPool) { 422 r#in(db, Some(IncomingSubject::Reserve(EddsaPublicKey::rand()))).await 423 } 424 425 #[tokio::test] 426 async fn admin_add_incoming() { 427 let (server, _) = setup().await; 428 admin_add_incoming_routine( 429 &server.prefix("/balances/42/taler-wire-gateway"), 430 &server.prefix("/balances/42/taler-prepared-transfer"), 431 &EXCHANGE, 432 &EXCHANGE, 433 ) 434 .await; 435 } 436 437 #[tokio::test] 438 async fn in_history() { 439 let (server, db) = &setup().await; 440 in_history_routine( 441 &server.prefix("/balances/42/taler-wire-gateway"), 442 &server.prefix("/balances/42/taler-prepared-transfer"), 443 &EXCHANGE, 444 &EXCHANGE, 445 tasks!({ in_talerable(db).await }), 446 tasks!({ in_malformed(db).await }), 447 ) 448 .await; 449 } 450 451 #[tokio::test] 452 async fn revenue() { 453 let (server, db) = &setup().await; 454 revenue_routine( 455 &server.prefix("/balances/42/taler-wire-gateway"), 456 &server.prefix("/balances/42/taler-revenue"), 457 &EXCHANGE, 458 tasks!({ in_malformed(db).await }, { in_talerable(db).await },), 459 tasks!(), 460 ) 461 .await; 462 } 463 }