db73d56
ZhenYi
refactor: add events dependency and enhance event handling for improved clarity and organization
db73d56
use export::AppState;
use export::pagination::{Page, PageQuery};
use models::pull_request::{pull_request_event_line, pull_requests};
use sea_orm::ActiveValue::Set;
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter,
QueryOrder, QuerySelect,
};
use serde::{Deserialize, Serialize};
use space::services::spaces::UserDto;
use std::collections::HashMap;
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::PullError;
use crate::services::service::PullService;
pub const EVENT_CREATED: &str = "created";
pub const EVENT_UPDATED: &str = "updated";
pub const EVENT_CLOSED: &str = "closed";
pub const EVENT_REOPENED: &str = "reopened";
pub const EVENT_MERGED: &str = "merged";
pub const EVENT_COMMENTED: &str = "commented";
pub const EVENT_REVIEW_SUBMITTED: &str = "review_submitted";
pub const EVENT_REVIEW_DISMISSED: &str = "review_dismissed";
pub const EVENT_REVIEW_REQUESTED: &str = "review_requested";
pub const EVENT_REVIEW_REQUEST_CANCELLED: &str = "review_request_cancelled";
pub const EVENT_HEAD_UPDATED: &str = "head_updated";
pub const EVENT_LABELED: &str = "labeled";
pub const EVENT_UNLABELED: &str = "unlabeled";
pub const EVENT_ASSIGNED: &str = "assigned";
pub const EVENT_UNASSIGNED: &str = "unassigned";
pub const EVENT_LOCKED: &str = "locked";
pub const EVENT_UNLOCKED: &str = "unlocked";
pub const EVENT_AUTO_MERGE_ENABLED: &str = "auto_merge_enabled";
pub const EVENT_AUTO_MERGE_DISABLED: &str = "auto_merge_disabled";
#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
pub struct PullEventDto {
pub id: Uuid,
pub pull_request_id: Uuid,
pub event_type: String,
pub actor: Option<UserDto>,
pub commit_sha: Option<String>,
pub content: Option<String>,
pub created_at: OffsetDateTime,
}
pub type EventPage = Page<PullEventDto>;
pub async fn write_event(
state: &AppState,
db: &impl ConnectionTrait,
pull: &pull_requests::Model,
actor: Option<i64>,
event_type: &str,
commit_sha: Option<String>,
content: Option<String>,
) -> Result<(), sea_orm::DbErr> {
let actor = match actor {
Some(actor) => actor,
None => pull.author_id,
};
let mut payload = serde_json::json!({
"pull_request_id": pull.id,
"user_id": actor,
"event_type": event_type,
"commit_sha": &commit_sha,
"content": &content,
"pull_request": {
"id": pull.id,
"repo_id": pull.repo_id,
"issue_id": pull.issue_id,
"number": pull.autoincrement_id,
"author_id": pull.author_id,
"title": &pull.title,
"body": &pull.body,
"head_branch": &pull.head_branch,
"head_sha": &pull.head_sha,
"head_label": &pull.head_label,
"base_branch": &pull.base_branch,
"base_sha": &pull.base_sha,
"is_draft": pull.is_draft,
"merge_method": pull.merge_method,
"has_merged": pull.has_merged,
"merged_commit_sha": &pull.merged_commit_sha,
"squash_commit_sha": &pull.squash_commit_sha,
"auto_merge": pull.auto_merge,
"additions": pull.additions,
"deletions": pull.deletions,
"files_changed": pull.files_changed,
"merger_id": pull.merger_id,
"merged_at": pull.merged_at,
"is_locked": pull.is_locked,
"created_at": pull.created_at,
"updated_at": pull.updated_at,
"closed_at": pull.closed_at,
},
});
if let Ok(Some(actor)) = export::users::get_public_user(state, actor).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),
});
}
let row = pull_request_event_line::ActiveModel {
id: Set(Uuid::now_v7()),
pull_request_id: Set(pull.id),
user_id: Set(actor),
event_type: Set(event_type.to_string()),
commit_sha: Set(commit_sha),
content: Set(content),
created_at: Set(OffsetDateTime::now_utc()),
};
row.insert(db).await?;
let Some(kind) = event_kind(event_type) else {
return Ok(());
};
let mut labels = queue::Labels::new()
.with("pull_id", pull.id)
.with("user_id", actor);
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!("pull:{}", pull.id));
events::outbox::enqueue_now(
db,
kind,
labels,
payload,
request_id.as_deref().unwrap_or_default(),
request_id.as_deref(),
)
.await
.map(|_| ())
}
fn event_kind(event_type: &str) -> Option<&'static str> {
Some(match event_type {
EVENT_CREATED => "pull.pull.create",
EVENT_UPDATED => "pull.pull.update",
EVENT_CLOSED => "pull.pull.close",
EVENT_REOPENED => "pull.pull.reopen",
EVENT_MERGED => "pull.pull.merge",
EVENT_COMMENTED => "pull.comment.create",
EVENT_REVIEW_SUBMITTED => "pull.review.create",
EVENT_REVIEW_DISMISSED => "pull.review.delete",
EVENT_REVIEW_REQUESTED => "pull.review-request.create",
EVENT_REVIEW_REQUEST_CANCELLED => "pull.review-request.delete",
EVENT_HEAD_UPDATED => "pull.head.update",
EVENT_LABELED => "pull.label.create",
EVENT_UNLABELED => "pull.label.delete",
EVENT_ASSIGNED => "pull.assignee.create",
EVENT_UNASSIGNED => "pull.assignee.delete",
EVENT_LOCKED => "pull.lock.create",
EVENT_UNLOCKED => "pull.lock.delete",
EVENT_AUTO_MERGE_ENABLED => "pull.auto-merge.create",
EVENT_AUTO_MERGE_DISABLED => "pull.auto-merge.delete",
_ => return None,
})
}
impl PullService {
pub async fn record_event(
&self,
pull: &pull_requests::Model,
actor: Option<i64>,
event_type: &str,
commit_sha: Option<String>,
content: Option<String>,
) -> Result<(), PullError> {
write_event(
&self.state,
self.db_write(),
pull,
actor,
event_type,
commit_sha,
content,
)
.await?;
Ok(())
}
pub async fn list_events(
&self,
pull_id: Uuid,
page: PageQuery,
) -> Result<EventPage, PullError> {
let (limit, offset) = page.page();
let total = pull_request_event_line::Entity::find()
.filter(pull_request_event_line::Column::PullRequestId.eq(pull_id))
.count(self.db_read())
.await? as i64;
let rows = pull_request_event_line::Entity::find()
.filter(pull_request_event_line::Column::PullRequestId.eq(pull_id))
.order_by_asc(pull_request_event_line::Column::CreatedAt)
.limit(limit)
.offset(offset)
.all(self.db_read())
.await?;
let actors = self.user_briefs(&unique_users(&rows)).await?;
let items = rows
.into_iter()
.map(|row| PullEventDto {
id: row.id,
pull_request_id: row.pull_request_id,
event_type: row.event_type,
commit_sha: row.commit_sha,
content: row.content,
created_at: row.created_at,
actor: actors.get(&row.user_id).cloned(),
})
.collect();
Ok(Page {
items,
total,
limit,
offset,
})
}
pub(crate) async fn user_briefs(
&self,
ids: &[i64],
) -> Result<HashMap<i64, UserDto>, PullError> {
if ids.is_empty() {
return Ok(HashMap::new());
}
let users = export::users::load_public_users(&self.state, ids).await?;
Ok(users
.into_iter()
.map(|(uid, user)| (uid, UserDto::from_public(&user)))
.collect())
}
}
fn unique_users(rows: &[pull_request_event_line::Model]) -> Vec<i64> {
let mut ids: Vec<i64> = rows.iter().map(|row| row.user_id).collect();
ids.sort();
ids.dedup();
ids
}