api.rs (22575B)
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 compact_str::CompactString; 18 use jiff::Timestamp; 19 use prometheus_client::{ 20 encoding::EncodeLabelSet, 21 metrics::{family::Family, gauge::Gauge}, 22 registry::{Registry, Unit}, 23 }; 24 use sqlx::PgPool; 25 use taler_api::{ 26 api::{ 27 TalerApi, observability::Observability, prepared::PreparedTransfer, revenue::Revenue, 28 wire::WireGateway, 29 }, 30 error::{ApiResult, failure_code}, 31 subject::{IncomingKey, fmt_in_subject}, 32 }; 33 use taler_common::{ 34 api::{ 35 params::{History, Page}, 36 prepared::{ 37 RegistrationRequest, RegistrationResponse, SubjectFormat, TransferSubject, 38 Unregistration, 39 }, 40 revenue::RevenueIncomingHistory, 41 wire::{ 42 AddIncomingRequest, AddIncomingResponse, AddKycauthRequest, AddMappedRequest, 43 IncomingHistory, OutgoingHistory, TransferList, TransferRequest, TransferResponse, 44 TransferState, TransferStatus, 45 }, 46 }, 47 db::IncomingType, 48 error_code::ErrorCode, 49 types::{amount::Currency, time::TalerTimestamp}, 50 }; 51 use tokio::sync::watch::Sender; 52 53 use crate::{ 54 db::{self, AddIncomingResult, Transfer, TxInAdmin}, 55 payto::{CyclosPayto, FullCyclosPayto}, 56 }; 57 58 pub struct CyclosApi { 59 pub pool: sqlx::PgPool, 60 pub currency: Currency, 61 pub payto: FullCyclosPayto, 62 pub in_channel: Sender<i64>, 63 pub taler_in_channel: Sender<i64>, 64 pub out_channel: Sender<i64>, 65 pub taler_out_channel: Sender<i64>, 66 pub root: CompactString, 67 metrics: Metrics, 68 registry: Registry, 69 } 70 71 #[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)] 72 struct TaskLabel { 73 name: &'static str, 74 } 75 76 #[derive(Default)] 77 struct Metrics { 78 db_access: Gauge, 79 task_execution: Family<TaskLabel, Gauge>, 80 task_success: Family<TaskLabel, Gauge>, 81 } 82 83 impl Metrics { 84 pub fn registry(&self) -> Registry { 85 let mut registry = Registry::default(); 86 87 registry.register( 88 "db_access", 89 "Whether the last database metrics refresh succeeded", 90 self.db_access.clone(), 91 ); 92 registry.register_with_unit( 93 "task_execution_timestamp_seconds", 94 "Unix timestamp of the last task execution", 95 Unit::Seconds, 96 self.task_execution.clone(), 97 ); 98 registry.register_with_unit( 99 "task_success_timestamp_seconds", 100 "Unix timestamp of the last successful task execution", 101 Unit::Seconds, 102 self.task_success.clone(), 103 ); 104 registry 105 } 106 107 pub async fn sync(&self, db: &PgPool) { 108 let result = sqlx::query_as::<_, (String, sqlx::types::Json<db::TaskStatus>)>( 109 "SELECT key, value FROM kv WHERE key IN ('sync', 'submit')", 110 ) 111 .fetch_all(db) 112 .await; 113 let success = match result { 114 Ok(rows) => { 115 for name in ["sync", "submit"] { 116 let label = TaskLabel { name }; 117 let status = rows.iter().find(|(key, _)| key == name).map(|(_, s)| &s.0); 118 if let Some(time) = status.and_then(|s| s.last_execution) { 119 self.task_execution 120 .get_or_create(&label) 121 .set(time.as_second()); 122 } else { 123 self.task_execution.remove(&label); 124 } 125 if let Some(time) = status.and_then(|s| s.last_success) { 126 self.task_success 127 .get_or_create(&label) 128 .set(time.as_second()); 129 } else { 130 self.task_success.remove(&label); 131 } 132 } 133 true 134 } 135 Err(_) => false, 136 }; 137 self.db_access.set(if success { 1 } else { 0 }); 138 } 139 } 140 141 impl CyclosApi { 142 pub fn start( 143 pool: sqlx::PgPool, 144 root: CompactString, 145 payto: FullCyclosPayto, 146 currency: Currency, 147 ) -> Self { 148 let in_channel = Sender::new(0); 149 let taler_in_channel = Sender::new(0); 150 let out_channel = Sender::new(0); 151 let taler_out_channel = Sender::new(0); 152 153 let metrics = Metrics::default(); 154 155 let tmp = Self { 156 pool: pool.clone(), 157 payto, 158 currency, 159 root, 160 in_channel: in_channel.clone(), 161 taler_in_channel: taler_in_channel.clone(), 162 out_channel: out_channel.clone(), 163 taler_out_channel: taler_out_channel.clone(), 164 registry: metrics.registry(), 165 metrics, 166 }; 167 tokio::spawn(db::notification_listener( 168 pool, 169 in_channel, 170 taler_in_channel, 171 out_channel, 172 taler_out_channel, 173 )); 174 tmp 175 } 176 } 177 178 impl TalerApi for CyclosApi { 179 fn currency(&self) -> Currency { 180 self.currency 181 } 182 183 fn implementation(&self) -> &'static str { 184 "urn:net:taler:specs:taler-cyclos:taler-rust" 185 } 186 } 187 188 impl WireGateway for CyclosApi { 189 async fn transfer(&self, req: TransferRequest) -> ApiResult<TransferResponse> { 190 let creditor = FullCyclosPayto::try_from(&req.credit_account)?; 191 let result = db::make_transfer( 192 &self.pool, 193 &Transfer { 194 request_uid: req.request_uid, 195 amount: req.amount.decimal(), 196 exchange_base_url: req.exchange_base_url, 197 metadata: req.metadata, 198 wtid: req.wtid, 199 creditor_id: *creditor.id, 200 creditor_name: creditor.name, 201 }, 202 &Timestamp::now(), 203 ) 204 .await?; 205 match result { 206 db::TransferResult::Success { id, initiated_at } => Ok(TransferResponse { 207 timestamp: initiated_at.into(), 208 row_id: id, 209 }), 210 db::TransferResult::RequestUidReuse => { 211 Err(failure_code(ErrorCode::BANK_TRANSFER_REQUEST_UID_REUSED)) 212 } 213 db::TransferResult::WtidReuse => { 214 Err(failure_code(ErrorCode::BANK_TRANSFER_WTID_REUSED)) 215 } 216 } 217 } 218 219 async fn transfer_page( 220 &self, 221 page: Page, 222 status: Option<TransferState>, 223 ) -> ApiResult<TransferList> { 224 Ok(TransferList { 225 transfers: db::transfer_page(&self.pool, &status, &self.currency, &self.root, &page) 226 .await?, 227 debit_account: self.payto.as_uri(), 228 }) 229 } 230 231 async fn transfer_by_id(&self, id: u64) -> ApiResult<Option<TransferStatus>> { 232 Ok(db::transfer_by_id(&self.pool, id, &self.currency, &self.root).await?) 233 } 234 235 async fn outgoing_history(&self, params: History) -> ApiResult<OutgoingHistory> { 236 Ok(OutgoingHistory { 237 outgoing_transactions: db::outgoing_history( 238 &self.pool, 239 ¶ms, 240 &self.currency, 241 &self.root, 242 || self.taler_out_channel.subscribe(), 243 ) 244 .await?, 245 debit_account: self.payto.as_uri(), 246 }) 247 } 248 249 async fn incoming_history(&self, params: History) -> ApiResult<IncomingHistory> { 250 Ok(IncomingHistory { 251 incoming_transactions: db::incoming_history( 252 &self.pool, 253 ¶ms, 254 &self.currency, 255 &self.root, 256 || self.taler_in_channel.subscribe(), 257 ) 258 .await?, 259 credit_account: self.payto.as_uri(), 260 }) 261 } 262 263 async fn add_incoming_reserve( 264 &self, 265 req: AddIncomingRequest, 266 ) -> ApiResult<AddIncomingResponse> { 267 let debtor = FullCyclosPayto::try_from(&req.debit_account)?; 268 let res = db::register_tx_in_admin( 269 &self.pool, 270 &TxInAdmin { 271 amount: req.amount.decimal(), 272 subject: format!("Admin incoming {}", req.reserve_pub), 273 debtor_id: *debtor.id, 274 debtor_name: debtor.name, 275 metadata: IncomingKey::reserve(req.reserve_pub), 276 }, 277 &Timestamp::now(), 278 ) 279 .await?; 280 match res { 281 AddIncomingResult::Success { 282 row_id, valued_at, .. 283 } => Ok(AddIncomingResponse { 284 row_id, 285 timestamp: valued_at.into(), 286 }), 287 AddIncomingResult::ReservePubReuse => { 288 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 289 } 290 AddIncomingResult::UnknownMapping | AddIncomingResult::MappingReuse => { 291 unreachable!("mapping unused") 292 } 293 } 294 } 295 296 async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddIncomingResponse> { 297 let debtor = FullCyclosPayto::try_from(&req.debit_account)?; 298 let res = db::register_tx_in_admin( 299 &self.pool, 300 &TxInAdmin { 301 amount: req.amount.decimal(), 302 subject: format!("Admin incoming KYC:{}", req.account_pub), 303 debtor_id: *debtor.id, 304 debtor_name: debtor.name, 305 metadata: IncomingKey::kyc(req.account_pub), 306 }, 307 &Timestamp::now(), 308 ) 309 .await?; 310 match res { 311 AddIncomingResult::Success { 312 row_id, valued_at, .. 313 } => Ok(AddIncomingResponse { 314 row_id, 315 timestamp: valued_at.into(), 316 }), 317 AddIncomingResult::ReservePubReuse => { 318 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 319 } 320 AddIncomingResult::UnknownMapping | AddIncomingResult::MappingReuse => { 321 unreachable!("mapping unused") 322 } 323 } 324 } 325 326 async fn add_incoming_mapped(&self, req: AddMappedRequest) -> ApiResult<AddIncomingResponse> { 327 let debtor = FullCyclosPayto::try_from(&req.debit_account)?; 328 let res = db::register_tx_in_admin( 329 &self.pool, 330 &TxInAdmin { 331 amount: req.amount.decimal(), 332 subject: format!("Admin incoming MAP:{}", req.authorization_pub), 333 debtor_id: *debtor.id, 334 debtor_name: debtor.name, 335 metadata: IncomingKey::map(req.authorization_pub), 336 }, 337 &Timestamp::now(), 338 ) 339 .await?; 340 match res { 341 AddIncomingResult::Success { 342 row_id, valued_at, .. 343 } => Ok(AddIncomingResponse { 344 row_id, 345 timestamp: valued_at.into(), 346 }), 347 AddIncomingResult::ReservePubReuse => { 348 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 349 } 350 AddIncomingResult::UnknownMapping => { 351 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_UNKNOWN)) 352 } 353 AddIncomingResult::MappingReuse => { 354 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_REUSED)) 355 } 356 } 357 } 358 359 fn support_account_check(&self) -> bool { 360 false 361 } 362 } 363 364 impl Revenue for CyclosApi { 365 async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> { 366 Ok(RevenueIncomingHistory { 367 incoming_transactions: db::revenue_history( 368 &self.pool, 369 ¶ms, 370 &self.currency, 371 &self.root, 372 || self.in_channel.subscribe(), 373 ) 374 .await?, 375 credit_account: self.payto.as_uri(), 376 }) 377 } 378 } 379 380 impl PreparedTransfer for CyclosApi { 381 fn supported_formats(&self) -> &[SubjectFormat] { 382 &[SubjectFormat::SIMPLE] 383 } 384 385 async fn registration(&self, req: RegistrationRequest) -> ApiResult<RegistrationResponse> { 386 let creditor = CyclosPayto::try_from(&req.credit_account)?; 387 if *creditor != *self.payto { 388 return Err(failure_code(ErrorCode::BANK_UNKNOWN_CREDITOR)); 389 } 390 match db::transfer_register(&self.pool, &req).await? { 391 db::RegistrationResult::Success => { 392 let simple = TransferSubject::Simple { 393 credit_amount: req.credit_amount, 394 subject: if req.authorization_pub == req.account_pub && !req.recurrent { 395 fmt_in_subject(req.r#type.into(), &req.account_pub).to_string() 396 } else { 397 fmt_in_subject(IncomingType::map, &req.authorization_pub).to_string() 398 }, 399 }; 400 ApiResult::Ok(RegistrationResponse { 401 subjects: vec![simple], 402 expiration: TalerTimestamp::Never, 403 }) 404 } 405 db::RegistrationResult::ReservePubReuse => { 406 ApiResult::Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 407 } 408 } 409 } 410 411 async fn unregistration(&self, req: Unregistration) -> ApiResult<bool> { 412 Ok(db::transfer_unregister(&self.pool, &req).await?) 413 } 414 } 415 416 impl Observability for CyclosApi { 417 async fn metrics(&self) -> ApiResult<&Registry> { 418 self.metrics.sync(&self.pool).await; 419 Ok(&self.registry) 420 } 421 } 422 423 #[cfg(test)] 424 mod test { 425 use std::sync::{ 426 LazyLock, 427 atomic::{AtomicI64, Ordering}, 428 }; 429 430 use compact_str::CompactString; 431 use jiff::Timestamp; 432 use sqlx::{PgPool, Row as _, postgres::PgRow}; 433 use taler_api::{ 434 Serve, 435 api::TalerRouter as _, 436 config::{ApiCfg, AuthCfg}, 437 db::TypeHelper as _, 438 subject::{IncomingKey, OutgoingSubject}, 439 }; 440 use taler_common::{ 441 api::{ 442 EddsaPublicKey, 443 observability::Config, 444 prepared::PreparedTransferConfig, 445 revenue::RevenueConfig, 446 wire::{TransferState, WireConfig}, 447 }, 448 db::IncomingType, 449 types::{ 450 amount::{Currency, decimal}, 451 payto::{PaytoURI, payto}, 452 }, 453 }; 454 use taler_macros::db_test; 455 use taler_test_utils::{ 456 Router, 457 routine::{ 458 Status, admin_add_incoming_routine, in_history_routine, out_history_routine, 459 registration_routine, revenue_routine, transfer_routine, 460 }, 461 server::TestServer as _, 462 tasks, 463 }; 464 465 use crate::{ 466 build_router, 467 config::ServeCfg, 468 db::{self, TxIn, TxOutKind}, 469 payto::{FullCyclosPayto, cyclos_payto}, 470 }; 471 472 static PAYTO: LazyLock<FullCyclosPayto> = LazyLock::new(|| { 473 cyclos_payto("payto://cyclos/localhost/7762070814178012479?receiver-name=Smith") 474 }); 475 static EXCHANGE: LazyLock<PaytoURI> = LazyLock::new(|| PAYTO.as_uri()); 476 static UNKNOWN: LazyLock<PaytoURI> = LazyLock::new(|| { 477 payto("payto://cyclos/localhost/7762070814178012478?receiver-name=Unknown") 478 }); 479 480 async fn setup(pool: &PgPool) -> Router { 481 let cfg = ServeCfg { 482 payto: PAYTO.clone(), 483 root: CompactString::const_new("localhost"), 484 currency: Currency::TEST, 485 serve: Serve::Systemd, 486 wire_gateway: Some(ApiCfg { 487 auth: AuthCfg::None, 488 }), 489 revenue: Some(ApiCfg { 490 auth: AuthCfg::None, 491 }), 492 observability: Some(ApiCfg { 493 auth: AuthCfg::None, 494 }), 495 }; 496 build_router(&cfg, pool.clone()).await.finalize() 497 } 498 499 #[db_test] 500 async fn config(pool: PgPool) { 501 let server = setup(&pool).await; 502 server 503 .get("/taler-wire-gateway/config") 504 .await 505 .assert_ok_json::<WireConfig>(); 506 server 507 .get("/taler-prepared-transfer/config") 508 .await 509 .assert_ok_json::<PreparedTransferConfig>(); 510 server 511 .get("/taler-revenue/config") 512 .await 513 .assert_ok_json::<RevenueConfig>(); 514 server 515 .get("/taler-observability/config") 516 .await 517 .assert_ok_json::<Config>(); 518 server.get("/taler-observability/metrics").await.assert_ok(); 519 } 520 521 #[db_test] 522 async fn transfer(pool: PgPool) { 523 let server = setup(&pool).await; 524 transfer_routine( 525 &server.prefix("/taler-wire-gateway"), 526 TransferState::pending, 527 &EXCHANGE, 528 ) 529 .await; 530 } 531 532 static CODE: AtomicI64 = AtomicI64::new(0); 533 534 async fn r#in(db: &PgPool, subject: Option<IncomingKey>) { 535 let now = Timestamp::now(); 536 db::register_tx_in( 537 &mut db.acquire().await.unwrap(), 538 &TxIn { 539 transfer_id: CODE.fetch_add(1, Ordering::Relaxed), 540 tx_id: None, 541 amount: decimal("10"), 542 subject: "subject".to_owned(), 543 debtor_id: 31000163100000000, 544 debtor_name: "Name".into(), 545 valued_at: Timestamp::now(), 546 }, 547 &subject, 548 &now, 549 ) 550 .await 551 .unwrap(); 552 } 553 554 async fn in_malformed(db: &PgPool) { 555 r#in(db, None).await 556 } 557 558 async fn in_talerable(db: &PgPool) { 559 r#in(db, Some(IncomingKey::reserve(EddsaPublicKey::rand()))).await 560 } 561 562 async fn out(db: &PgPool, kind: &TxOutKind) { 563 let i = CODE.fetch_add(1, Ordering::Relaxed); 564 let now = Timestamp::now(); 565 db::register_tx_out( 566 &mut db.acquire().await.unwrap(), 567 &db::TxOut { 568 transfer_id: i, 569 tx_id: if i % 2 == 0 { Some(i % 2) } else { None }, 570 amount: decimal("10"), 571 subject: "subject".to_owned(), 572 creditor_id: 31000163100000000, 573 creditor_name: "Name".into(), 574 valued_at: now, 575 }, 576 kind, 577 &now, 578 ) 579 .await 580 .unwrap(); 581 } 582 583 async fn out_talerable(db: &PgPool) { 584 out(db, &TxOutKind::Talerable(OutgoingSubject::rand())).await 585 } 586 587 async fn out_bounce(db: &PgPool) { 588 out(db, &TxOutKind::Bounce(CODE.load(Ordering::Relaxed))).await 589 } 590 591 async fn out_malformed(db: &PgPool) { 592 out(db, &TxOutKind::Simple).await 593 } 594 595 #[db_test] 596 async fn outgoing_history(db: PgPool) { 597 let db = &db; 598 let server = setup(db).await; 599 600 out_history_routine( 601 &server.prefix("/taler-wire-gateway"), 602 tasks!({ out_talerable(db).await }), 603 tasks!( 604 { out_bounce(db).await }, 605 { out_malformed(db).await }, 606 { in_malformed(db).await }, 607 { in_talerable(db).await } 608 ), 609 ) 610 .await; 611 } 612 613 #[db_test] 614 async fn admin_add_incoming(db: PgPool) { 615 let server = setup(&db).await; 616 admin_add_incoming_routine( 617 &server.prefix("/taler-wire-gateway"), 618 &server.prefix("/taler-prepared-transfer"), 619 &EXCHANGE, 620 &EXCHANGE, 621 ) 622 .await; 623 } 624 625 #[db_test] 626 async fn in_history(db: PgPool) { 627 let db = &db; 628 let server = setup(db).await; 629 in_history_routine( 630 &server.prefix("/taler-wire-gateway"), 631 &server.prefix("/taler-prepared-transfer"), 632 &EXCHANGE, 633 &EXCHANGE, 634 tasks!({ in_talerable(db).await }), 635 tasks!( 636 { out_malformed(db).await }, 637 { out_talerable(db).await }, 638 { out_bounce(db).await }, 639 { in_malformed(db).await } 640 ), 641 ) 642 .await; 643 } 644 645 #[db_test] 646 async fn revenue(db: PgPool) { 647 let db = &db; 648 let server = setup(db).await; 649 revenue_routine( 650 &server.prefix("/taler-wire-gateway"), 651 &server.prefix("/taler-revenue"), 652 &EXCHANGE, 653 tasks!({ in_malformed(db).await }, { in_talerable(db).await },), 654 tasks!({ out_malformed(db).await }, { out_talerable(db).await }, { 655 out_bounce(db).await 656 }), 657 ) 658 .await; 659 } 660 661 async fn check_in(pool: &PgPool) -> Vec<Status> { 662 sqlx::query( 663 " 664 SELECT pending_recurrent_in.authorization_pub IS NOT NULL, bounced.tx_in_id IS NOT NULL, type, metadata 665 FROM tx_in 666 LEFT JOIN taler_in USING (tx_in_id) 667 LEFT JOIN pending_recurrent_in USING (tx_in_id) 668 LEFT JOIN bounced USING (tx_in_id) 669 ORDER BY tx_in.tx_in_id 670 ", 671 ) 672 .try_map(|r: PgRow| { 673 Ok( 674 if r.try_get_flag(0)? { 675 Status::Pending 676 } else if r.try_get_flag(1)? { 677 Status::Bounced 678 } else { 679 match r.try_get(2)? { 680 None => Status::Simple, 681 Some(IncomingType::reserve) => Status::Reserve(r.try_get(3)?), 682 Some(IncomingType::kyc) => Status::Kyc(r.try_get(3)?), 683 Some(e) => unreachable!("{e:?}") 684 } 685 } 686 ) 687 }) 688 .fetch_all(pool) 689 .await 690 .unwrap() 691 } 692 693 #[db_test] 694 async fn registration(pool: PgPool) { 695 let server = setup(&pool).await; 696 registration_routine( 697 &server.prefix("/taler-wire-gateway"), 698 &server.prefix("/taler-prepared-transfer"), 699 &EXCHANGE, 700 &EXCHANGE, 701 &UNKNOWN, 702 || check_in(&pool), 703 ) 704 .await; 705 } 706 }