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, pub owner_id: i64, pub owner: Option, pub kind: RoomKind, pub name: String, pub topic: Option, pub position: i64, pub parent_id: Option, pub is_private: bool, pub nsfw: bool, pub rate_limit_per_user: Option, pub last_message_id: Option, 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, #[serde(default)] pub topic: Option, #[serde(default)] pub is_private: Option, #[serde(default)] pub nsfw: Option, #[serde(default)] pub rate_limit_per_user: Option, #[serde(default)] pub position: Option, /// 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>, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct AcceptRoomInviteCmd { pub code: String, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct AddMembersCmd { pub user_ids: Vec, } #[derive(Debug, Clone, Serialize, ToSchema)] pub struct RoomMemberDto { pub user: space::services::spaces::UserDto, pub role: RoomMemberRole, pub nick: Option, pub joined_at: OffsetDateTime, } #[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)] pub struct RoomMemberQuery { pub limit: Option, pub offset: Option, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct CreateRoomInviteCmd { #[serde(default)] pub invitee_id: Option, #[serde(default)] pub expires_secs: Option, } #[derive(Debug, Clone, Serialize, ToSchema)] pub struct RoomInviteDto { pub id: Uuid, pub room_id: Uuid, pub inviter_id: i64, pub inviter: Option, pub invitee_id: Option, pub invitee: Option, pub accept_code: String, pub created_at: OffsetDateTime, pub expires_at: Option, } #[derive(Debug, Clone, Serialize, ToSchema)] pub struct RoomInviteListItemDto { pub id: Uuid, pub room_id: Uuid, pub inviter_id: i64, pub inviter: Option, pub invitee_id: Option, pub invitee: Option, pub created_at: OffsetDateTime, pub expires_at: Option, pub accepted_at: Option, pub revoked_at: Option, } 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, 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 = 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 = 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 = 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)> = 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 = 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> = 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 = member_counts.into_iter().collect(); let owner_ids: Vec = rooms .iter() .map(|r| r.owner_id) .collect::>() .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 { 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 { 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)> = 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?; } } } 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, 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 = 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 { 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).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, 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 = 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 = 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 { 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 } }