Skip to content
Sign up

flagon-io/g1t

public
0
382 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 self.publish_as(
311 "checks.completed",
312 &a.repo_id,
313 None,
314 ChecksEvent {
315 pull_id: head.id,
316 repo_id: a.repo_id.clone(),
317 number: head.number,
318 status: if facts.failed.is_empty() { "passed" } else { "failed" },
319 commit: a.sha.clone(),
320 },
321 )
322 .await?;
323 }
324 Ok(Outcome::Ok(true))
325 }
326}
327
328#[cfg(test)]
329mod tests {
330 use super::*;
331
332 fn status(context: &str, state: &str) -> CommitStatus {
333 CommitStatus {
334 context: context.into(),
335 state: state.into(),
336 description: None,
337 target_url: None,
338 updated_at: String::new(),
339 source: None,
340 check_run_id: None,
341 }
342 }
343
344 fn names(list: &[&str]) -> Vec<String> {
345 list.iter().map(|name| (*name).to_owned()).collect()
346 }
347
348 #[test]
349 fn only_required_checks_hold_a_merge() {
350 let statuses = [status("CI / push", "pending"), status("Lint / pull_request", "failure"), status("Docs", "success")];
351 // Nothing required: nothing holds it, whatever failed.
352 let free = WorkflowFacts::of(&statuses, &[]);
353 assert_eq!(free.pending, ["CI / push"]);
354 assert_eq!(free.failed, ["Lint / pull_request"]);
355 assert!(free.required_failed().is_empty() && free.expected().is_empty());
356 let both = WorkflowFacts::of(&statuses, &names(&["CI", "Lint"]));
357 assert_eq!(both.required_failed(), ["Lint"]);
358 assert!(WorkflowFacts::of(&[status("Docs", "success")], &names(&["Docs"])).required_failed().is_empty());
359 }
360
361 #[test]
362 fn a_required_check_nothing_reported_is_expected() {
363 let facts = WorkflowFacts::of(&[status("CI / pull_request", "success")], &names(&["CI", "Deploy"]));
364 assert_eq!(facts.expected(), ["Deploy"]);
365 }
366
367 #[test]
368 fn seen_checks_are_named_once_with_their_events() {
369 let rows = vec![
370 ("CI / push".to_owned(), "2026-10-01T00:00:00Z".to_owned()),
371 ("CI / pull_request".to_owned(), "2026-10-03T00:00:00Z".to_owned()),
372 ("g1t / deploy".to_owned(), "2026-10-02T00:00:00Z".to_owned()),
373 ];
374 let seen = seen(rows);
375 assert_eq!(seen.len(), 2);
376 assert_eq!(seen[0].name, "CI");
377 assert_eq!(seen[0].events, ["pull_request", "push"]);
378 assert_eq!(seen[0].last_seen, "2026-10-03T00:00:00Z");
379 assert_eq!(seen[1].name, "g1t / deploy");
380 assert!(seen[1].events.is_empty());
381 }
382}