| 1 | //! `get_work`: everything about a work item an agent needs to pick it up, |
| 2 | //! in one call. What it is for and what done means, the repositories it is |
| 3 | //! linked to with their pull requests for it and what landed where, what |
| 4 | //! is still missing, where else it may belong (suggested targets), who is |
| 5 | //! on it, what it waits on and what waits on it, the decisions made about |
| 6 | //! it, and its latest coordination events. |
| 7 | |
| 8 | use std::collections::HashMap; |
| 9 | |
| 10 | use futures_util::future::{try_join, try_join_all, try_join3}; |
| 11 | use g1t_contracts::access::Capability; |
| 12 | use g1t_contracts::decisions::{Decision, DecisionKind, ListDecisionsArgs}; |
| 13 | use g1t_contracts::events::{Event, ListArgs as ListEventsArgs}; |
| 14 | use g1t_contracts::repos::{ReadableArgs, Repo}; |
| 15 | use g1t_contracts::work::*; |
| 16 | use g1t_contracts::{Outcome, Viewer}; |
| 17 | use g1t_kit::now_ms; |
| 18 | use serde::Deserialize; |
| 19 | use worker::Result; |
| 20 | |
| 21 | use super::{SESSION_COLUMNS, SessionRow, unwrap_or_fail}; |
| 22 | use crate::Work; |
| 23 | use crate::items::Held; |
| 24 | use crate::rows::{PULL_COLUMNS, PullRow}; |
| 25 | |
| 26 | /// How many sessions, decisions, events and suggestions `get_work` gives |
| 27 | /// at most. |
| 28 | const SESSIONS: u32 = 50; |
| 29 | const DECISIONS: u32 = 20; |
| 30 | const EVENTS: u32 = 20; |
| 31 | const SUGGESTIONS: usize = 5; |
| 32 | /// How many earlier landings suggestions are drawn from. |
| 33 | const LANDINGS: u32 = 200; |
| 34 | |
| 35 | #[derive(Deserialize)] |
| 36 | struct PlaceRow { |
| 37 | state: QueueState, |
| 38 | position: u32, |
| 39 | } |
| 40 | |
| 41 | /// A repository another work item of the workspace landed in. |
| 42 | #[derive(Clone, Debug, Deserialize)] |
| 43 | pub(crate) struct LandingRow { |
| 44 | pub repo_id: String, |
| 45 | pub title: String, |
| 46 | /// JSON array of label names. |
| 47 | pub labels: String, |
| 48 | } |
| 49 | |
| 50 | /// A repository of the workspace, and how many files sessions on the work |
| 51 | /// item touched in it. |
| 52 | #[derive(Clone, Debug, Deserialize)] |
| 53 | pub(crate) struct TouchedRow { |
| 54 | pub repo_id: String, |
| 55 | pub files: u32, |
| 56 | } |
| 57 | |
| 58 | /// The repositories of workspace `?1` with what earlier work items (not |
| 59 | /// `?2`) landed in them, newest first. |
| 60 | const LANDED_BEFORE: &str = "SELECT r.repo_id AS repo_id, i.title AS title, i.labels AS labels |
| 61 | FROM work_item_repos r JOIN work_items i ON i.id = r.item_id |
| 62 | WHERE i.workspace = ?1 AND i.id <> ?2 AND r.landed_at IS NOT NULL |
| 63 | ORDER BY r.landed_at DESC LIMIT ?3"; |
| 64 | |
| 65 | /// The repositories where sessions on work item `?1` touched files. |
| 66 | const TOUCHED_FOR: &str = "SELECT s.repo_id AS repo_id, count(*) AS files |
| 67 | FROM session_touches t JOIN sessions s ON s.id = t.session_id |
| 68 | WHERE t.kind = 'file' |
| 69 | AND (s.item_id = ?1 OR s.id IN (SELECT session_id FROM session_links WHERE kind = 'item' AND item_id = ?1)) |
| 70 | GROUP BY s.repo_id"; |
| 71 | |
| 72 | /// The words of `text` that say something: lowercase, four letters or |
| 73 | /// more. |
| 74 | fn words(text: &str) -> Vec<String> { |
| 75 | let mut found: Vec<String> = Vec::new(); |
| 76 | for word in text.split(|c: char| !c.is_alphanumeric()).map(str::to_lowercase) { |
| 77 | if word.chars().count() >= 4 && !found.contains(&word) { |
| 78 | found.push(word); |
| 79 | } |
| 80 | } |
| 81 | found |
| 82 | } |
| 83 | |
| 84 | /// Where a work item may belong besides where it is meant to land already: |
| 85 | /// the workspace's repositories (`candidates`, by id and path), ranked by |
| 86 | /// where work items sharing its labels or title words landed before, by a |
| 87 | /// repository's name in its title or description, and by where sessions on |
| 88 | /// it touched files. Each says why; the likeliest come first. |
| 89 | pub(crate) fn suggest_targets( |
| 90 | item: &WorkItem, |
| 91 | candidates: &[(String, String)], |
| 92 | landed: &[LandingRow], |
| 93 | touched: &[TouchedRow], |
| 94 | ) -> Vec<SuggestedTarget> { |
| 95 | let title_words = words(&item.title); |
| 96 | let text = format!("{}\n{}", item.title, item.body).to_lowercase(); |
| 97 | let mut suggested: Vec<SuggestedTarget> = Vec::new(); |
| 98 | for (repo_id, path) in candidates { |
| 99 | if item.repos.iter().any(|repo| repo.repo_id == *repo_id && repo.role == LinkRole::Target) { |
| 100 | continue; |
| 101 | } |
| 102 | let mut score = 0; |
| 103 | let mut why: Vec<String> = Vec::new(); |
| 104 | let landings: Vec<&LandingRow> = landed.iter().filter(|landing| landing.repo_id == *repo_id).collect(); |
| 105 | let labelled = landings |
| 106 | .iter() |
| 107 | .filter(|landing| { |
| 108 | let labels: Vec<String> = serde_json::from_str(&landing.labels).unwrap_or_default(); |
| 109 | labels.iter().any(|label| item.labels.contains(label)) |
| 110 | }) |
| 111 | .count() as u32; |
| 112 | if labelled > 0 { |
| 113 | score += 3 * labelled; |
| 114 | why.push(format!("{labelled} work items with the same labels landed there")); |
| 115 | } |
| 116 | let alike = landings.iter().filter(|landing| words(&landing.title).iter().any(|word| title_words.contains(word))).count() as u32; |
| 117 | if alike > 0 { |
| 118 | score += 2 * alike; |
| 119 | why.push(format!("{alike} work items with like titles landed there")); |
| 120 | } |
| 121 | let name = path.rsplit('/').next().unwrap_or(path).to_lowercase(); |
| 122 | let names_it = text.split(|c: char| !(c.is_alphanumeric() || c == '-' || c == '_' || c == '.')).any(|word| word == name); |
| 123 | if names_it || text.contains(&path.to_lowercase()) { |
| 124 | score += 5; |
| 125 | why.push(format!("it names {name}")); |
| 126 | } |
| 127 | if let Some(found) = touched.iter().find(|found| found.repo_id == *repo_id) { |
| 128 | score += 4; |
| 129 | why.push(format!("sessions on it touched {} files there", found.files)); |
| 130 | } |
| 131 | if score > 0 { |
| 132 | let mut why = why.join("; "); |
| 133 | if let Some(first) = why.get_mut(0..1) { |
| 134 | first.make_ascii_uppercase(); |
| 135 | } |
| 136 | suggested.push(SuggestedTarget { repo_id: repo_id.clone(), path: path.clone(), why: format!("{why}."), score }); |
| 137 | } |
| 138 | } |
| 139 | suggested.sort_by(|a, b| b.score.cmp(&a.score).then_with(|| a.path.cmp(&b.path))); |
| 140 | suggested.truncate(SUGGESTIONS); |
| 141 | suggested |
| 142 | } |
| 143 | |
| 144 | impl Work { |
| 145 | pub(crate) async fn get_work(&self, a: WorkItemViewArgs) -> Result<Outcome<WorkContext>> { |
| 146 | let held = unwrap_or_fail!(self.item_for(&a.viewer, &a.workspace, a.number, Capability::Read).await?); |
| 147 | // What went quiet in its repositories is read as it is now. |
| 148 | for repo_id in held.repo_ids().iter().take(10) { |
| 149 | self.sweep(repo_id).await?; |
| 150 | } |
| 151 | let pulls: Vec<Pull> = self |
| 152 | .db |
| 153 | .prepare(format!("SELECT {PULL_COLUMNS} FROM pulls WHERE item_id = ? ORDER BY created_at")) |
| 154 | .bind(&[held.item.id.as_str().into()])? |
| 155 | .all() |
| 156 | .await? |
| 157 | .results::<PullRow>()? |
| 158 | .into_iter() |
| 159 | .map(Pull::from) |
| 160 | .collect(); |
| 161 | // The repositories it is linked to and has pull requests in, as far |
| 162 | // as the viewer may read them. |
| 163 | let mut ids = held.repo_ids(); |
| 164 | for pull in &pulls { |
| 165 | if !ids.contains(&pull.repo_id) { |
| 166 | ids.push(pull.repo_id.clone()); |
| 167 | } |
| 168 | } |
| 169 | let readable: Vec<Repo> = g1t_kit::call(&self.repos, "readable", &ReadableArgs { ids, viewer: a.viewer.clone() }).await?; |
| 170 | let pulls: Vec<Pull> = pulls.into_iter().filter(|pull| readable.iter().any(|repo| repo.id == pull.repo_id)).collect(); |
| 171 | let mut numbers = vec![held.item.number]; |
| 172 | numbers.extend(pulls.iter().map(|pull| pull.number)); |
| 173 | |
| 174 | let people = try_join3( |
| 175 | self.active_claims(&held.item.id), |
| 176 | self.work_sessions(&held, &pulls), |
| 177 | self.suggestions(&held), |
| 178 | ); |
| 179 | let dependencies = try_join(self.blockers(&held.item.workspace, &held.item.blocked_by), self.blocked_items(&held)); |
| 180 | let linked = held.repo_ids(); |
| 181 | let history = try_join( |
| 182 | self.coordination_events(&linked, &numbers, COORDINATION_EVENTS.to_vec(), EVENTS), |
| 183 | self.decisions_on(&held, &a.viewer), |
| 184 | ); |
| 185 | let repos = try_join_all(held.item.repos.iter().filter_map(|link| { |
| 186 | let repo = readable.iter().find(|repo| repo.id == link.repo_id)?; |
| 187 | let mine: Vec<&Pull> = pulls.iter().filter(|pull| pull.repo_id == repo.id).collect(); |
| 188 | Some(self.work_repo(link.clone(), repo, mine)) |
| 189 | })); |
| 190 | let (((claims, sessions, suggested_targets), (blocked_by, blocks)), ((recent, decisions), repos)) = |
| 191 | try_join(try_join(people, dependencies), try_join(history, repos)).await?; |
| 192 | Ok(Outcome::Ok(WorkContext { |
| 193 | definition_of_done: definition_of_done(&held.item.body), |
| 194 | missing: held.item.missing(), |
| 195 | item: held.item, |
| 196 | repos, |
| 197 | suggested_targets, |
| 198 | claims, |
| 199 | sessions, |
| 200 | blocked_by, |
| 201 | blocks, |
| 202 | last_integration_failure: decisions.iter().find(|decision| decision.kind == DecisionKind::IntegrationFailure).cloned(), |
| 203 | decisions, |
| 204 | recent, |
| 205 | })) |
| 206 | } |
| 207 | |
| 208 | /// One repository a work item is linked to: its required checks, and |
| 209 | /// its pull requests for the item. |
| 210 | async fn work_repo(&self, link: ItemRepo, repo: &Repo, pulls: Vec<&Pull>) -> Result<WorkRepo> { |
| 211 | let settings = self.default_branch_settings(repo).await?; |
| 212 | let pulls = try_join_all(pulls.into_iter().map(|pull| self.work_pull(repo, pull))).await?; |
| 213 | Ok(WorkRepo { link, required_checks: settings.required_checks, pulls }) |
| 214 | } |
| 215 | |
| 216 | /// Where else a work item may belong ([`suggest_targets`]). |
| 217 | async fn suggestions(&self, held: &Held) -> Result<Vec<SuggestedTarget>> { |
| 218 | #[derive(Deserialize)] |
| 219 | struct Candidate { |
| 220 | repo_id: String, |
| 221 | path: String, |
| 222 | } |
| 223 | let results = self |
| 224 | .db |
| 225 | .batch(vec![ |
| 226 | self.db |
| 227 | .prepare("SELECT repo_id, path FROM repo_workspaces WHERE workspace = ? ORDER BY path LIMIT 500") |
| 228 | .bind(&[held.item.workspace.as_str().into()])?, |
| 229 | self.db |
| 230 | .prepare(LANDED_BEFORE) |
| 231 | .bind(&[held.item.workspace.as_str().into(), held.item.id.as_str().into(), LANDINGS.into()])?, |
| 232 | self.db.prepare(TOUCHED_FOR).bind(&[held.item.id.as_str().into()])?, |
| 233 | ]) |
| 234 | .await?; |
| 235 | let mut results = results.into_iter(); |
| 236 | let candidates: Vec<(String, String)> = match results.next() { |
| 237 | Some(result) => result.results::<Candidate>()?.into_iter().map(|row| (row.repo_id, row.path)).collect(), |
| 238 | None => Vec::new(), |
| 239 | }; |
| 240 | let landed: Vec<LandingRow> = match results.next() { |
| 241 | Some(result) => result.results()?, |
| 242 | None => Vec::new(), |
| 243 | }; |
| 244 | let touched: Vec<TouchedRow> = match results.next() { |
| 245 | Some(result) => result.results()?, |
| 246 | None => Vec::new(), |
| 247 | }; |
| 248 | Ok(suggest_targets(&held.item, &candidates, &landed, &touched)) |
| 249 | } |
| 250 | |
| 251 | /// Sessions on a work item or its pull requests, in any of their |
| 252 | /// repositories, most recently heard from first. Subagents are listed |
| 253 | /// beside the sessions that started them, so sibling agents on one |
| 254 | /// work item see each other; so are sessions tied to it any other way |
| 255 | /// (a claim, a branch, a trailer, `#N`). |
| 256 | async fn work_sessions(&self, held: &Held, pulls: &[Pull]) -> Result<Vec<Session>> { |
| 257 | let now = now_ms(); |
| 258 | let mut pushed: Vec<String> = Vec::new(); |
| 259 | let mut by_repo: HashMap<&str, Vec<u32>> = HashMap::new(); |
| 260 | for pull in pulls { |
| 261 | by_repo.entry(pull.repo_id.as_str()).or_default().push(pull.number); |
| 262 | } |
| 263 | for (repo_id, numbers) in &by_repo { |
| 264 | pushed.extend(self.branch_sessions(repo_id, numbers).await?); |
| 265 | } |
| 266 | let pull_ids: Vec<&str> = pulls.iter().map(|pull| pull.id.as_str()).collect(); |
| 267 | let mut sessions: Vec<Session> = self |
| 268 | .db |
| 269 | .prepare(format!( |
| 270 | "SELECT {SESSION_COLUMNS} FROM sessions |
| 271 | WHERE item_id = ?1 |
| 272 | OR id IN (SELECT value FROM json_each(?3)) |
| 273 | OR EXISTS (SELECT 1 FROM pulls p WHERE p.id IN (SELECT value FROM json_each(?2)) |
| 274 | AND p.repo_id = sessions.repo_id AND p.number = sessions.pull) |
| 275 | OR id IN (SELECT session_id FROM session_links WHERE kind = 'item' AND item_id = ?1) |
| 276 | OR id IN (SELECT l.session_id FROM session_links l JOIN pulls p ON p.repo_id = l.repo_id AND p.number = l.number |
| 277 | WHERE l.kind = 'pull' AND p.id IN (SELECT value FROM json_each(?2))) |
| 278 | ORDER BY state = 'ended', last_seen_at DESC LIMIT ?4" |
| 279 | )) |
| 280 | .bind(&[ |
| 281 | held.item.id.as_str().into(), |
| 282 | serde_json::to_string(&pull_ids)?.into(), |
| 283 | serde_json::to_string(&pushed)?.into(), |
| 284 | SESSIONS.into(), |
| 285 | ])? |
| 286 | .all() |
| 287 | .await? |
| 288 | .results::<SessionRow>()? |
| 289 | .into_iter() |
| 290 | .map(|row| row.at(now)) |
| 291 | .collect(); |
| 292 | self.fill_subagents(&mut sessions).await?; |
| 293 | Ok(sessions) |
| 294 | } |
| 295 | |
| 296 | /// The latest events of `types` about `numbers` in `repos`, newest |
| 297 | /// first. |
| 298 | async fn coordination_events(&self, repos: &[String], numbers: &[u32], types: Vec<&str>, limit: u32) -> Result<Vec<Event>> { |
| 299 | if repos.is_empty() { |
| 300 | return Ok(Vec::new()); |
| 301 | } |
| 302 | g1t_kit::call( |
| 303 | &self.events, |
| 304 | "list", |
| 305 | &ListEventsArgs { |
| 306 | repo_ids: repos.to_vec(), |
| 307 | types: types.into_iter().map(str::to_owned).collect(), |
| 308 | numbers: numbers.to_vec(), |
| 309 | limit: Some(limit), |
| 310 | ..ListEventsArgs::default() |
| 311 | }, |
| 312 | ) |
| 313 | .await |
| 314 | } |
| 315 | |
| 316 | /// Decision records on a work item and its pull requests, newest |
| 317 | /// first, up to [`DECISIONS`]. |
| 318 | async fn decisions_on(&self, held: &Held, viewer: &Viewer) -> Result<Vec<Decision>> { |
| 319 | let args = ListDecisionsArgs { |
| 320 | repo: None, |
| 321 | workspace: held.item.workspace.clone(), |
| 322 | viewer: viewer.clone(), |
| 323 | number: held.item.number, |
| 324 | }; |
| 325 | let mut decisions = match self.list_decisions(args).await? { |
| 326 | Outcome::Ok(decisions) => decisions, |
| 327 | Outcome::Fail(_) => Vec::new(), |
| 328 | }; |
| 329 | decisions.reverse(); |
| 330 | decisions.truncate(DECISIONS as usize); |
| 331 | Ok(decisions) |
| 332 | } |
| 333 | |
| 334 | /// One of the work item's pull requests and where it stands: its |
| 335 | /// required checks, whether it merges, what it overlaps and its place |
| 336 | /// in the merge queue, for those still in progress. |
| 337 | async fn work_pull(&self, repo: &Repo, pull: &Pull) -> Result<WorkPull> { |
| 338 | let mut pull = pull.clone(); |
| 339 | crate::fill_base(&mut pull, repo); |
| 340 | let brief = |pull: &Pull| WorkPull { |
| 341 | repo: crate::items::path_of(repo), |
| 342 | number: pull.number, |
| 343 | title: pull.title.clone(), |
| 344 | status: pull.status, |
| 345 | branch: pull.branch.clone(), |
| 346 | head_commit: pull.head_commit.clone(), |
| 347 | author: pull.owner().into(), |
| 348 | check_status: pull.check_status, |
| 349 | required_checks: Vec::new(), |
| 350 | mergeable: Mergeable::Unknown, |
| 351 | overlaps: Vec::new(), |
| 352 | queue: None, |
| 353 | }; |
| 354 | if !pull.status.is_active() { |
| 355 | return Ok(brief(&pull)); |
| 356 | } |
| 357 | let required = async { |
| 358 | let settings = self.settings_on(repo, &pull).await?; |
| 359 | if settings.required_checks.is_empty() { |
| 360 | return Ok(Vec::new()); |
| 361 | } |
| 362 | let statuses = self.statuses(&repo.id, pull.head_commit.as_deref()).await?; |
| 363 | let checked = self.with_job_checks(&repo.id, pull.head_commit.as_deref(), statuses).await?; |
| 364 | Ok::<_, worker::Error>(required_checks(&settings.required_checks, &checked)) |
| 365 | }; |
| 366 | let place_query = self |
| 367 | .db |
| 368 | .prepare( |
| 369 | "SELECT mine.state AS state, |
| 370 | (SELECT count(*) FROM queue_entries ahead |
| 371 | WHERE ahead.repo_id = mine.repo_id AND ahead.state IN ('waiting', 'testing', 'passed') |
| 372 | AND (ahead.created_at < mine.created_at |
| 373 | OR (ahead.created_at = mine.created_at AND ahead.id < mine.id))) + 1 AS position |
| 374 | FROM queue_entries mine |
| 375 | WHERE mine.pull_id = ? AND mine.state IN ('waiting', 'testing', 'passed') LIMIT 1", |
| 376 | ) |
| 377 | .bind(&[pull.id.as_str().into()])?; |
| 378 | let place = place_query.first::<PlaceRow>(None); |
| 379 | let ((required_checks, (mergeable, _)), (overlaps, place)) = |
| 380 | try_join(try_join(required, self.mergeability(&pull)), try_join(self.overlaps(&pull), place)).await?; |
| 381 | Ok(WorkPull { |
| 382 | required_checks, |
| 383 | mergeable, |
| 384 | overlaps, |
| 385 | queue: place.map(|place| QueuePlace { state: place.state, position: place.position }), |
| 386 | ..brief(&pull) |
| 387 | }) |
| 388 | } |
| 389 | } |
| 390 | |
| 391 | #[cfg(test)] |
| 392 | mod tests { |
| 393 | use super::*; |
| 394 | |
| 395 | fn item(title: &str, body: &str, labels: &[&str], repos: Vec<ItemRepo>) -> WorkItem { |
| 396 | serde_json::from_value(serde_json::json!({ |
| 397 | "id": "wki_1", "workspace": "acme", "number": 7, "title": title, "body": body, |
| 398 | "labels": labels, "state": "open", "reason": null, "author": { "id": "usr_1", "username": "ana" }, |
| 399 | "createdAt": "", "updatedAt": "", "closedAt": null, "pullCount": 0, "commentCount": 0, |
| 400 | "repos": repos |
| 401 | })) |
| 402 | .unwrap() |
| 403 | } |
| 404 | |
| 405 | fn landing(repo: &str, title: &str, labels: &str) -> LandingRow { |
| 406 | LandingRow { repo_id: repo.into(), title: title.into(), labels: labels.into() } |
| 407 | } |
| 408 | |
| 409 | #[test] |
| 410 | fn suggestions_say_why_and_come_likeliest_first() { |
| 411 | let candidates = vec![ |
| 412 | ("rep_api".to_owned(), "acme/api".to_owned()), |
| 413 | ("rep_web".to_owned(), "acme/web".to_owned()), |
| 414 | ("rep_docs".to_owned(), "acme/docs".to_owned()), |
| 415 | ]; |
| 416 | let item = item("Login page shows a stale avatar", "The web client caches it.", &["bug"], Vec::new()); |
| 417 | let landed = vec![ |
| 418 | landing("rep_api", "Avatar upload fails", r#"["bug"]"#), |
| 419 | landing("rep_api", "Rate limits", r#"["enhancement"]"#), |
| 420 | landing("rep_docs", "Typo", "[]"), |
| 421 | ]; |
| 422 | let touched = vec![TouchedRow { repo_id: "rep_web".into(), files: 3 }]; |
| 423 | let suggested = suggest_targets(&item, &candidates, &landed, &touched); |
| 424 | let paths: Vec<&str> = suggested.iter().map(|s| s.path.as_str()).collect(); |
| 425 | assert_eq!(paths, ["acme/web", "acme/api"]); |
| 426 | assert_eq!(suggested[0].why, "It names web; sessions on it touched 3 files there."); |
| 427 | assert_eq!(suggested[1].why, "1 work items with the same labels landed there; 1 work items with like titles landed there."); |
| 428 | assert!(suggested[0].score > suggested[1].score); |
| 429 | } |
| 430 | |
| 431 | #[test] |
| 432 | fn a_target_already_linked_is_not_suggested() { |
| 433 | let candidates = vec![("rep_web".to_owned(), "acme/web".to_owned())]; |
| 434 | let target = ItemRepo { repo_id: "rep_web".into(), path: "acme/web".into(), role: LinkRole::Target, legacy_number: None, landed: false }; |
| 435 | let item = item("Fix web", "", &[], vec![target]); |
| 436 | assert!(suggest_targets(&item, &candidates, &[], &[]).is_empty()); |
| 437 | } |
| 438 | } |