libeufin

api.rs (17451B)

---

 /*
 * This file is part of LibEuFin.
 * Copyright (C) 2026 Taler Systems S.A.
 
 * LibEuFin 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.
 
 * LibEuFin 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 LibEuFin; see the file COPYING.  If not, see
 * <http://www.gnu.org/licenses/>
 */
 
 use const_format::formatcp;
 use jiff::Timestamp;
 use libeufin_ebics::{
     ebics::{TaskStatus, rand_ebics_id},
     iso20022::model::{InId, InTx},
 };
 use prometheus_client::{
     encoding::EncodeLabelSet,
     metrics::{family::Family, gauge::Gauge},
     registry::{Registry, Unit},
 };
 use sqlx::{PgPool, types::Json};
 use taler_api::{
     api::{
         TalerApi,
         observability::Observability,
         prepared::{PreparedTransfer, simple_subject},
         revenue::Revenue,
         wire::WireGateway,
     },
     error::{ApiResult, failure_code},
     subject::{IncomingKey, fmt_in_subject, subject_fmt_qr_bill},
 };
 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,
         },
     },
     error_code::ErrorCode,
     types::{
         amount::{Amount, Currency},
         iban::IBAN,
         payto::{FullIbanPayto, IbanPayto, PaytoURI},
         time::TalerTimestamp,
     },
 };
 use tokio::sync::watch::Sender;
 
 use crate::{
     constants::{FETCH_TASK_KEY, SUBMIT_TASK_KEY},
     db::{
         self,
         exchange::{
             TransferResult, incoming_history, outgoing_history, revenue_history, transfer,
             transfer_by_id, transfer_page,
         },
         payment::{IncomingRegistrationResult, register_in_talerable},
         transfer::{RegistrationResult, transfer_register, transfer_unregister},
     },
 };
 
 pub struct NexusApi {
     pub pool: sqlx::PgPool,
     pub currency: Currency,
     pub payto: FullIbanPayto,
     pub qr_iban: Option<IBAN>,
     pub in_channel: Sender<i64>,
     pub taler_in_channel: Sender<i64>,
     pub taler_out_channel: Sender<i64>,
     metrics: Metrics,
     registry: Registry,
 }
 
 #[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
 struct TaskLabel {
     name: &'static str,
 }
 
 #[derive(Default)]
 pub struct Metrics {
     db_access: Gauge,
     task_execution: Family<TaskLabel, Gauge>,
     task_success: Family<TaskLabel, 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.register_with_unit(
             "task_execution_timestamp_seconds",
             "Unix timestamp of the last task execution",
             Unit::Seconds,
             self.task_execution.clone(),
         );
 
         registry.register_with_unit(
             "task_success_timestamp_seconds",
             "Unix timestamp of the last successful task execution",
             Unit::Seconds,
             self.task_success.clone(),
         );
 
         registry
     }
 
     pub async fn sync(&self, db: &PgPool) {
         let success =
             match sqlx::query_as::<_, (Option<Json<TaskStatus>>, Option<Json<TaskStatus>>)>(
                 formatcp!(
                     "
             SELECT
             (SELECT value FROM kv WHERE key='{SUBMIT_TASK_KEY}'),
             (SELECT value FROM kv WHERE key='{FETCH_TASK_KEY}')
         "
                 ),
             )
             .fetch_one(db)
             .await
             {
                 Ok((submit, fetch)) => {
                     for (name, status) in [("submit", submit), ("fetch", fetch)] {
                         if let Some(status) = status {
                             if let Some(time) = status.last_trial {
                                 self.task_execution
                                     .get_or_create(&TaskLabel { name })
                                     .set(time.as_second());
                             }
                             if let Some(time) = status.last_successfull {
                                 self.task_success
                                     .get_or_create(&TaskLabel { name })
                                     .set(time.as_second());
                             }
                         }
                     }
                     true
                 }
                 Err(_) => false,
             };
         self.db_access.set(if success { 1 } else { 0 });
     }
 }
 
 impl NexusApi {
     pub async fn start(
         pool: sqlx::PgPool,
         currency: Currency,
         payto: FullIbanPayto,
         qr_iban: Option<IBAN>,
     ) -> Self {
         let in_channel = Sender::new(0);
         let taler_in_channel = Sender::new(0);
         let taler_out_channel = Sender::new(0);
 
         let metrics = Metrics::default();
 
         let tmp = Self {
             pool: pool.clone(),
             payto,
             qr_iban,
             currency,
             in_channel: in_channel.clone(),
             taler_in_channel: taler_in_channel.clone(),
             taler_out_channel: taler_out_channel.clone(),
             registry: metrics.registry(),
             metrics,
         };
         tokio::spawn(db::notification_listener(
             pool,
             in_channel,
             taler_in_channel,
             taler_out_channel,
         ));
         tmp
     }
 }
 
 impl TalerApi for NexusApi {
     fn currency(&self) -> Currency {
         self.currency
     }
 
     fn implementation(&self) -> &'static str {
         "urn:net:taler:specs:libeufin-nexus:taler-rust"
     }
 }
 
 async fn add_incoming(
     db: &PgPool,
     subject: &IncomingKey,
     amount: Amount,
     debit_account: PaytoURI,
 ) -> ApiResult<AddIncomingResponse> {
     FullIbanPayto::try_from(&debit_account)?;
     let now = Timestamp::now();
     match register_in_talerable(
         db,
         &InTx {
             id: InId {
                 uetr: None,
                 tx_id: Some(rand_ebics_id()),
                 sref: None,
             },
             amount,
             credit_fee: Amount::zero(&amount.currency),
             subject: Some(format!(
                 "Manual incoming {}",
                 fmt_in_subject(subject.ty, &subject.key)
             )),
             execution_time: now,
             debtor: Some(debit_account),
         },
         subject,
     )
     .await?
     {
         IncomingRegistrationResult::Success(in_result) => Ok(AddIncomingResponse {
             row_id: in_result.id,
             timestamp: now.into(),
         }),
         IncomingRegistrationResult::ReservePubReuse => {
             Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT))
         }
         IncomingRegistrationResult::MappingReuse => {
             Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_REUSED))
         }
         IncomingRegistrationResult::UnknownMapping => {
             Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_UNKNOWN))
         }
     }
 }
 
 impl WireGateway for NexusApi {
     async fn transfer(&self, req: TransferRequest) -> ApiResult<TransferResponse> {
         FullIbanPayto::try_from(&req.credit_account)?;
         let result = transfer(&self.pool, &req, &rand_ebics_id(), &Timestamp::now()).await?;
         match result {
             TransferResult::Success { id, timestamp } => Ok(TransferResponse {
                 timestamp: timestamp.into(),
                 row_id: id,
             }),
             TransferResult::RequestUidReuse => {
                 Err(failure_code(ErrorCode::BANK_TRANSFER_REQUEST_UID_REUSED))
             }
             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: transfer_page(&self.pool, &self.currency, &page, &status).await?,
             debit_account: self.payto.as_uri(),
         })
     }
 
     async fn transfer_by_id(&self, id: u64) -> ApiResult<Option<TransferStatus>> {
         Ok(transfer_by_id(&self.pool, &self.currency, id).await?)
     }
 
     async fn outgoing_history(&self, params: History) -> ApiResult<OutgoingHistory> {
         Ok(OutgoingHistory {
             outgoing_transactions: outgoing_history(&self.pool, &self.currency, &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: incoming_history(&self.pool, &self.currency, &params, || {
                 self.taler_in_channel.subscribe()
             })
             .await?,
             credit_account: self.payto.as_uri(),
         })
     }
 
     async fn add_incoming_reserve(
         &self,
         req: AddIncomingRequest,
     ) -> ApiResult<AddIncomingResponse> {
         add_incoming(
             &self.pool,
             &IncomingKey::reserve(req.reserve_pub),
             req.amount,
             req.debit_account,
         )
         .await
     }
 
     async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddIncomingResponse> {
         add_incoming(
             &self.pool,
             &IncomingKey::kyc(req.account_pub),
             req.amount,
             req.debit_account,
         )
         .await
     }
 
     async fn add_incoming_mapped(&self, req: AddMappedRequest) -> ApiResult<AddIncomingResponse> {
         add_incoming(
             &self.pool,
             &IncomingKey::map(req.authorization_pub),
             req.amount,
             req.debit_account,
         )
         .await
     }
 
     fn support_account_check(&self) -> bool {
         false
     }
 }
 
 impl Revenue for NexusApi {
     async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> {
         Ok(RevenueIncomingHistory {
             incoming_transactions: revenue_history(&self.pool, &self.currency, &params, || {
                 self.in_channel.subscribe()
             })
             .await?,
             credit_account: self.payto.as_uri(),
         })
     }
 }
 
 impl PreparedTransfer for NexusApi {
     fn supported_formats(&self) -> &[SubjectFormat] {
         if self.qr_iban.is_some() {
             &[SubjectFormat::SIMPLE, SubjectFormat::CH_QR_BILL]
         } else {
             &[SubjectFormat::SIMPLE]
         }
     }
 
     async fn registration(&self, req: RegistrationRequest) -> ApiResult<RegistrationResponse> {
         let creditor = IbanPayto::try_from(&req.credit_account)?;
         let reference_number = if creditor.iban == self.payto.iban {
             None
         } else if Some(creditor.iban) == self.qr_iban {
             Some(subject_fmt_qr_bill(req.authorization_pub.as_ref()))
         } else {
             return Err(failure_code(ErrorCode::BANK_UNKNOWN_CREDITOR));
         };
         match transfer_register(
             &self.pool,
             req.r#type.into(),
             &req.account_pub,
             &req.authorization_pub,
             &req.authorization_sig,
             req.recurrent,
             reference_number.as_deref(),
             &Timestamp::now(),
         )
         .await?
         {
             RegistrationResult::Success => ApiResult::Ok(RegistrationResponse {
                 subjects: vec![if let Some(qr_reference_number) = reference_number {
                     TransferSubject::QrBill {
                         credit_amount: req.credit_amount,
                         qr_reference_number,
                     }
                 } else {
                     simple_subject(req)
                 }],
                 expiration: TalerTimestamp::Never,
             }),
             RegistrationResult::ReservePubReuse => {
                 ApiResult::Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT))
             }
             RegistrationResult::SubjectReuse => {
                 ApiResult::Err(failure_code(ErrorCode::BANK_DERIVATION_REUSE))
             }
         }
     }
 
     async fn unregistration(&self, req: Unregistration) -> ApiResult<bool> {
         Ok(transfer_unregister(&self.pool, &req.authorization_pub, &Timestamp::now()).await?)
     }
 }
 
 impl Observability for NexusApi {
     async fn metrics(&self) -> ApiResult<&Registry> {
         self.metrics.sync(&self.pool).await;
         Ok(&self.registry)
     }
 }
 
 #[cfg(test)]
 pub mod test {
     use std::sync::Arc;
 
     use sqlx::PgPool;
     use taler_api::{api::TalerRouter as _, auth::AuthMethod, subject::OutgoingSubject};
     use taler_common::api::{
         prepared::PreparedTransferConfig,
         revenue::RevenueConfig,
         wire::{TransferState, WireConfig},
     };
     use taler_macros::db_test;
     use taler_test_utils::{
         Router,
         routine::{
             admin_add_incoming_routine, in_history_routine, out_history_routine,
             registration_routine, revenue_routine, transfer_routine,
         },
         server::TestServer as _,
         tasks,
     };
 
     use crate::{
         api::NexusApi,
         db::{payment::register_out_tx, test::check_in},
         test::{CLIENT, CURR, EXCHANGE, UNKNOWN, gen_out_pay},
     };
 
     pub async fn api_setup(db: &PgPool) -> Router {
         let api = Arc::new(NexusApi::start(db.clone(), CURR, EXCHANGE.clone(), None).await);
         Router::new()
             .wire_gateway(api.clone(), AuthMethod::None)
             .prepared_transfer(api.clone())
             .revenue(api, AuthMethod::None)
             .finalize()
     }
 
     #[db_test]
     async fn config(db: PgPool) {
         let server = api_setup(&db).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>();
     }
 
     #[db_test]
     async fn transfer(db: PgPool) {
         let server = api_setup(&db).await;
         transfer_routine(
             &server.prefix("/taler-wire-gateway"),
             TransferState::pending,
             &EXCHANGE.as_uri(),
         )
         .await;
         // TODO
         /*db.initiated.batchSubmissionSuccess(1, Instant.now(), "ORDER1")
         db.initiated.batchSubmissionFailure(2, Instant.now(), "Failure")
         db.initiated.batchSubmissionFailure(3, Instant.now(), "Failure")
         client.getA("/taler-wire-gateway/transfers?status=transient_failure").assertOkJson<TransferList> {
             assertEquals(2, it.transfers.size)
         }
         client.getA("/taler-wire-gateway/transfers?status=pending").assertOkJson<TransferList> {
             assertEquals(4, it.transfers.size)
         }*/
     }
 
     #[db_test]
     async fn outgoing_history(db: PgPool) {
         let server = api_setup(&db).await;
         out_history_routine(
             &server.prefix("/taler-wire-gateway"),
             tasks!({
                 register_out_tx(&db, &gen_out_pay("subject"), Some(&OutgoingSubject::rand()))
                     .await
                     .unwrap();
             }),
             tasks!(),
         )
         .await;
     }
 
     #[db_test]
     async fn incoming_history(db: PgPool) {
         let server = api_setup(&db).await;
         in_history_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-prepared-transfer"),
             &CLIENT.as_uri(),
             &EXCHANGE.as_uri(),
             tasks!(),
             tasks!(),
         )
         .await;
     }
 
     #[db_test]
     async fn admin_add_incoming(db: PgPool) {
         let server = api_setup(&db).await;
         admin_add_incoming_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-prepared-transfer"),
             &CLIENT.as_uri(),
             &EXCHANGE.as_uri(),
         )
         .await;
     }
 
     #[db_test]
     async fn revenue(db: PgPool) {
         let server = api_setup(&db).await;
         revenue_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-revenue"),
             &CLIENT.as_uri(),
             tasks!(),
             tasks!(),
         )
         .await;
     }
 
     #[db_test]
     async fn registration(db: PgPool) {
         let server = api_setup(&db).await;
         registration_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-prepared-transfer"),
             &CLIENT.as_uri(),
             &EXCHANGE.as_uri(),
             &UNKNOWN,
             || check_in(&db),
         )
         .await;
     }
 }