194c44c
use models::chat::messages::{self, ChatMessageType};
use models::chat::read_states;
use models::chat::room_seq;
use models::chat::room_user_settings;
use models::dir::dirs::{self, DirKind};
use models::room::room_invites;
use models::room::room_members::{self, RoomMemberRole};
use models::room::rooms::{self, RoomKind};
use models::users::user_notifications::NotificationType;
use rand::Rng;
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, ExprTrait, QueryFilter, QueryOrder, QuerySelect,
Set, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use space::services::spaces::UserDto;
use space::share::authz::{MANAGE_INVITES, effective_permissions};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::authz::can_manage_space;
use crate::share::lookup::{
is_room_member, load_room_context, lookup_room, lookup_space_by_name, require_room_readable,
require_space_member,
};
use crate::share::notify::{next_room_seq, notify_user};
use crate::share::realtime::RoomEvent;
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct RoomBriefDto {
pub id: Uuid,
pub space_id: Option<Uuid>,
pub owner_id: i64,
pub owner: Option<UserDto>,
pub kind: RoomKind,
pub name: String,
pub topic: Option<String>,
pub position: i64,
pub parent_id: Option<Uuid>,
pub is_private: bool,
pub nsfw: bool,
pub rate_limit_per_user: Option<i32>,
pub last_message_id: Option<Uuid>,
pub unread_count: i64,
pub unread_mentions: i64,
pub favorite: bool,
pub muted: bool,
pub hidden: bool,
pub member_count: i64,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateRoomCmd {
#[serde(default)]
pub name: Option<String>,
#[serde(default)]
pub topic: Option<String>,
#[serde(default)]
pub is_private: Option<bool>,
#[serde(default)]
pub nsfw: Option<bool>,
#[serde(default)]
pub rate_limit_per_user: Option<i32>,
#[serde(default)]
pub position: Option<i64>,
/// Move the room under another directory node of the same space; `null` puts
/// it back at the space root. Absent means "leave the parent alone".
#[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
pub parent_id: Option<Option<Uuid>>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct AcceptRoomInviteCmd {
pub code: String,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct AddMembersCmd {
pub user_ids: Vec<i64>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct RoomMemberDto {
pub user: space::services::spaces::UserDto,
pub role: RoomMemberRole,
pub nick: Option<String>,
pub joined_at: OffsetDateTime,
}
#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct RoomMemberQuery {
pub limit: Option<u64>,
pub offset: Option<u64>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateRoomInviteCmd {
#[serde(default)]
pub invitee_id: Option<i64>,
#[serde(default)]
pub expires_secs: Option<i64>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct RoomInviteDto {
pub id: Uuid,
pub room_id: Uuid,
pub inviter_id: i64,
pub inviter: Option<UserDto>,
pub invitee_id: Option<i64>,
pub invitee: Option<UserDto>,
pub accept_code: String,
pub created_at: OffsetDateTime,
pub expires_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct RoomInviteListItemDto {
pub id: Uuid,
pub room_id: Uuid,
pub inviter_id: i64,
pub inviter: Option<UserDto>,
pub invitee_id: Option<i64>,
pub invitee: Option<UserDto>,
pub created_at: OffsetDateTime,
pub expires_at: Option<OffsetDateTime>,
pub accepted_at: Option<OffsetDateTime>,
pub revoked_at: Option<OffsetDateTime>,
}
fn random_accept_code() -> String {
const CHARSET: &[u8] = b"abcdefghjkmnpqrstuvwxyzABCDEFGHJKMNPQRSTUVWXYZ23456789";
let mut rng = rand::thread_rng();
(0..12)
.map(|_| CHARSET[rng.gen_range(0..CHARSET.len())] as char)
.collect()
}
impl ChatService {
pub async fn list_space_rooms(
&self,
caller: i64,
space_name: &str,
) -> Result<Vec<RoomBriefDto>, ChatError> {
let space = lookup_space_by_name(self, space_name).await?;
let member = require_space_member(self, &space, caller).await?;
let rooms = rooms::Entity::find()
.filter(rooms::Column::SpaceId.eq(space.id))
.filter(rooms::Column::IsActive.eq(true))
.order_by_asc(rooms::Column::Position)
.all(self.db().read())
.await?;
let room_ids: Vec<Uuid> = rooms.iter().map(|r| r.id).collect();
let memberships = room_members::Entity::find()
.filter(room_members::Column::RoomId.is_in(room_ids.clone()))
.filter(room_members::Column::UserId.eq(caller))
.all(self.db().read())
.await?;
let member_of: std::collections::HashSet<Uuid> = memberships
.into_iter()
.filter(|m| m.left_at.is_none())
.map(|m| m.room_id)
.collect();
let settings = room_user_settings::Entity::find()
.filter(room_user_settings::Column::UserId.eq(caller))
.filter(room_user_settings::Column::RoomId.is_in(room_ids.clone()))
.all(self.db().read())
.await?;
let settings_map: std::collections::HashMap<Uuid, room_user_settings::Model> =
settings.into_iter().map(|s| (s.room_id, s)).collect();
let read = read_states::Entity::find()
.filter(read_states::Column::UserId.eq(caller))
.filter(read_states::Column::RoomId.is_in(room_ids.clone()))
.all(self.db().read())
.await?;
let read_map: std::collections::HashMap<Uuid, (i64, Option<OffsetDateTime>)> = read
.into_iter()
.map(|r| (r.room_id, (r.last_read_seq, r.last_viewed_at)))
.collect();
let seqs = room_seq::Entity::find()
.filter(room_seq::Column::RoomId.is_in(room_ids.clone()))
.all(self.db().read())
.await?;
let seq_map: std::collections::HashMap<Uuid, i64> =
seqs.into_iter().map(|s| (s.room_id, s.seq)).collect();
let mention_rows = models::users::user_recent_mentions::Entity::find()
.filter(models::users::user_recent_mentions::Column::UserId.eq(caller))
.filter(models::users::user_recent_mentions::Column::ChannelId.is_in(room_ids.clone()))
.select_only()
.column(models::users::user_recent_mentions::Column::ChannelId)
.column(models::users::user_recent_mentions::Column::CreatedAt)
.into_tuple::<(Uuid, OffsetDateTime)>()
.all(self.db().read())
.await?;
let mut mentions_by_room: std::collections::HashMap<Uuid, Vec<OffsetDateTime>> =
std::collections::HashMap::new();
for (channel_id, created_at) in mention_rows {
mentions_by_room
.entry(channel_id)
.or_default()
.push(created_at);
}
let member_counts: Vec<(Uuid, i64)> = room_members::Entity::find()
.filter(room_members::Column::RoomId.is_in(room_ids.clone()))
.filter(room_members::Column::LeftAt.is_null())
.select_only()
.column(room_members::Column::RoomId)
.column_as(Expr::col(room_members::Column::UserId).count(), "cnt")
.group_by(room_members::Column::RoomId)
.into_tuple()
.all(self.db().read())
.await?;
let count_map: std::collections::HashMap<Uuid, i64> = member_counts.into_iter().collect();
let owner_ids: Vec<i64> = rooms
.iter()
.map(|r| r.owner_id)
.collect::<Vec<_>>()
.into_iter()
.collect();
let owner_map = self.load_user_briefs(&owner_ids).await?;
let mut result = Vec::new();
for room in rooms {
if room.is_private && !member_of.contains(&room.id) && room.owner_id != caller {
continue;
}
let (last_read, last_viewed) = read_map.get(&room.id).copied().unwrap_or((0, None));
let tip = seq_map.get(&room.id).copied().unwrap_or(0);
let unread_count = std::cmp::max(tip - last_read, 0);
let unread_mentions = mentions_by_room
.get(&room.id)
.map(|list| match last_viewed {
Some(at) => list.iter().filter(|c| **c > at).count() as i64,
None => list.len() as i64,
})
.unwrap_or(0);
let setting = settings_map.get(&room.id);
if member.hide_muted_channels
&& setting.is_some_and(|s| crate::share::notify::mute_active(s.muted, s.mute_until))
{
continue;
}
result.push(RoomBriefDto {
id: room.id,
owner_id: room.owner_id,
owner: owner_map.get(&room.owner_id).cloned(),
space_id: room.space_id,
kind: room.kind,
name: room.name,
topic: room.topic,
position: room.position,
parent_id: room.parent_id,
is_private: room.is_private,
nsfw: room.nsfw,
rate_limit_per_user: room.rate_limit_per_user,
last_message_id: room.last_message_id,
unread_count,
unread_mentions,
favorite: setting.is_some_and(|s| s.favorite),
muted: setting.is_some_and(|s| s.muted),
hidden: setting.is_some_and(|s| s.hidden),
member_count: count_map.get(&room.id).copied().unwrap_or(0),
created_at: room.created_at,
});
}
Ok(result)
}
pub async fn get_room(&self, caller: i64, room_id: Uuid) -> Result<RoomBriefDto, ChatError> {
let ctx = require_room_readable(self, room_id, caller).await?;
let room = ctx.room;
let tip = room_seq::Entity::find_by_id(room.id)
.one(self.db().read())
.await?
.map(|s| s.seq)
.unwrap_or(0);
let last_read = read_states::Entity::find_by_id((caller, room.id))
.one(self.db().read())
.await?
.map(|r| r.last_read_seq)
.unwrap_or(0);
let setting = room_user_settings::Entity::find_by_id((caller, room.id))
.one(self.db().read())
.await?;
let member_count = self.room_member_count(room.id).await? as i64;
let owner_map = self.load_user_briefs(&[room.owner_id]).await?;
Ok(RoomBriefDto {
id: room.id,
owner_id: room.owner_id,
owner: owner_map.get(&room.owner_id).cloned(),
space_id: room.space_id,
kind: room.kind,
name: room.name,
topic: room.topic,
position: room.position,
parent_id: room.parent_id,
is_private: room.is_private,
nsfw: room.nsfw,
rate_limit_per_user: room.rate_limit_per_user,
last_message_id: room.last_message_id,
unread_count: std::cmp::max(tip - last_read, 0),
unread_mentions: 0,
favorite: setting.as_ref().is_some_and(|s| s.favorite),
muted: setting.as_ref().is_some_and(|s| s.muted),
hidden: setting.as_ref().is_some_and(|s| s.hidden),
member_count,
created_at: room.created_at,
})
}
pub async fn update_room(
&self,
caller: i64,
room_id: Uuid,
cmd: UpdateRoomCmd,
) -> Result<RoomBriefDto, ChatError> {
let ctx = require_room_readable(self, room_id, caller).await?;
// Anything beyond a plain topic tweak on a DM is a management operation.
let needs_manage = ctx.space.is_some()
&& (cmd.name.is_some()
|| cmd.is_private.is_some()
|| cmd.position.is_some()
|| cmd.topic.is_some()
|| cmd.nsfw.is_some()
|| cmd.rate_limit_per_user.is_some()
|| cmd.parent_id.is_some());
if needs_manage {
match &ctx.space {
Some(space) => {
if !can_manage_space(self, space, caller).await? {
return Err(ChatError::Forbidden);
}
}
None => return Err(ChatError::Forbidden),
}
}
if ctx.space.is_none() && (cmd.name.is_some() || cmd.is_private.is_some()) {
return Err(ChatError::Forbidden);
}
if let Some(rate) = cmd.rate_limit_per_user
&& !(0..=3600).contains(&rate)
{
return Err(ChatError::BadRequest(
"rate_limit_per_user must be within 0..=3600".to_string(),
));
}
let now = OffsetDateTime::now_utc();
let room = ctx.room.clone();
let mut am: rooms::ActiveModel = room.clone().into();
let mut topic_changed: Option<(Option<String>, String)> = None;
if let Some(name) = cmd.name {
let name = name.trim().to_string();
if name.is_empty() || name.chars().count() > 100 {
return Err(ChatError::BadRequest("invalid room name".to_string()));
}
let lower = name.to_lowercase();
if !lower
.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
{
return Err(ChatError::BadRequest("invalid room name".to_string()));
}
am.name = Set(name);
am.lower_name = Set(lower);
}
if let Some(topic) = cmd.topic {
if topic.chars().count() > self.config().chat.max_topic_len {
return Err(ChatError::BadRequest("topic too long".to_string()));
}
if topic != room.topic.clone().unwrap_or_default() {
topic_changed = Some((room.topic.clone(), topic.clone()));
}
am.topic = Set(if topic.is_empty() { None } else { Some(topic) });
}
if let Some(is_private) = cmd.is_private {
am.is_private = Set(is_private);
}
if let Some(nsfw) = cmd.nsfw {
am.nsfw = Set(nsfw);
}
if let Some(rate) = cmd.rate_limit_per_user {
am.rate_limit_per_user = Set(Some(rate));
}
if let Some(position) = cmd.position {
am.position = Set(position);
}
// The sidebar's drag-and-drop files channels into folders, so the target
// must be a real directory node of *this* room's space (never another
// space's, and never an object node such as a repo placeholder).
if let Some(parent) = cmd.parent_id {
let next = match parent {
None => None,
Some(dir_id) => {
let space_id = ctx
.space
.as_ref()
.map(|space| space.id)
.ok_or(ChatError::Forbidden)?;
let node = dirs::Entity::find_by_id(dir_id)
.filter(dirs::Column::SpaceId.eq(space_id))
.one(self.db().read())
.await?
.ok_or(ChatError::NotFound)?;
if node.kind != DirKind::Dir {
return Err(ChatError::BadRequest(
"room parent must be a directory node".to_string(),
));
}
Some(dir_id)
}
};
if next != ctx.room.parent_id {
am.parent_id = Set(next);
}
}
am.updated_at = Set(now);
am.update(self.db().write()).await?;
if let Some((_old, new_topic)) = topic_changed {
let body = format!("set the channel topic to: {new_topic}");
let _ = self
.post_system_message(caller, room_id, ChatMessageType::ChannelTopic, body)
.await;
}
let event = RoomEvent::RoomUpdated { room_id };
self.realtime.publish_room(&event).await;
self.get_room(caller, room_id).await
}
pub(crate) async fn post_system_message(
&self,
author: i64,
room_id: Uuid,
mtype: ChatMessageType,
content: String,
) -> Result<(), ChatError> {
let now = OffsetDateTime::now_utc();
let message_id = Uuid::now_v7();
let seq = self
.db()
.primary()
.transaction::<_, i64, ChatError>(move |txn| {
Box::pin(async move {
let seq = next_room_seq(txn, room_id).await?;
messages::ActiveModel {
id: Set(message_id),
room_id: Set(room_id),
author_id: Set(author),
author_type: Set(models::chat::messages::AuthorType::User),
webhook_id: Set(None),
thread_id: Set(None),
thread_broadcast: Set(false),
reply_to_id: Set(None),
forwarded_from_id: Set(None),
message_type: Set(mtype),
content: Set(content),
mentions: Set(serde_json::json!([])),
mention_everyone: Set(false),
mention_group_ids: Set(serde_json::json!([])),
attachments: Set(serde_json::json!([])),
embeds: Set(serde_json::json!([])),
flags: Set(0),
nonce: Set(None),
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),
}
.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?;
Ok(seq)
})
})
.await?;
let row = messages::Entity::find_by_id(message_id)
.one(self.db().read())
.await?;
if let Some(row) = row {
let items = self.build_message_dto(vec![row], author).await?;
if let Some(dto) = items.into_iter().next() {
let value = serde_json::to_value(&dto)?;
crate::share::notify::publish_message_event(self, room_id, seq, value).await;
}
}
Ok(())
}
pub async fn join_room(&self, caller: i64, room_id: Uuid) -> Result<(), ChatError> {
let room = lookup_room(self, room_id).await?;
let ctx = load_room_context(self, room).await?;
match &ctx.space {
Some(space) => {
require_space_member(self, space, caller).await?;
if ctx.room.is_private {
return Err(ChatError::Forbidden);
}
}
None => return Err(ChatError::Forbidden),
}
let now = OffsetDateTime::now_utc();
let existing = room_members::Entity::find_by_id((room_id, caller))
.one(self.db().write())
.await?;
match existing {
Some(row) if row.left_at.is_some() => {
let mut am: room_members::ActiveModel = row.into();
am.left_at = Set(None);
am.updated_at = Set(Some(now));
am.update(self.db().write()).await?;
}
Some(_) => return Ok(()),
None => {
room_members::ActiveModel {
room_id: Set(room_id),
user_id: Set(caller),
role: Set(RoomMemberRole::Member),
nick: Set(None),
joined_at: Set(now),
left_at: Set(None),
created_at: Set(now),
updated_at: Set(Some(now)),
}
.insert(self.db().write())
.await?;
}
}
let event = RoomEvent::MemberJoined {
room_id,
user_id: caller,
};
self.realtime.publish_room(&event).await;
Ok(())
}
pub async fn leave_room(&self, caller: i64, room_id: Uuid) -> Result<(), ChatError> {
let room = lookup_room(self, room_id).await?;
if room.space_id.is_none() {
return Err(ChatError::BadRequest("use DM close instead".to_string()));
}
let now = OffsetDateTime::now_utc();
if room.owner_id == caller {
let successor = room_members::Entity::find()
.filter(room_members::Column::RoomId.eq(room_id))
.filter(room_members::Column::UserId.ne(caller))
.filter(room_members::Column::LeftAt.is_null())
.order_by_asc(room_members::Column::JoinedAt)
.one(self.db().write())
.await?;
match successor {
Some(next) => {
let next_uid = next.user_id;
let mut am: room_members::ActiveModel = next.into();
am.role = Set(RoomMemberRole::Owner);
am.updated_at = Set(Some(now));
am.update(self.db().write()).await?;
let mut room_am: rooms::ActiveModel = room.clone().into();
room_am.owner_id = Set(next_uid);
room_am.updated_at = Set(now);
room_am.update(self.db().write()).await?;
}
None => {
let mut room_am: rooms::ActiveModel = room.clone().into();
room_am.is_active = Set(false);
room_am.updated_at = Set(now);
room_am.update(self.db().write()).await?;
// 房间没有成员后停用，房间级索引任务会先清空它的消息文档。
if let Err(error) =
search::outbox::enqueue_room_messages(self.db().write(), room.id).await
{
tracing::warn!(%room_id, error = %error, "search room purge enqueue failed");
}
}
}
}
let row = room_members::Entity::find_by_id((room_id, caller))
.one(self.db().write())
.await?
.ok_or(ChatError::NotFound)?;
if row.left_at.is_none() {
let mut am: room_members::ActiveModel = row.into();
am.left_at = Set(Some(now));
am.updated_at = Set(Some(now));
am.update(self.db().write()).await?;
let event = RoomEvent::MemberLeft {
room_id,
user_id: caller,
};
self.realtime.publish_room(&event).await;
}
Ok(())
}
pub async fn add_room_members(
&self,
caller: i64,
room_id: Uuid,
cmd: AddMembersCmd,
) -> Result<(), ChatError> {
if cmd.user_ids.is_empty() || cmd.user_ids.len() > self.config().chat.max_members_per_add {
return Err(ChatError::BadRequest(format!(
"user_ids must contain 1..={} entries",
self.config().chat.max_members_per_add
)));
}
let ctx = require_room_readable(self, room_id, caller).await?;
match &ctx.space {
Some(space) => {
if !can_manage_space(self, space, caller).await? {
return Err(ChatError::Forbidden);
}
}
None => {
if !is_room_member(self, room_id, caller).await? {
return Err(ChatError::Forbidden);
}
let limit = ctx.room.user_limit.unwrap_or(i32::MAX);
let mut active = self.room_member_count(room_id).await? as i64;
for uid in &cmd.user_ids {
if !is_room_member(self, room_id, *uid).await? {
active += 1;
}
if active > limit as i64 {
return Err(ChatError::BadRequest(
"room member limit reached".to_string(),
));
}
}
}
}
for uid in cmd.user_ids {
if let Some(space) = &ctx.space
&& require_space_member(self, space, uid).await.is_err()
{
continue;
}
self.ensure_room_membership(room_id, uid).await?;
}
Ok(())
}
pub(crate) async fn ensure_room_membership(
&self,
room_id: Uuid,
uid: i64,
) -> Result<(), ChatError> {
let now = OffsetDateTime::now_utc();
let existing = room_members::Entity::find_by_id((room_id, uid))
.one(self.db().write())
.await?;
match existing {
Some(row) if row.left_at.is_some() => {
let mut am: room_members::ActiveModel = row.into();
am.left_at = Set(None);
am.updated_at = Set(Some(now));
am.update(self.db().write()).await?;
}
Some(_) => return Ok(()),
None => {
room_members::ActiveModel {
room_id: Set(room_id),
user_id: Set(uid),
role: Set(RoomMemberRole::Member),
nick: Set(None),
joined_at: Set(now),
left_at: Set(None),
created_at: Set(now),
updated_at: Set(Some(now)),
}
.insert(self.db().write())
.await?;
}
}
let event = RoomEvent::MemberJoined {
room_id,
user_id: uid,
};
self.realtime.publish_room(&event).await;
self.realtime
.publish_user(
uid,
&serde_json::json!({"kind": "room_added", "room_id": room_id}),
)
.await;
Ok(())
}
pub async fn list_room_members(
&self,
caller: i64,
room_id: Uuid,
q: RoomMemberQuery,
) -> Result<Vec<RoomMemberDto>, ChatError> {
let _ctx = require_room_readable(self, room_id, caller).await?;
let limit = q
.limit
.unwrap_or(self.config().chat.default_pagination)
.clamp(1, self.config().chat.max_pagination);
let rows = room_members::Entity::find()
.filter(room_members::Column::RoomId.eq(room_id))
.filter(room_members::Column::LeftAt.is_null())
.order_by_asc(room_members::Column::JoinedAt)
.limit(limit)
.offset(
q.offset
.unwrap_or(0)
.clamp(0, export::pagination::MAX_OFFSET),
)
.all(self.db().read())
.await?;
let uids: Vec<i64> = rows.iter().map(|r| r.user_id).collect();
let user_map = self.load_user_briefs(&uids).await?;
Ok(rows
.into_iter()
.filter_map(|r| {
user_map.get(&r.user_id).map(|u| RoomMemberDto {
role: r.role,
user: u.clone(),
nick: r.nick,
joined_at: r.joined_at,
})
})
.collect())
}
pub async fn create_room_invite(
&self,
caller: i64,
room_id: Uuid,
cmd: CreateRoomInviteCmd,
) -> Result<RoomInviteDto, ChatError> {
let ctx = require_room_readable(self, room_id, caller).await?;
let Some(space) = &ctx.space else {
return Err(ChatError::BadRequest("cannot invite to a DM".to_string()));
};
if !can_manage_space(self, space, caller).await?
&& !effective_permissions(&self.state, space, caller)
.await?
.has(MANAGE_INVITES)
{
return Err(ChatError::Forbidden);
}
let now = OffsetDateTime::now_utc();
let id = Uuid::now_v7();
let accept_code = random_accept_code();
let expires_at = match cmd.expires_secs {
Some(secs) if secs > 0 && secs <= 30 * 86_400 => {
Some(now + time::Duration::seconds(secs))
}
Some(_) => {
return Err(ChatError::BadRequest(
"expires_secs must be within 1..=2592000".to_string(),
));
}
None => None,
};
let row = room_invites::ActiveModel {
id: Set(id),
room_id: Set(room_id),
inviter_id: Set(caller),
invitee_user_id: Set(cmd.invitee_id),
invitee_email: Set(None),
accept_code: Set(accept_code.clone()),
created_at: Set(now),
expires_at: Set(expires_at),
accepted_at: Set(None),
revoked_at: Set(None),
}
.insert(self.db().write())
.await?;
if let Some(invitee) = cmd.invitee_id {
// The inbox renders real accept / reject controls, so the row has to
// carry the same credential the invite email links: `accept_code`.
// `space_name` lets the client deep-link into the channel once the
// invitee joins.
let data = serde_json::json!({
"room_id": room_id,
"invite_id": id,
"accept_code": accept_code,
"space_name": space.lower_name,
});
let pending = export::notify::PendingNotifications::default();
notify_user(
self.db().write(),
&pending,
invitee,
NotificationType::Invite,
i18n::catalog::notification_channel_invite_title(
self.recipient_locale(invitee).await,
)
.to_string(),
None,
Some(data),
Some(caller),
)
.await?;
pending.flush(&self.state.redis).await;
}
let mut uids = vec![row.inviter_id];
if let Some(invitee) = row.invitee_user_id {
uids.push(invitee);
}
let user_map = self.load_user_briefs(&uids).await?;
Ok(RoomInviteDto {
id: row.id,
room_id: row.room_id,
inviter_id: row.inviter_id,
inviter: user_map.get(&row.inviter_id).cloned(),
invitee_id: row.invitee_user_id,
invitee: row.invitee_user_id.and_then(|i| user_map.get(&i).cloned()),
accept_code,
created_at: row.created_at,
expires_at: row.expires_at,
})
}
/// The invitee declines: revoke the row so the inviter's invite list stops
/// reporting it as pending. Only the addressed invitee may do this, and an
/// already-accepted invite cannot be walked back from here.
pub async fn reject_room_invite(
&self,
caller: i64,
accept_code: &str,
) -> Result<(), ChatError> {
let invite = room_invites::Entity::find()
.filter(room_invites::Column::AcceptCode.eq(accept_code))
.one(self.db().read())
.await?
.ok_or(ChatError::NotFound)?;
if invite.invitee_user_id != Some(caller) {
return Err(ChatError::Forbidden);
}
if invite.accepted_at.is_some() {
return Err(ChatError::Conflict("invite already accepted".to_string()));
}
if invite.revoked_at.is_some() {
return Ok(());
}
room_invites::Entity::update_many()
.col_expr(
room_invites::Column::RevokedAt,
Expr::value(Some(OffsetDateTime::now_utc())),
)
.filter(room_invites::Column::Id.eq(invite.id))
.filter(room_invites::Column::AcceptedAt.is_null())
.filter(room_invites::Column::RevokedAt.is_null())
.exec(self.db().write())
.await?;
Ok(())
}
pub async fn list_room_invites(
&self,
caller: i64,
room_id: Uuid,
) -> Result<Vec<RoomInviteListItemDto>, ChatError> {
require_room_readable(self, room_id, caller).await?;
let rows = room_invites::Entity::find()
.filter(room_invites::Column::RoomId.eq(room_id))
.order_by_desc(room_invites::Column::CreatedAt)
.all(self.db().read())
.await?;
let mut items: Vec<RoomInviteListItemDto> = rows
.into_iter()
.map(|r| RoomInviteListItemDto {
id: r.id,
room_id: r.room_id,
inviter_id: r.inviter_id,
inviter: None,
invitee_id: r.invitee_user_id,
invitee: None,
created_at: r.created_at,
expires_at: r.expires_at,
accepted_at: r.accepted_at,
revoked_at: r.revoked_at,
})
.collect();
let mut uids: Vec<i64> = items
.iter()
.flat_map(|i| {
let mut v = vec![i.inviter_id];
if let Some(id) = i.invitee_id {
v.push(id);
}
v
})
.collect();
uids.sort();
uids.dedup();
if !uids.is_empty() {
let user_map = self.load_user_briefs(&uids).await?;
for item in items.iter_mut() {
item.inviter = user_map.get(&item.inviter_id).cloned();
item.invitee = item.invitee_id.and_then(|i| user_map.get(&i).cloned());
}
}
Ok(items)
}
pub async fn revoke_room_invite(
&self,
caller: i64,
room_id: Uuid,
invite_id: Uuid,
) -> Result<(), ChatError> {
let ctx = require_room_readable(self, room_id, caller).await?;
let row = room_invites::Entity::find_by_id(invite_id)
.filter(room_invites::Column::RoomId.eq(room_id))
.one(self.db().write())
.await?
.ok_or(ChatError::NotFound)?;
if row.accepted_at.is_some() {
return Err(ChatError::Conflict("invite already accepted".to_string()));
}
if row.inviter_id != caller {
match &ctx.space {
Some(space) => {
if !can_manage_space(self, space, caller).await? {
return Err(ChatError::Forbidden);
}
}
None => return Err(ChatError::Forbidden),
}
}
if row.revoked_at.is_none() {
let mut am: room_invites::ActiveModel = row.into();
am.revoked_at = Set(Some(OffsetDateTime::now_utc()));
am.update(self.db().write()).await?;
}
Ok(())
}
pub async fn accept_room_invite(
&self,
caller: i64,
accept_code: &str,
) -> Result<RoomBriefDto, ChatError> {
if !self
.state
.redis
.quota_allow(&format!("chat:invite_accept:{caller}"), 10, 60)
.await
{
return Err(ChatError::RateLimited(
"too many invite acceptance attempts, retry in 60s".to_string(),
));
}
let invite = room_invites::Entity::find()
.filter(room_invites::Column::AcceptCode.eq(accept_code))
.one(self.db().write())
.await?
.ok_or(ChatError::NotFound)?;
if invite.accepted_at.is_some() || invite.revoked_at.is_some() {
return Err(ChatError::NotFound);
}
if invite
.expires_at
.is_some_and(|at| at < OffsetDateTime::now_utc())
{
return Err(ChatError::NotFound);
}
if invite.invitee_user_id.is_some_and(|uid| uid != caller) {
return Err(ChatError::Forbidden);
}
let room = lookup_room(self, invite.room_id).await?;
if let Some(space_id) = room.space_id
&& require_space_member(
self,
&models::space::spaces::Entity::find_by_id(space_id)
.one(self.db().read())
.await?
.ok_or(ChatError::NotFound)?,
caller,
)
.await
.is_err()
{
return Err(ChatError::Forbidden);
}
let now = OffsetDateTime::now_utc();
let claimed = room_invites::Entity::update_many()
.col_expr(room_invites::Column::AcceptedAt, Expr::value(Some(now)))
.filter(room_invites::Column::Id.eq(invite.id))
.filter(room_invites::Column::AcceptedAt.is_null())
.exec(self.db().write())
.await?;
if claimed.rows_affected == 0 {
return Err(ChatError::NotFound);
}
self.ensure_room_membership(room.id, caller).await?;
self.get_room(caller, room.id).await
}
}