Skip to content
Sign up

flagon-io/g1t

public
0
192 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::decisions::{Decision, DecisionKind, ListDecisionsArgs};
11use g1t_contracts::work::*;
12use g1t_contracts::{FailureCode, Outcome};
13use g1t_kit::now_ms;
14use serde::Deserialize;
15use worker::Result;
16
17use super::{SESSION_COLUMNS, SessionRow, unwrap_or_fail};
18use crate::Work;
19use crate::rows::{PULL_COLUMNS, PullRow};
20
21/// How many sessions, decisions and events `get_work` gives at most.
22const SESSIONS: u32 = 50;
23const DECISIONS: u32 = 20;
24const EVENTS: u32 = 20;
25
26#[derive(Deserialize)]
27struct PlaceRow {
28 state: QueueState,
29 position: u32,
30}
31
32impl Work {
33 pub(crate) async fn get_work(&self, a: GetWorkArgs) -> Result<Outcome<WorkContext>> {
34 let repo = unwrap_or_fail!(self.coordination_repo(&a.repo, &a.viewer).await?);
35 let Some(issue) = self.issue(&repo.id, a.number).await? else {
36 return Ok(Outcome::fail(FailureCode::NotFound, "Issue not found."));
37 };
38 let pulls: Vec<Pull> = self
39 .db
40 .prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE issue_id = ? ORDER BY number"))
41 .bind(&[issue.id.as_str().into()])?
42 .all()
43 .await?
44 .results::<PullRow>()?
45 .into_iter()
46 .map(Pull::from)
47 .collect();
48 let pull_numbers: Vec<u32> = pulls.iter().map(|pull| pull.number).collect();
49 let mut numbers = vec![issue.number];
50 numbers.extend(&pull_numbers);
51
52 let people = try_join3(
53 self.active_claims(&repo.id, issue.number),
54 self.work_sessions(&repo.id, issue.number, &pull_numbers),
55 self.default_branch_settings(&repo),
56 );
57 let dependencies = try_join(self.blockers(&repo.id, &issue.blocked_by), self.blocked_issues(&repo.id, issue.number));
58 let history = try_join(
59 self.coordination_events(&repo.id, &numbers, COORDINATION_EVENTS.to_vec(), EVENTS),
60 self.decisions_on(&a.repo, &a.viewer, issue.number),
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, 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 last_integration_failure: decisions.iter().find(|decision| decision.kind == DecisionKind::IntegrationFailure).cloned(),
75 decisions,
76 recent,
77 }))
78 }
79
80 /// Sessions on an issue or its pull requests, most recently heard from
81 /// first.
82 /// Subagents are listed beside the sessions that started them, so
83 /// sibling agents on one issue see each other; so are sessions tied to
84 /// it any other way (a claim, a branch, a trailer, `#N`).
85 async fn work_sessions(&self, repo_id: &str, issue: u32, pulls: &[u32]) -> Result<Vec<Session>> {
86 let now = now_ms();
87 let pushed = self.branch_sessions(repo_id, pulls).await?;
88 let mut sessions: Vec<Session> = self
89 .db
90 .prepare(format!(
91 "SELECT {SESSION_COLUMNS} FROM sessions
92 WHERE repo_id = ?1 AND (issue = ?2 OR pull IN (SELECT value FROM json_each(?3))
93 OR id IN (SELECT value FROM json_each(?5))
94 OR id IN (SELECT session_id FROM session_links WHERE repo_id = ?1
95 AND ((kind = 'issue' AND number = ?2) OR (kind = 'pull' AND number IN (SELECT value FROM json_each(?3))))))
96 ORDER BY state = 'ended', last_seen_at DESC LIMIT ?4"
97 ))
98 .bind(&[repo_id.into(), issue.into(), serde_json::to_string(pulls)?.into(), SESSIONS.into(), serde_json::to_string(&pushed)?.into()])?
99 .all()
100 .await?
101 .results::<SessionRow>()?
102 .into_iter()
103 .map(|row| row.at(now))
104 .collect();
105 self.fill_subagents(&mut sessions).await?;
106 Ok(sessions)
107 }
108
109 /// The latest events of `types` about `numbers`, newest first.
110 async fn coordination_events(&self, repo_id: &str, numbers: &[u32], types: Vec<&str>, limit: u32) -> Result<Vec<Event>> {
111 g1t_kit::call(
112 &self.events,
113 "list",
114 &ListEventsArgs {
115 repo_id: Some(repo_id.to_owned()),
116 types: types.into_iter().map(str::to_owned).collect(),
117 numbers: numbers.to_vec(),
118 limit: Some(limit),
119 ..ListEventsArgs::default()
120 },
121 )
122 .await
123 }
124
125 /// Decision records on an issue and its pull requests, newest first,
126 /// up to [`DECISIONS`].
127 async fn decisions_on(&self, repo: &g1t_contracts::repos::RepoPath, viewer: &g1t_contracts::Viewer, number: u32) -> Result<Vec<Decision>> {
128 let args = ListDecisionsArgs { repo: repo.clone(), viewer: viewer.clone(), number };
129 let mut decisions = match self.list_decisions(args).await? {
130 Outcome::Ok(decisions) => decisions,
131 Outcome::Fail(_) => Vec::new(),
132 };
133 decisions.reverse();
134 decisions.truncate(DECISIONS as usize);
135 Ok(decisions)
136 }
137
138 /// One of the issue's pull requests and where it stands: its required
139 /// checks, whether it merges, what it overlaps and its place in the
140 /// merge queue, for those still in progress.
141 async fn work_pull(&self, repo: &Repo, pull: &Pull) -> Result<WorkPull> {
142 let mut pull = pull.clone();
143 crate::fill_base(&mut pull, repo);
144 let brief = |pull: &Pull| WorkPull {
145 number: pull.number,
146 title: pull.title.clone(),
147 status: pull.status,
148 branch: pull.branch.clone(),
149 head_commit: pull.head_commit.clone(),
150 author: pull.owner().into(),
151 check_status: pull.check_status,
152 required_checks: Vec::new(),
153 mergeable: Mergeable::Unknown,
154 overlaps: Vec::new(),
155 queue: None,
156 };
157 if !pull.status.is_active() {
158 return Ok(brief(&pull));
159 }
160 let required = async {
161 let settings = self.settings_on(repo, &pull).await?;
162 if settings.required_checks.is_empty() {
163 return Ok(Vec::new());
164 }
165 let statuses = self.statuses(&repo.id, pull.head_commit.as_deref()).await?;
166 let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses).await?;
167 Ok::<_, worker::Error>(required_checks(&settings.required_checks, &checked))
168 };
169 let place_query = self
170 .db
171 .prepare(
172 "SELECT mine.state AS state,
173 (SELECT count(*) FROM queue_entries ahead
174 WHERE ahead.repo_id = mine.repo_id AND ahead.state IN ('waiting', 'testing', 'passed')
175 AND (ahead.created_at < mine.created_at
176 OR (ahead.created_at = mine.created_at AND ahead.id < mine.id))) + 1 AS position
177 FROM queue_entries mine
178 WHERE mine.pull_id = ? AND mine.state IN ('waiting', 'testing', 'passed') LIMIT 1",
179 )
180 .bind(&[pull.id.as_str().into()])?;
181 let place = place_query.first::<PlaceRow>(None);
182 let ((required_checks, (mergeable, _)), (overlaps, place)) =
183 try_join(try_join(required, self.mergeability(&pull)), try_join(self.overlaps(&pull), place)).await?;
184 Ok(WorkPull {
185 required_checks,
186 mergeable,
187 overlaps,
188 queue: place.map(|place| QueuePlace { state: place.state, position: place.position }),
189 ..brief(&pull)
190 })
191 }
192}