use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use axum_core::body::Body;
use axum_core::extract::Request;
use axum_core::response::{IntoResponse, Response};
use http::{HeaderMap, HeaderValue, StatusCode};
use r402_core::facilitator::Facilitator;
use r402_core::wire;
use r402_core::wire::Base64Bytes;
use serde_json::json;
use tokio::sync::Notify;
use tower::Service;
#[cfg(feature = "telemetry")]
use tracing::{Instrument, instrument};
use url::Url;
use super::hooks::DynPaygateHooks;
const PAYMENT_HEADER: &str = "Payment-Signature";
#[derive(Debug, thiserror::Error)]
pub enum VerificationError {
#[error("Payment-Signature header is required")]
PaymentHeaderMissing,
#[error("Invalid or malformed payment header")]
InvalidPaymentHeader,
#[error("Unable to find matching payment requirements")]
NoPaymentMatching,
#[error("Verification failed: {0}")]
VerificationFailed(String),
}
#[derive(Debug, thiserror::Error)]
pub enum PaygateError {
#[error(transparent)]
Verification(#[from] VerificationError),
#[error("settlement failed: {}", settlement_failure_summary(.0))]
Settlement(Box<wire::SettleResponse>),
#[error("settlement aborted: {0}")]
SettlementAborted(String),
}
#[allow(
clippy::missing_const_for_fn,
reason = "const fn would prevent matching on `Box` indirection"
)]
fn settlement_failure_summary(resp: &wire::SettleResponse) -> String {
match resp {
wire::SettleResponse::Failure {
reason,
message,
network,
..
} => format!(
"{} ({}){}",
reason,
network,
message
.as_ref()
.map(|m| format!(": {m}"))
.unwrap_or_default(),
),
wire::SettleResponse::Success { .. } => "success returned via error path".to_owned(),
_ => "unknown settlement variant".to_owned(),
}
}
type PaymentPayload = wire::PaymentPayload<wire::PaymentRequirements, serde_json::Value>;
#[derive(Debug, Clone)]
pub struct ResourceTemplate {
pub description: String,
pub mime_type: String,
pub url: Option<String>,
}
impl Default for ResourceTemplate {
fn default() -> Self {
Self {
description: String::new(),
mime_type: "application/json".to_owned(),
url: None,
}
}
}
impl ResourceTemplate {
#[allow(clippy::unwrap_used, reason = "fallback URL is a hardcoded constant")]
pub fn resolve(&self, base_url: Option<&Url>, req: &Request) -> wire::ResourceInfo {
let url = self.url.clone().unwrap_or_else(|| {
let mut url = base_url.cloned().unwrap_or_else(|| {
let host = req
.headers()
.get("host")
.and_then(|h| h.to_str().ok())
.unwrap_or("localhost");
let origin = format!("http://{host}");
let url =
Url::parse(&origin).unwrap_or_else(|_| Url::parse("http://localhost").unwrap());
#[cfg(feature = "telemetry")]
tracing::warn!(
"X402Middleware base_url is not configured; \
using {url} as origin for resource resolution"
);
url
});
url.set_path(req.uri().path());
url.set_query(req.uri().query());
url.to_string()
});
let mut info = wire::ResourceInfo::new(url);
if !self.description.is_empty() {
info = info.with_description(self.description.clone());
}
if !self.mime_type.is_empty() {
info = info.with_mime_type(self.mime_type.clone());
}
info
}
}
#[allow(
missing_debug_implementations,
reason = "generic facilitator may not impl Debug"
)]
pub struct PaygateBuilder<TFacilitator> {
facilitator: TFacilitator,
accepts: Vec<wire::PriceTag>,
resource: Option<wire::ResourceInfo>,
hooks: Option<Arc<dyn DynPaygateHooks>>,
settlement_tracker: Option<BackgroundSettlementTracker>,
}
impl<TFacilitator> PaygateBuilder<TFacilitator> {
#[must_use]
pub fn accept(mut self, price_tag: wire::PriceTag) -> Self {
self.accepts.push(price_tag);
self
}
#[must_use]
pub fn accepts(mut self, price_tags: impl IntoIterator<Item = wire::PriceTag>) -> Self {
self.accepts.extend(price_tags);
self
}
#[must_use]
pub fn resource(mut self, resource: wire::ResourceInfo) -> Self {
self.resource = Some(resource);
self
}
#[must_use]
pub fn hooks<H>(mut self, hooks: H) -> Self
where
H: super::hooks::PaygateHooks + 'static,
{
self.hooks = Some(Arc::new(hooks));
self
}
#[must_use]
pub fn hooks_dyn(mut self, hooks: Arc<dyn DynPaygateHooks>) -> Self {
self.hooks = Some(hooks);
self
}
#[must_use]
pub fn with_settlement_tracker(mut self, tracker: BackgroundSettlementTracker) -> Self {
self.settlement_tracker = Some(tracker);
self
}
pub fn build(self) -> Paygate<TFacilitator> {
Paygate {
facilitator: self.facilitator,
accepts: self.accepts.into(),
resource: self
.resource
.unwrap_or_else(|| wire::ResourceInfo::new("").with_mime_type("application/json")),
hooks: self.hooks,
settlement_tracker: self.settlement_tracker,
}
}
}
#[allow(
missing_debug_implementations,
reason = "generic facilitator may not impl Debug"
)]
pub struct Paygate<TFacilitator> {
pub(crate) facilitator: TFacilitator,
pub(crate) accepts: Arc<[wire::PriceTag]>,
pub(crate) resource: wire::ResourceInfo,
pub(crate) hooks: Option<Arc<dyn DynPaygateHooks>>,
pub(crate) settlement_tracker: Option<BackgroundSettlementTracker>,
}
impl<TFacilitator> Paygate<TFacilitator> {
pub const fn builder(facilitator: TFacilitator) -> PaygateBuilder<TFacilitator> {
PaygateBuilder {
facilitator,
accepts: Vec::new(),
resource: None,
hooks: None,
settlement_tracker: None,
}
}
pub const fn facilitator(&self) -> &TFacilitator {
&self.facilitator
}
pub fn accepts(&self) -> &[wire::PriceTag] {
&self.accepts
}
#[must_use]
pub const fn settlement_tracker(&self) -> Option<&BackgroundSettlementTracker> {
self.settlement_tracker.as_ref()
}
pub const fn resource(&self) -> &wire::ResourceInfo {
&self.resource
}
#[must_use]
pub fn hooks(&self) -> Option<&Arc<dyn DynPaygateHooks>> {
self.hooks.as_ref()
}
#[must_use]
#[allow(
clippy::expect_used,
reason = "infallible JSON/HTTP construction; panic indicates a bug"
)]
pub fn error_response(&self, err: PaygateError) -> Response {
match err {
PaygateError::Verification(ve) => {
let (status, payment_required) = {
let status = inferred_status(&ve);
let payment_required = wire::PaymentRequired::new(self.resource.clone())
.with_error(ve.to_string())
.with_accepts(
self.accepts
.iter()
.map(|pt| pt.requirements.clone())
.collect(),
);
(status, payment_required)
};
let body_bytes =
serde_json::to_vec(&payment_required).expect("serialization failed");
let header_value =
HeaderValue::from_bytes(Base64Bytes::encode(&body_bytes).as_ref())
.expect("invalid header value");
let mut response = Response::builder()
.status(status)
.header("Payment-Required", header_value)
.header("Content-Type", "application/json")
.body(Body::from(body_bytes))
.expect("failed to construct response");
super::cors::ensure_expose_headers(response.headers_mut());
response
}
PaygateError::Settlement(failure) => {
#[cfg(feature = "telemetry")]
tracing::error!(failure = ?failure, "Settlement failed");
let body_bytes = serde_json::to_vec(&*failure).expect("serialization failed");
let header_value = failure
.encode_base64_any()
.and_then(|b64| HeaderValue::from_bytes(b64.as_ref()).ok());
let mut builder = Response::builder()
.status(StatusCode::PAYMENT_REQUIRED)
.header("Content-Type", "application/json");
if let Some(header_value) = header_value {
builder = builder.header("Payment-Response", header_value);
}
let mut response = builder
.body(Body::from(body_bytes))
.expect("failed to construct response");
super::cors::ensure_expose_headers(response.headers_mut());
response
}
PaygateError::SettlementAborted(ref detail) => {
#[cfg(feature = "telemetry")]
tracing::error!(details = %detail, "Settlement aborted");
let body = json!({
"error": "settlement aborted",
"details": detail,
})
.to_string();
let mut response = Response::builder()
.status(StatusCode::PAYMENT_REQUIRED)
.header("Content-Type", "application/json")
.body(Body::from(body))
.expect("failed to construct response");
super::cors::ensure_expose_headers(response.headers_mut());
response
}
}
}
}
impl<TFacilitator> Paygate<TFacilitator>
where
TFacilitator: Facilitator + Sync,
{
pub async fn enrich_accepts(&mut self) {
let capabilities = self.facilitator.supported().await.unwrap_or_default();
let accepts: Vec<_> = self
.accepts
.iter()
.cloned()
.map(|mut pt| {
pt.enrich(&capabilities);
pt
})
.collect();
self.accepts = accepts.into();
}
#[cfg_attr(feature = "telemetry", instrument(name = "x402.verify_only", skip_all))]
pub async fn verify_only(&self, headers: &HeaderMap) -> Result<VerifiedPayment, PaygateError> {
let header_bytes = headers
.get(PAYMENT_HEADER)
.map(HeaderValue::as_bytes)
.ok_or(VerificationError::PaymentHeaderMissing)?;
let payload: PaymentPayload =
decode_payment_payload(header_bytes).ok_or(VerificationError::InvalidPaymentHeader)?;
let verify_request = build_verify_request(payload, &self.accepts)?;
let verify_response = self
.facilitator
.verify(verify_request.clone())
.await
.map_err(|e| VerificationError::VerificationFailed(format!("{e}")))?;
if let wire::VerifyResponse::Invalid { reason, .. } = verify_response {
return Err(VerificationError::VerificationFailed(reason.to_string()).into());
}
Ok(VerifiedPayment {
settle_request: verify_request.into(),
})
}
#[cfg_attr(
feature = "telemetry",
instrument(name = "x402.handle_request", skip_all)
)]
pub async fn handle_request<
ReqBody,
ResBody,
S: Service<http::Request<ReqBody>, Response = http::Response<ResBody>>,
>(
&self,
inner: S,
req: http::Request<ReqBody>,
) -> Result<Response, PaygateError>
where
S::Response: IntoResponse,
S::Error: IntoResponse,
S::Future: Send,
{
let verified = self.verify_only(req.headers()).await?;
let response = match call_inner(inner, req).await {
Ok(r) => r,
Err(err) => return Ok(err.into_response()),
};
if response.status().is_client_error() || response.status().is_server_error() {
return Ok(response.into_response());
}
let mut response = response.into_response();
let override_amount = response
.extensions_mut()
.remove::<super::upto::UptoActualAmount>();
let settlement = verified
.settle_with_override(
&self.facilitator,
override_amount
.as_ref()
.map(super::upto::UptoActualAmount::as_str),
)
.await?;
let header_value = settlement_to_header(&settlement)?;
response
.headers_mut()
.insert("Payment-Response", header_value);
Ok(response)
}
}
impl<TFacilitator> Paygate<TFacilitator>
where
TFacilitator: Facilitator + Clone + Send + Sync + 'static,
{
#[cfg_attr(
feature = "telemetry",
instrument(name = "x402.handle_request_concurrent", skip_all)
)]
pub async fn handle_request_concurrent<
ReqBody,
ResBody,
S: Service<http::Request<ReqBody>, Response = http::Response<ResBody>>,
>(
&self,
inner: S,
req: http::Request<ReqBody>,
) -> Result<Response, PaygateError>
where
S::Response: IntoResponse,
S::Error: IntoResponse,
S::Future: Send + 'static,
ReqBody: Send + 'static,
{
let verified = self.verify_only(req.headers()).await?;
let facilitator = self.facilitator.clone();
let settle_handle = tokio::spawn(async move { verified.settle(&facilitator).await });
let response = match call_inner(inner, req).await {
Ok(r) => r,
Err(err) => {
drop(settle_handle);
return Ok(err.into_response());
}
};
if response.status().is_client_error() || response.status().is_server_error() {
drop(settle_handle);
return Ok(response.into_response());
}
let settlement = settle_handle
.await
.map_err(|e| PaygateError::SettlementAborted(format!("settle task panicked: {e}")))??;
let header_value = settlement_to_header(&settlement)?;
let mut res = response;
res.headers_mut().insert("Payment-Response", header_value);
Ok(res.into_response())
}
#[cfg_attr(
feature = "telemetry",
instrument(name = "x402.handle_request_background", skip_all)
)]
pub async fn handle_request_background<
ReqBody,
ResBody,
S: Service<http::Request<ReqBody>, Response = http::Response<ResBody>>,
>(
&self,
inner: S,
req: http::Request<ReqBody>,
) -> Result<Response, PaygateError>
where
S::Response: IntoResponse,
S::Error: IntoResponse,
S::Future: Send + 'static,
ReqBody: Send + 'static,
{
let verified = self.verify_only(req.headers()).await?;
let facilitator = self.facilitator.clone();
let settle_handle = tokio::spawn(async move { verified.settle(&facilitator).await });
let tracker_guard = self
.settlement_tracker
.as_ref()
.map(BackgroundSettlementTracker::start);
drop(tokio::spawn(supervise_background_settle(
settle_handle,
tracker_guard,
)));
match call_inner(inner, req).await {
Ok(r) => Ok(r.into_response()),
Err(err) => Ok(err.into_response()),
}
}
}
#[derive(Debug)]
pub struct VerifiedPayment {
settle_request: wire::SettleRequest,
}
impl VerifiedPayment {
pub async fn settle<F: Facilitator>(
self,
facilitator: &F,
) -> Result<wire::SettleResponse, PaygateError> {
self.settle_with_override(facilitator, None).await
}
pub async fn settle_with_override<F: Facilitator>(
mut self,
facilitator: &F,
actual_amount: Option<&str>,
) -> Result<wire::SettleResponse, PaygateError> {
if let Some(amount) = actual_amount {
self.settle_request
.set_settlement_amount(amount)
.map_err(|e| {
PaygateError::SettlementAborted(format!("upto amount override failed: {e}"))
})?;
}
let settlement = facilitator
.settle(self.settle_request)
.await
.map_err(|e| PaygateError::SettlementAborted(format!("{e}")))?;
if matches!(settlement, wire::SettleResponse::Failure { .. }) {
return Err(PaygateError::Settlement(Box::new(settlement)));
}
Ok(settlement)
}
#[must_use]
pub const fn settle_request(&self) -> &wire::SettleRequest {
&self.settle_request
}
}
#[derive(Clone, Debug)]
pub struct BackgroundSettlementTracker {
inner: Arc<TrackerInner>,
}
#[derive(Debug)]
struct TrackerInner {
in_flight: AtomicUsize,
drained: Notify,
}
impl Default for BackgroundSettlementTracker {
fn default() -> Self {
Self::new()
}
}
impl BackgroundSettlementTracker {
#[must_use]
pub fn new() -> Self {
Self {
inner: Arc::new(TrackerInner {
in_flight: AtomicUsize::new(0),
drained: Notify::new(),
}),
}
}
#[must_use]
pub fn in_flight(&self) -> usize {
self.inner.in_flight.load(Ordering::SeqCst)
}
fn start(&self) -> SettlementInFlightGuard {
let _previous = self.inner.in_flight.fetch_add(1, Ordering::SeqCst);
SettlementInFlightGuard {
inner: Arc::clone(&self.inner),
}
}
pub async fn wait_for_drain(&self, timeout: Duration) -> Result<(), usize> {
if self.in_flight() == 0 {
return Ok(());
}
let deadline = tokio::time::Instant::now() + timeout;
loop {
let notified = self.inner.drained.notified();
tokio::pin!(notified);
tokio::select! {
() = &mut notified => {}
() = tokio::time::sleep_until(deadline) => {
let remaining = self.in_flight();
return if remaining == 0 { Ok(()) } else { Err(remaining) };
}
}
if self.in_flight() == 0 {
return Ok(());
}
}
}
}
#[derive(Debug)]
pub(crate) struct SettlementInFlightGuard {
inner: Arc<TrackerInner>,
}
impl Drop for SettlementInFlightGuard {
fn drop(&mut self) {
let previous = self.inner.in_flight.fetch_sub(1, Ordering::SeqCst);
if previous == 1 {
self.inner.drained.notify_waiters();
}
}
}
async fn supervise_background_settle(
handle: tokio::task::JoinHandle<Result<wire::SettleResponse, PaygateError>>,
_tracker: Option<SettlementInFlightGuard>,
) {
let outcome = handle.await;
log_background_settle_outcome(outcome);
}
fn log_background_settle_outcome(
outcome: Result<Result<wire::SettleResponse, PaygateError>, tokio::task::JoinError>,
) {
match outcome {
Ok(Ok(_settlement)) => {
#[cfg(feature = "telemetry")]
tracing::debug!("background settlement completed");
record_background_settle_metric("ok");
}
Ok(Err(err)) => {
log_background_settle_error(&err);
record_background_settle_metric("error");
}
Err(join_err) => {
let label = if join_err.is_panic() {
"panic"
} else {
"cancelled"
};
log_background_settle_join_error(&join_err);
record_background_settle_metric(label);
}
}
}
#[cfg(feature = "metrics")]
fn record_background_settle_metric(result: &'static str) {
::metrics::counter!(
r402_core::metrics::PAYGATE_BACKGROUND_SETTLE_TOTAL,
"result" => result,
)
.increment(1);
}
#[cfg(not(feature = "metrics"))]
fn record_background_settle_metric(_result: &'static str) {}
#[cfg(feature = "telemetry")]
fn log_background_settle_error(err: &PaygateError) {
tracing::error!(error = %err, "background settlement returned error");
}
#[cfg(not(feature = "telemetry"))]
fn log_background_settle_error(_err: &PaygateError) {}
#[cfg(feature = "telemetry")]
fn log_background_settle_join_error(join_err: &tokio::task::JoinError) {
if join_err.is_panic() {
tracing::error!(error = %join_err, "background settlement task panicked");
} else {
tracing::warn!(error = %join_err, "background settlement task cancelled");
}
}
#[cfg(not(feature = "telemetry"))]
fn log_background_settle_join_error(_join_err: &tokio::task::JoinError) {}
pub fn settlement_to_header(
settlement: &wire::SettleResponse,
) -> Result<HeaderValue, PaygateError> {
let encoded = settlement.encode_base64().ok_or_else(|| {
PaygateError::SettlementAborted("cannot encode error settlement".to_owned())
})?;
HeaderValue::from_bytes(encoded.as_ref())
.map_err(|e| PaygateError::SettlementAborted(e.to_string()))
}
async fn call_inner<
ReqBody,
ResBody,
S: Service<http::Request<ReqBody>, Response = http::Response<ResBody>>,
>(
mut inner: S,
req: http::Request<ReqBody>,
) -> Result<http::Response<ResBody>, S::Error>
where
S::Future: Send,
{
#[cfg(feature = "telemetry")]
{
inner
.call(req)
.instrument(tracing::info_span!("inner"))
.await
}
#[cfg(not(feature = "telemetry"))]
{
inner.call(req).await
}
}
fn decode_payment_payload<T: serde::de::DeserializeOwned>(header_bytes: &[u8]) -> Option<T> {
let decoded = Base64Bytes::from(header_bytes).decode().ok()?;
serde_json::from_slice(decoded.as_ref()).ok()
}
fn inferred_status(ve: &VerificationError) -> StatusCode {
if let VerificationError::VerificationFailed(message) = ve
&& message.contains("permit2_allowance_required")
{
return StatusCode::PRECONDITION_FAILED;
}
StatusCode::PAYMENT_REQUIRED
}
fn build_verify_request(
payload: PaymentPayload,
accepts: &[wire::PriceTag],
) -> Result<wire::VerifyRequest, VerificationError> {
let selected = accepts
.iter()
.find(|pt| **pt == payload.accepted)
.ok_or(VerificationError::NoPaymentMatching)?;
let verify: wire::TypedVerifyRequest<2, PaymentPayload, wire::PaymentRequirements> =
wire::TypedVerifyRequest {
x402_version: wire::V2,
payment_payload: payload,
payment_requirements: selected.requirements.clone(),
};
let json = serde_json::to_value(&verify)
.map_err(|e| VerificationError::VerificationFailed(format!("{e}")))?;
Ok(wire::VerifyRequest::from(json))
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn empty_tracker_drains_immediately() {
let tracker = BackgroundSettlementTracker::new();
assert_eq!(tracker.in_flight(), 0);
tracker.wait_for_drain(Duration::ZERO).await.unwrap();
}
#[tokio::test]
async fn drain_waits_for_guard_drop() {
let tracker = BackgroundSettlementTracker::new();
let guard = tracker.start();
assert_eq!(tracker.in_flight(), 1);
let tracker_clone = tracker.clone();
let drop_task = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(10)).await;
drop(guard);
assert_eq!(tracker_clone.in_flight(), 0);
});
tracker
.wait_for_drain(Duration::from_secs(1))
.await
.expect("drain should complete after the guard drops");
drop_task.await.unwrap();
}
#[tokio::test]
async fn drain_times_out_when_guards_outlive_deadline() {
let tracker = BackgroundSettlementTracker::new();
let _guard = tracker.start();
let result = tracker.wait_for_drain(Duration::from_millis(20)).await;
assert_eq!(result, Err(1), "deadline elapses with the guard alive");
}
#[tokio::test]
async fn nested_guards_decrement_in_order() {
let tracker = BackgroundSettlementTracker::new();
let g1 = tracker.start();
let g2 = tracker.start();
let g3 = tracker.start();
assert_eq!(tracker.in_flight(), 3);
drop(g2);
assert_eq!(tracker.in_flight(), 2);
drop(g1);
assert_eq!(tracker.in_flight(), 1);
drop(g3);
assert_eq!(tracker.in_flight(), 0);
}
}