Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
201 linesCodeBlameRaw
1//! `get_work`: everything about an issue an agent needs to pick it up, in
2//! one call. What it is for and what done means, who is on it, the pull
3//! requests already made for it and where each stands, what it waits on
4//! and what waits on it, the decisions made about it, and its latest
5//! coordination events.
6
7use futures_util::future::{try_join, try_join_all, try_join3};
8use g1t_contracts::events::{Event, ListArgs as ListEventsArgs};
9use g1t_contracts::repos::Repo;
10use g1t_contracts::work::*;
11use g1t_contracts::{FailureCode, Outcome};
12use g1t_kit::now_ms;
13use serde::Deserialize;
14use worker::Result;
15
16use super::{SESSION_COLUMNS, SessionRow, unwrap_or_fail};
17use crate::Work;
18use crate::rows::{PULL_COLUMNS, PullRow};
19
20/// How many sessions, decisions and events `get_work` gives at most.
21const SESSIONS: u32 = 20;
22const DECISIONS: u32 = 20;
23const EVENTS: u32 = 20;
24
25#[derive(Deserialize)]
26struct PlaceRow {
27 state: QueueState,
28 position: u32,
29}
30
31impl Work {
32 pub(crate) async fn get_work(&self, a: GetWorkArgs) -> Result<Outcome<WorkContext>> {
33 let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?);
34 let Some(issue) = self.issue(&repo.id, a.number).await? else {
35 return Ok(Outcome::fail(FailureCode::NotFound, "Issue not found."));
36 };
37 let pulls: Vec<Pull> = self
38 .db
39 .prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE issue_id = ? ORDER BY number"))
40 .bind(&[issue.id.as_str().into()])?
41 .all()
42 .await?
43 .results::<PullRow>()?
44 .into_iter()
45 .map(Pull::from)
46 .collect();
47 let pull_numbers: Vec<u32> = pulls.iter().map(|pull| pull.number).collect();
48 let mut numbers = vec![issue.number];
49 numbers.extend(&pull_numbers);
50
51 let people = try_join3(
52 self.active_claims(&repo.id, issue.number),
53 self.work_sessions(&repo.id, issue.number, &pull_numbers),
54 self.default_branch_settings(&repo),
55 );
56 let dependencies = try_join(self.blockers(&repo.id, &issue.blocked_by), self.blocked_issues(&repo.id, issue.number));
57 let history = try_join3(
58 self.coordination_events(&repo.id, &numbers, COORDINATION_EVENTS.to_vec(), EVENTS),
59 self.coordination_events(&repo.id, &numbers, vec!["integration.failed"], 1),
60 self.decisions_on(&repo.id, issue.number, &pull_numbers),
61 );
62 let work_pulls = try_join_all(pulls.iter().map(|pull| self.work_pull(&repo, pull)));
63 let (((claims, sessions, settings), (blocked_by, blocks)), ((recent, failures, decisions), pulls)) =
64 try_join(try_join(people, dependencies), try_join(history, work_pulls)).await?;
65 Ok(Outcome::Ok(WorkContext {
66 definition_of_done: definition_of_done(&issue.body),
67 required_checks: settings.required_checks,
68 issue,
69 claims,
70 sessions,
71 pulls,
72 blocked_by,
73 blocks,
74 decisions,
75 last_integration_failure: failures.into_iter().next(),
76 recent,
77 }))
78 }
79
80 /// Sessions on an issue or its pull requests, most recently heard from
81 /// first.
82 async fn work_sessions(&self, repo_id: &str, issue: u32, pulls: &[u32]) -> Result<Vec<Session>> {
83 let now = now_ms();
84 Ok(self
85 .db
86 .prepare(format!(
87 "SELECT {SESSION_COLUMNS} FROM sessions
88 WHERE repo_id = ? AND (issue = ? OR pull IN (SELECT value FROM json_each(?)))
89 ORDER BY last_seen_at DESC LIMIT ?"
90 ))
91 .bind(&[repo_id.into(), issue.into(), serde_json::to_string(pulls)?.into(), SESSIONS.into()])?
92 .all()
93 .await?
94 .results::<SessionRow>()?
95 .into_iter()
96 .map(|row| row.at(now))
97 .collect())
98 }
99
100 /// The latest events of `types` about `numbers`, newest first.
101 async fn coordination_events(&self, repo_id: &str, numbers: &[u32], types: Vec<&str>, limit: u32) -> Result<Vec<Event>> {
102 g1t_kit::call(
103 &self.events,
104 "list",
105 &ListEventsArgs {
106 repo_id: Some(repo_id.to_owned()),
107 types: types.into_iter().map(str::to_owned).collect(),
108 numbers: numbers.to_vec(),
109 limit: Some(limit),
110 ..ListEventsArgs::default()
111 },
112 )
113 .await
114 }
115
116 /// Decision records on an issue and its pull requests, newest first,
117 /// as stored: none until the repository has any. Each `evidence` is
118 /// given as JSON.
119 async fn decisions_on(&self, repo_id: &str, issue: u32, pulls: &[u32]) -> Result<Vec<serde_json::Value>> {
120 let statement = self
121 .db
122 .prepare(
123 "SELECT * FROM decisions
124 WHERE repo_id = ? AND (issue = ? OR pull IN (SELECT value FROM json_each(?)))
125 ORDER BY created_at DESC LIMIT ?",
126 )
127 .bind(&[repo_id.into(), issue.into(), serde_json::to_string(pulls)?.into(), DECISIONS.into()])?;
128 // Decision records are kept by the merge queue's side of
129 // coordination; until they are, there are none to give.
130 let rows: Vec<serde_json::Value> = match statement.all().await {
131 Ok(result) => result.results()?,
132 Err(_) => return Ok(Vec::new()),
133 };
134 Ok(rows
135 .into_iter()
136 .map(|mut row| {
137 if let Some(evidence) = row.get("evidence").and_then(serde_json::Value::as_str)
138 && let Ok(parsed) = serde_json::from_str::<serde_json::Value>(evidence)
139 {
140 row["evidence"] = parsed;
141 }
142 row
143 })
144 .collect())
145 }
146
147 /// One of the issue's pull requests and where it stands: its required
148 /// checks, whether it merges, what it overlaps and its place in the
149 /// merge queue, for those still in progress.
150 async fn work_pull(&self, repo: &Repo, pull: &Pull) -> Result<WorkPull> {
151 let mut pull = pull.clone();
152 crate::fill_base(&mut pull, repo);
153 let brief = |pull: &Pull| WorkPull {
154 number: pull.number,
155 title: pull.title.clone(),
156 status: pull.status,
157 branch: pull.branch.clone(),
158 head_commit: pull.head_commit.clone(),
159 author: pull.owner().into(),
160 check_status: pull.check_status,
161 required_checks: Vec::new(),
162 mergeable: Mergeable::Unknown,
163 overlaps: Vec::new(),
164 queue: None,
165 };
166 if !pull.status.is_active() {
167 return Ok(brief(&pull));
168 }
169 let required = async {
170 let settings = self.settings_on(repo, &pull).await?;
171 if settings.required_checks.is_empty() {
172 return Ok(Vec::new());
173 }
174 let statuses = self.statuses(&repo.id, pull.head_commit.as_deref()).await?;
175 let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses).await?;
176 Ok::<_, worker::Error>(required_checks(&settings.required_checks, &checked))
177 };
178 let place_query = self
179 .db
180 .prepare(
181 "SELECT mine.state AS state,
182 (SELECT count(*) FROM queue_entries ahead
183 WHERE ahead.repo_id = mine.repo_id AND ahead.state IN ('waiting', 'testing', 'passed')
184 AND (ahead.created_at < mine.created_at
185 OR (ahead.created_at = mine.created_at AND ahead.id < mine.id))) + 1 AS position
186 FROM queue_entries mine
187 WHERE mine.pull_id = ? AND mine.state IN ('waiting', 'testing', 'passed') LIMIT 1",
188 )
189 .bind(&[pull.id.as_str().into()])?;
190 let place = place_query.first::<PlaceRow>(None);
191 let ((required_checks, (mergeable, _)), (overlaps, place)) =
192 try_join(try_join(required, self.mergeability(&pull)), try_join(self.overlaps(&pull), place)).await?;
193 Ok(WorkPull {
194 required_checks,
195 mergeable,
196 overlaps,
197 queue: place.map(|place| QueuePlace { state: place.state, position: place.position }),
198 ..brief(&pull)
199 })
200 }
201}