main
use api::users::UserDto;
use models::chat::messages::{self, FLAG_HAS_THREAD};
use models::chat::thread_members::{self, ThreadNotifyLevel};
use models::chat::threads;
use models::room::room_members;
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, ExprTrait, PaginatorTrait,
QueryFilter, QueryOrder, QuerySelect, Set, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::realtime::RoomEvent;
use crate::share::{authz, lookup};
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ThreadDto {
pub id: Uuid,
pub room_id: Uuid,
pub root_message_id: Uuid,
pub owner: Option<UserDto>,
pub owner_id: i64,
pub name: Option<String>,
pub reply_count: i64,
pub participant_count: i64,
pub last_reply_id: Option<Uuid>,
pub last_reply_at: Option<OffsetDateTime>,
pub archived: bool,
pub locked: bool,
pub created_at: OffsetDateTime,
pub updated_at: OffsetDateTime,
}
pub type ThreadPage = api::pagination::Page<ThreadDto>;
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateThreadCmd {
#[serde(default)]
pub name: Option<String>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateThreadCmd {
#[serde(default)]
pub name: Option<String>,
#[serde(default)]
pub archived: Option<bool>,
#[serde(default)]
pub locked: Option<bool>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct ThreadMemberCmd {
#[serde(default)]
pub notify_level: Option<ThreadNotifyLevel>,
}
#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ListThreadQuery {
pub archived: Option<bool>,
pub limit: Option<u64>,
pub offset: Option<u64>,
}
pub(crate) async fn maintain_thread_on_reply(
txn: &impl ConnectionTrait,
thread: &threads::Model,
message_id: Uuid,
author: i64,
now: OffsetDateTime,
) -> Result<(), ChatError> {
threads::Entity::update_many()
.filter(threads::Column::Id.eq(thread.id))
.col_expr(
threads::Column::ReplyCount,
Expr::col(threads::Column::ReplyCount).add(1),
)
.col_expr(threads::Column::LastReplyId, Expr::value(message_id))
.col_expr(threads::Column::LastReplyAt, Expr::value(now))
.col_expr(threads::Column::UpdatedAt, Expr::value(now))
.exec(txn)
.await?;
let member = thread_members::Entity::find_by_id((thread.id, author))
.one(txn)
.await?;
match member {
Some(_) => {}
None => {
thread_members::ActiveModel {
thread_id: Set(thread.id),
user_id: Set(author),
notify_level: Set(ThreadNotifyLevel::All),
last_read_message_id: Set(None),
last_read_seq: Set(0),
created_at: Set(now),
updated_at: Set(now),
}
.insert(txn)
.await?;
threads::Entity::update_many()
.filter(threads::Column::Id.eq(thread.id))
.col_expr(
threads::Column::ParticipantCount,
Expr::col(threads::Column::ParticipantCount).add(1),
)
.exec(txn)
.await?;
}
}
Ok(())
}
impl ChatService {
async fn build_thread_dto(
&self,
rows: Vec<threads::Model>,
) -> Result<Vec<ThreadDto>, ChatError> {
let owner_ids: Vec<i64> = rows.iter().map(|t| t.owner_id).collect();
let user_map = self.audience.load_user_briefs(&owner_ids).await?;
Ok(rows
.into_iter()
.map(|t| ThreadDto {
id: t.id,
room_id: t.room_id,
root_message_id: t.root_message_id,
owner: user_map.get(&t.owner_id).cloned(),
owner_id: t.owner_id,
name: t.name,
reply_count: t.reply_count,
participant_count: t.participant_count,
last_reply_id: t.last_reply_id,
last_reply_at: t.last_reply_at,
archived: t.archived,
locked: t.locked,
created_at: t.created_at,
updated_at: t.updated_at,
})
.collect())
}
pub async fn create_thread(
&self,
caller: i64,
room_id: Uuid,
message_id: Uuid,
cmd: CreateThreadCmd,
) -> Result<ThreadDto, ChatError> {
let _ctx = lookup::require_room_readable(self, room_id, caller).await?;
let root = 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 existing = threads::Entity::find()
.filter(threads::Column::RootMessageId.eq(message_id))
.one(self.database.read())
.await?;
if let Some(existing) = existing {
let items = self.build_thread_dto(vec![existing]).await?;
return Ok(items.into_iter().next().unwrap());
}
let now = OffsetDateTime::now_utc();
let thread_id = Uuid::now_v7();
let name = cmd
.name
.clone()
.map(|n| n.trim().to_string())
.filter(|n| !n.is_empty());
if name
.as_ref()
.is_some_and(|n| n.chars().count() > self.settings.chat.max_thread_name_len)
{
return Err(ChatError::BadRequest("thread name too long".to_string()));
}
self.database
.primary()
.transaction::<_, (), ChatError>(move |txn| {
Box::pin(async move {
threads::ActiveModel {
id: Set(thread_id),
room_id: Set(room_id),
root_message_id: Set(message_id),
owner_id: Set(caller),
name: Set(name),
reply_count: Set(0),
participant_count: Set(1),
last_reply_id: Set(None),
last_reply_at: Set(None),
archived: Set(false),
locked: Set(false),
auto_archive_minutes: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(txn)
.await?;
thread_members::ActiveModel {
thread_id: Set(thread_id),
user_id: Set(caller),
notify_level: Set(ThreadNotifyLevel::All),
last_read_message_id: Set(None),
last_read_seq: Set(0),
created_at: Set(now),
updated_at: Set(now),
}
.insert(txn)
.await?;
let new_flags = root.flags | FLAG_HAS_THREAD;
messages::Entity::update_many()
.filter(messages::Column::Id.eq(message_id))
.col_expr(messages::Column::Flags, Expr::value(new_flags))
.exec(txn)
.await?;
Ok(())
})
})
.await?;
let thread = threads::Entity::find_by_id(thread_id)
.one(self.database.read())
.await?
.ok_or(ChatError::Internal("thread vanished after insert".into()))?;
let event = RoomEvent::ThreadUpdated { room_id, thread_id };
self.audience.realtime.publish_room(&event).await;
let items = self.build_thread_dto(vec![thread]).await?;
Ok(items.into_iter().next().unwrap())
}
pub async fn list_threads(
&self,
caller: i64,
room_id: Uuid,
q: ListThreadQuery,
) -> Result<ThreadPage, ChatError> {
let _ctx = lookup::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 = threads::Entity::find().filter(threads::Column::RoomId.eq(room_id));
if let Some(archived) = q.archived {
select = select.filter(threads::Column::Archived.eq(archived));
}
let total = select.clone().count(self.database.read()).await?;
let rows = select
.order_by_desc(threads::Column::LastReplyAt)
.order_by_desc(threads::Column::CreatedAt)
.limit(limit)
.offset(offset)
.all(self.database.read())
.await?;
let items = self.build_thread_dto(rows).await?;
Ok(ThreadPage {
items,
total,
limit,
offset,
})
}
pub async fn list_my_threads(&self, caller: i64) -> Result<Vec<ThreadDto>, ChatError> {
let memberships = thread_members::Entity::find()
.filter(thread_members::Column::UserId.eq(caller))
.order_by_desc(thread_members::Column::UpdatedAt)
.limit(100)
.all(self.database.read())
.await?;
let thread_ids: Vec<Uuid> = memberships.into_iter().map(|m| m.thread_id).collect();
if thread_ids.is_empty() {
return Ok(Vec::new());
}
let rows = threads::Entity::find()
.filter(threads::Column::Id.is_in(thread_ids))
.all(self.database.read())
.await?;
self.build_thread_dto(rows).await
}
pub async fn get_thread(&self, caller: i64, thread_id: Uuid) -> Result<ThreadDto, ChatError> {
let thread = threads::Entity::find_by_id(thread_id)
.one(self.database.read())
.await?
.ok_or(ChatError::NotFound)?;
let _ctx = lookup::require_room_readable(self, thread.room_id, caller).await?;
let items = self.build_thread_dto(vec![thread]).await?;
Ok(items.into_iter().next().unwrap())
}
pub async fn update_thread(
&self,
caller: i64,
thread_id: Uuid,
cmd: UpdateThreadCmd,
) -> Result<ThreadDto, ChatError> {
let thread = threads::Entity::find_by_id(thread_id)
.one(self.database.read())
.await?
.ok_or(ChatError::NotFound)?;
let ctx = lookup::require_room_readable(self, thread.room_id, caller).await?;
let is_manager = match &ctx.space {
Some(space) => authz::can_manage_space(self, space, caller).await?,
None => false,
};
if thread.owner_id != caller && !is_manager {
return Err(ChatError::Forbidden);
}
let now = OffsetDateTime::now_utc();
let mut am: threads::ActiveModel = thread.clone().into();
if let Some(name) = cmd.name {
let name = name.trim().to_string();
if name.chars().count() > self.settings.chat.max_thread_name_len {
return Err(ChatError::BadRequest("thread name too long".to_string()));
}
am.name = Set(if name.is_empty() { None } else { Some(name) });
}
if let Some(archived) = cmd.archived {
am.archived = Set(archived);
}
if let Some(locked) = cmd.locked {
am.locked = Set(locked);
}
am.updated_at = Set(now);
am.update(self.database.write()).await?;
let event = RoomEvent::ThreadUpdated {
room_id: thread.room_id,
thread_id,
};
self.audience.realtime.publish_room(&event).await;
let updated = threads::Entity::find_by_id(thread_id)
.one(self.database.read())
.await?
.ok_or(ChatError::NotFound)?;
let items = self.build_thread_dto(vec![updated]).await?;
Ok(items.into_iter().next().unwrap())
}
pub async fn set_thread_membership(
&self,
caller: i64,
thread_id: Uuid,
cmd: ThreadMemberCmd,
) -> Result<(), ChatError> {
let thread = threads::Entity::find_by_id(thread_id)
.one(self.database.read())
.await?
.ok_or(ChatError::NotFound)?;
let _ctx = lookup::require_room_readable(self, thread.room_id, caller).await?;
let now = OffsetDateTime::now_utc();
let notify_level = cmd.notify_level.unwrap_or(ThreadNotifyLevel::All);
self.database
.primary()
.transaction::<_, (), ChatError>(move |txn| {
Box::pin(async move {
let member = thread_members::Entity::find_by_id((thread_id, caller))
.one(txn)
.await?;
match member {
Some(row) => {
let mut am: thread_members::ActiveModel = row.into();
am.notify_level = Set(notify_level);
am.updated_at = Set(now);
am.update(txn).await?;
}
None => {
thread_members::ActiveModel {
thread_id: Set(thread_id),
user_id: Set(caller),
notify_level: Set(notify_level),
last_read_message_id: Set(None),
last_read_seq: Set(0),
created_at: Set(now),
updated_at: Set(now),
}
.insert(txn)
.await?;
threads::Entity::update_many()
.filter(threads::Column::Id.eq(thread_id))
.col_expr(
threads::Column::ParticipantCount,
Expr::col(threads::Column::ParticipantCount).add(1),
)
.exec(txn)
.await?;
}
}
Ok(())
})
})
.await?;
Ok(())
}
pub async fn leave_thread(&self, caller: i64, thread_id: Uuid) -> Result<(), ChatError> {
let thread = threads::Entity::find_by_id(thread_id)
.one(self.database.read())
.await?
.ok_or(ChatError::NotFound)?;
let _ctx = lookup::require_room_readable(self, thread.room_id, caller).await?;
self.database
.primary()
.transaction::<_, (), ChatError>(move |txn| {
Box::pin(async move {
let deleted = thread_members::Entity::delete_by_id((thread_id, caller))
.exec(txn)
.await?;
if deleted.rows_affected > 0 {
threads::Entity::update_many()
.filter(threads::Column::Id.eq(thread_id))
.col_expr(
threads::Column::ParticipantCount,
Expr::col(threads::Column::ParticipantCount).sub(1),
)
.exec(txn)
.await?;
}
Ok(())
})
})
.await?;
Ok(())
}
pub(crate) async fn upsert_thread_read(
&self,
caller: i64,
thread_id: Uuid,
message_id: Uuid,
seq: i64,
) -> Result<(), ChatError> {
let now = OffsetDateTime::now_utc();
let member = thread_members::Entity::find_by_id((thread_id, caller))
.one(self.database.write())
.await?;
match member {
Some(row) if row.last_read_seq < seq => {
let mut am: thread_members::ActiveModel = row.into();
am.last_read_message_id = Set(Some(message_id));
am.last_read_seq = Set(seq);
am.updated_at = Set(now);
am.update(self.database.write()).await?;
}
Some(_) => {}
None => {
let am = thread_members::ActiveModel {
thread_id: Set(thread_id),
user_id: Set(caller),
notify_level: Set(ThreadNotifyLevel::All),
last_read_message_id: Set(Some(message_id)),
last_read_seq: Set(seq),
created_at: Set(now),
updated_at: Set(now),
};
match am.insert(self.database.write()).await {
Ok(_) => {
threads::Entity::update_many()
.filter(threads::Column::Id.eq(thread_id))
.col_expr(
threads::Column::ParticipantCount,
Expr::col(threads::Column::ParticipantCount).add(1),
)
.exec(self.database.write())
.await?;
}
Err(err) => {
let msg = err.to_string().to_lowercase();
if !(msg.contains("unique") || msg.contains("duplicate")) {
return Err(ChatError::from(err));
}
}
}
}
}
Ok(())
}
pub(crate) async fn room_member_count(&self, room_id: Uuid) -> Result<u64, ChatError> {
Ok(room_members::Entity::find()
.filter(room_members::Column::RoomId.eq(room_id))
.filter(room_members::Column::LeftAt.is_null())
.count(self.database.read())
.await?)
}
}