feat(reminders): add delivering status for message reminders and enhance scheduled message handling

Modified lib/chat/src/handlers/bookmarks.rs +3 −1 View lib/chat/src/handlers/bookmarks.rs

@@ -71,7 +71,9 @@

7171 Json(cmd): Json<UpdateBookmarkCmd>,

7272) -> Result<AppResponseData<BookmarkDto>, ApiError> {

7373 let uid = require_uid(&session)?;

74 let dto = service.update_bookmark(uid, room_id, bookmark_id, cmd).await?;

74 let dto = service

75 .update_bookmark(uid, room_id, bookmark_id, cmd)

76 .await?;

7577 Ok(AppResponseData::ok(dto))

7678}

7779

Modified lib/chat/src/handlers/messages.rs +6 −2 View lib/chat/src/handlers/messages.rs

@@ -71,7 +71,9 @@

7171 Path(path): Path<MessagePath>,

7272) -> Result<AppResponseData<MessageDto>, ApiError> {

7373 let uid = require_uid(&session)?;

74 let dto = service.get_message(uid, path.room_id, path.message_id).await?;

74 let dto = service

75 .get_message(uid, path.room_id, path.message_id)

76 .await?;

7577 Ok(AppResponseData::ok(dto))

7678}

7779

@@ -115,7 +117,9 @@

115117 Path(path): Path<MessagePath>,

116118) -> Result<AppResponseData<NoData>, ApiError> {

117119 let uid = require_uid(&session)?;

118 service.delete_message(uid, path.room_id, path.message_id).await?;

120 service

121 .delete_message(uid, path.room_id, path.message_id)

122 .await?;

119123 Ok(AppResponseData::empty())

120124}

121125

Modified lib/chat/src/handlers/mod.rs +1 −1 View lib/chat/src/handlers/mod.rs

@@ -1,7 +1,7 @@

11pub mod attachments;

22pub mod bookmarks;

3pub mod dms;

34pub mod drafts;

4pub mod dms;

55pub mod emojis;

66pub mod messages;

77pub mod presence;

Modified lib/chat/src/handlers/reactions.rs +6 −2 View lib/chat/src/handlers/reactions.rs

@@ -54,7 +54,9 @@

5454 Json(cmd): Json<ReactionCmd>,

5555) -> Result<AppResponseData<Vec<ReactionAggDto>>, ApiError> {

5656 let uid = require_uid(&session)?;

57 let agg = service.remove_reaction(uid, room_id, message_id, cmd).await?;

57 let agg = service

58 .remove_reaction(uid, room_id, message_id, cmd)

59 .await?;

5860 Ok(AppResponseData::ok(agg))

5961}

6062

@@ -77,6 +79,8 @@

7779 Path((room_id, message_id)): Path<(Uuid, Uuid)>,

7880) -> Result<AppResponseData<Vec<ReactionUserDto>>, ApiError> {

7981 let uid = require_uid(&session)?;

80 let rows = service.list_message_reactions(uid, room_id, message_id).await?;

82 let rows = service

83 .list_message_reactions(uid, room_id, message_id)

84 .await?;

8185 Ok(AppResponseData::ok(rows))

8286}

Modified lib/chat/src/handlers/read_state.rs +3 −1 View lib/chat/src/handlers/read_state.rs

@@ -92,6 +92,8 @@

9292 Query(q): Query<MarkThreadReadQuery>,

9393) -> Result<AppResponseData<NoData>, ApiError> {

9494 let uid = require_uid(&session)?;

95 service.mark_thread_read(uid, thread_id, q.message_id).await?;

95 service

96 .mark_thread_read(uid, thread_id, q.message_id)

97 .await?;

9698 Ok(AppResponseData::empty())

9799}

Modified lib/chat/src/handlers/reminders.rs +3 −1 View lib/chat/src/handlers/reminders.rs

@@ -29,7 +29,9 @@

2929 Json(cmd): Json<CreateReminderCmd>,

