1use anyhow::{Context, Result, anyhow};
2use base64::Engine;
3use chrono::{DateTime, Utc};
4use hmac::{Hmac, Mac, digest::KeyInit};
5use reqwest::{
6 StatusCode,
7 header::{AUTHORIZATION, CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue},
8};
9use serde::{Deserialize, Serialize};
10use serde_json::{Value, json};
11use sha2::Sha256;
12use std::collections::BTreeMap;
13
14type HmacSha256 = Hmac<Sha256>;
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(rename_all = "snake_case")]
18pub enum ChannelProvider {
19 Ntfy,
20 GenericWebhook,
21 Dingtalk,
22}
23
24#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
25pub struct NtfyChannelConfig {
26 pub base_url: String,
27 pub topic: String,
28}
29
30#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
31pub struct GenericWebhookChannelConfig {
32 pub url: String,
33 #[serde(default)]
34 pub extra_headers: BTreeMap<String, String>,
35}
36
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38pub struct DingtalkChannelConfig {
39 pub webhook_url: String,
40 #[serde(default)]
41 pub keywords: Vec<String>,
42}
43
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45#[serde(tag = "channel_type", rename_all = "snake_case")]
46pub enum ChannelConfig {
47 Ntfy(NtfyChannelConfig),
48 GenericWebhook(GenericWebhookChannelConfig),
49 Dingtalk(DingtalkChannelConfig),
50}
51
52#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
53pub struct NtfyChannelSecret {
54 pub auth_token: Option<String>,
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
58pub struct GenericWebhookChannelSecret {
59 pub bearer_token: Option<String>,
60}
61
62#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
63pub struct DingtalkChannelSecret {
64 pub secret: Option<String>,
65}
66
67#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
68#[serde(tag = "channel_type", rename_all = "snake_case")]
69pub enum ChannelSecret {
70 Ntfy(NtfyChannelSecret),
71 GenericWebhook(GenericWebhookChannelSecret),
72 Dingtalk(DingtalkChannelSecret),
73}
74
75#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
76pub struct NotificationChannel {
77 pub name: String,
78 pub provider: ChannelProvider,
79 pub config: ChannelConfig,
80 pub secret: Option<ChannelSecret>,
81 pub enabled: bool,
82}
83
84#[derive(Debug, Clone, Serialize, Deserialize)]
85pub struct NotificationRuleRef {
86 pub id: i64,
87 pub name: String,
88 pub mode: String,
89}
90
91#[derive(Debug, Clone, Serialize, Deserialize)]
92pub struct NotificationStockRef {
93 pub ts_code: String,
94 pub name: Option<String>,
95}
96
97#[derive(Debug, Clone, Serialize, Deserialize)]
98pub struct NotificationSignalRef {
99 pub message: String,
100 pub raw_message: String,
101 pub metric_value: Option<f64>,
102 pub threshold_value: Option<f64>,
103}
104
105#[derive(Debug, Clone, Serialize, Deserialize)]
106pub struct NotificationMessage {
107 pub event_id: i64,
108 pub triggered_at: DateTime<Utc>,
109 pub title: String,
110 pub rule: NotificationRuleRef,
111 pub stock: NotificationStockRef,
112 pub signal: NotificationSignalRef,
113}
114
115#[derive(Debug, Clone)]
116pub struct DeliveryResult {
117 pub delivery_status: String,
118 pub http_status: Option<i32>,
119}
120
121#[derive(Debug, Clone)]
122pub struct Notifier {
123 http_client: reqwest::Client,
124}
125
126impl Notifier {
127 pub fn new(http_client: reqwest::Client) -> Self {
128 Self { http_client }
129 }
130
131 pub async fn send(
132 &self,
133 channel: &NotificationChannel,
134 message: &NotificationMessage,
135 ) -> Result<DeliveryResult> {
136 match channel.provider {
137 ChannelProvider::Ntfy => self.send_ntfy(channel, message).await,
138 ChannelProvider::GenericWebhook => self.send_generic_webhook(channel, message).await,
139 ChannelProvider::Dingtalk => self.send_dingtalk(channel, message).await,
140 }
141 }
142
143 async fn send_ntfy(
144 &self,
145 channel: &NotificationChannel,
146 message: &NotificationMessage,
147 ) -> Result<DeliveryResult> {
148 let config = match &channel.config {
149 ChannelConfig::Ntfy(config) => config,
150 _ => return Err(anyhow!("channel config does not match ntfy type")),
151 };
152 let secret = match channel.secret.as_ref() {
153 Some(ChannelSecret::Ntfy(secret)) => secret.auth_token.as_deref(),
154 Some(_) => return Err(anyhow!("channel secret does not match ntfy type")),
155 None => None,
156 };
157
158 let mut request = self.http_client.post(format!(
159 "{}/{}",
160 config.base_url.trim_end_matches('/'),
161 config.topic
162 ));
163 request = request.header("Title", &message.title);
164 if let Some(token) = secret {
165 request = request.bearer_auth(token);
166 }
167 let response = request
168 .body(format_ntfy_body(message))
169 .send()
170 .await
171 .context("failed to send ntfy request")?;
172 ensure_success_status("ntfy", response.status())?;
173 Ok(DeliveryResult {
174 delivery_status: "delivered".to_string(),
175 http_status: Some(response.status().as_u16() as i32),
176 })
177 }
178
179 async fn send_generic_webhook(
180 &self,
181 channel: &NotificationChannel,
182 message: &NotificationMessage,
183 ) -> Result<DeliveryResult> {
184 let config = match &channel.config {
185 ChannelConfig::GenericWebhook(config) => config,
186 _ => {
187 return Err(anyhow!(
188 "channel config does not match generic_webhook type"
189 ));
190 }
191 };
192 let secret = match channel.secret.as_ref() {
193 Some(ChannelSecret::GenericWebhook(secret)) => secret.bearer_token.as_deref(),
194 Some(_) => {
195 return Err(anyhow!(
196 "channel secret does not match generic_webhook type"
197 ));
198 }
199 None => None,
200 };
201 validate_http_url(&config.url)?;
202
203 let mut headers = HeaderMap::new();
204 headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
205 if let Some(token) = secret {
206 let value = HeaderValue::from_str(&format!("Bearer {token}"))
207 .context("failed to encode webhook bearer token header")?;
208 headers.insert(AUTHORIZATION, value);
209 }
210 for (key, value) in &config.extra_headers {
211 if is_reserved_header(key) {
212 return Err(anyhow!(
213 "extra_headers cannot override reserved header `{key}`"
214 ));
215 }
216 let header_name = HeaderName::from_bytes(key.as_bytes())
217 .with_context(|| format!("invalid webhook header name `{key}`"))?;
218 let header_value = HeaderValue::from_str(value)
219 .with_context(|| format!("invalid webhook header value for `{key}`"))?;
220 headers.insert(header_name, header_value);
221 }
222
223 let response = self
224 .http_client
225 .post(&config.url)
226 .headers(headers)
227 .json(&generic_webhook_payload(message))
228 .send()
229 .await
230 .context("failed to send generic webhook request")?;
231 ensure_success_status("generic_webhook", response.status())?;
232 Ok(DeliveryResult {
233 delivery_status: "delivered".to_string(),
234 http_status: Some(response.status().as_u16() as i32),
235 })
236 }
237
238 async fn send_dingtalk(
239 &self,
240 channel: &NotificationChannel,
241 message: &NotificationMessage,
242 ) -> Result<DeliveryResult> {
243 let config = match &channel.config {
244 ChannelConfig::Dingtalk(config) => config,
245 _ => return Err(anyhow!("channel config does not match dingtalk type")),
246 };
247 let secret = match channel.secret.as_ref() {
248 Some(ChannelSecret::Dingtalk(secret)) => secret.secret.as_deref(),
249 Some(_) => return Err(anyhow!("channel secret does not match dingtalk type")),
250 None => None,
251 };
252 validate_http_url(&config.webhook_url)?;
253
254 let mut url = config.webhook_url.clone();
255 if let Some(secret) = secret {
256 let timestamp = Utc::now().timestamp_millis().to_string();
257 let sign = sign_dingtalk(×tamp, secret)?;
258 let separator = if url.contains('?') { "&" } else { "?" };
259 url.push_str(separator);
260 url.push_str(&format!("timestamp={timestamp}&sign={sign}"));
261 }
262
263 let response = self
264 .http_client
265 .post(url)
266 .json(&json!({
267 "msgtype": "text",
268 "text": {
269 "content": format_dingtalk_body(message, &config.keywords)?,
270 }
271 }))
272 .send()
273 .await
274 .context("failed to send dingtalk webhook request")?;
275 let status = response.status();
276 ensure_success_status("dingtalk", status)?;
277 let body: DingtalkResponse = response
278 .json()
279 .await
280 .context("failed to decode dingtalk webhook response")?;
281 if body.errcode != 0 {
282 return Err(anyhow!(
283 "dingtalk webhook rejected request with errcode {}: {}",
284 body.errcode,
285 body.errmsg
286 ));
287 }
288 Ok(DeliveryResult {
289 delivery_status: "delivered".to_string(),
290 http_status: Some(status.as_u16() as i32),
291 })
292 }
293}
294
295#[derive(Debug, Deserialize)]
296struct DingtalkResponse {
297 errcode: i64,
298 errmsg: String,
299}
300
301fn generic_webhook_payload(message: &NotificationMessage) -> Value {
302 json!({
303 "event_id": message.event_id,
304 "triggered_at": message.triggered_at.to_rfc3339(),
305 "title": message.title,
306 "rule": {
307 "id": message.rule.id,
308 "name": message.rule.name,
309 "mode": message.rule.mode,
310 },
311 "stock": {
312 "ts_code": message.stock.ts_code,
313 "name": message.stock.name,
314 },
315 "signal": {
316 "message": message.signal.message,
317 "raw_message": message.signal.raw_message,
318 "metric_value": message.signal.metric_value,
319 "threshold_value": message.signal.threshold_value,
320 }
321 })
322}
323
324fn format_ntfy_body(message: &NotificationMessage) -> String {
325 format!(
326 "{}\n{}\n{}",
327 message.stock.ts_code,
328 message.signal.message,
329 message.triggered_at.to_rfc3339()
330 )
331}
332
333fn format_dingtalk_body(message: &NotificationMessage, keywords: &[String]) -> Result<String> {
334 if keywords.len() > 10 {
335 return Err(anyhow!("dingtalk keywords cannot exceed 10"));
336 }
337 let stock_name = message.stock.name.as_deref().unwrap_or("-");
338 let keyword_prefix = keywords
339 .iter()
340 .map(|item| item.trim())
341 .filter(|item| !item.is_empty())
342 .collect::<Vec<_>>();
343 Ok(format!(
344 "{}股票: {} {}\n规则: {} ({})\n信号: {}\n当前值: {}\n阈值: {}\n触发时间: {}",
345 if keyword_prefix.is_empty() {
346 String::new()
347 } else {
348 format!("{} ", keyword_prefix.join(" "))
349 },
350 message.stock.ts_code,
351 stock_name,
352 message.rule.name,
353 message.rule.mode,
354 message.signal.message,
355 message
356 .signal
357 .metric_value
358 .map(|value| value.to_string())
359 .unwrap_or_else(|| "-".to_string()),
360 message
361 .signal
362 .threshold_value
363 .map(|value| value.to_string())
364 .unwrap_or_else(|| "-".to_string()),
365 message.triggered_at.to_rfc3339()
366 ))
367}
368
369fn sign_dingtalk(timestamp: &str, secret: &str) -> Result<String> {
370 let mut mac =
371 HmacSha256::new_from_slice(secret.as_bytes()).context("invalid dingtalk secret")?;
372 mac.update(format!("{timestamp}\n{secret}").as_bytes());
373 let encoded = base64::engine::general_purpose::STANDARD.encode(mac.finalize().into_bytes());
374 Ok(urlencoding::encode(&encoded).into_owned())
375}
376
377fn ensure_success_status(provider: &str, status: StatusCode) -> Result<()> {
378 if status.is_success() {
379 Ok(())
380 } else {
381 Err(anyhow!(
382 "{provider} request failed with status {}",
383 status.as_u16()
384 ))
385 }
386}
387
388fn validate_http_url(url: &str) -> Result<()> {
389 let parsed = reqwest::Url::parse(url).with_context(|| format!("invalid url `{url}`"))?;
390 match parsed.scheme() {
391 "http" | "https" => Ok(()),
392 other => Err(anyhow!("unsupported url scheme `{other}`")),
393 }
394}
395
396fn is_reserved_header(header: &str) -> bool {
397 header.eq_ignore_ascii_case("content-type") || header.eq_ignore_ascii_case("authorization")
398}
399
400#[cfg(test)]
401mod tests {
402 use chrono::TimeZone;
403
404 use super::{
405 ChannelConfig, ChannelProvider, DeliveryResult, DingtalkChannelConfig,
406 GenericWebhookChannelConfig, NotificationChannel, NotificationMessage, NotificationRuleRef,
407 NotificationSignalRef, NotificationStockRef, generic_webhook_payload, is_reserved_header,
408 sign_dingtalk,
409 };
410
411 fn sample_message() -> NotificationMessage {
412 NotificationMessage {
413 event_id: 1,
414 triggered_at: chrono::Utc
415 .with_ymd_and_hms(2026, 5, 7, 12, 34, 56)
416 .unwrap(),
417 title: "Stock alert 600519.SH".to_string(),
418 rule: NotificationRuleRef {
419 id: 2,
420 name: "price".to_string(),
421 mode: "simple".to_string(),
422 },
423 stock: NotificationStockRef {
424 ts_code: "600519.SH".to_string(),
425 name: Some("贵州茅台".to_string()),
426 },
427 signal: NotificationSignalRef {
428 message: "latest_price >= 1500".to_string(),
429 raw_message: "latest_price >= 1500".to_string(),
430 metric_value: Some(1501.0),
431 threshold_value: Some(1500.0),
432 },
433 }
434 }
435
436 #[test]
437 fn dingtalk_sign_is_non_empty() {
438 let sign = sign_dingtalk("1715000000000", "SECabc").unwrap();
439 assert!(!sign.is_empty());
440 }
441
442 #[test]
443 fn reserved_headers_are_detected_case_insensitively() {
444 assert!(is_reserved_header("Content-Type"));
445 assert!(is_reserved_header("authorization"));
446 assert!(!is_reserved_header("x-cloudiful-notifier"));
447 }
448
449 #[test]
450 fn generic_payload_has_expected_shape() {
451 let payload = generic_webhook_payload(&sample_message());
452 assert_eq!(payload["rule"]["mode"], "simple");
453 assert_eq!(payload["stock"]["ts_code"], "600519.SH");
454 assert_eq!(payload["signal"]["metric_value"], 1501.0);
455 }
456
457 #[test]
458 fn config_types_are_constructible() {
459 let _ = GenericWebhookChannelConfig {
460 url: "https://example.com".to_string(),
461 extra_headers: Default::default(),
462 };
463 let _ = DingtalkChannelConfig {
464 webhook_url: "https://oapi.dingtalk.com/robot/send?access_token=abc".to_string(),
465 keywords: vec!["监控报警".to_string()],
466 };
467 let _ = NotificationChannel {
468 name: "ops".to_string(),
469 provider: ChannelProvider::GenericWebhook,
470 config: ChannelConfig::GenericWebhook(GenericWebhookChannelConfig {
471 url: "https://example.com".to_string(),
472 extra_headers: Default::default(),
473 }),
474 secret: None,
475 enabled: true,
476 };
477 let _ = DeliveryResult {
478 delivery_status: "delivered".to_string(),
479 http_status: Some(200),
480 };
481 }
482}