Skip to content
Sign up

flagon-io/g1t

public
0
387 linesCodeBlameRaw
1//! Statuses on commits: what workflow runs (and other tools, such as
2//! deployments) say about a pull request's head. These are its checks.
3//!
4//! The rules of the branch a pull request merges into name the checks that
5//! must pass (rulesets.rs; `RepoSettings::required_checks` as they stack):
6//! a required check that failed, is still running or has not reported
7//! refuses the merge, for everyone and for the merge queue (`g1t_rules`
8//! says so).
9
10use g1t_contracts::checks::StatusEvent;
11use g1t_contracts::events::ChecksEvent;
12use g1t_contracts::time::rfc3339;
13use g1t_contracts::work::{
14 CommitStatus, RequiredCheck, RequiredState, SeenCheck, SeenChecksArgs, SetCommitStatusArgs, check_name,
15 required_checks,
16};
17use g1t_contracts::{FailureCode, Outcome};
18use g1t_kit::now_ms;
19use serde::Deserialize;
20use worker::Result;
21
22use crate::Work;
23
24#[derive(Deserialize)]
25struct StatusRow {
26 context: String,
27 state: String,
28 description: Option<String>,
29 target_url: Option<String>,
30 updated_at: String,
31 #[serde(default)]
32 source: Option<String>,
33 #[serde(default)]
34 check_run_id: Option<String>,
35}
36
37impl From<StatusRow> for CommitStatus {
38 fn from(row: StatusRow) -> Self {
39 CommitStatus {
40 context: row.context,
41 state: row.state,
42 description: row.description,
43 target_url: row.target_url,
44 updated_at: row.updated_at,
45 source: row.source,
46 check_run_id: row.check_run_id,
47 }
48 }
49}
50
51#[derive(Deserialize)]
52struct HeadRow {
53 id: String,
54 number: u32,
55}
56
57/// What a commit's checks say: every status still running and every one
58/// that failed, by context, and where each required check stands.
59#[derive(Clone, Debug, Default, PartialEq)]
60pub(crate) struct WorkflowFacts {
61 pub(crate) pending: Vec<String>,
62 pub(crate) failed: Vec<String>,
63 pub(crate) required: Vec<RequiredCheck>,
64}
65
66impl WorkflowFacts {
67 /// `required` names the checks the default branch's protection requires.
68 pub(crate) fn of(statuses: &[CommitStatus], required: &[String]) -> WorkflowFacts {
69 WorkflowFacts {
70 pending: statuses.iter().filter(|s| s.state == "pending").map(|s| s.context.clone()).collect(),
71 failed: statuses.iter().filter(|s| s.state == "failure" || s.state == "error").map(|s| s.context.clone()).collect(),
72 required: required_checks(required, statuses),
73 }
74 }
75
76 fn required_in(&self, state: RequiredState) -> Vec<String> {
77 self.required.iter().filter(|check| check.state == state).map(|check| check.name.clone()).collect()
78 }
79
80 /// The required checks that failed, by name.
81 pub(crate) fn required_failed(&self) -> Vec<String> {
82 self.required_in(RequiredState::Failure)
83 }
84
85 /// The required checks nothing has reported on the commit yet.
86 pub(crate) fn expected(&self) -> Vec<String> {
87 self.required_in(RequiredState::Expected)
88 }
89}
90
91/// The check names in `(context, last reported)` rows, most recent first:
92/// each name once, with the events it was reported for.
93pub(crate) fn seen(rows: Vec<(String, String)>) -> Vec<SeenCheck> {
94 let mut rows = rows;
95 rows.sort_by(|a, b| b.1.cmp(&a.1));
96 let mut out: Vec<SeenCheck> = Vec::new();
97 for (context, at) in rows {
98 let (name, event) = check_name(&context);
99 match out.iter_mut().find(|seen| seen.name.eq_ignore_ascii_case(name)) {
100 Some(seen) => {
101 if let Some(event) = event
102 && !seen.events.iter().any(|known| known == event)
103 {
104 seen.events.push(event.to_owned());
105 }
106 }
107 None => out.push(SeenCheck {
108 name: name.to_owned(),
109 events: event.map(|event| vec![event.to_owned()]).unwrap_or_default(),
110 last_seen: at,
111 }),
112 }
113 }
114 out
115}
116
117/// How far back `seen_checks` looks, and the most contexts it reads.
118const SEEN_DAYS: u64 = 30;
119const SEEN_LIMIT: u32 = 200;
120
121#[derive(Deserialize)]
122struct SeenRow {
123 context: String,
124 at: String,
125}
126
127pub(crate) fn list(names: &[String]) -> String {
128 match names {
129 [] => String::new(),
130 [one] => one.clone(),
131 [rest @ .., last] => format!("{} and {last}", rest.join(", ")),
132 }
133}
134
135impl Work {
136 /// Where a commit's checks stand, against the repository's required ones.
137 pub(crate) async fn facts(&self, repo_id: &str, sha: Option<&str>) -> Result<WorkflowFacts> {
138 let (statuses, settings) =
139 futures_util::future::try_join(self.statuses(repo_id, sha), self.settings_by_id(repo_id)).await?;
140 self.workflow_facts(repo_id, sha, &statuses, &settings.required_checks).await
141 }
142
143 /// What `statuses`, a commit's, say, with each required check also met
144 /// by the g1t Actions jobs of its name ([`Work::with_job_checks`]). Jobs
145 /// count only toward required checks: whether workflows are running or
146 /// failed is their runs' statuses' to say.
147 pub(crate) async fn workflow_facts(
148 &self,
149 repo_id: &str,
150 sha: Option<&str>,
151 statuses: &[CommitStatus],
152 required: &[String],
153 ) -> Result<WorkflowFacts> {
154 let mut facts = WorkflowFacts::of(statuses, required);
155 if !required.is_empty() {
156 facts.required = required_checks(required, &self.with_job_checks(repo_id, sha, statuses.to_vec()).await?);
157 }
158 Ok(facts)
159 }
160
161 /// `statuses` and, for each g1t Actions job on `sha`, the status its
162 /// check run stands as (`g1t_contracts::checks::job_statuses`), so a
163 /// required check named for a job, as GitHub's are, is met by it.
164 pub(crate) async fn with_job_checks(&self, repo_id: &str, sha: Option<&str>, mut statuses: Vec<CommitStatus>) -> Result<Vec<CommitStatus>> {
165 let Some(sha) = sha else { return Ok(statuses) };
166 let details = self
167 .actions_runs(g1t_contracts::checks::ActionsChecksArgs { repo_id: repo_id.to_owned(), shas: vec![sha.to_owned()], ..Default::default() })
168 .await?;
169 let runs: Vec<_> = details.into_iter().map(|detail| (detail.run, detail.jobs)).collect();
170 statuses.extend(g1t_contracts::checks::job_statuses(&runs, sha));
171 Ok(statuses)
172 }
173
174 /// The check names reported on a repository's commits lately, for
175 /// choosing which to require.
176 pub(crate) async fn seen_checks(&self, a: SeenChecksArgs) -> Result<Outcome<Vec<SeenCheck>>> {
177 let repo = match self.repo(&a.repo, &a.viewer).await? {
178 Outcome::Ok(repo) => repo,
179 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
180 };
181 let since = rfc3339(now_ms().saturating_sub(SEEN_DAYS * 24 * 60 * 60 * 1000));
182 let rows = self
183 .db
184 .prepare(
185 "SELECT context, MAX(updated_at) AS at FROM commit_statuses
186 WHERE repo_id = ? AND updated_at >= ? GROUP BY context ORDER BY at DESC LIMIT ?",
187 )
188 .bind(&[repo.id.as_str().into(), since.into(), SEEN_LIMIT.into()])?
189 .all()
190 .await?
191 .results::<SeenRow>()?;
192 Ok(Outcome::Ok(seen(rows.into_iter().map(|row| (row.context, row.at)).collect())))
193 }
194
195 pub(crate) async fn statuses(&self, repo_id: &str, sha: Option<&str>) -> Result<Vec<CommitStatus>> {
196 let Some(sha) = sha else { return Ok(Vec::new()) };
197 if let Some(found) = self.prefetched_repo(repo_id).filter(|found| found.head.as_deref() == Some(sha)) {
198 return Ok(found
199 .rows::<StatusRow>(crate::prefetch::Slot::Statuses)?
200 .into_iter()
201 .map(CommitStatus::from)
202 .collect());
203 }
204 Ok(self
205 .db
206 .prepare("SELECT context, state, description, target_url, updated_at, source, check_run_id FROM commit_statuses WHERE repo_id = ? AND sha = ? ORDER BY context")
207 .bind(&[repo_id.into(), sha.into()])?
208 .all()
209 .await?
210 .results::<StatusRow>()?
211 .into_iter()
212 .map(CommitStatus::from)
213 .collect())
214 }
215
216 /// A status from another part of g1t: a workflow run, a deploy, a scan.
217 /// Published as `status.created`, as one set through the API is, so
218 /// webhooks hear it.
219 /// Publishing is best effort: the status is kept either way.
220 pub(crate) async fn set_commit_status(&self, a: SetCommitStatusArgs) -> Result<Outcome<bool>> {
221 let event = StatusEvent {
222 repo_id: a.repo_id.clone(),
223 sha: a.sha.clone(),
224 context: a.context.clone(),
225 state: a.state.clone(),
226 description: a.description.clone(),
227 target_url: a.target_url.clone(),
228 };
229 let repo_id = a.repo_id.clone();
230 let stored = self.store_status(a, None).await?;
231 if matches!(stored, Outcome::Ok(_))
232 && let Err(error) = self.publish_as("status.created", &repo_id, None, event).await
233 {
234 worker::console_error!("status.created not published: {error}");
235 }
236 Ok(stored)
237 }
238
239 /// Sets a status, standing for check run `check_run_id` if it does,
240 /// and moves on what was waiting for the commit's checks.
241 pub(crate) async fn store_status(&self, a: SetCommitStatusArgs, check_run_id: Option<&str>) -> Result<Outcome<bool>> {
242 if !matches!(a.state.as_str(), "pending" | "success" | "failure" | "error") {
243 return Ok(Outcome::fail(FailureCode::Invalid, "`state` is pending, success, failure or error."));
244 }
245 self.db
246 .prepare(
247 "INSERT INTO commit_statuses (repo_id, sha, context, state, description, target_url, updated_at, source, check_run_id)
248 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
249 ON CONFLICT (repo_id, sha, context) DO UPDATE SET
250 state = excluded.state, description = excluded.description,
251 target_url = excluded.target_url, updated_at = excluded.updated_at,
252 source = excluded.source, check_run_id = excluded.check_run_id",
253 )
254 .bind(&[
255 a.repo_id.as_str().into(),
256 a.sha.as_str().into(),
257 a.context.as_str().into(),
258 a.state.as_str().into(),
259 a.description.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
260 a.target_url.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
261 rfc3339(now_ms()).into(),
262 a.source.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
263 check_run_id.map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
264 ])?
265 .run()
266 .await?;
267 if a.state == "pending" {
268 return Ok(Outcome::Ok(true));
269 }
270 // Once every workflow on a pull request's head has finished, its
271 // lifecycle moves on, as it does when its checks finish.
272 let facts = self.facts(&a.repo_id, Some(&a.sha)).await?;
273 if !facts.pending.is_empty() {
274 return Ok(Outcome::Ok(true));
275 }
276 // A merge queue state waiting on its merge_group workflows.
277 self.merge_group_finished(&a.repo_id, &a.sha, &facts).await?;
278 let heads = self
279 .db
280 .prepare("SELECT id, number FROM pulls WHERE repo_id = ? AND head_commit = ? AND status IN ('draft', 'open')")
281 .bind(&[a.repo_id.as_str().into(), a.sha.as_str().into()])?
282 .all()
283 .await?
284 .results::<HeadRow>()?;
285 for head in heads {
286 // A pull request g1t stopped on picks back up once what stopped
287 // it passes: its workflows, and its checks if it has any. The
288 // lifecycle then decides again, within its usual limits.
289 if facts.failed.is_empty() {
290 let resumed = self
291 .db
292 .prepare(
293 "UPDATE pulls SET stalled = NULL WHERE id = ? AND managed = 1 AND stalled IS NOT NULL
294 AND (check_status IS NULL OR check_status = 'passed') RETURNING id AS value",
295 )
296 .bind(&[head.id.as_str().into()])?
297 .first::<crate::rows::ValueRow>(None)
298 .await?;
299 if resumed.is_some() {
300 self.announce_resumed(&head.id, None).await?;
301 self.note(
302 &a.repo_id,
303 head.number,
304 (crate::lifecycle::POLICY_ACTOR_ID, crate::lifecycle::POLICY_ACTOR_NAME),
305 "picked this back up: its workflows pass now",
306 )
307 .await?;
308 }
309 }
310 // A pull request the merge queue took out rejoins once a new
311 // commit's checks all pass (integration.rs).
312 if facts.failed.is_empty() && facts.required_failed().is_empty() && facts.expected().is_empty() {
313 self.requeue_if_fixed(&head.id, &a.sha).await?;
314 }
315 self.publish_as(
316 "checks.completed",
317 &a.repo_id,
318 None,
319 ChecksEvent {
320 pull_id: head.id,
321 repo_id: a.repo_id.clone(),
322 number: head.number,
323 status: if facts.failed.is_empty() { "passed" } else { "failed" },
324 commit: a.sha.clone(),
325 },
326 )
327 .await?;
328 }
329 Ok(Outcome::Ok(true))
330 }
331}
332
333#[cfg(test)]
334mod tests {
335 use super::*;
336
337 fn status(context: &str, state: &str) -> CommitStatus {
338 CommitStatus {
339 context: context.into(),
340 state: state.into(),
341 description: None,
342 target_url: None,
343 updated_at: String::new(),
344 source: None,
345 check_run_id: None,
346 }
347 }
348
349 fn names(list: &[&str]) -> Vec<String> {
350 list.iter().map(|name| (*name).to_owned()).collect()
351 }
352
353 #[test]
354 fn only_required_checks_hold_a_merge() {
355 let statuses = [status("CI / push", "pending"), status("Lint / pull_request", "failure"), status("Docs", "success")];
356 // Nothing required: nothing holds it, whatever failed.
357 let free = WorkflowFacts::of(&statuses, &[]);
358 assert_eq!(free.pending, ["CI / push"]);
359 assert_eq!(free.failed, ["Lint / pull_request"]);
360 assert!(free.required_failed().is_empty() && free.expected().is_empty());
361 let both = WorkflowFacts::of(&statuses, &names(&["CI", "Lint"]));
362 assert_eq!(both.required_failed(), ["Lint"]);
363 assert!(WorkflowFacts::of(&[status("Docs", "success")], &names(&["Docs"])).required_failed().is_empty());
364 }
365
366 #[test]
367 fn a_required_check_nothing_reported_is_expected() {
368 let facts = WorkflowFacts::of(&[status("CI / pull_request", "success")], &names(&["CI", "Deploy"]));
369 assert_eq!(facts.expected(), ["Deploy"]);
370 }
371
372 #[test]
373 fn seen_checks_are_named_once_with_their_events() {
374 let rows = vec![
375 ("CI / push".to_owned(), "2026-10-01T00:00:00Z".to_owned()),
376 ("CI / pull_request".to_owned(), "2026-10-03T00:00:00Z".to_owned()),
377 ("g1t / deploy".to_owned(), "2026-10-02T00:00:00Z".to_owned()),
378 ];
379 let seen = seen(rows);
380 assert_eq!(seen.len(), 2);
381 assert_eq!(seen[0].name, "CI");
382 assert_eq!(seen[0].events, ["pull_request", "push"]);
383 assert_eq!(seen[0].last_seen, "2026-10-03T00:00:00Z");
384 assert_eq!(seen[1].name, "g1t / deploy");
385 assert!(seen[1].events.is_empty());
386 }
387}