f0a1552
ZhenYi
feat(chat): add chat module with room management, message handling, and session revocation
f0a1552
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use async_trait::async_trait;
use config::AppConfig;
use email::EmailService;
use export::AppState;
use export::database::AppDatabase;
use models::issue::issue_event_line::{self, IssueEventType};
use models::space::space_members::{self, MemberStatus};
use models::space::spaces;
use models::users::user;
use models::users::user_notifications::{self, NotificationType};
use sea_orm::{ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, Set};
use session::SessionError;
use session::share::guard::{SessionGuard, SessionWithUser};
use space::services::spaces::UserDto;
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::IssueError;
#[derive(Clone)]
pub struct IssueService {
pub state: AppState,
}
impl IssueService {
pub fn new(state: AppState) -> Self {
Self { state }
}
pub fn config(&self) -> &AppConfig {
&self.state.config
}
pub fn paginate(&self, limit: Option<u64>, offset: Option<u64>) -> (u64, u64) {
let max_pagination = self.config().issues.max_pagination;
let default_pagination = self.config().issues.default_pagination;
(
limit.unwrap_or(default_pagination).clamp(1, max_pagination),
offset.unwrap_or(0).clamp(0, 5000),
)
}
pub fn db(&self) -> &Arc<AppDatabase> {
&self.state.database
}
pub async fn queue_search_issue(&self, issue_id: Uuid) -> Result<(), IssueError> {
search::outbox::enqueue_issue(self.db().write(), issue_id).await?;
Ok(())
}
pub fn email(&self) -> &EmailService {
&self.state.email
}
pub async fn load_user_briefs(
&self,
uids: &[Uuid],
) -> Result<HashMap<Uuid, UserDto>, IssueError> {
if uids.is_empty() {
return Ok(HashMap::new());
}
let mut set: Vec<Uuid> = uids.to_vec();
set.sort();
set.dedup();
let users = user::Entity::find()
.filter(user::Column::Uid.is_in(set))
.all(self.db().read())
.await?;
let map = users
.into_iter()
.map(|u| (u.uid, UserDto::from(u)))
.collect();
Ok(map)
}
#[allow(clippy::too_many_arguments)]
pub async fn resolve_issue_ref(
&self,
caller: Uuid,
space_name: &str,
board_id: Uuid,
issue_ref: &str,
) -> Result<Uuid, IssueError> {
let space = crate::share::lookup::lookup_space_by_name(self, space_name).await?;
crate::share::lookup::require_member(self, &space, caller).await?;
let board = crate::share::lookup::lookup_board(self, &space, board_id, caller).await?;
Ok(
crate::share::lookup::lookup_issue_by_ref(self, &board, issue_ref)
.await?
.id,
)
}
pub(crate) async fn write_event(
&self,
txn: &impl ConnectionTrait,
issue_id: Uuid,
user_id: Uuid,
event_type: IssueEventType,
comment_id: Option<Uuid>,
old_title: Option<String>,
new_title: Option<String>,
content: Option<String>,
) -> Result<(), IssueError> {
issue_event_line::ActiveModel {
id: Set(Uuid::now_v7()),
issue_id: Set(Some(issue_id)),
project_id: Set(None),
user_id: Set(user_id),
event_type: Set(event_type),
comment_id: Set(comment_id),
commit_sha: Set(None),
old_title: Set(old_title),
new_title: Set(new_title),
old_ref: Set(None),
new_ref: Set(None),
content: Set(content),
created_at: Set(OffsetDateTime::now_utc()),
}
.insert(txn)
.await?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn notify_user(
&self,
txn: &impl ConnectionTrait,
pending: &export::notify::PendingNotifications,
recipient: Uuid,
ntype: NotificationType,
title: String,
body: Option<String>,
data: Option<serde_json::Value>,
actor_uid: Option<Uuid>,
) -> Result<(), IssueError> {
let row = user_notifications::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(recipient),
notification_type: Set(ntype),
title: Set(title),
body: Set(body),
data_json: Set(data),
actor_uid: Set(actor_uid),
read_at: Set(None),
created_at: Set(OffsetDateTime::now_utc()),
}
.insert(txn)
.await?;
pending.push(recipient, row.to_event_value());
Ok(())
}
pub(crate) async fn collect_mentioned_uids(
&self,
space: &spaces::Model,
text: &str,
exclude: Uuid,
) -> Result<Vec<Uuid>, IssueError> {
let members = space_members::Entity::find()
.filter(space_members::Column::SpaceId.eq(space.id))
.filter(space_members::Column::Status.eq(MemberStatus::Active))
.all(self.db().read())
.await?;
let mut member_uids: Vec<Uuid> = members.into_iter().map(|m| m.user_id).collect();
if !member_uids.contains(&space.owner_id) {
member_uids.push(space.owner_id);
}
if member_uids.is_empty() {
return Ok(Vec::new());
}
let users = user::Entity::find()
.filter(user::Column::Uid.is_in(member_uids))
.all(self.db().read())
.await?;
let mut name_to_uid: HashMap<String, Uuid> = HashMap::new();
for u in users {
name_to_uid.insert(u.username.to_lowercase(), u.uid);
name_to_uid.insert(u.lower_name.to_lowercase(), u.uid);
name_to_uid.insert(u.uid.to_string(), u.uid);
}
let mut result_set: HashSet<Uuid> = HashSet::new();
let punctuation = [
',', '.', ';', ':', '!', '?', '\'', '"', '“', '”', '`', ')', ']', '}', '>',
];
for token in text.split_whitespace() {
if let Some(rest) = token.strip_prefix('@') {
let clean = rest.trim_end_matches(&punctuation[..]);
if clean.is_empty() {
continue;
}
let clean_lower = clean.to_lowercase();
if let Some(&uid) = name_to_uid.get(&clean_lower)
&& uid != exclude
{
result_set.insert(uid);
}
}
}
let mut uids: Vec<Uuid> = result_set.into_iter().collect();
uids.sort();
let max_mentions = self.config().issues.max_mentions_per_message;
if uids.len() > max_mentions {
uids.truncate(max_mentions);
}
Ok(uids)
}
pub(crate) async fn primary_emails(
&self,
uids: &[Uuid],
) -> Result<HashMap<Uuid, String>, IssueError> {
if uids.is_empty() {
return Ok(HashMap::new());
}
models::users::queries::primary_emails(self.db().read(), uids)
.await
.map_err(IssueError::from)
}
pub(crate) async fn send_email(
&self,
to: &str,
subject: &str,
text: &str,
html: &str,
) -> Result<(), IssueError> {
self.state
.email
.send_mail(to, subject, text, html)
.await
.map_err(|error| IssueError::Internal(format!("email send failed: {error}")))
}
}
#[async_trait]
impl SessionGuard for IssueService {
async fn lookup_session_with_user(
&self,
token_hash: &str,
) -> Result<Option<SessionWithUser>, SessionError> {
session::lookup_session_with_user(self.db().read(), token_hash).await
}
}