| 1 | //! What follows for coordination when a blocker resolves, and what an |
| 2 | //! agent picking up a work item sees of its blockers. What a work item is |
| 3 | //! blocked by, and changing it, is items/dependencies.rs. |
| 4 | //! |
| 5 | //! Each blocker that closes or has a pull request merged for it is a |
| 6 | //! `dependency.changed` event on the work item it blocks. |
| 7 | |
| 8 | use std::collections::HashSet; |
| 9 | |
| 10 | use g1t_contracts::work::*; |
| 11 | use g1t_contracts::User; |
| 12 | use serde::Deserialize; |
| 13 | use worker::Result; |
| 14 | |
| 15 | use super::{CLAIM_COLUMNS, ClaimRow}; |
| 16 | use crate::Work; |
| 17 | use crate::items::{self, Held}; |
| 18 | |
| 19 | #[derive(Deserialize)] |
| 20 | struct PullBriefRow { |
| 21 | number: u32, |
| 22 | title: String, |
| 23 | status: PullStatus, |
| 24 | item_id: String, |
| 25 | repo_id: String, |
| 26 | } |
| 27 | |
| 28 | impl Work { |
| 29 | /// A work item closed (`closed`), or landed in one of its targets by |
| 30 | /// `landing` (a pull request merged, or a commit pushed) and stays |
| 31 | /// open: its claims end if it closed, and each open work item it |
| 32 | /// blocks hears. A close clears it; a landing that leaves it open |
| 33 | /// clears nothing, and it is still among what they wait on. Never |
| 34 | /// fails what caused it: a problem is logged. |
| 35 | pub(crate) async fn item_resolved(&self, held: &Held, actor: &User, closed: bool, landing: Option<&items::Landing>) { |
| 36 | let number = held.item.number; |
| 37 | let merged_pull = landing.and_then(|landing| landing.pull); |
| 38 | let resolved = async { |
| 39 | if closed { |
| 40 | self.release_claims_on_close(held, Some(actor)).await?; |
| 41 | } |
| 42 | for dependent in items::dependents(self, held).await? { |
| 43 | let blocked_by = dependent.item.blocked_by.clone(); |
| 44 | let remaining = items::still_open(self, &dependent.item.workspace, &blocked_by, closed.then_some(number)).await?; |
| 45 | let change = DependencyChanged { |
| 46 | workspace: dependent.item.workspace.clone(), |
| 47 | number: dependent.item.number, |
| 48 | item: dependent.item.number, |
| 49 | repo_ids: dependent.repo_ids(), |
| 50 | blocked_by, |
| 51 | cleared: if closed { vec![number] } else { Vec::new() }, |
| 52 | remaining, |
| 53 | merged_pull, |
| 54 | ..DependencyChanged::default() |
| 55 | }; |
| 56 | // Follows from a close or landing already made: named by it, |
| 57 | // so a repeat of the same one tells each session once. |
| 58 | let key = resolution_key(&dependent.item.id, &held.item.id, closed, landing); |
| 59 | let data = g1t_contracts::events::marked(change, Some(actor)); |
| 60 | self.publish_keyed_to("dependency.changed", &dependent.repo_ids(), Some(actor.id.clone()), data, &key).await?; |
| 61 | } |
| 62 | Ok::<(), worker::Error>(()) |
| 63 | }; |
| 64 | if let Err(error) = resolved.await { |
| 65 | worker::console_error!("#{number} resolved, but its dependents were not told: {error}"); |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | /// The work items of `workspace` that `numbers` names and `viewer` may |
| 70 | /// see, each with its active claims and its pull requests in |
| 71 | /// repositories `viewer` may read, in the order given. |
| 72 | pub(crate) async fn blockers(&self, viewer: Option<&User>, workspace: &str, numbers: &[u32]) -> Result<Vec<Blocker>> { |
| 73 | let items = items::visible_numbered(self, workspace, numbers, viewer).await?; |
| 74 | if items.is_empty() { |
| 75 | return Ok(Vec::new()); |
| 76 | } |
| 77 | let ids = serde_json::to_string(&items.iter().map(|held| held.item.id.as_str()).collect::<Vec<_>>())?; |
| 78 | let bind = |sql: String| -> Result<worker::D1PreparedStatement> { self.db.prepare(sql).bind(&[ids.as_str().into()]) }; |
| 79 | let results = self |
| 80 | .db |
| 81 | .batch(vec![ |
| 82 | bind(format!( |
| 83 | "SELECT {CLAIM_COLUMNS} FROM claims |
| 84 | WHERE state = 'active' AND item_id IN (SELECT value FROM json_each(?1)) |
| 85 | ORDER BY created_at, id" |
| 86 | ))?, |
| 87 | bind( |
| 88 | "SELECT number, title, status, item_id, repo_id FROM pulls |
| 89 | WHERE item_id IN (SELECT value FROM json_each(?1)) |
| 90 | ORDER BY created_at" |
| 91 | .to_owned(), |
| 92 | )?, |
| 93 | ]) |
| 94 | .await?; |
| 95 | let mut results = results.into_iter(); |
| 96 | let mut next = || results.next(); |
| 97 | let claims: Vec<(String, Claim)> = match next() { |
| 98 | Some(result) => result.results::<ClaimRow>()?.into_iter().map(|row| (row.item_id.clone(), Claim::from(row))).collect(), |
| 99 | None => Vec::new(), |
| 100 | }; |
| 101 | let pulls: Vec<PullBriefRow> = match next() { |
| 102 | Some(result) => result.results()?, |
| 103 | None => Vec::new(), |
| 104 | }; |
| 105 | let mut repo_ids: Vec<String> = pulls.iter().map(|pull| pull.repo_id.clone()).collect(); |
| 106 | repo_ids.extend(claims.iter().filter_map(|(_, claim)| claim.repo_id.clone())); |
| 107 | repo_ids.sort(); |
| 108 | repo_ids.dedup(); |
| 109 | let readable = items::readable_repo_ids(self, viewer, &repo_ids).await?; |
| 110 | Ok(assemble_blockers(numbers, &items, &claims, &pulls, &readable)) |
| 111 | } |
| 112 | } |
| 113 | |
| 114 | /// What tells `dependent` once of `blocker` resolving: closing (`:closed`, |
| 115 | /// or `:merged:N` when a merge closed it), or landing somewhere while it |
| 116 | /// stays open (`:landed:` and what landed), which never stands in for the |
| 117 | /// close that comes later. |
| 118 | fn resolution_key(dependent: &str, blocker: &str, closed: bool, landing: Option<&items::Landing>) -> String { |
| 119 | let pull = landing.and_then(|landing| landing.pull); |
| 120 | let commit = landing.and_then(|landing| landing.commit.as_deref()); |
| 121 | let how = match (closed, pull, commit) { |
| 122 | (true, Some(pull), _) => format!("merged:{pull}"), |
| 123 | (true, None, _) => "closed".to_owned(), |
| 124 | (false, Some(pull), _) => format!("landed:pull:{pull}"), |
| 125 | (false, None, Some(commit)) => format!("landed:commit:{commit}"), |
| 126 | (false, None, None) => "landed".to_owned(), |
| 127 | }; |
| 128 | format!("{dependent}:cleared:{blocker}:{how}") |
| 129 | } |
| 130 | |
| 131 | /// The blockers `numbers` names among `items` (those the viewer may see), |
| 132 | /// in the order given, each with the claims and pull requests that are not |
| 133 | /// in a repository outside `readable`. |
| 134 | fn assemble_blockers( |
| 135 | numbers: &[u32], |
| 136 | items: &[Held], |
| 137 | claims: &[(String, Claim)], |
| 138 | pulls: &[PullBriefRow], |
| 139 | readable: &HashSet<String>, |
| 140 | ) -> Vec<Blocker> { |
| 141 | numbers |
| 142 | .iter() |
| 143 | .filter_map(|number| { |
| 144 | let held = items.iter().find(|held| held.item.number == *number)?; |
| 145 | Some(Blocker { |
| 146 | number: held.item.number, |
| 147 | title: held.item.title.clone(), |
| 148 | state: held.item.state, |
| 149 | claims: claims |
| 150 | .iter() |
| 151 | .filter(|(id, claim)| *id == held.item.id && claim.repo_id.as_ref().is_none_or(|repo| readable.contains(repo))) |
| 152 | .map(|(_, claim)| claim.clone()) |
| 153 | .collect(), |
| 154 | pulls: pulls |
| 155 | .iter() |
| 156 | .filter(|pull| pull.item_id == held.item.id && readable.contains(&pull.repo_id)) |
| 157 | .map(|pull| PullBrief { number: pull.number, title: pull.title.clone(), status: pull.status }) |
| 158 | .collect(), |
| 159 | }) |
| 160 | }) |
| 161 | .collect() |
| 162 | } |
| 163 | |
| 164 | #[cfg(test)] |
| 165 | mod tests { |
| 166 | use super::*; |
| 167 | use crate::items::rows_for_test::held_for_test; |
| 168 | |
| 169 | #[test] |
| 170 | fn a_landing_that_leaves_a_blocker_open_never_stands_in_for_its_close() { |
| 171 | let commit = items::Landing { pull: None, commit: Some("abc".into()), pushed_by: None }; |
| 172 | let merged = items::Landing { pull: Some(5), commit: None, pushed_by: None }; |
| 173 | let landed = resolution_key("wki_2", "wki_1", false, Some(&commit)); |
| 174 | let closed = resolution_key("wki_2", "wki_1", true, None); |
| 175 | assert_ne!(landed, closed); |
| 176 | assert_eq!(closed, "wki_2:cleared:wki_1:closed"); |
| 177 | assert_eq!(landed, "wki_2:cleared:wki_1:landed:commit:abc"); |
| 178 | assert_ne!(resolution_key("wki_2", "wki_1", false, Some(&merged)), resolution_key("wki_2", "wki_1", true, Some(&merged))); |
| 179 | } |
| 180 | |
| 181 | #[test] |
| 182 | fn blockers_show_only_what_the_viewer_may_read() { |
| 183 | // #1 is visible (asked for by the caller, who filtered), #2 is not |
| 184 | // among `items`: the viewer may not see it. |
| 185 | let items = vec![held_for_test(1, &[("rep_web", "origin", "acme", false)], &[])]; |
| 186 | let pull = |number: u32, item: &str, repo: &str| PullBriefRow { |
| 187 | number, |
| 188 | title: format!("Pull {number}"), |
| 189 | status: PullStatus::Open, |
| 190 | item_id: item.to_owned(), |
| 191 | repo_id: repo.to_owned(), |
| 192 | }; |
| 193 | let pulls = vec![pull(10, "wki_1", "rep_web"), pull(11, "wki_1", "rep_secret"), pull(12, "wki_2", "rep_web")]; |
| 194 | let readable: HashSet<String> = ["rep_web".to_owned()].into(); |
| 195 | let blockers = assemble_blockers(&[2, 1], &items, &[], &pulls, &readable); |
| 196 | assert_eq!(blockers.len(), 1); |
| 197 | assert_eq!(blockers[0].number, 1); |
| 198 | assert_eq!(blockers[0].pulls.iter().map(|pull| pull.number).collect::<Vec<_>>(), vec![10]); |
| 199 | } |
| 200 | } |