| 1 | //! Announcing pull requests that start changing the same files. |
| 2 | //! |
| 3 | //! Each push works out what a pull request changes (reviews.rs) and so |
| 4 | //! which other open or draft pull requests it overlaps. A pair is announced |
| 5 | //! as `overlap.detected` the first time it overlaps, once for each side so |
| 6 | //! that whoever follows either hears of it, and remembered in |
| 7 | //! `overlap_pairs`; it is announced again only after a push shows it |
| 8 | //! stopped overlapping and it was forgotten. |
| 9 | |
| 10 | use g1t_contracts::time::rfc3339; |
| 11 | use g1t_contracts::work::*; |
| 12 | use g1t_kit::now_ms; |
| 13 | use serde::Deserialize; |
| 14 | use worker::Result; |
| 15 | |
| 16 | use crate::Work; |
| 17 | |
| 18 | #[derive(Deserialize)] |
| 19 | struct PairRow { |
| 20 | low: u32, |
| 21 | high: u32, |
| 22 | } |
| 23 | |
| 24 | /// A pair, lower number first. |
| 25 | fn pair(a: u32, b: u32) -> (u32, u32) { |
| 26 | (a.min(b), a.max(b)) |
| 27 | } |
| 28 | |
| 29 | impl Work { |
| 30 | /// Announces the overlaps `pull` (with its files just worked out) has |
| 31 | /// that were not announced before, and forgets the ones it no longer |
| 32 | /// has. Never fails the push it follows: a problem is logged. |
| 33 | pub(crate) async fn announce_overlaps(&self, pull: &Pull, actor: Option<String>) { |
| 34 | if let Err(error) = self.announce_overlaps_or_fail(pull, actor).await { |
| 35 | worker::console_error!("#{} overlaps were not announced: {error}", pull.number); |
| 36 | } |
| 37 | } |
| 38 | |
| 39 | async fn announce_overlaps_or_fail(&self, pull: &Pull, actor: Option<String>) -> Result<()> { |
| 40 | if !pull.status.is_active() { |
| 41 | return Ok(()); |
| 42 | } |
| 43 | let overlaps = self.overlaps(pull).await?; |
| 44 | let announced: Vec<u32> = self |
| 45 | .db |
| 46 | .prepare("SELECT low, high FROM overlap_pairs WHERE repo_id = ?1 AND (low = ?2 OR high = ?2)") |
| 47 | .bind(&[pull.repo_id.as_str().into(), pull.number.into()])? |
| 48 | .all() |
| 49 | .await? |
| 50 | .results::<PairRow>()? |
| 51 | .into_iter() |
| 52 | .map(|row| if row.low == pull.number { row.high } else { row.low }) |
| 53 | .collect(); |
| 54 | let current: Vec<u32> = overlaps.iter().map(|overlap| overlap.number).collect(); |
| 55 | let (new, cleared) = overlap_changes(¤t, &announced); |
| 56 | if new.is_empty() && cleared.is_empty() { |
| 57 | return Ok(()); |
| 58 | } |
| 59 | let at = rfc3339(now_ms()); |
| 60 | let mut statements = Vec::new(); |
| 61 | for other in &cleared { |
| 62 | let (low, high) = pair(pull.number, *other); |
| 63 | statements.push( |
| 64 | self.db |
| 65 | .prepare("DELETE FROM overlap_pairs WHERE repo_id = ? AND low = ? AND high = ?") |
| 66 | .bind(&[pull.repo_id.as_str().into(), low.into(), high.into()])?, |
| 67 | ); |
| 68 | } |
| 69 | // Each pair is noticed, and each side told, in one change. |
| 70 | let guard = "EXISTS (SELECT 1 FROM overlap_pairs WHERE repo_id = ?7 AND low = ?8 AND high = ?9 AND noticed_at = ?10)"; |
| 71 | let mut announced = Vec::new(); |
| 72 | for other in &new { |
| 73 | let (low, high) = pair(pull.number, *other); |
| 74 | let Some(overlap) = overlaps.iter().find(|overlap| overlap.number == *other) else { |
| 75 | continue; |
| 76 | }; |
| 77 | statements.push( |
| 78 | self.db |
| 79 | .prepare( |
| 80 | "INSERT INTO overlap_pairs (repo_id, low, high, files, noticed_at) VALUES (?, ?, ?, ?, ?) |
| 81 | ON CONFLICT (repo_id, low, high) DO NOTHING RETURNING low", |
| 82 | ) |
| 83 | .bind(&[ |
| 84 | pull.repo_id.as_str().into(), |
| 85 | low.into(), |
| 86 | high.into(), |
| 87 | serde_json::to_string(&overlap.paths)?.into(), |
| 88 | at.as_str().into(), |
| 89 | ])?, |
| 90 | ); |
| 91 | let index = statements.len() - 1; |
| 92 | let ours = OverlapDetected { |
| 93 | repo_id: pull.repo_id.clone(), |
| 94 | number: pull.number, |
| 95 | pull: pull.number, |
| 96 | other_pull: overlap.number, |
| 97 | files: overlap.paths.clone(), |
| 98 | issue: pull.issue, |
| 99 | other_issue: overlap.issue, |
| 100 | }; |
| 101 | let theirs = OverlapDetected { |
| 102 | number: overlap.number, |
| 103 | pull: overlap.number, |
| 104 | other_pull: pull.number, |
| 105 | issue: overlap.issue, |
| 106 | other_issue: pull.issue, |
| 107 | ..ours.clone() |
| 108 | }; |
| 109 | let mut sent = Vec::new(); |
| 110 | for event in [&ours, &theirs] { |
| 111 | let delivery = self.delivery( |
| 112 | "overlap.detected", |
| 113 | &pull.repo_id, |
| 114 | serde_json::to_value(event)?, |
| 115 | &format!("pulls:{low}:{high}:{at}:{}", event.pull), |
| 116 | guard, |
| 117 | vec![pull.repo_id.as_str().into(), low.into(), high.into(), at.as_str().into()], |
| 118 | )?; |
| 119 | sent.push(super::updates::push_delivery(&mut statements, delivery)); |
| 120 | } |
| 121 | announced.push((index, [ours, theirs], sent)); |
| 122 | } |
| 123 | let results = self.db.batch(statements).await?; |
| 124 | for (index, events, sent) in announced { |
| 125 | // Announced by the other side's push at the same moment. |
| 126 | match results.get(index) { |
| 127 | Some(result) if !result.results::<PairRow>()?.is_empty() => {} |
| 128 | _ => continue, |
| 129 | } |
| 130 | for (event, sent) in events.into_iter().zip(sent) { |
| 131 | self.delivered(sent, "overlap.detected", &pull.repo_id, actor.clone(), serde_json::to_value(event)?).await?; |
| 132 | } |
| 133 | } |
| 134 | Ok(()) |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | #[cfg(test)] |
| 139 | mod tests { |
| 140 | use super::*; |
| 141 | |
| 142 | #[test] |
| 143 | fn a_pair_is_the_same_from_either_side() { |
| 144 | assert_eq!(pair(9, 4), (4, 9)); |
| 145 | assert_eq!(pair(4, 9), (4, 9)); |
| 146 | } |
| 147 | } |