depolymerization

api.rs (18647B)

---

 /*
   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 std::sync::{
     Arc,
     atomic::{AtomicBool, Ordering},
 };
 
 use axum::{
     extract::{Request, State},
     http::StatusCode,
     middleware::Next,
     response::{IntoResponse as _, Response},
 };
 use jiff::Timestamp;
 use prometheus_client::{metrics::gauge::Gauge, registry::Registry};
 use sqlx::{PgPool, postgres::PgListener};
 use taler_api::{
     api::{
         TalerApi,
         observability::Observability,
         prepared::{PreparedTransfer, simple_subject},
         revenue::Revenue,
         wire::WireGateway,
     },
     error::{ApiResult, failure_code, failure_status},
     subject::IncomingKey,
 };
 use taler_common::{
     ExpoBackoffDecorr,
     api::{
         params::{History, Page},
         prepared::{RegistrationRequest, RegistrationResponse, SubjectFormat, Unregistration},
         revenue::RevenueIncomingHistory,
         wire::{
             AddIncomingRequest, AddIncomingResponse, AddKycauthRequest, AddMappedRequest,
             IncomingHistory, OutgoingHistory, TransferList, TransferRequest, TransferResponse,
             TransferState, TransferStatus,
         },
     },
     error_code::ErrorCode,
     types::{
         amount::{Amount, Currency},
         payto::PaytoURI,
         time::TalerTimestamp,
     },
 };
 use tokio::{sync::watch::Sender, time::sleep};
 use tracing::{debug, error, warn};
 
 use crate::{
     db::{
         self, AddIncomingResult, RegistrationResult, TransferResult, get_status,
         register_tx_in_admin, revenue_history, transfer, transfer_register, transfer_unregister,
     },
     payto::{BtcPayto, FullBtcPayto},
 };
 
 pub struct ServerState {
     pool: PgPool,
     payto: FullBtcPayto,
     currency: Currency,
     status: AtomicBool,
     in_channel: Sender<i64>,
     taler_in_channel: Sender<i64>,
     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 });
     }
 }
 
 pub async fn notification_listener(
     pool: PgPool,
     in_channel: Sender<i64>,
     taler_in_channel: Sender<i64>,
     taler_out_channel: Sender<i64>,
 ) -> sqlx::Result<()> {
     taler_api::notification::notification_listener!(&pool,
         "tx_in" => (row_id: i64) {
             in_channel.send_replace(row_id);
         },
         "taler_in" => (row_id: i64) {
             taler_in_channel.send_replace(row_id);
         },
         "taler_out" => (row_id: i64) {
             taler_out_channel.send_replace(row_id);
         }
     )
 }
 
 impl ServerState {
     pub async fn start(pool: sqlx::PgPool, payto: FullBtcPayto, currency: Currency) -> Arc<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,
             currency,
             status: AtomicBool::new(false),
             in_channel: in_channel.clone(),
             taler_in_channel: taler_in_channel.clone(),
             taler_out_channel: taler_out_channel.clone(),
             registry: metrics.registry(),
             metrics,
         };
         let state = Arc::new(tmp);
         tokio::spawn(status_watcher(state.clone()));
         tokio::spawn(notification_listener(
             pool,
             in_channel,
             taler_in_channel,
             taler_out_channel,
         ));
         state
     }
 }
 
 impl TalerApi for ServerState {
     fn currency(&self) -> Currency {
         self.currency
     }
 
     fn implementation(&self) -> &'static str {
         "urn:net:taler:specs:depolymerizer-bitcoin:depolymerization"
     }
 }
 
 async fn add_incoming(
     db: &PgPool,
     amount: Amount,
     debit_account: PaytoURI,
     subject: &IncomingKey,
 ) -> ApiResult<AddIncomingResponse> {
     let debtor = FullBtcPayto::try_from(&debit_account)?;
     match register_tx_in_admin(db, &amount, &debtor.0, &Timestamp::now(), subject).await? {
         AddIncomingResult::Success {
             row_id, valued_at, ..
         } => Ok(AddIncomingResponse {
             row_id,
             timestamp: valued_at.into(),
         }),
         AddIncomingResult::ReservePubReuse => {
             Err(failure_code(ErrorCode::BANK_DUPLICATE_RESERVE_PUB_SUBJECT))
         }
         AddIncomingResult::MappingReuse => {
             Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_REUSED))
         }
         AddIncomingResult::UnknownMapping => {
             Err(failure_code(ErrorCode::BANK_TRANSFER_MAPPING_UNKNOWN))
         }
     }
 }
 
 impl WireGateway for ServerState {
     async fn transfer(&self, req: TransferRequest) -> ApiResult<TransferResponse> {
         let creditor = FullBtcPayto::try_from(&req.credit_account)?;
         match transfer(&self.pool, &creditor, &req).await? {
             TransferResult::Success(transfer_response) => Ok(transfer_response),
             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,
         params: Page,
         status: Option<TransferState>,
     ) -> ApiResult<TransferList> {
         let transfers = db::transfer_page(&self.pool, &status, &params, &self.currency).await?;
         Ok(TransferList {
             transfers,
             debit_account: self.payto.as_uri(),
         })
     }
 
     async fn transfer_by_id(&self, id: u64) -> ApiResult<Option<TransferStatus>> {
         let status = db::transfer_by_id(&self.pool, id, &self.currency).await?;
         Ok(status)
     }
 
     async fn outgoing_history(&self, params: History) -> ApiResult<OutgoingHistory> {
         let outgoing_transactions =
             db::outgoing_history(&self.pool, &params, &self.currency, || {
                 self.taler_out_channel.subscribe()
             })
             .await?;
         Ok(OutgoingHistory {
             debit_account: self.payto.as_uri(),
             outgoing_transactions,
         })
     }
 
     async fn incoming_history(&self, params: History) -> ApiResult<IncomingHistory> {
         let incoming_transactions =
             db::incoming_history(&self.pool, &params, &self.currency, || {
                 self.taler_in_channel.subscribe()
             })
             .await?;
         Ok(IncomingHistory {
             credit_account: self.payto.as_uri(),
             incoming_transactions,
         })
     }
 
     async fn add_incoming_reserve(
         &self,
         req: AddIncomingRequest,
     ) -> ApiResult<AddIncomingResponse> {
         add_incoming(
             &self.pool,
             req.amount,
             req.debit_account,
             &IncomingKey::reserve(req.reserve_pub),
         )
         .await
     }
 
     async fn add_incoming_kyc(&self, req: AddKycauthRequest) -> ApiResult<AddIncomingResponse> {
         add_incoming(
             &self.pool,
             req.amount,
             req.debit_account,
             &IncomingKey::kyc(req.account_pub),
         )
         .await
     }
 
     async fn add_incoming_mapped(&self, req: AddMappedRequest) -> ApiResult<AddIncomingResponse> {
         add_incoming(
             &self.pool,
             req.amount,
             req.debit_account,
             &IncomingKey::map(req.authorization_pub),
         )
         .await
     }
 
     fn support_account_check(&self) -> bool {
         // TODO we might be able to check this ?
         false
     }
 }
 
 impl Revenue for ServerState {
     async fn history(&self, params: History) -> ApiResult<RevenueIncomingHistory> {
         Ok(RevenueIncomingHistory {
             incoming_transactions: revenue_history(&self.pool, &params, &self.currency, || {
                 self.in_channel.subscribe()
             })
             .await?,
             credit_account: self.payto.as_uri(),
         })
     }
 }
 
 impl PreparedTransfer for ServerState {
     // TODO bitcoin subject format
     fn supported_formats(&self) -> &[SubjectFormat] {
         &[SubjectFormat::SIMPLE]
     }
 
     async fn registration(&self, req: RegistrationRequest) -> ApiResult<RegistrationResponse> {
         let creditor = BtcPayto::try_from(&req.credit_account)?;
         if creditor.0 != self.payto.0 {
             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,
             &Timestamp::now(),
         )
         .await?
         {
             RegistrationResult::Success => ApiResult::Ok(RegistrationResponse {
                 subjects: vec![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 ServerState {
     async fn metrics(&self) -> ApiResult<&Registry> {
         self.metrics.sync(&self.pool).await;
         Ok(&self.registry)
     }
 }
 
 pub async fn status_middleware(
     State(state): State<Arc<ServerState>>,
     request: Request,
     next: Next,
 ) -> Response {
     if !state.status.load(Ordering::Relaxed) {
         failure_status(
             ErrorCode::GENERIC_INTERNAL_INVARIANT_FAILURE,
             "Currency backing is compromised until the transaction reappear",
             StatusCode::BAD_GATEWAY,
         )
         .into_response()
     } else {
         next.run(request).await
     }
 }
 
 /// Listen to backend status change
 async fn status_watcher(state: Arc<ServerState>) {
     let mut jitter = ExpoBackoffDecorr::default();
     async fn inner(
         state: &ServerState,
         jitter: &mut ExpoBackoffDecorr,
     ) -> Result<(), sqlx::error::Error> {
         let mut listener = PgListener::connect_with(&state.pool).await?;
         listener.listen("status").await?;
         loop {
             // Sync state
             if let Some([status]) = get_status(&state.pool).await? {
                 assert!(status < 2);
                 if status == 1 {
                     debug!(target: "status-watcher", "Worker healthy");
                 } else {
                     debug!(target: "status-watcher", "Worker down");
                 }
                 state.status.store(status == 1, Ordering::SeqCst);
             } else {
                 warn!(target: "status-watcher", "Status not setup");
             }
             // Wait for next notification
             listener.recv().await?;
             jitter.reset();
         }
     }
 
     loop {
         if let Err(err) = inner(&state, &mut jitter).await {
             error!(target: "status-watcher", "{err}");
             sleep(jitter.backoff()).await;
         }
     }
 }
 
 #[cfg(test)]
 pub mod test {
 
     use std::{str::FromStr, sync::LazyLock};
 
     use axum::Router;
     use jiff::Timestamp;
     use sqlx::{PgPool, Row, postgres::PgRow};
     use taler_api::{
         api::TalerRouter as _, auth::AuthMethod, db::TypeHelper, subject::OutgoingSubject,
     };
     use taler_common::{
         api::{
             observability::Config,
             prepared::PreparedTransferConfig,
             revenue::RevenueConfig,
             wire::{TransferState, WireConfig},
         },
         db::IncomingType,
         types::amount::{Currency, amount},
     };
     use taler_test_utils::{
         db::db_test_setup,
         routine::{
             Status, admin_add_incoming_routine, out_history_routine, registration_routine,
             revenue_routine, transfer_routine,
         },
         server::TestServer,
         tasks,
     };
 
     use crate::{
         CONFIG_SOURCE,
         api::ServerState,
         db::{TxOut, TxOutKind, sync_out, test::rand_tx_id},
         payto::FullBtcPayto,
     };
 
     pub static EXCHANGE: LazyLock<FullBtcPayto> = LazyLock::new(|| {
         FullBtcPayto::from_str(
             "payto://bitcoin/1FfmbHfnpaZjKFvyi1okTjJJusN455paPH?receiver-name=Exchange",
         )
         .unwrap()
     });
 
     pub static CLIENT: LazyLock<FullBtcPayto> = LazyLock::new(|| {
         FullBtcPayto::from_str(
             "payto://bitcoin/1FfmbHfnpaZjKFvyi1okTjJJusN455paPH?receiver-name=Anonymous",
         )
         .unwrap()
     });
 
     pub static UNKNOWN: LazyLock<FullBtcPayto> = LazyLock::new(|| {
         FullBtcPayto::from_str(
             "payto://bitcoin/1Q2TWHE3GMdB6BZKafqwxXtWAWgFt5Jvm3?receiver-name=Unknown",
         )
         .unwrap()
     });
 
     async fn setup() -> (Router, PgPool) {
         let (_, pool) = db_test_setup(CONFIG_SOURCE).await;
         let api = ServerState::start(
             pool.clone(),
             EXCHANGE.clone(),
             Currency::from_str("BTC").unwrap(),
         )
         .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-revenue/config")
             .await
             .assert_ok_json::<RevenueConfig>();
         server
             .get("/taler-prepared-transfer/config")
             .await
             .assert_ok_json::<PreparedTransferConfig>();
         server
             .get("/taler-observability/config")
             .await
             .assert_ok_json::<Config>();
         server.get("/taler-observability/metrics").await.assert_ok();
     }
 
     #[tokio::test]
     async fn transfer() {
         let (server, _) = setup().await;
         transfer_routine(
             &server.prefix("/taler-wire-gateway"),
             TransferState::pending,
             &CLIENT.as_uri(),
         )
         .await;
     }
 
     #[tokio::test]
     async fn outgoing_history() {
         let (server, db) = setup().await;
         out_history_routine(
             &server.prefix("/taler-wire-gateway"),
             tasks!({
                 let sub = &OutgoingSubject::rand();
                 sync_out(
                     &db,
                     &TxOut {
                         id: rand_tx_id(),
                         replaces_txid: None,
                         amount: amount("BTC:10"),
                         credit_acc: &EXCHANGE.0,
                         block_time: Timestamp::now(),
                     },
                     &TxOutKind::Talerable {
                         wtid: &sub.wtid,
                         url: &sub.exchange_base_url,
                         metadata: sub.metadata.as_deref(),
                     },
                     true,
                 )
                 .await
                 .unwrap();
             }),
             tasks!(),
         )
         .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"),
             &CLIENT.as_uri(),
             &EXCHANGE.as_uri(),
         )
         .await;
     }
 
     #[tokio::test]
     async fn revenue() {
         let (server, _) = &setup().await;
         revenue_routine(
             &server.prefix("/taler-wire-gateway"),
             &server.prefix("/taler-revenue"),
             &EXCHANGE.as_uri(),
             tasks!(),
             tasks!(),
         )
         .await;
     }
 
     async fn check_in(pool: &PgPool) -> Vec<Status> {
         sqlx::query(
             "
             SELECT pending_recurrent_in.authorization_pub IS NOT NULL, bounced.reason 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"),
             &CLIENT.as_uri(),
             &EXCHANGE.as_uri(),
             &UNKNOWN.as_uri(),
             || check_in(&pool),
         )
         .await;
     }
 }