refactor: remove unused dependencies and clean up code structure
ZhenYi committed
Deleted lib/auth/src/share/crypto.rs lib/auth/src/share/crypto.rs +0 −147
@@ -1,147 +0,0 @@
1use std::time::{SystemTime, UNIX_EPOCH};
2
3use aes_gcm::Aes256Gcm;
4use aes_gcm::aead::{Aead, AeadCore, KeyInit, OsRng};
5use argon2::{Argon2, PasswordHasher, PasswordVerifier};
6use base64::Engine;
7use base64::engine::general_purpose::{STANDARD, URL_SAFE_NO_PAD};
8use config::auth::PasswordPolicy;
9use rand::Rng;
10use sha2::{Digest, Sha256};
11
12use crate::errors::AuthError;
13
14pub fn hash_password(password: &str) -> Result<String, AuthError> {
15 Argon2::default()
16 .hash_password(password.as_bytes())
17 .map(|hash| hash.to_string())
18 .map_err(|error| AuthError::Internal(format!("password hashing failed: {error}")))
19}
20
21pub fn verify_password(password: &str, hash: &str) -> bool {
22 Argon2::default()
23 .verify_password(password.as_bytes(), hash)
24 .is_ok()
25}
26
27pub fn validate_password(password: &str, policy: &PasswordPolicy) -> Result<(), AuthError> {
28 if password.len() < policy.min_length {
29 return Err(AuthError::BadRequest(format!(
30 "password must be at least {} characters",
31 policy.min_length
32 )));
33 }
34 if policy.require_upper && !password.chars().any(|c| c.is_ascii_uppercase()) {
35 return Err(AuthError::BadRequest(
36 "password must contain an uppercase letter".to_string(),
37 ));
38 }
39 if policy.require_lower && !password.chars().any(|c| c.is_ascii_lowercase()) {
40 return Err(AuthError::BadRequest(
41 "password must contain a lowercase letter".to_string(),
42 ));
43 }
44 if policy.require_digit && !password.chars().any(|c| c.is_ascii_digit()) {
45 return Err(AuthError::BadRequest(
46 "password must contain a digit".to_string(),
47 ));
48 }
49 if policy.require_symbol && !password.chars().any(|c| !c.is_ascii_alphanumeric()) {
50 return Err(AuthError::BadRequest(
51 "password must contain a symbol".to_string(),
52 ));
53 }
54 Ok(())
55}
56
57pub fn generate_numeric_code(len: usize) -> String {
58 let mut rng = rand::thread_rng();
59 (0..len)
60 .map(|_| rng.gen_range(0..10u32).to_string())
61 .collect()
62}
63
64pub fn random_bytes(len: usize) -> Vec<u8> {
65 let mut bytes = vec![0u8; len];
66 rand::thread_rng().fill(&mut bytes[..]);
67 bytes
68}
69
70pub fn random_token(len: usize) -> String {
71 URL_SAFE_NO_PAD.encode(random_bytes(len))
72}
73
74pub fn sha256_b64url(data: &[u8]) -> String {
75 URL_SAFE_NO_PAD.encode(Sha256::digest(data))
76}
77
78pub fn sha256_hex(data: &[u8]) -> String {
79 Sha256::digest(data)
80 .iter()
81 .map(|byte| format!("{byte:02x}"))
82 .collect()
83}
84
85pub fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
86 if a.len() != b.len() {
87 return false;
88 }
89 let mut diff = 0u8;
90 for (x, y) in a.iter().zip(b.iter()) {
91 diff |= x ^ y;
92 }
93 diff == 0
94}
95
96pub fn aes_encrypt(key_material: &str, plaintext: &str) -> Result<String, AuthError> {
97 let key = aes_key(key_material);
98 let cipher = Aes256Gcm::new((&key).into());
99 let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
100 let ciphertext = cipher
101 .encrypt(&nonce, plaintext.as_bytes())
102 .map_err(|error| {
103 tracing::error!(%error, "aes encryption failed");
104 AuthError::Internal("aes encryption failed".to_string())
105 })?;
106 let mut output = Vec::with_capacity(nonce.len() + ciphertext.len());
107 output.extend_from_slice(&nonce);
108 output.extend_from_slice(&ciphertext);
109 Ok(STANDARD.encode(output))
110}
111
112pub fn aes_decrypt(key_material: &str, stored: &str) -> Result<String, AuthError> {
113 let bytes = STANDARD.decode(stored).map_err(|error| {
114 tracing::error!(%error, "malformed encrypted value");
115 AuthError::Internal("malformed encrypted value".to_string())
116 })?;
117 if bytes.len() < 12 {
118 return Err(AuthError::Internal("malformed encrypted value".to_string()));
119 }
120 let (nonce_bytes, ciphertext) = bytes.split_at(12);
121 let Ok(nonce_arr): Result<&[u8; 12], _> = nonce_bytes.try_into() else {
122 return Err(AuthError::Internal("malformed encrypted value".to_string()));
123 };
124 let key = aes_key(key_material);
125 let cipher = Aes256Gcm::new((&key).into());
126 let plaintext = cipher
127 .decrypt(nonce_arr.into(), ciphertext)
128 .map_err(|error| {
129 tracing::error!(%error, "aes decryption failed");
130 AuthError::Internal("aes decryption failed".to_string())
131 })?;
132 String::from_utf8(plaintext).map_err(|error| {
133 tracing::error!(%error, "decrypted value is not utf8");
134 AuthError::Internal("decrypted value is not utf8".to_string())
135 })
136}
137
138fn aes_key(material: &str) -> [u8; 32] {
139 Sha256::digest(material.as_bytes()).into()
140}
141
142pub fn now_secs() -> i64 {
143 SystemTime::now()
144 .duration_since(UNIX_EPOCH)
145 .map(|duration| duration.as_secs() as i64)
146 .unwrap_or(0)
147}
Modified lib/auth/src/share/mod.rs lib/auth/src/share/mod.rs +0 −1 View lib/auth/src/share/mod.rs
@@ -1,4 +1,3 @@
1pub mod crypto;
21pub mod guard;
32pub mod rpc;
43pub mod types;
Modified lib/auth/Cargo.toml lib/auth/Cargo.toml +0 −1 View lib/auth/Cargo.toml
@@ -26,7 +26,6 @@
2626hkdf = { workspace = true }
2727time = { workspace = true }
2828argon2 = { workspace = true }
29aes-gcm = { workspace = true }
3029rand = { workspace = true }
3130tracing = { workspace = true }
3231kenauth = { workspace = true }
Modified lib/config/src/identity/mod.rs lib/config/src/identity/mod.rs +2 −2 View lib/config/src/identity/mod.rs
@@ -6,9 +6,9 @@
66pub mod error;
77pub mod jwt;
88pub mod loader;
9pub mod pow;
9pub use crate::pow;
1010pub mod redis;
11pub mod rocketmq;
11pub use crate::rocketmq;
1212pub mod server;
1313pub mod service_auth;
1414pub mod session;
Deleted lib/config/src/identity/pow.rs lib/config/src/identity/pow.rs +0 −98
@@ -1,98 +0,0 @@
1use base64::Engine;
2use serde::Deserialize;
3
4#[derive(Clone, Deserialize)]
5#[serde(default)]
6pub struct PowConfig {
7 pub difficulty: u32,
8 pub max_difficulty: u32,
9 pub expires_secs: u64,
10 pub mac_key: Option<String>,
11}
12
13impl std::fmt::Debug for PowConfig {
14 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
15 formatter
16 .debug_struct("PowConfig")
17 .field("difficulty", &self.difficulty)
18 .field("max_difficulty", &self.max_difficulty)
19 .field("expires_secs", &self.expires_secs)
20 .field("mac_key", &self.mac_key.as_ref().map(|_| "<redacted>"))
21 .finish()
22 }
23}
24
25impl Default for PowConfig {
26 fn default() -> Self {
27 Self {
28 difficulty: 16,
29 max_difficulty: 24,
30 expires_secs: 60,
31 mac_key: None,
32 }
33 }
34}
35
36impl PowConfig {
37 pub fn new() -> Self {
38 Self::default()
39 }
40
41 pub fn with_difficulty(mut self, difficulty: u32) -> Self {
42 self.difficulty = difficulty;
43 self
44 }
45
46 pub fn with_max_difficulty(mut self, max_difficulty: u32) -> Self {
47 self.max_difficulty = max_difficulty;
48 self
49 }
50
51 pub fn with_expires_secs(mut self, expires_secs: u64) -> Self {
52 self.expires_secs = expires_secs;
53 self
54 }
55
56 pub fn with_mac_key(mut self, mac_key: impl Into<String>) -> Self {
57 self.mac_key = Some(mac_key.into());
58 self
59 }
60
61 pub fn validate(&self) -> Result<(), String> {
62 if self.difficulty == 0 {
63 return Err("pow.difficulty must be greater than 0".to_string());
64 }
65 if self.max_difficulty > 256 {
66 return Err("pow.max_difficulty must be <= 256".to_string());
67 }
68 if self.difficulty > self.max_difficulty {
69 return Err("pow.difficulty must be <= pow.max_difficulty".to_string());
70 }
71 if self.expires_secs == 0 {
72 return Err("pow.expires_secs must be greater than 0".to_string());
73 }
74 match &self.mac_key {
75 Some(key) => {
76 if key.trim().starts_with("REPLACE_WITH_") {
77 return Err(
78 "pow.mac_key is still set to the placeholder from config.yaml.example"
79 .to_string(),
80 );
81 }
82 let length = base64::engine::general_purpose::STANDARD
83 .decode(key)
84 .map(|bytes| bytes.len())
85 .unwrap_or(0);
86 if length < 32 {
87 return Err("pow.mac_key must be base64 of at least 32 bytes".to_string());
88 }
89 }
90 None => return Err("pow.mac_key is required".to_string()),
91 }
92 Ok(())
93 }
94}
95
96impl crate::identity::loader::Section for PowConfig {
97 const KEY: &'static str = "pow";
98}
Deleted lib/config/src/identity/rocketmq.rs lib/config/src/identity/rocketmq.rs +0 −121
@@ -1,121 +0,0 @@
1use serde::{Deserialize, Serialize};
2
3pub const DEFAULT_NAME_SERVER: &str = "127.0.0.1:9876";
4pub const DEFAULT_PRODUCER_GROUP: &str = "kenoauth";
5pub const DEFAULT_TOPIC_PREFIX: &str = "oauth";
6pub const DEFAULT_SEND_TIMEOUT_MS: u64 = 5_000;
7pub const DEFAULT_MAX_MESSAGE_BYTES: usize = 4 * 1024 * 1024;
8pub const MAX_TOPIC_LEN: usize = 127;
9
10fn default_name_server() -> String {
11 DEFAULT_NAME_SERVER.to_string()
12}
13
14fn default_producer_group() -> String {
15 DEFAULT_PRODUCER_GROUP.to_string()
16}
17
18fn default_topic_prefix() -> String {
19 DEFAULT_TOPIC_PREFIX.to_string()
20}
21
22fn default_send_timeout_ms() -> u64 {
23 DEFAULT_SEND_TIMEOUT_MS
24}
25
26fn default_max_message_bytes() -> usize {
27 DEFAULT_MAX_MESSAGE_BYTES
28}
29
30#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
31pub struct RocketMqConfig {
32 #[serde(default = "default_name_server")]
33 pub name_server: String,
34 #[serde(default = "default_producer_group")]
35 pub producer_group: String,
36 #[serde(default = "default_topic_prefix")]
37 pub topic_prefix: String,
38 #[serde(default = "default_send_timeout_ms")]
39 pub send_timeout_ms: u64,
40 #[serde(default = "default_max_message_bytes")]
41 pub max_message_bytes: usize,
42}
43
44impl Default for RocketMqConfig {
45 fn default() -> Self {
46 Self {
47 name_server: default_name_server(),
48 producer_group: default_producer_group(),
49 topic_prefix: default_topic_prefix(),
50 send_timeout_ms: default_send_timeout_ms(),
51 max_message_bytes: default_max_message_bytes(),
52 }
53 }
54}
55
56impl RocketMqConfig {
57 pub fn name_servers(&self) -> Vec<&str> {
58 self.name_server
59 .split([';', ','])
60 .map(str::trim)
61 .filter(|server| !server.is_empty())
62 .collect()
63 }
64
65 pub fn topic(&self, area: &str) -> String {
66 format!("{}-{area}", self.topic_prefix)
67 }
68
69 pub fn validate(&self) -> Result<(), String> {
70 let servers = self.name_servers();
71 if servers.is_empty() {
72 return Err("rocketmq.name_server must not be blank".to_string());
73 }
74 for server in servers {
75 let (host, port) = server.rsplit_once(':').ok_or_else(|| {
76 format!("rocketmq.name_server `{server}` must use the `host:port` form")
77 })?;
78 if host.trim().is_empty() {
79 return Err(format!("rocketmq.name_server `{server}` has an empty host"));
80 }
81 let port = port
82 .parse::<u16>()
83 .map_err(|_| format!("rocketmq.name_server `{server}` has an invalid port"))?;
84 if port == 0 {
85 return Err(format!("rocketmq.name_server `{server}` has a zero port"));
86 }
87 }
88
89 if !is_valid_identifier(&self.producer_group) {
90 return Err("rocketmq.producer_group must use letters, digits, `_` or `-`".to_string());
91 }
92 if !is_valid_identifier(&self.topic_prefix) {
93 return Err("rocketmq.topic_prefix must use letters, digits, `_` or `-`".to_string());
94 }
95 if self.topic_prefix.len() + 1 + 64 > MAX_TOPIC_LEN {
96 return Err(format!(
97 "rocketmq.topic_prefix leaves no room for a topic under {MAX_TOPIC_LEN} characters"
98 ));
99 }
100 if self.send_timeout_ms == 0 {
101 return Err("rocketmq.send_timeout_ms must be greater than zero".to_string());
102 }
103 if self.max_message_bytes == 0 {
104 return Err("rocketmq.max_message_bytes must be greater than zero".to_string());
105 }
106
107 Ok(())
108 }
109}
110
111fn is_valid_identifier(value: &str) -> bool {
112 !value.is_empty()
113 && value.len() <= MAX_TOPIC_LEN
114 && value
115 .chars()
116 .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-'))
117}
118
119impl crate::identity::loader::Section for RocketMqConfig {
120 const KEY: &'static str = "rocketmq";
121}
Modified lib/config/src/pow.rs lib/config/src/pow.rs +26 −1 View lib/config/src/pow.rs
@@ -1,7 +1,7 @@
11use base64::Engine;
22use serde::Deserialize;
33
4#[derive(Debug, Clone, Deserialize)]
4#[derive(Clone, Deserialize)]
55#[serde(default)]
66pub struct PowConfig {
77 pub difficulty: u32,
@@ -10,6 +10,18 @@
1010 pub mac_key: Option<String>,
1111}
1212
13impl std::fmt::Debug for PowConfig {
14 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
15 formatter
16 .debug_struct("PowConfig")
17 .field("difficulty", &self.difficulty)
18 .field("max_difficulty", &self.max_difficulty)
19 .field("expires_secs", &self.expires_secs)
20 .field("mac_key", &self.mac_key.as_ref().map(|_| "<redacted>"))
21 .finish()
22 }
23}
24
1325impl Default for PowConfig {
1426 fn default() -> Self {
1527 Self {
@@ -47,6 +59,9 @@
4759 }
4860
4961 pub fn validate(&self) -> Result<(), String> {
62 if self.difficulty == 0 {
63 return Err("pow.difficulty must be greater than 0".to_string());
64 }
5065 if self.max_difficulty > 256 {
5166 return Err("pow.max_difficulty must be <= 256".to_string());
5267 }
@@ -58,6 +73,12 @@
5873 }
5974 match &self.mac_key {
6075 Some(key) => {
76 if key.trim().starts_with("REPLACE_WITH_") {
77 return Err(
78 "pow.mac_key is still set to the placeholder from config.yaml.example"
79 .to_string(),
80 );
81 }
6182 let length = base64::engine::general_purpose::STANDARD
6283 .decode(key)
6384 .map(|bytes| bytes.len())
@@ -71,3 +92,7 @@
7192 Ok(())
7293 }
7394}
95
96impl crate::identity::loader::Section for PowConfig {
97 const KEY: &'static str = "pow";
98}
Modified lib/config/src/rocketmq.rs lib/config/src/rocketmq.rs +10 −2 View lib/config/src/rocketmq.rs
@@ -1,4 +1,4 @@
1use serde::Deserialize;
1use serde::{Deserialize, Serialize};
22
33const DEFAULT_NAME_SERVER: &str = "127.0.0.1:9876";
44const DEFAULT_PRODUCER_GROUP: &str = "gitdata";
@@ -27,7 +27,7 @@
2727 DEFAULT_MAX_MESSAGE_BYTES
2828}
2929
30#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
30#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
3131#[serde(default)]
3232pub struct RocketMqConfig {
3333 #[serde(default = "default_name_server")]
@@ -67,6 +67,10 @@
6767 format!("{}-{area}", self.topic_prefix)
6868 }
6969
70 pub fn validate(&self) -> Result<(), String> {
71 self.is_valid()
72 }
73
7074 pub fn is_valid(&self) -> Result<(), String> {
7175 let servers = self.name_servers();
7276 if servers.is_empty() {
@@ -107,6 +111,10 @@
107111 }
108112}
109113
114impl crate::identity::loader::Section for RocketMqConfig {
115 const KEY: &'static str = "rocketmq";
116}
117
110118fn is_valid_identifier(value: &str) -> bool {
111119 !value.is_empty()
112120 && value.len() <= MAX_TOPIC_LEN
Modified lib/email/src/error.rs lib/email/src/error.rs +1 −1 View lib/email/src/error.rs
@@ -1,6 +1,6 @@
11use std::fmt;
22
3use ::queue::identity::QueueError;
3use ::queue::QueueError;
44use lettre::address::AddressError;
55use lettre::error::Error as LettreError;
66use lettre::transport::smtp::Error as SmtpError;
Modified lib/events/src/event/mod.rs lib/events/src/event/mod.rs +2 −2 View lib/events/src/event/mod.rs
@@ -31,8 +31,8 @@
3131pub use user::UserEvent;
3232pub use webhook::WebhookEvent;
3333
34pub use queue::identity::DeliveryClass;
35pub use queue::identity::envelope::EVENT_PREFIX;
34pub use queue::DeliveryClass;
35pub use queue::envelope::EVENT_PREFIX;
3636
3737mod runtime;
3838
Modified lib/events/src/handler.rs lib/events/src/handler.rs +1 −1 View lib/events/src/handler.rs
@@ -3,7 +3,7 @@
33
44use crate::event::{EVENT_PREFIX, RuntimeEvent};
55use config::identity::rocketmq::RocketMqConfig;
6use queue::identity::{Envelope, Labels, Publisher, QueueError, QueueResult, RocketMqConn};
6use queue::{Envelope, Labels, Publisher, QueueError, QueueResult, RocketMqConn};
77
88pub const DURABLE_DOMAINS: &[&str] = &[
99 "auth",
Modified lib/events/src/lib.rs lib/events/src/lib.rs +1 −1 View lib/events/src/lib.rs
@@ -7,4 +7,4 @@
77
88pub use event::*;
99pub use handler::{EventHandler, retry_publish};
10pub use queue::identity::{Envelope, Labels};
10pub use queue::{Envelope, Labels};
Modified lib/events/src/outbox.rs lib/events/src/outbox.rs +10 −10 View lib/events/src/outbox.rs
@@ -6,8 +6,7 @@
66use models::event_outbox;
77use opentelemetry::metrics::Gauge;
88use opentelemetry::{KeyValue, global};
9use queue::identity::{DeliveryClass, Envelope, SCHEMA_VERSION};
10use queue::{Envelope as AppEnvelope, Publisher as AppPublisher};
9use queue::{DeliveryClass, Envelope, Publisher, SCHEMA_VERSION};
1110use sea_orm::ActiveValue::Set;
1211use sea_orm::{
1312 ColumnTrait, ConnectionTrait, DatabaseConnection, DatabaseTransaction, DbBackend, DbErr,
@@ -26,7 +25,7 @@
2625const PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);
2726const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
2827const POLL_INTERVAL: Duration = Duration::from_millis(500);
29const MAX_ENVELOPE_BYTES: usize = queue::identity::DEFAULT_MAX_MESSAGE_BYTES - 4096;
28const MAX_ENVELOPE_BYTES: usize = queue::DEFAULT_MAX_MESSAGE_BYTES - 4096;
3029const BACKLOG_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);
3130/// 单条事件的重试上限。超过后丢弃并打 ERROR：否则一条毒消息会永久堵住它所在的
3231/// ordering_key（`auth:{user_id}` 之类），后面的事件永远发不出去。
@@ -283,9 +282,9 @@
283282 .map_err(|error| error.to_string())
284283}
285284
286async fn publish_app(publisher: Option<&AppPublisher>, row: &PendingEvent) -> Result<(), String> {
285async fn publish_app(publisher: Option<&Publisher>, row: &PendingEvent) -> Result<(), String> {
287286 let publisher = publisher.ok_or_else(|| "app event publisher is not configured".to_owned())?;
288 let envelope: AppEnvelope<Value> = serde_json::from_value(row.envelope.clone())
287 let envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())
289288 .map_err(|error| format!("invalid outbox envelope: {error}"))?;
290289 if envelope.id != row.id || envelope.kind != row.kind {
291290 return Err("outbox event identity is invalid".to_owned());
@@ -295,12 +294,13 @@
295294 let body = envelope
296295 .summary_body(&route.tag)
297296 .map_err(|error| error.to_string())?;
297 let payload = serde_json::to_value(&envelope.payload).ok();
298298 publisher
299299 .publish(
300300 &topic,
301301 &route.tag,
302302 vec![row.id.to_string()],
303 envelope.properties(),
303 envelope.properties(payload.as_ref(), None),
304304 body,
305305 )
306306 .await
@@ -310,7 +310,7 @@
310310async fn publish(
311311 db: &DatabaseConnection,
312312 identity: Option<&EventHandler>,
313 app: Option<&AppPublisher>,
313 app: Option<&Publisher>,
314314 lease_id: Uuid,
315315 row: PendingEvent,
316316) -> Result<(), DbErr> {
@@ -393,7 +393,7 @@
393393async fn run_batch(
394394 db: &DatabaseConnection,
395395 identity: Option<&EventHandler>,
396 app: Option<&AppPublisher>,
396 app: Option<&Publisher>,
397397 _request_id: Option<&str>,
398398) -> Result<usize, DbErr> {
399399 let lease_id = Uuid::now_v7();
@@ -449,7 +449,7 @@
449449 domain: &'static str,
450450 db: DatabaseConnection,
451451 identity: Option<EventHandler>,
452 app: Option<AppPublisher>,
452 app: Option<Publisher>,
453453 cancel: CancellationToken,
454454) {
455455 let mut sample_at = Instant::now();
@@ -501,7 +501,7 @@
501501impl OutboxRelay {
502502 pub fn start(
503503 identity: Option<EventHandler>,
504 app: Option<AppPublisher>,
504 app: Option<Publisher>,
505505 databases: Vec<(&'static str, DatabaseConnection)>,
506506 ) -> Self {
507507 let cancel = CancellationToken::new();
Modified lib/events/src/route.rs lib/events/src/route.rs +1 −1 View lib/events/src/route.rs
@@ -1,6 +1,6 @@
11use serde_json::Value;
22
3use queue::identity::{QueueError, QueueResult};
3use queue::{QueueError, QueueResult};
44
55pub struct Route {
66 pub area: String,
Modified lib/export/src/events.rs lib/export/src/events.rs +2 −6 View lib/export/src/events.rs
@@ -4,10 +4,6 @@
44use serde::Serialize;
55use uuid::Uuid;
66
7fn timestamp(value: time::OffsetDateTime) -> chrono::DateTime<chrono::Utc> {
8 chrono::DateTime::from_timestamp(value.unix_timestamp(), value.nanosecond()).unwrap_or_default()
9}
10
117fn active_model<T: Serialize>(
128 kind: &str,
139 labels: Labels,
@@ -30,8 +26,8 @@
3026 kind: Set(kind.to_string()),
3127 ordering_key: Set(ordering_key.to_string()),
3228 envelope: Set(stored),
33 occurred_at: Set(timestamp(envelope.occurred_at)),
34 available_at: Set(timestamp(envelope.occurred_at)),
29 occurred_at: Set(envelope.occurred_at),
30 available_at: Set(envelope.occurred_at),
3531 ..Default::default()
3632 })
3733}
Modified lib/export/Cargo.toml lib/export/Cargo.toml +0 −1 View lib/export/Cargo.toml
@@ -29,7 +29,6 @@
2929utoipa = { workspace = true }
3030tower = { workspace = true }
3131serde_json = { workspace = true }
32chrono = { version = "0.4.45", features = ["serde"] }
3332uuid = { workspace = true }
3433bytes = { workspace = true }
3534thiserror = { workspace = true }
Modified lib/identity/src/lib.rs lib/identity/src/lib.rs +1 −1 View lib/identity/src/lib.rs
@@ -21,6 +21,6 @@
2121pub use models;
2222pub use oltp::{self, shutdown};
2323pub use proto as protos;
24pub use queue::identity as queue;
24pub use queue;
2525pub use session;
2626pub use token as jwt;
Deleted lib/queue/src/identity/conn.rs lib/queue/src/identity/conn.rs +0 −113
@@ -1,113 +0,0 @@
1use std::sync::Arc;
2
3use rocketmq_client_rust::producer::default_mq_producer::DefaultMQProducer;
4use rocketmq_common::common::message::message_single::Message;
5use tokio::sync::Mutex;
6
7use crate::identity::error::{QueueError, QueueResult};
8use config::identity::rocketmq::RocketMqConfig;
9
10pub const DEFAULT_MAX_MESSAGE_BYTES: usize = 4 * 1024 * 1024;
11
12#[derive(Clone)]
13pub struct RocketMqConn {
14 producer: Arc<dyn Producer>,
15 config: RocketMqConfig,
16}
17
18impl std::fmt::Debug for RocketMqConn {
19 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
20 formatter
21 .debug_struct("RocketMqConn")
22 .field("name_server", &self.config.name_server)
23 .field("producer_group", &self.config.producer_group)
24 .field("topic_prefix", &self.config.topic_prefix)
25 .finish()
26 }
27}
28
29impl RocketMqConn {
30 pub async fn init(config: &RocketMqConfig) -> QueueResult<Self> {
31 config.validate().map_err(QueueError::config)?;
32
33 let mut producer = DefaultMQProducer::builder()
34 .producer_group(config.producer_group.clone())
35 .name_server_addr(config.name_server.replace(',', ";"))
36 .build();
37 producer.start().await.map_err(|error| {
38 QueueError::connect(format!("failed to start the rocketmq producer: {error}"))
39 })?;
40
41 Ok(Self {
42 producer: Arc::new(RocketProducer(Mutex::new(producer))),
43 config: config.clone(),
44 })
45 }
46
47 pub fn config(&self) -> &RocketMqConfig {
48 &self.config
49 }
50
51 pub fn max_payload(&self) -> usize {
52 self.config.max_message_bytes
53 }
54
55 pub async fn send(&self, topic: &str, message: Message) -> QueueResult<()> {
56 self.producer
57 .send(topic, message, self.config.send_timeout_ms)
58 .await
59 }
60
61 pub async fn health(&self, topic: &str) -> QueueResult<()> {
62 self.producer.health(topic).await
63 }
64
65 pub async fn shutdown(&self) {
66 self.producer.shutdown().await;
67 }
68}
69
70#[async_trait::async_trait]
71trait Producer: Send + Sync {
72 async fn send(&self, topic: &str, message: Message, timeout_ms: u64) -> QueueResult<()>;
73 async fn health(&self, topic: &str) -> QueueResult<()>;
74 async fn shutdown(&self);
75}
76
77struct RocketProducer(Mutex<DefaultMQProducer>);
78
79#[async_trait::async_trait]
80impl Producer for RocketProducer {
81 async fn send(&self, topic: &str, message: Message, timeout_ms: u64) -> QueueResult<()> {
82 let mut producer = self.0.lock().await;
83 let result = producer
84 .send_with_timeout(message, timeout_ms)
85 .await
86 .map_err(|error| {
87 QueueError::publish(format!("failed to publish to `{topic}`: {error}"))
88 })?;
89 if result.is_none() {
90 return Err(QueueError::publish(format!(
91 "producer returned no send result for `{topic}`"
92 )));
93 }
94 Ok(())
95 }
96
97 async fn health(&self, topic: &str) -> QueueResult<()> {
98 let mut producer = self.0.lock().await;
99 producer
100 .fetch_publish_message_queues(topic)
101 .await
102 .map(|_| ())
103 .map_err(|error| {
104 QueueError::connect(format!(
105 "rocketmq route lookup for `{topic}` failed: {error}"
106 ))
107 })
108 }
109
110 async fn shutdown(&self) {
111 self.0.lock().await.shutdown().await;
112 }
113}
Deleted lib/queue/src/identity/envelope.rs lib/queue/src/identity/envelope.rs +0 −270
@@ -1,270 +0,0 @@
1use std::collections::BTreeMap;
2
3use bytes::Bytes;
4use chrono::{DateTime, Utc};
5use serde::de::DeserializeOwned;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use uuid::Uuid;
9
10use crate::identity::error::{QueueError, QueueResult};
11use crate::identity::subject;
12
13pub const PROPERTY_USER: &str = "user";
14pub const PROPERTY_ORG: &str = "org";
15pub const PROPERTY_REQUEST_ID: &str = "request_id";
16pub const SCHEMA_VERSION: u16 = 1;
17
18pub const EVENT_PREFIX: &str = "events.";
19
20#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
21#[serde(transparent)]
22pub struct Labels(BTreeMap<String, String>);
23
24impl Labels {
25 pub fn new() -> Self {
26 Self::default()
27 }
28
29 pub fn with(mut self, key: impl Into<String>, value: impl std::fmt::Display) -> Self {
30 self.insert(key, value);
31 self
32 }
33
34 pub fn with_opt(
35 mut self,
36 key: impl Into<String>,
37 value: Option<impl std::fmt::Display>,
38 ) -> Self {
39 if let Some(value) = value {
40 self.insert(key, value);
41 }
42 self
43 }
44
45 pub fn insert(&mut self, key: impl Into<String>, value: impl std::fmt::Display) {
46 self.0.insert(key.into(), value.to_string());
47 }
48
49 pub fn get(&self, key: &str) -> Option<&str> {
50 self.0.get(key).map(String::as_str)
51 }
52
53 pub fn is_empty(&self) -> bool {
54 self.0.is_empty()
55 }
56
57 pub fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
58 self.0
59 .iter()
60 .map(|(key, value)| (key.as_str(), value.as_str()))
61 }
62}
63
64#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
65#[serde(rename_all = "snake_case")]
66pub enum DeliveryClass {
67 #[default]
68 BusinessEvent,
69 TaskCommand,
70 Broadcast,
71}
72
73const fn schema_version() -> u16 {
74 SCHEMA_VERSION
75}
76
77#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
78pub struct Envelope<T> {
79 pub id: Uuid,
80 #[serde(default = "schema_version")]
81 pub schema_version: u16,
82 #[serde(default)]
83 pub delivery: DeliveryClass,
84 pub kind: String,
85 #[serde(default)]
86 pub source: String,
87 #[serde(default, skip_serializing_if = "Option::is_none")]
88 pub idempotency_key: Option<String>,
89 #[serde(default, skip_serializing_if = "Option::is_none")]
90 pub trace_id: Option<String>,
91 #[serde(default, skip_serializing_if = "Labels::is_empty")]
92 pub labels: Labels,
93 pub occurred_at: DateTime<Utc>,
94 #[serde(default)]
95 pub attempt: u32,
96 pub payload: T,
97}
98
99impl<T> Envelope<T> {
100 pub fn new(kind: impl Into<String>, payload: T) -> Self {
101 Self {
102 id: Uuid::now_v7(),
103 schema_version: SCHEMA_VERSION,
104 delivery: DeliveryClass::BusinessEvent,
105 kind: kind.into(),
106 source: String::new(),
107 idempotency_key: None,
108 trace_id: None,
109 labels: Labels::new(),
110 occurred_at: Utc::now(),
111 attempt: 0,
112 payload,
113 }
114 }
115
116 pub fn with_delivery(mut self, delivery: DeliveryClass) -> Self {
117 self.delivery = delivery;
118 self
119 }
120
121 pub fn with_source(mut self, source: impl Into<String>) -> Self {
122 self.source = source.into();
123 self
124 }
125
126 pub fn with_idempotency_key(mut self, key: impl Into<String>) -> Self {
127 self.idempotency_key = Some(key.into());
128 self
129 }
130
131 pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
132 self.trace_id = Some(trace_id.into());
133 self
134 }
135
136 pub fn with_labels(mut self, labels: Labels) -> Self {
137 self.labels = labels;
138 self
139 }
140
141 pub fn with_label(mut self, key: impl Into<String>, value: impl std::fmt::Display) -> Self {
142 self.labels.insert(key, value);
143 self
144 }
145
146 pub fn with_attempt(mut self, attempt: u32) -> Self {
147 self.attempt = attempt;
148 self
149 }
150
151 pub fn message_id(&self) -> String {
152 if self.delivery == DeliveryClass::TaskCommand {
153 return self
154 .idempotency_key
155 .clone()
156 .unwrap_or_else(|| self.id.to_string());
157 }
158 self.id.to_string()
159 }
160
161 pub fn bound_subject(&self, prefix: &str) -> QueueResult<String> {
162 let kind = subject::normalize(&self.kind);
163 if !subject::is_specific(&kind) {
164 return Err(QueueError::payload(format!(
165 "envelope kind `{}` is not a concrete subject",
166 self.kind
167 )));
168 }
169
170 Ok(format!("{prefix}{kind}"))
171 }
172
173 pub fn verify_subject(&self, prefix: &str, subject_name: &str) -> QueueResult<()> {
174 let expected = self.bound_subject(prefix)?;
175 let actual = subject_name.trim().trim_matches(subject::SEPARATOR);
176 if expected != actual {
177 return Err(QueueError::payload(format!(
178 "subject `{subject_name}` does not carry kind `{}`",
179 self.kind
180 )));
181 }
182
183 Ok(())
184 }
185
186 pub fn verify_event_subject(&self, subject_name: &str) -> QueueResult<()> {
187 self.verify_subject(EVENT_PREFIX, subject_name)
188 }
189
190 pub fn properties(
191 &self,
192 payload: Option<&Value>,
193 request_id: Option<&str>,
194 ) -> Vec<(String, String)> {
195 let mut properties = Vec::with_capacity(3);
196
197 let user = self
198 .labels
199 .get("user_id")
200 .map(str::to_owned)
201 .or_else(|| payload.and_then(|value| field_text(value, "user_id")))
202 .or_else(|| payload.and_then(|value| field_text(value, "target_user_id")))
203 .or_else(|| self.labels.get("actor_user_id").map(str::to_owned));
204 if let Some(user) = user {
205 properties.push((PROPERTY_USER.to_string(), user));
206 }
207
208 let org = self
209 .labels
210 .get("organization_id")
211 .map(str::to_owned)
212 .or_else(|| payload.and_then(|value| field_text(value, "organization_id")));
213 if let Some(org) = org {
214 properties.push((PROPERTY_ORG.to_string(), org));
215 }
216
217 let request_id = self
218 .labels
219 .get("request_id")
220 .map(str::to_owned)
221 .or_else(|| payload.and_then(|value| field_text(value, "request_id")))
222 .or_else(|| request_id.map(str::to_owned));
223 if let Some(request_id) = request_id {
224 properties.push((PROPERTY_REQUEST_ID.to_string(), request_id));
225 }
226
227 properties
228 }
229}
230
231impl<T: Serialize> Envelope<T> {
232 pub fn encode(&self) -> QueueResult<Bytes> {
233 serde_json::to_vec(self)
234 .map(Bytes::from)
235 .map_err(QueueError::from)
236 }
237
238 pub fn summary_body(&self, summary: &str) -> QueueResult<Bytes> {
239 serde_json::to_vec(&Summary {
240 id: self.id,
241 kind: &self.kind,
242 occurred_at: &self.occurred_at,
243 summary,
244 })
245 .map(Bytes::from)
246 .map_err(QueueError::from)
247 }
248}
249
250impl<T: DeserializeOwned> Envelope<T> {
251 pub fn decode(payload: &[u8]) -> QueueResult<Self> {
252 serde_json::from_slice(payload).map_err(QueueError::from)
253 }
254}
255
256#[derive(Serialize)]
257struct Summary<'a> {
258 id: Uuid,
259 kind: &'a str,
260 occurred_at: &'a DateTime<Utc>,
261 summary: &'a str,
262}
263
264fn field_text(value: &Value, key: &str) -> Option<String> {
265 match value.get(key)? {
266 Value::String(text) => Some(text.clone()),
267 Value::Number(number) => Some(number.to_string()),
268 _ => None,
269 }
270}
Deleted lib/queue/src/identity/error.rs lib/queue/src/identity/error.rs +0 −71
@@ -1,71 +0,0 @@
1use std::fmt;
2
3pub type QueueResult<T> = Result<T, QueueError>;
4
5#[derive(Debug, Clone, PartialEq, Eq)]
6pub enum QueueError {
7 Config(String),
8 Connect(String),
9 Publish(String),
10 Payload(String),
11}
12
13impl QueueError {
14 pub fn config(message: impl Into<String>) -> Self {
15 Self::Config(message.into())
16 }
17
18 pub fn connect(message: impl Into<String>) -> Self {
19 Self::Connect(message.into())
20 }
21
22 pub fn publish(message: impl Into<String>) -> Self {
23 Self::Publish(message.into())
24 }
25
26 pub fn payload(message: impl Into<String>) -> Self {
27 Self::Payload(message.into())
28 }
29
30 pub fn message(&self) -> &str {
31 match self {
32 Self::Config(message)
33 | Self::Connect(message)
34 | Self::Publish(message)
35 | Self::Payload(message) => message,
36 }
37 }
38
39 pub fn is_config(&self) -> bool {
40 matches!(self, Self::Config(_))
41 }
42
43 pub fn is_connect(&self) -> bool {
44 matches!(self, Self::Connect(_))
45 }
46
47 pub fn is_retryable(&self) -> bool {
48 matches!(self, Self::Connect(_) | Self::Publish(_))
49 }
50}
51
52impl fmt::Display for QueueError {
53 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
54 let label = match self {
55 Self::Config(_) => "config",
56 Self::Connect(_) => "connect",
57 Self::Publish(_) => "publish",
58 Self::Payload(_) => "payload",
59 };
60
61 write!(formatter, "rocketmq {label}: {}", self.message())
62 }
63}
64
65impl std::error::Error for QueueError {}
66
67impl From<serde_json::Error> for QueueError {
68 fn from(error: serde_json::Error) -> Self {
69 Self::Payload(error.to_string())
70 }
71}
Deleted lib/queue/src/identity/mod.rs lib/queue/src/identity/mod.rs +0 −13
@@ -1,13 +0,0 @@
1pub mod conn;
2pub mod envelope;
3pub mod error;
4pub mod publisher;
5pub mod subject;
6
7pub use conn::{DEFAULT_MAX_MESSAGE_BYTES, RocketMqConn};
8pub use envelope::{
9 DeliveryClass, Envelope, Labels, PROPERTY_ORG, PROPERTY_REQUEST_ID, PROPERTY_USER,
10 SCHEMA_VERSION,
11};
12pub use error::{QueueError, QueueResult};
13pub use publisher::Publisher;
Deleted lib/queue/src/identity/publisher.rs lib/queue/src/identity/publisher.rs +0 −65
@@ -1,65 +0,0 @@
1use bytes::Bytes;
2use rocketmq_common::common::message::message_single::Message;
3
4use crate::identity::conn::RocketMqConn;
5use crate::identity::error::{QueueError, QueueResult};
6
7#[derive(Clone, Debug)]
8pub struct Publisher {
9 conn: RocketMqConn,
10}
11
12impl Publisher {
13 pub fn new(conn: RocketMqConn) -> Self {
14 Self { conn }
15 }
16
17 pub fn conn(&self) -> &RocketMqConn {
18 &self.conn
19 }
20
21 pub fn max_payload(&self) -> usize {
22 self.conn.max_payload()
23 }
24
25 pub fn topic(&self, area: &str) -> String {
26 self.conn.config().topic(area)
27 }
28
29 pub async fn publish(
30 &self,
31 topic: &str,
32 tag: &str,
33 keys: Vec<String>,
34 properties: Vec<(String, String)>,
35 body: Bytes,
36 ) -> QueueResult<()> {
37 let max_payload = self.max_payload();
38 if body.len() > max_payload {
39 return Err(QueueError::payload(format!(
40 "event body for `{topic}` is {} bytes, exceeding the {max_payload}-byte limit",
41 body.len()
42 )));
43 }
44
45 let mut builder = Message::builder()
46 .topic(topic)
47 .tags(tag)
48 .keys(keys)
49 .body(body.to_vec());
50 for (key, value) in properties {
51 builder = builder.raw_property(key.clone(), value).map_err(|error| {
52 QueueError::publish(format!(
53 "invalid message property `{key}` for `{topic}`: {error}"
54 ))
55 })?;
56 }
57 let message = builder.build().map_err(|error| {
58 QueueError::publish(format!(
59 "failed to build the message for `{topic}`: {error}"
60 ))
61 })?;
62
63 self.conn.send(topic, message).await
64 }
65}
Deleted lib/queue/src/identity/subject.rs lib/queue/src/identity/subject.rs +0 −19
@@ -1,19 +0,0 @@
1pub const SEPARATOR: char = '.';
2pub const MAX_SUBJECT_LEN: usize = 255;
3pub const MAX_TOKEN_LEN: usize = 64;
4
5pub fn is_token_char(c: char) -> bool {
6 !(c.is_whitespace() || c.is_control() || matches!(c, '.' | '*' | '>' | '/' | '\\' | '$'))
7}
8
9pub fn is_specific(subject: &str) -> bool {
10 !subject.is_empty()
11 && subject.len() <= MAX_SUBJECT_LEN
12 && subject.split(SEPARATOR).all(|token| {
13 !token.is_empty() && token.len() <= MAX_TOKEN_LEN && token.chars().all(is_token_char)
14 })
15}
16
17pub fn normalize(value: &str) -> String {
18 value.trim().trim_matches(SEPARATOR).to_owned()
19}
Modified lib/queue/src/conn.rs lib/queue/src/conn.rs +48 −8 View lib/queue/src/conn.rs
@@ -7,9 +7,11 @@
77
88use crate::error::{QueueError, QueueResult};
99
10pub const DEFAULT_MAX_MESSAGE_BYTES: usize = 4 * 1024 * 1024;
11
1012#[derive(Clone)]
1113pub struct RocketMqConn {
12 producer: Arc<Mutex<DefaultMQProducer>>,
14 producer: Arc<dyn Producer>,
1315 config: RocketMqConfig,
1416}
1517
@@ -26,7 +28,7 @@
2628
2729impl RocketMqConn {
2830 pub async fn init(config: &RocketMqConfig) -> QueueResult<Self> {
29 config.is_valid().map_err(QueueError::config)?;
31 config.validate().map_err(QueueError::config)?;
3032 probe_nameservers(config).await?;
3133
3234 let mut producer = DefaultMQProducer::builder()
@@ -38,7 +40,7 @@
3840 })?;
3941
4042 Ok(Self {
41 producer: Arc::new(Mutex::new(producer)),
43 producer: Arc::new(RocketProducer(Mutex::new(producer))),
4244 config: config.clone(),
4345 })
4446 }
@@ -52,9 +54,35 @@
5254 }
5355
5456 pub async fn send(&self, topic: &str, message: Message) -> QueueResult<()> {
55 let mut producer = self.producer.lock().await;
57 self.producer
58 .send(topic, message, self.config.send_timeout_ms)
59 .await
60 }
61
62 pub async fn health(&self, topic: &str) -> QueueResult<()> {
63 self.producer.health(topic).await
64 }
65
66 pub async fn shutdown(&self) {
67 self.producer.shutdown().await;
68 }
69}
70
71#[async_trait::async_trait]
72trait Producer: Send + Sync {
73 async fn send(&self, topic: &str, message: Message, timeout_ms: u64) -> QueueResult<()>;
74 async fn health(&self, topic: &str) -> QueueResult<()>;
75 async fn shutdown(&self);
76}
77
78struct RocketProducer(Mutex<DefaultMQProducer>);
79
80#[async_trait::async_trait]
81impl Producer for RocketProducer {
82 async fn send(&self, topic: &str, message: Message, timeout_ms: u64) -> QueueResult<()> {
83 let mut producer = self.0.lock().await;
5684 let result = producer
57 .send_with_timeout(message, self.config.send_timeout_ms)
85 .send_with_timeout(message, timeout_ms)
5886 .await
5987 .map_err(|error| {
6088 QueueError::publish(format!("failed to publish to `{topic}`: {error}"))
@@ -67,9 +95,21 @@
6795 Ok(())
6896 }
6997
70 pub async fn shutdown(&self) {
71 let mut producer = self.producer.lock().await;
72 producer.shutdown().await;
98 async fn health(&self, topic: &str) -> QueueResult<()> {
99 let mut producer = self.0.lock().await;
100 producer
101 .fetch_publish_message_queues(topic)
102 .await
103 .map(|_| ())
104 .map_err(|error| {
105 QueueError::connect(format!(
106 "rocketmq route lookup for `{topic}` failed: {error}"
107 ))
108 })
109 }
110
111 async fn shutdown(&self) {
112 self.0.lock().await.shutdown().await;
73113 }
74114}
75115
Modified lib/queue/src/envelope.rs lib/queue/src/envelope.rs +122 −56 View lib/queue/src/envelope.rs
@@ -1,22 +1,22 @@
11use std::collections::BTreeMap;
22
33use bytes::Bytes;
4use chrono::{DateTime, Utc};
45use serde::de::DeserializeOwned;
56use serde::{Deserialize, Serialize};
67use serde_json::Value;
7use time::OffsetDateTime;
88use uuid::Uuid;
99
1010use crate::error::{QueueError, QueueResult};
1111use crate::subject;
1212
13pub const PROPERTY_USER: &str = "user";
14pub const PROPERTY_ORG: &str = "org";
15pub const PROPERTY_REQUEST_ID: &str = "request_id";
1316pub const SCHEMA_VERSION: u16 = 1;
17
1418pub const EVENT_PREFIX: &str = "events.";
1519
16fn schema_version() -> u16 {
17 SCHEMA_VERSION
18}
19
2020#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
2121#[serde(transparent)]
2222pub struct Labels(BTreeMap<String, String>);
@@ -53,19 +53,46 @@
5353 pub fn is_empty(&self) -> bool {
5454 self.0.is_empty()
5555 }
56
57 pub fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
58 self.0
59 .iter()
60 .map(|(key, value)| (key.as_str(), value.as_str()))
61 }
5662}
5763
58#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
64#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
65#[serde(rename_all = "snake_case")]
66pub enum DeliveryClass {
67 #[default]
68 BusinessEvent,
69 TaskCommand,
70 Broadcast,
71}
72
73const fn schema_version() -> u16 {
74 SCHEMA_VERSION
75}
76
77#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
5978pub struct Envelope<T> {
6079 pub id: Uuid,
6180 #[serde(default = "schema_version")]
6281 pub schema_version: u16,
82 #[serde(default)]
83 pub delivery: DeliveryClass,
6384 pub kind: String,
6485 #[serde(default)]
6586 pub source: String,
87 #[serde(default, skip_serializing_if = "Option::is_none")]
88 pub idempotency_key: Option<String>,
89 #[serde(default, skip_serializing_if = "Option::is_none")]
90 pub trace_id: Option<String>,
6691 #[serde(default, skip_serializing_if = "Labels::is_empty")]
6792 pub labels: Labels,
68 pub occurred_at: OffsetDateTime,
93 pub occurred_at: DateTime<Utc>,
94 #[serde(default)]
95 pub attempt: u32,
6996 pub payload: T,
7097}
7198
@@ -74,25 +101,68 @@
74101 Self {
75102 id: Uuid::now_v7(),
76103 schema_version: SCHEMA_VERSION,
104 delivery: DeliveryClass::BusinessEvent,
77105 kind: kind.into(),
78106 source: String::new(),
107 idempotency_key: None,
108 trace_id: None,
79109 labels: Labels::new(),
80 occurred_at: OffsetDateTime::now_utc(),
110 occurred_at: Utc::now(),
111 attempt: 0,
81112 payload,
82113 }
83114 }
84115
116 pub fn with_delivery(mut self, delivery: DeliveryClass) -> Self {
117 self.delivery = delivery;
118 self
119 }
120
85121 pub fn with_source(mut self, source: impl Into<String>) -> Self {
86122 self.source = source.into();
87123 self
88124 }
89125
126 pub fn with_idempotency_key(mut self, key: impl Into<String>) -> Self {
127 self.idempotency_key = Some(key.into());
128 self
129 }
130
131 pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
132 self.trace_id = Some(trace_id.into());
133 self
134 }
135
90136 pub fn with_labels(mut self, labels: Labels) -> Self {
91137 self.labels = labels;
92138 self
93139 }
94140
141 pub fn with_label(mut self, key: impl Into<String>, value: impl std::fmt::Display) -> Self {
142 self.labels.insert(key, value);
143 self
144 }
145
146 pub fn with_attempt(mut self, attempt: u32) -> Self {
147 self.attempt = attempt;
148 self
149 }
150
151 pub fn message_id(&self) -> String {
152 if self.delivery == DeliveryClass::TaskCommand {
153 return self
154 .idempotency_key
155 .clone()
156 .unwrap_or_else(|| self.id.to_string());
157 }
158 self.id.to_string()
159 }
160
95161 pub fn subject(&self) -> QueueResult<String> {
162 self.bound_subject(EVENT_PREFIX)
163 }
164
165 pub fn bound_subject(&self, prefix: &str) -> QueueResult<String> {
96166 let kind = subject::normalize(&self.kind);
97167 if !subject::is_specific(&kind) {
98168 return Err(QueueError::payload(format!(
@@ -100,37 +170,33 @@
100170 self.kind
101171 )));
102172 }
103 Ok(format!("{EVENT_PREFIX}{kind}"))
173
174 Ok(format!("{prefix}{kind}"))
104175 }
105}
106176
107impl<T: Serialize> Envelope<T> {
108 pub fn encode(&self) -> QueueResult<Bytes> {
109 serde_json::to_vec(self)
110 .map(Bytes::from)
111 .map_err(QueueError::from)
112 }
177 pub fn verify_subject(&self, prefix: &str, subject_name: &str) -> QueueResult<()> {
178 let expected = self.bound_subject(prefix)?;
179 let actual = subject_name.trim().trim_matches(subject::SEPARATOR);
180 if expected != actual {
181 return Err(QueueError::payload(format!(
182 "subject `{subject_name}` does not carry kind `{}`",
183 self.kind
184 )));
185 }
113186
114 pub fn summary_body(&self, summary: &str) -> QueueResult<Bytes> {
115 serde_json::to_vec(&Summary {
116 id: self.id,
117 kind: &self.kind,
118 occurred_at: &self.occurred_at,
119 summary,
120 })
121 .map(Bytes::from)
122 .map_err(QueueError::from)
187 Ok(())
123188 }
124189
125 pub fn properties(&self) -> Vec<(String, String)> {
126 let payload = serde_json::to_value(&self.payload).ok();
127 self.properties_of(payload.as_ref())
190 pub fn verify_event_subject(&self, subject_name: &str) -> QueueResult<()> {
191 self.verify_subject(EVENT_PREFIX, subject_name)
128192 }
129}
130193
131impl<T> Envelope<T> {
132 fn properties_of(&self, payload: Option<&Value>) -> Vec<(String, String)> {
133 let mut properties = Vec::with_capacity(4);
194 pub fn properties(
195 &self,
196 payload: Option<&Value>,
197 request_id: Option<&str>,
198 ) -> Vec<(String, String)> {
199 let mut properties = Vec::with_capacity(3);
134200
135201 let user = self
136202 .labels
@@ -140,7 +206,7 @@
140206 .or_else(|| payload.and_then(|value| field_text(value, "target_user_id")))
141207 .or_else(|| self.labels.get("actor_user_id").map(str::to_owned));
142208 if let Some(user) = user {
143 properties.push(("user".to_string(), user));
209 properties.push((PROPERTY_USER.to_string(), user));
144210 }
145211
146212 let org = self
@@ -149,39 +215,39 @@
149215 .map(str::to_owned)
150216 .or_else(|| payload.and_then(|value| field_text(value, "organization_id")));
151217 if let Some(org) = org {
152 properties.push(("org".to_string(), org));
218 properties.push((PROPERTY_ORG.to_string(), org));
153219 }
154220
155221 let request_id = self
156222 .labels
157223 .get("request_id")
158224 .map(str::to_owned)
159 .or_else(|| payload.and_then(|value| field_text(value, "request_id")));
225 .or_else(|| payload.and_then(|value| field_text(value, "request_id")))
226 .or_else(|| request_id.map(str::to_owned));
160227 if let Some(request_id) = request_id {
161 properties.push(("request_id".to_string(), request_id));
228 properties.push((PROPERTY_REQUEST_ID.to_string(), request_id));
162229 }
163230
164 let room = self
165 .labels
166 .get("room_id")
167 .map(str::to_owned)
168 .or_else(|| payload.and_then(|value| field_text(value, "room_id")));
169 if let Some(room) = room {
170 properties.push(("room_id".to_string(), room));
171 }
231 properties
232 }
233}
172234
173 for key in ["issue_id", "pull_id", "message_id", "thread_id"] {
174 let value = self
175 .labels
176 .get(key)
177 .map(str::to_owned)
178 .or_else(|| payload.and_then(|value| field_text(value, key)));
179 if let Some(value) = value {
180 properties.push((key.to_string(), value));
181 }
182 }
235impl<T: Serialize> Envelope<T> {
236 pub fn encode(&self) -> QueueResult<Bytes> {
237 serde_json::to_vec(self)
238 .map(Bytes::from)
239 .map_err(QueueError::from)
240 }
183241
184 properties
242 pub fn summary_body(&self, summary: &str) -> QueueResult<Bytes> {
243 serde_json::to_vec(&Summary {
244 id: self.id,
245 kind: &self.kind,
246 occurred_at: &self.occurred_at,
247 summary,
248 })
249 .map(Bytes::from)
250 .map_err(QueueError::from)
185251 }
186252}
187253
@@ -195,7 +261,7 @@
195261struct Summary<'a> {
196262 id: Uuid,
197263 kind: &'a str,
198 occurred_at: &'a OffsetDateTime,
264 occurred_at: &'a DateTime<Utc>,
199265 summary: &'a str,
200266}
201267
Modified lib/queue/src/error.rs lib/queue/src/error.rs +12 −0 View lib/queue/src/error.rs
@@ -35,6 +35,18 @@
3535 | Self::Payload(message) => message,
3636 }
3737 }
38
39 pub fn is_config(&self) -> bool {
40 matches!(self, Self::Config(_))
41 }
42
43 pub fn is_connect(&self) -> bool {
44 matches!(self, Self::Connect(_))
45 }
46
47 pub fn is_retryable(&self) -> bool {
48 matches!(self, Self::Connect(_) | Self::Publish(_))
49 }
3850}
3951
4052impl fmt::Display for QueueError {
Modified lib/queue/src/lib.rs lib/queue/src/lib.rs +2 −3 View lib/queue/src/lib.rs
@@ -3,13 +3,12 @@
33pub mod conn;
44pub mod envelope;
55pub mod error;
6pub mod identity;
76pub mod publisher;
87pub mod route;
98pub mod subject;
109
11pub use conn::RocketMqConn;
12pub use envelope::{EVENT_PREFIX, Envelope, Labels, SCHEMA_VERSION};
10pub use conn::{DEFAULT_MAX_MESSAGE_BYTES, RocketMqConn};
11pub use envelope::{DeliveryClass, EVENT_PREFIX, Envelope, Labels, SCHEMA_VERSION};
1312pub use error::{QueueError, QueueResult};
1413pub use publisher::Publisher;
1514pub use route::Route;
Modified lib/queue/src/subject.rs lib/queue/src/subject.rs +1 −1 View lib/queue/src/subject.rs
@@ -2,7 +2,7 @@
22pub const MAX_SUBJECT_LEN: usize = 255;
33pub const MAX_TOKEN_LEN: usize = 64;
44
5fn is_token_char(c: char) -> bool {
5pub fn is_token_char(c: char) -> bool {
66 !(c.is_whitespace() || c.is_control() || matches!(c, '.' | '*' | '>' | '/' | '\\' | '$'))
77}
88
Modified lib/queue/Cargo.toml lib/queue/Cargo.toml +0 −4 View lib/queue/Cargo.toml
@@ -7,15 +7,11 @@
77config = { workspace = true }
88async-trait = { workspace = true }
99chrono = { version = "0.4.45", features = ["serde"] }
10sea-orm = { workspace = true }
11thiserror = { workspace = true }
1210serde = { workspace = true, features = ["derive"] }
1311serde_json = { workspace = true }
1412bytes = { workspace = true }
1513uuid = { workspace = true }
16time = { workspace = true }
1714tokio = { workspace = true, features = ["time", "sync", "rt", "net"] }
18tokio-util = { workspace = true }
1915tracing = { workspace = true }
2016rocketmq-client-rust = { workspace = true }
2117rocketmq-common = { workspace = true }
Deleted lib/user/src/share/crypto.rs lib/user/src/share/crypto.rs +0 −27
@@ -1,27 +0,0 @@
1use base64::Engine;
2use base64::engine::general_purpose::URL_SAFE_NO_PAD;
3use rand::Rng;
4use rand::rngs::OsRng;
5use sha2::{Digest, Sha256};
6
7pub fn generate_numeric_code(len: usize) -> String {
8 let mut rng = OsRng;
9 (0..len)
10 .map(|_| rng.gen_range(0..10u32).to_string())
11 .collect()
12}
13
14pub fn sha256_b64url(data: &[u8]) -> String {
15 URL_SAFE_NO_PAD.encode(Sha256::digest(data))
16}
17
18pub fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
19 if a.len() != b.len() {
20 return false;
21 }
22 let mut diff = 0u8;
23 for (x, y) in a.iter().zip(b.iter()) {
24 diff |= x ^ y;
25 }
26 diff == 0
27}
Modified lib/user/src/share/mod.rs lib/user/src/share/mod.rs +0 −1 View lib/user/src/share/mod.rs
@@ -1,4 +1,3 @@
1pub mod crypto;
21pub mod guard;
32pub mod page;
43pub mod types;
Modified Cargo.lock Cargo.lock +0 −6 View Cargo.lock
@@ -762,7 +762,6 @@
762762version = "1.0.0"
763763dependencies = [
764764 "actix-web",
765 "aes-gcm 0.10.3",
766765 "argon2",
767766 "async-trait",
768767 "base64 0.23.1",
@@ -2385,7 +2384,6 @@
23852384dependencies = [
23862385 "actix-web",
23872386 "bytes",
2388 "chrono",
23892387 "config 1.0.0",
23902388 "http 1.5.0",
23912389 "imaging",
@@ -5929,13 +5927,9 @@
59295927 "config 1.0.0",
59305928 "rocketmq-client-rust",
59315929 "rocketmq-common",
5932 "sea-orm",
59335930 "serde",
59345931 "serde_json",
5935 "thiserror 2.0.21",
5936 "time 0.3.55",
59375932 "tokio",
5938 "tokio-util",
59395933 "tracing",
59405934 "uuid",
59415935]
Modified Cargo.toml Cargo.toml +0 −1 View Cargo.toml
@@ -78,7 +78,6 @@
7878hkdf = "0.13"
7979uuid = { version = "1", features = ["v7", "serde"] }
8080argon2 = "0.6.0"
81aes-gcm = "0.10"
8281time = { version = "0.3", features = ["serde", "serde-human-readable"] }
8382bytes = "1"
8483futures = "0.3"