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

feat: restructure notification services

Browse Source
This commit is contained in:
fawney19
2026-05-22 01:45:46 +08:00
parent d5e64d6ad9
commit 504c1ccb37
28 changed files with 1941 additions and 460 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/important_notification.rs
+422 -154
View File
@@ -2,8 +2,9 @@ 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::handlers::shared::{system_config_bool, system_config_string};
use crate::server_chan_push::{
read_server_chan_push_config, send_server_chan_push, ServerChanPushConfig,
};
use crate::{AppState, GatewayError};
use axum::body::Bytes;
@@ -17,14 +18,10 @@ 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";
pub(crate) const IMPORTANT_NOTIFICATION_DEFAULT_CHANNEL_KEY: &str =
"module.important_notification.default_channel";
pub(crate) const IMPORTANT_NOTIFICATION_ITEMS_KEY: &str = "module.important_notification.items";
pub(crate) const PROVIDER_QUOTA_ALERT_ITEM_KEY: &str = "provider_quota_alert";
#[derive(Debug, Clone)]
pub(crate) struct ImportantNotification {
@@ -45,9 +42,27 @@ 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>,
default_channel: ImportantNotificationChannelFilter,
items: Vec<ImportantNotificationItemConfig>,
server_chan: ServerChanPushConfig,
}
#[derive(Debug, Clone)]
struct ImportantNotificationItemConfig {
key: String,
name: String,
enabled: bool,
channel: Option<ImportantNotificationChannelFilter>,
title_template: Option<String>,
markdown_template: Option<String>,
text_template: Option<String>,
user_email_enabled: bool,
}
#[derive(Debug, Clone, Copy)]
struct NotificationChannelReadiness {
email: bool,
server_chan: bool,
}
#[derive(Debug, Clone, Serialize)]
@@ -67,6 +82,8 @@ pub(crate) struct ImportantNotificationDeliveryReport {
struct ImportantNotificationTestRequest {
#[serde(default)]
channel: Option<String>,
#[serde(default)]
item_key: Option<String>,
}
pub(crate) async fn important_notification_module_enabled(
@@ -91,25 +108,24 @@ pub(crate) async fn important_notification_configured(
important_notification_has_configured_channel(state, &config).await
}
pub(crate) async fn important_notification_dispatch_ready(
pub(crate) async fn important_notification_dispatch_ready_for_item(
state: &AppState,
item_key: &str,
) -> Result<bool, GatewayError> {
let config = read_important_notification_config(state).await?;
if !config.module_enabled {
return Ok(false);
}
important_notification_has_configured_channel(state, &config).await
}
async fn important_notification_has_configured_channel(
state: &AppState,
config: &ImportantNotificationConfig,
) -> Result<bool, GatewayError> {
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()),
)
if let Some(item) = find_notification_item(&config, item_key) {
if !item.enabled {
return Ok(false);
}
}
let readiness = read_notification_channel_readiness(state, &config).await?;
Ok(channel_filter_has_ready_channel(
notification_item_channel_filter(&config, item_key),
readiness,
))
}
pub(crate) async fn send_important_notification(
@@ -124,32 +140,129 @@ pub(crate) async fn send_important_notification(
.await
}
pub(crate) async fn send_important_notification_for_item(
state: &AppState,
item_key: &str,
notification: ImportantNotification,
variables: &[(&str, String)],
) -> Result<ImportantNotificationDeliveryReport, GatewayError> {
dispatch_important_notification(state, Some(item_key), notification, variables, None, false)
.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
dispatch_important_notification(state, None, notification, &[], Some(channel_filter), false)
.await
}
pub(crate) async fn send_user_important_notification_email(
state: &AppState,
item_key: &str,
user_email: &str,
notification: ImportantNotification,
variables: &[(&str, String)],
) -> Result<ImportantNotificationDeliveryReport, GatewayError> {
let config = read_important_notification_config(state).await?;
if !config.module_enabled {
return Ok(single_report("module", false, "通知服务未启用"));
}
let Some(item) = find_notification_item(&config, item_key) else {
return Ok(single_report("item", false, "通知项未定义"));
};
if !item.enabled {
return Ok(single_report("item", false, "通知项未启用"));
}
if !item.user_email_enabled {
return Ok(single_report("user_email", false, "通知项未启用用户邮件"));
}
let notification = apply_notification_item_template(Some(item), notification, variables);
let smtp_config = match read_smtp_delivery_config(state).await? {
Some(config) => config,
None => return Ok(single_report("user_email", false, "SMTP 配置不完整")),
};
let user_email = user_email.trim();
if user_email.is_empty() {
return Ok(single_report("user_email", false, "用户邮箱为空"));
}
match send_single_email_notification(smtp_config, user_email, &notification).await {
Ok(()) => Ok(single_report("user_email", true, "用户邮件通知已发送")),
Err(err) => {
warn!(error = ?err, user_email = %user_email, "failed to send user notification email");
Ok(single_report(
"user_email",
false,
format!("用户邮件通知发送失败: {err:?}"),
))
}
}
}
async fn important_notification_has_configured_channel(
state: &AppState,
config: &ImportantNotificationConfig,
) -> Result<bool, GatewayError> {
let readiness = read_notification_channel_readiness(state, config).await?;
if channel_filter_has_ready_channel(config.default_channel, readiness) {
return Ok(true);
}
Ok(config.items.iter().any(|item| {
item.enabled
&& channel_filter_has_ready_channel(
item.channel.unwrap_or(config.default_channel),
readiness,
)
}))
}
async fn read_notification_channel_readiness(
state: &AppState,
config: &ImportantNotificationConfig,
) -> Result<NotificationChannelReadiness, GatewayError> {
let smtp_config = read_smtp_delivery_config(state).await?;
Ok(NotificationChannelReadiness {
email: config.email_enabled && !config.email_recipients.is_empty() && smtp_config.is_some(),
server_chan: config.server_chan.enabled && config.server_chan.send_key.is_some(),
})
}
fn channel_filter_has_ready_channel(
filter: ImportantNotificationChannelFilter,
readiness: NotificationChannelReadiness,
) -> bool {
match filter {
ImportantNotificationChannelFilter::All => readiness.email || readiness.server_chan,
ImportantNotificationChannelFilter::Email => readiness.email,
ImportantNotificationChannelFilter::ServerChan => readiness.server_chan,
}
}
async fn dispatch_important_notification(
state: &AppState,
item_key: Option<&str>,
notification: ImportantNotification,
channel_filter: ImportantNotificationChannelFilter,
variables: &[(&str, String)],
channel_override: Option<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(),
}],
});
return Ok(single_report("module", false, "通知服务未启用"));
}
let item = item_key.and_then(|key| find_notification_item(&config, key));
if !bypass_enable_checks && item.is_some_and(|item| !item.enabled) {
return Ok(single_report("item", false, "通知项未启用"));
}
let notification = apply_notification_item_template(item, notification, variables);
let channel_filter = channel_override.unwrap_or_else(|| {
item.and_then(|item| item.channel)
.unwrap_or(config.default_channel)
});
let mut reports = Vec::new();
if matches!(
channel_filter,
@@ -198,31 +311,40 @@ pub(crate) async fn build_important_notification_test_payload(
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 },
Some(body) => serde_json::from_slice::<ImportantNotificationTestRequest>(body).unwrap_or(
ImportantNotificationTestRequest {
channel: None,
item_key: None,
},
),
None => ImportantNotificationTestRequest {
channel: None,
item_key: None,
},
};
let filter = match request
.channel
let filter = request.channel.as_deref().and_then(parse_channel_filter);
let item_key = request
.item_key
.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,
};
.filter(|value| !value.is_empty());
let variables = vec![
("provider_name", "示例 Provider".to_string()),
("provider_id", "provider-demo".to_string()),
("total_available", "8.0000".to_string()),
("threshold_amount", "10.0000".to_string()),
("user_email", "[email protected]".to_string()),
("balance", "1.0000".to_string()),
];
let report = dispatch_important_notification(
state.app(),
item_key,
ImportantNotification {
title: "Aether 重要通知测试".to_string(),
markdown_body: "这是一条来自 Aether 的重要通知测试。".to_string(),
text_body: "这是一条来自 Aether 的重要通知测试。".to_string(),
title: "Aether 通知服务测试".to_string(),
markdown_body: "这是一条来自 Aether 的通知服务测试。".to_string(),
text_body: "这是一条来自 Aether 的通知服务测试。".to_string(),
},
&variables,
filter,
true,
)
@@ -245,28 +367,195 @@ async fn read_important_notification_config(
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)
let default_channel = state
.read_system_config_json_value(IMPORTANT_NOTIFICATION_DEFAULT_CHANNEL_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)
let items = state
.read_system_config_json_value(IMPORTANT_NOTIFICATION_ITEMS_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()),
default_channel: default_channel
.as_ref()
.and_then(|value| value.as_str())
.and_then(parse_channel_filter)
.unwrap_or(ImportantNotificationChannelFilter::All),
items: parse_notification_items(items.as_ref()),
server_chan: read_server_chan_push_config(state).await?,
})
}
fn parse_channel_filter(raw: &str) -> Option<ImportantNotificationChannelFilter> {
match raw.trim().to_ascii_lowercase().as_str() {
"all" => Some(ImportantNotificationChannelFilter::All),
"email" => Some(ImportantNotificationChannelFilter::Email),
"server_chan" | "serverchan" | "serve_chan" => {
Some(ImportantNotificationChannelFilter::ServerChan)
}
"global" | "" => None,
_ => None,
}
}
fn parse_notification_items(value: Option<&Value>) -> Vec<ImportantNotificationItemConfig> {
let Some(Value::Array(items)) = value else {
return default_notification_items();
};
items
.iter()
.filter_map(parse_notification_item)
.collect::<Vec<_>>()
}
fn parse_notification_item(value: &Value) -> Option<ImportantNotificationItemConfig> {
let item = value.as_object()?;
let key = item.get("key")?.as_str()?.trim();
if key.is_empty() {
return None;
}
let name = item
.get("name")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or(key);
Some(ImportantNotificationItemConfig {
key: key.to_string(),
name: name.to_string(),
enabled: item.get("enabled").and_then(Value::as_bool).unwrap_or(true),
channel: item
.get("channel")
.and_then(Value::as_str)
.and_then(parse_channel_filter),
title_template: optional_non_empty_string(item.get("title_template")),
markdown_template: optional_non_empty_string(item.get("markdown_template")),
text_template: optional_non_empty_string(item.get("text_template")),
user_email_enabled: item
.get("user_email_enabled")
.and_then(Value::as_bool)
.unwrap_or(false),
})
}
fn default_notification_items() -> Vec<ImportantNotificationItemConfig> {
vec![
ImportantNotificationItemConfig {
key: PROVIDER_QUOTA_ALERT_ITEM_KEY.to_string(),
name: "号池额度不足".to_string(),
enabled: true,
channel: None,
title_template: None,
markdown_template: None,
text_template: None,
user_email_enabled: false,
},
ImportantNotificationItemConfig {
key: "provider_pool_abnormal".to_string(),
name: "号池异常".to_string(),
enabled: true,
channel: None,
title_template: Some("号池异常：{provider_name}".to_string()),
markdown_template: Some(
"号池 `{provider_name}` 出现异常，请检查服务状态。".to_string(),
),
text_template: Some("号池 {provider_name} 出现异常，请检查服务状态。".to_string()),
user_email_enabled: false,
},
ImportantNotificationItemConfig {
key: "user_balance_low".to_string(),
name: "用户余额不足".to_string(),
enabled: true,
channel: Some(ImportantNotificationChannelFilter::Email),
title_template: Some("余额不足提醒".to_string()),
markdown_template: Some("你的账户余额已低于提醒阈值，请及时处理。".to_string()),
text_template: Some("你的账户余额已低于提醒阈值，请及时处理。".to_string()),
user_email_enabled: true,
},
]
}
fn optional_non_empty_string(value: Option<&Value>) -> Option<String> {
value
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
}
fn find_notification_item<'a>(
config: &'a ImportantNotificationConfig,
item_key: &str,
) -> Option<&'a ImportantNotificationItemConfig> {
let item_key = item_key.trim();
config.items.iter().find(|item| item.key == item_key)
}
fn notification_item_channel_filter(
config: &ImportantNotificationConfig,
item_key: &str,
) -> ImportantNotificationChannelFilter {
find_notification_item(config, item_key)
.and_then(|item| item.channel)
.unwrap_or(config.default_channel)
}
fn apply_notification_item_template(
item: Option<&ImportantNotificationItemConfig>,
notification: ImportantNotification,
variables: &[(&str, String)],
) -> ImportantNotification {
let Some(item) = item else {
return notification;
};
let title = render_template(
item.title_template.as_deref(),
&notification.title,
&notification,
variables,
);
let markdown_body = render_template(
item.markdown_template.as_deref(),
&notification.markdown_body,
&notification,
variables,
);
let text_body = render_template(
item.text_template.as_deref(),
&notification.text_body,
&notification,
variables,
);
ImportantNotification {
title,
markdown_body,
text_body,
}
}
fn render_template(
template: Option<&str>,
fallback: &str,
notification: &ImportantNotification,
variables: &[(&str, String)],
) -> String {
let mut rendered = template
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or(fallback)
.to_string();
rendered = rendered
.replace("{title}", &notification.title)
.replace("{body}", &notification.markdown_body)
.replace("{text_body}", &notification.text_body);
for (key, value) in variables {
rendered = rendered.replace(&format!("{{{}}}", key.trim()), value);
}
rendered
}
async fn maybe_send_email_notification(
state: &AppState,
config: &ImportantNotificationConfig,
@@ -356,10 +645,10 @@ async fn maybe_send_server_chan_notification(
bypass_channel_toggle: bool,
reports: &mut Vec<ImportantNotificationChannelReport>,
) {
if !bypass_channel_toggle && !config.server_chan_enabled {
if !bypass_channel_toggle && !config.server_chan.enabled {
return;
}
let Some(send_key) = config.server_chan_send_key.as_deref() else {
if config.server_chan.send_key.is_none() {
reports.push(ImportantNotificationChannelReport {
channel: "server_chan",
success: false,
@@ -367,11 +656,11 @@ async fn maybe_send_server_chan_notification(
});
return;
};
match send_server_chan_notification(
match send_server_chan_push(
state,
send_key,
config.server_chan_template.as_deref(),
notification,
&config.server_chan,
&notification.title,
&notification.markdown_body,
)
.await
{
@@ -391,65 +680,18 @@ async fn maybe_send_server_chan_notification(
}
}
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 single_report(
channel: &'static str,
success: bool,
message: impl Into<String>,
) -> ImportantNotificationDeliveryReport {
ImportantNotificationDeliveryReport {
success,
channels: vec![ImportantNotificationChannelReport {
channel,
success,
message: message.into(),
}],
}
}
@@ -500,7 +742,10 @@ fn escape_html(value: &str) -> String {
#[cfg(test)]
mod tests {
use super::{parse_recipient_list, render_server_chan_desp, ImportantNotification};
use super::{
apply_notification_item_template, parse_notification_items, parse_recipient_list,
ImportantNotification, ImportantNotificationChannelFilter,
};
use serde_json::json;
#[test]
@@ -517,32 +762,55 @@ mod tests {
);
}
fn sample_notification() -> ImportantNotification {
ImportantNotification {
title: "告警".to_string(),
markdown_body: "原始正文".to_string(),
text_body: "原始正文".to_string(),
}
#[test]
fn parse_notification_items_reads_channel_and_user_email_flag() {
let items = parse_notification_items(Some(&json!([
{
"key": "user_balance_low",
"name": "用户余额不足",
"enabled": true,
"channel": "email",
"title_template": "余额提醒",
"markdown_template": "{user_email}: {balance}",
"user_email_enabled": true
}
])));
assert_eq!(items.len(), 1);
assert_eq!(items[0].key, "user_balance_low");
assert_eq!(
items[0].channel,
Some(ImportantNotificationChannelFilter::Email)
);
assert!(items[0].user_email_enabled);
}
#[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(),
fn item_template_renders_fallback_and_variables() {
let items = parse_notification_items(Some(&json!([
{
"key": "provider_quota_alert",
"name": "号池额度不足",
"title_template": "额度提醒：{provider_name}",
"markdown_template": "{body}\n剩余：{total_available}",
"text_template": "{text_body}\n剩余：{total_available}"
}
])));
let rendered = apply_notification_item_template(
Some(&items[0]),
ImportantNotification {
title: "默认标题".to_string(),
markdown_body: "默认正文".to_string(),
text_body: "默认文本".to_string(),
},
&[
("provider_name", "示例 Provider".to_string()),
("total_available", "8.0000".to_string()),
],
);
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()),
"原始正文"
);
assert_eq!(rendered.title, "额度提醒：示例 Provider");
assert_eq!(rendered.markdown_body, "默认正文\n剩余：8.0000");
assert_eq!(rendered.text_body, "默认文本\n剩余：8.0000");
}
}