Skip to content
Sign up

flagon-io/g1t

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