i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-03 09:50:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'entropy-xu/payment-billing-plans'

Browse Source 
# Conflicts:
#	crates/aether-data/src/lifecycle/bootstrap/postgres.rs
#	crates/aether-data/src/lifecycle/migrate/tests.rs
#	frontend/src/views/admin/Users.vue
This commit is contained in:
fawney19
2026-05-15 16:37:17 +08:00
parent 38240f77ff 3fb007a56d
commit e4f427f921
97 changed files with 14795 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
apps/aether-gateway/src/handlers/public/support.rs
View File

@@ -26,6 +26,8 @@ use std::time::{SystemTime, UNIX_EPOCH};
mod support_announcements;
#[path = "support/auth.rs"]
mod support_auth;
#[path = "support/billing.rs"]
mod support_billing;
#[path = "support/dashboard.rs"]
mod support_dashboard;
#[path = "support/install.rs"]
@@ -62,6 +64,7 @@ use self::support_auth::{
build_auth_error_response, build_auth_json_response, build_auth_registration_settings_payload,
build_auth_settings_payload, extract_client_device_id, maybe_build_local_auth_response,
};
use self::support_billing::maybe_build_local_billing_response;
use self::support_dashboard::maybe_build_local_dashboard_response;
pub(crate) use self::support_install::{
build_api_key_install_session_response, build_proxy_node_install_session_response,
@@ -151,6 +154,15 @@ pub(crate) async fn maybe_build_local_public_support_response(
return Some(build_unhandled_public_support_response(request_context));
}
if decision.route_family.as_deref() == Some("billing") {
if let Some(response) =
maybe_build_local_billing_response(state, request_context, headers, request_body).await
{
return Some(response);
}
return Some(build_unhandled_public_support_response(request_context));
}
if decision.route_family.as_deref() == Some("users_me") {
return maybe_build_local_users_me_response(state, request_context, headers, request_body)
.await;

451
apps/aether-gateway/src/handlers/public/support/billing.rs Normal file
View File

@@ -0,0 +1,451 @@
use super::support_payment::payment_epay::{
build_epay_checkout_url, epay_callback_base_url, load_epay_config, resolve_epay_channel,
EpayCheckoutInput,
};
use super::{
build_auth_error_response, build_auth_json_response, resolve_authenticated_local_user,
sanitize_wallet_gateway_response, unix_secs_to_rfc3339, AppState, GatewayPublicRequestContext,
};
use axum::{
body::{Body, Bytes},
http,
response::{IntoResponse, Response},
Json,
};
use chrono::Utc;
use serde::Deserialize;
use serde_json::json;
use uuid::Uuid;
const BILLING_STORAGE_UNAVAILABLE_DETAIL: &str = "套餐后端暂不可用";
#[derive(Debug, Deserialize, Default)]
struct BillingPlanCheckoutRequest {
#[serde(default)]
payment_method: Option<String>,
#[serde(default)]
payment_provider: Option<String>,
#[serde(default)]
payment_channel: Option<String>,
}
#[derive(Debug, Clone)]
struct NormalizedBillingPlanCheckoutRequest {
payment_method: String,
payment_provider: String,
payment_channel: Option<String>,
}
fn billing_storage_unavailable_response() -> Response<Body> {
build_auth_error_response(
http::StatusCode::SERVICE_UNAVAILABLE,
BILLING_STORAGE_UNAVAILABLE_DETAIL,
false,
)
}
fn normalize_optional_checkout_string(value: Option<String>, max_len: usize) -> Option<String> {
value
.map(|value| value.trim().to_ascii_lowercase())
.filter(|value| !value.is_empty() && value.chars().count() <= max_len)
}
fn normalize_checkout_request(
payload: BillingPlanCheckoutRequest,
) -> Result<NormalizedBillingPlanCheckoutRequest, &'static str> {
let payment_provider = normalize_optional_checkout_string(payload.payment_provider, 30)
.or_else(|| normalize_optional_checkout_string(payload.payment_method.clone(), 30))
.unwrap_or_else(|| "epay".to_string());
if payment_provider != "epay" {
return Err("unsupported payment_provider");
}
let payment_method = normalize_optional_checkout_string(payload.payment_method, 30)
.unwrap_or_else(|| "epay".to_string());
let payment_channel = normalize_optional_checkout_string(payload.payment_channel, 30)
.or_else(|| (payment_method != "epay").then_some(payment_method.clone()));
Ok(NormalizedBillingPlanCheckoutRequest {
payment_method: "epay".to_string(),
payment_provider,
payment_channel,
})
}
fn plan_id_from_checkout_path(path: &str) -> Option<String> {
let trimmed = path.trim_end_matches('/');
let rest = trimmed.strip_prefix("/api/billing/plans/")?;
let plan_id = rest.strip_suffix("/checkout")?.trim_matches('/');
if plan_id.is_empty() || plan_id.contains('/') {
None
} else {
Some(plan_id.to_string())
}
}
fn billing_order_no(now: chrono::DateTime<chrono::Utc>) -> String {
format!(
"pp_{}_{}",
now.format("%Y%m%d%H%M%S%6f"),
&Uuid::new_v4().simple().to_string()[..12]
)
}
fn billing_plan_payload(
record: &aether_data_contracts::repository::billing::BillingPlanRecord,
) -> serde_json::Value {
json!({
"id": record.id,
"title": record.title,
"description": record.description,
"price_amount": record.price_amount,
"price_currency": record.price_currency,
"duration_unit": record.duration_unit,
"duration_value": record.duration_value,
"enabled": record.enabled,
"sort_order": record.sort_order,
"max_active_per_user": record.max_active_per_user,
"purchase_limit_scope": record.purchase_limit_scope,
"entitlements": record.entitlements_json,
"created_at": record.created_at_unix_secs,
"updated_at": record.updated_at_unix_secs,
})
}
fn billing_plan_snapshot(
record: &aether_data_contracts::repository::billing::BillingPlanRecord,
) -> serde_json::Value {
json!({
"id": record.id,
"title": record.title,
"description": record.description,
"price_amount": record.price_amount,
"price_currency": record.price_currency,
"duration_unit": record.duration_unit,
"duration_value": record.duration_value,
"max_active_per_user": record.max_active_per_user,
"purchase_limit_scope": record.purchase_limit_scope,
"entitlements": record.entitlements_json,
})
}
fn plan_has_package_rights(
record: &aether_data_contracts::repository::billing::BillingPlanRecord,
) -> bool {
record.entitlements_json.as_array().is_some_and(|items| {
items.iter().any(|item| {
matches!(
item.get("type").and_then(|value| value.as_str()),
Some("daily_quota" | "membership_group")
)
})
})
}
fn payment_order_payload(
record: &aether_data::repository::wallet::StoredAdminPaymentOrder,
plan: &aether_data_contracts::repository::billing::BillingPlanRecord,
) -> serde_json::Value {
json!({
"id": record.id,
"order_no": record.order_no,
"wallet_id": record.wallet_id,
"user_id": record.user_id,
"amount_usd": record.amount_usd,
"pay_amount": record.pay_amount,
"pay_currency": record.pay_currency,
"exchange_rate": record.exchange_rate,
"payment_method": record.payment_method,
"gateway_order_id": record.gateway_order_id,
"gateway_response": sanitize_wallet_gateway_response(record.gateway_response.clone()),
"status": record.status,
"order_kind": "plan_purchase",
"product_id": plan.id,
"product": billing_plan_payload(plan),
"created_at": unix_secs_to_rfc3339(record.created_at_unix_ms),
"paid_at": record.paid_at_unix_secs.and_then(unix_secs_to_rfc3339),
"credited_at": record.credited_at_unix_secs.and_then(unix_secs_to_rfc3339),
"expires_at": record.expires_at_unix_secs.and_then(unix_secs_to_rfc3339),
})
}
fn entitlement_payload(
record: &aether_data_contracts::repository::billing::UserPlanEntitlementRecord,
) -> serde_json::Value {
json!({
"id": record.id,
"user_id": record.user_id,
"plan_id": record.plan_id,
"payment_order_id": record.payment_order_id,
"status": record.status,
"starts_at": unix_secs_to_rfc3339(record.starts_at_unix_secs),
"expires_at": unix_secs_to_rfc3339(record.expires_at_unix_secs),
"entitlements": record.entitlements_snapshot,
"created_at": unix_secs_to_rfc3339(record.created_at_unix_secs),
"updated_at": unix_secs_to_rfc3339(record.updated_at_unix_secs),
})
}
fn compute_plan_payment_amounts(
plan: &aether_data_contracts::repository::billing::BillingPlanRecord,
pay_currency: &str,
usd_exchange_rate: f64,
) -> Result<(f64, f64), &'static str> {
if !plan.price_amount.is_finite() || plan.price_amount <= 0.0 || usd_exchange_rate <= 0.0 {
return Err("套餐价格配置无效");
}
if plan.price_currency.eq_ignore_ascii_case(pay_currency) {
let amount_usd =
(plan.price_amount / usd_exchange_rate * 100_000_000.0).round() / 100_000_000.0;
let pay_amount = (plan.price_amount * 100.0).round() / 100.0;
return Ok((amount_usd, pay_amount));
}
if plan.price_currency.eq_ignore_ascii_case("USD") {
let amount_usd = (plan.price_amount * 100_000_000.0).round() / 100_000_000.0;
let pay_amount = (plan.price_amount * usd_exchange_rate * 100.0).round() / 100.0;
return Ok((amount_usd, pay_amount));
}
Err("套餐币种与支付网关币种不匹配")
}
pub(super) async fn handle_billing_plans_list(state: &AppState) -> Response<Body> {
let plans = match state.list_billing_plans(false).await {
Ok(Some(value)) => value,
Ok(None) => return billing_storage_unavailable_response(),
Err(err) => {
return build_auth_error_response(
http::StatusCode::INTERNAL_SERVER_ERROR,
format!("billing plan lookup failed: {err:?}"),
false,
)
}
};
let items = plans
.iter()
.filter(|plan| plan_has_package_rights(plan))
.map(billing_plan_payload)
.collect::<Vec<_>>();
Json(json!({"items": items, "total": items.len()})).into_response()
}
pub(super) async fn handle_billing_entitlements(
state: &AppState,
request_context: &GatewayPublicRequestContext,
headers: &http::HeaderMap,
) -> Response<Body> {
let auth = match resolve_authenticated_local_user(state, request_context, headers).await {
Ok(value) => value,
Err(response) => return response,
};
let entitlements = match state.list_user_plan_entitlements(&auth.user.id).await {
Ok(Some(value)) => value,
Ok(None) => return billing_storage_unavailable_response(),
Err(err) => {
return build_auth_error_response(
http::StatusCode::INTERNAL_SERVER_ERROR,
format!("billing entitlement lookup failed: {err:?}"),
false,
)
}
};
let now = Utc::now().timestamp().max(0) as u64;
let items = entitlements
.iter()
.map(|record| {
let mut payload = entitlement_payload(record);
payload["active"] = json!(
record.status == "active"
&& record.starts_at_unix_secs <= now
&& record.expires_at_unix_secs > now
);
payload
})
.collect::<Vec<_>>();
Json(json!({"items": items, "total": items.len()})).into_response()
}
pub(super) async fn handle_billing_plan_checkout(
state: &AppState,
request_context: &GatewayPublicRequestContext,
headers: &http::HeaderMap,
request_body: Option<&Bytes>,
) -> Response<Body> {
let auth = match resolve_authenticated_local_user(state, request_context, headers).await {
Ok(value) => value,
Err(response) => return response,
};
let Some(plan_id) = plan_id_from_checkout_path(&request_context.request_path) else {
return build_auth_error_response(http::StatusCode::BAD_REQUEST, "缺少套餐ID", false);
};
let payload = match request_body {
Some(body) if !body.is_empty() => {
match serde_json::from_slice::<BillingPlanCheckoutRequest>(body) {
Ok(value) => value,
Err(_) => {
return build_auth_error_response(
http::StatusCode::BAD_REQUEST,
"输入验证失败",
false,
)
}
}
}
_ => BillingPlanCheckoutRequest::default(),
};
let checkout_request = match normalize_checkout_request(payload) {
Ok(value) => value,
Err(detail) => {
return build_auth_error_response(http::StatusCode::BAD_REQUEST, detail, false)
}
};
let plan = match state.find_billing_plan(&plan_id).await {
Ok(Some(value)) if value.enabled => value,
Ok(Some(_)) => {
return build_auth_error_response(http::StatusCode::BAD_REQUEST, "套餐已下架", false)
}
Ok(None) => {
return build_auth_error_response(http::StatusCode::NOT_FOUND, "套餐不存在", false)
}
Err(err) => {
return build_auth_error_response(
http::StatusCode::INTERNAL_SERVER_ERROR,
format!("billing plan lookup failed: {err:?}"),
false,
)
}
};
if !plan_has_package_rights(&plan) {
return build_auth_error_response(
http::StatusCode::BAD_REQUEST,
"余额包已移除，请使用钱包充值功能",
false,
);
}
let config = match load_epay_config(state).await {
Ok(value) => value,
Err(detail) => {
return build_auth_error_response(http::StatusCode::BAD_REQUEST, detail, false)
}
};
let payment_channel =
match resolve_epay_channel(&config, checkout_request.payment_channel.as_deref()) {
Ok(value) => value,
Err(detail) => {
return build_auth_error_response(http::StatusCode::BAD_REQUEST, detail, false);
}
};
let (amount_usd, pay_amount) =
match compute_plan_payment_amounts(&plan, &config.pay_currency, config.usd_exchange_rate) {
Ok(value) => value,
Err(detail) => {
return build_auth_error_response(http::StatusCode::BAD_REQUEST, detail, false)
}
};
let Some(callback_base_url) = epay_callback_base_url(
config.callback_base_url.as_deref(),
headers,
request_context,
) else {
return build_auth_error_response(
http::StatusCode::BAD_REQUEST,
"epay callback_base_url is required",
false,
);
};
let now = Utc::now();
let order_no = billing_order_no(now);
let expires_at = now + chrono::Duration::minutes(30);
let checkout = build_epay_checkout_url(
&config,
&EpayCheckoutInput {
order_no: order_no.clone(),
channel: payment_channel.clone(),
subject: plan.title.clone(),
pay_amount,
notify_url: format!("{callback_base_url}/api/payment/epay/notify"),
return_url: format!("{callback_base_url}/api/payment/epay/return"),
},
);
let outcome = match state
.create_plan_purchase_order(
aether_data::repository::wallet::CreatePlanPurchaseOrderInput {
preferred_wallet_id: None,
user_id: auth.user.id.clone(),
amount_usd,
pay_amount,
pay_currency: config.pay_currency.clone(),
exchange_rate: config.usd_exchange_rate,
payment_method: checkout_request.payment_method,
payment_provider: Some(checkout_request.payment_provider),
payment_channel: Some(payment_channel),
gateway_order_id: order_no.clone(),
gateway_response: checkout.clone(),
order_no,
product_id: plan.id.clone(),
product_snapshot: billing_plan_snapshot(&plan),
expires_at_unix_secs: expires_at.timestamp().max(0) as u64,
},
)
.await
{
Ok(Some(value)) => value,
Ok(None) => return billing_storage_unavailable_response(),
Err(err) => {
return build_auth_error_response(
http::StatusCode::INTERNAL_SERVER_ERROR,
format!("billing checkout create failed: {err:?}"),
false,
)
}
};
let order = match outcome {
aether_data::repository::wallet::CreatePlanPurchaseOrderOutcome::Created(order) => {
payment_order_payload(&order, &plan)
}
aether_data::repository::wallet::CreatePlanPurchaseOrderOutcome::WalletInactive => {
return build_auth_error_response(
http::StatusCode::BAD_REQUEST,
"wallet is not active",
false,
)
}
aether_data::repository::wallet::CreatePlanPurchaseOrderOutcome::ActivePlanLimitReached => {
return build_auth_error_response(
http::StatusCode::CONFLICT,
"套餐购买限制已达到上限",
false,
)
}
};
build_auth_json_response(
http::StatusCode::OK,
json!({
"order": order,
"payment_instructions": sanitize_wallet_gateway_response(Some(checkout)),
}),
None,
)
}
pub(super) async fn maybe_build_local_billing_response(
state: &AppState,
request_context: &GatewayPublicRequestContext,
headers: &http::HeaderMap,
request_body: Option<&Bytes>,
) -> Option<Response<Body>> {
let decision = request_context.control_decision.as_ref()?;
if decision.route_family.as_deref() != Some("billing") {
return None;
}
match decision.route_kind.as_deref() {
Some("plans") if request_context.request_path == "/api/billing/plans" => {
Some(handle_billing_plans_list(state).await)
}
Some("plan_checkout") => {
Some(handle_billing_plan_checkout(state, request_context, headers, request_body).await)
}
Some("entitlements") if request_context.request_path == "/api/billing/entitlements" => {
Some(handle_billing_entitlements(state, request_context, headers).await)
}
_ => None,
}
}

