taler-rust

api.rs (20559B)

---

 /*
   This file is part of TALER
   Copyright (C) 2025-2026 Taler Systems SA
 
   TALER is free software; you can redistribute it and/or modify it under the
   terms of the GNU Affero General Public License as published by the Free Software
   Foundation; either version 3, or (at your option) any later version.
 
   TALER is distributed in the hope that it will be useful, but WITHOUT ANY
   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
   A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more details.
 
   You should have received a copy of the GNU Affero General Public License along with
   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
 */
 
 use compact_str::CompactString;
 use jiff::Timestamp;
 use prometheus_client::{metrics::gauge::Gauge, registry::Registry};
 use sqlx::PgPool;
 use taler_api::{
     api::{
         TalerApi, observability::Observability, prepared::PreparedTransfer, revenue::Revenue,
         wire::WireGateway,
     },
     error::{ApiResult, failure_code},
     subject::{IncomingKey, fmt_in_subject},
 };
 use taler_common::{
     api::{
         params::{History, Page},
         prepared::{
             RegistrationRequest, RegistrationResponse, SubjectFormat, TransferSubject,
             Unregistration,
         },
         revenue::RevenueIncomingHistory,
         wire::{
             AddIncomingRequest, AddIncomingResponse, AddKycauthRequest, AddMappedRequest,
             IncomingHistory, OutgoingHistory, TransferList, TransferRequest, TransferResponse,
             TransferState, TransferStatus,
         },
     },
     db::IncomingType,
     error_code::ErrorCode,
     types::{amount::Currency, time::TalerTimestamp},
 };
 use tokio::sync::watch::Sender;
 
 use crate::{
     db::{self, AddIncomingResult, Transfer, TxInAdmin},
     payto::{CyclosPayto, FullCyclosPayto},
 };
 
 pub struct CyclosApi {
     pub pool: sqlx::PgPool,
     pub currency: Currency,
     pub payto: FullCyclosPayto,
     pub in_channel: Sender<i64>,
     pub taler_in_channel: Sender<i64>,
     pub out_channel: Sender<i64>,
     pub taler_out_channel: Sender<i64>,
     pub root: CompactString,
     metrics: Metrics,
     registry: Registry,
 }
 
 #[derive(Default)]
 struct Metrics {
     db_access: Gauge,
 }
 
 impl Metrics {
     pub fn registry(&self) -> Registry {
         let mut registry = Registry::default();
 
         registry.register(
             "db_access",
             "Whether the last database metrics refresh succeeded",
             self.db_access.clone(),
         );
         registry
     }
 
     pub async fn sync(&self, db: &PgPool) {
         let test = sqlx::query("SELECT 1").fetch_one(db).await.is_ok();
         self.db_access.set(if test { 1 } else { 0 });
     }
 }
 
 impl CyclosApi {
     pub fn start(
         pool: sqlx::PgPool,
         root: CompactString,
         payto: FullCyclosPayto,
         currency: Currency,
     ) -> Self {
         let in_channel = Sender::new(0);
         let taler_in_channel = Sender::new(0);
         let out_channel = Sender::new(0);
         let taler_out_channel = Sender::new(0);
 
         let metrics = Metrics::default();
 
         let tmp = Self {
             pool: pool.clone(),
             payto,
             currency,
             root,
             in_channel: in_channel.clone(),
             taler_in_channel: taler_in_channel.clone(),
             out_channel: out_channel.clone(),
             taler_out_channel: taler_out_channel.clone(),
             registry: metrics.registry(),
             metrics,
         };
         tokio::spawn(db::notification_listener(
             pool,
             in_channel,
             taler_in_channel,
             out_channel,
             taler_out_channel,
         ));
         tmp
     }
 }
 
 impl TalerApi for CyclosApi {
     fn currency(&self) -> Currency {
         self.currency
     }
 
     fn implementation(&self) -> &'static str {
         "urn:net:taler:specs:taler-cyclos:taler-rust"
     }
 }
 
 impl WireGateway for CyclosApi {
     async fn transfer(&self, req: TransferRequest) -> ApiResult<TransferResponse> {
         let creditor = FullCyclosPayto::try_from(&req.credit_account)?;
         let result = db::make_transfer(
             &self.pool,
             &Transfer {
                 request_uid: req.request_uid,
                 amount: req.amount.decimal(),
                 exchange_base_url: req.exchange_base_url,
                 metadata: req.metadata,
                 wtid: req.wtid,
                 creditor_id: *creditor.id,
                 creditor_name: creditor.name,
             },
             &Timestamp::now(),
         )
         .await?;
         match result {
             db::TransferResult::Success { id, initiated_at } => Ok(TransferResponse {
                 timestamp: initiated_at.into(),
                 row_id: id,
             }),
             db::TransferResult::RequestUidReuse => {
                 Err(failure_code(ErrorCode::BANK_TRANSFER_REQUEST_UID_REUSED))
             }
             db::TransferResult::WtidReuse => {
                 Err(failure_code(ErrorCode::BANK_TRANSFER_WTID_REUSED))
             }
         }
     }
 
     async fn transfer_page(
         &self,
         page: Page,
         status: Option<TransferState>,
     ) -> ApiResult<TransferList> {
         Ok(TransferList {
             transfers: db::transfer_page(&self.pool, &status, &self.currency, &self.root, &page)
                 .await?,
             debit_account: self.payto.as_uri(),
         })
     }
 
     async fn transfer_by_id(&self, id: u64) -> ApiResult<Option<TransferStatus>> {
         Ok(db::transfer_by_id(&self.pool, id, &self.currency, &self.root).await?)
     }
 
     async fn outgoing_history(&self, params: History) -> ApiResult<OutgoingHistory> {
         Ok(OutgoingHistory {
             outgoing_transactions: db::outgoing_history(
                 &self.pool,
                 &params,
                 &self.currency,
                 &self.root,
                 || self.taler_out_channel.subscribe(),
             )
             .await?,
             debit_account: self.payto.as_uri(),
         })
     }
 
     async fn incoming_history(&self, params: History) -> ApiResult<IncomingHistory> {
         Ok(IncomingHistory {
             incoming_transactions: db::incoming_history(
                 &self.pool,
                 &params,
                 &self.currency,
                 &self.root,
                 || self.taler_in_channel.subscribe(),
             )
             .await?,
             credit_account: self.payto.as_uri(),
         })
     }
 
     async fn add_incoming_reserve(
         &self,
         req: AddIncomingRequest,
     ) -> ApiResult<AddIncomingResponse> {
         let debtor = FullCyclosPayto::try_from(&req.debit_account)?;
         let res = db::register_tx_in_admin(
             &self.pool,
             &TxInAdmin {
                 amount: req.amount.decimal(),
                 subject: format!("Admin incoming {}", req.reserve_pub),
                 debtor_id: *debtor.id,
                 debtor_name: debtor.name,
                 metadata: IncomingKey::reserve(req.reserve_pub),
             },
             &Timestamp::now(),
         )
         .await?;
         match res {
             AddIncomingResult::Success {
                 row_id, valued_at, ..
             } => Ok(AddIncomingResponse {
                 row_id,
                 timestamp: valued_at.into(),
             }),
             AddIncomingResult::ReservePubReuse => {
                 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT))
             }
             AddIncomingResult::UnknownMapping | AddIncomingResult::MappingReuse => {
                 unreachable!("mapping unused")
             }
         }
     }
 
     async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddIncomingResponse> {
         let debtor = FullCyclosPayto::try_from(&req.debit_account)?;
         let res = db::register_tx_in_admin(
             &self.pool,
             &TxInAdmin {
                 amount: req.amount.decimal(),
                 subject: format!("Admin incoming KYC:{}", req.account_pub),
                 debtor_id: *debtor.id,
                 debtor_name: debtor.name,
                 metadata: IncomingKey::kyc(req.account_pub),
             },
             &Timestamp::now(),
         )
         .await?;
         match res {
             AddIncomingResult::Success {
                 row_id, valued_at, ..
             } => Ok(AddIncomingResponse {
                 row_id,
                 timestamp: valued_at.into(),
             }),
             AddIncomingResult::ReservePubReuse => {
                 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT))
             }
             AddIncomingResult::UnknownMapping | AddIncomingResult::MappingReuse => {
                 unreachable!("mapping unused")
             }
         }
     }
 
     async fn add_incoming_mapped(&self, req: AddMappedRequest) -> ApiResult<AddIncomingResponse> {
         let debtor = FullCyclosPayto::try_from(&req.debit_account)?;
         let res = db::register_tx_in_admin(
             &self.pool,
             &TxInAdmin {
                 amount: req.amount.decimal(),
                 subject: format!("Admin incoming MAP:{}", req.authorization_pub),
                 debtor_id: *debtor.id,
                 debtor_name: debtor.name,
                 metadata: IncomingKey::map(req.authorization_pub),
             },
             &Timestamp::now(),
         )
         .await?;
         match res {
             AddIncomingResult::Success {
                 row_id, valued_at, ..
             } => Ok(AddIncomingResponse {
                 row_id,
                 timestamp: valued_at.into(),
             }),
             AddIncomingResult::ReservePubReuse => {
                 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT))
             }
             AddIncomingResult::UnknownMapping => {
                 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_UNKNOWN))
             }
             AddIncomingResult::MappingReuse => {
                 Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_REUSED))
             }
         }
     }
 
     fn support_account_check(&self) -> bool {
         false
     }
 }
 
 impl Revenue for CyclosApi {
     async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> {
         Ok(RevenueIncomingHistory {
             incoming_transactions: db::revenue_history(
                 &self.pool,
                 &params,
                 &self.currency,
                 &self.root,
                 || self.in_channel.subscribe(),
             )
             .await?,
             credit_account: self.payto.as_uri(),
         })
     }
 }
 
 impl PreparedTransfer for CyclosApi {
     fn supported_formats(&self) -> &[SubjectFormat] {
         &[SubjectFormat::SIMPLE]
     }
 
     async fn registration(&self, req: RegistrationRequest) -> ApiResult<RegistrationResponse> {
         let creditor = CyclosPayto::try_from(&req.credit_account)?;
         if *creditor != *self.payto {
             return Err(failure_code(ErrorCode::BANK_UNKNOWN_CREDITOR));
         }
         match db::transfer_register(&self.pool, &req).await? {
             db::RegistrationResult::Success => {
                 let simple = TransferSubject::Simple {
                     credit_amount: req.credit_amount,
                     subject: if req.authorization_pub == req.account_pub && !req.recurrent {
                         fmt_in_subject(req.r#type.into(), &req.account_pub).to_string()
                     } else {
                         fmt_in_subject(IncomingType::map, &req.authorization_pub).to_string()
                     },
                 };
                 ApiResult::Ok(RegistrationResponse {
                     subjects: vec![simple],
                     expiration: TalerTimestamp::Never,
                 })
             }
             db::RegistrationResult::ReservePubReuse => {
                 ApiResult::Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT))
             }
         }
     }
 
     async fn unregistration(&self, req: Unregistration) -> ApiResult<bool> {
         Ok(db::transfer_unregister(&self.pool, &req).await?)
     }
 }
 
 impl Observability for CyclosApi {
     async fn metrics(&self) -> ApiResult<&Registry> {
         self.metrics.sync(&self.pool).await;
         Ok(&self.registry)
     }
 }
 
 #[cfg(test)]
 mod test {
     use std::sync::{
         Arc, LazyLock,
         atomic::{AtomicI64, Ordering},
     };
 
     use compact_str::CompactString;
     use jiff::Timestamp;
     use sqlx::{PgPool, Row as _, postgres::PgRow};
     use taler_api::{
         api::TalerRouter as _,
         auth::AuthMethod,
         db::TypeHelper as _,
         subject::{IncomingKey, OutgoingSubject},
     };
     use taler_common::{
         api::{
             EddsaPublicKey,
             observability::Config,
             prepared::PreparedTransferConfig,
             revenue::RevenueConfig,
             wire::{TransferState, WireConfig},
         },
         db::IncomingType,
         types::{
             amount::{Currency, decimal},
             payto::{PaytoURI, payto},
         },
     };
     use taler_macros::db_test;
     use taler_test_utils::{
         Router,
         routine::{
             Status, admin_add_incoming_routine, in_history_routine, out_history_routine,
             registration_routine, revenue_routine, transfer_routine,
         },
         server::TestServer as _,
         tasks,
     };
 
     use crate::{
         api::CyclosApi,
         db::{self, TxIn, TxOutKind},
         payto::{FullCyclosPayto, cyclos_payto},
     };
 
     static PAYTO: LazyLock<FullCyclosPayto> = LazyLock::new(|| {
         cyclos_payto("payto://cyclos/localhost/7762070814178012479?receiver-name=Smith")
     });
     static EXCHANGE: LazyLock<PaytoURI> = LazyLock::new(|| PAYTO.as_uri());
     static UNKNOWN: LazyLock<PaytoURI> = LazyLock::new(|| {
         payto("payto://cyclos/localhost/7762070814178012478?receiver-name=Unknown")
     });
 
     async fn setup(pool: &PgPool) -> Router {
         let api = Arc::new(CyclosApi::start(
             pool.clone(),
             CompactString::const_new("localhost"),
             PAYTO.clone(),
             Currency::TEST,
         ));
         Router::new()
             .wire_gateway(api.clone(), AuthMethod::None)
             .prepared_transfer(api.clone())
             .revenue(api.clone(), AuthMethod::None)
             .observability(api, AuthMethod::None)
             .finalize()
     }
 
     #[db_test]
     async fn config(pool: PgPool) {
         let server = setup(&pool).await;
         server
             .get("/taler-wire-gateway/config")
             .await
             .assert_ok_json::<WireConfig>();
         server
             .get("/taler-prepared-transfer/config")
             .await
             .assert_ok_json::<PreparedTransferConfig>();
         server
             .get("/taler-revenue/config")
             .await
             .assert_ok_json::<RevenueConfig>();
         server
             .get("/taler-observability/config")
             .await
             .assert_ok_json::<Config>();
         server.get("/taler-observability/metrics").await.assert_ok();
     }
 
     #[db_test]
     async fn transfer(pool: PgPool) {
         let server = setup(&pool).await;
         transfer_routine(
             &server.prefix("/taler-wire-gateway"),
             TransferState::pending,
             &EXCHANGE,
         )
         .await;
     }
 
     static CODE: AtomicI64 = AtomicI64::new(0);
 
     async fn r#in(db: &PgPool, subject: Option<IncomingKey>) {
         let now = Timestamp::now();
         db::register_tx_in(
             &mut db.acquire().await.unwrap(),
             &TxIn {
                 transfer_id: CODE.fetch_add(1, Ordering::Relaxed),
                 tx_id: None,
                 amount: decimal("10"),
                 subject: "subject".to_owned(),
                 debtor_id: 31000163100000000,
                 debtor_name: "Name".into(),
                 valued_at: Timestamp::now(),
             },
             &subject,
             &now,
         )
         .await
         .unwrap();
     }
 
     async fn in_malformed(db: &PgPool) {
         r#in(db, None).await
     }
 
     async fn in_talerable(db: &PgPool) {
         r#in(db, Some(IncomingKey::reserve(EddsaPublicKey::rand()))).await
     }
 
     async fn out(db: &PgPool, kind: &TxOutKind) {
         let i = CODE.fetch_add(1, Ordering::Relaxed);
         let now = Timestamp::now();
         db::register_tx_out(
             &mut db.acquire().await.unwrap(),
             &db::TxOut {
                 transfer_id: i,
                 tx_id: if i % 2 == 0 { Some(i % 2) } else { None },
                 amount: decimal("10"),
                 subject: "subject".to_owned(),
                 creditor_id: 31000163100000000,
                 creditor_name: "Name".into(),
                 valued_at: now,
             },
             kind,
             &now,
         )
         .await
         .unwrap();
     }
 
     async fn out_talerable(db: &PgPool) {
         out(db, &TxOutKind::Talerable(OutgoingSubject::rand())).await
     }
 
     async fn out_bounce(db: &PgPool) {
         out(db, &TxOutKind::Bounce(CODE.load(Ordering::Relaxed))).await
     }
 
     async fn out_malformed(db: &PgPool) {
         out(db, &TxOutKind::Simple).await
     }
 
     #[db_test]
     async fn outgoing_history(db: PgPool) {
         let db = &db;
         let server = setup(db).await;
 
         out_history_routine(
             &server.prefix("/taler-wire-gateway"),
             tasks!({ out_talerable(db).await }),
             tasks!(
                 { out_bounce(db).await },
                 { out_malformed(db).await },
                 { in_malformed(db).await },
                 { in_talerable(db).await }
             ),
         )
         .await;
     }
 
     #[db_test]
     async fn admin_add_incoming(db: PgPool) {
         let server = setup(&db).await;
         admin_add_incoming_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-prepared-transfer"),
             &EXCHANGE,
             &EXCHANGE,
         )
         .await;
     }
 
     #[db_test]
     async fn in_history(db: PgPool) {
         let db = &db;
         let server = setup(db).await;
         in_history_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-prepared-transfer"),
             &EXCHANGE,
             &EXCHANGE,
             tasks!({ in_talerable(db).await }),
             tasks!(
                 { out_malformed(db).await },
                 { out_talerable(db).await },
                 { out_bounce(db).await },
                 { in_malformed(db).await }
             ),
         )
         .await;
     }
 
     #[db_test]
     async fn revenue(db: PgPool) {
         let db = &db;
         let server = setup(db).await;
         revenue_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-revenue"),
             &EXCHANGE,
             tasks!({ in_malformed(db).await }, { in_talerable(db).await },),
             tasks!({ out_malformed(db).await }, { out_talerable(db).await }, {
                 out_bounce(db).await
             }),
         )
         .await;
     }
 
     async fn check_in(pool: &PgPool) -> Vec<Status> {
         sqlx::query(
             "
             SELECT pending_recurrent_in.authorization_pub IS NOT NULL, bounced.tx_in_id IS NOT NULL, type, metadata
             FROM tx_in
                 LEFT JOIN taler_in USING (tx_in_id)
                 LEFT JOIN pending_recurrent_in USING (tx_in_id)
                 LEFT JOIN bounced USING (tx_in_id)
             ORDER BY tx_in.tx_in_id
         ",
         )
         .try_map(|r: PgRow| {
             Ok(
                 if r.try_get_flag(0)? {
                     Status::Pending
                 } else if r.try_get_flag(1)? {
                     Status::Bounced
                 } else {
                     match r.try_get(2)? {
                         None => Status::Simple,
                         Some(IncomingType::reserve) => Status::Reserve(r.try_get(3)?),
                         Some(IncomingType::kyc) => Status::Kyc(r.try_get(3)?),
                         Some(e) => unreachable!("{e:?}")
                     }
                 }
             )
         })
         .fetch_all(pool)
         .await
         .unwrap()
     }
 
     #[db_test]
     async fn registration(pool: PgPool) {
         let server = setup(&pool).await;
         registration_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-prepared-transfer"),
             &EXCHANGE,
             &EXCHANGE,
             &UNKNOWN,
             || check_in(&pool),
         )
         .await;
     }
 }