feat(notify): implement pending notifications system for user events

Modified lib/chat/src/services/messages.rs +8 −0 View lib/chat/src/services/messages.rs

@@ -403,6 +403,8 @@

403403 let svc = self.clone();

404404 let space_id = room.space_id;

405405 let nonce_retry = cmd.nonce.clone();

406 let pending = export::notify::PendingNotifications::default();

407 let pending_txn = pending.clone();

406408

407409 let txn_result = self

408410 .db()

@@ -516,6 +518,7 @@

516518 }

517519 notify_user(

518520 txn,

521 &pending_txn,

519522 uid,

520523 NotificationType::Mention,

521524 "You were mentioned in a message".to_string(),

@@ -548,6 +551,7 @@

548551 }

549552 }

550553 txn_result?;

554 pending.flush(&self.state.redis).await;

551555

552556 let row = messages::Entity::find_by_id(message_id)

553557 .one(self.db().read())

@@ -789,6 +793,8 @@

789793

790794 let this = self.clone();

791795 let this_txn = self.clone();

796 let pending = export::notify::PendingNotifications::default();

797 let pending_txn = pending.clone();

792798 self.db()

793799 .primary()

794800 .transaction::<_, (), ChatError>(move |txn| {

@@ -840,6 +846,7 @@

840846 }

841847 notify_user(

842848 txn,

849 &pending_txn,

843850 uid,

844851 NotificationType::Mention,

845852 "You were mentioned in a message".to_string(),

@@ -855,6 +862,7 @@

855862 })

856863 })

857864 .await?;

865 pending.flush(&self.state.redis).await;

858866

859867 let row = messages::Entity::find_by_id(message_id)

860868 .one(self.db().read())

Modified lib/chat/src/services/reminders.rs +3 −0 View lib/chat/src/services/reminders.rs

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

197197 .all(self.db().write())

198198 .await?;

199199 let count = due.len() as u64;

200 let pending = export::notify::PendingNotifications::default();

200201 for reminder in due {

201202 // Atomic claim so concurrent replicas notify at most once.

202203 let claimed = message_reminders::Entity::update_many()

@@ -218,6 +219,7 @@

218219 });

219220 if let Err(error) = notify_user(

220221 self.db().write(),

222 &pending,

221223 reminder.user_id,

222224 NotificationType::System,

223225 "Message reminder".to_string(),

@@ -258,6 +260,7 @@

258260 }),

259261 )

260262 .await;

263 pending.flush(&self.state.redis).await;

261264 }

262265 Ok(count)

263266 }

Modified lib/chat/src/services/rooms.rs +35 −1 View lib/chat/src/services/rooms.rs

@@ -1,4 +1,5 @@

11use models::chat::messages::{self, ChatMessageType};

2use models::dir::dirs::{self, DirKind};

23use models::chat::read_states;

34use models::chat::room_seq;

45use models::chat::room_user_settings;

@@ -67,6 +68,10 @@

6768 pub rate_limit_per_user: Option<i32>,

6869 #[serde(default)]

6970 pub position: Option<i64>,

71 /// Move the room under another directory node of the same space; `null` puts

72 /// it back at the space root. Absent means "leave the parent alone".

73 #[serde(default)]

74 pub parent_id: Option<Option<Uuid>>,

7075}

7176

7277#[derive(Debug, Clone, Deserialize, ToSchema)]

@@ -327,7 +332,8 @@

327332 || cmd.position.is_some()

328333 || cmd.topic.is_some()

329334 || cmd.nsfw.is_some()

330 || cmd.rate_limit_per_user.is_some());

335 || cmd.rate_limit_per_user.is_some()

336 || cmd.parent_id.is_some());

