eeff292
use api::users::UserDto;
use models::chat::messages::{self, ChatMessageType};
use models::chat::{read_states, room_seq, room_user_settings};
use models::room::rooms::{self, RoomKind};
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QueryOrder,
QuerySelect, Set, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::authz;
use crate::share::lookup::{lookup_space_by_name, require_room_readable, require_space_member};
use crate::share::notify::next_room_seq;
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 icon: Option<String>,
pub color: Option<String>,
pub position: i64,
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>,
#[serde(default, deserialize_with = "api::serde_patch::deserialize_opt_opt")]
pub icon: Option<Option<String>>,
#[serde(default, deserialize_with = "api::serde_patch::deserialize_opt_opt")]
pub color: Option<Option<String>>,
}
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.database.read())
.await?;
let room_ids: Vec<Uuid> = rooms.iter().map(|r| r.id).collect();
let my_rooms: std::collections::HashSet<Uuid> = models::room::room_members::Entity::find()
.filter(models::room::room_members::Column::UserId.eq(caller))
.filter(models::room::room_members::Column::RoomId.is_in(room_ids.clone()))
.filter(models::room::room_members::Column::LeftAt.is_null())
.all(self.database.read())
.await?
.into_iter()
.map(|row| row.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.database.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.database.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.database.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.database.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 space_member_count = space.members(caller).await?.len() as i64;
let owner_ids: Vec<i64> = rooms
.iter()
.map(|r| r.owner_id)
.collect::<Vec<_>>()
.into_iter()
.collect();
let owner_map = self.audience.load_user_briefs(&owner_ids).await?;
let mut result = Vec::new();
for room in rooms {
let visible =
!room.is_private || room.owner_id == caller || my_rooms.contains(&room.id);
if !visible {
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,
icon: room.icon,
color: room.color,
position: room.position,
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: space_member_count,
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.database.read())
.await?
.map(|s| s.seq)
.unwrap_or(0);
let last_read = read_states::Entity::find_by_id((caller, room.id))
.one(self.database.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.database.read())
.await?;
let member_count = match &ctx.space {
Some(space) => space.members(caller).await?.len() as i64,
None => self.room_member_count(room.id).await? as i64,
};
let owner_map = self.audience.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,
icon: room.icon,
color: room.color,
position: room.position,
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());
if needs_manage {
match &ctx.space {
Some(space) => {
if !authz::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.clone());
am.lower_name = Set(lower);
}
if let Some(topic) = cmd.topic {
if topic.chars().count() > self.settings.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()));
}
let stored = if topic.is_empty() { None } else { Some(topic) };
am.topic = Set(stored);
}
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 {
if position < 0 {
return Err(ChatError::BadRequest(
"position must be non-negative".to_string(),
));
}
am.position = Set(position);
}
if let Some(icon) = cmd.icon {
am.icon = Set(icon);
}
if let Some(color) = cmd.color {
am.color = Set(color);
}
am.updated_at = Set(now);
am.update(self.database.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.audience.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
.database
.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.database.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(())
}
}
#[derive(Clone, Serialize)]
pub struct SpaceContext {
pub id: Uuid,
pub lower_name: String,
pub name: String,
pub avatar_url: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct CreateRoomCmd {
pub name: String,
#[serde(default)]
pub kind: RoomKind,
#[serde(default)]
pub topic: Option<String>,
#[serde(default)]
pub is_private: bool,
#[serde(default)]
pub nsfw: bool,
#[serde(default)]
pub user_limit: Option<i32>,
#[serde(default)]
pub rate_limit_per_user: Option<i32>,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct RoomDto {
pub id: Uuid,
pub space_id: Option<Uuid>,
pub kind: RoomKind,
pub name: String,
pub topic: Option<String>,
pub icon: Option<String>,
pub color: Option<String>,
pub position: i64,
pub owner_id: i64,
pub is_private: bool,
pub nsfw: bool,
pub user_limit: Option<i32>,
pub rate_limit_per_user: Option<i32>,
pub is_active: bool,
pub created_at: OffsetDateTime,
pub updated_at: OffsetDateTime,
}
impl From<rooms::Model> for RoomDto {
fn from(r: rooms::Model) -> Self {
Self {
id: r.id,
space_id: r.space_id,
kind: r.kind,
name: r.name,
topic: r.topic,
icon: r.icon,
color: r.color,
position: r.position,
owner_id: r.owner_id,
is_private: r.is_private,
nsfw: r.nsfw,
user_limit: r.user_limit,
rate_limit_per_user: r.rate_limit_per_user,
is_active: r.is_active,
created_at: r.created_at,
updated_at: r.updated_at,
}
}
}
async fn room(tx: &impl ConnectionTrait, id: Uuid) -> Result<RoomDto, ChatError> {
rooms::Entity::find_by_id(id)
.one(tx)
.await?
.filter(|room| room.is_active)
.map(RoomDto::from)
.ok_or(ChatError::NotFound)
}
async fn create_space_room(
tx: &impl ConnectionTrait,
space: &SpaceContext,
actor: i64,
cmd: CreateRoomCmd,
) -> Result<RoomDto, ChatError> {
if !matches!(
cmd.kind,
RoomKind::GuildRoom
| RoomKind::Text
| RoomKind::Category
| RoomKind::News
| RoomKind::Voice
| RoomKind::Stage
) {
return Err(ChatError::BadRequest("unsupported room kind: only guild_room, text, category, news, voice and stage can be created".to_owned()));
}
let name = cmd.name.trim();
if name.is_empty() || name.chars().count() > 100 {
return Err(ChatError::BadRequest(
"room name is empty or too long".to_owned(),
));
}
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: must contain only lowercase letters, digits, and hyphens"
.to_owned(),
));
}
if rooms::Entity::find()
.filter(rooms::Column::SpaceId.eq(space.id))
.filter(rooms::Column::LowerName.eq(&lower))
.one(tx)
.await?
.is_some()
{
return Err(ChatError::Conflict(format!(
"room name {lower} already exists in this space"
)));
}
let position = rooms::Entity::find()
.filter(rooms::Column::SpaceId.eq(space.id))
.order_by_desc(rooms::Column::Position)
.one(tx)
.await?
.map(|row| row.position + 1)
.unwrap_or(0);
let now = OffsetDateTime::now_utc();
let created = rooms::ActiveModel {
id: Set(Uuid::now_v7()),
space_id: Set(Some(space.id)),
kind: Set(cmd.kind),
lower_name: Set(lower),
name: Set(name.to_owned()),
topic: Set(cmd.topic),
icon: Set(None),
color: Set(None),
position: Set(position),
owner_id: Set(actor),
is_private: Set(cmd.is_private),
nsfw: Set(cmd.nsfw),
user_limit: Set(cmd.user_limit),
rate_limit_per_user: Set(cmd.rate_limit_per_user),
bitrate: Set(None),
last_message_id: Set(None),
is_active: Set(true),
created_at: Set(now),
updated_at: Set(now),
}
.insert(tx)
.await?;
Ok(created.into())
}
async fn deactivate_space_room(
tx: &impl ConnectionTrait,
space: &SpaceContext,
actor: i64,
room_id: Uuid,
) -> Result<(), ChatError> {
let row = rooms::Entity::find_by_id(room_id)
.one(tx)
.await?
.filter(|row| row.is_active && row.space_id == Some(space.id))
.ok_or(ChatError::NotFound)?;
let detail = room_detail(&row);
let mut active: rooms::ActiveModel = row.into();
active.is_active = Set(false);
active.updated_at = Set(OffsetDateTime::now_utc());
active.update(tx).await?;
events::emit(
tx,
events::EventKind::ChatRoomUpdate,
events::Labels::new()
.with(events::label::ROOM_ID, room_id)
.with(events::label::ACTOR_USER_ID, actor),
serde_json::json!({
"room_id": room_id, "deleted": true, "room": detail, "space": space,
}),
)
.await?;
Ok(())
}
fn room_detail(room: &rooms::Model) -> serde_json::Value {
serde_json::json!({ "id": room.id, "space_id": room.space_id, "kind": room.kind, "name": room.name, "topic": room.topic })
}
impl ChatService {
pub async fn create_room(
&self,
caller: i64,
space_name: &str,
cmd: CreateRoomCmd,
) -> Result<RoomDto, ChatError> {
self.space
.allow_room_creation(caller)
.await
.map_err(|error| match error {
space::errors::SpaceError::RateLimited(message) => ChatError::RateLimited(message),
other => ChatError::Internal(other.to_string()),
})?;
let space = lookup_space_by_name(self, space_name).await?;
if !authz::can_manage_space(self, &space, caller).await? {
return Err(ChatError::Forbidden);
}
let context = SpaceContext {
id: space.id(),
lower_name: space.slug().to_owned(),
name: space.name().to_owned(),
avatar_url: space.avatar_url().await?,
};
let created = self
.database
.write()
.transaction::<_, RoomDto, ChatError>(move |tx| {
Box::pin(async move { create_space_room(tx, &context, caller, cmd).await })
})
.await?;
Ok(created)
}
pub async fn delete_room(&self, caller: i64, room_id: Uuid) -> Result<(), ChatError> {
let room = room(self.database.read(), room_id).await?;
let space_id = room.space_id.ok_or(ChatError::NotFound)?;
let space = self.participants.access.by_id(space_id).await?;
if !authz::can_manage_space(self, &space, caller).await? {
return Err(ChatError::Forbidden);
}
let context = SpaceContext {
id: space.id(),
lower_name: space.slug().to_owned(),
name: space.name().to_owned(),
avatar_url: space.avatar_url().await?,
};
self.database
.write()
.transaction::<_, (), ChatError>(move |tx| {
Box::pin(async move { deactivate_space_room(tx, &context, caller, room_id).await })
})
.await?;
Ok(())
}
}