use models::chat::messages; use models::chat::read_states; use models::chat::room_seq; use models::room::rooms; use models::users::user_recent_mentions; use sea_orm::{ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter}; 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::lookup::{lookup_space_by_name, require_room_readable, require_space_member}; use crate::share::notify::advance_read_state; #[derive(Debug, Clone, Serialize, ToSchema)] pub struct ReadStateDto { pub room_id: Uuid, pub last_read_message_id: Option, pub last_read_seq: i64, pub last_viewed_at: Option, pub tip_seq: i64, pub unread_count: i64, pub unread_mentions: i64, } #[derive(Debug, Clone, Serialize, ToSchema)] pub struct SpaceReadStatePage { pub items: Vec, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct MarkReadCmd { #[serde(default)] pub message_id: Option, #[serde(default)] pub seq: Option, } #[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)] pub struct ThreadReadCmd { pub thread_id: Uuid, } impl ChatService { pub async fn mark_read( &self, caller: i64, room_id: Uuid, cmd: MarkReadCmd, ) -> Result { let _ctx = require_room_readable(self, room_id, caller).await?; let (message_id, seq) = match (cmd.message_id, cmd.seq) { (Some(message_id), _) => { let msg = messages::Entity::find_by_id(message_id) .filter(messages::Column::RoomId.eq(room_id)) .one(self.db().read()) .await? .ok_or(ChatError::NotFound)?; (Some(message_id), msg.seq) } (None, Some(seq)) => { let tip = room_seq::Entity::find_by_id(room_id) .one(self.db().read()) .await? .map(|s| s.seq) .unwrap_or(0); (None, seq.clamp(0, tip)) } (None, None) => { let tip = room_seq::Entity::find_by_id(room_id) .one(self.db().read()) .await? .map(|s| s.seq) .unwrap_or(0); let last_msg = rooms::Entity::find_by_id(room_id) .one(self.db().read()) .await? .and_then(|r| r.last_message_id); (last_msg, tip) } }; let changed = advance_read_state(self.db().write(), caller, room_id, message_id, seq).await?; if changed { self.realtime .publish_user( caller, &serde_json::json!({ "kind": "read_marker", "room_id": room_id, "last_read_message_id": message_id, "last_read_seq": seq, }), ) .await; } self.get_read_state(caller, room_id).await } pub async fn get_read_state( &self, caller: i64, room_id: Uuid, ) -> Result { require_room_readable(self, room_id, caller).await?; let tip = room_seq::Entity::find_by_id(room_id) .one(self.db().read()) .await? .map(|s| s.seq) .unwrap_or(0); let state = read_states::Entity::find_by_id((caller, room_id)) .one(self.db().read()) .await?; let last_viewed_at = state.as_ref().and_then(|s| s.last_viewed_at); let unread_mentions = match last_viewed_at { Some(at) => { user_recent_mentions::Entity::find() .filter(user_recent_mentions::Column::UserId.eq(caller)) .filter(user_recent_mentions::Column::ChannelId.eq(room_id)) .filter(user_recent_mentions::Column::CreatedAt.gt(at)) .count(self.db().read()) .await? as i64 } None => { user_recent_mentions::Entity::find() .filter(user_recent_mentions::Column::UserId.eq(caller)) .filter(user_recent_mentions::Column::ChannelId.eq(room_id)) .count(self.db().read()) .await? as i64 } }; let last_read_seq = state.as_ref().map(|s| s.last_read_seq).unwrap_or(0); Ok(ReadStateDto { room_id, last_read_message_id: state.as_ref().and_then(|s| s.last_read_message_id), last_read_seq, last_viewed_at, tip_seq: tip, unread_count: (tip - last_read_seq).max(0), unread_mentions, }) } pub async fn list_space_read_states( &self, caller: i64, space_name: &str, ) -> Result { let space = lookup_space_by_name(self, space_name).await?; require_space_member(self, &space, caller).await?; let space_rooms = rooms::Entity::find() .filter(rooms::Column::SpaceId.eq(space.id)) .filter(rooms::Column::IsActive.eq(true)) .all(self.db().read()) .await?; let mut items = Vec::new(); for room in space_rooms { if room.is_private && room.owner_id != caller && !crate::share::lookup::is_room_member(self, room.id, caller).await? { continue; } items.push(self.get_read_state(caller, room.id).await?); } Ok(SpaceReadStatePage { items }) } pub async fn mark_thread_read( &self, caller: i64, thread_id: Uuid, message_id: Option, ) -> Result<(), ChatError> { let thread = models::chat::threads::Entity::find_by_id(thread_id) .one(self.db().read()) .await? .ok_or(ChatError::NotFound)?; let _ctx = require_room_readable(self, thread.room_id, caller).await?; let (mid, seq) = match message_id { Some(message_id) => { let msg = messages::Entity::find_by_id(message_id) .one(self.db().read()) .await? .filter(|m| m.thread_id == Some(thread_id)) .ok_or(ChatError::NotFound)?; (message_id, msg.seq) } None => match thread.last_reply_id { Some(last) => { let msg = messages::Entity::find_by_id(last) .one(self.db().read()) .await? .ok_or(ChatError::NotFound)?; (last, msg.seq) } None => return Ok(()), }, }; self.upsert_thread_read(caller, thread_id, mid, seq).await } }