Skip to content
Sign up

flagon-io/g1t

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