api.rs (16637B)
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 taler_api::{ 20 api::{TalerApi, revenue::Revenue, transfer::PreparedTransfer, wire::WireGateway}, 21 error::{ApiResult, failure, failure_code}, 22 subject::{IncomingSubject, fmt_in_subject}, 23 }; 24 use taler_common::{ 25 api_common::safe_u64, 26 api_params::{History, Page}, 27 api_revenue::RevenueIncomingHistory, 28 api_transfer::{ 29 RegistrationRequest, RegistrationResponse, SubjectFormat, TransferSubject, Unregistration, 30 }, 31 api_wire::{ 32 AddIncomingRequest, AddIncomingResponse, AddKycauthRequest, AddMappedRequest, 33 IncomingHistory, OutgoingHistory, TransferList, TransferRequest, TransferResponse, 34 TransferState, TransferStatus, 35 }, 36 db::IncomingType, 37 error_code::ErrorCode, 38 types::{amount::Currency, payto::PaytoURI, timestamp::TalerTimestamp}, 39 }; 40 use tokio::sync::watch::Sender; 41 42 use crate::{ 43 db::{self, AddIncomingResult, Transfer, TxInAdmin}, 44 payto::FullCyclosPayto, 45 }; 46 47 pub struct CyclosApi { 48 pub pool: sqlx::PgPool, 49 pub currency: Currency, 50 pub payto: PaytoURI, 51 pub in_channel: Sender<i64>, 52 pub taler_in_channel: Sender<i64>, 53 pub out_channel: Sender<i64>, 54 pub taler_out_channel: Sender<i64>, 55 pub root: CompactString, 56 } 57 58 impl CyclosApi { 59 pub fn start( 60 pool: sqlx::PgPool, 61 root: CompactString, 62 payto: PaytoURI, 63 currency: Currency, 64 ) -> Self { 65 let in_channel = Sender::new(0); 66 let taler_in_channel = Sender::new(0); 67 let out_channel = Sender::new(0); 68 let taler_out_channel = Sender::new(0); 69 let tmp = Self { 70 pool: pool.clone(), 71 payto, 72 currency, 73 root, 74 in_channel: in_channel.clone(), 75 taler_in_channel: taler_in_channel.clone(), 76 out_channel: out_channel.clone(), 77 taler_out_channel: taler_out_channel.clone(), 78 }; 79 tokio::spawn(db::notification_listener( 80 pool, 81 in_channel, 82 taler_in_channel, 83 out_channel, 84 taler_out_channel, 85 )); 86 tmp 87 } 88 } 89 90 impl TalerApi for CyclosApi { 91 fn currency(&self) -> &str { 92 self.currency.as_ref() 93 } 94 95 fn implementation(&self) -> &'static str { 96 "urn:net:taler:specs:taler-cyclos:taler-rust" 97 } 98 } 99 100 impl WireGateway for CyclosApi { 101 async fn transfer(&self, req: TransferRequest) -> ApiResult<TransferResponse> { 102 let creditor = FullCyclosPayto::try_from(&req.credit_account)?; 103 let result = db::make_transfer( 104 &self.pool, 105 &Transfer { 106 request_uid: req.request_uid, 107 amount: req.amount.decimal(), 108 exchange_base_url: req.exchange_base_url, 109 metadata: req.metadata, 110 wtid: req.wtid, 111 creditor_id: *creditor.id, 112 creditor_name: creditor.name, 113 }, 114 &Timestamp::now(), 115 ) 116 .await?; 117 match result { 118 db::TransferResult::Success { id, initiated_at } => Ok(TransferResponse { 119 timestamp: initiated_at.into(), 120 row_id: safe_u64(id), 121 }), 122 db::TransferResult::RequestUidReuse => { 123 Err(failure_code(ErrorCode::BANK_TRANSFER_REQUEST_UID_REUSED)) 124 } 125 db::TransferResult::WtidReuse => { 126 Err(failure_code(ErrorCode::BANK_TRANSFER_WTID_REUSED)) 127 } 128 } 129 } 130 131 async fn transfer_page( 132 &self, 133 page: Page, 134 status: Option<TransferState>, 135 ) -> ApiResult<TransferList> { 136 Ok(TransferList { 137 transfers: db::transfer_page(&self.pool, &status, &self.currency, &self.root, &page) 138 .await?, 139 debit_account: self.payto.clone(), 140 }) 141 } 142 143 async fn transfer_by_id(&self, id: u64) -> ApiResult<Option<TransferStatus>> { 144 Ok(db::transfer_by_id(&self.pool, id, &self.currency, &self.root).await?) 145 } 146 147 async fn outgoing_history(&self, params: History) -> ApiResult<OutgoingHistory> { 148 Ok(OutgoingHistory { 149 outgoing_transactions: db::outgoing_history( 150 &self.pool, 151 ¶ms, 152 &self.currency, 153 &self.root, 154 || self.taler_out_channel.subscribe(), 155 ) 156 .await?, 157 debit_account: self.payto.clone(), 158 }) 159 } 160 161 async fn incoming_history(&self, params: History) -> ApiResult<IncomingHistory> { 162 Ok(IncomingHistory { 163 incoming_transactions: db::incoming_history( 164 &self.pool, 165 ¶ms, 166 &self.currency, 167 &self.root, 168 || self.taler_in_channel.subscribe(), 169 ) 170 .await?, 171 credit_account: self.payto.clone(), 172 }) 173 } 174 175 async fn add_incoming_reserve( 176 &self, 177 req: AddIncomingRequest, 178 ) -> ApiResult<AddIncomingResponse> { 179 let debtor = FullCyclosPayto::try_from(&req.debit_account)?; 180 let res = db::register_tx_in_admin( 181 &self.pool, 182 &TxInAdmin { 183 amount: req.amount.decimal(), 184 subject: format!("Admin incoming {}", req.reserve_pub), 185 debtor_id: *debtor.id, 186 debtor_name: debtor.name, 187 metadata: IncomingSubject::Reserve(req.reserve_pub), 188 }, 189 &Timestamp::now(), 190 ) 191 .await?; 192 match res { 193 AddIncomingResult::Success { 194 row_id, valued_at, .. 195 } => Ok(AddIncomingResponse { 196 row_id: safe_u64(row_id), 197 timestamp: valued_at.into(), 198 }), 199 AddIncomingResult::ReservePubReuse => { 200 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 201 } 202 AddIncomingResult::UnknownMapping | AddIncomingResult::MappingReuse => { 203 unreachable!("mapping unused") 204 } 205 } 206 } 207 208 async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddIncomingResponse> { 209 let debtor = FullCyclosPayto::try_from(&req.debit_account)?; 210 let res = db::register_tx_in_admin( 211 &self.pool, 212 &TxInAdmin { 213 amount: req.amount.decimal(), 214 subject: format!("Admin incoming KYC:{}", req.account_pub), 215 debtor_id: *debtor.id, 216 debtor_name: debtor.name, 217 metadata: IncomingSubject::Kyc(req.account_pub), 218 }, 219 &Timestamp::now(), 220 ) 221 .await?; 222 match res { 223 AddIncomingResult::Success { 224 row_id, valued_at, .. 225 } => Ok(AddIncomingResponse { 226 row_id: safe_u64(row_id), 227 timestamp: valued_at.into(), 228 }), 229 AddIncomingResult::ReservePubReuse => { 230 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 231 } 232 AddIncomingResult::UnknownMapping | AddIncomingResult::MappingReuse => { 233 unreachable!("mapping unused") 234 } 235 } 236 } 237 238 async fn add_incoming_mapped(&self, req: AddMappedRequest) -> ApiResult<AddIncomingResponse> { 239 let debtor = FullCyclosPayto::try_from(&req.debit_account)?; 240 let res = db::register_tx_in_admin( 241 &self.pool, 242 &TxInAdmin { 243 amount: req.amount.decimal(), 244 subject: format!("Admin incoming MAP:{}", req.authorization_pub), 245 debtor_id: *debtor.id, 246 debtor_name: debtor.name, 247 metadata: IncomingSubject::Map(req.authorization_pub), 248 }, 249 &Timestamp::now(), 250 ) 251 .await?; 252 match res { 253 AddIncomingResult::Success { 254 row_id, valued_at, .. 255 } => Ok(AddIncomingResponse { 256 row_id: safe_u64(row_id), 257 timestamp: valued_at.into(), 258 }), 259 AddIncomingResult::ReservePubReuse => { 260 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 261 } 262 AddIncomingResult::UnknownMapping => { 263 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_UNKNOWN)) 264 } 265 AddIncomingResult::MappingReuse => { 266 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_REUSED)) 267 } 268 } 269 } 270 271 fn support_account_check(&self) -> bool { 272 false 273 } 274 } 275 276 impl Revenue for CyclosApi { 277 async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> { 278 Ok(RevenueIncomingHistory { 279 incoming_transactions: db::revenue_history( 280 &self.pool, 281 ¶ms, 282 &self.currency, 283 &self.root, 284 || self.in_channel.subscribe(), 285 ) 286 .await?, 287 credit_account: self.payto.clone(), 288 }) 289 } 290 } 291 292 impl PreparedTransfer for CyclosApi { 293 fn supported_formats(&self) -> &[SubjectFormat] { 294 &[SubjectFormat::SIMPLE] 295 } 296 297 async fn registration(&self, req: RegistrationRequest) -> ApiResult<RegistrationResponse> { 298 match db::transfer_register(&self.pool, &req).await? { 299 db::RegistrationResult::Success => { 300 let simple = TransferSubject::Simple { 301 credit_amount: req.credit_amount, 302 subject: if req.authorization_pub == req.account_pub && !req.recurrent { 303 fmt_in_subject(req.r#type.into(), &req.account_pub) 304 } else { 305 fmt_in_subject(IncomingType::map, &req.authorization_pub) 306 }, 307 }; 308 ApiResult::Ok(RegistrationResponse { 309 subjects: vec![simple], 310 expiration: TalerTimestamp::Never, 311 }) 312 } 313 db::RegistrationResult::ReservePubReuse => { 314 ApiResult::Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT)) 315 } 316 } 317 } 318 319 async fn unregistration(&self, req: Unregistration) -> ApiResult<()> { 320 if !db::transfer_unregister(&self.pool, &req).await? { 321 Err(failure( 322 ErrorCode::BANK_TRANSACTION_NOT_FOUND, 323 format!("Prepared transfer '{}' not found", req.authorization_pub), 324 )) 325 } else { 326 Ok(()) 327 } 328 } 329 } 330 331 #[cfg(test)] 332 mod test { 333 use std::{ 334 str::FromStr as _, 335 sync::{Arc, LazyLock}, 336 }; 337 338 use compact_str::CompactString; 339 use jiff::Timestamp; 340 use sqlx::{PgPool, Row as _, postgres::PgRow}; 341 use taler_api::{ 342 api::TalerRouter as _, auth::AuthMethod, db::TypeHelper as _, subject::OutgoingSubject, 343 }; 344 use taler_common::{ 345 api_revenue::RevenueConfig, 346 api_transfer::PreparedTransferConfig, 347 api_wire::{OutgoingHistory, TransferState, WireConfig}, 348 db::IncomingType, 349 types::{ 350 amount::{Currency, decimal}, 351 payto::{PaytoURI, payto}, 352 }, 353 }; 354 use taler_test_utils::{ 355 Router, 356 db::db_test_setup, 357 routine::{ 358 Status, admin_add_incoming_routine, in_history_routine, registration_routine, 359 revenue_routine, routine_pagination, transfer_routine, 360 }, 361 server::TestServer as _, 362 }; 363 364 use crate::{ 365 api::CyclosApi, 366 constants::CONFIG_SOURCE, 367 db::{self, TxOutKind}, 368 }; 369 370 static ACCOUNT: LazyLock<PaytoURI> = 371 LazyLock::new(|| payto("payto://cyclos/localhost/7762070814178012479?receiver-name=name")); 372 373 async fn setup() -> (Router, PgPool) { 374 let (_, pool) = db_test_setup(CONFIG_SOURCE).await; 375 let api = Arc::new(CyclosApi::start( 376 pool.clone(), 377 CompactString::const_new("localhost"), 378 ACCOUNT.clone(), 379 Currency::from_str("TEST").unwrap(), 380 )); 381 let server = Router::new() 382 .wire_gateway(api.clone(), AuthMethod::None) 383 .prepared_transfer(api.clone()) 384 .revenue(api, AuthMethod::None) 385 .finalize(); 386 387 (server, pool) 388 } 389 390 #[tokio::test] 391 async fn config() { 392 let (server, _) = setup().await; 393 server 394 .get("/taler-wire-gateway/config") 395 .await 396 .assert_ok_json::<WireConfig>(); 397 server 398 .get("/taler-prepared-transfer/config") 399 .await 400 .assert_ok_json::<PreparedTransferConfig>(); 401 server 402 .get("/taler-revenue/config") 403 .await 404 .assert_ok_json::<RevenueConfig>(); 405 } 406 407 #[tokio::test] 408 async fn transfer() { 409 let (server, _) = setup().await; 410 transfer_routine(&server, TransferState::pending, &ACCOUNT).await; 411 } 412 413 #[tokio::test] 414 async fn outgoing_history() { 415 let (server, pool) = setup().await; 416 routine_pagination::<OutgoingHistory>( 417 &server, 418 "/taler-wire-gateway/history/outgoing", 419 async |i| { 420 db::register_tx_out( 421 &mut pool.acquire().await.unwrap(), 422 &db::TxOut { 423 transfer_id: i as i64, 424 tx_id: if i % 2 == 0 { 425 Some((i % 2) as i64) 426 } else { 427 None 428 }, 429 amount: decimal("10"), 430 subject: "subject".to_owned(), 431 creditor_id: 31000163100000000, 432 creditor_name: "Name".into(), 433 valued_at: Timestamp::now(), 434 }, 435 &TxOutKind::Talerable(OutgoingSubject::rand()), 436 &Timestamp::now(), 437 ) 438 .await 439 .unwrap(); 440 }, 441 ) 442 .await; 443 } 444 445 #[tokio::test] 446 async fn admin_add_incoming() { 447 let (server, _) = setup().await; 448 admin_add_incoming_routine(&server, &ACCOUNT, true).await; 449 } 450 451 #[tokio::test] 452 async fn in_history() { 453 let (server, _) = setup().await; 454 in_history_routine(&server, &ACCOUNT, true).await; 455 } 456 457 #[tokio::test] 458 async fn revenue() { 459 let (server, _) = setup().await; 460 revenue_routine(&server, &ACCOUNT, true).await; 461 } 462 463 async fn check_in(pool: &PgPool) -> Vec<Status> { 464 sqlx::query( 465 " 466 SELECT pending_recurrent_in.authorization_pub IS NOT NULL, bounced.tx_in_id IS NOT NULL, type, metadata 467 FROM tx_in 468 LEFT JOIN taler_in USING (tx_in_id) 469 LEFT JOIN pending_recurrent_in USING (tx_in_id) 470 LEFT JOIN bounced USING (tx_in_id) 471 ORDER BY tx_in.tx_in_id 472 ", 473 ) 474 .try_map(|r: PgRow| { 475 Ok( 476 if r.try_get_flag(0)? { 477 Status::Pending 478 } else if r.try_get_flag(1)? { 479 Status::Bounced 480 } else { 481 match r.try_get(2)? { 482 None => Status::Simple, 483 Some(IncomingType::reserve) => Status::Reserve(r.try_get(3)?), 484 Some(IncomingType::kyc) => Status::Kyc(r.try_get(3)?), 485 Some(e) => unreachable!("{e:?}") 486 } 487 } 488 ) 489 }) 490 .fetch_all(pool) 491 .await 492 .unwrap() 493 } 494 495 #[tokio::test] 496 async fn registration() { 497 let (server, pool) = setup().await; 498 registration_routine(&server, &ACCOUNT, || check_in(&pool)).await; 499 } 500 }