331337 if needs_manage {

332338 match &ctx.space {

333339 Some(space) => {

@@ -390,6 +396,31 @@

390396 if let Some(position) = cmd.position {

391397 am.position = Set(position);

392398 }

399 // The sidebar's drag-and-drop files channels into folders, so the target

400 // must be a real directory node of *this* room's space (never another

401 // space's, and never an object node such as a repo placeholder).

402 if let Some(parent) = cmd.parent_id {

403 let next = match parent {

404 None => None,

405 Some(dir_id) => {

406 let space_id = ctx.space.as_ref().map(|space| space.id).ok_or(ChatError::Forbidden)?;

407 let node = dirs::Entity::find_by_id(dir_id)

408 .filter(dirs::Column::SpaceId.eq(space_id))

409 .one(self.db().read())

410 .await?

411 .ok_or(ChatError::NotFound)?;

412 if node.kind != DirKind::Dir {

413 return Err(ChatError::BadRequest(

414 "room parent must be a directory node".to_string(),

415 ));

416 }

417 Some(dir_id)

418 }

419 };

420 if next != ctx.room.parent_id {

421 am.parent_id = Set(next);

422 }

423 }

393424 am.updated_at = Set(now);

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

395426

@@ -753,8 +784,10 @@

753784 .await?;

754785 if let Some(invitee) = cmd.invitee_id {

755786 let data = serde_json::json!({"room_id": room_id, "invite_id": id});

787 let pending = export::notify::PendingNotifications::default();

756788 notify_user(

757789 self.db().write(),

790 &pending,

758791 invitee,

759792 NotificationType::Invite,

760793 "You were invited to a channel".to_string(),

@@ -763,6 +796,7 @@

763796 Some(caller),

764797 )

765798 .await?;

799 pending.flush(&self.state.redis).await;

766800 }

767801 let mut uids = vec![row.inviter_id];

768802 if let Some(invitee) = row.invitee_user_id {

Modified lib/chat/src/share/notify.rs +7 −1 View lib/chat/src/share/notify.rs

@@ -65,8 +65,10 @@

6565 Ok(())

6666}

6767

68#[allow(clippy::too_many_arguments)]

6869pub async fn notify_user(

6970 db: &impl ConnectionTrait,

71 pending: &export::notify::PendingNotifications,

7072 recipient: Uuid,

7173 ntype: NotificationType,

7274 title: String,

@@ -74,7 +76,7 @@

7476 data: Option<serde_json::Value>,

7577 actor_uid: Option<Uuid>,

7678) -> Result<(), ChatError> {

77 user_notifications::ActiveModel {

79 let row = user_notifications::ActiveModel {

7880 id: Set(Uuid::now_v7()),

7981 user_id: Set(recipient),

8082 notification_type: Set(ntype),

@@ -87,6 +89,7 @@

8789 }

8890 .insert(db)

8991 .await?;

92 pending.push(recipient, row.to_event_value());

9093 Ok(())

9194}

9295

@@ -200,6 +203,7 @@

200203 "message_id": message_id,

201204 "everyone": true,

202205 });

206 let pending = export::notify::PendingNotifications::default();

203207 for uid in candidates {

204208 if uid == actor || explicit.contains(&uid) {

205209 continue;

@@ -220,6 +224,7 @@

220224 .await;

221225 let _ = notify_user(

222226 service.db().write(),

227 &pending,

223228 uid,

224229 NotificationType::Mention,

225230 "You were mentioned in a message".to_string(),

@@ -236,6 +241,7 @@

236241 )

237242 .await;

238243 }

244 pending.flush(&service.state.redis).await;

239245}

240246

241247/// Mention e-mails honour per-room / per-space mute and notification level:

Modified lib/chat/src/share/realtime.rs +1 −1 View lib/chat/src/share/realtime.rs

@@ -13,7 +13,7 @@

1313}

1414

1515pub fn user_channel(uid: Uuid) -> String {

16 format!("chat:user:{uid}")

16 export::notify::user_channel(uid)

1717}

1818

1919#[derive(Debug, Clone, Serialize, Deserialize)]

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

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

22pub mod error;

33pub mod health;

44pub mod middleware;

5pub mod notify;

56pub mod pagination;

67pub mod redis;

78pub mod rest;

Added lib/export/src/notify.rs +65 −0 View lib/export/src/notify.rs

@@ -0,0 +1,65 @@

1use std::sync::{Arc, Mutex};

2

3use serde_json::Value;

4use uuid::Uuid;

5

6use crate::redis::AppRedis;

7

8/// The chat WS per-user channel; every crate that writes a notification

9/// publishes to it so connected clients receive the row in real time.

10pub fn user_channel(uid: Uuid) -> String {

11 format!("chat:user:{uid}")

12}

13

14pub fn notification_event_json(notification: Value) -> Value {

15 serde_json::json!({

16 "kind": "notification",

17 "notification": notification,

18 })

19}

20

21pub async fn publish_user_event(redis: &AppRedis, uid: Uuid, event: &Value) {

22 let payload = match serde_json::to_string(event) {

23 Ok(payload) => payload,

24 Err(error) => {

25 tracing::warn!(%error, %uid, "user event serialize failed");

26 return;

27 }

28 };

29 let channel = user_channel(uid);

30 if let Err(error) = redis.publish(&channel, &payload).await {

31 tracing::warn!(%error, %channel, "user event publish failed");

32 }

33}

34

35pub async fn publish_notification(redis: &AppRedis, uid: Uuid, notification: Value) {

36 let event = notification_event_json(notification);

37 publish_user_event(redis, uid, &event).await;

38}

39

40/// Notifications created inside a DB transaction; flush after commit.

41#[derive(Clone, Default)]

42pub struct PendingNotifications {

43 inner: Arc<Mutex<Vec<(Uuid, Value)>>>,

44}

45

46impl PendingNotifications {

47 pub fn push(&self, uid: Uuid, notification: Value) {

48 match self.inner.lock() {

49 Ok(mut pending) => pending.push((uid, notification_event_json(notification))),

50 Err(_) => {

51 tracing::warn!("pending notifications lock poisoned");

52 }

53 }

54 }

55

56 pub async fn flush(&self, redis: &AppRedis) {

57 let batch: Vec<(Uuid, Value)> = match self.inner.lock() {

58 Ok(mut pending) => std::mem::take(&mut *pending),

59 Err(_) => return,

60 };

61 for (uid, event) in batch {

62 publish_user_event(redis, uid, &event).await;

63 }

64 }

65}

Modified lib/export/src/redis.rs +21 −0 View lib/export/src/redis.rs

@@ -111,6 +111,27 @@

111111 self.connections[index].clone()

112112 }

