refactor: add events dependency and enhance event handling for improved clarity and organization

Modified lib/chat/src/share/events.rs +15 −6 View lib/chat/src/share/events.rs

@@ -90,7 +90,8 @@

9090 if let Some(thread_id) = room_thread_id(event) {

9191 labels.insert("thread_id", thread_id);

9292 }

93 if let Some(request_id) = export::events::request_id() {

93 let request_id = export::middleware::request_id();

94 if let Some(request_id) = request_id.as_deref() {

9495 labels.insert("request_id", request_id);

9596 }

9697

@@ -114,12 +115,17 @@

114115 payload["message"] = message;

115116 }

116117

117 export::events::enqueue(

118 labels.insert(

119 events::label::ORDERING_KEY,

120 format!("chat:room:{}", event.room_id()),

121 );

122 events::outbox::enqueue_now(

118123 state.database.write(),

119124 room_kind(event),

120125 labels,

121126 &payload,

122 format!("chat:room:{}", event.room_id()),

127 request_id.as_deref().unwrap_or_default(),

128 request_id.as_deref(),

123129 )

124130 .await?;

125131 Ok(())

@@ -139,7 +145,8 @@

139145 if let Some(room_id) = event.get("room_id").and_then(Value::as_str) {

140146 labels.insert("room_id", room_id);

141147 }

142 if let Some(request_id) = export::events::request_id() {

148 let request_id = export::middleware::request_id();

149 if let Some(request_id) = request_id.as_deref() {

143150 labels.insert("request_id", request_id);

144151 }

145152

@@ -166,12 +173,14 @@

166173 payload["message"] = message;

167174 }

168175

169 export::events::enqueue(

176 labels.insert(events::label::ORDERING_KEY, format!("chat:user:{uid}"));

177 events::outbox::enqueue_now(

170178 state.database.write(),

171179 kind,

172180 labels,

173181 &payload,

174 format!("chat:user:{uid}"),

182 request_id.as_deref().unwrap_or_default(),

183 request_id.as_deref(),

175184 )

176185 .await?;

177186 Ok(())

Modified lib/chat/Cargo.toml +1 −0 View lib/chat/Cargo.toml

@@ -7,6 +7,7 @@

77async-trait = { workspace = true }

88config = { workspace = true }

99export = { workspace = true }

10events = { workspace = true }

1011i18n = { workspace = true }

1112imaging = { workspace = true }

1213models = { workspace = true }

Modified lib/events/src/handler.rs +2 −0 View lib/events/src/handler.rs

@@ -11,7 +11,9 @@

1111 "chat",

1212 "git",

1313 "issue",

14 "notify",

1415 "organization",

16 "pull",

1517 "pull_request",

1618];

1719

Modified lib/events/src/label.rs +1 −0 View lib/events/src/label.rs

@@ -11,6 +11,7 @@

1111pub const BOARD_ID: &str = "board_id";

1212pub const CHANNEL_ID: &str = "channel_id";

1313pub const ROOM_ID: &str = "room_id";

14pub const ORDERING_KEY: &str = "ordering_key";

1415

1516pub const AUTH_REGISTER_CREATE: &str = "auth.register.create";

1617pub const AUTH_LOGIN_CREATE: &str = "auth.login.create";

Modified lib/events/src/outbox.rs +121 −8 View lib/events/src/outbox.rs

@@ -90,10 +90,20 @@

9090/// 组织事件按组织串行，认证事件按用户串行（没有用户主体时退回请求 id，

9191/// 相当于不串行），邮件事件按发信请求串行。其余域沿用资源 id。

9292fn ordering_key(kind: &str, labels: &Labels) -> Result<String, DbErr> {

93 if let Some(explicit) = labels

94 .get(label::ORDERING_KEY)

95 .filter(|key| !key.is_empty())

96 {

97 return Ok(explicit.to_string());

98 }

9399 let domain = domain(kind);

94100 let key = match domain {

95101 "guild" => labels.get(label::GUILD_ID),

96102 "git" => labels.get(label::REPOSITORY_ID),

103 "pull" => labels

104 .get("pull_id")

105 .or_else(|| labels.get(label::PULL_REQUEST_ID))

106 .or_else(|| labels.get(label::REPOSITORY_ID)),

97107 "pull_request" => labels

98108 .get(label::PULL_REQUEST_ID)

99109 .or_else(|| labels.get(label::REPOSITORY_ID)),

@@ -160,12 +170,54 @@

160170 "event `{kind}` requires an outbox-enabled business domain"

161171 )));

