merge(pr287): absorb remaining rust branch changes

Absorb the remaining changes from PR #287 into aether-rust-pioneer after resolving conflicts locally and preserving the admin fixes already landed in this branch.

Closes #287
Co-authored-by: AAEE86 <ppk0227@hotmail.com>
This commit is contained in:
fawney19
2026-04-11 03:07:26 +08:00
64 changed files with 8004 additions and 2530 deletions

View File

@@ -59,4 +59,5 @@ uuid.workspace = true
webpki-roots.workspace = true
[dev-dependencies]
aether-testkit.workspace = true
tracing-subscriber.workspace = true

View File

@@ -77,7 +77,7 @@ mod tests {
);
assert_eq!(
sync_cli_response_conversion_kind("claude:cli", "openai:compact"),
None
Some(SyncCliResponseConversionKind::ToOpenAIFamilyCli)
);
assert_eq!(
sync_cli_response_conversion_kind("openai:compact", "claude:cli"),

View File

@@ -192,7 +192,7 @@ async fn resolve_tunnel_owner_affinity_from_transport(
state: PlannerAppState<'_>,
transport: &GatewayProviderTransportSnapshot,
) -> TunnelOwnerAffinityBucket {
let Some(proxy) = resolve_transport_proxy_snapshot(&transport) else {
let Some(proxy) = resolve_transport_proxy_snapshot(transport) else {
return TunnelOwnerAffinityBucket::Neutral;
};
if proxy.enabled == Some(false) {

View File

@@ -53,8 +53,7 @@ pub(super) async fn build_cross_format_local_openai_chat_decision_payload_for_ca
else {
return None;
};
if !request_pair_allowed_for_transport(&transport, "openai:chat", provider_api_format.as_str())
{
if !request_pair_allowed_for_transport(transport, "openai:chat", provider_api_format.as_str()) {
let skip_reason =
if request_conversion_requires_enable_flag("openai:chat", provider_api_format.as_str())
&& !transport.provider.enable_format_conversion

View File

@@ -3,8 +3,8 @@ use aether_data_contracts::repository::candidate_selection::StoredMinimalCandida
use aether_scheduler_core::{
auth_constraints_allow_api_format, build_minimal_candidate_selection,
collect_global_model_names_for_required_capability, normalize_api_format,
resolve_requested_global_model_name, SchedulerAuthConstraints,
SchedulerMinimalCandidateSelectionCandidate, SchedulerPriorityMode,
resolve_requested_global_model_name, BuildMinimalCandidateSelectionInput,
SchedulerAuthConstraints, SchedulerMinimalCandidateSelectionCandidate, SchedulerPriorityMode,
};
use async_trait::async_trait;
use std::collections::BTreeSet;
@@ -167,17 +167,17 @@ pub(crate) async fn read_minimal_candidate_selection_with_priority_mode_and_affi
return Ok(Vec::new());
};
let auth_constraints = auth_snapshot.map(auth_snapshot_constraints);
build_minimal_candidate_selection(
build_minimal_candidate_selection(BuildMinimalCandidateSelectionInput {
rows,
&normalized_api_format,
normalized_api_format: &normalized_api_format,
requested_model_name,
resolved_global_model_name.as_str(),
resolved_global_model_name: resolved_global_model_name.as_str(),
require_streaming,
required_capabilities,
auth_constraints.as_ref(),
auth_constraints: auth_constraints.as_ref(),
affinity_key,
priority_mode,
)
})
}
pub(crate) async fn read_global_model_names_for_required_capability(

View File

@@ -29,7 +29,8 @@ use aether_data_contracts::repository::video_tasks::{
UpsertVideoTask, VideoTaskLookupKey, VideoTaskStatus, VideoTaskWriteRepository,
};
use aether_scheduler_core::{
build_minimal_candidate_selection, SchedulerAuthConstraints, SchedulerPriorityMode,
build_minimal_candidate_selection, BuildMinimalCandidateSelectionInput,
SchedulerAuthConstraints, SchedulerPriorityMode,
};
use async_trait::async_trait;
use serde_json::json;
@@ -664,17 +665,17 @@ async fn data_state_reads_minimal_candidate_selection_with_auth_filters() {
.map(|items| items.to_vec()),
};
let selection = build_minimal_candidate_selection(
let selection = build_minimal_candidate_selection(BuildMinimalCandidateSelectionInput {
rows,
"openai:chat",
"gpt-4.1",
"gpt-4.1",
false,
None,
Some(&auth_constraints),
Some(auth_snapshot.api_key_id.as_str()),
SchedulerPriorityMode::Provider,
)
normalized_api_format: "openai:chat",
requested_model_name: "gpt-4.1",
resolved_global_model_name: "gpt-4.1",
require_streaming: false,
required_capabilities: None,
auth_constraints: Some(&auth_constraints),
affinity_key: Some(auth_snapshot.api_key_id.as_str()),
priority_mode: SchedulerPriorityMode::Provider,
})
.expect("selection should read");
assert_eq!(selection.len(), 2);

View File

@@ -922,7 +922,6 @@ async fn execute_stream_from_frame_stream(
let request_id_for_report = request_id.to_string();
let request_id_for_report_log = short_request_id(request_id);
let candidate_id_for_report = candidate_id.map(ToOwned::to_owned);
let mut buffered_frames = buffered_frames;
tokio::spawn(async move {
let mut provider_buffered_body = provider_prefetched_body_for_report;
let mut buffered_body = prefetched_body_for_report;

View File

@@ -93,3 +93,45 @@ pub(crate) async fn build_admin_global_model_payload(
}
Some(payload)
}
#[cfg(test)]
mod tests {
use super::build_admin_global_model_response;
use aether_data_contracts::repository::global_models::StoredAdminGlobalModel;
use serde_json::json;
#[test]
fn admin_global_model_response_uses_stored_stats_and_usage_count() {
let global_model = StoredAdminGlobalModel::new(
"global-minimax".to_string(),
"MiniMax-M2.7".to_string(),
"MiniMax-M2.7".to_string(),
true,
None,
Some(json!({
"tiers": [{
"up_to": null,
"input_price_per_1m": 0.3,
"output_price_per_1m": 1.2,
}]
})),
Some(json!(["cache_1h"])),
Some(json!({
"family": "minimax",
"streaming": true,
})),
3,
3,
1,
Some(1_711_000_000),
Some(1_711_000_100),
)
.expect("global model should build");
let payload = build_admin_global_model_response(&global_model, 1_711_000_100);
assert_eq!(payload["provider_count"], 3);
assert_eq!(payload["active_provider_count"], 3);
assert_eq!(payload["usage_count"], 1);
}
}

View File

@@ -3,35 +3,23 @@ use crate::handlers::admin::provider::shared::paths::{
};
use crate::handlers::admin::request::AdminRequestContext;
use crate::GatewayError;
use aether_admin::provider::ops::{get_architecture, list_architectures};
use axum::{
body::Body,
http,
response::{IntoResponse, Response},
Json,
};
use serde_json::{json, Value};
static ADMIN_PROVIDER_OPS_ARCHITECTURES_ALL: std::sync::LazyLock<Vec<Value>> =
std::sync::LazyLock::new(|| {
serde_json::from_str(include_str!("architectures.all.json"))
.expect("admin provider ops architectures fixture should parse")
});
fn admin_provider_ops_architectures_list_payload() -> Vec<Value> {
ADMIN_PROVIDER_OPS_ARCHITECTURES_ALL
.iter()
.filter(|item| item.get("architecture_id").and_then(Value::as_str) != Some("generic_api"))
.cloned()
fn admin_provider_ops_architectures_list_payload() -> Vec<serde_json::Value> {
list_architectures(false)
.into_iter()
.map(|architecture| architecture.api_payload())
.collect()
}
fn admin_provider_ops_architecture_payload(architecture_id: &str) -> Option<Value> {
ADMIN_PROVIDER_OPS_ARCHITECTURES_ALL
.iter()
.find_map(|item| {
(item.get("architecture_id").and_then(Value::as_str) == Some(architecture_id))
.then(|| item.clone())
})
fn admin_provider_ops_architecture_payload(architecture_id: &str) -> Option<serde_json::Value> {
get_architecture(architecture_id).map(|architecture| architecture.api_payload())
}
pub(super) async fn maybe_build_local_admin_provider_ops_architectures_response(
@@ -61,7 +49,7 @@ pub(super) async fn maybe_build_local_admin_provider_ops_architectures_response(
return Ok(Some(
(
http::StatusCode::NOT_FOUND,
Json(json!({ "detail": "架构不存在" })),
Json(serde_json::json!({ "detail": "架构不存在" })),
)
.into_response(),
));
@@ -72,7 +60,7 @@ pub(super) async fn maybe_build_local_admin_provider_ops_architectures_response(
Some(payload) => Json(payload).into_response(),
None => (
http::StatusCode::NOT_FOUND,
Json(json!({ "detail": format!("架构 {architecture_id} 不存在") })),
Json(serde_json::json!({ "detail": format!("架构 {architecture_id} 不存在") })),
)
.into_response(),
},

View File

@@ -1,9 +1,11 @@
use super::super::super::support::AdminProviderOpsCheckinOutcome;
use super::super::super::verify::admin_provider_ops_execute_proxy_json_request;
use super::super::support::{admin_provider_ops_json_object_map, admin_provider_ops_request_url};
use super::shared::{
admin_provider_ops_checkin_already_done, admin_provider_ops_checkin_auth_failure,
};
use crate::handlers::admin::request::AdminAppState;
use aether_contracts::ProxySnapshot;
use serde_json::json;
pub(in super::super) async fn admin_provider_ops_probe_new_api_checkin(
@@ -12,6 +14,7 @@ pub(in super::super) async fn admin_provider_ops_probe_new_api_checkin(
action_config: &serde_json::Map<String, serde_json::Value>,
headers: &reqwest::header::HeaderMap,
has_cookie: bool,
proxy_snapshot: Option<&ProxySnapshot>,
) -> Option<AdminProviderOpsCheckinOutcome> {
let endpoint = action_config
.get("checkin_endpoint")
@@ -24,22 +27,47 @@ pub(in super::super) async fn admin_provider_ops_probe_new_api_checkin(
&admin_provider_ops_json_object_map(json!({ "endpoint": endpoint })),
endpoint,
);
let response = match state
.http_client()
.request(reqwest::Method::POST, url)
.headers(headers.clone())
.send()
let (status, response_json) = if let Some(proxy_snapshot) = proxy_snapshot {
match admin_provider_ops_execute_proxy_json_request(
state,
"provider-ops-action:probe_checkin",
reqwest::Method::POST,
&url,
headers,
None,
proxy_snapshot,
)
.await
{
Ok(response) => response,
Err(_) => return None,
{
Ok(result) => result,
Err(_) => return None,
}
} else {
let response = match state
.http_client()
.request(reqwest::Method::POST, url)
.headers(headers.clone())
.send()
.await
{
Ok(response) => response,
Err(_) => return None,
};
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => {
serde_json::from_slice::<serde_json::Value>(&bytes).unwrap_or_else(|_| json!({}))
}
Err(_) => json!({}),
};
(status, response_json)
};
if response.status() == http::StatusCode::NOT_FOUND {
if status == http::StatusCode::NOT_FOUND {
return None;
}
if matches!(
response.status(),
status,
http::StatusCode::UNAUTHORIZED | http::StatusCode::FORBIDDEN
) {
return has_cookie.then(|| AdminProviderOpsCheckinOutcome {
@@ -49,12 +77,6 @@ pub(in super::super) async fn admin_provider_ops_probe_new_api_checkin(
});
}
let response_json = match response.bytes().await {
Ok(bytes) => {
serde_json::from_slice::<serde_json::Value>(&bytes).unwrap_or_else(|_| json!({}))
}
Err(_) => json!({}),
};
let message = response_json
.get("message")
.and_then(serde_json::Value::as_str)

View File

@@ -1,4 +1,5 @@
use super::super::super::support::ADMIN_PROVIDER_OPS_ACTION_RUST_ONLY_MESSAGE;
use super::super::super::verify::admin_provider_ops_execute_proxy_json_request;
use super::super::responses::{
admin_provider_ops_action_error, admin_provider_ops_action_not_supported,
admin_provider_ops_action_response,
@@ -9,17 +10,20 @@ use super::shared::{
admin_provider_ops_checkin_payload,
};
use crate::handlers::admin::request::AdminAppState;
use aether_admin::provider::ops::{ProviderOpsArchitectureSpec, ProviderOpsCheckinMode};
use aether_contracts::ProxySnapshot;
pub(in super::super) async fn admin_provider_ops_run_checkin_action(
state: &AdminAppState<'_>,
base_url: &str,
architecture_id: &str,
architecture: &ProviderOpsArchitectureSpec,
action_config: &serde_json::Map<String, serde_json::Value>,
headers: &reqwest::header::HeaderMap,
has_cookie: bool,
proxy_snapshot: Option<&ProxySnapshot>,
) -> serde_json::Value {
let start = std::time::Instant::now();
if !matches!(architecture_id, "generic_api" | "new_api") {
if architecture.checkin_mode != ProviderOpsCheckinMode::NewApiCompatible {
return admin_provider_ops_action_not_supported(
"checkin",
ADMIN_PROVIDER_OPS_ACTION_RUST_ONLY_MESSAGE,
@@ -28,49 +32,82 @@ pub(in super::super) async fn admin_provider_ops_run_checkin_action(
let url = admin_provider_ops_request_url(base_url, action_config, "/api/user/checkin");
let method = admin_provider_ops_request_method(action_config, "POST");
let response = match state
.http_client()
.request(method, url)
.headers(headers.clone())
.send()
let (status, response_json) = if let Some(proxy_snapshot) = proxy_snapshot {
match admin_provider_ops_execute_proxy_json_request(
state,
&format!(
"provider-ops-action:{}:checkin",
architecture.architecture_id
),
method,
&url,
headers,
None,
proxy_snapshot,
)
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => {
return admin_provider_ops_action_error("network_error", "checkin", "请求超时", None);
}
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"checkin",
format!("网络错误: {err}"),
None,
);
}
};
let response_time_ms = Some(start.elapsed().as_millis() as u64);
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => match serde_json::from_slice::<serde_json::Value>(&bytes) {
Ok(value) => value,
Err(_) => {
{
Ok(result) => result,
Err(err) => {
return admin_provider_ops_action_error(
"parse_error",
"network_error",
"checkin",
"响应不是有效的 JSON",
response_time_ms,
admin_provider_ops_network_error_message(&err),
None,
);
}
},
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"checkin",
format!("网络错误: {err}"),
response_time_ms,
);
}
} else {
let response = match state
.http_client()
.request(method, url)
.headers(headers.clone())
.send()
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => {
return admin_provider_ops_action_error(
"network_error",
"checkin",
"请求超时",
None,
);
}
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"checkin",
format!("网络错误: {err}"),
None,
);
}
};
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => match serde_json::from_slice::<serde_json::Value>(&bytes) {
Ok(value) => value,
Err(_) => {
return admin_provider_ops_action_error(
"parse_error",
"checkin",
"响应不是有效的 JSON",
Some(start.elapsed().as_millis() as u64),
);
}
},
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"checkin",
format!("网络错误: {err}"),
Some(start.elapsed().as_millis() as u64),
);
}
};
(status, response_json)
};
let response_time_ms = Some(start.elapsed().as_millis() as u64);
if status == http::StatusCode::NOT_FOUND {
return admin_provider_ops_action_error(
@@ -194,3 +231,12 @@ pub(in super::super) async fn admin_provider_ops_run_checkin_action(
response_time_ms,
)
}
fn admin_provider_ops_network_error_message(error: &str) -> String {
let normalized = error.trim();
let lower = normalized.to_ascii_lowercase();
if lower.contains("timeout") || normalized.contains("超时") {
return "请求超时".to_string();
}
format!("网络错误: {normalized}")
}

View File

@@ -1,5 +1,5 @@
use super::super::super::verify::admin_provider_ops_value_as_f64;
use super::super::support::admin_provider_ops_checkin_data;
use aether_admin::provider::ops::admin_provider_ops_value_as_f64;
fn admin_provider_ops_message_contains_any(message: &str, indicators: &[&str]) -> bool {
let normalized = message.trim().to_ascii_lowercase();

View File

@@ -8,8 +8,13 @@ use super::config::{
admin_provider_ops_decrypted_credentials, resolve_admin_provider_ops_base_url,
};
use super::support::ADMIN_PROVIDER_OPS_ACTION_RUST_ONLY_MESSAGE;
use super::verify::admin_provider_ops_verify_headers;
use super::verify::{
admin_provider_ops_anyrouter_acw_cookie, admin_provider_ops_resolve_proxy_snapshot,
};
use crate::handlers::admin::request::AdminAppState;
use aether_admin::provider::ops::{
build_headers, get_architecture, normalize_architecture_id, resolve_action_config,
};
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogProvider,
};
@@ -29,7 +34,7 @@ pub(super) fn admin_provider_ops_is_valid_action_type(action_type: &str) -> bool
pub(crate) async fn admin_provider_ops_local_action_response(
state: &AdminAppState<'_>,
_provider_id: &str,
provider_id: &str,
provider: Option<&StoredProviderCatalogProvider>,
endpoints: &[StoredProviderCatalogEndpoint],
action_type: &str,
@@ -44,24 +49,14 @@ pub(crate) async fn admin_provider_ops_local_action_response(
let architecture_id = provider_ops_config
.get("architecture_id")
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("generic_api");
let connector_config = admin_provider_ops_connector_object(provider_ops_config)
.and_then(|connector| connector.get("config"))
.and_then(serde_json::Value::as_object)
.cloned()
.unwrap_or_default();
if support::admin_provider_ops_should_use_rust_only_action_stub(
architecture_id,
&connector_config,
) {
let architecture_id = normalize_architecture_id(architecture_id);
let Some(architecture) = get_architecture(architecture_id) else {
return responses::admin_provider_ops_action_not_supported(
action_type,
ADMIN_PROVIDER_OPS_ACTION_RUST_ONLY_MESSAGE,
);
}
};
let Some(base_url) =
resolve_admin_provider_ops_base_url(provider, endpoints, Some(provider_ops_config))
else {
@@ -70,20 +65,42 @@ pub(crate) async fn admin_provider_ops_local_action_response(
"Provider 未配置 base_url",
);
};
let mut connector_config = admin_provider_ops_connector_object(provider_ops_config)
.and_then(|connector| connector.get("config"))
.and_then(serde_json::Value::as_object)
.cloned()
.unwrap_or_default();
if architecture_id == "anyrouter" {
if let Some(challenge) =
admin_provider_ops_anyrouter_acw_cookie(state, &base_url, Some(&connector_config)).await
{
connector_config.insert(
"acw_cookie".to_string(),
serde_json::Value::String(challenge.acw_cookie),
);
}
}
let proxy_snapshot =
admin_provider_ops_resolve_proxy_snapshot(state, Some(&connector_config)).await;
let credentials = admin_provider_ops_decrypted_credentials(
state,
admin_provider_ops_config_object(provider)
.and_then(admin_provider_ops_connector_object)
.and_then(|connector| connector.get("credentials")),
);
let headers =
match admin_provider_ops_verify_headers(architecture_id, &connector_config, &credentials) {
Ok(headers) => headers,
Err(message) => {
return responses::admin_provider_ops_action_not_configured(action_type, message);
}
};
let Some(action_config) = support::admin_provider_ops_resolved_action_config(
let headers = match build_headers(
architecture.architecture_id,
&connector_config,
&credentials,
) {
Ok(headers) => headers,
Err(message) => {
return responses::admin_provider_ops_action_not_configured(action_type, message);
}
};
let Some(action_config) = resolve_action_config(
architecture_id,
provider_ops_config,
action_type,
@@ -99,26 +116,32 @@ pub(crate) async fn admin_provider_ops_local_action_response(
"query_balance" => {
query_balance::admin_provider_ops_run_query_balance_action(
state,
provider_id,
provider,
&architecture,
&base_url,
architecture_id,
&action_config,
&headers,
&credentials,
proxy_snapshot.as_ref(),
)
.await
}
"checkin" => {
let has_cookie = credentials
.get("cookie")
.and_then(serde_json::Value::as_str)
.is_some_and(|value| !value.trim().is_empty());
let has_cookie = ["cookie", "session_cookie"].into_iter().any(|key| {
credentials
.get(key)
.and_then(serde_json::Value::as_str)
.is_some_and(|value| !value.trim().is_empty())
});
checkin::admin_provider_ops_run_checkin_action(
state,
&base_url,
architecture_id,
&architecture,
&action_config,
&headers,
has_cookie,
proxy_snapshot.as_ref(),
)
.await
}

View File

@@ -1,50 +1,71 @@
mod parsers;
mod sub2api;
mod yescode;
use super::super::support::{
AdminProviderOpsCheckinOutcome, ADMIN_PROVIDER_OPS_ACTION_RUST_ONLY_MESSAGE,
};
use super::super::support::AdminProviderOpsCheckinOutcome;
use super::super::verify::admin_provider_ops_execute_proxy_json_request;
use super::checkin::admin_provider_ops_probe_new_api_checkin;
use super::responses::{
admin_provider_ops_action_error, admin_provider_ops_action_not_supported,
admin_provider_ops_action_response,
};
use super::support::{
admin_provider_ops_is_cookie_auth_architecture, admin_provider_ops_request_method,
admin_provider_ops_request_url,
};
use super::responses::{admin_provider_ops_action_error, admin_provider_ops_action_response};
use super::support::{admin_provider_ops_request_method, admin_provider_ops_request_url};
use crate::handlers::admin::request::AdminAppState;
use aether_admin::provider::ops::{
attach_balance_checkin_outcome, parse_query_balance_payload, ProviderOpsArchitectureSpec,
ProviderOpsBalanceMode, ProviderOpsCheckinMode,
};
use aether_contracts::ProxySnapshot;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider;
pub(super) async fn admin_provider_ops_run_query_balance_action(
state: &AdminAppState<'_>,
provider_id: &str,
provider: &StoredProviderCatalogProvider,
architecture: &ProviderOpsArchitectureSpec,
base_url: &str,
architecture_id: &str,
action_config: &serde_json::Map<String, serde_json::Value>,
headers: &reqwest::header::HeaderMap,
credentials: &serde_json::Map<String, serde_json::Value>,
proxy_snapshot: Option<&ProxySnapshot>,
) -> serde_json::Value {
if architecture_id == "yescode" {
return yescode::admin_provider_ops_yescode_balance_payload(
state,
base_url,
headers,
action_config,
)
.await;
match architecture.balance_mode {
ProviderOpsBalanceMode::YescodeCombined => {
return yescode::admin_provider_ops_yescode_balance_payload(
state,
base_url,
headers,
action_config,
proxy_snapshot,
)
.await;
}
ProviderOpsBalanceMode::Sub2ApiDualRequest => {
return sub2api::admin_provider_ops_sub2api_balance_payload(
state,
provider_id,
provider,
base_url,
action_config,
credentials,
proxy_snapshot,
)
.await;
}
ProviderOpsBalanceMode::SingleRequest => {}
}
let mut balance_checkin = None::<AdminProviderOpsCheckinOutcome>;
if matches!(architecture_id, "generic_api" | "new_api") {
let has_cookie = credentials
.get("cookie")
.and_then(serde_json::Value::as_str)
.is_some_and(|value| !value.trim().is_empty());
if architecture.checkin_mode == ProviderOpsCheckinMode::NewApiCompatible {
let has_cookie = ["session_cookie", "cookie"].into_iter().any(|key| {
credentials
.get(key)
.and_then(serde_json::Value::as_str)
.is_some_and(|value| !value.trim().is_empty())
});
balance_checkin = admin_provider_ops_probe_new_api_checkin(
state,
base_url,
action_config,
headers,
has_cookie,
proxy_snapshot,
)
.await;
}
@@ -52,57 +73,85 @@ pub(super) async fn admin_provider_ops_run_query_balance_action(
let start = std::time::Instant::now();
let url = admin_provider_ops_request_url(base_url, action_config, "/api/user/balance");
let method = admin_provider_ops_request_method(action_config, "GET");
let response = match state
.http_client()
.request(method, url)
.headers(headers.clone())
.send()
let (status, response_json) = if let Some(proxy_snapshot) = proxy_snapshot {
match admin_provider_ops_execute_proxy_json_request(
state,
&format!(
"provider-ops-action:{}:query_balance:{provider_id}",
architecture.architecture_id
),
method,
&url,
headers,
None,
proxy_snapshot,
)
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
"请求超时",
None,
);
}
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
format!("网络错误: {err}"),
None,
);
}
};
let response_time_ms = Some(start.elapsed().as_millis() as u64);
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => match serde_json::from_slice::<serde_json::Value>(&bytes) {
Ok(value) => value,
Err(_) => {
{
Ok(result) => result,
Err(err) => {
return admin_provider_ops_action_error(
"parse_error",
"network_error",
"query_balance",
"响应不是有效的 JSON",
response_time_ms,
admin_provider_ops_network_error_message(&err),
None,
);
}
},
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
format!("网络错误: {err}"),
response_time_ms,
);
}
} else {
let response = match state
.http_client()
.request(method, url)
.headers(headers.clone())
.send()
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
"请求超时",
None,
);
}
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
format!("网络错误: {err}"),
None,
);
}
};
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => match serde_json::from_slice::<serde_json::Value>(&bytes) {
Ok(value) => value,
Err(_) => {
return admin_provider_ops_action_error(
"parse_error",
"query_balance",
"响应不是有效的 JSON",
Some(start.elapsed().as_millis() as u64),
);
}
},
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
format!("网络错误: {err}"),
Some(start.elapsed().as_millis() as u64),
);
}
};
(status, response_json)
};
let response_time_ms = Some(start.elapsed().as_millis() as u64);
if status != http::StatusCode::OK {
let cookie_auth = admin_provider_ops_is_cookie_auth_architecture(architecture_id);
let cookie_auth = architecture.query_balance_cookie_auth_errors;
return match status {
http::StatusCode::UNAUTHORIZED => admin_provider_ops_action_error(
"auth_failed",
@@ -149,50 +198,25 @@ pub(super) async fn admin_provider_ops_run_query_balance_action(
};
}
let data = match architecture_id {
"generic_api" | "new_api" => {
match parsers::admin_provider_ops_new_api_balance_payload(action_config, &response_json)
{
Ok(data) => data,
Err(message) => {
return admin_provider_ops_action_error(
"unknown_error",
"query_balance",
message,
response_time_ms,
);
}
}
}
"cubence" => {
match parsers::admin_provider_ops_cubence_balance_payload(action_config, &response_json)
{
Ok(data) => data,
Err(message) => {
return admin_provider_ops_action_error(
"parse_error",
"query_balance",
message,
response_time_ms,
);
}
}
}
"nekocode" => match parsers::admin_provider_ops_nekocode_balance_payload(&response_json) {
Ok(data) => data,
Err(message) => {
return admin_provider_ops_action_error(
"parse_error",
"query_balance",
message,
response_time_ms,
);
}
},
_ => {
return admin_provider_ops_action_not_supported(
let data = match parse_query_balance_payload(
architecture.architecture_id,
action_config,
&response_json,
) {
Ok(data) => data,
Err(message) => {
return admin_provider_ops_action_error(
if architecture.architecture_id == "generic_api"
|| architecture.architecture_id == "new_api"
|| architecture.architecture_id == "anyrouter"
{
"unknown_error"
} else {
"parse_error"
},
"query_balance",
ADMIN_PROVIDER_OPS_ACTION_RUST_ONLY_MESSAGE,
message,
response_time_ms,
);
}
};
@@ -206,7 +230,16 @@ pub(super) async fn admin_provider_ops_run_query_balance_action(
86400,
);
if let Some(outcome) = balance_checkin.as_ref() {
parsers::admin_provider_ops_attach_balance_checkin_outcome(&mut payload, outcome);
attach_balance_checkin_outcome(&mut payload, outcome);
}
payload
}
fn admin_provider_ops_network_error_message(error: &str) -> String {
let normalized = error.trim();
let lower = normalized.to_ascii_lowercase();
if lower.contains("timeout") || normalized.contains("超时") {
return "请求超时".to_string();
}
format!("网络错误: {normalized}")
}

View File

@@ -1,312 +0,0 @@
use super::super::super::support::AdminProviderOpsCheckinOutcome;
use super::super::super::verify::admin_provider_ops_value_as_f64;
use super::super::support::{
admin_provider_ops_balance_data, admin_provider_ops_parse_rfc3339_unix_secs,
admin_provider_ops_quota_divisor,
};
use serde_json::json;
pub(super) fn admin_provider_ops_new_api_balance_payload(
action_config: &serde_json::Map<String, serde_json::Value>,
response_json: &serde_json::Value,
) -> Result<serde_json::Value, String> {
let user_data = if response_json
.get("success")
.and_then(serde_json::Value::as_bool)
== Some(true)
&& response_json
.get("data")
.is_some_and(serde_json::Value::is_object)
{
response_json.get("data")
} else if response_json
.get("success")
.and_then(serde_json::Value::as_bool)
== Some(false)
{
return Err(response_json
.get("message")
.and_then(serde_json::Value::as_str)
.unwrap_or("业务状态码表示失败")
.to_string());
} else {
Some(response_json)
};
let Some(user_data) = user_data.and_then(serde_json::Value::as_object) else {
return Err("响应格式无效".to_string());
};
let quota_divisor = admin_provider_ops_quota_divisor(action_config);
let total_available =
admin_provider_ops_value_as_f64(user_data.get("quota")).map(|value| value / quota_divisor);
let total_used = admin_provider_ops_value_as_f64(user_data.get("used_quota"))
.map(|value| value / quota_divisor);
Ok(admin_provider_ops_balance_data(
None,
total_used,
total_available,
action_config
.get("currency")
.and_then(serde_json::Value::as_str)
.unwrap_or("USD"),
serde_json::Map::new(),
))
}
pub(super) fn admin_provider_ops_cubence_balance_payload(
action_config: &serde_json::Map<String, serde_json::Value>,
response_json: &serde_json::Value,
) -> Result<serde_json::Value, String> {
let response_data = response_json
.get("data")
.and_then(serde_json::Value::as_object)
.ok_or_else(|| "响应格式无效".to_string())?;
let balance_data = response_data
.get("balance")
.and_then(serde_json::Value::as_object)
.cloned()
.unwrap_or_default();
let subscription_limits = response_data
.get("subscription_limits")
.and_then(serde_json::Value::as_object)
.cloned()
.unwrap_or_default();
let mut extra = serde_json::Map::new();
if let Some(five_hour) = subscription_limits
.get("five_hour")
.and_then(serde_json::Value::as_object)
{
extra.insert(
"five_hour_limit".to_string(),
json!({
"limit": five_hour.get("limit"),
"used": five_hour.get("used"),
"remaining": five_hour.get("remaining"),
"resets_at": five_hour.get("resets_at"),
}),
);
}
if let Some(weekly) = subscription_limits
.get("weekly")
.and_then(serde_json::Value::as_object)
{
extra.insert(
"weekly_limit".to_string(),
json!({
"limit": weekly.get("limit"),
"used": weekly.get("used"),
"remaining": weekly.get("remaining"),
"resets_at": weekly.get("resets_at"),
}),
);
}
for key in [
"normal_balance_dollar",
"subscription_balance_dollar",
"charity_balance_dollar",
] {
if let Some(value) = balance_data.get(key) {
extra.insert(
key.trim_end_matches("_dollar").replace("_dollar", ""),
value.clone(),
);
}
}
if let Some(value) = balance_data.get("normal_balance_dollar") {
extra.insert("normal_balance".to_string(), value.clone());
}
if let Some(value) = balance_data.get("subscription_balance_dollar") {
extra.insert("subscription_balance".to_string(), value.clone());
}
if let Some(value) = balance_data.get("charity_balance_dollar") {
extra.insert("charity_balance".to_string(), value.clone());
}
Ok(admin_provider_ops_balance_data(
None,
None,
admin_provider_ops_value_as_f64(balance_data.get("total_balance_dollar")),
action_config
.get("currency")
.and_then(serde_json::Value::as_str)
.unwrap_or("USD"),
extra,
))
}
pub(super) fn admin_provider_ops_nekocode_balance_payload(
response_json: &serde_json::Value,
) -> Result<serde_json::Value, String> {
let response_data = response_json
.get("data")
.and_then(serde_json::Value::as_object)
.ok_or_else(|| "响应格式无效".to_string())?;
let subscription = response_data
.get("subscription")
.and_then(serde_json::Value::as_object)
.cloned()
.unwrap_or_default();
let balance = admin_provider_ops_value_as_f64(response_data.get("balance"));
let daily_quota_limit = admin_provider_ops_value_as_f64(subscription.get("daily_quota_limit"));
let daily_remaining_quota =
admin_provider_ops_value_as_f64(subscription.get("daily_remaining_quota"));
let daily_used = match (daily_quota_limit, daily_remaining_quota) {
(Some(limit), Some(remaining)) => Some(limit - remaining),
_ => None,
};
let mut extra = serde_json::Map::new();
for key in [
"plan_name",
"status",
"daily_quota_limit",
"daily_remaining_quota",
"effective_start_date",
"effective_end_date",
] {
if let Some(value) = subscription.get(key) {
extra.insert(
match key {
"status" => "subscription_status",
other => other,
}
.to_string(),
value.clone(),
);
}
}
if let Some(value) = daily_used {
extra.insert("daily_used_quota".to_string(), json!(value));
}
if let Some(month_data) = response_data
.get("month")
.and_then(serde_json::Value::as_object)
{
extra.insert(
"month_stats".to_string(),
json!({
"total_input_tokens": month_data.get("total_input_tokens"),
"total_output_tokens": month_data.get("total_output_tokens"),
"total_quota": month_data.get("total_quota"),
"total_requests": month_data.get("total_requests"),
}),
);
}
if let Some(today_data) = response_data
.get("today")
.and_then(serde_json::Value::as_object)
{
if let Some(stats) = today_data.get("stats") {
extra.insert("today_stats".to_string(), stats.clone());
}
}
Ok(admin_provider_ops_balance_data(
daily_quota_limit,
daily_used,
balance,
"USD",
extra,
))
}
pub(super) fn admin_provider_ops_yescode_balance_extra(
combined_data: &serde_json::Map<String, serde_json::Value>,
) -> serde_json::Map<String, serde_json::Value> {
let pay_as_you_go =
admin_provider_ops_value_as_f64(combined_data.get("pay_as_you_go_balance")).unwrap_or(0.0);
let subscription =
admin_provider_ops_value_as_f64(combined_data.get("subscription_balance")).unwrap_or(0.0);
let plan = combined_data
.get("subscription_plan")
.and_then(serde_json::Value::as_object)
.cloned()
.unwrap_or_default();
let daily_balance =
admin_provider_ops_value_as_f64(plan.get("daily_balance")).unwrap_or(subscription);
let weekly_limit = admin_provider_ops_value_as_f64(
combined_data
.get("weekly_limit")
.or_else(|| plan.get("weekly_limit")),
);
let weekly_spent =
admin_provider_ops_value_as_f64(combined_data.get("weekly_spent_balance")).unwrap_or(0.0);
let subscription_available = weekly_limit
.map(|limit| (limit - weekly_spent).max(0.0).min(subscription))
.unwrap_or(subscription);
let mut extra = serde_json::Map::new();
extra.insert("pay_as_you_go_balance".to_string(), json!(pay_as_you_go));
extra.insert("daily_limit".to_string(), json!(daily_balance));
if let Some(limit) = weekly_limit {
extra.insert("weekly_limit".to_string(), json!(limit));
}
extra.insert("weekly_spent".to_string(), json!(weekly_spent));
if let Some(last_week_reset) =
admin_provider_ops_parse_rfc3339_unix_secs(combined_data.get("last_week_reset"))
{
extra.insert(
"weekly_resets_at".to_string(),
json!(last_week_reset + 7 * 24 * 3600),
);
}
if let Some(last_daily_add) =
admin_provider_ops_parse_rfc3339_unix_secs(combined_data.get("last_daily_balance_add"))
{
extra.insert(
"daily_resets_at".to_string(),
json!(last_daily_add + 24 * 3600),
);
}
let daily_spent = if let Some(limit) = weekly_limit {
daily_balance - daily_balance.min(subscription_available.min(limit.max(0.0)))
} else {
(daily_balance - subscription).max(0.0)
};
extra.insert("daily_spent".to_string(), json!(daily_spent));
extra.insert(
"_subscription_available".to_string(),
json!(subscription_available),
);
extra.insert(
"_total_available".to_string(),
json!(pay_as_you_go + subscription_available),
);
extra
}
pub(super) fn admin_provider_ops_attach_balance_checkin_outcome(
action_payload: &mut serde_json::Value,
outcome: &AdminProviderOpsCheckinOutcome,
) {
if let Some(data) = action_payload
.get_mut("data")
.and_then(serde_json::Value::as_object_mut)
{
let extra = data
.entry("extra".to_string())
.or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
if let Some(extra) = extra.as_object_mut() {
if outcome.cookie_expired {
extra.insert("cookie_expired".to_string(), serde_json::Value::Bool(true));
extra.insert(
"cookie_expired_message".to_string(),
serde_json::Value::String(outcome.message.clone()),
);
} else {
extra.insert(
"checkin_success".to_string(),
outcome
.success
.map(serde_json::Value::Bool)
.unwrap_or(serde_json::Value::Null),
);
extra.insert(
"checkin_message".to_string(),
serde_json::Value::String(outcome.message.clone()),
);
}
}
}
if outcome.cookie_expired {
if let Some(object) = action_payload.as_object_mut() {
object.insert("status".to_string(), json!("auth_expired"));
}
}
}

View File

@@ -0,0 +1,264 @@
use super::super::super::config::persist_admin_provider_ops_runtime_credentials;
use super::super::super::verify::{
admin_provider_ops_execute_proxy_json_request, admin_provider_ops_sub2api_exchange_token,
admin_provider_ops_sub2api_request_url,
};
use super::super::responses::{
admin_provider_ops_action_error, admin_provider_ops_action_response,
};
use super::super::support::admin_provider_ops_json_object_map;
use crate::handlers::admin::request::AdminAppState;
use aether_admin::provider::ops::parse_sub2api_balance_payload;
use aether_contracts::ProxySnapshot;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider;
use serde_json::{json, Value};
use tracing::warn;
pub(super) async fn admin_provider_ops_sub2api_balance_payload(
state: &AdminAppState<'_>,
provider_id: &str,
provider: &StoredProviderCatalogProvider,
base_url: &str,
action_config: &serde_json::Map<String, serde_json::Value>,
credentials: &serde_json::Map<String, serde_json::Value>,
proxy_snapshot: Option<&ProxySnapshot>,
) -> serde_json::Value {
let start = std::time::Instant::now();
let (access_token, updated_credentials, _frontend_updated_credentials) =
match admin_provider_ops_sub2api_exchange_token(
state,
base_url,
credentials,
proxy_snapshot,
)
.await
{
Ok(value) => value,
Err(message) => {
return admin_provider_ops_action_error(
"auth_failed",
"query_balance",
message,
None,
)
}
};
if !updated_credentials.is_empty() {
if let Err(err) =
persist_admin_provider_ops_runtime_credentials(state, provider, &updated_credentials)
.await
{
warn!(
provider_id = %provider_id,
error = ?err,
"failed to persist sub2api runtime credentials"
);
}
}
let me_endpoint = action_config
.get("endpoint")
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("/api/v1/auth/me?timezone=Asia/Shanghai");
let me_url = admin_provider_ops_sub2api_request_url(base_url, me_endpoint);
let subscription_endpoint = admin_provider_ops_json_object_map(json!({
"endpoint": action_config
.get("subscription_endpoint")
.cloned()
.unwrap_or_else(|| json!("/api/v1/subscriptions/summary")),
}))
.get("endpoint")
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("/api/v1/subscriptions/summary")
.to_string();
let subscription_url =
admin_provider_ops_sub2api_request_url(base_url, subscription_endpoint.as_str());
let auth_value = match reqwest::header::HeaderValue::from_str(&format!("Bearer {access_token}"))
{
Ok(value) => value,
Err(_) => {
return admin_provider_ops_action_error(
"parse_error",
"query_balance",
"访问令牌格式无效",
None,
);
}
};
let auth_headers =
reqwest::header::HeaderMap::from_iter([(reqwest::header::AUTHORIZATION, auth_value)]);
let (me_result, subscription_result) = if let Some(proxy_snapshot) = proxy_snapshot {
let me_request_id = format!("provider-ops-action:sub2api:me:{provider_id}");
let subscription_request_id =
format!("provider-ops-action:sub2api:subscriptions:{provider_id}");
tokio::join!(
admin_provider_ops_execute_proxy_json_request(
state,
&me_request_id,
reqwest::Method::GET,
&me_url,
&auth_headers,
None,
proxy_snapshot,
),
admin_provider_ops_execute_proxy_json_request(
state,
&subscription_request_id,
reqwest::Method::GET,
&subscription_url,
&auth_headers,
None,
proxy_snapshot,
)
)
} else {
let http_client = state.http_client();
let (me_response, subscription_response) = tokio::join!(
http_client.get(me_url).bearer_auth(&access_token).send(),
http_client
.get(subscription_url)
.bearer_auth(&access_token)
.send()
);
let me_result = match me_response {
Ok(response) => {
let status = response.status();
let value = match response.bytes().await {
Ok(bytes) => {
serde_json::from_slice::<Value>(&bytes).unwrap_or_else(|_| json!({}))
}
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
format!("网络错误: {err}"),
Some(start.elapsed().as_millis() as u64),
)
}
};
Ok((status, value))
}
Err(err) if err.is_timeout() => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
"请求超时",
Some(start.elapsed().as_millis() as u64),
);
}
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
format!("网络错误: {err}"),
Some(start.elapsed().as_millis() as u64),
);
}
};
let subscription_result = match subscription_response {
Ok(response) => {
let status = response.status();
let value = match response.bytes().await {
Ok(bytes) => {
serde_json::from_slice::<Value>(&bytes).unwrap_or_else(|_| json!({}))
}
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
format!("网络错误: {err}"),
Some(start.elapsed().as_millis() as u64),
)
}
};
Ok((status, value))
}
Err(err) if err.is_timeout() => Err("请求超时".to_string()),
Err(err) => Err(format!("网络错误: {err}")),
};
(me_result, subscription_result)
};
let response_time_ms = Some(start.elapsed().as_millis() as u64);
let (me_status, me_json) = match me_result {
Ok(result) => result,
Err(err) => {
return admin_provider_ops_action_error(
"network_error",
"query_balance",
network_error_message(&err),
response_time_ms,
);
}
};
if matches!(
me_status,
http::StatusCode::UNAUTHORIZED | http::StatusCode::FORBIDDEN
) {
return admin_provider_ops_action_error(
"auth_failed",
"query_balance",
"认证失败,请检查凭据配置",
response_time_ms,
);
}
if me_status != http::StatusCode::OK {
return admin_provider_ops_action_error(
"unknown_error",
"query_balance",
format!(
"HTTP {}: {}",
me_status.as_u16(),
me_status.canonical_reason().unwrap_or("Unknown")
),
response_time_ms,
);
}
let subscription_json = subscription_result
.ok()
.and_then(|(status, payload)| (status == http::StatusCode::OK).then_some(payload));
let data =
match parse_sub2api_balance_payload(action_config, &me_json, subscription_json.as_ref()) {
Ok(data) => data,
Err(message) => {
return admin_provider_ops_action_error(
if message == "响应格式无效" {
"parse_error"
} else {
"unknown_error"
},
"query_balance",
message,
response_time_ms,
);
}
};
admin_provider_ops_action_response(
"success",
"query_balance",
data,
None,
response_time_ms,
86400,
)
}
fn network_error_message(error: &str) -> String {
let normalized = error.trim();
let lower = normalized.to_ascii_lowercase();
if lower.contains("timeout") || normalized.contains("超时") {
return "请求超时".to_string();
}
if normalized.starts_with("网络错误:") {
return normalized.to_string();
}
format!("网络错误: {normalized}")
}

View File

@@ -1,10 +1,10 @@
use super::super::super::verify::admin_provider_ops_value_as_f64;
use super::super::super::verify::admin_provider_ops_execute_proxy_json_request;
use super::super::responses::{
admin_provider_ops_action_error, admin_provider_ops_action_response,
};
use super::super::support::admin_provider_ops_balance_data;
use super::parsers::admin_provider_ops_yescode_balance_extra;
use crate::handlers::admin::request::AdminAppState;
use aether_admin::provider::ops::parse_yescode_combined_balance_payload;
use aether_contracts::ProxySnapshot;
use serde_json::json;
pub(super) async fn admin_provider_ops_yescode_balance_payload(
@@ -12,120 +12,161 @@ pub(super) async fn admin_provider_ops_yescode_balance_payload(
base_url: &str,
headers: &reqwest::header::HeaderMap,
action_config: &serde_json::Map<String, serde_json::Value>,
proxy_snapshot: Option<&ProxySnapshot>,
) -> serde_json::Value {
let start = std::time::Instant::now();
let balance_url = format!("{}/api/v1/user/balance", base_url.trim_end_matches('/'));
let profile_url = format!("{}/api/v1/auth/profile", base_url.trim_end_matches('/'));
let balance_future = state
.http_client()
.request(reqwest::Method::GET, balance_url)
.headers(headers.clone())
.send();
let profile_future = state
.http_client()
.request(reqwest::Method::GET, profile_url)
.headers(headers.clone())
.send();
let (balance_result, profile_result) = tokio::join!(balance_future, profile_future);
let (balance_result, profile_result) = if let Some(proxy_snapshot) = proxy_snapshot {
tokio::join!(
admin_provider_ops_execute_proxy_json_request(
state,
"provider-ops-action:yescode:balance",
reqwest::Method::GET,
&balance_url,
headers,
None,
proxy_snapshot,
),
admin_provider_ops_execute_proxy_json_request(
state,
"provider-ops-action:yescode:profile",
reqwest::Method::GET,
&profile_url,
headers,
None,
proxy_snapshot,
)
)
} else {
let balance_future = state
.http_client()
.request(reqwest::Method::GET, balance_url)
.headers(headers.clone())
.send();
let profile_future = state
.http_client()
.request(reqwest::Method::GET, profile_url)
.headers(headers.clone())
.send();
let (balance_result, profile_result) = tokio::join!(balance_future, profile_future);
let balance_result = match balance_result {
Ok(response) => {
let status = response.status();
let value = match response.bytes().await {
Ok(bytes) => serde_json::from_slice::<serde_json::Value>(&bytes)
.unwrap_or_else(|_| json!({})),
Err(_) => json!({}),
};
Ok((status, value))
}
Err(err) => Err(err.to_string()),
};
let profile_result = match profile_result {
Ok(response) => {
let status = response.status();
let value = match response.bytes().await {
Ok(bytes) => serde_json::from_slice::<serde_json::Value>(&bytes)
.unwrap_or_else(|_| json!({})),
Err(_) => json!({}),
};
Ok((status, value))
}
Err(err) => Err(err.to_string()),
};
(balance_result, profile_result)
};
let response_time_ms = Some(start.elapsed().as_millis() as u64);
let mut combined = serde_json::Map::new();
let mut has_any = false;
if let Ok(balance_response) = balance_result {
if balance_response.status() == http::StatusCode::OK {
if let Ok(bytes) = balance_response.bytes().await {
if let Ok(value) = serde_json::from_slice::<serde_json::Value>(&bytes) {
if let Some(object) = value.as_object() {
has_any = true;
combined.insert(
"_balance_data".to_string(),
serde_json::Value::Object(object.clone()),
);
combined.insert(
"pay_as_you_go_balance".to_string(),
object
.get("pay_as_you_go_balance")
.cloned()
.unwrap_or_else(|| json!(0)),
);
combined.insert(
"subscription_balance".to_string(),
object
.get("subscription_balance")
.cloned()
.unwrap_or_else(|| json!(0)),
);
if let Some(limit) = object.get("weekly_limit") {
combined.insert("weekly_limit".to_string(), limit.clone());
}
combined.insert(
"weekly_spent_balance".to_string(),
object
.get("weekly_spent_balance")
.cloned()
.unwrap_or_else(|| json!(0)),
);
}
if let Ok((status, value)) = balance_result {
if status == http::StatusCode::OK {
if let Some(object) = value.as_object() {
has_any = true;
combined.insert(
"_balance_data".to_string(),
serde_json::Value::Object(object.clone()),
);
combined.insert(
"pay_as_you_go_balance".to_string(),
object
.get("pay_as_you_go_balance")
.cloned()
.unwrap_or_else(|| json!(0)),
);
combined.insert(
"subscription_balance".to_string(),
object
.get("subscription_balance")
.cloned()
.unwrap_or_else(|| json!(0)),
);
if let Some(limit) = object.get("weekly_limit") {
combined.insert("weekly_limit".to_string(), limit.clone());
}
combined.insert(
"weekly_spent_balance".to_string(),
object
.get("weekly_spent_balance")
.cloned()
.unwrap_or_else(|| json!(0)),
);
}
}
}
if let Ok(profile_response) = profile_result {
if profile_response.status() == http::StatusCode::OK {
if let Ok(bytes) = profile_response.bytes().await {
if let Ok(value) = serde_json::from_slice::<serde_json::Value>(&bytes) {
if let Some(object) = value.as_object() {
has_any = true;
combined.insert(
"_profile_data".to_string(),
serde_json::Value::Object(object.clone()),
);
for key in [
"username",
"email",
"last_week_reset",
"last_daily_balance_add",
"subscription_plan",
] {
if let Some(value) = object.get(key) {
combined.insert(key.to_string(), value.clone());
}
}
combined
.entry("pay_as_you_go_balance".to_string())
.or_insert_with(|| {
object
.get("pay_as_you_go_balance")
.cloned()
.unwrap_or_else(|| json!(0))
});
combined
.entry("subscription_balance".to_string())
.or_insert_with(|| {
object
.get("subscription_balance")
.cloned()
.unwrap_or_else(|| json!(0))
});
combined
.entry("weekly_spent_balance".to_string())
.or_insert_with(|| {
object
.get("current_week_spend")
.cloned()
.unwrap_or_else(|| json!(0))
});
if !combined.contains_key("weekly_limit") {
if let Some(limit) = object
.get("subscription_plan")
.and_then(serde_json::Value::as_object)
.and_then(|plan| plan.get("weekly_limit"))
{
combined.insert("weekly_limit".to_string(), limit.clone());
}
}
if let Ok((status, value)) = profile_result {
if status == http::StatusCode::OK {
if let Some(object) = value.as_object() {
has_any = true;
combined.insert(
"_profile_data".to_string(),
serde_json::Value::Object(object.clone()),
);
for key in [
"username",
"email",
"last_week_reset",
"last_daily_balance_add",
"subscription_plan",
] {
if let Some(value) = object.get(key) {
combined.insert(key.to_string(), value.clone());
}
}
combined
.entry("pay_as_you_go_balance".to_string())
.or_insert_with(|| {
object
.get("pay_as_you_go_balance")
.cloned()
.unwrap_or_else(|| json!(0))
});
combined
.entry("subscription_balance".to_string())
.or_insert_with(|| {
object
.get("subscription_balance")
.cloned()
.unwrap_or_else(|| json!(0))
});
combined
.entry("weekly_spent_balance".to_string())
.or_insert_with(|| {
object
.get("current_week_spend")
.cloned()
.unwrap_or_else(|| json!(0))
});
if !combined.contains_key("weekly_limit") {
if let Some(limit) = object
.get("subscription_plan")
.and_then(serde_json::Value::as_object)
.and_then(|plan| plan.get("weekly_limit"))
{
combined.insert("weekly_limit".to_string(), limit.clone());
}
}
}
@@ -141,23 +182,10 @@ pub(super) async fn admin_provider_ops_yescode_balance_payload(
);
}
let mut extra = admin_provider_ops_yescode_balance_extra(&combined);
let total_available = admin_provider_ops_value_as_f64(extra.get("_total_available"));
extra.remove("_subscription_available");
extra.remove("_total_available");
admin_provider_ops_action_response(
"success",
"query_balance",
admin_provider_ops_balance_data(
None,
None,
total_available,
action_config
.get("currency")
.and_then(serde_json::Value::as_str)
.unwrap_or("USD"),
extra,
),
parse_yescode_combined_balance_payload(action_config, &combined),
None,
response_time_ms,
86400,

View File

@@ -1,24 +1,5 @@
use super::super::config::admin_provider_ops_uses_python_verify_fallback;
use super::super::verify::admin_provider_ops_value_as_f64;
use serde_json::json;
pub(super) fn admin_provider_ops_balance_data(
total_granted: Option<f64>,
total_used: Option<f64>,
total_available: Option<f64>,
currency: &str,
extra: serde_json::Map<String, serde_json::Value>,
) -> serde_json::Value {
json!({
"total_granted": total_granted,
"total_used": total_used,
"total_available": total_available,
"expires_at": serde_json::Value::Null,
"currency": currency,
"extra": extra,
})
}
pub(super) fn admin_provider_ops_checkin_data(
reward: Option<f64>,
streak_days: Option<i64>,
@@ -35,79 +16,12 @@ pub(super) fn admin_provider_ops_checkin_data(
})
}
pub(super) fn admin_provider_ops_action_config_object<'a>(
provider_ops_config: &'a serde_json::Map<String, serde_json::Value>,
action_type: &str,
) -> Option<&'a serde_json::Map<String, serde_json::Value>> {
provider_ops_config
.get("actions")
.and_then(serde_json::Value::as_object)
.and_then(|actions| actions.get(action_type))
.and_then(serde_json::Value::as_object)
.and_then(|action| action.get("config"))
.and_then(serde_json::Value::as_object)
}
pub(super) fn admin_provider_ops_default_action_config(
architecture_id: &str,
action_type: &str,
) -> Option<serde_json::Map<String, serde_json::Value>> {
let value = match (architecture_id, action_type) {
("generic_api", "query_balance") => {
json!({ "endpoint": "/api/user/balance", "method": "GET" })
}
("generic_api", "checkin") => {
json!({ "endpoint": "/api/user/checkin", "method": "POST" })
}
("new_api", "query_balance") => json!({
"endpoint": "/api/user/self",
"method": "GET",
"quota_divisor": 500000,
"checkin_endpoint": "/api/user/checkin",
"currency": "USD",
}),
("new_api", "checkin") => json!({ "endpoint": "/api/user/checkin", "method": "POST" }),
("cubence", "query_balance") => {
json!({ "endpoint": "/api/v1/dashboard/overview", "method": "GET", "currency": "USD" })
}
("yescode", "query_balance") => {
json!({ "endpoint": "/api/v1/user/balance", "method": "GET", "currency": "USD" })
}
("nekocode", "query_balance") => {
json!({ "endpoint": "/api/usage/summary", "method": "GET", "currency": "USD" })
}
_ => return None,
};
value.as_object().cloned()
}
pub(super) fn admin_provider_ops_json_object_map(
value: serde_json::Value,
) -> serde_json::Map<String, serde_json::Value> {
value.as_object().cloned().unwrap_or_default()
}
pub(super) fn admin_provider_ops_resolved_action_config(
architecture_id: &str,
provider_ops_config: &serde_json::Map<String, serde_json::Value>,
action_type: &str,
request_config: Option<&serde_json::Map<String, serde_json::Value>>,
) -> Option<serde_json::Map<String, serde_json::Value>> {
let mut resolved =
admin_provider_ops_default_action_config(architecture_id, action_type).unwrap_or_default();
if let Some(saved) = admin_provider_ops_action_config_object(provider_ops_config, action_type) {
for (key, value) in saved {
resolved.insert(key.clone(), value.clone());
}
}
if let Some(overrides) = request_config {
for (key, value) in overrides {
resolved.insert(key.clone(), value.clone());
}
}
(!resolved.is_empty()).then_some(resolved)
}
pub(super) fn admin_provider_ops_request_url(
base_url: &str,
action_config: &serde_json::Map<String, serde_json::Value>,
@@ -150,26 +64,3 @@ pub(super) fn admin_provider_ops_parse_rfc3339_unix_secs(
.ok()
.map(|value| value.timestamp())
}
pub(super) fn admin_provider_ops_is_cookie_auth_architecture(architecture_id: &str) -> bool {
matches!(architecture_id, "cubence" | "yescode" | "nekocode")
}
pub(super) fn admin_provider_ops_should_use_rust_only_action_stub(
architecture_id: &str,
config: &serde_json::Map<String, serde_json::Value>,
) -> bool {
!matches!(
architecture_id,
"generic_api" | "new_api" | "cubence" | "yescode" | "nekocode"
) || admin_provider_ops_uses_python_verify_fallback(architecture_id, config)
}
#[allow(dead_code)]
pub(super) fn admin_provider_ops_quota_divisor(
action_config: &serde_json::Map<String, serde_json::Value>,
) -> f64 {
admin_provider_ops_value_as_f64(action_config.get("quota_divisor"))
.filter(|value| *value > 0.0)
.unwrap_or(500000.0)
}

View File

@@ -0,0 +1,303 @@
use super::actions::admin_provider_ops_local_action_response;
use crate::handlers::admin::request::AdminAppState;
use serde_json::{json, Value};
use std::collections::HashSet;
use std::time::Duration;
use tokio::sync::{Mutex, Semaphore};
use tracing::{debug, warn};
const ADMIN_PROVIDER_OPS_BALANCE_CACHE_PREFIX: &str = "provider_ops:balance:";
const ADMIN_PROVIDER_OPS_BALANCE_CACHE_TTL_SECS: u64 = 86_400;
const ADMIN_PROVIDER_OPS_BALANCE_AUTH_FAILED_CACHE_TTL_SECS: u64 = 60;
const ADMIN_PROVIDER_OPS_BALANCE_REFRESH_CONCURRENCY: usize = 3;
static ADMIN_PROVIDER_OPS_BALANCE_REFRESH_SEMAPHORE: std::sync::LazyLock<Semaphore> =
std::sync::LazyLock::new(|| Semaphore::new(ADMIN_PROVIDER_OPS_BALANCE_REFRESH_CONCURRENCY));
static ADMIN_PROVIDER_OPS_REFRESHING_PROVIDERS: std::sync::LazyLock<Mutex<HashSet<String>>> =
std::sync::LazyLock::new(|| Mutex::new(HashSet::new()));
#[derive(Debug)]
pub(super) enum AdminProviderOpsBalanceCacheLookup {
Hit(Value),
Miss,
Unavailable,
}
pub(super) fn admin_provider_ops_batch_balance_concurrency() -> usize {
std::env::var("BATCH_BALANCE_CONCURRENCY")
.ok()
.and_then(|value| value.trim().parse::<usize>().ok())
.map(|value| value.max(1))
.unwrap_or(3)
}
pub(super) fn admin_provider_ops_pending_balance_response(message: &str) -> Value {
json!({
"status": "pending",
"action_type": "query_balance",
"data": Value::Null,
"message": message,
"executed_at": chrono::Utc::now()
.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"response_time_ms": Value::Null,
"cache_ttl_seconds": 0,
})
}
pub(super) async fn read_admin_provider_ops_balance_cache(
state: &AdminAppState<'_>,
provider_id: &str,
) -> AdminProviderOpsBalanceCacheLookup {
let Some(runner) = state.redis_kv_runner() else {
return AdminProviderOpsBalanceCacheLookup::Unavailable;
};
let mut connection = match runner.client().get_multiplexed_async_connection().await {
Ok(connection) => connection,
Err(err) => {
warn!(error = %err, provider_id, "failed to connect to redis for provider ops balance cache");
return AdminProviderOpsBalanceCacheLookup::Unavailable;
}
};
let namespaced_key = runner.keyspace().key(&format!(
"{ADMIN_PROVIDER_OPS_BALANCE_CACHE_PREFIX}{provider_id}"
));
let raw = match redis::cmd("GET")
.arg(&namespaced_key)
.query_async::<Option<String>>(&mut connection)
.await
{
Ok(raw) => raw,
Err(err) => {
warn!(error = %err, provider_id, "failed to read provider ops balance cache");
return AdminProviderOpsBalanceCacheLookup::Unavailable;
}
};
let Some(raw) = raw else {
return AdminProviderOpsBalanceCacheLookup::Miss;
};
match serde_json::from_str::<Value>(&raw) {
Ok(payload) => AdminProviderOpsBalanceCacheLookup::Hit(payload),
Err(err) => {
warn!(error = %err, provider_id, "failed to parse provider ops balance cache payload");
AdminProviderOpsBalanceCacheLookup::Miss
}
}
}
pub(super) async fn store_admin_provider_ops_balance_cache(
state: &AdminAppState<'_>,
provider_id: &str,
payload: &Value,
) {
let Some(ttl_seconds) = balance_cache_ttl_seconds(payload) else {
return;
};
let Some(runner) = state.redis_kv_runner() else {
return;
};
let serialized = match serde_json::to_string(payload) {
Ok(serialized) => serialized,
Err(err) => {
warn!(
error = %err,
provider_id,
"failed to serialize provider ops balance payload"
);
return;
}
};
if let Err(err) = runner
.setex(
&format!("{ADMIN_PROVIDER_OPS_BALANCE_CACHE_PREFIX}{provider_id}"),
&serialized,
Some(ttl_seconds),
)
.await
{
warn!(error = %err, provider_id, "failed to store provider ops balance cache");
}
}
pub(super) async fn clear_admin_provider_ops_balance_cache(
state: &AdminAppState<'_>,
provider_id: &str,
) {
let Some(runner) = state.redis_kv_runner() else {
return;
};
if let Err(err) = runner
.del(&format!(
"{ADMIN_PROVIDER_OPS_BALANCE_CACHE_PREFIX}{provider_id}"
))
.await
{
warn!(error = %err, provider_id, "failed to clear provider ops balance cache");
}
}
pub(super) async fn spawn_admin_provider_ops_balance_refresh(
state: &AdminAppState<'_>,
provider_id: &str,
) {
let mut guard = ADMIN_PROVIDER_OPS_REFRESHING_PROVIDERS.lock().await;
if !guard.insert(provider_id.to_string()) {
debug!(provider_id, "provider ops balance refresh already running");
return;
}
drop(guard);
let app = state.cloned_app();
let provider_id = provider_id.to_string();
tokio::spawn(async move {
let permit = match tokio::time::timeout(
Duration::from_secs(5),
ADMIN_PROVIDER_OPS_BALANCE_REFRESH_SEMAPHORE.acquire(),
)
.await
{
Ok(Ok(permit)) => permit,
Ok(Err(err)) => {
warn!(
provider_id = %provider_id,
error = %err,
"provider ops balance refresh semaphore closed"
);
finish_refresh_provider(&provider_id).await;
return;
}
Err(_) => {
debug!(provider_id = %provider_id, "provider ops balance refresh skipped by concurrency limit");
finish_refresh_provider(&provider_id).await;
return;
}
};
let admin_state = AdminAppState::new(&app);
let provider_ids = [provider_id.clone()];
let providers = match admin_state
.read_provider_catalog_providers_by_ids(&provider_ids)
.await
{
Ok(providers) => providers,
Err(err) => {
warn!(
provider_id = %provider_id,
error = ?err,
"failed to load provider for balance refresh"
);
drop(permit);
finish_refresh_provider(&provider_id).await;
return;
}
};
let provider = providers.first();
let endpoints = if provider.is_some() {
match admin_state
.list_provider_catalog_endpoints_by_provider_ids(&provider_ids)
.await
{
Ok(endpoints) => endpoints,
Err(err) => {
warn!(
provider_id = %provider_id,
error = ?err,
"failed to load endpoints for balance refresh"
);
drop(permit);
finish_refresh_provider(&provider_id).await;
return;
}
}
} else {
Vec::new()
};
let payload = admin_provider_ops_local_action_response(
&admin_state,
&provider_id,
provider,
&endpoints,
"query_balance",
None,
)
.await;
store_admin_provider_ops_balance_cache(&admin_state, &provider_id, &payload).await;
drop(permit);
finish_refresh_provider(&provider_id).await;
});
}
fn balance_cache_ttl_seconds(payload: &Value) -> Option<u64> {
match payload
.get("status")
.and_then(Value::as_str)
.unwrap_or_default()
{
"success" | "auth_expired" => Some(ADMIN_PROVIDER_OPS_BALANCE_CACHE_TTL_SECS),
"auth_failed" => Some(ADMIN_PROVIDER_OPS_BALANCE_AUTH_FAILED_CACHE_TTL_SECS),
_ => None,
}
}
async fn finish_refresh_provider(provider_id: &str) {
ADMIN_PROVIDER_OPS_REFRESHING_PROVIDERS
.lock()
.await
.remove(provider_id);
}
fn admin_provider_ops_action_response(
total_available: f64,
extra: serde_json::Map<String, Value>,
) -> Value {
json!({
"status": "success",
"action_type": "query_balance",
"data": {
"total_granted": Value::Null,
"total_used": Value::Null,
"total_available": total_available,
"expires_at": Value::Null,
"currency": "USD",
"extra": extra,
},
"message": Value::Null,
"executed_at": chrono::Utc::now()
.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
"response_time_ms": Value::Null,
"cache_ttl_seconds": ADMIN_PROVIDER_OPS_BALANCE_CACHE_TTL_SECS,
})
}
#[cfg(test)]
mod tests {
use super::{admin_provider_ops_pending_balance_response, balance_cache_ttl_seconds};
use serde_json::json;
#[test]
fn pending_balance_response_uses_pending_status() {
let payload = admin_provider_ops_pending_balance_response("余额数据加载中,请稍后刷新");
assert_eq!(payload["status"], json!("pending"));
assert_eq!(payload["action_type"], json!("query_balance"));
}
#[test]
fn balance_cache_ttl_matches_status_contract() {
assert_eq!(
balance_cache_ttl_seconds(&json!({ "status": "success" })),
Some(86400)
);
assert_eq!(
balance_cache_ttl_seconds(&json!({ "status": "auth_expired" })),
Some(86400)
);
assert_eq!(
balance_cache_ttl_seconds(&json!({ "status": "auth_failed" })),
Some(60)
);
assert_eq!(
balance_cache_ttl_seconds(&json!({ "status": "network_error" })),
None
);
}
}

View File

@@ -1,10 +1,12 @@
use super::support::{AdminProviderOpsSaveConfigRequest, ADMIN_PROVIDER_OPS_SENSITIVE_FIELDS};
use crate::handlers::admin::request::AdminAppState;
use crate::GatewayError;
use aether_admin::provider::ops as admin_provider_ops_pure;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogProvider,
};
use serde_json::json;
use std::time::{SystemTime, UNIX_EPOCH};
pub(super) fn admin_provider_ops_config_object(
provider: &StoredProviderCatalogProvider,
@@ -61,6 +63,9 @@ fn admin_provider_ops_masked_credentials(
let mut masked = serde_json::Map::new();
for (key, value) in credentials {
if key.starts_with('_') {
continue;
}
if ADMIN_PROVIDER_OPS_SENSITIVE_FIELDS.contains(&key.as_str()) {
if let Some(ciphertext) = value.as_str().filter(|value| !value.is_empty()) {
masked.insert(
@@ -79,13 +84,6 @@ fn admin_provider_ops_is_supported_auth_type(auth_type: &str) -> bool {
admin_provider_ops_pure::admin_provider_ops_is_supported_auth_type(auth_type)
}
pub(super) fn admin_provider_ops_uses_python_verify_fallback(
architecture_id: &str,
config: &serde_json::Map<String, serde_json::Value>,
) -> bool {
admin_provider_ops_pure::admin_provider_ops_uses_python_verify_fallback(architecture_id, config)
}
pub(super) fn admin_provider_ops_decrypted_credentials(
state: &AdminAppState<'_>,
raw_credentials: Option<&serde_json::Value>,
@@ -116,17 +114,26 @@ fn admin_provider_ops_sensitive_placeholder_or_empty(value: Option<&serde_json::
pub(super) fn admin_provider_ops_merge_credentials(
state: &AdminAppState<'_>,
architecture_id: &str,
provider: &StoredProviderCatalogProvider,
mut request_credentials: serde_json::Map<String, serde_json::Value>,
) -> serde_json::Map<String, serde_json::Value> {
let saved_credentials = admin_provider_ops_decrypted_credentials(
let mut saved_credentials = admin_provider_ops_decrypted_credentials(
state,
admin_provider_ops_config_object(provider)
.and_then(admin_provider_ops_connector_object)
.and_then(|connector| connector.get("credentials")),
);
let preserve_internal_runtime_fields =
admin_provider_ops_pure::normalize_architecture_id(architecture_id) == "sub2api";
if !preserve_internal_runtime_fields {
saved_credentials.retain(|key, _| !key.starts_with('_'));
}
for field in ADMIN_PROVIDER_OPS_SENSITIVE_FIELDS {
if field.starts_with('_') {
continue;
}
if admin_provider_ops_sensitive_placeholder_or_empty(request_credentials.get(*field))
&& saved_credentials.contains_key(*field)
{
@@ -136,9 +143,11 @@ pub(super) fn admin_provider_ops_merge_credentials(
}
}
for (key, value) in saved_credentials {
if key.starts_with('_') && !request_credentials.contains_key(&key) {
request_credentials.insert(key, value);
if preserve_internal_runtime_fields {
for (key, value) in saved_credentials {
if key.starts_with('_') && !request_credentials.contains_key(&key) {
request_credentials.insert(key, value);
}
}
}
@@ -169,6 +178,73 @@ fn admin_provider_ops_encrypt_credentials(
Ok(encrypted)
}
pub(super) async fn persist_admin_provider_ops_runtime_credentials(
state: &AdminAppState<'_>,
provider: &StoredProviderCatalogProvider,
updated_credentials: &serde_json::Map<String, serde_json::Value>,
) -> Result<Option<StoredProviderCatalogProvider>, GatewayError> {
if updated_credentials.is_empty() || !state.has_provider_catalog_data_writer() {
return Ok(None);
}
let mut updated_provider = provider.clone();
let mut provider_config = updated_provider
.config
.as_ref()
.and_then(serde_json::Value::as_object)
.cloned()
.unwrap_or_default();
let Some(provider_ops_config) = provider_config
.get("provider_ops")
.and_then(serde_json::Value::as_object)
.cloned()
else {
return Ok(None);
};
let Some(connector_config) = provider_ops_config
.get("connector")
.and_then(serde_json::Value::as_object)
.cloned()
else {
return Ok(None);
};
let mut decrypted_credentials =
admin_provider_ops_decrypted_credentials(state, connector_config.get("credentials"));
for (key, value) in updated_credentials {
decrypted_credentials.insert(key.clone(), value.clone());
}
let encrypted_credentials =
admin_provider_ops_encrypt_credentials(state, decrypted_credentials)
.map_err(GatewayError::Internal)?;
let mut updated_connector = connector_config.clone();
updated_connector.insert(
"credentials".to_string(),
serde_json::Value::Object(encrypted_credentials),
);
let mut updated_provider_ops = provider_ops_config.clone();
updated_provider_ops.insert(
"connector".to_string(),
serde_json::Value::Object(updated_connector),
);
provider_config.insert(
"provider_ops".to_string(),
serde_json::Value::Object(updated_provider_ops),
);
updated_provider.config = Some(serde_json::Value::Object(provider_config));
updated_provider.updated_at_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()
.map(|duration| duration.as_secs());
state
.update_provider_catalog_provider(&updated_provider)
.await
}
pub(super) fn build_admin_provider_ops_saved_config_value(
state: &AdminAppState<'_>,
provider: &StoredProviderCatalogProvider,
@@ -179,8 +255,12 @@ pub(super) fn build_admin_provider_ops_saved_config_value(
return Err("connector.auth_type 必须是合法的认证类型".to_string());
}
let merged_credentials =
admin_provider_ops_merge_credentials(state, provider, payload.connector.credentials);
let merged_credentials = admin_provider_ops_merge_credentials(
state,
payload.architecture_id.as_str(),
provider,
payload.connector.credentials,
);
let encrypted_credentials = admin_provider_ops_encrypt_credentials(state, merged_credentials)?;
let actions = payload

View File

@@ -1,4 +1,5 @@
pub(crate) mod actions;
mod balance_cache;
mod config;
mod routes;
mod support;

View File

@@ -1,6 +1,12 @@
use super::super::actions::{
admin_provider_ops_is_valid_action_type, admin_provider_ops_local_action_response,
};
use super::super::balance_cache::{
admin_provider_ops_pending_balance_response, read_admin_provider_ops_balance_cache,
spawn_admin_provider_ops_balance_refresh, store_admin_provider_ops_balance_cache,
AdminProviderOpsBalanceCacheLookup,
};
use super::super::config::admin_provider_ops_config_object;
use super::super::support::AdminProviderOpsExecuteActionRequest;
use crate::handlers::admin::request::AdminAppState;
use crate::GatewayError;
@@ -17,6 +23,7 @@ pub(super) async fn handle_admin_provider_ops_action(
provider_id: &str,
route_kind: &str,
action_route: Option<&(String, String)>,
query_string: Option<&str>,
request_body: Option<&Bytes>,
) -> Result<Option<Response<Body>>, GatewayError> {
let action_type = if route_kind == "provider_checkin" {
@@ -83,19 +90,81 @@ pub(super) async fn handle_admin_provider_ops_action(
} else {
Vec::new()
};
let payload = admin_provider_ops_local_action_response(
state,
provider_id,
provider,
&endpoints,
&action_type,
request_config.as_ref(),
)
.await;
let payload = if action_type == "query_balance"
&& route_kind == "get_provider_balance"
&& provider.is_some_and(|provider| admin_provider_ops_config_object(provider).is_some())
{
match read_admin_provider_ops_balance_cache(state, provider_id).await {
AdminProviderOpsBalanceCacheLookup::Hit(cached) => {
if query_param_bool(query_string, "refresh", true) {
spawn_admin_provider_ops_balance_refresh(state, provider_id).await;
}
cached
}
AdminProviderOpsBalanceCacheLookup::Miss => {
if query_param_bool(query_string, "refresh", true) {
spawn_admin_provider_ops_balance_refresh(state, provider_id).await;
admin_provider_ops_pending_balance_response("余额数据加载中,请稍后刷新")
} else {
let payload = admin_provider_ops_local_action_response(
state,
provider_id,
provider,
&endpoints,
&action_type,
request_config.as_ref(),
)
.await;
store_admin_provider_ops_balance_cache(state, provider_id, &payload).await;
payload
}
}
AdminProviderOpsBalanceCacheLookup::Unavailable => {
let payload = admin_provider_ops_local_action_response(
state,
provider_id,
provider,
&endpoints,
&action_type,
request_config.as_ref(),
)
.await;
store_admin_provider_ops_balance_cache(state, provider_id, &payload).await;
payload
}
}
} else {
let payload = admin_provider_ops_local_action_response(
state,
provider_id,
provider,
&endpoints,
&action_type,
request_config.as_ref(),
)
.await;
if action_type == "query_balance" && route_kind == "refresh_provider_balance" {
store_admin_provider_ops_balance_cache(state, provider_id, &payload).await;
}
payload
};
Ok(Some(Json(payload).into_response()))
}
fn query_param_bool(query: Option<&str>, key: &str, default: bool) -> bool {
let Some(query) = query else {
return default;
};
for (entry_key, value) in url::form_urlencoded::parse(query.as_bytes()) {
if entry_key == key {
let normalized = value.trim().to_ascii_lowercase();
return matches!(normalized.as_str(), "1" | "true" | "yes" | "on");
}
}
default
}
fn bad_request_detail_response(detail: &str) -> Response<Body> {
(
http::StatusCode::BAD_REQUEST,

View File

@@ -1,4 +1,10 @@
use super::super::actions::admin_provider_ops_local_action_response;
use super::super::balance_cache::{
admin_provider_ops_batch_balance_concurrency, admin_provider_ops_pending_balance_response,
read_admin_provider_ops_balance_cache, spawn_admin_provider_ops_balance_refresh,
store_admin_provider_ops_balance_cache, AdminProviderOpsBalanceCacheLookup,
};
use super::super::config::admin_provider_ops_config_object;
use crate::handlers::admin::request::AdminAppState;
use crate::GatewayError;
use axum::{
@@ -7,7 +13,9 @@ use axum::{
response::{IntoResponse, Response},
Json,
};
use futures_util::stream::{self, StreamExt};
use serde_json::json;
use std::collections::HashMap;
pub(super) async fn handle_admin_provider_ops_batch_balance(
state: &AdminAppState<'_>,
@@ -49,28 +57,72 @@ pub(super) async fn handle_admin_provider_ops_batch_balance(
let endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(&provider_ids)
.await?;
let mut payload = serde_json::Map::new();
for provider_id in &provider_ids {
let provider = providers
.iter()
.find(|provider| provider.id == *provider_id);
let provider_endpoints = endpoints
.iter()
.filter(|endpoint| endpoint.provider_id == *provider_id)
.cloned()
.collect::<Vec<_>>();
let result = admin_provider_ops_local_action_response(
state,
provider_id,
provider,
&provider_endpoints,
"query_balance",
None,
)
.await;
payload.insert(provider_id.clone(), result);
let providers_by_id = providers
.into_iter()
.map(|provider| (provider.id.clone(), provider))
.collect::<HashMap<_, _>>();
let mut endpoints_by_provider = HashMap::<String, Vec<_>>::new();
for endpoint in endpoints {
endpoints_by_provider
.entry(endpoint.provider_id.clone())
.or_default()
.push(endpoint);
}
let results = stream::iter(provider_ids.into_iter().map(|provider_id| {
let provider = providers_by_id.get(&provider_id).cloned();
let provider_endpoints = endpoints_by_provider
.get(&provider_id)
.cloned()
.unwrap_or_default();
async move {
let result = if provider
.as_ref()
.is_some_and(|provider| admin_provider_ops_config_object(provider).is_some())
{
match read_admin_provider_ops_balance_cache(state, &provider_id).await {
AdminProviderOpsBalanceCacheLookup::Hit(cached) => {
spawn_admin_provider_ops_balance_refresh(state, &provider_id).await;
cached
}
AdminProviderOpsBalanceCacheLookup::Miss => {
spawn_admin_provider_ops_balance_refresh(state, &provider_id).await;
admin_provider_ops_pending_balance_response("余额数据加载中,请稍后刷新")
}
AdminProviderOpsBalanceCacheLookup::Unavailable => {
let payload = admin_provider_ops_local_action_response(
state,
&provider_id,
provider.as_ref(),
&provider_endpoints,
"query_balance",
None,
)
.await;
store_admin_provider_ops_balance_cache(state, &provider_id, &payload).await;
payload
}
}
} else {
admin_provider_ops_local_action_response(
state,
&provider_id,
provider.as_ref(),
&provider_endpoints,
"query_balance",
None,
)
.await
};
(provider_id, result)
}
}))
.buffer_unordered(admin_provider_ops_batch_balance_concurrency())
.collect::<Vec<_>>()
.await;
let payload = results.into_iter().collect::<serde_json::Map<_, _>>();
Ok(Json(serde_json::Value::Object(payload)).into_response())
}

View File

@@ -1,3 +1,4 @@
use super::super::balance_cache::clear_admin_provider_ops_balance_cache;
use super::super::config::build_admin_provider_ops_saved_config_value;
use super::super::support::AdminProviderOpsSaveConfigRequest;
use crate::handlers::admin::request::AdminAppState;
@@ -56,6 +57,7 @@ pub(super) async fn handle_admin_provider_ops_save_config(
else {
return Ok(None);
};
clear_admin_provider_ops_balance_cache(state, provider_id).await;
Ok(Some(
Json(json!({
@@ -99,6 +101,7 @@ pub(super) async fn handle_admin_provider_ops_delete_config(
else {
return Ok(None);
};
clear_admin_provider_ops_balance_cache(state, provider_id).await;
}
Ok(Some(

View File

@@ -79,6 +79,7 @@ pub(crate) async fn maybe_build_local_admin_provider_ops_providers_response(
&provider_id,
route_kind,
action_route.as_ref(),
request_context.query_string(),
request_body,
)
.await?

View File

@@ -3,13 +3,11 @@ use super::super::config::{
resolve_admin_provider_ops_base_url,
};
use super::super::support::AdminProviderOpsSaveConfigRequest;
use super::super::verify::{
admin_provider_ops_local_verify_response, admin_provider_ops_normalized_verify_architecture_id,
admin_provider_ops_verify_failure,
};
use super::super::verify::admin_provider_ops_local_verify_response;
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::admin::shared::attach_admin_audit_response;
use crate::GatewayError;
use aether_admin::provider::ops::{admin_provider_ops_verify_failure, normalize_architecture_id};
use axum::{
body::{Body, Bytes},
http,
@@ -60,13 +58,13 @@ pub(super) async fn handle_admin_provider_ops_verify(
return Ok(Json(admin_provider_ops_verify_failure("请提供 API 地址")).into_response());
};
let architecture_id =
admin_provider_ops_normalized_verify_architecture_id(&payload.architecture_id);
let architecture_id = normalize_architecture_id(&payload.architecture_id);
let credentials = existing_provider.as_ref().map_or_else(
|| payload.connector.credentials.clone(),
|provider| {
admin_provider_ops_merge_credentials(
state,
architecture_id,
provider,
payload.connector.credentials.clone(),
)
@@ -74,6 +72,7 @@ pub(super) async fn handle_admin_provider_ops_verify(
);
let payload = admin_provider_ops_local_verify_response(
state,
existing_provider.as_ref(),
&base_url,
architecture_id,
&payload.connector.config,

View File

@@ -1,10 +1,13 @@
use serde::Deserialize;
use std::collections::BTreeMap;
pub(super) use aether_admin::provider::ops::ProviderOpsCheckinOutcome as AdminProviderOpsCheckinOutcome;
pub(super) const ADMIN_PROVIDER_OPS_SENSITIVE_FIELDS: &[&str] = &[
"api_key",
"password",
"refresh_token",
"_cached_access_token",
"session_token",
"session_cookie",
"token_cookie",
@@ -61,13 +64,6 @@ pub(super) struct AdminProviderOpsExecuteActionRequest {
pub(crate) config: Option<serde_json::Map<String, serde_json::Value>>,
}
#[derive(Debug, Clone)]
pub(super) struct AdminProviderOpsCheckinOutcome {
pub(crate) success: Option<bool>,
pub(crate) message: String,
pub(crate) cookie_expired: bool,
}
fn default_admin_provider_ops_architecture_id() -> String {
"generic_api".to_string()
}

View File

@@ -1,15 +0,0 @@
use aether_admin::provider::verify as admin_provider_verify_pure;
use reqwest::header::HeaderMap;
use serde_json::Value;
pub(super) fn admin_provider_ops_verify_headers(
architecture_id: &str,
config: &serde_json::Map<String, Value>,
credentials: &serde_json::Map<String, Value>,
) -> Result<HeaderMap, String> {
admin_provider_verify_pure::admin_provider_ops_verify_headers(
architecture_id,
config,
credentials,
)
}

View File

@@ -1,192 +0,0 @@
use crate::handlers::admin::request::AdminAppState;
use aether_admin::provider::verify as admin_provider_verify_pure;
use regex::Regex;
use serde_json::{json, Map, Value};
pub(super) fn admin_provider_ops_normalized_verify_architecture_id(architecture_id: &str) -> &str {
admin_provider_verify_pure::admin_provider_ops_normalized_verify_architecture_id(
architecture_id,
)
}
pub(super) fn admin_provider_ops_extract_cookie_value(cookie_input: &str, key: &str) -> String {
admin_provider_verify_pure::admin_provider_ops_extract_cookie_value(cookie_input, key)
}
pub(super) fn admin_provider_ops_yescode_cookie_header(cookie_input: &str) -> String {
admin_provider_verify_pure::admin_provider_ops_yescode_cookie_header(cookie_input)
}
pub(super) fn admin_provider_ops_anyrouter_compute_acw_sc_v2(arg1: &str) -> Option<String> {
admin_provider_verify_pure::admin_provider_ops_anyrouter_compute_acw_sc_v2(arg1)
}
pub(super) fn admin_provider_ops_anyrouter_parse_session_user_id(
cookie_input: &str,
) -> Option<String> {
admin_provider_verify_pure::admin_provider_ops_anyrouter_parse_session_user_id(cookie_input)
}
pub(super) async fn admin_provider_ops_anyrouter_acw_cookie(
state: &AdminAppState<'_>,
base_url: &str,
) -> Option<String> {
let response = state
.http_client()
.get(base_url.trim_end_matches('/'))
.header(
reqwest::header::USER_AGENT,
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
)
.send()
.await
.ok()?;
let body = response.text().await.ok()?;
let compiled = Regex::new(r"var\s+arg1\s*=\s*'([0-9a-fA-F]{40})'").ok()?;
let captures = compiled.captures(&body)?;
let arg1 = captures.get(1)?.as_str();
admin_provider_ops_anyrouter_compute_acw_sc_v2(arg1).map(|value| format!("acw_sc__v2={value}"))
}
pub(super) fn admin_provider_ops_verify_failure(message: impl Into<String>) -> Value {
admin_provider_verify_pure::admin_provider_ops_verify_failure(message)
}
pub(super) fn admin_provider_ops_verify_success(
data: Value,
updated_credentials: Option<Map<String, Value>>,
) -> Value {
admin_provider_verify_pure::admin_provider_ops_verify_success(data, updated_credentials)
}
pub(super) fn admin_provider_ops_verify_user_payload(
username: Option<String>,
display_name: Option<String>,
email: Option<String>,
quota: Option<f64>,
extra: Option<Map<String, Value>>,
) -> Value {
admin_provider_verify_pure::admin_provider_ops_verify_user_payload(
username,
display_name,
email,
quota,
extra,
)
}
pub(super) fn admin_provider_ops_value_as_f64(value: Option<&Value>) -> Option<f64> {
admin_provider_verify_pure::admin_provider_ops_value_as_f64(value)
}
pub(super) fn admin_provider_ops_json_object(
value: &Value,
) -> Option<&serde_json::Map<String, Value>> {
admin_provider_verify_pure::admin_provider_ops_json_object(value)
}
pub(super) fn admin_provider_ops_frontend_updated_credentials(
credentials: Map<String, Value>,
) -> Option<Map<String, Value>> {
admin_provider_verify_pure::admin_provider_ops_frontend_updated_credentials(credentials)
}
pub(super) async fn admin_provider_ops_sub2api_exchange_token(
state: &AdminAppState<'_>,
base_url: &str,
credentials: &Map<String, Value>,
) -> Result<(String, Option<Map<String, Value>>), String> {
let email = credentials
.get("email")
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default();
let password = credentials
.get("password")
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default();
let refresh_token = credentials
.get("refresh_token")
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default();
let (path, body, default_error, previous_refresh_token) =
if !email.is_empty() && !password.is_empty() {
(
"/api/v1/auth/login",
json!({ "email": email, "password": password }),
"登录失败",
None,
)
} else if !refresh_token.is_empty() {
(
"/api/v1/auth/refresh",
json!({ "refresh_token": refresh_token }),
"Refresh Token 无效或已过期",
Some(refresh_token),
)
} else {
return Err("请填写账号密码或 Refresh Token".to_string());
};
let response = match state
.http_client()
.post(format!("{}{path}", base_url.trim_end_matches('/')))
.json(&body)
.send()
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => return Err("连接超时".to_string()),
Err(err) if err.is_connect() => return Err(format!("连接失败: {err}")),
Err(err) => return Err(format!("验证失败: {err}")),
};
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => serde_json::from_slice::<Value>(&bytes).unwrap_or_else(|_| json!({})),
Err(_) => json!({}),
};
let payload = response_json.as_object().cloned().unwrap_or_default();
if status != http::StatusCode::OK
|| payload.get("code").and_then(Value::as_i64).unwrap_or(-1) != 0
{
let message = payload
.get("message")
.and_then(Value::as_str)
.unwrap_or(default_error);
return Err(message.to_string());
}
let Some(token_data) = payload.get("data").and_then(Value::as_object) else {
return Err("响应格式无效".to_string());
};
let access_token = token_data
.get("access_token")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| "响应格式无效".to_string())?;
let mut updated_credentials = Map::new();
if let Some(new_refresh_token) = token_data
.get("refresh_token")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
{
if previous_refresh_token != Some(new_refresh_token) {
updated_credentials.insert(
"refresh_token".to_string(),
Value::String(new_refresh_token.to_string()),
);
}
}
Ok((
access_token.to_string(),
admin_provider_ops_frontend_updated_credentials(updated_credentials),
))
}

View File

@@ -1,119 +1,87 @@
mod headers;
mod helpers;
mod payload;
mod proxy;
mod request;
mod sub2api;
use crate::handlers::admin::provider::ops::providers::support::ADMIN_PROVIDER_OPS_VERIFY_RUST_ONLY_MESSAGE;
use crate::handlers::admin::request::AdminAppState;
use axum::{
body::Body,
http,
response::{IntoResponse, Response},
use aether_admin::provider::ops::{
admin_provider_ops_verify_failure, build_headers, get_architecture, normalize_architecture_id,
parse_verify_payload, ProviderOpsVerifyMode,
};
use serde_json::json;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider;
pub(super) fn admin_provider_ops_verify_headers(
architecture_id: &str,
config: &serde_json::Map<String, serde_json::Value>,
credentials: &serde_json::Map<String, serde_json::Value>,
) -> Result<reqwest::header::HeaderMap, String> {
headers::admin_provider_ops_verify_headers(architecture_id, config, credentials)
}
pub(super) fn admin_provider_ops_normalized_verify_architecture_id(architecture_id: &str) -> &str {
helpers::admin_provider_ops_normalized_verify_architecture_id(architecture_id)
}
pub(super) fn admin_provider_ops_verify_failure(message: impl Into<String>) -> serde_json::Value {
helpers::admin_provider_ops_verify_failure(message)
}
pub(super) fn admin_provider_ops_value_as_f64(value: Option<&serde_json::Value>) -> Option<f64> {
helpers::admin_provider_ops_value_as_f64(value)
}
pub(super) use proxy::{
admin_provider_ops_anyrouter_acw_cookie, admin_provider_ops_resolve_proxy_snapshot,
};
pub(super) use request::admin_provider_ops_execute_proxy_json_request;
pub(super) use sub2api::{
admin_provider_ops_sub2api_exchange_token, admin_provider_ops_sub2api_request_url,
};
pub(super) async fn admin_provider_ops_local_verify_response(
state: &AdminAppState<'_>,
provider: Option<&StoredProviderCatalogProvider>,
base_url: &str,
architecture_id: &str,
config: &serde_json::Map<String, serde_json::Value>,
credentials: &serde_json::Map<String, serde_json::Value>,
) -> serde_json::Value {
if architecture_id == "sub2api" {
return payload::admin_provider_ops_local_sub2api_verify_response(
let architecture_id = normalize_architecture_id(architecture_id);
let Some(architecture) = get_architecture(architecture_id) else {
return admin_provider_ops_verify_failure("认证验证仅支持 Rust execution runtime");
};
let base_url = base_url.trim().trim_end_matches('/');
if base_url.is_empty() {
return admin_provider_ops_verify_failure("请提供 API 地址");
}
let proxy_snapshot =
proxy::admin_provider_ops_resolve_proxy_snapshot(state, Some(config)).await;
if architecture.verify_mode == ProviderOpsVerifyMode::Sub2ApiExchange {
return sub2api::admin_provider_ops_local_sub2api_verify_response(
state,
provider,
base_url,
architecture.verify_endpoint,
credentials,
proxy_snapshot.as_ref(),
)
.await;
}
let mut resolved_config = config.clone();
if architecture_id == "anyrouter" {
if let Some(acw_cookie) =
helpers::admin_provider_ops_anyrouter_acw_cookie(state, base_url).await
if architecture.architecture_id == "anyrouter" {
if let Some(challenge) =
proxy::admin_provider_ops_anyrouter_acw_cookie(state, base_url, Some(config)).await
{
resolved_config.insert(
"acw_cookie".to_string(),
serde_json::Value::String(acw_cookie),
serde_json::Value::String(challenge.acw_cookie),
);
}
}
let verify_path = match architecture_id {
"anyrouter" => "/api/user/self",
"cubence" => "/api/v1/dashboard/overview",
"yescode" => "/api/v1/auth/profile",
"nekocode" => "/api/user/self",
"new_api" | "generic_api" => "/api/user/self",
_ => {
return helpers::admin_provider_ops_verify_failure(
ADMIN_PROVIDER_OPS_VERIFY_RUST_ONLY_MESSAGE,
)
}
};
let base_url = base_url.trim().trim_end_matches('/');
if base_url.is_empty() {
return helpers::admin_provider_ops_verify_failure("请提供 API 地址");
}
let headers = match headers::admin_provider_ops_verify_headers(
architecture_id,
&resolved_config,
credentials,
) {
let headers = match build_headers(architecture.architecture_id, &resolved_config, credentials) {
Ok(headers) => headers,
Err(message) => return helpers::admin_provider_ops_verify_failure(message),
Err(message) => return admin_provider_ops_verify_failure(message),
};
let response = match state
.http_client()
.get(format!("{base_url}{verify_path}"))
.headers(headers)
.send()
.await
let verify_url = format!("{base_url}{}", architecture.verify_endpoint);
let (status, response_json) = match request::admin_provider_ops_execute_get_json(
state,
&format!("provider-ops-verify:{}", architecture.architecture_id),
&verify_url,
&headers,
proxy_snapshot.as_ref(),
)
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => {
return helpers::admin_provider_ops_verify_failure("连接超时")
Ok(result) => result,
Err(error) => {
return admin_provider_ops_verify_failure(
request::admin_provider_ops_verify_execution_error_message(&error),
);
}
Err(err) if err.is_connect() => {
return helpers::admin_provider_ops_verify_failure(format!("连接失败: {err}"))
}
Err(err) => return helpers::admin_provider_ops_verify_failure(format!("验证失败: {err}")),
};
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => {
serde_json::from_slice::<serde_json::Value>(&bytes).unwrap_or_else(|_| json!({}))
}
Err(_) => json!({}),
};
match architecture_id {
"cubence" => payload::admin_provider_ops_cubence_verify_payload(status, &response_json),
"yescode" => payload::admin_provider_ops_yescode_verify_payload(status, &response_json),
"nekocode" => payload::admin_provider_ops_nekocode_verify_payload(status, &response_json),
_ => payload::admin_provider_ops_generic_verify_payload(status, &response_json),
}
parse_verify_payload(architecture.architecture_id, status, &response_json, None)
}

View File

@@ -1,88 +0,0 @@
use super::helpers::{
admin_provider_ops_frontend_updated_credentials, admin_provider_ops_json_object,
admin_provider_ops_sub2api_exchange_token, admin_provider_ops_value_as_f64,
admin_provider_ops_verify_failure, admin_provider_ops_verify_success,
admin_provider_ops_verify_user_payload,
};
use crate::handlers::admin::request::AdminAppState;
use aether_admin::provider::verify as admin_provider_verify_pure;
use serde_json::{json, Map, Value};
pub(super) fn admin_provider_ops_generic_verify_payload(
status: http::StatusCode,
response_json: &Value,
) -> Value {
admin_provider_verify_pure::admin_provider_ops_generic_verify_payload(status, response_json)
}
pub(super) fn admin_provider_ops_cubence_verify_payload(
status: http::StatusCode,
response_json: &Value,
) -> Value {
admin_provider_verify_pure::admin_provider_ops_cubence_verify_payload(status, response_json)
}
pub(super) fn admin_provider_ops_yescode_verify_payload(
status: http::StatusCode,
response_json: &Value,
) -> Value {
admin_provider_verify_pure::admin_provider_ops_yescode_verify_payload(status, response_json)
}
pub(super) fn admin_provider_ops_nekocode_verify_payload(
status: http::StatusCode,
response_json: &Value,
) -> Value {
admin_provider_verify_pure::admin_provider_ops_nekocode_verify_payload(status, response_json)
}
pub(super) fn admin_provider_ops_sub2api_verify_payload(
status: http::StatusCode,
response_json: &Value,
updated_credentials: Option<Map<String, Value>>,
) -> Value {
admin_provider_verify_pure::admin_provider_ops_sub2api_verify_payload(
status,
response_json,
updated_credentials,
)
}
pub(super) async fn admin_provider_ops_local_sub2api_verify_response(
state: &AdminAppState<'_>,
base_url: &str,
credentials: &Map<String, Value>,
) -> Value {
let base_url = base_url.trim().trim_end_matches('/');
if base_url.is_empty() {
return admin_provider_ops_verify_failure("请提供 API 地址");
}
let (access_token, updated_credentials) =
match admin_provider_ops_sub2api_exchange_token(state, base_url, credentials).await {
Ok(value) => value,
Err(message) => return admin_provider_ops_verify_failure(message),
};
let response = match state
.http_client()
.get(format!("{base_url}/api/v1/auth/me?timezone=Asia/Shanghai"))
.bearer_auth(access_token)
.send()
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => return admin_provider_ops_verify_failure("连接超时"),
Err(err) if err.is_connect() => {
return admin_provider_ops_verify_failure(format!("连接失败: {err}"));
}
Err(err) => return admin_provider_ops_verify_failure(format!("验证失败: {err}")),
};
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => serde_json::from_slice::<Value>(&bytes).unwrap_or_else(|_| json!({})),
Err(_) => json!({}),
};
admin_provider_ops_sub2api_verify_payload(status, &response_json, updated_credentials)
}

View File

@@ -0,0 +1,284 @@
use super::request::{
admin_provider_ops_execute_get_text, admin_provider_ops_execute_get_text_no_redirect,
};
use crate::handlers::admin::request::AdminAppState;
use aether_admin::provider::ops::admin_provider_ops_anyrouter_compute_acw_sc_v2;
use aether_contracts::ProxySnapshot;
use aether_data::repository::proxy_nodes::StoredProxyNode;
use aether_provider_transport::TransportTunnelAffinityLookup;
use regex::Regex;
use serde_json::{json, Map, Value};
use url::Url;
const TUNNEL_BASE_URL_EXTRA_KEY: &str = "tunnel_base_url";
const TUNNEL_OWNER_INSTANCE_ID_EXTRA_KEY: &str = "tunnel_owner_instance_id";
const TUNNEL_OWNER_OBSERVED_AT_EXTRA_KEY: &str = "tunnel_owner_observed_at_unix_secs";
pub(in super::super) struct AdminProviderOpsAnyrouterChallenge {
pub(in super::super) acw_cookie: String,
}
pub(in super::super) async fn admin_provider_ops_anyrouter_acw_cookie(
state: &AdminAppState<'_>,
base_url: &str,
connector_config: Option<&Map<String, Value>>,
) -> Option<AdminProviderOpsAnyrouterChallenge> {
let proxy_snapshot = admin_provider_ops_resolve_proxy_snapshot(state, connector_config).await;
let headers = reqwest::header::HeaderMap::from_iter([(
reqwest::header::USER_AGENT,
reqwest::header::HeaderValue::from_static(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
),
)]);
let response = if admin_provider_ops_proxy_uses_tunnel(proxy_snapshot.as_ref()) {
admin_provider_ops_execute_get_text(
state,
"provider-ops-acw:anyrouter",
base_url.trim_end_matches('/'),
&headers,
proxy_snapshot.as_ref(),
)
.await
.ok()?
} else {
admin_provider_ops_execute_get_text_no_redirect(
base_url.trim_end_matches('/'),
&headers,
proxy_snapshot.as_ref(),
)
.await
.ok()?
};
let compiled = Regex::new(r"var\s+arg1\s*=\s*'([0-9a-fA-F]{40})'").ok()?;
let captures = compiled.captures(&response.body)?;
let arg1 = captures.get(1)?.as_str();
admin_provider_ops_anyrouter_compute_acw_sc_v2(arg1).map(|value| {
AdminProviderOpsAnyrouterChallenge {
acw_cookie: format!("acw_sc__v2={value}"),
}
})
}
pub(in super::super) async fn admin_provider_ops_resolve_proxy_snapshot(
state: &AdminAppState<'_>,
connector_config: Option<&Map<String, Value>>,
) -> Option<ProxySnapshot> {
let explicit_node_id = connector_config
.and_then(|config| admin_provider_ops_string_field(config, "proxy_node_id"));
if let Some(snapshot) =
admin_provider_ops_resolve_proxy_node_snapshot(state, explicit_node_id.as_deref()).await
{
return Some(snapshot);
}
if explicit_node_id.is_none() {
let system_node_id = state
.read_system_config_json_value("system_proxy_node_id")
.await
.ok()
.flatten()
.and_then(|value| value.as_str().map(str::trim).map(ToOwned::to_owned))
.filter(|value| !value.is_empty());
if let Some(snapshot) =
admin_provider_ops_resolve_proxy_node_snapshot(state, system_node_id.as_deref()).await
{
return Some(snapshot);
}
}
connector_config
.and_then(|config| config.get("proxy"))
.and_then(admin_provider_ops_legacy_proxy_snapshot)
}
async fn admin_provider_ops_resolve_proxy_node_snapshot(
state: &AdminAppState<'_>,
node_id: Option<&str>,
) -> Option<ProxySnapshot> {
let node_id = node_id.map(str::trim).filter(|value| !value.is_empty())?;
let node = state.find_proxy_node(node_id).await.ok().flatten()?;
if node.status.trim() != "online" {
return None;
}
if node.tunnel_mode && node.tunnel_connected {
let mut extra = Map::new();
if let Ok(Some(owner)) = state.app().lookup_tunnel_attachment_owner(node_id).await {
extra.insert(
TUNNEL_BASE_URL_EXTRA_KEY.to_string(),
Value::String(owner.relay_base_url),
);
extra.insert(
TUNNEL_OWNER_INSTANCE_ID_EXTRA_KEY.to_string(),
Value::String(owner.gateway_instance_id),
);
extra.insert(
TUNNEL_OWNER_OBSERVED_AT_EXTRA_KEY.to_string(),
json!(owner.observed_at_unix_secs),
);
}
return Some(ProxySnapshot {
enabled: Some(true),
mode: Some("tunnel".to_string()),
node_id: Some(node_id.to_string()),
label: Some(node.name),
url: None,
extra: if extra.is_empty() {
None
} else {
Some(Value::Object(extra))
},
});
}
if !node.is_manual {
return None;
}
let proxy_url = node
.proxy_url
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())?;
Some(ProxySnapshot {
enabled: Some(true),
mode: admin_provider_ops_proxy_mode(Some(proxy_url)),
node_id: Some(node.id.clone()),
label: Some(node.name.clone()),
url: admin_provider_ops_proxy_url_with_node_auth(&node),
extra: None,
})
}
fn admin_provider_ops_legacy_proxy_snapshot(value: &Value) -> Option<ProxySnapshot> {
match value {
Value::String(proxy_url) => {
let proxy_url = proxy_url.trim();
if proxy_url.is_empty() {
return None;
}
Some(ProxySnapshot {
enabled: Some(true),
mode: admin_provider_ops_proxy_mode(Some(proxy_url)),
node_id: None,
label: None,
url: Some(proxy_url.to_string()),
extra: None,
})
}
Value::Object(object) => {
if object.get("enabled").and_then(Value::as_bool) == Some(false) {
return None;
}
let proxy_url = object
.get("url")
.or_else(|| object.get("proxy_url"))
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())?;
let username = object
.get("username")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty());
let password = object
.get("password")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty());
Some(ProxySnapshot {
enabled: Some(true),
mode: object
.get("mode")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| admin_provider_ops_proxy_mode(Some(proxy_url))),
node_id: None,
label: object
.get("label")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned),
url: admin_provider_ops_inject_proxy_auth(proxy_url, username, password)
.or_else(|| Some(proxy_url.to_string())),
extra: None,
})
}
_ => None,
}
}
fn admin_provider_ops_proxy_url_with_node_auth(node: &StoredProxyNode) -> Option<String> {
let proxy_url = node
.proxy_url
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())?;
let username = node
.proxy_username
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty());
let password = node
.proxy_password
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty());
admin_provider_ops_inject_proxy_auth(proxy_url, username, password)
.or_else(|| Some(proxy_url.to_string()))
}
fn admin_provider_ops_inject_proxy_auth(
proxy_url: &str,
username: Option<&str>,
password: Option<&str>,
) -> Option<String> {
let username = username.filter(|value| !value.is_empty())?;
let mut parsed = Url::parse(proxy_url).ok()?;
parsed.set_username(username).ok()?;
parsed.set_password(password).ok()?;
Some(parsed.to_string())
}
fn admin_provider_ops_proxy_mode(proxy_url: Option<&str>) -> Option<String> {
proxy_url
.and_then(|value| {
Url::parse(value)
.ok()
.map(|parsed| parsed.scheme().to_string())
})
.or_else(|| {
proxy_url.and_then(|value| {
value
.split_once("://")
.map(|(scheme, _)| scheme.trim().to_ascii_lowercase())
.filter(|scheme| !scheme.is_empty())
})
})
}
fn admin_provider_ops_proxy_uses_tunnel(proxy_snapshot: Option<&ProxySnapshot>) -> bool {
proxy_snapshot.is_some_and(|proxy| {
proxy.mode.as_deref().map(str::trim) == Some("tunnel")
|| (proxy
.url
.as_deref()
.map(str::trim)
.unwrap_or_default()
.is_empty()
&& proxy
.node_id
.as_deref()
.map(str::trim)
.is_some_and(|value| !value.is_empty()))
})
}
fn admin_provider_ops_string_field(config: &Map<String, Value>, key: &str) -> Option<String> {
config
.get(key)
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}

View File

@@ -0,0 +1,325 @@
use crate::handlers::admin::request::AdminAppState;
use crate::GatewayError;
use aether_contracts::{
ExecutionPlan, ExecutionResult, ExecutionTimeouts, ProxySnapshot, RequestBody,
};
use aether_http::{apply_http_client_config, HttpClientConfig};
use base64::{engine::general_purpose::STANDARD, Engine as _};
use flate2::read::{DeflateDecoder, GzDecoder};
use reqwest::redirect::Policy;
use serde_json::{json, Value};
use std::collections::BTreeMap;
use std::io::Read;
const ADMIN_PROVIDER_OPS_VERIFY_TIMEOUT_MS: u64 = 30_000;
pub(super) struct AdminProviderOpsTextResponse {
pub(super) body: String,
}
pub(super) async fn admin_provider_ops_execute_get_json(
state: &AdminAppState<'_>,
request_id: &str,
url: &str,
headers: &reqwest::header::HeaderMap,
proxy_snapshot: Option<&ProxySnapshot>,
) -> Result<(http::StatusCode, Value), String> {
if proxy_snapshot.is_none() {
let response = match state
.http_client()
.get(url)
.headers(headers.clone())
.send()
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => return Err("timeout".to_string()),
Err(err) => return Err(err.to_string()),
};
let status = response.status();
let content_encoding = response
.headers()
.get(reqwest::header::CONTENT_ENCODING)
.and_then(|value| value.to_str().ok())
.map(ToOwned::to_owned);
let bytes = response.bytes().await.map_err(|err| err.to_string())?;
let decoded_bytes =
admin_provider_ops_decode_response_bytes(bytes.as_ref(), content_encoding.as_deref())
.unwrap_or_else(|| bytes.to_vec());
let response_json = match serde_json::from_slice::<Value>(&decoded_bytes) {
Ok(value) => value,
Err(err) if status != http::StatusCode::OK => json!({}),
Err(err) => return Err(format!("upstream response is not valid JSON: {err}")),
};
return Ok((status, response_json));
}
let result = admin_provider_ops_execute_request(
state,
request_id,
reqwest::Method::GET,
url,
headers,
None,
proxy_snapshot,
)
.await?;
Ok((
admin_provider_ops_execution_status_code(&result),
admin_provider_ops_execution_json_body(&result),
))
}
pub(in super::super) async fn admin_provider_ops_execute_proxy_json_request(
state: &AdminAppState<'_>,
request_id: &str,
method: reqwest::Method,
url: &str,
headers: &reqwest::header::HeaderMap,
json_body: Option<Value>,
proxy_snapshot: &ProxySnapshot,
) -> Result<(http::StatusCode, Value), String> {
let result = admin_provider_ops_execute_request(
state,
request_id,
method,
url,
headers,
json_body,
Some(proxy_snapshot),
)
.await?;
Ok((
admin_provider_ops_execution_status_code(&result),
admin_provider_ops_execution_json_body(&result),
))
}
pub(super) async fn admin_provider_ops_execute_get_text(
state: &AdminAppState<'_>,
request_id: &str,
url: &str,
headers: &reqwest::header::HeaderMap,
proxy_snapshot: Option<&ProxySnapshot>,
) -> Result<AdminProviderOpsTextResponse, String> {
let result = admin_provider_ops_execute_request(
state,
request_id,
reqwest::Method::GET,
url,
headers,
None,
proxy_snapshot,
)
.await?;
let body = result
.body
.as_ref()
.and_then(|body| admin_provider_ops_execution_body_bytes(&result.headers, body))
.map(|bytes| String::from_utf8_lossy(&bytes).to_string())
.or_else(|| {
result
.body
.as_ref()
.and_then(|body| body.json_body.as_ref())
.and_then(|value| serde_json::to_string(value).ok())
})
.unwrap_or_default();
Ok(AdminProviderOpsTextResponse { body })
}
pub(super) async fn admin_provider_ops_execute_get_text_no_redirect(
url: &str,
headers: &reqwest::header::HeaderMap,
proxy_snapshot: Option<&ProxySnapshot>,
) -> Result<AdminProviderOpsTextResponse, String> {
let mut builder = apply_http_client_config(
reqwest::Client::builder().redirect(Policy::none()),
&HttpClientConfig {
connect_timeout_ms: Some(10_000),
request_timeout_ms: Some(ADMIN_PROVIDER_OPS_VERIFY_TIMEOUT_MS),
use_rustls_tls: true,
http2_adaptive_window: true,
..HttpClientConfig::default()
},
);
if let Some(proxy_url) = proxy_snapshot
.and_then(|proxy| proxy.url.as_deref())
.map(str::trim)
.filter(|value| !value.is_empty())
{
let proxy = reqwest::Proxy::all(proxy_url).map_err(|err| format!("连接失败: {err}"))?;
builder = builder.proxy(proxy);
}
let client = builder.build().map_err(|err| format!("验证失败: {err}"))?;
let response = match client.get(url).headers(headers.clone()).send().await {
Ok(response) => response,
Err(err) if err.is_timeout() => return Err("连接超时".to_string()),
Err(err) if err.is_connect() => return Err(format!("连接失败: {err}")),
Err(err) => return Err(format!("验证失败: {err}")),
};
let content_encoding = response
.headers()
.get(reqwest::header::CONTENT_ENCODING)
.and_then(|value| value.to_str().ok())
.map(ToOwned::to_owned);
let body = response
.bytes()
.await
.map_err(|err| format!("验证失败: {err}"))
.map(|bytes| {
admin_provider_ops_decode_response_bytes(bytes.as_ref(), content_encoding.as_deref())
.unwrap_or_else(|| bytes.to_vec())
})
.map(|bytes| String::from_utf8_lossy(&bytes).to_string())?;
Ok(AdminProviderOpsTextResponse { body })
}
async fn admin_provider_ops_execute_request(
state: &AdminAppState<'_>,
request_id: &str,
method: reqwest::Method,
url: &str,
headers: &reqwest::header::HeaderMap,
json_body: Option<Value>,
proxy_snapshot: Option<&ProxySnapshot>,
) -> Result<ExecutionResult, String> {
let has_json_body = json_body.is_some();
let body = json_body
.map(RequestBody::from_json)
.unwrap_or(RequestBody {
json_body: None,
body_bytes_b64: None,
body_ref: None,
});
let plan = ExecutionPlan {
request_id: request_id.to_string(),
candidate_id: None,
provider_name: Some("provider_ops".to_string()),
provider_id: String::new(),
endpoint_id: String::new(),
key_id: String::new(),
method: method.as_str().to_string(),
url: url.to_string(),
headers: admin_provider_ops_execution_headers(headers),
content_type: has_json_body.then(|| "application/json".to_string()),
content_encoding: None,
body,
stream: false,
client_api_format: "provider_ops:verify".to_string(),
provider_api_format: "provider_ops:verify".to_string(),
model_name: Some("verify-auth".to_string()),
proxy: proxy_snapshot.cloned(),
tls_profile: None,
timeouts: Some(ExecutionTimeouts {
connect_ms: Some(ADMIN_PROVIDER_OPS_VERIFY_TIMEOUT_MS),
read_ms: Some(ADMIN_PROVIDER_OPS_VERIFY_TIMEOUT_MS),
write_ms: Some(ADMIN_PROVIDER_OPS_VERIFY_TIMEOUT_MS),
pool_ms: Some(ADMIN_PROVIDER_OPS_VERIFY_TIMEOUT_MS),
total_ms: Some(ADMIN_PROVIDER_OPS_VERIFY_TIMEOUT_MS),
..ExecutionTimeouts::default()
}),
};
state
.execute_execution_runtime_sync_plan(None, &plan)
.await
.map_err(admin_provider_ops_gateway_error_message)
}
fn admin_provider_ops_execution_headers(
headers: &reqwest::header::HeaderMap,
) -> BTreeMap<String, String> {
headers
.iter()
.filter_map(|(name, value)| {
value
.to_str()
.ok()
.map(|text| (name.as_str().to_string(), text.to_string()))
})
.collect()
}
fn admin_provider_ops_execution_status_code(result: &ExecutionResult) -> http::StatusCode {
http::StatusCode::from_u16(result.status_code).unwrap_or(http::StatusCode::BAD_GATEWAY)
}
fn admin_provider_ops_execution_json_body(result: &ExecutionResult) -> Value {
result
.body
.as_ref()
.and_then(|body| body.json_body.clone())
.or_else(|| {
result
.body
.as_ref()
.and_then(|body| admin_provider_ops_execution_body_bytes(&result.headers, body))
.and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok())
})
.unwrap_or_else(|| json!({}))
}
fn admin_provider_ops_execution_body_bytes(
headers: &BTreeMap<String, String>,
body: &aether_contracts::ResponseBody,
) -> Option<Vec<u8>> {
let bytes = body
.body_bytes_b64
.as_deref()
.and_then(|value| STANDARD.decode(value).ok())?;
admin_provider_ops_decode_response_bytes(
&bytes,
headers.get("content-encoding").map(String::as_str),
)
.or(Some(bytes))
}
fn admin_provider_ops_decode_response_bytes(
bytes: &[u8],
content_encoding: Option<&str>,
) -> Option<Vec<u8>> {
let encoding = content_encoding
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase());
match encoding.as_deref() {
Some("gzip") => {
let mut decoder = GzDecoder::new(bytes);
let mut out = Vec::new();
decoder.read_to_end(&mut out).ok()?;
Some(out)
}
Some("deflate") => {
let mut decoder = DeflateDecoder::new(bytes);
let mut out = Vec::new();
decoder.read_to_end(&mut out).ok()?;
Some(out)
}
_ => None,
}
}
fn admin_provider_ops_gateway_error_message(error: GatewayError) -> String {
match error {
GatewayError::UpstreamUnavailable { message, .. }
| GatewayError::ControlUnavailable { message, .. }
| GatewayError::Internal(message) => message,
}
}
pub(super) fn admin_provider_ops_verify_execution_error_message(error: &str) -> String {
let normalized = error.trim();
let lower = normalized.to_ascii_lowercase();
if lower.contains("timeout") || lower.contains("timed out") {
return "连接超时".to_string();
}
if lower.contains("connect")
|| lower.contains("connection")
|| lower.contains("dns")
|| lower.contains("proxy")
|| lower.contains("relay")
{
return format!("连接失败: {normalized}");
}
format!("验证失败: {normalized}")
}

View File

@@ -0,0 +1,401 @@
use super::request::{
admin_provider_ops_execute_proxy_json_request,
admin_provider_ops_verify_execution_error_message,
};
use crate::handlers::admin::provider::ops::providers::config::persist_admin_provider_ops_runtime_credentials;
use crate::handlers::admin::request::AdminAppState;
use aether_admin::provider::ops::{
admin_provider_ops_frontend_updated_credentials, admin_provider_ops_verify_failure,
parse_verify_payload, ADMIN_PROVIDER_OPS_USER_AGENT,
};
use aether_contracts::ProxySnapshot;
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider;
use aether_http::{apply_http_client_config, HttpClientConfig};
use serde_json::{json, Map, Value};
use tracing::warn;
pub(super) async fn admin_provider_ops_local_sub2api_verify_response(
state: &AdminAppState<'_>,
provider: Option<&StoredProviderCatalogProvider>,
base_url: &str,
verify_endpoint: &str,
credentials: &Map<String, Value>,
proxy_snapshot: Option<&ProxySnapshot>,
) -> Value {
let (access_token, updated_credentials, frontend_updated_credentials) =
match admin_provider_ops_sub2api_exchange_token(
state,
base_url,
credentials,
proxy_snapshot,
)
.await
{
Ok(value) => value,
Err(message) => return admin_provider_ops_verify_failure(message),
};
if let Some(provider) = provider.filter(|_| !updated_credentials.is_empty()) {
if let Err(err) =
persist_admin_provider_ops_runtime_credentials(state, provider, &updated_credentials)
.await
{
warn!(
provider_id = %provider.id,
error = ?err,
"failed to persist sub2api verify runtime credentials"
);
}
}
let verify_url = admin_provider_ops_sub2api_request_url(base_url, verify_endpoint);
let auth_value = match reqwest::header::HeaderValue::from_str(&format!("Bearer {access_token}"))
{
Ok(value) => value,
Err(_) => return admin_provider_ops_verify_failure("访问令牌格式无效"),
};
let auth_headers = reqwest::header::HeaderMap::from_iter([
(reqwest::header::AUTHORIZATION, auth_value),
(
reqwest::header::USER_AGENT,
reqwest::header::HeaderValue::from_static(ADMIN_PROVIDER_OPS_USER_AGENT),
),
(
reqwest::header::ACCEPT,
reqwest::header::HeaderValue::from_static("*/*"),
),
]);
let (status, response_json) = if let Some(proxy_snapshot) = proxy_snapshot {
match admin_provider_ops_execute_proxy_json_request(
state,
"provider-ops-verify:sub2api",
reqwest::Method::GET,
&verify_url,
&auth_headers,
None,
proxy_snapshot,
)
.await
{
Ok(result) => result,
Err(error) => {
return admin_provider_ops_verify_failure(
admin_provider_ops_verify_execution_error_message(&error),
);
}
}
} else {
let http_client = match admin_provider_ops_sub2api_http_client() {
Ok(client) => client,
Err(err) => {
return admin_provider_ops_verify_failure(format!("验证失败: {err}"));
}
};
let response = match http_client
.get(&verify_url)
.headers(auth_headers)
.send()
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => return admin_provider_ops_verify_failure("连接超时"),
Err(err) if err.is_connect() => {
return admin_provider_ops_verify_failure(format!("连接失败: {err}"));
}
Err(err) => return admin_provider_ops_verify_failure(format!("验证失败: {err}")),
};
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => serde_json::from_slice::<Value>(&bytes).unwrap_or_else(|_| json!({})),
Err(_) => json!({}),
};
(status, response_json)
};
parse_verify_payload(
"sub2api",
status,
&response_json,
frontend_updated_credentials,
)
}
fn admin_provider_ops_sub2api_http_client() -> Result<reqwest::Client, reqwest::Error> {
let builder = apply_http_client_config(
reqwest::Client::builder().http1_only(),
&HttpClientConfig {
connect_timeout_ms: Some(10_000),
request_timeout_ms: Some(30_000),
use_rustls_tls: true,
user_agent: Some(ADMIN_PROVIDER_OPS_USER_AGENT.to_string()),
..HttpClientConfig::default()
},
);
builder.build()
}
// 对齐 Python httpx.AsyncClient(base_url=...) 的行为:
// 以 "/" 开头的端点始终相对站点根路径解析,而不是简单字符串拼接。
pub(in super::super) fn admin_provider_ops_sub2api_request_url(
base_url: &str,
endpoint: &str,
) -> String {
let trimmed_base_url = base_url.trim().trim_end_matches('/');
let trimmed_endpoint = endpoint.trim();
if trimmed_endpoint.is_empty() {
return trimmed_base_url.to_string();
}
if trimmed_endpoint.starts_with("http://") || trimmed_endpoint.starts_with("https://") {
return trimmed_endpoint.to_string();
}
reqwest::Url::parse(trimmed_base_url)
.and_then(|base| base.join(trimmed_endpoint))
.map(|url| url.to_string())
.unwrap_or_else(|_| format!("{trimmed_base_url}{trimmed_endpoint}"))
}
fn admin_provider_ops_sub2api_updated_credentials(
token_data: &Map<String, Value>,
previous_refresh_token: Option<&str>,
) -> Map<String, Value> {
let mut updated_credentials = Map::new();
if let Some(new_refresh_token) = token_data
.get("refresh_token")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
{
if previous_refresh_token != Some(new_refresh_token) {
updated_credentials.insert(
"refresh_token".to_string(),
Value::String(new_refresh_token.to_string()),
);
}
}
if let Some(access_token) = token_data
.get("access_token")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
{
updated_credentials.insert(
"_cached_access_token".to_string(),
Value::String(access_token.to_string()),
);
updated_credentials.insert(
"_cached_token_expires_at".to_string(),
Value::from(admin_provider_ops_sub2api_cached_token_expires_at(
token_data,
)),
);
}
updated_credentials
}
fn admin_provider_ops_sub2api_cached_token_expires_at(token_data: &Map<String, Value>) -> f64 {
if let Some(token_expires_at) = token_data
.get("token_expires_at")
.and_then(admin_provider_ops_sub2api_json_number)
{
return token_expires_at / 1000.0 - 60.0;
}
let expires_in = token_data
.get("expires_in")
.and_then(admin_provider_ops_sub2api_json_number)
.unwrap_or(900.0);
admin_provider_ops_sub2api_unix_timestamp_secs() + expires_in - 60.0
}
fn admin_provider_ops_sub2api_json_number(value: &Value) -> Option<f64> {
value
.as_f64()
.or_else(|| value.as_i64().map(|value| value as f64))
.or_else(|| value.as_u64().map(|value| value as f64))
.or_else(|| {
value
.as_str()
.and_then(|value| value.trim().parse::<f64>().ok())
})
}
fn admin_provider_ops_sub2api_unix_timestamp_secs() -> f64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|duration| duration.as_secs_f64())
.unwrap_or_default()
}
fn admin_provider_ops_sub2api_cached_access_token(
credentials: &Map<String, Value>,
) -> Option<String> {
let cached_access_token = credentials
.get("_cached_access_token")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())?;
let cached_expires_at = credentials
.get("_cached_token_expires_at")
.and_then(admin_provider_ops_sub2api_json_number)
.unwrap_or_default();
if admin_provider_ops_sub2api_unix_timestamp_secs() >= cached_expires_at {
return None;
}
Some(cached_access_token.to_string())
}
async fn admin_provider_ops_sub2api_token_request(
state: &AdminAppState<'_>,
base_url: &str,
path: &str,
body: Value,
default_error: &str,
proxy_snapshot: Option<&ProxySnapshot>,
) -> Result<Map<String, Value>, String> {
let url = admin_provider_ops_sub2api_request_url(base_url, path);
let default_headers = reqwest::header::HeaderMap::from_iter([
(
reqwest::header::USER_AGENT,
reqwest::header::HeaderValue::from_static(ADMIN_PROVIDER_OPS_USER_AGENT),
),
(
reqwest::header::ACCEPT,
reqwest::header::HeaderValue::from_static("*/*"),
),
]);
let (status, response_json) = if let Some(proxy_snapshot) = proxy_snapshot {
admin_provider_ops_execute_proxy_json_request(
state,
&format!("provider-ops-sub2api:{path}"),
reqwest::Method::POST,
&url,
&default_headers,
Some(body),
proxy_snapshot,
)
.await
.map_err(|error| admin_provider_ops_verify_execution_error_message(&error))?
} else {
let client =
admin_provider_ops_sub2api_http_client().map_err(|err| format!("验证失败: {err}"))?;
let response = match client
.post(url)
.headers(default_headers)
.json(&body)
.send()
.await
{
Ok(response) => response,
Err(err) if err.is_timeout() => return Err("连接超时".to_string()),
Err(err) if err.is_connect() => return Err(format!("连接失败: {err}")),
Err(err) => return Err(format!("验证失败: {err}")),
};
let status = response.status();
let response_json = match response.bytes().await {
Ok(bytes) => serde_json::from_slice::<Value>(&bytes).unwrap_or_else(|_| json!({})),
Err(_) => json!({}),
};
(status, response_json)
};
let payload = response_json.as_object().cloned().unwrap_or_default();
if status != http::StatusCode::OK
|| payload.get("code").and_then(Value::as_i64).unwrap_or(-1) != 0
{
let message = payload
.get("message")
.and_then(Value::as_str)
.unwrap_or(default_error);
return Err(message.to_string());
}
payload
.get("data")
.and_then(Value::as_object)
.cloned()
.ok_or_else(|| "响应格式无效".to_string())
}
pub(in super::super) async fn admin_provider_ops_sub2api_exchange_token(
state: &AdminAppState<'_>,
base_url: &str,
credentials: &Map<String, Value>,
proxy_snapshot: Option<&ProxySnapshot>,
) -> Result<(String, Map<String, Value>, Option<Map<String, Value>>), String> {
if let Some(cached_access_token) = admin_provider_ops_sub2api_cached_access_token(credentials) {
return Ok((cached_access_token, Map::new(), None));
}
let email = credentials
.get("email")
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default();
let password = credentials
.get("password")
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default();
let refresh_token = credentials
.get("refresh_token")
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default();
let mut refresh_error = None::<String>;
let token_data = if !refresh_token.is_empty() {
match admin_provider_ops_sub2api_token_request(
state,
base_url,
"/api/v1/auth/refresh",
json!({ "refresh_token": refresh_token }),
"Refresh Token 无效或已过期",
proxy_snapshot,
)
.await
{
Ok(token_data) => token_data,
Err(err) => {
if email.is_empty() || password.is_empty() {
return Err(err);
}
refresh_error = Some(err);
admin_provider_ops_sub2api_token_request(
state,
base_url,
"/api/v1/auth/login",
json!({ "email": email, "password": password }),
"登录失败",
proxy_snapshot,
)
.await?
}
}
} else if !email.is_empty() && !password.is_empty() {
admin_provider_ops_sub2api_token_request(
state,
base_url,
"/api/v1/auth/login",
json!({ "email": email, "password": password }),
"登录失败",
proxy_snapshot,
)
.await?
} else {
return Err(refresh_error.unwrap_or_else(|| "请填写账号密码或 Refresh Token".to_string()));
};
let access_token = token_data
.get("access_token")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| "响应格式无效".to_string())?;
let updated_credentials =
admin_provider_ops_sub2api_updated_credentials(&token_data, Some(refresh_token));
Ok((
access_token.to_string(),
updated_credentials.clone(),
admin_provider_ops_frontend_updated_credentials(updated_credentials),
))
}

View File

@@ -8,6 +8,7 @@ use aether_data_contracts::repository::candidates::{
UpsertRequestCandidateRecord,
};
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider;
use axum::body::Body;
use axum::http::StatusCode;
use axum::routing::{get, post};
use axum::{Json, Router};
@@ -253,6 +254,158 @@ async fn gateway_provider_checkin_runs_local_query_balance_for_configured_provid
ops_handle.abort();
}
#[tokio::test]
async fn gateway_provider_checkin_counts_anyrouter_auto_signin_as_success() {
fn sample_provider(provider_id: &str, ops_url: &str) -> StoredProviderCatalogProvider {
StoredProviderCatalogProvider::new(
provider_id.to_string(),
"openai".to_string(),
Some("https://example.com".to_string()),
"custom".to_string(),
)
.expect("provider should build")
.with_routing_fields(10)
.with_transport_fields(
true,
false,
true,
None,
None,
None,
None,
None,
Some(json!({
"provider_ops": {
"architecture_id": "anyrouter",
"base_url": ops_url,
"connector": {
"auth_type": "cookie",
"config": {},
"credentials": {
"session_cookie": encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
"session=MTIzfGVIaDRlQUpwWkFOcGJuU3F1d0RfVkhsNWVYa0lkWE5sY201aGJXVUdjM1J5YVc1bkRCQUFCV0ZzYVdObHxzaWc",
).expect("session cookie should encrypt"),
}
}
}
})),
)
}
let sign_in_hits = Arc::new(Mutex::new(0usize));
let sign_in_hits_clone = Arc::clone(&sign_in_hits);
let balance_hits = Arc::new(Mutex::new(0usize));
let balance_hits_clone = Arc::clone(&balance_hits);
let ops = Router::new()
.route(
"/",
get(|| async move {
(
StatusCode::OK,
Body::from(
"<html><script>var arg1 = '0123456789abcdef0123456789abcdef01234567';</script></html>",
),
)
}),
)
.route(
"/api/user/sign_in",
post(move |headers: axum::http::HeaderMap| {
let sign_in_hits_inner = Arc::clone(&sign_in_hits_clone);
async move {
*sign_in_hits_inner.lock().expect("mutex should lock") += 1;
let cookie = headers
.get(axum::http::header::COOKIE)
.and_then(|value| value.to_str().ok())
.expect("cookie header should exist");
assert!(cookie.contains("acw_sc__v2="));
assert!(cookie.contains(
"session=MTIzfGVIaDRlQUpwWkFOcGJuU3F1d0RfVkhsNWVYa0lkWE5sY201aGJXVUdjM1J5YVc1bkRCQUFCV0ZzYVdObHxzaWc"
));
assert_eq!(
headers
.get("New-Api-User")
.and_then(|value| value.to_str().ok()),
Some("42")
);
(
StatusCode::OK,
Json(json!({
"success": true,
"message": "Anyrouter 签到成功",
})),
)
}
}),
)
.route(
"/api/user/self",
get(move |headers: axum::http::HeaderMap| {
let balance_hits_inner = Arc::clone(&balance_hits_clone);
async move {
*balance_hits_inner.lock().expect("mutex should lock") += 1;
let cookie = headers
.get(axum::http::header::COOKIE)
.and_then(|value| value.to_str().ok())
.expect("cookie header should exist");
assert!(cookie.contains("acw_sc__v2="));
assert_eq!(
headers
.get("New-Api-User")
.and_then(|value| value.to_str().ok()),
Some("42")
);
(
StatusCode::OK,
Json(json!({
"success": true,
"data": {
"quota": 2500000,
"used_quota": 500000
}
})),
)
}
}),
);
let (ops_url, ops_handle) = start_server(ops).await;
let repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![sample_provider("provider-anyrouter", &ops_url)],
vec![],
vec![],
));
let gateway_state = AppState::new()
.expect("gateway state should build")
.with_data_state_for_tests(
crate::data::GatewayDataState::with_provider_catalog_repository_for_tests(repository)
.with_system_config_values_for_tests([
("enable_provider_checkin".to_string(), json!(true)),
("provider_checkin_time".to_string(), json!("01:05")),
])
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
);
let summary = crate::maintenance::perform_provider_checkin_once(&gateway_state)
.await
.expect("provider checkin should succeed");
assert_eq!(
summary,
ProviderCheckinRunSummary {
attempted: 1,
succeeded: 1,
failed: 0,
skipped: 0,
}
);
assert_eq!(*sign_in_hits.lock().expect("mutex should lock"), 1);
assert_eq!(*balance_hits.lock().expect("mutex should lock"), 1);
ops_handle.abort();
}
#[tokio::test]
async fn gateway_provider_checkin_skips_when_disabled_via_system_config() {
let repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(

View File

@@ -2,19 +2,14 @@ use aether_scheduler_core::SchedulerPriorityMode;
use crate::{AppState, GatewayError};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub(crate) enum SchedulerSchedulingMode {
FixedOrder,
#[default]
CacheAffinity,
LoadBalance,
}
impl Default for SchedulerSchedulingMode {
fn default() -> Self {
Self::CacheAffinity
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct SchedulerOrderingConfig {
pub(crate) priority_mode: SchedulerPriorityMode,

View File

@@ -1772,7 +1772,7 @@ async fn gateway_retries_next_local_openai_chat_stream_candidate_after_retryable
"provider-openai-local-stream-backup",
"endpoint-openai-local-stream-backup",
"key-openai-local-stream-backup",
10,
20,
2,
"gpt-5-upstream-backup",
),

View File

@@ -1012,7 +1012,7 @@ async fn gateway_retries_next_local_openai_chat_sync_candidate_with_local_failov
"provider-openai-local-backup",
"endpoint-openai-local-backup",
"key-openai-local-backup",
10,
20,
2,
"gpt-5-upstream-backup",
),

View File

@@ -385,8 +385,8 @@ fn admin_provider_ops_route_owners_stay_explicit() {
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/routes/verify.rs",
);
assert!(
verify.contains("super::super::verify::{"),
"verify.rs should depend directly on verify owner"
verify.contains("super::super::verify::admin_provider_ops_local_verify_response"),
"verify.rs should depend directly on gateway verify runtime owner"
);
let connect = read_workspace_file(
@@ -414,6 +414,28 @@ fn admin_provider_ops_route_owners_stay_explicit() {
);
}
#[test]
fn admin_provider_ops_architecture_registry_uses_pure_owner() {
let architectures =
read_workspace_file("apps/aether-gateway/src/handlers/admin/provider/ops/architectures.rs");
for pattern in [
"use aether_admin::provider::ops::{get_architecture, list_architectures};",
"list_architectures(false)",
"get_architecture(architecture_id)",
] {
assert!(
architectures.contains(pattern),
"handlers/admin/provider/ops/architectures.rs should delegate architecture registry to pure owner {pattern}"
);
}
assert!(
!workspace_file_exists(
"apps/aether-gateway/src/handlers/admin/provider/ops/architectures.all.json"
),
"handlers/admin/provider/ops/architectures.all.json should be removed after moving architecture registry into aether-admin"
);
}
#[test]
fn admin_provider_summary_mod_stays_thin() {
let summary_mod =
@@ -1150,7 +1172,7 @@ fn admin_provider_ops_providers_mod_stays_thin() {
"pub(super) struct AdminProviderOpsSaveConfigRequest",
"pub(super) struct AdminProviderOpsConnectRequest",
"pub(super) struct AdminProviderOpsExecuteActionRequest",
"pub(super) struct AdminProviderOpsCheckinOutcome",
"ProviderOpsCheckinOutcome as AdminProviderOpsCheckinOutcome",
] {
assert!(
providers_support.contains(pattern),
@@ -1234,10 +1256,11 @@ fn admin_provider_ops_actions_mod_stays_thin() {
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/actions/support.rs",
);
for pattern in [
"pub(super) fn admin_provider_ops_resolved_action_config(",
"pub(super) fn admin_provider_ops_checkin_data(",
"pub(super) fn admin_provider_ops_json_object_map(",
"pub(super) fn admin_provider_ops_request_url(",
"pub(super) fn admin_provider_ops_request_method(",
"pub(super) fn admin_provider_ops_should_use_rust_only_action_stub(",
"pub(super) fn admin_provider_ops_parse_rfc3339_unix_secs(",
] {
assert!(
actions_support.contains(pattern),
@@ -1298,29 +1321,34 @@ fn admin_provider_ops_actions_mod_stays_thin() {
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/actions/query_balance/mod.rs",
);
for pattern in [
"mod parsers;",
"mod sub2api;",
"mod yescode;",
"pub(super) async fn admin_provider_ops_run_query_balance_action(",
"parsers::admin_provider_ops_new_api_balance_payload(",
"parse_query_balance_payload(",
"yescode::admin_provider_ops_yescode_balance_payload(",
"sub2api::admin_provider_ops_sub2api_balance_payload(",
] {
assert!(
actions_query_balance_mod.contains(pattern),
"handlers/admin/provider/ops/providers/actions/query_balance/mod.rs should keep thin query_balance entry seam {pattern}"
);
}
let actions_query_balance_parsers = read_workspace_file(
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/actions/query_balance/parsers.rs",
assert!(
!workspace_file_exists(
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/actions/query_balance/parsers.rs"
),
"handlers/admin/provider/ops/providers/actions/query_balance/parsers.rs should be removed after moving balance parsing into aether-admin"
);
let pure_actions = read_workspace_file("crates/aether-admin/src/provider/ops/actions.rs");
for pattern in [
"pub(super) fn admin_provider_ops_new_api_balance_payload(",
"pub(super) fn admin_provider_ops_cubence_balance_payload(",
"pub(super) fn admin_provider_ops_nekocode_balance_payload(",
"pub(super) fn admin_provider_ops_attach_balance_checkin_outcome(",
"pub fn parse_query_balance_payload(",
"pub fn parse_sub2api_balance_payload(",
"pub fn parse_yescode_combined_balance_payload(",
"pub fn attach_balance_checkin_outcome(",
] {
assert!(
actions_query_balance_parsers.contains(pattern),
"handlers/admin/provider/ops/providers/actions/query_balance/parsers.rs should own {pattern}"
pure_actions.contains(pattern),
"crates/aether-admin/src/provider/ops/actions.rs should own {pattern}"
);
}
let actions_query_balance_yescode = read_workspace_file(
@@ -1339,6 +1367,84 @@ fn admin_provider_ops_actions_mod_stays_thin() {
);
}
#[test]
fn admin_provider_ops_verify_runtime_and_pure_owners_stay_explicit() {
let verify_mod = read_workspace_file(
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/verify/mod.rs",
);
for pattern in [
"mod proxy;",
"mod request;",
"mod sub2api;",
"pub(super) async fn admin_provider_ops_local_verify_response(",
] {
assert!(
verify_mod.contains(pattern),
"handlers/admin/provider/ops/providers/verify/mod.rs should keep runtime verify entry seam {pattern}"
);
}
let verify_proxy = read_workspace_file(
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/verify/proxy.rs",
);
for pattern in [
"struct AdminProviderOpsAnyrouterChallenge",
"fn admin_provider_ops_anyrouter_acw_cookie(",
"fn admin_provider_ops_resolve_proxy_snapshot(",
] {
assert!(
verify_proxy.contains(pattern),
"handlers/admin/provider/ops/providers/verify/proxy.rs should own {pattern}"
);
}
let verify_request = read_workspace_file(
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/verify/request.rs",
);
for pattern in [
"fn admin_provider_ops_execute_get_json(",
"fn admin_provider_ops_execute_proxy_json_request(",
"fn admin_provider_ops_verify_execution_error_message(",
] {
assert!(
verify_request.contains(pattern),
"handlers/admin/provider/ops/providers/verify/request.rs should own {pattern}"
);
}
let verify_sub2api = read_workspace_file(
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/verify/sub2api.rs",
);
for pattern in [
"fn admin_provider_ops_local_sub2api_verify_response(",
"fn admin_provider_ops_sub2api_exchange_token(",
] {
assert!(
verify_sub2api.contains(pattern),
"handlers/admin/provider/ops/providers/verify/sub2api.rs should own {pattern}"
);
}
for path in [
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/verify/helpers.rs",
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/verify/headers.rs",
"apps/aether-gateway/src/handlers/admin/provider/ops/providers/verify/payload.rs",
] {
assert!(
!workspace_file_exists(path),
"{path} should be removed after splitting verify runtime owners"
);
}
let pure_verify = read_workspace_file("crates/aether-admin/src/provider/ops/verify.rs");
for pattern in ["pub fn build_headers(", "pub fn parse_verify_payload("] {
assert!(
pure_verify.contains(pattern),
"crates/aether-admin/src/provider/ops/verify.rs should own {pattern}"
);
}
}
#[test]
fn admin_provider_oauth_dispatch_uses_helper_owner() {
let dispatch_mod = read_workspace_file(

File diff suppressed because it is too large Load Diff