2687ace
use models::pull_request::{pull_request_event_line, pull_requests};
use models::repo::repos;
use sea_orm::{
ActiveModelTrait, ActiveValue::Set, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter,
};
use time::OffsetDateTime;
use crate::sync::error::SyncResult;
use crate::sync::event::PostReceiveTask;
use export::AppDatabase;
pub const EVENT_HEAD_UPDATED: &str = "head_updated";
pub const EVENT_HEAD_DELETED: &str = "head_deleted";
pub const EVENT_BASE_UPDATED: &str = "base_updated";
pub const ZERO_OID_LEN: usize = 40;
/// push 之后把分支最新提交同步到受影响的开放合并请求。
pub async fn refresh_pulls_after_sync(
db: &AppDatabase,
repo: &repos::Model,
task: &PostReceiveTask,
) -> SyncResult<u64> {
let mut touched = 0u64;
for update in &task.refs {
let Some(branch) = update.name.strip_prefix("refs/heads/") else {
continue;
};
let deleted = update.new_oid.chars().all(|c| c == '0');
if let Err(error) =
refresh_for_branch(db, repo, branch, &update.new_oid, deleted, task.actor_id).await
{
tracing::warn!(
error = %error,
repo_id = %repo.id,
branch,
"pull request sync skipped"
);
} else {
touched += 1;
}
}
if touched > 0 {
recount_open_pulls(db, repo.id).await?;
}
Ok(touched)
}
async fn refresh_for_branch(
db: &AppDatabase,
repo: &repos::Model,
branch: &str,
new_oid: &str,
deleted: bool,
actor: Option<uuid::Uuid>,
) -> SyncResult<()> {
let open = pull_requests::Entity::find()
.filter(pull_requests::Column::RepoId.eq(repo.id))
.filter(pull_requests::Column::HasMerged.eq(false))
.filter(pull_requests::Column::ClosedAt.is_null())
.filter(pull_requests::Column::HeadRepoId.is_null())
.all(db.read())
.await?;
for pull in open {
if pull.head_branch == branch {
if deleted {
record_event(db, &pull, actor, EVENT_HEAD_DELETED, None).await?;
continue;
}
if pull.head_sha != new_oid {
let mut model: pull_requests::ActiveModel = pull.clone().into();
model.head_sha = Set(new_oid.to_string());
model.updated_at = Set(Some(OffsetDateTime::now_utc()));
model.update(db.write()).await?;
record_event(db, &pull, actor, EVENT_HEAD_UPDATED, Some(new_oid)).await?;
}
} else if pull.base_branch == branch && !deleted && pull.base_sha != new_oid {
let mut model: pull_requests::ActiveModel = pull.clone().into();
model.base_sha = Set(new_oid.to_string());
model.update(db.write()).await?;
record_event(db, &pull, actor, EVENT_BASE_UPDATED, Some(new_oid)).await?;
}
}
Ok(())
}
async fn record_event(
db: &AppDatabase,
pull: &pull_requests::Model,
actor: Option<uuid::Uuid>,
event_type: &str,
commit_sha: Option<&str>,
) -> SyncResult<()> {
let row = pull_request_event_line::ActiveModel {
id: Set(uuid::Uuid::now_v7()),
pull_request_id: Set(pull.id),
user_id: Set(actor.unwrap_or(pull.author_id)),
event_type: Set(event_type.to_string()),
commit_sha: Set(commit_sha.map(|sha| sha.to_string())),
content: Set(None),
created_at: Set(OffsetDateTime::now_utc()),
};
row.insert(db.write()).await?;
Ok(())
}
/// 重新统计开放合并请求数量，写回 `repos.nums_pr`。
pub async fn recount_open_pulls(db: &AppDatabase, repo_id: uuid::Uuid) -> SyncResult<()> {
let open = pull_requests::Entity::find()
.filter(pull_requests::Column::RepoId.eq(repo_id))
.filter(pull_requests::Column::HasMerged.eq(false))
.filter(pull_requests::Column::ClosedAt.is_null())
.count(db.read())
.await?;
let model = repos::ActiveModel {
id: Set(repo_id),
nums_pr: Set(i32::try_from(open).unwrap_or(i32::MAX)),
// A partial update leaves every other column `NotSet`.
..Default::default()
};
model.update(db.write()).await?;
Ok(())
}