feat(git): refresh open pull requests after push
After a sync task completes, map the pushed refs/heads/* updates onto open pull requests: move head_sha (head_updated), re-snapshot base_sha (base_updated), record head_deleted, and recount repos.nums_pr.
ZhenYi committed
Modified lib/git/src/sync/mod.rs lib/git/src/sync/mod.rs +1 −0 View lib/git/src/sync/mod.rs
@@ -7,6 +7,7 @@
77pub mod license;
88pub mod lock;
99pub mod post_receive;
10pub mod pulls;
1011pub mod queue;
1112pub mod stats;
1213#[allow(clippy::module_inception)]
Added lib/git/src/sync/pulls.rs lib/git/src/sync/pulls.rs +123 −0 View lib/git/src/sync/pulls.rs
@@ -0,0 +1,123 @@
1use models::pull_request::{pull_request_event_line, pull_requests};
2use models::repo::repos;
3use sea_orm::{
4 ActiveModelTrait, ActiveValue::Set, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter,
5};
6use time::OffsetDateTime;
7
8use crate::sync::error::SyncResult;
9use crate::sync::event::PostReceiveTask;
10use export::AppDatabase;
11
12pub const EVENT_HEAD_UPDATED: &str = "head_updated";
13pub const EVENT_HEAD_DELETED: &str = "head_deleted";
14pub const EVENT_BASE_UPDATED: &str = "base_updated";
15pub const ZERO_OID_LEN: usize = 40;
16
17/// push 之后把分支最新提交同步到受影响的开放合并请求。
18pub async fn refresh_pulls_after_sync(
19 db: &AppDatabase,
20 repo: &repos::Model,
21 task: &PostReceiveTask,
22) -> SyncResult<u64> {
23 let mut touched = 0u64;
24 for update in &task.refs {
25 let Some(branch) = update.name.strip_prefix("refs/heads/") else {
26 continue;
27 };
28 let deleted = update.new_oid.chars().all(|c| c == '0');
29 if let Err(error) =
30 refresh_for_branch(db, repo, branch, &update.new_oid, deleted, task.actor_id).await
31 {
32 tracing::warn!(
33 error = %error,
34 repo_id = %repo.id,
35 branch,
36 "pull request sync skipped"
37 );
38 } else {
39 touched += 1;
40 }
41 }
42 if touched > 0 {
43 recount_open_pulls(db, repo.id).await?;
44 }
45 Ok(touched)
46}
47
48async fn refresh_for_branch(
49 db: &AppDatabase,
50 repo: &repos::Model,
51 branch: &str,
52 new_oid: &str,
53 deleted: bool,
54 actor: Option<uuid::Uuid>,
55) -> SyncResult<()> {
56 let open = pull_requests::Entity::find()
57 .filter(pull_requests::Column::RepoId.eq(repo.id))
58 .filter(pull_requests::Column::HasMerged.eq(false))
59 .filter(pull_requests::Column::ClosedAt.is_null())
60 .filter(pull_requests::Column::HeadRepoId.is_null())
61 .all(db.read())
62 .await?;
63
64 for pull in open {
65 if pull.head_branch == branch {
66 if deleted {
67 record_event(db, &pull, actor, EVENT_HEAD_DELETED, None).await?;
68 continue;
69 }
70 if pull.head_sha != new_oid {
71 let mut model: pull_requests::ActiveModel = pull.clone().into();
72 model.head_sha = Set(new_oid.to_string());
73 model.updated_at = Set(Some(OffsetDateTime::now_utc()));
74 model.update(db.write()).await?;
75 record_event(db, &pull, actor, EVENT_HEAD_UPDATED, Some(new_oid)).await?;
76 }
77 } else if pull.base_branch == branch && !deleted && pull.base_sha != new_oid {
78 let mut model: pull_requests::ActiveModel = pull.clone().into();
79 model.base_sha = Set(new_oid.to_string());
80 model.update(db.write()).await?;
81 record_event(db, &pull, actor, EVENT_BASE_UPDATED, Some(new_oid)).await?;
82 }
83 }
84 Ok(())
85}
86
87async fn record_event(
88 db: &AppDatabase,
89 pull: &pull_requests::Model,
90 actor: Option<uuid::Uuid>,
91 event_type: &str,
92 commit_sha: Option<&str>,
93) -> SyncResult<()> {
94 let row = pull_request_event_line::ActiveModel {
95 id: Set(uuid::Uuid::now_v7()),
96 pull_request_id: Set(pull.id),
97 user_id: Set(actor.unwrap_or(pull.author_id)),
98 event_type: Set(event_type.to_string()),
99 commit_sha: Set(commit_sha.map(|sha| sha.to_string())),
100 content: Set(None),
101 created_at: Set(OffsetDateTime::now_utc()),
102 ..Default::default()
103 };
104 row.insert(db.write()).await?;
105 Ok(())
106}
107
108/// 重新统计开放合并请求数量，写回 `repos.nums_pr`。
109pub async fn recount_open_pulls(db: &AppDatabase, repo_id: uuid::Uuid) -> SyncResult<()> {
110 let open = pull_requests::Entity::find()
111 .filter(pull_requests::Column::RepoId.eq(repo_id))
112 .filter(pull_requests::Column::HasMerged.eq(false))
113 .filter(pull_requests::Column::ClosedAt.is_null())
114 .count(db.read())
115 .await?;
116 let model = repos::ActiveModel {
117 id: Set(repo_id),
118 nums_pr: Set(i32::try_from(open).unwrap_or(i32::MAX)),
119 ..Default::default()
120 };
121 model.update(db.write()).await?;
122 Ok(())
123}
Modified lib/git/src/sync/sync.rs lib/git/src/sync/sync.rs +5 −0 View lib/git/src/sync/sync.rs
@@ -73,6 +73,11 @@
7373 }
7474
7575 #[must_use]
76 pub fn repo_model(&self) -> &repos::Model {
77 &self.repo
78 }
79
80 #[must_use]
7681 pub fn repo_id(&self) -> Uuid {
7782 self.repo.id
7883 }
Modified lib/git/src/sync/worker.rs lib/git/src/sync/worker.rs +12 −0 View lib/git/src/sync/worker.rs
@@ -209,6 +209,18 @@
209209 elapsed_ms = started.elapsed().as_millis() as u64,
210210 "sync task completed"
211211 );
212 if let Err(e) =
213 crate::sync::pulls::refresh_pulls_after_sync(&db, rsync.repo_model(), &task)
214 .await
215 {
216 warn!(
217 consumer = id,
218 task_id = %task.id,
219 repo_id = %task.repo_id,
220 error = %e,
221 "pull request refresh failed"
222 );
223 }
212224 if let Err(e) = webhook.dispatch(&task).await {
213225 warn!(
214226 consumer = id,