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

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

@@ -43,8 +43,8 @@

4343 session: Session,

4444 Path(user_id): Path<Uuid>,

4545) -> Result<AppResponseData<PresenceDto>, ApiError> {

46 let _uid = require_uid(&session)?;

47 let dto = service.get_presence(user_id).await?;

46 let uid = require_uid(&session)?;

47 let dto = service.get_presence_for(uid, user_id).await?;

4848 Ok(AppResponseData::ok(dto))

4949}

5050

Modified lib/chat/src/services/attachments.rs +78 −19 View lib/chat/src/services/attachments.rs

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

99

1010use crate::errors::ChatError;

1111use crate::services::service::ChatService;

12use crate::share::lookup::{is_room_member, lookup_room};

1312

1413pub const ATTACHMENT_PREFIX: &str = "chat/attachments";

1514

@@ -92,6 +91,37 @@

9291 }

9392}

9493

94/// Content types that browsers may execute/script inline; never accepted or

95/// served embedded.

96const UNSAFE_INLINE_TYPES: &[&str] = &[

97 "text/html",

98 "application/xhtml+xml",

99 "image/svg+xml",

100 "text/xml",

101 "application/xml",

102 "text/javascript",

103 "application/javascript",

104 "application/x-javascript",

105 "application/ecmascript",

106];

107

108fn is_unsafe_inline(content_type: &str) -> bool {

109 let main = content_type.split(';').next().unwrap_or("").trim();

110 UNSAFE_INLINE_TYPES

111 .iter()

112 .any(|blocked| main.eq_ignore_ascii_case(blocked))

113}

114

115fn is_inline_renderable(content_type: &str) -> bool {

116 let main = content_type.split(';').next().unwrap_or("").trim();

117 main.starts_with("image/")

118 || main.starts_with("video/")

119 || main.starts_with("audio/")

120 || main.eq_ignore_ascii_case("text/plain")

121 || main.eq_ignore_ascii_case("text/markdown")

122 || main.eq_ignore_ascii_case("application/pdf")

123}

124

