e1df092
use models::chat::{read_states, room_seq};
use models::room::room_members;
use models::room::rooms::{self, RoomKind};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use space::services::spaces::UserDto;
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::realtime::RoomEvent;
const MAX_GROUP_DM_MEMBERS: usize = 10;
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct DmDto {
pub id: Uuid,
pub kind: RoomKind,
pub name: String,
pub peer: Option<UserDto>,
#[serde(default)]
pub members: Vec<UserDto>,
pub last_message_id: Option<Uuid>,
pub last_read_seq: i64,
pub tip_seq: i64,
pub unread_count: i64,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct OpenDmCmd {
#[serde(default)]
pub peer_user_id: Option<i64>,
#[serde(default)]
pub peer_user_ids: Option<Vec<i64>>,
}
fn dm_pair_key(a: i64, b: i64) -> String {
let (lo, hi) = if a < b { (a, b) } else { (b, a) };
format!("dm:{lo}:{hi}")
}
fn group_dm_key(mut uids: Vec<i64>) -> String {
uids.sort();
let mut hasher = Sha256::new();
for uid in &uids {
hasher.update(uid.to_le_bytes());
}
let hex: String = hasher
.finalize()
.iter()
.map(|b| format!("{b:02x}"))
.collect();
format!("gdm:{hex}")
}
impl ChatService {
pub async fn open_dm(&self, caller: i64, cmd: OpenDmCmd) -> Result<DmDto, ChatError> {
let mut peers: Vec<i64> = Vec::new();
if let Some(peer) = cmd.peer_user_id {
peers.push(peer);
}
if let Some(list) = cmd.peer_user_ids {
peers.extend(list);
}
let mut uniq: Vec<i64> = Vec::new();
for peer in peers {
if peer != caller && !uniq.contains(&peer) {
uniq.push(peer);
}
}
if uniq.is_empty() {
return Err(ChatError::BadRequest(
"peer_user_id or peer_user_ids is required".to_string(),
));
}
if uniq.len() == 1 {
return self.open_direct_dm(caller, uniq[0]).await;
}
self.open_group_dm(caller, uniq).await
}
async fn check_dm_quota(&self, caller: i64) -> Result<(), ChatError> {
if !self
.state
.redis
.quota_allow(
&format!("quota:dm_open:{caller}"),
self.config().quota.dm_open_per_day,
86_400,
)
.await
{
return Err(ChatError::RateLimited(
"DM creation quota exceeded, retry tomorrow".to_string(),
));
}
Ok(())
}
async fn open_direct_dm(&self, caller: i64, peer_user_id: i64) -> Result<DmDto, ChatError> {
self.check_dm_quota(caller).await?;
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| ChatError::Internal("kenoauth is not configured".to_string()))?;
let peer = export::users::get_public_user(&self.state, peer_user_id)
.await?
.ok_or(ChatError::NotFound)?;
let peer_status = client
.get_user_status(peer_user_id)
.await
.map_err(|error| ChatError::Internal(error.to_string()))?;
if !peer_status.allow_use {
return Err(ChatError::NotFound);
}
let pair = dm_pair_key(caller, peer.user_id);
let existing = rooms::Entity::find()
.filter(rooms::Column::LowerName.eq(&pair))
.filter(rooms::Column::Kind.eq(RoomKind::DirectMessage))
.one(self.db().read())
.await?;
let room = match existing {
Some(room) if room.is_active => room,
Some(room) => self.reactivate_room(room).await?,
None => {
let now = OffsetDateTime::now_utc();
let room_id = Uuid::now_v7();
rooms::ActiveModel {
id: Set(room_id),
space_id: Set(None),
kind: Set(RoomKind::DirectMessage),
lower_name: Set(pair.clone()),
name: Set(peer.username.clone()),
topic: Set(None),
position: Set(0),
parent_id: Set(None),
owner_id: Set(caller),
is_private: Set(true),
nsfw: Set(false),
user_limit: Set(Some(2)),
rate_limit_per_user: Set(None),
bitrate: Set(None),
last_message_id: Set(None),
is_active: Set(true),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db().write())
.await?
}
};
self.ensure_dm_members(room.id, caller, &[caller, peer.user_id])
.await?;
let peer_dto = UserDto::from_public(&peer);
self.build_dm_dto(caller, room, Some(peer_dto), Vec::new())
.await
}
async fn open_group_dm(&self, caller: i64, peers: Vec<i64>) -> Result<DmDto, ChatError> {
self.check_dm_quota(caller).await?;
if peers.len() + 1 > MAX_GROUP_DM_MEMBERS {
return Err(ChatError::BadRequest(format!(
"group DM supports at most {MAX_GROUP_DM_MEMBERS} members"
)));
}
let client = self
.state
.kenauth
.as_ref()
.ok_or_else(|| ChatError::Internal("kenoauth is not configured".to_string()))?;
let publics = export::users::load_public_users(&self.state, &peers).await?;
if publics.len() != peers.len() {
return Err(ChatError::NotFound);
}
let authorities = client
.batch_get_authority_users(peers.clone())
.await
.map_err(|error| ChatError::Internal(error.to_string()))?;
if authorities.iter().any(|user| !user.allow_use) {
return Err(ChatError::NotFound);
}
let users: Vec<_> = peers
.iter()
.filter_map(|uid| publics.get(uid).cloned())
.collect();
let mut member_uids = peers.clone();
member_uids.push(caller);
let key = group_dm_key(member_uids.clone());
let existing = rooms::Entity::find()
.filter(rooms::Column::LowerName.eq(&key))
.filter(rooms::Column::Kind.eq(RoomKind::GroupDm))
.one(self.db().read())
.await?;
let room = match existing {
Some(room) if room.is_active => room,
Some(room) => self.reactivate_room(room).await?,
None => {
let now = OffsetDateTime::now_utc();
let room_id = Uuid::now_v7();
let name: String = users
.iter()
.map(|u| u.username.clone())
.collect::<Vec<_>>()
.join(", ")
.chars()
.take(100)
.collect();
rooms::ActiveModel {
id: Set(room_id),
space_id: Set(None),
kind: Set(RoomKind::GroupDm),
lower_name: Set(key.clone()),
name: Set(name),
topic: Set(None),
position: Set(0),
parent_id: Set(None),
owner_id: Set(caller),
is_private: Set(true),
nsfw: Set(false),
user_limit: Set(Some(member_uids.len() as i32)),
rate_limit_per_user: Set(None),
bitrate: Set(None),
last_message_id: Set(None),
is_active: Set(true),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db().write())
.await?
}
};
self.ensure_dm_members(room.id, caller, &member_uids)
.await?;
let members = users.iter().map(UserDto::from_public).collect::<Vec<_>>();
self.build_dm_dto(caller, room, None, members).await
}
async fn reactivate_room(&self, room: rooms::Model) -> Result<rooms::Model, ChatError> {
let mut room = room;
room.is_active = true;
let mut am: rooms::ActiveModel = room.clone().into();
am.is_active = Set(true);
am.updated_at = Set(OffsetDateTime::now_utc());
am.update(self.db().write()).await?;
Ok(room)
}
async fn ensure_dm_members(
&self,
room_id: Uuid,
caller: i64,
uids: &[i64],
) -> Result<(), ChatError> {
let now = OffsetDateTime::now_utc();
for &uid in uids {
let existing_member = room_members::Entity::find_by_id((room_id, uid))
.one(self.db().write())
.await?;
match existing_member {
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(_) => {}
None => {
room_members::ActiveModel {
room_id: Set(room_id),
user_id: Set(uid),
role: Set(if uid == caller {
models::room::room_members::RoomMemberRole::Owner
} else {
models::room::room_members::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?;
}
}
}
Ok(())
}
pub async fn list_dms(&self, caller: i64) -> Result<Vec<DmDto>, ChatError> {
let memberships = room_members::Entity::find()
.filter(room_members::Column::UserId.eq(caller))
.filter(room_members::Column::LeftAt.is_null())
.all(self.db().read())
.await?;
let room_ids: Vec<Uuid> = memberships.into_iter().map(|m| m.room_id).collect();
if room_ids.is_empty() {
return Ok(Vec::new());
}
let rooms = rooms::Entity::find()
.filter(rooms::Column::Id.is_in(room_ids))
.filter(rooms::Column::IsActive.eq(true))
.filter(rooms::Column::SpaceId.is_null())
.all(self.db().read())
.await?;
let dm_room_ids: Vec<Uuid> = rooms.iter().map(|r| r.id).collect();
let peer_rows = room_members::Entity::find()
.filter(room_members::Column::RoomId.is_in(dm_room_ids.clone()))
.filter(room_members::Column::UserId.ne(caller))
.filter(room_members::Column::LeftAt.is_null())
.all(self.db().read())
.await?;
let peer_uids: Vec<i64> = peer_rows.iter().map(|m| m.user_id).collect();
let user_map = self.load_user_briefs(&peer_uids).await?;
let mut peers_by_room: std::collections::HashMap<Uuid, Vec<UserDto>> =
std::collections::HashMap::new();
for m in peer_rows {
if let Some(dto) = user_map.get(&m.user_id).cloned() {
peers_by_room.entry(m.room_id).or_default().push(dto);
}
}
let tips = room_seq::Entity::find()
.filter(room_seq::Column::RoomId.is_in(dm_room_ids.clone()))
.all(self.db().read())
.await?;
let tip_map: std::collections::HashMap<Uuid, i64> =
tips.into_iter().map(|s| (s.room_id, s.seq)).collect();
let reads = read_states::Entity::find()
.filter(read_states::Column::UserId.eq(caller))
.filter(read_states::Column::RoomId.is_in(dm_room_ids))
.all(self.db().read())
.await?;
let read_map: std::collections::HashMap<Uuid, i64> = reads
.into_iter()
.map(|r| (r.room_id, r.last_read_seq))
.collect();
let mut result = Vec::new();
for room in rooms {
let tip = tip_map.get(&room.id).copied().unwrap_or(0);
let last_read = read_map.get(&room.id).copied().unwrap_or(0);
let mut members = peers_by_room.remove(&room.id).unwrap_or_default();
let peer = if room.kind == RoomKind::DirectMessage {
members.first().cloned()
} else {
None
};
if room.kind == RoomKind::DirectMessage {
members.clear();
}
result.push(DmDto {
id: room.id,
kind: room.kind,
name: room.name,
peer,
members,
last_message_id: room.last_message_id,
last_read_seq: last_read,
tip_seq: tip,
unread_count: (tip - last_read).max(0),
created_at: room.created_at,
});
}
result.sort_by_key(|a| std::cmp::Reverse(a.tip_seq));
Ok(result)
}
async fn build_dm_dto(
&self,
caller: i64,
room: rooms::Model,
peer: Option<UserDto>,
members: Vec<UserDto>,
) -> Result<DmDto, ChatError> {
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);
Ok(DmDto {
id: room.id,
kind: room.kind,
name: room.name,
peer,
members,
last_message_id: room.last_message_id,
last_read_seq: last_read,
tip_seq: tip,
unread_count: (tip - last_read).max(0),
created_at: room.created_at,
})
}
pub async fn close_dm(&self, caller: i64, room_id: Uuid) -> Result<(), ChatError> {
let room = rooms::Entity::find_by_id(room_id)
.one(self.db().read())
.await?
.filter(|r| r.is_active && r.space_id.is_none())
.ok_or(ChatError::NotFound)?;
let member = room_members::Entity::find_by_id((room_id, caller))
.one(self.db().write())
.await?
.ok_or(ChatError::NotFound)?;
if member.left_at.is_none() {
let now = OffsetDateTime::now_utc();
let mut am: room_members::ActiveModel = member.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;
}
let _ = room;
Ok(())
}
}