| 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 | |
| 7 | use futures_util::future::{try_join, try_join_all, try_join3}; |
| 8 | use g1t_contracts::events::{Event, ListArgs as ListEventsArgs}; |
| 9 | use g1t_contracts::repos::Repo; |
| 10 | use g1t_contracts::decisions::{Decision, DecisionKind, ListDecisionsArgs}; |
| 11 | use g1t_contracts::work::*; |
| 12 | use g1t_contracts::{FailureCode, Outcome}; |
| 13 | use g1t_kit::now_ms; |
| 14 | use serde::Deserialize; |
| 15 | use worker::Result; |
| 16 | |
| 17 | use super::{SESSION_COLUMNS, SessionRow, unwrap_or_fail}; |
| 18 | use crate::Work; |
| 19 | use crate::rows::{PULL_COLUMNS, PullRow}; |
| 20 | |
| 21 | /// How many sessions, decisions and events `get_work` gives at most. |
| 22 | const SESSIONS: u32 = 20; |
| 23 | const DECISIONS: u32 = 20; |
| 24 | const EVENTS: u32 = 20; |
| 25 | |
| 26 | #[derive(Deserialize)] |
| 27 | struct PlaceRow { |
| 28 | state: QueueState, |
| 29 | position: u32, |
| 30 | } |
| 31 | |
| 32 | impl 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 | 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 | /// up to [`DECISIONS`]. |
| 118 | async fn decisions_on(&self, repo: &g1t_contracts::repos::RepoPath, viewer: &g1t_contracts::Viewer, number: u32) -> Result<Vec<Decision>> { |
| 119 | let args = ListDecisionsArgs { repo: repo.clone(), viewer: viewer.clone(), number }; |
| 120 | let mut decisions = match self.list_decisions(args).await? { |
| 121 | Outcome::Ok(decisions) => decisions, |
| 122 | Outcome::Fail(_) => Vec::new(), |
| 123 | }; |
| 124 | decisions.reverse(); |
| 125 | decisions.truncate(DECISIONS as usize); |
| 126 | Ok(decisions) |
| 127 | } |
| 128 | |
| 129 | /// One of the issue's pull requests and where it stands: its required |
| 130 | /// checks, whether it merges, what it overlaps and its place in the |
| 131 | /// merge queue, for those still in progress. |
| 132 | async fn work_pull(&self, repo: &Repo, pull: &Pull) -> Result<WorkPull> { |
| 133 | let mut pull = pull.clone(); |
| 134 | crate::fill_base(&mut pull, repo); |
| 135 | let brief = |pull: &Pull| WorkPull { |
| 136 | number: pull.number, |
| 137 | title: pull.title.clone(), |
| 138 | status: pull.status, |
| 139 | branch: pull.branch.clone(), |
| 140 | head_commit: pull.head_commit.clone(), |
| 141 | author: pull.owner().into(), |
| 142 | check_status: pull.check_status, |
| 143 | required_checks: Vec::new(), |
| 144 | mergeable: Mergeable::Unknown, |
| 145 | overlaps: Vec::new(), |
| 146 | queue: None, |
| 147 | }; |
| 148 | if !pull.status.is_active() { |
| 149 | return Ok(brief(&pull)); |
| 150 | } |
| 151 | let required = async { |
| 152 | let settings = self.settings_on(repo, &pull).await?; |
| 153 | if settings.required_checks.is_empty() { |
| 154 | return Ok(Vec::new()); |
| 155 | } |
| 156 | let statuses = self.statuses(&repo.id, pull.head_commit.as_deref()).await?; |
| 157 | let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses).await?; |
| 158 | Ok::<_, worker::Error>(required_checks(&settings.required_checks, &checked)) |
| 159 | }; |
| 160 | let place_query = self |
| 161 | .db |
| 162 | .prepare( |
| 163 | "SELECT mine.state AS state, |
| 164 | (SELECT count(*) FROM queue_entries ahead |
| 165 | WHERE ahead.repo_id = mine.repo_id AND ahead.state IN ('waiting', 'testing', 'passed') |
| 166 | AND (ahead.created_at < mine.created_at |
| 167 | OR (ahead.created_at = mine.created_at AND ahead.id < mine.id))) + 1 AS position |
| 168 | FROM queue_entries mine |
| 169 | WHERE mine.pull_id = ? AND mine.state IN ('waiting', 'testing', 'passed') LIMIT 1", |
| 170 | ) |
| 171 | .bind(&[pull.id.as_str().into()])?; |
| 172 | let place = place_query.first::<PlaceRow>(None); |
| 173 | let ((required_checks, (mergeable, _)), (overlaps, place)) = |
| 174 | try_join(try_join(required, self.mergeability(&pull)), try_join(self.overlaps(&pull), place)).await?; |
| 175 | Ok(WorkPull { |
| 176 | required_checks, |
| 177 | mergeable, |
| 178 | overlaps, |
| 179 | queue: place.map(|place| QueuePlace { state: place.state, position: place.position }), |
| 180 | ..brief(&pull) |
| 181 | }) |
| 182 | } |
| 183 | } |