main
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<Uuid>,
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<UserDto>,
pub author_id: i64,
pub author_type: AuthorType,
pub webhook_id: Option<Uuid>,
pub thread_id: Option<Uuid>,
pub thread_broadcast: bool,
pub reply_to_id: Option<Uuid>,
pub forwarded_from_id: Option<Uuid>,
pub message_type: ChatMessageType,
pub content: String,
pub mentions: Vec<i64>,
pub mention_everyone: bool,
pub attachments: Vec<AttachmentDto>,
#[serde(default)]
pub embeds: Vec<MessageEmbedDto>,
pub has_thread: bool,
pub pinned: bool,
pub pinned_by_id: Option<i64>,
pub pinned_at: Option<OffsetDateTime>,
pub edited_at: Option<OffsetDateTime>,
pub deleted_at: Option<OffsetDateTime>,
pub seq: i64,
pub created_at: OffsetDateTime,
#[serde(default)]
pub reactions: Vec<ReactionAggDto>,
}
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<MessageDto>,
pub total: u64,
pub limit: u64,
pub offset: u64,
pub has_more: bool,
pub oldest_seq: Option<i64>,
pub newest_seq: Option<i64>,
}
impl MessagePage {
pub fn new(items: Vec<MessageDto>, 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<Uuid>,
#[serde(default)]
pub thread_id: Option<Uuid>,
#[serde(default)]
pub thread_broadcast: bool,
#[serde(default)]
pub attachments: Vec<Uuid>,
/// client-generated idempotency key, unique per room; retrying with the
/// same nonce returns the original message instead of duplicating
#[serde(default)]
pub nonce: Option<String>,
#[serde(default)]
pub urgent: bool,
#[serde(default)]
pub forwarded_from: Option<Uuid>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateMessageCmd {
pub content: String,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct BulkDeleteCmd {
pub message_ids: Vec<Uuid>,
}
#[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<String>,
}
#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
#[serde(deny_unknown_fields)]
pub struct ListMessageQuery {
pub before_seq: Option<i64>,
pub after_seq: Option<i64>,
pub limit: Option<u64>,
pub offset: Option<u64>,
pub thread_id: Option<Uuid>,
pub author_id: Option<i64>,
}
#[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<UserDto>,
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<messages::Model>,
viewer: i64,
) -> Result<Vec<MessageDto>, ChatError> {
if rows.is_empty() {
return Ok(Vec::new());
}
let author_ids: Vec<i64> = rows.iter().map(|m| m.author_id).collect();
let user_map = self.audience.load_user_briefs(&author_ids).await?;
let message_ids: Vec<Uuid> = 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<Uuid, Vec<AttachmentDto>> =
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<i64> =
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<MessageDto, ChatError> {
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<attachment_rows::Model>,
) -> Result<MessageDto, ChatError> {
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<threads::Model> = 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::<String>()
} 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<i64> = 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::<String>()
} 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<i64> = 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::<String>()
} 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<i64> = 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::<String>()
} 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<MessagePage, ChatError> {
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<MessageDto, ChatError> {
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<MessageDto, ChatError> {
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<i64> =
serde_json::from_value(row.mentions.clone()).unwrap_or_default();
let new_mentions: Vec<i64> = 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::<String>()
} 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<Uuid>,
) -> 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<Uuid> = rows.iter().map(|r| r.id).collect();
let single = row_ids.len() == 1;
let first_id = row_ids[0];
let thread_roots: Vec<Uuid> = 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<MessageDto, ChatError> {
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::<i64>))
.col_expr(
messages::Column::PinnedAt,
Expr::value(None::<OffsetDateTime>),
);
}
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<Vec<MessageDto>, 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<MessageDto, ChatError> {
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<Vec<MessageEditDto>, 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<i64> = 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(())
}
}