162172 }

173 let (id, model) = build_model(kind, labels, payload, trace_id, available_at)?;

174 event_outbox::Entity::insert(model)

175 .exec_without_returning(conn)

176 .await?;

177 Ok(id)

178}

179

180pub async fn enqueue_now<T: Serialize>(

181 conn: &impl ConnectionTrait,

182 kind: &str,

183 labels: Labels,

184 payload: T,

185 trace_id: &str,

186 request_id: Option<&str>,

187) -> Result<Uuid, DbErr> {

188 enqueue_at(

189 conn,

190 kind,

191 labels,

192 payload,

193 trace_id,

194 Utc::now(),

195 request_id,

196 )

197 .await

198}

199

200fn build_model<T: Serialize>(

201 kind: &str,

202 labels: Labels,

203 payload: T,

204 trace_id: &str,

205 available_at: DateTime<Utc>,

206) -> Result<(Uuid, event_outbox::ActiveModel), DbErr> {

207 if !is_durable_kind(kind) {

208 return Err(DbErr::Custom(format!(

209 "event `{kind}` requires an outbox-enabled business domain"

210 )));

211 }

163212 let ordering_key = ordering_key(kind, &labels)?;

164 let envelope = Envelope::new(kind, payload)

213 let labels = labels.without(label::ORDERING_KEY);

214 let mut envelope = Envelope::new(kind, payload)

165215 .with_delivery(DeliveryClass::BusinessEvent)

166216 .with_source(domain(kind))

167 .with_labels(labels)

168 .with_trace_id(trace_id);

217 .with_labels(labels);

218 if !trace_id.is_empty() {

219 envelope = envelope.with_trace_id(trace_id);

220 }

169221 let subject = format!("{EVENT_PREFIX}{}", envelope.kind);

170222 let encoded = envelope

171223 .encode()

@@ -177,7 +229,7 @@

177229 }

178230 let stored = serde_json::from_slice(&encoded)

179231 .map_err(|error| DbErr::Custom(format!("storing outbox event: {error}")))?;

180 event_outbox::Entity::insert(event_outbox::ActiveModel {

232 let model = event_outbox::ActiveModel {

181233 id: Set(envelope.id),

182234 subject: Set(subject),

183235 kind: Set(envelope.kind.clone()),

@@ -186,10 +238,71 @@

186238 occurred_at: Set(envelope.occurred_at),

187239 available_at: Set(available_at),

188240 ..Default::default()

189 })

190 .exec_without_returning(conn)

191 .await?;

192 Ok(envelope.id)

241 };

242 Ok((envelope.id, model))

243}

244

245fn chat_labels(room_id: Uuid, request_id: Option<&str>) -> Labels {

246 let mut labels = Labels::new()

247 .with(label::ROOM_ID, room_id)

248 .with(label::ORDERING_KEY, format!("chat:room:{room_id}"));

249 if let Some(request_id) = request_id {

250 labels.insert(label::REQUEST_ID, request_id);

251 }

252 labels

253}

254

255pub async fn enqueue_room<T: Serialize>(

256 conn: &impl ConnectionTrait,

257 kind: &str,

258 room_id: Uuid,

259 payload: T,

260 request_id: Option<&str>,

261) -> Result<Uuid, DbErr> {

262 enqueue_at(

263 conn,

264 kind,

265 chat_labels(room_id, request_id),

266 payload,

267 request_id.unwrap_or_default(),

268 Utc::now(),

269 request_id,

270 )

271 .await

272}