95125impl ChatService {

96126 pub async fn create_attachment(

97127 &self,

@@ -107,6 +137,11 @@

107137 if cmd.filename.trim().is_empty() || cmd.filename.chars().count() > 255 {

108138 return Err(ChatError::BadRequest("invalid filename".to_string()));

109139 }

140 if cmd.content_type.trim().is_empty() || is_unsafe_inline(&cmd.content_type) {

141 return Err(ChatError::BadRequest(

142 "unsupported or unsafe content_type".to_string(),

143 ));

144 }

110145

111146 let now = OffsetDateTime::now_utc();

112147 let id = Uuid::now_v7();

@@ -137,10 +172,14 @@

137172 .await?;

138173

139174 let ttl = self.config().chat.upload.presign_ttl_secs.min(604_800);

175 let mut put_headers = http::HeaderMap::new();

176 if let Ok(value) = cmd.content_type.parse() {

177 put_headers.insert(http::header::CONTENT_TYPE, value);

178 }

140179 let upload_url = self

141180 .storage()

142181 .client()

143 .presign_put(&key, ttl as u32)

182 .presign_put_with_headers(&key, ttl as u32, put_headers)

144183 .map_err(ChatError::from)?;

145184

146185 let _ = row;

@@ -163,13 +202,13 @@

163202 .filter(|a| a.user_id == caller)

164203 .ok_or(ChatError::NotFound)?;

165204 if row.status != AttachmentStatus::Pending {

166 return Err(ChatError::Conflict("attachment already finalized".to_string()));

205 return Err(ChatError::Conflict(

206 "attachment already finalized".to_string(),

207 ));

167208 }

168209 let exists = self.storage().client().exists(&row.s3_key).await?;

169210 if !exists {

170 return Err(ChatError::BadRequest(

171 "object not uploaded yet".to_string(),

172 ));

211 return Err(ChatError::BadRequest("object not uploaded yet".to_string()));

173212 }

174213 let now = OffsetDateTime::now_utc();

175214 let mut am: attachments::ActiveModel = row.into();

@@ -224,22 +263,42 @@

224263 if row.user_id == caller {

225264 return Ok(());

226265 }

227 if let Some(room_id) = row.room_id

228 && lookup_room(self, room_id).await.is_ok()

229 && is_room_member(self, room_id, caller).await?

230 {

231 return Ok(());

266 if let Some(room_id) = row.room_id {

267 // Mirror message visibility: anyone who can read the room can read

268 // its attachments (public rooms do not require a membership row).

269 if crate::share::lookup::require_room_readable(self, room_id, caller)

270 .await

271 .is_ok()

272 {

273 return Ok(());

274 }

232275 }

233276 Err(ChatError::Forbidden)

234277 }

235278

236 pub fn attachment_content_headers(row: &attachments::Model) -> [(header::HeaderName, String); 2] {

237 [

238 (header::CONTENT_TYPE, row.content_type.clone()),

239 (

240 header::CONTENT_DISPOSITION,

241 format!("inline; filename=\"{}\"", row.filename.replace('"', "_")),

242 ),

243 ]

279 pub fn attachment_content_headers(

280 row: &attachments::Model,

281 ) -> Vec<(header::HeaderName, String)> {

282 let mut headers = Vec::new();

283 let content_type = if is_unsafe_inline(&row.content_type) {

284 "application/octet-stream".to_string()

285 } else {

286 row.content_type.clone()

287 };

288 headers.push((header::CONTENT_TYPE, content_type));

289 headers.push((

290 header::HeaderName::from_static("x-content-type-options"),

291 "nosniff".to_string(),

292 ));

293 let disposition = if is_inline_renderable(&row.content_type) {

294 format!("inline; filename=\"{}\"", row.filename.replace('"', "_"))

295 } else {

296 format!(

297 "attachment; filename=\"{}\"",

298 row.filename.replace('"', "_")

299 )

300 };

301 headers.push((header::CONTENT_DISPOSITION, disposition));

302 headers

244303 }

245304}

Modified lib/chat/src/services/bookmarks.rs +24 −8 View lib/chat/src/services/bookmarks.rs

@@ -128,12 +128,15 @@

128128 bookmark_id: Uuid,

129129 cmd: UpdateBookmarkCmd,

130130 ) -> Result<BookmarkDto, ChatError> {

131 let _ctx = require_room_readable(self, room_id, caller).await?;

131 let ctx = require_room_readable(self, room_id, caller).await?;

132132 let row = room_bookmarks::Entity::find_by_id(bookmark_id)

133133 .filter(room_bookmarks::Column::RoomId.eq(room_id))

134134 .one(self.db().write())

135135 .await?

136136 .ok_or(ChatError::NotFound)?;

137 if row.user_id != caller && !self.bookmark_manager(&ctx, caller).await? {

138 return Err(ChatError::Forbidden);

139 }

137140 let mut am: room_bookmarks::ActiveModel = row.into();

138141 if let Some(title) = cmd.title {

139142 let title = title.trim().to_string();

@@ -163,16 +166,29 @@

163166 room_id: Uuid,

164167 bookmark_id: Uuid,

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

166 let _ctx = require_room_readable(self, room_id, caller).await?;

167 let deleted = room_bookmarks::Entity::delete_many()

168 .filter(room_bookmarks::Column::Id.eq(bookmark_id))

169 let ctx = require_room_readable(self, room_id, caller).await?;

170 let row = room_bookmarks::Entity::find_by_id(bookmark_id)

169171 .filter(room_bookmarks::Column::RoomId.eq(room_id))

172 .one(self.db().write())

173 .await?

174 .ok_or(ChatError::NotFound)?;

175 if row.user_id != caller && !self.bookmark_manager(&ctx, caller).await? {

176 return Err(ChatError::Forbidden);

177 }

178 room_bookmarks::Entity::delete_by_id(row.id)

170179 .exec(self.db().write())

171180 .await?;

172 if deleted.rows_affected == 0 {

173 return Err(ChatError::NotFound);

181 Ok(())

182 }

183

184 async fn bookmark_manager(

185 &self,

186 ctx: &crate::share::lookup::RoomContext,

187 caller: Uuid,

188 ) -> Result<bool, ChatError> {

189 match &ctx.space {

190 Some(space) => crate::share::authz::can_manage_space(self, space, caller).await,

191 None => Ok(false),

174192 }

175 let _ = caller;

176 Ok(())

177193 }

178194}

Modified lib/chat/src/services/dms.rs +50 −16 View lib/chat/src/services/dms.rs

@@ -40,7 +40,9 @@

4040impl ChatService {

4141 pub async fn open_dm(&self, caller: Uuid, cmd: OpenDmCmd) -> Result<DmDto, ChatError> {

4242 if cmd.peer_user_id == caller {

43 return Err(ChatError::BadRequest("cannot open DM with self".to_string()));

43 return Err(ChatError::BadRequest(

44 "cannot open DM with self".to_string(),

45 ));

4446 }

4547 let peer = user::Entity::find_by_id(cmd.peer_user_id)

4648 .one(self.db().read())

@@ -141,25 +143,57 @@

141143 .filter(rooms::Column::SpaceId.is_null())

142144 .all(self.db().read())

143145 .await?;

146 let dm_room_ids: Vec<Uuid> = rooms.iter().map(|r| r.id).collect();

147

148 let peer_rows = room_members::Entity::find()

149 .filter(room_members::Column::RoomId.is_in(dm_room_ids.clone()))

150 .filter(room_members::Column::UserId.ne(caller))

151 .filter(room_members::Column::LeftAt.is_null())

152 .all(self.db().read())

153 .await?;

154 let peer_uids: Vec<Uuid> = peer_rows.iter().map(|m| m.user_id).collect();

155 let user_map = self.load_user_briefs(&peer_uids).await?;

156 let mut peer_by_room: std::collections::HashMap<Uuid, Option<UserBriefDto>> =

157 std::collections::HashMap::new();

158 for m in peer_rows {

159 peer_by_room

160 .entry(m.room_id)

161 .or_insert_with(|| user_map.get(&m.user_id).cloned());

162 }

163

164 let tips = room_seq::Entity::find()

165 .filter(room_seq::Column::RoomId.is_in(dm_room_ids.clone()))

166 .all(self.db().read())

167 .await?;

168 let tip_map: std::collections::HashMap<Uuid, i64> =

169 tips.into_iter().map(|s| (s.room_id, s.seq)).collect();

170 let reads = read_states::Entity::find()

171 .filter(read_states::Column::UserId.eq(caller))

172 .filter(read_states::Column::RoomId.is_in(dm_room_ids))

173 .all(self.db().read())

174 .await?;

175 let read_map: std::collections::HashMap<Uuid, i64> = reads

176 .into_iter()

177 .map(|r| (r.room_id, r.last_read_seq))

178 .collect();

144179

145180 let mut result = Vec::new();

146181 for room in rooms {

147 let peers = room_members::Entity::find()

148 .filter(room_members::Column::RoomId.eq(room.id))

149 .filter(room_members::Column::UserId.ne(caller))

150 .filter(room_members::Column::LeftAt.is_null())

151 .one(self.db().read())

152 .await?;

153 let peer_brief = match peers {

154 Some(peer) => user::Entity::find_by_id(peer.user_id)

155 .one(self.db().read())

156 .await?

157 .map(UserBriefDto::from),

158 None => None,

159 };

160 result.push(self.build_dm_dto(caller, room, peer_brief).await?);

182 let tip = tip_map.get(&room.id).copied().unwrap_or(0);

183 let last_read = read_map.get(&room.id).copied().unwrap_or(0);

184 result.push(DmDto {

185 id: room.id,

186 kind: room.kind,

187 name: room.name,

188 peer: peer_by_room.remove(&room.id).flatten(),

189 last_message_id: room.last_message_id,

190 last_read_seq: last_read,

191 tip_seq: tip,

192 unread_count: (tip - last_read).max(0),

193 created_at: room.created_at,

194 });

161195 }

162 result.sort_by_key(|a| std::cmp::Reverse(a.last_message_id));

196 result.sort_by_key(|a| std::cmp::Reverse(a.tip_seq));

163197 Ok(result)

164198 }

165199

Modified lib/chat/src/services/emojis.rs +9 −8 View lib/chat/src/services/emojis.rs

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

88

99use crate::errors::ChatError;

1010use crate::services::service::ChatService;

11use crate::share::authz::require_space_manager;

1112use crate::share::lookup::{lookup_space_by_name, require_space_member};

1213

1314pub const EMOJI_PREFIX: &str = "chat/emojis";

@@ -67,9 +68,7 @@

6768 ) -> Result<EmojiDto, ChatError> {

6869 let space = lookup_space_by_name(self, space_name).await?;

6970 require_space_member(self, &space, caller).await?;

70 if space.owner_id != caller {

71 return Err(ChatError::Forbidden);

72 }

71 require_space_manager(self, &space, caller).await?;

7372

7473 let name = cmd.name.trim().to_string();

7574 let lower = name.to_lowercase();

@@ -85,7 +84,9 @@

8584 cmd.content_type.as_str(),

8685 "image/png" | "image/jpeg" | "image/gif" | "image/webp"

8786 ) {

88 return Err(ChatError::BadRequest("unsupported emoji content_type".to_string()));

87 return Err(ChatError::BadRequest(

88 "unsupported emoji content_type".to_string(),

89 ));

8990 }

9091 let bytes = base64::Engine::decode(

9192 &base64::engine::general_purpose::STANDARD,

@@ -93,7 +94,9 @@

9394 )

9495 .map_err(|e| ChatError::BadRequest(format!("invalid base64 payload: {e}")))?;

9596 if bytes.is_empty() || bytes.len() > MAX_EMOJI_BYTES {

96 return Err(ChatError::BadRequest("emoji payload size invalid".to_string()));

97 return Err(ChatError::BadRequest(

98 "emoji payload size invalid".to_string(),

99 ));

97100 }

98101

99102 let duplicate = emojis::Entity::find()

@@ -153,9 +156,7 @@

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

154157 let space = lookup_space_by_name(self, space_name).await?;

155158 require_space_member(self, &space, caller).await?;

156 if space.owner_id != caller {

157 return Err(ChatError::Forbidden);

158 }

159 require_space_manager(self, &space, caller).await?;

159160 let row = emojis::Entity::find_by_id(emoji_id)

160161 .filter(emojis::Column::SpaceId.eq(space.id))

161162 .one(self.db().write())

Modified lib/chat/src/services/messages.rs +144 −71 View lib/chat/src/services/messages.rs

@@ -17,7 +17,7 @@

1717use utoipa::{IntoParams, ToSchema};

1818use uuid::Uuid;

1919

20use crate::errors::ChatError;

20use crate::errors::{ChatError, is_unique_violation};

2121use crate::services::attachments::AttachmentDto;

2222use crate::services::reactions::ReactionAggDto;

2323use crate::services::service::ChatService;

@@ -26,10 +26,13 @@

2626use crate::share::lookup::{RoomContext, require_room_readable};

2727use crate::share::mention::parse_mentions;

2828use crate::share::notify::{

29 advance_read_state, next_room_seq, notify_user, publish_message_event, record_mentions,

29 NotifyDecision, advance_read_state, current_room_seq, mention_email_allowed, next_room_seq,

30 notification_decision, notify_user, publish_message_event, record_mentions,

3031};

3132use crate::share::realtime::RoomEvent;

3233

34const NONCE_CONFLICT: &str = "chat-nonce-conflict";

35

3336#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

3437pub struct MessageDto {

3538 pub id: Uuid,

@@ -83,6 +86,8 @@

8386 pub nonce: Option<String>,

8487 #[serde(default)]

8588 pub urgent: bool,

89 #[serde(default)]

90 pub forwarded_from: Option<Uuid>,

8691}

8792

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

@@ -163,12 +168,12 @@

163168 .filter(reactions::Column::MessageId.is_in(message_ids.clone()))

164169 .all(self.db().read())

165170 .await?;

166 let mut reaction_agg: std::collections::HashMap<

167 (Uuid, String),

168 (i64, bool),

169 > = std::collections::HashMap::new();

171 let mut reaction_agg: std::collections::HashMap<(Uuid, String), (i64, bool)> =

172 std::collections::HashMap::new();

170173 for r in reaction_rows {

171 let entry = reaction_agg.entry((r.message_id, r.emoji.clone())).or_insert((0, false));

174 let entry = reaction_agg

175 .entry((r.message_id, r.emoji.clone()))

176 .or_insert((0, false));

172177 entry.0 += 1;

173178 if r.user_id == viewer {

174179 entry.1 = true;

@@ -178,8 +183,8 @@

178183 let items = rows

179184 .into_iter()

180185 .map(|m| {

181 let mentions: Vec<Uuid> = serde_json::from_value(m.mentions.clone())

182 .unwrap_or_default();

186 let mentions: Vec<Uuid> =

187 serde_json::from_value(m.mentions.clone()).unwrap_or_default();

183188 MessageDto {

184189 id: m.id,

185190 room_id: m.room_id,

@@ -219,11 +224,7 @@

219224 Ok(items)

220225 }

221226

222 async fn check_slow_mode(

223 &self,

224 ctx: &RoomContext,

225 caller: Uuid,

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

227 async fn check_slow_mode(&self, ctx: &RoomContext, caller: Uuid) -> Result<(), ChatError> {

227228 let Some(limit) = ctx.room.rate_limit_per_user else {

228229 return Ok(());

229230 };

@@ -238,14 +239,16 @@

238239 let last = messages::Entity::find()

239240 .filter(messages::Column::RoomId.eq(ctx.room.id))

240241 .filter(messages::Column::AuthorId.eq(caller))

241 .filter(messages::Column::DeletedAt.is_null())

242242 .order_by_desc(messages::Column::Seq)

243243 .one(self.db().read())

244244 .await?;

245245 if let Some(last) = last {

246246 let elapsed = OffsetDateTime::now_utc() - last.created_at;

247247 if elapsed.whole_seconds() < limit as i64 {

248 return Err(ChatError::Forbidden);

248 let retry_after = limit as i64 - elapsed.whole_seconds();

249 return Err(ChatError::RateLimited(format!(

250 "slow mode active, retry in {retry_after}s"

251 )));

249252 }

250253 }

251254 Ok(())

@@ -260,8 +263,20 @@

260263 let ctx = require_room_readable(self, room_id, caller).await?;

261264 self.check_slow_mode(&ctx, caller).await?;

262265

266 let forwarded_from = match cmd.forwarded_from {

267 Some(orig_id) => {

268 let orig = messages::Entity::find_by_id(orig_id)

269 .one(self.db().read())

270 .await?

271 .filter(|m| m.deleted_at.is_none())

272 .ok_or(ChatError::NotFound)?;

273 Some(orig.id)

274 }

275 None => None,

276 };

277

263278 let content_len = cmd.content.chars().count();

264 if cmd.content.trim().is_empty() && cmd.attachments.is_empty() {

279 if cmd.content.trim().is_empty() && cmd.attachments.is_empty() && forwarded_from.is_none() {

265280 return Err(ChatError::BadRequest(

266281 "message content is empty".to_string(),

267282 ));

@@ -273,6 +288,7 @@

273288 if let Some(nonce) = &cmd.nonce {

274289 let existing = messages::Entity::find()

275290 .filter(messages::Column::RoomId.eq(room_id))

291 .filter(messages::Column::AuthorId.eq(caller))

276292 .filter(messages::Column::Nonce.eq(nonce.as_str()))

277293 .one(self.db().read())

278294 .await?;

@@ -312,6 +328,9 @@

312328 .filter(attachment_rows::Column::Id.is_in(cmd.attachments.clone()))

313329 .filter(attachment_rows::Column::UserId.eq(caller))

314330 .filter(attachment_rows::Column::MessageId.is_null())

331 .filter(

332 attachment_rows::Column::Status.eq(attachment_rows::AttachmentStatus::Uploaded),

333 )

315334 .count(self.db().read())

316335 .await?;

317336 if owned != cmd.attachments.len() as u64 {

@@ -345,8 +364,12 @@

345364 let thread_clone = thread.clone();

346365 let mention_uids_for_txn = mention_uids.clone();

347366 let content_for_email = content.clone();

367 let svc = self.clone();

368 let space_id = room.space_id;

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

348370

349 self.db()

371 let txn_result = self

372 .db()

350373 .primary()

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

352375 Box::pin(async move {

@@ -361,7 +384,7 @@

361384 thread_id: Set(cmd.thread_id),

362385 thread_broadcast: Set(cmd.thread_broadcast),

363386 reply_to_id: Set(cmd.reply_to_id),

364 forwarded_from_id: Set(None),

387 forwarded_from_id: Set(cmd.forwarded_from),

365388 message_type: Set(message_type),

366389 content: Set(content.clone()),

367390 mentions: Set(mentions_json),

@@ -379,16 +402,22 @@

379402 seq: Set(seq),

380403 created_at: Set(now),

381404 };

382 row.insert(txn).await?;

405 match row.insert(txn).await {

406 Ok(_) => {}

407 Err(err)

408 if is_unique_violation(&err)

409 && err.to_string().contains("idx_chat_message_room_nonce") =>

410 {

411 return Err(ChatError::Conflict(NONCE_CONFLICT.to_string()));

412 }

413 Err(err) => return Err(ChatError::from(err)),

414 }

383415

384416 if !cmd.attachments.is_empty() {

385417 attachment_rows::Entity::update_many()

386418 .filter(attachment_rows::Column::Id.is_in(cmd.attachments.clone()))

387419 .filter(attachment_rows::Column::UserId.eq(caller))

388 .col_expr(

389 attachment_rows::Column::MessageId,

390 Expr::value(message_id),

391 )

420 .col_expr(attachment_rows::Column::MessageId, Expr::value(message_id))

392421 .col_expr(attachment_rows::Column::RoomId, Expr::value(room.id))

393422 .col_expr(attachment_rows::Column::UpdatedAt, Expr::value(now))

394423 .exec(txn)

@@ -420,6 +449,11 @@

420449 content.clone()

421450 };

422451 for &uid in &mention_uids_for_txn {

452 if notification_decision(&svc, room.id, space_id, uid, true).await?

453 == NotifyDecision::Skip

454 {

455 continue;

456 }

423457 notify_user(

424458 txn,

425459 uid,

@@ -435,7 +469,25 @@

435469 Ok(())

436470 })

437471 })

438 .await?;

472 .await

473 .map_err(ChatError::from);

474 if let Err(ChatError::Conflict(ref msg)) = txn_result {

475 if msg == NONCE_CONFLICT {

476 if let Some(nonce) = &nonce_retry {

477 let existing = messages::Entity::find()

478 .filter(messages::Column::RoomId.eq(room_id))

479 .filter(messages::Column::AuthorId.eq(caller))

480 .filter(messages::Column::Nonce.eq(nonce.as_str()))

481 .one(self.db().read())

482 .await?;

483 if let Some(existing) = existing {

484 let items = self.build_message_dto(vec![existing], caller).await?;

485 return Ok(items.into_iter().next().unwrap());

486 }

487 }

488 }

489 }

490 txn_result?;

439491

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

441493 .one(self.db().read())

@@ -458,33 +510,52 @@

458510 .collect();

459511 for uid in recipients {

460512 self.realtime

461 .publish_user(uid, &json!({"kind": "mention", "room_id": room_id, "message_id": message_id}))

513 .publish_user(

514 uid,

515 &json!({"kind": "mention", "room_id": room_id, "message_id": message_id}),

516 )

462517 .await;

463518 }

464519

465520 if !mention_uids.is_empty() {

466521 let room_name = ctx.room.name.clone();

467 if let Ok(emails) = self.primary_emails(&mention_uids).await {

468 for (_uid, addr) in emails {

522 let email_recipients: Vec<Uuid> = mention_uids

523 .iter()

524 .copied()

525 .filter(|uid| *uid != caller)

526 .collect();

527 let email_body = if content_for_email.chars().count() > 200 {

528 content_for_email.chars().take(200).collect::<String>()

529 } else {

530 content_for_email.clone()

531 };

532 let email_svc = self.clone();

533 tokio::spawn(async move {

534 let base_url = email_svc.config().web.web_base_url.clone();

535 for uid in email_recipients {

536 if !mention_email_allowed(&email_svc, room_id, space_id, uid)

537 .await

538 .unwrap_or(false)

539 {

540 continue;

541 }

542 let Ok(mut emails) = email_svc.primary_emails(&[uid]).await else {

543 continue;

544 };

545 let Some(addr) = emails.remove(&uid) else {

546 continue;

547 };

469548 let subject = format!("You were mentioned in #{room_name}");

470 let link = format!(

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

472 self.config().web.web_base_url

473 );

474 let body = if content_for_email.chars().count() > 200 {

475 content_for_email.chars().take(200).collect::<String>()

476 } else {

477 content_for_email.clone()

478 };

479 let text = format!("{subject}\n\n{body}\n\nView channel: {link}");

549 let link = format!("{base_url}/chat/rooms/{room_id}");

550 let text = format!("{subject}\n\n{email_body}\n\nView channel: {link}");

480551 let html = format!(

481 r#"<p>{subject}</p><p>{body}</p><p><a href="{link}">View channel</a></p>"#

552 r#"<p>{subject}</p><p>{email_body}</p><p><a href="{link}">View channel</a></p>"#

482553 );

483 if let Err(err) = self.send_email(&addr, &subject, &text, &html).await {

554 if let Err(err) = email_svc.send_email(&addr, &subject, &text, &html).await {

484555 tracing::warn!(error = %err, "failed to send mention email");

485556 }

486557 }

487 }

558 });

488559 }

489560

490561 Ok(dto)

@@ -619,8 +690,6 @@

619690 .primary()

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

621692 Box::pin(async move {

622 let seq = next_room_seq(txn, room_id).await?;

623

624693 message_edits::ActiveModel {

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

626695 message_id: Set(message_id),

@@ -636,7 +705,6 @@

636705 .filter(messages::Column::Id.eq(message_id))

637706 .col_expr(messages::Column::Content, Expr::value(content))

638707 .col_expr(messages::Column::EditedAt, Expr::value(now))

639 .col_expr(messages::Column::Seq, Expr::value(seq))

640708 .exec(txn)

641709 .await?;

642710

@@ -649,7 +717,7 @@

649717 .one(self.db().read())

650718 .await?

651719 .ok_or(ChatError::Internal("message vanished after update".into()))?;

652 let seq = row.seq;

720 let seq = current_room_seq(self.db().read(), room_id).await;

653721 let items = self.build_message_dto(vec![row], caller).await?;

654722 let dto = items.into_iter().next().unwrap();

655723 let dto_value = serde_json::to_value(&dto)?;

@@ -736,19 +804,15 @@

736804

737805 let now = OffsetDateTime::now_utc();

738806 let ids = row_ids.clone();

739 let seq = self

740 .db()

807 self.db()

741808 .primary()

742 .transaction::<_, i64, ChatError>(move |txn| {

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

743810 Box::pin(async move {

744 let seq = next_room_seq(txn, room_id).await?;

745

746811 messages::Entity::update_many()

747812 .filter(messages::Column::Id.is_in(ids.clone()))

748813 .col_expr(messages::Column::Content, Expr::value(String::new()))

749814 .col_expr(messages::Column::Attachments, Expr::value(json!([])))

750815 .col_expr(messages::Column::DeletedAt, Expr::value(now))

751 .col_expr(messages::Column::Seq, Expr::value(seq))

752816 .exec(txn)

753817 .await?;

754818

@@ -761,11 +825,13 @@

761825 .await?;

762826 }

763827

764 Ok(seq)

828 Ok(())

765829 })

766830 })

767831 .await?;

768832

833 let seq = current_room_seq(self.db().read(), room_id).await;

834

769835 let event = if single {

770836 RoomEvent::MessageDeleted {

771837 room_id,

@@ -792,7 +858,12 @@

792858 message_id: Uuid,

793859 cmd: PinCmd,

794860 ) -> Result<MessageDto, ChatError> {

795 let _ctx = require_room_readable(self, room_id, caller).await?;

861 let ctx = require_room_readable(self, room_id, caller).await?;

862 if ctx.space.is_some()

863 && !can_manage_space(self, ctx.space.as_ref().unwrap(), caller).await?

864 {

865 return Err(ChatError::Forbidden);

866 }

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

797868 .filter(messages::Column::RoomId.eq(room_id))

798869 .one(self.db().read())

@@ -806,16 +877,13 @@

806877 }

807878

808879 let now = OffsetDateTime::now_utc();

809 let seq = self

810 .db()

880 self.db()

811881 .primary()

812 .transaction::<_, i64, ChatError>(move |txn| {

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

813883 Box::pin(async move {

814 let seq = next_room_seq(txn, room_id).await?;

815884 let mut update = messages::Entity::update_many()

816885 .filter(messages::Column::Id.eq(message_id))

817 .col_expr(messages::Column::Pinned, Expr::value(cmd.pinned))

818 .col_expr(messages::Column::Seq, Expr::value(seq));

886 .col_expr(messages::Column::Pinned, Expr::value(cmd.pinned));

819887 if cmd.pinned {

820888 update = update

821889 .col_expr(messages::Column::PinnedBy, Expr::value(caller))

@@ -823,14 +891,19 @@

823891 } else {

824892 update = update

825893 .col_expr(messages::Column::PinnedBy, Expr::value(None::<Uuid>))

826 .col_expr(messages::Column::PinnedAt, Expr::value(None::<OffsetDateTime>));

894 .col_expr(

895 messages::Column::PinnedAt,

896 Expr::value(None::<OffsetDateTime>),

897 );

827898 }

828899 update.exec(txn).await?;

829 Ok(seq)

900 Ok(())

830901 })

831902 })

832903 .await?;

833904

905 let seq = current_room_seq(self.db().read(), room_id).await;

906

834907 let event = RoomEvent::MessagePinned {

835908 room_id,

836909 seq,

@@ -881,18 +954,11 @@

881954 attachments: Vec::new(),

882955 nonce: None,

883956 urgent: false,

957 forwarded_from: Some(original.id),

884958 };

885 let mut dto = self.send_message(caller, cmd.target_room_id, send_cmd).await?;

886

887 messages::Entity::update_many()

888 .filter(messages::Column::Id.eq(dto.id))

889 .col_expr(

890 messages::Column::ForwardedFromId,

891 Expr::value(original.id),

892 )

893 .exec(self.db().write())

959 let dto = self

960 .send_message(caller, cmd.target_room_id, send_cmd)

894961 .await?;

895 dto.forwarded_from_id = Some(original.id);

896962 Ok(dto)

897963 }

898964

@@ -903,6 +969,13 @@

903969 message_id: Uuid,

904970 ) -> Result<Vec<MessageEditDto>, ChatError> {

905971 let _ctx = require_room_readable(self, room_id, caller).await?;

972 let message = messages::Entity::find_by_id(message_id)

973 .filter(messages::Column::RoomId.eq(room_id))

974 .one(self.db().read())

975 .await?

976 .filter(|m| m.deleted_at.is_none())

977 .ok_or(ChatError::NotFound)?;

978 let _ = message;

906979 let rows = message_edits::Entity::find()

907980 .filter(message_edits::Column::MessageId.eq(message_id))

908981 .order_by_desc(message_edits::Column::CreatedAt)

Modified lib/chat/src/services/presence.rs +137 −13 View lib/chat/src/services/presence.rs

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

11use models::chat::presence_states::{self, PresenceStatus};

22use models::space::space_members::{self, MemberStatus};

33use redis::AsyncCommands;

4use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};

4use sea_orm::{

5 ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QuerySelect, Set,

6};

57use serde::{Deserialize, Serialize};

68use time::OffsetDateTime;

79use utoipa::ToSchema;

@@ -57,12 +59,8 @@

5759 let key = presence_key(caller);

5860 let value = serde_json::to_string(&payload)?;

5961 let result: Result<(), redis::RedisError> = match &mut conn {

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

61 c.set_ex(&key, &value, ttl).await

62 }

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

64 c.set_ex(&key, &value, ttl).await

65 }

62 export::RedisConnection::Standalone(c) => c.set_ex(&key, &value, ttl).await,

63 export::RedisConnection::Cluster(c) => c.set_ex(&key, &value, ttl).await,

6664 };

6765 if let Err(error) = result {

6866 tracing::warn!(%error, "failed to write presence to redis");

@@ -123,7 +121,10 @@

123121 user_id: uid,

124122 status: payload.status,

125123 activities: payload.activities,

126 last_seen_at: Some(OffsetDateTime::from_unix_timestamp(payload.at).unwrap_or_else(|_| OffsetDateTime::now_utc())),

124 last_seen_at: Some(

125 OffsetDateTime::from_unix_timestamp(payload.at)

126 .unwrap_or_else(|_| OffsetDateTime::now_utc()),

127 ),

127128 online: !matches!(payload.status, PresenceStatus::Offline),

128129 });

129130 }

@@ -148,6 +149,133 @@

148149 }

149150 }

150151

152 /// Presence of another user is only visible when both share an active

153 /// space or a DM room; otherwise the requester gets NotFound.

154 pub async fn get_presence_for(

155 &self,

156 caller: Uuid,

157 target: Uuid,

158 ) -> Result<PresenceDto, ChatError> {

159 if caller == target || self.shares_room_with(caller, target).await? {

160 return self.get_presence(target).await;

161 }

162 Err(ChatError::NotFound)

163 }

164

165 async fn shares_room_with(&self, caller: Uuid, target: Uuid) -> Result<bool, ChatError> {

166 use models::room::room_members;

167 use models::space::space_members as sm;

168 let my_spaces: Vec<Uuid> = sm::Entity::find()

169 .filter(sm::Column::UserId.eq(caller))

170 .select_only()

171 .column(sm::Column::SpaceId)

172 .into_tuple()

173 .all(self.db().read())

174 .await?;

175 if !my_spaces.is_empty() {

176 let shared = sm::Entity::find()

177 .filter(sm::Column::UserId.eq(target))

178 .filter(sm::Column::SpaceId.is_in(my_spaces))

179 .count(self.db().read())

180 .await?;

181 if shared > 0 {

182 return Ok(true);

183 }

184 }

185 use models::room::rooms;

186 use sea_orm::Condition;

187 let my_rooms: Vec<Uuid> = room_members::Entity::find()

188 .filter(room_members::Column::UserId.eq(caller))

189 .filter(room_members::Column::LeftAt.is_null())

190 .select_only()

191 .column(room_members::Column::RoomId)

192 .into_tuple()

193 .all(self.db().read())

194 .await?;

195 if my_rooms.is_empty() {

196 return Ok(false);

197 }

198 let shared = rooms::Entity::find()

199 .filter(rooms::Column::Id.is_in(my_rooms.clone()))

200 .filter(rooms::Column::SpaceId.is_null())

201 .filter(

202 Condition::any()

203 .add(rooms::Column::OwnerId.eq(caller))

204 .add(rooms::Column::OwnerId.eq(target)),

205 )

206 .count(self.db().read())

207 .await?;

208 if shared > 0 {

209 return Ok(true);

210 }

211 let dm_peer = room_members::Entity::find()

212 .filter(room_members::Column::UserId.eq(target))

213 .filter(room_members::Column::RoomId.is_in(my_rooms))

214 .filter(room_members::Column::LeftAt.is_null())

215 .count(self.db().read())

216 .await?;

217 Ok(dm_peer > 0)

218 }

219

220 pub async fn get_presences(&self, uids: &[Uuid]) -> Result<Vec<PresenceDto>, ChatError> {

221 let mut by_uid: std::collections::HashMap<Uuid, PresencePayload> =

222 std::collections::HashMap::new();

223 if !uids.is_empty() {

224 let mut conn = self.state.redis.connection_clone();

225 for uid in uids {

226 let key = presence_key(*uid);

227 let value: Result<Option<String>, redis::RedisError> = match &mut conn {

228 export::RedisConnection::Standalone(c) => c.get(&key).await,

229 export::RedisConnection::Cluster(c) => c.get(&key).await,

230 };

231 if let Ok(Some(value)) = value

232 && let Ok(payload) = serde_json::from_str::<PresencePayload>(&value)

233 {

234 by_uid.insert(*uid, payload);

235 }

236 }

237 let missing: Vec<Uuid> = uids

238 .iter()

239 .filter(|u| !by_uid.contains_key(u))

240 .copied()

241 .collect();

242 let rows = if missing.is_empty() {

243 Vec::new()

244 } else {

245 presence_states::Entity::find()

246 .filter(presence_states::Column::UserId.is_in(missing))

247 .all(self.db().read())

248 .await?

249 };

250 let mut result = Vec::with_capacity(uids.len());

251 for uid in uids {

252 if let Some(payload) = by_uid.get(uid) {

253 result.push(PresenceDto {

254 user_id: *uid,

255 status: payload.status,

256 activities: payload.activities.clone(),

257 last_seen_at: Some(

258 OffsetDateTime::from_unix_timestamp(payload.at)

259 .unwrap_or_else(|_| OffsetDateTime::now_utc()),

260 ),

261 online: !matches!(payload.status, PresenceStatus::Offline),

262 });

263 continue;

264 }

265 let row = rows.iter().find(|r| r.user_id == *uid);

266 result.push(PresenceDto {

267 user_id: *uid,

268 status: PresenceStatus::Offline,

269 activities: serde_json::json!([]),

270 last_seen_at: row.and_then(|r| r.last_seen_at),

271 online: false,

272 });

273 }

274 return Ok(result);

275 }

276 Ok(Vec::new())

277 }

278

151279 pub async fn list_space_presences(

152280 &self,

153281 caller: Uuid,

@@ -164,10 +292,6 @@

164292 if !uids.contains(&space.owner_id) {

165293 uids.push(space.owner_id);

166294 }

167 let mut result = Vec::new();

168 for uid in uids {

169 result.push(self.get_presence(uid).await?);

170 }

171 Ok(result)

295 self.get_presences(&uids).await

172296 }

173297}

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

@@ -138,12 +138,19 @@

138138 }

139139 }

140140 let mut result: Vec<ReactionAggDto> = agg

141 .clone()

141142 .into_iter()

142143 .map(|(emoji, (count, me))| ReactionAggDto { emoji, count, me })

143144 .collect();

144145 result.sort_by(|a, b| a.emoji.cmp(&b.emoji));

145146

146 let value = serde_json::to_value(&result)?;

147 // The room-wide event must be viewer-neutral: `me` is per-recipient

148 // state and only belongs in the HTTP response above.

149 let shared: Vec<serde_json::Value> = agg

150 .into_iter()

151 .map(|(emoji, (count, _))| serde_json::json!({ "emoji": emoji, "count": count }))

152 .collect();

153 let value = serde_json::to_value(&shared)?;

147154 let event = RoomEvent::ReactionUpdated {

148155 room_id,

149156 message_id,

Modified lib/chat/src/services/read_state.rs +28 −22 View lib/chat/src/services/read_state.rs

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

1111

1212use crate::errors::ChatError;

1313use crate::services::service::ChatService;

14use crate::share::lookup::{lookup_room, require_room_readable, require_space_member, lookup_space_by_name};

14use crate::share::lookup::{lookup_space_by_name, require_room_readable, require_space_member};

1515use crate::share::notify::advance_read_state;

1616

1717#[derive(Debug, Clone, Serialize, ToSchema)]

@@ -61,7 +61,14 @@

6161 .ok_or(ChatError::NotFound)?;

6262 (Some(message_id), msg.seq)

6363 }

64 (None, Some(seq)) => (None, seq),

64 (None, Some(seq)) => {

65 let tip = room_seq::Entity::find_by_id(room_id)

66 .one(self.db().read())

67 .await?

68 .map(|s| s.seq)

69 .unwrap_or(0);

70 (None, seq.clamp(0, tip))

71 }

6572 (None, None) => {

6673 let tip = room_seq::Entity::find_by_id(room_id)

6774 .one(self.db().read())

@@ -76,14 +83,8 @@

7683 }

7784 };

7885

79 let changed = advance_read_state(

80 self.db().write(),

81 caller,

82 room_id,

83 message_id,

84 seq,

85 )

86 .await?;

86 let changed =

87 advance_read_state(self.db().write(), caller, room_id, message_id, seq).await?;

8788 if changed {

8889 self.realtime

8990 .publish_user(

@@ -105,7 +106,7 @@

105106 caller: Uuid,

106107 room_id: Uuid,

107108 ) -> Result<ReadStateDto, ChatError> {

108 let _ = lookup_room(self, room_id).await?;

109 require_room_readable(self, room_id, caller).await?;

109110 let tip = room_seq::Entity::find_by_id(room_id)

110111 .one(self.db().read())

111112 .await?

@@ -116,17 +117,21 @@

116117 .await?;

117118 let last_viewed_at = state.as_ref().and_then(|s| s.last_viewed_at);

118119 let unread_mentions = match last_viewed_at {

119 Some(at) => user_recent_mentions::Entity::find()

120 .filter(user_recent_mentions::Column::UserId.eq(caller))

121 .filter(user_recent_mentions::Column::ChannelId.eq(room_id))

122 .filter(user_recent_mentions::Column::CreatedAt.gt(at))

123 .count(self.db().read())

124 .await? as i64,

125 None => user_recent_mentions::Entity::find()

126 .filter(user_recent_mentions::Column::UserId.eq(caller))

127 .filter(user_recent_mentions::Column::ChannelId.eq(room_id))

128 .count(self.db().read())

129 .await? as i64,

120 Some(at) => {

121 user_recent_mentions::Entity::find()

122 .filter(user_recent_mentions::Column::UserId.eq(caller))

123 .filter(user_recent_mentions::Column::ChannelId.eq(room_id))

124 .filter(user_recent_mentions::Column::CreatedAt.gt(at))

125 .count(self.db().read())

126 .await? as i64

127 }

128 None => {

129 user_recent_mentions::Entity::find()

130 .filter(user_recent_mentions::Column::UserId.eq(caller))

131 .filter(user_recent_mentions::Column::ChannelId.eq(room_id))

132 .count(self.db().read())

133 .await? as i64

134 }

130135 };

131136 let last_read_seq = state.as_ref().map(|s| s.last_read_seq).unwrap_or(0);

132137 Ok(ReadStateDto {

@@ -156,6 +161,7 @@

156161 let mut items = Vec::new();

157162 for room in space_rooms {

158163 if room.is_private

164 && room.owner_id != caller

159165 && !crate::share::lookup::is_room_member(self, room.id, caller).await?

160166 {

161167 continue;

Modified lib/chat/src/services/reminders.rs +41 −4 View lib/chat/src/services/reminders.rs

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

11use models::chat::message_reminders::{self, ReminderStatus};

22use models::chat::messages;

33use models::users::user_notifications::NotificationType;

4use sea_orm::sea_query::Expr;

45use sea_orm::{

56 ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,

67};

@@ -58,7 +59,9 @@

5859 .filter(|m| m.deleted_at.is_none())

5960 .ok_or(ChatError::NotFound)?;

6061 if cmd.remind_at <= OffsetDateTime::now_utc() {

61 return Err(ChatError::BadRequest("remind_at must be in the future".to_string()));

62 return Err(ChatError::BadRequest(

63 "remind_at must be in the future".to_string(),

64 ));

6265 }

6366 let row = message_reminders::ActiveModel {

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

@@ -101,6 +104,21 @@

101104

102105 pub async fn deliver_due_reminders(&self) -> Result<u64, ChatError> {

103106 let now = OffsetDateTime::now_utc();

107

108 // Recover claims abandoned by a crashed instance.

109 let _ = message_reminders::Entity::update_many()

110 .col_expr(

111 message_reminders::Column::Status,

112 Expr::value(ReminderStatus::Pending),

113 )

114 .filter(message_reminders::Column::Status.eq(ReminderStatus::Delivering))

115 .filter(

116 message_reminders::Column::RemindAt

117 .lt(now - time::Duration::minutes(crate::worker::STUCK_CLAIM_MINUTES)),

118 )

119 .exec(self.db().write())

120 .await?;

121

104122 let due = message_reminders::Entity::find()

105123 .filter(message_reminders::Column::Status.eq(ReminderStatus::Pending))

106124 .filter(message_reminders::Column::RemindAt.lte(now))

@@ -109,9 +127,19 @@

109127 .await?;

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

111129 for reminder in due {

112 let mut am: message_reminders::ActiveModel = reminder.clone().into();

113 am.status = Set(ReminderStatus::Done);

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

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

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

132 .col_expr(

133 message_reminders::Column::Status,

134 Expr::value(ReminderStatus::Delivering),

135 )

136 .filter(message_reminders::Column::Id.eq(reminder.id))

137 .filter(message_reminders::Column::Status.eq(ReminderStatus::Pending))

138 .exec(self.db().write())

139 .await?;

140 if claimed.rows_affected == 0 {

141 continue;

142 }

115143

116144 let data = serde_json::json!({

117145 "room_id": reminder.room_id,

@@ -130,6 +158,15 @@

130158 {

131159 tracing::warn!(%error, "failed to write reminder notification");

132160 }

161 message_reminders::Entity::update_many()

162 .col_expr(

163 message_reminders::Column::Status,

164 Expr::value(ReminderStatus::Done),

165 )

166 .filter(message_reminders::Column::Id.eq(reminder.id))

167 .filter(message_reminders::Column::Status.eq(ReminderStatus::Delivering))

168 .exec(self.db().write())

169 .await?;

133170 self.realtime

134171 .publish_user(

135172 reminder.user_id,

Modified lib/chat/src/services/rooms.rs +141 −72 View lib/chat/src/services/rooms.rs

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

99use rand::Rng;

1010use sea_orm::sea_query::Expr;

1111use sea_orm::{

12 ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,

13 TransactionTrait,

12 ActiveModelTrait, ColumnTrait, EntityTrait, ExprTrait, QueryFilter, QueryOrder, QuerySelect,

13 Set, TransactionTrait,

1414};

1515use serde::{Deserialize, Serialize};

1616use time::OffsetDateTime;

@@ -143,19 +143,17 @@

143143 .filter(room_user_settings::Column::RoomId.is_in(room_ids.clone()))

144144 .all(self.db().read())

145145 .await?;

146 let settings_map: std::collections::HashMap<Uuid, room_user_settings::Model> = settings

147 .into_iter()

148 .map(|s| (s.room_id, s))

149 .collect();

146 let settings_map: std::collections::HashMap<Uuid, room_user_settings::Model> =

147 settings.into_iter().map(|s| (s.room_id, s)).collect();

150148

151149 let read = read_states::Entity::find()

152150 .filter(read_states::Column::UserId.eq(caller))

153151 .filter(read_states::Column::RoomId.is_in(room_ids.clone()))

154152 .all(self.db().read())

155153 .await?;

156 let read_map: std::collections::HashMap<Uuid, i64> = read

154 let read_map: std::collections::HashMap<Uuid, (i64, Option<OffsetDateTime>)> = read

157155 .into_iter()

158 .map(|r| (r.room_id, r.last_read_seq))

156 .map(|r| (r.room_id, (r.last_read_seq, r.last_viewed_at)))

159157 .collect();

160158

161159 let seqs = room_seq::Entity::find()

@@ -165,34 +163,51 @@

165163 let seq_map: std::collections::HashMap<Uuid, i64> =

166164 seqs.into_iter().map(|s| (s.room_id, s.seq)).collect();

167165

168 let mention_counts = models::users::user_recent_mentions::Entity::find()

166 let mention_rows = models::users::user_recent_mentions::Entity::find()

169167 .filter(models::users::user_recent_mentions::Column::UserId.eq(caller))

170168 .filter(models::users::user_recent_mentions::Column::ChannelId.is_in(room_ids.clone()))

169 .select_only()

170 .column(models::users::user_recent_mentions::Column::ChannelId)

171 .column(models::users::user_recent_mentions::Column::CreatedAt)

172 .into_tuple::<(Uuid, OffsetDateTime)>()

171173 .all(self.db().read())

172174 .await?;

175 let mut mentions_by_room: std::collections::HashMap<Uuid, Vec<OffsetDateTime>> =

176 std::collections::HashMap::new();

177 for (channel_id, created_at) in mention_rows {

178 mentions_by_room

179 .entry(channel_id)

180 .or_default()

181 .push(created_at);

182 }

173183

174 let member_counts = room_members::Entity::find()

184 let member_counts: Vec<(Uuid, i64)> = room_members::Entity::find()

175185 .filter(room_members::Column::RoomId.is_in(room_ids.clone()))

176186 .filter(room_members::Column::LeftAt.is_null())

187 .select_only()

188 .column(room_members::Column::RoomId)

189 .column_as(Expr::col(room_members::Column::UserId).count(), "cnt")

190 .group_by(room_members::Column::RoomId)

191 .into_tuple()

177192 .all(self.db().read())

178193 .await?;

179 let mut count_map: std::collections::HashMap<Uuid, i64> = std::collections::HashMap::new();

180 for m in member_counts {

181 *count_map.entry(m.room_id).or_default() += 1;

182 }

194 let count_map: std::collections::HashMap<Uuid, i64> = member_counts.into_iter().collect();

183195

184196 let mut result = Vec::new();

185197 for room in rooms {

186198 if room.is_private && !member_of.contains(&room.id) && room.owner_id != caller {

187199 continue;

188200 }

189 let last_read = read_map.get(&room.id).copied().unwrap_or(0);

201 let (last_read, last_viewed) = read_map.get(&room.id).copied().unwrap_or((0, None));

190202 let tip = seq_map.get(&room.id).copied().unwrap_or(0);

191 let unread_count = (tip - last_read).max(0);

192 let unread_mentions = mention_counts

193 .iter()

194 .filter(|m| m.channel_id == room.id)

195 .count() as i64;

203 let unread_count = std::cmp::max(tip - last_read, 0);

204 let unread_mentions = mentions_by_room

205 .get(&room.id)

206 .map(|list| match last_viewed {

207 Some(at) => list.iter().filter(|c| **c > at).count() as i64,

208 None => list.len() as i64,

209 })

210 .unwrap_or(0);

196211 let setting = settings_map.get(&room.id);

197212 result.push(RoomBriefDto {

198213 id: room.id,

@@ -218,11 +233,7 @@

218233 Ok(result)

219234 }

220235

221 pub async fn get_room(

222 &self,

223 caller: Uuid,

224 room_id: Uuid,

225 ) -> Result<RoomBriefDto, ChatError> {

236 pub async fn get_room(&self, caller: Uuid, room_id: Uuid) -> Result<RoomBriefDto, ChatError> {

226237 let ctx = require_room_readable(self, room_id, caller).await?;

227238 let room = ctx.room;

228239 let tip = room_seq::Entity::find_by_id(room.id)

@@ -251,7 +262,7 @@

251262 nsfw: room.nsfw,

252263 rate_limit_per_user: room.rate_limit_per_user,

253264 last_message_id: room.last_message_id,

254 unread_count: (tip - last_read).max(0),

265 unread_count: std::cmp::max(tip - last_read, 0),

255266 unread_mentions: 0,

256267 favorite: setting.as_ref().is_some_and(|s| s.favorite),

257268 muted: setting.as_ref().is_some_and(|s| s.muted),

@@ -268,7 +279,14 @@

268279 cmd: UpdateRoomCmd,

269280 ) -> Result<RoomBriefDto, ChatError> {

270281 let ctx = require_room_readable(self, room_id, caller).await?;

271 let needs_manage = cmd.name.is_some() || cmd.is_private.is_some() || cmd.position.is_some();

282 // Anything beyond a plain topic tweak on a DM is a management operation.

283 let needs_manage = ctx.space.is_some()

284 && (cmd.name.is_some()

285 || cmd.is_private.is_some()

286 || cmd.position.is_some()

287 || cmd.topic.is_some()

288 || cmd.nsfw.is_some()

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

272290 if needs_manage {

273291 match &ctx.space {

274292 Some(space) => {

@@ -279,6 +297,16 @@

279297 None => return Err(ChatError::Forbidden),

280298 }

281299 }

300 if ctx.space.is_none() && (cmd.name.is_some() || cmd.is_private.is_some()) {

301 return Err(ChatError::Forbidden);

302 }

303 if let Some(rate) = cmd.rate_limit_per_user {

304 if !(0..=3600).contains(&rate) {

305 return Err(ChatError::BadRequest(

306 "rate_limit_per_user must be within 0..=3600".to_string(),

307 ));

308 }

309 }

282310

283311 let now = OffsetDateTime::now_utc();

284312 let room = ctx.room.clone();

@@ -495,50 +523,79 @@

495523 if !is_room_member(self, room_id, caller).await? {

496524 return Err(ChatError::Forbidden);

497525 }

526 let limit = ctx.room.user_limit.unwrap_or(i32::MAX);

527 let mut active = self.room_member_count(room_id).await? as i64;

528 if active >= limit as i64 && limit <= 2 {

529 return Err(ChatError::BadRequest(

530 "this conversation does not accept more members".to_string(),

531 ));

532 }

533 for uid in &cmd.user_ids {

534 if !is_room_member(self, room_id, *uid).await? {

535 active += 1;

536 }

537 if active > limit as i64 {

538 return Err(ChatError::BadRequest(

539 "room member limit reached".to_string(),

540 ));

541 }

542 }

498543 }

499544 }

500545

501 let now = OffsetDateTime::now_utc();

502546 for uid in cmd.user_ids {

503547 if let Some(space) = &ctx.space

504548 && require_space_member(self, space, uid).await.is_err()

505549 {

506550 continue;

507551 }

508 let existing = room_members::Entity::find_by_id((room_id, uid))

509 .one(self.db().write())

510 .await?;

511 match existing {

512 Some(row) if row.left_at.is_some() => {

513 let mut am: room_members::ActiveModel = row.into();

514 am.left_at = Set(None);

515 am.updated_at = Set(Some(now));

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

517 }

518 Some(_) => {}

519 None => {

520 room_members::ActiveModel {

521 room_id: Set(room_id),

522 user_id: Set(uid),

523 nick: Set(None),

524 joined_at: Set(now),

525 left_at: Set(None),

526 created_at: Set(now),

527 updated_at: Set(Some(now)),

528 }

529 .insert(self.db().write())

530 .await?;

552 self.ensure_room_membership(room_id, uid).await?;

553 }

554 Ok(())

555 }

556

557 pub(crate) async fn ensure_room_membership(

558 &self,

559 room_id: Uuid,

560 uid: Uuid,

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

562 let now = OffsetDateTime::now_utc();

563 let existing = room_members::Entity::find_by_id((room_id, uid))

564 .one(self.db().write())

565 .await?;

566 match existing {

567 Some(row) if row.left_at.is_some() => {

568 let mut am: room_members::ActiveModel = row.into();

569 am.left_at = Set(None);

570 am.updated_at = Set(Some(now));

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

572 }

573 Some(_) => return Ok(()),

574 None => {

575 room_members::ActiveModel {

576 room_id: Set(room_id),

577 user_id: Set(uid),

578 nick: Set(None),

579 joined_at: Set(now),

580 left_at: Set(None),

581 created_at: Set(now),

582 updated_at: Set(Some(now)),

531583 }

584 .insert(self.db().write())

585 .await?;

532586 }

533 let event = RoomEvent::MemberJoined {

534 room_id,

535 user_id: uid,

536 };

537 self.realtime.publish_room(&event).await;

538 self.realtime

539 .publish_user(uid, &serde_json::json!({"kind": "room_added", "room_id": room_id}))

540 .await;

541587 }

588 let event = RoomEvent::MemberJoined {

589 room_id,

590 user_id: uid,

591 };

592 self.realtime.publish_room(&event).await;

593 self.realtime

594 .publish_user(

595 uid,

596 &serde_json::json!({"kind": "room_added", "room_id": room_id}),

597 )

598 .await;

542599 Ok(())

543600 }

544601

@@ -588,7 +645,17 @@

588645 let now = OffsetDateTime::now_utc();

589646 let id = Uuid::now_v7();

590647 let accept_code = random_accept_code();

591 let expires_at = cmd.expires_secs.map(|secs| now + time::Duration::seconds(secs));

648 let expires_at = match cmd.expires_secs {

649 Some(secs) if secs > 0 && secs <= 30 * 86_400 => {

650 Some(now + time::Duration::seconds(secs))

651 }

652 Some(_) => {

653 return Err(ChatError::BadRequest(

654 "expires_secs must be within 1..=2592000".to_string(),

655 ));

656 }

657 None => None,

658 };

592659 let row = room_invites::ActiveModel {

593660 id: Set(id),

594661 room_id: Set(room_id),

@@ -640,7 +707,10 @@

640707 if invite.accepted_at.is_some() || invite.revoked_at.is_some() {

641708 return Err(ChatError::NotFound);

642709 }

643 if invite.expires_at.is_some_and(|at| at < OffsetDateTime::now_utc()) {

710 if invite

711 .expires_at

712 .is_some_and(|at| at < OffsetDateTime::now_utc())

713 {

644714 return Err(ChatError::NotFound);

645715 }

646716 if invite.invitee_user_id.is_some_and(|uid| uid != caller) {

@@ -663,18 +733,17 @@

663733 }

664734

665735 let now = OffsetDateTime::now_utc();

666 let mut am: room_invites::ActiveModel = invite.into();

667 am.accepted_at = Set(Some(now));

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

736 let claimed = room_invites::Entity::update_many()

737 .col_expr(room_invites::Column::AcceptedAt, Expr::value(Some(now)))

738 .filter(room_invites::Column::Id.eq(invite.id))

739 .filter(room_invites::Column::AcceptedAt.is_null())

740 .exec(self.db().write())

741 .await?;

742 if claimed.rows_affected == 0 {

743 return Err(ChatError::NotFound);

744 }

669745

670 self.add_room_members(

671 caller,

672 room.id,

673 AddMembersCmd {

674 user_ids: vec![caller],

675 },

676 )

677 .await?;

746 self.ensure_room_membership(room.id, caller).await?;

678747 self.get_room(caller, room.id).await

679748 }

680749}

Modified lib/chat/src/services/search.rs +26 −20 View lib/chat/src/services/search.rs

@@ -68,28 +68,34 @@

6868 .clamp(1, self.config().chat.max_pagination);

6969 let offset = q.offset.unwrap_or(0);

7070

71 let room_filter = q

72 .room_id

73 .map(|id| format!("AND m.room_id = '{}'", id.as_simple()))

74 .unwrap_or_default();

75 let author_filter = q

76 .author_id

77 .map(|id| format!("AND m.author_id = '{}'", id.as_simple()))

78 .unwrap_or_default();

71 // Private rooms must be restricted to their members; public space

72 // rooms are readable by every space member.

73 const FILTERS: &str = r#"

74 AND r.space_id = $1

75 AND m.deleted_at IS NULL

76 AND m.search_vector @@ plainto_tsquery('simple', $2)

77 AND ($5::uuid IS NULL OR m.room_id = $5)

78 AND ($6::uuid IS NULL OR m.author_id = $6)

79 AND (r.is_private = false OR r.owner_id = $7

80 OR EXISTS (SELECT 1 FROM room_members rm

81 WHERE rm.room_id = m.room_id AND rm.user_id = $7

82 AND rm.left_at IS NULL))"#;

7983

8084 let count_sql = format!(

81 r#"SELECT count(*)::bigint AS total

82 FROM chat_message m

83 JOIN rooms r ON r.id = m.room_id

84 WHERE r.space_id = $1

85 AND m.deleted_at IS NULL

86 AND m.search_vector @@ plainto_tsquery('simple', $2)

87 {room_filter} {author_filter}"#

85 "SELECT count(*)::bigint AS total FROM chat_message m JOIN rooms r ON r.id = m.room_id WHERE 1=1{FILTERS}"

8886 );

8987 let count_row = CountRow::find_by_statement(Statement::from_sql_and_values(

9088 DatabaseBackend::Postgres,

9189 &count_sql,

92 [space.id.into(), query.into()],

90 [

91 space.id.into(),

92 query.into(),

93 0i64.into(),

94 0i64.into(),

95 q.room_id.into(),

96 q.author_id.into(),

97 caller.into(),

98 ],

9399 ))

94100 .one(self.db().read())

95101 .await?

@@ -99,10 +105,7 @@

99105 r#"SELECT m.id AS message_id, m.room_id, m.author_id, m.seq, m.content, m.created_at

100106 FROM chat_message m

101107 JOIN rooms r ON r.id = m.room_id

102 WHERE r.space_id = $1

103 AND m.deleted_at IS NULL

104 AND m.search_vector @@ plainto_tsquery('simple', $2)

105 {room_filter} {author_filter}

108 WHERE 1=1{FILTERS}

106109 ORDER BY ts_rank(m.search_vector, plainto_tsquery('simple', $2)) DESC, m.created_at DESC

107110 LIMIT $3 OFFSET $4"#

108111 );

@@ -114,6 +117,9 @@

114117 query.into(),

115118 (limit as i64).into(),

116119 (offset as i64).into(),

120 q.room_id.into(),

121 q.author_id.into(),

122 caller.into(),

117123 ],

118124 ))

119125 .all(self.db().read())

Modified lib/chat/src/services/threads.rs +64 −52 View lib/chat/src/services/threads.rs

@@ -293,11 +293,7 @@

293293 self.build_thread_dto(rows).await

294294 }

295295

296 pub async fn get_thread(

297 &self,

298 caller: Uuid,

299 thread_id: Uuid,

300 ) -> Result<ThreadDto, ChatError> {

296 pub async fn get_thread(&self, caller: Uuid, thread_id: Uuid) -> Result<ThreadDto, ChatError> {

301297 let thread = threads::Entity::find_by_id(thread_id)

302298 .one(self.db().read())

303299 .await?

@@ -323,6 +319,9 @@

323319 Some(space) => can_manage_space(self, space, caller).await?,

324320 None => false,

325321 };

322 if ctx.space.is_some() && thread.owner_id != caller && !is_manager {

323 return Err(ChatError::Forbidden);

324 }

326325 if cmd.locked.is_some() && thread.owner_id != caller && !is_manager {

327326 return Err(ChatError::Forbidden);

328327 }

@@ -373,38 +372,46 @@

373372

374373 let now = OffsetDateTime::now_utc();

375374 let notify_level = cmd.notify_level.unwrap_or(ThreadNotifyLevel::All);

376 let member = thread_members::Entity::find_by_id((thread_id, caller))

377 .one(self.db().write())

375 self.db()

376 .primary()

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

378 Box::pin(async move {

379 let member = thread_members::Entity::find_by_id((thread_id, caller))

380 .one(txn)

381 .await?;

382 match member {

383 Some(row) => {

384 let mut am: thread_members::ActiveModel = row.into();

385 am.notify_level = Set(notify_level);

386 am.updated_at = Set(now);

387 am.update(txn).await?;

388 }

389 None => {

390 thread_members::ActiveModel {

391 thread_id: Set(thread_id),

392 user_id: Set(caller),

393 notify_level: Set(notify_level),

394 last_read_message_id: Set(None),

395 last_read_seq: Set(0),

396 created_at: Set(now),

397 updated_at: Set(now),

398 }

399 .insert(txn)

400 .await?;

401 threads::Entity::update_many()

402 .filter(threads::Column::Id.eq(thread_id))

403 .col_expr(

404 threads::Column::ParticipantCount,

405 Expr::col(threads::Column::ParticipantCount).add(1),

406 )

407 .exec(txn)

408 .await?;

409 }

410 }

411 Ok(())

412 })

413 })

378414 .await?;

379 match member {

380 Some(row) => {

381 let mut am: thread_members::ActiveModel = row.into();

382 am.notify_level = Set(notify_level);

383 am.updated_at = Set(now);

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

385 }

386 None => {

387 thread_members::ActiveModel {

388 thread_id: Set(thread_id),

389 user_id: Set(caller),

390 notify_level: Set(notify_level),

391 last_read_message_id: Set(None),

392 last_read_seq: Set(0),

393 created_at: Set(now),

394 updated_at: Set(now),

395 }

396 .insert(self.db().write())

397 .await?;

398 threads::Entity::update_many()

399 .filter(threads::Column::Id.eq(thread_id))

400 .col_expr(

401 threads::Column::ParticipantCount,

402 Expr::col(threads::Column::ParticipantCount).add(1),

403 )

404 .exec(self.db().write())

405 .await?;

406 }

407 }

408415 Ok(())

409416 }

410417

@@ -415,19 +422,27 @@

415422 .ok_or(ChatError::NotFound)?;

416423 let _ctx = require_room_readable(self, thread.room_id, caller).await?;

417424

418 let deleted = thread_members::Entity::delete_by_id((thread_id, caller))

419 .exec(self.db().write())

425 self.db()

426 .primary()

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

428 Box::pin(async move {

429 let deleted = thread_members::Entity::delete_by_id((thread_id, caller))

430 .exec(txn)

431 .await?;

432 if deleted.rows_affected > 0 {

433 threads::Entity::update_many()

434 .filter(threads::Column::Id.eq(thread_id))

435 .col_expr(

436 threads::Column::ParticipantCount,

437 Expr::col(threads::Column::ParticipantCount).sub(1),

438 )

439 .exec(txn)

440 .await?;

441 }

442 Ok(())

443 })

444 })

420445 .await?;

421 if deleted.rows_affected > 0 {

422 threads::Entity::update_many()

423 .filter(threads::Column::Id.eq(thread_id))

424 .col_expr(

425 threads::Column::ParticipantCount,

426 Expr::col(threads::Column::ParticipantCount).sub(1),

427 )

428 .exec(self.db().write())

429 .await?;

430 }

431446 Ok(())

432447 }

433448

@@ -484,10 +499,7 @@

484499 Ok(())

485500 }

486501

487 pub(crate) async fn room_member_count(

488 &self,

489 room_id: Uuid,

490 ) -> Result<u64, ChatError> {

502 pub(crate) async fn room_member_count(&self, room_id: Uuid) -> Result<u64, ChatError> {

491503 Ok(room_members::Entity::find()

492504 .filter(room_members::Column::RoomId.eq(room_id))

493505 .filter(room_members::Column::LeftAt.is_null())

Modified lib/chat/src/services/webhooks.rs +50 −11 View lib/chat/src/services/webhooks.rs

@@ -63,6 +63,32 @@

6363 )

6464}

6565

66/// Fixed-window limiter for the unauthenticated incoming-webhook endpoint.

67const HOOK_WINDOW_SECS: u64 = 60;

68const HOOK_MAX_PER_WINDOW: u32 = 30;

69

70fn hook_allow(bucket_key: &str) -> bool {

71 static BUCKETS: std::sync::OnceLock<

72 std::sync::Mutex<std::collections::HashMap<String, (u32, std::time::Instant)>>,

73 > = std::sync::OnceLock::new();

74 let map = BUCKETS.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()));

75 let Ok(mut guard) = map.lock() else {

76 return true;

77 };

78 if guard.len() > 10_000 {

79 guard.retain(|_, (_, start)| start.elapsed().as_secs() < HOOK_WINDOW_SECS);

80 }

81 let now = std::time::Instant::now();

82 let entry = guard.entry(bucket_key.to_string()).or_insert((0, now));

83 if entry.1.elapsed().as_secs() >= HOOK_WINDOW_SECS {

84 entry.0 = 1;

85 entry.1 = now;

86 return true;

87 }

88 entry.0 = entry.0.saturating_add(1);

89 entry.0 <= HOOK_MAX_PER_WINDOW

90}

91

6692impl From<room_webhooks::Model> for WebhookDto {

6793 fn from(w: room_webhooks::Model) -> Self {

6894 Self {

@@ -177,6 +203,11 @@

177203 cmd: WebhookPostCmd,

178204 ) -> Result<serde_json::Value, ChatError> {

179205 let hash = token_hash(token);

206 if !hook_allow(&hash) {

207 return Err(ChatError::RateLimited(

208 "webhook rate limit exceeded, retry in 60s".to_string(),

209 ));

210 }

180211 let webhook = room_webhooks::Entity::find()

181212 .filter(room_webhooks::Column::TokenHash.eq(&hash))

182213 .one(self.db().write())

@@ -196,15 +227,20 @@

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

197228

198229 let message = self

199 .post_webhook_message(webhook.room_id, &webhook, SendMessageCmd {

200 content: cmd.content,

201 reply_to_id: None,

202 thread_id: None,

203 thread_broadcast: false,

204 attachments: Vec::new(),

205 nonce: cmd.nonce,

206 urgent: false,

207 })

230 .post_webhook_message(

231 webhook.room_id,

232 &webhook,

233 SendMessageCmd {

234 content: cmd.content,

235 reply_to_id: None,

236 thread_id: None,

237 thread_broadcast: false,

238 attachments: Vec::new(),

239 nonce: cmd.nonce,

240 urgent: false,

241 forwarded_from: None,

242 },

243 )

208244 .await?;

209245 Ok(serde_json::to_value(message)?)

210246 }

@@ -217,8 +253,8 @@

217253 ) -> Result<crate::services::messages::MessageDto, ChatError> {

218254 use models::chat::messages::{self as chat_messages, AuthorType, ChatMessageType};

219255 use models::room::rooms;

256 use sea_orm::TransactionTrait;

220257 use sea_orm::sea_query::Expr;

221 use sea_orm::TransactionTrait;

222258

223259 let content = cmd.content.trim().to_string();

224260 if content.is_empty() {

@@ -231,11 +267,14 @@

231267 if let Some(nonce) = &cmd.nonce {

232268 let existing = chat_messages::Entity::find()

233269 .filter(chat_messages::Column::RoomId.eq(room_id))

270 .filter(chat_messages::Column::AuthorId.eq(webhook.user_id))

234271 .filter(chat_messages::Column::Nonce.eq(nonce.as_str()))

235272 .one(self.db().read())

236273 .await?;

237274 if let Some(existing) = existing {

238 let items = self.build_message_dto(vec![existing], webhook.user_id).await?;

275 let items = self

276 .build_message_dto(vec![existing], webhook.user_id)

277 .await?;

239278 return Ok(items.into_iter().next().unwrap());

240279 }

241280 }

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

@@ -124,7 +124,10 @@

124124 match &ctx.space {

125125 Some(space) => {

126126 require_space_member(service, space, uid).await?;

127 if ctx.room.is_private && !is_room_member(service, room_id, uid).await? {

127 if ctx.room.is_private

128 && ctx.room.owner_id != uid

129 && !is_room_member(service, room_id, uid).await?

130 {

128131 return Err(ChatError::Forbidden);

129132 }

130133 }

Modified lib/chat/src/share/mention.rs +25 −42 View lib/chat/src/share/mention.rs

@@ -1,10 +1,8 @@

1use std::collections::HashMap;

2

31use models::room::room_members;

42use models::space::space_groups;

53use models::space::space_members::{self, MemberStatus};

64use models::users::user;

7use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QuerySelect};

5use sea_orm::{ColumnTrait, Condition, EntityTrait, QueryFilter};

86use uuid::Uuid;

97

108use crate::errors::ChatError;

@@ -76,31 +74,26 @@

7674 ctx: &RoomContext,

7775 lower: &str,

7876) -> Result<Resolved, ChatError> {

77 // Targeted lookup by username/name, then verify the resolved user is

78 // actually visible in this room's context (avoids full member scans).

79 let candidates = user::Entity::find()

80 .filter(

81 Condition::any()

82 .add(user::Column::LowerName.eq(lower))

83 .add(user::Column::Username.eq(lower)),

84 )

85 .all(service.db().read())

86 .await?;

7987 if let Some(space) = &ctx.space {

80 let members = space_members::Entity::find()

81 .filter(space_members::Column::SpaceId.eq(space.id))

82 .filter(space_members::Column::Status.eq(MemberStatus::Active))

83 .select_only()

84 .column(space_members::Column::UserId)

85 .into_tuple::<Uuid>()

86 .all(service.db().read())

87 .await?;

88 let mut uids = members;

89 if !uids.contains(&space.owner_id) {

90 uids.push(space.owner_id);

91 }

92 if !uids.is_empty() {

93 let users = user::Entity::find()

94 .filter(user::Column::Uid.is_in(uids))

95 .all(service.db().read())

96 .await?;

97 let mut name_map: HashMap<String, Uuid> = HashMap::new();

98 for u in users {

99 name_map.insert(u.lower_name.to_lowercase(), u.uid);

100 name_map.insert(u.username.to_lowercase(), u.uid);

88 for candidate in candidates {

89 if space.owner_id == candidate.uid {

90 return Ok(Resolved::User(candidate.uid));

10191 }

102 if let Some(&uid) = name_map.get(lower) {

103 return Ok(Resolved::User(uid));

92 let member = space_members::Entity::find_by_id((space.id, candidate.uid))

93 .one(service.db().read())

94 .await?;

95 if member.is_some_and(|m| m.status == MemberStatus::Active) {

96 return Ok(Resolved::User(candidate.uid));

10497 }

10598 }

10699 let group = space_groups::Entity::find()

@@ -114,22 +107,12 @@

114107 return Ok(Resolved::None);

115108 }

116109

117 let peers = room_members::Entity::find()

118 .filter(room_members::Column::RoomId.eq(ctx.room.id))

119 .filter(room_members::Column::LeftAt.is_null())

120 .all(service.db().read())

121 .await?;

122 let uids: Vec<Uuid> = peers.into_iter().map(|m| m.user_id).collect();

123 if uids.is_empty() {

124 return Ok(Resolved::None);

125 }

126 let users = user::Entity::find()

127 .filter(user::Column::Uid.is_in(uids))

128 .all(service.db().read())

129 .await?;

130 for u in users {

131 if u.lower_name.to_lowercase() == lower || u.username.to_lowercase() == lower {

132 return Ok(Resolved::User(u.uid));

110 for candidate in candidates {

111 let member = room_members::Entity::find_by_id((ctx.room.id, candidate.uid))

112 .one(service.db().read())

113 .await?;

114 if member.is_some_and(|m| m.left_at.is_none()) {

115 return Ok(Resolved::User(candidate.uid));

133116 }

134117 }

135118 Ok(Resolved::None)

Modified lib/chat/src/share/notify.rs +68 −37 View lib/chat/src/share/notify.rs

@@ -1,10 +1,9 @@

1use models::chat::read_states;

1use models::chat::room_seq;

22use models::chat::room_user_settings::{self, RoomNotificationLevel};

33use models::space::space_members::{self, MessageNotifications};

44use models::users::user_notifications::{self, NotificationType};

55use models::users::user_recent_mentions;

6use sea_orm::sea_query::Expr;

7use sea_orm::{ActiveModelTrait, ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait, QueryFilter, Set, Statement};

6use sea_orm::{ActiveModelTrait, ConnectionTrait, DatabaseBackend, EntityTrait, Set, Statement};

87use time::OffsetDateTime;

98use uuid::Uuid;

109

@@ -12,10 +11,7 @@

1211use crate::services::service::ChatService;

1312use crate::share::realtime::RoomEvent;

1413

15pub async fn next_room_seq(

16 db: &impl ConnectionTrait,

17 room_id: Uuid,

18) -> Result<i64, ChatError> {

14pub async fn next_room_seq(db: &impl ConnectionTrait, room_id: Uuid) -> Result<i64, ChatError> {

1915 let now = OffsetDateTime::now_utc();

2016 let row = db

2117 .query_one_raw(Statement::from_sql_and_values(

@@ -33,6 +29,18 @@

3329 Ok(seq)

3430}

3531

32/// Current room tip without consuming a seq; used for non-message events so

33/// edits/pins/deletes never inflate unread counters or reorder listings.

34pub async fn current_room_seq(db: &impl ConnectionTrait, room_id: Uuid) -> i64 {

35 room_seq::Entity::find_by_id(room_id)

36 .one(db)

37 .await

38 .ok()

39 .flatten()

40 .map(|s| s.seq)

41 .unwrap_or(0)

42}

43

3644pub async fn record_mentions(

3745 db: &impl ConnectionTrait,

3846 room_id: Uuid,

@@ -143,6 +151,43 @@

143151 Ok(NotifyDecision::Notify)

144152}

145153

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

155/// Nothing, muted (even mid-window) and blocked settings opt out.

156pub async fn mention_email_allowed(

157 service: &ChatService,

158 room_id: Uuid,

159 space_id: Option<Uuid>,

160 uid: Uuid,

161) -> Result<bool, ChatError> {

162 if let Some(setting) = room_user_settings::Entity::find_by_id((uid, room_id))

163 .one(service.db().read())

164 .await?

165 {

166 if mute_active(setting.muted, setting.mute_until) {

167 return Ok(false);

168 }

169 match setting.notification_level {

170 RoomNotificationLevel::Nothing => return Ok(false),

171 RoomNotificationLevel::All | RoomNotificationLevel::Mentions => return Ok(true),

172 RoomNotificationLevel::Default => {}

173 }

174 }

175 if let Some(space_id) = space_id

176 && let Some(member) = space_members::Entity::find_by_id((space_id, uid))

177 .one(service.db().read())

178 .await?

179 {

180 if mute_active(member.muted, member.mute_until) {

181 return Ok(false);

182 }

183 return Ok(!matches!(

184 member.message_notifications,

185 MessageNotifications::Nothing

186 ));

187 }

188 Ok(true)

189}

190

146191pub async fn publish_message_event(

147192 service: &ChatService,

148193 room_id: Uuid,

@@ -165,35 +210,21 @@

165210 seq: i64,

166211) -> Result<bool, ChatError> {

167212 let now = OffsetDateTime::now_utc();

168 let result = read_states::Entity::update_many()

169 .filter(read_states::Column::UserId.eq(uid))

170 .filter(read_states::Column::RoomId.eq(room_id))

171 .filter(read_states::Column::LastReadSeq.lt(seq))

172 .col_expr(read_states::Column::LastReadSeq, Expr::value(seq))

173 .col_expr(read_states::Column::LastReadMessageId, Expr::value(message_id))

174 .col_expr(read_states::Column::LastViewedAt, Expr::value(now))

175 .col_expr(read_states::Column::UpdatedAt, Expr::value(now))

176 .exec(db)

213 let row = db

214 .query_one_raw(Statement::from_sql_and_values(

215 DatabaseBackend::Postgres,

216 r#"INSERT INTO chat_read_state

217 (user_id, room_id, last_read_message_id, last_read_seq, last_viewed_at, updated_at)

218 VALUES ($1, $2, $3, $4, $5, $5)

219 ON CONFLICT (user_id, room_id) DO UPDATE SET

220 last_read_message_id = EXCLUDED.last_read_message_id,

221 last_read_seq = EXCLUDED.last_read_seq,

222 last_viewed_at = EXCLUDED.last_viewed_at,

223 updated_at = EXCLUDED.updated_at

224 WHERE chat_read_state.last_read_seq < EXCLUDED.last_read_seq

225 RETURNING room_id"#,

226 [uid.into(), room_id.into(), message_id.into(), seq.into(), now.into()],

227 ))

177228 .await?;

178 if result.rows_affected > 0 {

179 return Ok(true);

180 }

181 let exists = read_states::Entity::find_by_id((uid, room_id))

182 .one(db)

183 .await?

184 .is_some();

185 if !exists {

186 read_states::ActiveModel {

187 user_id: Set(uid),

188 room_id: Set(room_id),

189 last_read_message_id: Set(message_id),

190 last_read_seq: Set(seq),

191 last_viewed_at: Set(Some(now)),

192 updated_at: Set(now),

193 }

194 .insert(db)

195 .await?;

196 return Ok(true);

197 }

198 Ok(false)

229 Ok(row.is_some())

199230}

Modified lib/chat/src/ws/mod.rs +132 −38 View lib/chat/src/ws/mod.rs

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

2626 pub ticket: String,

2727}

2828

29#[derive(Clone)]

2930enum Inbox {

3031 Room(Uuid, Result<String, ()>),

3132 User(String),

@@ -33,6 +34,7 @@

3334

3435struct Subscription {

3536 cancel: CancellationToken,

37 handle: tokio::task::JoinHandle<()>,

3638}

3739

3840pub async fn ws_upgrade(

@@ -42,15 +44,39 @@

4244) -> Response {

4345 let (uid, session_id) = match verify_ticket(&service.ticket_secret(), &query.ticket) {

4446 Ok(value) => value,

45 Err(_) => return ws.on_upgrade(|socket| async move {

46 let _ = close_with(socket, "invalid ticket").await;

47 }),

47 Err(_) => {

48 return ws.on_upgrade(|socket| async move {

49 let _ = close_with(socket, "invalid ticket").await;

50 });

51 }

4852 };

4953

5054 ws.max_message_size(service.config().chat.ws.max_frame_bytes)

5155 .on_upgrade(move |socket| handle_socket(socket, service, uid, session_id))

5256}

5357

58/// Push into the socket inbox without ever deadlocking the forwarder:

59/// bounded retry, then drop the payload (gap detection resyncs the client).

60async fn best_effort_send(

61 tx: &mpsc::Sender<Inbox>,

62 item: Inbox,

63 cancel: &CancellationToken,

64) -> bool {

65 for _ in 0..1000 {

66 match tx.try_send(item.clone()) {

67 Ok(()) => return true,

68 Err(mpsc::error::TrySendError::Full(_)) => {

69 tokio::select! {

70 _ = cancel.cancelled() => return false,

71 _ = tokio::time::sleep(Duration::from_millis(1)) => {}

72 }

73 }

74 Err(mpsc::error::TrySendError::Closed(_)) => return false,

75 }

76 }

77 false

78}

79

5480async fn close_with(mut socket: WebSocket, message: &str) -> Result<(), axum::Error> {

5581 let frame = ServerFrame::Bye {

5682 message: message.to_string(),

@@ -60,16 +86,12 @@

6086 socket.close().await

6187}

6288

63async fn handle_socket(

64 socket: WebSocket,

65 service: ChatService,

66 uid: Uuid,

67 session_id: Uuid,

68) {

89async fn handle_socket(socket: WebSocket, service: ChatService, uid: Uuid, session_id: Uuid) {

6990 let (mut sink, mut stream) = socket.split();

7091

7192 let (inbox_tx, mut inbox_rx) = mpsc::channel::<Inbox>(1024);

7293

94 let user_cancel = CancellationToken::new();

7395 let mut user_rx = match service.realtime.subscribe_user(uid).await {

7496 Ok(rx) => rx,

7597 Err(error) => {

@@ -78,22 +100,32 @@

78100 return;

79101 }

80102 };

81 {

103 let user_handle = {

82104 let tx = inbox_tx.clone();

105 let cancel = user_cancel.clone();

83106 tokio::spawn(async move {

84 while let Ok(payload) = user_rx.recv().await {

85 if tx.send(Inbox::User(payload)).await.is_err() {

86 break;

107 loop {

108 tokio::select! {

109 _ = cancel.cancelled() => break,

110 payload = user_rx.recv() => {

111 let Ok(payload) = payload else { break };

112 if !best_effort_send(&tx, Inbox::User(payload), &cancel).await {

113 break;

114 }

115 }

87116 }

88117 }

89 });

90 }

118 })

119 };

91120

92121 let ready = ServerFrame::Ready {

93122 user_id: uid,

94123 session_id,

95124 };

96125 if send_frame(&mut sink, &ready).await.is_err() {

126 user_cancel.cancel();

127 let _ = user_handle.await;

128 service.realtime.unsubscribe_user(uid).await;

97129 return;

98130 }

99131

@@ -106,6 +138,8 @@

106138 let mut subscriptions: HashMap<Uuid, Subscription> = HashMap::new();

107139 let mut last_seq: HashMap<Uuid, i64> = HashMap::new();

108140 let max_subscriptions = service.config().chat.ws.max_subscriptions;

141 let mut last_typing: Option<tokio::time::Instant> = None;

142 let mut tick_count: u32 = 0;

109143

110144 loop {

111145 tokio::select! {

@@ -125,6 +159,7 @@

125159 frame,

126160 &mut subscriptions,

127161 &mut last_seq,

162 &mut last_typing,

128163 &inbox_tx,

129164 max_subscriptions,

130165 )

@@ -191,12 +226,51 @@

191226 }).await;

192227 break;

193228 }

229 tick_count = tick_count.wrapping_add(1);

230 if tick_count % 2 == 0 {

231 let mut revoked: Vec<Uuid> = Vec::new();

232 for &room_id in subscriptions.keys() {

233 if service_can_read(&service, uid, room_id).await.is_err() {

234 revoked.push(room_id);

235 }

236 }

237 for room_id in revoked {

238 if let Some(sub) = subscriptions.remove(&room_id) {

239 sub.cancel.cancel();

240 let _ = sub.handle.await;

241 service.realtime.unsubscribe_room(room_id).await;

242 last_seq.remove(&room_id);

243 if send_frame(&mut sink, &ServerFrame::ResyncRequired {

244 room_id,

245 reason: "access revoked".to_string(),

246 })

247 .await

248 .is_err()

249 {

250 break;

251 }

252 }

253 }

254 }

194255 }

195256 }

196257 }

197258

198 for (_, sub) in subscriptions {

259 let mut teardown: Vec<(Uuid, tokio::task::JoinHandle<()>)> = Vec::new();

260 for (room_id, sub) in subscriptions {

199261 sub.cancel.cancel();

262 teardown.push((room_id, sub.handle));

263 }

264 user_cancel.cancel();

265 // Releasing the local sender unblocks any forwarder parked on a full

266 // inbox, guaranteeing the awaits below terminate.

267 drop(inbox_tx);

268 let _ = user_handle.await;

269 for (room_id, handle) in teardown {

270 // Await the forwarder so its broadcast receiver is released before

271 // the unsubscribe refcount check.

272 let _ = handle.await;

273 service.realtime.unsubscribe_room(room_id).await;

200274 }

201275 service.realtime.unsubscribe_user(uid).await;

202276}

@@ -209,6 +283,10 @@

209283 let event: serde_json::Value = serde_json::from_str(payload).ok()?;

210284 let seq = event.get("seq")?.as_i64()?;

211285 let last = last_seq.get(&room_id).copied();

286 // Only advance the cursor; edit/pin events may carry older seq values.

287 if last.is_some_and(|l| seq <= l) {

288 return None;

289 }

212290 last_seq.insert(room_id, seq);

213291 match last {

214292 Some(last) if seq > last + 1 => Some(ServerFrame::Gap {

@@ -228,6 +306,7 @@

228306 frame: ClientFrame,

229307 subscriptions: &mut HashMap<Uuid, Subscription>,

230308 last_seq: &mut HashMap<Uuid, i64>,

309 last_typing: &mut Option<tokio::time::Instant>,

231310 inbox_tx: &mpsc::Sender<Inbox>,

232311 max_subscriptions: usize,

233312) where

@@ -239,31 +318,40 @@

239318 return;

240319 }

241320 if subscriptions.len() >= max_subscriptions {

242 let _ = send_frame(sink, &ServerFrame::Error {

243 message: "subscription limit reached".to_string(),

244 })

321 let _ = send_frame(

322 sink,

323 &ServerFrame::Error {

324 message: "subscription limit reached".to_string(),

325 },

326 )

245327 .await;

246328 return;

247329 }

248330 if service_can_read(service, uid, room_id).await.is_err() {

249 let _ = send_frame(sink, &ServerFrame::Error {

250 message: "forbidden".to_string(),

251 })

331 let _ = send_frame(

332 sink,

333 &ServerFrame::Error {

334 message: "forbidden".to_string(),

335 },

336 )

252337 .await;

253338 return;

254339 }

255340 let rx = match service.realtime.subscribe_room(room_id).await {

256341 Ok(rx) => rx,

257342 Err(_) => {

258 let _ = send_frame(sink, &ServerFrame::Error {

259 message: "subscribe failed".to_string(),

260 })

343 let _ = send_frame(

344 sink,

345 &ServerFrame::Error {

346 message: "subscribe failed".to_string(),

347 },

348 )

261349 .await;

262350 return;

263351 }

264352 };

265353 let cancel = CancellationToken::new();

266 {

354 let handle = {

267355 let tx = inbox_tx.clone();

268356 let child = cancel.clone();

269357 let mut rx = rx;

@@ -272,25 +360,24 @@

272360 tokio::select! {

273361 _ = child.cancelled() => break,

274362 result = rx.recv() => {

275 match result {

276 Ok(payload) => {

277 if tx.send(Inbox::Room(room_id, Ok(payload))).await.is_err() {

278 break;

279 }

280 }

363 let item = match result {

364 Ok(payload) => Inbox::Room(room_id, Ok(payload)),

281365 Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {

282 if tx.send(Inbox::Room(room_id, Err(()))).await.is_err() {

283 break;

284 }

366 Inbox::Room(room_id, Err(()))

285367 }

286368 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,

369 };

370 // Non-blocking fallback: overflow drops the

371 // frame; seq gap detection triggers resync.

372 if !best_effort_send(&tx, item, &child).await {

373 break;

287374 }

288375 }

289376 }

290377 }

291 });

292 }

293 subscriptions.insert(room_id, Subscription { cancel });

378 })

379 };

380 subscriptions.insert(room_id, Subscription { cancel, handle });

294381

295382 let history_limit = service.config().chat.ws.history_limit;

296383 if let Ok(page) = service

@@ -322,6 +409,7 @@

322409 ClientFrame::Unsubscribe { room_id } => {

323410 if let Some(sub) = subscriptions.remove(&room_id) {

324411 sub.cancel.cancel();

412 let _ = sub.handle.await;

325413 last_seq.remove(&room_id);

326414 service.realtime.unsubscribe_room(room_id).await;

327415 }

@@ -364,7 +452,13 @@

364452 .await;

365453 }

366454 ClientFrame::Typing { room_id, thread_id } => {

455 // Coarse throttle: at most one typing broadcast per 3s per socket.

456 let now = tokio::time::Instant::now();

457 if last_typing.is_some_and(|prev| now.duration_since(prev).as_secs() < 3) {

458 return;

459 }

367460 if service_can_read(service, uid, room_id).await.is_ok() {

461 *last_typing = Some(now);

368462 let event = RoomEvent::Typing {

369463 room_id,

370464 user_id: uid,

Modified lib/chat/src/errors.rs +13 −1 View lib/chat/src/errors.rs

@@ -14,9 +14,16 @@

1414 Forbidden,

1515 NotFound,

1616 Conflict(String),

17 RateLimited(String),

1718 Internal(String),

1819}

1920

21impl ChatError {

22 pub fn is_rate_limited(&self) -> bool {

23 matches!(self, Self::RateLimited(_))

24 }

25}

26

2027impl fmt::Display for ChatError {

2128 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

2229 match self {

@@ -25,6 +32,7 @@

2532 Self::Forbidden => write!(f, "forbidden"),

2633 Self::NotFound => write!(f, "not found"),

2734 Self::Conflict(message) => write!(f, "{message}"),

35 Self::RateLimited(message) => write!(f, "{message}"),

2836 Self::Internal(message) => write!(f, "{message}"),

2937 }

3038 }

@@ -61,7 +69,7 @@

6169 }

6270}

6371

64fn is_unique_violation(err: &sea_orm::DbErr) -> bool {

72pub(crate) fn is_unique_violation(err: &sea_orm::DbErr) -> bool {

6573 let msg = err.to_string().to_lowercase();

6674 msg.contains("unique") || msg.contains("duplicate") || msg.contains("23505")

6775}

@@ -115,6 +123,8 @@

115123 NotFound(#[to_schema] AppResponseData<NoData>),

116124 #[response(status = 409, description = "resource conflict")]

117125 Conflict(#[to_schema] AppResponseData<NoData>),

126 #[response(status = 429, description = "rate limit exceeded")]

127 TooManyRequests(#[to_schema] AppResponseData<NoData>),

118128 #[response(status = 500, description = "internal server error")]

119129 Internal(#[to_schema] AppResponseData<NoData>),

120130}

@@ -131,6 +141,7 @@

131141 Self::Forbidden(envelope) => (403, envelope),

132142 Self::NotFound(envelope) => (404, envelope),

133143 Self::Conflict(envelope) => (409, envelope),

144 Self::TooManyRequests(envelope) => (429, envelope),

134145 Self::Internal(envelope) => (500, envelope),

135146 }

136147 }

@@ -155,6 +166,7 @@

155166 ChatError::Forbidden => Self::Forbidden(Self::payload(403, "forbidden")),

156167 ChatError::NotFound => Self::NotFound(Self::payload(404, "not found")),

157168 ChatError::Conflict(msg) => Self::Conflict(AppResponseData::err(409, msg)),

169 ChatError::RateLimited(msg) => Self::TooManyRequests(AppResponseData::err(429, msg)),

158170 ChatError::Internal(message) => {

159171 tracing::warn!(error = %message, "chat handler failed");

160172 Self::Internal(Self::payload(500, "internal server error"))

Modified lib/chat/src/worker.rs +74 −12 View lib/chat/src/worker.rs

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

11use std::time::Duration;

22

33use export::AppState;

4use models::users::user_recent_mentions;

45use models::users::user_scheduled_messages::{self, ScheduledMessageStatus};

6use sea_orm::sea_query::Expr;

57use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set};

68use time::OffsetDateTime;

79

@@ -9,9 +11,27 @@

911use crate::services::messages::SendMessageCmd;

1012use crate::services::service::ChatService;

1113

14pub(crate) const STUCK_CLAIM_MINUTES: i64 = 10;

15const MENTION_RETENTION_DAYS: i64 = 90;

16

1217impl ChatService {

1318 pub async fn deliver_due_scheduled_messages(&self) -> Result<u64, ChatError> {

1419 let now = OffsetDateTime::now_utc();

20

21 // Recover claims abandoned by a crashed instance.

22 let _ = user_scheduled_messages::Entity::update_many()

23 .col_expr(

24 user_scheduled_messages::Column::Status,

25 Expr::value(ScheduledMessageStatus::Pending),

26 )

27 .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Sending))

28 .filter(

29 user_scheduled_messages::Column::ScheduledAt

30 .lt(now - time::Duration::minutes(STUCK_CLAIM_MINUTES)),

31 )

32 .exec(self.db().write())

33 .await?;

34

1535 let due = user_scheduled_messages::Entity::find()

1636 .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Pending))

1737 .filter(user_scheduled_messages::Column::ScheduledAt.lte(now))

@@ -20,38 +40,74 @@

2040 .await?;

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

2242 for scheduled in due {

43 // Atomic claim so concurrent replicas deliver each row at most once.

44 let claimed = user_scheduled_messages::Entity::update_many()

45 .col_expr(

46 user_scheduled_messages::Column::Status,

47 Expr::value(ScheduledMessageStatus::Sending),

48 )

49 .filter(user_scheduled_messages::Column::Id.eq(scheduled.id))

50 .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Pending))

51 .exec(self.db().write())

52 .await?;

53 if claimed.rows_affected == 0 {

54 continue;

55 }

56

57 let id = scheduled.id;

58 let user_id = scheduled.user_id;

59 let channel_id = scheduled.target_channel_id;

2360 let cmd: SendMessageCmd = match serde_json::from_value(scheduled.payload.clone()) {

2461 Ok(cmd) => cmd,

2562 Err(error) => {

26 tracing::warn!(%error, id = %scheduled.id, "invalid scheduled message payload");

63 tracing::warn!(%error, %id, "invalid scheduled message payload");

2764 let mut am: user_scheduled_messages::ActiveModel = scheduled.into();

2865 am.status = Set(ScheduledMessageStatus::Cancelled);

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

3067 continue;

3168 }

3269 };

33 match self

34 .send_message(scheduled.user_id, scheduled.target_channel_id, cmd)

35 .await

36 {

37 Ok(_) => {

38 let mut am: user_scheduled_messages::ActiveModel = scheduled.into();

39 am.status = Set(ScheduledMessageStatus::Sent);

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

41 }

70 let next_status = match self.send_message(user_id, channel_id, cmd).await {

71 Ok(_) => ScheduledMessageStatus::Sent,

4272 Err(error) => {

43 tracing::warn!(%error, id = %scheduled.id, "scheduled message delivery failed");

73 tracing::warn!(%error, %id, "scheduled message delivery failed");

74 // Authz/validation failures are permanent; transient slow

75 // mode retries later.

76 if error.is_rate_limited() {

77 ScheduledMessageStatus::Pending

78 } else {

79 ScheduledMessageStatus::Failed

80 }

4481 }

45 }

82 };

83 user_scheduled_messages::Entity::update_many()

84 .col_expr(

85 user_scheduled_messages::Column::Status,

86 Expr::value(next_status),

87 )

88 .filter(user_scheduled_messages::Column::Id.eq(id))

89 .filter(user_scheduled_messages::Column::Status.eq(ScheduledMessageStatus::Sending))

90 .exec(self.db().write())

91 .await?;

4692 }

4793 Ok(count)

4894 }

95

96 pub async fn prune_old_mentions(&self) -> Result<(), ChatError> {

97 let cutoff = OffsetDateTime::now_utc() - time::Duration::days(MENTION_RETENTION_DAYS);

98 user_recent_mentions::Entity::delete_many()

99 .filter(user_recent_mentions::Column::CreatedAt.lt(cutoff))

100 .exec(self.db().write())

101 .await?;

102 Ok(())

103 }

49104}

50105

51106pub async fn run_background_tasks(state: AppState) {

52107 let service = ChatService::new(state);

53108 let mut interval = tokio::time::interval(Duration::from_secs(30));

54109 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

110 let mut tick: u32 = 0;

55111 loop {

56112 interval.tick().await;

57113 if let Err(error) = service.deliver_due_reminders().await {

@@ -60,5 +116,11 @@

60116 if let Err(error) = service.deliver_due_scheduled_messages().await {

61117 tracing::warn!(%error, "chat scheduled message delivery failed");

62118 }

119 tick = tick.wrapping_add(1);

120 if tick % 120 == 0

121 && let Err(error) = service.prune_old_mentions().await

122 {

123 tracing::warn!(%error, "chat mention prune failed");

124 }

63125 }

64126}

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

@@ -28,3 +28,4 @@

2828tokio-util = { workspace = true }

2929futures = { workspace = true }

3030bytes = { workspace = true }

31http = { workspace = true }

Modified lib/models/src/chat/message_reminders.rs +2 −0 View lib/models/src/chat/message_reminders.rs

@@ -25,6 +25,8 @@

2525 Done,

2626 #[sea_orm(num_value = 2)]

2727 Cancelled,

28 #[sea_orm(num_value = 3)]

29 Delivering,

2830}

2931

3032#[sea_orm::model]

Modified lib/models/src/users/user_scheduled_messages.rs +4 −0 View lib/models/src/users/user_scheduled_messages.rs

@@ -26,6 +26,10 @@

2626 Sent,

2727 #[sea_orm(string_value = "cancelled")]

2828 Cancelled,

29 #[sea_orm(string_value = "sending")]

30 Sending,

31 #[sea_orm(string_value = "failed")]

32 Failed,

2933}

3034

3135#[sea_orm::model]

Modified lib/models/schema_extras.sql +3 −0 View lib/models/schema_extras.sql

@@ -166,6 +166,9 @@

166166CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_dm_pair

167167 ON rooms (lower_name) WHERE kind = 15 AND space_id IS NULL;

168168

169CREATE UNIQUE INDEX IF NOT EXISTS idx_chat_message_room_nonce

170 ON chat_message (room_id, author_id, nonce) WHERE nonce IS NOT NULL;

171

169172ALTER TABLE chat_message ADD COLUMN IF NOT EXISTS search_vector tsvector;

170173CREATE INDEX IF NOT EXISTS idx_chat_message_fts

171174 ON chat_message USING gin (search_vector);

Modified lib/webapi/src/lib.rs +5 −0 View lib/webapi/src/lib.rs

@@ -3,9 +3,13 @@

33use axum::Router;

44use export::AppState;

55use session::{SessionLayer, store};

6use tower_http::limit::RequestBodyLimitLayer;

67use tower_http::trace::{DefaultMakeSpan, DefaultOnRequest, DefaultOnResponse, TraceLayer};

78use tracing::Level;

89

10/// Max chat JSON request body (covers the 512 KiB base64 emoji upload).

11const MAX_REQUEST_BODY_BYTES: usize = 2 * 1024 * 1024;

12

913pub async fn router(state: AppState) -> Result<Router, axum_session::SessionError> {

1014 let store = store::connect(state.redis.clone(), &state.config.session).await?;

1115

@@ -24,6 +28,7 @@

2428 .merge(user::routes::router(user.clone()).with_state(user))

2529 .merge(git::routes::router(git.clone()).with_state(git))

2630 .layer(SessionLayer::new(store))

31 .layer(RequestBodyLimitLayer::new(MAX_REQUEST_BODY_BYTES))

2732 .layer(

2833 TraceLayer::new_for_http()

2934 .make_span_with(

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

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

10851085 "export",

10861086 "futures",

10871087 "hmac 0.13.0",

1088 "http 0.2.12",

10881089 "models",

10891090 "rand 0.8.7",

10901091 "redis",

@@ -8133,6 +8134,7 @@

81338134 "bytes",

81348135 "http 1.5.0",

81358136 "http-body 1.1.0",

8137 "http-body-util",

81368138 "percent-encoding 2.3.2",

81378139 "pin-project-lite",

81388140 "tower-layer",

Modified Cargo.toml +1 −1 View Cargo.toml

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

5454serde_json = "1.0.151"

5555axum = { version = "0.8.9", features = ["ws"] }

5656axum_session = { version = "0.21.0", features = ["key-store"] }

57tower-http = { version = "0.7.0", features = ["trace"] }

57tower-http = { version = "0.7.0", features = ["trace", "limit"] }

5858tracing = "0.1.44"

5959figment = { version = "0.10", features = ["toml", "env"] }

6060dotenvy = "0.15"

Modified openapi.json View openapi.json

Diff too large to display.