44c6130
ZhenYi
refactor: update database handling and introduce BlobStore trait for improved abstraction and flexibility
44c6130
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use config::AppConfig;
use export::AppState;
use export::database::Database;
use i18n;
use models::issue::issue_event_line::{self, IssueEventType};
use models::issue::issue_item;
use models::space::spaces;
use models::users::user_notifications::{self, NotificationType};
use sea_orm::{ActiveModelTrait, ConnectionTrait, EntityTrait, Set};
use space::services::spaces::UserDto;
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::IssueError;
#[derive(Clone)]
pub struct IssueService {
pub state: AppState,
}
/// The optional "what changed" columns of an issue activity row. Grouped into a
/// struct so `write_event` does not take four positionally-interchangeable
/// `Option` arguments.
#[derive(Default)]
pub(crate) struct IssueEventDetail {
pub comment_id: Option<Uuid>,
pub old_title: Option<String>,
pub new_title: Option<String>,
pub content: Option<String>,
}
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) {
export::pagination::page_bounds(
limit,
offset,
self.config().issues.default_pagination,
self.config().issues.max_pagination,
)
}
pub fn db(&self) -> &Arc<dyn Database> {
&self.state.database
}
pub async fn load_user_briefs(
&self,
uids: &[i64],
) -> Result<HashMap<i64, UserDto>, IssueError> {
Ok(export::users::load_user_briefs(&self.state, uids).await?)
}
#[allow(clippy::too_many_arguments)]
pub async fn resolve_issue_ref(
&self,
caller: i64,
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,
)
}
/// Record one activity row. The optional columns all describe "what changed"
/// and are grouped so the call sites stay readable.
pub(crate) async fn write_event(
&self,
txn: &impl ConnectionTrait,
issue_id: Uuid,
user_id: i64,
event_type: IssueEventType,
detail: IssueEventDetail,
) -> Result<(), IssueError> {
let IssueEventDetail {
comment_id,
old_title,
new_title,
content,
} = detail;
let mut payload = serde_json::json!({
"issue_id": issue_id,
"user_id": user_id,
"event_type": event_type,
"comment_id": &comment_id,
"old_title": &old_title,
"new_title": &new_title,
"content": &content,
});
if let Ok(Some(issue)) = issue_item::Entity::find_by_id(issue_id).one(txn).await {
payload["issue"] = serde_json::json!({
"id": issue.id,
"number": issue.autoincrement_id,
"collection_id": issue.collection_id,
"author_id": issue.author_id,
"title": &issue.title,
"description": &issue.description,
"priority": issue.priority,
"milestone_id": issue.milestone_id,
"created_at": issue.created_at,
"updated_at": issue.updated_at,
"closed_at": issue.closed_at,
});
}
if let Ok(Some(actor)) = export::users::get_public_user(&self.state, user_id).await {
payload["actor"] = serde_json::json!({
"id": actor.user_id,
"username": actor.username,
"display_name": actor.displayname,
"avatar_url": export::resolve_user_avatar_url(actor.user_id, &actor.avatar_url),
});
}
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?;
let mut labels = queue::Labels::new()
.with("issue_id", issue_id)
.with("user_id", user_id);
let request_id = export::middleware::request_id();
if let Some(request_id) = request_id.as_deref() {
labels.insert("request_id", request_id);
}
labels.insert(events::label::ORDERING_KEY, format!("issue:{issue_id}"));
events::outbox::enqueue_now(
txn,
event_kind(event_type),
labels,
payload,
request_id.as_deref().unwrap_or_default(),
request_id.as_deref(),
)
.await?;
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn notify_user(
&self,
txn: &impl ConnectionTrait,
pending: &export::notify::PendingNotifications,
recipient: i64,
ntype: NotificationType,
title: String,
body: Option<String>,
data: Option<serde_json::Value>,
actor_uid: Option<i64>,
) -> 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: i64,
) -> Result<Vec<i64>, IssueError> {
let members =
space::authority::list_members(&self.state, space.owner_id, space.id, false, false)
.await?;
let mut member_uids: Vec<i64> = 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 = export::users::load_public_users(&self.state, &member_uids).await?;
let mut name_to_uid: HashMap<String, i64> = HashMap::new();
for (uid, user) in users {
name_to_uid.insert(user.username.to_lowercase(), uid);
name_to_uid.insert(uid.to_string(), uid);
}
let mut result_set: HashSet<i64> = 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<i64> = 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)
}
/// Language for one recipient's issue mail.
///
/// Account preference only: an issue event fans out to watchers, the
/// assignee and mentioned users, each of whom may read a different language.
pub(crate) async fn recipient_locale(&self, uid: i64) -> i18n::UiLocale {
let fallback = i18n::config_fallback(self.config().email.default_locale());
i18n::resolve_user_locale(self.db().read(), uid, fallback).await
}
pub(crate) async fn primary_emails(
&self,
uids: &[i64],
) -> Result<HashMap<i64, String>, IssueError> {
if uids.is_empty() {
return Ok(HashMap::new());
}
let users = export::users::load_public_users(&self.state, uids).await?;
Ok(users
.into_iter()
.filter(|(_, user)| user.email_verified && !user.email.trim().is_empty())
.map(|(uid, user)| (uid, user.email.to_lowercase()))
.collect())
}
pub(crate) async fn send_email(
&self,
to: &str,
subject: &str,
text: &str,
html: &str,
) -> Result<(), IssueError> {
let Some(client) = self.state.mail.as_ref() else {
tracing::warn!(to, "email rpc disabled; skipping issue mail");
return Ok(());
};
client
.send_mail(to, subject, text, html, None)
.await
.map_err(|error| IssueError::Internal(format!("email send failed: {error}")))
}
}
fn event_kind(event_type: IssueEventType) -> &'static str {
match event_type {
IssueEventType::Created => "issue.issue.create",
IssueEventType::TitleUpdated => "issue.title.update",
IssueEventType::DescriptionUpdated => "issue.description.update",
IssueEventType::ContentUpdated => "issue.content.update",
IssueEventType::MilestoneUpdated => "issue.milestone.update",
IssueEventType::PriorityUpdated => "issue.priority.update",
IssueEventType::DueUpdated => "issue.due.update",
IssueEventType::Closed => "issue.issue.close",
IssueEventType::Reopened => "issue.issue.reopen",
IssueEventType::CommentCreated => "issue.comment.create",
IssueEventType::CommentUpdated => "issue.comment.update",
IssueEventType::CommentDeleted => "issue.comment.delete",
IssueEventType::MemberAdded => "issue.member.create",
IssueEventType::MemberRemoved => "issue.member.delete",
IssueEventType::LabelAdded => "issue.label.create",
IssueEventType::LabelRemoved => "issue.label.delete",
IssueEventType::DependencyAdded => "issue.dependency.create",
IssueEventType::DependencyRemoved => "issue.dependency.delete",
IssueEventType::Pinned => "issue.pin.create",
IssueEventType::Unpinned => "issue.pin.delete",
IssueEventType::RepoRelationAdded => "issue.repo-relation.create",
IssueEventType::RepoRelationRemoved => "issue.repo-relation.delete",
IssueEventType::PullLinked => "issue.pull-link.create",
IssueEventType::PullUnlinked => "issue.pull-link.delete",
}
}