| 1 | //! What an issue is blocked by: the issues of the same repository that have |
| 2 | //! to land first (`issues.blocked_by`, a JSON array of numbers). |
| 3 | //! |
| 4 | //! `update_issue` replaces the set, refusing an issue that is not one of |
| 5 | //! the repository's, the issue itself, or a set that would make a cycle. |
| 6 | //! Each change, and each blocker that closes or has its pull request |
| 7 | //! merged, is a `dependency.changed` event on the issue it blocks. |
| 8 | |
| 9 | use std::collections::{HashMap, HashSet}; |
| 10 | |
| 11 | use g1t_contracts::work::*; |
| 12 | use g1t_contracts::{FailureCode, Outcome, User}; |
| 13 | use serde::Deserialize; |
| 14 | use worker::Result; |
| 15 | use worker::wasm_bindgen::JsValue; |
| 16 | |
| 17 | use super::{CLAIM_COLUMNS, ClaimRow}; |
| 18 | use crate::Work; |
| 19 | |
| 20 | #[derive(Deserialize)] |
| 21 | struct EdgeRow { |
| 22 | number: u32, |
| 23 | blocked_by: String, |
| 24 | } |
| 25 | |
| 26 | #[derive(Deserialize)] |
| 27 | struct BriefRow { |
| 28 | number: u32, |
| 29 | title: String, |
| 30 | state: State, |
| 31 | } |
| 32 | |
| 33 | #[derive(Deserialize)] |
| 34 | struct PullBriefRow { |
| 35 | number: u32, |
| 36 | title: String, |
| 37 | status: PullStatus, |
| 38 | issue_number: u32, |
| 39 | } |
| 40 | |
| 41 | fn numbers(json: &str) -> Vec<u32> { |
| 42 | serde_json::from_str(json).unwrap_or_default() |
| 43 | } |
| 44 | |
| 45 | impl Work { |
| 46 | /// `asked` as `number`'s new `blocked_by`, or why it cannot be. |
| 47 | pub(crate) async fn checked_blocked_by(&self, repo_id: &str, number: u32, asked: &[u32]) -> Result<Outcome<Vec<u32>>> { |
| 48 | let wanted = serde_json::to_string(asked)?; |
| 49 | let results = self |
| 50 | .db |
| 51 | .batch(vec![ |
| 52 | self.db |
| 53 | .prepare( |
| 54 | "SELECT number, title, state FROM issues |
| 55 | WHERE repo_id = ? AND number IN (SELECT value FROM json_each(?))", |
| 56 | ) |
| 57 | .bind(&[repo_id.into(), wanted.into()])?, |
| 58 | self.db |
| 59 | .prepare("SELECT number, blocked_by FROM issues WHERE repo_id = ? AND blocked_by <> '[]'") |
| 60 | .bind(&[repo_id.into()])?, |
| 61 | ]) |
| 62 | .await?; |
| 63 | let mut results = results.into_iter(); |
| 64 | let found: Vec<BriefRow> = match results.next() { |
| 65 | Some(result) => result.results()?, |
| 66 | None => Vec::new(), |
| 67 | }; |
| 68 | let edges: Vec<EdgeRow> = match results.next() { |
| 69 | Some(result) => result.results()?, |
| 70 | None => Vec::new(), |
| 71 | }; |
| 72 | let issues: HashSet<u32> = found.iter().map(|row| row.number).collect(); |
| 73 | let graph: HashMap<u32, Vec<u32>> = edges.into_iter().map(|row| (row.number, numbers(&row.blocked_by))).collect(); |
| 74 | Ok(match check_blocked_by(number, asked, &issues, &graph) { |
| 75 | Ok(set) => Outcome::Ok(set), |
| 76 | Err(message) => Outcome::fail(FailureCode::Invalid, message), |
| 77 | }) |
| 78 | } |
| 79 | |
| 80 | /// The `dependency.changed` that setting `issue`'s `blocked_by` to |
| 81 | /// `after` makes, worked out before the change so that its updates are |
| 82 | /// written with it; none when nothing changes. |
| 83 | pub(crate) async fn dependency_change(&self, issue: &Issue, after: &[u32], actor: &User) -> Result<Option<serde_json::Value>> { |
| 84 | let before = &issue.blocked_by; |
| 85 | let added: Vec<u32> = after.iter().filter(|n| !before.contains(n)).copied().collect(); |
| 86 | let removed: Vec<u32> = before.iter().filter(|n| !after.contains(n)).copied().collect(); |
| 87 | if added.is_empty() && removed.is_empty() { |
| 88 | return Ok(None); |
| 89 | } |
| 90 | let remaining = self.still_open(&issue.repo_id, after, None).await?; |
| 91 | let change = DependencyChanged { |
| 92 | repo_id: issue.repo_id.clone(), |
| 93 | number: issue.number, |
| 94 | issue: issue.number, |
| 95 | blocked_by: after.to_vec(), |
| 96 | added, |
| 97 | removed, |
| 98 | cleared: Vec::new(), |
| 99 | remaining, |
| 100 | merged_pull: None, |
| 101 | }; |
| 102 | Ok(Some(g1t_contracts::events::marked(change, Some(actor)))) |
| 103 | } |
| 104 | |
| 105 | /// Notes a change to `issue`'s `blocked_by` (from `before`) in the |
| 106 | /// issue's conversation. |
| 107 | pub(crate) async fn blocked_by_changed(&self, issue: &Issue, before: &[u32], actor: &User) -> Result<()> { |
| 108 | let added: Vec<u32> = issue.blocked_by.iter().filter(|n| !before.contains(n)).copied().collect(); |
| 109 | let removed: Vec<u32> = before.iter().filter(|n| !issue.blocked_by.contains(n)).copied().collect(); |
| 110 | if added.is_empty() && removed.is_empty() { |
| 111 | return Ok(()); |
| 112 | } |
| 113 | let list = |numbers: &[u32]| numbers.iter().map(|n| format!("#{n}")).collect::<Vec<_>>().join(", "); |
| 114 | let who = (actor.id.as_str(), actor.username.as_str()); |
| 115 | if !added.is_empty() { |
| 116 | self.note(&issue.repo_id, issue.number, who, &format!("marked this as blocked by {}", list(&added))).await?; |
| 117 | } |
| 118 | if !removed.is_empty() { |
| 119 | self.note(&issue.repo_id, issue.number, who, &format!("marked this as no longer blocked by {}", list(&removed))) |
| 120 | .await?; |
| 121 | } |
| 122 | Ok(()) |
| 123 | } |
| 124 | |
| 125 | /// Those of `numbers` whose issue is still open, leaving out `cleared` |
| 126 | /// (a blocker whose pull request just merged). |
| 127 | async fn still_open(&self, repo_id: &str, numbers: &[u32], cleared: Option<u32>) -> Result<Vec<u32>> { |
| 128 | if numbers.is_empty() { |
| 129 | return Ok(Vec::new()); |
| 130 | } |
| 131 | let open: Vec<BriefRow> = self |
| 132 | .db |
| 133 | .prepare( |
| 134 | "SELECT number, title, state FROM issues |
| 135 | WHERE repo_id = ? AND state = 'open' AND number IN (SELECT value FROM json_each(?))", |
| 136 | ) |
| 137 | .bind(&[repo_id.into(), serde_json::to_string(numbers)?.into()])? |
| 138 | .all() |
| 139 | .await? |
| 140 | .results()?; |
| 141 | Ok(numbers |
| 142 | .iter() |
| 143 | .filter(|number| Some(**number) != cleared && open.iter().any(|row| row.number == **number)) |
| 144 | .copied() |
| 145 | .collect()) |
| 146 | } |
| 147 | |
| 148 | /// An issue closed (`closed`) or had a pull request merged for it |
| 149 | /// (`merged_pull`): its claims end if it closed, and each open issue it |
| 150 | /// blocks hears that it cleared. Never fails what caused it: a problem |
| 151 | /// is logged. |
| 152 | pub(crate) async fn issue_resolved(&self, repo_id: &str, number: u32, actor: &User, closed: bool, merged_pull: Option<u32>) { |
| 153 | let resolved = async { |
| 154 | if closed { |
| 155 | self.release_claims_on_close(repo_id, number, Some(actor)).await?; |
| 156 | } |
| 157 | let dependents: Vec<EdgeRow> = self |
| 158 | .db |
| 159 | .prepare( |
| 160 | "SELECT number, blocked_by FROM issues |
| 161 | WHERE repo_id = ? AND state = 'open' |
| 162 | AND EXISTS (SELECT 1 FROM json_each(issues.blocked_by) WHERE json_each.value = ?)", |
| 163 | ) |
| 164 | .bind(&[repo_id.into(), number.into()])? |
| 165 | .all() |
| 166 | .await? |
| 167 | .results()?; |
| 168 | for dependent in dependents { |
| 169 | let blocked_by = numbers(&dependent.blocked_by); |
| 170 | let remaining = self.still_open(repo_id, &blocked_by, Some(number)).await?; |
| 171 | let change = DependencyChanged { |
| 172 | repo_id: repo_id.to_owned(), |
| 173 | number: dependent.number, |
| 174 | issue: dependent.number, |
| 175 | blocked_by, |
| 176 | cleared: vec![number], |
| 177 | remaining, |
| 178 | merged_pull, |
| 179 | ..DependencyChanged::default() |
| 180 | }; |
| 181 | // Follows from a close or merge already made: named by it, so |
| 182 | // a repeat of the same close or merge tells each session once. |
| 183 | let how = merged_pull.map_or_else(|| "closed".to_owned(), |pull| format!("merged:{pull}")); |
| 184 | let key = format!("{}:cleared:{number}:{how}", dependent.number); |
| 185 | let data = g1t_contracts::events::marked(change, Some(actor)); |
| 186 | self.publish_keyed("dependency.changed", repo_id, Some(actor.id.clone()), data, &key).await?; |
| 187 | } |
| 188 | Ok::<(), worker::Error>(()) |
| 189 | }; |
| 190 | if let Err(error) = resolved.await { |
| 191 | worker::console_error!("#{number} resolved, but its dependents were not told: {error}"); |
| 192 | } |
| 193 | } |
| 194 | |
| 195 | /// The issues `numbers` names, each with its active claims and its pull |
| 196 | /// requests, in the order given. |
| 197 | pub(crate) async fn blockers(&self, repo_id: &str, numbers: &[u32]) -> Result<Vec<Blocker>> { |
| 198 | if numbers.is_empty() { |
| 199 | return Ok(Vec::new()); |
| 200 | } |
| 201 | let list = serde_json::to_string(numbers)?; |
| 202 | let bind = |sql: String| -> Result<worker::D1PreparedStatement> { |
| 203 | self.db.prepare(sql).bind(&[repo_id.into(), JsValue::from(list.as_str())]) |
| 204 | }; |
| 205 | let results = self |
| 206 | .db |
| 207 | .batch(vec![ |
| 208 | bind("SELECT number, title, state FROM issues |
| 209 | WHERE repo_id = ? AND number IN (SELECT value FROM json_each(?))" |
| 210 | .to_owned())?, |
| 211 | bind(format!( |
| 212 | "SELECT {CLAIM_COLUMNS} FROM claims |
| 213 | WHERE repo_id = ? AND state = 'active' AND issue IN (SELECT value FROM json_each(?)) |
| 214 | ORDER BY created_at, id" |
| 215 | ))?, |
| 216 | bind("SELECT number, title, status, issue_number FROM pulls |
| 217 | WHERE repo_id = ? AND issue_number IN (SELECT value FROM json_each(?)) |
| 218 | ORDER BY number" |
| 219 | .to_owned())?, |
| 220 | ]) |
| 221 | .await?; |
| 222 | let mut results = results.into_iter(); |
| 223 | let mut next = || results.next(); |
| 224 | let issues: Vec<BriefRow> = match next() { |
| 225 | Some(result) => result.results()?, |
| 226 | None => Vec::new(), |
| 227 | }; |
| 228 | let claims: Vec<Claim> = match next() { |
| 229 | Some(result) => result.results::<ClaimRow>()?.into_iter().map(Claim::from).collect(), |
| 230 | None => Vec::new(), |
| 231 | }; |
| 232 | let pulls: Vec<PullBriefRow> = match next() { |
| 233 | Some(result) => result.results()?, |
| 234 | None => Vec::new(), |
| 235 | }; |
| 236 | Ok(numbers |
| 237 | .iter() |
| 238 | .filter_map(|number| { |
| 239 | let issue = issues.iter().find(|issue| issue.number == *number)?; |
| 240 | Some(Blocker { |
| 241 | number: issue.number, |
| 242 | title: issue.title.clone(), |
| 243 | state: issue.state, |
| 244 | claims: claims.iter().filter(|claim| claim.issue == *number).cloned().collect(), |
| 245 | pulls: pulls |
| 246 | .iter() |
| 247 | .filter(|pull| pull.issue_number == *number) |
| 248 | .map(|pull| PullBrief { number: pull.number, title: pull.title.clone(), status: pull.status }) |
| 249 | .collect(), |
| 250 | }) |
| 251 | }) |
| 252 | .collect()) |
| 253 | } |
| 254 | |
| 255 | /// The issues that name `number` in their `blocked_by`, by number. |
| 256 | pub(crate) async fn blocked_issues(&self, repo_id: &str, number: u32) -> Result<Vec<IssueBrief>> { |
| 257 | Ok(self |
| 258 | .db |
| 259 | .prepare( |
| 260 | "SELECT number, title, state FROM issues |
| 261 | WHERE repo_id = ? AND EXISTS (SELECT 1 FROM json_each(issues.blocked_by) WHERE json_each.value = ?) |
| 262 | ORDER BY number LIMIT 100", |
| 263 | ) |
| 264 | .bind(&[repo_id.into(), number.into()])? |
| 265 | .all() |
| 266 | .await? |
| 267 | .results::<BriefRow>()? |
| 268 | .into_iter() |
| 269 | .map(|row| IssueBrief { number: row.number, title: row.title, state: row.state }) |
| 270 | .collect()) |
| 271 | } |
| 272 | } |