273

274pub async fn enqueue_chat_users<T: Serialize>(

275 conn: &impl ConnectionTrait,

276 kind: &str,

277 uids: &[i64],

278 room_id: Uuid,

279 payload: &T,

280 request_id: Option<&str>,

281) -> Result<(), DbErr> {

282 if uids.is_empty() {

283 return Ok(());

284 }

285 if !is_durable_kind(kind) {

286 return Err(DbErr::Custom(format!(

287 "event `{kind}` requires an outbox-enabled business domain"

288 )));

289 }

290 let mut models = Vec::with_capacity(uids.len());

291 for uid in uids {

292 let labels = chat_labels(room_id, request_id)

293 .with(label::USER_ID, *uid)

294 .with(label::ORDERING_KEY, format!("chat:user:{uid}"));

295 let (_, model) = build_model(

296 kind,

297 labels,

298 payload,

299 request_id.unwrap_or_default(),

300 Utc::now(),

301 )?;

302 models.push(model);

303 }

304 event_outbox::Entity::insert_many(models).exec(conn).await?;

305 Ok(())

193306}

194307

195308pub async fn release(db: &DatabaseConnection, ids: &[Uuid]) -> Result<(), DbErr> {

Deleted lib/export/src/events.rs +0 −111

@@ -1,111 +0,0 @@

1use models::event_outbox;

2use queue::{Envelope, Labels};

3use sea_orm::{ActiveModelTrait, ConnectionTrait, DbErr, EntityTrait, Set};

4use serde::Serialize;

5use uuid::Uuid;

6

7fn active_model<T: Serialize>(

8 kind: &str,

9 labels: Labels,

10 payload: &T,

11 ordering_key: &str,

12) -> Result<event_outbox::ActiveModel, DbErr> {

13 let source = kind.split('.').next().unwrap_or_default().to_string();

14 let envelope = Envelope::new(kind, payload)

15 .with_source(source)

16 .with_labels(labels);

17 let subject = envelope

18 .subject()

19 .map_err(|error| DbErr::Custom(error.to_string()))?;

20 let stored = serde_json::to_value(&envelope)

21 .map_err(|error| DbErr::Custom(format!("encoding outbox event: {error}")))?;

22

23 Ok(event_outbox::ActiveModel {

24 id: Set(envelope.id),

25 subject: Set(subject),

26 kind: Set(kind.to_string()),

27 ordering_key: Set(ordering_key.to_string()),

28 envelope: Set(stored),

29 occurred_at: Set(envelope.occurred_at),

30 available_at: Set(envelope.occurred_at),

31 ..Default::default()

32 })

33}

34

35pub async fn enqueue<T: Serialize>(

36 db: &impl ConnectionTrait,

37 kind: &str,

38 labels: Labels,

39 payload: T,

40 ordering_key: impl Into<String>,

41) -> Result<(), DbErr> {

42 active_model(kind, labels, &payload, &ordering_key.into())?

43 .insert(db)

44 .await?;

45

46 Ok(())

47}

48

49pub fn request_id() -> Option<String> {

50 crate::middleware::request_id()

51}

52

53fn room_labels(room_id: Uuid) -> Labels {

54 let mut labels = Labels::new().with("room_id", room_id);

55 if let Some(request_id) = request_id() {

56 labels.insert("request_id", request_id);

57 }

58 labels

59}

60

61/// 房间级事件的统一入队口：标签固定 `room_id`，ordering key 为 `chat:room:{room_id}`。

62///

63/// 供不在 `chat` crate 里的写入点（例如 `space` 的频道删除）复用，

64/// 避免它们为了一个标签类型去直接依赖 `queue`。

65pub async fn enqueue_room<T: Serialize>(

66 db: &impl ConnectionTrait,

67 kind: &str,

68 room_id: Uuid,

69 payload: T,

70) -> Result<(), DbErr> {

71 active_model(

72 kind,

73 room_labels(room_id),

74 &payload,

75 &format!("chat:room:{room_id}"),

76 )?

77 .insert(db)

78 .await?;

79

80 Ok(())

81}

82

83/// 房间变化扇出给一批收件人：每条一个 `ordering_key`，一次批量写入 outbox。

84///

85/// 用批量而不是逐条 `enqueue`，是因为公开频道创建时收件人等于整个空间的

86/// 活跃成员数——逐条写入会变成 N 次往返。

87pub async fn enqueue_chat_users<T: Serialize>(

88 db: &impl ConnectionTrait,

89 kind: &str,

90 uids: &[i64],

91 room_id: Uuid,

92 payload: &T,

93) -> Result<(), DbErr> {

94 if uids.is_empty() {

95 return Ok(());

96 }

97

98 let mut models = Vec::with_capacity(uids.len());

99 for uid in uids {

100 let labels = room_labels(room_id).with("user_id", *uid);

101 models.push(active_model(

102 kind,

103 labels,

104 payload,

105 &format!("chat:user:{uid}"),

106 )?);

107 }

108

109 event_outbox::Entity::insert_many(models).exec(db).await?;

110 Ok(())

111}

Modified lib/export/src/lib.rs +0 −1 View lib/export/src/lib.rs

@@ -2,7 +2,6 @@

22pub mod auth_api;

33pub mod database;

44pub mod error;

5pub mod events;

65pub mod health;

76pub mod http;

87pub mod identity_client;

Modified lib/export/src/notify.rs +6 −3 View lib/export/src/notify.rs

@@ -36,15 +36,18 @@

3636 crate::push::enqueue(redis, &job).await;

3737 }