3030) -> Result<AppResponseData<ReminderDto>, ApiError> {

3131 let uid = require_uid(&session)?;

32 let dto = service.create_reminder(uid, room_id, message_id, cmd).await?;

32 let dto = service

33 .create_reminder(uid, room_id, message_id, cmd)

34 .await?;

3335 Ok(AppResponseData::ok(dto))

3436}

3537

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

@@ -1,7 +1,7 @@

11pub mod attachments;

22pub mod bookmarks;

3pub mod dms;

34pub mod drafts;

4pub mod dms;

55pub mod emojis;

66pub mod messages;

77pub mod presence;

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

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

22use std::sync::Arc;

33

44use serde::{Deserialize, Serialize};

5use tokio::sync::{broadcast, Mutex};

5use tokio::sync::{Mutex, broadcast};

66use tokio_util::sync::CancellationToken;

77use uuid::Uuid;

88

@@ -155,16 +155,20 @@

155155 };

156156 let mut conn = self.inner.redis.connection_clone();

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

158 export::RedisConnection::Standalone(c) => redis::cmd("PUBLISH")

159 .arg(channel)

160 .arg(payload.as_slice())

161 .query_async(c)

162 .await,

163 export::RedisConnection::Cluster(c) => redis::cmd("PUBLISH")

164 .arg(channel)

165 .arg(payload.as_slice())

166 .query_async(c)

167 .await,

158 export::RedisConnection::Standalone(c) => {

159 redis::cmd("PUBLISH")

160 .arg(channel)

161 .arg(payload.as_slice())

162 .query_async(c)

163 .await

164 }

165 export::RedisConnection::Cluster(c) => {

166 redis::cmd("PUBLISH")

167 .arg(channel)

168 .arg(payload.as_slice())

169 .query_async(c)

170 .await

171 }

168172 };

