use api::users::UserDto; use models::chat::attachments::{self as attachment_rows}; use models::chat::messages::{self, AuthorType, ChatMessageType}; use models::chat::thread_members::{self, ThreadNotifyLevel}; use models::chat::{message_edits, reactions, threads}; use models::room::rooms::{self, RoomKind}; use models::users::user_notifications::NotificationType; use sea_orm::sea_query::Expr; use sea_orm::{ ActiveModelTrait, ColumnTrait, Condition, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect, Set, TransactionTrait, }; use serde::{Deserialize, Serialize}; use serde_json::json; use time::OffsetDateTime; use utoipa::{IntoParams, ToSchema}; use uuid::Uuid; use crate::errors::{ChatError, is_unique_violation}; use crate::services::attachments::AttachmentDto; /// Drop both encodings of an attachment's derivative when the last reference to /// its original disappears. Declared here because message deletion owns the /// lifetime of the attachment rows. async fn delete_attachment_previews(service: &ChatService, original_key: &str) { let key_base = ::media::preview_key_base_of(original_key); for candidate in imaging::layout::preview_candidate_keys(&key_base) { if let Err(error) = service.media.client.as_ref().delete(&candidate).await { tracing::warn!(%error, key = %candidate, "failed to delete attachment preview"); } } } use crate::services::reactions::ReactionAggDto; use crate::services::service::ChatService; use crate::share::fanout::{self, Job}; use crate::share::lookup::{ RoomContext, require_room_readable, require_room_writable, space_slug_for, }; use crate::share::notify::{ NotifyDecision, advance_read_state, current_room_seq, next_room_seq, notification_decision, notify_user, publish_message_event, record_mentions, }; use crate::share::realtime::RoomEvent; use crate::share::{authz, mention}; const NONCE_CONFLICT: &str = "chat-nonce-conflict"; #[derive(Clone, Copy)] pub(crate) enum DirectMessageKind { Direct, Group, } /// Fan a DM message out as OS pushes. /// /// Runs after commit, best-effort: a push that cannot be enqueued never fails /// the message write. The per-recipient gate is deliberately the same /// `notification_decision` the mention path uses, so "muted" and "mentions /// only" mean the same thing here as everywhere else. #[allow(clippy::too_many_arguments)] pub(crate) async fn broadcast_direct_message_push( service: &ChatService, kind: DirectMessageKind, room_id: Uuid, space_id: Option, message_id: Uuid, actor: i64, body: String, recipients: &[i64], ) { for &uid in recipients { match notification_decision(service, room_id, space_id, uid, false, false).await { Ok(NotifyDecision::Notify) => {} _ => continue, } let locale = service.audience.recipient_locale(uid).await; let title = match kind { DirectMessageKind::Direct => i18n::catalog::notification_direct_message_title(locale), DirectMessageKind::Group => i18n::catalog::notification_group_dm_title(locale), }; push::protocol::enqueue( &service.participants.redis, &push::protocol::PushJob { user_id: uid, category: push::protocol::PushCategory::Dm, title: title.to_string(), body: Some(body.clone()), data: Some(json!({ "room_id": room_id, "message_id": message_id, "space_name": space_slug_for(service, space_id).await, })), actor_id: Some(actor), notification_id: None, trace: oltp::propagation::traceparent(), request_id: api::middleware::request_id(), targets: None, }, ) .await; } } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct MessageDto { pub id: Uuid, pub room_id: Uuid, pub author: Option, pub author_id: i64, pub author_type: AuthorType, pub webhook_id: Option, pub thread_id: Option, pub thread_broadcast: bool, pub reply_to_id: Option, pub forwarded_from_id: Option, pub message_type: ChatMessageType, pub content: String, pub mentions: Vec, pub mention_everyone: bool, pub attachments: Vec, #[serde(default)] pub embeds: Vec, pub has_thread: bool, pub pinned: bool, pub pinned_by_id: Option, pub pinned_at: Option, pub edited_at: Option, pub deleted_at: Option, pub seq: i64, pub created_at: OffsetDateTime, #[serde(default)] pub reactions: Vec, } pub use social::MessageEmbed as MessageEmbedDto; /// 消息页：在统一分页字段之上补回游标翻页所需的派生字段。 /// /// `has_more`/`oldest_seq`/`newest_seq` 由本页内容推导，客户端无需自己算； /// `items` 恒按 `seq` 升序，因此 `oldest_seq` 取首条、`newest_seq` 取末条。 #[derive(Debug, Clone, Serialize, ToSchema)] pub struct MessagePage { pub items: Vec, pub total: u64, pub limit: u64, pub offset: u64, pub has_more: bool, pub oldest_seq: Option, pub newest_seq: Option, } impl MessagePage { pub fn new(items: Vec, total: u64, limit: u64, offset: u64) -> Self { let has_more = offset.saturating_add(items.len() as u64) < total; let oldest_seq = items.first().map(|message| message.seq); let newest_seq = items.last().map(|message| message.seq); Self { items, total, limit, offset, has_more, oldest_seq, newest_seq, } } } #[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] pub struct SendMessageCmd { #[serde(default)] pub content: String, #[serde(default)] pub reply_to_id: Option, #[serde(default)] pub thread_id: Option, #[serde(default)] pub thread_broadcast: bool, #[serde(default)] pub attachments: Vec, /// client-generated idempotency key, unique per room; retrying with the /// same nonce returns the original message instead of duplicating #[serde(default)] pub nonce: Option, #[serde(default)] pub urgent: bool, #[serde(default)] pub forwarded_from: Option, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct UpdateMessageCmd { pub content: String, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct BulkDeleteCmd { pub message_ids: Vec, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct PinCmd { pub pinned: bool, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct ForwardMessageCmd { pub target_room_id: Uuid, #[serde(default)] pub comment: Option, } #[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)] #[serde(deny_unknown_fields)] pub struct ListMessageQuery { pub before_seq: Option, pub after_seq: Option, pub limit: Option, pub offset: Option, pub thread_id: Option, pub author_id: Option, } #[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)] pub struct MessagePath { pub room_id: Uuid, pub message_id: Uuid, } #[derive(Debug, Clone, Serialize, ToSchema)] pub struct MessageEditDto { pub id: Uuid, pub message_id: Uuid, pub editor: Option, pub content: String, pub created_at: OffsetDateTime, } const MAX_BULK_DELETE: usize = 100; impl ChatService { pub(crate) async fn build_message_dto( &self, rows: Vec, viewer: i64, ) -> Result, ChatError> { if rows.is_empty() { return Ok(Vec::new()); } let author_ids: Vec = rows.iter().map(|m| m.author_id).collect(); let user_map = self.audience.load_user_briefs(&author_ids).await?; let message_ids: Vec = rows.iter().map(|m| m.id).collect(); let attachment_rows = attachment_rows::Entity::find() .filter(attachment_rows::Column::MessageId.is_in(message_ids.clone())) .all(self.database.read()) .await?; let mut attachments_by_msg: std::collections::HashMap> = std::collections::HashMap::new(); for a in attachment_rows { attachments_by_msg .entry(a.message_id.unwrap_or(Uuid::nil())) .or_default() .push(AttachmentDto::from(a)); } let reaction_rows = reactions::Entity::find() .filter(reactions::Column::MessageId.is_in(message_ids.clone())) .all(self.database.read()) .await?; let mut reaction_agg: std::collections::HashMap<(Uuid, String), (i64, bool)> = std::collections::HashMap::new(); for r in reaction_rows { let entry = reaction_agg .entry((r.message_id, r.emoji.clone())) .or_insert((0, false)); entry.0 += 1; if r.user_id == viewer { entry.1 = true; } } let items = rows .into_iter() .map(|m| { let mentions: Vec = serde_json::from_value(m.mentions.clone()).unwrap_or_default(); MessageDto { id: m.id, room_id: m.room_id, author: user_map.get(&m.author_id).cloned(), author_id: m.author_id, author_type: m.author_type, webhook_id: m.webhook_id, thread_id: m.thread_id, thread_broadcast: m.thread_broadcast, reply_to_id: m.reply_to_id, forwarded_from_id: m.forwarded_from_id, message_type: m.message_type, content: m.content, mentions, mention_everyone: m.mention_everyone, attachments: attachments_by_msg.remove(&m.id).unwrap_or_default(), embeds: serde_json::from_value(m.embeds.clone()).unwrap_or_default(), has_thread: m.flags & messages::FLAG_HAS_THREAD != 0, pinned: m.pinned, pinned_by_id: m.pinned_by, pinned_at: m.pinned_at, edited_at: m.edited_at, deleted_at: m.deleted_at, seq: m.seq, created_at: m.created_at, reactions: reaction_agg .iter() .filter(|((mid, _), _)| *mid == m.id) .map(|((_, emoji), (count, me))| ReactionAggDto { emoji: emoji.clone(), count: *count, me: *me, }) .collect(), } }) .collect(); Ok(items) } async fn check_slow_mode(&self, ctx: &RoomContext, caller: i64) -> Result<(), ChatError> { let Some(limit) = ctx.room.rate_limit_per_user else { return Ok(()); }; if limit <= 0 { return Ok(()); } if let Some(space) = &ctx.space && authz::can_manage_space(self, space, caller).await? { return Ok(()); } let last = messages::Entity::find() .filter(messages::Column::RoomId.eq(ctx.room.id)) .filter(messages::Column::AuthorId.eq(caller)) .order_by_desc(messages::Column::Seq) .one(self.database.read()) .await?; if let Some(last) = last { let elapsed = OffsetDateTime::now_utc() - last.created_at; if elapsed.whole_seconds() < limit as i64 { let retry_after = limit as i64 - elapsed.whole_seconds(); return Err(ChatError::RateLimited(format!( "slow mode active, retry in {retry_after}s" ))); } } Ok(()) } #[tracing::instrument( name = "chat.send_message", skip_all, fields(caller = %caller, room_id = %room_id) )] pub async fn send_message( &self, caller: i64, room_id: Uuid, cmd: SendMessageCmd, ) -> Result { self.send_message_internal(caller, room_id, cmd, Vec::new()) .await } pub(crate) async fn send_message_internal( &self, caller: i64, room_id: Uuid, cmd: SendMessageCmd, attachment_copies: Vec, ) -> Result { let ctx = require_room_writable(self, room_id, caller).await?; self.check_slow_mode(&ctx, caller).await?; if cmd.attachments.len() > self.settings.chat.max_attachments_per_message { return Err(ChatError::BadRequest(format!( "attachments must be at most {}", self.settings.chat.max_attachments_per_message ))); } let forwarded_from = match cmd.forwarded_from { Some(orig_id) => { let orig = messages::Entity::find_by_id(orig_id) .one(self.database.read()) .await? .filter(|m| m.deleted_at.is_none()) .ok_or(ChatError::NotFound)?; require_room_readable(self, orig.room_id, caller).await?; Some(orig.id) } None => None, }; let content_len = cmd.content.chars().count(); if cmd.content.trim().is_empty() && cmd.attachments.is_empty() && forwarded_from.is_none() { return Err(ChatError::BadRequest( "message content is empty".to_string(), )); } if content_len > self.settings.chat.max_message_len { return Err(ChatError::BadRequest("message too long".to_string())); } if let Some(nonce) = &cmd.nonce { let existing = messages::Entity::find() .filter(messages::Column::RoomId.eq(room_id)) .filter(messages::Column::AuthorId.eq(caller)) .filter(messages::Column::Nonce.eq(nonce.as_str())) .one(self.database.read()) .await?; if let Some(existing) = existing { let items = self.build_message_dto(vec![existing], caller).await?; return Ok(items.into_iter().next().unwrap()); } } let mut thread: Option = None; if let Some(thread_id) = cmd.thread_id { let t = threads::Entity::find_by_id(thread_id) .one(self.database.read()) .await? .filter(|t| t.room_id == room_id) .ok_or(ChatError::NotFound)?; if t.archived || t.locked { return Err(ChatError::Forbidden); } thread = Some(t); } // The parent author is a notification target in its own right: replying // to someone is the same kind of address as mentioning them, and the // `reply` notification type exists precisely for it. let reply_parent_author = if let Some(reply_to) = cmd.reply_to_id { let parent = messages::Entity::find_by_id(reply_to) .one(self.database.read()) .await? .filter(|p| p.room_id == room_id && p.deleted_at.is_none()) .ok_or(ChatError::NotFound)?; Some(parent.author_id) } else { None }; let parsed = mention::parse_mentions(self, &ctx, &cmd.content, caller).await?; let mention_uids = parsed.users.clone(); // Resolve every mentioned recipient's language up front: the // notification rows are written inside the transaction and each one has // to carry the recipient's own copy. let fallback_locale = self.audience.default_locale; let mention_locales = i18n::resolve_user_locales(self.database.read(), &mention_uids, fallback_locale).await; if !cmd.attachments.is_empty() { let owned = attachment_rows::Entity::find() .filter(attachment_rows::Column::Id.is_in(cmd.attachments.clone())) .filter(attachment_rows::Column::UserId.eq(caller)) .filter(attachment_rows::Column::MessageId.is_null()) .filter( attachment_rows::Column::Status.eq(attachment_rows::AttachmentStatus::Uploaded), ) .count(self.database.read()) .await?; if owned != cmd.attachments.len() as u64 { return Err(ChatError::BadRequest( "attachment not found or already used".to_string(), )); } } let now = OffsetDateTime::now_utc(); let message_id = Uuid::now_v7(); let content = cmd.content.clone(); let mentions_json = json!(parsed.users); let groups_json = json!(parsed.groups); let copy_ids: Vec<(Uuid, attachment_rows::Model)> = attachment_copies .into_iter() .map(|row| (Uuid::now_v7(), row)) .collect(); let mut attachment_ids = cmd.attachments.clone(); attachment_ids.extend(copy_ids.iter().map(|(id, _)| *id)); let attachments_json = json!(attachment_ids); let embeds_json = json!([]); let mention_everyone = if parsed.everyone { match &ctx.space { Some(space) => authz::can_manage_space(self, space, caller).await?, None => true, } } else { false }; let flags = if cmd.urgent { models::chat::messages::FLAG_URGENT } else { 0 }; let message_type = if cmd.thread_id.is_some() && cmd.reply_to_id.is_none() { ChatMessageType::Default } else if cmd.reply_to_id.is_some() { ChatMessageType::Reply } else { ChatMessageType::Default }; let room = ctx.room.clone(); let thread_clone = thread.clone(); let mention_uids_for_txn = mention_uids.clone(); let mention_locales_for_txn = mention_locales.clone(); let reply_parent_for_txn = reply_parent_author; let thread_for_replies = thread.clone(); let content_for_email = content.clone(); let copy_ids_for_txn = copy_ids.clone(); let svc = self.clone(); let space_id = room.space_id; let nonce_retry = cmd.nonce.clone(); let pending = push::notify::PendingNotifications::default(); let pending_txn = pending.clone(); let txn_result = self .database .primary() .transaction::<_, (), ChatError>(move |txn| { Box::pin(async move { let seq = next_room_seq(txn, room.id).await?; let row = messages::ActiveModel { id: Set(message_id), room_id: Set(room.id), author_id: Set(caller), author_type: Set(AuthorType::User), webhook_id: Set(None), thread_id: Set(cmd.thread_id), thread_broadcast: Set(cmd.thread_broadcast), reply_to_id: Set(cmd.reply_to_id), forwarded_from_id: Set(cmd.forwarded_from), message_type: Set(message_type), content: Set(content.clone()), mentions: Set(mentions_json), mention_everyone: Set(mention_everyone), mention_group_ids: Set(groups_json), attachments: Set(attachments_json), embeds: Set(embeds_json), flags: Set(flags), nonce: Set(cmd.nonce.clone()), pinned: Set(false), pinned_by: Set(None), pinned_at: Set(None), edited_at: Set(None), deleted_at: Set(None), seq: Set(seq), created_at: Set(now), }; match row.insert(txn).await { Ok(_) => {} Err(err) if is_unique_violation(&err) && err.to_string().contains("idx_chat_message_room_nonce") => { return Err(ChatError::Conflict(NONCE_CONFLICT.to_string())); } Err(err) => return Err(ChatError::from(err)), } if !cmd.attachments.is_empty() { attachment_rows::Entity::update_many() .filter(attachment_rows::Column::Id.is_in(cmd.attachments.clone())) .filter(attachment_rows::Column::UserId.eq(caller)) .col_expr(attachment_rows::Column::MessageId, Expr::value(message_id)) .col_expr(attachment_rows::Column::RoomId, Expr::value(room.id)) .col_expr(attachment_rows::Column::UpdatedAt, Expr::value(now)) .exec(txn) .await?; } for (new_id, src) in ©_ids_for_txn { attachment_rows::ActiveModel { id: Set(*new_id), user_id: Set(caller), room_id: Set(Some(room.id)), message_id: Set(Some(message_id)), filename: Set(src.filename.clone()), description: Set(src.description.clone()), content_type: Set(src.content_type.clone()), size: Set(src.size), bucket: Set(src.bucket.clone()), s3_key: Set(src.s3_key.clone()), status: Set(attachment_rows::AttachmentStatus::Uploaded), width: Set(src.width), height: Set(src.height), duration_secs: Set(src.duration_secs), variants: Set(src.variants.clone()), created_at: Set(now), updated_at: Set(now), } .insert(txn) .await?; } rooms::Entity::update_many() .filter(rooms::Column::Id.eq(room.id)) .col_expr(rooms::Column::LastMessageId, Expr::value(message_id)) .col_expr(rooms::Column::UpdatedAt, Expr::value(now)) .exec(txn) .await?; if let Some(t) = &thread_clone { crate::services::threads::maintain_thread_on_reply( txn, t, message_id, caller, now, ) .await?; } record_mentions(txn, room.id, message_id, &mention_uids_for_txn).await?; let data = json!({ "room_id": room.id, "message_id": message_id, "seq": seq, "thread_id": cmd.thread_id, "space_name": space_slug_for(&svc, space_id).await, }); let body = if content.chars().count() > 120 { content.chars().take(120).collect::() } else { content.clone() }; for &uid in &mention_uids_for_txn { if notification_decision(&svc, room.id, space_id, uid, true, false).await? == NotifyDecision::Skip { continue; } let locale = mention_locales_for_txn .get(&uid) .copied() .unwrap_or(fallback_locale); notify_user( txn, &pending_txn, uid, NotificationType::Mention, i18n::catalog::notification_mention_message_title(locale).to_string(), Some(body.clone()), Some(data.clone()), Some(caller), ) .await?; } // Reply notifications: the parent's author plus everyone // following the thread. A person who was also @-mentioned // gets one row, not two — the mention is the more specific // address, so it wins. let mut reply_targets: Vec = Vec::new(); if let Some(author) = reply_parent_for_txn && author != caller && !mention_uids_for_txn.contains(&author) { reply_targets.push(author); } if let Some(thread) = &thread_for_replies { let watchers = thread_members::Entity::find() .filter(thread_members::Column::ThreadId.eq(thread.id)) .filter(thread_members::Column::NotifyLevel.eq(ThreadNotifyLevel::All)) .all(txn) .await .unwrap_or_default(); for watcher in watchers { if watcher.user_id != caller && !mention_uids_for_txn.contains(&watcher.user_id) && !reply_targets.contains(&watcher.user_id) { reply_targets.push(watcher.user_id); } } } if !reply_targets.is_empty() { let reply_locales = i18n::resolve_user_locales(txn, &reply_targets, fallback_locale).await; for &uid in &reply_targets { if notification_decision(&svc, room.id, space_id, uid, true, false) .await? == NotifyDecision::Skip { continue; } notify_user( txn, &pending_txn, uid, NotificationType::Reply, i18n::catalog::notification_reply_title( reply_locales.get(&uid).copied().unwrap_or(fallback_locale), ) .to_string(), Some(body.clone()), Some(data.clone()), Some(caller), ) .await?; } } Ok(()) }) }) .await .map_err(ChatError::from); if let Err(ChatError::Conflict(ref msg)) = txn_result && msg == NONCE_CONFLICT && let Some(nonce) = &nonce_retry { let existing = messages::Entity::find() .filter(messages::Column::RoomId.eq(room_id)) .filter(messages::Column::AuthorId.eq(caller)) .filter(messages::Column::Nonce.eq(nonce.as_str())) .one(self.database.read()) .await?; if let Some(existing) = existing { let items = self.build_message_dto(vec![existing], caller).await?; return Ok(items.into_iter().next().unwrap()); } } txn_result?; pending.flush(&self.audience.notifications).await; let row = messages::Entity::find_by_id(message_id) .one(self.database.read()) .await? .ok_or(ChatError::Internal("message vanished after insert".into()))?; let seq = row.seq; let items = self.build_message_dto(vec![row.clone()], caller).await?; let dto = items.into_iter().next().unwrap(); let dto_value = serde_json::to_value(&dto)?; publish_message_event(self, room_id, seq, dto_value).await; // A thread reply moves the topic's counters, but the message event alone // cannot refresh them; the room index listens for this one. if let Some(t) = &thread { let event = RoomEvent::ThreadUpdated { room_id, thread_id: t.id, }; self.audience.realtime.publish_room(&event).await; } let _ = self .sender_mark_read(caller, room_id, message_id, seq) .await; if mention_everyone { let body = if content_for_email.chars().count() > 120 { content_for_email.chars().take(120).collect::() } else { content_for_email.clone() }; fanout::submit(Job::EveryoneStart { scope: crate::share::notify::EveryoneScope { room_id, space_id: ctx.room.space_id, is_private: ctx.room.is_private, message_id, body, actor: caller, explicit: mention_uids.clone(), request_id: api::middleware::request_id(), trace: oltp::propagation::traceparent(), }, }); } crate::services::unfurl::spawn_unfurl(self, room_id, message_id, content_for_email.clone()); let recipients: Vec = mention_uids .iter() .copied() .filter(|uid| *uid != caller) .collect(); for uid in recipients { self.audience .realtime .publish_user( uid, &json!({"kind": "mention", "room_id": room_id, "message_id": message_id}), ) .await; } // DMs and group DMs get an OS push but no inbox row: the inbox is for // things that ask something of you, and a direct message is already // counted by the chat unread badge. Mute and notification level still // apply, so a muted conversation stays silent. if let Some(kind) = { let room = &room; match room.kind { RoomKind::DirectMessage => Some(DirectMessageKind::Direct), RoomKind::GroupDm => Some(DirectMessageKind::Group), _ => None, } } { let recipients = crate::share::notify::dm_recipients(self, room_id, caller).await; let body = if content_for_email.chars().count() > 120 { content_for_email.chars().take(120).collect::() } else { content_for_email.clone() }; if !recipients.is_empty() { fanout::submit(Job::DirectMessagePush { kind, room_id, space_id, message_id, actor: caller, body, recipients, request_id: api::middleware::request_id(), trace: oltp::propagation::traceparent(), }); } } if !mention_uids.is_empty() { let room_name = ctx.room.name.clone(); let email_recipients: Vec = mention_uids .iter() .copied() .filter(|uid| *uid != caller) .collect(); if !email_recipients.is_empty() { let email_body = if content_for_email.chars().count() > 200 { content_for_email.chars().take(200).collect::() } else { content_for_email.clone() }; fanout::submit(Job::MentionEmails { room_id, space_id, room_name, body: email_body, recipients: email_recipients, request_id: api::middleware::request_id(), trace: oltp::propagation::traceparent(), }); } } Ok(dto) } pub async fn list_messages( &self, caller: i64, room_id: Uuid, q: ListMessageQuery, ) -> Result { let _ctx = require_room_readable(self, room_id, caller).await?; let (limit, offset) = api::pagination::page_bounds( q.limit, q.offset, self.settings.chat.default_pagination, self.settings.chat.max_pagination, )?; let mut select = messages::Entity::find() .filter(messages::Column::RoomId.eq(room_id)) .filter(messages::Column::DeletedAt.is_null()); match q.thread_id { Some(thread_id) => { select = select.filter(messages::Column::ThreadId.eq(thread_id)); } None => { select = select.filter( Condition::any() .add(messages::Column::ThreadId.is_null()) .add(messages::Column::ThreadBroadcast.eq(true)), ); } } if let Some(author_id) = q.author_id { select = select.filter(messages::Column::AuthorId.eq(author_id)); } let (rows, total) = if let Some(after_seq) = q.after_seq { let select = select.filter(messages::Column::Seq.gt(after_seq)); let total = select.clone().count(self.database.read()).await?; let rows = select .order_by_asc(messages::Column::Seq) .limit(limit) .offset(offset) .all(self.database.read()) .await?; (rows, total) } else { let mut select = select; if let Some(before_seq) = q.before_seq { select = select.filter(messages::Column::Seq.lt(before_seq)); } let total = select.clone().count(self.database.read()).await?; let mut rows = select .order_by_desc(messages::Column::Seq) .limit(limit) .offset(offset) .all(self.database.read()) .await?; rows.reverse(); (rows, total) }; let items = self.build_message_dto(rows, caller).await?; Ok(MessagePage::new(items, total, limit, offset)) } pub async fn get_message( &self, caller: i64, room_id: Uuid, message_id: Uuid, ) -> Result { let _ctx = require_room_readable(self, room_id, caller).await?; let row = messages::Entity::find_by_id(message_id) .filter(messages::Column::RoomId.eq(room_id)) .one(self.database.read()) .await? .filter(|m| m.deleted_at.is_none()) .ok_or(ChatError::NotFound)?; let items = self.build_message_dto(vec![row], caller).await?; Ok(items.into_iter().next().unwrap()) } pub async fn update_message( &self, caller: i64, room_id: Uuid, message_id: Uuid, cmd: UpdateMessageCmd, ) -> Result { let ctx = require_room_readable(self, room_id, caller).await?; let row = messages::Entity::find_by_id(message_id) .filter(messages::Column::RoomId.eq(room_id)) .one(self.database.read()) .await? .filter(|m| m.deleted_at.is_none()) .ok_or(ChatError::NotFound)?; let is_manager = match &ctx.space { Some(space) => authz::can_manage_space(self, space, caller).await?, None => false, }; if row.author_id != caller && !is_manager { return Err(ChatError::Forbidden); } let content = cmd.content.trim().to_string(); if content.is_empty() { return Err(ChatError::BadRequest("content is empty".to_string())); } if content.chars().count() > self.settings.chat.max_message_len { return Err(ChatError::BadRequest("message too long".to_string())); } if content == row.content { let items = self.build_message_dto(vec![row], caller).await?; return Ok(items.into_iter().next().unwrap()); } let old_content = row.content.clone(); let old_attachments = row.attachments.clone(); let now = OffsetDateTime::now_utc(); let parsed = mention::parse_mentions(self, &ctx, &content, caller).await?; let fallback_locale = self.audience.default_locale; let new_mention_locales = i18n::resolve_user_locales(self.database.read(), &parsed.users, fallback_locale).await; let mention_everyone = if parsed.everyone { match &ctx.space { Some(space) => authz::can_manage_space(self, space, caller).await?, None => true, } } else { false }; let old_mentions: Vec = serde_json::from_value(row.mentions.clone()).unwrap_or_default(); let new_mentions: Vec = parsed .users .iter() .copied() .filter(|u| !old_mentions.contains(u)) .collect(); let mentions_json = json!(parsed.users); let groups_json = json!(parsed.groups); let new_mentions_txn = new_mentions.clone(); let space_id = ctx.space.as_ref().map(|s| s.id()); let content_for_notify = content.clone(); let this = self.clone(); let this_txn = self.clone(); let pending = push::notify::PendingNotifications::default(); let pending_txn = pending.clone(); self.database .primary() .transaction::<_, (), ChatError>(move |txn| { Box::pin(async move { message_edits::ActiveModel { id: Set(Uuid::now_v7()), message_id: Set(message_id), editor_id: Set(caller), content: Set(old_content), attachments: Set(old_attachments), created_at: Set(now), } .insert(txn) .await?; messages::Entity::update_many() .filter(messages::Column::Id.eq(message_id)) .col_expr(messages::Column::Content, Expr::value(content)) .col_expr(messages::Column::Mentions, Expr::value(mentions_json)) .col_expr( messages::Column::MentionEveryone, Expr::value(mention_everyone), ) .col_expr(messages::Column::MentionGroupIds, Expr::value(groups_json)) .col_expr(messages::Column::EditedAt, Expr::value(now)) .exec(txn) .await?; if !new_mentions_txn.is_empty() { record_mentions(txn, room_id, message_id, &new_mentions_txn).await?; let body = if content_for_notify.chars().count() > 120 { content_for_notify.chars().take(120).collect::() } else { content_for_notify.clone() }; let data = json!({ "room_id": room_id, "message_id": message_id, "space_name": space_slug_for(&this_txn, space_id).await, }); for &uid in &new_mentions_txn { if notification_decision(&this_txn, room_id, space_id, uid, true, false) .await? == NotifyDecision::Skip { continue; } notify_user( txn, &pending_txn, uid, NotificationType::Mention, i18n::catalog::notification_mention_message_title( new_mention_locales .get(&uid) .copied() .unwrap_or(fallback_locale), ) .to_string(), Some(body.clone()), Some(data.clone()), Some(caller), ) .await?; } } Ok(()) }) }) .await?; pending.flush(&self.audience.notifications).await; let row = messages::Entity::find_by_id(message_id) .one(self.database.read()) .await? .ok_or(ChatError::Internal("message vanished after update".into()))?; let seq = current_room_seq(self.database.read(), room_id).await; let items = self.build_message_dto(vec![row], caller).await?; let dto = items.into_iter().next().unwrap(); let dto_value = serde_json::to_value(&dto)?; let event = RoomEvent::MessageEdited { room_id, seq, message: dto_value, }; this.audience.realtime.publish_room(&event).await; crate::services::outbound::enqueue_outbound_event( self, room_id, "message_edited", &serde_json::to_value(&event)?, ) .await; for &uid in &new_mentions { self.audience .realtime .publish_user( uid, &json!({"kind": "mention", "room_id": room_id, "message_id": message_id}), ) .await; } crate::services::unfurl::spawn_unfurl(self, room_id, message_id, dto.content.clone()); Ok(dto) } pub async fn delete_message( &self, caller: i64, room_id: Uuid, message_id: Uuid, ) -> Result<(), ChatError> { self.delete_messages_internal(caller, room_id, vec![message_id]) .await } pub async fn bulk_delete_messages( &self, caller: i64, room_id: Uuid, cmd: BulkDeleteCmd, ) -> Result<(), ChatError> { if cmd.message_ids.is_empty() || cmd.message_ids.len() > MAX_BULK_DELETE { return Err(ChatError::BadRequest(format!( "message_ids must contain 1..={MAX_BULK_DELETE} entries" ))); } self.delete_messages_internal(caller, room_id, cmd.message_ids) .await } async fn delete_messages_internal( &self, caller: i64, room_id: Uuid, message_ids: Vec, ) -> Result<(), ChatError> { let ctx = require_room_readable(self, room_id, caller).await?; let is_manager = match &ctx.space { Some(space) => authz::can_manage_space(self, space, caller).await?, None => false, }; let rows = messages::Entity::find() .filter(messages::Column::Id.is_in(message_ids.clone())) .filter(messages::Column::RoomId.eq(room_id)) .filter(messages::Column::DeletedAt.is_null()) .all(self.database.read()) .await?; if rows.is_empty() { return Ok(()); } for row in &rows { if row.author_id != caller && !is_manager { return Err(ChatError::Forbidden); } } let row_ids: Vec = rows.iter().map(|r| r.id).collect(); let single = row_ids.len() == 1; let first_id = row_ids[0]; let thread_roots: Vec = threads::Entity::find() .filter(threads::Column::RootMessageId.is_in(row_ids.clone())) .all(self.database.read()) .await? .into_iter() .map(|t| t.id) .collect(); let now = OffsetDateTime::now_utc(); let ids = row_ids.clone(); self.database .primary() .transaction::<_, (), ChatError>(move |txn| { Box::pin(async move { messages::Entity::update_many() .filter(messages::Column::Id.is_in(ids.clone())) .col_expr(messages::Column::Content, Expr::value(String::new())) .col_expr(messages::Column::Attachments, Expr::value(json!([]))) .col_expr(messages::Column::DeletedAt, Expr::value(now)) .exec(txn) .await?; if !thread_roots.is_empty() { threads::Entity::update_many() .filter(threads::Column::Id.is_in(thread_roots.clone())) .col_expr(threads::Column::Archived, Expr::value(true)) .col_expr(threads::Column::UpdatedAt, Expr::value(now)) .exec(txn) .await?; } Ok(()) }) }) .await?; let attached = attachment_rows::Entity::find() .filter(attachment_rows::Column::MessageId.is_in(row_ids.clone())) .all(self.database.read()) .await?; for attachment in attached { let shared = attachment_rows::Entity::find() .filter(attachment_rows::Column::S3Key.eq(attachment.s3_key.clone())) .filter(attachment_rows::Column::Id.ne(attachment.id)) .count(self.database.read()) .await .unwrap_or(1); let _ = attachment_rows::Entity::delete_by_id(attachment.id) .exec(self.database.write()) .await; if shared == 0 { let _ = self.media.client.as_ref().delete(&attachment.s3_key).await; // The derivative is an orphan once the last reference to the // original is gone; leaving it behind would leak an object that // nothing can ever address again. delete_attachment_previews(self, &attachment.s3_key).await; } } let seq = current_room_seq(self.database.read(), room_id).await; let event = if single { RoomEvent::MessageDeleted { room_id, seq, user_id: caller, message_id: first_id, } } else { RoomEvent::MessagesDeleted { room_id, seq, user_id: caller, message_ids: row_ids.clone(), } }; self.audience.realtime.publish_room(&event).await; crate::services::outbound::enqueue_outbound_event( self, room_id, if single { "message_deleted" } else { "messages_deleted" }, &serde_json::to_value(&event)?, ) .await; let _ = ctx; Ok(()) } pub async fn set_message_pin( &self, caller: i64, room_id: Uuid, message_id: Uuid, cmd: PinCmd, ) -> Result { let ctx = require_room_readable(self, room_id, caller).await?; if ctx.space.is_some() && !authz::can_manage_space(self, ctx.space.as_ref().unwrap(), caller).await? { return Err(ChatError::Forbidden); } let row = messages::Entity::find_by_id(message_id) .filter(messages::Column::RoomId.eq(room_id)) .one(self.database.read()) .await? .filter(|m| m.deleted_at.is_none()) .ok_or(ChatError::NotFound)?; if row.pinned == cmd.pinned { let items = self.build_message_dto(vec![row], caller).await?; return Ok(items.into_iter().next().unwrap()); } let now = OffsetDateTime::now_utc(); self.database .primary() .transaction::<_, (), ChatError>(move |txn| { Box::pin(async move { let mut update = messages::Entity::update_many() .filter(messages::Column::Id.eq(message_id)) .col_expr(messages::Column::Pinned, Expr::value(cmd.pinned)); if cmd.pinned { update = update .col_expr(messages::Column::PinnedBy, Expr::value(caller)) .col_expr(messages::Column::PinnedAt, Expr::value(now)); } else { update = update .col_expr(messages::Column::PinnedBy, Expr::value(None::)) .col_expr( messages::Column::PinnedAt, Expr::value(None::), ); } update.exec(txn).await?; Ok(()) }) }) .await?; let seq = current_room_seq(self.database.read(), room_id).await; let event = RoomEvent::MessagePinned { room_id, seq, user_id: caller, message_id, pinned: cmd.pinned, }; self.audience.realtime.publish_room(&event).await; self.get_message(caller, room_id, message_id).await } pub async fn list_pins( &self, caller: i64, room_id: Uuid, ) -> Result, ChatError> { let _ctx = require_room_readable(self, room_id, caller).await?; let rows = messages::Entity::find() .filter(messages::Column::RoomId.eq(room_id)) .filter(messages::Column::Pinned.eq(true)) .filter(messages::Column::DeletedAt.is_null()) .order_by_desc(messages::Column::PinnedAt) .all(self.database.read()) .await?; self.build_message_dto(rows, caller).await } pub async fn forward_message( &self, caller: i64, room_id: Uuid, message_id: Uuid, cmd: ForwardMessageCmd, ) -> Result { let _ctx = require_room_readable(self, room_id, caller).await?; let original = messages::Entity::find_by_id(message_id) .filter(messages::Column::RoomId.eq(room_id)) .one(self.database.read()) .await? .filter(|m| m.deleted_at.is_none()) .ok_or(ChatError::NotFound)?; let original_attachments = attachment_rows::Entity::find() .filter(attachment_rows::Column::MessageId.eq(original.id)) .all(self.database.read()) .await?; let send_cmd = SendMessageCmd { content: cmd.comment.unwrap_or_default(), reply_to_id: None, thread_id: None, thread_broadcast: false, attachments: Vec::new(), nonce: None, urgent: false, forwarded_from: Some(original.id), }; let dto = self .send_message_internal(caller, cmd.target_room_id, send_cmd, original_attachments) .await?; Ok(dto) } pub async fn list_message_history( &self, caller: i64, room_id: Uuid, message_id: Uuid, ) -> Result, ChatError> { let _ctx = require_room_readable(self, room_id, caller).await?; let message = messages::Entity::find_by_id(message_id) .filter(messages::Column::RoomId.eq(room_id)) .one(self.database.read()) .await? .filter(|m| m.deleted_at.is_none()) .ok_or(ChatError::NotFound)?; let _ = message; let rows = message_edits::Entity::find() .filter(message_edits::Column::MessageId.eq(message_id)) .order_by_desc(message_edits::Column::CreatedAt) .all(self.database.read()) .await?; let editor_ids: Vec = rows.iter().map(|r| r.editor_id).collect(); let user_map = self.audience.load_user_briefs(&editor_ids).await?; Ok(rows .into_iter() .map(|r| MessageEditDto { id: r.id, message_id: r.message_id, editor: user_map.get(&r.editor_id).cloned(), content: r.content, created_at: r.created_at, }) .collect()) } pub(crate) async fn sender_mark_read( &self, caller: i64, room_id: Uuid, message_id: Uuid, seq: i64, ) -> Result<(), ChatError> { advance_read_state( self.database.write(), caller, room_id, Some(message_id), seq, ) .await?; Ok(()) } }