taler-rust

api.rs (19660B)

---

 /*
   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 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, utils::date_to_utc_ts},
 };
 use tokio::sync::watch::Sender;
 
 use crate::{
     FullHuPayto,
     constants::CURR,
     db::{self, AddIncomingResult, Transfer, TxInAdmin},
 };
 
 pub struct MagnetApi {
     pub pool: sqlx::PgPool,
     pub payto: FullHuPayto,
     pub in_channel: Sender<i64>,
     pub taler_in_channel: Sender<i64>,
     pub out_channel: Sender<i64>,
     pub taler_out_channel: Sender<i64>,
     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 MagnetApi {
     pub async fn start(pool: sqlx::PgPool, payto: FullHuPayto) -> 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,
             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 MagnetApi {
     fn currency(&self) -> Currency {
         CURR
     }
 
     fn implementation(&self) -> &'static str {
         "urn:net:taler:specs:taler-magnet-bank:taler-rust"
     }
 }
 
 impl WireGateway for MagnetApi {
     async fn transfer(&self, req: TransferRequest) -> ApiResult<TransferResponse> {
         let creditor = FullHuPayto::try_from(&req.credit_account)?;
         let result = db::make_transfer(
             &self.pool,
             &Transfer {
                 request_uid: req.request_uid,
                 wtid: req.wtid,
                 amount: req.amount.decimal(),
                 metadata: req.metadata,
                 creditor,
                 exchange_base_url: req.exchange_base_url,
             },
             &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, &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).await?)
     }
 
     async fn outgoing_history(&self, params: History) -> ApiResult<OutgoingHistory> {
         Ok(OutgoingHistory {
             outgoing_transactions: db::outgoing_history(&self.pool, &params, || {
                 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.taler_in_channel.subscribe()
             })
             .await?,
             credit_account: self.payto.as_uri(),
         })
     }
 
     async fn add_incoming_reserve(
         &self,
         req: AddIncomingRequest,
     ) -> ApiResult<AddIncomingResponse> {
         let debtor = FullHuPayto::try_from(&req.debit_account)?;
         let res = db::register_tx_in_admin(
             &self.pool,
             &TxInAdmin {
                 amount: req.amount,
                 subject: format!("Admin incoming {}", req.reserve_pub),
                 debtor,
                 metadata: IncomingKey::reserve(req.reserve_pub),
             },
             &Timestamp::now(),
         )
         .await?;
         match res {
             AddIncomingResult::Success {
                 row_id, valued_at, ..
             } => Ok(AddIncomingResponse {
                 row_id,
                 timestamp: date_to_utc_ts(&valued_at).into(),
             }),
             AddIncomingResult::ReservePubReuse => {
                 Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT))
             }
             AddIncomingResult::UnknownMapping | AddIncomingResult::MappingReuse => {
                 unreachable!("mapping not used")
             }
         }
     }
 
     async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddIncomingResponse> {
         let debtor = FullHuPayto::try_from(&req.debit_account)?;
         let res = db::register_tx_in_admin(
             &self.pool,
             &TxInAdmin {
                 amount: req.amount,
                 subject: format!("Admin incoming KYC:{}", req.account_pub),
                 debtor,
                 metadata: IncomingKey::kyc(req.account_pub),
             },
             &Timestamp::now(),
         )
         .await?;
         match res {
             AddIncomingResult::Success {
                 row_id, valued_at, ..
             } => Ok(AddIncomingResponse {
                 row_id,
                 timestamp: date_to_utc_ts(&valued_at).into(),
             }),
             AddIncomingResult::ReservePubReuse => unreachable!("kyc"),
             AddIncomingResult::UnknownMapping | AddIncomingResult::MappingReuse => {
                 unreachable!("mapping not used")
             }
         }
     }
 
     async fn add_incoming_mapped(&self, req: AddMappedRequest) -> ApiResult<AddIncomingResponse> {
         let debtor = FullHuPayto::try_from(&req.debit_account)?;
         let res = db::register_tx_in_admin(
             &self.pool,
             &TxInAdmin {
                 amount: req.amount,
                 subject: format!("Admin incoming MAP:{}", req.authorization_pub),
                 debtor,
                 metadata: IncomingKey::map(req.authorization_pub),
             },
             &Timestamp::now(),
         )
         .await?;
         match res {
             AddIncomingResult::Success {
                 row_id, valued_at, ..
             } => Ok(AddIncomingResponse {
                 row_id,
                 timestamp: date_to_utc_ts(&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 MagnetApi {
     async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> {
         Ok(RevenueIncomingHistory {
             incoming_transactions: db::revenue_history(&self.pool, &params, || {
                 self.in_channel.subscribe()
             })
             .await?,
             credit_account: self.payto.as_uri(),
         })
     }
 }
 
 impl PreparedTransfer for MagnetApi {
     fn supported_formats(&self) -> &[SubjectFormat] {
         &[SubjectFormat::SIMPLE]
     }
 
     async fn registration(&self, req: RegistrationRequest) -> ApiResult<RegistrationResponse> {
         let creditor = FullHuPayto::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 MagnetApi {
     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::{AtomicU64, Ordering},
     };
 
     use jiff::{Timestamp, Zoned};
     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::amount,
             payto::{PaytoURI, payto},
         },
     };
     use taler_test_utils::{
         Router,
         db::db_test_setup,
         routine::{
             Status, admin_add_incoming_routine, in_history_routine, out_history_routine,
             registration_routine, revenue_routine, transfer_routine,
         },
         server::TestServer,
         tasks,
     };
 
     use crate::{
         FullHuPayto,
         api::MagnetApi,
         constants::CONFIG_SOURCE,
         db::{self, TxIn, TxOutKind},
         magnet_api::types::TxStatus,
         magnet_payto,
     };
 
     static PAYTO: LazyLock<FullHuPayto> = LazyLock::new(|| {
         magnet_payto("payto://iban/HU02162000031000164800000000?receiver-name=Smith")
     });
     static EXCHANGE: LazyLock<PaytoURI> = LazyLock::new(|| PAYTO.as_uri());
     static UNKNOWN: LazyLock<PaytoURI> =
         LazyLock::new(|| payto("payto://iban/HU60162006491000639900000000?receiver-name=Unknown"));
 
     async fn setup() -> (Router, PgPool) {
         let (_, pool) = db_test_setup(CONFIG_SOURCE).await;
         let api = Arc::new(MagnetApi::start(pool.clone(), PAYTO.clone()).await);
         let server = Router::new()
             .wire_gateway(api.clone(), AuthMethod::None)
             .prepared_transfer(api.clone())
             .revenue(api.clone(), AuthMethod::None)
             .observability(api, AuthMethod::None)
             .finalize();
 
         (server, pool)
     }
 
     #[tokio::test]
     async fn config() {
         let (server, _) = setup().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(format!("/taler-observability/config"))
             .await
             .assert_ok_json::<Config>();
         server
             .get(format!("/taler-observability/metrics"))
             .await
             .assert_ok();
     }
 
     #[tokio::test]
     async fn transfer() {
         let (server, _) = setup().await;
         transfer_routine(
             &server.prefix("/taler-wire-gateway"),
             TransferState::pending,
             &payto("payto://iban/HU02162000031000164800000000?receiver-name=name"),
         )
         .await;
     }
 
     static CODE: AtomicU64 = AtomicU64::new(0);
 
     async fn r#in(db: &PgPool, subject: Option<IncomingKey>) {
         db::register_tx_in(
             &mut db.acquire().await.unwrap(),
             &TxIn {
                 code: CODE.fetch_add(1, Ordering::Relaxed),
                 amount: amount("EUR:10"),
                 subject: "subject".into(),
                 debtor: magnet_payto(
                     "payto://iban/HU30162000031000163100000000?receiver-name=name",
                 ),
                 value_date: Zoned::now().date(),
                 status: TxStatus::Completed,
             },
             &subject,
             &Timestamp::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) {
         db::register_tx_out(
             &mut db.acquire().await.unwrap(),
             &db::TxOut {
                 code: CODE.fetch_add(1, Ordering::Relaxed),
                 amount: amount("EUR:10"),
                 subject: "subject".into(),
                 creditor: PAYTO.clone(),
                 value_date: Zoned::now().date(),
                 status: TxStatus::Completed,
             },
             kind,
             &Timestamp::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) as u32)).await
     }
 
     async fn out_malformed(db: &PgPool) {
         out(db, &TxOutKind::Simple).await
     }
 
     #[tokio::test]
     async fn outgoing_history() {
         let (server, db) = &setup().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;
     }
 
     #[tokio::test]
     async fn admin_add_incoming() {
         let (server, _) = setup().await;
         admin_add_incoming_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-prepared-transfer"),
             &EXCHANGE,
             &EXCHANGE,
         )
         .await;
     }
 
     #[tokio::test]
     async fn in_history() {
         let (server, db) = &setup().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;
     }
 
     #[tokio::test]
     async fn revenue() {
         let (server, db) = &setup().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, initiated_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()
     }
 
     #[tokio::test]
     async fn registration() {
         let (server, pool) = setup().await;
         registration_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-prepared-transfer"),
             &EXCHANGE,
             &EXCHANGE,
             &UNKNOWN,
             || check_in(&pool),
         )
         .await;
     }
 }