taler-rust

api.rs (8169B)

---

 /*
   This file is part of TALER
   Copyright (C) 2025, 2026 Taler Systems SA
 
   TALER is free software; you can redistribute it and/or modify it under the
   terms of the GNU Affero General Public License as published by the Free Software
   Foundation; either version 3, or (at your option) any later version.
 
   TALER is distributed in the hope that it will be useful, but WITHOUT ANY
   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
   A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more details.
 
   You should have received a copy of the GNU Affero General Public License along with
   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
 */
 
 use compact_str::CompactString;
 use jiff::Timestamp;
 use taler_api::{
     api::{TalerApi, revenue::Revenue, wire::WireGateway},
     error::{ApiResult, failure},
     subject::IncomingSubject,
 };
 use taler_common::{
     api_common::{SafeU64, safe_u64},
     api_params::{History, Page},
     api_revenue::RevenueIncomingHistory,
     api_wire::{
         AddIncomingRequest, AddIncomingResponse, AddKycauthRequest, AddKycauthResponse,
         IncomingHistory, OutgoingHistory, TransferList, TransferRequest, TransferResponse,
         TransferState, TransferStatus,
     },
     error_code::ErrorCode,
     types::{amount::Currency, payto::PaytoURI},
 };
 use tokio::sync::watch::Sender;
 
 use crate::{
     db::{self, AddIncomingResult, Transfer, TxInAdmin},
     payto::FullCyclosPayto,
 };
 
 pub struct CyclosApi {
     pub pool: sqlx::PgPool,
     pub currency: Currency,
     pub payto: PaytoURI,
     pub in_channel: Sender<i64>,
     pub taler_in_channel: Sender<i64>,
     pub out_channel: Sender<i64>,
     pub taler_out_channel: Sender<i64>,
     pub root: CompactString,
 }
 
 impl CyclosApi {
     pub async fn start(
         pool: sqlx::PgPool,
         root: CompactString,
         payto: PaytoURI,
         currency: Currency,
     ) -> Self {
         let in_channel = Sender::new(0);
         let taler_in_channel = Sender::new(0);
         let out_channel = Sender::new(0);
         let taler_out_channel = Sender::new(0);
         let tmp = Self {
             pool: pool.clone(),
             payto,
             currency,
             root,
             in_channel: in_channel.clone(),
             taler_in_channel: taler_in_channel.clone(),
             out_channel: out_channel.clone(),
             taler_out_channel: taler_out_channel.clone(),
         };
         tokio::spawn(db::notification_listener(
             pool,
             in_channel,
             taler_in_channel,
             out_channel,
             taler_out_channel,
         ));
         tmp
     }
 }
 
 impl TalerApi for CyclosApi {
     fn currency(&self) -> &str {
         self.currency.as_ref()
     }
 
     fn implementation(&self) -> &'static str {
         "urn:net:taler:specs:taler-cyclos:taler-rust"
     }
 }
 
 impl WireGateway for CyclosApi {
     async fn transfer(&self, req: TransferRequest) -> ApiResult<TransferResponse> {
         let creditor = FullCyclosPayto::try_from(&req.credit_account)?;
         let result = db::make_transfer(
             &self.pool,
             &Transfer {
                 request_uid: req.request_uid,
                 amount: req.amount.decimal(),
                 exchange_base_url: req.exchange_base_url,
                 wtid: req.wtid,
                 creditor_id: *creditor.id,
                 creditor_name: creditor.name,
             },
             &Timestamp::now(),
         )
         .await?;
         match result {
             db::TransferResult::Success { id, initiated_at } => Ok(TransferResponse {
                 timestamp: initiated_at.into(),
                 row_id: SafeU64::try_from(id).unwrap(),
             }),
             db::TransferResult::RequestUidReuse => Err(failure(
                 ErrorCode::BANK_TRANSFER_REQUEST_UID_REUSED,
                 "request_uid used already",
             )),
             db::TransferResult::WtidReuse => Err(failure(
                 ErrorCode::BANK_TRANSFER_WTID_REUSED,
                 "wtid used already",
             )),
         }
     }
 
     async fn transfer_page(
         &self,
         page: Page,
         status: Option<TransferState>,
     ) -> ApiResult<TransferList> {
         Ok(TransferList {
             transfers: db::transfer_page(&self.pool, &status, &self.currency, &self.root, &page)
                 .await?,
             debit_account: self.payto.clone(),
         })
     }
 
     async fn transfer_by_id(&self, id: u64) -> ApiResult<Option<TransferStatus>> {
         Ok(db::transfer_by_id(&self.pool, id, &self.currency, &self.root).await?)
     }
 
     async fn outgoing_history(&self, params: History) -> ApiResult<OutgoingHistory> {
         Ok(OutgoingHistory {
             outgoing_transactions: db::outgoing_history(
                 &self.pool,
                 &params,
                 &self.currency,
                 &self.root,
                 || self.taler_out_channel.subscribe(),
             )
             .await?,
             debit_account: self.payto.clone(),
         })
     }
 
     async fn incoming_history(&self, params: History) -> ApiResult<IncomingHistory> {
         Ok(IncomingHistory {
             incoming_transactions: db::incoming_history(
                 &self.pool,
                 &params,
                 &self.currency,
                 &self.root,
                 || self.taler_in_channel.subscribe(),
             )
             .await?,
             credit_account: self.payto.clone(),
         })
     }
 
     async fn add_incoming_reserve(
         &self,
         req: AddIncomingRequest,
     ) -> ApiResult<AddIncomingResponse> {
         let debtor = FullCyclosPayto::try_from(&req.debit_account)?;
         let res = db::register_tx_in_admin(
             &self.pool,
             &TxInAdmin {
                 amount: req.amount.decimal(),
                 subject: format!("Admin incoming {}", req.reserve_pub),
                 debtor_id: *debtor.id,
                 debtor_name: debtor.name,
                 metadata: IncomingSubject::Reserve(req.reserve_pub),
             },
             &Timestamp::now(),
         )
         .await?;
         match res {
             AddIncomingResult::Success {
                 row_id, valued_at, ..
             } => Ok(AddIncomingResponse {
                 row_id: safe_u64(row_id),
                 timestamp: valued_at.into(),
             }),
             AddIncomingResult::ReservePubReuse => Err(failure(
                 ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT,
                 "reserve_pub used already".to_owned(),
             )),
         }
     }
 
     async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddKycauthResponse> {
         let debtor = FullCyclosPayto::try_from(&req.debit_account)?;
         let res = db::register_tx_in_admin(
             &self.pool,
             &TxInAdmin {
                 amount: req.amount.decimal(),
                 subject: format!("Admin incoming KYC:{}", req.account_pub),
                 debtor_id: *debtor.id,
                 debtor_name: debtor.name,
                 metadata: IncomingSubject::Kyc(req.account_pub),
             },
             &Timestamp::now(),
         )
         .await?;
         match res {
             AddIncomingResult::Success {
                 row_id, valued_at, ..
             } => Ok(AddKycauthResponse {
                 row_id: safe_u64(row_id),
                 timestamp: valued_at.into(),
             }),
             AddIncomingResult::ReservePubReuse => Err(failure(
                 ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT,
                 "reserve_pub used already".to_owned(),
             )),
         }
     }
 
     fn support_account_check(&self) -> bool {
         false
     }
 }
 
 impl Revenue for CyclosApi {
     async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> {
         Ok(RevenueIncomingHistory {
             incoming_transactions: db::revenue_history(
                 &self.pool,
                 &params,
                 &self.currency,
                 &self.root,
                 || self.in_channel.subscribe(),
             )
             .await?,
             credit_account: self.payto.clone(),
         })
     }
 }