8cc1265
use export::users::UserDto;
use models::issue::issue_event_line::IssueEventType;
use models::issue::issue_member::{self, IssueMemberKind};
use models::issue::{comment, issue_content_history};
use models::users::user_notifications::NotificationType;
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
QuerySelect, Set, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::errors::IssueError;
use crate::services::service::{IssueEventDetail, IssueService};
use crate::share::authz;
use crate::share::lookup::{issue_context, require_issue_access};
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct CommentDto {
pub id: Uuid,
pub issue_id: Option<Uuid>,
pub parent_id: Option<Uuid>,
pub poster: Option<UserDto>,
pub content: String,
pub created_at: OffsetDateTime,
pub updated_at: Option<OffsetDateTime>,
}
pub type CommentPage = export::pagination::Page<CommentDto>;
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct CommentTreeNode {
pub comment: CommentDto,
#[schema(no_recursion)]
pub replies: Vec<CommentTreeNode>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateCommentCmd {
pub content: String,
#[serde(default)]
pub parent_id: Option<Uuid>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateCommentCmd {
pub content: String,
}
#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ListCommentQuery {
pub parent_id: Option<Uuid>,
pub limit: Option<u64>,
pub offset: Option<u64>,
pub sort: Option<String>,
}
#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct CommentPath {
pub space_name: String,
pub board_id: Uuid,
pub issue_id: String,
pub comment_id: Uuid,
}
impl IssueService {
#[tracing::instrument(name = "issues.create_comment", skip_all, fields(caller = %caller, space = %space_name, board_id = %board_id, issue_id = %issue_id))]
pub async fn create_comment(
&self,
caller: i64,
space_name: &str,
board_id: Uuid,
issue_id: Uuid,
cmd: CreateCommentCmd,
) -> Result<CommentDto, IssueError> {
self.access.allow_comment(caller).await?;
let (space, board, item) =
issue_context(self, caller, space_name, board_id, issue_id).await?;
let content = cmd.content.trim();
if content.is_empty() || content.chars().count() > self.settings.issues.max_comment_len {
return Err(IssueError::BadRequest(
"comment content is empty or too long".to_string(),
));
}
let mut parent_poster_id: Option<i64> = None;
if let Some(pid) = cmd.parent_id {
let parent_comment = comment::Entity::find_by_id(pid)
.one(self.database.read())
.await?
.ok_or_else(|| IssueError::BadRequest("parent comment not found".to_string()))?;
if parent_comment.issue_id != Some(issue_id) {
return Err(IssueError::BadRequest(
"parent comment must belong to the same issue".to_string(),
));
}
if parent_comment.poster_id != caller {
parent_poster_id = Some(parent_comment.poster_id);
}
}
let mentioned_uids = self.collect_mentioned_uids(&space, content, caller).await?;
let now = OffsetDateTime::now_utc();
let comment_id = Uuid::now_v7();
let content_str = content.to_string();
let parent_opt = cmd.parent_id;
let this = self.clone();
let space_lower_for_txn = space.slug().to_owned();
let item_title_for_txn = item.title.clone();
let mentioned_for_txn = mentioned_uids.clone();
let pending = export::notify::PendingNotifications::default();
let pending_txn = pending.clone();
let inserted =
self.database
.primary()
.transaction::<_, comment::Model, IssueError>(move |txn| {
let content_clone = content_str.clone();
let mentioned_clone = mentioned_for_txn;
let space_name_clone = space_lower_for_txn;
let issue_num = item.autoincrement_id;
let issue_title_clone = item_title_for_txn;
Box::pin(async move {
let row = comment::ActiveModel {
id: Set(comment_id),
issue_id: Set(Some(issue_id)),
project_id: Set(None),
parent_id: Set(parent_opt),
poster_id: Set(caller),
content: Set(content_clone.clone()),
created_at: Set(now),
updated_at: Set(None),
};
let ins = row.insert(txn).await?;
this.write_event(
txn,
issue_id,
caller,
IssueEventType::CommentCreated,
IssueEventDetail {
comment_id: Some(comment_id),
old_title: None,
new_title: None,
content: None,
},
)
.await?;
if let Some(pp_uid) = parent_poster_id {
let data = serde_json::json!({
"space_name": space_name_clone,
"board_id": board.id,
"issue_id": issue_id,
"comment_id": comment_id,
"autoincrement_id": issue_num,
});
this.audience
.notify_user(
txn,
&pending_txn,
pp_uid,
NotificationType::Reply,
format!("New reply on issue #{issue_num}: {issue_title_clone}"),
Some(content_clone.clone()),
Some(data),
Some(caller),
)
.await?;
}
for m_uid in mentioned_clone {
if Some(m_uid) == parent_poster_id {
continue;
}
let existing_member = issue_member::Entity::find()
.filter(issue_member::Column::IssueId.eq(issue_id))
.filter(issue_member::Column::UserId.eq(m_uid))
.filter(issue_member::Column::Kind.eq(IssueMemberKind::Mentioned))
.one(txn)
.await?;
if existing_member.is_none() {
let m_row = issue_member::ActiveModel {
id: Set(Uuid::now_v7()),
issue_id: Set(issue_id),
user_id: Set(m_uid),
kind: Set(IssueMemberKind::Mentioned),
created_at: Set(now),
updated_at: Set(None),
};
m_row.insert(txn).await?;
}
let data = serde_json::json!({
"space_name": space_name_clone,
"board_id": board.id,
"issue_id": issue_id,
"comment_id": comment_id,
"autoincrement_id": issue_num,
});
this.audience.notify_user(
txn,
&pending_txn,
m_uid,
NotificationType::Mention,
format!(
"You were mentioned in issue #{issue_num}: {issue_title_clone}"
),
Some(content_clone.clone()),
Some(data),
Some(caller),
)
.await?;
}
Ok(ins)
})
})
.await?;
pending.flush(&self.audience.notifications).await;
let mut email_recipients = Vec::new();
if let Some(pp_uid) = parent_poster_id {
email_recipients.push(pp_uid);
}
for m_uid in &mentioned_uids {
if !email_recipients.contains(m_uid) {
email_recipients.push(*m_uid);
}
}
if !email_recipients.is_empty() {
let emails = self.audience.primary_emails(&email_recipients).await?;
for (uid, addr) in emails {
let locale = self.audience.recipient_locale(uid).await;
let is_reply = Some(uid) == parent_poster_id;
let subject = if is_reply {
i18n::catalog::issue_reply_subject(locale, item.autoincrement_id, &item.title)
} else {
i18n::catalog::issue_mention_subject(locale, item.autoincrement_id, &item.title)
};
let link = format!(
"{}/issues/{}/{}/{}",
self.settings.web.web_base_url,
space.slug(),
board.id,
item.id
);
let text = format!("{subject}\n\n{}: {link}", i18n::catalog::view_issue(locale));
let html = format!(
r#"<p>{subject}</p><p><a href="{link}">{}</a></p>"#,
i18n::catalog::view_issue(locale)
);
if let Err(err) = self
.audience
.send_email(&addr, &subject, &text, &html)
.await
{
tracing::warn!(error = %err, "failed to send comment notification email");
}
}
}
let user_map = self
.audience
.load_user_briefs(&[inserted.poster_id])
.await?;
let poster = user_map.get(&inserted.poster_id).cloned();
Ok(CommentDto {
id: inserted.id,
issue_id: inserted.issue_id,
parent_id: inserted.parent_id,
poster,
content: inserted.content,
created_at: inserted.created_at,
updated_at: inserted.updated_at,
})
}
pub async fn list_comments(
&self,
caller: i64,
space_name: &str,
board_id: Uuid,
issue_id: Uuid,
q: ListCommentQuery,
) -> Result<CommentPage, IssueError> {
require_issue_access(self, caller, space_name, board_id, issue_id).await?;
let (limit, offset) = self.paginate(q.limit, q.offset);
let mut select =
comment::Entity::find().filter(comment::Column::IssueId.eq(Some(issue_id)));
if let Some(pid) = q.parent_id {
select = select.filter(comment::Column::ParentId.eq(Some(pid)));
}
match q.sort.as_deref() {
Some("-created_at") => select = select.order_by_desc(comment::Column::CreatedAt),
Some("updated_at") => select = select.order_by_desc(comment::Column::UpdatedAt),
_ => select = select.order_by_asc(comment::Column::CreatedAt),
}
let total = select.clone().count(self.database.read()).await? as i64;
let rows = select
.limit(limit)
.offset(offset)
.all(self.database.read())
.await?;
let poster_ids: Vec<i64> = rows.iter().map(|c| c.poster_id).collect();
let user_map = self.audience.load_user_briefs(&poster_ids).await?;
let items = rows
.into_iter()
.map(|c| {
let poster = user_map.get(&c.poster_id).cloned();
CommentDto {
id: c.id,
issue_id: c.issue_id,
parent_id: c.parent_id,
poster,
content: c.content,
created_at: c.created_at,
updated_at: c.updated_at,
}
})
.collect();
Ok(CommentPage {
items,
total,
limit,
offset,
})
}
pub async fn list_comments_tree(
&self,
caller: i64,
space_name: &str,
board_id: Uuid,
issue_id: Uuid,
) -> Result<Vec<CommentTreeNode>, IssueError> {
require_issue_access(self, caller, space_name, board_id, issue_id).await?;
let rows = comment::Entity::find()
.filter(comment::Column::IssueId.eq(Some(issue_id)))
.order_by_asc(comment::Column::CreatedAt)
.all(self.database.read())
.await?;
let poster_ids: Vec<i64> = rows.iter().map(|c| c.poster_id).collect();
let user_map = self.audience.load_user_briefs(&poster_ids).await?;
let dtos: Vec<CommentDto> = rows
.into_iter()
.map(|c| {
let poster = user_map.get(&c.poster_id).cloned();
CommentDto {
id: c.id,
issue_id: c.issue_id,
parent_id: c.parent_id,
poster,
content: c.content,
created_at: c.created_at,
updated_at: c.updated_at,
}
})
.collect();
let mut children: std::collections::HashMap<Option<Uuid>, Vec<CommentDto>> =
std::collections::HashMap::new();
for dto in dtos {
children.entry(dto.parent_id).or_default().push(dto);
}
fn build(
parent: Option<Uuid>,
children: &mut std::collections::HashMap<Option<Uuid>, Vec<CommentDto>>,
) -> Vec<CommentTreeNode> {
let nodes = children.remove(&parent).unwrap_or_default();
nodes
.into_iter()
.map(|comment| CommentTreeNode {
comment: comment.clone(),
replies: build(Some(comment.id), children),
})
.collect()
}
Ok(build(None, &mut children))
}
pub async fn get_comment(
&self,
caller: i64,
space_name: &str,
board_id: Uuid,
issue_id: Uuid,
comment_id: Uuid,
) -> Result<CommentDto, IssueError> {
require_issue_access(self, caller, space_name, board_id, issue_id).await?;
let c = comment::Entity::find_by_id(comment_id)
.filter(comment::Column::IssueId.eq(Some(issue_id)))
.one(self.database.read())
.await?
.ok_or(IssueError::NotFound)?;
let user_map = self.audience.load_user_briefs(&[c.poster_id]).await?;
let poster = user_map.get(&c.poster_id).cloned();
Ok(CommentDto {
id: c.id,
issue_id: c.issue_id,
parent_id: c.parent_id,
poster,
content: c.content,
created_at: c.created_at,
updated_at: c.updated_at,
})
}
#[tracing::instrument(name = "issues.update_comment", skip_all, fields(caller = %caller, space = %space_name, board_id = %board_id, issue_id = %issue_id, comment_id = %comment_id))]
pub async fn update_comment(
&self,
caller: i64,
space_name: &str,
board_id: Uuid,
issue_id: Uuid,
comment_id: Uuid,
cmd: UpdateCommentCmd,
) -> Result<CommentDto, IssueError> {
let (space, _, _) = issue_context(self, caller, space_name, board_id, issue_id).await?;
let c = comment::Entity::find_by_id(comment_id)
.filter(comment::Column::IssueId.eq(Some(issue_id)))
.one(self.database.read())
.await?
.ok_or(IssueError::NotFound)?;
if c.poster_id != caller && !authz::is_board_manager(self, &space, caller).await? {
return Err(IssueError::Forbidden);
}
let content = cmd.content.trim();
if content.is_empty() || content.chars().count() > self.settings.issues.max_comment_len {
return Err(IssueError::BadRequest(
"comment content is empty or too long".to_string(),
));
}
let old_content = c.content.clone();
let content_changed: Option<String> = (content != old_content).then(|| content.to_string());
let now = OffsetDateTime::now_utc();
let mut am: comment::ActiveModel = c.into();
am.content = Set(content.to_string());
am.updated_at = Set(Some(now));
let this = self.clone();
let updated = self
.database
.primary()
.transaction::<_, comment::Model, IssueError>(move |txn| {
let content_change_clone = content_changed.clone();
Box::pin(async move {
let u = am.update(txn).await?;
if let Some(new_content) = content_change_clone {
issue_content_history::ActiveModel {
id: Set(Uuid::now_v7()),
issue_id: Set(issue_id),
poster_id: Set(caller),
comment_id: Set(Some(comment_id)),
content_before: Set(Some(old_content)),
content_after: Set(Some(new_content)),
edited_at: Set(now),
created_at: Set(now),
}
.insert(txn)
.await?;
}
this.write_event(
txn,
issue_id,
caller,
IssueEventType::CommentUpdated,
IssueEventDetail {
comment_id: Some(comment_id),
old_title: None,
new_title: None,
content: None,
},
)
.await?;
Ok(u)
})
})
.await?;
let user_map = self.audience.load_user_briefs(&[updated.poster_id]).await?;
let poster = user_map.get(&updated.poster_id).cloned();
Ok(CommentDto {
id: updated.id,
issue_id: updated.issue_id,
parent_id: updated.parent_id,
poster,
content: updated.content,
created_at: updated.created_at,
updated_at: updated.updated_at,
})
}
#[tracing::instrument(name = "issues.delete_comment", skip_all, fields(caller = %caller, space = %space_name, board_id = %board_id, issue_id = %issue_id, comment_id = %comment_id))]
pub async fn delete_comment(
&self,
caller: i64,
space_name: &str,
board_id: Uuid,
issue_id: Uuid,
comment_id: Uuid,
) -> Result<(), IssueError> {
let (space, _, _) = issue_context(self, caller, space_name, board_id, issue_id).await?;
let c = comment::Entity::find_by_id(comment_id)
.filter(comment::Column::IssueId.eq(Some(issue_id)))
.one(self.database.read())
.await?
.ok_or(IssueError::NotFound)?;
if c.poster_id != caller && !authz::is_board_manager(self, &space, caller).await? {
return Err(IssueError::Forbidden);
}
let this = self.clone();
self.database
.primary()
.transaction::<_, (), IssueError>(move |txn| {
Box::pin(async move {
comment::Entity::update_many()
.filter(comment::Column::ParentId.eq(comment_id))
.col_expr(comment::Column::ParentId, Expr::value(None::<Uuid>))
.exec(txn)
.await?;
comment::Entity::delete_by_id(comment_id).exec(txn).await?;
this.write_event(
txn,
issue_id,
caller,
IssueEventType::CommentDeleted,
IssueEventDetail {
comment_id: Some(comment_id),
old_title: None,
new_title: None,
content: None,
},
)
.await?;
Ok(())
})
})
.await?;
Ok(())
}
}