3838 }

39 let request_id = crate::middleware::request_id();

3940 let labels = Labels::new()

4041 .with("user_id", uid)

41 .with_opt("request_id", crate::events::request_id());

42 if let Err(error) = crate::events::enqueue(

42 .with_opt("request_id", request_id.clone())

43 .with(events::label::ORDERING_KEY, format!("notify:user:{uid}"));

44 if let Err(error) = events::outbox::enqueue_now(

4345 state.database.write(),

4446 "notify.notification.create",

4547 labels,

4648 &event,

47 format!("notify:user:{uid}"),

49 request_id.as_deref().unwrap_or_default(),

50 request_id.as_deref(),

4851 )

4952 .await

5053 {

Modified lib/issues/src/services/service.rs +6 −3 View lib/issues/src/services/service.rs

@@ -148,15 +148,18 @@

148148 let mut labels = queue::Labels::new()

149149 .with("issue_id", issue_id)

150150 .with("user_id", user_id);

151 if let Some(request_id) = export::events::request_id() {

151 let request_id = export::middleware::request_id();

152 if let Some(request_id) = request_id.as_deref() {

152153 labels.insert("request_id", request_id);

153154 }

154 export::events::enqueue(

155 labels.insert(events::label::ORDERING_KEY, format!("issue:{issue_id}"));

156 events::outbox::enqueue_now(

155157 txn,

156158 event_kind(event_type),

157159 labels,

158160 payload,

159 format!("issue:{issue_id}"),

161 request_id.as_deref().unwrap_or_default(),

162 request_id.as_deref(),

160163 )

161164 .await?;

162165 Ok(())

Modified lib/issues/Cargo.toml +1 −0 View lib/issues/Cargo.toml

@@ -9,6 +9,7 @@

99async-trait = { workspace = true }

1010config = { workspace = true }

1111export = { workspace = true }

12events = { workspace = true }

1213i18n = { workspace = true }

1314models = { workspace = true }

1415queue = { workspace = true }

Modified lib/pulls/src/services/events.rs +13 −2 View lib/pulls/src/services/events.rs

@@ -122,10 +122,21 @@

122122 let mut labels = queue::Labels::new()

123123 .with("pull_id", pull.id)

124124 .with("user_id", actor);

125 if let Some(request_id) = export::events::request_id() {

125 let request_id = export::middleware::request_id();

126 if let Some(request_id) = request_id.as_deref() {

126127 labels.insert("request_id", request_id);

127128 }

128 export::events::enqueue(db, kind, labels, payload, format!("pull:{}", pull.id)).await

129 labels.insert(events::label::ORDERING_KEY, format!("pull:{}", pull.id));

130 events::outbox::enqueue_now(

131 db,

132 kind,

133 labels,

134 payload,

135 request_id.as_deref().unwrap_or_default(),

136 request_id.as_deref(),

137 )

138 .await

139 .map(|_| ())

129140}

130141

131142fn event_kind(event_type: &str) -> Option<&'static str> {

Modified lib/pulls/Cargo.toml +1 −0 View lib/pulls/Cargo.toml

@@ -6,6 +6,7 @@

66[dependencies]

77config = { workspace = true }

88export = { workspace = true }

9events = { workspace = true }

910git = { workspace = true }

1011issues = { workspace = true }

1112models = { workspace = true }

Modified lib/queue/src/envelope.rs +5 −0 View lib/queue/src/envelope.rs

@@ -46,6 +46,11 @@

4646 self.0.insert(key.into(), value.to_string());

4747 }

4848

49 pub fn without(mut self, key: &str) -> Self {

50 self.0.remove(key);

51 self

52 }

53

4954 pub fn get(&self, key: &str) -> Option<&str> {

5055 self.0.get(key).map(String::as_str)

5156 }

Modified lib/space/src/services/rooms.rs +6 −2 View lib/space/src/services/rooms.rs

@@ -238,7 +238,8 @@

238238 "avatar_url": export::resolve_user_avatar_url(user.user_id, &user.avatar_url),

239239 })

240240 });

