Sign in

flagon-io/g1t

public

g1t/services/work/src/statuses.rs

174 lines6,276 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

GitHub Actions on g1t, part two: running workflows1//! Statuses on commits: what workflow runs say about a pull request's head.
2//! A pending status holds the pull request, a failed one sends its agent
3//! back (or, for anyone else's, refuses the merge), as acceptance checks do.
4
5use g1t_contracts::events::ChecksEvent;
6use g1t_contracts::time::rfc3339;
7use g1t_contracts::work::{CommitStatus, SetCommitStatusArgs};
8use g1t_contracts::{FailureCode, Outcome};
9use g1t_kit::now_ms;
10use serde::Deserialize;
11use worker::Result;
12
13use crate::Work;
14
15#[derive(Deserialize)]
16struct StatusRow {
17 context: String,
18 state: String,
19 description: Option<String>,
20 target_url: Option<String>,
21 updated_at: String,
22}
23
24impl From<StatusRow> for CommitStatus {
25 fn from(row: StatusRow) -> Self {
26 CommitStatus {
27 context: row.context,
28 state: row.state,
29 description: row.description,
30 target_url: row.target_url,
31 updated_at: row.updated_at,
32 }
33 }
34}
35
36#[derive(Deserialize)]
37struct HeadRow {
38 id: String,
39 number: u32,
40}
41
42/// The workflows still running and the ones that failed, by name.
43#[derive(Clone, Debug, Default, PartialEq, Eq)]
44pub(crate) struct WorkflowFacts {
45 pub(crate) pending: Vec<String>,
46 pub(crate) failed: Vec<String>,
47}
48
49impl WorkflowFacts {
50 pub(crate) fn of(statuses: &[CommitStatus]) -> WorkflowFacts {
51 WorkflowFacts {
52 pending: statuses.iter().filter(|s| s.state == "pending").map(|s| s.context.clone()).collect(),
53 failed: statuses.iter().filter(|s| s.state == "failure" || s.state == "error").map(|s| s.context.clone()).collect(),
54 }
55 }
56
57 /// Why a merge has to wait, if it does.
58 pub(crate) fn refusal(&self) -> Option<String> {
59 if !self.failed.is_empty() {
60 return Some(format!("{} failed.", list(&self.failed)));
61 }
62 if !self.pending.is_empty() {
63 return Some(format!("{} {} still running.", list(&self.pending), if self.pending.len() == 1 { "is" } else { "are" }));
64 }
65 None
66 }
67}
68
69pub(crate) fn list(names: &[String]) -> String {
70 match names {
71 [] => String::new(),
72 [one] => one.clone(),
73 [rest @ .., last] => format!("{} and {last}", rest.join(", ")),
74 }
75}
76
77impl Work {
78 pub(crate) async fn statuses(&self, repo_id: &str, sha: Option<&str>) -> Result<Vec<CommitStatus>> {
79 let Some(sha) = sha else { return Ok(Vec::new()) };
80 Ok(self
81 .db
82 .prepare("SELECT context, state, description, target_url, updated_at FROM commit_statuses WHERE repo_id = ? AND sha = ? ORDER BY context")
83 .bind(&[repo_id.into(), sha.into()])?
84 .all()
85 .await?
86 .results::<StatusRow>()?
87 .into_iter()
88 .map(CommitStatus::from)
89 .collect())
90 }
91
92 pub(crate) async fn set_commit_status(&self, a: SetCommitStatusArgs) -> Result<Outcome<bool>> {
93 if !matches!(a.state.as_str(), "pending" | "success" | "failure" | "error") {
94 return Ok(Outcome::fail(FailureCode::Invalid, "`state` is pending, success, failure or error."));
95 }
96 self.db
97 .prepare(
98 "INSERT INTO commit_statuses (repo_id, sha, context, state, description, target_url, updated_at)
99 VALUES (?, ?, ?, ?, ?, ?, ?)
100 ON CONFLICT (repo_id, sha, context) DO UPDATE SET
101 state = excluded.state, description = excluded.description,
102 target_url = excluded.target_url, updated_at = excluded.updated_at",
103 )
104 .bind(&[
105 a.repo_id.as_str().into(),
106 a.sha.as_str().into(),
107 a.context.as_str().into(),
108 a.state.as_str().into(),
109 a.description.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
110 a.target_url.as_deref().map_or(worker::wasm_bindgen::JsValue::NULL, Into::into),
111 rfc3339(now_ms()).into(),
112 ])?
113 .run()
114 .await?;
115 if a.state == "pending" {
116 return Ok(Outcome::Ok(true));
117 }
118 // Once every workflow on a pull request's head has finished, its
119 // lifecycle moves on, as it does when its checks finish.
120 let facts = WorkflowFacts::of(&self.statuses(&a.repo_id, Some(&a.sha)).await?);
121 if !facts.pending.is_empty() {
122 return Ok(Outcome::Ok(true));
123 }
124 let heads = self
125 .db
126 .prepare("SELECT id, number FROM pulls WHERE repo_id = ? AND head_commit = ? AND status IN ('draft', 'open')")
127 .bind(&[a.repo_id.as_str().into(), a.sha.as_str().into()])?
128 .all()
129 .await?
130 .results::<HeadRow>()?;
131 for head in heads {
132 self.publish_as(
133 "checks.completed",
134 &a.repo_id,
135 None,
136 ChecksEvent {
137 pull_id: head.id,
138 repo_id: a.repo_id.clone(),
139 number: head.number,
140 status: if facts.failed.is_empty() { "passed" } else { "failed" },
141 commit: a.sha.clone(),
142 },
143 )
144 .await?;
145 }
146 Ok(Outcome::Ok(true))
147 }
148}
149
150#[cfg(test)]
151mod tests {
152 use super::*;
153
154 fn status(context: &str, state: &str) -> CommitStatus {
155 CommitStatus {
156 context: context.into(),
157 state: state.into(),
158 description: None,
159 target_url: None,
160 updated_at: String::new(),
161 }
162 }
163
164 #[test]
165 fn failures_come_before_waiting() {
166 let facts = WorkflowFacts::of(&[status("CI / push", "pending"), status("Lint / pull_request", "failure"), status("Docs", "success")]);
167 assert_eq!(facts.pending, ["CI / push"]);
168 assert_eq!(facts.failed, ["Lint / pull_request"]);
169 assert_eq!(facts.refusal().unwrap(), "Lint / pull_request failed.");
170 let waiting = WorkflowFacts::of(&[status("A", "pending"), status("B", "pending")]);
171 assert_eq!(waiting.refusal().unwrap(), "A and B are still running.");
172 assert!(WorkflowFacts::of(&[status("A", "success")]).refusal().is_none());
173 }
174}