|
1 | | -use std::{borrow::Cow, collections::HashMap, sync::Arc}; |
| 1 | +use std::{collections::HashMap, sync::Arc, time::Duration}; |
2 | 2 |
|
3 | | -use http::{header::CONTENT_TYPE, HeaderName, HeaderValue, Uri}; |
4 | 3 | use serde::{Deserialize, Serialize}; |
5 | | -use spacegate_shell::{kernel::backend_service::http_client_service::HttpClient, SgBody, SgRequest}; |
6 | | -use tardis::{log as tracing, serde_json}; |
| 4 | +use tardis::{basic::dto::TardisContext, log as tracing, tokio}; |
7 | 5 |
|
8 | 6 | /// Context to call notification api |
9 | 7 | /// |
10 | | -/// Extract it from request extensions, and call [`NotificationContext::notify`] to send notification |
| 8 | +/// Extract it from request extensions, and call [`NotificationContext::report`] to send notification |
11 | 9 | #[derive(Debug, Clone)] |
12 | 10 | pub struct NotificationContext { |
13 | | - pub(crate) api: Arc<Uri>, |
14 | | - pub(crate) headers: Arc<HashMap<HeaderName, HeaderValue>>, |
15 | | - pub(crate) client: HttpClient, |
| 11 | + pub(crate) reach_api: Arc<str>, |
| 12 | + pub(crate) log_api: Arc<str>, |
| 13 | + pub(crate) spi_app_id: Arc<str>, |
| 14 | + pub(crate) audit_log_tag: Arc<str>, |
| 15 | + pub(crate) audit_log_token_header_name: Arc<str>, |
| 16 | + pub(crate) templates: Arc<HashMap<String, NotifyPluginConfigTemplatesItem>>, |
| 17 | + pub(crate) audit_param: Arc<AuditLogParam>, |
| 18 | + pub(crate) cache_client: RedisClient, |
| 19 | + pub(crate) dedup_cache_cool_down: Duration, |
16 | 20 | } |
17 | 21 |
|
18 | 22 | impl NotificationContext { |
19 | | - fn build_notification_request(&self, req: &ReachMsgSendReq) -> SgRequest { |
20 | | - let req_bytes = serde_json::to_vec(&req).expect("ReachMsgSendReq is a valid json"); |
21 | | - let body = SgBody::full(req_bytes); |
22 | | - let mut req = SgRequest::new(body); |
23 | | - req.headers_mut().insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); |
24 | | - *req.uri_mut() = self.api.as_ref().clone(); |
25 | | - for (k, v) in self.headers.iter() { |
26 | | - req.headers_mut().insert(k.clone(), v.clone()); |
| 23 | + pub fn submit_notify(&self, req: ReachRequest, dedup_hash: u64) { |
| 24 | + if self.reach_api.is_empty() { |
| 25 | + tracing::debug!("reach api is empty, skip sending notification"); |
| 26 | + return; |
27 | 27 | } |
28 | | - req |
| 28 | + let cache_client = self.cache_client.clone(); |
| 29 | + let ctx = TardisContext { |
| 30 | + ak: self.spi_app_id.to_string(), |
| 31 | + own_paths: self.spi_app_id.to_string(), |
| 32 | + ..Default::default() |
| 33 | + }; |
| 34 | + let cool_down = self.dedup_cache_cool_down.as_secs().min(1); |
| 35 | + tokio::spawn(async move { |
| 36 | + let key = format!("sg:plugin:{}:dedup:{}", NotifyPlugin::CODE, dedup_hash); |
| 37 | + let mut conn = cache_client.get_conn().await; |
| 38 | + // check if the key exists |
| 39 | + if let Ok(Some(_)) = conn.get::<'_, _, Option<String>>(&key).await { |
| 40 | + tracing::debug!("dedup cache hit, skip sending notification"); |
| 41 | + return; |
| 42 | + } |
| 43 | + |
| 44 | + // set the dedup key |
| 45 | + if let Err(e) = conn.set_ex::<'_, _, _,Option<String>>(&key, "1", cool_down).await { |
| 46 | + tracing::error!(error = ?e, "set dedup cache failed"); |
| 47 | + return; |
| 48 | + } |
| 49 | + |
| 50 | + let funs = NotifyPlugin::get_funs_inst_by_plugin_code(); |
| 51 | + tracing::debug!(?req, "submit notify"); |
| 52 | + let response = match req { |
| 53 | + ReachRequest::ByScene(req) => { |
| 54 | + bios_sdk_invoke::clients::reach_client::ReachClient::send_message(&req.into(), &funs, &ctx).await |
| 55 | + |
| 56 | + } |
| 57 | + ReachRequest::ByTemplate { contact, template_id, replace } => { |
| 58 | + bios_sdk_invoke::clients::reach_client::ReachClient::general_send(&contact, &template_id, &replace, &funs, &ctx).await |
| 59 | + } |
| 60 | + }; |
| 61 | + if let Err(e) = response { |
| 62 | + tracing::error!(error = ?e, "send notification failed"); |
| 63 | + } |
| 64 | + }); |
29 | 65 | } |
30 | | - pub async fn notify(&self, req: &ReachMsgSendReq) { |
31 | | - let notify_response = self.client.clone().request(self.build_notification_request(req)).await; |
32 | | - if !notify_response.status().is_success() { |
33 | | - tracing::warn!(response = ?notify_response, "send notification failed"); |
34 | | - } |
| 66 | + pub fn report<R: Report>(&self, response: &SgResponse, report: R) { |
| 67 | + let replace = report.get_replacement(); |
| 68 | + let key = report.key(); |
35 | 69 |
|
36 | | - let Ok(response) = notify_response.into_body().dump().await.inspect_err(|e| { |
37 | | - tracing::error!(error = ?e, "failed to read response body"); |
38 | | - }) else { |
39 | | - return; |
40 | | - }; |
41 | | - let response_str = String::from_utf8_lossy(response.get_dumped().expect("just dump body")); |
42 | | - tracing::debug!(response = ?response_str, "receive notification api response"); |
| 70 | + if let Some(template) = self.templates.get(key) { |
| 71 | + if let Some(notify_req) = template.reach.as_ref() { |
| 72 | + let mut req = notify_req.clone(); |
| 73 | + req.merge_replace(replace.clone()); |
| 74 | + let context = self.clone(); |
| 75 | + context.submit_notify(req, report.dedup_hash()); |
| 76 | + } |
| 77 | + if let Some(log_template) = template.audit_log.as_ref() { |
| 78 | + let formatted = format_template(log_template, &replace); |
| 79 | + self.submit_audit_log(response, Some(formatted)); |
| 80 | + } |
| 81 | + } |
| 82 | + } |
| 83 | + pub fn submit_audit_log(&self, response: &SgResponse, extra_info: Option<String>) { |
| 84 | + let mut log_param_content = self.audit_param.as_ref().clone().merge_audit_log_param_content(response, true, &self.audit_log_token_header_name); |
| 85 | + if let Some(extra_info) = extra_info { |
| 86 | + log_param_content.op = extra_info; |
| 87 | + } |
| 88 | + log_param_content.send_audit_log::<NotifyPlugin>(&self.spi_app_id, &self.log_api, &self.audit_log_tag); |
43 | 89 | } |
44 | 90 | } |
45 | 91 |
|
46 | | -#[derive(Debug, Serialize, Deserialize)] |
| 92 | +pub struct TamperReport {} |
| 93 | + |
| 94 | +pub struct UnauthorizedOperationReport {} |
| 95 | + |
| 96 | +pub struct CertLockReport {} |
| 97 | + |
| 98 | +#[derive(Debug, Clone)] |
| 99 | +pub struct ContentFilterForbiddenReport { |
| 100 | + pub(crate) forbidden_reason: String, |
| 101 | +} |
| 102 | + |
| 103 | +use spacegate_shell::{ |
| 104 | + ext_redis::{redis::AsyncCommands, RedisClient}, |
| 105 | + plugin::{ |
| 106 | + schemars::{self, JsonSchema}, |
| 107 | + Plugin, |
| 108 | + }, |
| 109 | + SgResponse, |
| 110 | +}; |
| 111 | + |
| 112 | +use crate::plugin::{ |
| 113 | + notify::{format_template, NotifyPlugin, NotifyPluginConfigTemplatesItem, ReachRequest, Report}, |
| 114 | + PluginBiosExt, |
| 115 | +}; |
| 116 | + |
| 117 | +use super::audit_log_param::AuditLogParam; |
| 118 | +#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone)] |
47 | 119 | pub struct ReachMsgSendReq { |
48 | 120 | pub scene_code: String, |
49 | 121 | pub receives: Vec<ReachMsgReceive>, |
50 | 122 | pub rel_item_id: String, |
51 | 123 | pub replace: HashMap<String, String>, |
52 | 124 | } |
53 | 125 |
|
54 | | -#[derive(Debug, Serialize, Deserialize)] |
| 126 | +impl ReachMsgSendReq { |
| 127 | + pub fn merge_replace<K: Into<String>>(&mut self, replace: impl IntoIterator<Item = (K, String)>) { |
| 128 | + self.replace.extend(replace.into_iter().map(|(k, v)| (k.into(), v))); |
| 129 | + } |
| 130 | +} |
| 131 | + |
| 132 | +impl From<ReachMsgSendReq> for bios_sdk_invoke::clients::reach_client::ReachMsgSendReq { |
| 133 | + fn from(val: ReachMsgSendReq) -> Self { |
| 134 | + bios_sdk_invoke::clients::reach_client::ReachMsgSendReq { |
| 135 | + scene_code: val.scene_code, |
| 136 | + receives: val.receives.into_iter().map(Into::into).collect(), |
| 137 | + rel_item_id: val.rel_item_id, |
| 138 | + replace: val.replace, |
| 139 | + } |
| 140 | + } |
| 141 | +} |
| 142 | +#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone)] |
55 | 143 | pub struct ReachMsgReceive { |
56 | 144 | pub receive_group_code: String, |
57 | 145 | pub receive_kind: String, |
58 | 146 | pub receive_ids: Vec<String>, |
59 | 147 | } |
| 148 | + |
| 149 | +impl From<ReachMsgReceive> for bios_sdk_invoke::clients::reach_client::ReachMsgReceive { |
| 150 | + fn from(val: ReachMsgReceive) -> Self { |
| 151 | + bios_sdk_invoke::clients::reach_client::ReachMsgReceive { |
| 152 | + receive_group_code: val.receive_group_code, |
| 153 | + receive_kind: val.receive_kind, |
| 154 | + receive_ids: val.receive_ids, |
| 155 | + } |
| 156 | + } |
| 157 | +} |
0 commit comments