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

api.rs (21733B)
      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         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         Serve,
    414         api::TalerRouter as _,
    415         config::{ApiCfg, AuthCfg},
    416         db::TypeHelper as _,
    417         subject::{IncomingKey, OutgoingSubject},
    418     };
    419     use taler_common::{
    420         api::{
    421             EddsaPublicKey,
    422             observability::Config,
    423             prepared::PreparedTransferConfig,
    424             revenue::RevenueConfig,
    425             wire::{TransferState, WireConfig},
    426         },
    427         db::IncomingType,
    428         types::{
    429             amount::amount,
    430             payto::{PaytoURI, payto},
    431         },
    432     };
    433     use taler_macros::db_test;
    434     use taler_test_utils::{
    435         Router,
    436         routine::{
    437             Status, admin_add_incoming_routine, in_history_routine, out_history_routine,
    438             registration_routine, revenue_routine, transfer_routine,
    439         },
    440         server::TestServer,
    441         tasks,
    442     };
    443 
    444     use crate::{
    445         FullHuPayto, build_router,
    446         config::ServeCfg,
    447         db::{self, TxIn, TxOutKind},
    448         magnet_api::types::TxStatus,
    449         magnet_payto,
    450     };
    451 
    452     static PAYTO: LazyLock<FullHuPayto> = LazyLock::new(|| {
    453         magnet_payto("payto://iban/HU02162000031000164800000000?receiver-name=Smith")
    454     });
    455     static EXCHANGE: LazyLock<PaytoURI> = LazyLock::new(|| PAYTO.as_uri());
    456     static UNKNOWN: LazyLock<PaytoURI> =
    457         LazyLock::new(|| payto("payto://iban/HU60162006491000639900000000?receiver-name=Unknown"));
    458 
    459     async fn setup(pool: &PgPool) -> Router {
    460         let cfg = ServeCfg {
    461             payto: PAYTO.clone(),
    462             serve: Serve::Systemd,
    463             wire_gateway: Some(ApiCfg {
    464                 auth: AuthCfg::None,
    465             }),
    466             revenue: Some(ApiCfg {
    467                 auth: AuthCfg::None,
    468             }),
    469             observability: Some(ApiCfg {
    470                 auth: AuthCfg::None,
    471             }),
    472         };
    473         build_router(&cfg, pool.clone()).await.finalize()
    474     }
    475 
    476     #[db_test]
    477     async fn config(pool: PgPool) {
    478         let server = setup(&pool).await;
    479         server
    480             .get("/taler-wire-gateway/config")
    481             .await
    482             .assert_ok_json::<WireConfig>();
    483         server
    484             .get("/taler-prepared-transfer/config")
    485             .await
    486             .assert_ok_json::<PreparedTransferConfig>();
    487         server
    488             .get("/taler-revenue/config")
    489             .await
    490             .assert_ok_json::<RevenueConfig>();
    491         server
    492             .get("/taler-observability/config")
    493             .await
    494             .assert_ok_json::<Config>();
    495         server.get("/taler-observability/metrics").await.assert_ok();
    496     }
    497 
    498     #[db_test]
    499     async fn transfer(pool: PgPool) {
    500         let server = setup(&pool).await;
    501         transfer_routine(
    502             &server.prefix("/taler-wire-gateway"),
    503             TransferState::pending,
    504             &payto("payto://iban/HU02162000031000164800000000?receiver-name=name"),
    505         )
    506         .await;
    507     }
    508 
    509     static CODE: AtomicU64 = AtomicU64::new(0);
    510 
    511     async fn r#in(db: &PgPool, subject: Option<IncomingKey>) {
    512         db::register_tx_in(
    513             &mut db.acquire().await.unwrap(),
    514             &TxIn {
    515                 code: CODE.fetch_add(1, Ordering::Relaxed),
    516                 amount: amount("EUR:10"),
    517                 subject: "subject".into(),
    518                 debtor: magnet_payto(
    519                     "payto://iban/HU30162000031000163100000000?receiver-name=name",
    520                 ),
    521                 value_date: Zoned::now().date(),
    522                 status: TxStatus::Completed,
    523             },
    524             &subject,
    525             &Timestamp::now(),
    526         )
    527         .await
    528         .unwrap();
    529     }
    530 
    531     async fn in_malformed(db: &PgPool) {
    532         r#in(db, None).await
    533     }
    534 
    535     async fn in_talerable(db: &PgPool) {
    536         r#in(db, Some(IncomingKey::reserve(EddsaPublicKey::rand()))).await
    537     }
    538 
    539     async fn out(db: &PgPool, kind: &TxOutKind) {
    540         db::register_tx_out(
    541             &mut db.acquire().await.unwrap(),
    542             &db::TxOut {
    543                 code: CODE.fetch_add(1, Ordering::Relaxed),
    544                 amount: amount("EUR:10"),
    545                 subject: "subject".into(),
    546                 creditor: PAYTO.clone(),
    547                 value_date: Zoned::now().date(),
    548                 status: TxStatus::Completed,
    549             },
    550             kind,
    551             &Timestamp::now(),
    552         )
    553         .await
    554         .unwrap();
    555     }
    556 
    557     async fn out_talerable(db: &PgPool) {
    558         out(db, &TxOutKind::Talerable(OutgoingSubject::rand())).await
    559     }
    560 
    561     async fn out_bounce(db: &PgPool) {
    562         out(db, &TxOutKind::Bounce(CODE.load(Ordering::Relaxed) as u32)).await
    563     }
    564 
    565     async fn out_malformed(db: &PgPool) {
    566         out(db, &TxOutKind::Simple).await
    567     }
    568 
    569     #[db_test]
    570     async fn outgoing_history(db: PgPool) {
    571         let db = &db;
    572         let server = setup(db).await;
    573         out_history_routine(
    574             &server.prefix("/taler-wire-gateway"),
    575             tasks!({ out_talerable(db).await }),
    576             tasks!(
    577                 { out_bounce(db).await },
    578                 { out_malformed(db).await },
    579                 { in_malformed(db).await },
    580                 { in_talerable(db).await }
    581             ),
    582         )
    583         .await;
    584     }
    585 
    586     #[db_test]
    587     async fn admin_add_incoming(db: PgPool) {
    588         let server = setup(&db).await;
    589         admin_add_incoming_routine(
    590             &server.prefix("/taler-wire-gateway"),
    591             &server.prefix("/taler-prepared-transfer"),
    592             &EXCHANGE,
    593             &EXCHANGE,
    594         )
    595         .await;
    596     }
    597 
    598     #[db_test]
    599     async fn in_history(db: PgPool) {
    600         let db = &db;
    601         let server = setup(db).await;
    602         in_history_routine(
    603             &server.prefix("/taler-wire-gateway"),
    604             &server.prefix("/taler-prepared-transfer"),
    605             &EXCHANGE,
    606             &EXCHANGE,
    607             tasks!({ in_talerable(db).await }),
    608             tasks!(
    609                 { out_malformed(db).await },
    610                 { out_talerable(db).await },
    611                 { out_bounce(db).await },
    612                 { in_malformed(db).await }
    613             ),
    614         )
    615         .await;
    616     }
    617 
    618     #[db_test]
    619     async fn revenue(db: PgPool) {
    620         let db = &db;
    621         let server = setup(db).await;
    622         revenue_routine(
    623             &server.prefix("/taler-wire-gateway"),
    624             &server.prefix("/taler-revenue"),
    625             &EXCHANGE,
    626             tasks!({ in_malformed(db).await }, { in_talerable(db).await },),
    627             tasks!({ out_malformed(db).await }, { out_talerable(db).await }, {
    628                 out_bounce(db).await
    629             }),
    630         )
    631         .await;
    632     }
    633 
    634     async fn check_in(pool: &PgPool) -> Vec<Status> {
    635         sqlx::query(
    636             "
    637             SELECT pending_recurrent_in.authorization_pub IS NOT NULL, initiated_id IS NOT NULL, type, metadata
    638             FROM tx_in
    639                 LEFT JOIN taler_in USING (tx_in_id)
    640                 LEFT JOIN pending_recurrent_in USING (tx_in_id)
    641                 LEFT JOIN bounced USING (tx_in_id)
    642             ORDER BY tx_in.tx_in_id
    643         ",
    644         )
    645         .try_map(|r: PgRow| {
    646             Ok(
    647                 if r.try_get_flag(0)? {
    648                     Status::Pending
    649                 } else if r.try_get_flag(1)? {
    650                     Status::Bounced
    651                 } else {
    652                     match r.try_get(2)? {
    653                         None => Status::Simple,
    654                         Some(IncomingType::reserve) => Status::Reserve(r.try_get(3)?),
    655                         Some(IncomingType::kyc) => Status::Kyc(r.try_get(3)?),
    656                         Some(e) => unreachable!("{e:?}")
    657                     }
    658                 }
    659             )
    660         })
    661         .fetch_all(pool)
    662         .await
    663         .unwrap()
    664     }
    665 
    666     #[db_test]
    667     async fn registration(db: PgPool) {
    668         let server = setup(&db).await;
    669         registration_routine(
    670             &server.prefix("/taler-wire-gateway"),
    671             &server.prefix("/taler-prepared-transfer"),
    672             &EXCHANGE,
    673             &EXCHANGE,
    674             &UNKNOWN,
    675             || check_in(&db),
    676         )
    677         .await;
    678     }
    679 }
	taler-rust GNU Taler code in Rust. Largely core banking integrations.
	Log \| Files \| Refs \| Submodules \| README \| LICENSE