113113

114 pub async fn publish(&self, channel: &str, payload: &str) -> Result<(), ExportError> {

115 let mut conn = self.connection_clone();

116 let result: Result<i64, redis::RedisError> = match &mut conn {

117 RedisConnection::Standalone(c) => {

118 redis::cmd("PUBLISH")

119 .arg(channel)

120 .arg(payload)

121 .query_async(c)

122 .await

123 }

124 RedisConnection::Cluster(c) => {

125 redis::cmd("PUBLISH")

126 .arg(channel)

127 .arg(payload)

128 .query_async(c)

129 .await

130 }

131 };

132 result.map(|_| ()).map_err(ExportError::from)

133 }

134

114135 /// `SET key 1 NX EX ttl`; returns false when the key already exists

115136 /// (already consumed / inside an active window) or redis fails.

116137 pub async fn set_nx_ex(&self, key: &str, ttl_secs: u64) -> bool {

Modified lib/git/src/context/facade.rs +4 −0 View lib/git/src/context/facade.rs

@@ -22,6 +22,10 @@

2222 pub fn repo_dir(&self) -> PathBuf {

2323 self.repo_root.join(self.model.id.to_string())

2424 }

25

26 pub fn repo_uninitialized(&self) -> bool {

27 !self.repo_dir().join("HEAD").exists()

28 }

2529}

2630

