use export::pagination::{Page, PageQuery}; use models::pull_request::{pull_request_event_line, pull_requests}; use models::users::user; use sea_orm::{ ActiveModelTrait, ActiveValue::Set, 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, pub commit_sha: Option, pub content: Option, pub created_at: OffsetDateTime, } pub type EventPage = Page; pub async fn write_event( 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 payload = serde_json::json!({ "pull_request_id": pull.id, "user_id": actor, "event_type": event_type, "commit_sha": &commit_sha, "content": &content, }); 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); if let Some(request_id) = export::events::request_id() { labels.insert("request_id", request_id); } export::events::enqueue(db, kind, labels, payload, format!("pull:{}", pull.id)).await } 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.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 }