api.rs - taler-rust - GNU Taler code in Rust. Largely core banking integrations.

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