b453d5d
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<Uuid>,
pub last_read_seq: i64,
pub last_viewed_at: Option<OffsetDateTime>,
pub tip_seq: i64,
pub unread_count: i64,
pub unread_mentions: i64,
}
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct SpaceReadStatePage {
pub items: Vec<ReadStateDto>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct MarkReadCmd {
#[serde(default)]
pub message_id: Option<Uuid>,
#[serde(default)]
pub seq: Option<i64>,
}
#[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<ReadStateDto, ChatError> {
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<ReadStateDto, ChatError> {
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<SpaceReadStatePage, ChatError> {
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<Uuid>,
) -> 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
}
}