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

feat: 重要通知模块、Server 酱独立配置与额度提醒

Browse Source 
- 新增重要通知统一模块（邮件 + Server 酱）作为后台任务通知出口
- 拆出独立的 Server 酱 配置页（SendKey + Markdown 模板，支持 {title}/{body} 变量替换），通过仪表盘内置工具入口进入
- 新增提供商额度提醒后台 worker：余额低于阈值时通过重要通知推送，提供商配置页加入额度提醒开关与阈值
- 重要通知页加入配置可用性守卫：未配置任一通道时禁用总开关，未配置邮件/SendKey 时禁用对应通道开关
- 测试通知端点支持 channel 过滤（all/email/server_chan），并绕过总开关与通道开关，便于配置阶段先验证通道
- 修复：测试通知路由未在 buffered-body 白名单导致 channel 参数丢失、测试时邮件分支被误触发
- 修复：sub2api 验证响应中 username 为 null 时正确回退到 email，避免误报"验证响应缺少: 用户信息"

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
yangrs
2026-05-18 23:51:14 +08:00
parent d51b44d642
commit 6c16f399d4
32 changed files with 2441 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
apps/aether-gateway/src/admin_api.rs
View File

@@ -7,10 +7,11 @@ pub(crate) use crate::handlers::admin::{
provider_quota_refresh_endpoint_for_provider, provider_type_supports_quota_refresh,
reconcile_admin_fixed_provider_template_endpoints,
refresh_provider_oauth_account_state_after_update, refresh_provider_pool_quota_locally,
update_existing_provider_oauth_catalog_key, AdminAppState,
AdminGatewayProviderTransportSnapshot, AdminLocalOAuthRefreshError, AdminRequestContext,
AdminRouteRequest, AdminRouteResponse, AdminRouteResult, AdminStatsTimeRange,
AdminStatsUsageFilter, OAUTH_ACCOUNT_BLOCK_PREFIX, OAUTH_REQUEST_FAILED_PREFIX,
store_admin_provider_ops_balance_cache, update_existing_provider_oauth_catalog_key,
AdminAppState, AdminGatewayProviderTransportSnapshot, AdminLocalOAuthRefreshError,
AdminRequestContext, AdminRouteRequest, AdminRouteResponse, AdminRouteResult,
AdminStatsTimeRange, AdminStatsUsageFilter, OAUTH_ACCOUNT_BLOCK_PREFIX,
OAUTH_REQUEST_FAILED_PREFIX,
};
use crate::handlers::admin::{

10
apps/aether-gateway/src/control/route/admin/system_families.rs
View File

@@ -103,6 +103,16 @@ pub(super) fn classify_admin_system_family_route(
"admin:system",
false,
))
} else if method == http::Method::POST
&& normalized_path == "/api/admin/system/important-notification/test"
{
Some(classified(
"admin_proxy",
"system_manage",
"important_notification_test",
"admin:system",
false,
))
} else if method == http::Method::POST && normalized_path == "/api/admin/system/cleanup" {
Some(classified(
"admin_proxy",

4
apps/aether-gateway/src/control/tests/admin_core.rs
View File

@@ -183,6 +183,10 @@ fn classifies_admin_system_maintenance_write_routes_as_admin_proxy_route() {
("/api/admin/system/users/import", "users_import"),
("/api/admin/system/data/import", "data_import"),
("/api/admin/system/smtp/test", "smtp_test"),
(
"/api/admin/system/important-notification/test",
"important_notification_test",
),
("/api/admin/system/cleanup", "cleanup"),
("/api/admin/system/purge/config", "purge_config"),
("/api/admin/system/purge/users", "purge_users"),

339
apps/aether-gateway/src/email_delivery.rs Normal file
View File

@@ -0,0 +1,339 @@
use base64::Engine;
use crate::handlers::shared::{
decrypt_catalog_secret_with_fallbacks, system_config_bool, system_config_string,
};
use crate::{AppState, GatewayError};
const SMTP_TIMEOUT_SECS: u64 = 15;
#[derive(Debug, Clone)]
pub(crate) struct SmtpDeliveryConfig {
pub(crate) host: String,
pub(crate) port: u16,
pub(crate) user: Option<String>,
pub(crate) password: Option<String>,
pub(crate) use_tls: bool,
pub(crate) use_ssl: bool,
pub(crate) from_email: String,
pub(crate) from_name: String,
}
#[derive(Debug, Clone)]
pub(crate) struct ComposedEmail {
pub(crate) to_email: String,
pub(crate) subject: String,
pub(crate) html_body: String,
pub(crate) text_body: String,
}
pub(crate) async fn read_smtp_delivery_config(
state: &AppState,
) -> Result<Option<SmtpDeliveryConfig>, GatewayError> {
let smtp_host = state.read_system_config_json_value("smtp_host").await?;
let smtp_from_email = state
.read_system_config_json_value("smtp_from_email")
.await?;
let Some(host) = system_config_string(smtp_host.as_ref()) else {
return Ok(None);
};
let Some(from_email) = system_config_string(smtp_from_email.as_ref()) else {
return Ok(None);
};
let smtp_port = state.read_system_config_json_value("smtp_port").await?;
let smtp_user = state.read_system_config_json_value("smtp_user").await?;
let smtp_password = state.read_system_config_json_value("smtp_password").await?;
let smtp_use_tls = state.read_system_config_json_value("smtp_use_tls").await?;
let smtp_use_ssl = state.read_system_config_json_value("smtp_use_ssl").await?;
let smtp_from_name = state
.read_system_config_json_value("smtp_from_name")
.await?;
let password = system_config_string(smtp_password.as_ref()).map(|value| {
decrypt_catalog_secret_with_fallbacks(state.encryption_key(), &value).unwrap_or(value)
});
Ok(Some(SmtpDeliveryConfig {
host,
port: system_config_u16(smtp_port.as_ref(), 587),
user: system_config_string(smtp_user.as_ref()),
password,
use_tls: system_config_bool(smtp_use_tls.as_ref(), true),
use_ssl: system_config_bool(smtp_use_ssl.as_ref(), false),
from_email,
from_name: system_config_string(smtp_from_name.as_ref())
.unwrap_or_else(|| "Aether".to_string()),
}))
}
pub(crate) async fn send_smtp_email(
config: SmtpDeliveryConfig,
email: ComposedEmail,
) -> Result<(), GatewayError> {
tokio::task::spawn_blocking(move || send_smtp_email_blocking(config, email))
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?
}
fn system_config_u16(value: Option<&serde_json::Value>, default: u16) -> u16 {
match value {
Some(serde_json::Value::Number(value)) => value
.as_u64()
.and_then(|value| u16::try_from(value).ok())
.unwrap_or(default),
Some(serde_json::Value::String(value)) => value.trim().parse::<u16>().unwrap_or(default),
_ => default,
}
}
fn encode_mime_header(value: &str) -> String {
if value.is_ascii() {
return value.to_string();
}
format!(
"=?UTF-8?B?{}?=",
base64::engine::general_purpose::STANDARD.encode(value.as_bytes())
)
}
fn wrap_base64(value: &str) -> String {
let mut wrapped = String::new();
for chunk in value.as_bytes().chunks(76) {
wrapped.push_str(std::str::from_utf8(chunk).unwrap_or_default());
wrapped.push_str("\r\n");
}
wrapped
}
fn build_tls_config() -> std::sync::Arc<rustls::ClientConfig> {
let _ = rustls::crypto::ring::default_provider().install_default();
let root_store =
rustls::RootCertStore::from_iter(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
let config = rustls::ClientConfig::builder()
.with_root_certificates(root_store)
.with_no_client_auth();
std::sync::Arc::new(config)
}
fn resolve_server_name(host: &str) -> Result<rustls::pki_types::ServerName<'static>, GatewayError> {
let host = host.trim().trim_start_matches('[').trim_end_matches(']');
if let Ok(ip) = host.parse::<std::net::IpAddr>() {
return Ok(rustls::pki_types::ServerName::from(ip));
}
rustls::pki_types::ServerName::try_from(host.to_string())
.map_err(|err| GatewayError::Internal(err.to_string()))
}
fn connect_tcp_stream(config: &SmtpDeliveryConfig) -> Result<std::net::TcpStream, GatewayError> {
let stream = std::net::TcpStream::connect((config.host.as_str(), config.port))
.map_err(|err| GatewayError::Internal(err.to_string()))?;
stream
.set_read_timeout(Some(std::time::Duration::from_secs(SMTP_TIMEOUT_SECS)))
.map_err(|err| GatewayError::Internal(err.to_string()))?;
stream
.set_write_timeout(Some(std::time::Duration::from_secs(SMTP_TIMEOUT_SECS)))
.map_err(|err| GatewayError::Internal(err.to_string()))?;
Ok(stream)
}
fn wrap_tls_stream(
stream: std::net::TcpStream,
host: &str,
) -> Result<rustls::StreamOwned<rustls::ClientConnection, std::net::TcpStream>, GatewayError> {
let server_name = resolve_server_name(host)?;
let connection = rustls::ClientConnection::new(build_tls_config(), server_name)
.map_err(|err| GatewayError::Internal(err.to_string()))?;
Ok(rustls::StreamOwned::new(connection, stream))
}
fn smtp_read_response<T: std::io::BufRead>(reader: &mut T) -> Result<(u16, String), GatewayError> {
let mut message = String::new();
let code = loop {
let parsed_code;
let continuation;
let trimmed;
{
let mut line = String::new();
let bytes = reader
.read_line(&mut line)
.map_err(|err| GatewayError::Internal(err.to_string()))?;
if bytes == 0 {
return Err(GatewayError::Internal(
"smtp connection closed unexpectedly".to_string(),
));
}
trimmed = line.trim_end_matches(['\r', '\n']).to_string();
if trimmed.len() < 3 {
return Err(GatewayError::Internal("invalid smtp response".to_string()));
}
parsed_code = trimmed[..3]
.parse::<u16>()
.map_err(|err| GatewayError::Internal(err.to_string()))?;
continuation = trimmed.as_bytes().get(3).copied() == Some(b'-');
}
if !message.is_empty() {
message.push('\n');
}
message.push_str(&trimmed);
if !continuation {
break parsed_code;
}
};
Ok((code, message))
}
fn smtp_expect<T: std::io::BufRead>(
reader: &mut T,
allowed_codes: &[u16],
) -> Result<String, GatewayError> {
let (code, message) = smtp_read_response(reader)?;
if allowed_codes.contains(&code) {
return Ok(message);
}
Err(GatewayError::Internal(format!(
"unexpected smtp response {code}: {message}"
)))
}
fn smtp_write_line<T: std::io::Write>(writer: &mut T, line: &str) -> Result<(), GatewayError> {
writer
.write_all(line.as_bytes())
.map_err(|err| GatewayError::Internal(err.to_string()))?;
writer
.write_all(b"\r\n")
.map_err(|err| GatewayError::Internal(err.to_string()))?;
writer
.flush()
.map_err(|err| GatewayError::Internal(err.to_string()))
}
fn smtp_send_command<S: std::io::Read + std::io::Write>(
reader: &mut std::io::BufReader<S>,
command: &str,
allowed_codes: &[u16],
) -> Result<String, GatewayError> {
smtp_write_line(reader.get_mut(), command)?;
smtp_expect(reader, allowed_codes)
}
fn build_email_message(config: &SmtpDeliveryConfig, email: &ComposedEmail) -> String {
let boundary = format!("aether-{}", uuid::Uuid::new_v4().simple());
let text_body =
wrap_base64(&base64::engine::general_purpose::STANDARD.encode(email.text_body.as_bytes()));
let html_body =
wrap_base64(&base64::engine::general_purpose::STANDARD.encode(email.html_body.as_bytes()));
let from_header = if config.from_name.trim().is_empty() {
format!("<{}>", config.from_email)
} else {
format!(
"{} <{}>",
encode_mime_header(config.from_name.trim()),
config.from_email
)
};
format!(
"From: {from_header}\r\nTo: <{to_email}>\r\nSubject: {subject}\r\nMIME-Version: 1.0\r\nContent-Type: multipart/alternative; boundary=\"{boundary}\"\r\n\r\n--{boundary}\r\nContent-Type: text/plain; charset=\"utf-8\"\r\nContent-Transfer-Encoding: base64\r\n\r\n{text_body}--{boundary}\r\nContent-Type: text/html; charset=\"utf-8\"\r\nContent-Transfer-Encoding: base64\r\n\r\n{html_body}--{boundary}--\r\n",
to_email = email.to_email,
subject = encode_mime_header(&email.subject),
)
}
fn smtp_authenticate<S: std::io::Read + std::io::Write>(
reader: &mut std::io::BufReader<S>,
config: &SmtpDeliveryConfig,
) -> Result<(), GatewayError> {
let Some(username) = config
.user
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
else {
return Ok(());
};
let password = config.password.as_deref().unwrap_or("");
smtp_send_command(reader, "AUTH LOGIN", &[334])?;
smtp_send_command(
reader,
&base64::engine::general_purpose::STANDARD.encode(username.as_bytes()),
&[334],
)?;
smtp_send_command(
reader,
&base64::engine::general_purpose::STANDARD.encode(password.as_bytes()),
&[235],
)?;
Ok(())
}
fn smtp_deliver_message<S: std::io::Read + std::io::Write>(
reader: &mut std::io::BufReader<S>,
config: &SmtpDeliveryConfig,
email: &ComposedEmail,
) -> Result<(), GatewayError> {
smtp_send_command(
reader,
&format!("MAIL FROM:<{}>", config.from_email),
&[250],
)?;
smtp_send_command(
reader,
&format!("RCPT TO:<{}>", email.to_email),
&[250, 251],
)?;
smtp_send_command(reader, "DATA", &[354])?;
let message = build_email_message(config, email);
reader
.get_mut()
.write_all(message.as_bytes())
.map_err(|err| GatewayError::Internal(err.to_string()))?;
reader
.get_mut()
.write_all(b"\r\n.\r\n")
.map_err(|err| GatewayError::Internal(err.to_string()))?;
reader
.get_mut()
.flush()
.map_err(|err| GatewayError::Internal(err.to_string()))?;
let _ = smtp_expect(reader, &[250])?;
let _ = smtp_send_command(reader, "QUIT", &[221]);
Ok(())
}
fn smtp_send_message<S: std::io::Read + std::io::Write>(
reader: &mut std::io::BufReader<S>,
config: &SmtpDeliveryConfig,
email: &ComposedEmail,
) -> Result<(), GatewayError> {
smtp_send_command(reader, "EHLO aether.local", &[250])?;
smtp_authenticate(reader, config)?;
smtp_deliver_message(reader, config, email)
}
fn send_smtp_email_blocking(
config: SmtpDeliveryConfig,
email: ComposedEmail,
) -> Result<(), GatewayError> {
if config.use_ssl {
let stream = connect_tcp_stream(&config)?;
let tls_stream = wrap_tls_stream(stream, &config.host)?;
let mut reader = std::io::BufReader::new(tls_stream);
let _ = smtp_expect(&mut reader, &[220])?;
return smtp_send_message(&mut reader, &config, &email);
}
let stream = connect_tcp_stream(&config)?;
let mut reader = std::io::BufReader::new(stream);
let _ = smtp_expect(&mut reader, &[220])?;
let _ = smtp_send_command(&mut reader, "EHLO aether.local", &[250])?;
if config.use_tls {
let _ = smtp_send_command(&mut reader, "STARTTLS", &[220])?;
let stream = reader.into_inner();
let tls_stream = wrap_tls_stream(stream, &config.host)?;
let mut reader = std::io::BufReader::new(tls_stream);
return smtp_send_message(&mut reader, &config, &email);
}
smtp_authenticate(&mut reader, &config)?;
smtp_deliver_message(&mut reader, &config, &email)
}

1
apps/aether-gateway/src/handlers/admin/mod.rs
View File

@@ -36,6 +36,7 @@ pub(crate) use self::provider::oauth::runtime::{
refresh_provider_oauth_account_state_after_update,
};
pub(crate) use self::provider::ops::providers::actions::admin_provider_ops_local_action_response;
pub(crate) use self::provider::ops::providers::store_admin_provider_ops_balance_cache;
pub(crate) use self::provider::pool::config::admin_provider_pool_config;
pub(crate) use self::provider::pool_admin::maybe_build_local_admin_pool_response;
pub(crate) use self::provider::shared::payloads::{

2
apps/aether-gateway/src/handlers/admin/provider/ops/providers/balance_cache.rs
View File

@@ -70,7 +70,7 @@ pub(super) async fn read_admin_provider_ops_balance_cache(
}
}
pub(super) async fn store_admin_provider_ops_balance_cache(
pub(crate) async fn store_admin_provider_ops_balance_cache(
state: &AdminAppState<'_>,
provider_id: &str,
payload: &Value,

54
apps/aether-gateway/src/handlers/admin/provider/ops/providers/config.rs
View File

@@ -1,4 +1,7 @@
use super::support::{AdminProviderOpsSaveConfigRequest, ADMIN_PROVIDER_OPS_SENSITIVE_FIELDS};
use super::support::{
AdminProviderOpsQuotaAlertConfigRequest, 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;
@@ -8,6 +11,10 @@ use aether_data_contracts::repository::provider_catalog::{
use serde_json::json;
use std::time::{SystemTime, UNIX_EPOCH};
const PROVIDER_OPS_QUOTA_ALERT_DEFAULT_FETCH_INTERVAL_SECS: u64 = 30;
const PROVIDER_OPS_QUOTA_ALERT_MIN_FETCH_INTERVAL_SECS: u64 = 30;
const PROVIDER_OPS_QUOTA_ALERT_MAX_FETCH_INTERVAL_SECS: u64 = 86_400;
pub(super) fn admin_provider_ops_config_object(
provider: &StoredProviderCatalogProvider,
) -> Option<&serde_json::Map<String, serde_json::Value>> {
@@ -276,6 +283,7 @@ pub(super) fn build_admin_provider_ops_saved_config_value(
)
})
.collect::<serde_json::Map<String, serde_json::Value>>();
let quota_alert = normalize_admin_provider_ops_quota_alert(payload.quota_alert)?;
Ok(json!({
"architecture_id": payload.architecture_id,
@@ -287,9 +295,48 @@ pub(super) fn build_admin_provider_ops_saved_config_value(
},
"actions": actions,
"schedule": payload.schedule,
"quota_alert": quota_alert,
}))
}
fn normalize_admin_provider_ops_quota_alert(
request: Option<AdminProviderOpsQuotaAlertConfigRequest>,
) -> Result<serde_json::Value, String> {
let Some(request) = request else {
return Ok(default_admin_provider_ops_quota_alert());
};
let threshold_amount = request.threshold_amount.unwrap_or(0.0);
if threshold_amount < 0.0 {
return Err("quota_alert.threshold_amount 必须大于等于 0".to_string());
}
let fetch_interval_seconds = request
.fetch_interval_seconds
.unwrap_or(PROVIDER_OPS_QUOTA_ALERT_DEFAULT_FETCH_INTERVAL_SECS);
if !(PROVIDER_OPS_QUOTA_ALERT_MIN_FETCH_INTERVAL_SECS
..=PROVIDER_OPS_QUOTA_ALERT_MAX_FETCH_INTERVAL_SECS)
.contains(&fetch_interval_seconds)
{
return Err(format!(
"quota_alert.fetch_interval_seconds 必须在 {} 到 {} 秒之间",
PROVIDER_OPS_QUOTA_ALERT_MIN_FETCH_INTERVAL_SECS,
PROVIDER_OPS_QUOTA_ALERT_MAX_FETCH_INTERVAL_SECS
));
}
Ok(json!({
"enabled": request.enabled,
"threshold_amount": threshold_amount,
"fetch_interval_seconds": fetch_interval_seconds,
}))
}
fn default_admin_provider_ops_quota_alert() -> serde_json::Value {
json!({
"enabled": false,
"threshold_amount": 0.0,
"fetch_interval_seconds": PROVIDER_OPS_QUOTA_ALERT_DEFAULT_FETCH_INTERVAL_SECS,
})
}
pub(super) fn resolve_admin_provider_ops_base_url(
provider: &StoredProviderCatalogProvider,
endpoints: &[StoredProviderCatalogEndpoint],
@@ -356,5 +403,10 @@ pub(super) fn build_admin_provider_ops_config_payload(
connector.and_then(|connector| connector.get("credentials")),
),
},
"quota_alert": provider_ops_config
.get("quota_alert")
.filter(|value| value.is_object())
.cloned()
.unwrap_or_else(default_admin_provider_ops_quota_alert),
})
}

1
apps/aether-gateway/src/handlers/admin/provider/ops/providers/mod.rs
View File

@@ -4,4 +4,5 @@ mod config;
mod routes;
mod support;
mod verify;
pub(crate) use self::balance_cache::store_admin_provider_ops_balance_cache;
pub(super) use self::routes::maybe_build_local_admin_provider_ops_providers_response;

37
apps/aether-gateway/src/handlers/admin/provider/ops/providers/support.rs
View File

@@ -33,6 +33,8 @@ pub(super) struct AdminProviderOpsSaveConfigRequest {
pub(crate) actions: BTreeMap<String, AdminProviderOpsActionConfigRequest>,
#[serde(default)]
pub(crate) schedule: BTreeMap<String, String>,
#[serde(default)]
pub(crate) quota_alert: Option<AdminProviderOpsQuotaAlertConfigRequest>,
}
#[derive(Debug, Deserialize)]
@@ -52,6 +54,16 @@ pub(super) struct AdminProviderOpsActionConfigRequest {
pub(crate) config: serde_json::Map<String, serde_json::Value>,
}
#[derive(Debug, Deserialize)]
pub(super) struct AdminProviderOpsQuotaAlertConfigRequest {
#[serde(default)]
pub(crate) enabled: bool,
#[serde(default, deserialize_with = "deserialize_optional_f64_from_number")]
pub(crate) threshold_amount: Option<f64>,
#[serde(default)]
pub(crate) fetch_interval_seconds: Option<u64>,
}
#[derive(Debug, Deserialize)]
pub(super) struct AdminProviderOpsConnectRequest {
#[serde(default)]
@@ -71,3 +83,28 @@ fn default_admin_provider_ops_architecture_id() -> String {
fn default_admin_provider_ops_action_enabled() -> bool {
true
}
fn deserialize_optional_f64_from_number<'de, D>(deserializer: D) -> Result<Option<f64>, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = Option::<serde_json::Value>::deserialize(deserializer)?;
match value {
None | Some(serde_json::Value::Null) => Ok(None),
Some(serde_json::Value::Number(number)) => number
.as_f64()
.filter(|value| value.is_finite())
.map(Some)
.ok_or_else(|| serde::de::Error::custom("expected a finite number")),
Some(serde_json::Value::String(raw)) => raw
.trim()
.parse::<f64>()
.ok()
.filter(|value| value.is_finite())
.map(Some)
.ok_or_else(|| serde::de::Error::custom("expected a finite number or numeric string")),
Some(_) => Err(serde::de::Error::custom(
"expected a finite number or numeric string",
)),
}
}

8
apps/aether-gateway/src/handlers/admin/provider/summary/value.rs
View File

@@ -132,6 +132,13 @@ pub(crate) fn build_admin_provider_summary_value(
.and_then(|cfg| cfg.get("architecture_id"))
.and_then(serde_json::Value::as_str)
.map(ToOwned::to_owned);
let ops_quota_alert_enabled = provider_ops_config
.and_then(serde_json::Value::as_object)
.and_then(|cfg| cfg.get("quota_alert"))
.and_then(serde_json::Value::as_object)
.and_then(|cfg| cfg.get("enabled"))
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let billing_type = quota_snapshot
.map(|quota| quota.billing_type.clone())
.or_else(|| provider.billing_type.clone());
@@ -190,6 +197,7 @@ pub(crate) fn build_admin_provider_summary_value(
"endpoint_health_details": endpoint_health_details,
"ops_configured": ops_configured,
"ops_architecture_id": ops_architecture_id,
"ops_quota_alert_enabled": ops_quota_alert_enabled,
"created_at": endpoint_timestamp_or_now(provider.created_at_unix_ms, now_unix_secs),
"updated_at": endpoint_timestamp_or_now(provider.updated_at_unix_secs, now_unix_secs),
})

11
apps/aether-gateway/src/handlers/admin/system/core/system_routes.rs
View File

@@ -17,6 +17,7 @@ use crate::handlers::admin::system::shared::settings::{
build_admin_system_stats_payload, current_aether_version, fetch_latest_admin_system_release,
};
use crate::handlers::admin::system::shared::smtp::build_admin_smtp_test_payload;
use crate::important_notification::build_important_notification_test_payload;
use crate::maintenance::{ManualUsageCleanupMode, ManualUsageCleanupOptions};
use crate::GatewayError;
use aether_data_contracts::repository::usage::UsageCleanupTargets;
@@ -241,6 +242,16 @@ pub(super) async fn maybe_build_local_admin_core_system_response(
));
}
if decision.route_kind.as_deref() == Some("important_notification_test")
&& request_method == http::Method::POST
&& request_path == "/api/admin/system/important-notification/test"
{
return Ok(Some(
Json(build_important_notification_test_payload(state, request_body).await?)
.into_response(),
));
}
if decision.route_kind.as_deref() == Some("cleanup") && request_method == http::Method::POST {
return Ok(Some(attach_admin_audit_response(
Json(build_admin_system_cleanup_payload(state).await?).into_response(),

55
apps/aether-gateway/src/handlers/admin/system/shared/modules.rs
View File

@@ -1,5 +1,9 @@
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::shared::{module_available_from_env, system_config_bool};
use crate::important_notification::{
important_notification_configured, IMPORTANT_NOTIFICATION_ENABLED_KEY,
LEGACY_NOTIFICATION_EMAIL_ENABLED_KEY,
};
use crate::system_features::ENABLE_MODEL_DIRECTIVES_CONFIG_KEY;
use crate::GatewayError;
use aether_admin::system as admin_system_kernel;
@@ -68,14 +72,14 @@ pub(crate) const ADMIN_MODULE_DEFINITIONS: &[AdminModuleDefinition] = &[
admin_menu_order: 59,
},
AdminModuleDefinition {
name: "notification_email",
display_name: "异常通知",
description: "为 5xx 异常发送邮件通知，可在模块管理中启用或禁用",
name: "important_notification",
display_name: "重要通知",
description: "统一发送邮件和 Server 酱重要通知，供额度提醒等后台任务使用",
category: "integration",
env_key: "NOTIFICATION_EMAIL_AVAILABLE",
default_available: true,
admin_route: None,
admin_menu_icon: Some("Mail"),
admin_route: Some("/admin/modules/important-notification"),
admin_menu_icon: Some("BellRing"),
admin_menu_group: Some("system"),
admin_menu_order: 58,
},
@@ -126,10 +130,15 @@ pub(crate) struct AdminModuleRuntimeState {
oauth_providers: Vec<aether_data::repository::auth_modules::StoredOAuthProviderModuleConfig>,
ldap_config: Option<aether_data::repository::auth_modules::StoredLdapModuleConfig>,
gemini_files_has_capable_key: bool,
smtp_configured: bool,
important_notification_configured: bool,
}
pub(crate) fn admin_module_by_name(name: &str) -> Option<&'static AdminModuleDefinition> {
let name = if name == "notification_email" {
"important_notification"
} else {
name
};
ADMIN_MODULE_DEFINITIONS
.iter()
.find(|module| module.name == name)
@@ -146,6 +155,8 @@ pub(crate) fn admin_module_name_from_enabled_path(request_path: &str) -> Option<
pub(crate) fn admin_module_enabled_config_key(module: &AdminModuleDefinition) -> String {
if module.name == "model_directives" {
ENABLE_MODEL_DIRECTIVES_CONFIG_KEY.to_string()
} else if module.name == "important_notification" {
IMPORTANT_NOTIFICATION_ENABLED_KEY.to_string()
} else {
format!("module.{}.enabled", module.name)
}
@@ -197,28 +208,13 @@ pub(crate) async fn build_admin_module_runtime_state(
})
};
let smtp_host = state.read_system_config_json_value("smtp_host").await?;
let smtp_from_email = state
.read_system_config_json_value("smtp_from_email")
.await?;
let smtp_configured = smtp_host
.as_ref()
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.is_some()
&& smtp_from_email
.as_ref()
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.is_some();
let notification_configured = important_notification_configured(state.app()).await?;
Ok(AdminModuleRuntimeState {
oauth_providers,
ldap_config,
gemini_files_has_capable_key,
smtp_configured,
important_notification_configured: notification_configured,
})
}
@@ -231,7 +227,7 @@ pub(crate) fn build_admin_module_validation_result(
&runtime.oauth_providers,
runtime.ldap_config.as_ref(),
runtime.gemini_files_has_capable_key,
runtime.smtp_configured,
runtime.important_notification_configured,
)
}
@@ -252,10 +248,17 @@ pub(crate) async fn build_admin_module_status_payload(
) -> Result<serde_json::Value, GatewayError> {
let available = module_available_from_env(module.env_key, module.default_available);
let enabled = if available {
let enabled = state
let enabled_value = state
.read_system_config_json_value(&admin_module_enabled_config_key(module))
.await?;
system_config_bool(enabled.as_ref(), false)
let enabled_value = if module.name == "important_notification" && enabled_value.is_none() {
state
.read_system_config_json_value(LEGACY_NOTIFICATION_EMAIL_ENABLED_KEY)
.await?
} else {
enabled_value
};
system_config_bool(enabled_value.as_ref(), false)
} else {
false
};

5
apps/aether-gateway/src/handlers/shared/request_utils.rs
View File

@@ -254,6 +254,11 @@ pub(crate) fn admin_proxy_local_requires_buffered_body(
| (Some("system_manage"), http::Method::PUT, Some("config_set"))
| (Some("system_manage"), http::Method::PUT, Some("email_template_set"))
| (Some("system_manage"), http::Method::POST, Some("email_template_preview"))
| (
Some("system_manage"),
http::Method::POST,
Some("important_notification_test"),
)
| (
Some("provider_models_manage"),
http::Method::POST,

531
apps/aether-gateway/src/important_notification.rs Normal file
View File

@@ -0,0 +1,531 @@
use crate::admin_api::AdminAppState;
use crate::email_delivery::{
read_smtp_delivery_config, send_smtp_email, ComposedEmail, SmtpDeliveryConfig,
};
use crate::handlers::shared::{
decrypt_catalog_secret_with_fallbacks, system_config_bool, system_config_string,
};
use crate::{AppState, GatewayError};
use axum::body::Bytes;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tracing::warn;
pub(crate) const IMPORTANT_NOTIFICATION_ENABLED_KEY: &str = "module.important_notification.enabled";
pub(crate) const LEGACY_NOTIFICATION_EMAIL_ENABLED_KEY: &str = "module.notification_email.enabled";
pub(crate) const IMPORTANT_NOTIFICATION_EMAIL_ENABLED_KEY: &str =
"module.important_notification.email_enabled";
pub(crate) const IMPORTANT_NOTIFICATION_EMAIL_RECIPIENTS_KEY: &str =
"module.important_notification.email_recipients";
pub(crate) const IMPORTANT_NOTIFICATION_SERVER_CHAN_ENABLED_KEY: &str =
"module.important_notification.server_chan_enabled";
pub(crate) const IMPORTANT_NOTIFICATION_SERVER_CHAN_SEND_KEY_KEY: &str =
"module.important_notification.server_chan_send_key";
pub(crate) const IMPORTANT_NOTIFICATION_SERVER_CHAN_TEMPLATE_KEY: &str =
"module.important_notification.server_chan_template";
const SERVER_CHAN_API_BASE: &str = "https://sctapi.ftqq.com";
#[derive(Debug, Clone)]
pub(crate) struct ImportantNotification {
pub(crate) title: String,
pub(crate) markdown_body: String,
pub(crate) text_body: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ImportantNotificationChannelFilter {
All,
Email,
ServerChan,
}
#[derive(Debug, Clone)]
struct ImportantNotificationConfig {
module_enabled: bool,
email_enabled: bool,
email_recipients: Vec<String>,
server_chan_enabled: bool,
server_chan_send_key: Option<String>,
server_chan_template: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
pub(crate) struct ImportantNotificationChannelReport {
pub(crate) channel: &'static str,
pub(crate) success: bool,
pub(crate) message: String,
}
#[derive(Debug, Clone, Serialize)]
pub(crate) struct ImportantNotificationDeliveryReport {
pub(crate) success: bool,
pub(crate) channels: Vec<ImportantNotificationChannelReport>,
}
#[derive(Debug, Deserialize)]
struct ImportantNotificationTestRequest {
#[serde(default)]
channel: Option<String>,
}
pub(crate) async fn important_notification_module_enabled(
state: &AppState,
) -> Result<bool, GatewayError> {
let canonical = state
.read_system_config_json_value(IMPORTANT_NOTIFICATION_ENABLED_KEY)
.await?;
if canonical.is_some() {
return Ok(system_config_bool(canonical.as_ref(), false));
}
let legacy = state
.read_system_config_json_value(LEGACY_NOTIFICATION_EMAIL_ENABLED_KEY)
.await?;
Ok(system_config_bool(legacy.as_ref(), false))
}
pub(crate) async fn important_notification_configured(
state: &AppState,
) -> Result<bool, GatewayError> {
let config = read_important_notification_config(state).await?;
let smtp_config = read_smtp_delivery_config(state).await?;
Ok(
(config.email_enabled && !config.email_recipients.is_empty() && smtp_config.is_some())
|| (config.server_chan_enabled && config.server_chan_send_key.is_some()),
)
}
pub(crate) async fn send_important_notification(
state: &AppState,
notification: ImportantNotification,
) -> Result<ImportantNotificationDeliveryReport, GatewayError> {
send_important_notification_with_filter(
state,
notification,
ImportantNotificationChannelFilter::All,
)
.await
}
pub(crate) async fn send_important_notification_with_filter(
state: &AppState,
notification: ImportantNotification,
channel_filter: ImportantNotificationChannelFilter,
) -> Result<ImportantNotificationDeliveryReport, GatewayError> {
dispatch_important_notification(state, notification, channel_filter, false).await
}
async fn dispatch_important_notification(
state: &AppState,
notification: ImportantNotification,
channel_filter: ImportantNotificationChannelFilter,
bypass_enable_checks: bool,
) -> Result<ImportantNotificationDeliveryReport, GatewayError> {
let config = read_important_notification_config(state).await?;
if !bypass_enable_checks && !config.module_enabled {
return Ok(ImportantNotificationDeliveryReport {
success: false,
channels: vec![ImportantNotificationChannelReport {
channel: "module",
success: false,
message: "重要通知模块未启用".to_string(),
}],
});
}
let mut reports = Vec::new();
if matches!(
channel_filter,
ImportantNotificationChannelFilter::All | ImportantNotificationChannelFilter::Email
) {
maybe_send_email_notification(
state,
&config,
&notification,
bypass_enable_checks,
&mut reports,
)
.await;
}
if matches!(
channel_filter,
ImportantNotificationChannelFilter::All | ImportantNotificationChannelFilter::ServerChan
) {
maybe_send_server_chan_notification(
state,
&config,
&notification,
bypass_enable_checks,
&mut reports,
)
.await;
}
if reports.is_empty() {
reports.push(ImportantNotificationChannelReport {
channel: "none",
success: false,
message: "未启用可用的通知通道".to_string(),
});
}
let success = reports.iter().any(|report| report.success);
Ok(ImportantNotificationDeliveryReport {
success,
channels: reports,
})
}
pub(crate) async fn build_important_notification_test_payload(
state: &AdminAppState<'_>,
request_body: Option<&Bytes>,
) -> Result<Value, GatewayError> {
let request = match request_body.filter(|body| !body.is_empty()) {
Some(body) => serde_json::from_slice::<ImportantNotificationTestRequest>(body)
.unwrap_or(ImportantNotificationTestRequest { channel: None }),
None => ImportantNotificationTestRequest { channel: None },
};
let filter = match request
.channel
.as_deref()
.map(str::trim)
.unwrap_or("all")
.to_ascii_lowercase()
.as_str()
{
"email" => ImportantNotificationChannelFilter::Email,
"server_chan" | "serverchan" | "serve_chan" => {
ImportantNotificationChannelFilter::ServerChan
}
_ => ImportantNotificationChannelFilter::All,
};
let report = dispatch_important_notification(
state.app(),
ImportantNotification {
title: "Aether 重要通知测试".to_string(),
markdown_body: "这是一条来自 Aether 的重要通知测试。".to_string(),
text_body: "这是一条来自 Aether 的重要通知测试。".to_string(),
},
filter,
true,
)
.await?;
Ok(json!({
"success": report.success,
"message": if report.success { "测试通知已发送" } else { "测试通知发送失败" },
"channels": report.channels,
}))
}
async fn read_important_notification_config(
state: &AppState,
) -> Result<ImportantNotificationConfig, GatewayError> {
let module_enabled = important_notification_module_enabled(state).await?;
let email_enabled = state
.read_system_config_json_value(IMPORTANT_NOTIFICATION_EMAIL_ENABLED_KEY)
.await?;
let email_recipients = state
.read_system_config_json_value(IMPORTANT_NOTIFICATION_EMAIL_RECIPIENTS_KEY)
.await?;
let server_chan_enabled = state
.read_system_config_json_value(IMPORTANT_NOTIFICATION_SERVER_CHAN_ENABLED_KEY)
.await?;
let server_chan_send_key = state
.read_system_config_json_value(IMPORTANT_NOTIFICATION_SERVER_CHAN_SEND_KEY_KEY)
.await?;
let server_chan_template = state
.read_system_config_json_value(IMPORTANT_NOTIFICATION_SERVER_CHAN_TEMPLATE_KEY)
.await?;
Ok(ImportantNotificationConfig {
module_enabled,
email_enabled: system_config_bool(email_enabled.as_ref(), false),
email_recipients: parse_recipient_list(email_recipients.as_ref()),
server_chan_enabled: system_config_bool(server_chan_enabled.as_ref(), false),
server_chan_send_key: system_config_string(server_chan_send_key.as_ref()).map(|value| {
decrypt_catalog_secret_with_fallbacks(state.encryption_key(), &value).unwrap_or(value)
}),
server_chan_template: system_config_string(server_chan_template.as_ref()),
})
}
async fn maybe_send_email_notification(
state: &AppState,
config: &ImportantNotificationConfig,
notification: &ImportantNotification,
bypass_channel_toggle: bool,
reports: &mut Vec<ImportantNotificationChannelReport>,
) {
if !bypass_channel_toggle && !config.email_enabled {
return;
}
if config.email_recipients.is_empty() {
reports.push(ImportantNotificationChannelReport {
channel: "email",
success: false,
message: "未配置邮件收件人".to_string(),
});
return;
}
let smtp_config = match read_smtp_delivery_config(state).await {
Ok(Some(config)) => config,
Ok(None) => {
reports.push(ImportantNotificationChannelReport {
channel: "email",
success: false,
message: "SMTP 配置不完整".to_string(),
});
return;
}
Err(err) => {
reports.push(ImportantNotificationChannelReport {
channel: "email",
success: false,
message: format!("读取 SMTP 配置失败: {err:?}"),
});
return;
}
};
let mut sent = 0usize;
let mut failed = 0usize;
for recipient in &config.email_recipients {
match send_single_email_notification(smtp_config.clone(), recipient, notification).await {
Ok(()) => sent += 1,
Err(err) => {
failed += 1;
warn!(
error = ?err,
recipient = %recipient,
"failed to send important notification email"
);
}
}
}
reports.push(ImportantNotificationChannelReport {
channel: "email",
success: sent > 0,
message: if failed == 0 {
format!("邮件通知已发送给 {sent} 个收件人")
} else {
format!("邮件通知成功 {sent} 个，失败 {failed} 个")
},
});
}
async fn send_single_email_notification(
smtp_config: SmtpDeliveryConfig,
recipient: &str,
notification: &ImportantNotification,
) -> Result<(), GatewayError> {
send_smtp_email(
smtp_config,
ComposedEmail {
to_email: recipient.to_string(),
subject: notification.title.clone(),
html_body: build_notification_html(notification),
text_body: notification.text_body.clone(),
},
)
.await
}
async fn maybe_send_server_chan_notification(
state: &AppState,
config: &ImportantNotificationConfig,
notification: &ImportantNotification,
bypass_channel_toggle: bool,
reports: &mut Vec<ImportantNotificationChannelReport>,
) {
if !bypass_channel_toggle && !config.server_chan_enabled {
return;
}
let Some(send_key) = config.server_chan_send_key.as_deref() else {
reports.push(ImportantNotificationChannelReport {
channel: "server_chan",
success: false,
message: "未配置 Server 酱 SendKey".to_string(),
});
return;
};
match send_server_chan_notification(
state,
send_key,
config.server_chan_template.as_deref(),
notification,
)
.await
{
Ok(()) => reports.push(ImportantNotificationChannelReport {
channel: "server_chan",
success: true,
message: "Server 酱通知已发送".to_string(),
}),
Err(err) => {
warn!(error = ?err, "failed to send server chan important notification");
reports.push(ImportantNotificationChannelReport {
channel: "server_chan",
success: false,
message: format!("Server 酱通知发送失败: {err:?}"),
});
}
}
}
async fn send_server_chan_notification(
state: &AppState,
send_key: &str,
template: Option<&str>,
notification: &ImportantNotification,
) -> Result<(), GatewayError> {
let send_key = send_key.trim();
if send_key.is_empty() {
return Err(GatewayError::Internal(
"Server 酱 SendKey 不能为空".to_string(),
));
}
let desp = render_server_chan_desp(template, notification);
let url = format!("{SERVER_CHAN_API_BASE}/{send_key}.send");
let response = state
.client
.post(url)
.form(&[
("title", notification.title.as_str()),
("desp", desp.as_str()),
])
.send()
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
let status = response.status();
let text = response
.text()
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
if !status.is_success() {
return Err(GatewayError::Internal(format!(
"Server 酱返回 HTTP {status}: {text}"
)));
}
if let Ok(payload) = serde_json::from_str::<Value>(&text) {
let code_is_ok = payload
.get("code")
.and_then(|value| {
value
.as_i64()
.map(|code| code == 0)
.or_else(|| value.as_str().map(|code| code.trim() == "0"))
})
.unwrap_or(true);
if !code_is_ok {
return Err(GatewayError::Internal(format!(
"Server 酱返回失败: {payload}"
)));
}
}
Ok(())
}
fn render_server_chan_desp(template: Option<&str>, notification: &ImportantNotification) -> String {
match template {
Some(template) if !template.trim().is_empty() => template
.replace("{title}", &notification.title)
.replace("{body}", &notification.markdown_body),
_ => notification.markdown_body.clone(),
}
}
fn parse_recipient_list(value: Option<&Value>) -> Vec<String> {
let mut recipients = Vec::new();
match value {
Some(Value::Array(items)) => {
for item in items {
if let Some(raw) = item.as_str() {
push_recipient_parts(&mut recipients, raw);
}
}
}
Some(Value::String(raw)) => push_recipient_parts(&mut recipients, raw),
_ => {}
}
recipients.sort();
recipients.dedup();
recipients
}
fn push_recipient_parts(recipients: &mut Vec<String>, raw: &str) {
for item in raw
.split([',', ';', '\n', '\r'])
.map(str::trim)
.filter(|value| !value.is_empty())
{
recipients.push(item.to_string());
}
}
fn build_notification_html(notification: &ImportantNotification) -> String {
format!(
"<!doctype html><html><body><h2>{}</h2><pre style=\"font-family:ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;white-space:pre-wrap;line-height:1.6\">{}</pre></body></html>",
escape_html(&notification.title),
escape_html(&notification.text_body),
)
}
fn escape_html(value: &str) -> String {
value
.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
.replace('"', "&quot;")
.replace('\'', "&#39;")
}
#[cfg(test)]
mod tests {
use super::{parse_recipient_list, render_server_chan_desp, ImportantNotification};
use serde_json::json;
#[test]
fn parse_recipient_list_accepts_arrays_and_delimiters() {
assert_eq!(
parse_recipient_list(Some(&json!([
"ops@example.com, admin@example.com",
"ops@example.com"
]))),
vec![
"admin@example.com".to_string(),
"ops@example.com".to_string()
]
);
}
fn sample_notification() -> ImportantNotification {
ImportantNotification {
title: "告警".to_string(),
markdown_body: "原始正文".to_string(),
text_body: "原始正文".to_string(),
}
}
#[test]
fn server_chan_desp_uses_template_when_provided() {
let rendered = render_server_chan_desp(
Some("**{title}**\n\n{body}\n\n--end--"),
&sample_notification(),
);
assert_eq!(rendered, "**告警**\n\n原始正文\n\n--end--");
}
#[test]
fn server_chan_desp_falls_back_to_markdown_body_for_empty_template() {
assert_eq!(
render_server_chan_desp(None, &sample_notification()),
"原始正文"
);
assert_eq!(
render_server_chan_desp(Some(" "), &sample_notification()),
"原始正文"
);
}
}

2
apps/aether-gateway/src/lib.rs
View File

@@ -37,6 +37,7 @@ mod constants;
mod control;
mod data;
mod dispatch;
mod email_delivery;
mod error;
mod execution_runtime;
mod executor;
@@ -45,6 +46,7 @@ mod frontdoor_loop_guard;
mod handlers;
mod headers;
mod hooks;
mod important_notification;
mod log_ids;
mod maintenance;
pub(crate) mod middleware;

25
apps/aether-gateway/src/maintenance/mod.rs
View File

@@ -7,8 +7,8 @@ pub(crate) use runtime::{
ensure_provider_key_pool_scores_for_keys, inspect_proxy_upgrade_rollout,
list_admin_cleanup_run_records, perform_account_self_check_once,
perform_oauth_token_refresh_once, perform_pool_quota_probe_once, perform_provider_checkin_once,
pool_quota_probe_target_count, preview_manual_usage_cleanup, rebuild_admin_stats_once,
record_completed_cleanup_run, record_proxy_upgrade_traffic_success,
perform_provider_quota_alert_once, pool_quota_probe_target_count, preview_manual_usage_cleanup,
rebuild_admin_stats_once, record_completed_cleanup_run, record_proxy_upgrade_traffic_success,
restore_proxy_upgrade_rollout_skipped_nodes, retry_proxy_upgrade_rollout_node,
run_admin_system_cleanup_once, run_manual_usage_cleanup_once, skip_proxy_upgrade_rollout_node,
spawn_account_self_check_worker, spawn_audit_cleanup_worker, spawn_db_maintenance_worker,
@@ -16,16 +16,17 @@ pub(crate) use runtime::{
spawn_pending_cleanup_worker, spawn_pool_monitor_worker,
spawn_pool_quota_probe_replenish_for_request, spawn_pool_quota_probe_worker,
spawn_pool_score_rebuild_worker, spawn_provider_checkin_worker,
spawn_proxy_node_metrics_cleanup_worker, spawn_proxy_node_stale_cleanup_worker,
spawn_proxy_upgrade_rollout_worker, spawn_request_candidate_cleanup_worker,
spawn_stats_aggregation_worker, spawn_stats_hourly_aggregation_worker,
spawn_usage_cleanup_worker, spawn_wallet_daily_usage_aggregation_worker,
start_admin_request_body_cleanup_task, start_admin_system_purge_task,
start_manual_usage_cleanup_task, start_proxy_upgrade_rollout, AccountSelfCheckRunSummary,
AdminCleanupRunRecord, AdminCleanupTaskKind, AdminStatsRebuildSummary,
AdminSystemCleanupSummary, ManualUsageCleanupError, ManualUsageCleanupMode,
ManualUsageCleanupOptions, OAuthTokenRefreshRunSummary, PoolQuotaProbeRunSummary,
PoolQuotaProbeWorkerConfig, ProviderCheckinRunSummary, ProxyUpgradeRolloutCancelSummary,
spawn_provider_quota_alert_worker, spawn_proxy_node_metrics_cleanup_worker,
spawn_proxy_node_stale_cleanup_worker, spawn_proxy_upgrade_rollout_worker,
spawn_request_candidate_cleanup_worker, spawn_stats_aggregation_worker,
spawn_stats_hourly_aggregation_worker, spawn_usage_cleanup_worker,
spawn_wallet_daily_usage_aggregation_worker, start_admin_request_body_cleanup_task,
start_admin_system_purge_task, start_manual_usage_cleanup_task, start_proxy_upgrade_rollout,
AccountSelfCheckRunSummary, AdminCleanupRunRecord, AdminCleanupTaskKind,
AdminStatsRebuildSummary, AdminSystemCleanupSummary, ManualUsageCleanupError,
ManualUsageCleanupMode, ManualUsageCleanupOptions, OAuthTokenRefreshRunSummary,
PoolQuotaProbeRunSummary, PoolQuotaProbeWorkerConfig, ProviderCheckinRunSummary,
ProviderQuotaAlertRunSummary, ProxyUpgradeRolloutCancelSummary,
ProxyUpgradeRolloutConflictClearSummary, ProxyUpgradeRolloutNodeActionSummary,
ProxyUpgradeRolloutProbeConfig, ProxyUpgradeRolloutSkippedRestoreSummary,
ProxyUpgradeRolloutStatus, ProxyUpgradeRolloutTrackedNodeState,

7
apps/aether-gateway/src/maintenance/runtime.rs
View File

@@ -26,6 +26,8 @@ mod pool_quota_probe;
mod pool_score_rebuild;
#[path = "runtime/provider_checkin.rs"]
mod provider_checkin;
#[path = "runtime/provider_quota_alert.rs"]
mod provider_quota_alert;
#[path = "runtime/proxy_node_metrics_cleanup.rs"]
mod proxy_node_metrics_cleanup;
#[path = "runtime/proxy_node_staleness.rs"]
@@ -83,6 +85,9 @@ pub(crate) use pool_score_rebuild::{
PoolScoreRebuildRunSummary, PoolScoreRebuildWorkerConfig,
};
pub(crate) use provider_checkin::{perform_provider_checkin_once, ProviderCheckinRunSummary};
pub(crate) use provider_quota_alert::{
perform_provider_quota_alert_once, ProviderQuotaAlertRunSummary,
};
use proxy_node_metrics_cleanup::*;
use proxy_node_staleness::*;
use proxy_upgrade_rollout::*;
@@ -129,6 +134,8 @@ const PROXY_NODE_STALE_MISSED_HEARTBEATS: u64 = 3;
const POOL_MONITOR_INTERVAL: Duration = Duration::from_secs(5 * 60);
const OAUTH_TOKEN_REFRESH_INTERVAL: Duration = Duration::from_secs(60);
const PROVIDER_CHECKIN_CONCURRENCY: usize = 3;
const PROVIDER_QUOTA_ALERT_CONCURRENCY: usize = 3;
const PROVIDER_QUOTA_ALERT_INTERVAL: Duration = Duration::from_secs(5);
const PROVIDER_CHECKIN_DEFAULT_TIME: &str = "01:05";
const REQUEST_CANDIDATE_CLEANUP_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);
const STATS_DAILY_AGGREGATION_HOUR: u32 = 0;

453
apps/aether-gateway/src/maintenance/runtime/provider_quota_alert.rs Normal file
View File

@@ -0,0 +1,453 @@
use std::collections::HashMap;
use aether_data_contracts::repository::provider_catalog::{
StoredProviderCatalogEndpoint, StoredProviderCatalogProvider,
};
use futures_util::stream::{self, StreamExt};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tracing::warn;
use crate::admin_api::{
admin_provider_ops_local_action_response, store_admin_provider_ops_balance_cache, AdminAppState,
};
use crate::important_notification::{send_important_notification, ImportantNotification};
use crate::{AppState, GatewayError};
use super::PROVIDER_QUOTA_ALERT_CONCURRENCY;
const PROVIDER_QUOTA_ALERT_STATE_PREFIX: &str = "provider_ops:quota_alert:";
const PROVIDER_QUOTA_ALERT_DEFAULT_FETCH_INTERVAL_SECS: u64 = 30;
const PROVIDER_QUOTA_ALERT_MIN_FETCH_INTERVAL_SECS: u64 = 30;
const PROVIDER_QUOTA_ALERT_REPEAT_COOLDOWN_SECS: u64 = 24 * 60 * 60;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ProviderQuotaAlertRunSummary {
pub(crate) checked: usize,
pub(crate) alerted: usize,
pub(crate) skipped: usize,
pub(crate) failed: usize,
}
#[derive(Debug, Clone)]
struct ProviderQuotaAlertTarget {
provider: StoredProviderCatalogProvider,
config: ProviderQuotaAlertConfig,
}
#[derive(Debug, Clone, Copy)]
struct ProviderQuotaAlertConfig {
enabled: bool,
threshold_amount: f64,
fetch_interval_seconds: u64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
struct ProviderQuotaAlertRuntimeState {
#[serde(default)]
last_checked_at: u64,
#[serde(default)]
last_available: Option<f64>,
#[serde(default)]
below_threshold: bool,
#[serde(default)]
last_notified_at: Option<u64>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProviderQuotaAlertStatus {
Checked,
Alerted,
Skipped,
Failed,
}
pub(crate) async fn perform_provider_quota_alert_once(
state: &AppState,
) -> Result<ProviderQuotaAlertRunSummary, GatewayError> {
if !state.has_provider_catalog_data_reader() {
return Ok(ProviderQuotaAlertRunSummary {
checked: 0,
alerted: 0,
skipped: 0,
failed: 0,
});
}
let now_unix_secs = now_unix_secs();
let targets = select_provider_quota_alert_targets(state, now_unix_secs).await?;
if targets.is_empty() {
return Ok(ProviderQuotaAlertRunSummary {
checked: 0,
alerted: 0,
skipped: 0,
failed: 0,
});
}
let provider_ids = targets
.iter()
.map(|target| target.provider.id.clone())
.collect::<Vec<_>>();
let mut endpoints_by_provider = HashMap::<String, Vec<StoredProviderCatalogEndpoint>>::new();
for endpoint in state
.list_provider_catalog_endpoints_by_provider_ids(&provider_ids)
.await?
{
endpoints_by_provider
.entry(endpoint.provider_id.clone())
.or_default()
.push(endpoint);
}
let mut results = stream::iter(targets.into_iter().map(|target| {
let state = state.clone();
let provider_id = target.provider.id.clone();
let endpoints = endpoints_by_provider
.get(&provider_id)
.cloned()
.unwrap_or_default();
async move { run_provider_quota_alert_for_provider(&state, target, endpoints).await }
}))
.buffer_unordered(PROVIDER_QUOTA_ALERT_CONCURRENCY);
let mut summary = ProviderQuotaAlertRunSummary {
checked: 0,
alerted: 0,
skipped: 0,
failed: 0,
};
while let Some(status) = results.next().await {
match status {
ProviderQuotaAlertStatus::Checked => summary.checked += 1,
ProviderQuotaAlertStatus::Alerted => {
summary.checked += 1;
summary.alerted += 1;
}
ProviderQuotaAlertStatus::Skipped => summary.skipped += 1,
ProviderQuotaAlertStatus::Failed => summary.failed += 1,
}
}
Ok(summary)
}
async fn select_provider_quota_alert_targets(
state: &AppState,
now_unix_secs: u64,
) -> Result<Vec<ProviderQuotaAlertTarget>, GatewayError> {
let providers = state
.list_provider_catalog_providers(true)
.await?
.into_iter()
.filter_map(|provider| {
let config = provider_quota_alert_config(&provider)?;
(config.enabled).then_some(ProviderQuotaAlertTarget { provider, config })
})
.collect::<Vec<_>>();
let mut due = Vec::new();
for target in providers {
let runtime = read_quota_alert_runtime_state(state, &target.provider.id).await;
let last_checked_at = runtime
.as_ref()
.map(|state| state.last_checked_at)
.unwrap_or(0);
if now_unix_secs.saturating_sub(last_checked_at)
>= target
.config
.fetch_interval_seconds
.max(PROVIDER_QUOTA_ALERT_MIN_FETCH_INTERVAL_SECS)
{
due.push(target);
}
}
Ok(due)
}
async fn run_provider_quota_alert_for_provider(
state: &AppState,
target: ProviderQuotaAlertTarget,
endpoints: Vec<StoredProviderCatalogEndpoint>,
) -> ProviderQuotaAlertStatus {
let provider_id = target.provider.id.clone();
let admin_state = AdminAppState::new(state);
let payload = admin_provider_ops_local_action_response(
&admin_state,
&provider_id,
Some(&target.provider),
&endpoints,
"query_balance",
None,
)
.await;
store_admin_provider_ops_balance_cache(&admin_state, &provider_id, &payload).await;
let now_unix_secs = now_unix_secs();
let payload_status_success = payload.get("status").and_then(Value::as_str) == Some("success");
let Some(total_available) = extract_total_available(&payload) else {
warn!(
provider_id = %provider_id,
payload = %payload,
"provider quota alert skipped because balance payload has no total_available"
);
write_checked_runtime_state_without_balance(state, &provider_id, now_unix_secs).await;
return if payload_status_success {
ProviderQuotaAlertStatus::Skipped
} else {
ProviderQuotaAlertStatus::Failed
};
};
let previous = read_quota_alert_runtime_state(state, &provider_id).await;
let should_notify = provider_quota_alert_should_notify(
now_unix_secs,
total_available,
target.config.threshold_amount,
previous.as_ref(),
);
let mut next = ProviderQuotaAlertRuntimeState {
last_checked_at: now_unix_secs,
last_available: Some(total_available),
below_threshold: total_available <= target.config.threshold_amount,
last_notified_at: previous.and_then(|state| state.last_notified_at),
};
if should_notify {
let report = send_important_notification(
state,
build_provider_quota_alert_notification(
&target.provider,
total_available,
target.config.threshold_amount,
),
)
.await;
let delivered = match &report {
Ok(report) if report.success => true,
Ok(report) => {
warn!(
provider_id = %provider_id,
report = ?report,
"provider quota alert notification did not reach any channel"
);
false
}
Err(err) => {
warn!(
provider_id = %provider_id,
error = ?err,
"provider quota alert notification failed"
);
false
}
};
if delivered {
next.last_notified_at = Some(now_unix_secs);
write_quota_alert_runtime_state(state, &provider_id, &next).await;
return ProviderQuotaAlertStatus::Alerted;
}
write_quota_alert_runtime_state(state, &provider_id, &next).await;
return ProviderQuotaAlertStatus::Failed;
}
write_quota_alert_runtime_state(state, &provider_id, &next).await;
ProviderQuotaAlertStatus::Checked
}
fn provider_quota_alert_config(
provider: &StoredProviderCatalogProvider,
) -> Option<ProviderQuotaAlertConfig> {
let quota_alert = provider
.config
.as_ref()
.and_then(Value::as_object)
.and_then(|config| config.get("provider_ops"))
.and_then(Value::as_object)
.and_then(|provider_ops| provider_ops.get("quota_alert"))
.and_then(Value::as_object)?;
let enabled = quota_alert
.get("enabled")
.and_then(Value::as_bool)
.unwrap_or(false);
let threshold_amount = quota_alert
.get("threshold_amount")
.and_then(value_as_f64)
.filter(|value| value.is_finite() && *value >= 0.0)
.unwrap_or(0.0);
let fetch_interval_seconds = quota_alert
.get("fetch_interval_seconds")
.and_then(Value::as_u64)
.unwrap_or(PROVIDER_QUOTA_ALERT_DEFAULT_FETCH_INTERVAL_SECS)
.max(PROVIDER_QUOTA_ALERT_MIN_FETCH_INTERVAL_SECS);
Some(ProviderQuotaAlertConfig {
enabled,
threshold_amount,
fetch_interval_seconds,
})
}
fn provider_quota_alert_should_notify(
now_unix_secs: u64,
total_available: f64,
threshold_amount: f64,
previous: Option<&ProviderQuotaAlertRuntimeState>,
) -> bool {
if total_available > threshold_amount {
return false;
}
let Some(previous) = previous else {
return true;
};
if !previous.below_threshold {
return true;
}
previous
.last_notified_at
.map(|last| now_unix_secs.saturating_sub(last) >= PROVIDER_QUOTA_ALERT_REPEAT_COOLDOWN_SECS)
.unwrap_or(true)
}
fn extract_total_available(payload: &Value) -> Option<f64> {
if payload.get("status").and_then(Value::as_str) != Some("success") {
return None;
}
payload
.get("data")
.and_then(|data| data.get("total_available"))
.and_then(value_as_f64)
.filter(|value| value.is_finite())
}
fn value_as_f64(value: &Value) -> Option<f64> {
value.as_f64().or_else(|| {
value
.as_str()
.and_then(|raw| raw.trim().parse::<f64>().ok())
})
}
fn build_provider_quota_alert_notification(
provider: &StoredProviderCatalogProvider,
total_available: f64,
threshold_amount: f64,
) -> ImportantNotification {
let title = format!("提供商额度提醒：{}", provider.name);
let body = format!(
"提供商 `{}` 当前剩余额度为 `{:.4}`，已低于或等于提醒阈值 `{:.4}`。\n\nProvider ID: `{}`",
provider.name, total_available, threshold_amount, provider.id
);
let text_body = format!(
"提供商 {} 当前剩余额度为 {:.4}，已低于或等于提醒阈值 {:.4}。\n\nProvider ID: {}",
provider.name, total_available, threshold_amount, provider.id
);
ImportantNotification {
title,
markdown_body: body,
text_body,
}
}
async fn read_quota_alert_runtime_state(
state: &AppState,
provider_id: &str,
) -> Option<ProviderQuotaAlertRuntimeState> {
let key = provider_quota_alert_state_key(provider_id);
state
.runtime_kv_get(&key)
.await
.ok()
.flatten()
.and_then(|raw| serde_json::from_str::<ProviderQuotaAlertRuntimeState>(&raw).ok())
}
async fn write_checked_runtime_state_without_balance(
state: &AppState,
provider_id: &str,
now_unix_secs: u64,
) {
let mut next = read_quota_alert_runtime_state(state, provider_id)
.await
.unwrap_or_default();
next.last_checked_at = now_unix_secs;
write_quota_alert_runtime_state(state, provider_id, &next).await;
}
async fn write_quota_alert_runtime_state(
state: &AppState,
provider_id: &str,
runtime_state: &ProviderQuotaAlertRuntimeState,
) {
let Ok(serialized) = serde_json::to_string(runtime_state) else {
return;
};
if let Err(err) = state
.runtime_state()
.kv_set(
&provider_quota_alert_state_key(provider_id),
serialized,
None,
)
.await
{
warn!(
error = %err,
provider_id,
"failed to write provider quota alert runtime state"
);
}
}
fn provider_quota_alert_state_key(provider_id: &str) -> String {
format!("{PROVIDER_QUOTA_ALERT_STATE_PREFIX}{provider_id}")
}
fn now_unix_secs() -> u64 {
chrono::Utc::now().timestamp().max(0) as u64
}
#[cfg(test)]
mod tests {
use super::{
provider_quota_alert_should_notify, ProviderQuotaAlertRuntimeState,
PROVIDER_QUOTA_ALERT_REPEAT_COOLDOWN_SECS,
};
#[test]
fn quota_alert_notifies_on_first_drop_and_after_cooldown() {
assert!(provider_quota_alert_should_notify(100, 3.0, 5.0, None));
assert!(provider_quota_alert_should_notify(
100,
3.0,
5.0,
Some(&ProviderQuotaAlertRuntimeState {
below_threshold: false,
..ProviderQuotaAlertRuntimeState::default()
})
));
assert!(!provider_quota_alert_should_notify(
100,
3.0,
5.0,
Some(&ProviderQuotaAlertRuntimeState {
below_threshold: true,
last_notified_at: Some(90),
..ProviderQuotaAlertRuntimeState::default()
})
));
assert!(provider_quota_alert_should_notify(
100 + PROVIDER_QUOTA_ALERT_REPEAT_COOLDOWN_SECS,
3.0,
5.0,
Some(&ProviderQuotaAlertRuntimeState {
below_threshold: true,
last_notified_at: Some(100),
..ProviderQuotaAlertRuntimeState::default()
})
));
}
#[test]
fn quota_alert_does_not_notify_above_threshold() {
assert!(!provider_quota_alert_should_notify(100, 6.0, 5.0, None));
}
}

36
apps/aether-gateway/src/maintenance/runtime/workers.rs
View File

@@ -10,19 +10,19 @@ use super::{
duration_until_next_daily_run, duration_until_next_db_maintenance_run,
duration_until_next_stats_aggregation_run, duration_until_next_stats_hourly_aggregation_run,
maintenance_timezone, parse_hhmm_time, perform_oauth_token_refresh_once,
provider_checkin_schedule, run_audit_cleanup_once, run_db_maintenance_once,
run_gemini_file_mapping_cleanup_once, run_pending_cleanup_once, run_pool_monitor_once,
run_provider_checkin_once, run_proxy_node_metrics_cleanup_once,
perform_provider_quota_alert_once, provider_checkin_schedule, run_audit_cleanup_once,
run_db_maintenance_once, run_gemini_file_mapping_cleanup_once, run_pending_cleanup_once,
run_pool_monitor_once, run_provider_checkin_once, run_proxy_node_metrics_cleanup_once,
run_proxy_node_stale_cleanup_once, run_proxy_upgrade_rollout_once,
run_request_candidate_cleanup_once, run_stats_aggregation_once,
run_stats_hourly_aggregation_once, run_usage_cleanup_once,
run_wallet_daily_usage_aggregation_once, AUDIT_LOG_CLEANUP_INTERVAL,
GEMINI_FILE_MAPPING_CLEANUP_INTERVAL, OAUTH_TOKEN_REFRESH_INTERVAL, PENDING_CLEANUP_INTERVAL,
POOL_MONITOR_INTERVAL, PROVIDER_CHECKIN_DEFAULT_TIME, PROXY_NODE_METRICS_CLEANUP_HOUR,
PROXY_NODE_METRICS_CLEANUP_MINUTE, PROXY_NODE_STALE_SWEEP_INTERVAL,
PROXY_UPGRADE_ROLLOUT_INTERVAL, REQUEST_CANDIDATE_CLEANUP_INTERVAL, USAGE_CLEANUP_HOUR,
USAGE_CLEANUP_MINUTE, WALLET_DAILY_USAGE_AGGREGATION_HOUR,
WALLET_DAILY_USAGE_AGGREGATION_MINUTE,
POOL_MONITOR_INTERVAL, PROVIDER_CHECKIN_DEFAULT_TIME, PROVIDER_QUOTA_ALERT_INTERVAL,
PROXY_NODE_METRICS_CLEANUP_HOUR, PROXY_NODE_METRICS_CLEANUP_MINUTE,
PROXY_NODE_STALE_SWEEP_INTERVAL, PROXY_UPGRADE_ROLLOUT_INTERVAL,
REQUEST_CANDIDATE_CLEANUP_INTERVAL, USAGE_CLEANUP_HOUR, USAGE_CLEANUP_MINUTE,
WALLET_DAILY_USAGE_AGGREGATION_HOUR, WALLET_DAILY_USAGE_AGGREGATION_MINUTE,
};
const STATS_DAILY_CATCH_UP_BURST_LIMIT: usize = 14;
@@ -202,6 +202,26 @@ pub(crate) fn spawn_provider_checkin_worker(
}))
}
pub(crate) fn spawn_provider_quota_alert_worker(
state: AppState,
) -> Option<tokio::task::JoinHandle<()>> {
if !state.has_provider_catalog_data_reader() {
return None;
}
Some(tokio::spawn(async move {
let mut interval = tokio::time::interval(PROVIDER_QUOTA_ALERT_INTERVAL);
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
interval.tick().await;
loop {
interval.tick().await;
if let Err(err) = perform_provider_quota_alert_once(&state).await {
log_maintenance_worker_failure("provider_quota_alert", "tick", &err);
}
}
}))
}
pub(crate) fn spawn_oauth_token_refresh_worker(
state: AppState,
) -> Option<tokio::task::JoinHandle<()>> {

5
apps/aether-gateway/src/state/core.rs
View File

@@ -49,6 +49,7 @@ use crate::maintenance::spawn_pending_cleanup_worker;
use crate::maintenance::spawn_pool_monitor_worker;
use crate::maintenance::spawn_pool_score_rebuild_worker;
use crate::maintenance::spawn_provider_checkin_worker;
use crate::maintenance::spawn_provider_quota_alert_worker;
use crate::maintenance::spawn_proxy_node_metrics_cleanup_worker;
use crate::maintenance::spawn_proxy_node_stale_cleanup_worker;
use crate::maintenance::spawn_proxy_upgrade_rollout_worker;
@@ -1204,6 +1205,10 @@ impl AppState {
crate::task_runtime::TASK_KEY_PROVIDER_CHECKIN,
spawn_provider_checkin_worker(self.clone()),
);
supervise_worker(
crate::task_runtime::TASK_KEY_PROVIDER_QUOTA_ALERT,
spawn_provider_quota_alert_worker(self.clone()),
);
supervise_worker(
crate::task_runtime::TASK_KEY_OAUTH_TOKEN_REFRESH,
spawn_oauth_token_refresh_worker(self.clone()),

9
apps/aether-gateway/src/task_runtime/mod.rs
View File

@@ -35,6 +35,7 @@ pub(crate) const TASK_KEY_PROXY_NODE_METRICS_CLEANUP: &str =
"maintenance.proxy.node.metrics.cleanup";
pub(crate) const TASK_KEY_PROXY_UPGRADE_ROLLOUT: &str = "maintenance.proxy.upgrade.rollout";
pub(crate) const TASK_KEY_PROVIDER_CHECKIN: &str = "maintenance.provider.checkin";
pub(crate) const TASK_KEY_PROVIDER_QUOTA_ALERT: &str = "maintenance.provider.quota_alert";
pub(crate) const TASK_KEY_USAGE_CLEANUP: &str = "maintenance.usage.cleanup";
pub(crate) const TASK_KEY_WALLET_DAILY_USAGE_AGG: &str = "maintenance.wallet.daily.usage.agg";
pub(crate) const TASK_KEY_STATS_DAILY_AGG: &str = "maintenance.stats.daily.agg";
@@ -198,6 +199,14 @@ const TASK_DEFINITIONS: &[TaskDefinition] = &[
true,
RETRY_ONCE,
),
TaskDefinition::new(
TASK_KEY_PROVIDER_QUOTA_ALERT,
TaskKind::Scheduled,
"interval",
true,
true,
RETRY_ONCE,
),
TaskDefinition::new(
TASK_KEY_USAGE_CLEANUP,
TaskKind::Scheduled,

24
apps/aether-gateway/src/tests/control/admin/health_access.rs
View File

@@ -754,6 +754,14 @@ async fn gateway_handles_admin_modules_status_locally_with_trusted_admin_princip
.with_system_config_values_for_tests(vec![
("module.oauth.enabled".to_string(), json!(true)),
("module.management_tokens.enabled".to_string(), json!(true)),
(
"module.important_notification.email_enabled".to_string(),
json!(true),
),
(
"module.important_notification.email_recipients".to_string(),
json!("ops@example.com"),
),
("smtp_host".to_string(), json!("smtp.example.com")),
("smtp_from_email".to_string(), json!("ops@example.com")),
]);
@@ -796,9 +804,13 @@ async fn gateway_handles_admin_modules_status_locally_with_trusted_admin_princip
"/admin/modules/chat-pii-redaction"
);
assert_eq!(
payload["notification_email"]["config_validated"],
payload["important_notification"]["config_validated"],
json!(true)
);
assert_eq!(
payload["important_notification"]["admin_route"],
"/admin/modules/important-notification"
);
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
@@ -828,6 +840,14 @@ async fn gateway_handles_admin_modules_status_locally_with_bearer_admin_session(
.with_system_config_values_for_tests(vec![
("module.oauth.enabled".to_string(), json!(true)),
("module.management_tokens.enabled".to_string(), json!(true)),
(
"module.important_notification.email_enabled".to_string(),
json!(true),
),
(
"module.important_notification.email_recipients".to_string(),
json!("ops@example.com"),
),
("smtp_host".to_string(), json!("smtp.example.com")),
("smtp_from_email".to_string(), json!("ops@example.com")),
]);
@@ -855,7 +875,7 @@ async fn gateway_handles_admin_modules_status_locally_with_bearer_admin_session(
assert_eq!(payload["oauth"]["config_validated"], json!(true));
assert_eq!(payload["management_tokens"]["active"], json!(true));
assert_eq!(
payload["notification_email"]["config_validated"],
payload["important_notification"]["config_validated"],
json!(true)
);
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);