169173 if let Err(error) = result {

170174 tracing::warn!(%error, %channel, "chat realtime publish failed");

@@ -191,9 +195,7 @@

191195 let remove = map

192196 .get(&room_id)

193197 .is_some_and(|entry| entry.tx.receiver_count() == 0);

194 if remove

195 && let Some(entry) = map.remove(&room_id)

196 {

198 if remove && let Some(entry) = map.remove(&room_id) {

197199 entry.cancel.cancel();

198200 }

199201 }

@@ -215,10 +217,10 @@

215217

216218 pub async fn unsubscribe_user(&self, uid: Uuid) {

217219 let mut map = self.inner.users.lock().await;

218 let remove = map.get(&uid).is_some_and(|entry| entry.tx.receiver_count() == 0);

219 if remove

220 && let Some(entry) = map.remove(&uid)

221 {

220 let remove = map

221 .get(&uid)

222 .is_some_and(|entry| entry.tx.receiver_count() == 0);

223 if remove && let Some(entry) = map.remove(&uid) {

222224 entry.cancel.cancel();

223225 }

224226 }

Modified lib/chat/src/ws/ticket.rs +9 −6 View lib/chat/src/ws/ticket.rs

@@ -18,15 +18,17 @@

1818

1919pub fn issue_ticket(secret: &str, uid: Uuid, ttl_secs: u64) -> Result<String, ChatError> {

2020 if secret.is_empty() {

21 return Err(ChatError::Internal("chat ticket_secret not configured".into()));

21 return Err(ChatError::Internal(

22 "chat ticket_secret not configured".into(),

23 ));

2224 }

2325 let payload = TicketPayload {

2426 uid,

2527 sid: Uuid::now_v7(),

2628 exp: OffsetDateTime::now_utc().unix_timestamp() + ttl_secs as i64,

2729 };

28 let payload_b64 = base64::engine::general_purpose::URL_SAFE_NO_PAD

29 .encode(serde_json::to_vec(&payload)?);

30 let payload_b64 =

31 base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(serde_json::to_vec(&payload)?);

3032 let mut mac = HmacSha256::new_from_slice(secret.as_bytes())

3133 .map_err(|e| ChatError::Internal(format!("hmac init failed: {e}")))?;

3234 mac.update(payload_b64.as_bytes());

@@ -41,13 +43,14 @@

4143 let Some((payload_b64, sig_b64)) = ticket.split_once('.') else {

4244 return Err(ChatError::Unauthorized);

4345 };

44 let mut mac = HmacSha256::new_from_slice(secret.as_bytes())

45 .map_err(|_| ChatError::Unauthorized)?;

46 let mut mac =

47 HmacSha256::new_from_slice(secret.as_bytes()).map_err(|_| ChatError::Unauthorized)?;

4648 mac.update(payload_b64.as_bytes());

4749 let sig = base64::engine::general_purpose::URL_SAFE_NO_PAD

4850 .decode(sig_b64)

4951 .map_err(|_| ChatError::Unauthorized)?;

50 mac.verify_slice(&sig).map_err(|_| ChatError::Unauthorized)?;

52 mac.verify_slice(&sig)

53 .map_err(|_| ChatError::Unauthorized)?;

5154

5255 let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD

5356 .decode(payload_b64)

Modified lib/chat/src/routes.rs +8 −13 View lib/chat/src/routes.rs

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

77use crate::handlers::bookmarks::{

88 create_bookmark, delete_bookmark, list_bookmarks, update_bookmark,

99};

10use crate::handlers::dms::{close_dm, list_dms, open_dm};

1011use crate::handlers::drafts::{get_draft, list_my_drafts, upsert_draft};

11use crate::handlers::dms::{close_dm, list_dms, open_dm};

1212use crate::handlers::emojis::{delete_emoji, emoji_image, list_emojis, upload_emoji};

1313use crate::handlers::messages::{

1414 bulk_delete_messages, delete_message, forward_message, get_message, list_message_history,

@@ -29,10 +29,8 @@

2929use crate::handlers::threads::{

3030 create_thread, follow_thread, get_thread, list_my_threads, list_threads, unfollow_thread,

3131 update_thread,

32};

33use crate::handlers::webhooks::{

34 create_webhook, delete_webhook, list_webhooks, post_via_webhook,

3532};

33use crate::handlers::webhooks::{create_webhook, delete_webhook, list_webhooks, post_via_webhook};

3634use crate::handlers::ws::issue_ws_ticket;

3735use crate::services::service::ChatService;

3836use crate::share::guard::chat_guard;

@@ -41,10 +39,7 @@

4139pub fn router(service: ChatService) -> Router<ChatService> {

4240 let guarded = Router::new()

4341 .route("/chat/spaces/{space_name}/rooms", get(list_space_rooms))

44 .route(

45 "/chat/rooms/{room_id}",

46 get(get_room).patch(update_room),

47 )

42 .route("/chat/rooms/{room_id}", get(get_room).patch(update_room))

4843 .route(

4944 "/chat/rooms/{room_id}/members/@me",

5045 axum::routing::put(join_room).delete(leave_room),

@@ -54,8 +49,11 @@

5449 get(list_room_members).post(add_room_members),

5550 )

5651 .route("/chat/rooms/{room_id}/invites", post(create_room_invite))

57 .route("/chat/invites/{accept_code}/accept", post(accept_room_invite))

5852 .route(

53 "/chat/invites/{accept_code}/accept",

54 post(accept_room_invite),

55 )

56 .route(

5957 "/chat/rooms/{room_id}/messages",

6058 get(list_messages).post(send_message),

6159 )

@@ -128,10 +126,7 @@

128126 .route("/chat/dms", get(list_dms).post(open_dm))

129127 .route("/chat/dms/{room_id}", axum::routing::delete(close_dm))

130128 .route("/chat/attachments", post(create_attachment))

131 .route(

132 "/chat/attachments/{attachment_id}",

133 get(get_attachment),

134 )

129 .route("/chat/attachments/{attachment_id}", get(get_attachment))

135130 .route(

136131 "/chat/attachments/{attachment_id}/finalize",

137132 post(finalize_attachment),