2731pub fn repo_dir_for(repo_root: &std::path::Path, repo_id: Uuid) -> PathBuf {

Modified lib/git/src/service/analysis.rs +6 −0 View lib/git/src/service/analysis.rs

@@ -104,6 +104,9 @@

104104 repo_name: String,

105105 ) -> Result<Vec<LanguageStatDto>, GitError> {

106106 let ctx = self.context(space_name, repo_name).await?;

107 if ctx.repo_uninitialized() {

108 return Ok(Vec::new());

109 }

107110 let baby = self.baby(&ctx);

108111 let Some(revision) =

109112 resolve_analysis_revision(&baby, None, &ctx.model.default_branch).await?

@@ -211,6 +214,9 @@

211214 repo_name: String,

212215 ) -> Result<Vec<ContributorStatDto>, GitError> {

213216 let ctx = self.context(space_name, repo_name).await?;

217 if ctx.repo_uninitialized() {

218 return Ok(Vec::new());

219 }

214220 let baby = self.baby(&ctx);

215221 let Some(revision) =

216222 resolve_analysis_revision(&baby, None, &ctx.model.default_branch).await?

Modified lib/git/src/service/branches.rs +3 −0 View lib/git/src/service/branches.rs

@@ -63,6 +63,9 @@

6363 repo_name: String,

6464 ) -> Result<Vec<BranchDto>, GitError> {

6565 let ctx = self.context(space_name, repo_name).await?;

66 if ctx.repo_uninitialized() {

67 return Ok(Vec::new());

68 }

6669 let baby = self.baby(&ctx);

6770 let branches = baby

6871 .list_branches(ListBranchesOptions::default())

Modified lib/git/src/service/tags.rs +3 −0 View lib/git/src/service/tags.rs

@@ -216,6 +216,9 @@

216216 repo_name: String,

217217 ) -> Result<Vec<TagDto>, GitError> {

218218 let ctx = self.context(space_name, repo_name).await?;

219 if ctx.repo_uninitialized() {

220 return Ok(Vec::new());

221 }

219222 let baby = self.baby(&ctx);

220223 let tags = baby

221224 .list_tags(ListTagsOptions {

Modified lib/git/src/service/trees.rs +6 −0 View lib/git/src/service/trees.rs

@@ -190,6 +190,9 @@

190190 revision: &str,

191191 ) -> Result<Vec<gitbaby::tree::types::TreeEntryWithLatest>, GitError> {

192192 validate_revision(revision)?;

193 if ctx.repo_uninitialized() {

194 return Ok(Vec::new());

195 }

193196 let baby = self.baby(ctx);

194197 let rev_oid = baby.rev_parse(revision).await.map_err(map_baby)?;

195198 let key = format!("repo.{}.tree_entries_latest_html.{}", ctx.model.id, rev_oid);

@@ -215,6 +218,9 @@

215218 {

216219 return Err(GitError::InvalidArgument("invalid path".into()));

217220 }

221 if ctx.repo_uninitialized() {

222 return Ok(Vec::new());

223 }

218224 let baby = self.baby(ctx);

219225 let rev_oid = baby.rev_parse(revision).await.map_err(map_baby)?;

220226 let key = format!(

Modified lib/issues/src/services/comments.rs +5 −0 View lib/issues/src/services/comments.rs

@@ -129,6 +129,8 @@

129129 let space_lower_for_txn = space.lower_name.clone();

130130 let item_title_for_txn = item.title.clone();

131131 let mentioned_for_txn = mentioned_uids.clone();

132 let pending = export::notify::PendingNotifications::default();

133 let pending_txn = pending.clone();

132134

133135 let inserted = self

134136 .db()

@@ -174,6 +176,7 @@

174176 });

175177 this.notify_user(

176178 txn,

179 &pending_txn,

177180 pp_uid,

178181 NotificationType::Reply,

179182 format!("New reply on issue #{issue_num}: {issue_title_clone}"),

@@ -216,6 +219,7 @@

216219 });

217220 this.notify_user(

218221 txn,

222 &pending_txn,

219223 m_uid,

220224 NotificationType::Mention,

221225 format!(

@@ -232,6 +236,7 @@

232236 })

233237 })

234238 .await?;

239 pending.flush(&self.state.redis).await;

235240

236241 let mut email_recipients = Vec::new();

237242 if let Some(pp_uid) = parent_poster_id {

Modified lib/issues/src/services/issues.rs +4 −0 View lib/issues/src/services/issues.rs

@@ -393,6 +393,8 @@

393393 let space_lower_for_txn = space.lower_name.clone();

394394 let mentioned_for_txn = mentioned_uids.clone();

395395 let this = self.clone();

396 let pending = export::notify::PendingNotifications::default();

397 let pending_txn = pending.clone();

396398

397399 let inserted = self

398400 .db()

@@ -497,6 +499,7 @@

497499 });

498500 this.notify_user(

499501 txn,

502 &pending_txn,

500503 m_uid,

501504 NotificationType::Mention,

502505 format!("You were mentioned in issue #{next_num}: {title_clone}"),

@@ -511,6 +514,7 @@

511514 })

512515 })

513516 .await?;

517 pending.flush(&self.state.redis).await;

514518

515519 if !mentioned_uids.is_empty() {

516520 let emails = self.primary_emails(&mentioned_uids).await?;

Modified lib/issues/src/services/members.rs +4 −0 View lib/issues/src/services/members.rs

@@ -96,6 +96,8 @@

9696 let this = self.clone();

9797 let item_title_for_txn = item.title.clone();

9898 let space_lower_for_txn = space.lower_name.clone();

99 let pending = export::notify::PendingNotifications::default();

100 let pending_txn = pending.clone();

99101

100102 let inserted = self

101103 .db()

@@ -137,6 +139,7 @@

137139 });

