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::dto::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, pub commit_sha: Option, pub content: Option, pub created_at: OffsetDateTime, } pub type EventPage = Page; pub async fn write_event( state: &AppState, db: &impl ConnectionTrait, pull: &pull_requests::Model, actor: Option, event_type: &str, commit_sha: Option, content: Option, ) -> 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, event_type: &str, commit_sha: Option, content: Option, ) -> 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 { 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, 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 { let mut ids: Vec = rows.iter().map(|row| row.user_id).collect(); ids.sort(); ids.dedup(); ids }