api.rs (16135B)
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 prometheus_client::{metrics::gauge::Gauge, registry::Registry}; 22 use sqlx::PgPool; 23 use taler_api::{ 24 api::{ 25 TalerApi, TalerRouter as _, observability::Observability, prepared::PreparedTransfer, 26 revenue::Revenue, wire::WireGateway, 27 }, 28 config::ApiCfg, 29 error::{ApiResult, failure_code, not_implemented}, 30 subject::{IncomingKey, fmt_in_subject}, 31 }; 32 use taler_common::{ 33 api::{ 34 params::{History, Page}, 35 prepared::{ 36 RegistrationRequest, RegistrationResponse, SubjectFormat, TransferSubject, 37 Unregistration, 38 }, 39 revenue::RevenueIncomingHistory, 40 wire::{ 41 AddIncomingRequest, AddIncomingResponse, AddKycauthRequest, AddMappedRequest, 42 IncomingHistory, OutgoingHistory, TransferList, TransferRequest, TransferResponse, 43 TransferState, TransferStatus, 44 }, 45 }, 46 db::IncomingType, 47 error_code::ErrorCode, 48 types::{amount::Currency, payto::PaytoImpl, time::TalerTimestamp}, 49 }; 50 use taler_test_utils::Router; 51 use tokio::sync::watch::Sender; 52 53 use crate::{ 54 config::WiseBalance, 55 db::{self, AddIncomingResult, TxIn}, 56 payto::FullWisePayto, 57 }; 58 59 pub async fn start( 60 pool: sqlx::PgPool, 61 name: &str, 62 balances: Vec<WiseBalance>, 63 wire_gateway: &Option<ApiCfg>, 64 revenue: &Option<ApiCfg>, 65 observability: &Option<ApiCfg>, 66 ) -> Router { 67 let apis: Vec<_> = balances 68 .into_iter() 69 .map(|b| { 70 let metrics = Metrics::default(); 71 WiseBalanceApi { 72 id: b.id, 73 currency: b.currency, 74 pool: pool.clone(), 75 payto: b.payto.full(name), 76 in_channel: Sender::new(0), 77 taler_in_channel: Sender::new(0), 78 registry: metrics.registry(), 79 metrics, 80 } 81 }) 82 .collect(); 83 let in_channels = Arc::new(BTreeMap::from_iter( 84 apis.iter().map(|b| (b.id, b.in_channel.clone())), 85 )); 86 let taler_in_channels = Arc::new(BTreeMap::from_iter( 87 apis.iter().map(|b| (b.id, b.taler_in_channel.clone())), 88 )); 89 tokio::spawn(db::notification_listener( 90 pool, 91 in_channels, 92 taler_in_channels, 93 )); 94 let mut main_router = Router::new(); 95 for api in apis { 96 let api = Arc::new(api); 97 let mut balance_router = Router::new(); 98 if let Some(cfg) = wire_gateway { 99 balance_router = balance_router 100 .wire_gateway(api.clone(), cfg.auth.method()) 101 .prepared_transfer(api.clone()); 102 } 103 if let Some(cfg) = revenue { 104 balance_router = balance_router.revenue(api.clone(), cfg.auth.method()); 105 } 106 if let Some(cfg) = observability { 107 balance_router = balance_router.observability(api.clone(), cfg.auth.method()); 108 } 109 main_router = main_router.nest(&format!("/balances/{}", api.id), balance_router); 110 } 111 main_router 112 } 113 114 pub struct WiseBalanceApi { 115 pub id: u32, 116 pub currency: Currency, 117 pub pool: sqlx::PgPool, 118 pub payto: FullWisePayto, 119 pub in_channel: Sender<i64>, 120 pub taler_in_channel: Sender<i64>, 121 metrics: Metrics, 122 registry: Registry, 123 } 124 125 #[derive(Default)] 126 struct Metrics { 127 db_access: Gauge, 128 } 129 130 impl Metrics { 131 pub fn registry(&self) -> Registry { 132 let mut registry = Registry::default(); 133 134 registry.register( 135 "db_access", 136 "Whether the last database metrics refresh succeeded", 137 self.db_access.clone(), 138 ); 139 registry 140 } 141 142 pub async fn sync(&self, db: &PgPool) { 143 let test = sqlx::query("SELECT 1").fetch_one(db).await.is_ok(); 144 self.db_access.set(if test { 1 } else { 0 }); 145 } 146 } 147 148 impl TalerApi for WiseBalanceApi { 149 fn currency(&self) -> Currency { 150 self.currency 151 } 152 153 fn implementation(&self) -> &'static str { 154 "urn:net:taler:specs:taler-wise:taler-rust" 155 } 156 } 157 158 impl WiseBalanceApi { 159 async fn add_incoming( 160 &self, 161 tx: &TxIn, 162 subject: IncomingKey, 163 ) -> ApiResult<AddIncomingResponse> { 164 let now = Timestamp::now(); 165 match db::register_tx_in(&self.pool, tx, &Some(subject), &now).await? { 166 AddIncomingResult::Success { 167 row_id, valued_at, .. 168 } => Ok(AddIncomingResponse { 169 row_id, 170 timestamp: valued_at.into(), 171 }), 172 AddIncomingResult::ReservePubReuse => { 173 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 174 } 175 AddIncomingResult::UnknownMapping => { 176 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_UNKNOWN)) 177 } 178 AddIncomingResult::MappingReuse => { 179 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_REUSED)) 180 } 181 } 182 } 183 } 184 185 impl WireGateway for WiseBalanceApi { 186 async fn transfer(&self, _: TransferRequest) -> ApiResult<TransferResponse> { 187 Err(not_implemented()) 188 } 189 190 async fn transfer_page(&self, _: Page, _: Option<TransferState>) -> ApiResult<TransferList> { 191 Ok(TransferList { 192 transfers: Vec::new(), 193 debit_account: self.payto.as_uri(), 194 }) 195 } 196 197 async fn transfer_by_id(&self, _: u64) -> ApiResult<Option<TransferStatus>> { 198 Ok(None) 199 } 200 201 async fn outgoing_history(&self, _: History) -> ApiResult<OutgoingHistory> { 202 Ok(OutgoingHistory { 203 outgoing_transactions: Vec::new(), 204 debit_account: self.payto.as_uri(), 205 }) 206 } 207 208 async fn incoming_history(&self, params: History) -> ApiResult<IncomingHistory> { 209 Ok(IncomingHistory { 210 incoming_transactions: db::incoming_history( 211 &self.pool, 212 self.id, 213 &self.currency, 214 ¶ms, 215 || self.taler_in_channel.subscribe(), 216 ) 217 .await?, 218 credit_account: self.payto.as_uri(), 219 }) 220 } 221 222 async fn add_incoming_reserve( 223 &self, 224 req: AddIncomingRequest, 225 ) -> ApiResult<AddIncomingResponse> { 226 let (account, name) = FullWisePayto::try_from(&req.debit_account)?.into_inner(); 227 let subject = format_compact!("Admin incoming {}", req.reserve_pub); 228 self.add_incoming( 229 &TxIn { 230 balance_id: self.id, 231 wise_ref: None, 232 amount: req.amount, 233 subject, 234 name, 235 debtor: Some(account), 236 value_at: Timestamp::now(), 237 }, 238 IncomingKey::reserve(req.reserve_pub), 239 ) 240 .await 241 } 242 243 async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddIncomingResponse> { 244 let (account, name) = FullWisePayto::try_from(&req.debit_account)?.into_inner(); 245 let subject = format_compact!("Admin incoming KYC:{}", req.account_pub); 246 self.add_incoming( 247 &TxIn { 248 balance_id: self.id, 249 wise_ref: None, 250 amount: req.amount, 251 subject, 252 name, 253 debtor: Some(account), 254 value_at: Timestamp::now(), 255 }, 256 IncomingKey::kyc(req.account_pub), 257 ) 258 .await 259 } 260 261 async fn add_incoming_mapped(&self, req: AddMappedRequest) -> ApiResult<AddIncomingResponse> { 262 let (account, name) = FullWisePayto::try_from(&req.debit_account)?.into_inner(); 263 let subject = format_compact!("Admin incoming MAP:{}", req.authorization_pub); 264 self.add_incoming( 265 &TxIn { 266 balance_id: self.id, 267 wise_ref: None, 268 amount: req.amount, 269 subject, 270 name, 271 debtor: Some(account), 272 value_at: Timestamp::now(), 273 }, 274 IncomingKey::map(req.authorization_pub), 275 ) 276 .await 277 } 278 279 fn support_account_check(&self) -> bool { 280 false 281 } 282 } 283 284 impl Revenue for WiseBalanceApi { 285 async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> { 286 Ok(RevenueIncomingHistory { 287 incoming_transactions: db::revenue_history( 288 &self.pool, 289 self.id, 290 &self.currency, 291 ¶ms, 292 || self.in_channel.subscribe(), 293 ) 294 .await?, 295 credit_account: self.payto.as_uri(), 296 }) 297 } 298 } 299 300 impl PreparedTransfer for WiseBalanceApi { 301 fn supported_formats(&self) -> &[SubjectFormat] { 302 &[SubjectFormat::SIMPLE] 303 } 304 305 async fn registration(&self, req: RegistrationRequest) -> ApiResult<RegistrationResponse> { 306 let creditor = FullWisePayto::try_from(&req.credit_account)?; 307 if creditor != self.payto { 308 return Err(failure_code(ErrorCode::BANK_UNKNOWN_CREDITOR)); 309 } 310 match db::transfer_register(&self.pool, &req).await? { 311 db::RegistrationResult::Success => { 312 let simple = TransferSubject::Simple { 313 credit_amount: req.credit_amount, 314 subject: if req.authorization_pub == req.account_pub && !req.recurrent { 315 fmt_in_subject(req.r#type.into(), &req.account_pub).to_string() 316 } else { 317 fmt_in_subject(IncomingType::map, &req.authorization_pub).to_string() 318 }, 319 }; 320 ApiResult::Ok(RegistrationResponse { 321 subjects: vec![simple], 322 expiration: TalerTimestamp::Never, 323 }) 324 } 325 db::RegistrationResult::ReservePubReuse => { 326 ApiResult::Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 327 } 328 } 329 } 330 331 async fn unregistration(&self, req: Unregistration) -> ApiResult<bool> { 332 Ok(db::transfer_unregister(&self.pool, &req).await?) 333 } 334 } 335 336 impl Observability for WiseBalanceApi { 337 async fn metrics(&self) -> ApiResult<&Registry> { 338 self.metrics.sync(&self.pool).await; 339 Ok(&self.registry) 340 } 341 } 342 343 #[cfg(test)] 344 mod test { 345 346 use std::sync::LazyLock; 347 348 use compact_str::CompactString; 349 use jiff::Timestamp; 350 use sqlx::PgPool; 351 use taler_api::{ 352 api::TalerRouter as _, 353 config::{ApiCfg, AuthCfg}, 354 subject::IncomingKey, 355 }; 356 use taler_common::{ 357 api::{ 358 EddsaPublicKey, observability::Config, prepared::PreparedTransferConfig, 359 revenue::RevenueConfig, wire::WireConfig, 360 }, 361 types::{ 362 amount::{Currency, amount}, 363 iban::iban, 364 payto::{BankID, PaytoURI}, 365 }, 366 }; 367 use taler_test_utils::{ 368 Router, 369 db::db_test_setup, 370 routine::{admin_add_incoming_routine, in_history_routine, revenue_routine}, 371 server::TestServer, 372 tasks, 373 }; 374 375 use crate::{ 376 CONFIG_SOURCE, api, 377 config::WiseBalance, 378 db::{TxIn, register_tx_in}, 379 payto::{FullWisePayto, WiseAccount}, 380 }; 381 382 static PAYTO: LazyLock<FullWisePayto> = LazyLock::new(|| { 383 "payto://ach/021000021/024030222?receiver-name=Exchange" 384 .parse() 385 .unwrap() 386 }); 387 static EXCHANGE: LazyLock<PaytoURI> = LazyLock::new(|| PAYTO.as_uri()); 388 389 async fn setup() -> (Router, PgPool) { 390 let (_, pool) = db_test_setup(CONFIG_SOURCE).await; 391 let balances = vec![ 392 WiseBalance { 393 id: 42, 394 currency: Currency::TEST, 395 payto: PAYTO.clone().into_inner().0, 396 }, 397 WiseBalance { 398 id: 34, 399 currency: Currency::KUDOS, 400 payto: PAYTO.clone().into_inner().0, 401 }, 402 ]; 403 let server = api::start( 404 pool.clone(), 405 "Exchange", 406 balances, 407 &Some(ApiCfg { 408 auth: AuthCfg::None, 409 }), 410 &Some(ApiCfg { 411 auth: AuthCfg::None, 412 }), 413 &Some(ApiCfg { 414 auth: AuthCfg::None, 415 }), 416 ) 417 .await 418 .finalize(); 419 420 (server, pool) 421 } 422 423 #[tokio::test] 424 async fn config() { 425 let (server, _) = setup().await; 426 for id in [42, 34] { 427 server 428 .get(format!("/balances/{id}/taler-wire-gateway/config")) 429 .await 430 .assert_ok_json::<WireConfig>(); 431 server 432 .get(format!("/balances/{id}/taler-prepared-transfer/config")) 433 .await 434 .assert_ok_json::<PreparedTransferConfig>(); 435 server 436 .get(format!("/balances/{id}/taler-revenue/config")) 437 .await 438 .assert_ok_json::<RevenueConfig>(); 439 server 440 .get(format!("/balances/{id}/taler-observability/config")) 441 .await 442 .assert_ok_json::<Config>(); 443 server 444 .get(format!("/balances/{id}/taler-observability/metrics")) 445 .await 446 .assert_ok(); 447 } 448 } 449 450 async fn r#in(db: &PgPool, subject: Option<IncomingKey>) { 451 register_tx_in( 452 db, 453 &TxIn { 454 balance_id: 42, 455 wise_ref: None, 456 amount: amount("EUR:10"), 457 subject: "subject".into(), 458 name: CompactString::const_new("Name"), 459 debtor: Some(WiseAccount::IBAN(BankID { 460 iban: iban("HU30162000031000163100000000"), 461 bic: None, 462 })), 463 value_at: Timestamp::now(), 464 }, 465 &subject, 466 &Timestamp::now(), 467 ) 468 .await 469 .unwrap(); 470 } 471 472 async fn in_malformed(db: &PgPool) { 473 r#in(db, None).await 474 } 475 476 async fn in_talerable(db: &PgPool) { 477 r#in(db, Some(IncomingKey::reserve(EddsaPublicKey::rand()))).await 478 } 479 480 #[tokio::test] 481 async fn admin_add_incoming() { 482 let (server, _) = setup().await; 483 admin_add_incoming_routine( 484 &server.prefix("/balances/42/taler-wire-gateway"), 485 &server.prefix("/balances/42/taler-prepared-transfer"), 486 &EXCHANGE, 487 &EXCHANGE, 488 ) 489 .await; 490 } 491 492 #[tokio::test] 493 async fn in_history() { 494 let (server, db) = &setup().await; 495 in_history_routine( 496 &server.prefix("/balances/42/taler-wire-gateway"), 497 &server.prefix("/balances/42/taler-prepared-transfer"), 498 &EXCHANGE, 499 &EXCHANGE, 500 tasks!({ in_talerable(db).await }), 501 tasks!({ in_malformed(db).await }), 502 ) 503 .await; 504 } 505 506 #[tokio::test] 507 async fn revenue() { 508 let (server, db) = &setup().await; 509 revenue_routine( 510 &server.prefix("/balances/42/taler-wire-gateway"), 511 &server.prefix("/balances/42/taler-revenue"), 512 &EXCHANGE, 513 tasks!({ in_malformed(db).await }, { in_talerable(db).await },), 514 tasks!(), 515 ) 516 .await; 517 } 518 }