138140 this.notify_user(

139141 txn,

142 &pending_txn,

140143 target_uid,

141144 NotificationType::Assignment,

142145 format!("You were assigned to issue #{issue_num}: {issue_title_clone}"),

@@ -151,6 +154,7 @@

151154 })

152155 })

153156 .await?;

157 pending.flush(&self.state.redis).await;

154158

155159 if kind_enum == IssueMemberKind::Assignee && target_uid != caller {

156160 let emails = self.primary_emails(&[target_uid]).await?;

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

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

116116 pub(crate) async fn notify_user(

117117 &self,

118118 txn: &impl ConnectionTrait,

119 pending: &export::notify::PendingNotifications,

119120 recipient: Uuid,

120121 ntype: NotificationType,

121122 title: String,

@@ -123,7 +124,7 @@

123124 data: Option<serde_json::Value>,

124125 actor_uid: Option<Uuid>,

125126 ) -> Result<(), IssueError> {

126 user_notifications::ActiveModel {

127 let row = user_notifications::ActiveModel {

127128 id: Set(Uuid::now_v7()),

128129 user_id: Set(recipient),

129130 notification_type: Set(ntype),

@@ -136,6 +137,7 @@

136137 }

137138 .insert(txn)

138139 .await?;

140 pending.push(recipient, row.to_event_value());

139141 Ok(())

140142 }

141143

Modified lib/models/src/users/user_notifications.rs +30 −0 View lib/models/src/users/user_notifications.rs

@@ -55,6 +55,36 @@

5555

5656impl ActiveModelBehavior for ActiveModel {}

5757

58impl Model {

59 /// Canonical payload for the WS `notification` user event; mirrors the

60 /// `NotificationDto` returned by `GET /me/notifications`.

61 pub fn to_event_value(&self) -> JsonValue {

62 #[derive(serde::Serialize)]

63 struct EventPayload<'a> {

64 id: Uuid,

65 notification_type: &'a NotificationType,

66 title: &'a str,

67 body: &'a Option<String>,

68 data: &'a Option<JsonValue>,

69 actor_id: &'a Option<Uuid>,

70 read_at: &'a Option<OffsetDateTime>,

71 created_at: &'a OffsetDateTime,

72 }

73

74 serde_json::to_value(EventPayload {

75 id: self.id,

76 notification_type: &self.notification_type,

77 title: &self.title,

78 body: &self.body,

79 data: &self.data_json,

80 actor_id: &self.actor_uid,

81 read_at: &self.read_at,

82 created_at: &self.created_at,

83 })

84 .unwrap_or(serde_json::Value::Null)

85 }

86}

87

5888impl Related<super::user::Entity> for Entity {

5989 fn to() -> RelationDef {

6090 Relation::User.def()

Modified lib/space/src/services/invites.rs +29 −0 View lib/space/src/services/invites.rs

@@ -9,6 +9,8 @@

99use models::space::spaces;

1010use models::users::user;

1111use models::users::user_emails;

12use models::users::user_notifications;

13use models::users::user_notifications::NotificationType;

1214use sea_orm::sea_query::Expr;

1315use sea_orm::{

1416 ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,

@@ -241,6 +243,33 @@

241243 })

242244 .await?;

243245

246 if let Some(invitee) = inserted.user_id {

247 let data = serde_json::json!({

248 "space_id": space.id,

249 "space_name": space.name,

250 "invite_id": invite_id,

251 });

252 let row = user_notifications::ActiveModel {

253 id: Set(Uuid::now_v7()),

254 user_id: Set(invitee),

255 notification_type: Set(NotificationType::Invite),

256 title: Set(format!("You were invited to join {}", space.name)),

257 body: Set(None),

258 data_json: Set(Some(data)),

259 actor_uid: Set(Some(caller)),

260 read_at: Set(None),

261 created_at: Set(OffsetDateTime::now_utc()),

262 }

263 .insert(self.db().write())

264 .await?;

265 export::notify::publish_notification(

266 &self.state.redis,

267 invitee,

268 row.to_event_value(),

269 )

270 .await;

271 }

272

244273 let mut recipient_email: Option<String> = email_norm;

245274 if recipient_email.is_none()

246275 && let Some(uid) = target_uid