Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

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