b7cd2b1
use export::pagination::Page;
use git::context::facade::GitContext;
use models::issue::issue_relation_pull;
use models::pull_request::{
pull_request_assignee, pull_request_comment, pull_request_event_line, pull_request_label,
pull_request_review, pull_request_reviewer, pull_requests,
};
use models::repo::repos;
use sea_orm::ActiveValue::Set;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
QuerySelect, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use space::services::spaces::UserDto;
use std::collections::HashMap;
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::errors::PullError;
use crate::services::events::{
EVENT_CLOSED, EVENT_CREATED, EVENT_MERGED, EVENT_REOPENED, EVENT_UPDATED, write_event,
};
use crate::services::service::PullService;
pub const MAX_TITLE_LEN: usize = 512;
pub const MAX_BODY_LEN: usize = 65_536;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum PullState {
Open,
Closed,
Merged,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct PullRefDto {
pub label: String,
pub branch: String,
pub sha: String,
pub repo_id: Option<Uuid>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct PullDto {
pub id: Uuid,
pub repo_id: Uuid,
pub number: i32,
pub title: String,
pub body: String,
pub state: PullState,
pub is_draft: bool,
pub auto_merge: bool,
pub locked: bool,
pub head: PullRefDto,
pub base: PullRefDto,
pub author: Option<UserDto>,
pub merger: Option<UserDto>,
pub merge_method: models::pull_request::pull_requests::MergeMethod,
pub additions: i32,
pub deletions: i32,
pub files_changed: i32,
pub merged_commit_sha: Option<String>,
pub squash_commit_sha: Option<String>,
pub issue_id: Option<Uuid>,
pub created_at: OffsetDateTime,
pub updated_at: Option<OffsetDateTime>,
pub closed_at: Option<OffsetDateTime>,
pub merged_at: Option<OffsetDateTime>,
#[serde(skip_serializing_if = "Option::is_none")]
pub mergeable: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub mergeable_state: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub conflicts: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub commits_ahead: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub commits_behind: Option<i64>,
}
pub type PullPage = Page<PullDto>;
#[derive(Debug, Clone, Default, Deserialize, IntoParams)]
pub struct ListPullQuery {
/// open | closed | merged | all
#[serde(default)]
pub state: Option<String>,
#[serde(default)]
pub author: Option<String>,
#[serde(default)]
pub base: Option<String>,
#[serde(default)]
pub head: Option<String>,
#[serde(default)]
pub draft: Option<bool>,
#[serde(default)]
pub search: Option<String>,
/// number | created | updated
#[serde(default)]
pub sort: Option<String>,
/// asc | desc
#[serde(default)]
pub order: Option<String>,
#[serde(default)]
pub limit: Option<u64>,
#[serde(default)]
pub offset: Option<u64>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreatePullCmd {
pub title: String,
#[serde(default)]
pub body: Option<String>,
pub head_branch: String,
#[serde(default)]
pub base_branch: Option<String>,
#[serde(default)]
pub draft: Option<bool>,
#[serde(default)]
pub issue_id: Option<Uuid>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdatePullCmd {
#[serde(default)]
pub title: Option<String>,
#[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
pub body: Option<Option<String>>,
#[serde(default)]
pub base_branch: Option<String>,
#[serde(default)]
pub draft: Option<bool>,
#[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
pub auto_merge: Option<Option<bool>>,
#[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
pub issue_id: Option<Option<Uuid>>,
}
fn state_of(pull: &pull_requests::Model) -> PullState {
if pull.has_merged {
PullState::Merged
} else if pull.closed_at.is_some() {
PullState::Closed
} else {
PullState::Open
}
}
impl PullService {
pub(crate) fn dto(pull: &pull_requests::Model, users: &HashMap<i64, UserDto>) -> PullDto {
PullDto {
id: pull.id,
repo_id: pull.repo_id,
number: pull.autoincrement_id,
title: pull.title.clone(),
body: pull.body.clone(),
state: state_of(pull),
is_draft: pull.is_draft,
auto_merge: pull.auto_merge,
locked: pull.is_locked,
head: PullRefDto {
label: pull
.head_label
.clone()
.unwrap_or_else(|| pull.head_branch.clone()),
branch: pull.head_branch.clone(),
sha: pull.head_sha.clone(),
repo_id: pull.head_repo_id,
},
base: PullRefDto {
label: pull.base_branch.clone(),
branch: pull.base_branch.clone(),
sha: pull.base_sha.clone(),
repo_id: Some(pull.repo_id),
},
author: users.get(&pull.author_id).cloned(),
merger: pull.merger_id.and_then(|id| users.get(&id).cloned()),
merge_method: pull.merge_method,
additions: pull.additions,
deletions: pull.deletions,
files_changed: pull.files_changed,
merged_commit_sha: pull.merged_commit_sha.clone(),
squash_commit_sha: pull.squash_commit_sha.clone(),
issue_id: pull.issue_id,
created_at: pull.created_at,
updated_at: pull.updated_at,
closed_at: pull.closed_at,
merged_at: pull.merged_at,
mergeable: None,
mergeable_state: None,
conflicts: None,
commits_ahead: None,
commits_behind: None,
}
}
pub async fn list_pulls(&self, repo_id: Uuid, q: ListPullQuery) -> Result<PullPage, PullError> {
let limit = q.limit.unwrap_or(50).clamp(1, 100);
let offset = q.offset.unwrap_or(0);
let state = q.state.clone().unwrap_or_else(|| "open".to_string());
let mut select =
pull_requests::Entity::find().filter(pull_requests::Column::RepoId.eq(repo_id));
match state.as_str() {
"all" => {}
"open" => {
select = select
.filter(pull_requests::Column::HasMerged.eq(false))
.filter(pull_requests::Column::ClosedAt.is_null());
}
"closed" => {
select = select
.filter(pull_requests::Column::HasMerged.eq(false))
.filter(pull_requests::Column::ClosedAt.is_not_null());
}
"merged" => select = select.filter(pull_requests::Column::HasMerged.eq(true)),
other => {
return Err(PullError::BadRequest(format!("unknown state `{other}`")));
}
}
if let Some(author) = q.author.as_deref().filter(|a| !a.trim().is_empty()) {
let uid = self.resolve_user_id(author).await?;
select = select.filter(pull_requests::Column::AuthorId.eq(uid));
}
if let Some(base) = q.base.as_deref().filter(|b| !b.trim().is_empty()) {
select = select.filter(pull_requests::Column::BaseBranch.eq(base));
}
if let Some(head) = q.head.as_deref().filter(|h| !h.trim().is_empty()) {
select = select.filter(pull_requests::Column::HeadBranch.eq(head));
}
if let Some(draft) = q.draft {
select = select.filter(pull_requests::Column::IsDraft.eq(draft));
}
if let Some(search) = q.search.as_deref().filter(|s| !s.trim().is_empty()) {
select = select.filter(pull_requests::Column::Title.contains(search));
}
let total = select.clone().count(self.db_read()).await? as i64;
let descending = q.order.as_deref().unwrap_or("desc") != "asc";
select = match q.sort.as_deref().unwrap_or("created") {
"number" => {
if descending {
select.order_by_desc(pull_requests::Column::AutoincrementId)
} else {
select.order_by_asc(pull_requests::Column::AutoincrementId)
}
}
"updated" => {
if descending {
select.order_by_desc(pull_requests::Column::UpdatedAt)
} else {
select.order_by_asc(pull_requests::Column::UpdatedAt)
}
}
_ => {
if descending {
select.order_by_desc(pull_requests::Column::CreatedAt)
} else {
select.order_by_asc(pull_requests::Column::CreatedAt)
}
}
};
let rows = select
.limit(limit)
.offset(offset)
.all(self.db_read())
.await?;
let users = self.pull_users(&rows).await?;
let items = rows.iter().map(|row| Self::dto(row, &users)).collect();
Ok(Page {
items,
total,
limit,
offset,
})
}
pub async fn create_pull(
&self,
space_name: String,
repo_name: String,
ctx: &GitContext,
caller: i64,
cmd: CreatePullCmd,
) -> Result<pull_requests::Model, PullError> {
if !ctx.model.allow_pull_requests {
return Err(PullError::BadRequest(
"pull requests are disabled for this repository".to_string(),
));
}
let title = cmd.title.trim();
if title.is_empty() || title.chars().count() > MAX_TITLE_LEN {
return Err(PullError::BadRequest(format!(
"title must be 1-{MAX_TITLE_LEN} characters"
)));
}
let body = cmd.body.clone().unwrap_or_default();
if body.chars().count() > MAX_BODY_LEN {
return Err(PullError::BadRequest("body too large".to_string()));
}
if !self
.state
.redis
.quota_allow(
&format!("quota:pull_create:{caller}"),
self.config().quota.pull_create_per_day,
86_400,
)
.await
{
return Err(PullError::RateLimited(
"pull request creation quota exceeded, retry tomorrow".to_string(),
));
}
let base_branch = cmd
.base_branch
.clone()
.unwrap_or_else(|| ctx.model.default_branch.clone());
if base_branch == cmd.head_branch {
return Err(PullError::BadRequest(
"head and base branch must differ".to_string(),
));
}
if let Some(issue_id) = cmd.issue_id
&& !models::issue::issue_item::Entity::find_by_id(issue_id)
.one(self.db_read())
.await?
.is_some()
{
return Err(PullError::BadRequest("issue not found".to_string()));
}
let plan = self
.git
.merge_plan(
ctx,
&format!("refs/heads/{base_branch}"),
&format!("refs/heads/{}", cmd.head_branch),
)
.await
.map_err(|e| match e {
git::share::errors::GitError::NotFound => PullError::BadRequest(
"head or base branch not found in this repository".to_string(),
),
other => PullError::from(other),
})?;
if plan.head_only == 0 {
return Err(PullError::Conflict(
"the head branch has no commits ahead of the base branch".to_string(),
));
}
let stats = self
.diff_stats_for(&space_name, &repo_name, &plan, &plan.head_sha)
.await?;
let txn = self.state.database.write().begin().await?;
let _ = repos::Entity::find_by_id(ctx.model.id)
.lock_exclusive()
.one(&txn)
.await?;
let last = pull_requests::Entity::find()
.filter(pull_requests::Column::RepoId.eq(ctx.model.id))
.order_by_desc(pull_requests::Column::AutoincrementId)
.one(&txn)
.await?;
let number = last.map(|row| row.autoincrement_id + 1).unwrap_or(1);
let now = OffsetDateTime::now_utc();
let created = pull_requests::ActiveModel {
id: Set(Uuid::now_v7()),
repo_id: Set(ctx.model.id),
issue_id: Set(cmd.issue_id),
autoincrement_id: Set(number),
author_id: Set(caller),
title: Set(title.to_string()),
body: Set(body),
head_branch: Set(cmd.head_branch.clone()),
head_sha: Set(plan.head_sha.clone()),
head_repo_id: Set(None),
head_label: Set(Some(cmd.head_branch.to_string())),
base_branch: Set(base_branch.clone()),
base_sha: Set(plan.base_sha.clone()),
is_draft: Set(cmd.draft.unwrap_or(false)),
merge_method: Set(models::pull_request::pull_requests::MergeMethod::Merge),
has_merged: Set(false),
merged_commit_sha: Set(None),
squash_commit_sha: Set(None),
auto_merge: Set(false),
additions: Set(stats.0),
deletions: Set(stats.1),
files_changed: Set(stats.2),
merger_id: Set(None),
merged_at: Set(None),
is_locked: Set(false),
created_at: Set(now),
updated_at: Set(None),
closed_at: Set(None),
};
let pull = created.insert(&txn).await?;
write_event(
&txn,
&pull,
Some(caller),
EVENT_CREATED,
Some(plan.head_sha.clone()),
None,
)
.await?;
txn.commit().await?;
self.recount_open(ctx.model.id).await?;
Ok(pull)
}
/// `handle` carries the resolved repository, pull row, caller and admin bit
/// so the signature does not repeat four positionally-similar arguments.
pub async fn update_pull(
&self,
space_name: String,
repo_name: String,
handle: &crate::share::lookup::PullHandle,
cmd: UpdatePullCmd,
) -> Result<pull_requests::Model, PullError> {
let crate::share::lookup::PullHandle {
ctx, pull, admin, ..
} = handle;
let caller = handle.caller.ok_or(PullError::Unauthorized)?;
let admin = *admin;
if pull.is_locked && !admin {
return Err(PullError::Forbidden);
}
let mut model: pull_requests::ActiveModel = pull.clone().into();
let mut changed = false;
if let Some(title) = cmd.title.as_deref() {
let title = title.trim();
if title.is_empty() || title.chars().count() > MAX_TITLE_LEN {
return Err(PullError::BadRequest("invalid title".to_string()));
}
model.title = Set(title.to_string());
changed = true;
}
if let Some(body) = cmd.body {
if body
.as_deref()
.is_some_and(|b| b.chars().count() > MAX_BODY_LEN)
{
return Err(PullError::BadRequest("body too large".to_string()));
}
model.body = Set(body.unwrap_or_default());
changed = true;
}
if let Some(draft) = cmd.draft {
model.is_draft = Set(draft);
changed = true;
}
if let Some(auto) = cmd.auto_merge {
if auto.is_some_and(|a| a) {
crate::share::authz::require_open(pull)?;
}
model.auto_merge = Set(auto.unwrap_or(false));
changed = true;
}
if let Some(issue_id) = cmd.issue_id {
if let Some(id) = issue_id
&& !models::issue::issue_item::Entity::find_by_id(id)
.one(self.db_read())
.await?
.is_some()
{
return Err(PullError::BadRequest("issue not found".to_string()));
}
model.issue_id = Set(issue_id);
changed = true;
}
if let Some(base_branch) = cmd.base_branch.as_deref() {
if base_branch == pull.head_branch {
return Err(PullError::BadRequest(
"base branch must differ from head branch".to_string(),
));
}
let plan = self
.git
.merge_plan(ctx, &format!("refs/heads/{base_branch}"), &pull.head_sha)
.await
.map_err(|_| {
PullError::BadRequest(format!("base branch `{base_branch}` not found"))
})?;
model.base_branch = Set(base_branch.to_string());
model.base_sha = Set(plan.base_sha);
changed = true;
}
if !changed {
return Ok(pull.clone());
}
model.updated_at = Set(Some(OffsetDateTime::now_utc()));
let updated = model.update(self.db_write()).await?;
self.record_event(&updated, Some(caller), EVENT_UPDATED, None, None)
.await?;
let _ = (space_name, repo_name);
Ok(updated)
}
pub async fn set_state(
&self,
pull: &pull_requests::Model,
caller: i64,
closed: bool,
) -> Result<pull_requests::Model, PullError> {
if pull.has_merged {
return Err(PullError::Conflict(
"a merged pull request cannot change state".to_string(),
));
}
if closed && pull.closed_at.is_some() {
return Ok(pull.clone());
}
if !closed && pull.closed_at.is_none() {
return Ok(pull.clone());
}
let mut model: pull_requests::ActiveModel = pull.clone().into();
model.closed_at = Set(if closed {
Some(OffsetDateTime::now_utc())
} else {
None
});
model.updated_at = Set(Some(OffsetDateTime::now_utc()));
let updated = model.update(self.db_write()).await?;
self.record_event(
&updated,
Some(caller),
if closed { EVENT_CLOSED } else { EVENT_REOPENED },
None,
None,
)
.await?;
self.recount_open(updated.repo_id).await?;
Ok(updated)
}
pub async fn delete_pull(
&self,
pull: &pull_requests::Model,
caller: i64,
) -> Result<(), PullError> {
let _ = caller;
let id = pull.id;
let txn = self.state.database.write().begin().await?;
pull_request_comment::Entity::delete_many()
.filter(pull_request_comment::Column::PullRequestId.eq(id))
.exec(&txn)
.await?;
pull_request_review::Entity::delete_many()
.filter(pull_request_review::Column::PullRequestId.eq(id))
.exec(&txn)
.await?;
pull_request_reviewer::Entity::delete_many()
.filter(pull_request_reviewer::Column::PullRequestId.eq(id))
.exec(&txn)
.await?;
pull_request_label::Entity::delete_many()
.filter(pull_request_label::Column::PullRequestId.eq(id))
.exec(&txn)
.await?;
pull_request_assignee::Entity::delete_many()
.filter(pull_request_assignee::Column::PullRequestId.eq(id))
.exec(&txn)
.await?;
pull_request_event_line::Entity::delete_many()
.filter(pull_request_event_line::Column::PullRequestId.eq(id))
.exec(&txn)
.await?;
issue_relation_pull::Entity::delete_many()
.filter(issue_relation_pull::Column::PullRequestId.eq(id))
.exec(&txn)
.await?;
pull_requests::Entity::delete_many()
.filter(pull_requests::Column::Id.eq(id))
.exec(&txn)
.await?;
txn.commit().await?;
self.recount_open(pull.repo_id).await?;
Ok(())
}
pub(crate) async fn pull_users(
&self,
rows: &[pull_requests::Model],
) -> Result<HashMap<i64, UserDto>, PullError> {
let mut ids: Vec<i64> = rows
.iter()
.flat_map(|row| [Some(row.author_id), row.merger_id])
.flatten()
.collect();
ids.sort();
ids.dedup();
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())
}
pub(crate) async fn resolve_user_id(&self, reference: &str) -> Result<i64, PullError> {
let reference = reference.trim();
if let Ok(uid) = reference.parse::<i64>() {
return Ok(uid);
}
let user = export::users::find_by_username(&self.state, reference)
.await?
.ok_or_else(|| PullError::BadRequest(format!("user `{reference}` not found")))?;
Ok(user.user_id)
}
/// (additions, deletions, files_changed)
pub(crate) async fn diff_stats_for(
&self,
space_name: &str,
repo_name: &str,
plan: &git::service::merge::MergePlanDto,
head: &str,
) -> Result<(i32, i32, i32), PullError> {
let from = plan
.merge_base
.clone()
.unwrap_or_else(|| plan.base_sha.clone());
let stat = self
.git
.diff_short_stat(
space_name.to_string(),
repo_name.to_string(),
from,
head.to_string(),
)
.await?;
Ok((
stat.additions as i32,
stat.deletions as i32,
stat.files_changed as i32,
))
}
pub async fn apply_merge_outcome(
&self,
pull: &pull_requests::Model,
caller: i64,
merged_sha: String,
squash_sha: Option<String>,
method: models::pull_request::pull_requests::MergeMethod,
) -> Result<pull_requests::Model, PullError> {
let mut model: pull_requests::ActiveModel = pull.clone().into();
model.has_merged = Set(true);
model.merger_id = Set(Some(caller));
model.merged_at = Set(Some(OffsetDateTime::now_utc()));
model.closed_at = Set(Some(OffsetDateTime::now_utc()));
model.merge_method = Set(method);
model.squash_commit_sha = Set(squash_sha);
model.merged_commit_sha = Set(Some(merged_sha.clone()));
model.auto_merge = Set(false);
model.updated_at = Set(Some(OffsetDateTime::now_utc()));
let updated = model.update(self.db_write()).await?;
self.record_event(&updated, Some(caller), EVENT_MERGED, Some(merged_sha), None)
.await?;
self.recount_open(updated.repo_id).await?;
Ok(updated)
}
}
impl PullService {
pub async fn set_lock(
&self,
pull: &pull_requests::Model,
caller: i64,
locked: bool,
) -> Result<pull_requests::Model, PullError> {
if pull.is_locked == locked {
return Ok(pull.clone());
}
let mut model: pull_requests::ActiveModel = pull.clone().into();
model.is_locked = Set(locked);
model.updated_at = Set(Some(OffsetDateTime::now_utc()));
let updated = model.update(self.db_write()).await?;
self.record_event(
&updated,
Some(caller),
if locked {
crate::services::events::EVENT_LOCKED
} else {
crate::services::events::EVENT_UNLOCKED
},
None,
None,
)
.await?;
Ok(updated)
}
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct LinkIssueCmd {
pub space_name: String,
pub repo_name: String,
/// PR 编号或 UUID。
pub pull_ref: String,
}