a06673b
//! Pull request feature use cases.
use std::collections::HashMap;
use db::conn::sea_orm::*;
use db::issue::issue as issue_entity;
use db::pull_request::pull_request_comment as comment_entity;
use db::pull_request::pull_request_event_line as event_entity;
use db::pull_request::pull_request_review as review_entity;
use db::pull_request::pull_requests as pr_entity;
use db::repo::repo_collaborations as collab_entity;
use db::repo::repo_collaborations::CollaborationMode;
use db::repo::repos as repo_entity;
use db::user::users as user_entity;
use gix::hash::ObjectId;
use gix::object::tree::diff::ChangeDetached;
use gix::refs::transaction::{Change, LogChange, PreviousValue, RefEdit, RefLog};
use gix::refs::{FullName, Target};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::error::{RepoError, RepoResult};
use crate::features::RepoService;
use crate::path::{git_bare_path, validate_branch_name};
use super::types::{
CreatePullRequestCommentOption, CreatePullRequestOption, ListPullRequestQuery,
MergePullRequestOption, MergeStyle, MergeView, PullRequestCommentView, PullRequestEventView,
PullRequestPage, PullRequestReviewView, PullRequestView, ReviewState as ReviewStateView,
SubmitPullRequestReviewOption, UpdatePullRequestOption,
};
fn require_authenticated(actor: Uuid) -> RepoResult<()> {
if actor.is_nil() {
return Err(RepoError::Unauthorized("authentication required".into()));
}
Ok(())
}
async fn load_repo(svc: &RepoService, repo_id: Uuid) -> RepoResult<repo_entity::Model> {
let db = svc.db.reader();
repo_entity::Entity::find()
.filter(repo_entity::Column::Id.eq(repo_id))
.one(&db)
.await?
.ok_or_else(|| RepoError::NotFound(format!("repository {repo_id}")))
}
async fn load_pr(svc: &RepoService, repo_id: Uuid, number: i64) -> RepoResult<pr_entity::Model> {
let db = svc.db.reader();
pr_entity::Entity::find()
.filter(pr_entity::Column::RepoId.eq(repo_id))
.filter(pr_entity::Column::AutoincrementId.eq(number as i32))
.one(&db)
.await?
.ok_or_else(|| RepoError::NotFound(format!("pull request #{number}")))
}
async fn load_user(svc: &RepoService, user_id: Uuid) -> RepoResult<Option<user_entity::Model>> {
let db = svc.db.reader();
user_entity::Entity::find()
.filter(user_entity::Column::Uid.eq(user_id))
.one(&db)
.await
.map_err(Into::into)
}
async fn collaborator_mode(
svc: &RepoService,
repo_id: Uuid,
actor: Uuid,
) -> RepoResult<Option<CollaborationMode>> {
let db = svc.db.reader();
let row = collab_entity::Entity::find()
.filter(collab_entity::Column::RepoId.eq(repo_id))
.filter(collab_entity::Column::UserId.eq(actor))
.one(&db)
.await?;
Ok(row.map(|r| r.mode))
}
fn can_manage_pr(pr: &pr_entity::Model, actor: Uuid, mode: Option<CollaborationMode>) -> bool {
pr.author_id == actor
|| matches!(
mode,
Some(CollaborationMode::Write | CollaborationMode::Admin)
)
}
fn open_git(svc: &RepoService, repo: &repo_entity::Model) -> RepoResult<gix::Repository> {
gix::open(git_bare_path(&svc.config, repo.channel_id, repo.id))
.map_err(|e| RepoError::Internal(format!("open repository: {e}")))
}
fn resolve_tree_id(repo: &gix::Repository, id: ObjectId) -> RepoResult<ObjectId> {
let commit = repo
.find_commit(id)
.map_err(|e| RepoError::Internal(format!("find commit {id}: {e}")))?;
commit
.tree_id()
.map(|t| t.detach())
.map_err(|e| RepoError::Internal(format!("tree of {id}: {e}")))
}
fn refs_resolve(repo: &gix::Repository, name: &str) -> RepoResult<ObjectId> {
let reference = repo
.find_reference(name)
.map_err(|_| RepoError::NotFound(format!("reference {name}")))?;
reference
.target()
.try_id()
.map(|o| o.to_owned())
.ok_or_else(|| RepoError::Internal(format!("non-object ref {name}")))
}
fn refs_update(
repo: &gix::Repository,
name: &str,
new: ObjectId,
expected: Option<ObjectId>,
) -> RepoResult<()> {
let previous = match expected {
Some(oid) => PreviousValue::MustExistAndMatch(Target::Object(oid)),
None => PreviousValue::Any,
};
let edit = RefEdit {
change: Change::Update {
log: LogChange::default(),
expected: previous,
new: Target::Object(new),
},
name: FullName::try_from(name)
.map_err(|e| RepoError::Internal(format!("invalid ref name {name}: {e}")))?,
deref: false,
};
repo.edit_reference(edit)
.map_err(|e| RepoError::Conflict(format!("reference update failed: {e}")))?;
Ok(())
}
fn refs_delete(repo: &gix::Repository, name: &str) -> RepoResult<()> {
let edit = RefEdit {
change: Change::Delete {
expected: PreviousValue::MustExist,
log: RefLog::AndReference,
},
name: FullName::try_from(name)
.map_err(|e| RepoError::Internal(format!("invalid ref name {name}: {e}")))?,
deref: false,
};
repo.edit_reference(edit)
.map_err(|e| RepoError::Internal(format!("reference delete {name}: {e}")))?;
Ok(())
}
fn changes_to_oid_map(changes: &[ChangeDetached]) -> HashMap<String, Option<ObjectId>> {
let mut map = HashMap::new();
for change in changes {
match change {
ChangeDetached::Addition { location, id, .. } => {
map.insert(location.to_string(), Some(*id));
}
ChangeDetached::Deletion { location, .. } => {
map.insert(location.to_string(), None);
}
ChangeDetached::Modification { location, id, .. } => {
map.insert(location.to_string(), Some(*id));
}
ChangeDetached::Rewrite { location, id, .. } => {
map.insert(location.to_string(), Some(*id));
}
}
}
map
}
fn threeway_merge_status(
repo: &gix::Repository,
ours: ObjectId,
theirs: ObjectId,
ancestor: ObjectId,
) -> RepoResult<git::conflict::types::MergeStatus> {
let fast_forwardable = repo
.merge_base(ours, theirs)
.map(|mb| mb.detach() == ours)
.unwrap_or(false);
let ours_tree = resolve_tree_id(repo, ours)?;
let theirs_tree = resolve_tree_id(repo, theirs)?;
let ancestor_tree = resolve_tree_id(repo, ancestor)?;
if ours_tree == theirs_tree {
return Ok(git::conflict::types::MergeStatus {
mergeable: true,
conflicts: Vec::new(),
fast_forwardable,
});
}
let ancestor_tree = repo
.find_tree(ancestor_tree)
.map_err(|e| RepoError::Internal(format!("find tree: {e}")))?;
let ours_tree = repo
.find_tree(ours_tree)
.map_err(|e| RepoError::Internal(format!("find tree: {e}")))?;
let theirs_tree = repo
.find_tree(theirs_tree)
.map_err(|e| RepoError::Internal(format!("find tree: {e}")))?;
let ours_changes = repo
.diff_tree_to_tree(Some(&ancestor_tree), Some(&ours_tree), None)
.map_err(|e| RepoError::Internal(format!("diff: {e}")))?;
let theirs_changes = repo
.diff_tree_to_tree(Some(&ancestor_tree), Some(&theirs_tree), None)
.map_err(|e| RepoError::Internal(format!("diff: {e}")))?;
let ours_map = changes_to_oid_map(&ours_changes);
let theirs_map = changes_to_oid_map(&theirs_changes);
let mut conflicts = Vec::new();
for (path, their_oid) in theirs_map.iter() {
if let Some(our_oid) = ours_map.get(path)
&& our_oid != their_oid
{
conflicts.push(git::conflict::types::ConflictFile {
path: path.clone(),
our_oid: *our_oid,
their_oid: *their_oid,
base_oid: None,
});
}
}
Ok(git::conflict::types::MergeStatus {
mergeable: conflicts.is_empty(),
conflicts,
fast_forwardable,
})
}
fn write_commit(
repo: &gix::Repository,
parents: Vec<ObjectId>,
tree_id: ObjectId,
message: &str,
signature: &gix::actor::Signature,
) -> RepoResult<ObjectId> {
let commit = gix::objs::Commit {
tree: tree_id,
parents: parents.into_iter().collect(),
author: signature.clone(),
committer: signature.clone(),
encoding: None,
message: message.as_bytes().into(),
extra_headers: Vec::new(),
};
repo.write_object(&commit)
.map(|id| id.detach())
.map_err(|e| RepoError::Internal(format!("write commit: {e}")))
}
fn merger_signature(user: Option<&user_entity::Model>) -> gix::actor::Signature {
let (name, ident) = match user {
Some(u) => (u.display_name.clone(), u.username.clone()),
None => ("Pull Request Merger".into(), "noreply".into()),
};
gix::actor::Signature {
name: name.into(),
email: format!("{ident}@noreply.local").into(),
time: gix::date::Time {
seconds: OffsetDateTime::now_utc().unix_timestamp(),
offset: 0,
},
}
}
fn merge_message(pr: &pr_entity::Model, opt: &MergePullRequestOption, squash: bool) -> String {
let title = opt.commit_title.clone().unwrap_or_else(|| {
if squash {
format!("{} (#{})", pr.title, pr.autoincrement_id)
} else {
format!(
"Merge pull request #{} from {}",
pr.autoincrement_id, pr.head_branch
)
}
});
match &opt.commit_message {
Some(body) if !body.is_empty() => format!("{title}\n\n{body}"),
_ => title,
}
}
fn review_state_to_db(state: ReviewStateView) -> review_entity::ReviewState {
match state {
ReviewStateView::Approved => review_entity::ReviewState::Approved,
ReviewStateView::ChangesRequested => review_entity::ReviewState::ChangesRequested,
ReviewStateView::Commented => review_entity::ReviewState::Commented,
}
}
async fn insert_event(
txn: &db::conn::sea_orm::DatabaseTransaction,
pr_id: Uuid,
user_id: Uuid,
event_type: &str,
commit_sha: Option<String>,
content: Option<String>,
) -> RepoResult<()> {
let active: event_entity::ActiveModel = event_entity::ActiveModel {
id: Set(Uuid::new_v4()),
pull_request_id: Set(pr_id),
user_id: Set(user_id),
event_type: Set(event_type.to_string()),
commit_sha: Set(commit_sha),
content: Set(content),
created_at: Set(OffsetDateTime::now_utc()),
};
active.insert(txn).await?;
Ok(())
}
pub async fn list_pull_requests(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
query: ListPullRequestQuery,
) -> RepoResult<PullRequestPage> {
let _ = actor;
let limit = query.limit.unwrap_or(svc.config.default_pagination);
let offset = query.page.unwrap_or(0).saturating_mul(limit);
let limit = std::cmp::min(limit, svc.config.max_pagination);
let db = svc.db.reader();
let mut base = pr_entity::Entity::find().filter(pr_entity::Column::RepoId.eq(repo_id));
if let Some(state) = query.state.as_deref() {
match state {
"open" => {
base = base
.filter(pr_entity::Column::HasMerged.eq(false))
.filter(pr_entity::Column::ClosedAt.is_null());
}
"closed" => {
base = base
.filter(pr_entity::Column::HasMerged.eq(false))
.filter(pr_entity::Column::ClosedAt.is_not_null());
}
"merged" => {
base = base.filter(pr_entity::Column::HasMerged.eq(true));
}
_ => {}
}
}
if let Some(keyword) = query.keyword.as_deref()
&& !keyword.is_empty()
{
base = base.filter(pr_entity::Column::Title.contains(keyword));
}
let total = base.clone().count(&db).await?;
let items: Vec<PullRequestView> = base
.clone()
.order_by_desc(pr_entity::Column::AutoincrementId)
.limit(Some(limit))
.offset(Some(offset))
.all(&db)
.await?
.into_iter()
.map(PullRequestView::from)
.collect();
Ok(PullRequestPage {
items,
total: total as i64,
})
}
pub async fn get_pull_request(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
) -> RepoResult<PullRequestView> {
let _ = actor;
let pr = load_pr(svc, repo_id, number).await?;
let pr = heal_merged(svc, repo_id, &pr).await;
Ok(PullRequestView::from(pr))
}
/// Heal the half-state where the base reference advanced to the merged commit
/// but the database transaction failed before recording `has_merged`.
async fn heal_merged(svc: &RepoService, repo_id: Uuid, pr: &pr_entity::Model) -> pr_entity::Model {
if pr.has_merged {
return pr.clone();
}
let Some(merged_sha) = pr.merged_commit_sha.clone() else {
return pr.clone();
};
let Ok(repo) = load_repo(svc, repo_id).await else {
return pr.clone();
};
let Ok(git) = open_git(svc, &repo) else {
return pr.clone();
};
let base_ref = format!("refs/heads/{}", pr.base_branch);
let Ok(base_oid) = refs_resolve(&git, &base_ref) else {
return pr.clone();
};
let Ok(target) = ObjectId::from_hex(merged_sha.as_bytes()) else {
return pr.clone();
};
if base_oid != target {
return pr.clone();
}
let db = svc.db.writer();
let now = OffsetDateTime::now_utc();
let mut active: pr_entity::ActiveModel = pr.clone().into();
active.has_merged = Set(true);
active.merged_at = Set(Some(now));
active.closed_at = Set(Some(now));
active.updated_at = Set(Some(now));
match active.update(&db).await {
Ok(model) => model,
Err(_) => pr.clone(),
}
}
pub async fn create_pull_request(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
input: CreatePullRequestOption,
) -> RepoResult<PullRequestView> {
require_authenticated(actor)?;
let title = input.title.trim().to_string();
if title.is_empty() || title.len() > svc.config.max_pr_title_len {
return Err(RepoError::BadRequest(format!(
"title length must be in 1..={}",
svc.config.max_pr_title_len
)));
}
let body = input.body.unwrap_or_default();
if body.len() > svc.config.max_pr_body_len {
return Err(RepoError::BadRequest(format!(
"body too long (max {})",
svc.config.max_pr_body_len
)));
}
let head = validate_branch_name(&input.head, svc.config.max_branch_name_len)?;
let base = validate_branch_name(&input.base, svc.config.max_branch_name_len)?;
if head == base {
return Err(RepoError::BadRequest(
"head and base branches must differ".into(),
));
}
let repo = load_repo(svc, repo_id).await?;
let git = open_git(svc, &repo)?;
let head_sha = refs_resolve(&git, &format!("refs/heads/{head}"))?;
let base_sha = refs_resolve(&git, &format!("refs/heads/{base}"))?;
let db = svc.db.writer();
let txn = db.begin().await?;
let max_id: Option<i32> = pr_entity::Entity::find()
.filter(pr_entity::Column::RepoId.eq(repo_id))
.lock_exclusive()
.select_only()
.column_as(pr_entity::Column::AutoincrementId.max(), "max")
.into_tuple::<Option<i32>>()
.one(&txn)
.await?
.flatten();
let next_id = max_id.map(|n| n + 1).unwrap_or(1);
let now = OffsetDateTime::now_utc();
let issue_id = Uuid::new_v4();
let issue_active: issue_entity::ActiveModel = issue_entity::ActiveModel {
id: Set(issue_id),
collection_id: Set(repo_id),
autoincrement_id: Set(next_id),
author_id: Set(actor),
title: Set(title.clone()),
description: Set(None),
body: Set(body.clone()),
milestone_id: Set(None),
created_at: Set(now),
updated_at: Set(None),
closed_at: Set(None),
};
issue_active.insert(&txn).await?;
let pr_id = Uuid::new_v4();
let pr_active: pr_entity::ActiveModel = pr_entity::ActiveModel {
id: Set(pr_id),
repo_id: Set(repo_id),
issue_id: Set(issue_id),
autoincrement_id: Set(next_id),
author_id: Set(actor),
title: Set(title),
body: Set(body),
head_branch: Set(head),
head_sha: Set(head_sha.to_string()),
base_branch: Set(base),
base_sha: Set(base_sha.to_string()),
has_merged: Set(false),
merged_commit_sha: Set(None),
additions: Set(0),
deletions: Set(0),
files_changed: Set(0),
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 saved = pr_active.insert(&txn).await?;
insert_event(
&txn,
pr_id,
actor,
"opened",
Some(head_sha.to_string()),
None,
)
.await?;
let nums_pr = repo.nums_pr;
let mut repo_active: repo_entity::ActiveModel = repo.into();
repo_active.nums_pr = Set(nums_pr + 1);
repo_active.update(&txn).await?;
txn.commit().await?;
Ok(PullRequestView::from(saved))
}
pub async fn update_pull_request(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
input: UpdatePullRequestOption,
) -> RepoResult<PullRequestView> {
require_authenticated(actor)?;
let pr = load_pr(svc, repo_id, number).await?;
let mode = collaborator_mode(svc, repo_id, actor).await?;
if !can_manage_pr(&pr, actor, mode) {
return Err(RepoError::Forbidden(
"only the author or a collaborator can update this pull request".into(),
));
}
if pr.has_merged || pr.closed_at.is_some() {
return Err(RepoError::Conflict(
"cannot update a closed or merged pull request".into(),
));
}
let db = svc.db.writer();
let mut active: pr_entity::ActiveModel = pr.into();
if let Some(title) = input.title {
let title = title.trim().to_string();
if title.is_empty() || title.len() > svc.config.max_pr_title_len {
return Err(RepoError::BadRequest(format!(
"title length must be in 1..={}",
svc.config.max_pr_title_len
)));
}
active.title = Set(title);
}
if let Some(body) = input.body {
if body.len() > svc.config.max_pr_body_len {
return Err(RepoError::BadRequest(format!(
"body too long (max {})",
svc.config.max_pr_body_len
)));
}
active.body = Set(body);
}
if let Some(base) = input.base {
let base = validate_branch_name(&base, svc.config.max_branch_name_len)?;
active.base_branch = Set(base);
}
if let Some(head) = input.head {
let head = validate_branch_name(&head, svc.config.max_branch_name_len)?;
active.head_branch = Set(head);
}
active.updated_at = Set(Some(OffsetDateTime::now_utc()));
let saved = active.update(&db).await?;
Ok(PullRequestView::from(saved))
}
pub async fn close_pull_request(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
) -> RepoResult<PullRequestView> {
require_authenticated(actor)?;
let pr = load_pr(svc, repo_id, number).await?;
let mode = collaborator_mode(svc, repo_id, actor).await?;
if !can_manage_pr(&pr, actor, mode) {
return Err(RepoError::Forbidden(
"only the author or a collaborator can close this pull request".into(),
));
}
if pr.has_merged || pr.closed_at.is_some() {
return Err(RepoError::Conflict(
"pull request already closed or merged".into(),
));
}
let db = svc.db.writer();
let txn = db.begin().await?;
let now = OffsetDateTime::now_utc();
let mut active: pr_entity::ActiveModel = pr.clone().into();
active.closed_at = Set(Some(now));
active.updated_at = Set(Some(now));
let saved = active.update(&txn).await?;
if let Some(issue) = issue_entity::Entity::find_by_id(pr.issue_id)
.one(&txn)
.await?
{
let mut ia: issue_entity::ActiveModel = issue.into();
ia.closed_at = Set(Some(now));
ia.updated_at = Set(Some(now));
ia.update(&txn).await?;
}
insert_event(&txn, pr.id, actor, "closed", None, None).await?;
txn.commit().await?;
Ok(PullRequestView::from(saved))
}
pub async fn reopen_pull_request(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
) -> RepoResult<PullRequestView> {
require_authenticated(actor)?;
let pr = load_pr(svc, repo_id, number).await?;
let mode = collaborator_mode(svc, repo_id, actor).await?;
if !can_manage_pr(&pr, actor, mode) {
return Err(RepoError::Forbidden(
"only the author or a collaborator can reopen this pull request".into(),
));
}
if pr.has_merged || pr.closed_at.is_none() {
return Err(RepoError::Conflict("pull request is not closed".into()));
}
let db = svc.db.writer();
let txn = db.begin().await?;
let now = OffsetDateTime::now_utc();
let mut active: pr_entity::ActiveModel = pr.clone().into();
active.closed_at = Set(None);
active.updated_at = Set(Some(now));
let saved = active.update(&txn).await?;
if let Some(issue) = issue_entity::Entity::find_by_id(pr.issue_id)
.one(&txn)
.await?
{
let mut ia: issue_entity::ActiveModel = issue.into();
ia.closed_at = Set(None);
ia.updated_at = Set(Some(now));
ia.update(&txn).await?;
}
insert_event(&txn, pr.id, actor, "reopened", None, None).await?;
txn.commit().await?;
Ok(PullRequestView::from(saved))
}
pub async fn lock_pull_request(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
) -> RepoResult<PullRequestView> {
require_authenticated(actor)?;
let pr = load_pr(svc, repo_id, number).await?;
let mode = collaborator_mode(svc, repo_id, actor).await?;
if !can_manage_pr(&pr, actor, mode) {
return Err(RepoError::Forbidden(
"only the author or a collaborator can lock this pull request".into(),
));
}
let db = svc.db.writer();
let mut active: pr_entity::ActiveModel = pr.into();
active.is_locked = Set(true);
active.updated_at = Set(Some(OffsetDateTime::now_utc()));
let saved = active.update(&db).await?;
Ok(PullRequestView::from(saved))
}
pub async fn unlock_pull_request(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
) -> RepoResult<PullRequestView> {
require_authenticated(actor)?;
let pr = load_pr(svc, repo_id, number).await?;
let mode = collaborator_mode(svc, repo_id, actor).await?;
if !can_manage_pr(&pr, actor, mode) {
return Err(RepoError::Forbidden(
"only the author or a collaborator can unlock this pull request".into(),
));
}
let db = svc.db.writer();
let mut active: pr_entity::ActiveModel = pr.into();
active.is_locked = Set(false);
active.updated_at = Set(Some(OffsetDateTime::now_utc()));
let saved = active.update(&db).await?;
Ok(PullRequestView::from(saved))
}
pub async fn create_pull_request_comment(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
input: CreatePullRequestCommentOption,
) -> RepoResult<PullRequestCommentView> {
require_authenticated(actor)?;
let pr = load_pr(svc, repo_id, number).await?;
let body = input.body.trim().to_string();
if body.is_empty() {
return Err(RepoError::BadRequest("comment body is empty".into()));
}
if body.len() > svc.config.max_pr_body_len {
return Err(RepoError::BadRequest(format!(
"comment too long (max {})",
svc.config.max_pr_body_len
)));
}
let db = svc.db.writer();
let txn = db.begin().await?;
let now = OffsetDateTime::now_utc();
let active: comment_entity::ActiveModel = comment_entity::ActiveModel {
id: Set(Uuid::new_v4()),
pull_request_id: Set(pr.id),
user_id: Set(actor),
review_id: Set(None),
commit_sha: Set(input.commit_sha.unwrap_or_default()),
path: Set(input.path.unwrap_or_default()),
line: Set(input.line),
body: Set(body.clone()),
created_at: Set(now),
updated_at: Set(None),
};
let saved = active.insert(&txn).await?;
insert_event(
&txn,
pr.id,
actor,
"comment",
(!saved.commit_sha.is_empty()).then_some(saved.commit_sha.clone()),
Some(body),
)
.await?;
txn.commit().await?;
Ok(PullRequestCommentView::from(saved))
}
pub async fn list_pull_request_comments(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
) -> RepoResult<Vec<PullRequestCommentView>> {
let _ = actor;
let pr = load_pr(svc, repo_id, number).await?;
let db = svc.db.reader();
let items: Vec<PullRequestCommentView> = comment_entity::Entity::find()
.filter(comment_entity::Column::PullRequestId.eq(pr.id))
.order_by_asc(comment_entity::Column::CreatedAt)
.all(&db)
.await?
.into_iter()
.map(PullRequestCommentView::from)
.collect();
Ok(items)
}
pub async fn submit_pull_request_review(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
input: SubmitPullRequestReviewOption,
) -> RepoResult<PullRequestReviewView> {
require_authenticated(actor)?;
let pr = load_pr(svc, repo_id, number).await?;
let db = svc.db.writer();
let txn = db.begin().await?;
let now = OffsetDateTime::now_utc();
let active: review_entity::ActiveModel = review_entity::ActiveModel {
id: Set(Uuid::new_v4()),
pull_request_id: Set(pr.id),
user_id: Set(actor),
commit_sha: Set(input.commit_sha.unwrap_or(pr.head_sha)),
official: Set(true),
content: Set(input.content.clone()),
state: Set(review_state_to_db(input.state)),
created_at: Set(now),
updated_at: Set(None),
};
let saved = active.insert(&txn).await?;
insert_event(
&txn,
pr.id,
actor,
"review",
(!saved.commit_sha.is_empty()).then_some(saved.commit_sha.clone()),
input.content,
)
.await?;
txn.commit().await?;
Ok(PullRequestReviewView::from(saved))
}
pub async fn list_pull_request_reviews(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
) -> RepoResult<Vec<PullRequestReviewView>> {
let _ = actor;
let pr = load_pr(svc, repo_id, number).await?;
let db = svc.db.reader();
let items: Vec<PullRequestReviewView> = review_entity::Entity::find()
.filter(review_entity::Column::PullRequestId.eq(pr.id))
.order_by_asc(review_entity::Column::CreatedAt)
.all(&db)
.await?
.into_iter()
.map(PullRequestReviewView::from)
.collect();
Ok(items)
}
pub async fn list_pull_request_events(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
) -> RepoResult<Vec<PullRequestEventView>> {
let _ = actor;
let pr = load_pr(svc, repo_id, number).await?;
let db = svc.db.reader();
let items: Vec<PullRequestEventView> = event_entity::Entity::find()
.filter(event_entity::Column::PullRequestId.eq(pr.id))
.order_by_asc(event_entity::Column::CreatedAt)
.all(&db)
.await?
.into_iter()
.map(PullRequestEventView::from)
.collect();
Ok(items)
}
pub async fn merge_pull_request(
svc: &RepoService,
actor: Uuid,
repo_id: Uuid,
number: i64,
opt: MergePullRequestOption,
) -> RepoResult<MergeView> {
require_authenticated(actor)?;
let pr = load_pr(svc, repo_id, number).await?;
let mode = collaborator_mode(svc, repo_id, actor).await?;
if !can_manage_pr(&pr, actor, mode) {
return Err(RepoError::Forbidden(
"only the author or a collaborator can merge this pull request".into(),
));
}
if pr.has_merged {
return Err(RepoError::Conflict("pull request already merged".into()));
}
let repo = load_repo(svc, repo_id).await?;
let git = open_git(svc, &repo)?;
let base_ref = format!("refs/heads/{}", pr.base_branch);
let head_ref = format!("refs/heads/{}", pr.head_branch);
let live_base = refs_resolve(&git, &base_ref)?;
let head = refs_resolve(&git, &head_ref)?;
let status = threeway_merge_status(&git, live_base, head, parse_oid(&pr.base_sha)?)?;
if !status.mergeable && opt.style != MergeStyle::Merge && opt.style != MergeStyle::FastForward {
let detail = serde_json::to_string(&status.conflicts).unwrap_or_default();
return Err(RepoError::Conflict(format!(
"pull request has conflicts; cannot merge: {detail}"
)));
}
if opt.style == MergeStyle::Squash && !repo.allow_squash_merge {
return Err(RepoError::Unprocessable(
"squash merge is disabled for this repository".into(),
));
}
if opt.style == MergeStyle::Merge && !repo.allow_merge_commits {
return Err(RepoError::Unprocessable(
"merge commits are disabled for this repository".into(),
));
}
let merger = load_user(svc, actor).await?;
let signature = merger_signature(merger.as_ref());
let new_sha = match opt.style {
MergeStyle::FastForward => {
if !status.fast_forwardable {
return Err(RepoError::Conflict(
"pull request cannot be fast-forwarded; base branch has diverged".into(),
));
}
head
}
MergeStyle::Squash => {
let tree = resolve_tree_id(&git, head)?;
let message = merge_message(&pr, &opt, true);
write_commit(&git, vec![live_base], tree, &message, &signature)?
}
MergeStyle::Merge => {
let tree = resolve_tree_id(&git, head)?;
let message = merge_message(&pr, &opt, false);
write_commit(&git, vec![live_base, head], tree, &message, &signature)?
}
MergeStyle::Rebase | MergeStyle::RebaseMerge | MergeStyle::ManuallyMerged => {
return Err(RepoError::BadRequest(format!(
"merge style {:?} is not supported yet",
opt.style
)));
}
};
refs_update(&git, &base_ref, new_sha, Some(live_base))?;
let db = svc.db.writer();
let txn = db.begin().await?;
let now = OffsetDateTime::now_utc();
let locked: pr_entity::Model = pr_entity::Entity::find()
.filter(pr_entity::Column::Id.eq(pr.id))
.lock_exclusive()
.one(&txn)
.await?
.ok_or_else(|| RepoError::NotFound(format!("pull request #{number}")))?;
if locked.has_merged {
return Err(RepoError::Conflict("pull request already merged".into()));
}
let mut active: pr_entity::ActiveModel = locked.into();
active.has_merged = Set(true);
active.merged_commit_sha = Set(Some(new_sha.to_string()));
active.base_sha = Set(new_sha.to_string());
active.merged_at = Set(Some(now));
active.merger_id = Set(Some(actor));
active.closed_at = Set(Some(now));
active.updated_at = Set(Some(now));
active.update(&txn).await?;
if let Some(issue) = issue_entity::Entity::find_by_id(pr.issue_id)
.one(&txn)
.await?
{
let mut ia: issue_entity::ActiveModel = issue.into();
ia.closed_at = Set(Some(now));
ia.updated_at = Set(Some(now));
ia.update(&txn).await?;
}
insert_event(
&txn,
pr.id,
actor,
"merged",
Some(new_sha.to_string()),
Some(serde_json::json!({ "style": format!("{:?}", opt.style), "mergeable": status.mergeable }).to_string()),
)
.await?;
txn.commit().await?;
if opt.delete_branch_after_merge {
let _ = refs_delete(&git, &head_ref);
}
Ok(MergeView {
sha: new_sha.to_string(),
merged: true,
message: None,
})
}
fn parse_oid(hex: &str) -> RepoResult<ObjectId> {
ObjectId::from_hex(hex.as_bytes())
.map_err(|e| RepoError::Internal(format!("invalid object id {hex}: {e}")))
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::Write;
use std::path::Path;
use std::process::{Command, Stdio};
struct TempRepo {
path: std::path::PathBuf,
}
impl TempRepo {
fn new() -> Self {
let path = std::env::temp_dir().join(format!(
"pr-repo-test-{}-{}",
std::process::id(),
Uuid::new_v4()
));
git(&path, &["init", "-q", "--bare", "-b", "main"]);
Self { path }
}
fn path(&self) -> &Path {
&self.path
}
fn write_blob(&self, content: &str) -> ObjectId {
let mut child = Command::new("git")
.args(["hash-object", "-w", "--stdin"])
.env("GIT_DIR", self.path())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.spawn()
.expect("spawn git hash-object");
child
.stdin
.take()
.expect("hash-object stdin")
.write_all(content.as_bytes())
.expect("write blob stdin");
let out = child.wait_with_output().expect("hash-object output");
assert!(out.status.success(), "hash-object failed: {:?}", out);
parse_oid(String::from_utf8_lossy(&out.stdout).trim()).expect("blob oid")
}
fn write_tree(&self, entries: &[(&str, ObjectId)]) -> ObjectId {
let mut input = String::new();
for (path, oid) in entries {
input.push_str(&format!("100644 blob {oid}\t{path}\n"));
}
let mut child = Command::new("git")
.args(["mktree"])
.env("GIT_DIR", self.path())
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.spawn()
.expect("spawn git mktree");
child
.stdin
.take()
.expect("mktree stdin")
.write_all(input.as_bytes())
.expect("write tree stdin");
let out = child.wait_with_output().expect("mktree output");
assert!(out.status.success(), "mktree failed: {:?}", out);
parse_oid(String::from_utf8_lossy(&out.stdout).trim()).expect("tree oid")
}
fn commit(&self, tree: ObjectId, parents: &[ObjectId], message: &str) -> ObjectId {
let mut args = vec!["commit-tree".to_string(), tree.to_string()];
for p in parents {
args.push("-p".to_string());
args.push(p.to_string());
}
args.push("-m".to_string());
args.push(message.to_string());
let out = git(
self.path(),
&args.iter().map(AsRef::as_ref).collect::<Vec<_>>(),
);
parse_oid(out.trim()).expect("commit oid")
}
}
impl Drop for TempRepo {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.path);
}
}
fn git(path: &Path, args: &[&str]) -> String {
let out = Command::new("git")
.args(args)
.env("GIT_DIR", path)
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.expect("git command");
assert!(
out.status.success(),
"git {args:?} failed: {}",
String::from_utf8_lossy(&out.stderr)
);
String::from_utf8_lossy(&out.stdout).into_owned()
}
fn open(repo: &TempRepo) -> gix::Repository {
gix::open(repo.path()).expect("open bare repo")
}
fn pr_model(number: i32) -> pr_entity::Model {
pr_entity::Model {
id: Uuid::new_v4(),
repo_id: Uuid::new_v4(),
issue_id: Uuid::new_v4(),
autoincrement_id: number,
author_id: Uuid::new_v4(),
title: "Add feature".to_string(),
body: String::new(),
head_branch: "feature".to_string(),
head_sha: String::new(),
base_branch: "main".to_string(),
base_sha: String::new(),
has_merged: false,
merged_commit_sha: None,
additions: 0,
deletions: 0,
files_changed: 0,
merger_id: None,
merged_at: None,
is_locked: false,
created_at: OffsetDateTime::now_utc(),
updated_at: None,
closed_at: None,
}
}
#[test]
fn threeway_merge_status_fast_forwardable() {
let repo = TempRepo::new();
let blob = repo.write_blob("v1\n");
let ancestor = repo.commit(repo.write_tree(&[("a.txt", blob)]), &[], "base");
let extra = repo.write_blob("v2\n");
let head_tree = repo.write_tree(&[("a.txt", blob), ("b.txt", extra)]);
let head = repo.commit(head_tree, &[ancestor], "feature work");
let status =
threeway_merge_status(&open(&repo), ancestor, head, ancestor).expect("threeway status");
assert!(status.mergeable, "fast-forward should be mergeable");
assert!(status.fast_forwardable, "should be fast-forwardable");
assert!(status.conflicts.is_empty(), "no conflicts expected");
}
#[test]
fn threeway_merge_status_conflict() {
let repo = TempRepo::new();
let blob = repo.write_blob("v1\n");
let ancestor = repo.commit(repo.write_tree(&[("a.txt", blob)]), &[], "base");
let ours_blob = repo.write_blob("ours\n");
let ours = repo.commit(
repo.write_tree(&[("a.txt", ours_blob)]),
&[ancestor],
"ours",
);
let theirs_blob = repo.write_blob("theirs\n");
let theirs = repo.commit(
repo.write_tree(&[("a.txt", theirs_blob)]),
&[ancestor],
"theirs",
);
let status =
threeway_merge_status(&open(&repo), ours, theirs, ancestor).expect("threeway status");
assert!(!status.mergeable, "conflicting branches should not merge");
assert!(!status.fast_forwardable, "conflicting branches are not ff");
assert_eq!(status.conflicts.len(), 1, "exactly one conflict expected");
assert_eq!(status.conflicts[0].path, "a.txt");
assert_eq!(status.conflicts[0].our_oid, Some(ours_blob));
assert_eq!(status.conflicts[0].their_oid, Some(theirs_blob));
}
#[test]
fn write_commit_roundtrip() {
let repo = TempRepo::new();
let blob = repo.write_blob("hello\n");
let tree = repo.write_tree(&[("hello.txt", blob)]);
let sig = gix::actor::Signature {
name: "Tester".into(),
email: "tester@example.com".into(),
time: gix::date::Time {
seconds: 1_700_000_000,
offset: 0,
},
};
let id = write_commit(&open(&repo), vec![], tree, "initial", &sig).expect("write commit");
let gix_repo = open(&repo);
let found = gix_repo.find_commit(id).expect("find written commit");
let commit = found.decode().expect("decode commit");
let tree_oid = ObjectId::from_hex(commit.tree.as_ref()).expect("tree hex");
assert_eq!(tree_oid, tree);
assert!(commit.parents.is_empty());
assert_eq!(commit.message, b"initial".as_slice());
}
#[test]
fn merge_message_formats() {
let pr = pr_model(42);
let opt = MergePullRequestOption {
style: MergeStyle::Squash,
delete_branch_after_merge: false,
commit_title: None,
commit_message: None,
};
assert_eq!(merge_message(&pr, &opt, true), "Add feature (#42)");
assert_eq!(
merge_message(&pr, &opt, false),
"Merge pull request #42 from feature"
);
}
#[test]
fn merge_message_overrides() {
let pr = pr_model(7);
let opt = MergePullRequestOption {
style: MergeStyle::Merge,
delete_branch_after_merge: true,
commit_title: Some("Custom title".to_string()),
commit_message: Some("Details".to_string()),
};
assert_eq!(merge_message(&pr, &opt, false), "Custom title\n\nDetails");
}
#[test]
fn review_state_to_db_maps() {
assert!(matches!(
review_state_to_db(ReviewStateView::Approved),
review_entity::ReviewState::Approved
));
assert!(matches!(
review_state_to_db(ReviewStateView::ChangesRequested),
review_entity::ReviewState::ChangesRequested
));
assert!(matches!(
review_state_to_db(ReviewStateView::Commented),
review_entity::ReviewState::Commented
));
}
#[test]
fn parse_oid_accepts_and_rejects() {
let hex = "a".repeat(40);
assert!(parse_oid(&hex).is_ok());
assert!(parse_oid("not-an-oid").is_err());
assert!(parse_oid("").is_err());
}
}