taler-rust

api.rs (7182B)

---

 /*
   This file is part of TALER
   Copyright (C) 2025 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::{AtomicU32, Ordering},
     },
     time::Instant,
 };
 
 use axum::{
     extract::{Request, State},
     middleware::{self, Next},
     response::Response,
 };
 use revenue::Revenue;
 use taler_common::{error_code::ErrorCode, types::amount::Amount};
 use tokio::signal;
 use tracing::{Level, info};
 use wire::WireGateway;
 
 use crate::{
     Listener, Serve,
     auth::{AuthMethod, AuthMiddlewareState},
     error::{ApiResult, failure, failure_code},
 };
 
 pub mod revenue;
 pub mod wire;
 
 pub use axum::Router;
 
 pub trait TalerApi: Send + Sync + 'static {
     fn currency(&self) -> &str;
     fn implementation(&self) -> Option<&str>;
     fn check_currency(&self, amount: &Amount) -> ApiResult<()> {
         let currency = self.currency();
         if amount.currency.as_ref() != currency {
             Err(failure(
                 ErrorCode::GENERIC_CURRENCY_MISMATCH,
                 format!(
                     "wrong currency expected {} got {}",
                     currency, amount.currency
                 ),
             ))
         } else {
             Ok(())
         }
     }
 }
 
 pub trait RouterUtils {
     fn auth(self, auth: AuthMethod, realm: &str) -> Self;
 }
 
 impl<S: Send + Clone + Sync + 'static> RouterUtils for Router<S> {
     fn auth(self, auth: AuthMethod, realm: &str) -> Self {
         self.route_layer(middleware::from_fn_with_state(
             Arc::new(AuthMiddlewareState::new(auth, realm)),
             crate::auth::auth_middleware,
         ))
     }
 }
 
 pub trait TalerRouter {
     fn wire_gateway<T: WireGateway>(self, api: Arc<T>, auth: AuthMethod) -> Self;
     fn revenue<T: Revenue>(self, api: Arc<T>, auth: AuthMethod) -> Self;
     fn finalize(self) -> Self;
     fn serve(
         self,
         serve: Serve,
         lifetime: Option<u32>,
     ) -> impl std::future::Future<Output = std::io::Result<()>> + Send;
 }
 
 impl TalerRouter for Router {
     fn wire_gateway<T: WireGateway>(self, api: Arc<T>, auth: AuthMethod) -> Self {
         self.nest("/taler-wire-gateway", wire::router(api, auth))
     }
 
     fn revenue<T: Revenue>(self, api: Arc<T>, auth: AuthMethod) -> Self {
         self.nest("/taler-revenue", revenue::router(api, auth))
     }
 
     fn finalize(self) -> Router {
         self.method_not_allowed_fallback(|| async {
             failure_code(ErrorCode::GENERIC_METHOD_INVALID)
         })
         .fallback(|| async { failure_code(ErrorCode::GENERIC_ENDPOINT_UNKNOWN) })
         .layer(middleware::from_fn(logger_middleware))
     }
 
     async fn serve(mut self, serve: Serve, lifetime: Option<u32>) -> std::io::Result<()> {
         let listener = serve.resolve()?;
 
         let notify = Arc::new(tokio::sync::Notify::new());
         if let Some(lifetime) = lifetime {
             self = self.layer(middleware::from_fn_with_state(
                 Arc::new(LifetimeMiddlewareState {
                     notify: notify.clone(),
                     lifetime: AtomicU32::new(lifetime),
                 }),
                 lifetime_middleware,
             ))
         }
         let router = self.finalize();
         let signal = shutdown_signal(notify);
         match listener {
             Listener::Tcp(tcp_listener) => {
                 axum::serve(tcp_listener, router)
                     .with_graceful_shutdown(signal)
                     .await?;
             }
             Listener::Unix(unix_listener) => {
                 axum::serve(unix_listener, router)
                     .with_graceful_shutdown(signal)
                     .await?;
             }
         }
 
         info!(target: "api", "Server stopped");
         Ok(())
     }
 }
 
 struct LifetimeMiddlewareState {
     lifetime: AtomicU32,
     notify: Arc<tokio::sync::Notify>,
 }
 
 async fn lifetime_middleware(
     State(state): State<Arc<LifetimeMiddlewareState>>,
     request: Request,
     next: Next,
 ) -> Response {
     let mut current = state.lifetime.load(Ordering::Relaxed);
     while current != 0 {
         match state.lifetime.compare_exchange_weak(
             current,
             current - 1,
             Ordering::Relaxed,
             Ordering::Relaxed,
         ) {
             Ok(_) => break,
             Err(new) => current = new,
         }
     }
     if current == 0 {
         state.notify.notify_one();
     }
     next.run(request).await
 }
 
 /** Wait for manual shutdown or system signal shutdown */
 async fn shutdown_signal(manual_shutdown: Arc<tokio::sync::Notify>) {
     let ctrl_c = async {
         signal::ctrl_c()
             .await
             .expect("failed to install Ctrl+C handler");
     };
 
     #[cfg(unix)]
     let terminate = async {
         signal::unix::signal(signal::unix::SignalKind::terminate())
             .expect("failed to install signal handler")
             .recv()
             .await;
     };
 
     #[cfg(not(unix))]
     let terminate = std::future::pending::<()>();
 
     let manual = async { manual_shutdown.notified().await };
 
     tokio::select! {
         _ = ctrl_c => {},
         _ = terminate => {},
         _ = manual => {}
     }
 }
 
 #[macro_export]
 macro_rules! dyn_event {
     ($lvl:ident, $($arg:tt)+) => {
         match $lvl {
             ::tracing::Level::TRACE => ::tracing::trace!($($arg)+),
             ::tracing::Level::DEBUG => ::tracing::debug!($($arg)+),
             ::tracing::Level::INFO => ::tracing::info!($($arg)+),
             ::tracing::Level::WARN => ::tracing::warn!($($arg)+),
             ::tracing::Level::ERROR => ::tracing::error!($($arg)+),
         }
     };
 }
 
 /** Taler API logger */
 async fn logger_middleware(request: Request, next: Next) -> Response {
     let request_info = format!(
         "{} {}",
         request.method(),
         request
             .uri()
             .path_and_query()
             .map(|it| it.as_str())
             .unwrap_or_default()
     );
     let now = Instant::now();
     let response = next.run(request).await;
     let elapsed = now.elapsed();
     let status = response.status();
     let level = match status.as_u16() {
         400..500 => Level::WARN,
         500..600 => Level::ERROR,
         _ => Level::INFO,
     };
 
     if let Some(log) = response.extensions().get::<Box<str>>() {
         dyn_event!(level, target: "api",
             "{} {request_info} {}ms: {log}",
             response.status(),
             elapsed.as_millis()
         );
     } else {
         dyn_event!(level, target: "api",
             "{} {request_info} {}ms",
             response.status(),
             elapsed.as_millis()
         );
     }
     response
 }