241 export::events::enqueue_chat_users(

241 let request_id = export::middleware::request_id();

242 events::outbox::enqueue_chat_users(

242243 self.db().write(),

243244 "chat.membership.create",

244245 &member_uids

@@ -265,6 +266,7 @@

265266 },

266267 "actor": actor,

267268 }),

269 request_id.as_deref(),

268270 )

269271 .await?;

270272

@@ -316,7 +318,8 @@

316318 am.updated_at = Set(now);

317319 am.update(self.db().write()).await?;

318320

319 export::events::enqueue_room(

321 let request_id = export::middleware::request_id();

322 events::outbox::enqueue_room(

320323 self.db().write(),

321324 "chat.room.update",

322325 room_id,

@@ -332,6 +335,7 @@

332335 },

333336 "actor": actor,

334337 }),

338 request_id.as_deref(),

335339 )

336340 .await?;

337341

Modified lib/space/Cargo.toml +1 −0 View lib/space/Cargo.toml

@@ -8,6 +8,7 @@

88config = { workspace = true }

99deletion = { workspace = true }

1010export = { workspace = true }

11events = { workspace = true }

1112proto = { workspace = true }

1213i18n = { workspace = true }

1314imaging = { workspace = true }

Modified Cargo.lock +4 −0 View Cargo.lock

@@ -1261,6 +1261,7 @@

12611261 "bytes",

12621262 "config 1.0.0",

12631263 "encoding_rs",

1264 "events",

12641265 "export",

12651266 "futures",

12661267 "hmac 0.13.0",

@@ -3577,6 +3578,7 @@

35773578 "async-trait",

35783579 "bytes",

35793580 "config 1.0.0",

3581 "events",

35803582 "export",

35813583 "futures",

35823584 "http 0.2.12",

@@ -5680,6 +5682,7 @@

56805682dependencies = [

56815683 "actix-web",

56825684 "config 1.0.0",

5685 "events",

56835686 "export",

56845687 "git",

56855688 "globset",

@@ -7183,6 +7186,7 @@

71837186 "async-trait",

71847187 "config 1.0.0",

71857188 "deletion",

7189 "events",

71867190 "export",

71877191 "i18n",

71887192 "imaging",