6
apps/aether-gateway/src/handlers/public/support/payment.rs
View File

@@ -1,7 +1,11 @@
use axum::{body::Body, http, response::Response};
pub(super) use super::{build_auth_error_response, AppState, GatewayPublicRequestContext};
pub(super) use super::{
build_auth_error_response, build_auth_json_response, AppState, GatewayPublicRequestContext,
};
#[path = "payment/epay.rs"]
pub(super) mod payment_epay;
#[path = "payment/gateway.rs"]
pub(super) mod payment_gateway;
#[path = "payment/repository.rs"]

528
apps/aether-gateway/src/handlers/public/support/payment/epay.rs Normal file
View File

@@ -0,0 +1,528 @@
use std::collections::BTreeMap;
use axum::{body::Body, http, response::Response};
use md5::{Digest, Md5};
use serde_json::json;
use super::{payment_shared::payment_callback_payload_hash, AppState, GatewayPublicRequestContext};
#[derive(Debug, Clone)]
pub(crate) struct EpayMerchantConfig {
pub(crate) endpoint_url: String,
pub(crate) callback_base_url: Option<String>,
pub(crate) merchant_id: String,
pub(crate) merchant_key: String,
pub(crate) pay_currency: String,
pub(crate) usd_exchange_rate: f64,
pub(crate) min_recharge_usd: f64,
pub(crate) channels: serde_json::Value,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct EpayChannelConfig {
pub(crate) channel: String,
pub(crate) display_name: String,
}
#[derive(Debug, Clone)]
pub(crate) struct EpayCheckoutInput {
pub(crate) order_no: String,
pub(crate) channel: String,
pub(crate) subject: String,
pub(crate) pay_amount: f64,
pub(crate) notify_url: String,
pub(crate) return_url: String,
}
pub(crate) fn configured_epay_channels(config: &EpayMerchantConfig) -> Vec<EpayChannelConfig> {
let Some(channels) = config.channels.as_array() else {
return Vec::new();
};
channels
.iter()
.filter_map(|channel| {
let channel_id = channel
.get("channel")
.or_else(|| channel.get("type"))
.and_then(|value| value.as_str())
.map(str::trim)
.filter(|value| !value.is_empty())?;
let display_name = channel
.get("display_name")
.and_then(|value| value.as_str())
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or(channel_id)
.to_string();
Some(EpayChannelConfig {
channel: channel_id.to_string(),
display_name,
})
})
.collect()
}
pub(crate) fn resolve_epay_channel(
config: &EpayMerchantConfig,
requested_channel: Option<&str>,
) -> Result<String, &'static str> {
let channels = configured_epay_channels(config);
if channels.is_empty() {
return Err("支付网关未配置可用通道");
}
let requested_channel = requested_channel
.map(|value| value.trim().to_ascii_lowercase())
.filter(|value| !value.is_empty());
if let Some(requested_channel) = requested_channel {
if let Some(channel) = channels
.iter()
.find(|channel| channel.channel.eq_ignore_ascii_case(&requested_channel))
{
return Ok(channel.channel.clone());
}
return Err("支付通道未配置或已停用");
}
Ok(channels[0].channel.clone())
}
pub(crate) fn epay_sign(params: &BTreeMap<String, String>, merchant_key: &str) -> String {
let canonical = params
.iter()
.filter(|(key, value)| {
key.as_str() != "sign" && key.as_str() != "sign_type" && !value.trim().is_empty()
})
.map(|(key, value)| format!("{key}={value}"))
.collect::<Vec<_>>()
.join("&");
let mut hasher = Md5::new();
hasher.update(canonical.as_bytes());
hasher.update(merchant_key.as_bytes());
format!("{:x}", hasher.finalize())
}
pub(crate) fn epay_signature_valid(params: &BTreeMap<String, String>, merchant_key: &str) -> bool {
let Some(sign) = params.get("sign") else {
return false;
};
epay_sign(params, merchant_key).eq_ignore_ascii_case(sign.trim())
}
fn epay_submit_url(endpoint_url: &str) -> String {
let trimmed = endpoint_url.trim();
if trimmed.is_empty() {
return trimmed.to_string();
}
let Ok(mut url) = url::Url::parse(trimmed) else {
return trimmed.trim_end_matches('/').to_string();
};
let path = url.path();
if path.is_empty() || path == "/" {
url.set_path("submit.php");
}
url.to_string()
}
fn normalize_epay_base_url(value: &str) -> Option<String> {
let trimmed = value.trim().trim_end_matches('/');
let parsed = url::Url::parse(trimmed).ok()?;
if !matches!(parsed.scheme(), "http" | "https") || parsed.host_str().is_none() {
return None;
}
Some(trimmed.to_string())
}
fn forwarded_header_first(value: String) -> Option<String> {
value
.split(',')
.next()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
pub(crate) fn epay_callback_base_url(
configured: Option<&str>,
headers: &http::HeaderMap,
request_context: &GatewayPublicRequestContext,
) -> Option<String> {
if let Some(value) = configured.and_then(normalize_epay_base_url) {
return Some(value);
}
if let Some(value) = std::env::var("AETHER_PUBLIC_BASE_URL")
.ok()
.or_else(|| std::env::var("PUBLIC_BASE_URL").ok())
.and_then(|value| normalize_epay_base_url(&value))
{
return Some(value);
}
let host = crate::headers::header_value_str(headers, crate::constants::FORWARDED_HOST_HEADER)
.and_then(forwarded_header_first)
.or_else(|| request_context.host_header.clone())
.map(|value| value.trim().trim_end_matches('/').to_string())
.filter(|value| {
!value.is_empty()
&& !value.contains('/')
&& !value.contains('\\')
&& !value.contains('@')
&& !value.contains(char::is_whitespace)
})?;
let proto = crate::headers::header_value_str(headers, crate::constants::FORWARDED_PROTO_HEADER)
.and_then(forwarded_header_first)
.map(|value| value.trim().trim_end_matches(':').to_ascii_lowercase())
.filter(|value| value == "http" || value == "https")
.unwrap_or_else(|| "http".to_string());
normalize_epay_base_url(&format!("{proto}://{host}"))
}
pub(crate) fn build_epay_checkout_url(
config: &EpayMerchantConfig,
input: &EpayCheckoutInput,
) -> serde_json::Value {
let money = format!("{:.2}", input.pay_amount);
let mut params = BTreeMap::new();
params.insert("pid".to_string(), config.merchant_id.clone());
params.insert("type".to_string(), input.channel.clone());
params.insert("out_trade_no".to_string(), input.order_no.clone());
params.insert("notify_url".to_string(), input.notify_url.clone());
params.insert("return_url".to_string(), input.return_url.clone());
params.insert("name".to_string(), input.subject.clone());
params.insert("money".to_string(), money.clone());
params.insert("sign_type".to_string(), "MD5".to_string());
let sign = epay_sign(&params, &config.merchant_key);
params.insert("sign".to_string(), sign);
let payment_url = epay_submit_url(&config.endpoint_url);
let payment_params = params
.iter()
.map(|(key, value)| (key.clone(), serde_json::Value::String(value.clone())))
.collect::<serde_json::Map<_, _>>();
json!({
"gateway": "epay",
"display_name": "易支付",
"gateway_order_id": input.order_no,
"payment_url": payment_url,
"submit_method": "POST",
"payment_params": serde_json::Value::Object(payment_params),
"qr_code": serde_json::Value::Null,
"pay_amount": input.pay_amount,
"pay_currency": config.pay_currency,
"payment_channel": input.channel,
})
}
pub(crate) fn parse_epay_params(
query: Option<&str>,
body: Option<&axum::body::Bytes>,
) -> BTreeMap<String, String> {
let raw = body
.filter(|bytes| !bytes.is_empty())
.and_then(|bytes| std::str::from_utf8(bytes).ok())
.or(query)
.unwrap_or("");
url::form_urlencoded::parse(raw.as_bytes())
.map(|(key, value)| (key.into_owned(), value.into_owned()))
.collect()
}
pub(crate) async fn load_epay_config(state: &AppState) -> Result<EpayMerchantConfig, String> {
let Some(record) = state
.find_payment_gateway_config("epay")
.await
.map_err(|err| format!("epay config lookup failed: {err:?}"))?
else {
return Err("epay is not configured".to_string());
};
if !record.enabled {
return Err("epay is disabled".to_string());
}
let Some(encrypted_key) = record.merchant_key_encrypted.as_deref() else {
return Err("epay merchant key is missing".to_string());
};
let Some(merchant_key) = crate::handlers::shared::decrypt_catalog_secret_with_fallbacks(
state.encryption_key(),
encrypted_key,
) else {
return Err("epay merchant key decrypt failed".to_string());
};
Ok(EpayMerchantConfig {
endpoint_url: record.endpoint_url,
callback_base_url: record.callback_base_url,
merchant_id: record.merchant_id,
merchant_key,
pay_currency: record.pay_currency,
usd_exchange_rate: record.usd_exchange_rate,
min_recharge_usd: record.min_recharge_usd,
channels: record.channels_json,
})
}
fn epay_plain(status: http::StatusCode, body: &'static str) -> Response<Body> {
Response::builder()
.status(status)
.header(http::header::CONTENT_TYPE, "text/plain; charset=utf-8")
.body(Body::from(body))
.expect("epay plain response should build")
}
fn epay_redirect(location: String) -> Response<Body> {
Response::builder()
.status(http::StatusCode::FOUND)
.header(http::header::LOCATION, location)
.body(Body::empty())
.expect("epay redirect response should build")
}
fn epay_return_location(params: &BTreeMap<String, String>, signature_valid: bool) -> String {
let order_no = params.get("out_trade_no").map(String::as_str).unwrap_or("");
let base = if order_no.starts_with("pp_") {
"/dashboard/billing"
} else {
"/dashboard/wallet"
};
let payment_status = if signature_valid
&& params.get("trade_status").map(String::as_str) == Some("TRADE_SUCCESS")
{
"success"
} else {
"pending"
};
let mut serializer = url::form_urlencoded::Serializer::new(String::new());
serializer.append_pair("payment_provider", "epay");
serializer.append_pair("payment_status", payment_status);
if !order_no.is_empty() {
serializer.append_pair("order_no", order_no);
}
if let Some(trade_no) = params
.get("trade_no")
.map(String::as_str)
.filter(|value| !value.trim().is_empty())
{
serializer.append_pair("trade_no", trade_no);
}
format!("{base}?{}", serializer.finish())
}
pub(super) async fn handle_epay_notify(
state: &AppState,
request_context: &GatewayPublicRequestContext,
request_body: Option<&axum::body::Bytes>,
) -> Response<Body> {
let config = match load_epay_config(state).await {
Ok(value) => value,
Err(_) => return epay_plain(http::StatusCode::OK, "fail"),
};
let params = parse_epay_params(
request_context.request_query_string.as_deref(),
request_body,
);
if !epay_signature_valid(&params, &config.merchant_key) {
return epay_plain(http::StatusCode::OK, "fail");
}
if params.get("trade_status").map(String::as_str) != Some("TRADE_SUCCESS") {
return epay_plain(http::StatusCode::OK, "fail");
}
let Some(order_no) = params.get("out_trade_no").cloned() else {
return epay_plain(http::StatusCode::OK, "fail");
};
let Some(pay_amount) = params
.get("money")
.and_then(|value| value.parse::<f64>().ok())
else {
return epay_plain(http::StatusCode::OK, "fail");
};
let channel = params
.get("type")
.map(|value| value.trim().to_ascii_lowercase())
.filter(|value| !value.is_empty());
let payload = serde_json::to_value(&params).unwrap_or_else(|_| json!({}));
let payload_hash = match payment_callback_payload_hash(&payload) {
Ok(value) => value,
Err(_) => return epay_plain(http::StatusCode::OK, "fail"),
};
let callback_key = params
.get("trade_no")
.cloned()
.unwrap_or_else(|| format!("epay:{order_no}:{payload_hash}"));
let amount_usd = if config.usd_exchange_rate > 0.0 {
pay_amount / config.usd_exchange_rate
} else {
pay_amount
};
let outcome = state
.process_payment_callback(
aether_data::repository::wallet::ProcessPaymentCallbackInput {
payment_method: "epay".to_string(),
payment_provider: Some("epay".to_string()),
payment_channel: channel,
callback_key,
order_no: Some(order_no),
gateway_order_id: params.get("trade_no").cloned(),
amount_usd,
pay_amount: Some(pay_amount),
pay_currency: Some(config.pay_currency),
exchange_rate: Some(config.usd_exchange_rate),
payload_hash,
payload,
signature_valid: true,
},
)
.await;
match outcome {
Ok(Some(aether_data::repository::wallet::ProcessPaymentCallbackOutcome::Applied {
..
}))
| Ok(Some(
aether_data::repository::wallet::ProcessPaymentCallbackOutcome::AlreadyCredited {
..
},
))
| Ok(Some(
aether_data::repository::wallet::ProcessPaymentCallbackOutcome::DuplicateProcessed {
..
},
)) => epay_plain(http::StatusCode::OK, "success"),
_ => epay_plain(http::StatusCode::OK, "fail"),
}
}
pub(super) async fn handle_epay_return(
state: &AppState,
request_context: &GatewayPublicRequestContext,
request_body: Option<&axum::body::Bytes>,
) -> Response<Body> {
let params = parse_epay_params(
request_context.request_query_string.as_deref(),
request_body,
);
let signature_valid = load_epay_config(state)
.await
.ok()
.is_some_and(|config| epay_signature_valid(&params, &config.merchant_key));
epay_redirect(epay_return_location(&params, signature_valid))
}
#[cfg(test)]
mod tests {
use super::{
build_epay_checkout_url, configured_epay_channels, epay_sign, epay_signature_valid,
resolve_epay_channel, EpayCheckoutInput, EpayMerchantConfig,
};
use chrono::Utc;
use serde_json::json;
use std::collections::BTreeMap;
#[test]
fn epay_sign_excludes_sign_type_sign_and_empty_values() {
let mut params = BTreeMap::new();
params.insert("pid".to_string(), "1001".to_string());
params.insert("out_trade_no".to_string(), "po_1".to_string());
params.insert("money".to_string(), "10.00".to_string());
params.insert("empty".to_string(), "".to_string());
params.insert("sign_type".to_string(), "MD5".to_string());
let sign = epay_sign(&params, "secret");
params.insert("sign".to_string(), sign.clone());
assert!(epay_signature_valid(&params, "secret"));
assert!(!epay_signature_valid(&params, "wrong"));
}
#[test]
fn configured_epay_channels_do_not_invent_defaults() {
let mut config = test_epay_config(json!([
{"channel": " Alipay ", "display_name": "支付宝"},
{"type": "wxpay", "display_name": ""},
{"display_name": "缺少通道值"}
]));
let channels = configured_epay_channels(&config);
assert_eq!(channels.len(), 2);
assert_eq!(channels[0].channel, "Alipay");
assert_eq!(channels[0].display_name, "支付宝");
assert_eq!(channels[1].channel, "wxpay");
assert_eq!(channels[1].display_name, "wxpay");
assert_eq!(
resolve_epay_channel(&config, None),
Ok("Alipay".to_string())
);
assert_eq!(
resolve_epay_channel(&config, Some("WXPAY")),
Ok("wxpay".to_string())
);
assert_eq!(
resolve_epay_channel(&config, Some("manual")),
Err("支付通道未配置或已停用")
);
config.channels = json!([]);
assert!(configured_epay_channels(&config).is_empty());
assert_eq!(
resolve_epay_channel(&config, None),
Err("支付网关未配置可用通道")
);
}
#[test]
fn epay_checkout_uses_post_form_payload_and_submit_endpoint() {
let mut config = test_epay_config(json!([]));
config.endpoint_url = "https://pay.example.com/".to_string();
let checkout = build_epay_checkout_url(
&config,
&EpayCheckoutInput {
order_no: "po_test".to_string(),
channel: "alipay".to_string(),
subject: "钱包充值".to_string(),
pay_amount: 10.0,
notify_url: "https://aether.example.com/api/payment/epay/notify".to_string(),
return_url: "https://aether.example.com/api/payment/epay/return".to_string(),
},
);
assert_eq!(
checkout["payment_url"],
"https://pay.example.com/submit.php"
);
assert_eq!(checkout["submit_method"], "POST");
assert_eq!(checkout["payment_params"]["pid"], "1000");
assert_eq!(checkout["payment_params"]["type"], "alipay");
assert_eq!(checkout["payment_params"]["out_trade_no"], "po_test");
assert_eq!(checkout["payment_params"]["money"], "10.00");
assert_eq!(checkout["payment_params"]["sign_type"], "MD5");
assert!(checkout["payment_params"]["sign"]
.as_str()
.is_some_and(|value| !value.is_empty()));
config.endpoint_url = "https://pay.example.com/submit.php".to_string();
let checkout = build_epay_checkout_url(
&config,
&EpayCheckoutInput {
order_no: format!("po_{}", Utc::now().timestamp()),
channel: "wxpay".to_string(),
subject: "钱包充值".to_string(),
pay_amount: 1.0,
notify_url: "https://aether.example.com/api/payment/epay/notify".to_string(),
return_url: "https://aether.example.com/api/payment/epay/return".to_string(),
},
);
assert_eq!(
checkout["payment_url"],
"https://pay.example.com/submit.php"
);
}
fn test_epay_config(channels: serde_json::Value) -> EpayMerchantConfig {
EpayMerchantConfig {
endpoint_url: "https://pay.example.com/submit.php".to_string(),
callback_base_url: Some("https://aether.example.com".to_string()),
merchant_id: "1000".to_string(),
merchant_key: "secret".to_string(),
pay_currency: "CNY".to_string(),
usd_exchange_rate: 7.2,
min_recharge_usd: 1.0,
channels,
}
}
}

2
apps/aether-gateway/src/handlers/public/support/payment/repository.rs
View File

@@ -32,6 +32,8 @@ pub(super) async fn handle_payment_callback_with_wallet_repository(
.process_payment_callback(
aether_data::repository::wallet::ProcessPaymentCallbackInput {
payment_method: payment_method.to_string(),
payment_provider: None,
payment_channel: None,
callback_key: payload.callback_key.clone(),
order_no: payload.order_no.clone(),
gateway_order_id: payload.gateway_order_id.clone(),

19
apps/aether-gateway/src/handlers/public/support/payment/route.rs
View File

@@ -7,7 +7,8 @@ use super::payment_shared::{
};
use super::{
build_auth_error_response, build_payment_callback_storage_unavailable_response,
handle_payment_callback_with_wallet_repository, AppState, GatewayPublicRequestContext,
handle_payment_callback_with_wallet_repository, payment_epay, AppState,
GatewayPublicRequestContext,
};
pub(super) async fn maybe_build_local_payment_callback_route_response(
@@ -17,9 +18,19 @@ pub(super) async fn maybe_build_local_payment_callback_route_response(
request_body: Option<&axum::body::Bytes>,
) -> Option<Response<Body>> {
let decision = request_context.control_decision.as_ref()?;
if decision.route_family.as_deref() != Some("payment_callback")
|| decision.route_kind.as_deref() != Some("callback")
{
if decision.route_family.as_deref() != Some("payment_callback") {
return None;
}
if decision.route_kind.as_deref() == Some("epay_notify") {
return Some(payment_epay::handle_epay_notify(state, request_context, request_body).await);
}
if decision.route_kind.as_deref() == Some("epay_return") {
return Some(payment_epay::handle_epay_return(state, request_context, request_body).await);
}
if decision.route_kind.as_deref() != Some("callback") {
return None;
}

13
apps/aether-gateway/src/handlers/public/support/wallet.rs
View File

@@ -37,7 +37,7 @@ use self::reads::{
pub(crate) use self::recharge::sanitize_wallet_gateway_response;
use self::recharge::{
handle_wallet_create_recharge, handle_wallet_recharge_detail, handle_wallet_recharge_list,
wallet_recharge_detail_path_matches,
handle_wallet_recharge_options, wallet_recharge_detail_path_matches,
};
use self::redeem::handle_wallet_redeem;
use self::refunds::{
@@ -53,8 +53,13 @@ const WALLET_SAFE_GATEWAY_RESPONSE_KEYS: &[&str] = &[
"display_name",
"gateway_order_id",
"payment_url",
"payment_params",
"submit_method",
"qr_code",
"expires_at",
"pay_amount",
"pay_currency",
"payment_channel",
"manual_credit",
];
@@ -154,6 +159,12 @@ pub(super) async fn maybe_build_local_wallet_response(
);
}
if decision.route_kind.as_deref() == Some("recharge_options")
&& request_context.request_path == "/api/wallet/recharge/options"
{
return Some(handle_wallet_recharge_options(state, request_context, headers).await);
}
if decision.route_kind.as_deref() == Some("redeem")
&& request_context.request_path == "/api/wallet/redeem"
{

60
apps/aether-gateway/src/handlers/public/support/wallet/reads.rs
View File

@@ -49,6 +49,64 @@ fn build_wallet_balance_payload(
payload
}
async fn build_wallet_balance_payload_for_user(
state: &AppState,
user_id: &str,
wallet: Option<&aether_data::repository::wallet::StoredWalletSnapshot>,
) -> serde_json::Value {
let mut payload = build_wallet_balance_payload(wallet);
let wallet_balance = wallet
.map(|value| value.balance + value.gift_balance)
.unwrap_or(0.0);
let daily_quota = state
.find_user_daily_quota_availability(user_id)
.await
.ok()
.flatten();
let (has_active_daily_quota, total_quota_usd, used_usd, remaining_usd, allow_wallet_overage) =
daily_quota
.map(|quota| {
(
quota.has_active_daily_quota,
quota.total_quota_usd,
quota.used_usd,
quota.remaining_usd,
quota.allow_wallet_overage,
)
})
.unwrap_or((false, 0.0, 0.0, 0.0, false));
let package_balance = if has_active_daily_quota {
remaining_usd.max(0.0)
} else {
0.0
};
let unlimited = payload
.get("unlimited")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
payload["daily_quota"] = json!({
"has_active": has_active_daily_quota,
"total_usd": round_to(total_quota_usd.max(0.0), 6),
"used_usd": round_to(used_usd.max(0.0), 6),
"remaining_usd": round_to(package_balance, 6),
"allow_wallet_overage": allow_wallet_overage,
});
payload["package_balance"] = json!(round_to(package_balance, 6));
payload["wallet_balance"] = json!(round_to(wallet_balance.max(0.0), 6));
payload["total_available_balance"] = if unlimited {
serde_json::Value::Null
} else {
json!(round_to((wallet_balance + package_balance).max(0.0), 6))
};
payload["deduction_order"] = json!([
"package_daily_quota",
"wallet_recharge_balance",
"wallet_gift_balance"
]);
payload
}
pub(super) fn parse_wallet_limit(query: Option<&str>) -> Result<usize, String> {
match query_param_value(query, "limit") {
Some(value) => {
@@ -173,7 +231,7 @@ pub(super) async fn handle_wallet_balance(
.flatten();
build_auth_json_response(
http::StatusCode::OK,
build_wallet_balance_payload(wallet.as_ref()),
build_wallet_balance_payload_for_user(state, &auth.user.id, wallet.as_ref()).await,
None,
)
}

200
apps/aether-gateway/src/handlers/public/support/wallet/recharge.rs
View File

@@ -1,3 +1,7 @@
use super::super::support_payment::payment_epay::{
build_epay_checkout_url, configured_epay_channels, epay_callback_base_url, load_epay_config,
resolve_epay_channel, EpayCheckoutInput,
};
use super::super::support_payment::payment_gateway::{
CreateCheckoutSessionInput, PaymentGatewayRegistry,
};
@@ -23,6 +27,10 @@ struct WalletCreateRechargeRequest {
amount_usd: f64,
payment_method: String,
#[serde(default)]
payment_provider: Option<String>,
#[serde(default)]
payment_channel: Option<String>,
#[serde(default)]
pay_amount: Option<f64>,
#[serde(default)]
pay_currency: Option<String>,
@@ -34,6 +42,8 @@ struct WalletCreateRechargeRequest {
struct NormalizedWalletCreateRechargeRequest {
amount_usd: f64,
payment_method: String,
payment_provider: Option<String>,
payment_channel: Option<String>,
pay_amount: Option<f64>,
pay_currency: Option<String>,
exchange_rate: Option<f64>,
@@ -49,6 +59,10 @@ fn normalize_wallet_create_recharge_request(
if payment_method.is_empty() || payment_method.chars().count() > 30 {
return Err("输入验证失败");
}
let payment_provider = wallet_normalize_optional_string_field(payload.payment_provider, 30)?
.map(|value| value.to_ascii_lowercase());
let payment_channel = wallet_normalize_optional_string_field(payload.payment_channel, 30)?
.map(|value| value.to_ascii_lowercase());
if matches!(payload.pay_amount, Some(value) if !value.is_finite() || value <= 0.0) {
return Err("输入验证失败");
}
@@ -63,6 +77,8 @@ fn normalize_wallet_create_recharge_request(
Ok(NormalizedWalletCreateRechargeRequest {
amount_usd: payload.amount_usd,
payment_method,
payment_provider,
payment_channel,
pay_amount: payload.pay_amount,
pay_currency,
exchange_rate: payload.exchange_rate,
@@ -300,73 +316,137 @@ pub(super) async fn handle_wallet_create_recharge(
let now = Utc::now();
let order_no = wallet_build_order_no(now);
let expires_at = now + chrono::Duration::minutes(30);
let Some(adapter) = PaymentGatewayRegistry::get(&payload.payment_method) else {
return build_auth_error_response(
http::StatusCode::BAD_REQUEST,
format!("unsupported payment_method: {}", payload.payment_method),
false,
);
};
let checkout = match adapter.create_checkout_session(&CreateCheckoutSessionInput {
order_no: order_no.clone(),
amount_usd: payload.amount_usd,
expires_at,
}) {
Ok(value) => value,
Err(detail) => {
return build_auth_error_response(http::StatusCode::BAD_REQUEST, detail, false);
}
};
let outcome = match state
.create_wallet_recharge_order(
aether_data::repository::wallet::CreateWalletRechargeOrderInput {
preferred_wallet_id: wallet.as_ref().map(|value| value.id.clone()),
user_id: auth.user.id.clone(),
amount_usd: payload.amount_usd,
pay_amount: payload.pay_amount,
pay_currency: payload.pay_currency.clone(),
exchange_rate: payload.exchange_rate,
payment_method: payload.payment_method.clone(),
gateway_order_id: checkout.gateway_order_id,
gateway_response: checkout.gateway_response.clone(),
order_no,
expires_at_unix_secs: expires_at.timestamp().max(0) as u64,
},
)
.await
{
Ok(Some(value)) => value,
Ok(None) => return build_wallet_recharge_storage_unavailable_response(),
Err(err) => {
return build_auth_error_response(
http::StatusCode::INTERNAL_SERVER_ERROR,
format!("wallet recharge create failed: {err:?}"),
false,
)
}
};
let order_payload = match outcome {
aether_data::repository::wallet::CreateWalletRechargeOrderOutcome::Created(order) => {
wallet_payment_order_payload_from_record(&order)
}
aether_data::repository::wallet::CreateWalletRechargeOrderOutcome::WalletInactive => {
let uses_epay =
payload.payment_provider.as_deref() == Some("epay") || payload.payment_method == "epay";
if uses_epay {
let config = match load_epay_config(state).await {
Ok(value) => value,
Err(detail) => {
return build_auth_error_response(http::StatusCode::BAD_REQUEST, detail, false);
}
};
if payload.amount_usd < config.min_recharge_usd {
return build_auth_error_response(
http::StatusCode::BAD_REQUEST,
"wallet is not active",
"充值金额低于支付网关最小金额",
false,
)
);
}
};
build_auth_json_response(
http::StatusCode::OK,
json!({
"order": order_payload,
"payment_instructions": sanitize_wallet_gateway_response(Some(checkout.gateway_response)),
}),
None,
let requested_channel = payload.payment_channel.as_deref().or_else(|| {
(payload.payment_method != "epay").then_some(payload.payment_method.as_str())
});
let payment_channel = match resolve_epay_channel(&config, requested_channel) {
Ok(value) => value,
Err(detail) => {
return build_auth_error_response(http::StatusCode::BAD_REQUEST, detail, false);
}
};
let pay_amount = (payload.amount_usd * config.usd_exchange_rate * 100.0).round() / 100.0;
let Some(callback_base_url) = epay_callback_base_url(
config.callback_base_url.as_deref(),
headers,
request_context,
) else {
return build_auth_error_response(
http::StatusCode::BAD_REQUEST,
"epay callback_base_url is required",
false,
);
};
let checkout = build_epay_checkout_url(
&config,
&EpayCheckoutInput {
order_no: order_no.clone(),
channel: payment_channel.clone(),
subject: "钱包充值".to_string(),
pay_amount,
notify_url: format!("{callback_base_url}/api/payment/epay/notify"),
return_url: format!("{callback_base_url}/api/payment/epay/return"),
},
);
let outcome = match state
.create_wallet_recharge_order(
aether_data::repository::wallet::CreateWalletRechargeOrderInput {
preferred_wallet_id: wallet.as_ref().map(|value| value.id.clone()),
user_id: auth.user.id.clone(),
amount_usd: payload.amount_usd,
pay_amount: Some(pay_amount),
pay_currency: Some(config.pay_currency.clone()),
exchange_rate: Some(config.usd_exchange_rate),
payment_method: "epay".to_string(),
payment_provider: Some("epay".to_string()),
payment_channel: Some(payment_channel),
gateway_order_id: order_no.clone(),
gateway_response: checkout.clone(),
order_no,
expires_at_unix_secs: expires_at.timestamp().max(0) as u64,
},
)
.await
{
Ok(Some(value)) => value,
Ok(None) => return build_wallet_recharge_storage_unavailable_response(),
Err(err) => {
return build_auth_error_response(
http::StatusCode::INTERNAL_SERVER_ERROR,
format!("wallet recharge create failed: {err:?}"),
false,
)
}
};
let order_payload = match outcome {
aether_data::repository::wallet::CreateWalletRechargeOrderOutcome::Created(order) => {
wallet_payment_order_payload_from_record(&order)
}
aether_data::repository::wallet::CreateWalletRechargeOrderOutcome::WalletInactive => {
return build_auth_error_response(
http::StatusCode::BAD_REQUEST,
"wallet is not active",
false,
)
}
};
return build_auth_json_response(
http::StatusCode::OK,
json!({
"order": order_payload,
"payment_instructions": sanitize_wallet_gateway_response(Some(checkout)),
}),
None,
);
}
build_auth_error_response(
http::StatusCode::BAD_REQUEST,
format!("unsupported payment_method: {}", payload.payment_method),
false,
)
}
pub(super) async fn handle_wallet_recharge_options(
state: &AppState,
request_context: &GatewayPublicRequestContext,
headers: &http::HeaderMap,
) -> Response<Body> {
if let Err(response) = resolve_authenticated_local_user(state, request_context, headers).await {
return response;
}
let mut methods = Vec::new();
if let Ok(config) = load_epay_config(state).await {
for channel in configured_epay_channels(&config) {
methods.push(json!({
"payment_method": "epay",
"payment_provider": "epay",
"payment_channel": channel.channel,
"display_name": channel.display_name,
"pay_currency": config.pay_currency,
"usd_exchange_rate": config.usd_exchange_rate,
"min_recharge_usd": config.min_recharge_usd,
}));
}
}
build_auth_json_response(http::StatusCode::OK, json!({ "items": methods }), None)
}
pub(super) async fn handle_wallet_recharge_list(
state: &AppState,
request_context: &GatewayPublicRequestContext,